mirror of
https://github.com/zvx-echo6/central.git
synced 2026-08-26 17:31:39 +00:00
v0.14.2: exempt global satellite telemetry from bbox filter
The v0.14.0 monitoring-area bbox filter is Idaho-shaped (correct for
terrestrial adapters) but globally-scoped satellite telemetry is a
worldwide firehose, not a regional feed. Post-v0.14.0 audit found the
filter dropping 462k+ sat_positions and 87k+ sat_orbits events at
supervisor publish alone -- only ~0.3% (sub-sat point over Idaho) survived.
meshAI's "where is the ISS" global queries against central.sat.position.*
need the full firehose.
Carve-out (global-by-design only):
bypass: sat_positions, sat_orbits
untouched: celestrak_tle (null-geom, already passes),
satpass_predict + n2yo_visualpasses (observer-anchored at
Idaho QTHs -> centroids always in-bounds, change is a no-op)
- SourceAdapter.bypass_bbox_filter (default False); sat_positions/sat_orbits
override True
- supervisor._run_adapter_loop skips classify_geom_areas when the flag is
set; separate self._bypassed_publish counter for observability
- archive._process_message mirrors via module-level _BYPASS_BBOX_ADAPTERS
(consumer has only the adapter name on the wire, not the class)
- drift-detection test asserts the two truth sources stay in sync
- INFO "Bypass-bbox adapters: [...]" logged at supervisor + archive startup
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
c2864a94ce
commit
facb940227
8 changed files with 212 additions and 45 deletions
|
|
@ -57,6 +57,15 @@ class SourceAdapter(ABC):
|
||||||
gauges) that would drown discrete-event signal; shown on /telemetry instead.
|
gauges) that would drown discrete-event signal; shown on /telemetry instead.
|
||||||
GUI-only — does not affect publishing or the events.json contract."""
|
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
|
@abstractmethod
|
||||||
async def poll(self) -> AsyncIterator[Event]:
|
async def poll(self) -> AsyncIterator[Event]:
|
||||||
"""
|
"""
|
||||||
|
|
|
||||||
|
|
@ -121,6 +121,11 @@ class SatOrbitsAdapter(SourceAdapter):
|
||||||
default_cadence_s = 300 # 5 min
|
default_cadence_s = 300 # 5 min
|
||||||
data_class = "telemetry"
|
data_class = "telemetry"
|
||||||
enrichment_locations = []
|
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__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
|
|
||||||
|
|
@ -130,6 +130,12 @@ class SatPositionsAdapter(SourceAdapter):
|
||||||
default_cadence_s = 60
|
default_cadence_s = 60
|
||||||
data_class = "telemetry"
|
data_class = "telemetry"
|
||||||
enrichment_locations = []
|
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__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
|
|
||||||
|
|
@ -36,6 +36,15 @@ BATCH_SIZE = 100
|
||||||
FETCH_TIMEOUT = 5.0
|
FETCH_TIMEOUT = 5.0
|
||||||
ACK_WAIT = 30
|
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:
|
def consumer_name_for(stream: str) -> str:
|
||||||
"""Generate consumer name for a stream."""
|
"""Generate consumer name for a stream."""
|
||||||
|
|
@ -89,6 +98,9 @@ class ArchiveConsumer:
|
||||||
self._shutdown_event = asyncio.Event()
|
self._shutdown_event = asyncio.Event()
|
||||||
self._monitoring_areas: list[MonitoringArea] = []
|
self._monitoring_areas: list[MonitoringArea] = []
|
||||||
self._dropped: dict[str, int] = {}
|
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
|
@property
|
||||||
def _monitoring_area(self) -> MonitoringArea | None:
|
def _monitoring_area(self) -> MonitoringArea | None:
|
||||||
|
|
@ -249,6 +261,15 @@ class ArchiveConsumer:
|
||||||
|
|
||||||
geom_json = build_geom_json(geo_data)
|
geom_json = build_geom_json(geo_data)
|
||||||
|
|
||||||
|
# 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)
|
verdict = classify_geom_areas(geom_json, self._monitoring_areas)
|
||||||
if verdict == "out-of-bounds":
|
if verdict == "out-of-bounds":
|
||||||
self._dropped[adapter] = self._dropped.get(adapter, 0) + 1
|
self._dropped[adapter] = self._dropped.get(adapter, 0) + 1
|
||||||
|
|
@ -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:
|
async def run(self) -> None:
|
||||||
"""Run consume loops for all streams until shutdown."""
|
"""Run consume loops for all streams until shutdown."""
|
||||||
|
|
|
||||||
|
|
@ -245,6 +245,10 @@ class Supervisor:
|
||||||
# every MONITORING_AREA_REFRESH_S from config.system.
|
# every MONITORING_AREA_REFRESH_S from config.system.
|
||||||
self._monitoring_areas: list[MonitoringArea] = []
|
self._monitoring_areas: list[MonitoringArea] = []
|
||||||
self._dropped_publish: dict[str, int] = {}
|
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
|
@property
|
||||||
def _monitoring_area(self) -> MonitoringArea | None:
|
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:
|
async def disconnect(self) -> None:
|
||||||
"""Disconnect from NATS."""
|
"""Disconnect from NATS."""
|
||||||
|
|
@ -334,10 +345,13 @@ class Supervisor:
|
||||||
"Could not refresh monitoring area; keeping previous value",
|
"Could not refresh monitoring area; keeping previous value",
|
||||||
extra={"error": str(e)},
|
extra={"error": str(e)},
|
||||||
)
|
)
|
||||||
if self._dropped_publish:
|
if self._dropped_publish or self._bypassed_publish:
|
||||||
logger.info(
|
logger.info(
|
||||||
"publish bbox filter drop summary (cumulative)",
|
"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:
|
def _create_adapter(self, config: AdapterConfig) -> SourceAdapter:
|
||||||
|
|
@ -425,12 +439,22 @@ class Supervisor:
|
||||||
|
|
||||||
subject = state.adapter.subject_for(event)
|
subject = state.adapter.subject_for(event)
|
||||||
|
|
||||||
|
# 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
|
||||||
|
)
|
||||||
|
else:
|
||||||
# v0.10.2 publish-time monitoring-area filter. Mirrors
|
# v0.10.2 publish-time monitoring-area filter. Mirrors
|
||||||
# archive's classify->ACK pattern but here we just `continue`
|
# archive's classify->ACK pattern but here we just
|
||||||
# without mark_published -- if the area widens later the
|
# `continue` without mark_published -- if the area widens
|
||||||
# next poll re-yields the same id and we'll publish it
|
# later the next poll re-yields the same id and we'll
|
||||||
# naturally. Marking published on drop would be a forward-
|
# publish it naturally. Marking published on drop would
|
||||||
# only blackhole.
|
# be a forward-only blackhole.
|
||||||
geom_json = build_geom_json(
|
geom_json = build_geom_json(
|
||||||
event.geo.model_dump() if event.geo else None
|
event.geo.model_dump() if event.geo else None
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -117,3 +117,30 @@ class TestProcessMessageMultiArea:
|
||||||
await c._process_message(_make_msg(_envelope("wzdx", -74.0, 40.7)), conn)
|
await c._process_message(_make_msg(_envelope("wzdx", -74.0, 40.7)), conn)
|
||||||
conn.execute.assert_awaited_once()
|
conn.execute.assert_awaited_once()
|
||||||
assert c._dropped == {}
|
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 == {}
|
||||||
|
|
|
||||||
23
tests/test_bypass_bbox_consistency.py
Normal file
23
tests/test_bypass_bbox_consistency.py
Normal file
|
|
@ -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)}"
|
||||||
|
)
|
||||||
|
|
@ -32,6 +32,9 @@ def _ev(eid: str, geo: Geo) -> Event:
|
||||||
class _MockAdapter:
|
class _MockAdapter:
|
||||||
requires_api_key = None
|
requires_api_key = None
|
||||||
enrichment_locations = ()
|
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:
|
def __init__(self, config, *_args) -> None:
|
||||||
self.config = config
|
self.config = config
|
||||||
|
|
@ -90,8 +93,8 @@ def sup_factory():
|
||||||
return _build
|
return _build
|
||||||
|
|
||||||
|
|
||||||
async def _drive(sup, events):
|
async def _drive(sup, events, adapter_cls=_MockAdapter):
|
||||||
adapter = _MockAdapter(MagicMock(cadence_s=3600))
|
adapter = adapter_cls(MagicMock(cadence_s=3600))
|
||||||
adapter.events = events
|
adapter.events = events
|
||||||
config = AdapterConfig(
|
config = AdapterConfig(
|
||||||
name="mock", enabled=True, cadence_s=3600, settings={},
|
name="mock", enabled=True, cadence_s=3600, settings={},
|
||||||
|
|
@ -187,3 +190,50 @@ async def test_refresh_loop_reloads_area_and_logs_summary(
|
||||||
assert any(
|
assert any(
|
||||||
"publish bbox filter drop summary" in r.message for r in caplog.records
|
"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 == {}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue