diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_indexer.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_indexer.py | 290 |
1 files changed, 289 insertions, 1 deletions
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index c304361..68ccc4c 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -9,7 +9,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import generate_gek -from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache +import meshbay_node.indexer.indexer as indexer_mod from conftest import one_root from meshbay_node.keystore import NodeKeys @@ -190,3 +191,290 @@ async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek): wire = indexer.index.serialize() recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) assert recovered.count == indexer.index.count + + +# ── Cache-aware scanning ─────────────────────────────────────────────────────── + +@pytest.fixture +async def index_cache(tmp_path): + c = IndexCache(db_path=tmp_path / "index_cache.db") + await c.open() + yield c + await c.close() + + +@pytest.mark.asyncio +async def test_second_scan_with_same_cache_hashes_nothing( + shared_dir, sk_node, gek, index_cache): + """ + The whole point of the cache: a "restart" (a fresh DirectoryIndexer, same + on-disk cache) that finds every file's (size, mtime) unchanged must not + read a single byte of file content. + """ + first = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + 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="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await second.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert calls == [], f"expected zero hash calls on a fully-cached rescan, got {calls}" + assert second.index.count == first.index.count + 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() + f = d / "movie.mkv" + f.write_bytes(b"original content") + + first = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await first.initial_scan() + old_id = first.index.entries[0].id + + # Change both content and mtime, as any real edit would. + f.write_bytes(b"a completely different, longer payload") + os.utime(f, (time.time() + 5, time.time() + 5)) + + second = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await second.initial_scan() + + assert second.index.count == 1 + assert second.index.entries[0].id != old_id + + +@pytest.mark.asyncio +async def test_scan_interrupted_partway_leaves_only_completed_files_cached( + tmp_path, sk_node, gek, index_cache): + """ + A cache row is only ever written after a file is fully hashed (cache.py), + so a crash mid-scan cannot leave a stale/partial row — the next scan just + treats the not-yet-cached files as new, and finishes the job. + """ + d = tmp_path / "shared" + d.mkdir() + names = [f"file{i}.bin" for i in range(5)] + for i, name in enumerate(names): + (d / name).write_bytes(os.urandom(64) * (i + 1)) + + real_scan_file = indexer_mod._scan_file + hashed_before_crash = [] + + def crash_after_three(root, path): + if len(hashed_before_crash) >= 3: + raise RuntimeError("simulated crash mid-scan") + entry = real_scan_file(root, path) + hashed_before_crash.append(path) + return entry + + indexer_mod._scan_file = crash_after_three + try: + crashing = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + with pytest.raises(RuntimeError): + await crashing.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert len(hashed_before_crash) == 3 + + # A normal rescan (same cache) must still end up with all 5 files + # correctly indexed, hashing only the ones the crash never got to. + calls = [] + + def spy(root, path): + calls.append(path) + return real_scan_file(root, path) + + indexer_mod._scan_file = spy + try: + resumed = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await resumed.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert resumed.index.count == 5 + assert len(calls) == 2, f"expected only the 2 not-yet-cached files to be hashed, got {len(calls)}" + + +# ── Progress state ─────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_progress_reflects_bytes_scanned(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek) + assert indexer.progress.scanning is False + + await indexer.initial_scan() + + total_size = sum(f.stat().st_size for f in shared_dir.rglob("*") if f.is_file()) + assert indexer.progress.scanning is False, "must end idle, not stuck scanning" + assert indexer.progress.scanned_bytes == total_size + assert indexer.progress.total_bytes == total_size + + +@pytest.mark.asyncio +async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek): + d = tmp_path / "shared" + d.mkdir() + (d / "a.bin").write_bytes(os.urandom(64)) + (d / "b.bin").write_bytes(os.urandom(64)) + + real_scan_file = indexer_mod._scan_file + + def boom(root, path): + raise RuntimeError("simulated failure mid-scan") + + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek) + indexer_mod._scan_file = boom + try: + with pytest.raises(RuntimeError): + await indexer.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert indexer.progress.scanning is False, \ + "an exception mid-scan must not leave the scanning flag stuck on" + + +# ── Off-loop directory walks, reconcile backoff ───────────────────────────── + +@pytest.mark.asyncio +async def test_walk_root_does_not_stall_the_event_loop(tmp_path, sk_node, gek): + d = tmp_path / "shared" + d.mkdir() + (d / "f.bin").write_bytes(b"x") + + real_walk = indexer_mod._walk_root + + def slow_walk(root): + time.sleep(0.2) + return real_walk(root) + + indexer_mod._walk_root = slow_walk + ticks = 0 + + async def ticker(): + nonlocal ticks + while True: + await asyncio.sleep(0.01) + ticks += 1 + + ticker_task = asyncio.create_task(ticker()) + try: + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek) + await indexer.initial_scan() + finally: + indexer_mod._walk_root = real_walk + ticker_task.cancel() + try: + await ticker_task + except asyncio.CancelledError: + pass + + assert ticks >= 5, ( + "the event loop must keep running other tasks while a directory " + f"walk is in progress in the executor — only {ticks} ticks happened " + "during a 0.2s walk") + + +@pytest.mark.asyncio +async def test_reconcile_backoff_grows_with_no_changes_then_caps(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=0.01) + await indexer.initial_scan() + assert indexer._reconcile_delay == 0.01 + + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.sleep(0.2) + assert indexer._reconcile_delay > 0.01, \ + "several no-change ticks must have grown the delay" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + # A low, instance-only cap so the clamp is observable without waiting + # through dozens of real doublings up to the real 7200s ceiling. + indexer.RECONCILE_BACKOFF_CAP = 0.05 + indexer._reconcile_delay = 0.04 + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.sleep(0.15) + assert indexer._reconcile_delay <= 0.05, "delay must never exceed the cap" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio +async def test_note_activity_resets_backoff(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=10.0) + await indexer.initial_scan() + indexer._reconcile_delay = 5000.0 # simulate a long-idle backoff + + indexer.note_activity() + + assert indexer._reconcile_delay == 10.0 + + +@pytest.mark.asyncio +async def test_reconcile_backoff_resets_when_something_actually_changes( + shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=0.02) + await indexer.initial_scan() + indexer._reconcile_delay = 5.0 # pretend it had already backed off a lot + + # A fake reconcile() rather than a real filesystem change: the real + # sweep's timing (disk I/O, the executor round trip) would race against + # this test's own sleeps. What matters here is only _reconcile_loop's + # reaction to "something changed", not reconcile()'s own detection logic + # — that is covered separately (test_root_availability.py). + reconciled_once = asyncio.Event() + + async def fake_reconcile(): + reconciled_once.set() + return True + + indexer._reconcile_delay = 0.01 + indexer.reconcile = fake_reconcile + + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.wait_for(reconciled_once.wait(), timeout=2.0) + assert indexer._reconcile_delay == 0.02, \ + "a real change must reset the delay back to the base interval" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass |