diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/indexer/indexer.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 108 |
1 files changed, 78 insertions, 30 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index b479f7e..b643046 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -226,6 +226,18 @@ class DirectoryIndexer: self._reconciler: asyncio.Task | None = None self._pending_timers: dict[str, asyncio.TimerHandle] = {} self.progress = IndexProgress() + # Real-time watchdog adds (_debounce/_update_entry below) previously + # never touched `progress` at all — a whole season dropped into an + # already-watched folder gave the operator no scanning indicator and + # no progress bar, files just appeared one at a time with no feedback + # (found live). `_burst_inflight` counts files currently scheduled or + # being hashed in the current burst; `scanning` only drops back to + # False once it reaches zero *and* no more timers are pending — a + # debounce timer firing is not the same as the hash it schedules + # having finished, and the whole point of a progress indicator is to + # stay up for exactly as long as the slow part (hashing) is running. + self._burst_inflight = 0 + self._burst_sizes: dict[str, int] = {} @property def index(self) -> GroupIndex: @@ -637,6 +649,28 @@ class DirectoryIndexer: if old: old.cancel() + # Progress accounting for the real-time path — see `_burst_inflight`'s + # docstring in __init__. Only on this key's *first* appearance in the + # current burst: a rapid re-trigger of the same path (several writes + # debounced together) cancels the old timer above and must not also + # double-count its size, so `_burst_sizes` (keyed the same as + # `_pending_timers`) is the source of truth for "already counted", + # not "does a timer currently exist for it" — a cancelled timer's + # bookkeeping still has to reach the `fire()` that actually runs. + if not deleted and key not in self._burst_sizes: + try: + 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 + self._burst_sizes[key] = size + self._burst_inflight += 1 + def fire() -> None: self._pending_timers.pop(key, None) asyncio.ensure_future(self._update_entry(file_path, deleted)) @@ -654,41 +688,55 @@ class DirectoryIndexer: self._index.remove_entry(entry.id) async def _update_entry(self, file_path: Path, deleted: bool) -> None: - root = self._root_for(file_path) - if root is None: - return + try: + root = self._root_for(file_path) + if root is None: + return - if deleted and not root.is_live(): - # The volume went away rather than the file. Freeze: mark the root - # and touch nothing. Every other event for this root will arrive - # here too and be dropped the same way, which is the intent — one - # unplugged drive must not empty a library. - if root.available: - root.available = False - self._index.roots = self.roots.describe() - log.warning("Root %r disappeared — ignoring deletion events and " - "freezing %d entries", root.name, - len(self._entries_under(root))) - self._index.version = int(time.time()) - if self.on_change: - await self.on_change(self) - return + if deleted and not root.is_live(): + # The volume went away rather than the file. Freeze: mark the + # root and touch nothing. Every other event for this root + # will arrive here too and be dropped the same way, which is + # the intent — one unplugged drive must not empty a library. + if root.available: + root.available = False + self._index.roots = self.roots.describe() + log.warning("Root %r disappeared — ignoring deletion events and " + "freezing %d entries", root.name, + len(self._entries_under(root))) + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + return - if not root.available: - return + if not root.available: + return - self._remove_by_path(root, file_path) + self._remove_by_path(root, file_path) - if not deleted: - entry = await self._hash_or_cached(root, file_path) - if entry: - self._index.add_entry(entry) - log.debug("Indexed: %s (%s, %d bytes)", - file_path.name, entry.id[:8], entry.size) + if not deleted: + entry = await self._hash_or_cached(root, file_path) + if entry: + self._index.add_entry(entry) + log.debug("Indexed: %s (%s, %d bytes)", + file_path.name, entry.id[:8], entry.size) - self._index.version = int(time.time()) - if self.on_change: - await self.on_change(self) + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + finally: + # Mirror image of the accounting in _debounce, run whichever way + # this method exits (including the several early returns above) — + # otherwise a frozen/unavailable root's files would leave + # `scanning` stuck True forever, the exact bug this is fixing but + # 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_inflight -= 1 + if self._burst_inflight <= 0 and not self._pending_timers: + self.progress.scanning = False + self.progress.current_dir = "" class _WatchdogHandler(FileSystemEventHandler): |