From e46960fae9f1d3bc368de8909ee01402e4c82f9f Mon Sep 17 00:00:00 2001 From: Babissimo Date: Thu, 10 Sep 2026 18:44:09 +0100 Subject: [PATCH 1/4] Carry the node_ref with mirrored detections A node whose detections are mirrored in from another environment has no registry row here, so the boundary could not name it and dropped it: on the test droplet that is all twelve real nodes, whose detections arrive over the bulk endpoint from production rather than from an enrolment here. The ref travels with the detections instead. Production resolves it from the row it owns and sends it alongside; the bulk endpoint records it on the connected-node entry; the resolver reads it only when the local registry has nothing, so a real row always wins. Nothing is minted locally, which keeps refs to one per node across environments rather than one per server, and the value is only as trusted as the API key the bulk endpoint is gated on. The reverse direction stays local: a mirrored ref does not resolve back to its node id, so the per-node path-parameter routes still answer only for nodes this server registered. Co-Authored-By: Claude Opus 5 --- backend/routes/radar.py | 5 +++ backend/services/detection_mirror.py | 9 +++++- backend/services/node_refs.py | 19 +++++++++++- backend/tests/test_node_refs.py | 46 ++++++++++++++++++++++++++++ 4 files changed, 77 insertions(+), 2 deletions(-) diff --git a/backend/routes/radar.py b/backend/routes/radar.py index ac53cfb1..fbf8ce2a 100644 --- a/backend/routes/radar.py +++ b/backend/routes/radar.py @@ -36,6 +36,10 @@ class DetectionRequest(BaseModel): class BulkNodeEntry(BaseModel): node_id: str = Field(default="http-node", max_length=128) + # The handle the sending environment publishes this node under. Mirrored + # nodes have no row here, so it is the only ref this server can name one + # by; see services/node_refs._mirrored_ref. + node_ref: str | None = Field(default=None, max_length=15) config: dict | None = None frames: list[dict] = Field(default_factory=list) @@ -214,6 +218,7 @@ async def ingest_detections_bulk( "peer": "http-bulk", "is_synthetic": is_synthetic_node(node_id), "capabilities": {}, + "node_ref": entry.node_ref, } # A cached pipeline was built from the config that was active when # it was created, and nothing else refreshes it. diff --git a/backend/services/detection_mirror.py b/backend/services/detection_mirror.py index 4d40b973..acfcac45 100644 --- a/backend/services/detection_mirror.py +++ b/backend/services/detection_mirror.py @@ -25,6 +25,7 @@ DETECTION_MIRROR_TIMEOUT_S, ) from core import state +from services import node_refs from services.node_pipeline import pipeline_frame if TYPE_CHECKING: @@ -140,7 +141,13 @@ def build_batch(items) -> list: config = dict(known["config"]) if known and known.get("config") else None if config is None: continue - entries.append({"node_id": node_id, "config": config, "frames": frames}) + # The receiving environment has no registry row for this node, so the + # ref goes with the detections or it cannot name the node at all. + entry = {"node_id": node_id, "config": config, "frames": frames} + ref = node_refs.ref_for(node_id) + if ref: + entry["node_ref"] = ref + entries.append(entry) return entries diff --git a/backend/services/node_refs.py b/backend/services/node_refs.py index 5adddd0f..8f589a6a 100644 --- a/backend/services/node_refs.py +++ b/backend/services/node_refs.py @@ -166,6 +166,23 @@ def public_name(name, fallback: str, known_ids: Iterable[str] = ()) -> str: return name +def _mirrored_ref(node_id: str) -> str | None: + """The ref another environment resolved for a node it mirrors to us. + + A mirrored node has no row here: its detections arrive over the bulk + endpoint (routes/radar.py) from the environment that holds its registry, + which sends the ref along with them. Read only when the local registry has + nothing, so a real row always wins, and the value is only as trusted as the + API key the bulk endpoint is gated on. + """ + from core import state + + with state.connected_nodes_lock: + known = state.connected_nodes.get(node_id) + ref = known.get("node_ref") if known else None + return ref if isinstance(ref, str) and ref else None + + def owner_identity(node_id: str | None) -> str | None: """What a node publishes as, for a caller that already knows the node. @@ -177,7 +194,7 @@ def owner_identity(node_id: str | None) -> str | None: return None if is_synthetic_node(node_id): return node_id - return ref_for(node_id) + return ref_for(node_id) or _mirrored_ref(node_id) def public_identity(node_id: str | None) -> str | None: diff --git a/backend/tests/test_node_refs.py b/backend/tests/test_node_refs.py index 4cc3853e..5f27f327 100644 --- a/backend/tests/test_node_refs.py +++ b/backend/tests/test_node_refs.py @@ -314,3 +314,49 @@ def test_an_unresolvable_id_inside_a_tuple_loses_its_place(self, seed): seed(ret1a2b3c4d="nde1a2b3c4d00") (out,) = node_refs.public_records([{"contributing_node_ids": ("ret1a2b3c4d", "ret0badcafe")}]) assert out == {"contributing_node_refs": ["nde1a2b3c4d00"]} + + +_MIRRORED = "ret0badcafe" + + +class TestMirroredRef: + """A node whose detections are mirrored in from another environment. + + It has no registry row here, so the registry cannot answer for it. The + mirror sends the ref the owning environment resolved, and the boundary + publishes that rather than dropping the node. + """ + + def _connected(self, **entry): + from core import state + + with state.connected_nodes_lock: + state.connected_nodes[_MIRRORED] = {"is_synthetic": False, "status": "active", **entry} + + def teardown_method(self): + from core import state + + with state.connected_nodes_lock: + state.connected_nodes.pop(_MIRRORED, None) + + def test_a_mirrored_ref_is_published_when_the_registry_has_none(self, seed): + seed(ret1a2b3c4d="nde1a2b3c4d00") + self._connected(node_ref="ndemirrored001") + assert node_refs.public_identity(_MIRRORED) == "ndemirrored001" + + def test_without_one_the_node_is_still_dropped(self, seed): + seed(ret1a2b3c4d="nde1a2b3c4d00") + self._connected() + assert node_refs.public_identity(_MIRRORED) is None + + def test_the_registry_wins_over_a_mirrored_ref(self, seed): + """A local row is the authority; a mirrored value cannot override it.""" + seed(**{_MIRRORED: "ndelocalrow001"}) + self._connected(node_ref="ndemirrored001") + assert node_refs.public_identity(_MIRRORED) == "ndelocalrow001" + + def test_a_mirrored_node_reaches_the_published_feed(self, seed): + seed(ret1a2b3c4d="nde1a2b3c4d00") + self._connected(node_ref="ndemirrored001") + out = node_refs.substitute_identities({"aircraft": [{"hex": "abc123", "node_id": _MIRRORED}]}) + assert out["aircraft"] == [{"hex": "abc123", "node_ref": "ndemirrored001"}] From b9ab45a60542bf25124b31a49b30fe9c5d2b2dc3 Mon Sep 17 00:00:00 2001 From: Babissimo Date: Thu, 10 Sep 2026 19:04:47 +0100 Subject: [PATCH 2/4] Validate the mirrored ref and record it on every call A review of the previous commit found two faults in it. The ingested node_ref was bounded only by length, and a node id is short enough to fit, so a bulk entry naming one would have had it published as that node's public handle: a raw node id on the public wire, reached through the fallback that exists to keep one off it. It now takes the same annotation a minted ref must satisfy, so a value of the wrong shape is refused at the edge. The ref was also recorded only where the node registers, and a ref arriving against an entry the server already holds moves neither `known` nor `changed`. That is the rolling upgrade this feature was written for, so it would have done nothing until a restart. Both branches record it now. Two smaller things while here: the collision case is refused rather than left to publish two nodes under one handle, and the lookup no longer takes connected_nodes_lock, which had put it on the 1 Hz publication path. Co-Authored-By: Claude Opus 5 --- backend/core/types.py | 1 + backend/routes/radar.py | 30 +++++++++++++++-- backend/services/node_refs.py | 19 ++++++----- backend/tests/test_detection_mirror.py | 38 +++++++++++++++++++++ backend/tests/test_radar_routes.py | 46 ++++++++++++++++++++++++++ 5 files changed, 123 insertions(+), 11 deletions(-) diff --git a/backend/core/types.py b/backend/core/types.py index 0ed09ce9..d812720c 100644 --- a/backend/core/types.py +++ b/backend/core/types.py @@ -20,6 +20,7 @@ class NodeState(TypedDict, total=False): peer: str is_synthetic: bool capabilities: dict + node_ref: str # mirrored nodes only; see services/node_refs._mirrored_ref class AircraftPosition(TypedDict, total=False): diff --git a/backend/routes/radar.py b/backend/routes/radar.py index fbf8ce2a..2f90c3c6 100644 --- a/backend/routes/radar.py +++ b/backend/routes/radar.py @@ -12,9 +12,10 @@ from config.constants import RATE_BUCKETS_MAX_IPS from core import state +from routes.node_schemas import NodeRef from core.users import require_admin from pipeline.passive_radar import PassiveRadarPipeline -from services import node_registration +from services import node_refs, node_registration from services.node_config import canonical_config from services.node_pipeline import config_hash from services.public_location import public_latlon @@ -38,8 +39,10 @@ class BulkNodeEntry(BaseModel): node_id: str = Field(default="http-node", max_length=128) # The handle the sending environment publishes this node under. Mirrored # nodes have no row here, so it is the only ref this server can name one - # by; see services/node_refs._mirrored_ref. - node_ref: str | None = Field(default=None, max_length=15) + # by (services/node_refs._mirrored_ref), which is why it is validated + # against the same pattern a minted ref must match: a value shaped like a + # node_id would otherwise be published as one. + node_ref: NodeRef | None = None config: dict | None = None frames: list[dict] = Field(default_factory=list) @@ -171,6 +174,21 @@ async def ingest_detections( } +def _record_mirrored_ref(node_id: str, node_ref: str | None) -> None: + """Record the handle the sending environment publishes this node under. + + Refused when another node already publishes under it: two nodes sharing a + handle would both answer to it on every surface while the reverse map named + only one, so the collision is dropped rather than resolved silently. + """ + if not node_ref or node_refs.id_for_ref(node_ref) not in (None, node_id): + return + with state.connected_nodes_lock: + known = state.connected_nodes.get(node_id) + if known is not None: + known["node_ref"] = node_ref + + @router.post("/api/radar/detections/bulk") async def ingest_detections_bulk( request: Request, @@ -220,6 +238,7 @@ async def ingest_detections_bulk( "capabilities": {}, "node_ref": entry.node_ref, } + _record_mirrored_ref(node_id, entry.node_ref) # A cached pipeline was built from the config that was active when # it was created, and nothing else refreshes it. node_registration.evict_pipeline(node_id) @@ -229,6 +248,11 @@ async def ingest_detections_bulk( with state.connected_nodes_lock: state.connected_nodes[node_id]["status"] = "active" state.connected_nodes[node_id]["last_heartbeat"] = datetime.now(timezone.utc).isoformat() + # Also on the unchanged path: a sender that starts sending refs + # against entries this server already holds moves neither `known` + # nor `changed`, so writing it only at registration would leave + # every existing node without one until a restart. + _record_mirrored_ref(node_id, entry.node_ref) for frame in frames: if "timestamp" not in frame: diff --git a/backend/services/node_refs.py b/backend/services/node_refs.py index 8f589a6a..74878ec4 100644 --- a/backend/services/node_refs.py +++ b/backend/services/node_refs.py @@ -24,6 +24,7 @@ from sqlalchemy import create_engine, select from sqlalchemy.pool import NullPool +from core import state from core.nodes import Node from core.users import DATABASE_URL from services.tcp_handler import is_synthetic_node @@ -171,15 +172,17 @@ def _mirrored_ref(node_id: str) -> str | None: A mirrored node has no row here: its detections arrive over the bulk endpoint (routes/radar.py) from the environment that holds its registry, - which sends the ref along with them. Read only when the local registry has - nothing, so a real row always wins, and the value is only as trusted as the - API key the bulk endpoint is gated on. + which sends the ref along with them, validated there against the same + pattern a minted one must match. Read only when the local registry has + nothing, so a real row always wins. + + Unlocked: both reads are single dict lookups, and an entry is replaced + wholesale rather than edited field by field, so a concurrent write yields + the old entry or the new one. Taking connected_nodes_lock here would put it + on the 1 Hz publication path, contending with the ingest that writes it. """ - from core import state - - with state.connected_nodes_lock: - known = state.connected_nodes.get(node_id) - ref = known.get("node_ref") if known else None + known = state.connected_nodes.get(node_id) + ref = known.get("node_ref") if known else None return ref if isinstance(ref, str) and ref else None diff --git a/backend/tests/test_detection_mirror.py b/backend/tests/test_detection_mirror.py index eae49d2a..8f2b6db4 100644 --- a/backend/tests/test_detection_mirror.py +++ b/backend/tests/test_detection_mirror.py @@ -492,3 +492,41 @@ def _task(): pass assert called == {"configured": 1, "task": 1} + + +class TestBatchCarriesTheRef: + """The receiving environment has no row for these nodes, so the ref has to + travel with the detections or it cannot name them at all.""" + + def _connected(self, node_id): + from core import state + + with state.connected_nodes_lock: + state.connected_nodes[node_id] = {"config": {"node_id": node_id}, "is_synthetic": False} + + def teardown_method(self): + from core import state + + with state.connected_nodes_lock: + for nid in ("ret1a2b3c4d", "ret9f8e7d6c"): + state.connected_nodes.pop(nid, None) + + def test_a_registered_node_sends_its_ref(self, monkeypatch): + from services import detection_mirror, node_refs + + self._connected("ret1a2b3c4d") + monkeypatch.setattr(node_refs, "ref_for", lambda nid: "nde1a2b3c4d00") + monkeypatch.setattr(detection_mirror, "pipeline_frame", lambda f: f) + (entry,) = detection_mirror.build_batch([("ret1a2b3c4d", {})]) + assert entry["node_ref"] == "nde1a2b3c4d00" + + def test_a_node_with_no_ref_sends_none_rather_than_a_null(self, monkeypatch): + """An absent key, not node_ref: null, so the receiver's own validation + never sees a value it would have to special-case.""" + from services import detection_mirror, node_refs + + self._connected("ret9f8e7d6c") + monkeypatch.setattr(node_refs, "ref_for", lambda nid: None) + monkeypatch.setattr(detection_mirror, "pipeline_frame", lambda f: f) + (entry,) = detection_mirror.build_batch([("ret9f8e7d6c", {})]) + assert "node_ref" not in entry diff --git a/backend/tests/test_radar_routes.py b/backend/tests/test_radar_routes.py index 299b1938..20ac0c53 100644 --- a/backend/tests/test_radar_routes.py +++ b/backend/tests/test_radar_routes.py @@ -362,3 +362,49 @@ def test_expired_timestamps_cleaned_up(self, monkeypatch): # Only the fresh timestamp added at the end of _check_rate_limit remains assert len(bucket) == 1 assert bucket[0] > old_ts + + +class TestBulkRecordsTheMirroredRef: + """The ingest end of the mirrored-ref path: a ref arriving with detections + is recorded on the connected-node entry, whether or not that call also + registers the node.""" + + NODE = "ret1a2b3c4d" + REF = "nde1a2b3c4d0000" + + def _entry(self): + from core import state + + with state.connected_nodes_lock: + return dict(state.connected_nodes.get(self.NODE) or {}) + + def teardown_method(self): + from core import state + + with state.connected_nodes_lock: + state.connected_nodes.pop(self.NODE, None) + + def _post(self, client, **extra): + body = {"nodes": [{"node_id": self.NODE, "frames": [], **extra}]} + return client.post("/api/radar/detections/bulk", json=body, headers=HEADERS_OK) + + def test_a_ref_arriving_at_registration_is_recorded(self, client): + assert self._post(client, node_ref=self.REF, config={"rx_lat": 34.0, "rx_lon": -82.0}).status_code == 200 + assert self._entry().get("node_ref") == self.REF + + def test_a_ref_arriving_later_is_still_recorded(self, client): + """The regression that mattered: a sender that starts sending refs + against entries this server already holds moves neither `known` nor + `changed`, so a registration-only write would never pick it up.""" + assert self._post(client, config={"rx_lat": 34.0, "rx_lon": -82.0}).status_code == 200 + assert self._entry().get("node_ref") is None + assert self._post(client, node_ref=self.REF).status_code == 200 + assert self._entry().get("node_ref") == self.REF + + def test_a_ref_shaped_like_a_node_id_is_refused(self, client): + """Publishing it would put a raw node id on the public wire through the + very fallback that exists to keep one off it.""" + assert self._post(client, node_ref="retdeadbeef").status_code == 422 + + def test_a_malformed_ref_is_refused(self, client): + assert self._post(client, node_ref="nde-not-valid!").status_code == 422 From 09a637bb28d3bf033294abf64f373c9c700e488f Mon Sep 17 00:00:00 2001 From: Babissimo Date: Mon, 14 Sep 2026 11:50:06 +0100 Subject: [PATCH 3/4] Sort the NodeRef import into the first-party block ruff's isort rule (I001) rejects the placement; the lint gate would have failed CI. Fold this into the commit above before merging. Co-Authored-By: Claude Opus 5 --- backend/routes/radar.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/routes/radar.py b/backend/routes/radar.py index 2f90c3c6..b851aa3c 100644 --- a/backend/routes/radar.py +++ b/backend/routes/radar.py @@ -12,9 +12,9 @@ from config.constants import RATE_BUCKETS_MAX_IPS from core import state -from routes.node_schemas import NodeRef from core.users import require_admin from pipeline.passive_radar import PassiveRadarPipeline +from routes.node_schemas import NodeRef from services import node_refs, node_registration from services.node_config import canonical_config from services.node_pipeline import config_hash From 5b8353d68f0925a067e1579df7a2f8ba4dde4503 Mon Sep 17 00:00:00 2001 From: Babissimo Date: Mon, 14 Sep 2026 12:02:45 +0100 Subject: [PATCH 4/4] Let the guard be the only writer of a mirrored ref The collision check lived in _record_mirrored_ref, but the registration branch had already written entry.node_ref straight into the new connected-node entry before calling it, and the guard refuses by returning rather than by clearing. So a colliding ref was seated anyway, and the check only ever bit on the already-known branch. Registration is first contact for a mirrored node, which made the unguarded path the common one rather than an edge case. Dropping node_ref from the dict literal leaves _record_mirrored_ref as the sole writer on both branches, so the two paths give the same guarantee. Two docstrings also claimed more than the code does. The guard consults the local registry only, so two mirrored nodes both claiming one ref are not caught; and _mirrored_ref justified its lock-free read on entries being replaced wholesale, which is not true of the single-key writes around it. The read is still safe, for a different reason, now stated. Found in review of this branch. Co-Authored-By: Claude Opus 5 --- backend/routes/radar.py | 13 +++++++++---- backend/services/node_refs.py | 8 ++++---- backend/tests/test_radar_routes.py | 9 +++++++++ 3 files changed, 22 insertions(+), 8 deletions(-) diff --git a/backend/routes/radar.py b/backend/routes/radar.py index b851aa3c..f875fc2c 100644 --- a/backend/routes/radar.py +++ b/backend/routes/radar.py @@ -177,9 +177,12 @@ async def ingest_detections( def _record_mirrored_ref(node_id: str, node_ref: str | None) -> None: """Record the handle the sending environment publishes this node under. - Refused when another node already publishes under it: two nodes sharing a - handle would both answer to it on every surface while the reverse map named - only one, so the collision is dropped rather than resolved silently. + Refused when the local registry already names another node under it: two + nodes sharing a handle would both answer to it on every surface while the + reverse map named only one, so the collision is dropped rather than + resolved silently. Only the registry is consulted, so two mirrored nodes + that both arrive claiming one ref, neither of them registered here, are not + caught; the sending environment's registry is what keeps refs unique. """ if not node_ref or node_refs.id_for_ref(node_ref) not in (None, node_id): return @@ -236,8 +239,10 @@ async def ingest_detections_bulk( "peer": "http-bulk", "is_synthetic": is_synthetic_node(node_id), "capabilities": {}, - "node_ref": entry.node_ref, } + # Sole writer of node_ref on both branches: setting it in the + # literal above would seat a colliding ref before the guard runs, + # and the guard refuses by returning rather than clearing. _record_mirrored_ref(node_id, entry.node_ref) # A cached pipeline was built from the config that was active when # it was created, and nothing else refreshes it. diff --git a/backend/services/node_refs.py b/backend/services/node_refs.py index 74878ec4..a9e1cdb4 100644 --- a/backend/services/node_refs.py +++ b/backend/services/node_refs.py @@ -176,10 +176,10 @@ def _mirrored_ref(node_id: str) -> str | None: pattern a minted one must match. Read only when the local registry has nothing, so a real row always wins. - Unlocked: both reads are single dict lookups, and an entry is replaced - wholesale rather than edited field by field, so a concurrent write yields - the old entry or the new one. Taking connected_nodes_lock here would put it - on the 1 Hz publication path, contending with the ingest that writes it. + Unlocked: both reads are single dict lookups, and a concurrent writer + either replaces the entry or assigns this one key, so a read yields the old + value or the new one. Taking connected_nodes_lock here would put it on the + 1 Hz publication path, contending with the ingest that writes it. """ known = state.connected_nodes.get(node_id) ref = known.get("node_ref") if known else None diff --git a/backend/tests/test_radar_routes.py b/backend/tests/test_radar_routes.py index 20ac0c53..15c7cf05 100644 --- a/backend/tests/test_radar_routes.py +++ b/backend/tests/test_radar_routes.py @@ -401,6 +401,15 @@ def test_a_ref_arriving_later_is_still_recorded(self, client): assert self._post(client, node_ref=self.REF).status_code == 200 assert self._entry().get("node_ref") == self.REF + def test_a_ref_the_registry_gives_to_another_node_is_refused(self, client, monkeypatch): + """Registration is first contact for a mirrored node, so a guard that + only held on the already-known path would never run for one.""" + from services import node_refs + + monkeypatch.setattr(node_refs, "id_for_ref", lambda ref: "retdeadbeef") + assert self._post(client, node_ref=self.REF, config={"rx_lat": 34.0, "rx_lon": -82.0}).status_code == 200 + assert self._entry().get("node_ref") is None + def test_a_ref_shaped_like_a_node_id_is_refused(self, client): """Publishing it would put a raw node id on the public wire through the very fallback that exists to keep one off it."""