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 | 76 |
1 files changed, 71 insertions, 5 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index da8ea82..2b68550 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -323,6 +323,14 @@ class DirectoryIndexer: # and no deletion and nothing downstream can tell the fields were # wiped. See `_drop_root_entries`. self.rescanned_ids: set[str] = set() + # Held by every whole-root walk — the initial scan, a scan of a root a + # retarget added, and reconcile. Reconcile compares disk against the + # index, so while a scan is part-way through a root every file it has + # not reached yet looks like a missed event: on a 900 GB drive that was + # thousands of "appeared" lines, the same root hashed twice over the + # one executor thread, and `progress` rewritten under the scan's feet. + self._scan_lock = asyncio.Lock() + self._scan_tasks: set[asyncio.Task] = set() @property def index(self) -> GroupIndex: @@ -350,6 +358,10 @@ class DirectoryIndexer: async def initial_scan(self) -> None: """Scan every available root. Run once at startup.""" + async with self._scan_lock: + await self._initial_scan() + + async def _initial_scan(self) -> None: self.roots.refresh_availability() total = 0 for root in self.roots: @@ -556,6 +568,10 @@ class DirectoryIndexer: except asyncio.CancelledError: pass self._reconciler = None + scans = list(self._scan_tasks) + for task in scans: + task.cancel() + await asyncio.gather(*scans, return_exceptions=True) for handle in self._pending_timers.values(): handle.cancel() self._pending_timers.clear() @@ -566,7 +582,7 @@ class DirectoryIndexer: self._executor.shutdown(wait=False) log.info("Indexer stopped") - async def retarget(self, roots: RootSet) -> None: + async def retarget(self, roots: RootSet, *, wait: bool = True) -> None: """ Point this indexer at a new set of roots, without a restart (14.8). @@ -574,6 +590,12 @@ class DirectoryIndexer: operator removed it deliberately, which is not the same event as a volume disappearing, and conflating the two is what §6.9 exists to prevent. Roots that survive keep their entries; new ones are scanned. + + The set takes effect before anything is scanned: the roots table, the + watcher and `self.roots` all move at once. With ``wait=False`` the scan + of the added roots runs in the background and this returns as soon as + the set is in place — the daemon's reload must not sit on its lock for + the hours a large drive takes to hash. """ old_names = {r.folded for r in self.roots} new_names = {r.folded for r in roots} @@ -588,22 +610,62 @@ class DirectoryIndexer: self.roots = roots roots.refresh_availability() - for root in roots: - if root.folded not in old_names and root.available: - await self._scan_root(root) + added = [r for r in roots if r.folded not in old_names and r.available] self._index.roots = roots.describe() self._index.version = int(time.time()) self._restart_observer() - if self.on_change: + + if not added: + if self.on_change: + await self.on_change(self) + return + + task = asyncio.create_task(self._scan_added_roots(added)) + self._scan_tasks.add(task) + task.add_done_callback(self._scan_tasks.discard) + if wait: + await task + elif self.on_change: + # The table now; the files when the scan ends. await self.on_change(self) + def _holds(self, root: Root) -> bool: + return any(r.folded == root.folded and r.path == root.path for r in self.roots) + + async def _scan_added_roots(self, added: list[Root]) -> None: + try: + async with self._scan_lock: + for root in added: + # A later retarget may have removed it while this waited. + if not self._holds(root): + continue + count = await self._scan_root(root) + if not self._holds(root): + # Removed while being scanned: that retarget's drop ran + # before these entries existed. + for entry in self._entries_under(root): + self._index.remove_entry(entry.id) + continue + log.info("Scan complete: %d files in root %r", count, root.name) + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + except asyncio.CancelledError: + raise + except Exception: + log.exception("Scanning the added root(s) failed") + # ── Reconciliation ──────────────────────────────────────────────────────── async def _reconcile_loop(self) -> None: while True: try: await asyncio.sleep(self._reconcile_delay) + if self._scan_lock.locked() or self._scan_tasks: + # A scan is walking a root right now. Skipped without + # backing off: the next tick after it ends is the useful one. + continue changed = await self.reconcile() if changed: self._reconcile_delay = self.reconcile_secs @@ -636,6 +698,10 @@ class DirectoryIndexer: to back off when a pass finds nothing to do, rather than running at the same cadence forever regardless of how quiet the root is. """ + async with self._scan_lock: + return await self._reconcile() + + async def _reconcile(self) -> bool: changed = self.roots.refresh_availability() touched = False |