diff --git a/src/central/adapter.py b/src/central/adapter.py index 3ee575b..8a82c94 100644 --- a/src/central/adapter.py +++ b/src/central/adapter.py @@ -57,6 +57,15 @@ class SourceAdapter(ABC): gauges) that would drown discrete-event signal; shown on /telemetry instead. GUI-only — does not affect publishing or the events.json contract.""" + bypass_bbox_filter: bool = False + """Set True for adapters whose events are global-by-design (satellite + telemetry, space weather) and should never be filtered by a geographic + monitoring area. The supervisor publishes such events unconditionally, + skipping the publish-time ``classify_geom_areas`` check (v0.14.2). The + archive consumer mirrors this via the module-level ``_BYPASS_BBOX_ADAPTERS`` + set in ``central.archive`` -- the two MUST stay in sync (enforced by + ``tests/test_bypass_bbox_consistency.py``).""" + @abstractmethod async def poll(self) -> AsyncIterator[Event]: """ diff --git a/src/central/adapters/sat_orbits.py b/src/central/adapters/sat_orbits.py index 505e4fe..8b74721 100644 --- a/src/central/adapters/sat_orbits.py +++ b/src/central/adapters/sat_orbits.py @@ -121,6 +121,11 @@ class SatOrbitsAdapter(SourceAdapter): default_cadence_s = 300 # 5 min data_class = "telemetry" enrichment_locations = [] + # Global-by-design: forward-orbit LineStrings (mostly polar) span the globe, + # not a region. A geographic monitoring area would drop nearly all of them. + # Skip the publish-time/archive bbox filter (v0.14.2). Keep in sync with + # archive._BYPASS_BBOX_ADAPTERS. + bypass_bbox_filter = True def __init__( self, diff --git a/src/central/adapters/sat_positions.py b/src/central/adapters/sat_positions.py index f635a3f..91d8bdb 100644 --- a/src/central/adapters/sat_positions.py +++ b/src/central/adapters/sat_positions.py @@ -130,6 +130,12 @@ class SatPositionsAdapter(SourceAdapter): default_cadence_s = 60 data_class = "telemetry" enrichment_locations = [] + # Global-by-design: one event per satellite per poll with the sub-satellite + # point as centroid -- a worldwide firehose. A geographic monitoring area + # (e.g. Idaho) would drop ~99.7% of it, defeating "where is the ISS" global + # queries. Skip the publish-time/archive bbox filter (v0.14.2). Keep in sync + # with archive._BYPASS_BBOX_ADAPTERS. + bypass_bbox_filter = True def __init__( self, diff --git a/src/central/archive.py b/src/central/archive.py index f448e91..4380fee 100644 --- a/src/central/archive.py +++ b/src/central/archive.py @@ -36,6 +36,15 @@ BATCH_SIZE = 100 FETCH_TIMEOUT = 5.0 ACK_WAIT = 30 +# v0.14.2: adapters whose events are global-by-design (satellite telemetry) and +# must bypass the geographic monitoring-area bbox filter. The archive consumer +# only has the adapter NAME at runtime (it reads off the wire, not the adapter +# class), so it can't reach SourceAdapter.bypass_bbox_filter directly -- this +# static set is the consumer-side mirror. It MUST stay in sync with the adapter +# classes that set bypass_bbox_filter = True; tests/test_bypass_bbox_consistency.py +# fails CI if the two drift. +_BYPASS_BBOX_ADAPTERS = {"sat_positions", "sat_orbits"} + def consumer_name_for(stream: str) -> str: """Generate consumer name for a stream.""" @@ -89,6 +98,9 @@ class ArchiveConsumer: self._shutdown_event = asyncio.Event() self._monitoring_areas: list[MonitoringArea] = [] self._dropped: dict[str, int] = {} + # v0.14.2: events archived unconditionally because their adapter is in + # _BYPASS_BBOX_ADAPTERS (global-by-design satellite telemetry). + self._bypassed: dict[str, int] = {} @property def _monitoring_area(self) -> MonitoringArea | None: @@ -249,20 +261,29 @@ class ArchiveConsumer: geom_json = build_geom_json(geo_data) - verdict = classify_geom_areas(geom_json, self._monitoring_areas) - if verdict == "out-of-bounds": - self._dropped[adapter] = self._dropped.get(adapter, 0) + 1 - logger.debug( - "Dropped out-of-bounds event (archive bbox filter)", - extra={"id": event_id, "adapter": adapter, "category": category}, - ) - await msg.ack() - return - if verdict == "invalid-geom": - logger.warning( - "Geom could not be evaluated for bbox filter; archiving", - extra={"id": event_id, "adapter": adapter}, - ) + # v0.14.2: global-by-design adapters (satellite telemetry) bypass the + # geographic monitoring-area filter entirely -- mirrors the supervisor's + # SourceAdapter.bypass_bbox_filter carve-out, but keyed on adapter NAME + # because the consumer reads events off the wire, not the adapter class. + # _BYPASS_BBOX_ADAPTERS must stay in sync with the adapter class attrs + # (tests/test_bypass_bbox_consistency.py enforces it). + if adapter in _BYPASS_BBOX_ADAPTERS: + self._bypassed[adapter] = self._bypassed.get(adapter, 0) + 1 + else: + verdict = classify_geom_areas(geom_json, self._monitoring_areas) + if verdict == "out-of-bounds": + self._dropped[adapter] = self._dropped.get(adapter, 0) + 1 + logger.debug( + "Dropped out-of-bounds event (archive bbox filter)", + extra={"id": event_id, "adapter": adapter, "category": category}, + ) + await msg.ack() + return + if verdict == "invalid-geom": + logger.warning( + "Geom could not be evaluated for bbox filter; archiving", + extra={"id": event_id, "adapter": adapter}, + ) try: if geom_json: @@ -377,6 +398,8 @@ class ArchiveConsumer: ], }, ) + # v0.14.2: surface the global-by-design carve-out for operators. + logger.info("Bypass-bbox adapters: %s", sorted(_BYPASS_BBOX_ADAPTERS)) async def run(self) -> None: """Run consume loops for all streams until shutdown.""" diff --git a/src/central/supervisor.py b/src/central/supervisor.py index 407f0c2..7b269c6 100644 --- a/src/central/supervisor.py +++ b/src/central/supervisor.py @@ -245,6 +245,10 @@ class Supervisor: # every MONITORING_AREA_REFRESH_S from config.system. self._monitoring_areas: list[MonitoringArea] = [] self._dropped_publish: dict[str, int] = {} + # v0.14.2: events published unconditionally because their adapter sets + # bypass_bbox_filter (global-by-design satellite telemetry). Counted + # separately so operators can confirm the carve-out is doing work. + self._bypassed_publish: dict[str, int] = {} @property def _monitoring_area(self) -> MonitoringArea | None: @@ -286,6 +290,13 @@ class Supervisor: ], }, ) + # v0.14.2: surface the global-by-design carve-out so operators can see + # which adapters skip the bbox filter. + bypass_adapters = sorted( + name for name, cls in self._adapters.items() + if getattr(cls, "bypass_bbox_filter", False) + ) + logger.info("Bypass-bbox adapters: %s", bypass_adapters) async def disconnect(self) -> None: """Disconnect from NATS.""" @@ -334,10 +345,13 @@ class Supervisor: "Could not refresh monitoring area; keeping previous value", extra={"error": str(e)}, ) - if self._dropped_publish: + if self._dropped_publish or self._bypassed_publish: logger.info( "publish bbox filter drop summary (cumulative)", - extra={"dropped_by_adapter": dict(self._dropped_publish)}, + extra={ + "dropped_by_adapter": dict(self._dropped_publish), + "bypassed_by_adapter": dict(self._bypassed_publish), + }, ) def _create_adapter(self, config: AdapterConfig) -> SourceAdapter: @@ -425,35 +439,45 @@ class Supervisor: subject = state.adapter.subject_for(event) - # v0.10.2 publish-time monitoring-area filter. Mirrors - # archive's classify->ACK pattern but here we just `continue` - # without mark_published -- if the area widens later the - # next poll re-yields the same id and we'll publish it - # naturally. Marking published on drop would be a forward- - # only blackhole. - geom_json = build_geom_json( - event.geo.model_dump() if event.geo else None - ) - verdict = classify_geom_areas(geom_json, self._monitoring_areas) - if verdict == "out-of-bounds": - self._dropped_publish[state.name] = ( - self._dropped_publish.get(state.name, 0) + 1 + # v0.14.2: global-by-design adapters (satellite telemetry) + # bypass the geographic monitoring-area filter entirely -- + # their events are worldwide and a regional bbox would drop + # nearly all of them. Count separately for observability, then + # fall through to the shared publish path below. + if state.adapter.bypass_bbox_filter: + self._bypassed_publish[state.name] = ( + self._bypassed_publish.get(state.name, 0) + 1 ) - logger.debug( - "Dropped out-of-bounds event at publish (monitoring-area filter)", - extra={ - "id": event.id, - "adapter": state.name, - "category": event.category, - "subject": subject, - }, - ) - continue - if verdict == "invalid-geom": - logger.warning( - "Geom could not be evaluated for publish-time bbox filter; publishing", - extra={"id": event.id, "adapter": state.name}, + else: + # v0.10.2 publish-time monitoring-area filter. Mirrors + # archive's classify->ACK pattern but here we just + # `continue` without mark_published -- if the area widens + # later the next poll re-yields the same id and we'll + # publish it naturally. Marking published on drop would + # be a forward-only blackhole. + geom_json = build_geom_json( + event.geo.model_dump() if event.geo else None ) + verdict = classify_geom_areas(geom_json, self._monitoring_areas) + if verdict == "out-of-bounds": + self._dropped_publish[state.name] = ( + self._dropped_publish.get(state.name, 0) + 1 + ) + logger.debug( + "Dropped out-of-bounds event at publish (monitoring-area filter)", + extra={ + "id": event.id, + "adapter": state.name, + "category": event.category, + "subject": subject, + }, + ) + continue + if verdict == "invalid-geom": + logger.warning( + "Geom could not be evaluated for publish-time bbox filter; publishing", + extra={"id": event.id, "adapter": state.name}, + ) # Publish await self._publish_event(subject, envelope, msg_id) diff --git a/tests/test_archive_bbox_filter.py b/tests/test_archive_bbox_filter.py index e4a51ea..2ebf2a1 100644 --- a/tests/test_archive_bbox_filter.py +++ b/tests/test_archive_bbox_filter.py @@ -117,3 +117,30 @@ class TestProcessMessageMultiArea: await c._process_message(_make_msg(_envelope("wzdx", -74.0, 40.7)), conn) conn.execute.assert_awaited_once() assert c._dropped == {} + + +class TestBypassBboxAdapters: + """v0.14.2: _BYPASS_BBOX_ADAPTERS skip the bbox filter at archive time.""" + + @pytest.mark.asyncio + async def test_sat_positions_archived_when_out_of_bounds(self): + # ISS sub-sat point over NYC, only IDAHO configured -> normally dropped, + # but sat_positions is global-by-design so it bypasses and inserts. + c = ArchiveConsumer("nats://x", "postgresql://x") + c._monitoring_area = IDAHO + conn = AsyncMock() + await c._process_message(_make_msg(_envelope("sat_positions", -74.0, 40.7)), conn) + conn.execute.assert_awaited_once() + assert c._dropped == {} + assert c._bypassed == {"sat_positions": 1} + + @pytest.mark.asyncio + async def test_tomtom_flow_still_dropped(self): + # Regression guard: a non-bypass adapter is still filtered out-of-bounds. + c = ArchiveConsumer("nats://x", "postgresql://x") + c._monitoring_area = IDAHO + conn = AsyncMock() + await c._process_message(_make_msg(_envelope("tomtom_flow", -74.0, 40.7)), conn) + conn.execute.assert_not_called() + assert c._dropped == {"tomtom_flow": 1} + assert c._bypassed == {} diff --git a/tests/test_bypass_bbox_consistency.py b/tests/test_bypass_bbox_consistency.py new file mode 100644 index 0000000..182cd41 --- /dev/null +++ b/tests/test_bypass_bbox_consistency.py @@ -0,0 +1,23 @@ +"""v0.14.2 drift guard: the archive consumer mirrors the per-adapter +``bypass_bbox_filter`` flag as a static module-level set +(``archive._BYPASS_BBOX_ADAPTERS``) because the consumer reads events off the +wire and has no access to the adapter class at runtime. The two truth sources +must never diverge -- if someone adds a global-by-design adapter and sets +``bypass_bbox_filter = True`` but forgets the archive set (or vice versa), this +test fails at CI time. +""" +from central.adapter_discovery import discover_adapters +from central.archive import _BYPASS_BBOX_ADAPTERS + + +def test_archive_static_set_matches_adapter_class_attr(): + bypass_from_classes = { + name + for name, cls in discover_adapters().items() + if getattr(cls, "bypass_bbox_filter", False) + } + assert bypass_from_classes == _BYPASS_BBOX_ADAPTERS, ( + "archive._BYPASS_BBOX_ADAPTERS is out of sync with the adapter classes " + "that set bypass_bbox_filter=True. Update the set in central.archive to " + f"match: {sorted(bypass_from_classes)}" + ) diff --git a/tests/test_supervisor_publish_filter.py b/tests/test_supervisor_publish_filter.py index 0a936ac..fed2c46 100644 --- a/tests/test_supervisor_publish_filter.py +++ b/tests/test_supervisor_publish_filter.py @@ -32,6 +32,9 @@ def _ev(eid: str, geo: Geo) -> Event: class _MockAdapter: requires_api_key = None enrichment_locations = () + bypass_bbox_filter = False # v0.14.2: real adapters inherit this from + # SourceAdapter; the mock declares it so _run_adapter_loop's direct attr + # access works. Bypass tests use _BypassMockAdapter (overrides True). def __init__(self, config, *_args) -> None: self.config = config @@ -90,8 +93,8 @@ def sup_factory(): return _build -async def _drive(sup, events): - adapter = _MockAdapter(MagicMock(cadence_s=3600)) +async def _drive(sup, events, adapter_cls=_MockAdapter): + adapter = adapter_cls(MagicMock(cadence_s=3600)) adapter.events = events config = AdapterConfig( name="mock", enabled=True, cadence_s=3600, settings={}, @@ -187,3 +190,50 @@ async def test_refresh_loop_reloads_area_and_logs_summary( assert any( "publish bbox filter drop summary" in r.message for r in caplog.records ) + + +# --- v0.14.2: global-by-design satellite telemetry bypasses the bbox filter --- + +class _BypassMockAdapter(_MockAdapter): + """Stand-in for a global-by-design adapter (sat_positions/sat_orbits).""" + bypass_bbox_filter = True + + +class TestBypassBboxAdapters: + @pytest.mark.asyncio + async def test_sat_positions_publishes_when_out_of_bounds(self, sup_factory): + """Centroid over NYC (outside Idaho) still publishes; not dropped.""" + sup = sup_factory(IDAHO) + a = await _drive( + sup, [_ev("iss", Geo(centroid=NYC))], adapter_cls=_BypassMockAdapter + ) + assert sup._publish_event.await_count == 1 + assert a.published == {"iss"} + assert sup._dropped_publish == {} + assert sup._bypassed_publish == {"mock": 1} + + @pytest.mark.asyncio + async def test_sat_orbits_publishes_when_linestring_outside_idaho(self, sup_factory): + """An equatorial forward-track LineString never intersects Idaho, yet publishes.""" + sup = sup_factory(IDAHO) + equator = Geo(geometry={ + "type": "LineString", + "coordinates": [[0.0, 0.0], [10.0, 0.0], [20.0, 1.0]], + }) + a = await _drive( + sup, [_ev("orbit1", equator)], adapter_cls=_BypassMockAdapter + ) + assert sup._publish_event.await_count == 1 + assert a.published == {"orbit1"} + assert sup._dropped_publish == {} + assert sup._bypassed_publish == {"mock": 1} + + @pytest.mark.asyncio + async def test_tomtom_flow_still_drops_when_out_of_bounds(self, sup_factory): + """Regression guard: a non-bypass adapter is still filtered (NYC drops).""" + sup = sup_factory(IDAHO) + a = await _drive(sup, [_ev("flow1", Geo(centroid=NYC))]) # _MockAdapter: bypass False + sup._publish_event.assert_not_called() + assert a.published == set() + assert sup._dropped_publish == {"mock": 1} + assert sup._bypassed_publish == {}