diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
4 files changed, 129 insertions, 10 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/cli/lifecycle.py b/packages/meshbay-node/src/meshbay_node/cli/lifecycle.py index 901c3d5..b461055 100644 --- a/packages/meshbay-node/src/meshbay_node/cli/lifecycle.py +++ b/packages/meshbay-node/src/meshbay_node/cli/lifecycle.py @@ -55,23 +55,36 @@ def _running_status(cfg) -> dict | None: return _await_daemon(cfg, since=0, timeout=0.1) +class NodeStillRunning(RuntimeError): + pass + + def _stop_node(cfg) -> str: """Stop the node, whichever session it runs in: through its own control API - first, so it shuts down properly, then by force. Returns how it went.""" + first, so it shuts down properly, then by force. Returns how it went, and + raises NodeStillRunning when a node process is still there at the end -- + never "stopped" about a process nobody saw go.""" import time from meshbay_node.platform import ( + _pid_alive, autostart_end, + kill_pid, + node_pids, request_graceful_stop, service_end, service_state, service_status, ) + answering = _running_status(cfg) + pid = answering.get("pid") if answering else None if request_graceful_stop(cfg.data_dir, cfg.node.ui_port): how = "stopped" - elif _running_status(cfg) is None: + elif answering is None and not node_pids(): how = "not running" else: + # Asked and not gone: its control API closes first, so it may answer + # nothing while its process is still there. how = "stopped (forced)" if service_status()["installed"]: # Also leaves the task "Ready": a /run while it still reads "Running" @@ -80,7 +93,17 @@ def _stop_node(cfg) -> str: deadline = time.monotonic() + 15 while service_state().lower() == "running" and time.monotonic() < deadline: time.sleep(0.25) + if pid is not None and _pid_alive(int(pid)): + kill_pid(int(pid)) # the one that answered, by pid, wherever it came from autostart_end() # a node in this session that would not stop + deadline = time.monotonic() + 5 + while (left := node_pids()) and time.monotonic() < deadline: + time.sleep(0.25) + if left: + raise NodeStillRunning( + f"meshbay-node is still running (pid {', '.join(map(str, left))}) and could " + "not be stopped from here -- it was probably started from an " + "administrator prompt. Stop it there, or end it in Task Manager.") return how @@ -94,7 +117,12 @@ def _start_and_confirm(args, action: str) -> None: cfg = load_config(args.config or DEFAULT_CONFIG_PATH) if action == "restart": - _stop_node(cfg) + try: + _stop_node(cfg) + except NodeStillRunning as e: + # Starting another would find the port taken and quit at once. + print(f"Could not restart the node: {e}") + sys.exit(1) else: already = _running_status(cfg) if already is not None: @@ -122,6 +150,21 @@ def _start_and_confirm(args, action: str) -> None: "its files were not replaced, or another copy was started.") +def _print_stop(args) -> None: + try: + print(_stop_node(load_config(args.config or DEFAULT_CONFIG_PATH))) + except NodeStillRunning as e: + print(e) + sys.exit(1) + + +def _stop_node_or_warn(args) -> None: + try: + _stop_node(load_config(args.config or DEFAULT_CONFIG_PATH)) + except NodeStillRunning as e: + print(f"warning: {e}") + + def restart_daemon(args) -> None: if sys.platform == "win32": _start_and_confirm(args, "restart") @@ -161,7 +204,7 @@ def autostart(args) -> None: elif sub == "start": _start_and_confirm(args, "start") elif sub == "stop": - print(_stop_node(load_config(args.config or DEFAULT_CONFIG_PATH))) + _print_stop(args) elif sub == "status": st = _plat.autostart_status() if st["installed"]: @@ -186,7 +229,7 @@ def service(args) -> None: if sub == "install": # A node already running in this session holds the control API's port: # the service's own would exit at once, leaving the old one in charge. - _stop_node(load_config(args.config or DEFAULT_CONFIG_PATH)) + _stop_node_or_warn(args) try: _plat.service_install() except RuntimeError as e: @@ -201,7 +244,7 @@ def service(args) -> None: elif sub == "remove": # Deleting a task does not end its running instance: stop the node # first, or it runs on in session 0 with nothing left to stop it. - _stop_node(load_config(args.config or DEFAULT_CONFIG_PATH)) + _stop_node_or_warn(args) _plat.service_remove() print(f"Removed the {_plat.TASK_NAME!r} scheduled task.") elif sub == "start": @@ -211,7 +254,7 @@ def service(args) -> None: sys.exit(1) _start_and_confirm(args, "start") elif sub == "stop": - print(_stop_node(load_config(args.config or DEFAULT_CONFIG_PATH))) + _print_stop(args) elif sub == "status": st = _plat.service_status() if st["installed"]: diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index 2cd8a8f..eb694ae 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -403,7 +403,11 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg = Config() if path.exists(): - raw = tomllib.loads(path.read_text(encoding="utf-8")) + # utf-8-sig: Windows PowerShell 5.1's `Set-Content -Encoding utf8` and + # older Notepads write a BOM, which TOML refuses -- and a service node + # then stopped before it had said anything (found editing node.toml + # on a real install). + raw = tomllib.loads(path.read_text(encoding="utf-8-sig")) hub = raw.get("hub", {}) cfg.hub.url = hub.get("url", cfg.hub.url) diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index e1aac41..6fba120 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -31,6 +31,7 @@ import os import signal import socket import sys +import threading import time from dataclasses import asdict from pathlib import Path @@ -81,6 +82,31 @@ if WEBRTC_AVAILABLE: log = logging.getLogger(__name__) +# How long a stopped node's process may outlive its _shutdown(), and the most +# _shutdown() itself may take. Python's own exit waits for every worker thread, +# and one still walking or hashing a large tree kept the process alive for as +# long as that took, with its control API already closed: the desktop app saw +# no node and said "Stopped" about a process that was still there. +EXIT_GRACE_SECS = 3.0 +SHUTDOWN_DEADLINE_SECS = 30.0 + + +def exit_after(seconds: float, why: str, code: int = 0) -> threading.Timer: + """End this process in `seconds`, whatever is still running in it.""" + def _exit() -> None: + busy = sorted(t.name for t in threading.enumerate() + if t is not threading.current_thread() and not t.daemon + and t is not threading.main_thread()) + log.warning("%s -- ending the process now (still busy: %s)", + why, ", ".join(busy) or "nothing") + logging.shutdown() + os._exit(code) + timer = threading.Timer(seconds, _exit) + timer.daemon = True + timer.start() + return timer + + # ── Hub WS sender bridge ───────────────────────────────────────────────────── class _WsSender: @@ -101,6 +127,9 @@ def _root_shape(roots) -> set[tuple]: class NodeDaemon(EnrichmentMixin): + # Set by main(), never by a test that runs a daemon in its own process. + exit_process_when_stopped = False + def __init__(self, config: Config, config_path: Path = DEFAULT_CONFIG_PATH): self._config = config self._config_path = config_path @@ -1564,6 +1593,9 @@ class NodeDaemon(EnrichmentMixin): async def _shutdown(self) -> None: log.info("Shutting down...") self._state["status"] = "stopping" + if self.exit_process_when_stopped: + exit_after(SHUTDOWN_DEADLINE_SECS, + f"shutdown took more than {SHUTDOWN_DEADLINE_SECS:.0f}s", code=1) for handle in self._pending_broadcasts.values(): handle.cancel() @@ -1624,6 +1656,8 @@ class NodeDaemon(EnrichmentMixin): token_file.unlink(missing_ok=True) log.info("Node stopped") + if self.exit_process_when_stopped: + exit_after(EXIT_GRACE_SECS, "the node has stopped") # ── Entry point ─────────────────────────────────────────────────────────────── @@ -1668,7 +1702,15 @@ def main() -> None: # Printed for whoever ran it, and logged too: a daemon started by the # service task has no console, and these are why it would refuse to start. - cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + config_path = args.config or DEFAULT_CONFIG_PATH + try: + cfg = load_config(config_path) + except Exception as e: + # Uncaught, this went to a stderr that Task Scheduler discards: the + # node died with "Logging to ..." as its last word. + print(f"Error: cannot read {config_path}: {e}") + log.error("cannot read %s: %s", config_path, e) + sys.exit(1) if not cfg.hub.username: print("Error: hub.username not set in config. Run: meshbay-node init") log.error("hub.username not set in config. Run: meshbay-node init") @@ -1681,7 +1723,8 @@ def main() -> None: log.error("%s", e) sys.exit(1) - daemon = NodeDaemon(cfg, Path(args.config or DEFAULT_CONFIG_PATH)) + daemon = NodeDaemon(cfg, Path(config_path)) + daemon.exit_process_when_stopped = True try: asyncio.run(daemon.run()) except ControlPortTaken as e: diff --git a/packages/meshbay-node/src/meshbay_node/platform.py b/packages/meshbay-node/src/meshbay_node/platform.py index 88db83d..c0e4d00 100644 --- a/packages/meshbay-node/src/meshbay_node/platform.py +++ b/packages/meshbay-node/src/meshbay_node/platform.py @@ -397,6 +397,35 @@ def autostart_end() -> None: subprocess.run(argv, capture_output=True) +def node_pids() -> list[int]: + """Every meshbay-node.exe running, in any session, except this process and + its parent (the CLI is meshbay-node.exe too). Empty off Windows. + + What says whether a node is still there after a stop: its control API + closes first, so a process that has not exited yet answers nothing and + used to count as gone. + """ + if sys.platform != "win32": + return [] + r = subprocess.run(["tasklist", "/FI", f"IMAGENAME eq {NODE_IMAGE}", "/NH", "/FO", "CSV"], + capture_output=True, text=True) + mine = {os.getpid(), os.getppid()} + pids = [] + for line in r.stdout.splitlines(): + cells = [c.strip('"') for c in line.split('","')] + if len(cells) > 1 and cells[1].isdigit() and int(cells[1]) not in mine: + pids.append(int(cells[1])) + return pids + + +def kill_pid(pid: int) -> None: + """Force-stop one process and its children (its ffmpeg transcodes). Reaches + this session only, unless the caller is elevated. Windows only: systemd + stops the node everywhere else.""" + if sys.platform == "win32": + subprocess.run(["taskkill", "/F", "/T", "/PID", str(pid)], capture_output=True) + + # ── Service mode (Windows, opt-in at install time) ─────────────────────────── # # The Startup-folder .vbs above only ever runs after *this* user signs in. A |