""" 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, **kwargs): if self.gate is not None: self.at_gate.set() await self.gate.wait() return await super()._scan_root(root, **kwargs) 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()