diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/indexer/indexer.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 546 |
1 files changed, 481 insertions, 65 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..55ec56a 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 os.path.isfile(long_path(path)) 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,57 @@ 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()] + return walk_files(root.path) + + +def walk_files(top: Path) -> list[Path]: + """ + Every file below `top`, links to files included, links to directories not + entered. Raises if `top` itself cannot be listed. Blocking. + + Each directory is listed under its long form and the paths returned are + built from the plain one, so a directory past Windows' MAX_PATH is listed + and everything downstream still sees, stores and compares plain paths. + """ + files: list[Path] = [] + pending = [top] + while pending: + directory = pending.pop() + _beat(directory) + try: + with os.scandir(long_path(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(directory / entry.name) + elif entry.is_file(): + files.append(directory / entry.name) + except OSError: + continue + except OSError as e: + if directory == top: + 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 +408,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, os.stat(long_path(p)))) except OSError: continue return sized @@ -215,7 +432,7 @@ def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: if not _is_indexable(file_path): return None try: - stat = file_path.stat() + stat = os.stat(long_path(file_path)) if not _is_indexable_size(file_path, stat.st_size): return None if stat.st_size > _PARTIAL_THRESHOLD: @@ -224,7 +441,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 +458,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 +493,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 +545,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 +627,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 +764,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 +781,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 +818,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 +889,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 +906,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 = os.stat(long_path(file_path)) + 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 +939,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, @@ -635,7 +1000,7 @@ class DirectoryIndexer: if self._cache is None or not user_id: return try: - st = file_path.stat() + st = os.stat(long_path(file_path)) except OSError: # Gone between the rename and here. Nothing to attribute, and # nothing for anyone to delete either. @@ -742,8 +1107,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 +1316,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 +1346,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 +1425,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 +1485,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.""" @@ -1168,7 +1577,7 @@ class DirectoryIndexer: return True for path, size, mtime in known: try: - st = Path(path).stat() + st = os.stat(long_path(path)) except OSError: continue if st.st_size == size and (mtime is None or st.st_mtime == mtime): @@ -1276,6 +1685,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() @@ -1335,7 +1747,7 @@ class DirectoryIndexer: # bookkeeping still has to reach the `fire()` that actually runs. if not deleted and key not in self._burst_sizes: try: - size = file_path.stat().st_size + size = os.stat(long_path(file_path)).st_size except OSError: size = 0 if self._burst_inflight <= 0: @@ -1390,7 +1802,8 @@ class DirectoryIndexer: covers this directory too. """ try: - return not dir_path.exists() and dir_path.parent.is_dir() + return (not os.path.exists(long_path(dir_path)) + and os.path.isdir(long_path(dir_path.parent))) except OSError: return False @@ -1417,6 +1830,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 |