""" 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 on a USB hard drive. This cache lets a scan skip the read entirely for a file whose size and mtime still match what was hashed last time. 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 import time from dataclasses import dataclass from pathlib import Path import aiosqlite log = logging.getLogger(__name__) _SCHEMA = """ CREATE TABLE IF NOT EXISTS files ( path TEXT PRIMARY KEY, mtime REAL NOT NULL, size INTEGER NOT NULL, hash TEXT NOT NULL, type TEXT NOT NULL, added_at INTEGER NOT NULL, hash_version INTEGER NOT NULL DEFAULT 1 ); -- Who sent a file. Written when an upload finishes, read when the file is -- indexed — two different moments, and the second is much later: the watchdog -- debounces for two seconds and then hashes, so the entry does not exist yet -- when the last chunk lands. Recording it in memory would also lose it at every -- restart, where the index is rebuilt from disk, and an owner the node forgets -- is an owner who cannot delete their own file tomorrow. -- -- Validated against a live stat() exactly as `files` is: a row whose size or -- mtime no longer match is a different file at that path, and attributes -- nothing. That is what makes a row left behind by a deleted file harmless. CREATE TABLE IF NOT EXISTS uploads ( path TEXT PRIMARY KEY, size INTEGER NOT NULL, mtime REAL NOT NULL, user_id TEXT NOT NULL, pk_ed25519 TEXT NOT NULL DEFAULT '', at INTEGER NOT NULL ); """ _MIGRATE_V2 = "ALTER TABLE files ADD COLUMN hash_version INTEGER NOT NULL DEFAULT 1" @dataclass class CachedEntry: hash: str type: str added_at: int hash_version: int = 1 class IndexCache: """Async SQLite (path, size, mtime) -> hash cache, shared node-wide.""" def __init__(self, db_path: Path): self._db_path = db_path self._db: aiosqlite.Connection | None = None # Whether `uploads` holds anything at all. Every entry the indexer # builds asks this cache who sent the file, and on a node that has # never received an upload — most of them, most of the time — that is # one query per file per scan for an answer that is always None. self._has_uploads = False async def open(self) -> None: self._db_path.parent.mkdir(parents=True, exist_ok=True) self._db = await aiosqlite.connect(str(self._db_path)) await self._db.executescript(_SCHEMA) try: await self._db.execute(_MIGRATE_V2) except Exception: pass # column already exists await self._db.commit() async with self._db.execute( "SELECT EXISTS(SELECT 1 FROM uploads)") as cur: row = await cur.fetchone() self._has_uploads = bool(row and row[0]) async def close(self) -> None: if self._db: await self._db.close() self._db = None async def __aenter__(self): await self.open() return self async def __aexit__(self, *_): await self.close() async def lookup(self, path: str, size: int, mtime: float, hash_version: int = 1) -> CachedEntry | None: """ A cache hit requires an EXACT match on size, mtime AND hash_version. A v1 cached hash won't serve a v2 lookup for the same path — the file is re-hashed with the new algorithm instead. """ async with self._db.execute( "SELECT hash, type, added_at, hash_version FROM files " "WHERE path = ? AND size = ? AND mtime = ? AND hash_version = ?", (path, size, mtime, hash_version)) as cur: row = await cur.fetchone() return CachedEntry(hash=row[0], type=row[1], added_at=row[2], hash_version=row[3]) if row else None async def put(self, path: str, size: int, mtime: float, hash: str, type: str, added_at: int, hash_version: int = 1) -> None: """ Written only once a file has been hashed — never partway through — so a crash mid-hash leaves no stale/partial row behind: the next scan simply finds no cache entry and hashes the file again. """ await self._db.execute( "INSERT INTO files (path, mtime, size, hash, type, added_at, hash_version) " "VALUES (?, ?, ?, ?, ?, ?, ?) " "ON CONFLICT(path) DO UPDATE SET " "mtime = excluded.mtime, size = excluded.size, hash = excluded.hash, " "type = excluded.type, added_at = excluded.added_at, " "hash_version = excluded.hash_version", (path, mtime, size, hash, type, added_at, hash_version)) await self._db.commit() # ── Who sent a file ────────────────────────────────────────────────────── async def record_upload(self, path: str, size: int, mtime: float, user_id: str, pk_ed25519: str) -> None: """Remember that this account put this file here. Written once the upload is complete and the file is at its final name, never partway through — a `.part` is not indexable and would key a row to a path that is about to change. """ await self._db.execute( "INSERT INTO uploads (path, size, mtime, user_id, pk_ed25519, at) " "VALUES (?, ?, ?, ?, ?, ?) " "ON CONFLICT(path) DO UPDATE SET " "size = excluded.size, mtime = excluded.mtime, " "user_id = excluded.user_id, pk_ed25519 = excluded.pk_ed25519, " "at = excluded.at", (path, size, mtime, user_id, pk_ed25519, int(time.time()))) await self._db.commit() self._has_uploads = True async def uploader(self, path: str, size: int, mtime: float) -> tuple[str, str] | None: """Who sent the file currently at this path, or None. The size and mtime are matched exactly, like `lookup`: the path alone would credit whoever last uploaded *a* file of that name for whatever occupies the name now — including something the operator put there themselves afterwards, which would hand a member the right to delete it. """ if not self._has_uploads: return None async with self._db.execute( "SELECT user_id, pk_ed25519 FROM uploads " "WHERE path = ? AND size = ? AND mtime = ?", (path, size, mtime)) as cur: row = await cur.fetchone() return (row[0], row[1]) if row else None # ── 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()