diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/indexer')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/group_index.py | 10 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 400 |
2 files changed, 360 insertions, 50 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py index a69429c..5dbdc5d 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py @@ -59,6 +59,12 @@ class GroupIndex: sk_node: Ed25519PrivateKey gek: bytes | None = None # None → public group (no encryption) version: int = 1 + # The group's roots and whether each is readable right now. Travels inside + # the encrypted payload because it names the operator's directories, and a + # member needs it to tell "temporarily unavailable" from "deleted" — a + # distinction the entries alone cannot carry, since an unavailable root's + # files are still listed. Absent in an index written before roots existed. + roots: list = field(default_factory=list) _entries: dict = field(default_factory=dict, repr=False) # id → IndexEntry # ── Entry management ────────────────────────────────────────────────────── @@ -90,6 +96,7 @@ class GroupIndex: payload = msgpack.packb({ "group_id": self.group_id, "version": self.version, + "roots": list(self.roots), "entries": [asdict(e) for e in self.entries], }, use_bin_type=True) @@ -170,6 +177,9 @@ class GroupIndex: sk_node=sk_node, gek=gek, version=payload["version"], + # Absent from an index written before roots existed; an empty list + # reads as "nothing known about availability", not "no roots". + roots=payload.get("roots") or [], ) for e in payload["entries"]: idx.add_entry(IndexEntry(**e)) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 60dc04b..d9b1d2c 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -1,15 +1,29 @@ """ -Directory indexer — watches a directory and maintains a GroupIndex. +Directory indexer — watches a group's roots and maintains a GroupIndex. -Uses watchdog for filesystem events. On any change (create/modify/delete/move), -the affected file is re-scanned and the GroupIndex is updated. -File metadata (blake3 hash, size, type, duration) is computed on first scan. -Heavy operations (hashing large files) run in a thread pool to avoid blocking. +Uses watchdog for filesystem events. On any change the affected file is +re-scanned and the GroupIndex is updated. File metadata (blake3 hash, size, +type) is computed on first scan; hashing runs in a thread pool. + +Two properties are worth stating because they are what the code is shaped +around, not incidental: + +**A root that goes away freezes; it never empties.** Unmounting a volume either +makes watchdog emit a deletion for every file under it or presents an empty +directory to the next scan. Both would propagate deletions for a whole library +as though the owner had erased it, to every member. So a deletion is acted on +only once the root it belongs to has been confirmed still readable, and a root +that is not is marked unavailable with its entries left exactly where they are. + +**Events are not trusted to be complete.** `ReadDirectoryChangesW` drops events +when its buffer overflows under a burst, and inotify on a FUSE mount misses +changes made outside it. Most users are on Windows sharing from exFAT, so both +apply. A periodic reconciliation scan is therefore not a belt-and-braces extra; +it is the only thing that recovers a missed event. """ import asyncio import logging -import mimetypes import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path @@ -20,8 +34,10 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from watchdog.events import FileSystemEvent, FileSystemEventHandler from watchdog.observers import Observer +from meshbay_common.paths import fold, find_fold_collisions, long_path from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roots import Root, RootSet log = logging.getLogger(__name__) @@ -59,7 +75,19 @@ def _is_indexable(path: Path) -> bool: _HASH_CHUNK = 8 * 1024 * 1024 # 8 MB streaming hash chunks -def _scan_file(root: Path, file_path: Path) -> IndexEntry | None: +def _virtual_dir(root: Root, file_path: Path) -> str: + """ + The directory a file appears in, as members see it: `"Films/2024"`. + + The root name is the first segment for every root, including the only one of + a single-root group — one path shape has to be got right once, two have to + be kept right forever. + """ + rel = file_path.parent.relative_to(root.path) + return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}" + + +def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: """Compute IndexEntry for a file. Blocking — run in executor. Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.) don't require loading the whole file into memory.""" @@ -68,33 +96,33 @@ def _scan_file(root: Path, file_path: Path) -> IndexEntry | None: try: stat = file_path.stat() hasher = blake3.blake3() - with open(file_path, "rb") as f: + # long_path is a no-op off Windows; there it is what lets a deep media + # library past MAX_PATH. + with open(long_path(file_path), "rb") as f: while chunk := f.read(_HASH_CHUNK): hasher.update(chunk) - file_id = hasher.hexdigest() - rel_path = str(file_path.parent.relative_to(root)) - if rel_path == ".": - rel_path = "" return IndexEntry( - id=file_id, + id=hasher.hexdigest(), + # Stored exactly as the filesystem gave it: this is the string that + # opens the file. Normalization is for comparison only. name=file_path.name, - path=rel_path, + path=_virtual_dir(root, file_path), size=stat.st_size, type=_detect_type(file_path), added_at=int(stat.st_mtime), ) - except (OSError, PermissionError) as e: + except (OSError, PermissionError, ValueError) as e: log.warning("Cannot index %s: %s", file_path, e) return None class DirectoryIndexer: """ - Watches a directory and keeps a GroupIndex up to date. + Watches a group's roots and keeps a GroupIndex up to date. Usage: indexer = DirectoryIndexer( - root=Path("/home/user/shared"), + roots=RootSet.build([{"path": "/home/user/shared", "upload": True}]), group_id="my-group", sk_node=sk, gek=gek_bytes, @@ -105,61 +133,147 @@ class DirectoryIndexer: await indexer.stop() """ + # How often to re-check which roots are readable and reconcile the index + # against what is actually on disk. Not a poll for changes — a backstop for + # the events the OS did not deliver, and the way a re-plugged drive is + # noticed. + RECONCILE_SECS = 60.0 + def __init__( self, - root: Path, + roots: RootSet, group_id: str, sk_node: Ed25519PrivateKey, gek: bytes | None, on_change: Callable[["DirectoryIndexer"], Awaitable[None]] | None = None, ): - self.root = root.resolve() + self.roots = roots self.group_id = group_id self.sk_node = sk_node self.gek = gek self.on_change = on_change self._index = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek) + self._index.roots = roots.describe() self._executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="indexer") self._observer: Observer | None = None self._loop: asyncio.AbstractEventLoop | None = None + self._reconciler: asyncio.Task | None = None + self._pending_timers: dict[str, asyncio.TimerHandle] = {} @property def index(self) -> GroupIndex: return self._index + # ── Root lookup ─────────────────────────────────────────────────────────── + + def _root_for(self, file_path: Path) -> Root | None: + """Which root a real path belongs to, longest match first.""" + try: + resolved = file_path.resolve() + except OSError: + resolved = file_path + best: Root | None = None + for root in self.roots: + try: + resolved.relative_to(root.path) + except ValueError: + continue + if best is None or len(root.path.parts) > len(best.path.parts): + best = root + return best + # ── Initial scan ────────────────────────────────────────────────────────── async def initial_scan(self) -> None: - """Scan the entire directory tree. Run once at startup.""" - log.info("Scanning %s ...", self.root) + """Scan every available root. Run once at startup.""" + self.roots.refresh_availability() + total = 0 + for root in self.roots: + if not root.available: + log.warning("Root %r is not readable at startup (%s) — its files " + "are not indexed yet and will appear when it returns", + root.name, root.path) + continue + total += await self._scan_root(root) + self._index.version = int(time.time()) + self._index.roots = self.roots.describe() + self._report_collisions() + log.info("Initial scan complete: %d files across %d root(s)", + total, len(self.roots)) + + async def _scan_root(self, root: Root) -> int: + log.info("Scanning %s (root %r) ...", root.path, root.name) loop = asyncio.get_event_loop() - files = [p for p in self.root.rglob("*") if p.is_file()] count = 0 + try: + files = [p for p in root.path.rglob("*") if p.is_file()] + except OSError as e: + log.warning("Cannot scan root %r: %s", root.name, e) + return 0 for file_path in files: entry = await loop.run_in_executor( - self._executor, _scan_file, self.root, file_path) + self._executor, _scan_file, root, file_path) if entry: self._index.add_entry(entry) count += 1 - self._index.version = int(time.time()) - log.info("Initial scan complete: %d files indexed", count) + return count + + def _report_collisions(self) -> None: + """ + Names that are the same file on a case-insensitive filesystem. + + Reported, never resolved: on ext4 both files exist and only the operator + knows which was meant. Left silent, the pair reaches somebody on Windows + who can save one of them. + """ + by_dir: dict[str, list[str]] = {} + for entry in self._index.entries: + by_dir.setdefault(fold(entry.path), []).append(entry.name) + for folded_dir, names in by_dir.items(): + for _, clashing in find_fold_collisions(names).items(): + log.warning( + "Names that differ only by case or accent form in %s: %s — " + "these are one file on NTFS or exFAT, and a member on Windows " + "can only keep one of them", + folded_dir or "/", ", ".join(sorted(clashing))) # ── Watchdog integration ────────────────────────────────────────────────── async def start(self) -> None: - """Start initial scan + filesystem watcher.""" + """Start initial scan + filesystem watcher + reconciler.""" self._loop = asyncio.get_event_loop() await self.initial_scan() + self._start_observer() + self._reconciler = asyncio.create_task(self._reconcile_loop()) + def _start_observer(self) -> None: handler = _WatchdogHandler(self) self._observer = Observer() - self._observer.schedule(handler, str(self.root), recursive=True) + watched = 0 + for root in self.roots: + if not root.available: + continue + try: + self._observer.schedule(handler, str(root.path), recursive=True) + watched += 1 + except OSError as e: + log.warning("Cannot watch root %r: %s", root.name, e) self._observer.start() - log.info("Watching %s for changes", self.root) + log.info("Watching %d root(s) for changes", watched) async def stop(self) -> None: - """Stop the filesystem watcher.""" + """Stop the filesystem watcher and the reconciler.""" + if self._reconciler: + self._reconciler.cancel() + try: + await self._reconciler + except asyncio.CancelledError: + pass + self._reconciler = None + for handle in self._pending_timers.values(): + handle.cancel() + self._pending_timers.clear() if self._observer: self._observer.stop() self._observer.join() @@ -167,6 +281,171 @@ class DirectoryIndexer: self._executor.shutdown(wait=False) log.info("Indexer stopped") + async def retarget(self, roots: RootSet) -> None: + """ + Point this indexer at a new set of roots, without a restart (14.8). + + Entries under a root that is gone from the config are dropped — the + 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. + """ + old_names = {r.folded for r in self.roots} + new_names = {r.folded for r in roots} + + for root in self.roots: + if root.folded not in new_names: + dropped = self._entries_under(root) + log.info("Root %r removed from the config — dropping %d entries", + root.name, len(dropped)) + for entry in dropped: + self._index.remove_entry(entry.id) + + 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) + + self._index.roots = roots.describe() + self._index.version = int(time.time()) + self._restart_observer() + if self.on_change: + await self.on_change(self) + + # ── Reconciliation ──────────────────────────────────────────────────────── + + async def _reconcile_loop(self) -> None: + while True: + try: + await asyncio.sleep(self.RECONCILE_SECS) + await self.reconcile() + except asyncio.CancelledError: + raise + except Exception: + log.exception("Reconcile failed — continuing") + + async def reconcile(self) -> None: + """ + Re-check availability, and rescan roots that came back. + + The only place a root's entries are dropped: when the root is readable + and the files are genuinely gone. A root that is not readable is left + untouched, which is the whole point. + """ + changed = self.roots.refresh_availability() + touched = False + + 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) + touched = True + else: + # Frozen: entries stay, marked unavailable to members through + # the roots table in the index payload. + log.warning("Root %r went away — %d entries frozen, not deleted", + root.name, len(self._entries_under(root))) + touched = True + + if changed: + self._restart_observer() + + if await self._sweep_available_roots(): + touched = True + + if touched: + self._index.roots = self.roots.describe() + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + + async def _sweep_available_roots(self) -> bool: + """ + Catch what the watcher missed: files gone, and files never announced. + + Only touches roots that are readable right now — a root whose volume is + absent has nothing to compare against, and comparing anyway is exactly + the mistake this module exists to avoid. + """ + loop = asyncio.get_event_loop() + changed = False + for root in self.roots: + if not root.available: + continue + try: + on_disk = {p.resolve() for p in root.path.rglob("*") + if _is_indexable(p)} + except OSError as e: + log.warning("Cannot reconcile root %r: %s", root.name, e) + continue + + known: dict[Path, str] = {} + for entry in self._entries_under(root): + abs_path = self._entry_path(root, entry) + if abs_path: + known[abs_path] = entry.id + + for missing in set(known) - on_disk: + # Duplicate content is handled without a special case here: the + # entry goes, and the add loop below re-indexes the surviving + # copy under its own path, because the id is then absent. A + # dedicated "find the survivor" lookup was written first and + # deleted — it rehashed every file under the root on any single + # deletion, and a test proved it changed nothing. + self._index.remove_entry(known[missing]) + log.info("Reconcile: %s is gone", missing) + changed = True + + for added in on_disk - set(known): + entry = await loop.run_in_executor( + self._executor, _scan_file, root, added) + if not entry: + continue + # The index is keyed by **content hash**, so two identical files + # at two paths are one entry and the path comparison above + # cannot see the second. Adding it anyway rewrites that entry's + # path every cycle, bumps the version, and pushes an index + # update to every connected peer once a minute — for ever. + # Measured on a live node: `clip.mp4` present at the root and in + # uploads/ with the same bytes. + if self._index.get_entry(entry.id) is not None: + log.debug("Reconcile: %s duplicates content already indexed " + "as %s — leaving the index alone", + added, entry.id[:8]) + continue + self._index.add_entry(entry) + log.info("Reconcile: %s appeared (missed event)", added) + changed = True + return changed + + def _entries_under(self, root: Root) -> list[IndexEntry]: + prefix = fold(root.name) + return [e for e in self._index.entries + if fold(e.path).split("/", 1)[0] == prefix] + + def _drop_root_entries(self, root: Root) -> None: + for entry in self._entries_under(root): + self._index.remove_entry(entry.id) + + @staticmethod + def _entry_path(root: Root, entry: IndexEntry) -> Path | None: + _, _, tail = entry.path.partition("/") + try: + return (root.path / tail / entry.name).resolve() if tail else \ + (root.path / entry.name).resolve() + except OSError: + return None + + def _restart_observer(self) -> None: + """Re-schedule watches after roots appeared or disappeared.""" + if self._observer: + self._observer.stop() + self._observer.join() + self._observer = None + self._start_observer() + # ── Internal update ─────────────────────────────────────────────────────── _DEBOUNCE_SECS = 2.0 @@ -175,39 +454,60 @@ class DirectoryIndexer: """Called from watchdog thread — schedule debounced async update.""" if not self._loop: return - key = str(file_path.resolve()) - self._loop.call_soon_threadsafe( - self._debounce, key, file_path, deleted) + key = str(file_path) + self._loop.call_soon_threadsafe(self._debounce, key, file_path, deleted) def _debounce(self, key: str, file_path: Path, deleted: bool) -> None: - if not hasattr(self, "_pending_timers"): - self._pending_timers: dict[str, asyncio.TimerHandle] = {} old = self._pending_timers.pop(key, None) if old: old.cancel() - handle = self._loop.call_later( - self._DEBOUNCE_SECS, - lambda: asyncio.ensure_future(self._update_entry(file_path, deleted)), - ) - self._pending_timers[key] = handle - def _remove_by_path(self, file_path: Path) -> None: - """Remove any existing entries that match this file's path + name.""" - resolved = file_path.resolve() - to_remove = [ - e.id for e in self._index.entries - if (self.root / e.path / e.name).resolve() == resolved - ] - for fid in to_remove: - self._index.remove_entry(fid) + def fire() -> None: + self._pending_timers.pop(key, None) + asyncio.ensure_future(self._update_entry(file_path, deleted)) + + self._pending_timers[key] = self._loop.call_later(self._DEBOUNCE_SECS, fire) + + def _remove_by_path(self, root: Root, file_path: Path) -> None: + """Remove any existing entries that point at this file.""" + try: + resolved = file_path.resolve() + except OSError: + resolved = file_path + for entry in self._entries_under(root): + if self._entry_path(root, entry) == resolved: + self._index.remove_entry(entry.id) async def _update_entry(self, file_path: Path, deleted: bool) -> None: - self._remove_by_path(file_path) + 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 not root.available: + return + + self._remove_by_path(root, file_path) if not deleted: loop = asyncio.get_event_loop() entry = await loop.run_in_executor( - self._executor, _scan_file, self.root, file_path) + self._executor, _scan_file, root, file_path) if entry: self._index.add_entry(entry) log.debug("Indexed: %s (%s, %d bytes)", |