""" Directory indexer — watches a group's roots and maintains a GroupIndex. 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 time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from pathlib import Path from typing import Callable, Awaitable import blake3 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.cache import IndexCache from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roots import Root, RootSet log = logging.getLogger(__name__) # File types we include in the index (skip hidden files, temp files, etc.) EXCLUDED_PREFIXES = (".", "~", "#") EXCLUDED_SUFFIXES = (".tmp", ".part", ".crdownload", ".download") MEDIA_EXTENSIONS = { "video": {".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv", ".webm", ".m4v"}, "audio": {".mp3", ".flac", ".ogg", ".wav", ".aac", ".m4a", ".opus"}, "image": {".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".bmp", ".tiff"}, "document": {".pdf", ".epub", ".mobi", ".txt", ".md", ".docx", ".odt"}, "archive": {".zip", ".tar", ".gz", ".bz2", ".xz", ".7z", ".rar"}, } def _detect_type(path: Path) -> str: suffix = path.suffix.lower() for ftype, exts in MEDIA_EXTENSIONS.items(): if suffix in exts: return ftype return "other" def _is_indexable(path: Path) -> bool: if not path.is_file(): return False name = path.name return ( not any(name.startswith(p) for p in EXCLUDED_PREFIXES) and not any(name.endswith(s) for s in EXCLUDED_SUFFIXES) ) _HASH_CHUNK = 8 * 1024 * 1024 # 8 MB streaming hash chunks @dataclass class IndexProgress: """ A snapshot of "is this indexer mid-scan right now", for the status shown to the operator (Create Group wizard, adding a directory) and pushed to connected members (a presence dot, never anything more specific — see daemon.py/webrtc_server.py). Reset per _scan_root() call rather than accumulated across a group's roots: the consumers that matter always watch exactly one root being scanned. Mutated only from the asyncio loop thread (the hashing itself runs in an executor thread, but never touches this), so no lock is needed. """ scanning: bool = False scanned_bytes: int = 0 total_bytes: int = 0 # Basename only, deliberately not the full path — enough to show progress # without broadcasting the operator's directory structure. current_dir: str = "" 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 _walk_root(root: Root) -> list[Path]: """ Blocking directory walk — always run via an executor, never awaited directly in the asyncio loop. A tree of tens of thousands of files (or one on a slow network share) can take seconds; run inline, that stalls every other thing the daemon is doing — WebRTC sessions, chat, the admin UI — for as long as it takes. """ return [p for p in root.path.rglob("*") if p.is_file()] 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.""" if not _is_indexable(file_path): return None try: stat = file_path.stat() hasher = blake3.blake3() # 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) return IndexEntry( 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=_virtual_dir(root, file_path), size=stat.st_size, type=_detect_type(file_path), added_at=int(stat.st_mtime), ) except (OSError, PermissionError, ValueError) as e: log.warning("Cannot index %s: %s", file_path, e) return None class DirectoryIndexer: """ Watches a group's roots and keeps a GroupIndex up to date. Usage: indexer = DirectoryIndexer( roots=RootSet.build([{"path": "/home/user/shared", "upload": True}]), group_id="my-group", sk_node=sk, gek=gek_bytes, on_change=async_callback, ) await indexer.start() # ... later 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. Watchdog already covers the common case in real time, so this # does not need to run often to do its job; it backs off further still # (see _reconcile_loop) when nothing has changed for a while, and a # per-group operator setting (roster.py SETTING_RECONCILE_INTERVAL) can # override the starting point. DEFAULT_RECONCILE_SECS = 600.0 # 10 min RECONCILE_BACKOFF_CAP = 7200.0 # 2 h — never sleeps longer than this # How long to wait after the last event on a given path before acting on # it — several writes to the same file in quick succession (a slow copy # in several passes) collapse into one hash instead of one per write. DEFAULT_DEBOUNCE_SECS = 2.0 def __init__( self, roots: RootSet, group_id: str, sk_node: Ed25519PrivateKey, gek: bytes | None, on_change: Callable[["DirectoryIndexer"], Awaitable[None]] | None = None, cache: IndexCache | None = None, reconcile_secs: float = DEFAULT_RECONCILE_SECS, debounce_secs: float = DEFAULT_DEBOUNCE_SECS, ): self.roots = roots self.group_id = group_id self.sk_node = sk_node self.gek = gek self.on_change = on_change self.reconcile_secs = reconcile_secs self.debounce_secs = debounce_secs # Current backoff delay — starts at reconcile_secs, doubles on every # tick that finds nothing changed (up to RECONCILE_BACKOFF_CAP), and # resets the moment something real happens (a change, or a peer # connecting — see note_activity()). self._reconcile_delay = reconcile_secs # Path -> (size, mtime, hash) accelerator, so a restart does not have # to re-read a file it already hashed last time (see cache.py). None # in tests that do not care about it — every hash is then a miss. self._cache = cache self._index = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek) self._index.roots = roots.describe() # One worker, deliberately, not a real pool: hashing two files at once # buys nothing here and can cost a lot. It only offloads the blocking # read+hash off the asyncio loop; it was never used for concurrency — # every call site awaits one run_in_executor before starting the next # (see _hash_or_cached below) — and measured on a spinning USB drive, # two interleaved multi-GB reads would seek-thrash against each other # rather than go faster. Left at 1 so the number does not promise a # concurrency this code never provided. self._executor = ThreadPoolExecutor(max_workers=1, 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] = {} self.progress = IndexProgress() @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 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) count = 0 loop = asyncio.get_event_loop() try: files = await loop.run_in_executor(self._executor, _walk_root, root) except OSError as e: log.warning("Cannot scan root %r: %s", root.name, e) return 0 # Sizes up front, off the same listing that already walked the tree — # the progress bar's denominator, not a second pass over the disk. sized: list[tuple[Path, int]] = [] for p in files: try: sized.append((p, p.stat().st_size)) except OSError: continue self.progress.scanning = True self.progress.scanned_bytes = 0 self.progress.total_bytes = sum(size for _, size in sized) self.progress.current_dir = "" try: for file_path, size in sized: self.progress.current_dir = file_path.parent.name entry = await self._hash_or_cached(root, file_path) self.progress.scanned_bytes += size if entry: self._index.add_entry(entry) count += 1 finally: # 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 = "" return count async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None: """ Cache-aware replacement for a bare _scan_file() call: skips the content read entirely when this path's (size, mtime) still match what was hashed last time — the difference between a redundant full rehash of a 100+ GB library on every restart and a stat()-only pass. The only place that decides to actually read a file's bytes. """ if not _is_indexable(file_path): return None try: st = file_path.stat() except OSError: return None if self._cache is not None: cached = await self._cache.lookup(str(file_path), st.st_size, st.st_mtime) if cached is not None: return IndexEntry( id=cached.hash, name=file_path.name, path=_virtual_dir(root, file_path), size=st.st_size, type=cached.type, added_at=cached.added_at, ) loop = asyncio.get_event_loop() entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path) if entry and self._cache is not None: await self._cache.put(str(file_path), st.st_size, st.st_mtime, entry.id, entry.type, entry.added_at) return entry 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, *, defer_scan: bool = False) -> None: """Start initial scan + filesystem watcher + reconciler. With ``defer_scan=True`` the watcher and reconciler start immediately but the initial scan is skipped — call :meth:`initial_scan` yourself when ready. """ self._loop = asyncio.get_event_loop() if not defer_scan: 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() 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 %d root(s) for changes", watched) async def stop(self) -> None: """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() self._observer = None 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_delay) changed = await self.reconcile() if changed: self._reconcile_delay = self.reconcile_secs else: self._reconcile_delay = min( self._reconcile_delay * 2, self.RECONCILE_BACKOFF_CAP) except asyncio.CancelledError: raise except Exception: log.exception("Reconcile failed — continuing") def note_activity(self) -> None: """ Called when something makes a prompt reconcile worth having again — today, a peer completing the handshake for this group (webrtc_server.py). Someone is looking, so the backstop should be at its normal cadence rather than however far backoff had stretched it. """ self._reconcile_delay = self.reconcile_secs async def reconcile(self) -> bool: """ 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. Returns whether anything actually changed — _reconcile_loop uses this 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. """ 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) return touched 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. """ changed = False loop = asyncio.get_event_loop() for root in self.roots: if not root.available: continue try: on_disk, known = await loop.run_in_executor( self._executor, self._sweep_scan_root, root) except OSError as e: log.warning("Cannot reconcile root %r: %s", root.name, e) continue 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 added_paths = on_disk - set(known) if not added_paths: continue added_sized: list[tuple[Path, int]] = [] for p in added_paths: try: added_sized.append((p, p.stat().st_size)) 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 = "" try: for added, size in added_sized: self.progress.current_dir = added.parent.name entry = await self._hash_or_cached(root, added) self.progress.scanned_bytes += size 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 finally: self.progress.scanning = False self.progress.current_dir = "" 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 _sweep_scan_root(self, root: Root) -> tuple[set[Path], dict[Path, str]]: """ Blocking: the two disk-touching pieces of one reconcile pass for a root, bundled so both run together in the executor rather than in the asyncio loop — the tree walk, and resolving the real path of every entry already known under this root (one syscall each). """ on_disk = {p.resolve() for p in root.path.rglob("*") if _is_indexable(p)} 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 return on_disk, known 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 ─────────────────────────────────────────────────────── def _schedule_update(self, file_path: Path, deleted: bool = False) -> None: """Called from watchdog thread — schedule debounced async update.""" if not self._loop: return 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: old = self._pending_timers.pop(key, None) if old: old.cancel() 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: 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: 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) class _WatchdogHandler(FileSystemEventHandler): def __init__(self, indexer: DirectoryIndexer): self._indexer = indexer def on_created(self, event: FileSystemEvent): if not event.is_directory: self._indexer._schedule_update(Path(event.src_path)) def on_modified(self, event: FileSystemEvent): if not event.is_directory: self._indexer._schedule_update(Path(event.src_path)) def on_deleted(self, event: FileSystemEvent): if not event.is_directory: self._indexer._schedule_update(Path(event.src_path), deleted=True) def on_moved(self, event: FileSystemEvent): if not event.is_directory: self._indexer._schedule_update(Path(event.src_path), deleted=True) self._indexer._schedule_update(Path(event.dest_path))