diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
12 files changed, 437 insertions, 87 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.""" |