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 | 125 |
1 files changed, 123 insertions, 2 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index b0a8e50..33e7210 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -263,12 +263,17 @@ class DirectoryIndexer: cache: IndexCache | None = None, reconcile_secs: float = DEFAULT_RECONCILE_SECS, debounce_secs: float = DEFAULT_DEBOUNCE_SECS, + on_root_ejected: Callable[[str, bool], Awaitable[None]] | None = None, ): self.roots = roots self.group_id = group_id self.sk_node = sk_node self.gek = gek self.on_change = on_change + # Called with (root_name, ejected) whenever this indexer changes a + # root's ejected state by itself — the surprise-unplug safety net. + # The daemon writes it to the roster, so a restart does not undo it. + self.on_root_ejected = on_root_ejected self.reconcile_secs = reconcile_secs self.debounce_secs = debounce_secs # Current backoff delay — starts at reconcile_secs, doubles on every @@ -309,6 +314,14 @@ 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] = {} + # 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, + # Music and Photos apps put there is gone — but its id is the file's + # content hash, so a diff against the last broadcast sees no addition + # and no deletion and nothing downstream can tell the fields were + # wiped. See `_drop_root_entries`. + self.rescanned_ids: set[str] = set() @property def index(self) -> GroupIndex: @@ -582,11 +595,21 @@ class DirectoryIndexer: changed = self.roots.refresh_availability() touched = False + # Drained before the loop below, because persisting the flag is what + # makes the safety net survive a restart — and a restart is exactly + # what an operator does after noticing a drive fell off. + while self.roots.auto_ejected: + name = self.roots.auto_ejected.pop(0) + if self.on_root_ejected: + try: + await self.on_root_ejected(name, True) + except Exception: + log.exception("Could not persist the auto-eject of root %r", name) + for root, available in changed: if available: log.info("Root %r is back — rescanning", root.name) - self._drop_root_entries(root) - await self._scan_root(root) + await self._rescan_root(root) touched = True else: # Frozen: entries stay, marked unavailable to members through @@ -687,10 +710,71 @@ class DirectoryIndexer: return [e for e in self._index.entries if fold(e.path).split("/", 1)[0] == prefix] + # Everything on an IndexEntry that a scan does not produce. `_scan_root` + # fills id/name/path/size/type/added_at/hash_version from the file itself; + # every field below was derived by one of the enrichment passes and is + # nowhere on disk to be read back. + _ENRICHED_FIELDS = ( + "duration", "thumb_hash", "width", "height", + "display_title", "season", "episode", + "artist", "album", "track_no", "taken_at", "camera", + "uploader_id", "uploader_pk", + ) + + async def _rescan_root(self, root: Root) -> int: + """ + Rebuild one root's entries from disk, keeping what the files still say. + + The two callers — `reconcile` when a root reappears, `plug_root` when + the operator plugs one back in — have to re-walk: the drive may have + changed while it was away. What they must not do is throw away the + enrichment. An entry's id is its content hash, so an entry that comes + back under the same id, name and path is the same bytes in the same + place, and every field the Videos, Music and Photos passes derived from + it still holds. Re-deriving them means minutes of tag reads, ffprobe + runs and rate-limited metadata lookups during which the operator's + library sits empty — which is exactly what a replug looked like. + + Anything that does *not* match is left bare on purpose: a different id + is different content, and a different name or path can change the + filename and folder fallbacks that `display_title`, `track_no`, + `artist` and `album` fall back to. Those are the entries + `daemon._broadcast_index_change` re-enriches, off `rescanned_ids`. + """ + carried = {(e.id, e.name, e.path): e for e in self._entries_under(root)} + self._drop_root_entries(root) + count = await self._scan_root(root) + for entry in self._entries_under(root): + old = carried.get((entry.id, entry.name, entry.path)) + if old is None: + continue + for field in self._ENRICHED_FIELDS: + setattr(entry, field, getattr(old, field)) + # It came back intact, so it is not one of the entries the daemon + # needs to enrich again. + self.rescanned_ids.discard(entry.id) + return count + def _drop_root_entries(self, root: Root) -> None: + """ + Throw away a root's entries. Only ever called to rebuild them. + + The ids are recorded because nothing outside can otherwise tell they + were rebuilt: no deletion is broadcast (the rescan is immediate) and + the entries come back under the same content-hash ids, so a diff + against the last broadcast reports neither an addition nor a deletion. + `_rescan_root` clears the ones it managed to carry over intact; what is + left is genuinely new to the apps and is re-enriched by the daemon. + """ for entry in self._entries_under(root): + self.rescanned_ids.add(entry.id) self._index.remove_entry(entry.id) + def drain_rescanned_ids(self) -> set[str]: + """Take the ids rebuilt since the last call; leave the set empty.""" + drained, self.rescanned_ids = self.rescanned_ids, set() + return drained + @staticmethod def _entry_path(root: Root, entry: IndexEntry) -> Path | None: _, _, tail = entry.path.partition("/") @@ -723,6 +807,43 @@ class DirectoryIndexer: self._observer = None self._start_observer() + def eject_root(self, root_name: str) -> None: + """Stop watching a root without touching its entries.""" + from meshbay_common.paths import fold + target = fold(root_name) + for root in self.roots: + if fold(root.name) == target: + root.ejected = True + root.available = False + frozen = len(self._entries_under(root)) + log.info("Root %r ejected — %d entries frozen", root.name, frozen) + break + self._restart_observer() + self._index.roots = self.roots.describe() + self._index.version = int(time.time()) + + async def plug_root(self, root_name: str) -> None: + """Restart watching a previously ejected root and reconcile.""" + from meshbay_common.paths import fold + target = fold(root_name) + root = None + for r in self.roots: + if fold(r.name) == target: + root = r + break + if root is None: + return + root.ejected = False + root.available = root.is_live() + if root.available: + log.info("Root %r plugged — rescanning", root.name) + await self._rescan_root(root) + self._restart_observer() + self._index.roots = self.roots.describe() + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + # ── Internal update ─────────────────────────────────────────────────────── def _schedule_update(self, file_path: Path, deleted: bool = False) -> None: |