fix(node): an added root is served before it is scanned

Adding a large directory to a running group made the reload await the
scan of the new root before putting the new RootSet in the group's
context, holding _reload_lock the whole time. For the hours a large drive
takes to hash, the node served the old set:

- a file request under the new root got None from entry_abs_path and the
  handler died on None.exists() without replying;
- a writable/removable toggle answered with the live table, still the
  old one, so the directory vanished from the operator's settings;
- reconcile saw every file the scan had not reached as a missed event and
  hashed it again on the same executor, rewriting progress under the scan.

retarget now applies the set, the roots table and the watcher first, and
with wait=False scans the added roots in the background; the daemon swaps
ctx["roots"] before calling it. A scan lock shared by the initial scan,
added-root scans and reconcile makes the reconcile loop sit out a running
scan without backing off. Every transport site that resolves an entry
answers ROOT_NOT_SERVED instead of crashing, and a delete keeps the entry.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T6jPTeocXA1BePekdsgPya

5 files changed, 103 insertions, 8 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 518f221..c9c362a 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -805,8 +805,15 @@ class NodeDaemon:
                 continue
             log.info("Group %r roots changed: %s", group_cfg.name,
                      ", ".join(f"{r.name}={r.path}" for r in roots))
-            await indexer.retarget(roots)
+            # The new set is what the node serves from this moment, and the
+            # scan of an added root is not waited for. Awaiting it here held
+            # `_reload_lock` and the old set for as long as the scan ran —
+            # hours for a large drive — so every file request under the new
+            # root found no root to resolve against, and any op answering with
+            # the live table (a writable/removable toggle) showed the directory
+            # gone from the operator's settings.
             ctx["roots"] = roots
+            await indexer.retarget(roots, wait=False)
             changed += 1
 
         # ── Hot-load new groups ──────────────────────────────────────────
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index da8ea82..2b68550 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -323,6 +323,14 @@ class DirectoryIndexer:
         # and no deletion and nothing downstream can tell the fields were
         # wiped. See `_drop_root_entries`.
         self.rescanned_ids: set[str] = set()
+        # Held by every whole-root walk — the initial scan, a scan of a root a
+        # retarget added, and reconcile. Reconcile compares disk against the
+        # index, so while a scan is part-way through a root every file it has
+        # not reached yet looks like a missed event: on a 900 GB drive that was
+        # thousands of "appeared" lines, the same root hashed twice over the
+        # one executor thread, and `progress` rewritten under the scan's feet.
+        self._scan_lock = asyncio.Lock()
+        self._scan_tasks: set[asyncio.Task] = set()
 
     @property
     def index(self) -> GroupIndex:
@@ -350,6 +358,10 @@ class DirectoryIndexer:
 
     async def initial_scan(self) -> None:
         """Scan every available root. Run once at startup."""
+        async with self._scan_lock:
+            await self._initial_scan()
+
+    async def _initial_scan(self) -> None:
         self.roots.refresh_availability()
         total = 0
         for root in self.roots:
@@ -556,6 +568,10 @@ class DirectoryIndexer:
             except asyncio.CancelledError:
                 pass
             self._reconciler = None
+        scans = list(self._scan_tasks)
+        for task in scans:
+            task.cancel()
+        await asyncio.gather(*scans, return_exceptions=True)
         for handle in self._pending_timers.values():
             handle.cancel()
         self._pending_timers.clear()
@@ -566,7 +582,7 @@ class DirectoryIndexer:
         self._executor.shutdown(wait=False)
         log.info("Indexer stopped")
 
-    async def retarget(self, roots: RootSet) -> None:
+    async def retarget(self, roots: RootSet, *, wait: bool = True) -> None:
         """
         Point this indexer at a new set of roots, without a restart (14.8).
 
@@ -574,6 +590,12 @@ class DirectoryIndexer:
         operator removed it deliberately, which is not the same event as a
         volume disappearing, and conflating the two is what §6.9 exists to
         prevent. Roots that survive keep their entries; new ones are scanned.
+
+        The set takes effect before anything is scanned: the roots table, the
+        watcher and `self.roots` all move at once. With ``wait=False`` the scan
+        of the added roots runs in the background and this returns as soon as
+        the set is in place — the daemon's reload must not sit on its lock for
+        the hours a large drive takes to hash.
         """
         old_names = {r.folded for r in self.roots}
         new_names = {r.folded for r in roots}
@@ -588,22 +610,62 @@ class DirectoryIndexer:
 
         self.roots = roots
         roots.refresh_availability()
-        for root in roots:
-            if root.folded not in old_names and root.available:
-                await self._scan_root(root)
+        added = [r for r in roots if r.folded not in old_names and r.available]
 
         self._index.roots = roots.describe()
         self._index.version = int(time.time())
         self._restart_observer()
-        if self.on_change:
+
+        if not added:
+            if self.on_change:
+                await self.on_change(self)
+            return
+
+        task = asyncio.create_task(self._scan_added_roots(added))
+        self._scan_tasks.add(task)
+        task.add_done_callback(self._scan_tasks.discard)
+        if wait:
+            await task
+        elif self.on_change:
+            # The table now; the files when the scan ends.
             await self.on_change(self)
 
+    def _holds(self, root: Root) -> bool:
+        return any(r.folded == root.folded and r.path == root.path for r in self.roots)
+
+    async def _scan_added_roots(self, added: list[Root]) -> None:
+        try:
+            async with self._scan_lock:
+                for root in added:
+                    # A later retarget may have removed it while this waited.
+                    if not self._holds(root):
+                        continue
+                    count = await self._scan_root(root)
+                    if not self._holds(root):
+                        # Removed while being scanned: that retarget's drop ran
+                        # before these entries existed.
+                        for entry in self._entries_under(root):
+                            self._index.remove_entry(entry.id)
+                        continue
+                    log.info("Scan complete: %d files in root %r", count, root.name)
+                self._index.version = int(time.time())
+            if self.on_change:
+                await self.on_change(self)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            log.exception("Scanning the added root(s) failed")
+
     # ── Reconciliation ────────────────────────────────────────────────────────
 
     async def _reconcile_loop(self) -> None:
         while True:
             try:
                 await asyncio.sleep(self._reconcile_delay)
+                if self._scan_lock.locked() or self._scan_tasks:
+                    # A scan is walking a root right now. Skipped without
+                    # backing off: the next tick after it ends is the useful one.
+                    continue
                 changed = await self.reconcile()
                 if changed:
                     self._reconcile_delay = self.reconcile_secs
@@ -636,6 +698,10 @@ class DirectoryIndexer:
         to back off when a pass finds nothing to do, rather than running at
         the same cadence forever regardless of how quiet the root is.
         """
+        async with self._scan_lock:
+            return await self._reconcile()
+
+    async def _reconcile(self) -> bool:
         changed = self.roots.refresh_availability()
         touched = False
 
diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py
index 410fe68..7854e28 100644
--- a/packages/meshbay-node/src/meshbay_node/roots.py
+++ b/packages/meshbay-node/src/meshbay_node/roots.py
@@ -384,6 +384,11 @@ class RootSet:
         return out
 
 
+# What a transport answers when `entry_abs_path` gives None: the entry is in the
+# index but its root is ejected, unplugged, or not in the set being served.
+ROOT_NOT_SERVED = "File not available: its folder is not readable right now"
+
+
 def entry_abs_path(roots: RootSet, entry) -> Path | None:
     """
     Where an index entry actually lives, or None if its root is gone.
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
index 7b9d094..2a2b07a 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
@@ -34,7 +34,7 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived, StreamReset
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from meshbay_common import MNP_VERSION
-from meshbay_node.roots import RootSet, entry_abs_path
+from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path
 from meshbay_common.handshake import (
     MNP_MIN_SUPPORTED,
     NONCE_LEN,
@@ -418,6 +418,9 @@ class _MNPServerProtocol(QuicConnectionProtocol):
             return
 
         file_path = entry_abs_path(ctx["roots"], entry)
+        if file_path is None:
+            self._send(stream_id, {"type": "error", "detail": ROOT_NOT_SERVED})
+            return
         if not file_path.exists():
             self._send(stream_id, {"type": "error", "detail": "File not on disk"})
             return
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index bc78644..eff40f1 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -137,7 +137,7 @@ from meshbay_node.media_probe import (
     probe_video as _probe_video,
 )
 from meshbay_node.roots import (
-    RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name,
+    ROOT_NOT_SERVED, RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name,
 )
 
 log = logging.getLogger(__name__)
@@ -3643,6 +3643,9 @@ class WebRTCPeerSession:
             return
 
         file_path = entry_abs_path(ctx["roots"], entry)
+        if file_path is None:
+            self._send({"type": "error", "detail": ROOT_NOT_SERVED})
+            return
         if not file_path.exists():
             self._send({"type": "error", "detail": "File not on disk"})
             return
@@ -3770,6 +3773,9 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not found"})
             return
         file_path = entry_abs_path(ctx["roots"], entry)
+        if file_path is None:
+            self._send({"type": "error", "detail": ROOT_NOT_SERVED})
+            return
         if not file_path.exists():
             self._send({"type": "error", "detail": "File not on disk"})
             return
@@ -5524,6 +5530,11 @@ class WebRTCPeerSession:
 
     def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None:
         file_path = entry_abs_path(ctx["roots"], entry)
+        if file_path is None:
+            # Frozen, not gone: removing the entry would lose a file that is
+            # still on a drive the node cannot read right now.
+            self._send({"type": "error", "detail": ROOT_NOT_SERVED})
+            return
         if file_path.exists():
             file_path.unlink()
             log.info("File deleted: %s", entry.name)
@@ -5723,6 +5734,9 @@ class WebRTCPeerSession:
             return
 
         file_path = entry_abs_path(ctx["roots"], entry)
+        if file_path is None:
+            self._send({"type": "error", "detail": ROOT_NOT_SERVED})
+            return
         if not file_path.exists():
             self._send({"type": "error", "detail": "File not on disk"})
             return