diff options
Diffstat (limited to 'packages/meshbay-node')
16 files changed, 2036 insertions, 73 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 ─────────────────────────────────────────────────────── diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py index ab1613d..b367a20 100644 --- a/packages/meshbay-node/tests/test_daemon.py +++ b/packages/meshbay-node/tests/test_daemon.py @@ -237,6 +237,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 # real value would make this test wait 0.5s daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=2) daemon._state["endpoint_hint"] = "node123" @@ -256,6 +257,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu daemon._webrtc = mock_webrtc await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) # let the coalescing timer fire mock_session._send.assert_called_once() msg = mock_session._send.call_args[0][0] @@ -288,6 +290,7 @@ async def test_daemon_index_change_registers_swarm_for_public_group( data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=2) daemon._state["endpoint_hint"] = "node123" @@ -317,6 +320,7 @@ async def test_daemon_index_change_skips_other_group_peers( data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=0) daemon._state["endpoint_hint"] = "node123" @@ -340,6 +344,145 @@ async def test_daemon_index_change_skips_other_group_peers( daemon._webrtc = mock_webrtc await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) same_group._send.assert_called_once() other_group._send.assert_not_called() + + +# ── INDEX_DELTA (phase 4) ──────────────────────────────────────────────────── + +def _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32, + visibility="private"): + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), + groups=[GroupConfig( + id=group_id, name="test-group", shared_dir=str(shared_dir), + visibility=visibility, quic_port=29010, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 + daemon._hub = AsyncMock() + daemon._hub.register_swarm = AsyncMock(return_value=0) + daemon._state["endpoint_hint"] = "node123" + return daemon + + +@pytest.mark.asyncio +async def test_first_broadcast_is_full_sync_second_is_delta(tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + first = session._send.call_args_list[0].args[0] + assert first["type"] == "index_sync" + assert len(first["entries"]) == indexer.index.count + + # Nothing actually changed in the index between the two calls, but + # _on_index_change does not know or care why it was called — the + # SECOND broadcast must still be a delta, now that there is a + # previous snapshot to diff against. + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + second = session._send.call_args_list[1].args[0] + assert second["type"] == "index_delta" + assert second["additions"] == [] + assert second["deletions"] == [] + + +@pytest.mark.asyncio +async def test_delta_reflects_additions_and_deletions(tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + removed_id = indexer.index.entries[0].id + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + daemon._webrtc = MagicMock() + daemon._webrtc._sessions = {"p1": session} + + await daemon._on_index_change(indexer) # first: full sync, establishes the snapshot + await asyncio.sleep(0.05) + + # A real change: one entry removed, one added. + indexer.index.remove_entry(removed_id) + from meshbay_common.protocol import IndexEntry + new_entry = IndexEntry(id="new-file-id", name="new.mp4", path="shared", + size=10, type="video", added_at=0) + indexer.index.add_entry(new_entry) + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + delta_msg = session._send.call_args_list[1].args[0] + assert delta_msg["type"] == "index_delta" + assert delta_msg["deletions"] == [removed_id] + assert [a["id"] for a in delta_msg["additions"]] == ["new-file-id"] + + +@pytest.mark.asyncio +async def test_a_burst_of_changes_produces_one_broadcast(tmp_path, shared_dir, gek): + """Coalescing: several _on_index_change calls in quick succession (one + per debounced watchdog event) must collapse into a single push.""" + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + daemon._webrtc = MagicMock() + daemon._webrtc._sessions = {"p1": session} + + for _ in range(5): + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + session._send.assert_called_once() + + +@pytest.mark.asyncio +async def test_swarm_registration_only_sends_new_hashes_after_the_first( + tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, visibility="public") + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + total_files = indexer.index.count + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + assert len(daemon._hub.register_swarm.call_args_list[0].args[0]) == total_files + + from meshbay_common.protocol import IndexEntry + indexer.index.add_entry(IndexEntry(id="new-file-id", name="new.mp4", + path="shared", size=10, type="video", + added_at=0)) + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + assert daemon._hub.register_swarm.call_count == 2 + assert daemon._hub.register_swarm.call_args_list[1].args[0] == ["new-file-id"], \ + "only the newly added hash must be (re-)registered, not the whole library" diff --git a/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py new file mode 100644 index 0000000..7cb74cb --- /dev/null +++ b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py @@ -0,0 +1,329 @@ +""" +Adding a group (Create Group wizard, or "add a directory" to an existing +one) fires `_reload_config()` without awaiting it (`asyncio.ensure_future`, +ui/app.py) — the request handler, and whatever browser tab triggered it, +return immediately. This is deliberate: the initial scan behind it can take +a very long time (measured at 23 minutes for a 114 GB library on a slow +disk), and none of that work belongs to the HTTP request or the WebRTC +session that happened to start it. + +This test proves the scan is genuinely independent of its caller: it starts +the reload the same way the real endpoint does — schedules it and does not +await it, standing in for "the browser tab that made the call was closed" — +then does something else, and only afterwards checks that the reload +finished and the new group became available on its own. +""" + +import asyncio +import base64 +import os + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from unittest.mock import AsyncMock, MagicMock, patch + +from meshbay_node import ops +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +import meshbay_node.indexer.indexer as indexer_mod + + +def _free_port() -> int: + import socket + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _toml(data_dir, first_group_dir, second_group_id=None, second_group_dir=None) -> str: + # data_dir MUST come before any [section] header — TOML has no notion of + # "back to top-level" once a table is open, so a bare `key = value` line + # placed after [node] becomes node.data_dir, not the top-level data_dir + # load_config() actually reads. Silently falls back to the real default + # (~/.local/share/meshbay) instead of erroring, which is exactly how this + # test once ran a whole daemon — including _shutdown()'s unlink of + # ui-token — against the developer's real, already-running node. + text = f""" +data_dir = "{data_dir}" + +[hub] +url = "http://localhost:9999" +username = "testuser" + +[node] +quic_port = {_free_port()} +ui_port = {_free_port()} + +[[groups]] +id = "{"a" * 32}" +name = "first" +shared_dir = "{first_group_dir}" +visibility = "private" +""" + if second_group_id: + text += f""" +[[groups]] +id = "{second_group_id}" +name = "slow-new-group" +shared_dir = "{second_group_dir}" +visibility = "private" +""" + return text + + +def _mock_keystore_keys(sk_ed): + sk_x = X25519PrivateKey.generate() + pk_x_raw = sk_x.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + mock_keys = MagicMock() + mock_keys.sk_ed25519 = sk_ed + mock_keys.pk_ed25519_b64 = "test" + mock_keys.sk_x25519 = sk_x + mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode() + return mock_keys + + +@pytest.mark.asyncio +async def test_hot_loaded_group_finishes_scanning_without_anyone_awaiting_the_reload( + tmp_path): + first_dir = tmp_path / "first" + first_dir.mkdir() + (first_dir / "readme.txt").write_bytes(b"hello") + + second_dir = tmp_path / "second" + second_dir.mkdir() + for i in range(3): + (second_dir / f"file{i}.bin").write_bytes(os.urandom(64)) + second_group_id = "b" * 32 + + data_dir = tmp_path / "data" + config_path = tmp_path / "node.toml" + config_path.write_text(_toml(data_dir, first_dir)) + + from meshbay_node.config import load_config + daemon = NodeDaemon(load_config(config_path), config_path=config_path) + + sk_node = Ed25519PrivateKey.generate() + mock_keys = _mock_keystore_keys(sk_node) + mock_session = MagicMock() + mock_session.node_id = "node123" + mock_session.user_id = "user123" + mock_session.hub_pk_pem = sk_node.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + # Slow the second group's hashing down (a stand-in for a large/slow + # library) so there is a real window in which "nobody is awaiting this" + # actually means something, without needing a genuinely huge file. + real_scan_file = indexer_mod._scan_file + + def slow_scan_file(root, path): + import time + time.sleep(0.15) + return real_scan_file(root, path) + + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ + patch("meshbay_node.daemon.HubClient") as MockHub, \ + patch.object(indexer_mod, "_scan_file", slow_scan_file): + + hub_instance = AsyncMock() + hub_instance.startup = AsyncMock(return_value=mock_session) + hub_instance.send_ws = AsyncMock() + hub_instance._ws = None + hub_instance.close = AsyncMock() + hub_instance.__aenter__ = AsyncMock(return_value=hub_instance) + hub_instance.__aexit__ = AsyncMock(return_value=False) + MockHub.return_value = hub_instance + + shutdown_event = asyncio.Event() + + async def mock_maintain_ws(**kwargs): + await shutdown_event.wait() + hub_instance.maintain_ws = mock_maintain_ws + + async def run_daemon(): + with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15): + try: + await asyncio.wait_for(daemon.run(), timeout=15) + except (asyncio.TimeoutError, Exception): + pass + + run_task = asyncio.create_task(run_daemon()) + try: + for _ in range(50): + if daemon._state.get("status") == "running": + break + await asyncio.sleep(0.05) + assert daemon._state["status"] == "running" + assert second_group_id not in daemon._state.get("groups_ctx", {}) + + # Add the second group to the config on disk, the way the wizard's + # attach + /api/reload would leave it, then fire the reload exactly + # as ui/app.py does: scheduled, NOT awaited. + config_path.write_text(_toml(data_dir, first_dir, + second_group_id, second_dir)) + reload_task = asyncio.ensure_future(daemon._reload_config()) + + # Stand in for "the browser tab is gone": do something completely + # unrelated to the reload, and explicitly do not await it here. + await asyncio.sleep(0.01) + assert second_group_id not in daemon._state.get("groups_ctx", {}), \ + "the scan (3 files x 0.15s) cannot have finished yet" + + # Only now catch up with the background work, from a place that + # has no relationship to whatever originally triggered it. + await asyncio.wait_for(reload_task, timeout=5) + + assert second_group_id in daemon._state["groups_ctx"], \ + "the new group must be usable once its scan finishes, " \ + "regardless of whether anything was still watching the reload" + new_indexer = daemon._state["indexers"][second_group_id] + assert new_indexer.index.count == 3 + assert new_indexer.progress.scanning is False + finally: + shutdown_event.set() + await daemon._shutdown() + run_task.cancel() + try: + await run_task + except (asyncio.CancelledError, Exception): + pass + + +@pytest.mark.asyncio +async def test_group_scoped_ops_404_until_listed_then_succeed(tmp_path): + """ + The wizard's own sequence, reproduced against the real ops layer: attach + a brand-new group, fire the reload the way /api/reload now does + (ops.start_reload — scheduled, not awaited), and hit the group-scoped + calls that come right after in the UI (add a root, init the GEK) while + the scan is still running. + + Found live: "Attaching to node" no longer times out (ops.start_reload + returns immediately), but the very next wizard step then failed with + "Group not configured on this node" / "Group not hosted on this node" — + the group is not in daemon._state["config"].groups or ["groups_ctx"] + until _reload_config_inner() finishes, scan included, which is *after* + ops.start_reload has already returned. This locks in both halves: the + 404 while the scan runs, and success once ops.list_groups() actually + lists the group — the exact condition the wizard's own wait + (platform.waitForGroupHosted, app.js) polls for. + """ + first_dir = tmp_path / "first" + first_dir.mkdir() + (first_dir / "readme.txt").write_bytes(b"hello") + + second_dir = tmp_path / "second" + second_dir.mkdir() + for i in range(3): + (second_dir / f"file{i}.bin").write_bytes(os.urandom(64)) + second_group_id = "c" * 32 + extra_root_dir = tmp_path / "extra_root" + extra_root_dir.mkdir() + + data_dir = tmp_path / "data2" + config_path = tmp_path / "node2.toml" + config_path.write_text(_toml(data_dir, first_dir)) + + from meshbay_node.config import load_config + daemon = NodeDaemon(load_config(config_path), config_path=config_path) + + sk_node = Ed25519PrivateKey.generate() + mock_keys = _mock_keystore_keys(sk_node) + mock_session = MagicMock() + mock_session.node_id = "node123" + mock_session.user_id = "user123" + mock_session.hub_pk_pem = sk_node.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + real_scan_file = indexer_mod._scan_file + + def slow_scan_file(root, path): + import time + time.sleep(0.15) + return real_scan_file(root, path) + + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ + patch("meshbay_node.daemon.HubClient") as MockHub, \ + patch.object(indexer_mod, "_scan_file", slow_scan_file): + + hub_instance = AsyncMock() + hub_instance.startup = AsyncMock(return_value=mock_session) + hub_instance.send_ws = AsyncMock() + hub_instance._ws = None + hub_instance.close = AsyncMock() + hub_instance.__aenter__ = AsyncMock(return_value=hub_instance) + hub_instance.__aexit__ = AsyncMock(return_value=False) + MockHub.return_value = hub_instance + + shutdown_event = asyncio.Event() + + async def mock_maintain_ws(**kwargs): + await shutdown_event.wait() + hub_instance.maintain_ws = mock_maintain_ws + + async def run_daemon(): + with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15): + try: + await asyncio.wait_for(daemon.run(), timeout=15) + except (asyncio.TimeoutError, Exception): + pass + + run_task = asyncio.create_task(run_daemon()) + try: + for _ in range(50): + if daemon._state.get("status") == "running": + break + await asyncio.sleep(0.05) + assert daemon._state["status"] == "running" + + # The same file write ops.attach_group does (a raw text append), + # then the same fire-and-forget reload /api/reload now does. + config_path.write_text(_toml(data_dir, first_dir, + second_group_id, second_dir)) + reload_task = asyncio.ensure_future(ops.start_reload(daemon._state)) + + await asyncio.sleep(0.01) + listing = await ops.list_groups(daemon._state) + assert second_group_id not in [g["id"] for g in listing["groups"]], \ + "the scan (3 files x 0.15s) cannot have finished this fast" + + # Exactly the wizard's next two steps, hit mid-scan. + with pytest.raises(ops.OpError) as add_root_exc: + await ops.add_root(daemon._state, second_group_id, + str(extra_root_dir)) + assert add_root_exc.value.status == 404 + + with pytest.raises(ops.OpError) as gek_exc: + await ops.set_gek(daemon._state, second_group_id) + assert gek_exc.value.status == 404 + + # Now wait the way platform.waitForGroupHosted (app.js) does: + # poll list_groups(), not index-status, until the group is + # actually there. + for _ in range(100): + listing = await ops.list_groups(daemon._state) + if second_group_id in [g["id"] for g in listing["groups"]]: + break + await asyncio.sleep(0.05) + else: + pytest.fail("group never appeared in list_groups()") + + await asyncio.wait_for(reload_task, timeout=5) + + # Both calls that 404'd above must now succeed. + add_result = await ops.add_root(daemon._state, second_group_id, + str(extra_root_dir)) + assert add_result["status"] == "added" + + gek_result = await ops.set_gek(daemon._state, second_group_id) + assert gek_result["status"] == "ok" + finally: + shutdown_event.set() + await daemon._shutdown() + run_task.cancel() + try: + await run_task + except (asyncio.CancelledError, Exception): + pass diff --git a/packages/meshbay-node/tests/test_index_cache.py b/packages/meshbay-node/tests/test_index_cache.py new file mode 100644 index 0000000..db0e37e --- /dev/null +++ b/packages/meshbay-node/tests/test_index_cache.py @@ -0,0 +1,78 @@ +"""Tests for the (path, size, mtime) -> hash cache (indexer/cache.py).""" + +import pytest + +from meshbay_node.indexer.cache import IndexCache + + +@pytest.fixture +async def cache(tmp_path): + c = IndexCache(db_path=tmp_path / "index_cache.db") + await c.open() + yield c + await c.close() + + +@pytest.mark.asyncio +async def test_put_then_lookup_hits(cache): + await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123", + type="video", added_at=42) + + hit = await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) + + assert hit is not None + assert hit.hash == "abc123" + assert hit.type == "video" + assert hit.added_at == 42 + + +@pytest.mark.asyncio +async def test_lookup_misses_on_unknown_path(cache): + assert await cache.lookup("/lib/never-seen.mkv", size=1, mtime=1.0) is None + + +@pytest.mark.asyncio +async def test_lookup_misses_on_different_mtime(cache): + await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123", + type="video", added_at=42) + + assert await cache.lookup("/lib/a.mkv", size=1000, mtime=222.0) is None + + +@pytest.mark.asyncio +async def test_lookup_misses_on_different_size(cache): + await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123", + type="video", added_at=42) + + assert await cache.lookup("/lib/a.mkv", size=2000, mtime=111.0) is None + + +@pytest.mark.asyncio +async def test_put_overwrites_previous_row_for_same_path(cache): + await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="old", + type="video", added_at=1) + await cache.put("/lib/a.mkv", size=2000, mtime=222.0, hash="new", + type="video", added_at=2) + + assert await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) is None + hit = await cache.lookup("/lib/a.mkv", size=2000, mtime=222.0) + assert hit.hash == "new" + + +@pytest.mark.asyncio +async def test_cache_survives_reopen(tmp_path): + db_path = tmp_path / "index_cache.db" + + c1 = IndexCache(db_path=db_path) + await c1.open() + await c1.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123", + type="video", added_at=42) + await c1.close() + + c2 = IndexCache(db_path=db_path) + await c2.open() + hit = await c2.lookup("/lib/a.mkv", size=1000, mtime=111.0) + await c2.close() + + assert hit is not None + assert hit.hash == "abc123" diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py new file mode 100644 index 0000000..52cd78b --- /dev/null +++ b/packages/meshbay-node/tests/test_index_progress.py @@ -0,0 +1,165 @@ +""" +Indexing status visible node -> client: handshake ack field, the loopback +status route for the Create Group wizard / "add a directory", and the +periodic INDEX_PROGRESS push to already-connected peers. Never the index +itself (see test_daemon.py for that) and never anything sent to the hub. +""" + +import asyncio + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from unittest.mock import MagicMock +from fastapi.testclient import TestClient + +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer.indexer import IndexProgress +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from meshbay_node.ui.app import create_ui_app + + +def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" * 32): + index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate()) + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + group_ctx = {"index": index} + if progress is not None: + group_ctx["progress"] = progress + session._ctx = {"groups": {group_id: group_ctx}} + session._group_id = group_id + return session + + +# ── _indexing_status() ─────────────────────────────────────────────────────── + +def test_indexing_status_defaults_idle_when_no_progress_tracked(): + session = _session_with_progress(None) + assert session._indexing_status() == { + "scanning": False, "scanned_bytes": 0, "total_bytes": 0} + + +def test_indexing_status_reflects_live_progress(): + progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000, + current_dir="StarWars") + session = _session_with_progress(progress) + + status = session._indexing_status() + + assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000} + assert "current_dir" not in status, \ + "the directory name is operator-local detail, never sent to a member" + + +# ── /api/groups/{id}/index-status (loopback) ──────────────────────────────── + +def _ui_client(state: dict) -> TestClient: + return TestClient(create_ui_app({"status": "running", "groups_ctx": {}, + "indexes": {}, **state})) + + +def test_index_status_route_idle_for_unknown_group(): + client = _ui_client({"indexers": {}}) + resp = client.get("/api/groups/unknown-group/index-status") + assert resp.status_code == 200 + assert resp.json() == {"scanning": False, "scanned_bytes": 0, + "total_bytes": 0, "current_dir": ""} + + +def test_index_status_route_reflects_indexer_progress(): + fake_indexer = MagicMock() + fake_indexer.progress = IndexProgress( + scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000, + current_dir="2024") + client = _ui_client({"indexers": {"g" * 32: fake_indexer}}) + + resp = client.get(f"/api/groups/{'g' * 32}/index-status") + + assert resp.json() == { + "scanning": True, "scanned_bytes": 1_000_000, + "total_bytes": 4_000_000_000, "current_dir": "2024", + } + + +# ── _push_index_progress / _progress_pusher ───────────────────────────────── + +def _daemon(tmp_path) -> NodeDaemon: + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(), + groups=[], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + return NodeDaemon(config) + + +@pytest.mark.asyncio +async def test_push_index_progress_only_reaches_same_group_peers(tmp_path): + daemon = _daemon(tmp_path) + + same_group = MagicMock() + same_group._group_id = "a" * 32 + same_group._send = MagicMock() + other_group = MagicMock() + other_group._group_id = "b" * 32 + other_group._send = MagicMock() + + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": same_group, "p2": other_group} + daemon._webrtc = mock_webrtc + + progress = IndexProgress(scanning=True, scanned_bytes=10, total_bytes=100) + daemon._push_index_progress("a" * 32, progress) + + same_group._send.assert_called_once() + msg = same_group._send.call_args[0][0] + assert msg["type"] == "index_progress" + assert msg["group_id"] == "a" * 32 + assert msg["scanning"] is True + assert msg["scanned_bytes"] == 10 + assert msg["total_bytes"] == 100 + other_group._send.assert_not_called() + + +@pytest.mark.asyncio +async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path): + daemon = _daemon(tmp_path) + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + indexer = MagicMock() + indexer.group_id = "a" * 32 + indexer.progress = IndexProgress(scanning=True, scanned_bytes=0, total_bytes=100) + + task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05)) + try: + # Two ticks while still scanning. + await asyncio.sleep(0.12) + assert session._send.call_count >= 2 + assert all(c.args[0]["scanning"] is True for c in session._send.call_args_list) + + # Scan finishes between ticks. + indexer.progress.scanning = False + calls_before = session._send.call_count + await asyncio.sleep(0.07) + assert session._send.call_count == calls_before + 1, \ + "exactly one final push must follow the False transition" + assert session._send.call_args.args[0]["scanning"] is False + + # Nothing further once idle. + calls_after_final = session._send.call_count + await asyncio.sleep(0.15) + assert session._send.call_count == calls_after_final, \ + "no more pushes once idle and already reported" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index c304361..68ccc4c 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -9,7 +9,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import generate_gek -from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache +import meshbay_node.indexer.indexer as indexer_mod from conftest import one_root from meshbay_node.keystore import NodeKeys @@ -190,3 +191,290 @@ async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek): wire = indexer.index.serialize() recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) assert recovered.count == indexer.index.count + + +# ── Cache-aware scanning ─────────────────────────────────────────────────────── + +@pytest.fixture +async def index_cache(tmp_path): + c = IndexCache(db_path=tmp_path / "index_cache.db") + await c.open() + yield c + await c.close() + + +@pytest.mark.asyncio +async def test_second_scan_with_same_cache_hashes_nothing( + shared_dir, sk_node, gek, index_cache): + """ + The whole point of the cache: a "restart" (a fresh DirectoryIndexer, same + on-disk cache) that finds every file's (size, mtime) unchanged must not + read a single byte of file content. + """ + first = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await first.initial_scan() + assert first.index.count == 4 + + calls = [] + real_scan_file = indexer_mod._scan_file + + def spy(root, path): + calls.append(path) + return real_scan_file(root, path) + + indexer_mod._scan_file = spy + try: + second = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await second.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert calls == [], f"expected zero hash calls on a fully-cached rescan, got {calls}" + assert second.index.count == first.index.count + assert {e.id for e in second.index.entries} == {e.id for e in first.index.entries} + + +@pytest.mark.asyncio +async def test_modified_file_is_rehashed(tmp_path, sk_node, gek, index_cache): + d = tmp_path / "shared" + d.mkdir() + f = d / "movie.mkv" + f.write_bytes(b"original content") + + first = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await first.initial_scan() + old_id = first.index.entries[0].id + + # Change both content and mtime, as any real edit would. + f.write_bytes(b"a completely different, longer payload") + os.utime(f, (time.time() + 5, time.time() + 5)) + + second = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await second.initial_scan() + + assert second.index.count == 1 + assert second.index.entries[0].id != old_id + + +@pytest.mark.asyncio +async def test_scan_interrupted_partway_leaves_only_completed_files_cached( + tmp_path, sk_node, gek, index_cache): + """ + A cache row is only ever written after a file is fully hashed (cache.py), + so a crash mid-scan cannot leave a stale/partial row — the next scan just + treats the not-yet-cached files as new, and finishes the job. + """ + d = tmp_path / "shared" + d.mkdir() + names = [f"file{i}.bin" for i in range(5)] + for i, name in enumerate(names): + (d / name).write_bytes(os.urandom(64) * (i + 1)) + + real_scan_file = indexer_mod._scan_file + hashed_before_crash = [] + + def crash_after_three(root, path): + if len(hashed_before_crash) >= 3: + raise RuntimeError("simulated crash mid-scan") + entry = real_scan_file(root, path) + hashed_before_crash.append(path) + return entry + + indexer_mod._scan_file = crash_after_three + try: + crashing = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + with pytest.raises(RuntimeError): + await crashing.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert len(hashed_before_crash) == 3 + + # A normal rescan (same cache) must still end up with all 5 files + # correctly indexed, hashing only the ones the crash never got to. + calls = [] + + def spy(root, path): + calls.append(path) + return real_scan_file(root, path) + + indexer_mod._scan_file = spy + try: + resumed = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await resumed.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert resumed.index.count == 5 + assert len(calls) == 2, f"expected only the 2 not-yet-cached files to be hashed, got {len(calls)}" + + +# ── Progress state ─────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_progress_reflects_bytes_scanned(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek) + assert indexer.progress.scanning is False + + await indexer.initial_scan() + + total_size = sum(f.stat().st_size for f in shared_dir.rglob("*") if f.is_file()) + assert indexer.progress.scanning is False, "must end idle, not stuck scanning" + assert indexer.progress.scanned_bytes == total_size + assert indexer.progress.total_bytes == total_size + + +@pytest.mark.asyncio +async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek): + d = tmp_path / "shared" + d.mkdir() + (d / "a.bin").write_bytes(os.urandom(64)) + (d / "b.bin").write_bytes(os.urandom(64)) + + real_scan_file = indexer_mod._scan_file + + def boom(root, path): + raise RuntimeError("simulated failure mid-scan") + + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek) + indexer_mod._scan_file = boom + try: + with pytest.raises(RuntimeError): + await indexer.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert indexer.progress.scanning is False, \ + "an exception mid-scan must not leave the scanning flag stuck on" + + +# ── Off-loop directory walks, reconcile backoff ───────────────────────────── + +@pytest.mark.asyncio +async def test_walk_root_does_not_stall_the_event_loop(tmp_path, sk_node, gek): + d = tmp_path / "shared" + d.mkdir() + (d / "f.bin").write_bytes(b"x") + + real_walk = indexer_mod._walk_root + + def slow_walk(root): + time.sleep(0.2) + return real_walk(root) + + indexer_mod._walk_root = slow_walk + ticks = 0 + + async def ticker(): + nonlocal ticks + while True: + await asyncio.sleep(0.01) + ticks += 1 + + ticker_task = asyncio.create_task(ticker()) + try: + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek) + await indexer.initial_scan() + finally: + indexer_mod._walk_root = real_walk + ticker_task.cancel() + try: + await ticker_task + except asyncio.CancelledError: + pass + + assert ticks >= 5, ( + "the event loop must keep running other tasks while a directory " + f"walk is in progress in the executor — only {ticks} ticks happened " + "during a 0.2s walk") + + +@pytest.mark.asyncio +async def test_reconcile_backoff_grows_with_no_changes_then_caps(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=0.01) + await indexer.initial_scan() + assert indexer._reconcile_delay == 0.01 + + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.sleep(0.2) + assert indexer._reconcile_delay > 0.01, \ + "several no-change ticks must have grown the delay" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + # A low, instance-only cap so the clamp is observable without waiting + # through dozens of real doublings up to the real 7200s ceiling. + indexer.RECONCILE_BACKOFF_CAP = 0.05 + indexer._reconcile_delay = 0.04 + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.sleep(0.15) + assert indexer._reconcile_delay <= 0.05, "delay must never exceed the cap" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio +async def test_note_activity_resets_backoff(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=10.0) + await indexer.initial_scan() + indexer._reconcile_delay = 5000.0 # simulate a long-idle backoff + + indexer.note_activity() + + assert indexer._reconcile_delay == 10.0 + + +@pytest.mark.asyncio +async def test_reconcile_backoff_resets_when_something_actually_changes( + shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=0.02) + await indexer.initial_scan() + indexer._reconcile_delay = 5.0 # pretend it had already backed off a lot + + # A fake reconcile() rather than a real filesystem change: the real + # sweep's timing (disk I/O, the executor round trip) would race against + # this test's own sleeps. What matters here is only _reconcile_loop's + # reaction to "something changed", not reconcile()'s own detection logic + # — that is covered separately (test_root_availability.py). + reconciled_once = asyncio.Event() + + async def fake_reconcile(): + reconciled_once.set() + return True + + indexer._reconcile_delay = 0.01 + indexer.reconcile = fake_reconcile + + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.wait_for(reconciled_once.wait(), timeout=2.0) + assert indexer._reconcile_delay == 0.02, \ + "a real change must reset the delay back to the base interval" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py index d2ccc0d..83758ae 100644 --- a/packages/meshbay-node/tests/test_ops.py +++ b/packages/meshbay-node/tests/test_ops.py @@ -9,6 +9,7 @@ appeared: the adapters must be thin, and the operations must not decide who may call them. """ +import asyncio import inspect from pathlib import Path @@ -221,3 +222,32 @@ async def test_reload_config_without_fn_is_refused(tmp_path): state = _state(tmp_path) with pytest.raises(ops.OpError, match="Reload not available"): await ops.reload_config(state) + + +async def test_start_reload_returns_before_reload_fn_finishes(tmp_path): + """The loopback route uses this one: a brand-new group's initial scan + can take minutes, and the Electron bridge caps every loopback call at + 30s (main.js node:call) — start_reload must not block on it.""" + state = _state(tmp_path) + release = asyncio.Event() + called = [] + + async def slow_reload(): + await release.wait() + called.append(True) + state["reload_fn"] = slow_reload + + out = await asyncio.wait_for(ops.start_reload(state), timeout=1.0) + + assert out["status"] == "reloading" + assert not called, "start_reload must return before reload_fn finishes" + + release.set() + await asyncio.sleep(0) # let the still-running reload_fn task complete + assert called, "reload_fn must still actually run, just not be waited on" + + +async def test_start_reload_without_fn_is_refused(tmp_path): + state = _state(tmp_path) + with pytest.raises(ops.OpError, match="Reload not available"): + await ops.start_reload(state) diff --git a/packages/meshbay-node/tests/test_scan_settings_policy.py b/packages/meshbay-node/tests/test_scan_settings_policy.py new file mode 100644 index 0000000..719b988 --- /dev/null +++ b/packages/meshbay-node/tests/test_scan_settings_policy.py @@ -0,0 +1,205 @@ +""" +The operator can tune how often the indexer's reconciliation backstop runs, +and how long it waits after a file's last write before hashing it. + +Same shape as test_apps_enabled_policy.py / test_member_upload_policy.py: +changed by a signed operator instruction, stored on the node rather than the +hub. Unlike those two, there is also a *live* DirectoryIndexer object to +update — see test_set_scan_settings_updates_the_live_indexer below. +""" + +import os +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.adminop import OP_SET_SCAN_SETTINGS +from meshbay_common.crypto import generate_gek +from meshbay_node import ops +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roster import Roster +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +@pytest.fixture +def gek(): + return generate_gek() + + +@pytest.fixture +def shared_dir(tmp_path): + d = tmp_path / "shared" + d.mkdir() + (d / "video.mkv").write_bytes(os.urandom(256)) + return d + + +def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + ctx = { + "roots": one_root(shared_root), + "index": index, + "sk_node": index.sk_node, + "node_user_id": operator, + } + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = None + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +# ── Refused before a challenge is even issued ─────────────────────────────── + +async def test_out_of_range_reconcile_interval_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_set_scan_settings( + {"reconcile_interval_secs": 1.0, "debounce_secs": 2.0}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_out_of_range_debounce_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_set_scan_settings( + {"reconcile_interval_secs": 600.0, "debounce_secs": 99999.0}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_non_numeric_values_are_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_set_scan_settings( + {"reconcile_interval_secs": "not-a-number", "debounce_secs": 2.0}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._has_admin_authority = lambda: False + + session._do_set_scan_settings( + {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0}) + + assert [m for m in session.sent if m.get("type") == "error"] + + +# ── Who may change it ─────────────────────────────────────────────────────── + +async def test_changing_it_needs_a_signature(tmp_path): + """The request only ever produces a challenge — nothing is applied + until a signature over the transcript verifies.""" + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_set_scan_settings( + {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0}) + + assert issued == [(OP_SET_SCAN_SETTINGS, "600,2")] + + +# ── Where it is stored ────────────────────────────────────────────────────── + +async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + defaults = await roster.scan_settings("g1") + assert defaults == { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }, "unset must mean the indexer's own defaults, or an upgrade " \ + "changes behaviour for every existing group" + + await roster.set_scan_settings("g1", 1200.0, 5.0, set_by="op") + assert await roster.scan_settings("g1") == { + "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0} + finally: + await roster.close() + + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.scan_settings("g1") == { + "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0} + assert await reopened.scan_settings("g2") == { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }, "one group's setting must not answer for another" + finally: + await reopened.close() + + +# ── Applying it to the live indexer ───────────────────────────────────────── + +async def test_set_scan_settings_updates_the_live_indexer(tmp_path, shared_dir, gek): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="g1", + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + indexer._reconcile_delay = 5000.0 # simulate a long-idle backoff + state = {"roster": roster, "indexers": {"g1": indexer}} + + try: + result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0) + + assert result == {"reconcile_interval_secs": 1800.0, "debounce_secs": 3.0, + "group_id": "g1"} + assert indexer.reconcile_secs == 1800.0 + assert indexer.debounce_secs == 3.0 + assert indexer._reconcile_delay == 1800.0, \ + "the new interval must apply right away, not after whatever " \ + "backoff had already stretched the wait to" + assert await roster.scan_settings("g1") == { + "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0} + finally: + await roster.close() + + +async def test_set_scan_settings_without_a_live_indexer_still_persists(tmp_path): + """A group hosted on the node but with no running indexer in this + process (e.g. a test, or a group not yet hot-loaded) must not crash — + the setting still lands in roster.db for whenever it is.""" + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + state = {"roster": roster, "indexers": {}} + + try: + result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0) + assert result["reconcile_interval_secs"] == 1800.0 + assert await roster.scan_settings("g1") == { + "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0} + finally: + await roster.close() |