Publish the data archive under node_ref, not node_id

backend/routes/archive.py

@@ -4,23 +4,33 @@
 Doppler and SNR plus the node's rx/tx geometry, for one node at a time.  That
 makes it the most literal reading of "publish this node's detections" the
 server has, so a node whose owner registered it private is not listed here and
-its keys do not download.  The keys are Hive-partitioned
-(``year=/month=/day=/node_id=/part-*.parquet``), so the node id is in the key
-itself and both checks read it from there — see ``_key_node_id``.
+its keys do not download.
+
+The store is Hive-partitioned on the private id
+(``year=/month=/day=/node_id=/part-*.parquet``), so the id is in every key on
+disk.  The route publishes each key with that segment rewritten to the node's
+published identity (``node_ref=``), resolves it back on the way in, and renames
+the downloaded body's ``node_id`` the same way: no key, parameter or body it
+hands out names the private id.  See ``_published_key`` and ``_stored_key``.
 """
 
 import asyncio
 
 from fastapi import APIRouter, HTTPException, Query
 
+from services.node_refs import id_for_identity, public_identity
 from services.publication import private_node_ids
 from services.storage import list_archived_files, read_archived_file
 
 router = APIRouter(tags=["archive"])
 
+# The node segment's Hive name on disk, and the one it is published under.
+_STORED = "node_id="
+_PUBLISHED = "node_ref="
+
 
 def _key_node_id(key: str) -> str:
-    """The node id a key is partitioned under, or "" if it has none.
+    """The node id a stored key is partitioned under, or "" if it has none.
 
     Reads the second-to-last path segment, which is where both the Hive layout
     (``node_id=NNN``) and the legacy one (a bare directory name) put it, and
@@ -35,39 +45,97 @@ def _key_node_id(key: str) -> str:
     return parts[-2].split("=", 1)[-1]
 
 
+def _published_key(key: str) -> str | None:
+    """A stored key as the listing publishes it, or None if its node has no
+    published identity.  A key with no node segment is published as it is."""
+    parts = key.strip("/").split("/")
+    if len(parts) < 2:
+        return key
+    hive = parts[-2].startswith(_STORED)
+    ref = public_identity(parts[-2].removeprefix(_STORED))
+    if ref is None:
+        return None
+    parts[-2] = f"{_PUBLISHED}{ref}" if hive else ref
+    return "/".join(parts)
+
+
+def _stored_key(key: str) -> tuple[str, str] | None:
+    """The stored key behind a published one, and the node id it names ("" for
+    a key with no node segment); None if it names no published node.
+
+    Only the published spellings resolve: a ``node_id=`` segment, or a bare
+    one holding a private id, is a key this route never handed out.
+    """
+    parts = key.strip("/").split("/")
+    if len(parts) < 2:
+        return key, ""
+    hive = parts[-2].startswith(_PUBLISHED)
+    if "=" in parts[-2] and not hive:
+        return None
+    node_id = id_for_identity(parts[-2].removeprefix(_PUBLISHED))
+    if node_id is None:
+        return None
+    parts[-2] = f"{_STORED}{node_id}" if hive else node_id
+    return "/".join(parts), node_id
+
+
 @router.get("/api/data/archive")
 async def list_archive(
     date: str = Query(None, description="Date prefix, e.g. 2025/06/21"),
-    node_id: str = Query(None, description="Filter by node ID"),
+    node_ref: str = Query(None, description="Filter by node_ref"),
     limit: int = Query(50, ge=1, le=500, description="Page size"),
     offset: int = Query(0, ge=0, description="Pagination offset"),
 ):
+    private = private_node_ids()
+    node_id = None
+    if node_ref:
+        # An unknown ref and a private node's answer the same, so the filter
+        # cannot confirm that a private node exists.
+        node_id = id_for_identity(node_ref)
+        if node_id is None or node_id in private:
+            return {"files": [], "count": 0, "total": 0}
     loop = asyncio.get_running_loop()
     result = await loop.run_in_executor(
         None, lambda: list_archived_files(date_prefix=date, node_id=node_id, limit=limit, offset=offset)
     )
-    private = private_node_ids()
-    if not private:
-        return result
     # Filtered after paging rather than before it, which shortens a page
     # instead of resampling the store: list_archived_files walks the tree in
     # date order with an early exit, and pushing the predicate into it would
     # mean either teaching storage.py about publication or re-listing until a
     # page fills.  `count` follows what is actually returned; `total` is left
     # as the scan reported, the same approximation it already is for the
     # truncated no-date path.
-    files = [f for f in result.get("files", []) if _key_node_id(f.get("key", "")) not in private]
+    files = []
+    for f in result.get("files", []):
+        key = f.get("key", "")
+        if _key_node_id(key) in private:
+            continue
+        published = _published_key(key)
+        if published is not None:
+            files.append({**f, "key": published})
     return {**result, "files": files, "count": len(files)}
 
 
 @router.get("/api/data/archive/{key:path}")
 async def download_archive_file(key: str):
-    # Before the read, not after: the point is not to avoid serving the bytes
-    # but to give a private node's key the same answer as a key that does not
-    # exist, so the endpoint cannot be used to enumerate the private fleet.
-    if _key_node_id(key) in private_node_ids():
+    # The policy is read before the key is resolved, and both before the read:
+    # a private node's key gets the same answer as a key that does not exist,
+    # so the endpoint cannot be used to enumerate the private fleet.
+    private = private_node_ids()
+    stored = _stored_key(key)
+    if stored is None or stored[1] in private:
         raise HTTPException(status_code=404, detail="Archive file not found")
-    data = read_archived_file(key)
+    data = read_archived_file(stored[0])
     if data is None:
         raise HTTPException(status_code=404, detail="Archive file not found")
+    if isinstance(data, dict) and "node_id" in data:
+        data = dict(data)
+        body_id = data.pop("node_id")
+        # Named from the key where it has a node, which is already resolved: a
+        # zero-row file's own node_id is "".  A key with no node segment was
+        # never checked above, so the node its body names is checked here.
+        node_id = stored[1] or body_id
+        if node_id in private:
+            raise HTTPException(status_code=404, detail="Archive file not found")
+        data["node_ref"] = public_identity(node_id)
     return data

backend/routes/auth.py

@@ -43,7 +43,7 @@
     release_node,
 )
 from services.node_config import position_status
-from services.node_refs import owner_identity
+from services.node_refs import owner_identity, public_name
 from services.polled_radars import remove_polled_radar
 
 logger = logging.getLogger(__name__)
@@ -417,11 +417,14 @@ async def my_nodes(request: Request):
         info = snapshot.get(nid) or {}
         cfg = info.get("config", {}) or {}
         private, source = privacy.get(nid, (False, publication.SOURCE_DEFAULT))
+        ref = owner_identity(nid)
         out.append(
             {
                 "node_id": nid,
-                "node_ref": owner_identity(nid),
-                "name": cfg.get("name", nid),
+                "node_ref": ref,
+                # The console names a node by its ref, so an unnamed node, or
+                # one whose name is a node id, is called by that instead.
+                "name": public_name(cfg.get("name"), ref, node_ids),
                 "status": info.get("status", "never_connected"),
                 "last_heartbeat": info.get("last_heartbeat"),
                 "is_synthetic": info.get("is_synthetic", False),

backend/services/node_refs.py

@@ -131,16 +131,17 @@ def id_for_identity(identity: str | None) -> str | None:
     """The node behind a published identity, or None if nothing publishes as it.
 
     The inverse of `public_identity`, and what every public path parameter
-    resolves through. It is `id_for_ref` plus the synthetic passthrough that
+    resolves through. It is `id_for_ref` plus the two fallbacks
     `public_identity` applies on the way out: a synthetic node is published
-    under its own id, so that id is its handle and must keep resolving. A real
-    node is reachable only by the ref its registry row carries.
+    under its own id, so that id is its handle and must keep resolving, and a
+    mirrored node under the ref its sending environment resolved. A real node
+    is reachable only by the ref it is published under.
     """
     if not identity:
         return None
     if is_synthetic_node(identity):
         return identity
-    return id_for_ref(identity)
+    return id_for_ref(identity) or _mirrored_id(identity)
 
 
 def ref_to_id_map(node_ids: Iterable[str] = ()) -> dict[str, str]:
@@ -177,7 +178,7 @@ def _names_a_node(value: str, known_ids: Iterable[str]) -> bool:
     return value in _forward or value in known_ids
 
 
-def public_name(name, fallback: str, known_ids: Iterable[str] = ()) -> str:
+def public_name(name, fallback: str | None, known_ids: Iterable[str] = ()) -> str | None:
     """A node-supplied display name, or `fallback` when it names a node.
 
     Nodes fill this field in themselves and nothing validates it, so a name
@@ -207,6 +208,22 @@ def _mirrored_ref(node_id: str) -> str | None:
     return ref if isinstance(ref, str) and ref else None
 
 
+def _mirrored_id(node_ref: str) -> str | None:
+    """The mirrored node published under `node_ref`, the inverse of `_mirrored_ref`.
+
+    A scan of the connected nodes, which is where a mirrored ref lives; there
+    are tens of them. A candidate counts only if `owner_identity` names it by
+    this ref, so a registry row that outranks its mirrored ref on the way out
+    outranks it here too.
+    """
+    with state.connected_nodes_lock:
+        snapshot = list(state.connected_nodes.items())
+    for node_id, known in snapshot:
+        if known.get("node_ref") == node_ref and owner_identity(node_id) == node_ref:
+            return node_id
+    return None
+
+
 def owner_identity(node_id: str | None) -> str | None:
     """What a node publishes as, for a caller that already knows the node.
 

backend/tests/test_auth_routes.py

@@ -174,6 +174,43 @@ def test_an_owned_node_carries_the_ref_the_public_surfaces_key_on(self, client):
         finally:
             asyncio.run(set_node_owner(node_id, None))
 
+    @pytest.mark.parametrize("configured", [None, "ref-name-node"])
+    def test_a_node_without_a_name_of_its_own_is_called_by_its_ref(self, configured, client):
+        """An unnamed node, and one whose name is its own id, read as the ref:
+        the console never shows the node_id."""
+        from core import state
+        from core.auth import set_node_owner
+        from core.users import ANONYMOUS_USER
+
+        node_id = "ref-name-node"
+        ref = own(node_id, ANONYMOUS_USER["id"])
+        with state.connected_nodes_lock:
+            state.connected_nodes[node_id] = {"status": "active", "config": {"name": configured}}
+        try:
+            node = next(n for n in client.get("/api/auth/me/nodes").json() if n["node_id"] == node_id)
+            assert node["name"] == ref
+        finally:
+            with state.connected_nodes_lock:
+                state.connected_nodes.pop(node_id, None)
+            asyncio.run(set_node_owner(node_id, None))
+
+    def test_a_node_keeps_the_name_it_was_given(self, client):
+        from core import state
+        from core.auth import set_node_owner
+        from core.users import ANONYMOUS_USER
+
+        node_id = "named-node"
+        own(node_id, ANONYMOUS_USER["id"])
+        with state.connected_nodes_lock:
+            state.connected_nodes[node_id] = {"status": "active", "config": {"name": "Roof dish"}}
+        try:
+            node = next(n for n in client.get("/api/auth/me/nodes").json() if n["node_id"] == node_id)
+            assert node["name"] == "Roof dish"
+        finally:
+            with state.connected_nodes_lock:
+                state.connected_nodes.pop(node_id, None)
+            asyncio.run(set_node_owner(node_id, None))
+
     def test_my_nodes_entry_carries_position_status_for_a_private_node(self, client):
         """A private node is filtered out of /api/radar/nodes entirely, so
         its owner has nowhere else to learn it needs a position."""

backend/tests/test_node_refs.py

@@ -329,6 +329,12 @@ def _connected(self, **entry):
         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")
@@ -345,6 +351,18 @@ def test_the_registry_wins_over_a_mirrored_ref(self, seed):
         self._connected(node_ref="ndemirrored001")
         assert node_refs.public_identity(_MIRRORED) == "ndelocalrow001"
 
+    def test_a_mirrored_ref_resolves_back_to_its_node(self, seed):
+        """Public path parameters take the ref a node is published under."""
+        seed(ret1a2b3c4d="nde1a2b3c4d00")
+        self._connected(node_ref="ndemirrored001")
+        assert node_refs.id_for_identity("ndemirrored001") == _MIRRORED
+
+    def test_a_mirrored_ref_the_registry_outranks_resolves_to_nothing(self, seed):
+        seed(**{_MIRRORED: "ndelocalrow001"})
+        self._connected(node_ref="ndemirrored001")
+        assert node_refs.id_for_identity("ndemirrored001") is None
+        assert node_refs.id_for_identity("ndelocalrow001") == _MIRRORED
+
     def test_a_mirrored_node_reaches_the_published_feed(self, seed):
         seed(ret1a2b3c4d="nde1a2b3c4d00")
         self._connected(node_ref="ndemirrored001")