diff options
Diffstat (limited to 'packages/meshbay-node')
8 files changed, 1038 insertions, 78 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 4d68212..e4679d4 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -24,9 +24,12 @@ it is the only thing that recovers a missed event. import asyncio import logging +import os +import queue +import threading import time -from collections.abc import Awaitable, Callable -from concurrent.futures import ThreadPoolExecutor +from collections.abc import Awaitable, Callable, Iterable +from concurrent.futures import Executor, Future from dataclasses import dataclass from dataclasses import field as dataclass_field from pathlib import Path @@ -71,16 +74,171 @@ def _detect_type(path: Path) -> str: return "other" -def _is_indexable(path: Path) -> bool: - if not path.is_file(): - return False - name = path.name +def _is_indexable_name(name: str) -> bool: return ( not any(name.startswith(p) for p in EXCLUDED_PREFIXES) and not any(name.endswith(s) for s in EXCLUDED_SUFFIXES) ) +def _is_indexable(path: Path) -> bool: + return path.is_file() and _is_indexable_name(path.name) + + +# ── A disk that stops answering ─────────────────────────────────────────────── +# +# 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. Nothing here can interrupt a +# thread blocked in the kernel, so the indexer measures instead how long the +# disk has said nothing, and past a limit leaves that thread where it is and +# goes on with another (`DirectoryIndexer._on_disk`). + +_serving = threading.local() + + +def _beat(where: Path | str | None = None) -> None: + """ + The disk answered: called by a blocking routine for each directory entry + listed and each piece of a file read, on the thread doing it. + + What is timed is silence, not the call. A root of a million files and a + file on a slow link both take as long as they take; neither goes a minute + without a single entry or a single megabyte. + """ + call = getattr(_serving, "call", None) + if call is not None: + call.answered = time.monotonic() + if where is not None: + call.where = str(where) + + +def _failed(reason: str) -> None: + """Why the routine running on this thread gave up on its file.""" + call = getattr(_serving, "call", None) + if call is not None: + call.failure = reason + + +class _DiskCall: + """One blocking routine handed to the disk thread.""" + + def __init__(self, fn, args) -> None: + self.fn = fn + self.args = args + self.future: Future = Future() + # When the disk last answered it: None until a thread takes it, so a + # call waiting behind another is not timed for the wait. + self.answered: float | None = None + self.where = "" + self.failure = "" + + def silent_for(self) -> float: + answered = self.answered + return 0.0 if answered is None else time.monotonic() - answered + + +class _DiskThread(Executor): + """ + The one thread an indexer reads the disk on, which can be left behind. + + One, deliberately: hashing two files at once buys nothing and can cost a + lot. Measured on a spinning USB drive, two interleaved multi-GB reads + seek-thrash against each other rather than go faster. It only takes the + blocking read and hash off the asyncio loop. + + `abandon()` is what a ThreadPoolExecutor has no word for: the thread stays + in the call it is stuck in, for as long as the kernel keeps it there, and a + new one takes the calls that were waiting behind. The threads are daemons + for the same reason, since the interpreter joins a pool's threads on its + way out and a node with one read that never returns could not stop. + """ + + def __init__(self, name: str = "indexer") -> None: + self._name = name + self._calls: queue.SimpleQueue = queue.SimpleQueue() + self._closed = False + self.left_behind = 0 + self._start() + + def _start(self) -> None: + threading.Thread(target=self._serve, args=(self._calls,), daemon=True, + name=f"{self._name}_{self.left_behind}").start() + + @staticmethod + def _serve(calls: queue.SimpleQueue) -> None: + while (call := calls.get()) is not None: + if not call.future.set_running_or_notify_cancel(): + continue + _serving.call = call + call.answered = time.monotonic() + try: + result = call.fn(*call.args) + except BaseException as e: + call.future.set_exception(e) + else: + call.future.set_result(result) + finally: + _serving.call = None + + def call(self, fn, /, *args) -> _DiskCall: + if self._closed: + raise RuntimeError("cannot schedule new futures after shutdown") + call = _DiskCall(fn, args) + self._calls.put(call) + return call + + def submit(self, fn, /, *args, **kwargs) -> Future: + return self.call(fn, *args).future + + def abandon(self) -> None: + """Leave the thread to the call it is in; a new one takes the rest.""" + stuck, self._calls = self._calls, queue.SimpleQueue() + try: + while True: + self._calls.put(stuck.get_nowait()) + except queue.Empty: + pass + stuck.put(None) # so it ends, if the call it is in ever does + self.left_behind += 1 + self._start() + + def shutdown(self, wait: bool = True, *, cancel_futures: bool = False) -> None: + self._closed = True + self._calls.put(None) + + +class DiskSilent(Exception): + """The disk gave no answer for `DirectoryIndexer.STALL_SECS`.""" + + def __init__(self, where: str) -> None: + super().__init__(where) + self.where = where + + +@dataclass +class UnreadFile: + """A file a scan met and could not index.""" + path: str + root: str + # NO_ANSWER, or what the system said when it refused. + reason: str + size: int + mtime: float + + +NO_ANSWER = "no_answer" + + +def _stop_observer(observer) -> None: + """Blocking, and on a disk that answers nothing possibly for good: run on + a thread of its own.""" + try: + observer.stop() + observer.join() + except Exception: + log.exception("Stopping a file watcher failed") + + # Found live: a 1256-byte ".mp3" with no audio stream at all, just an ID3 # tag — a truncated/corrupted rip, sitting between two good tracks of the # same album (docs/MESHBAY_DESIGN.md §9.8). A source this small claiming to @@ -95,7 +253,11 @@ def _is_indexable_size(path: Path, size: int) -> bool: return not (_detect_type(path) == "audio" and size < MIN_AUDIO_SIZE_BYTES) -_HASH_CHUNK = 8 * 1024 * 1024 # 8 MB streaming hash chunks +# What one read asks for. Each is one answer from the disk (`_beat`), so this is +# also the least a file must give per `STALL_SECS` to count as answering: at +# 1 MB that is 17 KB/s, below any link worth serving from, where 8 MB asked +# for 140 KB/s and would have given up on a slow share that was still reading. +_HASH_CHUNK = 1024 * 1024 # On a spinning disk the cost of a sample is its three seeks, not its bytes: # measured cold on USB, 45 MB took 516 ms a file, 9 MB 94 ms, 3 MB 79 ms. The @@ -111,6 +273,7 @@ def _feed(hasher, f, nbytes: int) -> None: remaining = nbytes while remaining > 0: chunk = f.read(min(_HASH_CHUNK, remaining)) + _beat() if not chunk: break hasher.update(chunk) @@ -121,6 +284,7 @@ def _partial_hash(file_path: Path, size: int) -> str: hasher = blake3.blake3() hasher.update(size.to_bytes(8, "little")) with open(long_path(file_path), "rb") as f: + _beat() _feed(hasher, f, _PARTIAL_HEAD) f.seek(size - _PARTIAL_TAIL) _feed(hasher, f, _PARTIAL_TAIL) @@ -184,10 +348,45 @@ def _walk_root(root: Root) -> list[Path]: every other thing the daemon is doing — WebRTC sessions, chat, the admin UI — for as long as it takes. """ - return [p for p in root.path.rglob("*") if p.is_file()] + files: list[Path] = [] + pending = [root.path] + while pending: + directory = pending.pop() + _beat(directory) + try: + with os.scandir(directory) as listing: + below: list[Path] = [] + for entry in listing: + _beat() + # The same two questions `rglob("*")` asked, which this + # replaces: a directory is entered unless it is a link, and + # a file counts whether or not it is one. + try: + if entry.is_dir(follow_symlinks=False): + below.append(Path(entry.path)) + elif entry.is_file(): + files.append(Path(entry.path)) + except OSError: + continue + except OSError as e: + if directory == root.path: + raise + # One directory that cannot be listed costs that directory. It + # used to be silent for a refusal and to end the whole walk for + # anything else. Said once: reconcile walks the root again every + # few minutes, and a drive's own system folders refuse every time. + if len(_unlistable) < 1000 and str(directory) not in _unlistable: + _unlistable.add(str(directory)) + log.warning("Cannot list %s: %s", directory, e.strerror or e) + continue + pending.extend(reversed(below)) + return files + +_unlistable: set[str] = set() -def _size_files(files: list[Path]) -> list[tuple[Path, int]]: + +def _size_files(files: list[Path]) -> list[tuple[Path, os.stat_result]]: """ Blocking: stat() every file from an already-completed walk — run via an executor for the same reason `_walk_root` is (its own docstring above). @@ -197,11 +396,17 @@ def _size_files(files: list[Path]) -> list[tuple[Path, int]]: the admin UI included, for as long as the stat() calls took — and did so *before* `_scan_root` had even set `progress.scanning`, so a consumer polling it saw "not scanning" the whole time real, blocking work was happening. + + The whole stat is returned and handed down to `_hash_or_cached`, which + needs the size and the date for the hash cache: asking again there would be + a disk call on the event loop for every file, and on a disk that stops + answering it is the loop, and with it the node, that would wait. """ - sized: list[tuple[Path, int]] = [] + sized: list[tuple[Path, os.stat_result]] = [] for p in files: + _beat(p.parent) try: - sized.append((p, p.stat().st_size)) + sized.append((p, p.stat())) except OSError: continue return sized @@ -224,7 +429,9 @@ def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: else: hasher = blake3.blake3() with open(long_path(file_path), "rb") as f: + _beat() while chunk := f.read(_HASH_CHUNK): + _beat() hasher.update(chunk) hex_hash = hasher.hexdigest() hv = 1 @@ -239,6 +446,7 @@ def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: ) except (OSError, PermissionError, ValueError) as e: log.warning("Cannot index %s: %s", file_path, e) + _failed(getattr(e, "strerror", None) or str(e)) return None @@ -273,6 +481,22 @@ class DirectoryIndexer: # it — several writes to the same file in quick succession (a slow copy # in several passes) collapse into one hash instead of one per write. DEFAULT_DEBOUNCE_SECS = 2.0 + # How long the disk may say nothing — not one directory entry, not one + # megabyte of a file — before the indexer stops waiting for it. A healthy + # read answers in milliseconds and a drive woken from sleep in fifteen + # seconds or so; nothing that is going to answer takes a minute. + STALL_SECS = 60.0 + # From here the silence is logged and shown to the operator, with the + # path: the one thing they could not find out while a scan sat still. + SLOW_SECS = 10.0 + DISK_POLL_SECS = 1.0 + # This many files in a row that gave no answer, with none read in between, + # and the fault is the root's and no longer a file's: it is paused. Going + # on would cost a minute a file, a thread left behind each time, and weeks + # on a music library. + STALL_LIMIT = 3 + # Files remembered as unread, for the operator's list. + UNREAD_KEPT = 500 def __init__( self, @@ -309,15 +533,21 @@ class DirectoryIndexer: self._index = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek) self._index.roots = roots.describe() - # One worker, deliberately, not a real pool: hashing two files at once - # buys nothing here and can cost a lot. It only offloads the blocking - # read+hash off the asyncio loop; it was never used for concurrency — - # every call site awaits one run_in_executor before starting the next - # (see _hash_or_cached below) — and measured on a spinning USB drive, - # two interleaved multi-GB reads would seek-thrash against each other - # rather than go faster. Left at 1 so the number does not promise a - # concurrency this code never provided. - self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="indexer") + # One thread, see `_DiskThread`. Every blocking call goes to it through + # `_on_disk`, which is where a disk that stops answering is noticed. + self._executor = _DiskThread("indexer") + self._disk_calls: dict[_DiskCall, str] = {} + # Files met and not indexed, by path. One that gave no answer is not + # read again until it changes: reconcile meets it at every pass, and + # would otherwise wait a minute on it each time. + self.unread: dict[str, UnreadFile] = {} + self._unread_dropped = 0 + # How many files in a row gave no answer, by folded root name. + self._silent_run: dict[str, int] = {} + # Roots this indexer paused by itself, and where the disk went silent. + # Not persisted: a restart tries the root again, which is also what + # brings back one whose share or drive has returned. + self.stalled: dict[str, str] = {} self._observer: Observer | None = None self._loop: asyncio.AbstractEventLoop | None = None self._reconciler: asyncio.Task | None = None @@ -385,6 +615,97 @@ class DirectoryIndexer: best = root return best + # ── A disk that stops answering ─────────────────────────────────────────── + + async def _on_disk(self, fn, *args, what: Path | str) -> _DiskCall: + """ + Run one blocking disk routine on the indexer's thread and wait for it, + for as long as the disk keeps answering. + + Raises DiskSilent once it has said nothing for `STALL_SECS`. The thread + is then left in that call, and the calls behind it go to a new one. + The call is returned rather than its result so the caller can read why + a routine gave up on a file (`failure`). + """ + call = self._executor.call(fn, *args) + waiting = asyncio.wrap_future(call.future) + self._disk_calls[call] = str(what) + said = False + try: + while True: + done, _ = await asyncio.wait({waiting}, timeout=self.DISK_POLL_SECS) + if done: + waiting.result() + return call + silent = call.silent_for() + if silent >= self.STALL_SECS: + where = call.where or str(what) + log.warning("No answer from the disk for %d s: giving up on %s", + silent, where) + self._executor.abandon() + raise DiskSilent(where) + if silent >= self.SLOW_SECS and not said: + said = True + log.warning("No answer from the disk for %d s, still waiting on %s", + silent, call.where or what) + finally: + self._disk_calls.pop(call, None) + if not waiting.done(): + waiting.cancel() + + def disk_wait(self) -> tuple[str, float] | None: + """What the disk has been silent on for `SLOW_SECS` or more, and for + how long. For the operator's own screen: it is a path.""" + for call, what in self._disk_calls.items(): + silent = call.silent_for() + if silent >= self.SLOW_SECS: + return call.where or what, silent + return None + + def _note_unread(self, root: Root, file_path: Path, st, reason: str) -> None: + key = str(file_path) + if key not in self.unread and len(self.unread) >= self.UNREAD_KEPT: + self._unread_dropped += 1 + return + self.unread[key] = UnreadFile(path=key, root=root.name, reason=reason, + size=st.st_size, mtime=st.st_mtime) + + def unread_count(self) -> int: + return len(self.unread) + self._unread_dropped + + def _forget_root_trouble(self, root: Root) -> None: + """A root resumed, replugged or removed starts again with a clean slate.""" + self.stalled.pop(root.name, None) + self._silent_run.pop(root.folded, None) + + async def _stall_root(self, root: Root, where: str) -> None: + """ + Pause a root whose disk stopped answering, and say so. + + The operator's own pause, set by the node: the walk under way lists + the rest from the hash cache, the watcher and reconcile leave the root + alone, what is indexed stays served, and Resume starts it again. + """ + live = self._live(root) + if live is None or live.paused: + return + live.paused = True + self.stalled[live.name] = where + self._silent_run.pop(live.folded, None) + self.progress.queued[:] = [n for n in self.progress.queued + if fold(n) != live.folded] + log.warning("Root %r paused by the node: its disk gave no answer (%s). " + "Nothing more is indexed on it until it is resumed", + live.name, where) + # Not `_restart_observer`: it joins the old watcher on this thread, and + # a watcher on the disk that just stopped answering may never let go. + old, self._observer = self._observer, None + if old is not None: + self._start_observer() + threading.Thread(target=_stop_observer, args=(old,), daemon=True, + name="indexer_unwatch").start() + await self.publish_roots() + # ── Initial scan ────────────────────────────────────────────────────────── async def initial_scan(self) -> None: @@ -431,7 +752,6 @@ class DirectoryIndexer: """ log.info("Scanning %s (root %r) ...", root.path, root.name) count = 0 - loop = asyncio.get_event_loop() quiet = root.paused # `scanning` flips on here, before the walk — not after it and the @@ -449,23 +769,29 @@ class DirectoryIndexer: self._begin_job(root.name, root=root, kind=kind) try: try: - files = await loop.run_in_executor(self._executor, _walk_root, root) + files = (await self._on_disk(_walk_root, root, what=root.path) + ).future.result() + # Sizes up front, off the same listing that already walked the + # tree — the progress bar's denominator, not a second pass over + # the disk. Off-loop for the same reason the walk itself is + # (_size_files's own docstring: this used to be a synchronous + # loop right here, blocking the whole daemon for a large root). + sized = (await self._on_disk(_size_files, files, what=root.path) + ).future.result() + except DiskSilent as e: + # No list, so nothing to walk and nothing to list from the + # cache either (`_rescan_root` puts back what it had dropped). + await self._stall_root(root, e.where) + return 0 except OSError as e: log.warning("Cannot scan root %r: %s", root.name, e) return 0 - # Sizes up front, off the same listing that already walked the - # tree — the progress bar's denominator, not a second pass over - # the disk. Off-loop for the same reason the walk itself is - # (_size_files's own docstring: this used to be a synchronous - # loop right here, blocking the whole daemon for a large root). - sized = await loop.run_in_executor(self._executor, _size_files, files) - if not quiet: - self.progress.total_bytes = sum(size for _, size in sized) + self.progress.total_bytes = sum(st.st_size for _, st in sized) self.progress.files_total = len(sized) self.progress.current_dir = "" - for done, (file_path, size) in enumerate(sized): + for done, (file_path, st) in enumerate(sized): live = self._live(root) if live is None or live.ejected: log.info("Scan of root %r stopped after %d of %d files: " @@ -480,10 +806,10 @@ class DirectoryIndexer: quiet = True if not quiet: self.progress.current_dir = file_path.parent.name - entry = await (self._cached(root, file_path) if quiet - else self._hash_or_cached(root, file_path)) + entry = await (self._cached(root, file_path, st) if quiet + else self._hash_or_cached(root, file_path, st=st)) if not quiet: - self.progress.scanned_bytes += size + self.progress.scanned_bytes += st.st_size self.progress.files_done += 1 if entry: self._index.add_entry(entry) @@ -551,13 +877,15 @@ class DirectoryIndexer: if name in self.progress.queued: self.progress.queued.remove(name) - async def _cached(self, root: Root, file_path: Path) -> IndexEntry | None: + async def _cached(self, root: Root, file_path: Path, + st: os.stat_result | None = None) -> IndexEntry | None: """What `_hash_or_cached` would return without reading the file, or None: a paused root's file the cache does not know is left out.""" - return await self._hash_or_cached(root, file_path, read=False) + return await self._hash_or_cached(root, file_path, read=False, st=st) async def _hash_or_cached(self, root: Root, file_path: Path, *, - read: bool = True) -> IndexEntry | None: + read: bool = True, + st: os.stat_result | None = None) -> 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 @@ -566,12 +894,19 @@ class DirectoryIndexer: Whatever hash_version the cache holds is kept: TMDB matches, manual corrections, thumbnails and members' playlists are keyed by the id, so a file the node already knows never changes id because the scheme did. + + `st` is the file's stat when a walk already has it (`_size_files`), so + that a root's scan asks the disk nothing from the event loop. Without + it, for one file a watcher event names, it is asked for here. """ - if not _is_indexable(file_path): - return None - try: - st = file_path.stat() - except OSError: + if st is None: + if not _is_indexable(file_path): + return None + try: + st = file_path.stat() + except OSError: + return None + elif not _is_indexable_name(file_path.name): return None if not _is_indexable_size(file_path, st.st_size): return None @@ -592,8 +927,26 @@ class DirectoryIndexer: if not read: return None - loop = asyncio.get_event_loop() - entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path) + known = self.unread.get(str(file_path)) + if (known is not None and known.reason == NO_ANSWER + and (known.size, known.mtime) == (st.st_size, st.st_mtime)): + return None + try: + call = await self._on_disk(_scan_file, root, file_path, what=file_path) + except DiskSilent: + self._note_unread(root, file_path, st, NO_ANSWER) + run = self._silent_run.get(root.folded, 0) + 1 + self._silent_run[root.folded] = run + if run >= self.STALL_LIMIT: + await self._stall_root(root, str(file_path)) + return None + # The disk answered, whatever it said. + self._silent_run.pop(root.folded, None) + entry = call.future.result() + if entry is None and call.failure: + self._note_unread(root, file_path, st, call.failure) + elif entry is not None: + self.unread.pop(str(file_path), None) 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, @@ -742,8 +1095,17 @@ class DirectoryIndexer: root.name, len(dropped)) for entry in dropped: self._index.remove_entry(entry.id) + self._forget_root_trouble(root) + for path in [k for k, u in self.unread.items() + if fold(u.root) == root.folded]: + del self.unread[path] self.roots = roots + # The new set is built from the roster, which knows the operator's + # pauses and not this indexer's own. + for root in roots: + if root.name in self.stalled: + root.paused = True await off_disk(roots, roots.refresh_availability) added = [r for r in roots if r.folded not in old_names and r.available] @@ -942,14 +1304,21 @@ class DirectoryIndexer: entry it met and lose what the enrichment passes had put there. """ changed = False - loop = asyncio.get_event_loop() + unread = [path for path, u in self.unread.items() if fold(u.root) == root.folded] try: - on_disk, known = await loop.run_in_executor( - self._executor, self._sweep_scan_root, root) + on_disk, known, gone = (await self._on_disk( + self._sweep_scan_root, root, unread, what=root.path)).future.result() + except DiskSilent as e: + await self._stall_root(root, e.where) + return False except OSError as e: log.warning("Cannot reconcile root %r: %s", root.name, e) return False + # A file that could not be read and has since gone is no longer one. + for path in gone: + self.unread.pop(path, None) + for missing in set(known) - on_disk: # Duplicate content is handled without a special case here: the # entry goes, and the add loop below re-indexes the surviving @@ -965,24 +1334,24 @@ class DirectoryIndexer: if not added_paths: return changed - added_sized: list[tuple[Path, int]] = [] - for p in added_paths: - try: - added_sized.append((p, p.stat().st_size)) - except OSError: - added_sized.append((p, 0)) + try: + added_sized = (await self._on_disk( + _size_files, list(added_paths), what=root.path)).future.result() + except DiskSilent as e: + await self._stall_root(root, e.where) + return changed - self._begin_job("", sum(size for _, size in added_sized), root=root, + self._begin_job("", sum(st.st_size for _, st in added_sized), root=root, kind=kind, files_total=len(added_sized)) try: - for done, (added, size) in enumerate(added_sized): + for done, (added, st) in enumerate(added_sized): if self._stopped(root): log.info("Indexing root %r stopped after %d of %d new files", root.name, done, len(added_sized)) break self.progress.current_dir = added.parent.name - entry = await self._hash_or_cached(root, added) - self.progress.scanned_bytes += size + entry = await self._hash_or_cached(root, added, st=st) + self.progress.scanned_bytes += st.st_size self.progress.files_done += 1 if not entry: continue @@ -1044,7 +1413,18 @@ class DirectoryIndexer: """ carried = {(e.id, e.name, e.path): e for e in self._entries_under(root)} self._drop_root_entries(root) + # It went away and came back: what it refused before is asked again. + self._forget_root_trouble(root) + for path in [k for k, u in self.unread.items() if fold(u.root) == root.folded]: + del self.unread[path] count = await self._scan_root(root, kind="rescan") + if not count and root.name in self.stalled: + # The disk went silent before anything was read: nothing was + # learnt about the root, so nothing it held is forgotten. + for old in carried.values(): + self._index.add_entry(old) + self.rescanned_ids.discard(old.id) + return 0 for entry in self._entries_under(root): old = carried.get((entry.id, entry.name, entry.path)) if old is None: @@ -1093,20 +1473,37 @@ class DirectoryIndexer: except OSError: return None - def _sweep_scan_root(self, root: Root) -> tuple[set[Path], dict[Path, str]]: + def _sweep_scan_root(self, root: Root, unread: Iterable[str] = (), + ) -> tuple[set[Path], dict[Path, str], list[str]]: """ Blocking: the two disk-touching pieces of one reconcile pass for a root, bundled so both run together in the executor rather than in the asyncio loop — the tree walk, and resolving the real path of every entry already known under this root (one syscall each). + + Also which of the `unread` paths are no longer on the disk, since + telling takes the same resolution. """ - on_disk = {p.resolve() for p in root.path.rglob("*") if _is_indexable(p)} + on_disk = set() + for p in _walk_root(root): + if _is_indexable_name(p.name): + _beat(p.parent) + on_disk.add(p.resolve()) known: dict[Path, str] = {} for entry in self._entries_under(root): + _beat() abs_path = self._entry_path(root, entry) if abs_path: known[abs_path] = entry.id - return on_disk, known + gone = [] + for path in unread: + _beat() + try: + if Path(path).resolve() not in on_disk: + gone.append(path) + except OSError: + continue + return on_disk, known, gone def _restart_observer(self) -> None: """Re-schedule watches after roots appeared or disappeared.""" @@ -1276,6 +1673,9 @@ class DirectoryIndexer: if root is None or not root.paused: return root.paused = False + # Files that gave no answer stay in `unread` and are not read again by + # the walk this starts: it has to get past them to reach the rest. + self._forget_root_trouble(root) log.info("Root %r resumed", root.name) self._restart_observer() await self.publish_roots() @@ -1417,6 +1817,9 @@ class DirectoryIndexer: await self.on_change(self) return + if deleted: + self.unread.pop(str(file_path), None) + if not root.available or root.paused: return diff --git a/packages/meshbay-node/src/meshbay_node/ops/roots.py b/packages/meshbay-node/src/meshbay_node/ops/roots.py index 187e134..0cd2daa 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/roots.py +++ b/packages/meshbay-node/src/meshbay_node/ops/roots.py @@ -361,13 +361,16 @@ async def pause_root(state: dict, group_id: str, root_name: str) -> dict: async def resume_root(state: dict, group_id: str, root_name: str) -> dict: """Index a paused root again; the walk runs in the background.""" roots, root = _live_root(state, group_id, root_name) - if not root.paused: + indexer = state.get("indexers", {}).get(group_id) + # A root the node paused by itself, its disk having stopped answering, is + # paused on the indexer's own set and nowhere in the roster. + own = indexer.roots.by_name(root.name) if indexer else None + if not root.paused and not (own is not None and own.paused): return {"status": "already_resumed", "name": root.name, "group_id": group_id, "roots": roots.describe()} await _roster(state).set_root_paused( group_id, root.name, False, set_by=state.get("node_user_id", "")) - indexer = state.get("indexers", {}).get(group_id) if indexer: await indexer.resume_root(root.name) root.paused = False diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 9933750..f5cd5d0 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -23,6 +23,9 @@ from meshbay_node import __version__, ops log = logging.getLogger(__name__) +# Unread files sent to the operator's page at once; `unread_total` says the rest. +UNREAD_LISTED = 100 + def _op(coro): """ @@ -490,17 +493,28 @@ def create_ui_app(state: dict) -> FastAPI: is only added to groups_ctx once that scan finishes (it is not yet authorized for member connections either way — see _reload_config) — this is precisely the window the wizard needs to watch. + + Also what the scan could not do, with paths, which is why it is here + and nowhere a member can ask: `unread`, the files it met and did not + index, each with the reason (`no_answer` when the disk said nothing); + and `stalled`, the folders the node paused by itself because their + disk stopped answering, each with where. """ indexer = state.get("indexers", {}).get(group_id) progress = indexer.progress if indexer else None if progress is None: return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0, - "current_dir": ""} + "current_dir": "", "unread": [], "unread_total": 0, "stalled": []} return { "scanning": progress.scanning, "scanned_bytes": progress.scanned_bytes, "total_bytes": progress.total_bytes, "current_dir": progress.current_dir, + "unread": [{"root": u.root, "path": u.path, "reason": u.reason} + for u in list(indexer.unread.values())[:UNREAD_LISTED]], + "unread_total": indexer.unread_count(), + "stalled": [{"root": name, "path": where} + for name, where in indexer.stalled.items()], } @app.get("/api/index-status") @@ -512,12 +526,17 @@ def create_ui_app(state: dict) -> FastAPI: is on screen whatever page the operator is on, so it cannot ask per group. Names roots, like `current_dir` above: loopback only, the operator's own screen. Reads state["indexers"] for the same reason. + + `slow_path` and `slow_secs` say what the disk has given no answer on, + and for how long, once that passes ten seconds: a scan that sits still + names the file it is sitting on. """ config = state.get("config") names = {g.id: g.name for g in config.groups} if config else {} groups = [] for gid, indexer in list(state.get("indexers", {}).items()): p = indexer.progress + slow = indexer.disk_wait() groups.append({ "group_id": gid, "group_name": names.get(gid, gid[:8]), @@ -530,6 +549,10 @@ def create_ui_app(state: dict) -> FastAPI: "files_done": p.files_done, "files_total": p.files_total, "queued": list(p.queued), + "slow_path": slow[0] if slow else "", + "slow_secs": int(slow[1]) if slow else 0, + "unread": indexer.unread_count(), + "stalled": list(indexer.stalled), }) return {"groups": groups} 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) |