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

2 files changed, 224 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):
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
new file mode 100644
index 0000000..1f93ffa
--- /dev/null
+++ b/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py
@@ -0,0 +1,165 @@
+"""
+An indexer has one `progress`, and a whole-root walk owns it while it runs.
+
+Found while planning a progress band for the operator (2026-09-14), by reading
+who writes `progress`. Two writers had no business doing it during a scan:
+
+- A watchdog burst — a file dropped into a folder while a 900 GB root was being
+  hashed — added its size to the scan's total, and cleared `scanning` when its
+  own hash finished. Anything showing progress would have gone blank with hours
+  of hashing left.
+- A plug rescan took no lock, so it walked its root beside an added root's
+  scan: both reset the same counters, and both read the drive in turn.
+"""
+
+import asyncio
+import os
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+def _set(*specs) -> RootSet:
+    return RootSet.build([s if isinstance(s, dict) else {"path": str(s), "name": s.name}
+                          for s in specs])
+
+
+def _names(idx) -> list[str]:
+    return sorted(e.name for e in idx.index.entries)
+
+
+async def _until(predicate, timeout: float = 3.0) -> bool:
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not predicate():
+        if asyncio.get_running_loop().time() > deadline:
+            return False
+        await asyncio.sleep(0.02)
+    return True
+
+
+class _Held(DirectoryIndexer):
+    """Stops hashing under `hold_under` once `hold_after` of its files are done."""
+    hold_under: Path | None = None
+    hold_after = 0
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.gate = asyncio.Event()
+        self.at_gate = asyncio.Event()
+        self.rescanned: list[str] = []
+        self._seen = 0
+
+    async def _hash_or_cached(self, root, file_path):
+        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)
+
+    async def _rescan_root(self, root):
+        self.rescanned.append(root.name)
+        return await super()._rescan_root(root)
+
+
+def _tree(tmp_path: Path) -> tuple[Path, Path, Path]:
+    one, two, big = tmp_path / "one", tmp_path / "two", tmp_path / "big"
+    for d in (one, two, big):
+        d.mkdir()
+    (one / "a.txt").write_bytes(b"first root")
+    (two / "b.txt").write_bytes(b"removable root")
+    (big / "x.bin").write_bytes(os.urandom(3000))
+    (big / "y.bin").write_bytes(os.urandom(5000))
+    return one, two, big
+
+
+async def test_a_burst_during_a_scan_leaves_the_scan_on_screen(tmp_path):
+    one, _, big = _tree(tmp_path)
+    idx = _Held(roots=_set(one), group_id=GROUP, sk_node=Ed25519PrivateKey.generate(),
+                gek=None, debounce_secs=0.01)
+    await idx.initial_scan()
+    idx._loop = asyncio.get_running_loop()
+    idx.hold_under, idx.hold_after = big, 1
+    try:
+        await idx.retarget(_set(one, big), wait=False)
+        await asyncio.wait_for(idx.at_gate.wait(), 5)
+        total, scanned = idx.progress.total_bytes, idx.progress.scanned_bytes
+        assert total == 8000 and scanned in (3000, 5000)
+
+        dropped = one / "dropped.txt"
+        dropped.write_bytes(os.urandom(700))
+        idx._schedule_update(dropped)
+        assert await _until(lambda: "dropped.txt" in _names(idx))
+        assert await _until(lambda: idx._burst_inflight == 0)
+
+        assert idx.progress.scanning is True, (
+            "the dropped file's hash finishing turned the scan's progress off")
+        assert (idx.progress.total_bytes, idx.progress.scanned_bytes) == (total, scanned), (
+            "the dropped file was counted into the scan under way")
+
+        idx.gate.set()
+        await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
+        assert await _until(lambda: not idx.progress.scanning)
+        assert _names(idx) == ["a.txt", "dropped.txt", "x.bin", "y.bin"]
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+def _with_two_ejected(one: Path, two: Path, *more: Path) -> RootSet:
+    return _set(one, {"path": str(two), "name": "two", "ejected": True}, *more)
+
+
+async def test_a_plug_rescan_waits_for_the_scan_under_way(tmp_path):
+    one, two, big = _tree(tmp_path)
+    idx = _Held(roots=_set(one, two), group_id=GROUP,
+                sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    idx.eject_root("two")
+    idx.hold_under = big
+    try:
+        await idx.retarget(_with_two_ejected(one, two, big), wait=False)
+        await asyncio.wait_for(idx.at_gate.wait(), 5)
+
+        plug = asyncio.create_task(idx.plug_root("two"))
+        await asyncio.sleep(0.1)
+        assert idx.rescanned == [], "the plug walked its root beside the scan under way"
+
+        idx.gate.set()
+        await asyncio.wait_for(plug, 5)
+        assert idx.rescanned == ["two"]
+        assert _names(idx) == ["a.txt", "b.txt", "x.bin", "y.bin"]
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+async def test_a_root_ejected_again_while_its_plug_waited_is_not_emptied(tmp_path):
+    one, two, big = _tree(tmp_path)
+    idx = _Held(roots=_set(one, two), group_id=GROUP,
+                sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    idx.eject_root("two")
+    idx.hold_under = big
+    try:
+        await idx.retarget(_with_two_ejected(one, two, big), wait=False)
+        await asyncio.wait_for(idx.at_gate.wait(), 5)
+
+        plug = asyncio.create_task(idx.plug_root("two"))
+        await asyncio.sleep(0.1)
+        idx.eject_root("two")
+        idx.gate.set()
+        await asyncio.wait_for(plug, 5)
+
+        assert idx.rescanned == []
+        assert "b.txt" in _names(idx), "a root ejected while it waited was emptied"
+    finally:
+        idx.gate.set()
+        await idx.stop()