Carry the node_ref across the detection mirror

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):

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:

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
 
 

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:

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

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"}]

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