1 files changed, 485 insertions, 0 deletions
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}"