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

1 files changed, 240 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py b/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py
new file mode 100644
index 0000000..2dc31c9
--- /dev/null
+++ b/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py
@@ -0,0 +1,240 @@
+"""
+Adding a directory to a running group must not wait for that directory to be hashed.
+
+Found live on a 900 GB NTFS drive added to a group that already had one root.
+The daemon's reload awaited `DirectoryIndexer.retarget`, which scanned the new
+root before returning, and only then put the new `RootSet` into the group's
+context — holding `_reload_lock` the whole time. For the hours that took:
+
+- every file request under the new root resolved against the *old* set,
+  `entry_abs_path` answered None, and the handler died on `None.exists()`
+  without replying, so the client waited out its own timeout;
+- toggling the new root's "removable" switch answered with the live table,
+  still the old one, and the directory vanished from the operator's settings
+  while `meshbay-node status` — reading node.toml — still listed it;
+- reconcile, ten minutes in, found every file the scan had not reached yet
+  missing from the index and hashed them itself as "missed events", on the
+  same single executor thread, rewriting `progress` under the scan.
+
+These tests hold each scan at the door with an event, which is what makes
+"before the scan finishes" a state a test can stand in rather than a race.
+"""
+
+import asyncio
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.protocol import MNP
+from meshbay_node.config import load_config
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import ROOT_NOT_SERVED, RootSet
+from meshbay_node.transfers import LeaselessReads
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+def _dirs(tmp_path: Path) -> tuple[Path, Path]:
+    one, two = tmp_path / "one", tmp_path / "two"
+    one.mkdir()
+    two.mkdir()
+    (one / "a.txt").write_bytes(b"first root")
+    (two / "b.txt").write_bytes(b"second root, one")
+    (two / "c.txt").write_bytes(b"second root, two")
+    return one, two
+
+
+def _set(*dirs: Path) -> RootSet:
+    return RootSet.build([{"path": str(d), "name": d.name} for d in dirs])
+
+
+class _GatedIndexer(DirectoryIndexer):
+    """Every whole-root scan waits for `gate` before it reads anything."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.gate = asyncio.Event()
+        self.at_gate = asyncio.Event()
+
+    async def _scan_root(self, root):
+        self.at_gate.set()
+        await self.gate.wait()
+        return await super()._scan_root(root)
+
+
+async def _indexer(one: Path, **kwargs) -> _GatedIndexer:
+    idx = _GatedIndexer(roots=_set(one), group_id=GROUP,
+                        sk_node=Ed25519PrivateKey.generate(), gek=None, **kwargs)
+    idx.gate.set()
+    await idx.initial_scan()
+    idx.gate.clear()
+    idx.at_gate.clear()
+    return idx
+
+
+async def _finish(idx: _GatedIndexer) -> None:
+    idx.gate.set()
+    await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
+
+
+def _names(idx) -> list[str]:
+    return sorted(e.name for e in idx.index.entries)
+
+
+async def test_the_new_set_is_in_place_before_the_new_root_is_scanned(tmp_path):
+    one, two = _dirs(tmp_path)
+    idx = await _indexer(one)
+    try:
+        await asyncio.wait_for(idx.retarget(_set(one, two), wait=False), 2)
+        await asyncio.wait_for(idx.at_gate.wait(), 2)
+
+        assert [r.name for r in idx.roots] == ["one", "two"]
+        assert [r["name"] for r in idx.index.roots] == ["one", "two"]
+        assert _names(idx) == ["a.txt"]
+
+        await _finish(idx)
+        assert _names(idx) == ["a.txt", "b.txt", "c.txt"]
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+async def test_the_daemon_serves_the_new_set_and_releases_its_lock(tmp_path):
+    one, two = _dirs(tmp_path)
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'data_dir = "{(tmp_path / "data").as_posix()}"\n\n'
+        f'[[groups]]\nid = "{GROUP}"\nname = "plop"\n\n'
+        f'  [[groups.roots]]\n  path = "{one.as_posix()}"\n  name = "one"\n\n'
+        f'  [[groups.roots]]\n  path = "{two.as_posix()}"\n  name = "two"\n')
+
+    idx = await _indexer(one)
+    ctx = {"roots": idx.roots}
+    daemon = NodeDaemon.__new__(NodeDaemon)
+    daemon._config_path = conf
+    daemon._config = load_config(conf)
+    daemon._roster = None
+    daemon._hub = None
+    daemon._indexers = [idx]
+    daemon._reload_lock = asyncio.Lock()
+    daemon._state = {"groups_ctx": {GROUP: ctx}, "indexes": {}, "indexers": {}}
+    try:
+        await asyncio.wait_for(daemon._reload_config(), 2)
+
+        assert [r.name for r in ctx["roots"]] == ["one", "two"], (
+            "the reload returned with the group still served from the old set")
+        assert not daemon._reload_lock.locked()
+
+        await _finish(idx)
+        assert _names(idx) == ["a.txt", "b.txt", "c.txt"]
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+async def test_reconcile_sits_out_a_scan_instead_of_redoing_it(tmp_path):
+    one, two = _dirs(tmp_path)
+    idx = await _indexer(one, reconcile_secs=0.01)
+    calls: list[int] = []
+    real = idx.reconcile
+
+    async def counting():
+        calls.append(1)
+        return await real()
+
+    idx.reconcile = counting
+    loop_task = None
+    try:
+        await idx.retarget(_set(one, two), wait=False)
+        await asyncio.wait_for(idx.at_gate.wait(), 2)
+
+        loop_task = asyncio.create_task(idx._reconcile_loop())
+        await asyncio.sleep(0.2)
+        assert calls == [], "reconcile ran against a root that was mid-scan"
+        assert idx._reconcile_delay == 0.01, "a skipped tick must not back off"
+
+        await _finish(idx)
+        await asyncio.sleep(0.2)
+        assert calls, "reconcile never resumed once the scan was over"
+        assert _names(idx) == ["a.txt", "b.txt", "c.txt"]
+    finally:
+        if loop_task:
+            loop_task.cancel()
+        idx.gate.set()
+        await idx.stop()
+
+
+async def test_a_root_removed_while_it_was_being_scanned_leaves_nothing(tmp_path):
+    one, two = _dirs(tmp_path)
+    idx = await _indexer(one)
+    try:
+        await idx.retarget(_set(one, two), wait=False)
+        await asyncio.wait_for(idx.at_gate.wait(), 2)
+        await asyncio.wait_for(idx.retarget(_set(one), wait=False), 2)
+
+        await _finish(idx)
+        assert _names(idx) == ["a.txt"]
+        assert [r["name"] for r in idx.index.roots] == ["one"]
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+# ── A request for a file whose root is not being served ──────────────────────
+
+class _Channel:
+    readyState = "open"
+    bufferedAmount = 0
+
+
+class _Session(WebRTCPeerSession):
+    def __init__(self, ctx):
+        self._ctx = ctx
+        self._registry_key = "s1"
+        self._user_id = "alice"
+        self._username = "alice"
+        self._group_id = GROUP
+        self._channel = _Channel()
+        self._leaseless = LeaselessReads()
+        self._unleased_noted = False
+        self.sent: list[dict] = []
+
+    def _send(self, msg):
+        self.sent.append(msg)
+
+    def _audit(self, event, detail=""):
+        pass
+
+    def _spawn(self, coro):
+        coro.close()
+        return None
+
+
+async def _served_without_two(tmp_path):
+    """An index holding both roots' files, served from a set holding only one."""
+    one, two = _dirs(tmp_path)
+    idx = DirectoryIndexer(roots=_set(one, two), group_id=GROUP,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    entry = next(e for e in idx.index.entries if e.name == "b.txt")
+    ctx = {"_peers": {}, "roots": _set(one), "index": idx.index,
+           "sk_node": idx.sk_node, "gek": None}
+    return _Session(ctx), ctx, entry
+
+
+async def test_a_file_request_is_refused_not_crashed(tmp_path):
+    session, _, entry = await _served_without_two(tmp_path)
+    await session._do_file_request(
+        {"type": MNP.FILE_REQUEST, "file_id": entry.id, "chunk_index": 0})
+    assert [m.get("detail") for m in session.sent] == [ROOT_NOT_SERVED]
+
+
+async def test_a_delete_is_refused_and_the_entry_kept(tmp_path):
+    session, ctx, entry = await _served_without_two(tmp_path)
+    session._exec_file_delete(ctx, entry.id, entry)
+    assert [m.get("detail") for m in session.sent] == [ROOT_NOT_SERVED]
+    assert ctx["index"].get_entry(entry.id) is not None