fix(node): a reload and a plug rescan outlive the session that asked

A root added from the client arrives over MNP, and _retarget_indexer
started the daemon's reload with the session's own _spawn. When that
session closed - a client reconnecting 47 s into the scan of a 900 GB
root - shutdown_tasks() cancelled the reload mid-scan, and the reload
queued behind it, without a line in the log. The new root was in
node.toml and in the indexer's set but never in the group's context; the
lock was free and nothing retried, so the node served the old roots table
for hours while reconcile hashed the whole drive as missed events. One
loopback reload fixed the live node in 9 ms.

_reload_config now runs the work in a node-owned task and awaits it
through asyncio.shield, so a caller that goes away only stops waiting; a
cancelled reload is logged. plug_root does the same for its rescan, which
drops the root's entries before walking the disk and so left the root
empty when its admin op's session closed.

The existing MNP test replaced _spawn with a list and could not cancel
anything; the new tests close the session for real and fail without this.

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

3 files changed, 212 insertions, 4 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index c9c362a..932a90d 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -764,9 +764,27 @@ class NodeDaemon:
         Serialised by _reload_lock: fire-and-forget reloads from config-mutating
         endpoints can overlap with the wizard's explicit /api/reload call,
         and two concurrent hot-loads of the same group corrupt the runtime state.
+
+        The reload belongs to the node, never to whoever asked for it. A root
+        added from a browser reaches here through the operator's WebRTC session,
+        whose tasks are all cancelled when that session closes — and on
+        2026-09-14 one closed 47 s into the scan of a 900 GB root. The reload
+        died without a line in the log, the new root was in node.toml and in
+        the indexer but never in the group's context, and nothing ever tried
+        again: the lock was free, and nobody was waiting on it. So the work runs
+        in a task of its own, and a caller that goes away only stops waiting.
         """
-        async with self._reload_lock:
-            await self._reload_config_inner()
+        await asyncio.shield(spawn(self._reload_config_locked(), what="config reload"))
+
+    async def _reload_config_locked(self) -> None:
+        try:
+            async with self._reload_lock:
+                await self._reload_config_inner()
+        except asyncio.CancelledError:
+            log.warning("Config reload cancelled before it finished — the node may "
+                        "be serving part of the previous configuration until the "
+                        "next reload")
+            raise
 
     async def _reload_config_inner(self) -> None:
         log.info("Reloading config from %s", self._config_path)
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index 2b68550..c20902e 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -953,8 +953,22 @@ class DirectoryIndexer:
             return
         root.ejected = False
         root.available = root.is_live()
-        if root.available:
-            log.info("Root %r plugged — rescanning", root.name)
+        if not root.available:
+            await self._finish_plug(None)
+            return
+        log.info("Root %r plugged — rescanning", root.name)
+        # `_rescan_root` drops the root's entries before it walks the disk, and
+        # this is called from an admin op running in the operator's WebRTC
+        # session — which cancels everything it started when it closes. So the
+        # rescan runs in a task this indexer owns, and a caller that goes away
+        # only stops waiting for it instead of leaving the root emptied.
+        task = asyncio.create_task(self._finish_plug(root))
+        self._scan_tasks.add(task)
+        task.add_done_callback(self._scan_tasks.discard)
+        await asyncio.shield(task)
+
+    async def _finish_plug(self, root: Root | None) -> None:
+        if root is not None:
             await self._rescan_root(root)
         self._restart_observer()
         self._index.roots = self.roots.describe()
diff --git a/packages/meshbay-node/tests/test_root_work_outlives_the_session.py b/packages/meshbay-node/tests/test_root_work_outlives_the_session.py
new file mode 100644
index 0000000..31de12e
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_work_outlives_the_session.py
@@ -0,0 +1,176 @@
+"""
+Work on a group's roots belongs to the node, not to the session that asked for it.
+
+Found live on 2026-09-14. A 900 GB directory was added from the desktop client,
+so the op arrived over MNP and `_retarget_indexer` started the daemon's reload
+with the session's own `_spawn`. That session closed 47 seconds later — the
+client reconnected — and `shutdown_tasks()` cancelled everything it had started:
+the reload, part-way through hashing the 42nd file, and the reload the
+"removable" toggle had queued behind it. A cancelled task logs nothing. The new
+root was in node.toml and in the indexer's own set, and never in the group's
+context, so the node went on serving the old table for eight hours while
+reconcile hashed the whole drive as "missed events" — and a client reloaded in
+the morning showed nothing new. One loopback reload, which nothing could cancel,
+put it right in nine milliseconds.
+
+`test_root_ops_reach_the_live_set.py` covered this seam, with `_spawn` replaced
+by a list: a fixture that could not cancel anything, testing a path whose whole
+failure was being cancelled. These tests close the session for real.
+"""
+
+import asyncio
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+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 RootSet
+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])
+
+
+def _names(idx) -> list[str]:
+    return sorted(e.name for e in idx.index.entries)
+
+
+async def _daemon(tmp_path: Path, one: Path, two: Path):
+    """A daemon whose reload is held at the door, hosting one group whose
+    node.toml already names a second root the running set does not have."""
+    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 = DirectoryIndexer(roots=_set(one), group_id=GROUP,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+
+    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": {},
+                     "reload_fn": daemon._reload_config}
+
+    gate, entered = asyncio.Event(), asyncio.Event()
+
+    async def held_build_roots(group_cfg):
+        entered.set()
+        await gate.wait()
+        return await NodeDaemon._build_roots(daemon, group_cfg)
+
+    daemon._build_roots = held_build_roots
+    return daemon, idx, ctx, gate, entered
+
+
+async def _until(predicate, timeout: float = 3.0) -> bool:
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not predicate():
+        if asyncio.get_running_loop().time() > deadline:
+            return False
+        await asyncio.sleep(0.02)
+    return True
+
+
+async def test_a_reload_started_over_mnp_survives_the_session_closing(tmp_path):
+    one, two = _dirs(tmp_path)
+    daemon, idx, ctx, gate, entered = await _daemon(tmp_path, one, two)
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"daemon_state": daemon._state}
+    session._tasks = set()
+    session._stop_stream = lambda: None
+    session._release_transfers = lambda: None
+    try:
+        await session._retarget_indexer(GROUP)
+        await asyncio.wait_for(entered.wait(), 2)
+
+        await session.shutdown_tasks()      # what a closed client does
+        gate.set()
+
+        assert await _until(lambda: [r.name for r in ctx["roots"]] == ["one", "two"]), (
+            "closing the session that asked for the reload cancelled it")
+        assert await _until(lambda: not daemon._reload_lock.locked())
+        await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
+        assert _names(idx) == ["a.txt", "b.txt", "c.txt"]
+    finally:
+        gate.set()
+        await idx.stop()
+
+
+async def test_a_reload_whose_caller_is_cancelled_still_finishes(tmp_path):
+    """`ops.reload_config` awaits the reload inside the session's task too."""
+    one, two = _dirs(tmp_path)
+    daemon, idx, ctx, gate, entered = await _daemon(tmp_path, one, two)
+    try:
+        caller = asyncio.create_task(daemon._reload_config())
+        await asyncio.wait_for(entered.wait(), 2)
+        caller.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await caller
+        gate.set()
+
+        assert await _until(lambda: [r.name for r in ctx["roots"]] == ["one", "two"])
+        assert await _until(lambda: not daemon._reload_lock.locked())
+    finally:
+        gate.set()
+        await idx.stop()
+
+
+async def test_a_plug_whose_caller_goes_away_does_not_empty_the_root(tmp_path):
+    one, two = _dirs(tmp_path)
+
+    class _Held(DirectoryIndexer):
+        gate = None
+        at_gate = None
+
+        async def _scan_root(self, root):
+            if self.gate is not None:
+                self.at_gate.set()
+                await self.gate.wait()
+            return await super()._scan_root(root)
+
+    idx = _Held(roots=_set(one, two), group_id=GROUP,
+                sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    idx.eject_root("two")
+    idx.gate, idx.at_gate = asyncio.Event(), asyncio.Event()
+    try:
+        caller = asyncio.create_task(idx.plug_root("two"))
+        await asyncio.wait_for(idx.at_gate.wait(), 2)
+        caller.cancel()                     # the admin op's session closed
+        with pytest.raises(asyncio.CancelledError):
+            await caller
+        idx.gate.set()
+
+        await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
+        assert _names(idx) == ["a.txt", "b.txt", "c.txt"], (
+            "the plug's rescan was cancelled after it had dropped the root's entries")
+    finally:
+        idx.gate.set()
+        await idx.stop()