From a294c1d338ba4c20d66873d593d1c101e69c5a40 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Mon, 14 Sep 2026 11:08:12 +0200 Subject: feat(node): progress names the root under way and the roots waiting `IndexProgress` said "scanning, this many bytes of that many" and nothing more. A group's roots are walked one after another, so a second directory added during a large scan showed as the bar jumping back to 0 %. It now also carries the root being walked and its position in the roots table, the kind of walk (scan, rescan, reconcile, watch), file counts, and the roots waiting for the scan lock in order: queued by the initial scan, by a retarget, and by a plug; dropped when a root is removed. `GET /api/index-status` answers for every group at once, including a group still in its initial scan, so a client can show indexing on any page. It names roots: loopback only, like `current_dir`. `index_progress` and the handshake ack gain the same counters, still naming nothing (decision D3): the root is a position in the roots table the member already opened from the sealed index, and the queue is a count. The pusher keeps speaking while a root only waits for the lock. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01T6jPTeocXA1BePekdsgPya --- packages/meshbay-node/src/meshbay_node/daemon.py | 19 +- .../src/meshbay_node/indexer/indexer.py | 122 +++++++++--- .../src/meshbay_node/transport/webrtc_server.py | 18 +- packages/meshbay-node/src/meshbay_node/ui/app.py | 28 +++ .../test_added_root_does_not_wait_for_its_scan.py | 4 +- .../tests/test_index_jobs_are_described.py | 207 +++++++++++++++++++++ .../meshbay-node/tests/test_index_no_cleartext.py | 8 +- packages/meshbay-node/tests/test_index_progress.py | 91 ++++++++- .../tests/test_root_work_outlives_the_session.py | 4 +- 9 files changed, 457 insertions(+), 44 deletions(-) create mode 100644 packages/meshbay-node/tests/test_index_jobs_are_described.py (limited to 'packages/meshbay-node') 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 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 index 2dc31c9..a80645a 100644 --- 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 @@ -60,10 +60,10 @@ class _GatedIndexer(DirectoryIndexer): self.gate = asyncio.Event() self.at_gate = asyncio.Event() - async def _scan_root(self, root): + async def _scan_root(self, root, **kwargs): self.at_gate.set() await self.gate.wait() - return await super()._scan_root(root) + return await super()._scan_root(root, **kwargs) async def _indexer(one: Path, **kwargs) -> _GatedIndexer: diff --git a/packages/meshbay-node/tests/test_index_jobs_are_described.py b/packages/meshbay-node/tests/test_index_jobs_are_described.py new file mode 100644 index 0000000..dee6e39 --- /dev/null +++ b/packages/meshbay-node/tests/test_index_jobs_are_described.py @@ -0,0 +1,207 @@ +""" +What the operator's progress band is told about the indexing under way. + +`progress` said "scanning, this many bytes of that many" and nothing more. That +is one bar with no name on it. A node asked to add a second directory while the +first is still hashing does them one after the other — one scan lock, one +hashing thread — and an operator looking at a bar that jumps back to 0 % cannot +tell a second directory from a scan that started over. So it now says which +root, what kind of walk, how many files, and which roots wait their turn. +""" + +import asyncio +import os +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet + +pytestmark = pytest.mark.asyncio + +GROUP = "g" * 32 + + +def _set(*specs) -> RootSet: + return RootSet.build([s if isinstance(s, dict) else {"path": str(s), "name": s.name} + for s in specs]) + + +def _names(idx) -> list[str]: + return sorted(e.name for e in idx.index.entries) + + +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 + + +class _Held(DirectoryIndexer): + """Stops before hashing the first file under `hold_under`, and records walks.""" + hold_under: Path | None = None + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.gate = asyncio.Event() + self.at_gate = asyncio.Event() + self.walked: list[str] = [] + + async def _scan_root(self, root, *args, **kwargs): + self.walked.append(root.name) + return await super()._scan_root(root, *args, **kwargs) + + async def _hash_or_cached(self, root, file_path): + if self.hold_under is not None and file_path.is_relative_to(self.hold_under): + self.at_gate.set() + await self.gate.wait() + return await super()._hash_or_cached(root, file_path) + + +def _dirs(tmp_path: Path) -> tuple[Path, Path, Path]: + one, r1, r2 = tmp_path / "one", tmp_path / "r1", tmp_path / "r2" + for d in (one, r1, r2): + d.mkdir() + (one / "a.txt").write_bytes(b"first root") + (r1 / "b.bin").write_bytes(os.urandom(2000)) + (r1 / "c.bin").write_bytes(os.urandom(3000)) + (r2 / "d.bin").write_bytes(os.urandom(4000)) + return one, r1, r2 + + +async def _drain(idx) -> None: + await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5) + + +async def test_the_root_under_way_and_the_ones_waiting_are_named(tmp_path): + one, r1, r2 = _dirs(tmp_path) + idx = _Held(roots=_set(one), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + await idx.initial_scan() + idx.walked.clear() + idx.hold_under = r1 + try: + await idx.retarget(_set(one, r1), wait=False) + await asyncio.wait_for(idx.at_gate.wait(), 5) + p = idx.progress + assert (p.scanning, p.kind, p.root, p.root_pos) == (True, "scan", "r1", 1) + assert (p.files_done, p.files_total, p.total_bytes) == (0, 2, 5000) + assert p.queued == [] + + await idx.retarget(_set(one, r1, r2), wait=False) + assert p.queued == ["r2"] + + idx.gate.set() + await _drain(idx) + assert idx.walked == ["r1", "r2"], "the second root did not wait for the first" + assert (p.scanning, p.kind, p.root, p.queued) == (False, "", "", []) + assert _names(idx) == ["a.txt", "b.bin", "c.bin", "d.bin"] + finally: + idx.gate.set() + await idx.stop() + + +async def test_a_root_removed_while_it_waited_leaves_the_queue(tmp_path): + one, r1, r2 = _dirs(tmp_path) + idx = _Held(roots=_set(one), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + await idx.initial_scan() + idx.walked.clear() + idx.hold_under = r1 + try: + await idx.retarget(_set(one, r1), wait=False) + await asyncio.wait_for(idx.at_gate.wait(), 5) + await idx.retarget(_set(one, r1, r2), wait=False) + await idx.retarget(_set(one, r1), wait=False) + assert idx.progress.queued == [], "a removed root is still announced as next" + + idx.gate.set() + await _drain(idx) + assert idx.walked == ["r1"] + assert idx.progress.queued == [] + finally: + idx.gate.set() + await idx.stop() + + +async def test_the_initial_scan_names_the_roots_still_to_come(tmp_path): + _, r1, r2 = _dirs(tmp_path) + idx = _Held(roots=_set(r1, r2), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + idx.hold_under = r1 + scan = asyncio.create_task(idx.initial_scan()) + try: + await asyncio.wait_for(idx.at_gate.wait(), 5) + assert (idx.progress.root, idx.progress.queued) == ("r1", ["r2"]) + idx.gate.set() + await asyncio.wait_for(scan, 5) + assert idx.progress.queued == [] + finally: + idx.gate.set() + await idx.stop() + + +async def test_a_failed_initial_scan_leaves_nothing_announced(tmp_path): + _, r1, r2 = _dirs(tmp_path) + + class _Broken(DirectoryIndexer): + async def _hash_or_cached(self, root, file_path): + raise RuntimeError("simulated failure mid-scan") + + idx = _Broken(roots=_set(r1, r2), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + with pytest.raises(RuntimeError): + await idx.initial_scan() + p = idx.progress + assert (p.scanning, p.kind, p.root, p.queued) == (False, "", "", []) + + +async def test_a_plug_waiting_its_turn_is_announced(tmp_path): + one, r1, _ = _dirs(tmp_path) + two = tmp_path / "two" + two.mkdir() + (two / "e.txt").write_bytes(b"removable") + idx = _Held(roots=_set(one, two), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + await idx.initial_scan() + idx.eject_root("two") + idx.hold_under = r1 + try: + await idx.retarget( + _set(one, {"path": str(two), "name": "two", "ejected": True}, r1), wait=False) + await asyncio.wait_for(idx.at_gate.wait(), 5) + plug = asyncio.create_task(idx.plug_root("two")) + assert await _until(lambda: idx.progress.queued == ["two"]) + + idx.gate.set() + await asyncio.wait_for(plug, 5) + assert idx.progress.queued == [] + assert idx.walked[-1] == "two" + finally: + idx.gate.set() + await idx.stop() + + +async def test_a_burst_is_described_as_watching(tmp_path): + one, _, _ = _dirs(tmp_path) + idx = DirectoryIndexer(roots=_set(one), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None, + debounce_secs=0.01) + await idx.initial_scan() + idx._loop = asyncio.get_running_loop() + for i in range(2): + f = one / f"new{i}.bin" + f.write_bytes(os.urandom(1000)) + idx._schedule_update(f) + await asyncio.sleep(0) + + p = idx.progress + assert (p.scanning, p.kind, p.root, p.root_pos) == (True, "watch", "", -1) + assert (p.files_done, p.files_total) == (0, 2) + + assert await _until(lambda: not p.scanning) + assert (p.kind, p.files_done, p.files_total) == ("", 2, 2) diff --git a/packages/meshbay-node/tests/test_index_no_cleartext.py b/packages/meshbay-node/tests/test_index_no_cleartext.py index f510884..e5e4555 100644 --- a/packages/meshbay-node/tests/test_index_no_cleartext.py +++ b/packages/meshbay-node/tests/test_index_no_cleartext.py @@ -151,9 +151,15 @@ def test_index_progress_stays_clear_and_stays_counters(): dicts = [n for n in ast.walk(tree) if isinstance(n, ast.Dict)] assert len(dicts) == 1, "more than one message built here — re-read this test" keys = {k.value for k in dicts[0].keys} + # `kind` is one of four fixed words and `root_pos`/`queued` are integers — + # the root names themselves stay on the loopback API. assert keys == {"type", "v", "group_id", - "scanning", "scanned_bytes", "total_bytes"}, ( + "scanning", "scanned_bytes", "total_bytes", + "files_done", "files_total", "kind", "root_pos", "queued"}, ( f"index_progress now carries {keys} — re-read decision D3 before shipping it") + values = {k.value: v for k, v in zip(dicts[0].keys, dicts[0].values)} + assert ast.unparse(values["queued"]) == "len(progress.queued)", ( + "the roots waiting must go out as a count, never as their names") assert "seal(" not in source assert "D3" in source, "the reason it is not sealed must stay next to the code" diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py index 52cd78b..df51e85 100644 --- a/packages/meshbay-node/tests/test_index_progress.py +++ b/packages/meshbay-node/tests/test_index_progress.py @@ -36,19 +36,26 @@ def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" * def test_indexing_status_defaults_idle_when_no_progress_tracked(): session = _session_with_progress(None) assert session._indexing_status() == { - "scanning": False, "scanned_bytes": 0, "total_bytes": 0} + "scanning": False, "scanned_bytes": 0, "total_bytes": 0, + "files_done": 0, "files_total": 0, "kind": "", "root_pos": -1, "queued": 0} def test_indexing_status_reflects_live_progress(): progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000, - current_dir="StarWars") + current_dir="Season 2", root="series", root_pos=1, + kind="scan", files_done=3, files_total=9, + queued=["archive", "photos"]) session = _session_with_progress(progress) status = session._indexing_status() - assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000} + assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000, + "files_done": 3, "files_total": 9, "kind": "scan", + "root_pos": 1, "queued": 2} assert "current_dir" not in status, \ "the directory name is operator-local detail, never sent to a member" + assert not {"series", "archive", "photos"} & {str(v) for v in status.values()}, \ + "a root name reached a member" # ── /api/groups/{id}/index-status (loopback) ──────────────────────────────── @@ -81,6 +88,36 @@ def test_index_status_route_reflects_indexer_progress(): } +# ── /api/index-status (loopback, every group) ─────────────────────────────── + +def test_every_group_is_described_including_one_still_being_attached(): + """The band reads one route for the whole node. A group in its initial scan + is in state["indexers"] and not yet in groups_ctx, and must be there.""" + busy, idle = MagicMock(), MagicMock() + busy.progress = IndexProgress( + scanning=True, scanned_bytes=10, total_bytes=40, current_dir="2024", + root="results", root_pos=1, kind="scan", files_done=1, files_total=4, + queued=["archive"]) + idle.progress = IndexProgress() + config = MagicMock() + named = MagicMock() + named.id, named.name = "a" * 32, "outputs" + config.groups = [named] + client = _ui_client({"config": config, + "indexers": {"a" * 32: busy, "b" * 32: idle}}) + + groups = {g["group_id"]: g for g in client.get("/api/index-status").json()["groups"]} + + assert groups["a" * 32] == { + "group_id": "a" * 32, "group_name": "outputs", "scanning": True, + "kind": "scan", "root": "results", "current_dir": "2024", + "scanned_bytes": 10, "total_bytes": 40, "files_done": 1, "files_total": 4, + "queued": ["archive"], + } + assert groups["b" * 32]["group_name"] == "b" * 8 + assert groups["b" * 32]["scanning"] is False + + # ── _push_index_progress / _progress_pusher ───────────────────────────────── def _daemon(tmp_path) -> NodeDaemon: @@ -122,6 +159,54 @@ async def test_push_index_progress_only_reaches_same_group_peers(tmp_path): other_group._send.assert_not_called() +@pytest.mark.asyncio +async def test_push_index_progress_carries_counters_never_root_names(tmp_path): + daemon = _daemon(tmp_path) + session = MagicMock() + session._group_id = "a" * 32 + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + daemon._push_index_progress("a" * 32, IndexProgress( + scanning=True, scanned_bytes=10, total_bytes=100, current_dir="2024", + root="results", root_pos=2, kind="rescan", files_done=5, files_total=50, + queued=["archive", "photos"])) + + msg = session._send.call_args[0][0] + assert (msg["kind"], msg["root_pos"], msg["queued"]) == ("rescan", 2, 2) + assert (msg["files_done"], msg["files_total"]) == (5, 50) + assert not {"results", "archive", "photos", "2024"} & {str(v) for v in msg.values()} + + +@pytest.mark.asyncio +async def test_progress_pusher_speaks_while_a_root_only_waits(tmp_path): + """Between one root's scan ending and the next one taking the lock, nothing + is scanning — but there is work coming, and the band must not drop it.""" + daemon = _daemon(tmp_path) + session = MagicMock() + session._group_id = "a" * 32 + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + indexer = MagicMock() + indexer.group_id = "a" * 32 + indexer.progress = IndexProgress(scanning=False, queued=["archive"]) + + task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05)) + try: + await asyncio.sleep(0.12) + assert session._send.call_count >= 2 + assert session._send.call_args.args[0]["queued"] == 1 + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path): daemon = _daemon(tmp_path) 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 index 31de12e..1e31c31 100644 --- a/packages/meshbay-node/tests/test_root_work_outlives_the_session.py +++ b/packages/meshbay-node/tests/test_root_work_outlives_the_session.py @@ -149,11 +149,11 @@ async def test_a_plug_whose_caller_goes_away_does_not_empty_the_root(tmp_path): gate = None at_gate = None - async def _scan_root(self, root): + 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) + return await super()._scan_root(root, **kwargs) idx = _Held(roots=_set(one, two), group_id=GROUP, sk_node=Ed25519PrivateKey.generate(), gek=None) -- cgit v1.2.3