Fix exact hotel deal ranking and cache concurrency

app/flight_sellers.py

@@ -14,7 +14,7 @@ list Google shows under a fare (`booking_options`), each with the seller name,
 its own price and a booking request URL.
 
 Cost discipline: a search and a seller lookup are billed separately. The app
-checks result rows through a bounded two-request queue, while this module caches
+checks result rows through a bounded request pool, while this module caches
 each seller table by booking token for SELLERS_TTL_S. A repeated row/search
 therefore reuses its exact quote rather than spending another upstream call.
 

app/hotel_sellers.py

@@ -34,10 +34,18 @@ import hashlib
 import json
 import os
 import re
+import tempfile
+import threading
 import time
+from contextlib import contextmanager
 from datetime import datetime
 from typing import Dict, List, Optional, Tuple
 
+try:
+    import fcntl
+except ImportError:  # pragma: no cover - production and supported dev are Unix
+    fcntl = None
+
 SERPAPI_URL = "https://serpapi.com/search"
 
 SELLERS_TTL_S = int(os.getenv("HOTEL_SELLERS_TTL_S", "3600"))          # 1 hour
@@ -50,8 +58,8 @@ SELLERS_CACHE = os.getenv("HOTEL_SELLERS_CACHE", "/tmp/rp_hotel_sellers_cache.js
 # "MakeMyTrip.com" unmapped, because the entry had been derived by the old code.
 # The key now carries this version, so changing the logic changes the key and
 # stale derivations can never be served as current.
-CACHE_VERSION = "2"
-HTTP_TIMEOUT_S = float(os.getenv("HOTEL_SELLERS_TIMEOUT_S", "30"))
+CACHE_VERSION = "3"   # v3: base rows carry tax basis; cache writes are serialized
+HTTP_TIMEOUT_S = float(os.getenv("HOTEL_SELLERS_TIMEOUT_S", "45"))
 
 def _norm(name: str) -> str:
     return re.sub(r"[^a-z0-9]", "", (name or "").lower())
@@ -100,6 +108,32 @@ def is_live() -> bool:
     return bool(os.getenv("SERPAPI_KEY")) and os.getenv("HOTEL_SELLERS_MOCK") != "1"
 
 
+_CACHE_THREAD_LOCK = threading.Lock()
+_CACHE_LOCK_FILE = f"{SELLERS_CACHE}.lock"
+
+
+@contextmanager
+def _cache_guard():
+    """Serialize JSON read/merge/write across threads and uvicorn workers."""
+    with _CACHE_THREAD_LOCK:
+        lock_fh = None
+        try:
+            lock_fh = open(_CACHE_LOCK_FILE, "a+", encoding="utf-8")
+            if fcntl is not None:
+                fcntl.flock(lock_fh.fileno(), fcntl.LOCK_EX)
+        except OSError:
+            if lock_fh is not None:
+                lock_fh.close()
+            lock_fh = None
+        try:
+            yield
+        finally:
+            if lock_fh is not None:
+                if fcntl is not None:
+                    fcntl.flock(lock_fh.fileno(), fcntl.LOCK_UN)
+                lock_fh.close()
+
+
 def _cache_load() -> Dict:
     try:
         with open(SELLERS_CACHE, "r", encoding="utf-8") as fh:
@@ -109,13 +143,40 @@ def _cache_load() -> Dict:
 
 
 def _cache_save(data: Dict) -> None:
+    temp_path = None
     try:
         now = time.time()
         pruned = {k: v for k, v in data.items() if now - v.get("at", 0) < SELLERS_TTL_S}
-        with open(SELLERS_CACHE, "w", encoding="utf-8") as fh:
+        cache_dir = os.path.dirname(SELLERS_CACHE) or "."
+        fd, temp_path = tempfile.mkstemp(prefix=".rp-hotel-cache-", dir=cache_dir)
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
             json.dump(pruned, fh)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(temp_path, SELLERS_CACHE)
+        temp_path = None
     except Exception:
         pass
+    finally:
+        if temp_path:
+            try:
+                os.unlink(temp_path)
+            except OSError:
+                pass
+
+
+def _cache_get(key: str) -> Optional[Dict]:
+    with _cache_guard():
+        return _cache_load().get(key)
+
+
+def _cache_put(key: str, payload: Dict) -> None:
+    # Merge into the latest image while holding the cross-worker lock. Without
+    # this, simultaneous whole-list quote calls overwrite one another's token.
+    with _cache_guard():
+        cache = _cache_load()
+        cache[key] = {"at": time.time(), "payload": payload}
+        _cache_save(cache)
 
 
 def _params(city: str, checkin: str, checkout: str, adults: int) -> Dict:
@@ -233,8 +294,8 @@ def search_hotels(
     out: List[Dict] = []
     nights = _nights(checkin, checkout)
     for p in (data.get("properties") or []):
-        total = _amount(p.get("total_rate"))
-        night = _amount(p.get("rate_per_night"))
+        total, total_taxed = _amount_taxed(p.get("total_rate"))
+        night, night_taxed = _amount_taxed(p.get("rate_per_night"))
         stay, derived = _stay_total(total, night, nights)
         if stay is None:
             continue          # neither a total nor enough to derive one
@@ -244,6 +305,7 @@ def search_hotels(
             # A per-night rate is multiplied out, never substituted.
             "price_inr": stay,
             "total_derived": derived,
+            "taxes_included": total_taxed if total is not None else night_taxed,
             "nightly_inr": night,
             "currency": "INR",
             "name": p.get("name") or "",
@@ -257,9 +319,11 @@ def search_hotels(
             "vendors": [],
             "book_url": p.get("link"),
         })
-        if len(out) >= max_results:
-            break
-    return out, "serpapi" if out else "serpapi (no properties)"
+    # Google orders properties by its relevance blend. RewardPilot's base list
+    # promises cheapest-stay-first, so sort explicitly before applying the scan
+    # cap; truncating first can discard a cheaper property later in the page.
+    out.sort(key=lambda h: float(h["price_inr"]))
+    return out[:max_results], "serpapi" if out else "serpapi (no properties)"
 
 
 def sellers_for(
@@ -282,8 +346,7 @@ def sellers_for(
     ck = hashlib.sha1(
         f"v{CACHE_VERSION}|{property_token}|{checkin}|{checkout}|{adults}".encode()
     ).hexdigest()[:20]
-    cache = _cache_load()
-    hit = cache.get(ck)
+    hit = _cache_get(ck)
     if hit and time.time() - hit.get("at", 0) < SELLERS_TTL_S:
         return dict(hit["payload"], source=hit["payload"].get("source", "serpapi") + " (cached)")
 
@@ -346,6 +409,5 @@ def sellers_for(
             default=min((r["price_inr"] for r in rows), default=None),
         ),
     }
-    cache[ck] = {"at": time.time(), "payload": payload}
-    _cache_save(cache)
+    _cache_put(ck, payload)
     return payload

app/main.py

@@ -631,12 +631,12 @@ def flights_sellers(req: FlightSellersRequest, request: Request = None):
     platform with the loudest discount.
 
     Billed per call upstream. The client checks a result list through a bounded
-    two-request queue and reuses this endpoint's token cache, so base fares stay
+    request pool and reuses this endpoint's token cache, so base fares stay
     interactive while exact post-offer ranking is assembled.
     """
     if request is not None:
         # One round-trip search can contain 20 outbound + 20 return rows. The
-        # client never runs more than two concurrently, but cache hits can finish
+        # client bounds concurrency, but cache hits can finish
         # quickly; allow one complete scan without self-rate-limiting it.
         _limit(request, "fls", rate=60)     # 60/min per IP per worker
     import flight_sellers
@@ -798,11 +798,12 @@ def hotels_sellers(req: HotelSellersRequest, request: Request = None):
     hotel itself sold at 11,025. No card reward closes a gap like that, which is
     why the platform has to be priced and not just the card.
 
-    Billed per call upstream, so the client fetches this for the property the
-    user picked, never for a whole result list.
+    Billed per call upstream. The client checks its capped result list through a
+    bounded queue and this endpoint caches by property/dates, so post-offer
+    ranking never borrows one property's rate table for another.
     """
     if request is not None:
-        _limit(request, "hts", rate=20)
+        _limit(request, "hts", rate=60)     # one capped whole-list scan + fast retry/min
     import hotel_sellers
     from channels import compare_channels, hotel_brand_key
     from availability import normalize_city