diff options
Diffstat (limited to 'packages/meshbay-node/tests')
5 files changed, 546 insertions, 15 deletions
diff --git a/packages/meshbay-node/tests/test_index_jobs_are_described.py b/packages/meshbay-node/tests/test_index_jobs_are_described.py index dee6e39..1eb203e 100644 --- a/packages/meshbay-node/tests/test_index_jobs_are_described.py +++ b/packages/meshbay-node/tests/test_index_jobs_are_described.py @@ -55,11 +55,11 @@ class _Held(DirectoryIndexer): self.walked.append(root.name) return await super()._scan_root(root, *args, **kwargs) - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): if self.hold_under is not None and file_path.is_relative_to(self.hold_under): self.at_gate.set() await self.gate.wait() - return await super()._hash_or_cached(root, file_path) + return await super()._hash_or_cached(root, file_path, **kw) def _dirs(tmp_path: Path) -> tuple[Path, Path, Path]: @@ -149,7 +149,7 @@ async def test_a_failed_initial_scan_leaves_nothing_announced(tmp_path): _, r1, r2 = _dirs(tmp_path) class _Broken(DirectoryIndexer): - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): raise RuntimeError("simulated failure mid-scan") idx = _Broken(roots=_set(r1, r2), group_id=GROUP, diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py index ecdc223..2fa8a5c 100644 --- a/packages/meshbay-node/tests/test_index_progress.py +++ b/packages/meshbay-node/tests/test_index_progress.py @@ -14,7 +14,7 @@ from fastapi.testclient import TestClient from meshbay_node.config import Config, HubConfig, KeystoreConfig, NodeConfig from meshbay_node.daemon import NodeDaemon from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.indexer.indexer import IndexProgress +from meshbay_node.indexer.indexer import NO_ANSWER, IndexProgress, UnreadFile from meshbay_node.transport.webrtc_server import WebRTCPeerSession from meshbay_node.ui.app import create_ui_app @@ -64,19 +64,30 @@ def _ui_client(state: dict) -> TestClient: "indexes": {}, **state})) +def _indexer(progress: IndexProgress, *, unread=(), stalled=None, slow=None) -> MagicMock: + """What the routes read off an indexer, with nothing wrong unless said.""" + indexer = MagicMock() + indexer.progress = progress + indexer.unread = {u.path: u for u in unread} + indexer.unread_count.return_value = len(indexer.unread) + indexer.stalled = dict(stalled or {}) + indexer.disk_wait.return_value = slow + return indexer + + def test_index_status_route_idle_for_unknown_group(): client = _ui_client({"indexers": {}}) resp = client.get("/api/groups/unknown-group/index-status") assert resp.status_code == 200 assert resp.json() == {"scanning": False, "scanned_bytes": 0, - "total_bytes": 0, "current_dir": ""} + "total_bytes": 0, "current_dir": "", + "unread": [], "unread_total": 0, "stalled": []} def test_index_status_route_reflects_indexer_progress(): - fake_indexer = MagicMock() - fake_indexer.progress = IndexProgress( + fake_indexer = _indexer(IndexProgress( scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000, - current_dir="2024") + current_dir="2024")) client = _ui_client({"indexers": {"g" * 32: fake_indexer}}) resp = client.get(f"/api/groups/{'g' * 32}/index-status") @@ -84,20 +95,47 @@ def test_index_status_route_reflects_indexer_progress(): assert resp.json() == { "scanning": True, "scanned_bytes": 1_000_000, "total_bytes": 4_000_000_000, "current_dir": "2024", + "unread": [], "unread_total": 0, "stalled": [], } +def test_the_operator_is_shown_what_a_scan_could_not_read_and_where_it_stopped(): + """On the node's own machine only: these are paths. A scan that sat still + used to say nothing about the file it sat on, and a file it gave up on was + one line in a log the operator does not read.""" + silent = UnreadFile(path="/srv/music/a/one.flac", root="music", reason=NO_ANSWER, + size=1, mtime=1.0) + refused = UnreadFile(path="/srv/music/b/two.flac", root="music", + reason="Permission denied", size=1, mtime=1.0) + indexer = _indexer(IndexProgress(), unread=[silent, refused], + stalled={"music": "/srv/music/a/three.flac"}, + slow=("/srv/music/a/four.flac", 23.7)) + config = MagicMock() + config.groups = [] + client = _ui_client({"config": config, "indexers": {"g" * 32: indexer}}) + + one = client.get(f"/api/groups/{'g' * 32}/index-status").json() + assert one["unread"] == [ + {"root": "music", "path": "/srv/music/a/one.flac", "reason": "no_answer"}, + {"root": "music", "path": "/srv/music/b/two.flac", "reason": "Permission denied"}] + assert one["unread_total"] == 2 + assert one["stalled"] == [{"root": "music", "path": "/srv/music/a/three.flac"}] + + (band,) = client.get("/api/index-status").json()["groups"] + assert (band["slow_path"], band["slow_secs"]) == ("/srv/music/a/four.flac", 23) + assert (band["unread"], band["stalled"]) == (2, ["music"]) + + # ── /api/index-status (loopback, every group) ─────────────────────────────── def test_every_group_is_described_including_one_still_being_attached(): """The band reads one route for the whole node. A group in its initial scan is in state["indexers"] and not yet in groups_ctx, and must be there.""" - busy, idle = MagicMock(), MagicMock() - busy.progress = IndexProgress( + busy = _indexer(IndexProgress( scanning=True, scanned_bytes=10, total_bytes=40, current_dir="2024", root="results", root_pos=1, kind="scan", files_done=1, files_total=4, - queued=["archive"]) - idle.progress = IndexProgress() + queued=["archive"])) + idle = _indexer(IndexProgress()) config = MagicMock() named = MagicMock() named.id, named.name = "a" * 32, "outputs" @@ -112,6 +150,7 @@ def test_every_group_is_described_including_one_still_being_attached(): "kind": "scan", "root": "results", "current_dir": "2024", "scanned_bytes": 10, "total_bytes": 40, "files_done": 1, "files_total": 4, "queued": ["archive"], + "slow_path": "", "slow_secs": 0, "unread": 0, "stalled": [], } assert groups["b" * 32]["group_name"] == "b" * 8 assert groups["b" * 32]["scanning"] is False diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index fda40c8..5540be8 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -418,7 +418,14 @@ async def test_realtime_watchdog_add_reports_progress_like_a_bulk_scan(tmp_path, assert indexer.progress.total_bytes == total_size assert indexer.progress.scanned_bytes == 0 - await asyncio.sleep(0.05) # past debounce_secs; lets all three fire and finish + # Past debounce_secs, until all three have fired and finished. Not a fixed + # 50 ms: a Windows timer ticks every 15.6 ms, the debounce alone can take + # two of them, and the test failed one run in five there on a count of + # milliseconds that says nothing about the indexer. + for _ in range(200): + if not indexer.progress.scanning: + break + await asyncio.sleep(0.01) assert indexer.progress.scanning is False, "must end idle, not stuck scanning" assert indexer.progress.scanned_bytes == total_size diff --git a/packages/meshbay-node/tests/test_indexer_silent_disk.py b/packages/meshbay-node/tests/test_indexer_silent_disk.py new file mode 100644 index 0000000..bb618fc --- /dev/null +++ b/packages/meshbay-node/tests/test_indexer_silent_disk.py @@ -0,0 +1,485 @@ +""" +A disk that stops answering, and the scan that used to wait on it for ever. + +A read on a failing drive, a share that dropped or a file a sync client has not +fetched does not fail, it does not return. The hash ran on the indexer's one +thread with nothing to bound it, so one such file stopped the scan of its group +where it stood: every root behind it unread, the enrichment that follows a scan +never started, a pause unable to take effect (it is looked at between two +files), and nothing in the log to say which file, since a file is named there +only when it fails. + +What is timed is silence, not work: a listing of a million entries and a file +on a slow link both take as long as they take. The files below "never answer" +by waiting on an event the test sets when it ends. +""" + +import asyncio +import os +import threading +from pathlib import Path +from types import SimpleNamespace + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node import ops +from meshbay_node.indexer import indexer as indexer_mod +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.indexer.indexer import NO_ANSWER, DirectoryIndexer +from meshbay_node.roots import RootSet +from meshbay_node.roster import Roster + +pytestmark = pytest.mark.asyncio + +GID = "g" * 32 + + +@pytest.fixture(autouse=True) +def _short_patience(monkeypatch): + """Tenths of a second where the node waits a minute.""" + monkeypatch.setattr(DirectoryIndexer, "STALL_SECS", 0.3) + monkeypatch.setattr(DirectoryIndexer, "SLOW_SECS", 0.1) + monkeypatch.setattr(DirectoryIndexer, "DISK_POLL_SECS", 0.02) + + +@pytest.fixture +def never(): + """What a read that never returns waits on. Set at the end, so the threads + the tests leave behind do end.""" + event = threading.Event() + yield event + event.set() + + +def _folder(tmp_path: Path, name: str, files: list[str]) -> Path: + d = tmp_path / name + d.mkdir() + for f in files: + (d / f).write_bytes(f"{name}/{f}".encode()) + return d + + +def _indexer(roots: RootSet, **kw) -> DirectoryIndexer: + return DirectoryIndexer(roots=roots, group_id=GID, + sk_node=Ed25519PrivateKey.generate(), gek=None, **kw) + + +def _names(idx: DirectoryIndexer) -> set[str]: + return {e.name for e in idx.index.entries} + + +def _silent_files(monkeypatch, never: threading.Event, silent=lambda name: False) -> list[str]: + """Reads of the files `silent` names never return. Every read is recorded.""" + read: list[str] = [] + real = indexer_mod._scan_file + + def scan(root, path): + read.append(path.name) + if silent(path.name): + never.wait() + return None + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", scan) + return read + + +# ── One file ───────────────────────────────────────────────────────────────── + +async def test_a_read_that_never_returns_costs_its_file_not_the_scan( + tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv", "stuck.mkv", "c.mkv", "d.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert _names(idx) == {"a.mkv", "b.mkv", "c.mkv", "d.mkv"} + (unread,) = idx.unread.values() + assert (Path(unread.path).name, unread.root, unread.reason) == ( + "stuck.mkv", "Films", NO_ANSWER) + assert roots.roots[0].paused is False, "one file is not the root's fault" + assert idx.progress.scanning is False + assert idx.progress.files_done == 5, "the file given up on is done with, too" + + +async def test_reconcile_does_not_wait_on_that_file_again_until_it_changes( + tmp_path, monkeypatch, never): + """Reconcile indexes every file the index lacks, at every pass. Without a + memory of this one it would wait out the limit on it each time.""" + films = _folder(tmp_path, "Films", ["a.mkv", "stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + answering = indexer_mod._scan_file + read = _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + await idx.initial_scan() + assert read.count("stuck.mkv") == 1 + + await idx.reconcile() + await idx.reconcile() + assert read.count("stuck.mkv") == 1, "asked the disk again for a file it left unanswered" + + monkeypatch.setattr(indexer_mod, "_scan_file", answering) # the disk is back... + await idx.reconcile() + assert "stuck.mkv" not in _names(idx), "...which nothing has told the node" + + (films / "stuck.mkv").write_bytes(b"written again, and longer") + await idx.reconcile() + assert "stuck.mkv" in _names(idx) + assert idx.unread == {} + + +async def test_a_file_the_system_refuses_is_listed_with_what_it_said( + tmp_path, monkeypatch): + films = _folder(tmp_path, "Films", ["a.mkv", "locked.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + real = indexer_mod.long_path + + def refuse(path): + if Path(path).name == "locked.mkv": + raise PermissionError(13, "Permission denied") + return real(path) + + monkeypatch.setattr(indexer_mod, "long_path", refuse) + await idx.initial_scan() + + assert _names(idx) == {"a.mkv"} + (unread,) = idx.unread.values() + assert (Path(unread.path).name, unread.reason) == ("locked.mkv", "Permission denied") + + # A refusal costs nothing to ask again, and permissions change without the + # file changing: it is not remembered the way a silence is. + monkeypatch.setattr(indexer_mod, "long_path", real) + await idx.reconcile() + assert "locked.mkv" in _names(idx) + assert idx.unread == {} + + +async def test_a_file_that_has_gone_is_no_longer_unread(tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv", "stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + await idx.initial_scan() + assert len(idx.unread) == 1 + + (films / "stuck.mkv").unlink() + await idx.reconcile() + + assert idx.unread == {} + + +# ── What counts as silence ─────────────────────────────────────────────────── + +async def test_a_slow_file_that_keeps_arriving_is_not_given_up_on(tmp_path, monkeypatch): + """Twice the limit in all, never the limit without a megabyte.""" + films = _folder(tmp_path, "Films", ["slow.mkv"]) + root = RootSet.build([{"path": str(films)}]) + idx = _indexer(root) + real = indexer_mod._scan_file + + def trickle(root, path): + for _ in range(6): + threading.Event().wait(0.1) + indexer_mod._beat() + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", trickle) + await idx.initial_scan() + + assert _names(idx) == {"slow.mkv"} + assert idx._executor.left_behind == 0 + + +async def test_files_waiting_behind_a_slow_one_are_not_timed_for_the_wait( + tmp_path, monkeypatch): + """A season dropped into a watched folder is hashed one file at a time on + the one thread, each event its own task. Timed from when it was asked, the + last of them would be given up on for having queued.""" + names = [f"e{i}.mkv" for i in range(5)] + films = _folder(tmp_path, "Films", names) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + real = indexer_mod._scan_file + + def unhurried(root, path): + threading.Event().wait(0.15) + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", unhurried) + entries = await asyncio.gather(*( + idx._hash_or_cached(roots.roots[0], films / name) for name in names)) + + assert all(entries), "0.75 s of queue against a limit of 0.3 s" + assert idx.unread == {} and idx._executor.left_behind == 0 + + +# ── Three in a row ─────────────────────────────────────────────────────────── + +async def test_three_in_a_row_pause_the_root_and_the_next_root_is_indexed( + tmp_path, monkeypatch, never): + """A minute a file is weeks on a music library, with a thread left behind + each time. After three with none read in between it is the disk.""" + sick = _folder(tmp_path, "Sick", [f"s{i}.mkv" for i in range(6)]) + well = _folder(tmp_path, "Well", ["w0.mkv", "w1.mkv"]) + roots = RootSet.build([{"path": str(sick)}, {"path": str(well)}]) + told = [] + + async def on_change(indexer): + told.append([r["paused"] for r in indexer.index.roots]) + + idx = _indexer(roots, on_change=on_change) + read = _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert sum(name.startswith("s") for name in read) == 3, read + assert roots.by_name("Sick").paused is True + assert Path(idx.stalled["Sick"]).parent == sick + assert _names(idx) == {"w0.mkv", "w1.mkv"}, "the root behind it waited for ever" + assert told and told[0] == [True, False], "members are told the root is paused" + assert idx.progress.scanning is False and idx.progress.queued == [] + + +async def test_a_file_read_in_between_starts_the_count_again(tmp_path, monkeypatch, never): + names = ["s0", "s1", "good", "s2", "s3", "s4"] + films = _folder(tmp_path, "Films", [n + ".mkv" for n in names]) + roots = RootSet.build([{"path": str(films)}]) + root = roots.roots[0] + idx = _indexer(roots) + _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + + for name in names[:5]: + await idx._hash_or_cached(root, films / f"{name}.mkv") + assert root.paused is False, "two, one read, two: never three in a row" + + await idx._hash_or_cached(root, films / "s4.mkv") + assert root.paused is True + + +async def test_resume_gets_past_the_files_that_gave_no_answer(tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + read = _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + await idx.initial_scan() + assert roots.roots[0].paused is True + (films / "new.mkv").write_bytes(b"dropped in while it was paused") + + await idx.resume_root("Films") + await asyncio.gather(*idx._scan_tasks) + + assert _names(idx) == {"new.mkv"} + assert len(read) == 4, f"read again what gave no answer: {read}" + assert roots.roots[0].paused is False and idx.stalled == {} + assert len(idx.unread) == 3, "still listed for the operator" + + +async def test_a_reload_does_not_start_a_stalled_root_again(tmp_path, monkeypatch, never): + """A reload builds the set from the roster, which holds the operator's + pauses. This one is the node's own.""" + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + + again = RootSet.build([{"path": str(films)}]) + await idx.retarget(again) + + assert again.roots[0].paused is True + + +def _state(roots: RootSet, roster: Roster, idx: DirectoryIndexer) -> dict: + index = GroupIndex(group_id=GID, sk_node=Ed25519PrivateKey.generate()) + return { + "groups_ctx": {GID: {"index": index, "roots": roots, "gek": None}}, + "indexers": {GID: idx}, + "config": SimpleNamespace(groups=[SimpleNamespace(id=GID, roots=[])]), + "roster": roster, + "node_user_id": "operator", + } + + +async def test_the_operators_resume_reaches_a_root_the_node_paused( + tmp_path, monkeypatch, never): + """Paused on the indexer's set and nowhere in the roster; and after a + reload the set the operations look at may be another object.""" + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + seen_by_ops = RootSet.build([{"path": str(films)}]) + out = await ops.resume_root(_state(seen_by_ops, roster, idx), GID, "Films") + await asyncio.gather(*idx._scan_tasks) + + assert out["status"] == "resumed" + assert idx.roots.roots[0].paused is False and idx.stalled == {} + finally: + await roster.close() + + +# ── The listing ────────────────────────────────────────────────────────────── + +async def test_a_listing_that_stops_answering_pauses_the_root(tmp_path, monkeypatch, never): + sick = _folder(tmp_path, "Sick", ["a.mkv"]) + well = _folder(tmp_path, "Well", ["w.mkv"]) + roots = RootSet.build([{"path": str(sick)}, {"path": str(well)}]) + idx = _indexer(roots) + real = indexer_mod._walk_root + + def walk(root): + if root.name == "Sick": + indexer_mod._beat(sick / "deep" / "down") + never.wait() + return real(root) + + monkeypatch.setattr(indexer_mod, "_walk_root", walk) + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert roots.by_name("Sick").paused is True + assert idx.stalled == {"Sick": str(sick / "deep" / "down")}, "where it went silent" + assert _names(idx) == {"w.mkv"} + + +async def test_a_rescan_that_gets_no_listing_keeps_what_the_root_held( + tmp_path, monkeypatch, never): + """A rescan drops the root's entries before it walks. A walk that learns + nothing must not leave the root empty for everyone.""" + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + for entry in idx.index.entries: + entry.display_title = "Kept" + + monkeypatch.setattr(indexer_mod, "_walk_root", lambda root: never.wait()) + await asyncio.wait_for(idx._rescan_root(roots.roots[0]), timeout=10) + + assert _names(idx) == {"a.mkv", "b.mkv"} + assert all(e.display_title == "Kept" for e in idx.index.entries) + assert idx.drain_rescanned_ids() == set(), "nothing was rebuilt, nothing to enrich again" + + +async def test_a_reconcile_that_gets_no_listing_pauses_and_removes_nothing( + tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + + monkeypatch.setattr(indexer_mod, "_walk_root", lambda root: never.wait()) + await asyncio.wait_for(idx.reconcile(), timeout=10) + + assert roots.roots[0].paused is True + assert _names(idx) == {"a.mkv"} + + +def _listed_before(root_path: Path) -> set[Path]: + """The walk this replaced.""" + return {p for p in root_path.rglob("*") if p.is_file()} + + +async def test_the_walk_lists_what_it_always_listed(tmp_path): + root = _folder(tmp_path, "Root", ["top.mkv", ".hidden", "x.part"]) + (root / "a" / "b" / "c").mkdir(parents=True) + (root / "a" / "one.mkv").write_bytes(b"1") + (root / "a" / "b" / "c" / "deep.mkv").write_bytes(b"2") + (root / "empty").mkdir() + outside = _folder(tmp_path, "Outside", ["o.mkv"]) + try: + os.symlink(outside / "o.mkv", root / "link_to_file.mkv") + os.symlink(outside, root / "link_to_dir", target_is_directory=True) + except OSError: + pass # no right to make links here: the rest still holds + listed = indexer_mod._walk_root(RootSet.build([{"path": str(root)}]).roots[0]) + + assert set(listed) == _listed_before(root) + assert len(listed) == len(set(listed)) + + +async def test_a_directory_that_cannot_be_listed_costs_only_itself(tmp_path, monkeypatch): + """One refusal used to be silent, and any other error ended the walk of the + whole root: "Cannot scan root", and nothing indexed on it at all.""" + root = _folder(tmp_path, "Root", ["top.mkv"]) + for sub in ("ok", "broken"): + (root / sub).mkdir() + (root / sub / f"{sub}.mkv").write_bytes(sub.encode()) + real = os.scandir + + def scandir(path): + if Path(path).name == "broken": + raise OSError(5, "Input/output error") + return real(path) + + monkeypatch.setattr(indexer_mod.os, "scandir", scandir) + idx = _indexer(RootSet.build([{"path": str(root)}])) + await idx.initial_scan() + + assert _names(idx) == {"top.mkv", "ok.mkv"} + + +# ── Said, and survivable ───────────────────────────────────────────────────── + +async def test_the_operator_is_told_which_file_the_disk_is_silent_on( + tmp_path, monkeypatch, never): + monkeypatch.setattr(DirectoryIndexer, "STALL_SECS", 30) + films = _folder(tmp_path, "Films", ["stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + assert idx.disk_wait() is None + + scan = asyncio.create_task(idx.initial_scan()) + try: + for _ in range(200): + if idx.disk_wait(): + break + await asyncio.sleep(0.02) + path, seconds = idx.disk_wait() + assert Path(path) == films / "stuck.mkv" and seconds >= 0.1 + finally: + never.set() + await asyncio.wait_for(scan, timeout=10) + assert idx.disk_wait() is None + + +async def test_the_node_can_stop_with_a_read_still_stuck(tmp_path, monkeypatch, never): + """The interpreter joins the threads of a pool on its way out. One read + that never returned kept the node from stopping at all.""" + films = _folder(tmp_path, "Films", ["stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + assert idx._executor.left_behind == 1 + + await asyncio.wait_for(idx.stop(), timeout=5) + + stuck = [t for t in threading.enumerate() if t.name.startswith("indexer_")] + assert stuck and all(t.daemon for t in stuck) + + +async def test_a_scan_asks_the_disk_nothing_from_the_event_loop(tmp_path, monkeypatch): + """The hash was off the loop and the stat before it was not. On a disk that + stops answering that stat is where the loop would wait, and with it every + connection of the node: no limit on the indexer's thread helps a call made + somewhere else.""" + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv"]) + (films / "sub").mkdir() + (films / "sub" / "c.mkv").write_bytes(b"c") + idx = _indexer(RootSet.build([{"path": str(films)}])) + asked: set[str] = set() + real = Path.stat + + def stat(self, **kw): + if films in self.parents: + asked.add(threading.current_thread().name.split("_")[0]) + return real(self, **kw) + + monkeypatch.setattr(Path, "stat", stat) + await idx.initial_scan() + (films / "new.mkv").write_bytes(b"missed by the watcher") + await idx.reconcile() + + assert _names(idx) == {"a.mkv", "b.mkv", "c.mkv", "new.mkv"} + assert asked == {"indexer"}, f"a stat ran on: {asked}" diff --git a/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py b/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py index 1f93ffa..056c0b9 100644 --- a/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py +++ b/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py @@ -56,13 +56,13 @@ class _Held(DirectoryIndexer): self.rescanned: list[str] = [] self._seen = 0 - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): if self.hold_under is not None and file_path.is_relative_to(self.hold_under): if self._seen == self.hold_after: self.at_gate.set() await self.gate.wait() self._seen += 1 - return await super()._hash_or_cached(root, file_path) + return await super()._hash_or_cached(root, file_path, **kw) async def _rescan_root(self, root): self.rescanned.append(root.name) |