diff options
| -rw-r--r-- | docs/indexing-v2.md | 312 | ||||
| -rw-r--r-- | packages/meshbay-common/src/meshbay_common/protocol.py | 2 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/cache.py | 56 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 67 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_index_cache.py | 77 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_indexer.py | 145 |
6 files changed, 619 insertions, 40 deletions
diff --git a/docs/indexing-v2.md b/docs/indexing-v2.md new file mode 100644 index 0000000..6fc7990 --- /dev/null +++ b/docs/indexing-v2.md @@ -0,0 +1,312 @@ +# Indexing v2 — Partial-read hashing for large files + +> Status: **plan, not built.** Decision record and implementation checklist. + +--- + +## 0. Problem + +The current indexer reads every file in full to compute its blake3 content hash (`id`). +For a large media library (multi-terabyte, thousands of files), this means: + +- **Time**: initial indexing takes tens of minutes to hours. +- **Disk I/O**: every byte of every file is read, which wears SSDs and saturates spinning + drives for the entire duration. A USB hard drive serving a 4 TB library is pegged for + over an hour. +- **Blocking**: no connected peer receives a usable index until the full scan finishes. + +The hash exists for **content identity** (deduplication, cross-group search, file +requests). A 4 GB film does not need 4 GB of I/O to be identified with overwhelming +probability — 45 MB of well-chosen samples suffice. + +--- + +## 1. Design + +### 1.1 Hashing rules + +| File size | Method | `hash_version` | +|--------------------|-----------------------------------------------------|-----------------| +| <= 40 MB | Full read, blake3 of entire content (unchanged) | `1` | +| > 40 MB | Partial read, blake3 of 45 MB sampled (see below) | `2` | + +**Partial-read algorithm (hash_version 2):** + +Given a file of `S` bytes where `S > 40 MB`: + +1. Read the first **20 MB** (bytes `[0, 20 MB)`). +2. Append the last **20 MB** (bytes `[S - 20 MB, S)`). +3. Append **5 MB** starting at **50% of the file** (bytes `[S // 2, S // 2 + 5 MB)`). +4. Compute `blake3(concatenation of the three regions)`. + +The three regions may overlap for files just above 40 MB. This is fine — the concatenation +is deterministic for a given file, which is the only property that matters. + +**Why these offsets.** Head and tail catch container headers, trailers, and the common case +of files that differ only at one end (re-encoded, re-muxed, appended). The mid-sample +catches files that share a header and trailer but differ in content (same container, +different media stream). + +**Why 40 MB threshold.** Below 40 MB the partial read would sample the entire file anyway +(head + tail >= file size), so the full-read path is both simpler and produces the same +result. The boundary is inclusive: a 40 MB file is read in full. + +### 1.2 `hash_version` field + +A new field on `IndexEntry`: + +``` +hash_version: int = 1 +``` + +- `1` — the `id` is blake3 of the full file content. This is the only value any existing + node has ever produced. +- `2` — the `id` is blake3 of the 45 MB partial sample described above. + +**For files <= 40 MB on a v2 node, `hash_version` stays `1`.** The hash is identical to +what a v1 node produces, because both read the file in full. This preserves cross-group +search compatibility for small files across v1 and v2 nodes. + +**For files > 40 MB on a v2 node, `hash_version` is `2`.** The hash is different from +what a v1 node would produce for the same file. This is the accepted side effect. + +### 1.3 Backward compatibility + +| Scenario | Behaviour | +|---|---| +| v2 node sends `hash_version` to v1 client | Client ignores unknown field (JS objects are open) | +| v1 node sends entries without `hash_version` | Client/consumer treats it as `1` (dataclass default) | +| v2 `IndexEntry(**e)` where `e` lacks `hash_version` | Uses default `1` — existing serialized indexes deserialize correctly | +| Cross-group search: same file, one node v1, one node v2 | Different `id` for files > 40 MB — not merged. Accepted | +| Cross-group search: same small file, mixed nodes | Same `id` (both `hash_version=1`) — merged correctly | +| Hub tables (`swarm_sources`, `content_blocklist`, `content_reports`) | Store `content_hash` as an opaque string. No change needed | +| `GroupIndex.serialize()` / `deserialize()` | `asdict(e)` includes `hash_version`; `IndexEntry(**e)` with default handles missing field | + +**Nothing breaks.** A v1 node's data remains valid. A v2 node produces correct new hashes. +Mixed v1/v2 environments work, with the documented search side effect. + +--- + +## 2. Affected components + +### 2.1 `meshbay_common` — `protocol.py` + +| Change | Detail | +|---|---| +| `IndexEntry` dataclass | Add `hash_version: int = 1` field | +| `index_entry_wire()` | Add `"hash_version": e.hash_version` to the wire dict | + +### 2.2 `meshbay_node` — `indexer/indexer.py` + +| Change | Detail | +|---|---| +| Constants | `_PARTIAL_THRESHOLD = 40 * 1024 * 1024`, `_PARTIAL_HEAD = 20 * 1024 * 1024`, `_PARTIAL_TAIL = 20 * 1024 * 1024`, `_PARTIAL_MID = 5 * 1024 * 1024` | +| `_scan_file()` | After stat, if `size > _PARTIAL_THRESHOLD`: use partial-read blake3. Set `hash_version=2` on the returned `IndexEntry`. Otherwise: unchanged (full read, `hash_version=1`) | +| `_hash_or_cached()` | Compute expected `hash_version` from file size. Pass it to cache `lookup()`. Store it in cache `put()`. Set it on the returned `IndexEntry` | + +**`_scan_file` partial-read implementation:** + +```python +def _partial_hash(file_path: Path, size: int) -> str: + hasher = blake3.blake3() + with open(long_path(file_path), "rb") as f: + _feed(hasher, f, _PARTIAL_HEAD) + f.seek(size - _PARTIAL_TAIL) + _feed(hasher, f, _PARTIAL_TAIL) + f.seek(size // 2) + _feed(hasher, f, _PARTIAL_MID) + return hasher.hexdigest() + +def _feed(hasher, f, nbytes: int) -> None: + remaining = nbytes + while remaining > 0: + chunk = f.read(min(_HASH_CHUNK, remaining)) + if not chunk: + break + hasher.update(chunk) + remaining -= len(chunk) +``` + +### 2.3 `meshbay_node` — `indexer/cache.py` + +| Change | Detail | +|---|---| +| Schema | Add `hash_version INTEGER NOT NULL DEFAULT 1` column to `files` table | +| `_SCHEMA` | New databases get the column via `CREATE TABLE` | +| `_MIGRATE_V2` | `ALTER TABLE files ADD COLUMN hash_version INTEGER NOT NULL DEFAULT 1` — applied on `open()` if the column does not exist | +| `lookup()` | Add `hash_version` parameter. WHERE clause becomes `path = ? AND size = ? AND mtime = ? AND hash_version = ?` | +| `put()` | Add `hash_version` parameter. INSERT includes `hash_version` | +| `CachedEntry` | Add `hash_version: int` field | + +**Auto-migration on open:** `IndexCache.open()` runs the ALTER TABLE inside a try/except +(column already exists → no-op). This way a node upgrade just works — no manual step +needed for the cache. + +### 2.4 `meshbay_node` — `indexer/group_index.py` + +No code change needed. `serialize()` calls `asdict(e)` which includes `hash_version`. +`deserialize()` calls `IndexEntry(**e)` which uses the default `1` for entries written +before v2. + +### 2.5 `meshbay_node` — `transport/wire.py` + +No code change needed. `index_sync_message()` and `index_delta_message()` call +`index_entry_wire()` which is updated in §2.1. + +### 2.6 Client side (browser / desktop) + +**No code changes required.** Entries are JavaScript objects; the extra `hash_version` +field is carried through without needing explicit handling: + +- `transport.js` — `_applyIndexMessage()` passes the opened payload through. `hash_version` + rides along on each entry object. +- `group-page.js` — `applyIndex()` / `applyIndexDelta()` store entries as-is in state + and in IndexedDB. +- `hub-client.js` — IndexedDB cache stores entry objects verbatim. +- `source-merge.js` — merges by `id`. Different hashes naturally don't merge. +- `search-page.js` — aggregates entries from cached indexes. No change. +- `files-app.js` — renders entries. `hash_version` is ignored. + +### 2.7 Hub side + +**No code changes required.** `swarm_sources.content_hash`, `content_reports.content_hash`, +`content_blocklist.content_hash` are opaque `String(64)` columns. They store whatever +blake3 hex the node provides. No hub migration needed. + +### 2.8 Tests + +| Test | What it verifies | +|---|---| +| `test_indexer.py` — new cases | Partial hash for file > 40 MB produces `hash_version=2`. File <= 40 MB produces `hash_version=1`. Partial hash is deterministic. Partial hash differs from full hash for the same large file | +| `test_indexer.py` — existing cases | All existing tests still pass (small files, type detection, enrichment, etc.) | +| `test_index_cache.py` — new cases | Cache lookup with `hash_version` match. Cache miss when `hash_version` differs. Schema migration from v1 cache | +| `test_index_cache.py` — existing cases | Unchanged behaviour for v1 entries | +| `test_indexer.py` — round-trip | `GroupIndex.serialize()` → `deserialize()` preserves `hash_version` on entries | +| `test_index_seal_client.py` | Wire format includes `hash_version`, old entries without it deserialize as v1 | + +--- + +## 3. Migration + +### 3.1 What actually needs migrating + +The node has two relevant stores: + +| Store | Location | Content | Migration | +|---|---|---|---| +| `index_cache.db` | `data_dir/index_cache.db` | `(path, size, mtime) → hash` accelerator | Add `hash_version` column | +| In-memory `GroupIndex` | rebuilt from disk on every startup | Current file listing | No migration — rebuilt on next scan | + +**The IndexCache is the only persistent store that needs a schema change.** The GroupIndex +is rebuilt by scanning the filesystem on every daemon start. Once the code uses v2 hashing, +the next startup produces v2 hashes for large files automatically. + +**The cache auto-migrates.** `IndexCache.open()` adds the `hash_version` column if missing. +Existing rows get `DEFAULT 1`. When the v2 indexer looks up a large file with +`hash_version=2`, the cached v1 entry won't match (different hash_version in WHERE), so +the file is re-hashed with the partial algorithm and the new entry is written with +`hash_version=2`. + +This means: **large files are re-hashed lazily on first scan after upgrade.** The first +scan after upgrading to v2 re-reads 45 MB per large file instead of the full content — +already much faster than v1's full read. + +### 3.2 Migration script — `QE/migration/migrate_index_v2.sh` + +A standalone bash script (not in git — QE/ is gitignored) that: + +1. Detects the node's `data_dir` from `node.toml` (default `~/.local/share/meshbay-node/`) +2. Checks that `index_cache.db` exists +3. Runs `ALTER TABLE files ADD COLUMN hash_version INTEGER NOT NULL DEFAULT 1` +4. Optionally (`--purge-large`) deletes cache entries for files > 40 MB, forcing immediate + re-hash on next scan instead of lazy migration +5. Reports what it did + +**Works on Ubuntu and Fedora** — uses only `sqlite3` (present by default on both) and +standard bash. + +**Not strictly required** if the code's auto-migration in `IndexCache.open()` is +implemented. The script exists for operators who want to: +- Verify the schema change before restarting +- Force a clean re-hash of all large files in one pass +- Run the migration on a machine where the node isn't installed yet (preparing a data dir) + +### 3.3 What the operator does + +1. Update the node package (or `pip install -e` in dev) +2. (Optional) Run `QE/migration/migrate_index_v2.sh` to preview or force the migration +3. Restart the node daemon +4. The first scan re-hashes files > 40 MB with partial reads — much faster than before + +No client-side action needed. No hub-side action needed. + +--- + +## 4. What does NOT change + +- **Files <= 40 MB** — identical hashing, identical `id`, `hash_version=1`. +- **Enrichment** (thumbnails, duration, metadata) — unchanged, still runs after hashing. +- **File downloads** — `file_req` uses the current `id` from the index. After re-indexing, + clients get the new index with new hashes and request accordingly. +- **Watchdog / reconciliation** — filesystem events trigger the same code paths. New or + modified files are hashed with the appropriate method based on size. +- **Index encryption / signing** — `GroupIndex.serialize()` and the GEK-sealed wire + messages are unchanged. `hash_version` rides inside each entry via `asdict()`. +- **Index delta computation** — `GroupIndex.diff()` compares entries by all fields + (dataclass `__eq__`). A re-indexed file whose hash changed (v1 → v2) appears as a + deletion of the old id + addition of the new id, which is correct. +- **Hub tables** — opaque content_hash storage, untouched. +- **MNP version** — this is an additive field on index entries. No protocol version bump + needed. An older node that doesn't send `hash_version` is handled by the default. + +--- + +## 5. Side effects — documented and accepted + +1. **Cross-group search**: a file > 40 MB indexed on a v1 node and a v2 node produces + different `id`s. The search page will not merge them as the same file. The user sees + two entries instead of one, each from its own group. This resolves itself when both + nodes upgrade. + +2. **First scan after upgrade**: files > 40 MB are re-hashed. With v2 this reads 45 MB + per file (not the full content), so the re-index is fast — but it is not instant. A + 4 TB library with 1000 large files reads ~44 GB instead of 4 TB. + +3. **`content_hash` drift on hub tables**: if a public group's node upgrades, the hashes + it registers in `swarm_sources` change for large files. Old entries with v1 hashes + become stale. The swarm registration mechanism's `last_seen` update handles this — stale + entries age out. No explicit cleanup needed. + +4. **Index version bump**: re-indexing sets `GroupIndex.version = int(time.time())`, + which triggers a full `index_sync` to all connected peers. This is the normal path for + any index change — it is not new load. + +--- + +## 6. Indexing trigger points — verified safe + +| Trigger | Location | Impact of v2 | +|---|---|---| +| Daemon startup — initial scan | `daemon.py:648-674` | Uses `_hash_or_cached()` which applies v2 rules. Safe | +| Create Group wizard — Step 3 | `create-group-page.js:244-248` polls `index-status` | No change — polls progress, doesn't control hashing | +| Settings — add root | `group-settings.js:909-927` | Triggers `retarget()` → scan → `_hash_or_cached()`. Safe | +| Watchdog — file created/modified | `indexer.py:797-816` | Calls `_update_entry()` → `_hash_or_cached()`. Safe | +| Reconciliation loop | `indexer.py:513-650` | Calls `_sweep_available_roots()` → `_hash_or_cached()`. Safe | +| Hot reload — `_reload_config()` | `daemon.py:790-829` | Creates new `DirectoryIndexer` with v2 code. Safe | + +--- + +## 7. Implementation order + +1. **`protocol.py`** — add `hash_version` field to `IndexEntry` and `index_entry_wire()` +2. **`cache.py`** — add `hash_version` to schema, auto-migrate on open, update lookup/put +3. **`indexer.py`** — implement `_partial_hash()`, update `_scan_file()` and + `_hash_or_cached()` +4. **Tests** — new test cases for partial hashing, cache versioning, wire round-trip +5. **`QE/migration/migrate_index_v2.sh`** — standalone migration script +6. **Manual test** — run a node with a mixed library (small + large files), verify: + - Small files: same hash as before, `hash_version=1` + - Large files: different hash, `hash_version=2`, 45 MB read + - Cross-group search: small files merge, large files don't (across v1/v2 nodes) + - Cache hit on second scan: no re-read + - Index sync to connected peers: entries carry `hash_version` diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 4d2dd9d..e502379 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -204,6 +204,7 @@ class IndexEntry: track_no: int | None = None # tag or parsed, Music app taken_at: int | None = None # unix timestamp, EXIF DateTimeOriginal — Photos app camera: str | None = None # "Make Model", when both present — Photos app + hash_version: int = 1 # 1 = full-file blake3, 2 = partial-read (45 MB sample) def index_entry_wire(e: IndexEntry) -> dict: @@ -224,6 +225,7 @@ def index_entry_wire(e: IndexEntry) -> dict: "season": e.season, "episode": e.episode, "artist": e.artist, "album": e.album, "track_no": e.track_no, "taken_at": e.taken_at, "camera": e.camera, + "hash_version": e.hash_version, } diff --git a/packages/meshbay-node/src/meshbay_node/indexer/cache.py b/packages/meshbay-node/src/meshbay_node/indexer/cache.py index 31b9b15..c167db9 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/cache.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/cache.py @@ -32,21 +32,25 @@ 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 + 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 ); """ +_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: @@ -60,6 +64,10 @@ class IndexCache: 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 def close(self) -> None: @@ -74,34 +82,36 @@ class IndexCache: async def __aexit__(self, *_): await self.close() - async def lookup(self, path: str, size: int, mtime: float) -> CachedEntry | None: + async def lookup(self, path: str, size: int, mtime: float, + hash_version: int = 1) -> CachedEntry | None: """ - A cache hit requires an EXACT match on both size and mtime. A mtime - touched without a content change is a false negative (an unnecessary - rehash) — accepted, since the alternative (trusting a stale hash) is - a silent wrong answer instead of an occasional wasted read. + 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 FROM files " - "WHERE path = ? AND size = ? AND mtime = ?", - (path, size, mtime)) as cur: + "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]) if row else None + 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) -> None: + type: str, added_at: int, hash_version: int = 1) -> None: """ - Written only once a file has been hashed in full — 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. + 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) " - "VALUES (?, ?, ?, ?, ?, ?) " + "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", - (path, mtime, size, hash, type, added_at)) + "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() # ── Maintenance (node admin UI "prune index cache") ────────────────────── diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index f78465b..8376c23 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -94,6 +94,32 @@ def _is_indexable_size(path: Path, size: int) -> bool: _HASH_CHUNK = 8 * 1024 * 1024 # 8 MB streaming hash chunks +_PARTIAL_THRESHOLD = 40 * 1024 * 1024 # files above this use partial-read hashing +_PARTIAL_HEAD = 20 * 1024 * 1024 +_PARTIAL_TAIL = 20 * 1024 * 1024 +_PARTIAL_MID = 5 * 1024 * 1024 + + +def _feed(hasher, f, nbytes: int) -> None: + remaining = nbytes + while remaining > 0: + chunk = f.read(min(_HASH_CHUNK, remaining)) + if not chunk: + break + hasher.update(chunk) + remaining -= len(chunk) + + +def _partial_hash(file_path: Path, size: int) -> str: + hasher = blake3.blake3() + with open(long_path(file_path), "rb") as f: + _feed(hasher, f, _PARTIAL_HEAD) + f.seek(size - _PARTIAL_TAIL) + _feed(hasher, f, _PARTIAL_TAIL) + f.seek(size // 2) + _feed(hasher, f, _PARTIAL_MID) + return hasher.hexdigest() + @dataclass class IndexProgress: @@ -162,29 +188,33 @@ def _size_files(files: list[Path]) -> list[tuple[Path, int]]: def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: """Compute IndexEntry for a file. Blocking — run in executor. - Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.) - don't require loading the whole file into memory.""" + Files <= 40 MB are hashed in full (hash_version 1). Files > 40 MB use a + 45 MB partial read — first 20 MB, last 20 MB, 5 MB at 50% — for + hash_version 2.""" if not _is_indexable(file_path): return None try: stat = file_path.stat() if not _is_indexable_size(file_path, stat.st_size): return None - hasher = blake3.blake3() - # long_path is a no-op off Windows; there it is what lets a deep media - # library past MAX_PATH. - with open(long_path(file_path), "rb") as f: - while chunk := f.read(_HASH_CHUNK): - hasher.update(chunk) + if stat.st_size > _PARTIAL_THRESHOLD: + hex_hash = _partial_hash(file_path, stat.st_size) + hv = 2 + else: + hasher = blake3.blake3() + with open(long_path(file_path), "rb") as f: + while chunk := f.read(_HASH_CHUNK): + hasher.update(chunk) + hex_hash = hasher.hexdigest() + hv = 1 return IndexEntry( - id=hasher.hexdigest(), - # Stored exactly as the filesystem gave it: this is the string that - # opens the file. Normalization is for comparison only. + id=hex_hash, name=file_path.name, path=_virtual_dir(root, file_path), size=stat.st_size, type=_detect_type(file_path), added_at=int(stat.st_mtime), + hash_version=hv, ) except (OSError, PermissionError, ValueError) as e: log.warning("Cannot index %s: %s", file_path, e) @@ -375,10 +405,8 @@ class DirectoryIndexer: async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None: """ Cache-aware replacement for a bare _scan_file() call: skips the - content read entirely when this path's (size, mtime) still match - what was hashed last time — the difference between a redundant full - rehash of a 100+ GB library on every restart and a stat()-only pass. - The only place that decides to actually read a file's bytes. + content read entirely when this path's (size, mtime, hash_version) + still match what was hashed last time. """ if not _is_indexable(file_path): return None @@ -389,8 +417,11 @@ class DirectoryIndexer: if not _is_indexable_size(file_path, st.st_size): return None + expected_hv = 2 if st.st_size > _PARTIAL_THRESHOLD else 1 + if self._cache is not None: - cached = await self._cache.lookup(str(file_path), st.st_size, st.st_mtime) + cached = await self._cache.lookup( + str(file_path), st.st_size, st.st_mtime, expected_hv) if cached is not None: return IndexEntry( id=cached.hash, @@ -399,13 +430,15 @@ class DirectoryIndexer: size=st.st_size, type=cached.type, added_at=cached.added_at, + hash_version=cached.hash_version, ) loop = asyncio.get_event_loop() entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path) if entry and self._cache is not None: await self._cache.put(str(file_path), st.st_size, st.st_mtime, - entry.id, entry.type, entry.added_at) + entry.id, entry.type, entry.added_at, + entry.hash_version) return entry def _report_collisions(self) -> None: diff --git a/packages/meshbay-node/tests/test_index_cache.py b/packages/meshbay-node/tests/test_index_cache.py index 24e2f76..036b4d7 100644 --- a/packages/meshbay-node/tests/test_index_cache.py +++ b/packages/meshbay-node/tests/test_index_cache.py @@ -122,3 +122,80 @@ async def test_cache_survives_reopen(tmp_path): assert hit is not None assert hit.hash == "abc123" + + +# ── hash_version support (indexing v2) ───────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_put_v2_then_lookup_hits(cache): + await cache.put("/lib/big.mkv", size=50_000_000, mtime=111.0, + hash="partial_abc", type="video", added_at=42, + hash_version=2) + + hit = await cache.lookup("/lib/big.mkv", size=50_000_000, mtime=111.0, + hash_version=2) + assert hit is not None + assert hit.hash == "partial_abc" + assert hit.hash_version == 2 + + +@pytest.mark.asyncio +async def test_lookup_misses_on_wrong_hash_version(cache): + await cache.put("/lib/big.mkv", size=50_000_000, mtime=111.0, + hash="full_hash", type="video", added_at=42, + hash_version=1) + + assert await cache.lookup("/lib/big.mkv", size=50_000_000, mtime=111.0, + hash_version=2) is None + + +@pytest.mark.asyncio +async def test_put_v2_overwrites_v1_for_same_path(cache): + await cache.put("/lib/big.mkv", size=50_000_000, mtime=111.0, + hash="full_hash", type="video", added_at=42, + hash_version=1) + await cache.put("/lib/big.mkv", size=50_000_000, mtime=111.0, + hash="partial_hash", type="video", added_at=42, + hash_version=2) + + assert await cache.lookup("/lib/big.mkv", size=50_000_000, mtime=111.0, + hash_version=1) is None + hit = await cache.lookup("/lib/big.mkv", size=50_000_000, mtime=111.0, + hash_version=2) + assert hit is not None + assert hit.hash == "partial_hash" + + +@pytest.mark.asyncio +async def test_v1_schema_auto_migrates(tmp_path): + """An index_cache.db created by old code (no hash_version column) gains + the column on the next open(), and existing rows default to hash_version=1.""" + import aiosqlite + db_path = tmp_path / "old_cache.db" + async with aiosqlite.connect(str(db_path)) as db: + await db.executescript(""" + CREATE TABLE 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 + ); + """) + await db.execute( + "INSERT INTO files (path, mtime, size, hash, type, added_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + ("/lib/old.mkv", 111.0, 1000, "oldhash", "video", 42)) + await db.commit() + + cache = IndexCache(db_path=db_path) + await cache.open() + hit = await cache.lookup("/lib/old.mkv", size=1000, mtime=111.0, + hash_version=1) + await cache.close() + + assert hit is not None + assert hit.hash == "oldhash" + assert hit.hash_version == 1 diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index 729dade..6aee1b5 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -723,3 +723,148 @@ async def test_reconcile_backoff_resets_when_something_actually_changes( await task except asyncio.CancelledError: pass + + +# ── Indexing v2 — partial-read hashing ──────────────────────────────────────── + + +def test_small_file_gets_hash_version_1(tmp_path): + from meshbay_node.indexer.indexer import _scan_file + d = tmp_path / "root" + d.mkdir() + f = d / "small.mp4" + f.write_bytes(os.urandom(1024)) + + entry = _scan_file(one_root(d).roots[0], f) + assert entry is not None + assert entry.hash_version == 1 + + +def test_large_file_gets_hash_version_2(tmp_path): + from meshbay_node.indexer.indexer import _scan_file, _PARTIAL_THRESHOLD + d = tmp_path / "root" + d.mkdir() + f = d / "big.mkv" + size = _PARTIAL_THRESHOLD + 1 + f.write_bytes(os.urandom(size)) + + entry = _scan_file(one_root(d).roots[0], f) + assert entry is not None + assert entry.hash_version == 2 + assert entry.size == size + + +def test_file_at_threshold_gets_hash_version_1(tmp_path): + from meshbay_node.indexer.indexer import _scan_file, _PARTIAL_THRESHOLD + d = tmp_path / "root" + d.mkdir() + f = d / "exact.mkv" + f.write_bytes(os.urandom(_PARTIAL_THRESHOLD)) + + entry = _scan_file(one_root(d).roots[0], f) + assert entry is not None + assert entry.hash_version == 1 + + +def test_partial_hash_differs_from_full_hash(tmp_path): + """For a file above the threshold, the partial hash must differ from what + a full-file blake3 would produce (they read different bytes).""" + import blake3 as b3 + from meshbay_node.indexer.indexer import _scan_file, _PARTIAL_THRESHOLD + d = tmp_path / "root" + d.mkdir() + f = d / "big.mkv" + content = os.urandom(_PARTIAL_THRESHOLD + 1024 * 1024) + f.write_bytes(content) + + entry = _scan_file(one_root(d).roots[0], f) + full_hash = b3.blake3(content).hexdigest() + + assert entry.id != full_hash + assert entry.hash_version == 2 + + +def test_partial_hash_is_deterministic(tmp_path): + from meshbay_node.indexer.indexer import _scan_file, _PARTIAL_THRESHOLD + d = tmp_path / "root" + d.mkdir() + f = d / "big.mkv" + f.write_bytes(os.urandom(_PARTIAL_THRESHOLD + 1)) + + e1 = _scan_file(one_root(d).roots[0], f) + e2 = _scan_file(one_root(d).roots[0], f) + assert e1.id == e2.id + + +def test_group_index_roundtrip_preserves_hash_version(sk_node, gek): + from meshbay_common.protocol import IndexEntry + idx = GroupIndex(group_id="hv-test", sk_node=sk_node, gek=gek) + idx.add_entry(IndexEntry( + id="aaa", name="small.mp4", path="root", size=1024, + type="video", added_at=100, hash_version=1)) + idx.add_entry(IndexEntry( + id="bbb", name="big.mkv", path="root", size=50_000_000, + type="video", added_at=200, hash_version=2)) + + wire = idx.serialize() + recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) + + by_id = {e.id: e for e in recovered.entries} + assert by_id["aaa"].hash_version == 1 + assert by_id["bbb"].hash_version == 2 + + +def test_deserialize_without_hash_version_defaults_to_1(sk_node, gek): + """Entries serialized by old code (no hash_version field) must deserialize + as hash_version=1.""" + from meshbay_common.protocol import IndexEntry + idx = GroupIndex(group_id="compat", sk_node=sk_node, gek=gek) + idx.add_entry(IndexEntry( + id="old", name="f.mp4", path="root", size=1024, + type="video", added_at=100)) + wire = idx.serialize() + recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) + assert recovered.entries[0].hash_version == 1 + + +def test_index_entry_wire_includes_hash_version(): + from meshbay_common.protocol import IndexEntry, index_entry_wire + e = IndexEntry(id="x", name="f.mp4", path="root", size=1, + type="video", added_at=0, hash_version=2) + w = index_entry_wire(e) + assert w["hash_version"] == 2 + + +@pytest.mark.asyncio +async def test_cache_aware_scan_uses_hash_version(tmp_path, sk_node, gek): + from meshbay_node.indexer.indexer import _PARTIAL_THRESHOLD + d = tmp_path / "root" + d.mkdir() + small = d / "small.mp4" + small.write_bytes(os.urandom(1024)) + big = d / "big.mkv" + big.write_bytes(os.urandom(_PARTIAL_THRESHOLD + 1)) + + cache = IndexCache(db_path=tmp_path / "cache.db") + await cache.open() + + indexer = DirectoryIndexer( + roots=one_root(d), group_id="g", sk_node=sk_node, gek=gek, + cache=cache) + await indexer.initial_scan() + + by_name = {e.name: e for e in indexer.index.entries} + assert by_name["small.mp4"].hash_version == 1 + assert by_name["big.mkv"].hash_version == 2 + + hit_small = await cache.lookup( + str(small), small.stat().st_size, small.stat().st_mtime, + hash_version=1) + assert hit_small is not None + + hit_big = await cache.lookup( + str(big), big.stat().st_size, big.stat().st_mtime, + hash_version=2) + assert hit_big is not None + + await cache.close() |