1 files changed, 176 insertions, 0 deletions
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()