feat: let the operator pause and resume a root's indexing (MNP 6.2)

A whole-root walk now checks its root before every file: removed or
ejected, it stops instead of reading the drive to the end; paused, it
finishes from the hash cache alone. A paused root is not watched, not
swept by reconcile and survives a restart (roster), its indexed files
stay served. Resume runs the reconcile sweep over the root, so entries
and their enrichment are kept.

Signed root_pause / root_resume over MNP, loopback routes, CLI
`root pause|resume`, a pause button on the indexing dock and
pause/resume in the roots table of Group settings.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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