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, 205 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_scan_settings_policy.py b/packages/meshbay-node/tests/test_scan_settings_policy.py
new file mode 100644
index 0000000..719b988
--- /dev/null
+++ b/packages/meshbay-node/tests/test_scan_settings_policy.py
@@ -0,0 +1,205 @@
+"""
+The operator can tune how often the indexer's reconciliation backstop runs,
+and how long it waits after a file's last write before hashing it.
+
+Same shape as test_apps_enabled_policy.py / test_member_upload_policy.py:
+changed by a signed operator instruction, stored on the node rather than the
+hub. Unlike those two, there is also a *live* DirectoryIndexer object to
+update — see test_set_scan_settings_updates_the_live_indexer below.
+"""
+
+import os
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_SET_SCAN_SETTINGS
+from meshbay_common.crypto import generate_gek
+from meshbay_node import ops
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture
+def gek():
+    return generate_gek()
+
+
+@pytest.fixture
+def shared_dir(tmp_path):
+    d = tmp_path / "shared"
+    d.mkdir()
+    (d / "video.mkv").write_bytes(os.urandom(256))
+    return d
+
+
+def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {
+        "roots": one_root(shared_root),
+        "index": index,
+        "sk_node": index.sk_node,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = None
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+# ── Refused before a challenge is even issued ───────────────────────────────
+
+async def test_out_of_range_reconcile_interval_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": 1.0, "debounce_secs": 2.0})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_out_of_range_debounce_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": 600.0, "debounce_secs": 99999.0})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_non_numeric_values_are_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": "not-a-number", "debounce_secs": 2.0})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    session = _session(tmp_path, "member-1", operator="the-operator")
+    session._has_admin_authority = lambda: False
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0})
+
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Who may change it ───────────────────────────────────────────────────────
+
+async def test_changing_it_needs_a_signature(tmp_path):
+    """The request only ever produces a challenge — nothing is applied
+    until a signature over the transcript verifies."""
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0})
+
+    assert issued == [(OP_SET_SCAN_SETTINGS, "600,2")]
+
+
+# ── Where it is stored ──────────────────────────────────────────────────────
+
+async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        defaults = await roster.scan_settings("g1")
+        assert defaults == {
+            "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+            "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+        }, "unset must mean the indexer's own defaults, or an upgrade " \
+           "changes behaviour for every existing group"
+
+        await roster.set_scan_settings("g1", 1200.0, 5.0, set_by="op")
+        assert await roster.scan_settings("g1") == {
+            "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0}
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.scan_settings("g1") == {
+            "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0}
+        assert await reopened.scan_settings("g2") == {
+            "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+            "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+        }, "one group's setting must not answer for another"
+    finally:
+        await reopened.close()
+
+
+# ── Applying it to the live indexer ─────────────────────────────────────────
+
+async def test_set_scan_settings_updates_the_live_indexer(tmp_path, shared_dir, gek):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    indexer = DirectoryIndexer(
+        roots=one_root(shared_dir), group_id="g1",
+        sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer.initial_scan()
+    indexer._reconcile_delay = 5000.0  # simulate a long-idle backoff
+    state = {"roster": roster, "indexers": {"g1": indexer}}
+
+    try:
+        result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0)
+
+        assert result == {"reconcile_interval_secs": 1800.0, "debounce_secs": 3.0,
+                          "group_id": "g1"}
+        assert indexer.reconcile_secs == 1800.0
+        assert indexer.debounce_secs == 3.0
+        assert indexer._reconcile_delay == 1800.0, \
+            "the new interval must apply right away, not after whatever " \
+            "backoff had already stretched the wait to"
+        assert await roster.scan_settings("g1") == {
+            "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0}
+    finally:
+        await roster.close()
+
+
+async def test_set_scan_settings_without_a_live_indexer_still_persists(tmp_path):
+    """A group hosted on the node but with no running indexer in this
+    process (e.g. a test, or a group not yet hot-loaded) must not crash —
+    the setting still lands in roster.db for whenever it is."""
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state = {"roster": roster, "indexers": {}}
+
+    try:
+        result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0)
+        assert result["reconcile_interval_secs"] == 1800.0
+        assert await roster.scan_settings("g1") == {
+            "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0}
+    finally:
+        await roster.close()