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 ac53cfb1..f875fc2c 100644 --- a/backend/routes/radar.py +++ b/backend/routes/radar.py @@ -14,7 +14,8 @@ from core import state from core.users import require_admin from pipeline.passive_radar import PassiveRadarPipeline -from services import node_registration +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 services.public_location import public_latlon @@ -36,6 +37,12 @@ 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 (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) @@ -167,6 +174,24 @@ 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 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 + 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, @@ -215,6 +240,10 @@ async def ingest_detections_bulk( "is_synthetic": is_synthetic_node(node_id), "capabilities": {}, } + # 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. node_registration.evict_pipeline(node_id) @@ -224,6 +253,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/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..a9e1cdb4 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 @@ -166,6 +167,25 @@ 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, 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 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 + 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 +197,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_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_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"}] diff --git a/backend/tests/test_radar_routes.py b/backend/tests/test_radar_routes.py index 299b1938..15c7cf05 100644 --- a/backend/tests/test_radar_routes.py +++ b/backend/tests/test_radar_routes.py @@ -362,3 +362,58 @@ 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_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.""" + 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