diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 207 |
1 files changed, 184 insertions, 23 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 6d4f172..7e3ebc1 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -111,6 +111,7 @@ class NodeDaemon: "quic_port": config.node.quic_port, "endpoint_hint": None, "indexes": {}, + "indexers": {}, } self._quic_server = None self._webrtc = None @@ -243,6 +244,7 @@ class NodeDaemon: await indexer.start() self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index + self._state["indexers"][group_cfg.id] = indexer log.info("Indexing group %s: %s (%d files)", group_cfg.name, ", ".join(f"{r.name}={r.path}" for r in roots), @@ -416,12 +418,15 @@ class NodeDaemon: self._state["webrtc"] = self._webrtc self._state["quic_server"] = self._quic_server self._state["hub"] = hub + self._state["reload_fn"] = self._reload_config # Rotating a key has to reach every transport holding a copy of it, # and clearing the denylist has to reach the one the handshake # consults — so both are published rather than reachable only # through the object that happens to own them. self._state["denylist"] = self._denylist self._state["pk_x25519_raw"] = pk_x_raw + self._state["sk_x25519_raw"] = sk_x_raw + self._state["sk_ed25519"] = keys.sk_ed25519 self._state["status"] = "running" log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s", @@ -460,19 +465,13 @@ class NodeDaemon: async def _reload_config(self) -> None: """ - Re-read node.toml on SIGHUP. + Re-read node.toml and reconcile groups. - Deliberately narrow: it picks up **root changes on groups already - hosted**, which is what an operator adjusts day to day, and reports - anything else as needing a restart. Adding or removing a whole group - means new indexers, chat stores, GEK loads and transport contexts, and - doing that under a live daemon is how a half-built group ends up serving - content. Saying "restart for that" is honest and costs one restart. - - Nothing here touches connections: a member watching a film keeps - watching it. + Handles root changes on existing groups, hot-loads new groups, and + tears down removed groups. Existing connections are untouched: a member + watching a film keeps watching it. """ - log.info("SIGHUP — re-reading %s", self._config_path) + log.info("Reloading config from %s", self._config_path) try: fresh = load_config(self._config_path) except Exception as e: @@ -482,13 +481,8 @@ class NodeDaemon: groups_ctx = self._state.get("groups_ctx") or {} hosted = set(groups_ctx) incoming = {g.id for g in fresh.groups if g.id} - if incoming != hosted: - added = ", ".join(sorted(incoming - hosted)) or "none" - removed = ", ".join(sorted(hosted - incoming)) or "none" - log.warning("Group set changed (added: %s, removed: %s) — restart the " - "daemon for that; roots of existing groups reloaded anyway", - added, removed) + # ── Root changes on existing groups ────────────────────────────── changed = 0 for group_cfg in fresh.groups: ctx = groups_ctx.get(group_cfg.id) @@ -515,9 +509,112 @@ class NodeDaemon: ctx["roots"] = roots changed += 1 + # ── Hot-load new groups ────────────────────────────────────────── + added_names = [] + sk_ed = self._state.get("sk_ed25519") + sk_x_raw = self._state.get("sk_x25519_raw") + pk_x_raw = self._state.get("pk_x25519_raw") + node_user_id = self._state.get("node_user_id") + data_dir = fresh.data_dir + + for group_cfg in fresh.groups: + if group_cfg.id in hosted: + continue + if not group_cfg.id or not group_cfg.roots: + log.warning("New group %r has no id or roots — skipping", + group_cfg.name) + continue + if not sk_ed: + log.warning("Cannot hot-load %r — signing key not available", + group_cfg.name) + continue + + try: + roots = RootSet.build([asdict(r) for r in group_cfg.roots]) + except RootError as e: + log.error("New group %r: %s — skipping", group_cfg.name, e) + continue + roots.refresh_availability() + + gek = None + if group_cfg.visibility == "private" and sk_x_raw and pk_x_raw: + gek = await self._load_gek( + group_cfg.id, node_user_id, sk_x_raw, pk_x_raw) + if gek: + log.info("GEK loaded for new group %s", group_cfg.id[:8]) + + indexer = DirectoryIndexer( + roots=roots, + group_id=group_cfg.id, + sk_node=sk_ed, + gek=gek, + on_change=self._on_index_change, + ) + await indexer.start() + self._indexers.append(indexer) + self._state["indexes"][group_cfg.id] = indexer.index + self._state["indexers"][group_cfg.id] = indexer + + data_dir.mkdir(parents=True, exist_ok=True) + chat_db = data_dir / group_cfg.id[:16] / "chat.db" + store = ChatStore(db_path=chat_db) + await store.open() + self._chat_stores[group_cfg.id] = store + + new_ctx = { + "gek": gek, + "roots": roots, + "index": indexer.index, + "visibility": group_cfg.visibility, + "join_policy": group_cfg.join_policy, + "member_upload": ( + await self._roster.member_upload_allowed(group_cfg.id) + if self._roster else True), + "chat_store": store, + } + groups_ctx[group_cfg.id] = new_ctx + + if self._webrtc: + self._webrtc._ctx["groups"][group_cfg.id] = new_ctx + log.info("Hot-loaded group %s (%s, %d roots)", + group_cfg.name, group_cfg.id[:8], len(roots)) + added_names.append(group_cfg.name) + + # ── Tear down removed groups ───────────────────────────────────── + removed_names = [] + for gid in hosted - incoming: + indexer = next((i for i in self._indexers + if i.group_id == gid), None) + if indexer: + try: + await indexer.stop() + except Exception: + pass + self._indexers.remove(indexer) + store = self._chat_stores.pop(gid, None) + if store: + try: + await store.close() + except Exception: + pass + self._state["indexes"].pop(gid, None) + self._state["indexers"].pop(gid, None) + old_name = gid[:8] + for g_cfg in self._config.groups: + if g_cfg.id == gid: + old_name = g_cfg.name + break + groups_ctx.pop(gid, None) + if self._webrtc and self._webrtc._ctx.get("groups") is not groups_ctx: + self._webrtc._ctx["groups"].pop(gid, None) + log.info("Unloaded group %s (%s)", old_name, gid[:8]) + removed_names.append(old_name) + self._config = fresh self._state["config"] = fresh - log.info("Reload complete — %d group(s) re-rooted", changed) + self._state["groups"] = [g.name for g in fresh.groups] + log.info("Reload complete — %d re-rooted, %d added, %d removed", + changed, len(added_names), len(removed_names)) async def _login_with_retry(self, hub: HubClient): """Login to hub, retrying if the node key hasn't been linked yet.""" @@ -782,16 +879,18 @@ def main() -> None: parser.add_argument("command", nargs="?", choices=["init", "status", "ui", "gek-init", "gek", "operator", "member", "group", "file", - "denylist", "reload", "calibrate-argon2"], + "denylist", "reload", "restart-daemon", + "calibrate-argon2"], help="init: write example config | status: node state and keys " "| ui: print the admin UI URL | operator pair: pair a " "browser with this node | member list|invite|revoke|unpin " - "| group list|add | gek init|rotate | file list|rm " + "| group list|add|remove | gek init|rotate | file list|rm " "| denylist show|clear | reload: re-read node.toml " + "| restart-daemon: full stop + start " "| calibrate-argon2: benchmark") parser.add_argument("subcommand", nargs="?", help="'pair' for operator; list|invite|revoke|unpin for " - "member; list|add for group; init|rotate for gek; " + "member; list|add|remove for group; init|rotate for gek; " "list|rm for file; show|clear for denylist") parser.add_argument("target", nargs="?", help="username for member invite|revoke|unpin; group name " @@ -813,7 +912,8 @@ def main() -> None: # Query commands print a report; library logging would interleave with it. quiet = args.command in ("status", "ui", "gek-init", "gek", "operator", - "member", "group", "file", "denylist", "reload") + "member", "group", "file", "denylist", "reload", + "restart-daemon") logging.basicConfig( level=logging.ERROR if quiet else getattr(logging, args.log_level), format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", @@ -1051,6 +1151,49 @@ def main() -> None: print("watch the result: tail -f /tmp/meshbay-node.log") return + if args.command == "restart-daemon": + import os as _os + import signal as _signal + import subprocess as _subprocess + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + pid_out = _subprocess.run( + ["pgrep", "-f", "--", r"-m meshbay_node\.daemon$"], + capture_output=True, text=True) + pids = [int(x) for x in pid_out.stdout.split()] + if pids: + for pid in pids: + _os.kill(pid, _signal.SIGTERM) + print(f"stopped {len(pids)} daemon process(es)") + for pid in pids: + try: + _os.waitpid(pid, 0) + except ChildProcessError: + import time as _time + _time.sleep(2) + else: + print("no running daemon found — starting fresh") + log_path = "/tmp/meshbay-node.log" + config_flag = ["--config", str(args.config)] if args.config else [] + _subprocess.Popen( + [sys.executable, "-m", "meshbay_node.daemon"] + config_flag, + stdout=open(log_path, "a"), + stderr=_subprocess.STDOUT, + start_new_session=True, + ) + import time as _time + _time.sleep(3) + pid_out2 = _subprocess.run( + ["pgrep", "-f", "--", r"-m meshbay_node\.daemon$"], + capture_output=True, text=True) + new_pids = [int(x) for x in pid_out2.stdout.split()] + if new_pids: + print(f"daemon started (PID {new_pids[0]})") + print(f"log: tail -f {log_path}") + else: + print(f"daemon may have failed to start — check {log_path}") + sys.exit(1) + return + if args.command == "denylist": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "show" @@ -1159,8 +1302,26 @@ def main() -> None: f"--group {g['name']}") return + if args.subcommand == "remove": + if not args.target: + print("usage: meshbay-node group remove <name>") + sys.exit(1) + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + if not args.yes: + answer = input(f"Remove group '{args.target}' from this node? [y/N] ") + if answer.lower() not in ("y", "yes"): + print("cancelled") + return + out = _daemon_api(cfg, "/api/groups/detach", method="POST", + body={"name": args.target}) + print(f"{out['name']} ({out['group_id'][:8]}) removed from {out['config']}") + print() + print("Restart the daemon to stop hosting it:") + print(" meshbay-node restart-daemon") + return + if args.subcommand != "add": - print("usage: meshbay-node group list|add <name> --dir <path> [--upload-dir <path>]") + print("usage: meshbay-node group list|add|remove <name>") sys.exit(1) if not args.target or not args.dir: print("usage: meshbay-node group add <name> --dir <path> [--upload-dir <path>]") |