diff options
Diffstat (limited to 'packages')
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() |