feat(node): share the (path,size,mtime)->hash index cache across every group

An operator routinely shares the same physical folder into more than one
group (a music library, a Séries drive) — IndexCache used to be opened
once per group (data_dir/{group_id}/index_cache.db), so the second group
to reference an already-fully-hashed multi-terabyte folder paid the same
full content read the first one did. IndexCache itself carried no
group_id in its schema; only daemon.py's wiring did. Now one instance,
opened once at startup (data_dir/index_cache.db), shared by every group's
DirectoryIndexer.

Confirmed against a real deployment (2026-08-25/26): a group sharing an
already-indexed folder with an existing group indexes it instantly, with
zero rehashing.

Also fixes a related cross-group correctness gap found during this work:
media_cache.db (thumbnails, TMDB/MusicBrainz metadata — already node-wide,
untouched by this change) was pruned for a file the moment it left *one*
group's index, even if another group's index still held the same content
hash — forcing a redundant re-fetch/re-probe/re-thumbnail for a group that
never actually lost anything. Prune now runs only once no group's index
references the file_id any more.

Adds a node admin UI action ("Maintenance" card, prune-index-cache) to
drop cache rows that no longer belong to any group's roots — skips
anything under a root that is merely temporarily unavailable (indexer.py's
"a root that goes away freezes, never empties" rule extends to this
cache too, or a reconnected drive would pay a full rehash for no reason).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013XSohfUQQiaE77qyFLgSv3

4 files changed, 187 insertions, 26 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index cb3626b..74ab7b1 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -147,7 +147,9 @@ 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] = {}
+        # One instance, shared by every group's DirectoryIndexer — see
+        # indexer/cache.py's docstring for why this stopped being per-group.
+        self._index_cache: IndexCache | None = None
         # 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
@@ -245,6 +247,13 @@ class NodeDaemon:
             await self._bundle_store.open()
             log.info("Bundle store opened: %s", data_dir / "bundles.db")
 
+            # 4a. Path->hash cache, node-wide — opened once, shared by every
+            # group's DirectoryIndexer below (indexer/cache.py).
+            self._index_cache = IndexCache(db_path=data_dir / "index_cache.db")
+            await self._index_cache.open()
+            self._state["index_cache"] = self._index_cache
+            log.info("Index cache opened: %s", data_dir / "index_cache.db")
+
             # 4b. Roster — who this node recognises and which keys are theirs.
             # Node authority is established here, locally, and never learned from
             # the hub: a hub that could name the operator's key could install
@@ -296,11 +305,6 @@ 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
@@ -318,7 +322,7 @@ class NodeDaemon:
                     sk_node=keys.sk_ed25519,
                     gek=gek,
                     on_change=self._on_index_change,
-                    cache=index_cache,
+                    cache=self._index_cache,
                     reconcile_secs=scan_settings["reconcile_interval_secs"],
                     debounce_secs=scan_settings["debounce_secs"],
                 )
@@ -733,11 +737,6 @@ 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 {
@@ -751,7 +750,7 @@ class NodeDaemon:
                 sk_node=sk_ed,
                 gek=gek,
                 on_change=self._on_index_change,
-                cache=index_cache,
+                cache=self._index_cache,
                 reconcile_secs=scan_settings["reconcile_interval_secs"],
                 debounce_secs=scan_settings["debounce_secs"],
             )
@@ -836,12 +835,10 @@ class NodeDaemon:
                     await store.close()
                 except Exception:
                     pass
-            cache = self._index_caches.pop(gid, None)
-            if cache:
-                try:
-                    await cache.close()
-                except Exception:
-                    pass
+            # No per-group index cache to close here (2026-08-25): the
+            # (path, size, mtime) -> hash cache is now one shared instance,
+            # open for the life of the daemon, since another group may still
+            # reference the same physical folder — see indexer/cache.py.
             pending = self._pending_broadcasts.pop(gid, None)
             if pending:
                 pending.cancel()
@@ -1080,8 +1077,19 @@ class NodeDaemon:
         # that is content-addressed and simply moved effectively is.
         if delta is not None and delta.deletions and self._media_cache:
             for file_id in delta.deletions:
-                asyncio.ensure_future(self._media_cache.prune_file(file_id))
                 self._enriched_attempted.discard((indexer.group_id, file_id))
+                # media_cache.db is node-wide, keyed by content hash — a file
+                # shared into two groups is one row there, same reasoning as
+                # `_enriched_attempted`'s own docstring above. This group's
+                # copy is genuinely gone (that's what a deletion delta is),
+                # but another group may still hold the same content: only
+                # prune once *no* group's index has this file_id any more,
+                # or the surviving group pays for a redundant re-fetch/
+                # re-probe/re-thumbnail for content it never actually lost.
+                still_referenced = any(
+                    i.index.get_entry(file_id) is not None for i in self._indexers)
+                if not still_referenced:
+                    asyncio.ensure_future(self._media_cache.prune_file(file_id))
 
         # 11.5 — Push to connected WebRTC peers in this group
         if self._webrtc:
@@ -1426,8 +1434,8 @@ class NodeDaemon:
         for store in self._chat_stores.values():
             await store.close()
 
-        for cache in self._index_caches.values():
-            await cache.close()
+        if self._index_cache:
+            await self._index_cache.close()
 
         for indexer in self._indexers:
             await indexer.stop()
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/cache.py b/packages/meshbay-node/src/meshbay_node/indexer/cache.py
index c26ddf1..31b9b15 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/cache.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/cache.py
@@ -1,5 +1,5 @@
 """
-MeshBay Node — persistent (path, size, mtime) -> hash cache, one per group.
+MeshBay Node — persistent (path, size, mtime) -> hash cache, one per node.
 
 Without this, every node restart re-reads and re-hashes every file in every
 root, even when nothing changed — measured at 23 minutes for a 114 GB library
@@ -10,6 +10,16 @@ It is a path-keyed accelerator only. The GroupIndex itself stays keyed by
 content hash (see indexer.py's note on why two identical files are one
 entry) — this cache never changes that, it only avoids recomputing a hash
 that has not changed.
+
+**Shared by every group's DirectoryIndexer, one instance, one open
+connection** (2026-08-25) — an operator very often shares the same physical
+folder (a music library, a Séries drive) into more than one group, and a
+cache keyed purely by absolute path has no reason to care which group asked.
+It used to be opened once per group (`data_dir/{group_id}/index_cache.db`),
+which meant the second group to reference an already-fully-hashed multi-
+terabyte folder paid the same full read the first one did — exactly the cost
+this cache exists to avoid. The schema carries no group_id and never has;
+only daemon.py's wiring changed.
 """
 
 import logging
@@ -40,7 +50,7 @@ class CachedEntry:
 
 
 class IndexCache:
-    """Async SQLite (path, size, mtime) -> hash cache for one group."""
+    """Async SQLite (path, size, mtime) -> hash cache, shared node-wide."""
 
     def __init__(self, db_path: Path):
         self._db_path = db_path
@@ -93,3 +103,31 @@ class IndexCache:
             "type = excluded.type, added_at = excluded.added_at",
             (path, mtime, size, hash, type, added_at))
         await self._db.commit()
+
+    # ── Maintenance (node admin UI "prune index cache") ──────────────────────
+
+    async def count(self) -> int:
+        """Cheap — used for the dashboard stat, never for the prune decision
+        itself (that needs the actual paths, see all_paths)."""
+        async with self._db.execute("SELECT COUNT(*) FROM files") as cur:
+            row = await cur.fetchone()
+        return row[0] if row else 0
+
+    async def all_paths(self) -> list[str]:
+        """Every cached path, for a caller that decides staleness itself —
+        this cache has no notion of which paths are still claimed by a
+        group's roots, on purpose (see ops.prune_index_cache)."""
+        async with self._db.execute("SELECT path FROM files") as cur:
+            rows = await cur.fetchall()
+        return [row[0] for row in rows]
+
+    async def remove_many(self, paths: list[str]) -> None:
+        """Drop rows outright — used only for paths a caller has already
+        decided are gone for good. Losing one costs a rehash next time that
+        path is scanned, never a wrong answer (lookup() always re-validates
+        against a live stat())."""
+        if not paths:
+            return
+        await self._db.executemany(
+            "DELETE FROM files WHERE path = ?", [(p,) for p in paths])
+        await self._db.commit()
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index c9375b3..c880c75 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -922,6 +922,66 @@ async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs:
             "debounce_secs": debounce_secs, "group_id": group_id}
 
 
+# ── Index cache maintenance ───────────────────────────────────────────────────
+#
+# The (path, size, mtime) -> hash accelerator (indexer/cache.py) is node-wide
+# and grows for as long as a path was ever seen — a folder an operator later
+# stops sharing (root removed, or every group hosting it is deleted) leaves
+# its rows behind forever otherwise. Nothing about correctness needs this:
+# a stale row just sits unused (lookup() keys on the live path string, so a
+# path nothing scans any more is never looked up). This is disk space
+# hygiene the operator can run when they want it, not a background job.
+
+async def index_cache_stats(state: dict) -> dict:
+    """Row count only — cheap, safe to call on every dashboard render.
+    The actual staleness check (prune_index_cache) is not this cheap and
+    must never run implicitly."""
+    cache = state.get("index_cache")
+    return {"count": await cache.count() if cache else 0}
+
+
+async def prune_index_cache(state: dict) -> dict:
+    """
+    Drop cache rows that cannot be right for anything any more: the path is
+    not under any group's root at all, or it is under a root that is
+    available right now and the file is genuinely gone from disk.
+
+    Deliberately leaves alone anything under a root that is currently
+    *unavailable* (a disconnected drive) — indexer.py's own rule is that
+    such a root freezes rather than empties, precisely so it does not pay a
+    full rehash the moment it comes back. Pruning through an unavailable
+    root here would reintroduce exactly that cost via a different door, so
+    an owning-but-unavailable root wins over "the file isn't there right
+    now" every time, unconditionally.
+
+    A row lost here costs one rehash the next time that path is scanned,
+    never a wrong answer: lookup() (cache.py) always re-validates size and
+    mtime against a live stat() before trusting a cached hash.
+    """
+    cache = state.get("index_cache")
+    if cache is None:
+        raise OpError("No index cache in this process", status=503)
+
+    indexers = list((state.get("indexers") or {}).values())
+    roots = [root for indexer in indexers for root in indexer.roots]
+
+    def _is_stale(path_str: str) -> bool:
+        path = Path(path_str)
+        owning = [r for r in roots if r.path in path.parents]
+        if not owning:
+            return True
+        if any(not r.available for r in owning):
+            return False
+        return not path.exists()
+
+    paths = await cache.all_paths()
+    stale = await asyncio.to_thread(lambda: [p for p in paths if _is_stale(p)])
+    await cache.remove_many(stale)
+    log.info("Index cache pruned: %d stale row(s) removed, %d kept",
+             len(stale), len(paths) - len(stale))
+    return {"status": "pruned", "removed": len(stale), "kept": len(paths) - len(stale)}
+
+
 # ── Reload ──────────────────────────────────────────────────────────────────
 
 async def reload_config(state: dict) -> dict:
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index 18764e8..3dc320b 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -169,6 +169,14 @@ def create_ui_app(state: dict) -> FastAPI:
     async def api_denylist_clear(subject: str = ""):
         return await _op(lambda: ops.clear_denylist(state, subject=subject))
 
+    @app.get("/api/index-cache")
+    async def api_index_cache_stats():
+        return await _op(lambda: ops.index_cache_stats(state))
+
+    @app.post("/api/index-cache/prune")
+    async def api_index_cache_prune():
+        return await _op(lambda: ops.prune_index_cache(state))
+
     @app.get("/api/groups/{group_id}/files")
     async def api_group_files(group_id: str):
         groups_ctx = state.get("groups_ctx", {})
@@ -458,7 +466,9 @@ def create_ui_app(state: dict) -> FastAPI:
                 "members": await roster.list_members(),
                 "invites": await roster.list_invites(),
             }
-        return _render_page(state, roster_view)
+        index_cache = state.get("index_cache")
+        cache_count = await index_cache.count() if index_cache else None
+        return _render_page(state, roster_view, cache_count)
 
     @app.get("/audit", response_class=HTMLResponse)
     async def audit_page():
@@ -540,7 +550,8 @@ def _render_roster(roster_view: dict | None) -> str:
     </p>"""
 
 
-def _render_page(state: dict, roster_view: dict | None = None) -> str:
+def _render_page(state: dict, roster_view: dict | None = None,
+                 index_cache_count: int | None = None) -> str:
     token_js = json.dumps(state.get("ui_token", ""))
     status = state.get("status", "starting")
     indexes = state.get("indexes", {})
@@ -723,6 +734,26 @@ def _render_page(state: dict, roster_view: dict | None = None) -> str:
     <p><b>Node ID:</b> <code>{state.get("endpoint_hint") or "—"}</code></p>
   </div>
 
+  <h2>Maintenance</h2>
+  <div class="card">
+    <p><b>Index cache:</b> <span id="cacheCount">{
+      index_cache_count if index_cache_count is not None else "—"
+    }</span> path(s) remembered (size/mtime → hash, shared by every group)</p>
+    <p class="muted">Removes rows whose path no longer belongs to any group's
+       root, or whose file is genuinely gone from a root that is currently
+       reachable. Never touches a root that is temporarily unavailable
+       (unplugged drive) — that one still needs its full cache back the
+       moment it returns.</p>
+    <div style="margin:10px 0">
+      <button onclick="pruneIndexCache()" id="prune-cache-btn"
+        style="padding:8px 16px;background:#3b82f6;color:#fff;border:none;
+        border-radius:6px;cursor:pointer;font-size:0.85em">
+        Prune stale entries
+      </button>
+      <span id="prune-cache-status" class="muted" style="margin-left:8px"></span>
+    </div>
+  </div>
+
   <h2>Link Node to Hub Account</h2>
   <div class="card">
     <p>To connect to your group from a browser, link this node to your hub account.
@@ -774,6 +805,30 @@ async function initGEK(groupId) {{
     if (btn) btn.disabled = false;
   }}
 }}
+async function pruneIndexCache() {{
+  const btn = document.getElementById('prune-cache-btn');
+  const status = document.getElementById('prune-cache-status');
+  if (btn) btn.disabled = true;
+  if (status) {{ status.textContent = 'Pruning...'; status.style.color = ''; }}
+  try {{
+    const resp = await fetch('/api/index-cache/prune?t=' + TOKEN, {{ method: 'POST' }});
+    const data = await resp.json();
+    if (resp.ok) {{
+      if (status) status.textContent = 'Removed ' + data.removed + ', kept ' + data.kept;
+      if (status) status.style.color = '#22c55e';
+      const count = document.getElementById('cacheCount');
+      if (count) count.textContent = data.kept;
+    }} else {{
+      if (status) status.textContent = data.error || 'Failed';
+      if (status) status.style.color = '#ef4444';
+    }}
+  }} catch (e) {{
+    if (status) status.textContent = 'Error: ' + e.message;
+    if (status) status.style.color = '#ef4444';
+  }} finally {{
+    if (btn) btn.disabled = false;
+  }}
+}}
 setTimeout(()=>location.reload(), 10000);
 </script>
 </body>