From 37d8d9c15c982f2da17b2fad4ea1a90613b560a6 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 26 Aug 2026 00:40:20 +0200 Subject: feat(node): share the (path,size,mtime)->hash index cache across every group MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_013XSohfUQQiaE77qyFLgSv3 --- packages/meshbay-node/src/meshbay_node/daemon.py | 52 +++++++++++-------- .../meshbay-node/src/meshbay_node/indexer/cache.py | 42 ++++++++++++++- packages/meshbay-node/src/meshbay_node/ops.py | 60 ++++++++++++++++++++++ packages/meshbay-node/src/meshbay_node/ui/app.py | 59 ++++++++++++++++++++- 4 files changed, 187 insertions(+), 26 deletions(-) (limited to 'packages/meshbay-node/src/meshbay_node') 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:

""" -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:

Node ID: {state.get("endpoint_hint") or "—"}

+

Maintenance

+
+

Index cache: { + index_cache_count if index_cache_count is not None else "—" + } path(s) remembered (size/mtime → hash, shared by every group)

+

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.

+
+ + +
+
+

Link Node to Hub Account

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); -- cgit v1.2.3