From b3709ac4d362987a9d025616c95065ceed0d216b Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 23 Aug 2026 21:55:20 +0200 Subject: feat(node): persistent index cache, visible scan progress, adaptive reconcile, and delta sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Indexer performance work, in four parts: - Persistent (path, size, mtime) -> hash cache (indexer/cache.py) so a node restart no longer re-hashes every file — measured at 23 minutes for a 114 GB library on a slow disk before this, near-instant after. Hashing is deliberately kept sequential (max_workers=1): it was never actually concurrent despite the pool size, and two interleaved reads seek-thrash a spinning disk instead of going faster. - Byte-based scan progress (IndexProgress), surfaced via the loopback index-status route, the handshake ack, and a periodic INDEX_PROGRESS push to connected peers — drives a progress bar in the Create Group wizard and "add a directory" in Settings, and an animated presence dot. Guaranteed to settle back to idle via try/finally and a final push on the scanning->false transition. - The reconcile backstop's directory walks now run in the executor instead of blocking the daemon's event loop; its interval defaults to 10 min (was 60s) with adaptive backoff to 2h when nothing changes, reset on a real change or a peer connecting, and is now a per-group operator setting (signed op + group Settings UI). - INDEX_DELTA wired up (protocol support existed, nothing called it): _on_index_change now sends additions/deletions instead of rebuilding the full entries list, coalesced over a short window so a burst of file events produces one push, and the hub swarm registration for public groups only (re-)registers newly added hashes. Also fixes several bugs found while testing the above against real libraries (a 114 GB and a 100+ GB group on a USB HDD): - /api/reload blocked until the reload — including a brand-new group's full initial scan — finished, which the Electron bridge's fixed 30s call timeout turned into a hard failure on any real library. The route now fires the reload without waiting (ops.start_reload), matching add_root/remove_root's existing pattern; the wizard's own step order was fixed to wait for the group to actually appear hosted before the steps that need it (extra roots, GEK), with retries for the residual race between that and the daemon's own bookkeeping. - transport.js's hand-rolled msgpack codec had no case for uint64/int64 (0xcf/0xd3) and crashed decoding any message containing one — hit by IndexProgress.scanned_bytes/total_bytes for any group over ~4.3 GB. Verified against real msgpack-encoded bytes from the Python side. - chat_hist_resp, and this change's own index_progress and set_scan_settings_ack pushes, were not routed by message type and could be handed to an unrelated pending request by the transport's "oldest pending" fallback, stalling it until its own 30s timeout and corrupting whatever received the wrong reply in its place. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_016SF6RKNBKg9qejmoMJ9ybA --- packages/meshbay-node/tests/test_index_progress.py | 165 +++++++++++++++++++++ 1 file changed, 165 insertions(+) create mode 100644 packages/meshbay-node/tests/test_index_progress.py (limited to 'packages/meshbay-node/tests/test_index_progress.py') diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py new file mode 100644 index 0000000..52cd78b --- /dev/null +++ b/packages/meshbay-node/tests/test_index_progress.py @@ -0,0 +1,165 @@ +""" +Indexing status visible node -> client: handshake ack field, the loopback +status route for the Create Group wizard / "add a directory", and the +periodic INDEX_PROGRESS push to already-connected peers. Never the index +itself (see test_daemon.py for that) and never anything sent to the hub. +""" + +import asyncio + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from unittest.mock import MagicMock +from fastapi.testclient import TestClient + +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer.indexer import IndexProgress +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from meshbay_node.ui.app import create_ui_app + + +def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" * 32): + index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate()) + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + group_ctx = {"index": index} + if progress is not None: + group_ctx["progress"] = progress + session._ctx = {"groups": {group_id: group_ctx}} + session._group_id = group_id + return session + + +# ── _indexing_status() ─────────────────────────────────────────────────────── + +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} + + +def test_indexing_status_reflects_live_progress(): + progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000, + current_dir="StarWars") + session = _session_with_progress(progress) + + status = session._indexing_status() + + assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000} + assert "current_dir" not in status, \ + "the directory name is operator-local detail, never sent to a member" + + +# ── /api/groups/{id}/index-status (loopback) ──────────────────────────────── + +def _ui_client(state: dict) -> TestClient: + return TestClient(create_ui_app({"status": "running", "groups_ctx": {}, + "indexes": {}, **state})) + + +def test_index_status_route_idle_for_unknown_group(): + client = _ui_client({"indexers": {}}) + resp = client.get("/api/groups/unknown-group/index-status") + assert resp.status_code == 200 + assert resp.json() == {"scanning": False, "scanned_bytes": 0, + "total_bytes": 0, "current_dir": ""} + + +def test_index_status_route_reflects_indexer_progress(): + fake_indexer = MagicMock() + fake_indexer.progress = IndexProgress( + scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000, + current_dir="2024") + client = _ui_client({"indexers": {"g" * 32: fake_indexer}}) + + resp = client.get(f"/api/groups/{'g' * 32}/index-status") + + assert resp.json() == { + "scanning": True, "scanned_bytes": 1_000_000, + "total_bytes": 4_000_000_000, "current_dir": "2024", + } + + +# ── _push_index_progress / _progress_pusher ───────────────────────────────── + +def _daemon(tmp_path) -> NodeDaemon: + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(), + groups=[], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + return NodeDaemon(config) + + +@pytest.mark.asyncio +async def test_push_index_progress_only_reaches_same_group_peers(tmp_path): + daemon = _daemon(tmp_path) + + same_group = MagicMock() + same_group._group_id = "a" * 32 + same_group._send = MagicMock() + other_group = MagicMock() + other_group._group_id = "b" * 32 + other_group._send = MagicMock() + + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": same_group, "p2": other_group} + daemon._webrtc = mock_webrtc + + progress = IndexProgress(scanning=True, scanned_bytes=10, total_bytes=100) + daemon._push_index_progress("a" * 32, progress) + + same_group._send.assert_called_once() + msg = same_group._send.call_args[0][0] + assert msg["type"] == "index_progress" + assert msg["group_id"] == "a" * 32 + assert msg["scanning"] is True + assert msg["scanned_bytes"] == 10 + assert msg["total_bytes"] == 100 + other_group._send.assert_not_called() + + +@pytest.mark.asyncio +async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path): + daemon = _daemon(tmp_path) + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + indexer = MagicMock() + indexer.group_id = "a" * 32 + indexer.progress = IndexProgress(scanning=True, scanned_bytes=0, total_bytes=100) + + task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05)) + try: + # Two ticks while still scanning. + await asyncio.sleep(0.12) + assert session._send.call_count >= 2 + assert all(c.args[0]["scanning"] is True for c in session._send.call_args_list) + + # Scan finishes between ticks. + indexer.progress.scanning = False + calls_before = session._send.call_count + await asyncio.sleep(0.07) + assert session._send.call_count == calls_before + 1, \ + "exactly one final push must follow the False transition" + assert session._send.call_args.args[0]["scanning"] is False + + # Nothing further once idle. + calls_after_final = session._send.call_count + await asyncio.sleep(0.15) + assert session._send.call_count == calls_after_final, \ + "no more pushes once idle and already reported" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass -- cgit v1.2.3