diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/indexer')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 279 |
1 files changed, 209 insertions, 70 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 4eeeec0..4d68212 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -396,7 +396,7 @@ class DirectoryIndexer: await off_disk(self.roots, self.roots.refresh_availability) await self._plug_back_recognised() total = 0 - waiting = [r.name for r in self.roots if r.available] + waiting = [r.name for r in self.roots if r.available and not r.paused] self._queue(waiting) try: for root in self.roots: @@ -405,8 +405,9 @@ class DirectoryIndexer: "are not indexed yet and will appear when it returns", root.name, root.path) continue - waiting.remove(root.name) - self._unqueue([root.name]) + if root.name in waiting: + waiting.remove(root.name) + self._unqueue([root.name]) total += await self._scan_root(root) finally: self._unqueue(waiting) @@ -417,9 +418,21 @@ class DirectoryIndexer: total, len(self.roots)) async def _scan_root(self, root: Root, *, kind: str = "scan") -> int: + """ + Walk one root and index what is on it. + + A paused root — paused before the walk or part-way through it — is + listed from the hash cache alone: what the node already hashed comes + back, nothing new is read. It cannot simply be skipped, because the + index is rebuilt at every start and a rescan drops the root's entries + before it walks. A root removed or ejected while this runs is not + finished at all: its entries are dropped, or frozen on a drive that is + leaving, so reading the rest of it is work for nothing. + """ 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 # sizing pass, which for a large or slow (network/USB) root can @@ -432,7 +445,8 @@ class DirectoryIndexer: # had not yet turned on, even though the node was already several # seconds into walking and sizing a large root — total_bytes is # unknown at this point, so it starts at 0 and is corrected below. - self._begin_job(root.name, root=root, kind=kind) + if not quiet: + self._begin_job(root.name, root=root, kind=kind) try: try: files = await loop.run_in_executor(self._executor, _walk_root, root) @@ -447,14 +461,30 @@ class DirectoryIndexer: # loop right here, blocking the whole daemon for a large root). sized = await loop.run_in_executor(self._executor, _size_files, files) - self.progress.total_bytes = sum(size for _, size in sized) - self.progress.files_total = len(sized) - self.progress.current_dir = "" - for file_path, size in sized: - self.progress.current_dir = file_path.parent.name - entry = await self._hash_or_cached(root, file_path) - self.progress.scanned_bytes += size - self.progress.files_done += 1 + if not quiet: + self.progress.total_bytes = sum(size for _, size in sized) + self.progress.files_total = len(sized) + self.progress.current_dir = "" + for done, (file_path, size) 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: " + "removed or ejected", root.name, done, len(sized)) + break + if not quiet and live.paused: + log.info("Root %r paused after %d of %d files — the rest is " + "listed from the hash cache only", root.name, done, + len(sized)) + # Off the operator's screen now; what is left reads no file. + self._end_job() + 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)) + if not quiet: + self.progress.scanned_bytes += size + self.progress.files_done += 1 if entry: self._index.add_entry(entry) count += 1 @@ -462,7 +492,8 @@ class DirectoryIndexer: # Must run even if a hash/IO error propagates out of the loop # above — an indexing state that never turns back off is worse # than the scan itself failing. - self._end_job() + if not quiet: + self._end_job() return count def _begin_job(self, current_dir: str, total_bytes: int = 0, *, @@ -520,7 +551,13 @@ class DirectoryIndexer: if name in self.progress.queued: self.progress.queued.remove(name) - async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None: + async def _cached(self, root: Root, file_path: Path) -> 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) + + async def _hash_or_cached(self, root: Root, file_path: Path, *, + read: bool = True) -> 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 @@ -553,6 +590,8 @@ class DirectoryIndexer: hash_version=cached.hash_version, ), file_path, st) + if not read: + return None loop = asyncio.get_event_loop() entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path) if entry and self._cache is not None: @@ -643,7 +682,7 @@ class DirectoryIndexer: self._observer = Observer() watched = 0 for root in self.roots: - if not root.available: + if not root.available or root.paused: continue try: self._observer.schedule(handler, str(root.path), recursive=True) @@ -742,8 +781,26 @@ class DirectoryIndexer: if self.on_change: await self.on_change(self) + def _live(self, root: Root) -> Root | None: + """ + This set's own object for `root`, or None once it is gone from the set. + + A retarget replaces every Root object, so a walk that started before + one holds a stale copy: the flags an operator sets afterwards are on + this one. + """ + for r in self.roots: + if r.folded == root.folded and r.path == root.path: + return r + return None + def _holds(self, root: Root) -> bool: - return any(r.folded == root.folded and r.path == root.path for r in self.roots) + return self._live(root) is not None + + def _stopped(self, root: Root) -> bool: + """Removed, ejected or paused since this walk began: stop reading it.""" + live = self._live(root) + return live is None or live.ejected or live.paused async def _scan_added_roots(self, added: list[Root]) -> None: waiting = [r.name for r in added] @@ -867,69 +924,86 @@ class DirectoryIndexer: Only touches roots that are readable right now — a root whose volume is absent has nothing to compare against, and comparing anyway is exactly - the mistake this module exists to avoid. + the mistake this module exists to avoid. A paused root is left alone: + the operator stopped its indexing, and this pass is indexing. """ changed = False - loop = asyncio.get_event_loop() for root in self.roots: - if not root.available: - continue - try: - on_disk, known = await loop.run_in_executor( - self._executor, self._sweep_scan_root, root) - except OSError as e: - log.warning("Cannot reconcile root %r: %s", root.name, e) + if not root.available or root.paused: continue - - 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 - # copy under its own path, because the id is then absent. A - # dedicated "find the survivor" lookup was written first and - # deleted — it rehashed every file under the root on any single - # deletion, and a test proved it changed nothing. - self._index.remove_entry(known[missing]) - log.info("Reconcile: %s is gone", missing) + if await self._sweep_root(root): changed = True + return changed - added_paths = on_disk - set(known) - if not added_paths: - continue + async def _sweep_root(self, root: Root, *, kind: str = "reconcile") -> bool: + """ + One root's half of the sweep. Also what a resume runs: it indexes only + the files the index lacks, where a `_scan_root` would replace every + entry it met and lose what the enrichment passes had put there. + """ + changed = False + loop = asyncio.get_event_loop() + try: + on_disk, known = await loop.run_in_executor( + self._executor, self._sweep_scan_root, root) + except OSError as e: + log.warning("Cannot reconcile root %r: %s", root.name, e) + return False - 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)) + 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 + # copy under its own path, because the id is then absent. A + # dedicated "find the survivor" lookup was written first and + # deleted — it rehashed every file under the root on any single + # deletion, and a test proved it changed nothing. + self._index.remove_entry(known[missing]) + log.info("Reconcile: %s is gone", missing) + changed = True + + added_paths = on_disk - set(known) + if not added_paths: + return changed - self._begin_job("", sum(size for _, size in added_sized), root=root, - kind="reconcile", files_total=len(added_sized)) + added_sized: list[tuple[Path, int]] = [] + for p in added_paths: try: - for added, size in added_sized: - self.progress.current_dir = added.parent.name - entry = await self._hash_or_cached(root, added) - self.progress.scanned_bytes += size - self.progress.files_done += 1 - if not entry: - continue - # The index is keyed by **content hash**, so two identical - # files at two paths are one entry and the path comparison - # above cannot see the second. Adding it anyway rewrites - # that entry's path every cycle, bumps the version, and - # pushes an index update to every connected peer once a - # minute — for ever. Measured on a live node: `clip.mp4` - # present at the root and in uploads/ with the same bytes. - if self._index.get_entry(entry.id) is not None: - log.debug("Reconcile: %s duplicates content already " - "indexed as %s — leaving the index alone", - added, entry.id[:8]) - continue - self._index.add_entry(entry) + added_sized.append((p, p.stat().st_size)) + except OSError: + added_sized.append((p, 0)) + + self._begin_job("", sum(size for _, size in added_sized), root=root, + kind=kind, files_total=len(added_sized)) + try: + for done, (added, size) 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 + self.progress.files_done += 1 + if not entry: + continue + # The index is keyed by **content hash**, so two identical + # files at two paths are one entry and the path comparison + # above cannot see the second. Adding it anyway rewrites + # that entry's path every cycle, bumps the version, and + # pushes an index update to every connected peer once a + # minute — for ever. Measured on a live node: `clip.mp4` + # present at the root and in uploads/ with the same bytes. + if self._index.get_entry(entry.id) is not None: + log.debug("Reconcile: %s duplicates content already " + "indexed as %s — leaving the index alone", + added, entry.id[:8]) + continue + self._index.add_entry(entry) + if kind == "reconcile": log.info("Reconcile: %s appeared (missed event)", added) - changed = True - finally: - self._end_job() + changed = True + finally: + self._end_job() return changed def _entries_under(self, root: Root) -> list[IndexEntry]: @@ -1170,6 +1244,71 @@ class DirectoryIndexer: if self.on_change: await self.on_change(self) + # ── Pause and resume ────────────────────────────────────────────────────── + + async def pause_root(self, root_name: str) -> None: + """ + Stop indexing a root: the walk under way ends at the next file, and + neither the watcher nor reconcile touch it until it is resumed. Its + entries stay — they are files on a readable disk, served as before. + """ + root = self.roots.by_name(root_name) + if root is None or root.paused: + return + root.paused = True + self.progress.queued[:] = [n for n in self.progress.queued + if fold(n) != root.folded] + log.info("Root %r paused — %d entries kept, nothing more is indexed " + "until it is resumed", root.name, len(self._entries_under(root))) + self._restart_observer() + await self.publish_roots() + + async def resume_root(self, root_name: str) -> None: + """ + Index a paused root again, from where it stopped. + + Returns once the walk is queued, not when it ends: on a large drive + that is hours, and the operator's request has nothing to wait for — + the progress band shows the walk. Run as a task this indexer owns, like + a plug, so the session that asked can close without cancelling it. + """ + root = self.roots.by_name(root_name) + if root is None or not root.paused: + return + root.paused = False + log.info("Root %r resumed", root.name) + self._restart_observer() + await self.publish_roots() + if not root.available: + return + self._queue([root.name]) + task = asyncio.create_task(self._finish_resume(root)) + self._scan_tasks.add(task) + task.add_done_callback(self._scan_tasks.discard) + + async def _finish_resume(self, root: Root) -> None: + changed = False + waiting = True + try: + async with self._scan_lock: + self._unqueue([root.name]) + waiting = False + # Paused, ejected or removed again while it waited. + live = self._live(root) + if live is not None and live.available and not self._stopped(live): + changed = await self._sweep_root(live, kind="scan") + except asyncio.CancelledError: + raise + except Exception: + log.exception("Resuming root %r failed", root.name) + finally: + if waiting: + self._unqueue([root.name]) + if changed: + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + # ── Internal update ─────────────────────────────────────────────────────── def _schedule_update(self, file_path: Path, deleted: bool = False, @@ -1278,7 +1417,7 @@ class DirectoryIndexer: await self.on_change(self) return - if not root.available: + if not root.available or root.paused: return # On Windows watchdog cannot tell what a deleted path was -- it is |