diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 82 |
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): |