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 --- .../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 +- 5 files changed, 306 insertions(+), 8 deletions(-) create mode 100644 packages/meshbay-node/tests/test_index_jobs_are_described.py (limited to 'packages/meshbay-node/tests') 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