LAB-744: fix Review Panel findings (data race, ZK checkpoint leak, restore boot-loop, publish ordering)

ingester/README.md

@@ -33,7 +33,10 @@ Configuration (env or `.env`, via pydantic-settings; secrets are `SecretStr`):
 
 Each window republishes at TTL/2 (locked TTLs: 5m→60 s, 1h→300 s, 24h→900 s), so readers always
 hit. cachekit is decorator-only, so a publish is `invalidate_cache(window)` + call — the miss
-recomputes from the in-memory windows and writes fresh bytes.
+recomputes from the in-memory windows and writes fresh bytes. The recompute is probed *before*
+invalidating so a compute failure never deletes a live key; a backend **write** failure after the
+invalidate can still leave the key briefly deleted until the next tick — cachekit has no atomic
+set/replace, so that gap is inherent to the decorator API.
 
 Values are interop/v1 plain MessagePack, top-level maps with string keys. All carry
 `window` (str), `generated_at` (unix seconds, int), `total_posts` (int), plus:
@@ -51,7 +54,9 @@ Values are interop/v1 plain MessagePack, top-level maps with string keys. All ca
 `language_sentiment(window="1h")` — per-language lexicon sentiment `{lang: {avg, n}}` — is written
 via `@cache.secure(master_key=…)` auto mode, `namespace="bluesky-thinking"`. Its key is the
 Python-only 7-segment auto key (`ns:bluesky-thinking:func:…`), and the backend stores ciphertext
-only (asserted in tests). Ciphertext-only verification against the live SaaS is Stage 3.
+only (asserted in tests). Zero-knowledge holds end-to-end: the sentiment value is encrypted here and
+its plaintext source is never written to any other key (the checkpoint omits it — see below), so the
+backend never sees it in the clear. Ciphertext-only verification against the live SaaS is Stage 3.
 
 ### Checkpointing
 
@@ -61,6 +66,17 @@ window (the spec's Render-restart mitigation). Per-minute counters are truncated
 entries in the snapshot — long-tail trending counts are approximate after a restore;
 `posts_per_minute` and `lang_mix` stay exact.
 
+The checkpoint is stored **unencrypted**, so it deliberately omits the per-language sentiment
+totals: those are the cleartext source of the `@cache.secure` value, and persisting them in the
+plaintext checkpoint would let the backend reconstruct it (`avg = sum / count`), breaking the
+zero-knowledge property. Sentiment is not restart-critical — the secure 1h window repopulates within
+an hour of a restart; the aggregate counts above are unaffected.
+
+The checkpoint is equally **untrusted on read-back** (a backend operator can poison it): `restore()`
+validates and coerces every entry, skipping corrupt ones with a warning instead of crashing startup,
+and ignores any legacy `sent` field entirely — restoring it would let a poisoned checkpoint choose
+the plaintext that the next secure publish encrypts.
+
 ## Privacy
 
 Aggregate-only: the extractor reduces each post to counter inputs (tags, links, primary language,

ingester/src/skyline_ingester/jetstream.py

@@ -5,6 +5,8 @@
 import asyncio
 import json
 import logging
+import time
+from collections.abc import Callable
 from urllib.parse import urlencode
 
 import websockets
@@ -15,20 +17,38 @@
 logger = logging.getLogger(__name__)
 
 MAX_BACKOFF = 60.0
+# Payload timestamps are untrusted. One far-future time_us would (a) set a
+# retention floor in WindowStore._prune that instantly evicts every real bucket
+# — wiping the restart-critical 24h window and poisoning the next checkpoint —
+# and (b) as a resume cursor, skip every real event on the next reconnect.
+# Anything beyond this skew over wall-clock is dropped whole.
+MAX_FUTURE_SKEW_SECONDS = 300.0
 
 
-def ingest_raw(raw: str | bytes, store: WindowStore) -> int | None:
-    """Parse one Jetstream frame into the store; returns the event's time_us cursor."""
+def ingest_raw(raw: str | bytes, store: WindowStore, *, now_fn: Callable[[], float] = time.time) -> int | None:
+    """Parse one Jetstream frame into the store; returns the event's time_us cursor.
+
+    Future-dated events (beyond MAX_FUTURE_SKEW_SECONDS of wall-clock) are dropped
+    entirely — neither aggregated nor used to advance the cursor.
+    """
     try:
         event = json.loads(raw)
     except (ValueError, UnicodeDecodeError):
         logger.warning("unparseable Jetstream frame (%d bytes)", len(raw))
         return None
+    time_us = event.get("time_us") if isinstance(event, dict) else None
+    if not isinstance(time_us, int):
+        # Same visibility as the sibling drops: a Jetstream schema change here
+        # would otherwise be 100% silent data loss under a healthy-looking loop.
+        logger.warning("dropping Jetstream frame without int time_us")
+        return None
+    if time_us / 1_000_000 > now_fn() + MAX_FUTURE_SKEW_SECONDS:
+        logger.warning("dropping future-dated Jetstream event (time_us=%d)", time_us)
+        return None
     feats = extract_post(event)
     if feats is not None:
         store.add(feats)
-    time_us = event.get("time_us") if isinstance(event, dict) else None
-    return time_us if isinstance(time_us, int) else None
+    return time_us
 
 
 def subscribe_url(base: str, cursor: int | None = None) -> str:

ingester/src/skyline_ingester/publisher.py

@@ -90,39 +90,77 @@ def publish(window: str) -> dict:
     def secure_enabled(self) -> bool:
         return self._secure_fn is not None
 
+    def _refresh(self, fn: Callable, window: str, recompute: Callable[[], object], operation: str) -> int:
+        """Recompute the value, then invalidate + republish one wrapper.
+
+        cachekit is decorator-only, so a fresh write is invalidate-then-call — and
+        if the recompute would raise we must find out BEFORE invalidating, or a
+        failed recompute leaves the key deleted (a cache miss on the metered-miss
+        path) until the next tick.
+
+        Honest ceiling: the probe only covers RECOMPUTE failure. The wrapper call
+        after invalidate is itself recompute-then-backend-WRITE, and a write
+        failure at that point still leaves the key deleted until the next tick —
+        cachekit has no atomic set/replace (confirmed against 0.15.0), so this is
+        as close as the decorator API allows.
+        """
+        try:
+            recompute()
+            fn.invalidate_cache(window)
+            fn(window)
+            return 1
+        except Exception:
+            logger.exception("publish failed: %s/%s", operation, window, extra={"operation": operation, "window": window})
+            return 0
+
     def publish_window(self, window: str) -> int:
         """Refresh every operation for one window; returns how many published."""
         published = 0
         for operation in OPERATIONS:
             fn = self._publish_fns[(operation, window)]
-            try:
-                fn.invalidate_cache(window)
-                fn(window)
-                published += 1
-            except Exception:
-                logger.exception("publish failed: %s/%s", operation, window)
+            published += self._refresh(
+                fn,
+                window,
+                lambda operation=operation: self._store.build_value(operation, window, self._now(), self._top_n),
+                operation,
+            )
         if window == SECURE_WINDOW and self._secure_fn is not None:
-            try:
-                self._secure_fn.invalidate_cache(window)
-                self._secure_fn(window)
-                published += 1
-            except Exception:
-                logger.exception("secure publish failed: language_sentiment/%s", window)
+            published += self._refresh(
+                self._secure_fn,
+                window,
+                lambda: self._store.sentiment_value(window, self._now()),
+                "language_sentiment",
+            )
         return published
 
     def checkpoint(self) -> None:
-        """Force-write the current window state (restart insurance)."""
+        """Force-write the current window state (restart insurance).
+
+        No recompute probe here (unlike _refresh): snapshot() is a pure in-memory
+        walk of our own state — the only realistic failure after invalidate is the
+        backend WRITE, which no probe can cover (see _refresh's ceiling note), so a
+        probe would just double the snapshot cost for nothing.
+        """
         self._checkpoint_fn.invalidate_cache()
         self._checkpoint_fn()
 
     def restore_checkpoint(self) -> int:
         """Load the last checkpoint into the store; returns buckets restored.
 
         On a cold cache the call is a miss, which harmlessly writes a snapshot
-        of the (empty) store and restores 0 buckets.
+        of the (empty) store and restores 0 buckets. A failure anywhere in the
+        read OR the restore degrades to 0 rather than propagating — a bad
+        checkpoint must never crash startup into a permanent boot loop, and the
+        checkpoint outlives a bad deploy (26h TTL), so a crash here would loop
+        until the TTL expires.
         """
-        snap = self._checkpoint_fn()
-        restored = self._store.restore(snap, self._now())
+        try:
+            snap = self._checkpoint_fn()
+            restored = self._store.restore(snap, self._now())
+        except Exception:
+            extra = {"operation": "restore_checkpoint"}
+            logger.exception("checkpoint restore failed at startup; continuing with a cold window", extra=extra)
+            return 0
         if restored:
             logger.info("restored %d window buckets from checkpoint (saved_at=%s)", restored, snap.get("saved_at"))
         return restored