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, 217 insertions, 20 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index c68f999..af85340 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -45,7 +45,7 @@ from meshbay_node.chat.store import ChatStore
 from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
 from meshbay_node.roots import RootSet, RootError
 from meshbay_node.hub_client import HubClient, HubConfig
-from meshbay_node.indexer import DirectoryIndexer
+from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex
 from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore
 from meshbay_node.roster import Roster
 from meshbay_node.transport import (
@@ -119,6 +119,17 @@ class NodeDaemon:
         self._denylist = (
             Denylist(path=config.data_dir / "denylist.json") if Denylist else None)
         self._chat_stores: dict[str, ChatStore] = {}
+        self._index_caches: dict[str, IndexCache] = {}
+        # Coalesces a burst of index changes (one per debounced watchdog
+        # event) into a single broadcast — see _on_index_change. 0.5s is
+        # short enough nobody notices the wait, long enough that dropping a
+        # few hundred files into a watched folder produces one push instead
+        # of one per file.
+        self._broadcast_coalesce_secs = 0.5
+        self._pending_broadcasts: dict[str, asyncio.TimerHandle] = {}
+        # group_id -> (version, {id: entry}) as of the last thing actually
+        # broadcast — the comparison point for the next delta.
+        self._last_broadcast_snapshot: dict[str, tuple] = {}
         self._audit_store: AuditStore | None = None
         self._bundle_store: BundleStore | None = None
         self._roster: Roster | None = None
@@ -234,12 +245,31 @@ class NodeDaemon:
                     log.info("No GEK yet for group %s — will accept first setup",
                              group_cfg.name)
 
+                index_cache = IndexCache(
+                    db_path=data_dir / group_cfg.id[:16] / "index_cache.db")
+                await index_cache.open()
+                self._index_caches[group_cfg.id] = index_cache
+
+                # Read once at load, like member_upload/enabled_apps below —
+                # kept current in place afterwards by set_scan_settings
+                # (ops.py), which updates both this indexer object directly
+                # and roster.db, so a restart picks up the same values.
+                scan_settings = (
+                    await self._roster.scan_settings(group_cfg.id)
+                    if self._roster else {
+                        "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+                        "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+                    })
+
                 indexer = DirectoryIndexer(
                     roots=roots,
                     group_id=group_cfg.id,
                     sk_node=keys.sk_ed25519,
                     gek=gek,
                     on_change=self._on_index_change,
+                    cache=index_cache,
+                    reconcile_secs=scan_settings["reconcile_interval_secs"],
+                    debounce_secs=scan_settings["debounce_secs"],
                 )
                 await indexer.start(defer_scan=True)
                 self._indexers.append(indexer)
@@ -253,6 +283,20 @@ class NodeDaemon:
                     "gek": gek,
                     "roots": roots,
                     "index": indexer.index,
+                    # Live reference, mutated in place by the indexer itself
+                    # (see IndexProgress in indexer.py) — read, never copied,
+                    # by the handshake ack and the periodic progress pusher.
+                    "progress": indexer.progress,
+                    # Bound method, called when a peer completes the
+                    # handshake — resets reconcile's backoff (indexer.py
+                    # _reconcile_loop) so the backstop is prompt again now
+                    # that someone is actually looking.
+                    "note_activity": indexer.note_activity,
+                    # Shown to the operator in Settings, and kept current in
+                    # place by set_scan_settings (ops.py) — same reasoning as
+                    # member_upload below.
+                    "reconcile_interval_secs": scan_settings["reconcile_interval_secs"],
+                    "debounce_secs": scan_settings["debounce_secs"],
                     "visibility": group_cfg.visibility,
                     # Admission policy comes from node.toml, never from the hub:
                     # a hub that could declare a group open would be handed its key.
@@ -456,6 +500,8 @@ class NodeDaemon:
                 if gctx:
                     self._tasks.append(asyncio.create_task(
                         _bg_scan(idx, group_cfg.name, gctx)))
+                    self._tasks.append(asyncio.create_task(
+                        self._progress_pusher(idx)))
 
             # 12. Wait for shutdown
             stop_event = asyncio.Event()
@@ -561,18 +607,40 @@ class NodeDaemon:
                 if gek:
                     log.info("GEK loaded for new group %s", group_cfg.id[:8])
 
+            index_cache = IndexCache(
+                db_path=data_dir / group_cfg.id[:16] / "index_cache.db")
+            await index_cache.open()
+            self._index_caches[group_cfg.id] = index_cache
+
+            scan_settings = (
+                await self._roster.scan_settings(group_cfg.id)
+                if self._roster else {
+                    "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+                    "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+                })
+
             indexer = DirectoryIndexer(
                 roots=roots,
                 group_id=group_cfg.id,
                 sk_node=sk_ed,
                 gek=gek,
                 on_change=self._on_index_change,
+                cache=index_cache,
+                reconcile_secs=scan_settings["reconcile_interval_secs"],
+                debounce_secs=scan_settings["debounce_secs"],
             )
-            await indexer.start()
+            # Registered *before* start() runs its (blocking, possibly very
+            # long — see the StarWars benchmark) initial scan, specifically
+            # so /api/groups/{id}/index-status can see indexer.progress
+            # while a brand-new group is still scanning — this is the one
+            # group state that must stay visible during the very window the
+            # group is not yet authorized for member connections (below).
             self._indexers.append(indexer)
             self._state["indexes"][group_cfg.id] = indexer.index
             self._state["indexers"][group_cfg.id] = indexer
 
+            await indexer.start()
+
             data_dir.mkdir(parents=True, exist_ok=True)
             chat_db = data_dir / group_cfg.id[:16] / "chat.db"
             store = ChatStore(db_path=chat_db)
@@ -583,6 +651,10 @@ class NodeDaemon:
                 "gek": gek,
                 "roots": roots,
                 "index": indexer.index,
+                "progress": indexer.progress,
+                "note_activity": indexer.note_activity,
+                "reconcile_interval_secs": scan_settings["reconcile_interval_secs"],
+                "debounce_secs": scan_settings["debounce_secs"],
                 "visibility": group_cfg.visibility,
                 "join_policy": group_cfg.join_policy,
                 "member_upload": (
@@ -600,6 +672,11 @@ class NodeDaemon:
             log.info("Hot-loaded group %s (%s, %d roots)",
                      group_cfg.name, group_cfg.id[:8], len(roots))
             added_names.append(group_cfg.name)
+            # The initial scan above already ran to completion (indexer.start()
+            # is not deferred here), so this only matters for whatever scans
+            # this group as time goes on — a root added later, reconcile
+            # picking one back up.
+            self._tasks.append(asyncio.create_task(self._progress_pusher(indexer)))
 
         # ── Tear down removed groups ─────────────────────────────────────
         removed_names = []
@@ -618,6 +695,16 @@ class NodeDaemon:
                     await store.close()
                 except Exception:
                     pass
+            cache = self._index_caches.pop(gid, None)
+            if cache:
+                try:
+                    await cache.close()
+                except Exception:
+                    pass
+            pending = self._pending_broadcasts.pop(gid, None)
+            if pending:
+                pending.cancel()
+            self._last_broadcast_snapshot.pop(gid, None)
             self._state["indexes"].pop(gid, None)
             self._state["indexers"].pop(gid, None)
             old_name = gid[:8]
@@ -708,46 +795,149 @@ class NodeDaemon:
         log.warning("No unwrappable GEK bundle found for group %s", group_id[:8])
         return None
 
+    async def _progress_pusher(self, indexer: DirectoryIndexer,
+                               interval: float = 2.0) -> None:
+        """
+        Watches indexer.progress and pushes a light INDEX_PROGRESS message to
+        this group's connected peers — never the index itself, that stays
+        _on_index_change's job. Runs for the node's whole lifetime: a scan
+        can start from several places (initial scan, a root added later,
+        reconcile picking a root back up), and this only needs to notice the
+        flag, not why it changed.
+
+        The final push at the False transition is what lets a presence dot
+        reliably turn back off on an already-connected client — the
+        handshake ack only covers the moment of connecting. `interval` is a
+        parameter (not a bare constant) only so a test can drive this loop
+        without waiting on the real 2s cadence.
+        """
+        was_scanning = False
+        while True:
+            await asyncio.sleep(interval)
+            progress = indexer.progress
+            now_scanning = progress.scanning
+            if now_scanning or was_scanning:
+                self._push_index_progress(indexer.group_id, progress)
+            was_scanning = now_scanning
+
+    def _push_index_progress(self, group_id: str, progress) -> None:
+        if not self._webrtc:
+            return
+        msg = {
+            "type": MNP.INDEX_PROGRESS,
+            "v": MNP_VERSION,
+            "group_id": group_id,
+            "scanning": progress.scanning,
+            "scanned_bytes": progress.scanned_bytes,
+            "total_bytes": progress.total_bytes,
+        }
+        pushed = 0
+        for session in list(self._webrtc._sessions.values()):
+            if session._group_id == group_id:
+                try:
+                    session._send(msg)
+                    pushed += 1
+                except Exception:
+                    pass
+        if pushed:
+            log.debug("Index progress pushed to %d peer(s) for group %s",
+                      pushed, group_id[:8])
+
     async def _on_index_change(self, indexer: DirectoryIndexer) -> None:
-        """Called when a DirectoryIndexer detects file changes."""
+        """
+        Called when a DirectoryIndexer detects file changes — once per
+        debounced watchdog event, so dropping N files into a watched folder
+        calls this N times in quick succession. Coalesces those into one
+        broadcast (_broadcast_index_change) rather than one push per file:
+        the timer is reset on every call and only fires once calls stop
+        arriving for _broadcast_coalesce_secs.
+        """
+        group_id = indexer.group_id
+        loop = asyncio.get_event_loop()
+        pending = self._pending_broadcasts.pop(group_id, None)
+        if pending:
+            pending.cancel()
+
+        def fire() -> None:
+            self._pending_broadcasts.pop(group_id, None)
+            asyncio.ensure_future(self._broadcast_index_change(indexer))
+
+        self._pending_broadcasts[group_id] = loop.call_later(
+            self._broadcast_coalesce_secs, fire)
+
+    async def _broadcast_index_change(self, indexer: DirectoryIndexer) -> None:
+        """
+        The actual push, run once per coalesced burst. Sends a full
+        INDEX_SYNC the first time a group is ever broadcast (no previous
+        snapshot to diff against — the client's own first fetchIndex() call
+        already covers that case) and an INDEX_DELTA every time after,
+        computed against the last thing this method actually sent.
+        """
         group_id = indexer.group_id
         idx = indexer.index
         log.info("Index changed for group %s: %d files (v%d)",
                  group_id[:8], idx.count, idx.version)
 
-        # 11.5 — Push updated index to connected WebRTC peers in this group
+        prev = self._last_broadcast_snapshot.get(group_id)
+        delta = None
+        if prev is not None:
+            prev_version, prev_entries = prev
+            previous = GroupIndex._snapshot(
+                idx.group_id, idx.sk_node, idx.gek, prev_version, prev_entries)
+            delta = idx.diff(previous)
+        self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id())
+
+        # 11.5 — Push to connected WebRTC peers in this group
         if self._webrtc:
-            entries = [
-                {
-                    "id": e.id, "name": e.name, "path": e.path,
-                    "size": e.size, "type": e.type, "added_at": e.added_at,
+            if delta is not None:
+                msg = {
+                    "type": MNP.INDEX_DELTA,
+                    "v": MNP_VERSION,
+                    "group_id": idx.group_id,
+                    "base_version": delta.base_version,
+                    "version": delta.version,
+                    "additions": [
+                        {"id": e.id, "name": e.name, "path": e.path,
+                         "size": e.size, "type": e.type, "added_at": e.added_at}
+                        for e in delta.additions
+                    ],
+                    "deletions": delta.deletions,
+                }
+            else:
+                msg = {
+                    "type": MNP.INDEX_SYNC,
+                    "v": MNP_VERSION,
+                    "group_id": idx.group_id,
+                    "version": idx.version,
+                    "entries": [
+                        {"id": e.id, "name": e.name, "path": e.path,
+                         "size": e.size, "type": e.type, "added_at": e.added_at}
+                        for e in idx.entries
+                    ],
                 }
-                for e in idx.entries
-            ]
-            sync_msg = {
-                "type": MNP.INDEX_SYNC,
-                "v": MNP_VERSION,
-                "group_id": idx.group_id,
-                "version": idx.version,
-                "entries": entries,
-            }
             pushed = 0
             for session in list(self._webrtc._sessions.values()):
                 if session._group_id == group_id:
                     try:
-                        session._send(sync_msg)
+                        session._send(msg)
                         pushed += 1
                     except Exception:
                         pass
             if pushed:
-                log.info("Index pushed to %d WebRTC peers", pushed)
+                log.info("Index %s pushed to %d WebRTC peers",
+                         "delta" if delta is not None else "sync", pushed)
 
         # 11.9 — Register file hashes with hub swarm table (public groups only, H7)
         group_cfg = next(
             (g for g in self._config.groups if g.id == group_id), None)
         if (self._hub and self._state.get("endpoint_hint")
                 and group_cfg and group_cfg.visibility == "public"):
-            hashes = [e.id for e in idx.entries]
+            # Only the newly added hashes once there is a delta to know them
+            # from — registering the whole library again on every change is
+            # the same O(changes x library size) cost the delta above exists
+            # to avoid.
+            hashes = ([e.id for e in delta.additions] if delta is not None
+                     else [e.id for e in idx.entries])
             if hashes:
                 endpoint = f"webrtc:{self._config.node.quic_port}"
                 asyncio.ensure_future(self._register_swarm(hashes, endpoint))
@@ -772,6 +962,10 @@ class NodeDaemon:
         log.info("Shutting down...")
         self._state["status"] = "stopping"
 
+        for handle in self._pending_broadcasts.values():
+            handle.cancel()
+        self._pending_broadcasts.clear()
+
         for task in self._tasks:
             task.cancel()
         for task in self._tasks:
@@ -795,6 +989,9 @@ class NodeDaemon:
         for store in self._chat_stores.values():
             await store.close()
 
+        for cache in self._index_caches.values():
+            await cache.close()
+
         for indexer in self._indexers:
             await indexer.stop()