diff options
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 82 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py | 165 |
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() |