aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node
diff options
context:
space:
mode:
Diffstat (limited to 'packages/meshbay-node')
-rw-r--r--packages/meshbay-node/src/meshbay_node/daemon.py52
-rw-r--r--packages/meshbay-node/src/meshbay_node/indexer/cache.py42
-rw-r--r--packages/meshbay-node/src/meshbay_node/ops.py60
-rw-r--r--packages/meshbay-node/src/meshbay_node/ui/app.py59
-rw-r--r--packages/meshbay-node/tests/test_daemon.py65
-rw-r--r--packages/meshbay-node/tests/test_index_cache.py46
-rw-r--r--packages/meshbay-node/tests/test_indexer.py38
-rw-r--r--packages/meshbay-node/tests/test_ops_index_cache.py139
8 files changed, 475 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>
diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py
index b367a20..71aae78 100644
--- a/packages/meshbay-node/tests/test_daemon.py
+++ b/packages/meshbay-node/tests/test_daemon.py
@@ -440,6 +440,71 @@ async def test_delta_reflects_additions_and_deletions(tmp_path, shared_dir, gek)
@pytest.mark.asyncio
+async def test_media_cache_not_pruned_when_another_group_still_has_the_content(
+ tmp_path, shared_dir, gek):
+ """
+ media_cache.db is node-wide, keyed by content hash — a file shared into
+ two groups is one row there. Removing it from ONE group's index (root
+ unshared, group left) must not wipe the thumbnail/tmdb/mbid mapping the
+ OTHER group's copy still needs, or that surviving group pays for a
+ redundant re-fetch/re-probe/re-thumbnail for content it never lost.
+ """
+ daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32)
+ daemon._media_cache = AsyncMock()
+ daemon._webrtc = MagicMock()
+ daemon._webrtc._sessions = {}
+
+ indexer_a = DirectoryIndexer(roots=one_root(shared_dir), group_id="a" * 32,
+ sk_node=Ed25519PrivateKey.generate(), gek=gek)
+ await indexer_a.initial_scan()
+ shared_id = indexer_a.index.entries[0].id
+
+ indexer_b = DirectoryIndexer(roots=one_root(shared_dir), group_id="b" * 32,
+ sk_node=Ed25519PrivateKey.generate(), gek=gek)
+ await indexer_b.initial_scan()
+ assert indexer_b.index.get_entry(shared_id) is not None
+
+ daemon._indexers = [indexer_a, indexer_b]
+
+ await daemon._on_index_change(indexer_a) # establishes the snapshot
+ await asyncio.sleep(0.05)
+
+ indexer_a.index.remove_entry(shared_id)
+ await daemon._on_index_change(indexer_a)
+ await asyncio.sleep(0.05)
+
+ daemon._media_cache.prune_file.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_media_cache_pruned_once_no_group_has_the_content_left(
+ tmp_path, shared_dir, gek):
+ """Counterpart of the test above: with only one group ever having held
+ the content, its removal must still prune media_cache as before — the
+ fix only withholds pruning when the content genuinely survives
+ elsewhere, it must not make pruning stop happening altogether."""
+ daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32)
+ daemon._media_cache = AsyncMock()
+ daemon._webrtc = MagicMock()
+ daemon._webrtc._sessions = {}
+
+ indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="a" * 32,
+ sk_node=Ed25519PrivateKey.generate(), gek=gek)
+ await indexer.initial_scan()
+ removed_id = indexer.index.entries[0].id
+ daemon._indexers = [indexer]
+
+ await daemon._on_index_change(indexer)
+ await asyncio.sleep(0.05)
+
+ indexer.index.remove_entry(removed_id)
+ await daemon._on_index_change(indexer)
+ await asyncio.sleep(0.05)
+
+ daemon._media_cache.prune_file.assert_called_once_with(removed_id)
+
+
+@pytest.mark.asyncio
async def test_a_burst_of_changes_produces_one_broadcast(tmp_path, shared_dir, gek):
"""Coalescing: several _on_index_change calls in quick succession (one
per debounced watchdog event) must collapse into a single push."""
diff --git a/packages/meshbay-node/tests/test_index_cache.py b/packages/meshbay-node/tests/test_index_cache.py
index db0e37e..24e2f76 100644
--- a/packages/meshbay-node/tests/test_index_cache.py
+++ b/packages/meshbay-node/tests/test_index_cache.py
@@ -60,6 +60,52 @@ async def test_put_overwrites_previous_row_for_same_path(cache):
@pytest.mark.asyncio
+async def test_count_reflects_number_of_rows(cache):
+ assert await cache.count() == 0
+
+ await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="a",
+ type="video", added_at=1)
+ await cache.put("/lib/b.mkv", size=2000, mtime=222.0, hash="b",
+ type="video", added_at=2)
+
+ assert await cache.count() == 2
+
+
+@pytest.mark.asyncio
+async def test_all_paths_returns_every_row(cache):
+ await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="a",
+ type="video", added_at=1)
+ await cache.put("/lib/b.mkv", size=2000, mtime=222.0, hash="b",
+ type="video", added_at=2)
+
+ assert set(await cache.all_paths()) == {"/lib/a.mkv", "/lib/b.mkv"}
+
+
+@pytest.mark.asyncio
+async def test_remove_many_drops_only_the_given_paths(cache):
+ await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="a",
+ type="video", added_at=1)
+ await cache.put("/lib/b.mkv", size=2000, mtime=222.0, hash="b",
+ type="video", added_at=2)
+
+ await cache.remove_many(["/lib/a.mkv"])
+
+ assert await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) is None
+ assert await cache.lookup("/lib/b.mkv", size=2000, mtime=222.0) is not None
+ assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_remove_many_with_empty_list_is_a_no_op(cache):
+ await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="a",
+ type="video", added_at=1)
+
+ await cache.remove_many([])
+
+ assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
async def test_cache_survives_reopen(tmp_path):
db_path = tmp_path / "index_cache.db"
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index 9aa9fb9..729dade 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -284,6 +284,44 @@ async def test_second_scan_with_same_cache_hashes_nothing(
@pytest.mark.asyncio
+async def test_second_group_sharing_the_same_folder_hashes_nothing(
+ shared_dir, sk_node, gek, index_cache):
+ """
+ The scenario the cache was made node-wide for (2026-08-25): an operator
+ shares the same physical folder into a second group. `IndexCache` is
+ keyed purely by absolute path, with no notion of group_id at all — a
+ *different* group_id scanning the same folder through the same shared
+ cache instance must hit exactly as hard as a same-group restart does
+ (the test right above this one). Before this cache was shared node-wide,
+ each group got its own on-disk cache file and this scan would have
+ rehashed every byte again.
+ """
+ first = DirectoryIndexer(roots=one_root(shared_dir), group_id="group-a",
+ sk_node=sk_node, gek=gek, cache=index_cache)
+ await first.initial_scan()
+ assert first.index.count == 4
+
+ calls = []
+ real_scan_file = indexer_mod._scan_file
+
+ def spy(root, path):
+ calls.append(path)
+ return real_scan_file(root, path)
+
+ indexer_mod._scan_file = spy
+ try:
+ second = DirectoryIndexer(roots=one_root(shared_dir), group_id="group-b",
+ sk_node=sk_node, gek=gek, cache=index_cache)
+ await second.initial_scan()
+ finally:
+ indexer_mod._scan_file = real_scan_file
+
+ assert calls == [], (
+ f"a second group scanning the same folder must not rehash it, got {calls}")
+ assert {e.id for e in second.index.entries} == {e.id for e in first.index.entries}
+
+
+@pytest.mark.asyncio
async def test_modified_file_is_rehashed(tmp_path, sk_node, gek, index_cache):
d = tmp_path / "shared"
d.mkdir()
diff --git a/packages/meshbay-node/tests/test_ops_index_cache.py b/packages/meshbay-node/tests/test_ops_index_cache.py
new file mode 100644
index 0000000..0b28d69
--- /dev/null
+++ b/packages/meshbay-node/tests/test_ops_index_cache.py
@@ -0,0 +1,139 @@
+"""Tests for ops.index_cache_stats / ops.prune_index_cache."""
+
+import types
+
+import pytest
+from meshbay_node import ops
+from meshbay_node.indexer.cache import IndexCache
+from meshbay_node.roots import Root
+
+
+@pytest.fixture
+async def cache(tmp_path):
+ c = IndexCache(db_path=tmp_path / "index_cache.db")
+ await c.open()
+ yield c
+ await c.close()
+
+
+def _indexer(roots):
+ """A stand-in for DirectoryIndexer — prune_index_cache only ever reads
+ `.roots`, so a real one (with its watchdog Observer, executor, etc.)
+ would be pure overhead here."""
+ return types.SimpleNamespace(roots=roots)
+
+
+@pytest.mark.asyncio
+async def test_stats_with_no_cache_reports_zero():
+ assert await ops.index_cache_stats({}) == {"count": 0}
+
+
+@pytest.mark.asyncio
+async def test_stats_reports_row_count(cache):
+ await cache.put("/lib/a.mkv", size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+ assert await ops.index_cache_stats({"index_cache": cache}) == {"count": 1}
+
+
+@pytest.mark.asyncio
+async def test_prune_without_a_cache_raises():
+ with pytest.raises(ops.OpError):
+ await ops.prune_index_cache({})
+
+
+@pytest.mark.asyncio
+async def test_prune_removes_a_path_no_group_claims_any_more(cache):
+ """A folder no group's roots reference at all any more — every group
+ that once shared it was deleted, or the root was removed everywhere."""
+ await cache.put("/gone/orphan.mkv", size=1, mtime=1.0, hash="a",
+ type="video", added_at=1)
+ state = {"index_cache": cache, "indexers": {}}
+
+ result = await ops.prune_index_cache(state)
+
+ assert result == {"status": "pruned", "removed": 1, "kept": 0}
+ assert await cache.count() == 0
+
+
+@pytest.mark.asyncio
+async def test_prune_keeps_a_path_whose_file_still_exists_under_an_available_root(
+ tmp_path, cache):
+ root_dir = tmp_path / "lib"
+ root_dir.mkdir()
+ f = root_dir / "still-here.mkv"
+ f.write_bytes(b"x")
+ await cache.put(str(f), size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+ root = Root(name="lib", path=root_dir, available=True)
+ state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+ result = await ops.prune_index_cache(state)
+
+ assert result == {"status": "pruned", "removed": 0, "kept": 1}
+ assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_prune_removes_a_path_deleted_from_an_available_root(tmp_path, cache):
+ root_dir = tmp_path / "lib"
+ root_dir.mkdir()
+ gone = root_dir / "deleted.mkv"
+ # Never written to disk — simulates a file that was there when cached
+ # and has since been deleted.
+ await cache.put(str(gone), size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+ root = Root(name="lib", path=root_dir, available=True)
+ state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+ result = await ops.prune_index_cache(state)
+
+ assert result == {"status": "pruned", "removed": 1, "kept": 0}
+ assert await cache.count() == 0
+
+
+@pytest.mark.asyncio
+async def test_prune_never_touches_a_path_under_an_unavailable_root(tmp_path, cache):
+ """
+ indexer.py's own rule: a root that goes away freezes, it never empties.
+ A disconnected drive must not have its cache wiped just because the
+ files can't be verified right now — that would force a full rehash of
+ the whole drive the moment it comes back, exactly the cost this cache
+ exists to avoid. Checked with a root directory that does not even exist
+ on disk (the strongest form of "can't verify") to prove availability,
+ not on-disk state, is what decides this.
+ """
+ root_dir = tmp_path / "usb" # never created — simulates "unplugged"
+ unreachable = root_dir / "movie.mkv"
+ await cache.put(str(unreachable), size=1, mtime=1.0, hash="a",
+ type="video", added_at=1)
+
+ root = Root(name="usb", path=root_dir, available=False)
+ state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+ result = await ops.prune_index_cache(state)
+
+ assert result == {"status": "pruned", "removed": 0, "kept": 1}
+ assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_prune_across_two_groups_sharing_one_root_path(tmp_path, cache):
+ """The scenario this whole cache redesign is for: the same folder is a
+ root of two groups. A path under it must be kept as long as *either*
+ group's root still resolves it, available."""
+ root_dir = tmp_path / "shared"
+ root_dir.mkdir()
+ f = root_dir / "track.mp3"
+ f.write_bytes(b"x")
+ await cache.put(str(f), size=1, mtime=1.0, hash="a", type="audio", added_at=1)
+
+ root_a = Root(name="shared", path=root_dir, available=True)
+ root_b = Root(name="shared", path=root_dir, available=True)
+ state = {
+ "index_cache": cache,
+ "indexers": {"g1": _indexer([root_a]), "g2": _indexer([root_b])},
+ }
+
+ result = await ops.prune_index_cache(state)
+
+ assert result == {"status": "pruned", "removed": 0, "kept": 1}
generated by cgit v1.2.3 (git 2.51.0) at 2026-09-25 18:07:09 +0000