Merge branch 'feat/cross-group-file-cache'

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

1 files changed, 30 insertions, 22 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()