diff --git a/backend/api/content_routes.py b/backend/api/content_routes.py index 67e111d..cd682b1 100644 --- a/backend/api/content_routes.py +++ b/backend/api/content_routes.py @@ -11,7 +11,7 @@ from fastapi import APIRouter, HTTPException, Request, Response from api import content_payloads as payloads from api import sitemap as sitemap_mod -from api.payloads import history_token +from api.payloads import content_token from data import climate from data import cities from data import grid @@ -58,13 +58,21 @@ def _cached(request: Request, kind: str, cell_id: str, key: str, token: str, bui return _json_response(store.put_payload(kind, cell_id, key, token, payload), etag) -def _resolve_city(slug: str): - """(city, cell, history) for a slug, or raise 404 (unknown) / 503 (warming) -- - mirrors web/content.py's _resolve_city.""" +def _city_cell(slug: str): + """(city, cell) for a slug, or raise 404 — both steps are cheap (a dict + lookup and a pure grid snap) and, crucially, load no history, so the + derived-store token can be computed and a cache hit served without ever + touching the ~45-year archive.""" city = cities.get(slug) if city is None: raise HTTPException(status_code=404, detail="Unknown city.") - cell = grid.snap(city["lat"], city["lon"]) + return city, grid.snap(city["lat"], city["lon"]) + + +def _load_history(cell): + """The cell's ~45-year archive, or raise 503 while it's still warming -- + the expensive load that used to run on every request. Now called only from + the content handlers' build() closures, i.e. only on a derived-store miss.""" history = climate.load_cached_history(cell) if history is None or history.is_empty(): try: @@ -73,7 +81,7 @@ def _resolve_city(slug: str): history = None if history is None or history.is_empty(): raise HTTPException(status_code=503, detail="Climate data is warming up; please retry shortly.") - return city, cell, history + return history def _origin(request: Request) -> str: @@ -105,16 +113,17 @@ def content_home(request: Request): @router.get("/content/city/{slug}") def content_city(request: Request, slug: str): - city, cell, history = _resolve_city(slug) - recent = None - try: - recent = climate.get_recent_forecast(cell) - except Exception: # noqa: BLE001 - today_vs_normal degrades to None - pass - token = history_token(history) + city, cell = _city_cell(slug) + token = content_token(cell["id"]) origin = _origin(request) def build(): + history = _load_history(cell) + recent = None + try: + recent = climate.get_recent_forecast(cell) + except Exception: # noqa: BLE001 - today_vs_normal degrades to None + pass return payloads.city_payload(origin, BASE, city, history, recent) # origin is folded into the cache key (not just slug) because the payload's @@ -131,10 +140,11 @@ def content_city(request: Request, slug: str): def content_city_month(request: Request, slug: str, month: str): if month not in payloads.MONTH_INDEX: raise HTTPException(status_code=404, detail="Unknown month.") - city, cell, history = _resolve_city(slug) - token = history_token(history) + city, cell = _city_cell(slug) + token = content_token(cell["id"]) def build(): + history = _load_history(cell) return payloads.month_payload(BASE, city, history, payloads.MONTH_INDEX[month]) return _cached(request, "content-month", cell["id"], f"{slug}:{month}", token, build) @@ -142,11 +152,12 @@ def content_city_month(request: Request, slug: str, month: str): @router.get("/content/city/{slug}/records") def content_city_records(request: Request, slug: str): - city, cell, history = _resolve_city(slug) - token = history_token(history) + city, cell = _city_cell(slug) + token = content_token(cell["id"]) origin = _origin(request) def build(): + history = _load_history(cell) return payloads.records_payload(origin, BASE, city, history) # See content_city's comment: origin folded into the cache key for the diff --git a/backend/api/payloads.py b/backend/api/payloads.py index 420ab50..adcf9aa 100644 --- a/backend/api/payloads.py +++ b/backend/api/payloads.py @@ -73,6 +73,30 @@ def history_token(history) -> str: return f"{PAYLOAD_VER}:{hist_end(history)}" +# The content-shape version for the SEO content pages (/climate/[/month| +# /records]). Kept separate from PAYLOAD_VER so a content-only shape change need +# not orphan every other kind's cache, and vice-versa. Bump on change. +CONTENT_VER = "c1" + + +def content_token(cell_id: str) -> str: + """Validity for the SEO content-page payloads — cheap AND stable. + + Unlike history_token, this never loads the ~45-year archive: it keys on the + cell's newest archived DATE (climate.history_max_date — an indexed + ``MAX(date)`` on Postgres, a single-column read on the parquet backend), so it + survives the hourly tail top-ups (which only refresh intra-day freshness) and + turns over only when the archive's last day genuinely advances (≈1×/day). That + keeps content pages ≤1 day stale — acceptable for SEO — while sparing every + request the full-history load the old token forced even on a cache hit. + Fail-soft: a store/DB error yields a 'none' bucket rather than raising.""" + try: + max_date = climate.history_max_date(cell_id) + except Exception: # noqa: BLE001 - token computation must never fail a request + max_date = None + return f"{PAYLOAD_VER}:{CONTENT_VER}:{max_date or 'none'}" + + def recent_token(history, cell_id: str) -> str: """Validity for payloads that also grade the hourly recent/forecast bundle.""" return f"{history_token(history)}:{climate.recent_stamp(cell_id)}" diff --git a/backend/core/metrics.py b/backend/core/metrics.py index 36b085b..7652bb5 100644 --- a/backend/core/metrics.py +++ b/backend/core/metrics.py @@ -96,6 +96,12 @@ def classify_inbound(path: str, base: str = "") -> str: the app's mount prefix (e.g. ``/thermograph`` or ``""``). """ p = path or "/" + # The liveness probe (Dockerfile HEALTHCHECK, Caddy health_uri) is by far the + # highest-volume single path in prod (measured ~47% of the access log) and is + # never real traffic — same posture as the metrics check below. Never under + # BASE (see /healthz's own docstring in web/app.py), so check before stripping. + if p.rstrip("/") == "/healthz": + return "health" # The dashboard polls the metrics endpoint; never count it as traffic, whatever base # prefix it arrives under — e.g. a `/thermograph/api/v2/metrics` probe against a # root-served prod app would otherwise land in "other" and show as an inbound error. @@ -118,6 +124,13 @@ def classify_inbound(path: str, base: str = "") -> str: # category keeps record_inbound from double-counting every interaction. if seg == "event": return "event" + # The SSR content API (backend/api/content_routes.py) is called only by + # the frontend_ssr service's own api_client.py, never by a browser — a + # server-to-server hop, not a page view. Its own category is what lets + # that (measured ~32% of the access log on prod) be excluded from the + # access log below without also hiding real external traffic. + if seg == "content": + return "internal" return f"api:{seg}" if seg else "api:other" if p.endswith((".js", ".css", ".html", ".webmanifest", ".png", ".svg", ".ico", ".json", ".woff", ".woff2", ".map", ".txt", ".xml")): diff --git a/backend/data/climate.py b/backend/data/climate.py index 0a148ff..b198c23 100644 --- a/backend/data/climate.py +++ b/backend/data/climate.py @@ -862,6 +862,30 @@ def recent_stamp(cell_id: str) -> int: return 0 +def history_max_date(cell_id: str) -> "str | None": + """ISO date (YYYY-MM-DD) of the cell's newest cached archived day, or None when + nothing is cached — read WITHOUT loading the full multi-decade history. + + Backs the content-page cache token (api/payloads.content_token). On Postgres + it's an indexed ``MAX(date)`` over climate_history; on the parquet backend it's + the max of the cached file's date column (a single columnar read via + ``scan_parquet``, not a full frame load). Fail-soft: any error reads as None.""" + if climate_store.is_postgres(): + return climate_store.history_max_date(cell_id) + path = _cache_path(cell_id) + if not os.path.exists(path): + return None + try: + val = pl.scan_parquet(path).select(pl.col("date").max()).collect().item() + if val is None: + return None + if isinstance(val, datetime.datetime): # older files stored date as Datetime + val = val.date() + return val.isoformat() + except Exception: # noqa: BLE001 - a corrupt/absent cache reads as no max date + return None + + def get_recent_forecast(cell: dict) -> pl.DataFrame: """Recent observations + forward forecast, with humidity as absolute humidity (g/m³). Thin wrapper over the raw loader (see below).""" @@ -985,10 +1009,13 @@ _REVGEO_CACHE: dict[str, str | None] = {} _REVGEO_MIN_INTERVAL = 1.1 # seconds between successive Nominatim reverse calls _revgeo_last = 0.0 -_REVGEO_QUEUE: "queue.Queue[tuple[float, float, str, concurrent.futures.Future]]" = queue.Queue() +_REVGEO_QUEUE: "queue.Queue[tuple]" = queue.Queue() _REVGEO_WORKER_LOCK = threading.Lock() _revgeo_worker_started = False _REVGEO_WAIT_TIMEOUT = 10.0 # seconds a caller waits for ITS OWN request before giving up +_GEOCODE_WAIT_TIMEOUT = 35.0 # forward /geocode is a deliberate user search, not a map-pan + # enrichment -- worth a longer wait than reverse geocoding's, + # covering the 30s per-request timeout plus queue wait. def reverse_geocode_cached(lat: float, lon: float) -> tuple[bool, str | None]: @@ -1041,31 +1068,46 @@ def _fetch_revgeo_label(lat: float, lon: float) -> str | None: def _revgeo_worker() -> None: - """Drains _REVGEO_QUEUE one request at a time, forever. The sole caller of - _fetch_revgeo_label, so the ~1/sec pacing below is enforced just by doing the - work serially — no lock needed, since nothing else ever touches Nominatim. - Re-checks the cache before fetching (a request queued behind an identical one - is answered from what the earlier request just cached, no duplicate call), - and always finishes the fetch + persists the result even if the original - caller already gave up waiting (see reverse_geocode's timeout).""" + """Drains _REVGEO_QUEUE one job at a time, forever — both reverse ("revgeo") + and forward ("geocode") jobs, so the ~1/sec pacing below is enforced just by + doing the work serially on this one thread, no lock needed, since nothing + else ever touches Nominatim. Reverse jobs re-check the cache before fetching + (a request queued behind an identical one is answered from what the earlier + request just cached, no duplicate call) and always finish + persist even if + the original caller already gave up waiting (see reverse_geocode's timeout). + Forward jobs have no cache (see geocode_nominatim) but share the same pacer.""" global _revgeo_last while True: - lat, lon, key, fut = _REVGEO_QUEUE.get() + job = _REVGEO_QUEUE.get() + kind = job[0] try: - found, label = reverse_geocode_cached(lat, lon) - if not found: + if kind == "revgeo": + _, lat, lon, key, fut = job + found, label = reverse_geocode_cached(lat, lon) + if not found: + wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last) + if wait > 0: + time.sleep(wait) + label = _fetch_revgeo_label(lat, lon) + _revgeo_last = time.monotonic() + _REVGEO_CACHE[key] = label + store.put_revgeo(key, label) # survive restarts (a None label retries after its TTL) + if not fut.done(): + fut.set_result(label) + else: # "geocode" + _, name, count, fut = job wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last) if wait > 0: time.sleep(wait) - label = _fetch_revgeo_label(lat, lon) - _revgeo_last = time.monotonic() - _REVGEO_CACHE[key] = label - store.put_revgeo(key, label) # survive restarts (a None label retries after its TTL) - if not fut.done(): - fut.set_result(label) + try: + results = _fetch_geocode_forward(name, count) + finally: + _revgeo_last = time.monotonic() + if not fut.done(): + fut.set_result(results) except Exception: # noqa: BLE001 - never let a bad request kill the worker if not fut.done(): - fut.set_result(None) + fut.set_result(None if kind == "revgeo" else []) finally: _REVGEO_QUEUE.task_done() @@ -1105,44 +1147,28 @@ def reverse_geocode(lat: float, lon: float) -> str | None: return label _start_revgeo_worker() fut: "concurrent.futures.Future[str | None]" = concurrent.futures.Future() - _REVGEO_QUEUE.put((lat, lon, key, fut)) + _REVGEO_QUEUE.put(("revgeo", lat, lon, key, fut)) try: return fut.result(timeout=_REVGEO_WAIT_TIMEOUT) except concurrent.futures.TimeoutError: return None -def geocode_nominatim(name: str, count: int = 5) -> list[dict]: - """Forward place-name lookup via OpenStreetMap Nominatim (keyless). - - Replaces the former Open-Meteo geocoder. Nominatim covers the long tail the - local GeoNames index can't — neighbourhoods, postcodes, sub-1000-population - villages, and alternate/native-language spellings — so /geocode falls back to - it on a local miss. It carries no population, so results keep Nominatim's own - relevance order; name/admin/country map straight across. - - Shares the reverse geocoder's lock and ~1/sec pacing: both hit the same host, - so serializing them together keeps total Nominatim traffic under the usage - policy. Only the low-volume /geocode miss path reaches here — autocomplete - (/suggest) is served purely from the local index and never calls out. - """ - global _revgeo_last - with _REVGEO_LOCK: - wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last) - if wait > 0: - time.sleep(wait) - try: - r = _request( - "https://nominatim.openstreetmap.org/search", - {"q": name, "format": "jsonv2", "addressdetails": 1, - "limit": count, "accept-language": "en"}, - 30, - phase="geocode", - headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"}, - ) - rows = r.json() or [] - finally: - _revgeo_last = time.monotonic() +def _fetch_geocode_forward(name: str, count: int) -> list[dict]: + """The actual Nominatim forward-search HTTP call + result shaping. Called + ONLY from _revgeo_worker (never on a caller's thread), mirroring + _fetch_revgeo_label — but unlike that function, a bad response is allowed to + raise here; _revgeo_worker's own except clause turns it into an empty list, + same net effect without a second layer of exception-swallowing.""" + r = _request( + "https://nominatim.openstreetmap.org/search", + {"q": name, "format": "jsonv2", "addressdetails": 1, + "limit": count, "accept-language": "en"}, + 30, + phase="geocode", + headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"}, + ) + rows = r.json() or [] out = [] for g in rows: a = g.get("address", {}) or {} @@ -1163,3 +1189,28 @@ def geocode_nominatim(name: str, count: int = 5) -> list[dict]: "population": None, # Nominatim has no population; order is relevance-based }) return out + + +def geocode_nominatim(name: str, count: int = 5) -> list[dict]: + """Forward place-name lookup via OpenStreetMap Nominatim (keyless). + + Replaces the former Open-Meteo geocoder. Nominatim covers the long tail the + local GeoNames index can't — neighbourhoods, postcodes, sub-1000-population + villages, and alternate/native-language spellings — so /geocode falls back to + it on a local miss. It carries no population, so results keep Nominatim's own + relevance order; name/admin/country map straight across. + + Shares the reverse geocoder's worker thread and ~1/sec pacing: both hit the + same host, and both jobs are drained serially by the one thread in + _revgeo_worker, so there is exactly one writer of _revgeo_last — no lock + needed, same reasoning as reverse_geocode. Only the low-volume /geocode miss + path reaches here — autocomplete (/suggest) is served purely from the local + index and never calls out. + """ + _start_revgeo_worker() + fut: "concurrent.futures.Future[list[dict]]" = concurrent.futures.Future() + _REVGEO_QUEUE.put(("geocode", name, count, fut)) + try: + return fut.result(timeout=_GEOCODE_WAIT_TIMEOUT) or [] + except concurrent.futures.TimeoutError: + return [] diff --git a/backend/data/climate_store.py b/backend/data/climate_store.py index 19c883c..b6951fd 100644 --- a/backend/data/climate_store.py +++ b/backend/data/climate_store.py @@ -188,6 +188,29 @@ def recent_synced_at(cell_id: str) -> float: return 0.0 +def history_max_date(cell_id: str) -> "str | None": + """The ISO date (YYYY-MM-DD) of the cell's newest archived day, or None when + there is no cached history (or Postgres is off/unreachable). + + Backs the content-page cache token (api/payloads.content_token), so it must be + CHEAP — an indexed ``MAX(date)`` over the ``(cell_id, date)`` primary key (see + the 0002 migration), never a full-history load. Fail-soft: any error reads as + None, so the caller falls back to a 'none' bucket rather than raising.""" + try: + with _conn() as conn: + if conn is None: + return None + row = conn.execute( + "SELECT MAX(date) FROM climate_history WHERE cell_id = %s", + (cell_id,), + ).fetchone() + if not row or row[0] is None: + return None + return row[0].isoformat() + except Exception: # noqa: BLE001 + return None + + def cols_version(cell_id: str) -> int: """The stored schema version for a cell's history; 0 if absent.""" try: diff --git a/backend/deploy/entrypoint.sh b/backend/deploy/entrypoint.sh index 9b6f399..9baf4b3 100755 --- a/backend/deploy/entrypoint.sh +++ b/backend/deploy/entrypoint.sh @@ -85,4 +85,7 @@ if [ "${RUN_MIGRATIONS:-1}" != "0" ]; then fi echo "==> Starting uvicorn on 0.0.0.0:${PORT:-8137} with ${WORKERS:-4} worker(s)" -exec uvicorn app:app --host 0.0.0.0 --port "${PORT:-8137}" --workers "${WORKERS:-4}" +# --no-access-log: the app's own request-logging middleware (audit.log_access, +# web/app.py) already writes a structured line per request; uvicorn's own access +# log just duplicated every one of them for no benefit. +exec uvicorn app:app --host 0.0.0.0 --port "${PORT:-8137}" --workers "${WORKERS:-4}" --no-access-log diff --git a/backend/notifications/notify.py b/backend/notifications/notify.py index a26702f..b8fafbb 100644 --- a/backend/notifications/notify.py +++ b/backend/notifications/notify.py @@ -450,6 +450,37 @@ def _maybe_refresh_homepage() -> None: pass +# Pre-warming the SEO content derived-store rides this loop too, for the same +# reason the homepage sweep does: it's leader-only (the loop runs on the single +# elected leader) and cache-only, so it spends no upstream quota. The content +# token turns over when a cell's archive gains a day (~1x/day), so after each +# advance the first request to a /climate page would otherwise recompute a +# 45-year payload cold; warming rebuilds those rows off-request. It's paced and +# capped per tick (CONTENT_WARM_MAX_CITIES) so one tick can't stall the notifier, +# and the idempotent skip means a tick after everything is fresh is a cheap +# no-op — the ~1000-city set refreshes across a handful of ticks, well inside the +# <=1-day staleness the content token already tolerates. +CONTENT_WARM_INTERVAL = float(os.environ.get("THERMOGRAPH_CONTENT_WARM_INTERVAL", "1800")) # 30 min +CONTENT_WARM_MAX_CITIES = int(os.environ.get("THERMOGRAPH_CONTENT_WARM_MAX_CITIES", "50")) +_last_content_warm = 0.0 + + +def _maybe_warm_content() -> None: + global _last_content_warm + now = time.time() + if now - _last_content_warm < CONTENT_WARM_INTERVAL: + return + _last_content_warm = now + try: + # Local import: warm_cities is a script-style top-level module (mirrors + # api/internal_routes.py's warm-cities job), so only the leader pays its + # import cost, and it's kept off notify.py's import graph. + import warm_cities + warm_cities.warm_content(limit=CONTENT_WARM_MAX_CITIES) + except Exception: # noqa: BLE001 - warming is best-effort; pages self-heal on request + pass + + # The daily "most unusual right now" post to Discord rides this loop too: once per # day, after the feed is refreshed, leader-only like everything in run_loop. No # webhook configured => no-op. @@ -489,6 +520,7 @@ def run_loop(): try: run_pass() _maybe_refresh_homepage() + _maybe_warm_content() _maybe_post_discord() except Exception: # noqa: BLE001 - a bad iteration must never kill the loop pass diff --git a/backend/tests/api/test_content_routes.py b/backend/tests/api/test_content_routes.py new file mode 100644 index 0000000..236601a --- /dev/null +++ b/backend/tests/api/test_content_routes.py @@ -0,0 +1,130 @@ +"""Route tests for the SSR content JSON API (api/content_routes.py). + +The point of these is the latency fix: a derived-store cache HIT must be O(1) — +it must NOT load the ~45-year archive. So the weather layer is faked with call +counters, and the core assertion is that a second identical request loads the +history zero more times than the first. ETag/304 revalidation and the 404/payload +shapes are covered alongside, against a fresh per-test store. +""" +import pytest +from fastapi.testclient import TestClient + +from web import app as appmod +from data import climate +from data import cities + +BASE = "/thermograph/api/v2/content" + +CITY = { + "slug": "testville", + "name": "Testville", + "admin1": "Washington", + "country": "United States", + "country_code": "US", + "lat": 47.6062, + "lon": -122.3321, + "population": 100000, +} + + +@pytest.fixture +def counts(): + return {"load_cached_history": 0, "get_history": 0} + + +@pytest.fixture +def client(monkeypatch, history, recent, counts, tmp_store): + """TestClient with the weather + city layer faked and the history loaders + wrapped in counters. tmp_store gives each test a fresh derived store so the + miss->hit sequence is deterministic.""" + def load_cached_history(cell): + counts["load_cached_history"] += 1 + return history.clone() + + def get_history(cell): + counts["get_history"] += 1 + return history.clone(), {"cached": True, "cache_age_days": 3} + + monkeypatch.setattr(climate, "load_cached_history", load_cached_history) + monkeypatch.setattr(climate, "get_history", get_history) + monkeypatch.setattr(climate, "get_recent_forecast", lambda cell: recent.clone()) + # Stable token: content_token() -> payloads -> climate.history_max_date. + monkeypatch.setattr(climate, "history_max_date", lambda cell_id: "2026-07-01") + monkeypatch.setattr(climate, "reverse_geocode", lambda lat, lon: "Testville, Washington") + # Only "testville" is a known city; anything else -> 404. + monkeypatch.setattr(cities, "get", lambda slug: CITY if slug == "testville" else None) + return TestClient(appmod.app) + + +# --- the regression guard: a cache hit loads no history ---------------------- + +def test_cache_hit_does_not_load_history(client, counts): + """First request misses -> builds -> loads history exactly once. A second + identical request is served from the derived store and loads history ZERO + more times. This is the whole point of the fix.""" + r1 = client.get(f"{BASE}/city/testville") + assert r1.status_code == 200 + assert counts["load_cached_history"] == 1 + assert counts["get_history"] == 0 # cached load was non-empty, no live fetch + + r2 = client.get(f"{BASE}/city/testville") + assert r2.status_code == 200 + assert r2.content == r1.content # replayed from the store + assert counts["load_cached_history"] == 1 # unchanged: no history load on the hit + assert counts["get_history"] == 0 + + +def test_304_revalidation_loads_no_history(client, counts): + """If-None-Match on a stored payload -> 304 with no history load.""" + r1 = client.get(f"{BASE}/city/testville") + assert r1.status_code == 200 + etag = r1.headers["etag"] + before = counts["load_cached_history"] + + r304 = client.get(f"{BASE}/city/testville", headers={"If-None-Match": etag}) + assert r304.status_code == 304 + assert r304.headers["etag"] == etag + assert counts["load_cached_history"] == before # 304 loaded nothing + + +def test_month_and_records_hits_skip_history_load(client, counts): + """Same O(1)-hit guarantee for the month and records handlers.""" + for path in ("month/july", "records"): + counts["load_cached_history"] = 0 + assert client.get(f"{BASE}/city/testville/{path}").status_code == 200 + assert counts["load_cached_history"] == 1 + assert client.get(f"{BASE}/city/testville/{path}").status_code == 200 + assert counts["load_cached_history"] == 1 # second call is a store hit + + +# --- payload shapes ---------------------------------------------------------- + +def test_city_payload_shape(client): + body = client.get(f"{BASE}/city/testville").json() + assert body["city"]["slug"] == "testville" + assert body["canonical_path"] == "/climate/testville" + assert len(body["months"]) == 12 + assert body["today_vs_normal"] is not None # recent forecast folded in + + +def test_month_payload_shape(client): + body = client.get(f"{BASE}/city/testville/month/july").json() + assert body["month_slug"] == "july" + assert body["canonical_path"] == "/climate/testville/july" + + +def test_records_payload_shape(client): + body = client.get(f"{BASE}/city/testville/records").json() + assert body["canonical_path"] == "/climate/testville/records" + + +# --- 404s -------------------------------------------------------------------- + +def test_unknown_slug_is_404(client, counts): + for path in ("city/nope", "city/nope/month/july", "city/nope/records"): + assert client.get(f"{BASE}/{path}").status_code == 404 + assert counts["load_cached_history"] == 0 # never reached the archive + + +def test_unknown_month_is_404(client): + assert client.get(f"{BASE}/city/testville/month/smarch").status_code == 404 diff --git a/backend/tests/api/test_payloads.py b/backend/tests/api/test_payloads.py index 29ab153..0fdf428 100644 --- a/backend/tests/api/test_payloads.py +++ b/backend/tests/api/test_payloads.py @@ -39,6 +39,34 @@ def test_cache_identity_formats_are_pinned(history): assert payloads.history_token(history) == f"{payloads.PAYLOAD_VER}:{payloads.hist_end(history)}" +def test_content_token_is_stable_across_tail_topups(monkeypatch): + """The content token keys on the archive's newest DATE, not its row count/mtime, + so hourly tail top-ups that don't change the max date leave it unchanged — while + a genuinely-advanced last day turns it over.""" + date_box = {"v": "2026-06-15"} + monkeypatch.setattr(climate, "history_max_date", lambda cid: date_box["v"]) + first = payloads.content_token("1_2") + assert first == f"{payloads.PAYLOAD_VER}:{payloads.CONTENT_VER}:2026-06-15" + # Simulated hourly top-ups: same max date -> identical token every time. + assert payloads.content_token("1_2") == first + assert payloads.content_token("1_2") == first + # The archive's last day advances -> the token turns over. + date_box["v"] = "2026-06-16" + assert payloads.content_token("1_2") != first + assert payloads.content_token("1_2") == f"{payloads.PAYLOAD_VER}:{payloads.CONTENT_VER}:2026-06-16" + + +def test_content_token_fail_soft_buckets_to_none(monkeypatch): + """A store/DB error (or an uncached cell) never raises — it buckets to 'none'.""" + monkeypatch.setattr(climate, "history_max_date", lambda cid: None) + assert payloads.content_token("1_2") == f"{payloads.PAYLOAD_VER}:{payloads.CONTENT_VER}:none" + + def boom(cid): + raise RuntimeError("store down") + monkeypatch.setattr(climate, "history_max_date", boom) + assert payloads.content_token("1_2") == f"{payloads.PAYLOAD_VER}:{payloads.CONTENT_VER}:none" + + def test_recent_token_composes_history_and_stamp(history, monkeypatch): monkeypatch.setattr(climate, "recent_stamp", lambda cid: "stamp") assert payloads.recent_token(history, "1_2") == f"{payloads.history_token(history)}:stamp" diff --git a/backend/tests/core/test_metrics.py b/backend/tests/core/test_metrics.py index 010f246..d58ad47 100644 --- a/backend/tests/core/test_metrics.py +++ b/backend/tests/core/test_metrics.py @@ -40,6 +40,16 @@ def test_classify_inbound_categories(): # dashboard probing /thermograph/... against a root-served (base="") prod app. assert c("/thermograph/api/v2/metrics", "") == "metrics" assert c("/api/v2/metrics", "") == "metrics" + # The SSR content API (backend/api/content_routes.py) is a server-to-server + # hop from frontend_ssr's own api_client.py, never a browser -- its own + # category is what lets it be excluded from the access log without also + # hiding real external traffic under the generic "api:*" bucket. + assert c("/thermograph/api/v2/content/hub", "/thermograph") == "internal" + assert c("/api/v2/content/city/tokyo", "") == "internal" + # The liveness probe: never under BASE, whatever base the app happens to be + # mounted at (it's registered at a fixed path -- see web/app.py's healthz). + assert c("/healthz") == "health" + assert c("/healthz", "/thermograph") == "health" # pages, static, seo assert c("/thermograph/", "/thermograph") == "page" assert c("/thermograph/calendar", "/thermograph") == "page" diff --git a/backend/tests/data/test_climate.py b/backend/tests/data/test_climate.py index 95e869f..68d1491 100644 --- a/backend/tests/data/test_climate.py +++ b/backend/tests/data/test_climate.py @@ -615,3 +615,104 @@ def test_reverse_geocode_timeout_returns_none_without_blocking_the_caller(monkey elapsed = time_mod.monotonic() - t0 assert label is None assert elapsed < 0.2 # returned near the wait timeout, not after the 0.3s fetch + + +# --- forward geocode: shares the reverse-geocode worker, not a second lock ----- +# Regression coverage for the NameError _REVGEO_LOCK bug (geocode_nominatim +# referenced a lock that was removed when reverse geocoding moved to the +# worker/queue design, so every forward lookup 502'd in production). These +# exercise the real queue/worker plumbing rather than mocking geocode_nominatim +# itself away, so a reintroduced bare `with _REVGEO_LOCK:` or any other +# not-actually-defined-name bug fails loudly here instead of shipping unseen. + +def test_geocode_nominatim_resolves_via_the_worker_thread(monkeypatch): + monkeypatch.setattr( + climate, "_fetch_geocode_forward", + lambda name, count: [{"name": name, "admin1": None, "country": "Testland", + "country_code": "TL", "lat": 1.0, "lon": 2.0, + "population": None}], + ) + results = climate.geocode_nominatim("Nowheresville") + assert results[0]["name"] == "Nowheresville" + assert results[0]["country"] == "Testland" + + +def test_geocode_nominatim_timeout_returns_empty_list(monkeypatch): + """Same shape as reverse_geocode's timeout test: a caller waits only up to + _GEOCODE_WAIT_TIMEOUT, not as long as the fetch itself takes.""" + import time as time_mod + monkeypatch.setattr(climate, "_GEOCODE_WAIT_TIMEOUT", 0.05) + + def slow_fetch(name, count): + time_mod.sleep(0.3) + return [{"name": "Too Slow"}] + + monkeypatch.setattr(climate, "_fetch_geocode_forward", slow_fetch) + t0 = time_mod.monotonic() + results = climate.geocode_nominatim("anywhere") + elapsed = time_mod.monotonic() - t0 + assert results == [] + assert elapsed < 0.2 + + +def test_geocode_nominatim_a_bad_fetch_degrades_to_empty_list_not_a_crash(monkeypatch): + """_fetch_geocode_forward is allowed to raise (matches _fetch_revgeo_label's + contract loosely -- the worker's except clause is the actual safety net); + confirm a raising fetch never reaches the caller as an exception.""" + def boom(name, count): + raise RuntimeError("Nominatim is down") + monkeypatch.setattr(climate, "_fetch_geocode_forward", boom) + assert climate.geocode_nominatim("anywhere") == [] + + +def test_geocode_nominatim_shares_the_reverse_geocode_pacer(monkeypatch): + """Forward and reverse jobs are drained by the SAME worker thread off the + SAME queue, so a forward call advances _revgeo_last exactly like a reverse + one does -- this is what makes a second lock unnecessary.""" + monkeypatch.setattr(climate, "_revgeo_last", 0.0) + monkeypatch.setattr(climate, "_fetch_geocode_forward", lambda name, count: []) + before = climate._revgeo_last + climate.geocode_nominatim("anywhere") + assert climate._revgeo_last > before + + +def test_fetch_geocode_forward_parses_nominatim_response(monkeypatch): + """Executes _fetch_geocode_forward's real body (the function the NameError + bug prevented from ever running) against a stubbed HTTP transport -- not a + monkeypatch of geocode_nominatim itself, unlike the API-layer tests.""" + class _FakeResp: + def json(self): + return [ + {"name": "West Seattle", "lat": "47.57", "lon": "-122.38", + "display_name": "West Seattle, Seattle, King County, Washington, United States", + "address": {"suburb": "West Seattle", "city": "Seattle", + "state": "Washington", "country": "United States", + "country_code": "us"}}, + # No `name`, no recognized address component -- falls back to the + # head of display_name, exercising that branch too. + {"lat": "51.5", "lon": "-0.1", + "display_name": "Some Unnamed Place, Greater London, England", + "address": {"country": "United Kingdom", "country_code": "gb"}}, + ] + + captured = {} + + def fake_request(url, params, timeout, *, phase, headers=None, attempts=climate.MAX_ATTEMPTS): + captured["url"] = url + captured["params"] = params + captured["phase"] = phase + return _FakeResp() + + monkeypatch.setattr(climate, "_request", fake_request) + results = climate._fetch_geocode_forward("west seattle", 5) + + assert captured["url"] == "https://nominatim.openstreetmap.org/search" + assert captured["params"]["q"] == "west seattle" + assert captured["phase"] == "geocode" + + assert results[0] == { + "name": "West Seattle", "admin1": "Washington", "country": "United States", + "country_code": "US", "lat": 47.57, "lon": -122.38, "population": None, + } + assert results[1]["name"] == "Some Unnamed Place" + assert results[1]["country_code"] == "GB" diff --git a/backend/tests/data/test_climate_store.py b/backend/tests/data/test_climate_store.py index a6fbcd3..09eaa49 100644 --- a/backend/tests/data/test_climate_store.py +++ b/backend/tests/data/test_climate_store.py @@ -69,6 +69,17 @@ def test_recent_backend_roundtrips_through_parquet(monkeypatch, tmp_path): assert hit is not None and hit[0].height == 3 +def test_history_max_date_reads_parquet_tail_without_full_load(monkeypatch, tmp_path): + monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) + cell_id = "9_10" + assert climate.history_max_date(cell_id) is None # nothing cached -> None + climate._write_history_backed(cell_id, _hist_frame(n=5)) # dates 1995-01-01..05 + assert climate.history_max_date(cell_id) == "1995-01-05" + # A tail top-up that appends a newer day advances the max date. + climate._write_history_backed(cell_id, _hist_frame(n=7)) + assert climate.history_max_date(cell_id) == "1995-01-07" + + # --- gated Postgres integration --------------------------------------------- @pytest.fixture @@ -140,6 +151,15 @@ def test_pg_recent_stamp_advances_only_on_write(pg_store): assert pg_store.history_synced_at("1_2") == 6000.0 +def test_pg_history_max_date(pg_store): + assert pg_store.history_max_date("1_2") is None # no rows -> None + pg_store.write_history("1_2", _hist_frame(n=5), 1000.0) # dates 1995-01-01..05 + assert pg_store.history_max_date("1_2") == "1995-01-05" + # Another cell's rows must not leak into this cell's max. + pg_store.write_history("3_4", _hist_frame(n=9), 1000.0) + assert pg_store.history_max_date("1_2") == "1995-01-05" + + def test_pg_cols_version_gate(pg_store): pg_store.write_history("1_2", _hist_frame(n=2), 1000.0) assert pg_store.read_history("1_2") is not None diff --git a/backend/tests/test_warm_content.py b/backend/tests/test_warm_content.py new file mode 100644 index 0000000..ca8a95c --- /dev/null +++ b/backend/tests/test_warm_content.py @@ -0,0 +1,131 @@ +"""warm_cities.warm_content -- the off-request pre-warm that rebuilds the SEO +content derived-store (api/content_routes.py's content-city / content-month / +content-records rows) so the first request after each daily archive advance hits +a warm cache instead of a cold 45-year recompute. + +The invariants under test mirror the ones content_routes.py relies on: + * it populates all three kinds under the exact (kind, key, token) the routes use; + * a city with no cached archive is skipped and never triggers an upstream fetch + (warming must not spend quota); + * it is idempotent -- a second run with nothing changed writes zero payloads. +""" +import warm_cities +from api import content_payloads +from api import payloads as cache_ids +from data import cities as cities_mod +from data import climate +from data import grid +from data import store + + +# A fixed archive last-date -> a stable content token across a test's runs, so the +# idempotency check exercises the "already fresh for this token" fast path. +_MAX_DATE = "2026-07-20" + + +def _two_cities(): + """Two real curated cities (so the payload builders get the fields they read), + the first with a cached archive, the second without.""" + everyone = cities_mod.all_cities() + return everyone[0], everyone[1] + + +def _wire(monkeypatch, city_with_history, history): + """Point warm_content's cache-only reads at fixtures and make any upstream + fetch a hard failure, so a test proves warming never reaches the network.""" + have_cell = grid.snap(city_with_history["lat"], city_with_history["lon"])["id"] + + monkeypatch.setattr(climate, "history_max_date", lambda cell_id: _MAX_DATE) + + def fake_cached_history(cell): + return history if cell["id"] == have_cell else None + + monkeypatch.setattr(climate, "load_cached_history", fake_cached_history) + monkeypatch.setattr(climate, "load_cached_recent_forecast", lambda cell: None) + + def _no_upstream(*a, **k): # pragma: no cover - only fires on a regression + raise AssertionError("warm_content must not fetch upstream") + + monkeypatch.setattr(climate, "get_history", _no_upstream) + monkeypatch.setattr(climate, "get_recent_forecast", _no_upstream) + + +def test_warms_three_kinds_with_matching_key_and_token(tmp_store, monkeypatch, history): + warm, cold = _two_cities() + _wire(monkeypatch, warm, history) + monkeypatch.setattr(cities_mod, "all_cities", lambda: [warm, cold]) + + result = warm_cities.warm_content(origin="https://thermograph.org") + + origin = "https://thermograph.org" + cell_id = grid.snap(warm["lat"], warm["lon"])["id"] + token = cache_ids.content_token(cell_id) + slug = warm["slug"] + + # content-city + content-records under {slug}:{origin}, 12 months under {slug}:{month} + assert store.get_json("content-city", cell_id, f"{slug}:{origin}", token) is not None + assert store.get_json("content-records", cell_id, f"{slug}:{origin}", token) is not None + for month in content_payloads.MONTH_INDEX: + assert store.get_json("content-month", cell_id, f"{slug}:{month}", token) is not None + + # 1 city + 1 records + 12 months = 14 payloads for the one city with history. + assert result["built"] == 14 + assert result["warmed"] == 1 + + +def test_skips_city_without_cached_history_and_never_fetches(tmp_store, monkeypatch, history): + warm, cold = _two_cities() + _wire(monkeypatch, warm, history) + monkeypatch.setattr(cities_mod, "all_cities", lambda: [warm, cold]) + + result = warm_cities.warm_content(origin="https://thermograph.org") + + # The archive-less city produced no rows (and _no_upstream never fired, or the + # call above would have raised). + cold_cell = grid.snap(cold["lat"], cold["lon"])["id"] + token = cache_ids.content_token(cold_cell) + assert store.get_json("content-city", cold_cell, f"{cold['slug']}:https://thermograph.org", token) is None + assert result["empty"] == 1 + + +def test_idempotent_second_run_writes_nothing(tmp_store, monkeypatch, history): + warm, cold = _two_cities() + _wire(monkeypatch, warm, history) + monkeypatch.setattr(cities_mod, "all_cities", lambda: [warm, cold]) + + # Spy on put_payload while still persisting, so run 2 sees run 1's rows. + calls = [] + real_put = store.put_payload + + def spy(kind, cell_id, key, token, payload, cache=True): + calls.append((kind, cell_id, key, token)) + return real_put(kind, cell_id, key, token, payload, cache) + + monkeypatch.setattr(store, "put_payload", spy) + + first = warm_cities.warm_content(origin="https://thermograph.org") + assert len(calls) == 14 # one full city warmed + assert first["built"] == 14 + + calls.clear() + second = warm_cities.warm_content(origin="https://thermograph.org") + assert calls == [] # nothing rebuilt: every kind still fresh + assert second["built"] == 0 + assert second["warmed"] == 0 + assert second["skipped"] == 1 # the warm city skipped cheaply + assert second["empty"] == 1 # the archive-less city still empty + + +def test_limit_caps_cities_built_per_call(tmp_store, monkeypatch, history): + # Two cities, both with a cached archive; limit=1 should build exactly one. + a, b = _two_cities() + have = {grid.snap(a["lat"], a["lon"])["id"], grid.snap(b["lat"], b["lon"])["id"]} + monkeypatch.setattr(climate, "history_max_date", lambda cell_id: _MAX_DATE) + monkeypatch.setattr(climate, "load_cached_history", + lambda cell: history if cell["id"] in have else None) + monkeypatch.setattr(climate, "load_cached_recent_forecast", lambda cell: None) + monkeypatch.setattr(cities_mod, "all_cities", lambda: [a, b]) + + result = warm_cities.warm_content(limit=1, origin="https://thermograph.org") + assert result["warmed"] == 1 + assert result["built"] == 14 diff --git a/backend/tests/web/test_request_logging.py b/backend/tests/web/test_request_logging.py new file mode 100644 index 0000000..b14882b --- /dev/null +++ b/backend/tests/web/test_request_logging.py @@ -0,0 +1,87 @@ +"""The request-logging middleware (web/app.py's revalidate_static): what gets +counted, what gets written to the access log, and what gets truncated before +it's persisted. Companion to core/test_metrics.py's classify_inbound tests -- +these exercise the actual ASGI request path, not just the classifier function. +""" +import pytest +from fastapi.testclient import TestClient + +from web import app as appmod + +metrics = appmod.metrics # the exact module object app.py's own calls use -- +audit = appmod.audit # see the note on module-reload isolation below. + + +@pytest.fixture +def client(): + return TestClient(appmod.app) + + +def _patch_log_access(monkeypatch): + calls = [] + monkeypatch.setattr(audit, "log_access", lambda record: calls.append(record)) + return calls + + +def test_healthz_is_never_audited(monkeypatch, client): + """/healthz is ~47% of prod's daily access log; it must never even reach + audit.log_access (not just be cheap once there -- see classify_inbound's + "health" category and the exclusion in revalidate_static).""" + calls = _patch_log_access(monkeypatch) + r = client.get("/healthz") + assert r.status_code == 200 + assert calls == [] + + +def test_internal_category_is_never_audited(monkeypatch, client): + """The SSR content API's own category ("internal") is excluded from the + access log the same way -- it's a server-to-server hop (frontend_ssr's + api_client.py calling backend), not a page view.""" + calls = _patch_log_access(monkeypatch) + monkeypatch.setattr(metrics, "classify_inbound", lambda path, base: "internal") + r = client.get("/thermograph/api/version") + assert r.status_code == 200 + assert calls == [] + + +def test_access_log_ip_is_truncated_not_raw(monkeypatch, client): + calls = _patch_log_access(monkeypatch) + r = client.get("/thermograph/api/version", + headers={"X-Forwarded-For": "203.0.113.77, 10.0.0.1"}) + assert r.status_code == 200 + assert len(calls) == 1 + assert calls[0]["ip"] == "203.0.113.0" # /24, not the exact address + assert calls[0]["ip"] != "203.0.113.77" + + +def test_loggable_ip_truncation(): + assert appmod._loggable_ip("203.0.113.77") == "203.0.113.0" + assert appmod._loggable_ip("2001:db8:1234:5678::1") == "2001:db8:1234::" + assert appmod._loggable_ip(None) is None + assert appmod._loggable_ip("") == "" + assert appmod._loggable_ip("not-an-ip") == "not-an-ip" # best-effort, never raises + + +def test_rate_limiter_still_sees_the_full_precision_ip(monkeypatch, client): + """_client_ip (which feeds both the access log and the event rate limiter) + is never itself truncated -- only what audit.log_access persists is. A + /24-truncated rate-limit key would let one abuser exhaust a whole NAT'd + office's quota, which is exactly what must NOT happen here.""" + seen = {} + monkeypatch.setattr(metrics, "record_event", lambda name, **kw: seen.update(kw)) + r = client.post("/thermograph/api/v2/event", json={"event": "home.locate"}, + headers={"X-Forwarded-For": "203.0.113.77"}) + assert r.status_code == 204 + assert seen["ip"] == "203.0.113.77" + + +def test_event_route_records_end_to_end(client): + """Full HTTP round trip through the real ASGI route -- confirms the write + path backend/api/event -> metrics.record_event -> the counters store works + end to end (the existing suite only ever called record_event directly).""" + before = metrics.snapshot()["events_total"] + r = client.post("/thermograph/api/v2/event", json={"event": "home.locate"}) + assert r.status_code == 204 + after = metrics.snapshot() + assert after["events_total"] == before + 1 + assert after["events"]["home.locate"]["direct"] # no Referer sent -> "direct" diff --git a/backend/warm_cities.py b/backend/warm_cities.py index 199a726..f667968 100644 --- a/backend/warm_cities.py +++ b/backend/warm_cities.py @@ -19,13 +19,21 @@ import os import sys import time +from api import content_payloads from api import homepage +from api import payloads as cache_ids from core import singleton from data import cities from data import climate from data import grid +from data import store import paths +# The deployment base path, derived exactly as api/content_routes.py does so the +# breadcrumb hrefs / jsonld URLs the warmer bakes match what a live request bakes. +_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/") +BASE = f"/{_BASE}" if _BASE else "" + # core/singleton.py's flock guard, reused here for cross-*process* (not # cross-worker) exclusion: the first invocation holds this for its process # lifetime; a second one (an overlapping deploy) fails the non-blocking flock @@ -67,6 +75,100 @@ def main(limit: int | None = None, pace: float = 2.0) -> None: print(f"homepage feed: FAILED {e}") +# --- content derived-store pre-warm ------------------------------------------- +# The SEO /climate/[/month|/records] pages render their JSON from the +# derived store (api/content_routes.py), keyed by the cheap+stable content token +# (payloads.content_token -> PAYLOAD_VER:CONTENT_VER:archive_last_date). That token +# turns over only when a cell's archive gains a new day (~1x/day), so on the first +# request after each daily advance the page would otherwise recompute a 45-year +# payload cold. Pre-warming rebuilds those rows off-request — cache-only, so it +# never spends upstream quota — so users and crawlers land on a warm cache. + + +def _content_specs(slug: str, origin: str) -> "list[tuple[str, str, str, int | None]]": + """The (store-kind, store-key, builder-tag, month_idx) tuples for one city, + matching api/content_routes.py's kinds/keys exactly: content-city and + content-records key on ``{slug}:{origin}`` (the origin is folded in because the + payloads' jsonld.url is origin-qualified), content-month keys on + ``{slug}:{month}`` for each of the 12 months.""" + specs = [ + ("content-city", f"{slug}:{origin}", "city", None), + ("content-records", f"{slug}:{origin}", "records", None), + ] + for month, idx in content_payloads.MONTH_INDEX.items(): + specs.append(("content-month", f"{slug}:{month}", "month", idx)) + return specs + + +def _build_content(tag: str, city: dict, history, recent, origin: str, month_idx: "int | None") -> dict: + if tag == "city": + return content_payloads.city_payload(origin, BASE, city, history, recent) + if tag == "records": + return content_payloads.records_payload(origin, BASE, city, history) + return content_payloads.month_payload(BASE, city, history, month_idx) + + +def warm_content(limit: int | None = None, origin: str | None = None, pace: float = 0.05) -> dict: + """Rebuild any missing/stale content-page payloads in the derived store. + + For each curated city, compute the current content token (cheap — no history + load) and check whether every content kind (city, 12 months, records) already + has a row for that token. Cities that are wholly fresh are skipped *without* + loading the archive, so a re-run after everything is warm is near-instant and + only cells whose archive genuinely advanced (new token) do work — the same + idempotent contract as ``main``. + + Cache-only, exactly like ``main``: the history and recent/forecast bundles are + read from the cache; a cell with no cached archive is skipped rather than + fetched, so warming never spends a single upstream request. + + ``limit`` caps the number of cities actually (re)built this call (not scanned), + so a caller can bound one pass's wall-clock and rely on the idempotent skip to + resume from where it left off on the next call. Returns a small counts dict. + """ + if origin is None: + origin = os.environ.get("THERMOGRAPH_BASE_URL", "https://thermograph.org").rstrip("/") + built = warmed = skipped = empty = failed = 0 + for c in cities.all_cities(): + if limit is not None and warmed >= limit: + break + slug = c["slug"] + cell = grid.snap(c["lat"], c["lon"]) + cell_id = cell["id"] + token = cache_ids.content_token(cell_id) + specs = _content_specs(slug, origin) + # Cheap idempotency gate: which kinds are missing/stale for this token? + # get_payload is a cheap keyed lookup; loading the 45-year archive is not, + # so we defer that until we know at least one kind actually needs building. + stale = [s for s in specs if store.get_payload(s[0], cell_id, s[1], token) is None] + if not stale: + skipped += 1 + continue + history = climate.load_cached_history(cell) + if history is None or history.is_empty(): + # No cached archive yet -> nothing to build from, and warming must not + # fetch upstream. The cell self-heals on its first live request. + empty += 1 + continue + # today_vs_normal on the city payload needs the recent/forecast bundle; + # read it cache-only (never fetch) — None just drops that one block. + recent = climate.load_cached_recent_forecast(cell) + did_work = False + for kind, key, tag, month_idx in stale: + try: + payload = _build_content(tag, c, history, recent, origin, month_idx) + store.put_payload(kind, cell_id, key, token, payload) + built += 1 + did_work = True + except Exception: # noqa: BLE001 - one bad payload must not abort the sweep + failed += 1 + if did_work: + warmed += 1 + if pace: + time.sleep(pace) + return {"warmed": warmed, "built": built, "skipped": skipped, "empty": empty, "failed": failed} + + if __name__ == "__main__": if not singleton.claim(LOCK_PATH): # Not an error: an overlapping run just means a previous deploy's warm is diff --git a/backend/web/app.py b/backend/web/app.py index 0253319..b8a5d35 100644 --- a/backend/web/app.py +++ b/backend/web/app.py @@ -3,6 +3,7 @@ import contextlib import datetime import hashlib import hmac +import ipaddress import json import os import queue @@ -263,13 +264,34 @@ def api_version(): def _client_ip(request) -> "str | None": """The real client IP: the left-most X-Forwarded-For hop when a proxy fronts us - (Caddy on prod sets it), otherwise the direct peer address (LAN dev).""" + (Caddy on prod sets it), otherwise the direct peer address (LAN dev). Full + precision — the rate limiter (metrics._rate_ok) needs the exact address, since + a truncated key would let one abuser exhaust a whole NAT'd office's quota. + Truncate at the point of persistence instead (see _loggable_ip).""" xff = request.headers.get("x-forwarded-for") if xff: return xff.split(",")[0].strip() return request.client.host if request.client else None +def _loggable_ip(ip: "str | None") -> "str | None": + """Coarsen a client IP before it's written to the access log: /24 for IPv4, + /48 for IPv6. Keeps rough geo/abuse signal without keeping a full, joinable + address sitting in structured logs for the log's 30-day retention -- the + public privacy page promises IPs aren't logged beyond normal request handling, + and a raw address shipped to Loki was a live gap against that. Never the + input to the rate limiter (_client_ip's callers pass the untruncated value + there) -- this is only what gets persisted.""" + if not ip: + return ip + try: + addr = ipaddress.ip_address(ip) + except ValueError: + return ip # not a parseable address (e.g. a test/placeholder value) -- pass through + prefix = 24 if addr.version == 4 else 48 + return str(ipaddress.ip_network(f"{addr}/{prefix}", strict=False).network_address) + + @app.middleware("http") async def revalidate_static(request, call_next): """Serve the frontend with no-cache (NOT no-store): browsers may keep a copy @@ -288,11 +310,15 @@ async def revalidate_static(request, call_next): # loop so one request's instrumentation can't stall every other request # this worker is serving concurrently. await run_in_threadpool(metrics.record_inbound, cat, response.status_code) - # Retain per-request client IPs for later analysis; skip static assets and the - # dashboard's own metrics polling to keep the log to real, meaningful traffic. - if cat not in ("static", "metrics", "event"): + # Retain per-request client IPs for later analysis; skip static assets, the + # dashboard's own metrics polling, the liveness probe (health -- by far the + # highest-volume single path, and never real traffic), and the internal + # SSR->API hop (internal -- frontend_ssr's own server-to-server calls, not + # a page view) to keep the log to real, meaningful traffic. The IP itself is + # truncated (see _loggable_ip) -- coarse geo/abuse signal, not a full address. + if cat not in ("static", "metrics", "event", "health", "internal"): await run_in_threadpool(audit.log_access, { - "ip": _client_ip(request), "method": request.method, + "ip": _loggable_ip(_client_ip(request)), "method": request.method, "path": path, "status": response.status_code, "cat": cat}) # Threshold-gated slow-request line: RunAudit only times the 7 graded # endpoints, so this is the only latency signal for auth/account/content diff --git a/infra/ops/ICEBERG-HANDOFF.md b/infra/ops/ICEBERG-HANDOFF.md new file mode 100644 index 0000000..d27810b --- /dev/null +++ b/infra/ops/ICEBERG-HANDOFF.md @@ -0,0 +1,133 @@ +# Iceberg: implementing the agent-queryable service + +Handoff spec for whoever builds the Iceberg layer. The Postgres equivalent is +already shipped (`infra/ops/dbq.sh`) — **mirror it**. This document is the +contract, the environment facts you'll trip over, and how the result gets +verified. + +## Definition of done + +`infra/ops/iceberg.sh` exists and behaves like `dbq.sh`: + +```sh +infra/ops/iceberg.sh prod -c "select count(*) from ." +infra/ops/iceberg.sh dev -c "show tables" +echo "select 1" | infra/ops/iceberg.sh beta -f - +``` + +An agent (or operator) can query Iceberg in any environment, read-only, with no +new network exposure and no interactive steps. + +## The contract (non-negotiable) + +1. **Same interface shape as `dbq.sh`** — `iceberg.sh [flags] + ""`. Extra flags pass through to the engine; **stdin is forwarded** so + `-f -` works. Non-interactive, deterministic, safe to call from CI. +2. **Read-only enforced by the system, not by convention.** A dedicated + read-only identity — not an admin/superuser credential. You must be able to + *demonstrate* a refused write (see Acceptance). +3. **No new network endpoints.** Run the engine where the data is already + reachable (containerised), rather than publishing ports or opening firewall + rules. See the prod constraint below — this is not negotiable, it's physics. +4. **Env-keyed dispatch, names resolved at call time.** Prod is Docker Swarm; + task container names change on **every redeploy**. Resolve via + `docker ps --filter name=…`; never hardcode a container name. +5. **No secrets in the repo** or in workflow files. + +## Environment facts you need + +- Monorepo `emi/thermograph` (domain dirs: `backend/ frontend/ infra/ + observability/`). Ops tooling lives in `infra/ops/`. +- Environments: **dev** = LAN compose stack on the dev machine; **beta** = + `75.119.132.91`; **prod** = `169.58.46.181`. SSH as `agent` with + `~/.ssh/thermograph_agent_ed25519` (passwordless sudo on both boxes). +- **Prod runs Docker Swarm.** Its app database sits on the overlay network + `thermograph_internal` (`10.0.2.0/24`), which **the prod host itself cannot + route to**. Consequence, learned the hard way: `ssh -L` tunnelling works for + beta but is *impossible* for prod. Any design that depends on a tunnel or a + published port will fail on prod — that's why `dbq.sh` execs into the + container instead. Assume the same for anything you deploy there. +- The overlay is `attachable=true`, and the **dev machine is a Swarm worker** in + the prod cluster (NodeAddr `10.10.0.3`) — so `docker run --network + thermograph_internal …` from the dev box is a plausible route to prod-side + services. Untested; verify before relying on it. +- WireGuard mesh: prod `10.10.0.1`, beta `10.10.0.2`, dev `10.10.0.3`. +- `rclone` and `age` are already installed on prod and beta. +- Reference implementation to copy: **`infra/ops/dbq.sh`** + `infra/ops/README.md`. + +## Storage — almost certainly your warehouse + +Contabo Object Storage (S3-compatible), already provisioned and in use: + +- Endpoint `https://eu2.contabostorage.com` · bucket `era5-thermograph` · + region `default` · ~2 TB · **private** (keep it that way). +- ⚠️ **Contabo requires PATH-STYLE addressing.** Set + `force_path_style=true` / `s3.path-style-access=true` (whatever your engine's + FileIO calls it) and `region=default`. Virtual-host-style addressing **fails**. + This is validated, not theoretical. +- ⚠️ **The `backups/` prefix is in use** by the nightly backup jobs (prod DB + + Forgejo). Put Iceberg data under its own prefix (e.g. `iceberg/` or + `warehouse/`). Do not write outside your prefix. +- Credentials already exist — **reuse them, don't mint new ones silently**: + - SOPS vault: `infra/deploy/secrets/{prod,beta}.yaml` as `THERMOGRAPH_S3_*` + (rendered to `/etc/thermograph.env` at deploy by + `infra/deploy/render-secrets.sh`). + - Forgejo Actions secrets: `S3_ENDPOINT`, `S3_BUCKET`, `S3_ACCESS_KEY`, + `S3_SECRET_KEY`. + - If read-only S3 keys are needed for the query identity, ask the operator to + mint a scoped keypair rather than reusing the read-write one. + +## Secrets handling + +- Host-side: SOPS + age. Recipient `age1xx4dzs0dxlwvkv9sjuqzsphl7lfrxannkfken374yu2qvvcte9sqzktqt2`; + the private key is on each host at `/etc/thermograph/age.key` and on the + operator's machine at `~/.config/sops/age/keys.txt`. +- CI-side: Forgejo Actions secrets. +- ⚠️ **beta gotcha:** `/etc/thermograph.env` on beta is `agent:agent 0640` — the + `deploy` user **cannot read it**. If your service or job runs as `deploy` on + beta, pull credentials from Actions secrets instead, or change the ownership + deliberately and say so. + +## Decisions you must make — and report back + +1. **Catalog**: REST (Lakekeeper / Nessie / Polaris), JDBC-on-Postgres, Hive, or + filesystem/hadoop — and where it runs per environment. +2. **Engine**: DuckDB + iceberg extension, Trino, Spark, or pyiceberg. Prefer the + lightest that satisfies the contract; it must run containerised and headless. +3. **Warehouse location**: bucket + prefix, and whether environments are isolated + by separate prefixes, namespaces, or buckets. +4. **Read-only mechanism**: scoped S3 keys? catalog RBAC? engine restricted mode? + State which, and how a write is refused. +5. **Where the engine runs** for each env (local container on dev, on the box for + beta/prod, or attached to the overlay) — consistent with constraint #3. + +## Acceptance criteria + +The work is accepted when all of these are demonstrated with pasted output: + +- [ ] `infra/ops/iceberg.sh -c "
/metadata/`. `iceberg.sh` resolves that at call time +(numeric metadata version, newest wins) and exposes each table directory as a +view of the same name — which is also how it keeps reading fresh snapshots of +a table that is still being loaded. + +Credentials are resolved at call time and never stored in the repo: the +target host's `/etc/thermograph.env` (`THERMOGRAPH_LAKE_S3_*`) wins if +present; otherwise the calling machine decrypts the SOPS vault +(`infra/deploy/secrets/prod.yaml`) and hands the two values to the remote +shell over stdin — never argv, so never visible in `ps`, and never written to +a file on the target. + +### Safety + +Every session locks itself down *before* user SQL runs: + +```sql +SET allowed_directories=['s3://era5-thermograph/iceberg/', 's3://era5-thermograph/era5/']; +SET enable_external_access=false; +SET lock_configuration=true; +``` + +Both prefixes are required for reads: `era5_daily` was built with pyiceberg +`add_files` over the existing hive parquet, so its metadata lives under +`iceberg/` while its data files stay in place under `era5/daily/`. DuckDB +itself then refuses everything else — the live `backups/` prefix, all local +files — and the lockdown cannot be SET away by query text: + +``` +$ infra/ops/iceberg.sh prod -c "COPY (SELECT 1) TO 's3://era5-thermograph/backups/x.csv'" +Permission Error: Cannot access file "s3://era5-thermograph/backups/x.csv" - file system operations are disabled by configuration +$ infra/ops/iceberg.sh dev -c "SET enable_external_access=true" +Invalid Input Error: Cannot change configuration option "enable_external_access" - the configuration has been locked +``` + +**Known gap:** the only provisioned keypair for the bucket is read-write, so +a raw `COPY TO` targeting a path *inside* the two allowed prefixes is not +refused — the DuckDB iceberg extension cannot write Iceberg tables, but it +could still clobber raw objects under `iceberg/` or `era5/`. Path-based +restriction cannot both allow reading a prefix and forbid writing it; closing +the gap needs a scoped read-only keypair (the object-storage equivalent of +`thermograph_ro`). When the operator mints one, render it as +`THERMOGRAPH_LAKE_S3_*` on the hosts or swap it into the vault — +`iceberg.sh` needs no code change. diff --git a/infra/ops/dbq.sh b/infra/ops/dbq.sh new file mode 100755 index 0000000..c2c0cac --- /dev/null +++ b/infra/ops/dbq.sh @@ -0,0 +1,68 @@ +#!/usr/bin/env bash +# dbq -- run READ-ONLY SQL against any Thermograph Postgres (LAN dev / beta / prod). +# +# infra/ops/dbq.sh dev "select count(*) from climate_history" +# infra/ops/dbq.sh prod -tA "select max(date) from climate_history" +# infra/ops/dbq.sh beta -c '\dt' +# echo "select 1" | infra/ops/dbq.sh prod -f - +# +# Why exec-into-the-container instead of a connection string: +# none of the databases are exposed over TCP. Each listens only on its private +# docker network -- and prod's is a Swarm *overlay* (10.0.2.0/24) that the host +# itself cannot route to, so `ssh -L` works for beta but is impossible for prod. +# Publishing 5432 would mean new firewall + Swarm endpoint changes on production. +# Running psql *inside* the db container works identically in all three +# environments with no ports, no tunnels and no infra changes -- locally for LAN +# dev, over SSH for beta/prod. +# +# Safety: always connects as `thermograph_ro`, a NOSUPERUSER role granted only +# pg_read_all_data. Read-only is enforced by Postgres itself, not by convention, +# so a stray INSERT/DDL fails with "permission denied" even against prod. (The +# app's own `thermograph` role is a superuser -- deliberately not used here.) +# +# Any extra arguments are passed straight through to psql, so -tA, -c, -f, -x, +# --csv etc. all work. Stdin is forwarded, so `-f -` reads piped SQL. +set -euo pipefail + +KEYFILE="${THERMOGRAPH_AGENT_KEY:-$HOME/.ssh/thermograph_agent_ed25519}" +DB_USER="${THERMOGRAPH_DB_QUERY_USER:-thermograph_ro}" +DB_NAME="${THERMOGRAPH_DB_NAME:-thermograph}" + +usage() { + sed -n '2,9p' "$0" | sed 's/^# \{0,1\}//' >&2 + echo "environments: dev | beta | prod" >&2 + exit 2 +} + +env_name="${1:-}" +[ -n "$env_name" ] || usage +shift + +case "$env_name" in + dev) filter=thermograph-dev-db ; ssh_target= ;; + beta) filter=thermograph-db-1 ; ssh_target=agent@75.119.132.91 ;; + prod) filter=thermograph_db ; ssh_target=agent@169.58.46.181 ;; + -h|--help) usage ;; + *) echo "dbq: unknown environment '$env_name' (want: dev|beta|prod)" >&2; exit 2 ;; +esac + +[ $# -gt 0 ] || usage + +# Quote the psql arguments so they survive the remote shell intact. +psql_args=$(printf '%q ' "$@") + +# The container name is resolved on the target host at call time: prod's Swarm +# task name changes on every redeploy, so it can never be hardcoded. +remote=$(cat <&2; exit 1; } +exec docker exec -i "\$cid" psql -U ${DB_USER} -d ${DB_NAME} -v ON_ERROR_STOP=1 ${psql_args} +REMOTE +) + +if [ -z "$ssh_target" ]; then + exec bash -c "$remote" +else + exec ssh -i "$KEYFILE" -o StrictHostKeyChecking=no -o ConnectTimeout=15 \ + "$ssh_target" "$remote" +fi diff --git a/infra/ops/iceberg.sh b/infra/ops/iceberg.sh new file mode 100755 index 0000000..693e54d --- /dev/null +++ b/infra/ops/iceberg.sh @@ -0,0 +1,200 @@ +#!/usr/bin/env bash +# iceberg -- run READ-ONLY SQL against the Thermograph Iceberg lake (LAN dev / beta / prod). +# +# infra/ops/iceberg.sh dev -c "show tables" +# infra/ops/iceberg.sh prod -c "select count(*) from era5_daily" +# infra/ops/iceberg.sh beta -json -c "select * from era5_daily limit 3" +# echo "select 1" | infra/ops/iceberg.sh prod -f - +# +# Why an ephemeral container instead of a query service: the lake is one shared +# warehouse in Contabo object storage (s3://era5-thermograph/iceberg), readable +# from every environment, so the environments differ only in WHERE the engine +# runs -- a throwaway `docker run` of a small DuckDB image on the target box. +# No daemon, no listening port, no tunnel (prod's overlay cannot be tunnelled), +# and no container names to resolve, so prod redeploys cannot break it. +# +# There is no catalog service: a table's current state is its newest metadata +# JSON under /
/metadata/, resolved at call time and exposed +# as a view named after the table directory. +# +# Safety: before user SQL runs, the session pins external access to the lake +# prefixes (iceberg/ metadata + era5/ data files) and locks the configuration +# -- anything else, notably backups/ and all local files, is refused by +# DuckDB itself, and the lockdown cannot be SET away. The provisioned S3 +# keypair is read-write, so a write *inside* those prefixes is still possible +# until a scoped read-only keypair exists (see infra/ops/README.md). +# +# Credentials are resolved at call time, never stored here: the target host's +# /etc/thermograph.env (THERMOGRAPH_LAKE_S3_*) wins if present; otherwise the +# calling machine decrypts the SOPS vault and feeds the two values to the +# remote shell over stdin (never argv, so never visible in `ps`). +# +# Any extra arguments are passed straight through to the duckdb CLI, so -c, +# -json, -csv, -line, -f etc. all work. With `-f -`, stdin is forwarded and +# read as piped SQL. A single bare argument is treated as `-c` SQL. +set -euo pipefail + +KEYFILE="${THERMOGRAPH_AGENT_KEY:-$HOME/.ssh/thermograph_agent_ed25519}" + +usage() { + sed -n '2,8p' "$0" | sed 's/^# \{0,1\}//' >&2 + echo "environments: dev | beta | prod" >&2 + exit 2 +} + +env_name="${1:-}" +[ -n "$env_name" ] || usage +shift + +case "$env_name" in + dev) ssh_target= ;; + beta) ssh_target=agent@75.119.132.91 ;; + prod) ssh_target=agent@169.58.46.181 ;; + -h|--help) usage ;; + *) echo "iceberg: unknown environment '$env_name' (want: dev|beta|prod)" >&2; exit 2 ;; +esac + +[ $# -gt 0 ] || usage + +# The engine image: duckdb CLI with the httpfs + iceberg extensions installed +# at build time, so a query needs no downloads. Built on the target host the +# first time it's needed; the tag is a hash of this Dockerfile, so editing it +# here rolls every host forward automatically on the next call. +dockerfile=$(cat <<'DOCKERFILE' +FROM debian:bookworm-slim +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates curl \ + && rm -rf /var/lib/apt/lists/* +RUN curl -fsSL https://github.com/duckdb/duckdb/releases/download/v1.5.5/duckdb_cli-linux-amd64.gz \ + | gunzip > /usr/local/bin/duckdb \ + && chmod +x /usr/local/bin/duckdb +RUN duckdb -c "INSTALL httpfs; INSTALL iceberg;" +COPY <<'ENTRY' /usr/local/bin/iceberg-query +#!/bin/sh +# Entrypoint of the thermograph-iceberg image; run by iceberg.sh, not directly. +set -eu +: "${LAKE_S3_ACCESS_KEY:?}" "${LAKE_S3_SECRET_KEY:?}" +WAREHOUSE="${LAKE_WAREHOUSE:-s3://era5-thermograph/iceberg}" +ENDPOINT="${LAKE_S3_ENDPOINT:-eu2.contabostorage.com}" +# Prefixes a query may touch. era5/ is needed because era5_daily was built +# with add_files over the existing hive parquet: its metadata lives in the +# warehouse but its data files stay in place under era5/daily/. +ALLOWED="${LAKE_ALLOWED_DIRS:-$WAREHOUSE/,s3://era5-thermograph/era5/}" + +esc() { printf %s "$1" | sed "s/'/''/g"; } +SECRET="CREATE SECRET lake (TYPE s3, KEY_ID '$(esc "$LAKE_S3_ACCESS_KEY")', SECRET '$(esc "$LAKE_S3_SECRET_KEY")', ENDPOINT '$ENDPOINT', REGION 'default', URL_STYLE 'path');" + +# One view per table dir, over its newest metadata JSON (numeric version wins, +# so v10 beats v9 and 00010-... beats 00009-...). grep drops statement noise +# like CREATE SECRET's "true" row; an empty warehouse yields no views. +VIEWS=$(duckdb -batch -noheader -list \ + -cmd "LOAD httpfs; $SECRET" \ + -c "SELECT 'CREATE VIEW \"' || tbl || '\" AS SELECT * FROM iceberg_scan(''' || file || ''');' + FROM (SELECT file, + split_part(replace(file, '$WAREHOUSE/', ''), '/', 1) AS tbl, + row_number() OVER ( + PARTITION BY split_part(replace(file, '$WAREHOUSE/', ''), '/', 1) + ORDER BY coalesce(try_cast(regexp_extract(parse_filename(file), '[0-9]+') AS BIGINT), -1) DESC, + file DESC) AS rn + FROM glob('$WAREHOUSE/*/metadata/*.metadata.json')) + WHERE rn = 1;" | grep '^CREATE VIEW ' || true) + +# Lock the session down BEFORE user SQL: external access is pinned to the +# allowed prefixes, local files are unreachable, and the config can't be +# un-SET. Views bind lazily, so reads still work under the lockdown. +allowed_sql= +oifs=$IFS; IFS=, +for d in $ALLOWED; do allowed_sql="$allowed_sql${allowed_sql:+, }'$d'"; done +IFS=$oifs + +INIT="LOAD httpfs; LOAD iceberg; $SECRET $VIEWS +SET allowed_directories=[$allowed_sql]; +SET enable_external_access=false; +SET lock_configuration=true;" + +# A single bare argument is the SQL itself (dbq.sh calling convention). +if [ $# -eq 1 ] && [ "${1#-}" = "$1" ]; then set -- -c "$1"; fi + +# duckdb has no "-f - means stdin" convention; hand it the real thing. +i=0; n=$#; prev= +while [ "$i" -lt "$n" ]; do + a=$1; shift + if [ "$prev" = "-f" ] && [ "$a" = "-" ]; then a=/dev/stdin; fi + prev=$a + set -- "$@" "$a" + i=$((i+1)) +done + +# The .output pair silences the init's own result rows (CREATE SECRET prints +# a "true"); errors still reach stderr, and user SQL output is untouched. +exec duckdb -batch -cmd ".output /dev/null" -cmd "$INIT" -cmd ".output" "$@" +ENTRY +RUN chmod +x /usr/local/bin/iceberg-query +DOCKERFILE +) + +tag="thermograph-iceberg:$(printf '%s' "$dockerfile" | sha256sum | cut -c1-12)" +b64=$(printf '%s' "$dockerfile" | base64 -w0) + +# Quote the duckdb arguments so they survive the remote shell intact. +args=$(printf '%q ' "$@") + +# Credentials: env override first, then a call-time sops decrypt of the vault +# on the calling machine. Either may come up empty here -- the remote side +# still gets a chance to fill them from its own /etc/thermograph.env. +ak="${THERMOGRAPH_LAKE_S3_ACCESS_KEY:-}" +sk="${THERMOGRAPH_LAKE_S3_SECRET_KEY:-}" +if [ -z "$ak" ] || [ -z "$sk" ]; then + repo_root=$(cd -- "$(dirname -- "$0")/../.." && pwd) + vault="${THERMOGRAPH_LAKE_VAULT:-$repo_root/infra/deploy/secrets/prod.yaml}" + sops_bin=$(command -v sops || echo "$HOME/.local/bin/sops") + if [ -r "$vault" ] && [ -x "$sops_bin" ]; then + ak=$("$sops_bin" -d --extract '["THERMOGRAPH_LAKE_S3_ACCESS_KEY"]' "$vault" 2>/dev/null || true) + sk=$("$sops_bin" -d --extract '["THERMOGRAPH_LAKE_S3_SECRET_KEY"]' "$vault" 2>/dev/null || true) + fi +fi + +# The engine image is (re)built on the target host only when its tag is +# missing, then run with the credentials passed via environment -- docker +# inherits them from the remote shell, which read them from stdin, so they +# never appear on a command line. The rest of stdin flows through to duckdb. +remote=$(cat <&2 + exit 1 +fi +export LAKE_S3_ACCESS_KEY="\$ak" LAKE_S3_SECRET_KEY="\$sk" +docker image inspect $tag >/dev/null 2>&1 \ + || echo $b64 | base64 -d | docker build -q -t $tag - >/dev/null +exec docker run --rm -i -e LAKE_S3_ACCESS_KEY -e LAKE_S3_SECRET_KEY $tag iceberg-query $args +REMOTE +) + +# Stdin is forwarded only for the documented `-f -` form. Anything else gets +# just the credential lines -- unconditionally forwarding an idle stdin would +# leave the local pipeline waiting on input the query never reads. +want_stdin=false +prev= +for a in "$@"; do + if [ "$prev" = "-f" ] && [ "$a" = "-" ]; then want_stdin=true; fi + prev=$a +done + +feed() { + printf '%s\n%s\n' "$ak" "$sk" || true + if $want_stdin; then cat || true; fi +} + +if [ -z "$ssh_target" ]; then + feed | bash -c "$remote" +else + feed | ssh -i "$KEYFILE" -o StrictHostKeyChecking=no -o ConnectTimeout=15 \ + "$ssh_target" "$remote" +fi