diff --git a/ingester/README.md b/ingester/README.md index d2f5035..aa509cc 100644 --- a/ingester/README.md +++ b/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, diff --git a/ingester/src/skyline_ingester/jetstream.py b/ingester/src/skyline_ingester/jetstream.py index 8a53330..537a004 100644 --- a/ingester/src/skyline_ingester/jetstream.py +++ b/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: diff --git a/ingester/src/skyline_ingester/publisher.py b/ingester/src/skyline_ingester/publisher.py index a9fb94c..ffd4767 100644 --- a/ingester/src/skyline_ingester/publisher.py +++ b/ingester/src/skyline_ingester/publisher.py @@ -90,28 +90,57 @@ 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() @@ -119,10 +148,19 @@ 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 diff --git a/ingester/src/skyline_ingester/windows.py b/ingester/src/skyline_ingester/windows.py index c203fc1..207dab7 100644 --- a/ingester/src/skyline_ingester/windows.py +++ b/ingester/src/skyline_ingester/windows.py @@ -1,8 +1,11 @@ """Sliding minute-bucket windows and the five locked aggregates. One minute of posts = one Bucket of counters. A window aggregate merges the -buckets inside (now - window, now]; merges are memoised per (window, now) so -one publish tick computes each window's merge once for all five operations. +buckets inside (now - window, now]; merges are memoised per (window, now), so +on a quiet stream one publish tick computes each window's merge once for all +five operations. The memo is best-effort: an add() landing mid-merge suppresses +it (typical under live firehose load) and each caller then recomputes — correct +either way, just without the shortcut. ponytail: merge-on-demand walks up to 1440 buckets per 24h publish (~every 450 s). Move to incremental per-window running totals if that ever shows up @@ -11,11 +14,15 @@ from __future__ import annotations +import logging +import threading from collections import Counter from dataclasses import dataclass, field from skyline_ingester.extract import PostFeatures +logger = logging.getLogger(__name__) + WINDOW_MINUTES = {"5m": 5, "1h": 60, "24h": 1440} # Locked TTLs (docs/architecture.md): 5m -> 60 s, 1h -> 300 s, 24h -> 900 s. WINDOW_TTLS = {"5m": 60, "1h": 300, "24h": 900} @@ -29,6 +36,8 @@ @dataclass(slots=True) class Bucket: + """One minute of counters — also the shape a window merge accumulates into.""" + n: int = 0 tags: Counter = field(default_factory=Counter) links: Counter = field(default_factory=Counter) @@ -36,15 +45,19 @@ class Bucket: emoji: Counter = field(default_factory=Counter) sent: dict[str, list[float]] = field(default_factory=dict) # lang -> [sum, count] - -@dataclass(slots=True) -class Merged: - n: int = 0 - tags: Counter = field(default_factory=Counter) - links: Counter = field(default_factory=Counter) - langs: Counter = field(default_factory=Counter) - emoji: Counter = field(default_factory=Counter) - sent: dict[str, list[float]] = field(default_factory=dict) + def copy(self) -> Bucket: + # Shallow copies: enough isolation to read the copy while the original + # keeps being mutated under the store lock. Counter.copy() into an empty + # destination is a single C-level dict.update (Counter.update's empty + # fast path), so each copy stays cheap enough to run under the lock. + return Bucket( + n=self.n, + tags=self.tags.copy(), + links=self.links.copy(), + langs=self.langs.copy(), + emoji=self.emoji.copy(), + sent={lang: acc.copy() for lang, acc in self.sent.items()}, + ) class WindowStore: @@ -53,56 +66,109 @@ class WindowStore: def __init__(self, max_minutes: int = WINDOW_MINUTES["24h"]): self._max = max_minutes self._buckets: dict[int, Bucket] = {} - self._memo: dict[tuple[str, int], Merged] = {} + self._memo: dict[tuple[str, int], Bucket] = {} + # Bumped by every add(); a merge only memoises its result if no add() + # landed since it started, so a cleared memo can't be resurrected with + # a pre-add() view for the rest of that second. + self._gen = 0 + # The Jetstream consumer calls add() on the event-loop thread while the + # publish/checkpoint loops read the store from asyncio.to_thread workers; + # every access to _buckets/_memo is serialised through this lock. + self._lock = threading.Lock() def add(self, feats: PostFeatures) -> None: minute = int(feats.ts // 60) - bucket = self._buckets.get(minute) - if bucket is None: - bucket = self._buckets[minute] = Bucket() - self._prune(minute) - bucket.n += 1 - bucket.tags.update(feats.hashtags) - bucket.links.update(feats.links) - bucket.langs[feats.lang] += 1 - bucket.emoji.update(feats.emoji) - if feats.sentiment is not None: - acc = bucket.sent.setdefault(feats.lang, [0.0, 0]) - acc[0] += feats.sentiment - acc[1] += 1 - self._memo.clear() + with self._lock: + bucket = self._buckets.get(minute) + if bucket is None: + bucket = self._buckets[minute] = Bucket() + self._prune(minute) + bucket.n += 1 + bucket.tags.update(feats.hashtags) + bucket.links.update(feats.links) + bucket.langs[feats.lang] += 1 + bucket.emoji.update(feats.emoji) + if feats.sentiment is not None: + acc = bucket.sent.setdefault(feats.lang, [0.0, 0]) + acc[0] += feats.sentiment + acc[1] += 1 + self._memo.clear() + self._gen += 1 def _prune(self, newest_minute: int) -> None: - floor = max((m for m in self._buckets), default=newest_minute) - floor = max(floor, newest_minute) - self._max + # Caller holds self._lock. Anchor the retention floor to the minute being + # added, NOT max(self._buckets): one bogus far-future timestamp must not + # become a permanent anchor that evicts every real bucket forever. With + # this anchor a stray future bucket is excluded from every merged() query + # (which bounds by `now`) and real minutes re-accumulate on the next event. + floor = newest_minute - self._max for minute in [m for m in self._buckets if m <= floor]: del self._buckets[minute] - def merged(self, window: str, now: float) -> Merged: - """Merge the buckets inside (now - window, now].""" + def merged(self, window: str, now: float) -> Bucket: + """Merge the buckets inside (now - window, now] into one Bucket. + + Lock contract: self._lock is a non-reentrant threading.Lock — never call + merged()/snapshot()/add() while holding it. The lock is held only for + C-speed per-bucket copies; the O(window) Counter merge runs outside it so + add() on the event-loop thread never stalls behind a full 24h merge. The + returned (memoised) Bucket is read lock-free by callers and MUST NOT be + mutated. + """ key = (window, int(now)) - memo = self._memo.get(key) - if memo is not None: - return memo now_min = int(now // 60) lo = now_min - WINDOW_MINUTES[window] - out = Merged() - for minute, b in self._buckets.items(): - if lo < minute <= now_min: - out.n += b.n - out.tags.update(b.tags) - out.links.update(b.links) - out.langs.update(b.langs) - out.emoji.update(b.emoji) - for lang, (s, c) in b.sent.items(): - acc = out.sent.setdefault(lang, [0.0, 0]) - acc[0] += s - acc[1] += c - if len(self._memo) > 8: - self._memo.clear() - self._memo[key] = out + with self._lock: + memo = self._memo.get(key) + if memo is not None: + return memo + gen = self._gen + out = Bucket() + for _minute, b in self._copy_range(lo, now_min): + out.n += b.n + out.tags.update(b.tags) + out.links.update(b.links) + out.langs.update(b.langs) + out.emoji.update(b.emoji) + for lang, (s, c) in b.sent.items(): + acc = out.sent.setdefault(lang, [0.0, 0]) + acc[0] += s + acc[1] += c + with self._lock: + # Memoise only if no add() landed since the merge started: add() + # cleared the memo, and re-inserting this pre-add() view would serve + # it stale to every same-second caller. + if self._gen == gen: + if len(self._memo) > 8: + self._memo.clear() + self._memo[key] = out return out + # 16 buckets/chunk keeps each lock hold ~1-2 ms even at firehose-dense + # buckets; the per-chunk lock overhead itself is microseconds. + _COPY_CHUNK = 16 + + def _copy_range(self, lo: float = float("-inf"), hi: float = float("inf")) -> list[tuple[int, Bucket]]: + """Copy the buckets in (lo, hi] in chunks, releasing the lock between chunks. + + Copy, don't reference: add() mutates hot buckets' Counters in place, and + iterating a Counter that grows mid-merge raises "dictionary changed size + during iteration" (the round-1 bug class). Chunking bounds add()'s worst + stall to one chunk's copy (~few ms) instead of a full-window copy; a bucket + created or pruned between chunks simply lands in or out of this tick's view, + which periodic analytics tolerates. + """ + with self._lock: + keys = [m for m in self._buckets if lo < m <= hi] + copies: list[tuple[int, Bucket]] = [] + for i in range(0, len(keys), self._COPY_CHUNK): + with self._lock: + for m in keys[i : i + self._COPY_CHUNK]: + b = self._buckets.get(m) + if b is not None: # pruned between chunks + copies.append((m, b.copy())) + return copies + def build_value(self, operation: str, window: str, now: float, top_n: int = 50) -> dict: """The interop/v1 value for one (operation, window): a top-level map with string keys.""" m = self.merged(window, now) @@ -142,7 +208,17 @@ def snapshot(self, now: float) -> dict: Per-bucket counters are cut to their top-K entries, so long-tail counts are approximate after a restore; posts_per_minute and lang_mix totals stay exact (bucket n / langs are kept in full up to _K_LANGS languages). + + Per-language sentiment (`sent`) is deliberately NOT persisted: it is the + cleartext source of the @cache.secure sentiment cache, and this checkpoint + is stored unencrypted. Writing it here would let the backend reconstruct + the zero-knowledge value (avg = sum / count). The secure 1h window + repopulates within an hour of a restart; the restart-critical aggregate + counts below are unaffected. """ + # Same lock discipline as merged(): chunked copy-under-lock; the + # most_common() sorts and dict building run outside. + copies = sorted(self._copy_range()) return { "v": SNAPSHOT_VERSION, "saved_at": int(now), @@ -155,31 +231,73 @@ def snapshot(self, now: float) -> dict: "links": dict(b.links.most_common(_K_LINKS)), "langs": dict(b.langs.most_common(_K_LANGS)), "emoji": dict(b.emoji.most_common(_K_EMOJI)), - "sent": {lang: [s, c] for lang, (s, c) in b.sent.items()}, }, ] - for minute, b in sorted(self._buckets.items()) + for minute, b in copies ], } def restore(self, snap: dict, now: float) -> int: - """Load a snapshot(); returns the number of buckets restored (0 = nothing usable).""" + """Load a snapshot(); returns the number of buckets restored (0 = nothing usable). + + The checkpoint is untrusted input (plaintext, integrity-unprotected in the + backend), so every entry is validated and a malformed one is skipped with a + warning rather than raising — a corrupt or partial checkpoint must never + crash startup into a permanent boot loop. Legacy checkpoints may still carry + `sent`; it is IGNORED entirely: the checkpoint is operator-poisonable, and + restoring `sent` would let the backend operator choose the plaintext that the + next @cache.secure publish encrypts — the exact value the zero-knowledge + boundary exists to protect. Sentiment repopulates from live ingestion only. + """ if not isinstance(snap, dict) or snap.get("v") != SNAPSHOT_VERSION: + logger.warning("ignoring checkpoint with unexpected shape/version: %.80r", snap) + return 0 + buckets = snap.get("buckets") + if not isinstance(buckets, list): + logger.warning("ignoring checkpoint with malformed buckets: %.80r", buckets) return 0 - floor = int(now // 60) - self._max + now_min = int(now // 60) + floor = now_min - self._max + ceiling = now_min + 1 # a checkpoint can't legitimately hold future minutes restored = 0 - for minute, d in snap.get("buckets") or []: - if not isinstance(minute, int) or minute <= floor: - continue - b = Bucket( - n=int(d.get("n", 0)), - tags=Counter(d.get("tags") or {}), - links=Counter(d.get("links") or {}), - langs=Counter(d.get("langs") or {}), - emoji=Counter(d.get("emoji") or {}), - sent={lang: [float(s), int(c)] for lang, (s, c) in (d.get("sent") or {}).items()}, - ) - self._buckets[minute] = b - restored += 1 - self._memo.clear() + with self._lock: + for item in buckets: + try: + minute, d = item + if not isinstance(minute, int) or minute <= floor or minute > ceiling: + continue + # Coerce keys/values, not just presence: a poisoned-but-valid + # checkpoint (e.g. a counter value of "not-a-number", or a + # negative count that skews ppm/lang_mix) would pass restore + # and detonate later inside merged()/most_common(), where the + # publisher's except swallows it into silent misses for up to + # 24h. A bad entry must fail HERE, skipping only its bucket. + b = Bucket( + n=_non_negative(int(d.get("n", 0))), + tags=_coerced_counter(d.get("tags")), + links=_coerced_counter(d.get("links")), + langs=_coerced_counter(d.get("langs")), + emoji=_coerced_counter(d.get("emoji")), + ) + except (ValueError, TypeError, AttributeError, OverflowError) as exc: + # %.120r: entries come from the untrusted checkpoint and can + # be arbitrarily large — cap what one bad bucket puts in a log. + logger.warning("skipping corrupt checkpoint bucket: %s: %.120r", exc, item) + continue + self._buckets[minute] = b + restored += 1 + self._memo.clear() + self._gen += 1 return restored + + +def _non_negative(value: int) -> int: + if value < 0: + raise ValueError("negative count in checkpoint") + return value + + +def _coerced_counter(data) -> Counter: + # str keys / non-negative int values, or ValueError|TypeError|OverflowError + # (int(float("inf"))) — restore() skips the bucket. + return Counter({str(k): _non_negative(int(v)) for k, v in (data or {}).items()}) diff --git a/ingester/tests/test_checkpoint.py b/ingester/tests/test_checkpoint.py index 350d394..40cb3e1 100644 --- a/ingester/tests/test_checkpoint.py +++ b/ingester/tests/test_checkpoint.py @@ -1,7 +1,9 @@ """Checkpoint/restore: a restart must not zero the 24h window.""" +import logging + from skyline_ingester.publisher import Publisher -from skyline_ingester.windows import WindowStore +from skyline_ingester.windows import SNAPSHOT_VERSION, WindowStore from .conftest import FIXTURE_TOTALS, MASTER_KEY, NOW @@ -45,3 +47,83 @@ def test_snapshot_truncates_per_bucket_counters(store): def test_restore_rejects_unknown_version(store): assert store.restore({"v": 999, "buckets": []}, NOW) == 0 assert store.restore({}, NOW) == 0 + + +def test_restore_ignores_legacy_sent(): + # ZK (panel round 3): the plaintext checkpoint is operator-poisonable, so a + # restored `sent` would let the backend operator choose the plaintext of the + # next @cache.secure publish. Sentiment must come from live ingestion only. + good = int(NOW // 60) + legacy = {"v": SNAPSHOT_VERSION, "buckets": [[good, {"n": 2, "sent": {"en": [999.0, 1]}}]]} + store = WindowStore() + assert store.restore(legacy, NOW) == 1 # the bucket's counts still restore + assert store.sentiment_value("1h", NOW)["langs"] == {} + + +def test_restore_checkpoint_never_crashes_startup(publisher, backend, monkeypatch): + # The boot-loop guard end-to-end: nothing a poisoned checkpoint triggers inside + # restore() may propagate through asyncio.run and crash startup — the bad + # checkpoint outlives the crash (26h TTL), so a raise here loops until the TTL. + publisher.checkpoint() + store2 = WindowStore() + publisher2 = Publisher(store2, backend, master_key=MASTER_KEY, now_fn=lambda: NOW) + + def boom(snap, now): + raise RuntimeError("poisoned checkpoint detonated inside restore") + + monkeypatch.setattr(store2, "restore", boom) + assert publisher2.restore_checkpoint() == 0 + + +def test_snapshot_omits_sentiment_for_zero_knowledge(store): + # ZK: `sent` is the cleartext source of the @cache.secure value; the plaintext + # checkpoint must not carry it, or the backend reconstructs avg = sum / count. + snap = store.snapshot(NOW) + assert snap["buckets"], "fixture stream should produce buckets" + assert all("sent" not in d for _minute, d in snap["buckets"]) + + +def test_restore_tolerates_malformed_checkpoints(caplog): + # A corrupt / partial checkpoint must degrade to a skip, never raise — a raise + # here propagates through asyncio.run and crashes startup into a boot loop. + good = int(NOW // 60) + bad = [ + {"v": SNAPSHOT_VERSION, "buckets": "not-a-list"}, + {"v": SNAPSHOT_VERSION, "buckets": [[good]]}, # item is not a (minute, dict) pair + {"v": SNAPSHOT_VERSION, "buckets": [[good, "not-a-dict"]]}, + {"v": SNAPSHOT_VERSION, "buckets": [["not-an-int", {}]]}, + {"v": SNAPSHOT_VERSION, "buckets": [[good, {"n": "x"}]]}, # non-numeric count + {"v": SNAPSHOT_VERSION, "buckets": [[good, {"n": float("inf")}]]}, # int(inf) -> OverflowError + ] + for snap in bad: + assert WindowStore().restore(snap, NOW) == 0 # skipped, no raise + # a valid bucket alongside a broken one is still restored — and the skip is + # logged, not silent: an operator must be able to see checkpoint corruption. + mixed = {"v": SNAPSHOT_VERSION, "buckets": [[good, {"n": 5}], [good - 1, "broken"]]} + with caplog.at_level(logging.WARNING, logger="skyline_ingester.windows"): + assert WindowStore().restore(mixed, NOW) == 1 + assert any("corrupt checkpoint bucket" in r.getMessage() for r in caplog.records) + + +def test_restore_rejects_poisoned_but_valid_counter_values(): + # Panel round-2 MAJ: a structurally valid checkpoint with a non-numeric counter + # VALUE used to pass restore() and detonate later in merged()/most_common(), + # where the publisher's except turns it into silent misses for up to 24h. + # It must fail at restore, skipping only the poisoned bucket. + good = int(NOW // 60) + poisoned = { + "v": SNAPSHOT_VERSION, + "buckets": [ + [good, {"n": 3, "tags": {"x": "not-a-number"}}], # poisoned value -> skipped + [good - 1, {"n": 2, "tags": {"ok": 2}, "langs": {1: 2}}], # non-str key -> coerced + [good - 2, {"n": -1_000_000}], # negative count skews ppm -> skipped + [good - 3, {"n": 1, "tags": {"neg": -5}}], # negative counter value -> skipped + [good + 10_000, {"n": 1, "tags": {"future": 1}}], # future minute parks forever -> skipped + ], + } + store = WindowStore() + assert store.restore(poisoned, NOW) == 1 + merged = store.merged("24h", NOW) # must never raise + assert merged.tags.most_common(5) == [("ok", 2)] + assert merged.langs == {"1": 2} + assert merged.n == 2 diff --git a/ingester/tests/test_extract.py b/ingester/tests/test_extract.py index 1145c06..f68ad03 100644 --- a/ingester/tests/test_extract.py +++ b/ingester/tests/test_extract.py @@ -64,6 +64,35 @@ def test_ingest_raw_returns_cursor_and_skips_garbage(): assert ingest_raw('{"kind": "commit", "time_us": 123}', store) == 123 +def test_ingest_raw_drops_future_dated_events(fixture_lines): + # Panel round-2 CRIT: one far-future time_us sets a retention floor in _prune + # that wipes every real bucket (and, as a cursor, would skip everything on the + # next reconnect). Future-dated frames are dropped whole at the ingest boundary. + import json + + from .conftest import NOW + + store = WindowStore() + now_fn = lambda: NOW # noqa: E731 + for line in fixture_lines: + ingest_raw(line, store, now_fn=now_fn) + healthy = store.merged("24h", NOW).n + assert healthy == FIXTURE_TOTALS["24h"] + + poison = json.loads(fixture_lines[0]) + poison["time_us"] = int((NOW + 10_000_000 * 60) * 1_000_000) # ~19 years ahead + assert ingest_raw(json.dumps(poison), store, now_fn=now_fn) is None # no cursor advance + # NOW + 1 forces a fresh merge (same minute, different memo key) — asserting at + # NOW would just re-read the memoised Bucket and pass even on a wiped store. + assert store.merged("24h", NOW + 1).n == healthy # window NOT wiped + assert store.snapshot(NOW)["buckets"], "checkpoint still has the real buckets" + + # small clock skew stays acceptable + slight = json.loads(fixture_lines[0]) + slight["time_us"] = int((NOW + 60) * 1_000_000) + assert ingest_raw(json.dumps(slight), store, now_fn=now_fn) == slight["time_us"] + + def test_fixture_totals_match_windows(store): from .conftest import NOW diff --git a/ingester/tests/test_publisher.py b/ingester/tests/test_publisher.py index 29adb2c..8bc2bb7 100644 --- a/ingester/tests/test_publisher.py +++ b/ingester/tests/test_publisher.py @@ -60,3 +60,19 @@ def test_republish_refreshes_stale_values(publisher, backend, store): publisher.publish_window("5m") # invalidate + recompute, not a cache hit after = decode_interop_value(backend.get(key)) assert after["total_posts"] == before["total_posts"] + 1 + + +def test_failed_recompute_keeps_the_live_key(publisher, backend, store, monkeypatch): + # Regression: publish was invalidate-then-recompute, so a recompute failure left + # the key deleted (a miss on the metered-miss path). Recompute now runs first, so + # a failure leaves the previously published entry intact. + publisher.publish_window("5m") + key = generate_interop_key(NAMESPACE, "trending_hashtags", ["5m"]) + assert backend.get(key) is not None + + def boom(*_a, **_k): + raise RuntimeError("recompute failed") + + monkeypatch.setattr(store, "build_value", boom) + assert publisher.publish_window("5m") == 0 # every op fails its recompute probe + assert backend.get(key) is not None # ...and the live key was NOT invalidated diff --git a/ingester/tests/test_windows.py b/ingester/tests/test_windows.py index 1ab04d8..8448eda 100644 --- a/ingester/tests/test_windows.py +++ b/ingester/tests/test_windows.py @@ -1,5 +1,8 @@ """Window aggregation and expiry against the recorded fixture stream.""" +import sys +import threading + from skyline_ingester.extract import PostFeatures from skyline_ingester.windows import WindowStore @@ -65,6 +68,30 @@ def test_windows_expire(store): assert store.merged("24h", NOW + 65 * 60).n == FIXTURE_TOTALS["24h"] +def test_memo_is_not_resurrected_by_a_concurrent_add(store): + # Regression (CodeRabbit on PR #5): merged() computes outside the lock; if an + # add() lands mid-merge it clears the memo, and blindly re-inserting the + # pre-add() result would serve it stale to every same-second caller. The + # generation counter must suppress that memo insert. + before = store.merged("5m", NOW).n + + orig = store._copy_range + + def add_mid_merge(lo, hi): + copies = orig(lo, hi) + store.add(PostFeatures(ts=NOW, lang="en", hashtags=[], links=[], emoji=[], sentiment=None)) + return copies + + store._copy_range = add_mid_merge + try: + stale = store.merged("5m", NOW + 1) # computed from the pre-add copies... + finally: + store._copy_range = orig + assert stale.n == before + # ...but NOT memoised: the next same-second call recomputes and sees the add. + assert store.merged("5m", NOW + 1).n == before + 1 + + def test_memo_does_not_leak_across_now(store): a = store.merged("5m", NOW) b = store.merged("5m", NOW + 6 * 60) @@ -79,3 +106,72 @@ def test_prune_drops_buckets_beyond_24h(): s.add(PostFeatures(ts=base + 1441 * 60, lang="en", hashtags=["new"], links=[], emoji=[], sentiment=None)) assert len(s._buckets) == 1 # the 1441-min-old bucket was pruned on insert assert "new" in s.merged("24h", base + 1441 * 60).tags + + +def test_prune_recovers_after_a_future_timestamp(): + # Regression: max()-anchored pruning let one bogus far-future event set a + # permanent retention floor that dropped every subsequent real event on insert + # (window stuck at zero until restart). Anchoring the floor to the minute being + # added lets the window recover; the stray future bucket is excluded by merged(). + s = WindowStore() + base_min = 20_000_000 + s.add(PostFeatures(ts=(base_min + 10_000_000) * 60.0, lang="en", hashtags=["bogus"], links=[], emoji=[], sentiment=None)) + for _ in range(3): # real events arriving after the bogus one must still register + s.add(PostFeatures(ts=base_min * 60.0, lang="en", hashtags=["real"], links=[], emoji=[], sentiment=None)) + m = s.merged("5m", base_min * 60.0) + assert m.n == 3 + assert m.tags["real"] == 3 + assert "bogus" not in m.tags + + +def test_concurrent_add_and_read_is_race_free(): + # Regression: consume() calls add() on the event-loop thread while the publish/ + # checkpoint loops read via asyncio.to_thread. Unsynchronised, iterating _buckets + # while add() inserts/prunes raised "RuntimeError: dictionary changed size during + # iteration". A tiny GIL switch interval forces a thread hand-off mid-iteration so + # the race is deterministic without the lock; with the lock it can never happen. + def _post(offset: int) -> PostFeatures: + return PostFeatures( + ts=(20_000_000 + offset) * 60.0, + lang="en", + hashtags=[f"t{offset % 30}"], + links=[], + emoji=["🔥"], + sentiment=0.5, + ) + + store = WindowStore(max_minutes=400) + for i in range(400): # seed buckets so one iteration spans several switch points + store.add(_post(i)) + + errors: list[str] = [] + start = threading.Barrier(2) + + def writer(): + start.wait() + try: + for i in range(5000): + store.add(_post(400 + i)) + except Exception as exc: + errors.append(repr(exc)) + + old_interval = sys.getswitchinterval() + sys.setswitchinterval(1e-9) + # daemon: a genuinely deadlocked writer must fail the is_alive() assert + # below, not wedge interpreter shutdown after the join times out. + t = threading.Thread(target=writer, daemon=True) + try: + t.start() + start.wait() + for i in range(2000): + now = (20_000_400 + i) * 60.0 + store.snapshot(now) + store.merged("24h", now) + except Exception as exc: + errors.append(repr(exc)) + finally: + t.join(timeout=10) + sys.setswitchinterval(old_interval) + + assert not t.is_alive(), "writer thread did not finish (possible deadlock)" + assert not errors, f"race detected: {errors[:3]}"