diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-10-12 00:17:32 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-10-12 00:17:32 +0200 |
| commit | 14911aae21ec4975f7a805b91e3c4867cb41974b (patch) | |
| tree | 8c00230a88053905fd77938f3a9539efeb4c3094 /packages/meshbay-node | |
| parent | f6f107e02c7a4aa7326c9a82310a239327d81ed6 (diff) | |
| download | meshbay-14911aae21ec4975f7a805b91e3c4867cb41974b.tar.gz | |
feat: let the operator pause and resume a root's indexing (MNP 6.2)
A whole-root walk now checks its root before every file: removed or
ejected, it stops instead of reading the drive to the end; paused, it
finishes from the hash cache alone. A paused root is not watched, not
swept by reconcile and survives a restart (roster), its indexed files
stay served. Resume runs the reconcile sweep over the root, so entries
and their enrichment are kept.
Signed root_pause / root_resume over MNP, loopback routes, CLI
`root pause|resume`, a pause button on the indexing dock and
pause/resume in the roots table of Group settings.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-node')
18 files changed, 1563 insertions, 137 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/cli/groups.py b/packages/meshbay-node/src/meshbay_node/cli/groups.py index fca4800..b402930 100644 --- a/packages/meshbay-node/src/meshbay_node/cli/groups.py +++ b/packages/meshbay-node/src/meshbay_node/cli/groups.py @@ -33,6 +33,8 @@ def group(args) -> None: flags.append("removable") if r.get("ejected"): flags.append("ejected") + if r.get("paused"): + flags.append("paused") flag_str = f" ({', '.join(flags)})" if flags else "" live = "" if r.get("available", True) else " [UNAVAILABLE]" print(f" root {r['name']}{flag_str}{live}") @@ -128,6 +130,8 @@ def root(args) -> None: flags.append("removable") if r.get("ejected"): flags.append("EJECTED") + if r.get("paused"): + flags.append("PAUSED") avail = "available" if r.get("available", True) else "UNAVAILABLE" flags.append(avail) print(f" {r['name']:<20} {', '.join(flags)}") @@ -215,8 +219,29 @@ def root(args) -> None: print(f"plugged root {name} — files are visible again") return - print("usage: meshbay-node root list|add|remove|set|eject|plug [name] " - "[--group NAME]") + if sub == "pause": + name = args.target + if not name: + print("usage: meshbay-node root pause <name> [--group NAME]") + sys.exit(1) + _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/pause", + method="PUT") + print(f"paused root {name} — indexed files stay listed, nothing new " + f"is indexed until resumed") + return + + if sub == "resume": + name = args.target + if not name: + print("usage: meshbay-node root resume <name> [--group NAME]") + sys.exit(1) + _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/resume", + method="PUT") + print(f"resumed root {name} — indexing continues in the background") + return + + print("usage: meshbay-node root list|add|remove|set|eject|plug|pause|resume " + "[name] [--group NAME]") sys.exit(1) diff --git a/packages/meshbay-node/src/meshbay_node/cli/parser.py b/packages/meshbay-node/src/meshbay_node/cli/parser.py index cfc4704..2a593b6 100644 --- a/packages/meshbay-node/src/meshbay_node/cli/parser.py +++ b/packages/meshbay-node/src/meshbay_node/cli/parser.py @@ -20,7 +20,7 @@ def build_parser() -> argparse.ArgumentParser: "| operator pair: pair a " "browser with this node | member list|invite|cancel|revoke|unpin " "| group list|add|remove " - "| root list|add|remove|set|eject|plug " + "| root list|add|remove|set|eject|plug|pause|resume " "| gek init|rotate | file list|rm " "| video rematch: re-resolve TMDB matches for a group's " "videos " @@ -44,7 +44,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("subcommand", nargs="?", help="'pair' for operator; list|invite|revoke|unpin for " "member; list|add|remove for group; " - "list|add|remove|set|eject|plug for root; " + "list|add|remove|set|eject|plug|pause|resume for root; " "init|rotate for gek; " "list|rm for file; rematch for video; show|clear for " "denylist; list|add|remove|reset for stun; " diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 4d89f97..022553e 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -1286,7 +1286,8 @@ class NodeDaemon(EnrichmentMixin): async def _build_roots(self, group_cfg) -> RootSet: """ - Build a group's RootSet from node.toml, with the ejected state restored. + Build a group's RootSet from node.toml, with the ejected and paused + state restored. node.toml carries configuration (`writable`, `removable`); the roster carries the runtime answer to "is this drive ejected right now". They @@ -1298,12 +1299,14 @@ class NodeDaemon(EnrichmentMixin): if self._roster: ejected = await self._roster.ejected_roots(group_cfg.id) auto = await self._roster.auto_ejected_roots(group_cfg.id) - if ejected: - for spec in specs: - name = spec.get("name") or Path(spec.get("path", "")).name - if fold(name) in ejected: - spec["ejected"] = True - spec["ejected_auto"] = fold(name) in auto + paused = await self._roster.paused_roots(group_cfg.id) + for spec in specs: + name = spec.get("name") or Path(spec.get("path", "")).name + if fold(name) in ejected: + spec["ejected"] = True + spec["ejected_auto"] = fold(name) in auto + if fold(name) in paused: + spec["paused"] = True return RootSet.build(specs) # Every application that keeps directories. This is the one list, and it diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 4eeeec0..4d68212 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -396,7 +396,7 @@ class DirectoryIndexer: await off_disk(self.roots, self.roots.refresh_availability) await self._plug_back_recognised() total = 0 - waiting = [r.name for r in self.roots if r.available] + waiting = [r.name for r in self.roots if r.available and not r.paused] self._queue(waiting) try: for root in self.roots: @@ -405,8 +405,9 @@ class DirectoryIndexer: "are not indexed yet and will appear when it returns", root.name, root.path) continue - waiting.remove(root.name) - self._unqueue([root.name]) + if root.name in waiting: + waiting.remove(root.name) + self._unqueue([root.name]) total += await self._scan_root(root) finally: self._unqueue(waiting) @@ -417,9 +418,21 @@ class DirectoryIndexer: total, len(self.roots)) async def _scan_root(self, root: Root, *, kind: str = "scan") -> int: + """ + Walk one root and index what is on it. + + A paused root — paused before the walk or part-way through it — is + listed from the hash cache alone: what the node already hashed comes + back, nothing new is read. It cannot simply be skipped, because the + index is rebuilt at every start and a rescan drops the root's entries + before it walks. A root removed or ejected while this runs is not + finished at all: its entries are dropped, or frozen on a drive that is + leaving, so reading the rest of it is work for nothing. + """ log.info("Scanning %s (root %r) ...", root.path, root.name) count = 0 loop = asyncio.get_event_loop() + quiet = root.paused # `scanning` flips on here, before the walk — not after it and the # sizing pass, which for a large or slow (network/USB) root can @@ -432,7 +445,8 @@ class DirectoryIndexer: # had not yet turned on, even though the node was already several # seconds into walking and sizing a large root — total_bytes is # unknown at this point, so it starts at 0 and is corrected below. - self._begin_job(root.name, root=root, kind=kind) + if not quiet: + self._begin_job(root.name, root=root, kind=kind) try: try: files = await loop.run_in_executor(self._executor, _walk_root, root) @@ -447,14 +461,30 @@ class DirectoryIndexer: # loop right here, blocking the whole daemon for a large root). sized = await loop.run_in_executor(self._executor, _size_files, files) - self.progress.total_bytes = sum(size for _, size in sized) - self.progress.files_total = len(sized) - self.progress.current_dir = "" - 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 - self.progress.files_done += 1 + if not quiet: + self.progress.total_bytes = sum(size for _, size in sized) + self.progress.files_total = len(sized) + self.progress.current_dir = "" + for done, (file_path, size) in enumerate(sized): + live = self._live(root) + if live is None or live.ejected: + log.info("Scan of root %r stopped after %d of %d files: " + "removed or ejected", root.name, done, len(sized)) + break + if not quiet and live.paused: + log.info("Root %r paused after %d of %d files — the rest is " + "listed from the hash cache only", root.name, done, + len(sized)) + # Off the operator's screen now; what is left reads no file. + self._end_job() + quiet = True + if not quiet: + self.progress.current_dir = file_path.parent.name + entry = await (self._cached(root, file_path) if quiet + else self._hash_or_cached(root, file_path)) + if not quiet: + self.progress.scanned_bytes += size + self.progress.files_done += 1 if entry: self._index.add_entry(entry) count += 1 @@ -462,7 +492,8 @@ class DirectoryIndexer: # 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._end_job() + if not quiet: + self._end_job() return count def _begin_job(self, current_dir: str, total_bytes: int = 0, *, @@ -520,7 +551,13 @@ class DirectoryIndexer: if name in self.progress.queued: self.progress.queued.remove(name) - async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None: + async def _cached(self, root: Root, file_path: Path) -> IndexEntry | None: + """What `_hash_or_cached` would return without reading the file, or + None: a paused root's file the cache does not know is left out.""" + return await self._hash_or_cached(root, file_path, read=False) + + async def _hash_or_cached(self, root: Root, file_path: Path, *, + read: bool = True) -> 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 @@ -553,6 +590,8 @@ class DirectoryIndexer: hash_version=cached.hash_version, ), file_path, st) + if not read: + return None 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: @@ -643,7 +682,7 @@ class DirectoryIndexer: self._observer = Observer() watched = 0 for root in self.roots: - if not root.available: + if not root.available or root.paused: continue try: self._observer.schedule(handler, str(root.path), recursive=True) @@ -742,8 +781,26 @@ class DirectoryIndexer: if self.on_change: await self.on_change(self) + def _live(self, root: Root) -> Root | None: + """ + This set's own object for `root`, or None once it is gone from the set. + + A retarget replaces every Root object, so a walk that started before + one holds a stale copy: the flags an operator sets afterwards are on + this one. + """ + for r in self.roots: + if r.folded == root.folded and r.path == root.path: + return r + return None + def _holds(self, root: Root) -> bool: - return any(r.folded == root.folded and r.path == root.path for r in self.roots) + return self._live(root) is not None + + def _stopped(self, root: Root) -> bool: + """Removed, ejected or paused since this walk began: stop reading it.""" + live = self._live(root) + return live is None or live.ejected or live.paused async def _scan_added_roots(self, added: list[Root]) -> None: waiting = [r.name for r in added] @@ -867,69 +924,86 @@ class DirectoryIndexer: Only touches roots that are readable right now — a root whose volume is absent has nothing to compare against, and comparing anyway is exactly - the mistake this module exists to avoid. + the mistake this module exists to avoid. A paused root is left alone: + the operator stopped its indexing, and this pass is indexing. """ changed = False - loop = asyncio.get_event_loop() for root in self.roots: - if not root.available: - continue - try: - on_disk, known = await loop.run_in_executor( - self._executor, self._sweep_scan_root, root) - except OSError as e: - log.warning("Cannot reconcile root %r: %s", root.name, e) + if not root.available or root.paused: continue - - for missing in set(known) - on_disk: - # Duplicate content is handled without a special case here: the - # entry goes, and the add loop below re-indexes the surviving - # copy under its own path, because the id is then absent. A - # dedicated "find the survivor" lookup was written first and - # deleted — it rehashed every file under the root on any single - # deletion, and a test proved it changed nothing. - self._index.remove_entry(known[missing]) - log.info("Reconcile: %s is gone", missing) + if await self._sweep_root(root): changed = True + return changed - added_paths = on_disk - set(known) - if not added_paths: - continue + async def _sweep_root(self, root: Root, *, kind: str = "reconcile") -> bool: + """ + One root's half of the sweep. Also what a resume runs: it indexes only + the files the index lacks, where a `_scan_root` would replace every + entry it met and lose what the enrichment passes had put there. + """ + changed = False + loop = asyncio.get_event_loop() + try: + on_disk, known = await loop.run_in_executor( + self._executor, self._sweep_scan_root, root) + except OSError as e: + log.warning("Cannot reconcile root %r: %s", root.name, e) + return False - 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)) + for missing in set(known) - on_disk: + # Duplicate content is handled without a special case here: the + # entry goes, and the add loop below re-indexes the surviving + # copy under its own path, because the id is then absent. A + # dedicated "find the survivor" lookup was written first and + # deleted — it rehashed every file under the root on any single + # deletion, and a test proved it changed nothing. + self._index.remove_entry(known[missing]) + log.info("Reconcile: %s is gone", missing) + changed = True + + added_paths = on_disk - set(known) + if not added_paths: + return changed - self._begin_job("", sum(size for _, size in added_sized), root=root, - kind="reconcile", files_total=len(added_sized)) + added_sized: list[tuple[Path, int]] = [] + for p in added_paths: 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 - self.progress.files_done += 1 - 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) + added_sized.append((p, p.stat().st_size)) + except OSError: + added_sized.append((p, 0)) + + self._begin_job("", sum(size for _, size in added_sized), root=root, + kind=kind, files_total=len(added_sized)) + try: + for done, (added, size) in enumerate(added_sized): + if self._stopped(root): + log.info("Indexing root %r stopped after %d of %d new files", + root.name, done, len(added_sized)) + break + self.progress.current_dir = added.parent.name + entry = await self._hash_or_cached(root, added) + self.progress.scanned_bytes += size + self.progress.files_done += 1 + 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) + if kind == "reconcile": log.info("Reconcile: %s appeared (missed event)", added) - changed = True - finally: - self._end_job() + changed = True + finally: + self._end_job() return changed def _entries_under(self, root: Root) -> list[IndexEntry]: @@ -1170,6 +1244,71 @@ class DirectoryIndexer: if self.on_change: await self.on_change(self) + # ── Pause and resume ────────────────────────────────────────────────────── + + async def pause_root(self, root_name: str) -> None: + """ + Stop indexing a root: the walk under way ends at the next file, and + neither the watcher nor reconcile touch it until it is resumed. Its + entries stay — they are files on a readable disk, served as before. + """ + root = self.roots.by_name(root_name) + if root is None or root.paused: + return + root.paused = True + self.progress.queued[:] = [n for n in self.progress.queued + if fold(n) != root.folded] + log.info("Root %r paused — %d entries kept, nothing more is indexed " + "until it is resumed", root.name, len(self._entries_under(root))) + self._restart_observer() + await self.publish_roots() + + async def resume_root(self, root_name: str) -> None: + """ + Index a paused root again, from where it stopped. + + Returns once the walk is queued, not when it ends: on a large drive + that is hours, and the operator's request has nothing to wait for — + the progress band shows the walk. Run as a task this indexer owns, like + a plug, so the session that asked can close without cancelling it. + """ + root = self.roots.by_name(root_name) + if root is None or not root.paused: + return + root.paused = False + log.info("Root %r resumed", root.name) + self._restart_observer() + await self.publish_roots() + if not root.available: + return + self._queue([root.name]) + task = asyncio.create_task(self._finish_resume(root)) + self._scan_tasks.add(task) + task.add_done_callback(self._scan_tasks.discard) + + async def _finish_resume(self, root: Root) -> None: + changed = False + waiting = True + try: + async with self._scan_lock: + self._unqueue([root.name]) + waiting = False + # Paused, ejected or removed again while it waited. + live = self._live(root) + if live is not None and live.available and not self._stopped(live): + changed = await self._sweep_root(live, kind="scan") + except asyncio.CancelledError: + raise + except Exception: + log.exception("Resuming root %r failed", root.name) + finally: + if waiting: + self._unqueue([root.name]) + if changed: + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + # ── Internal update ─────────────────────────────────────────────────────── def _schedule_update(self, file_path: Path, deleted: bool = False, @@ -1278,7 +1417,7 @@ class DirectoryIndexer: await self.on_change(self) return - if not root.available: + if not root.available or root.paused: return # On Windows watchdog cannot tell what a deleted path was -- it is diff --git a/packages/meshbay-node/src/meshbay_node/ops/__init__.py b/packages/meshbay-node/src/meshbay_node/ops/__init__.py index fc5b7ef..c88a909 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/ops/__init__.py @@ -84,8 +84,10 @@ from meshbay_node.ops.node_toml import ( from meshbay_node.ops.roots import ( add_root, eject_root, + pause_root, plug_root, remove_root, + resume_root, update_root, ) from meshbay_node.ops.settings import ( @@ -139,6 +141,7 @@ __all__ = [ "list_transfers", "open_chat_epoch", "pair_operator", + "pause_root", "plug_root", "prune_chat", "purge_chat", @@ -149,6 +152,7 @@ __all__ = [ "rematch_video", "remove_root", "resolve_user", + "resume_root", "revoke_member", "set_app_directories", "set_app_directory", diff --git a/packages/meshbay-node/src/meshbay_node/ops/roots.py b/packages/meshbay-node/src/meshbay_node/ops/roots.py index c272978..187e134 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/roots.py +++ b/packages/meshbay-node/src/meshbay_node/ops/roots.py @@ -1,4 +1,4 @@ -"""A group's directories: adding, removing, changing and ejecting them.""" +"""A group's directories: adding, removing, changing, ejecting and pausing them.""" from __future__ import annotations @@ -126,6 +126,12 @@ async def remove_root(state: dict, group_id: str, root_name: str) -> dict: _remove_roots_block(conf_path, group_id, resolved) cfg.roots.pop(match_idx) + # A directory added back later under the same name starts indexed, not + # paused by a decision about the one removed here. + roster = state.get("roster") + if roster: + await roster.set_root_paused(group_id, rname, False, + set_by=state.get("node_user_id", "")) # Not mutating the live set here either — see `add_root`. Dropping the # root from it would leave `retarget` unable to tell that its entries @@ -304,3 +310,68 @@ async def plug_root(state: dict, group_id: str, root_name: str) -> dict: log.info("Root plugged: %s in group %s", root_name, group_id[:8]) return {"status": "plugged", "name": root_name, "group_id": group_id, "roots": roots.describe()} + + +def _live_root(state: dict, group_id: str, root_name: str): + """ + The group's live RootSet and the root named `root_name` in it. + + The indexer's set when the group has no context yet: a group added while + the node runs gets one only once its first scan ends, and that scan — a + whole new library — is the one most worth pausing. + """ + config = _config(state) + if not any(g.id == group_id for g in config.groups): + raise OpError("Group not configured on this node", status=404) + ctx = state.get("groups_ctx", {}).get(group_id) + indexer = state.get("indexers", {}).get(group_id) + roots: RootSet | None = (ctx.get("roots") if ctx else None) or ( + indexer.roots if indexer else None) + if not roots: + raise OpError("Group has no roots", status=503) + root = roots.by_name(root_name) + if root is None: + raise OpError(f"No root named {root_name!r} in this group", status=404) + return roots, root + + +async def pause_root(state: dict, group_id: str, root_name: str) -> dict: + """Stop indexing a root; what is already indexed stays listed.""" + roots, root = _live_root(state, group_id, root_name) + if root.paused: + return {"status": "already_paused", "name": root.name, + "group_id": group_id, "roots": roots.describe()} + + # Persisted first: a restart between the two must not start the scan the + # operator stopped. + await _roster(state).set_root_paused( + group_id, root.name, True, set_by=state.get("node_user_id", "")) + # The flag on the context's set too, as `eject_root` does, for a context + # whose indexer a retarget replaced. + indexer = state.get("indexers", {}).get(group_id) + if indexer: + await indexer.pause_root(root.name) + root.paused = True + + log.info("Root paused: %s in group %s", root.name, group_id[:8]) + return {"status": "paused", "name": root.name, "group_id": group_id, + "roots": roots.describe()} + + +async def resume_root(state: dict, group_id: str, root_name: str) -> dict: + """Index a paused root again; the walk runs in the background.""" + roots, root = _live_root(state, group_id, root_name) + if not root.paused: + return {"status": "already_resumed", "name": root.name, + "group_id": group_id, "roots": roots.describe()} + + await _roster(state).set_root_paused( + group_id, root.name, False, set_by=state.get("node_user_id", "")) + indexer = state.get("indexers", {}).get(group_id) + if indexer: + await indexer.resume_root(root.name) + root.paused = False + + log.info("Root resumed: %s in group %s", root.name, group_id[:8]) + return {"status": "resumed", "name": root.name, "group_id": group_id, + "roots": roots.describe()} diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py index 7dded1f..61fdc9b 100644 --- a/packages/meshbay-node/src/meshbay_node/roots.py +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -182,6 +182,10 @@ class Root: # such a root comes back on its own once its files are there again # (`DirectoryIndexer._reconcile`). An operator's eject never does. auto: bool = False + # The operator stopped indexing it: not walked, not watched, not re-checked, + # and what was already indexed stays listed. Independent of `available` — + # a paused root is readable, and its files are served. + paused: bool = False available: bool = True @property @@ -341,14 +345,16 @@ class RootSet: # Backward compat: old configs use `upload` instead of `writable` writable = bool(spec.get("writable", spec.get("upload", False))) - # `ejected` is runtime state, not configuration — it reaches here - # only from the roster, restored at startup so a drive ejected - # before a restart does not come back on its own. + # `ejected` and `paused` are runtime state, not configuration — they + # reach here only from the roster, restored at startup so a drive + # ejected before a restart does not come back on its own, and a + # paused scan does not start again. root = Root(name=name, path=path, kind=kind, writable=writable, removable=bool(spec.get("removable", False)), ejected=bool(spec.get("ejected", False)), auto=bool(spec.get("ejected_auto", False)), + paused=bool(spec.get("paused", False)), available=not bool(spec.get("ejected", False))) _refuse_nesting(root, roots) roots.append(root) @@ -496,7 +502,8 @@ class RootSet: "available": r.available, "writable": r.writable, "removable": r.removable, - "ejected": r.ejected} + "ejected": r.ejected, + "paused": r.paused} # `with_paths` is for the operator's *own* channels only — the # loopback API and the CLI reading it, both of which already # require being on this machine with the run token. A member is diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index a5d1967..9336e38 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -737,6 +737,31 @@ class Roster: return {r["key"][len(prefix):] for r in rows if r["key"].startswith(prefix) and r["value"] == "auto"} + # Whether the operator paused a root's indexing. Kept here beside + # `root_ejected:` for the same reasons: runtime state rather than + # configuration, and a pause that a restart undid would start the very scan + # the operator stopped. + SETTING_ROOT_PAUSED_PREFIX = "root_paused:" + + @classmethod + def root_paused_key(cls, root_name: str) -> str: + return cls.SETTING_ROOT_PAUSED_PREFIX + fold(root_name) + + async def set_root_paused(self, group_id: str, root_name: str, + paused: bool, set_by: str = "") -> None: + await self.set_setting(group_id, self.root_paused_key(root_name), + "1" if paused else "0", set_by) + + async def paused_roots(self, group_id: str) -> set[str]: + """The folded names of this group's paused roots — matched like `ejected_roots`.""" + prefix = self.SETTING_ROOT_PAUSED_PREFIX + async with self._db.execute( + "SELECT key, value FROM group_settings WHERE group_id = ?", + (group_id,)) as cur: + rows = await cur.fetchall() + return {r["key"][len(prefix):] for r in rows + if r["key"].startswith(prefix) and r["value"] == "1"} + async def get_setting(self, group_id: str, key: str, default: str | None = None) -> str | None: async with self._db.execute( diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py index 7b3b1c9..aab3691 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py @@ -24,8 +24,10 @@ from meshbay_common.adminop import ( OP_MEMBER_REVOKE, OP_MUSICBRAINZ_ENABLED, OP_ROOT_EJECT, + OP_ROOT_PAUSE, OP_ROOT_PLUG, OP_ROOT_REMOVE, + OP_ROOT_RESUME, OP_SEARCH_LISTED, OP_SET_SCAN_SETTINGS, OP_TMDB_CONFIG, @@ -72,6 +74,8 @@ _ADMIN_EXECUTORS = { OP_CHAT_PURGE: "_admin_exec_chat_purge", OP_ROOT_EJECT: "_admin_exec_root_eject", OP_ROOT_PLUG: "_admin_exec_root_plug", + OP_ROOT_PAUSE: "_admin_exec_root_pause", + OP_ROOT_RESUME: "_admin_exec_root_resume", } diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py index 5d4ed94..0ceefb5 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py @@ -71,6 +71,8 @@ _HANDLERS = { MNP.ROOT_REMOVE: ("_do_root_remove", INLINE), MNP.ROOT_EJECT: ("_do_root_eject", INLINE), MNP.ROOT_PLUG: ("_do_root_plug", INLINE), + MNP.ROOT_PAUSE: ("_do_root_pause", INLINE), + MNP.ROOT_RESUME: ("_do_root_resume", INLINE), MNP.KEYPAIR_BUNDLE_STORE: ("_do_keypair_bundle_store", SPAWNED), MNP.KEYPAIR_BUNDLE_DELETE: ("_do_keypair_bundle_delete", SPAWNED), MNP.USER_BLOB_STORE: ("_do_user_blob_store", SPAWNED), diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py index 237a359..c1f424d 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py @@ -1,5 +1,5 @@ """The operator's controls over a group's roots and scan pacing that MNP carries: -removing, ejecting and plugging a root. What widens the sharing, and the node's +removing, ejecting, plugging, pausing and resuming a root. What widens the sharing, and the node's own status, settings, roster and denylist, are the loopback API's and the CLI's (MNP 6.0).""" @@ -8,8 +8,10 @@ import logging from meshbay_common import MNP_VERSION from meshbay_common.adminop import ( OP_ROOT_EJECT, + OP_ROOT_PAUSE, OP_ROOT_PLUG, OP_ROOT_REMOVE, + OP_ROOT_RESUME, OP_SET_SCAN_SETTINGS, ) from meshbay_common.protocol import MNP @@ -201,6 +203,64 @@ class NodeOpsMixin: except Exception: pass + def _do_root_pause(self, msg: dict) -> None: + self._ask_root_index_op(msg, OP_ROOT_PAUSE) + + def _do_root_resume(self, msg: dict) -> None: + self._ask_root_index_op(msg, OP_ROOT_RESUME) + + def _ask_root_index_op(self, msg: dict, op: str) -> None: + target_group = str(msg.get("group_id", self._group_id or "")).strip() + root_name = str(msg.get("root_name", "")).strip() + if not target_group or not root_name: + self._send({"type": "error", "detail": "Missing group_id or root_name"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge( + op, root_name, + payload={"group_id": target_group, "root_name": root_name}, + group_id=target_group) + + async def _admin_exec_root_pause( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + await self._exec_root_index_op( + pending, transcript, sig, OP_ROOT_PAUSE, ops.pause_root, MNP.ROOT_PAUSE_ACK) + + async def _admin_exec_root_resume( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + await self._exec_root_index_op( + pending, transcript, sig, OP_ROOT_RESUME, ops.resume_root, + MNP.ROOT_RESUME_ACK) + + async def _exec_root_index_op(self, pending: dict, transcript: bytes, sig: bytes, + op: str, run, ack: str) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"{op}:{pending['subject'][:24]}") + return + p = pending["payload"] + try: + result = await self._run_op(run, p["group_id"], p["root_name"]) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + except Exception as e: + log.error("%s failed: %s", op, e, exc_info=True) + self._send({"type": "error", "detail": "Internal error"}) + return + self._audit(op, p["root_name"]) + # Broadcast, like eject's: every open roots table shows the new state. + notice = {"type": ack, "v": MNP_VERSION, **result} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + async def _retarget_indexer(self, group_id: str) -> None: """ Pick up a root that was just removed from node.toml. diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 03db8b6..9933750 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -449,6 +449,16 @@ def create_ui_app(state: dict) -> FastAPI: """Bring an ejected folder back.""" return await _op(lambda: ops.plug_root(state, group_id, root_name)) + @app.put("/api/groups/{group_id}/roots/{root_name}/pause") + async def pause_root(group_id: str, root_name: str): + """Stop indexing a folder. What is already indexed stays listed.""" + return await _op(lambda: ops.pause_root(state, group_id, root_name)) + + @app.put("/api/groups/{group_id}/roots/{root_name}/resume") + async def resume_root(group_id: str, root_name: str): + """Index a paused folder again, from where it stopped.""" + return await _op(lambda: ops.resume_root(state, group_id, root_name)) + @app.delete("/api/groups/{group_id}/roots/{root_name}") async def remove_root(group_id: str, root_name: str): """Remove a folder from a group. At least one must remain.""" diff --git a/packages/meshbay-node/tests/golden/cli.json b/packages/meshbay-node/tests/golden/cli.json index 48d2f74..d069510 100644 --- a/packages/meshbay-node/tests/golden/cli.json +++ b/packages/meshbay-node/tests/golden/cli.json @@ -4,7 +4,7 @@ "asked": [], "exit": 0, "stderr": "", - "stdout": "usage: meshbay-node [-h] [--hub-url HUB_URL] [--username USERNAME] [--dir DIR] [--yes]\n [--config CONFIG] [--group GROUP] [--link] [--writable] [--no-writable]\n [--removable] [--no-removable] [--open] [--name NAME]\n [--log-level {DEBUG,INFO,WARNING,ERROR}]\n [{init,reset,status,gek-init,gek,operator,member,group,root,file,video,chat,denylist,stun,transfers,reload,restart-daemon,autostart,service,calibrate-argon2}]\n [subcommand] [target] [value]\n\nMeshBay Node daemon\n\npositional arguments:\n {init,reset,status,gek-init,gek,operator,member,group,root,file,video,chat,denylist,stun,transfers,reload,restart-daemon,autostart,service,calibrate-argon2}\n init: provision config + keystore | reset: erase all node state | status:\n node state and keys | operator pair: pair a browser with this node |\n member list|invite|cancel|revoke|unpin | group list|add|remove | root\n list|add|remove|set|eject|plug | gek init|rotate | file list|rm | video\n rematch: re-resolve TMDB matches for a group's videos | chat\n status|rotate|encrypt-history|prune | denylist show|clear | stun\n list|add|remove|reset | transfers show|set|max-size|per-member: live\n transfer slots, the node-wide caps, the largest single upload, and how\n many one member may run at once in a group | reload: re-read node.toml\n (hot; systemd or the loopback API) | restart-daemon: restart the node\n (systemd unit, the Windows autostart launcher, or the service task,\n whichever applies) | autostart install|remove|start|stop|status (Windows:\n run meshbay-node at each sign-in, no admin) | service\n install|remove|start|stop|status (Windows: run at boot, before sign-in,\n needs admin once to install) | calibrate-argon2: benchmark\n subcommand 'pair' for operator; list|invite|revoke|unpin for member; list|add|remove\n for group; list|add|remove|set|eject|plug for root; init|rotate for gek;\n list|rm for file; rematch for video; show|clear for denylist;\n list|add|remove|reset for stun; show|set|max-size|per-member for\n transfers; install|remove|start|stop|status for autostart and for service\n target username for member invite|revoke|unpin (an optional e-mail label with\n --link, a link id for member cancel); group name for group add; file id\n for file rm; identifier for denylist clear; download cap for transfers\n set; size in GB for transfers max-size\n value the second value where a verb takes two: the upload cap for transfers set\n\noptions:\n -h, --help show this help message and exit\n --hub-url HUB_URL hub URL, for init (e.g. https://meshbay.org)\n --username USERNAME hub username, for init\n --dir DIR shared directory, for group add\n --yes skip the confirmation for destructive commands\n --config CONFIG Config file path\n --group GROUP group id (optional if only one is configured)\n --link member invite: an invitation link, for someone who may have no account yet\n (valid 7 days, single use)\n --writable root accepts member uploads (root add/set)\n --no-writable root is read-only (root add/set, group add)\n --removable mark root as removable (root set/add)\n --no-removable mark root as not removable (root set)\n --open group add: anyone the hub lists the group to may join (default: by\n invitation only)\n --name NAME root name (root add; defaults to directory basename)\n --log-level {DEBUG,INFO,WARNING,ERROR}\n", + "stdout": "usage: meshbay-node [-h] [--hub-url HUB_URL] [--username USERNAME] [--dir DIR] [--yes]\n [--config CONFIG] [--group GROUP] [--link] [--writable] [--no-writable]\n [--removable] [--no-removable] [--open] [--name NAME]\n [--log-level {DEBUG,INFO,WARNING,ERROR}]\n [{init,reset,status,gek-init,gek,operator,member,group,root,file,video,chat,denylist,stun,transfers,reload,restart-daemon,autostart,service,calibrate-argon2}]\n [subcommand] [target] [value]\n\nMeshBay Node daemon\n\npositional arguments:\n {init,reset,status,gek-init,gek,operator,member,group,root,file,video,chat,denylist,stun,transfers,reload,restart-daemon,autostart,service,calibrate-argon2}\n init: provision config + keystore | reset: erase all node state | status:\n node state and keys | operator pair: pair a browser with this node |\n member list|invite|cancel|revoke|unpin | group list|add|remove | root\n list|add|remove|set|eject|plug|pause|resume | gek init|rotate | file\n list|rm | video rematch: re-resolve TMDB matches for a group's videos |\n chat status|rotate|encrypt-history|prune | denylist show|clear | stun\n list|add|remove|reset | transfers show|set|max-size|per-member: live\n transfer slots, the node-wide caps, the largest single upload, and how\n many one member may run at once in a group | reload: re-read node.toml\n (hot; systemd or the loopback API) | restart-daemon: restart the node\n (systemd unit, the Windows autostart launcher, or the service task,\n whichever applies) | autostart install|remove|start|stop|status (Windows:\n run meshbay-node at each sign-in, no admin) | service\n install|remove|start|stop|status (Windows: run at boot, before sign-in,\n needs admin once to install) | calibrate-argon2: benchmark\n subcommand 'pair' for operator; list|invite|revoke|unpin for member; list|add|remove\n for group; list|add|remove|set|eject|plug|pause|resume for root;\n init|rotate for gek; list|rm for file; rematch for video; show|clear for\n denylist; list|add|remove|reset for stun; show|set|max-size|per-member for\n transfers; install|remove|start|stop|status for autostart and for service\n target username for member invite|revoke|unpin (an optional e-mail label with\n --link, a link id for member cancel); group name for group add; file id\n for file rm; identifier for denylist clear; download cap for transfers\n set; size in GB for transfers max-size\n value the second value where a verb takes two: the upload cap for transfers set\n\noptions:\n -h, --help show this help message and exit\n --hub-url HUB_URL hub URL, for init (e.g. https://meshbay.org)\n --username USERNAME hub username, for init\n --dir DIR shared directory, for group add\n --yes skip the confirmation for destructive commands\n --config CONFIG Config file path\n --group GROUP group id (optional if only one is configured)\n --link member invite: an invitation link, for someone who may have no account yet\n (valid 7 days, single use)\n --writable root accepts member uploads (root add/set)\n --no-writable root is read-only (root add/set, group add)\n --removable mark root as removable (root set/add)\n --no-removable mark root as not removable (root set)\n --open group add: anyone the hub lists the group to may join (default: by\n invitation only)\n --name NAME root name (root add; defaults to directory basename)\n --log-level {DEBUG,INFO,WARNING,ERROR}\n", "systemctl": [] }, "autostart no-such-sub": { @@ -550,7 +550,7 @@ "asked": [], "exit": 1, "stderr": "", - "stdout": "usage: meshbay-node root list|add|remove|set|eject|plug [name] [--group NAME]\n", + "stdout": "usage: meshbay-node root list|add|remove|set|eject|plug|pause|resume [name] [--group NAME]\n", "systemctl": [] }, "root plug media": { diff --git a/packages/meshbay-node/tests/golden/dispatch.json b/packages/meshbay-node/tests/golden/dispatch.json index a39c57d..384c38f 100644 --- a/packages/meshbay-node/tests/golden/dispatch.json +++ b/packages/meshbay-node/tests/golden/dispatch.json @@ -144,6 +144,14 @@ "_admin_exec_root_eject" ] }, + "root_pause": { + "audit": [], + "log": [], + "sent": [], + "spawned": [ + "_admin_exec_root_pause" + ] + }, "root_plug": { "audit": [], "log": [], @@ -160,6 +168,14 @@ "_admin_exec_root_remove" ] }, + "root_resume": { + "audit": [], + "log": [], + "sent": [], + "spawned": [ + "_admin_exec_root_resume" + ] + }, "search_listed": { "audit": [], "log": [], @@ -2020,7 +2036,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2343,7 +2359,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2362,7 +2378,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2381,7 +2397,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2400,7 +2416,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2670,7 +2686,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2684,7 +2700,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2698,7 +2714,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2712,7 +2728,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2726,7 +2742,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2740,7 +2756,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2754,7 +2770,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2768,7 +2784,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -3983,7 +3999,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -4002,7 +4018,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -4021,7 +4037,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -4040,7 +4056,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -10159,7 +10175,7 @@ "subject": "link:gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -12258,7 +12274,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -12277,7 +12293,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -12296,7 +12312,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13397,7 +13413,7 @@ "req_id": 4242, "token": null, "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13412,7 +13428,7 @@ "x" ], "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13425,7 +13441,7 @@ "req_id": 4242, "token": 7, "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13438,7 +13454,7 @@ "req_id": 4242, "token": "x", "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13451,7 +13467,7 @@ "req_id": 4242, "token": null, "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13466,7 +13482,7 @@ "x" ], "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13479,7 +13495,7 @@ "req_id": 4242, "token": 7, "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13492,7 +13508,7 @@ "req_id": 4242, "token": "x", "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13827,7 +13843,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13846,7 +13862,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13865,7 +13881,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14030,6 +14046,379 @@ "sent": [], "spawned": [] }, + "root_pause | challenged | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | challenged | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | challenged | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | challenged | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | fresh | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | fresh | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | fresh | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | fresh | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | member | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Missing group_id or root_name", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | member | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | member | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | member | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | operator | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Missing group_id or root_name", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | operator | lists": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "['x']", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_pause", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "['x']", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_pause | operator | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "7", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_pause", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "7", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_pause | operator | strings": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "x", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_pause", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "x", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_pause_ack | challenged | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | challenged | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | challenged | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | challenged | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | fresh | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | fresh | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | fresh | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | fresh | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | member | bare": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | member | lists": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | member | numbers": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | member | strings": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | operator | bare": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | operator | lists": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | operator | numbers": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | operator | strings": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, "root_plug | challenged | bare": { "audit": [], "log": [], @@ -14200,7 +14589,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14219,7 +14608,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14238,7 +14627,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14573,7 +14962,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14592,7 +14981,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14611,7 +15000,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14776,6 +15165,379 @@ "sent": [], "spawned": [] }, + "root_resume | challenged | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | challenged | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | challenged | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | challenged | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | fresh | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | fresh | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | fresh | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | fresh | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | member | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Missing group_id or root_name", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | member | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | member | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | member | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | operator | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Missing group_id or root_name", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | operator | lists": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "['x']", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_resume", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "['x']", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_resume | operator | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "7", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_resume", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "7", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_resume | operator | strings": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "x", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_resume", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "x", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_resume_ack | challenged | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | challenged | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | challenged | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | challenged | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | fresh | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | fresh | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | fresh | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | fresh | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | member | bare": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | member | lists": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | member | numbers": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | member | strings": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | operator | bare": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | operator | lists": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | operator | numbers": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | operator | strings": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, "search_listed | challenged | bare": { "audit": [], "log": [], @@ -17238,7 +18000,7 @@ "subject": "{\"language\":null,\"token\":null}", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -17281,7 +18043,7 @@ "subject": "{\"language\":\"x\",\"token\":\"sha256:2d711642b726b04401627ca9fbac32f5c8530fb1903cc4db02258717921a4881\"}", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] diff --git a/packages/meshbay-node/tests/test_root_availability.py b/packages/meshbay-node/tests/test_root_availability.py index 7cc890f..d31e11b 100644 --- a/packages/meshbay-node/tests/test_root_availability.py +++ b/packages/meshbay-node/tests/test_root_availability.py @@ -120,7 +120,7 @@ async def test_members_are_told_which_roots_are_unavailable(tmp_path): idx = await _indexer(_roots(films)) assert idx.index.roots == [ {"name": "Films", "kind": "generic", "available": True, - "writable": True, "removable": False, "ejected": False}] + "writable": True, "removable": False, "ejected": False, "paused": False}] (films / "a.mkv").unlink() films.rmdir() diff --git a/packages/meshbay-node/tests/test_root_paths_are_operator_only.py b/packages/meshbay-node/tests/test_root_paths_are_operator_only.py index 646c7b4..edbc62f 100644 --- a/packages/meshbay-node/tests/test_root_paths_are_operator_only.py +++ b/packages/meshbay-node/tests/test_root_paths_are_operator_only.py @@ -102,7 +102,7 @@ def test_the_cli_only_reads_fields_the_payload_carries(monkeypatch, tmp_path, ca path = Path("/tmp/x") name = "x" kind = "generic" - writable = removable = ejected = False + writable = removable = ejected = paused = False available = True offered = RootSet(roots=[_Any()]).describe(with_paths=True)[0] diff --git a/packages/meshbay-node/tests/test_root_pause.py b/packages/meshbay-node/tests/test_root_pause.py new file mode 100644 index 0000000..d9a4fcb --- /dev/null +++ b/packages/meshbay-node/tests/test_root_pause.py @@ -0,0 +1,313 @@ +""" +Pausing a root's indexing, and the walk that has to notice it. + +A scan of a large drive runs for hours, and nothing could stop one: the hashing +loop never looked at the root again once it had the file list. Removing the +directory let it read the whole drive anyway and threw the result away at the +end; ejecting it went on reading a drive the operator was about to unplug. + +A pause has to outlast the walk it stops. Reconcile indexes every file the index +lacks, so a walk that was merely cancelled was redone at the next tick; the flag +is what reconcile, the watcher and a restart all read. And it keeps what was +indexed: the index is rebuilt at every start, so a paused root is listed from the +hash cache rather than left empty. +""" + +import asyncio +from pathlib import Path +from types import SimpleNamespace + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node import ops +from meshbay_node.config import GroupConfig, RootSpec +from meshbay_node.indexer import indexer as indexer_mod +from meshbay_node.indexer.cache import IndexCache +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet +from meshbay_node.roster import Roster + +pytestmark = pytest.mark.asyncio + +GID = "g" * 32 + + +def _films(tmp_path: Path, n: int = 6) -> Path: + films = tmp_path / "Films" + films.mkdir() + for i in range(n): + (films / f"f{i}.mkv").write_bytes(f"film {i}".encode()) + return films + + +def _indexer(roots: RootSet, **kw) -> DirectoryIndexer: + return DirectoryIndexer(roots=roots, group_id=GID, + sk_node=Ed25519PrivateKey.generate(), gek=None, **kw) + + +def _names(idx: DirectoryIndexer) -> set[str]: + return {e.name for e in idx.index.entries} + + +def _count_reads(monkeypatch, after=None, at: int = 0) -> list[str]: + """Record every file actually read; run `after` once `at` have been.""" + read: list[str] = [] + real = indexer_mod._scan_file + + def counting(root, path): + read.append(path.name) + if after is not None and len(read) == at: + after() + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", counting) + return read + + +# ── The walk notices ───────────────────────────────────────────────────────── + +async def test_a_pause_stops_the_walk_at_the_next_file(tmp_path, monkeypatch): + roots = RootSet.build([{"path": str(_films(tmp_path))}]) + idx = _indexer(roots) + read = _count_reads(monkeypatch, lambda: setattr(roots.roots[0], "paused", True), at=2) + + await idx.initial_scan() + + assert len(read) == 2, f"read {len(read)} files after the pause" + assert len(_names(idx)) == 2, "the files hashed before the pause stay listed" + assert idx.progress.scanning is False + + +async def test_a_root_removed_mid_walk_is_not_read_to_the_end(tmp_path, monkeypatch): + films = _films(tmp_path) + other = tmp_path / "Other" + other.mkdir() + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + + def remove(): + idx.roots = RootSet.build([{"path": str(other)}]) + + read = _count_reads(monkeypatch, remove, at=1) + await idx.initial_scan() + + assert len(read) == 1, f"went on reading a removed root: {read}" + + +async def test_a_root_ejected_mid_walk_is_not_read_to_the_end(tmp_path, monkeypatch): + roots = RootSet.build([{"path": str(_films(tmp_path)), "removable": True}]) + idx = _indexer(roots) + read = _count_reads(monkeypatch, lambda: idx.eject_root("Films"), at=1) + + await idx.initial_scan() + + assert len(read) == 1, f"went on reading an ejected drive: {read}" + + +async def test_a_retarget_does_not_hide_the_pause_from_a_running_walk( + tmp_path, monkeypatch): + """The walk holds the Root it started with; a reload replaces every one.""" + films = _films(tmp_path) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + + def reload_then_pause(): + idx.roots = RootSet.build([{"path": str(films)}]) + idx.roots.roots[0].paused = True + + read = _count_reads(monkeypatch, reload_then_pause, at=1) + await idx.initial_scan() + + assert len(read) == 1 + + +# ── The pause outlasts the walk ────────────────────────────────────────────── + +async def test_reconcile_leaves_a_paused_root_alone(tmp_path): + films = _films(tmp_path, n=1) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + await idx.pause_root("Films") + + (films / "new.mkv").write_bytes(b"new film") + await idx.reconcile() + + assert "new.mkv" not in _names(idx), "reconcile redid the work the pause stopped" + + +async def test_the_watcher_leaves_a_paused_root_alone(tmp_path): + films = _films(tmp_path, n=1) + idx = _indexer(RootSet.build([{"path": str(films)}])) + await idx.initial_scan() + await idx.pause_root("Films") + + new = films / "new.mkv" + new.write_bytes(b"new film") + await idx._update_entry(new, deleted=False) + + assert "new.mkv" not in _names(idx) + + +async def test_a_paused_root_is_listed_from_the_cache_and_nothing_is_read( + tmp_path, monkeypatch): + """ + After a restart the index is empty until the walk refills it. Skipping a + paused root would empty it for members; reading it would undo the pause. + """ + films = _films(tmp_path, n=3) + cache = IndexCache(db_path=tmp_path / "cache.db") + await cache.open() + try: + await _indexer(RootSet.build([{"path": str(films)}]), cache=cache).initial_scan() + (films / "unhashed.mkv").write_bytes(b"never hashed") + + roots = RootSet.build([{"path": str(films), "paused": True}]) + idx = _indexer(roots, cache=cache) + read = _count_reads(monkeypatch) + await idx.initial_scan() + + assert read == [] + assert _names(idx) == {"f0.mkv", "f1.mkv", "f2.mkv"} + assert idx.progress.scanning is False + finally: + await cache.close() + + +async def test_a_pause_mid_rescan_keeps_the_entries_it_had(tmp_path, monkeypatch): + """A rescan drops the root's entries before it walks; stopping it there + would leave the operator with half a library.""" + films = _films(tmp_path, n=4) + cache = IndexCache(db_path=tmp_path / "cache.db") + await cache.open() + try: + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots, cache=cache) + await idx.initial_scan() + for f in films.iterdir(): + f.write_bytes(f.read_bytes() + b" edited") # every cache row stale + + read = _count_reads(monkeypatch, lambda: setattr(roots.roots[0], "paused", True), + at=1) + await idx._rescan_root(roots.roots[0]) + + assert len(read) == 1 + assert len(_names(idx)) == 1 + finally: + await cache.close() + + +# ── Resume ─────────────────────────────────────────────────────────────────── + +async def test_resume_indexes_what_is_missing_and_keeps_what_was_enriched(tmp_path): + films = _films(tmp_path, n=2) + idx = _indexer(RootSet.build([{"path": str(films)}])) + await idx.initial_scan() + for e in idx.index.entries: + e.artist = "Some Artist" + await idx.pause_root("Films") + (films / "new.mkv").write_bytes(b"new film") + + await idx.resume_root("Films") + await asyncio.gather(*idx._scan_tasks) + + assert "new.mkv" in _names(idx) + kept = [e for e in idx.index.entries if e.name != "new.mkv"] + assert all(e.artist == "Some Artist" for e in kept), \ + "resume replaced entries it should have left alone" + + +async def test_resume_returns_before_the_walk_ends(tmp_path): + films = _films(tmp_path, n=1) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + await idx.pause_root("Films") + + async with idx._scan_lock: + await asyncio.wait_for(idx.resume_root("Films"), timeout=1) + assert idx.progress.queued == ["Films"] + await asyncio.gather(*idx._scan_tasks) + assert idx.progress.queued == [] + + +# ── The flag is the operator's, and survives ───────────────────────────────── + +def _state(tmp_path: Path, roots: RootSet, roster: Roster) -> dict: + index = GroupIndex(group_id=GID, sk_node=Ed25519PrivateKey.generate()) + return { + "groups_ctx": {GID: {"index": index, "roots": roots, "gek": None}}, + "config": SimpleNamespace(groups=[SimpleNamespace(id=GID, roots=[])]), + "roster": roster, + "node_user_id": "operator", + } + + +async def test_pause_and_resume_persist_through_the_roster(tmp_path): + films = _films(tmp_path, n=1) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + state = _state(tmp_path, RootSet.build([{"path": str(films)}]), roster) + + out = await ops.pause_root(state, GID, "films") + assert out["status"] == "paused" + assert out["roots"][0]["paused"] is True + assert out["roots"][0]["available"] is True, "a paused root is still served" + assert await roster.paused_roots(GID) == {"films"} + assert (await ops.pause_root(state, GID, "Films"))["status"] == "already_paused" + + out = await ops.resume_root(state, GID, "Films") + assert out["status"] == "resumed" + assert await roster.paused_roots(GID) == set() + finally: + await roster.close() + + +async def test_removing_a_root_forgets_its_pause(tmp_path): + """A directory added back later under the same name starts indexed.""" + films = _films(tmp_path, n=1) + keep = tmp_path / "Keep" + keep.mkdir() + conf = tmp_path / "node.toml" + conf.write_text( + f'[[groups]]\nid = "{GID}"\nname = "g"\n\n' + f'[[groups.roots]]\npath = "{films.as_posix()}"\n\n' + f'[[groups.roots]]\npath = "{keep.as_posix()}"\n') + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + state = _state(tmp_path, RootSet.build( + [{"path": str(films)}, {"path": str(keep)}]), roster) + state["config_path"] = str(conf) + state["config"] = SimpleNamespace(groups=[GroupConfig( + id=GID, roots=[RootSpec(path=str(films)), RootSpec(path=str(keep))])]) + await ops.pause_root(state, GID, "Films") + + await ops.remove_root(state, GID, "Films") + + assert await roster.paused_roots(GID) == set() + finally: + await roster.close() + + +async def test_a_new_group_can_be_paused_during_its_first_scan(tmp_path): + """A group added while the node runs has an indexer, and no context until + that first scan ends — the scan most worth pausing.""" + films = _films(tmp_path, n=1) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + state = _state(tmp_path, roots, roster) + state["groups_ctx"] = {} + state["indexers"] = {GID: idx} + + out = await ops.pause_root(state, GID, "Films") + + assert out["status"] == "paused" + assert roots.roots[0].paused is True + finally: + await roster.close() diff --git a/packages/meshbay-node/tests/test_roots.py b/packages/meshbay-node/tests/test_roots.py index 0fe9f03..30bbc8a 100644 --- a/packages/meshbay-node/tests/test_roots.py +++ b/packages/meshbay-node/tests/test_roots.py @@ -290,9 +290,9 @@ def test_describe_reports_what_a_member_needs(tmp_path): described = roots.describe() assert described == [ {"name": "Media", "kind": "generic", "available": True, - "writable": True, "removable": False, "ejected": False}, + "writable": True, "removable": False, "ejected": False, "paused": False}, {"name": "Music", "kind": "audio", "available": True, - "writable": False, "removable": True, "ejected": False}, + "writable": False, "removable": True, "ejected": False, "paused": False}, ] # Deliberately no paths: a member is told what exists and whether it is # readable, not where on the operator's disk it lives. @@ -301,15 +301,16 @@ def test_describe_reports_what_a_member_needs(tmp_path): def test_describe_names_a_root_and_never_a_path(tmp_path): """ - What a member is told about a root: that it exists, and whether it is - readable and writable. Never where on the operator's disk it lives — that - is the operator's own view (`with_paths`), over their own channel. + What a member is told about a root: that it exists, whether it is + readable and writable, and whether its indexing is paused. Never where on + the operator's disk it lives — that is the operator's own view + (`with_paths`), over their own channel. """ (tmp_path / "Media").mkdir() roots = RootSet.build([_spec(tmp_path / "Media", writable=True)]) described = roots.describe()[0] assert set(described) == {"name", "kind", "available", "writable", - "removable", "ejected"} + "removable", "ejected", "paused"} # ── SAFE_UPLOAD_NAME ──────────────────────────────────────────────────────── |