diff options
Diffstat (limited to 'packages/meshbay-node')
9 files changed, 457 insertions, 44 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 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: @@ -123,6 +160,54 @@ async def test_push_index_progress_only_reaches_same_group_peers(tmp_path): @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) |