diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
4 files changed, 151 insertions, 36 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 932a90d..a43e3a3 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -1163,14 +1163,16 @@ class NodeDaemon: parameter (not a bare constant) only so a test can drive this loop without waiting on the real 2s cadence. """ - was_scanning = False + was_busy = False while True: await asyncio.sleep(interval) progress = indexer.progress - now_scanning = progress.scanning - if now_scanning or was_scanning: + # A root waiting for the scan lock is work the operator is owed a + # sight of, even in the gap where nothing is walking yet. + busy = progress.scanning or bool(progress.queued) + if busy or was_busy: self._push_index_progress(indexer.group_id, progress) - was_scanning = now_scanning + was_busy = busy def _push_index_progress(self, group_id: str, progress) -> None: if not self._webrtc: @@ -1181,7 +1183,9 @@ class NodeDaemon: # scan. Sealing it would buy an attacker's rough estimate of a library's # size and cost a key derivation and a decrypt per push. If a field that # names anything is ever added here, that trade is void and this message - # joins the other two. + # joins the other two. `kind` is one of four fixed words; the root under + # way is a position in the roots table the member already opened from + # the sealed index, and the roots waiting are a count, never names. msg = { "type": MNP.INDEX_PROGRESS, "v": MNP_VERSION, @@ -1189,6 +1193,11 @@ class NodeDaemon: "scanning": progress.scanning, "scanned_bytes": progress.scanned_bytes, "total_bytes": progress.total_bytes, + "files_done": progress.files_done, + "files_total": progress.files_total, + "kind": progress.kind, + "root_pos": progress.root_pos, + "queued": len(progress.queued), } pushed = 0 for session in list(self._webrtc._sessions.values()): diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index db1b11c..c1b151a 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -26,7 +26,7 @@ import asyncio import logging import time from concurrent.futures import ThreadPoolExecutor -from dataclasses import dataclass +from dataclasses import dataclass, field as dataclass_field from pathlib import Path from typing import Callable, Awaitable @@ -141,6 +141,20 @@ class IndexProgress: # Basename only, deliberately not the full path — enough to show progress # without broadcasting the operator's directory structure. current_dir: str = "" + # What the operator's progress band names. `root` and `queued` are root + # names, so they stay on the loopback API like `current_dir`; members get + # `root_pos`, a position in the roots table they already opened from the + # sealed index, and the length of the queue. + root: str = "" + root_pos: int = -1 + # "scan" (a root walked for the first time), "rescan" (one that came back), + # "reconcile" (files the watcher missed), "watch" (a burst of events), or + # "" when idle. + kind: str = "" + files_done: int = 0 + files_total: int = 0 + # Roots waiting for the scan lock, in the order they will be walked. + queued: list[str] = dataclass_field(default_factory=list) def _virtual_dir(root: Root, file_path: Path) -> str: @@ -323,6 +337,8 @@ class DirectoryIndexer: self._job_running = False self._burst_scanned = 0 self._burst_total = 0 + self._burst_files_done = 0 + self._burst_files_total = 0 self._burst_dir = "" # Ids whose entry this indexer threw away and rebuilt from disk, since # the last time a consumer drained this. A rebuilt entry carries only @@ -373,20 +389,27 @@ class DirectoryIndexer: async def _initial_scan(self) -> None: self.roots.refresh_availability() total = 0 - for root in self.roots: - if not root.available: - log.warning("Root %r is not readable at startup (%s) — its files " - "are not indexed yet and will appear when it returns", - root.name, root.path) - continue - total += await self._scan_root(root) + waiting = [r.name for r in self.roots if r.available] + self._queue(waiting) + try: + for root in self.roots: + if not root.available: + log.warning("Root %r is not readable at startup (%s) — its files " + "are not indexed yet and will appear when it returns", + root.name, root.path) + continue + waiting.remove(root.name) + self._unqueue([root.name]) + total += await self._scan_root(root) + finally: + self._unqueue(waiting) self._index.version = int(time.time()) self._index.roots = self.roots.describe() self._report_collisions() log.info("Initial scan complete: %d files across %d root(s)", total, len(self.roots)) - async def _scan_root(self, root: Root) -> int: + async def _scan_root(self, root: Root, *, kind: str = "scan") -> int: log.info("Scanning %s (root %r) ...", root.path, root.name) count = 0 loop = asyncio.get_event_loop() @@ -402,7 +425,7 @@ class DirectoryIndexer: # had not yet turned on, even though the node was already several # seconds into walking and sizing a large root — total_bytes is # unknown at this point, so it starts at 0 and is corrected below. - self._begin_job(root.name) + self._begin_job(root.name, root=root, kind=kind) try: try: files = await loop.run_in_executor(self._executor, _walk_root, root) @@ -418,11 +441,13 @@ class DirectoryIndexer: sized = await loop.run_in_executor(self._executor, _size_files, files) self.progress.total_bytes = sum(size for _, size in sized) + self.progress.files_total = len(sized) self.progress.current_dir = "" for file_path, size in sized: self.progress.current_dir = file_path.parent.name entry = await self._hash_or_cached(root, file_path) self.progress.scanned_bytes += size + self.progress.files_done += 1 if entry: self._index.add_entry(entry) count += 1 @@ -433,18 +458,30 @@ class DirectoryIndexer: self._end_job() return count - def _begin_job(self, current_dir: str, total_bytes: int = 0) -> None: + def _begin_job(self, current_dir: str, total_bytes: int = 0, *, + root: Root, kind: str, files_total: int = 0) -> None: """`progress` now describes a whole-root walk, whatever a burst is doing.""" self._job_running = True - self.progress.scanning = True - self.progress.scanned_bytes = 0 - self.progress.total_bytes = total_bytes - self.progress.current_dir = current_dir + p = self.progress + p.scanning = True + p.scanned_bytes = 0 + p.total_bytes = total_bytes + p.files_done = 0 + p.files_total = files_total + p.current_dir = current_dir + p.kind = kind + p.root = root.name + p.root_pos = next((i for i, r in enumerate(self.roots) + if r.folded == root.folded), -1) def _end_job(self) -> None: self._job_running = False - self.progress.scanning = False - self.progress.current_dir = "" + p = self.progress + p.scanning = False + p.current_dir = "" + p.kind = "" + p.root = "" + p.root_pos = -1 if self._burst_inflight > 0: # A burst that started during the walk is still hashing. self._show_burst() @@ -456,11 +493,25 @@ class DirectoryIndexer: if self._burst_inflight > 0: p.scanning = True p.current_dir = self._burst_dir + p.kind = "watch" + p.root = "" + p.root_pos = -1 elif not self._pending_timers: p.scanning = False p.current_dir = "" + p.kind = "" p.scanned_bytes = self._burst_scanned p.total_bytes = self._burst_total + p.files_done = self._burst_files_done + p.files_total = self._burst_files_total + + def _queue(self, names: list[str]) -> None: + self.progress.queued.extend(names) + + def _unqueue(self, names: list[str]) -> None: + for name in names: + if name in self.progress.queued: + self.progress.queued.remove(name) async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None: """ @@ -633,6 +684,7 @@ class DirectoryIndexer: """ old_names = {r.folded for r in self.roots} new_names = {r.folded for r in roots} + self.progress.queued[:] = [n for n in self.progress.queued if fold(n) in new_names] for root in self.roots: if root.folded not in new_names: @@ -655,6 +707,7 @@ class DirectoryIndexer: await self.on_change(self) return + self._queue([r.name for r in added]) task = asyncio.create_task(self._scan_added_roots(added)) self._scan_tasks.add(task) task.add_done_callback(self._scan_tasks.discard) @@ -668,9 +721,12 @@ class DirectoryIndexer: 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: + waiting = [r.name for r in added] try: async with self._scan_lock: for root in added: + waiting.remove(root.name) + self._unqueue([root.name]) # A later retarget may have removed it while this waited. if not self._holds(root): continue @@ -689,6 +745,8 @@ class DirectoryIndexer: raise except Exception: log.exception("Scanning the added root(s) failed") + finally: + self._unqueue(waiting) # ── Reconciliation ──────────────────────────────────────────────────────── @@ -818,12 +876,14 @@ class DirectoryIndexer: except OSError: added_sized.append((p, 0)) - self._begin_job("", sum(size for _, size in added_sized)) + self._begin_job("", sum(size for _, size in added_sized), root=root, + kind="reconcile", files_total=len(added_sized)) try: for added, size in added_sized: self.progress.current_dir = added.parent.name entry = await self._hash_or_cached(root, added) self.progress.scanned_bytes += size + self.progress.files_done += 1 if not entry: continue # The index is keyed by **content hash**, so two identical @@ -883,7 +943,7 @@ class DirectoryIndexer: """ carried = {(e.id, e.name, e.path): e for e in self._entries_under(root)} self._drop_root_entries(root) - count = await self._scan_root(root) + count = await self._scan_root(root, kind="rescan") for entry in self._entries_under(root): old = carried.get((entry.id, entry.name, entry.path)) if old is None: @@ -1002,12 +1062,20 @@ class DirectoryIndexer: # A whole-root walk like any other, so it waits its turn: run beside # an added root's scan, the two reset each other's progress and read # the same drive in alternation. - async with self._scan_lock: - # Ejected or removed again while it waited. `_rescan_root` drops - # the entries before it walks, so going ahead would empty a root - # that is not there to be read. - if self._holds(root) and not root.ejected and root.is_live(): - await self._rescan_root(root) + self._queue([root.name]) + waiting = True + try: + async with self._scan_lock: + self._unqueue([root.name]) + waiting = False + # Ejected or removed again while it waited. `_rescan_root` + # drops the entries before it walks, so going ahead would + # empty a root that is not there to be read. + if self._holds(root) and not root.ejected and root.is_live(): + await self._rescan_root(root) + finally: + if waiting: + self._unqueue([root.name]) self._restart_observer() self._index.roots = self.roots.describe() self._index.version = int(time.time()) @@ -1044,7 +1112,10 @@ class DirectoryIndexer: if self._burst_inflight <= 0: self._burst_scanned = 0 self._burst_total = 0 + self._burst_files_done = 0 + self._burst_files_total = 0 self._burst_total += size + self._burst_files_total += 1 self._burst_dir = file_path.parent.name self._burst_sizes[key] = size self._burst_inflight += 1 @@ -1118,6 +1189,7 @@ class DirectoryIndexer: size = self._burst_sizes.pop(str(file_path), None) if size is not None: self._burst_scanned += size + self._burst_files_done += 1 self._burst_inflight -= 1 self._show_burst() 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 eff40f1..79a97cc 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -3268,19 +3268,25 @@ class WebRTCPeerSession: def _indexing_status(self) -> dict: """ - {"scanning": bool, "scanned_bytes": int, "total_bytes": int} for the - handshake ack and INDEX_PROGRESS pushes — never a path or filename, - that stays local to the operator's own admin UI. Absent "progress" - (context not loaded, or a group with no indexer at all) reads as - idle rather than erroring. + The counters of `_push_index_progress` (daemon.py), for the handshake + ack — never a path, a filename or a root name, which stay local to the + operator's own admin UI. Absent "progress" (context not loaded, or a + group with no indexer at all) reads as idle rather than erroring. """ progress = self._group_ctx().get("progress") if progress is None: - return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0} + return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0, + "files_done": 0, "files_total": 0, "kind": "", + "root_pos": -1, "queued": 0} return { "scanning": progress.scanning, "scanned_bytes": progress.scanned_bytes, "total_bytes": progress.total_bytes, + "files_done": progress.files_done, + "files_total": progress.files_total, + "kind": progress.kind, + "root_pos": progress.root_pos, + "queued": len(progress.queued), } # ── Transfer slots ─────────────────────────────────────────────────────── diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index de2542e..22b5afb 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -437,6 +437,34 @@ def create_ui_app(state: dict) -> FastAPI: "current_dir": progress.current_dir, } + @app.get("/api/index-status") + async def index_status_all(): + """ + Every group's indexing at once, for the client's progress band — which + is on screen whatever page the operator is on, so it cannot ask per + group. Names roots, like `current_dir` above: loopback only, the + operator's own screen. Reads state["indexers"] for the same reason. + """ + config = state.get("config") + names = {g.id: g.name for g in config.groups} if config else {} + groups = [] + for gid, indexer in list(state.get("indexers", {}).items()): + p = indexer.progress + groups.append({ + "group_id": gid, + "group_name": names.get(gid, gid[:8]), + "scanning": p.scanning, + "kind": p.kind, + "root": p.root, + "current_dir": p.current_dir, + "scanned_bytes": p.scanned_bytes, + "total_bytes": p.total_bytes, + "files_done": p.files_done, + "files_total": p.files_total, + "queued": list(p.queued), + }) + return {"groups": groups} + # ── Enabled apps (operator only, localhost) ──────────────────────────── # # Same loopback shape as member-upload: the Create Group wizard sets this |