diff options
Diffstat (limited to 'packages/meshbay-node/src')
9 files changed, 797 insertions, 72 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index c68f999..af85340 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -45,7 +45,7 @@ from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config from meshbay_node.roots import RootSet, RootError from meshbay_node.hub_client import HubClient, HubConfig -from meshbay_node.indexer import DirectoryIndexer +from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore from meshbay_node.roster import Roster from meshbay_node.transport import ( @@ -119,6 +119,17 @@ class NodeDaemon: self._denylist = ( Denylist(path=config.data_dir / "denylist.json") if Denylist else None) self._chat_stores: dict[str, ChatStore] = {} + self._index_caches: dict[str, IndexCache] = {} + # Coalesces a burst of index changes (one per debounced watchdog + # event) into a single broadcast — see _on_index_change. 0.5s is + # short enough nobody notices the wait, long enough that dropping a + # few hundred files into a watched folder produces one push instead + # of one per file. + self._broadcast_coalesce_secs = 0.5 + self._pending_broadcasts: dict[str, asyncio.TimerHandle] = {} + # group_id -> (version, {id: entry}) as of the last thing actually + # broadcast — the comparison point for the next delta. + self._last_broadcast_snapshot: dict[str, tuple] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None self._roster: Roster | None = None @@ -234,12 +245,31 @@ class NodeDaemon: log.info("No GEK yet for group %s — will accept first setup", group_cfg.name) + index_cache = IndexCache( + db_path=data_dir / group_cfg.id[:16] / "index_cache.db") + await index_cache.open() + self._index_caches[group_cfg.id] = index_cache + + # Read once at load, like member_upload/enabled_apps below — + # kept current in place afterwards by set_scan_settings + # (ops.py), which updates both this indexer object directly + # and roster.db, so a restart picks up the same values. + scan_settings = ( + await self._roster.scan_settings(group_cfg.id) + if self._roster else { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }) + indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=keys.sk_ed25519, gek=gek, on_change=self._on_index_change, + cache=index_cache, + reconcile_secs=scan_settings["reconcile_interval_secs"], + debounce_secs=scan_settings["debounce_secs"], ) await indexer.start(defer_scan=True) self._indexers.append(indexer) @@ -253,6 +283,20 @@ class NodeDaemon: "gek": gek, "roots": roots, "index": indexer.index, + # Live reference, mutated in place by the indexer itself + # (see IndexProgress in indexer.py) — read, never copied, + # by the handshake ack and the periodic progress pusher. + "progress": indexer.progress, + # Bound method, called when a peer completes the + # handshake — resets reconcile's backoff (indexer.py + # _reconcile_loop) so the backstop is prompt again now + # that someone is actually looking. + "note_activity": indexer.note_activity, + # Shown to the operator in Settings, and kept current in + # place by set_scan_settings (ops.py) — same reasoning as + # member_upload below. + "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], + "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, # Admission policy comes from node.toml, never from the hub: # a hub that could declare a group open would be handed its key. @@ -456,6 +500,8 @@ class NodeDaemon: if gctx: self._tasks.append(asyncio.create_task( _bg_scan(idx, group_cfg.name, gctx))) + self._tasks.append(asyncio.create_task( + self._progress_pusher(idx))) # 12. Wait for shutdown stop_event = asyncio.Event() @@ -561,18 +607,40 @@ class NodeDaemon: if gek: log.info("GEK loaded for new group %s", group_cfg.id[:8]) + index_cache = IndexCache( + db_path=data_dir / group_cfg.id[:16] / "index_cache.db") + await index_cache.open() + self._index_caches[group_cfg.id] = index_cache + + scan_settings = ( + await self._roster.scan_settings(group_cfg.id) + if self._roster else { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }) + indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=sk_ed, gek=gek, on_change=self._on_index_change, + cache=index_cache, + reconcile_secs=scan_settings["reconcile_interval_secs"], + debounce_secs=scan_settings["debounce_secs"], ) - await indexer.start() + # Registered *before* start() runs its (blocking, possibly very + # long — see the StarWars benchmark) initial scan, specifically + # so /api/groups/{id}/index-status can see indexer.progress + # while a brand-new group is still scanning — this is the one + # group state that must stay visible during the very window the + # group is not yet authorized for member connections (below). self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index self._state["indexers"][group_cfg.id] = indexer + await indexer.start() + data_dir.mkdir(parents=True, exist_ok=True) chat_db = data_dir / group_cfg.id[:16] / "chat.db" store = ChatStore(db_path=chat_db) @@ -583,6 +651,10 @@ class NodeDaemon: "gek": gek, "roots": roots, "index": indexer.index, + "progress": indexer.progress, + "note_activity": indexer.note_activity, + "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], + "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, "join_policy": group_cfg.join_policy, "member_upload": ( @@ -600,6 +672,11 @@ class NodeDaemon: log.info("Hot-loaded group %s (%s, %d roots)", group_cfg.name, group_cfg.id[:8], len(roots)) added_names.append(group_cfg.name) + # The initial scan above already ran to completion (indexer.start() + # is not deferred here), so this only matters for whatever scans + # this group as time goes on — a root added later, reconcile + # picking one back up. + self._tasks.append(asyncio.create_task(self._progress_pusher(indexer))) # ── Tear down removed groups ───────────────────────────────────── removed_names = [] @@ -618,6 +695,16 @@ class NodeDaemon: await store.close() except Exception: pass + cache = self._index_caches.pop(gid, None) + if cache: + try: + await cache.close() + except Exception: + pass + pending = self._pending_broadcasts.pop(gid, None) + if pending: + pending.cancel() + self._last_broadcast_snapshot.pop(gid, None) self._state["indexes"].pop(gid, None) self._state["indexers"].pop(gid, None) old_name = gid[:8] @@ -708,46 +795,149 @@ class NodeDaemon: log.warning("No unwrappable GEK bundle found for group %s", group_id[:8]) return None + async def _progress_pusher(self, indexer: DirectoryIndexer, + interval: float = 2.0) -> None: + """ + Watches indexer.progress and pushes a light INDEX_PROGRESS message to + this group's connected peers — never the index itself, that stays + _on_index_change's job. Runs for the node's whole lifetime: a scan + can start from several places (initial scan, a root added later, + reconcile picking a root back up), and this only needs to notice the + flag, not why it changed. + + The final push at the False transition is what lets a presence dot + reliably turn back off on an already-connected client — the + handshake ack only covers the moment of connecting. `interval` is a + parameter (not a bare constant) only so a test can drive this loop + without waiting on the real 2s cadence. + """ + was_scanning = False + while True: + await asyncio.sleep(interval) + progress = indexer.progress + now_scanning = progress.scanning + if now_scanning or was_scanning: + self._push_index_progress(indexer.group_id, progress) + was_scanning = now_scanning + + def _push_index_progress(self, group_id: str, progress) -> None: + if not self._webrtc: + return + msg = { + "type": MNP.INDEX_PROGRESS, + "v": MNP_VERSION, + "group_id": group_id, + "scanning": progress.scanning, + "scanned_bytes": progress.scanned_bytes, + "total_bytes": progress.total_bytes, + } + pushed = 0 + for session in list(self._webrtc._sessions.values()): + if session._group_id == group_id: + try: + session._send(msg) + pushed += 1 + except Exception: + pass + if pushed: + log.debug("Index progress pushed to %d peer(s) for group %s", + pushed, group_id[:8]) + async def _on_index_change(self, indexer: DirectoryIndexer) -> None: - """Called when a DirectoryIndexer detects file changes.""" + """ + Called when a DirectoryIndexer detects file changes — once per + debounced watchdog event, so dropping N files into a watched folder + calls this N times in quick succession. Coalesces those into one + broadcast (_broadcast_index_change) rather than one push per file: + the timer is reset on every call and only fires once calls stop + arriving for _broadcast_coalesce_secs. + """ + group_id = indexer.group_id + loop = asyncio.get_event_loop() + pending = self._pending_broadcasts.pop(group_id, None) + if pending: + pending.cancel() + + def fire() -> None: + self._pending_broadcasts.pop(group_id, None) + asyncio.ensure_future(self._broadcast_index_change(indexer)) + + self._pending_broadcasts[group_id] = loop.call_later( + self._broadcast_coalesce_secs, fire) + + async def _broadcast_index_change(self, indexer: DirectoryIndexer) -> None: + """ + The actual push, run once per coalesced burst. Sends a full + INDEX_SYNC the first time a group is ever broadcast (no previous + snapshot to diff against — the client's own first fetchIndex() call + already covers that case) and an INDEX_DELTA every time after, + computed against the last thing this method actually sent. + """ group_id = indexer.group_id idx = indexer.index log.info("Index changed for group %s: %d files (v%d)", group_id[:8], idx.count, idx.version) - # 11.5 — Push updated index to connected WebRTC peers in this group + prev = self._last_broadcast_snapshot.get(group_id) + delta = None + if prev is not None: + prev_version, prev_entries = prev + previous = GroupIndex._snapshot( + idx.group_id, idx.sk_node, idx.gek, prev_version, prev_entries) + delta = idx.diff(previous) + self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id()) + + # 11.5 — Push to connected WebRTC peers in this group if self._webrtc: - entries = [ - { - "id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at, + if delta is not None: + msg = { + "type": MNP.INDEX_DELTA, + "v": MNP_VERSION, + "group_id": idx.group_id, + "base_version": delta.base_version, + "version": delta.version, + "additions": [ + {"id": e.id, "name": e.name, "path": e.path, + "size": e.size, "type": e.type, "added_at": e.added_at} + for e in delta.additions + ], + "deletions": delta.deletions, + } + else: + msg = { + "type": MNP.INDEX_SYNC, + "v": MNP_VERSION, + "group_id": idx.group_id, + "version": idx.version, + "entries": [ + {"id": e.id, "name": e.name, "path": e.path, + "size": e.size, "type": e.type, "added_at": e.added_at} + for e in idx.entries + ], } - for e in idx.entries - ] - sync_msg = { - "type": MNP.INDEX_SYNC, - "v": MNP_VERSION, - "group_id": idx.group_id, - "version": idx.version, - "entries": entries, - } pushed = 0 for session in list(self._webrtc._sessions.values()): if session._group_id == group_id: try: - session._send(sync_msg) + session._send(msg) pushed += 1 except Exception: pass if pushed: - log.info("Index pushed to %d WebRTC peers", pushed) + log.info("Index %s pushed to %d WebRTC peers", + "delta" if delta is not None else "sync", pushed) # 11.9 — Register file hashes with hub swarm table (public groups only, H7) group_cfg = next( (g for g in self._config.groups if g.id == group_id), None) if (self._hub and self._state.get("endpoint_hint") and group_cfg and group_cfg.visibility == "public"): - hashes = [e.id for e in idx.entries] + # Only the newly added hashes once there is a delta to know them + # from — registering the whole library again on every change is + # the same O(changes x library size) cost the delta above exists + # to avoid. + hashes = ([e.id for e in delta.additions] if delta is not None + else [e.id for e in idx.entries]) if hashes: endpoint = f"webrtc:{self._config.node.quic_port}" asyncio.ensure_future(self._register_swarm(hashes, endpoint)) @@ -772,6 +962,10 @@ class NodeDaemon: log.info("Shutting down...") self._state["status"] = "stopping" + for handle in self._pending_broadcasts.values(): + handle.cancel() + self._pending_broadcasts.clear() + for task in self._tasks: task.cancel() for task in self._tasks: @@ -795,6 +989,9 @@ class NodeDaemon: for store in self._chat_stores.values(): await store.close() + for cache in self._index_caches.values(): + await cache.close() + for indexer in self._indexers: await indexer.stop() diff --git a/packages/meshbay-node/src/meshbay_node/indexer/__init__.py b/packages/meshbay-node/src/meshbay_node/indexer/__init__.py index bb6231b..c92c730 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/__init__.py @@ -1,5 +1,6 @@ """Directory indexer and Mesh Group Index.""" from .indexer import DirectoryIndexer from .group_index import GroupIndex +from .cache import IndexCache -__all__ = ["DirectoryIndexer", "GroupIndex"] +__all__ = ["DirectoryIndexer", "GroupIndex", "IndexCache"] diff --git a/packages/meshbay-node/src/meshbay_node/indexer/cache.py b/packages/meshbay-node/src/meshbay_node/indexer/cache.py new file mode 100644 index 0000000..c26ddf1 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/indexer/cache.py @@ -0,0 +1,95 @@ +""" +MeshBay Node — persistent (path, size, mtime) -> hash cache, one per group. + +Without this, every node restart re-reads and re-hashes every file in every +root, even when nothing changed — measured at 23 minutes for a 114 GB library +on a USB hard drive. This cache lets a scan skip the read entirely for a file +whose size and mtime still match what was hashed last time. + +It is a path-keyed accelerator only. The GroupIndex itself stays keyed by +content hash (see indexer.py's note on why two identical files are one +entry) — this cache never changes that, it only avoids recomputing a hash +that has not changed. +""" + +import logging +from dataclasses import dataclass +from pathlib import Path + +import aiosqlite + +log = logging.getLogger(__name__) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS files ( + path TEXT PRIMARY KEY, + mtime REAL NOT NULL, + size INTEGER NOT NULL, + hash TEXT NOT NULL, + type TEXT NOT NULL, + added_at INTEGER NOT NULL +); +""" + + +@dataclass +class CachedEntry: + hash: str + type: str + added_at: int + + +class IndexCache: + """Async SQLite (path, size, mtime) -> hash cache for one group.""" + + def __init__(self, db_path: Path): + self._db_path = db_path + self._db: aiosqlite.Connection | None = None + + async def open(self) -> None: + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._db = await aiosqlite.connect(str(self._db_path)) + await self._db.executescript(_SCHEMA) + await self._db.commit() + + async def close(self) -> None: + if self._db: + await self._db.close() + self._db = None + + async def __aenter__(self): + await self.open() + return self + + async def __aexit__(self, *_): + await self.close() + + async def lookup(self, path: str, size: int, mtime: float) -> CachedEntry | None: + """ + A cache hit requires an EXACT match on both size and mtime. A mtime + touched without a content change is a false negative (an unnecessary + rehash) — accepted, since the alternative (trusting a stale hash) is + a silent wrong answer instead of an occasional wasted read. + """ + async with self._db.execute( + "SELECT hash, type, added_at FROM files " + "WHERE path = ? AND size = ? AND mtime = ?", + (path, size, mtime)) as cur: + row = await cur.fetchone() + return CachedEntry(hash=row[0], type=row[1], added_at=row[2]) if row else None + + async def put(self, path: str, size: int, mtime: float, hash: str, + type: str, added_at: int) -> None: + """ + Written only once a file has been hashed in full — never partway + through — so a crash mid-hash leaves no stale/partial row behind: the + next scan simply finds no cache entry and hashes the file again. + """ + await self._db.execute( + "INSERT INTO files (path, mtime, size, hash, type, added_at) " + "VALUES (?, ?, ?, ?, ?, ?) " + "ON CONFLICT(path) DO UPDATE SET " + "mtime = excluded.mtime, size = excluded.size, hash = excluded.hash, " + "type = excluded.type, added_at = excluded.added_at", + (path, mtime, size, hash, type, added_at)) + await self._db.commit() 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 5dbdc5d..ec98667 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py @@ -86,6 +86,23 @@ class GroupIndex: def count(self) -> int: return len(self._entries) + def entries_by_id(self) -> dict: + """A snapshot copy, for diff() to compare a later version against — + see daemon.py _on_index_change, the only caller.""" + return dict(self._entries) + + @classmethod + def _snapshot(cls, group_id: str, sk_node: Ed25519PrivateKey, gek: bytes | None, + version: int, entries_by_id: dict) -> "GroupIndex": + """ + A lightweight stand-in for diff()'s `previous` argument — never + serialized or sent anywhere, just a comparison point built from an + earlier entries_by_id() snapshot rather than a live GroupIndex. + """ + idx = cls(group_id=group_id, sk_node=sk_node, gek=gek, version=version) + idx._entries = dict(entries_by_id) + return idx + # ── Serialisation ───────────────────────────────────────────────────────── def serialize(self) -> bytes: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 482b556..b479f7e 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -26,6 +26,7 @@ import asyncio import logging import time from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass from pathlib import Path from typing import Callable, Awaitable @@ -36,6 +37,7 @@ 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 @@ -75,6 +77,27 @@ def _is_indexable(path: Path) -> bool: _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"`. @@ -87,6 +110,17 @@ def _virtual_dir(root: Root, file_path: Path) -> str: 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.) @@ -136,8 +170,17 @@ class DirectoryIndexer: # 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 + # 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, @@ -146,20 +189,43 @@ class DirectoryIndexer: 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() - self._executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="indexer") + # 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: @@ -204,21 +270,77 @@ class DirectoryIndexer: async def _scan_root(self, root: Root) -> int: log.info("Scanning %s (root %r) ...", root.path, root.name) - loop = asyncio.get_event_loop() count = 0 + loop = asyncio.get_event_loop() try: - files = [p for p in root.path.rglob("*") if p.is_file()] + 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 - for file_path in files: - entry = await loop.run_in_executor( - self._executor, _scan_file, root, file_path) - if entry: - self._index.add_entry(entry) - count += 1 + + # 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. @@ -324,20 +446,38 @@ class DirectoryIndexer: async def _reconcile_loop(self) -> None: while True: try: - await asyncio.sleep(self.RECONCILE_SECS) - await self.reconcile() + 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") - async def reconcile(self) -> None: + 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 @@ -367,6 +507,8 @@ class DirectoryIndexer: 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. @@ -375,24 +517,18 @@ class DirectoryIndexer: 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 + loop = asyncio.get_event_loop() 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)} + 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 - 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 @@ -404,26 +540,46 @@ class DirectoryIndexer: 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 + 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]: @@ -444,6 +600,21 @@ class DirectoryIndexer: 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: @@ -454,8 +625,6 @@ class DirectoryIndexer: # ── Internal update ─────────────────────────────────────────────────────── - _DEBOUNCE_SECS = 2.0 - def _schedule_update(self, file_path: Path, deleted: bool = False) -> None: """Called from watchdog thread — schedule debounced async update.""" if not self._loop: @@ -472,7 +641,7 @@ class DirectoryIndexer: 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) + 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.""" @@ -511,9 +680,7 @@ class DirectoryIndexer: 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, root, file_path) + entry = await self._hash_or_cached(root, file_path) if entry: self._index.add_entry(entry) log.debug("Indexed: %s (%s, %d bytes)", diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index c9d862a..154813d 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -22,6 +22,7 @@ in the adapter. from __future__ import annotations +import asyncio import logging import re from dataclasses import asdict @@ -730,12 +731,69 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict: return {"apps": apps, "group_id": group_id} +# ── Scan settings ──────────────────────────────────────────────────────────── + +async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs: float, + debounce_secs: float) -> dict: + """ + How often the indexer's reconciliation backstop runs, and how long a + changed file is left alone before being hashed (indexer.py + DirectoryIndexer). Persisted like set_member_upload/set_enabled_apps — + but there is also a *live* DirectoryIndexer object to update, since it + reads these once at construction and runs its own background loop with + them rather than consulting groups_ctx on every use. + """ + roster = _roster(state) + await roster.set_scan_settings(group_id, reconcile_interval_secs, debounce_secs, + set_by=state.get("node_user_id", "")) + indexer = state.get("indexers", {}).get(group_id) + if indexer: + indexer.reconcile_secs = reconcile_interval_secs + indexer.debounce_secs = debounce_secs + # Apply the new interval now rather than after whatever backoff had + # already stretched the wait to. + indexer.note_activity() + # Optional, unlike _group_ctx(): a group can be persisted here before it + # is hot-loaded (or in a test that only cares about the roster/indexer + # side), and that must not turn a successful write into a 404. + ctx = state.get("groups_ctx", {}).get(group_id) + if ctx is not None: + ctx["reconcile_interval_secs"] = reconcile_interval_secs + ctx["debounce_secs"] = debounce_secs + log.info("Scan settings for group %s: reconcile=%.0fs debounce=%.0fs", + group_id[:8], reconcile_interval_secs, debounce_secs) + return {"reconcile_interval_secs": reconcile_interval_secs, + "debounce_secs": debounce_secs, "group_id": group_id} + + # ── Reload ────────────────────────────────────────────────────────────────── async def reload_config(state: dict) -> dict: - """Hot-reload node.toml without dropping connections.""" + """Hot-reload node.toml without dropping connections. Blocks until the + reload actually finishes — see start_reload for why the loopback route + uses that instead.""" reload_fn = state.get("reload_fn") if not reload_fn: raise OpError("Reload not available", status=503) await reload_fn() return {"status": "reloaded"} + + +async def start_reload(state: dict) -> dict: + """ + Same as reload_config, but does not wait for the reload to finish. + + The loopback route uses this one: the Electron bridge caps every call at + a fixed 30s (main.js node:call), and hot-loading a brand-new group runs + its full initial scan synchronously inside _reload_config_inner() + (daemon.py) before that coroutine returns — minutes, not seconds, on a + real library (found against a 45 GB group on the same slow disk the + StarWars benchmark used). The reload keeps running on the daemon's own + event loop either way; add_root/remove_root below already fire it the + same way for exactly this reason. + """ + reload_fn = state.get("reload_fn") + if not reload_fn: + raise OpError("Reload not available", status=503) + asyncio.ensure_future(reload_fn()) + return {"status": "reloading"} diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index 1cc8cae..6eb7651 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -599,6 +599,39 @@ class Roster: json.dumps(sorted(apps)), set_by) return apps + # How often the indexer's reconciliation backstop runs, and how long it + # waits after the last change on a file before hashing it. Unset means + # the indexer's own defaults — an existing group's behaviour must not + # change because a node was upgraded. See indexer.py DirectoryIndexer + # for what these actually do and why the defaults are what they are. + SETTING_RECONCILE_INTERVAL = "reconcile_interval_secs" + SETTING_DEBOUNCE_SECS = "debounce_secs" + + async def scan_settings(self, group_id: str) -> dict: + # Imported here, not at module load: roster.py is loaded before the + # indexer package during startup, and this is the only place the two + # need each other's names. + from meshbay_node.indexer.indexer import DirectoryIndexer + + reconcile = await self.get_setting(group_id, self.SETTING_RECONCILE_INTERVAL) + debounce = await self.get_setting(group_id, self.SETTING_DEBOUNCE_SECS) + return { + "reconcile_interval_secs": ( + float(reconcile) if reconcile is not None + else DirectoryIndexer.DEFAULT_RECONCILE_SECS), + "debounce_secs": ( + float(debounce) if debounce is not None + else DirectoryIndexer.DEFAULT_DEBOUNCE_SECS), + } + + async def set_scan_settings(self, group_id: str, reconcile_interval_secs: float, + debounce_secs: float, set_by: str = "") -> dict: + await self.set_setting(group_id, self.SETTING_RECONCILE_INTERVAL, + str(float(reconcile_interval_secs)), set_by) + await self.set_setting(group_id, self.SETTING_DEBOUNCE_SECS, + str(float(debounce_secs)), set_by) + return await self.scan_settings(group_id) + async def create_invite( self, group_id: str, diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index b6f572a..fa6c3e9 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -63,6 +63,7 @@ from meshbay_common.adminop import ( OP_MEMBER_UNPIN, OP_MEMBER_UPLOAD, OP_APPS_ENABLED, + OP_SET_SCAN_SETTINGS, OP_ROOT_ADD, OP_ROOT_REMOVE, OP_GROUP_ATTACH, @@ -85,6 +86,7 @@ from meshbay_common.join import ( from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex +from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import ops from meshbay_node.roots import ( RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, @@ -409,6 +411,8 @@ class WebRTCPeerSession: self._do_member_upload(msg) elif mtype == MNP.APPS_ENABLED: self._do_apps_enabled(msg) + elif mtype == MNP.SET_SCAN_SETTINGS: + self._do_set_scan_settings(msg) elif mtype == MNP.MEMBER_UNPIN: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: @@ -659,6 +663,20 @@ class WebRTCPeerSession: # setting (or one whose context has not loaded it yet) hides # nothing. "enabled_apps": list(self._group_ctx().get("enabled_apps") or []), + # So a client that connects mid-scan shows the indexing state + # immediately, instead of waiting for the next periodic + # INDEX_PROGRESS push. Never a path or filename — see + # IndexProgress in indexer.py. + "indexing": self._indexing_status(), + # Current values only — not enforced from here, just shown to + # the operator in Settings so the number on screen matches what + # the indexer is actually doing (set_scan_settings, ops.py). + "scan_settings": { + "reconcile_interval_secs": self._group_ctx().get( + "reconcile_interval_secs", DirectoryIndexer.DEFAULT_RECONCILE_SECS), + "debounce_secs": self._group_ctx().get( + "debounce_secs", DirectoryIndexer.DEFAULT_DEBOUNCE_SECS), + }, } if node_user_id: ack["node_user_id"] = node_user_id @@ -668,6 +686,13 @@ class WebRTCPeerSession: self._send(ack) self._audit("handshake") + # Someone is here now — reconcile's backstop should be prompt again + # rather than however far its backoff had stretched while nobody + # was connected (indexer.py DirectoryIndexer.note_activity). + note_activity = self._group_ctx().get("note_activity") + if note_activity: + note_activity() + async def _do_gek_bundle_fetch(self) -> None: """Serve the caller's wrapped GEK bundle during the handshake window.""" bundle_store = self._ctx.get("bundle_store") @@ -1648,6 +1673,69 @@ class WebRTCPeerSession: except Exception: pass + # Reconcile's backstop and the watchdog debounce (indexer.py + # DirectoryIndexer) — how hard the node works on the operator's own + # disk, not a member-facing permission. Signed for the same reason as + # apps_enabled: consistency of the authorization model, not because a + # wrong value here is itself dangerous. + MIN_RECONCILE_SECS = 10.0 + MAX_RECONCILE_SECS = 24 * 3600.0 + MIN_DEBOUNCE_SECS = 0.0 + MAX_DEBOUNCE_SECS = 300.0 + + def _do_set_scan_settings(self, msg: dict) -> None: + try: + reconcile = float(msg.get("reconcile_interval_secs")) + debounce = float(msg.get("debounce_secs")) + except (TypeError, ValueError): + self._send({"type": "error", "detail": "Invalid scan settings"}) + return + if not (self.MIN_RECONCILE_SECS <= reconcile <= self.MAX_RECONCILE_SECS): + self._send({"type": "error", + "detail": f"reconcile_interval_secs must be between " + f"{self.MIN_RECONCILE_SECS:.0f} and " + f"{self.MAX_RECONCILE_SECS:.0f}"}) + return + if not (self.MIN_DEBOUNCE_SECS <= debounce <= self.MAX_DEBOUNCE_SECS): + self._send({"type": "error", + "detail": f"debounce_secs must be between " + f"{self.MIN_DEBOUNCE_SECS:.0f} and " + f"{self.MAX_DEBOUNCE_SECS:.0f}"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge( + OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}") + + async def _admin_exec_set_scan_settings( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + try: + reconcile_s, debounce_s = pending["subject"].split(",") + reconcile, debounce = float(reconcile_s), float(debounce_s) + except (ValueError, KeyError): + self._send({"type": "error", "detail": "Invalid scan settings"}) + return + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"set_scan_settings:{pending['subject']}") + return + try: + result = await self._run_op( + ops.set_scan_settings, self._group_id or "", reconcile, debounce) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("set_scan_settings", pending["subject"]) + + notice = {"type": MNP.SET_SCAN_SETTINGS_ACK, "v": MNP_VERSION, **result} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + # ── Node management (D5) ───────────────────────────────────────────────── async def _do_node_status(self, msg: dict) -> None: @@ -2012,6 +2100,23 @@ class WebRTCPeerSession: return self._ctx["groups"][self._group_id] return self._ctx + def _indexing_status(self) -> dict: + """ + {"scanning": bool, "scanned_bytes": int, "total_bytes": int} for the + handshake ack and INDEX_PROGRESS pushes — never a path or filename, + that stays local to the operator's own admin UI. Absent "progress" + (context not loaded, or a group with no indexer at all) reads as + idle rather than erroring. + """ + progress = self._group_ctx().get("progress") + if progress is None: + return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0} + return { + "scanning": progress.scanning, + "scanned_bytes": progress.scanned_bytes, + "total_bytes": progress.total_bytes, + } + def _peer_registry(self) -> dict: """ Connected peers for THIS group only. @@ -2626,6 +2731,9 @@ class WebRTCPeerSession: elif pending["op"] == OP_APPS_ENABLED: self._spawn( self._admin_exec_apps_enabled(pending, transcript, sig_bytes)) + elif pending["op"] == OP_SET_SCAN_SETTINGS: + self._spawn( + self._admin_exec_set_scan_settings(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_ADD: self._spawn( self._admin_exec_root_add(pending, transcript, sig_bytes)) diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index b505f25..d5b3f94 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -26,6 +26,7 @@ from fastapi.responses import HTMLResponse, JSONResponse from meshbay_node import __version__ from meshbay_node import ops from meshbay_node.config import DEFAULT_CONFIG_PATH +from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_common.crypto import generate_gek, wrap_gek_aes from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR @@ -319,6 +320,39 @@ def create_ui_app(state: dict) -> FastAPI: asyncio.ensure_future(reload_fn()) return result + # asyncio.ensure_future above schedules the reload (and whatever initial + # scan it triggers) on the daemon's own event loop — it has no link to + # this HTTP request or to any browser tab. Closing the client that made + # this call does not cancel it: the scan is the node's own background + # work, not something borrowed from the request that started it. + + @app.get("/api/groups/{group_id}/index-status") + async def index_status(group_id: str): + """ + Polled by the Create Group wizard and by "add a directory" in + Settings — the same source either way, since both just start a scan + on this group's indexer. `current_dir` is a basename only, and is + never sent over MNP (see IndexProgress in indexer.py) — this route + is loopback-only, for the operator's own screen. + + Reads state["indexers"] rather than groups_ctx: a brand-new group is + registered there before its (possibly long) initial scan runs, but + is only added to groups_ctx once that scan finishes (it is not yet + authorized for member connections either way — see _reload_config) + — this is precisely the window the wizard needs to watch. + """ + indexer = state.get("indexers", {}).get(group_id) + progress = indexer.progress if indexer else None + if progress is None: + return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0, + "current_dir": ""} + return { + "scanning": progress.scanning, + "scanned_bytes": progress.scanned_bytes, + "total_bytes": progress.total_bytes, + "current_dir": progress.current_dir, + } + # ── Upload toggle (operator only, localhost) ───────────────────────── @app.put("/api/groups/{group_id}/member-upload") @@ -327,11 +361,26 @@ def create_ui_app(state: dict) -> FastAPI: state, group_id, bool(payload.get("allowed", False)), )) + # ── Scan settings (operator only, localhost) ────────────────────────── + + @app.put("/api/groups/{group_id}/scan-settings") + async def set_scan_settings(group_id: str, payload: dict): + return await _op(lambda: ops.set_scan_settings( + state, group_id, + float(payload.get("reconcile_interval_secs", + DirectoryIndexer.DEFAULT_RECONCILE_SECS)), + float(payload.get("debounce_secs", + DirectoryIndexer.DEFAULT_DEBOUNCE_SECS)), + )) + # ── Reload config ──────────────────────────────────────────────────── @app.post("/api/reload") async def reload_config(): - return await _op(lambda: ops.reload_config(state)) + # start_reload, not reload_config: this must return before a + # brand-new group's synchronous initial scan finishes (minutes, not + # seconds, on a real library) — see ops.start_reload for why. + return await _op(lambda: ops.start_reload(state)) # ── Chat endpoints ─────────────────────────────────────────────────────── |