mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
fix(env): leak-proof received-delta — durable persistent baseline + non-empty seed guard
Two live backlog-broadcast leaks traced to the in-memory first-poll seed: (1) incremental-fetch adapters (wzdx: registry tick [0 events] then feeds tick [many]) got marked _seeded on the EMPTY first tick, so the real batch next tick all looked "new" and broadcast; (2) in-memory seed lost on restart. Fix — durable baseline + guard: - _seed_from_persistent() at store init: pre-load already-received item keys from the persistent hazard tables into self._seen, so nothing ever received can re-broadcast (immune to fetch staging + restart). Only sources whose native emit key PROVABLY equals a persistent key are durably seeded: wzdx (traffic_events.external_id) + usgs_quake (quake_events.event_id). Resilient (per-table try/except; missing table -> skip). - _seen_key() now namespaces by evt["source"] (matches persistent tables), via shared _key_ext/_key_eid helpers used by both seed and live emit so they can't drift. - non-empty-seed guard: _ingest marks only sources that carried >=1 event this poll as _seeded -> an empty first tick can never seed-then-leak. This is the root-cause fix; covers all adapters (roads511/traffic fetch atomically per tick, so the guard fully protects them). - storage untouched (self._events populated for every event); Central path/deciders untouched. Live-DB verified: seed pre-loads 784 wzdx + 8 quake keys -> a live wzdx poll of 784 known zones broadcasts 0. +6 tests (incremental staging, restart, persistent-preseed, fresh-DB fallback); suite at 10-failure baseline. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson
parent
0c76a806b8
commit
0b4c0e2814
3 changed files with 419 additions and 25 deletions
203
work/meshai/env/store.py
vendored
203
work/meshai/env/store.py
vendored
|
|
@ -12,6 +12,23 @@ if TYPE_CHECKING:
|
|||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── Received-delta key format ────────────────────────────────────────────────
|
||||
# A single, shared key format used by BOTH the live emit path (_seen_key) and
|
||||
# the durable startup pre-seed (_seed_from_persistent). Keeping the two on one
|
||||
# helper guarantees they can never drift apart — a drift would silently defeat
|
||||
# the pre-seed (keys wouldn't match), which is exactly the leak we are fixing.
|
||||
_SEP = "\x1e"
|
||||
|
||||
|
||||
def _key_ext(source: str, external_id) -> str:
|
||||
"""Seen-key for an item identified by its upstream external_id."""
|
||||
return f"{source}{_SEP}ext:{external_id}"
|
||||
|
||||
|
||||
def _key_eid(source: str, event_id) -> str:
|
||||
"""Seen-key for an item identified by the adapter's stable event_id."""
|
||||
return f"{source}{_SEP}eid:{event_id}"
|
||||
|
||||
|
||||
class EnvironmentalStore:
|
||||
"""Cache and tick-driver for all environmental feed adapters."""
|
||||
|
|
@ -32,19 +49,31 @@ class EnvironmentalStore:
|
|||
self._region_anchors = region_anchors or []
|
||||
|
||||
# ── Received-delta gate (NATIVE-only) ────────────────────────────
|
||||
# Per-adapter set of item keys seen in PRIOR polls, plus the set of
|
||||
# adapters whose first poll has completed. The model the operator
|
||||
# demanded: a native adapter broadcasts an item ONLY when it was newly
|
||||
# RECEIVED from the API this poll — never by scanning an accumulated
|
||||
# backlog. So the FIRST poll for an adapter records every current key
|
||||
# as "seen" and emits NOTHING (that batch is pre-existing backlog, not
|
||||
# ours to announce); every later poll emits only keys not already
|
||||
# seen. In-memory only (per process) — a restart empties the sets, so
|
||||
# the next poll is a fresh "first poll" that re-seeds silently. This
|
||||
# makes broadcasting a backlog item structurally impossible on cold
|
||||
# start, restart, or after the adapter was disabled for days.
|
||||
self._seen: dict[str, set] = {} # adapter name -> set of item keys
|
||||
self._seeded: set[str] = set() # adapters past their first poll
|
||||
# The model the operator demanded: a native adapter broadcasts an item
|
||||
# ONLY when it was newly RECEIVED from the API this poll — never by
|
||||
# scanning an accumulated backlog. Two layers enforce this:
|
||||
#
|
||||
# 1. DURABLE pre-seed (_seed_from_persistent, below). meshai already
|
||||
# has a durable record of everything it has ever received: the
|
||||
# persistent hazard tables. At startup we load the identifying
|
||||
# keys of every already-received item into `self._seen`, so nothing
|
||||
# ever received can re-broadcast — immune to fetch staging AND to
|
||||
# restarts. Keyed by the event's `source` (NOT adapter name) so the
|
||||
# seed and the live-emit key line up on the same value.
|
||||
# 2. IN-MEMORY first-poll seed. The FIRST *non-empty* poll for a
|
||||
# source records every current key as "seen" and emits NOTHING
|
||||
# (that batch is pre-existing backlog); later polls emit only keys
|
||||
# not already seen. Belt-and-suspenders: a source is marked
|
||||
# "seeded" ONLY after a non-empty ingest, so an empty first poll
|
||||
# (e.g. wzdx's registry-only tick) can never mark it seeded and
|
||||
# then leak the real batch on the next tick.
|
||||
#
|
||||
# `self._seen` is keyed by SOURCE (evt["source"]), matching the durable
|
||||
# pre-seed. Durable keys added at startup are never removed, so they
|
||||
# suppress a backlog item even AFTER the source is marked seeded (this
|
||||
# is what closes cross-tick, non-empty staging leaks).
|
||||
self._seen: dict[str, set] = {} # source -> set of item keys
|
||||
self._seeded: set[str] = set() # sources past their first non-empty poll
|
||||
|
||||
# Create adapter instances with error isolation
|
||||
self._register_adapter("nws", config.nws, ".nws", "NWSAlertsAdapter",
|
||||
|
|
@ -97,6 +126,11 @@ class EnvironmentalStore:
|
|||
logger.warning("Failed adapters: %s", list(self._failed_adapters.keys()))
|
||||
logger.info(f"EnvironmentalStore initialized with {len(self._adapters)} adapters")
|
||||
|
||||
# Durable pre-seed: load every already-received item key from the
|
||||
# persistent hazard tables into `self._seen` so nothing ever received
|
||||
# can re-broadcast, regardless of fetch staging or process restarts.
|
||||
self._seed_from_persistent()
|
||||
|
||||
|
||||
def _register_adapter(self, name: str, cfg, module_path: str, class_name: str, args_fn):
|
||||
"""Register a single adapter with error isolation."""
|
||||
|
|
@ -129,7 +163,7 @@ class EnvironmentalStore:
|
|||
self._purge_expired()
|
||||
return changed
|
||||
|
||||
def _seen_key(self, name: str, raw_evt: dict) -> str:
|
||||
def _seen_key(self, raw_evt: dict) -> str:
|
||||
"""Derive a STABLE per-item key for the received-delta gate.
|
||||
|
||||
Stability across polls is the whole point: the SAME real-world item
|
||||
|
|
@ -144,16 +178,23 @@ class EnvironmentalStore:
|
|||
``avy_<center>_<zone>``, etc. — all stable across polls).
|
||||
3. content hash — last-resort fallback if an item somehow carries no
|
||||
id at all.
|
||||
Namespaced by adapter so two feeds never cross-contaminate.
|
||||
|
||||
Namespaced by the event's ``source`` (NOT the adapter name). The
|
||||
persistent hazard tables key on ``source`` (e.g. ``wzdx``), so keying
|
||||
the live-emit path on the same value is what lets the durable startup
|
||||
pre-seed line up with what the adapter emits. Every native raw event
|
||||
carries a stable ``source``; if one somehow does not, fall back to a
|
||||
neutral namespace so two feeds still never cross-contaminate.
|
||||
"""
|
||||
source = raw_evt.get("source") or "?"
|
||||
ext = raw_evt.get("external_id")
|
||||
if ext:
|
||||
return f"{name}\x1eext:{ext}"
|
||||
return _key_ext(source, ext)
|
||||
eid = raw_evt.get("event_id")
|
||||
if eid:
|
||||
return f"{name}\x1eeid:{eid}"
|
||||
return _key_eid(source, eid)
|
||||
blob = json.dumps(raw_evt, sort_keys=True, default=str)
|
||||
return f"{name}\x1ehash:" + hashlib.sha1(blob.encode()).hexdigest()[:16]
|
||||
return f"{source}{_SEP}hash:" + hashlib.sha1(blob.encode()).hexdigest()[:16]
|
||||
|
||||
def _delta_emit(self, name: str, adapter, raw_evt: dict, force: bool = False):
|
||||
"""Received-delta gate — emit ONLY items newly received THIS poll.
|
||||
|
|
@ -172,15 +213,16 @@ class EnvironmentalStore:
|
|||
content filtering (severity/threshold/impact, decider gates) still runs
|
||||
downstream in ``_emit_event``.
|
||||
"""
|
||||
seen = self._seen.setdefault(name, set())
|
||||
key = self._seen_key(name, raw_evt)
|
||||
first_poll = name not in self._seeded
|
||||
source = raw_evt.get("source") or name
|
||||
seen = self._seen.setdefault(source, set())
|
||||
key = self._seen_key(raw_evt)
|
||||
first_poll = source not in self._seeded
|
||||
|
||||
if first_poll:
|
||||
seen.add(key) # seed silently — this is backlog
|
||||
return
|
||||
if key in seen and not force:
|
||||
return # already received in a prior poll
|
||||
return # already received (prior poll OR durable pre-seed)
|
||||
seen.add(key)
|
||||
|
||||
if self._event_bus is not None and hasattr(adapter, "to_event"):
|
||||
|
|
@ -193,19 +235,30 @@ class EnvironmentalStore:
|
|||
adapter's FIRST data-bearing poll seeds the seen-set and broadcasts
|
||||
nothing; only items that newly appear on later polls are broadcast.
|
||||
``self._events`` is still maintained for state/queries as before.
|
||||
|
||||
Belt-and-suspenders: a source is marked "seeded" (past its first poll)
|
||||
ONLY after a NON-EMPTY ingest for that source. An empty poll — e.g.
|
||||
wzdx's registry-only tick that yields 0 events but returns ``tick()``
|
||||
True — must never mark a source seeded, or the next tick's real batch
|
||||
would be treated as "new" and leak. We collect the sources that
|
||||
actually carried ≥1 event this ingest and mark only those.
|
||||
"""
|
||||
touched: set[str] = set()
|
||||
|
||||
if name == "swpc":
|
||||
self._swpc_status = adapter.get_status()
|
||||
# Also ingest any alert events (R-scale >= 3)
|
||||
for evt in adapter.get_events():
|
||||
key = (evt["source"], evt["event_id"])
|
||||
self._events[key] = evt
|
||||
touched.add(evt.get("source") or name)
|
||||
self._delta_emit(name, adapter, evt)
|
||||
elif name == "ducting":
|
||||
self._ducting_status = adapter.get_status()
|
||||
for evt in adapter.get_events():
|
||||
key = (evt["source"], evt["event_id"])
|
||||
self._events[key] = evt
|
||||
touched.add(evt.get("source") or name)
|
||||
self._delta_emit(name, adapter, evt)
|
||||
elif name == "avalanche":
|
||||
# Avalanche: re-emit on danger_level rise (Update:) not just new
|
||||
|
|
@ -219,16 +272,118 @@ class EnvironmentalStore:
|
|||
level_rose = (prior is not None) and (
|
||||
evt.get("danger_level", -1) > prior_level)
|
||||
evt["_is_update"] = level_rose # signal to to_event()
|
||||
touched.add(evt.get("source") or name)
|
||||
self._delta_emit(name, adapter, evt, force=level_rose)
|
||||
self._events[key] = evt # always update stored state
|
||||
else:
|
||||
for evt in adapter.get_events():
|
||||
key = (evt["source"], evt["event_id"])
|
||||
self._events[key] = evt
|
||||
touched.add(evt.get("source") or name)
|
||||
self._delta_emit(name, adapter, evt)
|
||||
|
||||
# First poll for this adapter is now complete: later polls may emit.
|
||||
self._seeded.add(name)
|
||||
# First (non-empty) poll for these sources is now complete: later polls
|
||||
# may emit. Empty ingests touch nothing and so never mark a source
|
||||
# seeded — the leak-proof invariant behind the wzdx staging fix.
|
||||
self._seeded.update(touched)
|
||||
|
||||
def _seed_from_persistent(self) -> None:
|
||||
"""Pre-seed ``self._seen`` from the durable hazard tables at startup.
|
||||
|
||||
A row exists in these tables iff meshai has ALREADY RECEIVED that item,
|
||||
so loading their identifying keys guarantees a native adapter can never
|
||||
re-broadcast a previously-received item — immune to fetch staging and
|
||||
to process restarts. Durable keys are added to ``self._seen`` and,
|
||||
when ≥1 key is loaded for a source, that source is marked ``_seeded``
|
||||
so the durable set becomes its baseline: the very next poll emits ONLY
|
||||
items NOT already received (the operator's "newly received → send it
|
||||
now") and suppresses everything in the table — no silent first-poll
|
||||
needed, and no dependency on how the fetch is staged. Because durable
|
||||
keys are never removed, they keep suppressing backlog across every
|
||||
later poll, which is what closes cross-tick, non-empty staging leaks.
|
||||
(When a source has 0 durable rows we do NOT mark it seeded, so it falls
|
||||
back to the leak-proof in-memory silent first-poll seed — a fresh DB
|
||||
can never cause the first real batch to be treated as new.)
|
||||
|
||||
Only sources whose native emit key PROVABLY equals the persistent key
|
||||
are seeded here (verified against the live schema + adapter code):
|
||||
|
||||
* ``wzdx`` — the native WZDx adapter carries a stable
|
||||
``external_id`` on its raw event and the incident decider persists
|
||||
it as ``traffic_events(source='wzdx', external_id)``. Same value on
|
||||
both sides → ``_key_ext('wzdx', external_id)`` matches exactly.
|
||||
* ``usgs_quake`` — the native adapter's raw ``event_id`` is the bare
|
||||
USGS id (e.g. ``us6000t9bn``) and the quake decider persists it as
|
||||
``quake_events.event_id`` verbatim → ``_key_eid('usgs_quake', id)``
|
||||
matches exactly.
|
||||
|
||||
DELIBERATELY NOT seeded here (native emit key ≠ any persistent key —
|
||||
see the report; these stay protected by the in-memory first-poll +
|
||||
non-empty-seed guard, which is sufficient because each fetches
|
||||
atomically per tick rather than across ticks):
|
||||
roads511 / traffic (persistent rows are Central-keyed:
|
||||
itd_511/tomtom_incidents with idaho_511:event:* external_ids, but the
|
||||
native adapters emit source '511'/'traffic' with derived event_ids),
|
||||
fires (native ``nifc_<name>_<state>`` vs persistent IRWIN GUID),
|
||||
firms (native ``firms_<lat>_<lon>_<date>_<time>`` — no matching PK),
|
||||
satpass (native ``<norad>:<bucket>`` vs persistent
|
||||
``<norad>:<observer>:<bucket>``), nws (native CAP url vs persistent
|
||||
bare urn:oid), swpc (native ``swpc_<scale><level>`` vs timestamped
|
||||
persistent ids), usgs hydro (time-series, no per-item PK).
|
||||
|
||||
Resilient by construction: each table is loaded in its own try/except
|
||||
so a missing/renamed table (or an unavailable DB) logs and continues —
|
||||
startup never crashes. The SELECTs are cheap key-only scans of bounded
|
||||
tables.
|
||||
"""
|
||||
try:
|
||||
from meshai.persistence import get_db
|
||||
conn = get_db()
|
||||
except Exception as e:
|
||||
logger.warning("received-delta pre-seed skipped (DB unavailable): %s", e)
|
||||
return
|
||||
|
||||
# (source, description, SQL, row->key) — one spec per verified source.
|
||||
specs = [
|
||||
(
|
||||
"wzdx",
|
||||
"traffic_events(source='wzdx')",
|
||||
"SELECT external_id FROM traffic_events "
|
||||
"WHERE source='wzdx' AND external_id IS NOT NULL",
|
||||
lambda row: _key_ext("wzdx", row[0]),
|
||||
),
|
||||
(
|
||||
"usgs_quake",
|
||||
"quake_events",
|
||||
"SELECT event_id FROM quake_events WHERE event_id IS NOT NULL",
|
||||
lambda row: _key_eid("usgs_quake", row[0]),
|
||||
),
|
||||
]
|
||||
|
||||
total = 0
|
||||
for source, desc, sql, key_fn in specs:
|
||||
try:
|
||||
seen = self._seen.setdefault(source, set())
|
||||
n = 0
|
||||
for row in conn.execute(sql).fetchall():
|
||||
seen.add(key_fn(row))
|
||||
n += 1
|
||||
total += n
|
||||
if n:
|
||||
# Durable baseline established → skip the silent first-poll
|
||||
# seed; the next poll emits only genuinely-new items.
|
||||
self._seeded.add(source)
|
||||
logger.info(
|
||||
"received-delta pre-seed: %d key(s) from %s -> source %r%s",
|
||||
n, desc, source, " (marked seeded)" if n else "",
|
||||
)
|
||||
except Exception as e:
|
||||
# Missing/renamed table or bad row — never fatal at startup.
|
||||
logger.warning(
|
||||
"received-delta pre-seed: %s failed (continuing): %s",
|
||||
desc, e,
|
||||
)
|
||||
logger.info("received-delta pre-seed complete: %d durable key(s) loaded", total)
|
||||
|
||||
def _emit_event(self, adapter, raw_evt: dict):
|
||||
"""Convert raw event to pipeline Event and emit to bus.
|
||||
|
|
|
|||
|
|
@ -335,8 +335,12 @@ class TestAdapterTickFusion:
|
|||
# as already past its first-poll seed — otherwise the store's
|
||||
# received-delta gate would (correctly) seed this first batch silently.
|
||||
# First-poll seed-silence is covered by test_store_received_delta.
|
||||
# The gate keys on the event's ``source`` (not the adapter name), and
|
||||
# this adapter emits BOTH "firms" (raw hotspots) and "firms_fusion"
|
||||
# (consolidated growth) — mark both seeded, exactly as a non-empty
|
||||
# ingest would (store._ingest marks every source it touched).
|
||||
store._seen = {}
|
||||
store._seeded = {"firms"}
|
||||
store._seeded = {"firms", "firms_fusion"}
|
||||
assert a.tick() is True
|
||||
store._ingest("firms", a)
|
||||
|
||||
|
|
|
|||
|
|
@ -154,3 +154,238 @@ def test_disabled_for_days_then_backlog_is_not_broadcast():
|
|||
adapter.set_batch(backlog + ["fresh"])
|
||||
store.refresh()
|
||||
assert _emitted_ids(captured) == ["fresh"]
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
# Durable persistent pre-seed (the robust anti-leak fix)
|
||||
#
|
||||
# meshai already has a durable record of everything it has ever received: the
|
||||
# persistent hazard tables. At startup the store loads their identifying keys
|
||||
# into self._seen so nothing ever received can re-broadcast, immune to how the
|
||||
# fetch is staged and immune to restarts. These tests use the REAL persistent
|
||||
# DB (conftest points MESHAI_DB_PATH at a fresh migrated tmp file per test).
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
from meshai.persistence import get_db
|
||||
|
||||
|
||||
class _FakeWZDx:
|
||||
"""Native-WZDx stand-in. Its raw events carry a stable ``external_id``
|
||||
(like the real adapter), so the seen-key is ``wzdx\x1eext:<id>`` — exactly
|
||||
what the durable pre-seed loads from traffic_events(source='wzdx')."""
|
||||
|
||||
def __init__(self):
|
||||
self._batch: list[dict] = []
|
||||
|
||||
def set_batch(self, ext_ids: list[str]) -> None:
|
||||
self._batch = [
|
||||
{"source": "wzdx", "event_id": f"wzdx_{x}",
|
||||
"external_id": x, "fetched_at": 0}
|
||||
for x in ext_ids
|
||||
]
|
||||
|
||||
def tick(self) -> bool:
|
||||
# Real wzdx.tick() returns True even on a 0-event (registry-only) tick.
|
||||
return True
|
||||
|
||||
def get_events(self) -> list:
|
||||
return list(self._batch)
|
||||
|
||||
def to_event(self, raw_evt: dict):
|
||||
eid = raw_evt["external_id"]
|
||||
return make_event(source="wzdx", category="test_delta",
|
||||
severity="routine", title=eid, summary=eid,
|
||||
group_key=eid)
|
||||
|
||||
|
||||
class _FakeQuake:
|
||||
"""Native usgs_quake stand-in. Raw events have NO external_id, so the
|
||||
seen-key falls to ``usgs_quake\x1eeid:<event_id>`` — exactly what the
|
||||
durable pre-seed loads from quake_events.event_id."""
|
||||
|
||||
def __init__(self):
|
||||
self._batch: list[dict] = []
|
||||
|
||||
def set_batch(self, event_ids: list[str]) -> None:
|
||||
self._batch = [
|
||||
{"source": "usgs_quake", "event_id": e, "fetched_at": 0}
|
||||
for e in event_ids
|
||||
]
|
||||
|
||||
def tick(self) -> bool:
|
||||
return True
|
||||
|
||||
def get_events(self) -> list:
|
||||
return list(self._batch)
|
||||
|
||||
def to_event(self, raw_evt: dict):
|
||||
eid = raw_evt["event_id"]
|
||||
return make_event(source="usgs_quake", category="test_delta",
|
||||
severity="routine", title=eid, summary=eid,
|
||||
group_key=eid)
|
||||
|
||||
|
||||
def _insert_traffic(external_ids: list[str], source: str = "wzdx") -> None:
|
||||
conn = get_db()
|
||||
for x in external_ids:
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO traffic_events"
|
||||
"(source, external_id, first_seen_at, last_seen_at) "
|
||||
"VALUES (?,?,?,?)",
|
||||
(source, x, 0, 0),
|
||||
)
|
||||
|
||||
|
||||
def _insert_quake(event_ids: list[str]) -> None:
|
||||
conn = get_db()
|
||||
for e in event_ids:
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO quake_events(event_id, first_seen_at) "
|
||||
"VALUES (?,?)",
|
||||
(e, 0),
|
||||
)
|
||||
|
||||
|
||||
def _build_store(adapter_name: str, adapter):
|
||||
"""Construct a store (runs the durable pre-seed against the current DB),
|
||||
then inject a fake adapter. Insert persistent rows BEFORE calling this."""
|
||||
bus = EventBus()
|
||||
captured: list = []
|
||||
bus.subscribe(lambda e: captured.append(e))
|
||||
store = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus)
|
||||
store._adapters[adapter_name] = adapter
|
||||
return store, captured
|
||||
|
||||
|
||||
def test_persistent_preseed_known_suppressed_new_emitted():
|
||||
# N durable rows in traffic_events(source=wzdx). A fresh store must treat
|
||||
# them as already-received: a poll of those same N broadcasts NOTHING; a
|
||||
# poll adding one external_id NOT in the table broadcasts only that one.
|
||||
known = [f"z{i}" for i in range(5)]
|
||||
_insert_traffic(known)
|
||||
|
||||
adapter = _FakeWZDx()
|
||||
store, captured = _build_store("wzdx", adapter)
|
||||
|
||||
# wzdx was pre-seeded from the durable table AND marked seeded (baseline).
|
||||
assert "wzdx" in store._seeded
|
||||
assert len(store._seen["wzdx"]) == 5
|
||||
|
||||
adapter.set_batch(known)
|
||||
store.refresh()
|
||||
assert captured == [], "all 5 are durably-known → zero broadcast"
|
||||
|
||||
adapter.set_batch(known + ["z_new"])
|
||||
store.refresh()
|
||||
assert _emitted_ids(captured) == ["z_new"], "only the not-in-table id broadcasts"
|
||||
|
||||
|
||||
def test_persistent_preseed_cross_tick_staging_no_leak():
|
||||
# The scenario the DURABLE seed uniquely fixes: a source is marked seeded
|
||||
# on a partial tick, then a LATER tick brings a backlog item that was never
|
||||
# seen in-process. Without the durable record it would leak.
|
||||
_insert_traffic(["A", "B"]) # both already RECEIVED (durable)
|
||||
adapter = _FakeWZDx()
|
||||
store, captured = _build_store("wzdx", adapter)
|
||||
|
||||
adapter.set_batch(["A"])
|
||||
store.refresh() # tick 1: only A present
|
||||
adapter.set_batch(["A", "B"])
|
||||
store.refresh() # tick 2: B appears (backlog)
|
||||
assert captured == [], "B is durably-known — must NOT leak on a later tick"
|
||||
|
||||
# CONTROL: identical staging but NO durable rows → B leaks (proves the
|
||||
# durable seed is what prevents it; in-memory alone cannot).
|
||||
ctrl = _FakeWZDx()
|
||||
bus = EventBus(); cap2: list = []
|
||||
bus.subscribe(lambda e: cap2.append(e))
|
||||
# A separate source name so its 0-row durable seed doesn't mark it seeded.
|
||||
ctrl._batch = []
|
||||
store2 = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus)
|
||||
store2._adapters["wzdx_ctrl"] = ctrl
|
||||
ctrl.set_batch(["A"])
|
||||
# Re-point ctrl events to a fresh source with no durable rows.
|
||||
for e in ctrl._batch:
|
||||
e["source"] = "wzdx_ctrl"
|
||||
store2.refresh()
|
||||
ctrl.set_batch(["A", "B"])
|
||||
for e in ctrl._batch:
|
||||
e["source"] = "wzdx_ctrl"
|
||||
store2.refresh()
|
||||
assert [e.title for e in cap2] == ["B"], "without a durable record, B leaks"
|
||||
|
||||
|
||||
def test_incremental_empty_first_tick_then_only_new_broadcasts():
|
||||
# Task's incremental case: tick 1 yields [] (e.g. wzdx registry-only tick),
|
||||
# tick 2 yields [A,B,C] where A,B are durably-known and C is new.
|
||||
# The empty tick must not mark-and-leak; the durable seed catches A,B; only
|
||||
# C — genuinely never received — broadcasts.
|
||||
_insert_traffic(["A", "B"])
|
||||
adapter = _FakeWZDx()
|
||||
store, captured = _build_store("wzdx", adapter)
|
||||
|
||||
adapter.set_batch([]) # empty first tick
|
||||
store.refresh()
|
||||
assert captured == [], "empty tick emits nothing"
|
||||
|
||||
adapter.set_batch(["A", "B", "C"]) # backlog A,B + new C
|
||||
store.refresh()
|
||||
assert _emitted_ids(captured) == ["C"], "only the never-received C broadcasts"
|
||||
|
||||
|
||||
def test_restart_against_same_persistent_db_never_rebroadcasts():
|
||||
# A full received backlog is durable. Process 1 broadcasts nothing for it.
|
||||
# After a RESTART (fresh store, same DB) the backlog still never broadcasts.
|
||||
backlog = ["A", "B", "C", "D"]
|
||||
_insert_traffic(backlog)
|
||||
|
||||
a1 = _FakeWZDx()
|
||||
store1, cap1 = _build_store("wzdx", a1)
|
||||
a1.set_batch(backlog)
|
||||
store1.refresh()
|
||||
assert cap1 == [], "process 1: durable backlog is silent"
|
||||
|
||||
# RESTART: brand-new store, same persistent DB → pre-seed reloads.
|
||||
a2 = _FakeWZDx()
|
||||
store2, cap2 = _build_store("wzdx", a2)
|
||||
a2.set_batch(backlog)
|
||||
store2.refresh()
|
||||
assert cap2 == [], "restart must NEVER re-broadcast the durable backlog"
|
||||
|
||||
a2.set_batch(backlog + ["E"])
|
||||
store2.refresh()
|
||||
assert _emitted_ids(cap2) == ["E"], "a genuinely-new item still broadcasts once"
|
||||
|
||||
|
||||
def test_persistent_preseed_quake_by_event_id():
|
||||
# Durable seed for the event_id-keyed path (no external_id): quake_events.
|
||||
_insert_quake(["us1000aaaa", "us1000bbbb"])
|
||||
adapter = _FakeQuake()
|
||||
store, captured = _build_store("usgs_quake", adapter)
|
||||
|
||||
assert "usgs_quake" in store._seeded
|
||||
assert len(store._seen["usgs_quake"]) == 2
|
||||
|
||||
adapter.set_batch(["us1000aaaa", "us1000bbbb"])
|
||||
store.refresh()
|
||||
assert captured == [], "both quakes already received → zero broadcast"
|
||||
|
||||
adapter.set_batch(["us1000aaaa", "us1000bbbb", "us1000cccc"])
|
||||
store.refresh()
|
||||
assert _emitted_ids(captured) == ["us1000cccc"], "only the new quake broadcasts"
|
||||
|
||||
|
||||
def test_no_durable_rows_falls_back_to_silent_first_poll():
|
||||
# Fresh DB (0 durable rows): the source must NOT be pre-marked seeded, so
|
||||
# the first real batch is silently seeded (never leaked) — the fresh-start
|
||||
# safety property.
|
||||
adapter = _FakeWZDx()
|
||||
store, captured = _build_store("wzdx", adapter)
|
||||
assert "wzdx" not in store._seeded, "0 durable rows → not pre-marked seeded"
|
||||
|
||||
adapter.set_batch(["A", "B"])
|
||||
store.refresh()
|
||||
assert captured == [], "first non-empty poll on a fresh DB is silent"
|
||||
|
||||
adapter.set_batch(["A", "B", "C"])
|
||||
store.refresh()
|
||||
assert _emitted_ids(captured) == ["C"]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue