diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index 237ca09..4f091d2 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -17,6 +17,7 @@ from __future__ import annotations import argparse import copy +import json import os import secrets import socket @@ -258,6 +259,85 @@ DEFAULT_PORT = 8000 PORT_TRIES = 20 +#: Where a serving engine records itself, beside the data it is serving. Read +#: to tell "this directory's engine is already up" from "something else has the +#: port", which are the two ways a second `serve` fails to be what was wanted. +PIDFILE = "serve.pid" + + +def write_pidfile(data_dir: Path, port: int) -> Path: + """Record which process is serving this directory, and where.""" + path = data_dir / PIDFILE + path.write_text(json.dumps({"pid": os.getpid(), "port": port})) + return path + + +def read_pidfile(data_dir: Path) -> dict[str, int] | None: + """The engine serving this directory, if one still is. + + A process killed outright leaves the file behind, so the pid is checked + rather than believed — a stale file is the same as no file. + """ + try: + record = json.loads((data_dir / PIDFILE).read_text()) + pid, port = int(record["pid"]), int(record["port"]) + except (OSError, ValueError, KeyError, TypeError): + return None + try: + os.kill(pid, 0) + except (OSError, ProcessLookupError): + return None + return {"pid": pid, "port": port} + + +def _token_for(data_dir: Path) -> str: + """The credential this directory's last engine wrote, if it wrote one.""" + from fluksio.sdk.client import config_path + + try: + return str(json.loads(config_path(data_dir).read_text()).get("token", "")) + except (OSError, ValueError, AttributeError): + return "" + + +def probe_engine(url: str, token: str, client: Any = None) -> str: + """Who is on this port: ``ours``, ``foreign``, or ``other``. + + ``ours`` means an engine serving *this* data directory, which is what + makes stopping it something this command may offer. The proof is the + token: it is signed with this directory's secret key, so an engine that + accepts it is one reading this directory's database. A Fluksio belonging + to another installation answers the health check and refuses the token, + and is only ever named — never stopped from here. + """ + import logging + + import httpx + + # Two requests nobody asked for, on a start that is otherwise quiet until + # uvicorn's own banner. httpx logs every one of them at INFO. + noisy = logging.getLogger("httpx") + was = noisy.level + noisy.setLevel(logging.WARNING) + http = client or httpx.Client(timeout=2.0) + try: + health = http.get(f"{url}/api/v1/utils/health-check/") + if health.status_code != 200: + return "other" + # No token at all is asked unauthenticated: `Bearer ` is not a legal + # header value, and a Fluksio this directory cannot prove is its own + # is foreign — which is the answer that never stops anything. + auth = {"Authorization": f"Bearer {token}"} if token else {} + answer = http.get(f"{url}/api/v1/observability/summary", headers=auth) + return "ours" if answer.status_code == 200 else "foreign" + except Exception: + return "other" + finally: + noisy.setLevel(was) + if client is None: + http.close() + + def _free_port(host: str, start: int) -> int: """The first port from ``start`` that nothing is listening on. @@ -312,15 +392,34 @@ def cmd_serve(args: argparse.Namespace) -> int: from fluksio.flow import modules from fluksio.main import app + # The client talks to this engine, and 0.0.0.0 is not an address to talk + # to — it is a statement about which interfaces to listen on. + reachable = "127.0.0.1" if args.host in ("0.0.0.0", "::", "") else args.host + port = args.port if port is None: port = _free_port(args.host, DEFAULT_PORT) if port != DEFAULT_PORT: - _say(f"Port {DEFAULT_PORT} is in use; serving on {port} instead.") + # Moving off the port quietly makes starting a second engine for + # one directory look like it worked. Two of them on one SQLite + # file is not a supported shape — two *installations* on one + # machine is — so the one already up is named instead. + url = f"http://{reachable}:{DEFAULT_PORT}" + who = probe_engine(url, _token_for(data_dir)) + if who == "ours": + running = read_pidfile(data_dir) + where = f" (pid {running['pid']})" if running else "" + _say(f"An engine for {data_dir} is already serving at {url}{where}.") + _say(" fluksio status talks to it; stop it to start another.") + return 0 + if who == "foreign": + _say( + f"Port {DEFAULT_PORT} holds another installation's Fluksio; " + f"serving on {port} instead." + ) + else: + _say(f"Port {DEFAULT_PORT} is in use; serving on {port} instead.") - # The client talks to this engine, and 0.0.0.0 is not an address to talk - # to — it is a statement about which interfaces to listen on. - reachable = "127.0.0.1" if args.host in ("0.0.0.0", "::", "") else args.host url = f"http://{reachable}:{port}" token_path = _sign_in(admin_id, url, data_dir) @@ -351,13 +450,17 @@ def cmd_serve(args: argparse.Namespace) -> int: # One process: it holds the flow engine, and a second worker would be a # second engine — duplicated subscriptions, cron ticks and webhooks. - uvicorn.run( - app, - host=args.host, - port=port, - log_level=args.log_level, - log_config=_log_config(args.log_level), - ) + pidfile = write_pidfile(data_dir, port) + try: + uvicorn.run( + app, + host=args.host, + port=port, + log_level=args.log_level, + log_config=_log_config(args.log_level), + ) + finally: + pidfile.unlink(missing_ok=True) return 0 diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 9d6c156..afca09e 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -567,6 +567,58 @@ def test_run_and_sweep_take_what_to_sync() -> None: assert parser.parse_args(["sweep", "train"]).sync == [] +def test_a_serving_engine_records_itself_until_it_stops(tmp_path) -> None: + """A pid nobody is running is the same as no pidfile at all.""" + import os + + from fluksio.cli import PIDFILE, read_pidfile, write_pidfile + + assert read_pidfile(tmp_path) is None + + written = write_pidfile(tmp_path, 8000) + assert read_pidfile(tmp_path) == {"pid": os.getpid(), "port": 8000} + + # Killed outright: the file outlives the process it names. + written.write_text('{"pid": 2147483646, "port": 8000}') + assert read_pidfile(tmp_path) is None + + written.write_text("not json") + assert read_pidfile(tmp_path) is None + + +def test_who_holds_the_port_is_told_apart_by_the_token(tmp_path) -> None: + """Only this directory's own engine may be reported as already up. + + The token is signed with this directory's secret key, so an engine that + accepts it is one reading this directory's database. Another + installation's Fluksio answers the health check and refuses it. + """ + import httpx + + from fluksio.cli import probe_engine + + def engine(health: int, summary: int): + def handle(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/health-check/"): + return httpx.Response(health) + return httpx.Response(summary) + + return httpx.Client(transport=httpx.MockTransport(handle)) + + with engine(200, 200) as client: + assert probe_engine("http://x", "t", client) == "ours" + with engine(200, 401) as client: + assert probe_engine("http://x", "t", client) == "foreign" + # A directory with no credential yet cannot prove anything is its own, and + # `Bearer ` is not a legal header value — so it asks without one. + with engine(200, 401) as client: + assert probe_engine("http://x", "", client) == "foreign" + # Somebody else's dev server, or nothing listening at all. + with engine(404, 404) as client: + assert probe_engine("http://x", "t", client) == "other" + assert probe_engine("http://127.0.0.1:1", "t") == "other" + + def test_serve_moves_off_a_port_that_is_taken() -> None: """A first start should not die on somebody else's dev server.""" import socket