fix(node): a scan keeps its progress while a burst or a plug runs beside it

A watchdog burst wrote the indexer's single `progress` directly. A file
dropped into a folder during a large scan added its size to the scan's
total, then cleared `scanning` when its own hash finished, so the progress
went blank with hours of hashing left. Bursts now keep their own counters,
shown only while no whole-root walk runs.

A plug rescan took no scan lock and walked its root beside an added root's
scan, both resetting the same counters and reading the drive in turn. It
now waits for the lock, and skips the rescan if the root was ejected or
removed while it waited, since the rescan drops the entries before it walks.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T6jPTeocXA1BePekdsgPya

1 files changed, 59 insertions, 23 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index c20902e..db1b11c 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -315,6 +315,15 @@ class DirectoryIndexer:
         # stay up for exactly as long as the slow part (hashing) is running.
         self._burst_inflight = 0
         self._burst_sizes: dict[str, int] = {}
+        # A burst keeps its own counters and `progress` shows them only while
+        # no whole-root job runs. They used to write `progress` directly, and a
+        # file dropped into a folder during a 900 GB scan added its size to the
+        # scan's total, then cleared `scanning` when its own hash finished —
+        # the bar went away with hours of hashing left.
+        self._job_running = False
+        self._burst_scanned = 0
+        self._burst_total = 0
+        self._burst_dir = ""
         # Ids whose entry this indexer threw away and rebuilt from disk, since
         # the last time a consumer drained this. A rebuilt entry carries only
         # what `_hash_or_cached` fills in — every enrichment field the Videos,
@@ -393,10 +402,7 @@ 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.progress.scanning = True
-        self.progress.scanned_bytes = 0
-        self.progress.total_bytes = 0
-        self.progress.current_dir = root.name
+        self._begin_job(root.name)
         try:
             try:
                 files = await loop.run_in_executor(self._executor, _walk_root, root)
@@ -424,10 +430,38 @@ 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.progress.scanning = False
-            self.progress.current_dir = ""
+            self._end_job()
         return count
 
+    def _begin_job(self, current_dir: str, total_bytes: int = 0) -> None:
+        """`progress` now describes a whole-root walk, whatever a burst is doing."""
+        self._job_running = True
+        self.progress.scanning = True
+        self.progress.scanned_bytes = 0
+        self.progress.total_bytes = total_bytes
+        self.progress.current_dir = current_dir
+
+    def _end_job(self) -> None:
+        self._job_running = False
+        self.progress.scanning = False
+        self.progress.current_dir = ""
+        if self._burst_inflight > 0:
+            # A burst that started during the walk is still hashing.
+            self._show_burst()
+
+    def _show_burst(self) -> None:
+        if self._job_running:
+            return
+        p = self.progress
+        if self._burst_inflight > 0:
+            p.scanning = True
+            p.current_dir = self._burst_dir
+        elif not self._pending_timers:
+            p.scanning = False
+            p.current_dir = ""
+        p.scanned_bytes = self._burst_scanned
+        p.total_bytes = self._burst_total
+
     async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None:
         """
         Cache-aware replacement for a bare _scan_file() call: skips the
@@ -784,10 +818,7 @@ class DirectoryIndexer:
                 except OSError:
                     added_sized.append((p, 0))
 
-            self.progress.scanning = True
-            self.progress.scanned_bytes = 0
-            self.progress.total_bytes = sum(size for _, size in added_sized)
-            self.progress.current_dir = ""
+            self._begin_job("", sum(size for _, size in added_sized))
             try:
                 for added, size in added_sized:
                     self.progress.current_dir = added.parent.name
@@ -811,8 +842,7 @@ class DirectoryIndexer:
                     log.info("Reconcile: %s appeared (missed event)", added)
                     changed = True
             finally:
-                self.progress.scanning = False
-                self.progress.current_dir = ""
+                self._end_job()
         return changed
 
     def _entries_under(self, root: Root) -> list[IndexEntry]:
@@ -969,7 +999,15 @@ class DirectoryIndexer:
 
     async def _finish_plug(self, root: Root | None) -> None:
         if root is not None:
-            await self._rescan_root(root)
+            # A whole-root walk like any other, so it waits its turn: run beside
+            # an added root's scan, the two reset each other's progress and read
+            # the same drive in alternation.
+            async with self._scan_lock:
+                # Ejected or removed again while it waited. `_rescan_root` drops
+                # the entries before it walks, so going ahead would empty a root
+                # that is not there to be read.
+                if self._holds(root) and not root.ejected and root.is_live():
+                    await self._rescan_root(root)
         self._restart_observer()
         self._index.roots = self.roots.describe()
         self._index.version = int(time.time())
@@ -1003,14 +1041,14 @@ class DirectoryIndexer:
                 size = file_path.stat().st_size
             except OSError:
                 size = 0
-            if not self.progress.scanning:
-                self.progress.scanning = True
-                self.progress.scanned_bytes = 0
-                self.progress.total_bytes = 0
-            self.progress.total_bytes += size
-            self.progress.current_dir = file_path.parent.name
+            if self._burst_inflight <= 0:
+                self._burst_scanned = 0
+                self._burst_total = 0
+            self._burst_total += size
+            self._burst_dir = file_path.parent.name
             self._burst_sizes[key] = size
             self._burst_inflight += 1
+            self._show_burst()
 
         def fire() -> None:
             self._pending_timers.pop(key, None)
@@ -1079,11 +1117,9 @@ class DirectoryIndexer:
             # in the other direction.
             size = self._burst_sizes.pop(str(file_path), None)
             if size is not None:
-                self.progress.scanned_bytes += size
+                self._burst_scanned += size
                 self._burst_inflight -= 1
-                if self._burst_inflight <= 0 and not self._pending_timers:
-                    self.progress.scanning = False
-                    self.progress.current_dir = ""
+                self._show_burst()
 
 
 class _WatchdogHandler(FileSystemEventHandler):