feat(node): indexing v2 — partial-read hashing for files above 40 MB

Files above 40 MB are no longer read in full. Instead, blake3 hashes
45 MB of samples (first 20 MB + last 20 MB + 5 MB at 50% offset).
Files at or below 40 MB are unchanged (full read, hash_version 1).

A new `hash_version` field on IndexEntry (default 1) travels on the
wire and through the cache so both versions coexist without breaking
existing nodes or clients.

The IndexCache auto-migrates its schema on open (ALTER TABLE), so no
manual step is required on upgrade. A standalone migration script is
available in QE/migration/ for operators who want to preview or force
a full re-hash.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

2 files changed, 222 insertions, 0 deletions

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()