diff options
Diffstat (limited to 'packages')
6 files changed, 343 insertions, 8 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 518f221..c9c362a 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -805,8 +805,15 @@ class NodeDaemon: continue log.info("Group %r roots changed: %s", group_cfg.name, ", ".join(f"{r.name}={r.path}" for r in roots)) - await indexer.retarget(roots) + # The new set is what the node serves from this moment, and the + # scan of an added root is not waited for. Awaiting it here held + # `_reload_lock` and the old set for as long as the scan ran — + # hours for a large drive — so every file request under the new + # root found no root to resolve against, and any op answering with + # the live table (a writable/removable toggle) showed the directory + # gone from the operator's settings. ctx["roots"] = roots + await indexer.retarget(roots, wait=False) changed += 1 # ── Hot-load new groups ────────────────────────────────────────── diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index da8ea82..2b68550 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -323,6 +323,14 @@ class DirectoryIndexer: # and no deletion and nothing downstream can tell the fields were # wiped. See `_drop_root_entries`. self.rescanned_ids: set[str] = set() + # Held by every whole-root walk — the initial scan, a scan of a root a + # retarget added, and reconcile. Reconcile compares disk against the + # index, so while a scan is part-way through a root every file it has + # not reached yet looks like a missed event: on a 900 GB drive that was + # thousands of "appeared" lines, the same root hashed twice over the + # one executor thread, and `progress` rewritten under the scan's feet. + self._scan_lock = asyncio.Lock() + self._scan_tasks: set[asyncio.Task] = set() @property def index(self) -> GroupIndex: @@ -350,6 +358,10 @@ class DirectoryIndexer: async def initial_scan(self) -> None: """Scan every available root. Run once at startup.""" + async with self._scan_lock: + await self._initial_scan() + + async def _initial_scan(self) -> None: self.roots.refresh_availability() total = 0 for root in self.roots: @@ -556,6 +568,10 @@ class DirectoryIndexer: except asyncio.CancelledError: pass self._reconciler = None + scans = list(self._scan_tasks) + for task in scans: + task.cancel() + await asyncio.gather(*scans, return_exceptions=True) for handle in self._pending_timers.values(): handle.cancel() self._pending_timers.clear() @@ -566,7 +582,7 @@ class DirectoryIndexer: self._executor.shutdown(wait=False) log.info("Indexer stopped") - async def retarget(self, roots: RootSet) -> None: + async def retarget(self, roots: RootSet, *, wait: bool = True) -> None: """ Point this indexer at a new set of roots, without a restart (14.8). @@ -574,6 +590,12 @@ class DirectoryIndexer: operator removed it deliberately, which is not the same event as a volume disappearing, and conflating the two is what §6.9 exists to prevent. Roots that survive keep their entries; new ones are scanned. + + The set takes effect before anything is scanned: the roots table, the + watcher and `self.roots` all move at once. With ``wait=False`` the scan + of the added roots runs in the background and this returns as soon as + the set is in place — the daemon's reload must not sit on its lock for + the hours a large drive takes to hash. """ old_names = {r.folded for r in self.roots} new_names = {r.folded for r in roots} @@ -588,22 +610,62 @@ class DirectoryIndexer: self.roots = roots roots.refresh_availability() - for root in roots: - if root.folded not in old_names and root.available: - await self._scan_root(root) + added = [r for r in roots if r.folded not in old_names and r.available] self._index.roots = roots.describe() self._index.version = int(time.time()) self._restart_observer() - if self.on_change: + + if not added: + if self.on_change: + await self.on_change(self) + return + + task = asyncio.create_task(self._scan_added_roots(added)) + self._scan_tasks.add(task) + task.add_done_callback(self._scan_tasks.discard) + if wait: + await task + elif self.on_change: + # The table now; the files when the scan ends. await self.on_change(self) + def _holds(self, root: Root) -> bool: + return any(r.folded == root.folded and r.path == root.path for r in self.roots) + + async def _scan_added_roots(self, added: list[Root]) -> None: + try: + async with self._scan_lock: + for root in added: + # A later retarget may have removed it while this waited. + if not self._holds(root): + continue + count = await self._scan_root(root) + if not self._holds(root): + # Removed while being scanned: that retarget's drop ran + # before these entries existed. + for entry in self._entries_under(root): + self._index.remove_entry(entry.id) + continue + log.info("Scan complete: %d files in root %r", count, root.name) + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + except asyncio.CancelledError: + raise + except Exception: + log.exception("Scanning the added root(s) failed") + # ── Reconciliation ──────────────────────────────────────────────────────── async def _reconcile_loop(self) -> None: while True: try: await asyncio.sleep(self._reconcile_delay) + if self._scan_lock.locked() or self._scan_tasks: + # A scan is walking a root right now. Skipped without + # backing off: the next tick after it ends is the useful one. + continue changed = await self.reconcile() if changed: self._reconcile_delay = self.reconcile_secs @@ -636,6 +698,10 @@ class DirectoryIndexer: to back off when a pass finds nothing to do, rather than running at the same cadence forever regardless of how quiet the root is. """ + async with self._scan_lock: + return await self._reconcile() + + async def _reconcile(self) -> bool: changed = self.roots.refresh_availability() touched = False diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py index 410fe68..7854e28 100644 --- a/packages/meshbay-node/src/meshbay_node/roots.py +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -384,6 +384,11 @@ class RootSet: return out +# What a transport answers when `entry_abs_path` gives None: the entry is in the +# index but its root is ejected, unplugged, or not in the set being served. +ROOT_NOT_SERVED = "File not available: its folder is not readable right now" + + def entry_abs_path(roots: RootSet, entry) -> Path | None: """ Where an index entry actually lives, or None if its root is gone. diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 7b9d094..2a2b07a 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -34,7 +34,7 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived, StreamReset from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION -from meshbay_node.roots import RootSet, entry_abs_path +from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path from meshbay_common.handshake import ( MNP_MIN_SUPPORTED, NONCE_LEN, @@ -418,6 +418,9 @@ class _MNPServerProtocol(QuicConnectionProtocol): return file_path = entry_abs_path(ctx["roots"], entry) + if file_path is None: + self._send(stream_id, {"type": "error", "detail": ROOT_NOT_SERVED}) + return if not file_path.exists(): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index bc78644..eff40f1 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -137,7 +137,7 @@ from meshbay_node.media_probe import ( probe_video as _probe_video, ) from meshbay_node.roots import ( - RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, + ROOT_NOT_SERVED, RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, ) log = logging.getLogger(__name__) @@ -3643,6 +3643,9 @@ class WebRTCPeerSession: return file_path = entry_abs_path(ctx["roots"], entry) + if file_path is None: + self._send({"type": "error", "detail": ROOT_NOT_SERVED}) + return if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return @@ -3770,6 +3773,9 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return file_path = entry_abs_path(ctx["roots"], entry) + if file_path is None: + self._send({"type": "error", "detail": ROOT_NOT_SERVED}) + return if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return @@ -5524,6 +5530,11 @@ class WebRTCPeerSession: def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: file_path = entry_abs_path(ctx["roots"], entry) + if file_path is None: + # Frozen, not gone: removing the entry would lose a file that is + # still on a drive the node cannot read right now. + self._send({"type": "error", "detail": ROOT_NOT_SERVED}) + return if file_path.exists(): file_path.unlink() log.info("File deleted: %s", entry.name) @@ -5723,6 +5734,9 @@ class WebRTCPeerSession: return file_path = entry_abs_path(ctx["roots"], entry) + if file_path is None: + self._send({"type": "error", "detail": ROOT_NOT_SERVED}) + return if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return 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 |