feat(node): persistent index cache, visible scan progress, adaptive reconcile, and delta sync

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016SF6RKNBKg9qejmoMJ9ybA

1 files changed, 289 insertions, 1 deletions

diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index c304361..68ccc4c 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -9,7 +9,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 
 from meshbay_common.crypto import generate_gek
-from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache
+import meshbay_node.indexer.indexer as indexer_mod
 from conftest import one_root
 from meshbay_node.keystore import NodeKeys
 
@@ -190,3 +191,290 @@ async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek):
     wire = indexer.index.serialize()
     recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
     assert recovered.count == indexer.index.count
+
+
+# ── Cache-aware scanning ───────────────────────────────────────────────────────
+
+@pytest.fixture
+async def index_cache(tmp_path):
+    c = IndexCache(db_path=tmp_path / "index_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+@pytest.mark.asyncio
+async def test_second_scan_with_same_cache_hashes_nothing(
+        shared_dir, sk_node, gek, index_cache):
+    """
+    The whole point of the cache: a "restart" (a fresh DirectoryIndexer, same
+    on-disk cache) that finds every file's (size, mtime) unchanged must not
+    read a single byte of file content.
+    """
+    first = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                             sk_node=sk_node, gek=gek, cache=index_cache)
+    await first.initial_scan()
+    assert first.index.count == 4
+
+    calls = []
+    real_scan_file = indexer_mod._scan_file
+
+    def spy(root, path):
+        calls.append(path)
+        return real_scan_file(root, path)
+
+    indexer_mod._scan_file = spy
+    try:
+        second = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                                  sk_node=sk_node, gek=gek, cache=index_cache)
+        await second.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert calls == [], f"expected zero hash calls on a fully-cached rescan, got {calls}"
+    assert second.index.count == first.index.count
+    assert {e.id for e in second.index.entries} == {e.id for e in first.index.entries}
+
+
+@pytest.mark.asyncio
+async def test_modified_file_is_rehashed(tmp_path, sk_node, gek, index_cache):
+    d = tmp_path / "shared"
+    d.mkdir()
+    f = d / "movie.mkv"
+    f.write_bytes(b"original content")
+
+    first = DirectoryIndexer(roots=one_root(d), group_id="g",
+                             sk_node=sk_node, gek=gek, cache=index_cache)
+    await first.initial_scan()
+    old_id = first.index.entries[0].id
+
+    # Change both content and mtime, as any real edit would.
+    f.write_bytes(b"a completely different, longer payload")
+    os.utime(f, (time.time() + 5, time.time() + 5))
+
+    second = DirectoryIndexer(roots=one_root(d), group_id="g",
+                              sk_node=sk_node, gek=gek, cache=index_cache)
+    await second.initial_scan()
+
+    assert second.index.count == 1
+    assert second.index.entries[0].id != old_id
+
+
+@pytest.mark.asyncio
+async def test_scan_interrupted_partway_leaves_only_completed_files_cached(
+        tmp_path, sk_node, gek, index_cache):
+    """
+    A cache row is only ever written after a file is fully hashed (cache.py),
+    so a crash mid-scan cannot leave a stale/partial row — the next scan just
+    treats the not-yet-cached files as new, and finishes the job.
+    """
+    d = tmp_path / "shared"
+    d.mkdir()
+    names = [f"file{i}.bin" for i in range(5)]
+    for i, name in enumerate(names):
+        (d / name).write_bytes(os.urandom(64) * (i + 1))
+
+    real_scan_file = indexer_mod._scan_file
+    hashed_before_crash = []
+
+    def crash_after_three(root, path):
+        if len(hashed_before_crash) >= 3:
+            raise RuntimeError("simulated crash mid-scan")
+        entry = real_scan_file(root, path)
+        hashed_before_crash.append(path)
+        return entry
+
+    indexer_mod._scan_file = crash_after_three
+    try:
+        crashing = DirectoryIndexer(roots=one_root(d), group_id="g",
+                                    sk_node=sk_node, gek=gek, cache=index_cache)
+        with pytest.raises(RuntimeError):
+            await crashing.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert len(hashed_before_crash) == 3
+
+    # A normal rescan (same cache) must still end up with all 5 files
+    # correctly indexed, hashing only the ones the crash never got to.
+    calls = []
+
+    def spy(root, path):
+        calls.append(path)
+        return real_scan_file(root, path)
+
+    indexer_mod._scan_file = spy
+    try:
+        resumed = DirectoryIndexer(roots=one_root(d), group_id="g",
+                                   sk_node=sk_node, gek=gek, cache=index_cache)
+        await resumed.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert resumed.index.count == 5
+    assert len(calls) == 2, f"expected only the 2 not-yet-cached files to be hashed, got {len(calls)}"
+
+
+# ── Progress state ───────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_progress_reflects_bytes_scanned(shared_dir, sk_node, gek):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek)
+    assert indexer.progress.scanning is False
+
+    await indexer.initial_scan()
+
+    total_size = sum(f.stat().st_size for f in shared_dir.rglob("*") if f.is_file())
+    assert indexer.progress.scanning is False, "must end idle, not stuck scanning"
+    assert indexer.progress.scanned_bytes == total_size
+    assert indexer.progress.total_bytes == total_size
+
+
+@pytest.mark.asyncio
+async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek):
+    d = tmp_path / "shared"
+    d.mkdir()
+    (d / "a.bin").write_bytes(os.urandom(64))
+    (d / "b.bin").write_bytes(os.urandom(64))
+
+    real_scan_file = indexer_mod._scan_file
+
+    def boom(root, path):
+        raise RuntimeError("simulated failure mid-scan")
+
+    indexer = DirectoryIndexer(roots=one_root(d), group_id="g",
+                               sk_node=sk_node, gek=gek)
+    indexer_mod._scan_file = boom
+    try:
+        with pytest.raises(RuntimeError):
+            await indexer.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert indexer.progress.scanning is False, \
+        "an exception mid-scan must not leave the scanning flag stuck on"
+
+
+# ── Off-loop directory walks, reconcile backoff ─────────────────────────────
+
+@pytest.mark.asyncio
+async def test_walk_root_does_not_stall_the_event_loop(tmp_path, sk_node, gek):
+    d = tmp_path / "shared"
+    d.mkdir()
+    (d / "f.bin").write_bytes(b"x")
+
+    real_walk = indexer_mod._walk_root
+
+    def slow_walk(root):
+        time.sleep(0.2)
+        return real_walk(root)
+
+    indexer_mod._walk_root = slow_walk
+    ticks = 0
+
+    async def ticker():
+        nonlocal ticks
+        while True:
+            await asyncio.sleep(0.01)
+            ticks += 1
+
+    ticker_task = asyncio.create_task(ticker())
+    try:
+        indexer = DirectoryIndexer(roots=one_root(d), group_id="g",
+                                   sk_node=sk_node, gek=gek)
+        await indexer.initial_scan()
+    finally:
+        indexer_mod._walk_root = real_walk
+        ticker_task.cancel()
+        try:
+            await ticker_task
+        except asyncio.CancelledError:
+            pass
+
+    assert ticks >= 5, (
+        "the event loop must keep running other tasks while a directory "
+        f"walk is in progress in the executor — only {ticks} ticks happened "
+        "during a 0.2s walk")
+
+
+@pytest.mark.asyncio
+async def test_reconcile_backoff_grows_with_no_changes_then_caps(shared_dir, sk_node, gek):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek, reconcile_secs=0.01)
+    await indexer.initial_scan()
+    assert indexer._reconcile_delay == 0.01
+
+    task = asyncio.create_task(indexer._reconcile_loop())
+    try:
+        await asyncio.sleep(0.2)
+        assert indexer._reconcile_delay > 0.01, \
+            "several no-change ticks must have grown the delay"
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    # A low, instance-only cap so the clamp is observable without waiting
+    # through dozens of real doublings up to the real 7200s ceiling.
+    indexer.RECONCILE_BACKOFF_CAP = 0.05
+    indexer._reconcile_delay = 0.04
+    task = asyncio.create_task(indexer._reconcile_loop())
+    try:
+        await asyncio.sleep(0.15)
+        assert indexer._reconcile_delay <= 0.05, "delay must never exceed the cap"
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+
+@pytest.mark.asyncio
+async def test_note_activity_resets_backoff(shared_dir, sk_node, gek):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek, reconcile_secs=10.0)
+    await indexer.initial_scan()
+    indexer._reconcile_delay = 5000.0  # simulate a long-idle backoff
+
+    indexer.note_activity()
+
+    assert indexer._reconcile_delay == 10.0
+
+
+@pytest.mark.asyncio
+async def test_reconcile_backoff_resets_when_something_actually_changes(
+        shared_dir, sk_node, gek):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek, reconcile_secs=0.02)
+    await indexer.initial_scan()
+    indexer._reconcile_delay = 5.0  # pretend it had already backed off a lot
+
+    # A fake reconcile() rather than a real filesystem change: the real
+    # sweep's timing (disk I/O, the executor round trip) would race against
+    # this test's own sleeps. What matters here is only _reconcile_loop's
+    # reaction to "something changed", not reconcile()'s own detection logic
+    # — that is covered separately (test_root_availability.py).
+    reconciled_once = asyncio.Event()
+
+    async def fake_reconcile():
+        reconciled_once.set()
+        return True
+
+    indexer._reconcile_delay = 0.01
+    indexer.reconcile = fake_reconcile
+
+    task = asyncio.create_task(indexer._reconcile_loop())
+    try:
+        await asyncio.wait_for(reconciled_once.wait(), timeout=2.0)
+        assert indexer._reconcile_delay == 0.02, \
+            "a real change must reset the delay back to the base interval"
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass