diff options
Diffstat (limited to 'packages/meshbay-node/src')
5 files changed, 103 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 |