diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index 54af8ec..2b86006 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -368,6 +368,9 @@ RUN_COLUMNS = ( "duration_ms", "seed", "group_id", + #: Set on a run made by retrying another, so a sweep that was completed + #: rather than repeated says which row replaced which. + "parent_id", "code_digest", "origin_commit", ) @@ -639,6 +642,20 @@ async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any: return run +@router.post("/{run_id}/retry", response_model=RunRow, status_code=202) +async def retry_run(run_id: str, request: Request, user: CurrentUser) -> Any: + """Run the same thing again, as a new run pointing back at this one.""" + service = _service(request) + try: + return await run_in_threadpool(service.retry, run_id, user.email) + except FlowNotFound as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except RunRejected as exc: + detail = str(exc) + status = 404 if detail.startswith("No run ") else 422 + raise HTTPException(status_code=status, detail=detail) from exc + + @router.delete("/{run_id}", status_code=204) def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response: """Forget a run and everything hanging off it. diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index 2877f1c..6e472f2 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -186,9 +186,9 @@ def _prepare(data_dir: Path) -> None: def _enroll(portal: str, code: str, as_email: str | None) -> int: from sqlmodel import Session + from fluksio.cloud import config as cloud_config from fluksio.cloud import enroll as enroll_mod from fluksio.core.bootstrap import ensure_superuser, pick_superuser - from fluksio.core.config import settings from fluksio.core.db import engine with Session(engine) as session: @@ -203,19 +203,15 @@ def _enroll(portal: str, code: str, as_email: str | None) -> int: # Read before the session closes: the instance is detached after it, # and touching an attribute then goes back to a database that is gone. email = user.email + previous = cloud_config.load() try: config = enroll_mod.enroll(session, user, portal, code) - except enroll_mod.AlreadyEnrolled as exc: - print( - f"error: {exc.detail} (see {settings.CLOUD_CONFIG_FILE}).", - file=sys.stderr, - flush=True, - ) - return 1 except enroll_mod.EnrollError as exc: print(f"error: {exc.detail}", file=sys.stderr, flush=True) return 1 + if previous is not None and previous.instance_id != config.instance_id: + _say(f"Replaced the connection to {previous.portal_url}.") _say( f"Connected to {config.portal_url} as {email} (instance {config.instance_id})." ) @@ -387,6 +383,26 @@ def _free_port(host: str, start: int) -> int: return start +def already_serving(data_dir: Path, host: str) -> str: + """Where this directory's engine is answering, if one is. + + Two engines on one SQLite file is not a supported shape, and the second + one does more than fail: it repoints `client.json` at itself, so every + later CLI call goes to a port that dies with it. The pidfile says where to + look and the token says whether what answers there is ours. + """ + running = read_pidfile(data_dir) + if running is None: + return "" + reachable = "127.0.0.1" if host in ("0.0.0.0", "::", "") else host + url = f"http://{reachable}:{running['port']}" + # A pid can be reused, so the file alone is not proof. What answers has to + # accept this directory's token as well. + if probe_engine(url, _token_for(data_dir)) != "ours": + return "" + return f"{url} (pid {running['pid']})" + + def cmd_serve(args: argparse.Namespace) -> int: # At a terminal this is a dashboard with the engine as a child of it. The # import is here rather than at the top because it is only ever needed on @@ -397,6 +413,14 @@ def cmd_serve(args: argparse.Namespace) -> int: return run_tui(args) data_dir = _data_dir(args.data_dir, args.shared) + # Before `_prepare`, so a refusal never migrates the database another + # engine is serving out of, and before `_sign_in`, so the credential keeps + # pointing at the engine that is actually up. + where = already_serving(data_dir, args.host) + if where: + _say(f"An engine for {data_dir} is already serving at {where}.") + _say(" fluksio status talks to it; stop it to start another.") + return 0 for flag, name in CONCURRENCY_FLAGS.items(): value = getattr(args, flag, None) if value is not None: diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index 3045c9e..100a074 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -60,6 +60,11 @@ MAX_IN_FLIGHT = 256 ENROL_POLL_S = 3.0 +def _identity(config: cloud_config.CloudConfig | None) -> tuple[str, str] | None: + """Which enrolment a link is running on, so a replaced one is noticed.""" + return (config.instance_id, config.token) if config is not None else None + + def start(app: FastAPI) -> None: """Dial the portal, replacing any link already up.""" existing = getattr(app.state, "cloud_task", None) @@ -67,6 +72,7 @@ def start(app: FastAPI) -> None: existing.cancel() connector = CloudConnector(app) app.state.cloud_connector = connector + app.state.cloud_identity = _identity(cloud_config.load()) app.state.cloud_task = asyncio.create_task( connector.serve_forever(), name="cloud-connector" ) @@ -85,7 +91,7 @@ async def watch_enrolment(app: FastAPI) -> None: task = getattr(app.state, "cloud_task", None) if task is not None and task.done(): # It returns of its own accord when the config goes away, which is - # what `fluksio disconnect` and the portal's own Disconnect do. + # what Disconnect, here or on the portal, does. app.state.cloud_task = None app.state.cloud_connector = None task = None @@ -94,7 +100,17 @@ async def watch_enrolment(app: FastAPI) -> None: # started — which, started from here, is a restart every few seconds. # A config that is fine but unreachable keeps its task, and the # retrying belongs to the connector rather than to this. - if task is None and cloud_config.load() is not None: + config = cloud_config.load() + if task is not None and _identity(config) != getattr( + app.state, "cloud_identity", None + ): + # Enrolled again, at this portal or another one. `fluksio enroll` + # is its own process and cannot cancel this task, so the link would + # otherwise stay up on the credential that was replaced. + logger.info("Enrolment replaced while running; redialling") + task.cancel() + task = None + if task is None and config is not None: logger.info("Enrolled while running; dialling the portal") start(app) diff --git a/backend/fluksio/cloud/enroll.py b/backend/fluksio/cloud/enroll.py index ae82dd5..9ef9639 100644 --- a/backend/fluksio/cloud/enroll.py +++ b/backend/fluksio/cloud/enroll.py @@ -14,7 +14,7 @@ from typing import Any from urllib.parse import urlsplit import httpx -from sqlmodel import Session, select +from sqlmodel import Session, col, select import fluksio from fluksio.cloud import config as cloud_config @@ -30,11 +30,6 @@ class EnrollError(Exception): self.detail = detail -class AlreadyEnrolled(EnrollError): - def __init__(self) -> None: - super().__init__(409, "This instance is already connected to a portal") - - def _is_local(url: str) -> bool: """Whether the address is this machine or a compose-internal service.""" host = urlsplit(url).hostname or "" @@ -83,10 +78,15 @@ def redeem_claim( def enroll( session: Session, user: User, portal_url: str, claim_code: str ) -> cloud_config.CloudConfig: - """Redeem the code and write the config the connector dials with.""" - if cloud_config.exists(): - raise AlreadyEnrolled() + """Redeem the code and write the config the connector dials with. + An instance that is already paired is re-paired rather than refused: a new + claim code is somebody asking for this, and refusing left deleting + `cloud.json` by hand as the only way through. The code is redeemed before + anything is written, so a code the portal rejects leaves a working + connection working. + """ + previous = cloud_config.load() data = redeem_claim(portal_url, claim_code) config = cloud_config.CloudConfig( portal_url=portal_url.rstrip("/"), @@ -104,6 +104,17 @@ def enroll( cloud_config.save(config) owner_id = str(data["owner_id"]) + if previous is not None and previous.issuer != config.issuer: + # A mapping is a subject of the portal that issued it, and nothing in + # the row says which portal that was — so against a different portal + # the old rows would both block the new owner from being adopted and + # resolve a stranger's subject onto a local account. + for stale in session.exec( + select(User).where(col(User.portal_sub).is_not(None)) + ): + stale.portal_sub = None + session.add(stale) + session.flush() # Re-enrolling from a different local account moves the mapping rather than # leaving two accounts claiming the same portal identity, which the unique # index would refuse and the lookup could not choose between anyway. diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index cfb4d6d..9aa6e36 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -799,6 +799,28 @@ class RunService: target=self._keep_leases, name="run-leases", daemon=True ) self._keeper.start() + self._requeue_queued() + + def _requeue_queued(self) -> None: + """Wake the engine for runs an earlier process was holding. + + A row is journaled `queued` and then a work item is added, and the two + are not one transaction — an in-memory queue loses the item with the + process, and a stream item nobody claimed is nobody's. The row is the + record, so this reads it back. Nothing is re-executed by it: claiming + is a compare-and-set out of `queued`, so an item that did survive is + claimed once whatever this adds. + """ + with Session(db_engine) as session: + waiting = session.exec( + select(Run.id, Run.flow).where(col(Run.status) == "queued") + ).all() + for run_id, flow in waiting: + self.queue.add(WorkItem(kind="run", node="", flow=flow, run_id=run_id)) + if waiting: + logger.info( + "Requeued %d run(s) left queued by an earlier engine", len(waiting) + ) def stop(self) -> None: self._stop.set() @@ -828,6 +850,7 @@ class RunService: draft: bool = False, no_cache: bool = False, idempotency_key: str | None = None, + parent_id: str | None = None, ) -> Run: """Journal a run and wake an engine up for it. Never blocks on it.""" # Before anything else, including reading the flow: a caller retrying a @@ -889,6 +912,7 @@ class RunService: created_at=datetime.now(UTC), actor=actor, idempotency_key=idempotency_key, + parent_id=parent_id, ) with Session(db_engine) as session: session.add(run) @@ -940,6 +964,34 @@ class RunService: self.controller.remote.cancel_run(run_id) return True + def retry(self, run_id: str, actor: str = "") -> Run: + """Submit the same run again, as a run of its own. + + What an engine that stopped mid-run leaves behind is a run marked + `abandoned`, and re-issuing the whole sweep to recover one config is + the blunt way back. This is the sharp one: the same flow, the same + inputs, the same group, so a sweep is completed rather than repeated. + The stage cache is what makes it cheap — the nodes that finished are + restored rather than run again. + """ + with Session(db_engine) as session: + run = session.get(Run, run_id) + if run is None: + raise RunRejected(f"No run {run_id}") + if run.status in ("queued", "running"): + raise RunRejected("That run has not finished; cancel it first") + source = run.model_copy() + return self.submit( + source.flow, + params=dict(source.params), + seed=source.seed, + group_id=source.group_id, + cause="retry", + actor=actor, + no_cache=source.no_cache, + parent_id=source.id, + ) + def _flow_of(self, run_id: str) -> str | None: with Session(db_engine) as session: run = session.get(Run, run_id) diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 2ece23a..e6df1ea 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -808,36 +808,49 @@ def _status_screen(client: Client) -> Any: parts.append( table if flows else Text("No flows yet. `fluksio sync` uploads yours.", "dim") ) + # Tables rather than padded strings: a flow named longer than the column + # used to push everything after it out of line. if runs: - parts += ["", Text("recent runs", style="dim")] + recent = Table(box=None, pad_edge=False, show_header=False) + recent.add_column("") + # Minimums rather than widths: the column keeps its shape when every + # value is short and grows for the one that is not. + recent.add_column("", min_width=9) + recent.add_column("", min_width=14) + recent.add_column("", justify="right", min_width=7) + recent.add_column("", justify="right", min_width=8) for row in runs: state = str(row.get("status", "")) - parts.append( - Text(f" {str(row.get('id', ''))[-8:]} ") - + Text( - f"{state:<9}", + recent.add_row( + f" {str(row.get('id', ''))[-8:]}", + Text( + state, style={"ok": "green", "cached": "cyan", "running": "cyan"}.get( state, "red" if state == "error" else "yellow" ), - ) - + Text( - f"{str(row.get('flow', '')):<16}" - f"{_dur(row.get('duration_ms')):>8}" - f" {_ago(row.get('finished_at') or row.get('created_at')):>9}", - style="dim", - ) + ), + Text(str(row.get("flow", "")), style="dim"), + Text(_dur(row.get("duration_ms")), style="dim"), + Text( + _ago(row.get("finished_at") or row.get("created_at")), style="dim" + ), ) + parts += ["", Text("recent runs", style="dim"), recent] if failures: - parts += ["", Text("recent failures", style="dim")] + broken = Table(box=None, pad_edge=False, show_header=False) + broken.add_column("") + broken.add_column("", justify="right") + broken.add_column("") for event in failures: where = " ".join( str(event.get(key, "")) for key in ("flow", "node") if event.get(key) ) - parts.append( - Text(f" {where} ", style="red") - + Text(f"{_ago(event.get('ts')):>9} ", style="dim") - + Text(str(event.get("detail", ""))[:100], style="dim") + broken.add_row( + Text(f" {where}", style="red"), + Text(_ago(event.get("ts")), style="dim"), + Text(str(event.get("detail", ""))[:100], style="dim"), ) + parts += ["", Text("recent failures", style="dim"), broken] return Group(*parts) @@ -986,6 +999,38 @@ def cmd_runs(args: argparse.Namespace) -> int: return 0 +#: The most runs of one group a retry looks at, which is the list route's own +#: ceiling. ponytail: a sweep larger than this needs paging, not a bigger number. +MAX_GROUP = 500 + +#: What a retry leaves alone. Everything else in a group — error, cancelled, +#: abandoned — is what "the ones that did not make it" means. +KEPT = frozenset({"ok", "cached", "queued", "running"}) + + +def cmd_retry(args: argparse.Namespace) -> int: + """Run the same thing again, one run or a group's unfinished ones.""" + try: + with _client_for(args) as client: + ids = list(args.run_id) + if args.group: + ids += [ + str(row["id"]) + for row in client.runs(limit=MAX_GROUP, group=args.group) + if str(row["status"]) not in KEPT + ] + if not ids: + return _fail("nothing to retry; name a run or a group with runs in it") + for run_id in ids: + made = client.retry(run_id) + _say(f"{made['id']} queued (retry of {run_id})") + except (SyncError, ApiError) as exc: + return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc) + return 0 + + def _artifacts(client: Client, args: argparse.Namespace) -> int: """List a run's files, or write one of them here.""" handle = RunHandle(client, args.run_id, client.run(args.run_id)) @@ -1401,6 +1446,19 @@ def add_parsers(subparsers: Any) -> None: with_engine(parser, local=True) parser.set_defaults(func=cmd_runs) + parser = subparsers.add_parser( + "retry", help="run something again: one run, or a group's unfinished ones" + ) + parser.add_argument("run_id", nargs="*", help="the runs to retry") + parser.add_argument( + "--group", + default="", + metavar="ID", + help="every run of this sweep that did not end ok", + ) + with_engine(parser) + parser.set_defaults(func=cmd_retry) + parser = subparsers.add_parser( "artifacts", help="the files a run produced; name one to download it" ) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 30b5ddf..bda939e 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -475,6 +475,10 @@ class Client: # Cancelling a cancelled run is cancelled. return self._call("POST", f"/runs/{run_id}/cancel", idempotent=True) + def retry(self, run_id: str) -> Any: + """Submit the same run again. Not idempotent: it makes a run.""" + return self._call("POST", f"/runs/{run_id}/retry") + def download(self, digest: str) -> bytes: response = self._request("GET", f"/artifacts/{digest}", idempotent=True) if response.status_code >= 400: diff --git a/backend/fluksio/tui/app.py b/backend/fluksio/tui/app.py index b5bd13d..91141b5 100644 --- a/backend/fluksio/tui/app.py +++ b/backend/fluksio/tui/app.py @@ -12,14 +12,26 @@ import signal import subprocess import sys import threading +import time from pathlib import Path from typing import Any from textual import work from textual.app import App, ComposeResult -from textual.containers import Horizontal, Vertical +from textual.containers import Horizontal, Vertical, VerticalScroll from textual.screen import ModalScreen -from textual.widgets import Button, DataTable, Footer, Header, Input, Label, RichLog +from textual.widgets import ( + Button, + DataTable, + Footer, + Header, + Input, + Label, + RichLog, + Static, + TabbedContent, + TabPane, +) from fluksio.cli import ( DEFAULT_PORT, @@ -43,6 +55,9 @@ STARTUP_TRIES = 60 #: split the browser makes for the same reason. MAX_PICKED = 20 +#: How many runs the table holds, now that it owns a whole tab. +RUNS_SHOWN = 50 + #: What the dashboard subscribes to. `message_value` is most of what the bus #: carries and none of what this screen draws. KINDS = frozenset( @@ -53,6 +68,33 @@ KINDS = frozenset( #: runs should cost one read of the engine, not twenty. COALESCE_S = 0.25 +#: Where the engine's own output goes, beside the data it is serving. A file +#: rather than a pipe because the screen is meant to be closed while the +#: engine keeps running: a pipe whose reader has gone breaks the next write, +#: and a node's `print` is one of those writes. It also means the log of an +#: engine this screen only *adopted* can be read here. +LOG_NAME = "serve.log" + +#: How much of it to read back when the screen opens. +LOG_TAIL_BYTES = 64 * 1024 + +#: What it is truncated to when this screen starts an engine of its own. +#: ponytail: a size check, not rotation — add rotation when somebody wants +#: yesterday's log. +LOG_KEEP_BYTES = 5 * 1024 * 1024 + +#: How long the tail waits when the file has nothing new. Also how long +#: closing the screen waits for that thread. +LOG_POLL_S = 0.25 + +#: How long a stop waits for the engine to be gone before killing it, and how +#: long it waits for an adopted one, which it can only ask. +STOP_WAIT_S = 10.0 + +#: The width of the runs table's five fixed columns, padding included, which +#: is what is left over for the inputs. +FIXED_COLUMNS = 54 + def child_argv(argv: list[str]) -> list[str]: """The same command, told to serve without a dashboard of its own. @@ -64,27 +106,98 @@ def child_argv(argv: list[str]) -> list[str]: return [sys.executable, "-m", "fluksio.cli", *passed, "--plain"] -class Enroll(ModalScreen[tuple[str, str] | None]): - """The claim code a portal minted, and which portal minted it.""" +class Enroll(ModalScreen[None]): + """The claim code a portal minted, and which portal minted it. + + The work happens here rather than back on the dashboard: enrolment is a + round trip to the portal, and the person who pressed the button is looking + at this modal while it happens. + """ BINDINGS = [("escape", "dismiss(None)", "cancel")] + def __init__(self, data_dir: Path) -> None: + super().__init__() + self.data_dir = data_dir + def compose(self) -> ComposeResult: with Vertical(id="enroll"): yield Label("Pair this instance with a portal") yield Input(placeholder="claim code", id="code") yield Input(value=DEFAULT_PORTAL, id="portal") + yield Label("", id="note") with Horizontal(): yield Button("Enroll", variant="primary", id="go") yield Button("Cancel", id="cancel") + def on_mount(self) -> None: + self.query_one("#code", Input).focus() + def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "cancel": self.dismiss(None) return + self.submit() + + def on_input_submitted(self, event: Input.Submitted) -> None: + self.submit() + + def submit(self) -> None: code = self.query_one("#code", Input).value.strip() - portal = self.query_one("#portal", Input).value.strip() - self.dismiss((code, portal or DEFAULT_PORTAL) if code else None) + portal = self.query_one("#portal", Input).value.strip() or DEFAULT_PORTAL + if not code: + self.note("A claim code is what pairs this instance.") + return + self.busy(True) + self.note(f"Enrolling with {portal}…") + self.enroll(code, portal) + + def note(self, message: str) -> None: + self.query_one("#note", Label).update(message) + + def busy(self, working: bool) -> None: + for one in self.query(Input): + one.disabled = working + for button in self.query(Button): + button.disabled = working + + @work(thread=True) + def enroll(self, code: str, portal: str) -> None: + """`fluksio enroll`, as its own process for the same reason serve is. + + A running engine picks the configuration up on its own; this screen + only has to report what the command said. + """ + done = subprocess.run( # noqa: S603 + [ + sys.executable, + "-m", + "fluksio.cli", + "enroll", + code, + "--portal", + portal, + "--data-dir", + str(self.data_dir), + ], + capture_output=True, + text=True, + ) + self.app.call_from_thread( + self.finished, done.returncode, done.stdout + done.stderr + ) + + def finished(self, code: int, output: str) -> None: + lines = [line for line in output.splitlines() if line.strip()] + for line in lines: + self.app.note(line) # type: ignore[attr-defined] + said = lines[-1] if lines else "" + if code == 0: + self.app.notify(said or "Connected to the portal.") + self.dismiss(None) + return + self.busy(False) + self.note(said or "Enrolment failed; the log has what it said.") class Artifacts(ModalScreen[None]): @@ -168,26 +281,41 @@ class Artifacts(ModalScreen[None]): self.app.call_from_thread(self.note, f"Wrote {out.resolve()}") +class RunsTable(DataTable[str]): + """The runs, and the keys that only mean anything while they are in front. + + Textual resolves a binding from the focused widget outwards, so hanging + them here is what makes the footer read as run tools on this tab and as + engine tools on the others. + """ + + BINDINGS = [ + ("space", "app.pick", "pick for comparison"), + ("c", "app.cancel_run", "cancel run"), + ("a", "app.artifacts", "artifacts"), + ] + + class ServeApp(App[int]): - """One screen: how the engine is, what has run, and what it is saying.""" + """Three tabs: how the engine is, what has run, and what it is saying.""" CSS = """ - #status { height: auto; padding: 0 1; } - #runs { height: 2fr; } - #log { height: 1fr; border-top: solid $panel; } + TabbedContent { height: 1fr; } + TabPane { height: 1fr; padding: 0; } + #overview { padding: 0 1; } #enroll { width: 60; height: auto; padding: 1 2; background: $surface; } + #enroll #note { color: $text-muted; height: auto; } #artifacts { width: 80; height: auto; padding: 1 2; background: $surface; } #artifacts DataTable { height: auto; max-height: 14; } """ BINDINGS = [ ("q", "quit", "quit (engine keeps running)"), + ("1", "tab('overview-tab')", "overview"), + ("2", "tab('runs-tab')", "runs"), + ("3", "tab('logs-tab')", "logs"), ("s", "stop_start", "stop/start"), ("r", "restart", "restart"), - ("c", "cancel_run", "cancel run"), - ("space", "pick", "pick for comparison"), - ("enter", "compare", "compare"), - ("a", "artifacts", "artifacts"), ("e", "enroll", "enroll"), ] @@ -195,7 +323,7 @@ class ServeApp(App[int]): super().__init__() self.args = args self.data_dir: Path = _data_dir(args.data_dir, args.shared) - self.child: subprocess.Popen[str] | None = None + self.child: subprocess.Popen[bytes] | None = None #: The pid of an engine this dashboard did not start. Only ever one #: whose data directory is this one — the probe is what proves it. self.adopted: int | None = None @@ -207,43 +335,92 @@ class ServeApp(App[int]): #: cascade of events costs one read of the engine rather than one each. self.stirred = False self.stop_stream = threading.Event() + self.stop_log = threading.Event() + #: The credential's mtime when the engine under this screen was + #: started, so its own is told from the one before it. + self.config_stamp = 0 # -- layout --------------------------------------------------------------- def compose(self) -> ComposeResult: yield Header() - yield RichLog(id="status", markup=True, wrap=True) - table: DataTable[str] = DataTable(id="runs", cursor_type="row") - yield table - yield RichLog(id="log", markup=False, highlight=False, max_lines=2000) + with TabbedContent(initial="overview-tab"): + with TabPane("Overview", id="overview-tab"): + with VerticalScroll(): + yield Static(id="overview") + with TabPane("Runs", id="runs-tab"): + yield RunsTable(id="runs", cursor_type="row") + with TabPane("Logs", id="logs-tab"): + yield RichLog(id="log", markup=False, highlight=False, max_lines=5000) yield Footer() def on_mount(self) -> None: self.title = f"fluksio — {self.data_dir}" - table = self.query_one("#runs", DataTable) - table.add_columns(" ", "run", "status", "flow", "took", "params") - table.focus() - self.start_engine(first=True) + table = self.query_one("#runs", RunsTable) + table.add_column(" ", width=1) + table.add_column("run", width=8) + table.add_column("status", width=9) + table.add_column("flow", width=16) + table.add_column("took", width=8) + table.add_column("params") + self.tail_log() + self.start_engine() # The socket is what makes a run that starts and finishes between two # ticks visible; this stays as the heartbeat for what is not on the bus. self.set_interval(WATCH_INTERVAL_S, self.refresh_panels) self.set_interval(COALESCE_S, self.drain) + def action_tab(self, tab: str) -> None: + self.query_one(TabbedContent).active = tab + + def on_tabbed_content_tab_activated( + self, event: TabbedContent.TabActivated + ) -> None: + """Focus what the tab is about, so its keys are the ones in the footer.""" + focusable = event.pane.query("RunsTable, RichLog, VerticalScroll") + if focusable: + focusable.first().focus() + + async def action_quit(self) -> None: + """Close the screen and leave the engine running. + + The threads are told first: both are worker threads on the loop's own + executor, which is joined while the loop closes — so a reader still + blocked on the log or the socket would hold the process after the + screen is gone, which is what `q` used to do. + """ + self.stop_log.set() + self.stop_stream.set() + self.exit(0) + + def on_unmount(self) -> None: + self.stop_log.set() + self.stop_stream.set() + # -- the engine under the screen ------------------------------------------ def note(self, message: str) -> None: self.query_one("#log", RichLog).write(message) - def start_engine(self, first: bool = False) -> None: + def say(self, message: str) -> None: + """`note`, from a worker thread.""" + self.call_from_thread(self.note, message) + + def log_path(self) -> Path: + return self.data_dir / LOG_NAME + + def start_engine(self) -> None: """Adopt whatever is already serving this directory, or start one.""" host = self.args.host reachable = "127.0.0.1" if host in ("0.0.0.0", "::", "") else host - wanted = self.args.port or DEFAULT_PORT + # Where this directory's engine last said it was, which is not always + # where it was asked to be: a taken port moves. + running = read_pidfile(self.data_dir) + wanted = running["port"] if running else (self.args.port or DEFAULT_PORT) url = f"http://{reachable}:{wanted}" - who = probe_engine(url, _token_for(self.data_dir)) if first else "other" + who = probe_engine(url, _token_for(self.data_dir)) if who == "ours": - running = read_pidfile(self.data_dir) self.adopted = running["pid"] if running else None self.url = url named = f" (pid {self.adopted})" if self.adopted else "" @@ -255,25 +432,47 @@ class ServeApp(App[int]): if who == "foreign": self.note(f"Port {wanted} holds another instance's Fluksio.") - self.child = subprocess.Popen( # noqa: S603 - child_argv(sys.argv[1:]), - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - bufsize=1, - env={**os.environ, "PYTHONUNBUFFERED": "1"}, - ) - self.tail_child(self.child) + # What the credential looked like before the child wrote its own, so + # the wait below cannot mistake the last engine's url for this one's. + self.config_stamp = self._config_stamp() + path = self.log_path() + if path.exists() and path.stat().st_size > LOG_KEEP_BYTES: + path.write_text("") + handle = path.open("ab", buffering=0) + try: + self.child = subprocess.Popen( # noqa: S603 + child_argv(sys.argv[1:]), + stdout=handle, + stderr=subprocess.STDOUT, + env={**os.environ, "PYTHONUNBUFFERED": "1"}, + ) + finally: + # The child holds a descriptor of its own; this one is what would + # otherwise keep the file open for as long as the screen lives. + handle.close() self.await_engine() - @work(thread=True, exclusive=False) - def tail_child(self, child: subprocess.Popen[str]) -> None: + @work(thread=True, exclusive=True, group="log") + def tail_log(self) -> None: """The engine's own output, which is why a second terminal was needed.""" - if child.stdout is None: - return - for line in child.stdout: - self.call_from_thread(self.note, line.rstrip()) - self.call_from_thread(self.note, f"The engine stopped ({child.wait()}).") + path = self.log_path() + while not self.stop_log.is_set() and self.is_running: + try: + with path.open("r", errors="replace") as handle: + handle.seek(max(0, path.stat().st_size - LOG_TAIL_BYTES)) + if handle.tell(): + handle.readline() # whatever line the seek landed in + while not self.stop_log.is_set() and self.is_running: + line = handle.readline() + if line: + self.call_from_thread(self.note, line.rstrip()) + continue + if path.stat().st_size < handle.tell(): + break # truncated under us; read it from the top + time.sleep(LOG_POLL_S) + except OSError: + # Not written yet, or gone. Either way it may appear. + time.sleep(LOG_POLL_S) @work(thread=True, exclusive=True, group="startup") def await_engine(self) -> None: @@ -282,8 +481,6 @@ class ServeApp(App[int]): The port is the child's to choose — it moves off a taken one — and `client.json` is where it says which it took. """ - import time - for _ in range(STARTUP_TRIES): if self.child is not None and self.child.poll() is not None: return @@ -291,6 +488,11 @@ class ServeApp(App[int]): stored = json.loads(config_path(self.data_dir).read_text()) except (OSError, ValueError): stored = {} + if self._config_stamp() == self.config_stamp: + # Still the one the last engine left, which names a port this + # one may not have taken. + time.sleep(0.5) + continue if stored.get("url") and stored.get("token"): self.url = str(stored["url"]) self.call_from_thread(self.connect) @@ -313,23 +515,56 @@ class ServeApp(App[int]): self.watch_engine(self.client.url, self.client.token) self.refresh_panels() + def _config_stamp(self) -> int: + """When the credential was last written, or 0 if it never was.""" + try: + return config_path(self.data_dir).stat().st_mtime_ns + except OSError: + return 0 + def engine_pid(self) -> int | None: return self.child.pid if self.child is not None else self.adopted def stop_engine(self) -> None: + """Stop the engine and wait for it to be gone. **From a thread.** + + Waiting is what makes `r` work: a new engine started while the old one + still holds the port would move off it, and the screen would end up + watching one engine while the client talks to another. An engine + draining its flows takes as long as it takes, which is why this is not + something to do on the loop that draws. + """ + pid = self.engine_pid() if self.child is not None: - self.note(f"Stopping the engine (pid {self.child.pid}).") + self.say(f"Stopping the engine (pid {self.child.pid}).") self.child.terminate() + try: + self.child.wait(timeout=STOP_WAIT_S) + except subprocess.TimeoutExpired: + self.child.kill() self.child = None elif self.adopted is not None: - self.note(f"Stopping the adopted engine (pid {self.adopted}).") + self.say(f"Stopping the adopted engine (pid {self.adopted}).") try: os.kill(self.adopted, signal.SIGTERM) except OSError as exc: - self.note(f"Could not stop it: {exc}") + self.say(f"Could not stop it: {exc}") + self._await_exit(self.adopted) self.adopted = None self.stop_stream.set() self.client = None + if pid is not None: + self.say("The engine stopped.") + + def _await_exit(self, pid: int) -> None: + deadline = time.time() + STOP_WAIT_S + while time.time() < deadline: + try: + os.kill(pid, 0) + except OSError: + return + time.sleep(0.2) + self.say(f"pid {pid} is still running.") # -- the engine's own events ---------------------------------------------- @@ -371,28 +606,30 @@ class ServeApp(App[int]): return try: screen = _status_screen(client) - rows = client.runs(limit=20) + rows = client.runs(limit=RUNS_SHOWN) except Exception as exc: self.call_from_thread(self.show_offline, exc) return self.call_from_thread(self.show, screen, rows) def show(self, screen: Any, rows: list[dict[str, Any]]) -> None: - status = self.query_one("#status", RichLog) - status.clear() - status.write(screen) - table = self.query_one("#runs", DataTable) + self.query_one("#overview", Static).update(screen) + table = self.query_one("#runs", RunsTable) cursor = table.cursor_row + # Whatever the fixed columns do not use, so a wide terminal shows the + # inputs rather than a stripe of empty table. + room = max(20, table.size.width - FIXED_COLUMNS) table.clear() for row in rows: run_id = str(row.get("id", "")) + params = json.dumps(row.get("params") or {}) table.add_row( "·" if run_id in self.picked else " ", run_id[-8:], str(row.get("status", "")), str(row.get("flow", "")), _dur(row.get("duration_ms")), - json.dumps(row.get("params") or {})[:60], + params.ljust(room) if len(params) <= room else params[: room - 1] + "…", key=run_id, ) # The rows are rewritten wholesale every refresh, and a cursor that @@ -402,22 +639,25 @@ class ServeApp(App[int]): table.move_cursor(row=cursor) def show_offline(self, exc: Exception) -> None: - status = self.query_one("#status", RichLog) - status.clear() which = "starting" if self.engine_pid() is not None else "not running" - status.write(f"[yellow]The engine is {which}.[/] ({type(exc).__name__})") + self.query_one("#overview", Static).update( + f"[yellow]The engine is {which}.[/] ({type(exc).__name__})" + ) # -- keys ----------------------------------------------------------------- + @work(thread=True, group="engine") def action_stop_start(self) -> None: if self.engine_pid() is not None: self.stop_engine() else: - self.start_engine() + self.call_from_thread(self.start_engine) + @work(thread=True, group="engine") def action_restart(self) -> None: + """Stop, wait for the port to come back, start.""" self.stop_engine() - self.start_engine() + self.call_from_thread(self.start_engine) def cursor_run(self) -> str: """The whole id of the run the cursor is on. @@ -425,7 +665,7 @@ class ServeApp(App[int]): The cell holds its tail, which is what fits in a column; the row's key is the whole of it, which is what the engine is asked about. """ - table = self.query_one("#runs", DataTable) + table = self.query_one("#runs", RunsTable) if self.client is None or not table.row_count: return "" key = table.coordinate_to_cell_key(table.cursor_coordinate).row_key @@ -482,38 +722,7 @@ class ServeApp(App[int]): self.call_from_thread(self.refresh_panels) def action_enroll(self) -> None: - self.push_screen(Enroll(), self.enrolled) - - def enrolled(self, answer: tuple[str, str] | None) -> None: - if answer is None: - return - code, portal = answer - self.run_enroll(code, portal) - - @work(thread=True) - def run_enroll(self, code: str, portal: str) -> None: - """`fluksio enroll`, as its own process for the same reason serve is. - - A running engine picks the configuration up on its own; this screen - only has to report what the command said. - """ - done = subprocess.run( # noqa: S603 - [ - sys.executable, - "-m", - "fluksio.cli", - "enroll", - code, - "--portal", - portal, - "--data-dir", - str(self.data_dir), - ], - capture_output=True, - text=True, - ) - for line in (done.stdout + done.stderr).splitlines(): - self.call_from_thread(self.note, line) + self.push_screen(Enroll(self.data_dir)) def run_tui(args: argparse.Namespace) -> int: @@ -524,4 +733,5 @@ def run_tui(args: argparse.Namespace) -> int: if pid is not None: print(f"The engine is still running: pid {pid} at {url or 'its port'}.") print(f" fluksio serve reattaches to it; kill {pid} stops it.") + print(f" Its log: {app.log_path()}") return 0 diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 2fba021..f4d20b4 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -492,6 +492,107 @@ def test_the_seed_is_recorded_the_same_way_however_it_arrived(): session.commit() +def test_a_retry_is_a_new_run_that_names_the_one_it_repeats(): + """The way back from a run an engine restart interrupted. + + Re-issuing the whole sweep is the blunt version; this keeps the group, so + what was missing is filled in rather than run twice. + """ + flow = FlowDef( + name="study", + mode="batch", + inputs=[ + FlowInput(spec=MessageSpec(name="lr", dtype=DType.FLOAT), initial=0.01) + ], + ) + queue = _Collect() + service = RunService(controller=_OneFlow(flow), queue=queue) + made = [] + with Session(db_engine) as session: + session.add( + Run( + id="abandoned-1", + flow="study", + status="abandoned", + params={"lr": 0.3}, + group_id="sweep-9", + created_at=datetime.now(UTC), + ) + ) + session.add( + Run( + id="going-1", + flow="study", + status="running", + created_at=datetime.now(UTC), + ) + ) + session.commit() + try: + again = service.retry("abandoned-1", actor="someone@example.com") + made.append(again.id) + assert again.parent_id == "abandoned-1" + assert (again.params, again.group_id, again.cause) == ( + {"lr": 0.3}, + "sweep-9", + "retry", + ) + assert queue.items[-1].run_id == again.id + + # A run that has not finished is cancelled, not retried. + with pytest.raises(RunRejected): + service.retry("going-1") + with pytest.raises(RunRejected): + service.retry("never-existed") + finally: + with Session(db_engine) as session: + for run in session.exec( + select(Run).where(col(Run.id).in_([*made, "abandoned-1", "going-1"])) + ).all(): + session.delete(run) + session.commit() + + +def test_runs_left_queued_are_woken_by_the_next_engine(): + """The row is the record, and the work item is not part of its transaction. + + An in-memory queue loses the item with the process and a stream item + nobody claimed is nobody's, so without this a run submitted just before a + restart waits for an engine that will never be told about it. + """ + queue = _Collect() + service = RunService(controller=_Unusable(), queue=queue) + with Session(db_engine) as session: + session.add( + Run( + id="orphan-1", + flow="study", + status="queued", + created_at=datetime.now(UTC), + ) + ) + session.add( + Run( + id="finished-1", + flow="study", + status="ok", + created_at=datetime.now(UTC), + ) + ) + session.commit() + try: + service._requeue_queued() + assert [item.run_id for item in queue.items] == ["orphan-1"] + assert queue.items[0].flow == "study" + finally: + with Session(db_engine) as session: + for run in session.exec( + select(Run).where(col(Run.id).in_(["orphan-1", "finished-1"])) + ).all(): + session.delete(run) + session.commit() + + def test_a_key_nobody_used_submits_normally( client, superuser_token_headers, monkeypatch ): diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 7403567..07e153d 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -4,6 +4,8 @@ import subprocess import sys from pathlib import Path +import pytest + from fluksio.cli import load_or_create_secret_key @@ -689,6 +691,41 @@ def test_the_dashboard_runs_the_engine_as_a_child_of_itself(monkeypatch) -> None assert opened == [] +@pytest.mark.anyio +async def test_the_dashboard_is_three_tabs_and_lets_go_of_its_threads( + tmp_path, monkeypatch +) -> None: + """The tabs, and what `q` has to do before the screen goes. + + Its two readers are worker threads on the event loop's own executor, which + is joined while the loop closes — so a reader still blocked on the log file + or the websocket holds the terminal after the dashboard has gone, which is + what `q` used to do. + """ + from textual.widgets import TabbedContent + + from fluksio import cli + from fluksio.tui.app import ServeApp + + # Nothing is started under this screen: the engine has its own tests. + monkeypatch.setattr(ServeApp, "start_engine", lambda self: None) + args = cli._parser().parse_args(["serve", "--data-dir", str(tmp_path / ".fluksio")]) + app = ServeApp(args) + async with app.run_test() as pilot: + tabs = app.query_one(TabbedContent) + assert tabs.active == "overview-tab" + await pilot.press("2") + assert tabs.active == "runs-tab" + # The run keys live on the table, so they are in the footer only here. + assert "app.pick" in {binding[1] for binding in app.query_one("#runs").BINDINGS} + await pilot.press("3") + assert tabs.active == "logs-tab" + await pilot.press("q") + + assert app.stop_log.is_set() + assert app.stop_stream.is_set() + + 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 @@ -741,6 +778,41 @@ def test_who_holds_the_port_is_told_apart_by_the_token() -> None: assert probe_engine("http://127.0.0.1:1", "t") == "other" +def test_a_second_engine_for_one_directory_is_refused(tmp_path, monkeypatch, capsys): + """One SQLite file, one engine — whatever port the second was asked for. + + The refusal is worth more than the duplicate it prevents: the second + engine signs in on the way up, so `client.json` would point at a port + that dies with it and every later command would reach nothing. + """ + import json + import os + + from fluksio import cli + + data_dir = tmp_path / ".fluksio" + data_dir.mkdir() + (data_dir / "client.json").write_text( + json.dumps({"url": "http://127.0.0.1:8000", "token": "the-first-one"}) + ) + cli.write_pidfile(data_dir, 8000) + monkeypatch.setattr(cli, "probe_engine", lambda *a, **k: "ours") + + args = cli._parser().parse_args( + ["serve", "--plain", "--port", "8123", "--data-dir", str(data_dir)] + ) + assert cli.cmd_serve(args) == 0 + said = capsys.readouterr().out + assert "already serving" in said and str(os.getpid()) in said + # The credential still names the engine that is actually up. + stored = json.loads((data_dir / "client.json").read_text()) + assert stored["url"] == "http://127.0.0.1:8000" + + # A pid that is alive but is not an engine of ours is not a refusal. + monkeypatch.setattr(cli, "probe_engine", lambda *a, **k: "foreign") + assert cli.already_serving(data_dir, "127.0.0.1") == "" + + 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 diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py index fcdbeb2..a66e5e8 100644 --- a/backend/tests/test_cloud.py +++ b/backend/tests/test_cloud.py @@ -25,7 +25,8 @@ from fluksio.cloud import config as cloud_config from fluksio.core.config import settings from fluksio.flow.panels import PanelDef, PanelsConfig, write_config from fluksio.models import User -from tests.utils.portal import ISSUER, portal_token +from tests.utils.portal import ISSUER, jwks, portal_token +from tests.utils.user import create_random_user def test_portal_token_is_refused_when_not_enrolled( @@ -101,6 +102,85 @@ def test_status_reports_not_enrolled( assert response.json()["enrolled"] is False +def test_enrolling_again_replaces_the_connection( + client: TestClient, + enrolled: User, + portal_key: rsa.RSAPrivateKey, + superuser_token_headers: dict[str, str], + db: Session, +) -> None: + """A new claim code re-pairs rather than being refused. + + Deleting cloud.json by hand used to be the only way through, and it takes + every remote connection with it. The mappings of the portal being left go + too: nothing in the row says which portal issued the subject, so one kept + from the old one would resolve a stranger onto a local account. + """ + stale = create_random_user(db) + stale.portal_sub = "portal-user-7" + db.add(stale) + db.commit() + + elsewhere = "https://other.example.test" + reply = Mock( + status_code=200, + json=Mock( + return_value={ + "ws_url": f"{elsewhere}/api/v1/tunnel/attach", + "instance_id": "11111111-2222-3333-4444-555555555555", + "instance_token": "the-new-token", + "issuer": elsewhere, + "jwks": jwks(portal_key), + "owner_id": "portal-user-9", + } + ), + ) + with patch("fluksio.cloud.enroll.httpx.post", return_value=reply): + again = client.post( + f"{settings.API_V1_STR}/cloud/enroll", + headers=superuser_token_headers, + json={"portal_url": elsewhere, "claim_code": "ABCD-EFGH"}, + ) + assert again.status_code == 200, again.text + + config = cloud_config.load() + assert config is not None + assert config.portal_url == elsewhere + assert config.token == "the-new-token" + + db.expire_all() + assert db.get(User, enrolled.id).portal_sub == "portal-user-9" + # The other portal's mapping is not carried over to this one. + assert db.get(User, stale.id).portal_sub is None + + +def test_a_claim_the_portal_refuses_keeps_the_connection( + client: TestClient, + enrolled: User, + superuser_token_headers: dict[str, str], + db: Session, +) -> None: + """Nothing is written until the portal has accepted the code. + + That ordering is what makes replacing safe: a mistyped code leaves the + instance connected to the portal it was connected to. + """ + before = cloud_config.load() + with patch("fluksio.cloud.enroll.httpx.post", return_value=Mock(status_code=404)): + refused = client.post( + f"{settings.API_V1_STR}/cloud/enroll", + headers=superuser_token_headers, + json={"portal_url": ISSUER, "claim_code": "NOPE-NOPE"}, + ) + assert refused.status_code == 400, refused.text + + after = cloud_config.load() + assert before is not None and after is not None + assert (before.instance_id, before.token) == (after.instance_id, after.token) + db.expire_all() + assert db.get(User, enrolled.id).portal_sub == "portal-user-1" + + def test_enrolling_needs_a_superuser( client: TestClient, normal_user_token_headers: dict[str, str] ) -> None: @@ -388,6 +468,39 @@ async def test_a_config_that_appears_while_running_is_dialled(monkeypatch) -> No assert started +@pytest.mark.anyio +async def test_an_enrolment_replaced_while_running_is_redialled(monkeypatch) -> None: + """`fluksio enroll` is its own process and cannot cancel the live link. + + Without this the tunnel would stay up on the credential that was replaced, + since the watcher only ever started a link where there was none. + """ + from types import SimpleNamespace + + from fluksio.cloud import connector + + live = asyncio.create_task(asyncio.sleep(30)) + app = Mock() + app.state = Mock( + cloud_task=live, cloud_connector=Mock(), cloud_identity=("old", "old-token") + ) + started: list[object] = [] + monkeypatch.setattr(connector, "ENROL_POLL_S", 0.01) + monkeypatch.setattr(connector, "start", lambda one: started.append(one)) + monkeypatch.setattr( + cloud_config, + "load", + lambda: SimpleNamespace(instance_id="new", token="new-token"), + ) + + watcher = asyncio.create_task(connector.watch_enrolment(app)) + await asyncio.sleep(0.05) + watcher.cancel() + + assert live.cancelled() or live.done() + assert started + + @pytest.mark.anyio async def test_a_config_that_cannot_be_read_is_not_dialled(monkeypatch) -> None: """Otherwise the connector gives up at once and this restarts it forever.""" diff --git a/docs/code/cli.md b/docs/code/cli.md index ad69fc1..fb499a4 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -55,10 +55,13 @@ The default port moves out of the way when something already has it (8001, moved off: `--port 9000` on a taken 9000 fails, because something else is there and you named it. -What it will *not* do is start a second engine for the same instance. If -the port is held by an engine already serving this directory, it says so and -stops, since one SQLite database wants one engine. Another instance's Fluksio -on that port is named, and the move happens as usual. +What it will *not* do is start a second engine for the same instance. Before +anything else, `serve` looks for an engine already serving this directory — +the pidfile beside the data says where, and a token this directory's key +signed says whether what answers there is ours — and stops if it finds one, +whatever port the second was asked for. One SQLite database wants one engine, +and the second would repoint `client.json` at a port that dies with it. +Another instance's Fluksio on the port is named, and the move happens as usual. | Option | Default | What it does | |---|---|---| @@ -115,25 +118,36 @@ dashboard is served from there rather than here. ### The dashboard -At a terminal, `serve` draws the health overview, the recent runs, and the -engine's own log in a pane below, so the output above is in there rather than -replaced by it. +At a terminal, `serve` opens three tabs: **Overview** is the health block and +the flows, **Runs** is the history, and **Logs** is the engine's own output. | Key | What it does | |---|---| +| `1` `2` `3` | the Overview, the Runs and the Logs | | `q` | close the dashboard. **The engine keeps running**, and the pid is printed | | `s` | stop the engine, or start it again | | `r` | restart it | -| `c` | cancel the run the cursor is on | +| `e` | pair with a portal, without leaving the screen | + +On the Runs tab the toolbar carries what a run is for: + +| Key | What it does | +|---|---| | `space` | tick the run under the cursor into a comparison | | `enter` | compare the ticked runs, or draw the one under the cursor | +| `c` | cancel the run the cursor is on | | `a` | list what the run left behind, and fetch it | -| `e` | pair with a portal, without leaving the screen | The engine is a child process rather than a thread, which is what makes those possible, and what makes `q` a way out of the screen rather than a way to stop the engine. Running `fluksio serve` again reattaches to it. +Its output goes to `serve.log` in the data directory rather than down a pipe, +which is what lets the screen be closed while the engine keeps running — a +pipe with nobody reading it breaks the next line the engine writes, and a +node's `print` is one of those. It also means the Logs tab has the output of +an engine this screen only adopted, and the scrollback of the one before it. + An engine started elsewhere is adopted rather than duplicated, and can be stopped from here only when it is this instance's own: both the pidfile beside the data and a token this directory's key signed have to agree. Another @@ -185,6 +199,14 @@ Get the code from the portal under **Instances → Add instance**. It is single-use and expires in fifteen minutes. `--as` matters when the instance has several superusers. Without it, enrolment refuses rather than guessing. +An instance that is already paired is re-paired: a new claim code replaces the +connection, at the same portal or another one, and a running engine drops the +old tunnel within a few seconds. The code is redeemed before anything is +written, so one the portal rejects leaves the connection as it was. Moving to +a different portal clears the local accounts' mappings to the old one, since +nothing in a mapping says which portal issued it — the people who had access +are admitted again from the new portal. + Afterwards, `fluksio serve` dials the portal as it comes up, and keeps dialling: a portal that restarts, a wifi that changes, a laptop that suspends and wakes somewhere else all end the same connection, and the link is put back up without @@ -405,6 +427,26 @@ everything else off the line. `Client.runs()` and `--local` reads the same history from an in-process engine, without one having to be served. +### `fluksio retry` + +```sh +fluksio retry [ ...] +fluksio retry --group +``` + +Runs the same thing again as a run of its own, carrying the flow, the inputs, +the seed and the group of the one it repeats, and recording it as its parent. +What it is for is a run an engine restart interrupted: those are marked +`abandoned` once their lease goes stale, and `--group` retries every run of a +sweep that did not end `ok`, so the missing combinations are filled in rather +than the whole grid being submitted again. The stage cache is what makes it +cheap — the nodes that finished are restored rather than recomputed. + +A run that has not finished is refused; cancel it first. + +Runs that were still *queued* when an engine stopped need none of this: the +next engine reads them back out of its own history and picks them up. + ### `fluksio flavors` ```sh @@ -493,6 +535,8 @@ The message name is the one to pass, since it is what addresses the bytes; ```text .fluksio/ (or ~/.fluksio, with `--global`) ├── client.json the token `serve` wrote, mode 600 +├── serve.pid the engine serving this directory, while one is +├── serve.log what the engine under the dashboard printed ├── .gitignore `*`, so a database and a credential are ignored from within ├── fluksio.db SQLite: users, runs, metrics, observability, agents ├── flows/ a git repository, one directory per flow diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 54e9cf9..99866f3 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -296,11 +296,20 @@ one the automations use: a burst of five hundred sweep runs must not stand between a house and its heating. An engine that is down when a run is submitted picks it up when it starts. +The row is what makes that true rather than the queue. A run is written before +the work item is added, so an engine reads its own history at startup and +wakes itself for anything still `queued` — which is what a run submitted +seconds before a restart is, and what an in-memory queue would otherwise have +lost with the process. + From the moment a run is claimed, its database row is the record and the queue is finished with it. Redelivering two hours of training because an acknowledgement was late is not recovery; instead a running run refreshes a lease, and one whose lease goes stale is marked `abandoned`, which is what a -run whose engine was killed mid-training becomes. +run whose engine was killed mid-training becomes. Starting it over is a +decision rather than something that happens: [`fluksio +retry`](../code/cli.md#fluksio-retry) submits it again, and `--group` does +that for the runs of a sweep that did not finish. ## Looking at what ran diff --git a/docs/interface/portal.md b/docs/interface/portal.md index 6721eab..1b486a4 100644 --- a/docs/interface/portal.md +++ b/docs/interface/portal.md @@ -37,7 +37,9 @@ interface of its own: fluksio enroll ABCD-1234 --portal https://hub.fluksio.com ``` -The code is single-use and expires in fifteen minutes. +The code is single-use and expires in fifteen minutes. A fresh code on an +instance that is already paired replaces the pairing, so moving an instance to +another portal is one command rather than a disconnect and a reconnect. A portal session then arrives as *that local account*, which the settings screen states plainly. Use `--as someone@example.com` to enrol as a specific local