From 1f7c6646f1d9455df09ecff2c240def72b514054 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 26 Aug 2026 21:19:09 +0200 Subject: [PATCH] Survive a busy engine: retry, idempotent submit, resilient waiting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A driver script died of one slow answer: httpx.ReadTimeout out of RunHandle.refresh() with a 30 s read timeout and no retry anywhere, which cost a sweep 78 of its 84 runs. - Split the timeout (5 s connect, 120 s read): a wrong URL fails at once, and a busy engine gets longer than the slowest thing it does on purpose (a 60 s compile, a 15 s rebuild wait). - Retry idempotent calls three times on a transport error or 502/503/504. 503 is the engine's own "ask again" — it is what RebuildBusy answers. - Submit carries a key the engine stores with the run, so a retry after a timeout returns that run instead of starting a second. A sweep keys every entry, so a half-created one recreates only what is missing. - wait() and --follow tolerate five failed polls in a row; a 404 still stops at once, because that is an answer rather than a gap. - CLI says "engine not answering" and names the run still on the engine, instead of printing a traceback. - runs: clamp the params column to 80 characters; events() takes the flow/since/until the endpoint already had; RunHandle.failures answers "what killed this run" from the run's own node rows. Co-Authored-By: Claude Opus 5 (1M context) --- .../f2c6a8d15e93_run_idempotency_key.py | 40 +++++ backend/fluksio/api/routes/runs.py | 8 + backend/fluksio/flow/runs.py | 29 ++- backend/fluksio/models.py | 6 + backend/fluksio/sdk/cli.py | 59 +++++- backend/fluksio/sdk/client.py | 168 +++++++++++++++--- backend/pyproject.toml | 2 +- backend/tests/api/routes/test_runs.py | 54 ++++++ backend/tests/sdk/test_client.py | 132 ++++++++++++++ docs/code/cli.md | 6 +- docs/getting-started/data-science.md | 28 ++- worker/pyproject.toml | 2 +- 12 files changed, 503 insertions(+), 31 deletions(-) create mode 100644 backend/fluksio/alembic/versions/f2c6a8d15e93_run_idempotency_key.py create mode 100644 backend/tests/sdk/test_client.py diff --git a/backend/fluksio/alembic/versions/f2c6a8d15e93_run_idempotency_key.py b/backend/fluksio/alembic/versions/f2c6a8d15e93_run_idempotency_key.py new file mode 100644 index 0000000..8454151 --- /dev/null +++ b/backend/fluksio/alembic/versions/f2c6a8d15e93_run_idempotency_key.py @@ -0,0 +1,40 @@ +"""run.idempotency_key + +Submitting is a POST, so a client that retries one after a timeout cannot know +whether the first attempt landed. A key the caller mints per submission makes +the answer a lookup: the same key returns the run it already created. + +Revision ID: f2c6a8d15e93 +Revises: d1f7a3c8b204 +Create Date: 2026-08-26 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "f2c6a8d15e93" +down_revision = "d1f7a3c8b204" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "run", + sa.Column( + "idempotency_key", + sqlmodel.sql.sqltypes.AutoString(length=64), + nullable=True, + ), + ) + # Nullable and unique: every run submitted without a key stays NULL, and + # both SQLite and Postgres allow as many of those as there are runs. + op.create_index("ix_run_idempotency_key", "run", ["idempotency_key"], unique=True) + + +def downgrade(): + op.drop_index("ix_run_idempotency_key", table_name="run") + op.drop_column("run", "idempotency_key") diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index 29e22c9..3eb5c58 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -50,11 +50,17 @@ class RunCreate(BaseModel): no_cache: bool = False #: Who is asking. The dashboard leaves it, and is the "api" default. cause: RunCause = "api" + #: A key the caller minted for this submission. Sending it again returns + #: the run it already made, so a retry after a timeout cannot double-submit. + idempotency_key: str | None = Field(default=None, max_length=64) class SweepEntry(BaseModel): params: dict[str, Any] = Field(default_factory=dict) seed: int | None = None + #: One per entry, so retrying a half-created sweep recreates only the runs + #: whose rows never landed. + idempotency_key: str | None = Field(default=None, max_length=64) class SweepCreate(BaseModel): @@ -184,6 +190,7 @@ async def create_run( actor=user.email, draft=body.draft, no_cache=body.no_cache, + idempotency_key=body.idempotency_key, ) except FlowNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @@ -221,6 +228,7 @@ async def create_sweep( actor=user.email, draft=body.draft, no_cache=body.no_cache, + idempotency_key=entry.idempotency_key, ) for entry in body.runs ] diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index a2540d8..2ab5827 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -42,6 +42,7 @@ from typing import Any from sqlalchemy import update from sqlalchemy.dialects.sqlite import insert as upsert +from sqlalchemy.exc import IntegrityError from sqlmodel import Session, col, select from fluksio.core.db import engine as db_engine @@ -549,8 +550,16 @@ class RunService: actor: str = "", draft: bool = False, no_cache: bool = False, + idempotency_key: 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 + # submit it never got an answer for is owed the run that answer was + # about, whatever the flow says now. + if idempotency_key: + existing = self._by_key(idempotency_key) + if existing is not None: + return existing flow = self.controller.store.read_flow(flow_name, draft=draft) issues = batch_issues(flow) if issues: @@ -580,15 +589,33 @@ class RunService: labels=required_labels(flow), created_at=datetime.now(UTC), actor=actor, + idempotency_key=idempotency_key, ) with Session(db_engine) as session: session.add(run) - session.commit() + try: + session.commit() + except IntegrityError: + # Two retries of one submit raced here. The unique index is + # what decided which of them is the run; this one reads it + # back rather than queueing a second execution of it. + session.rollback() + existing = self._by_key(idempotency_key) if idempotency_key else None + if existing is None: + raise + return existing session.refresh(run) self.queue.add(WorkItem(kind="run", node="", flow=flow.name, run_id=run.id)) return run + def _by_key(self, idempotency_key: str) -> Run | None: + """The run a key already made, if it made one.""" + with Session(db_engine) as session: + return session.exec( + select(Run).where(col(Run.idempotency_key) == idempotency_key) + ).first() + def cancel(self, run_id: str) -> bool: """Stop a run: kill what it is executing, schedule nothing further.""" with self._lock: diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 7e999b9..6116bd4 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -340,6 +340,12 @@ class Run(SQLModel, table=True): group_id: str | None = Field(default=None, index=True, max_length=64) #: The run this one was made from, on a retry. parent_id: str | None = Field(default=None, max_length=64) + #: A key the caller made up for one submission, so retrying a submit that + #: may already have landed returns that run instead of starting a second. + #: Unique where it is set; null for anything submitted without one. + idempotency_key: str | None = Field( + default=None, unique=True, index=True, max_length=64 + ) #: Where it was asked for: api, cli, sdk, hook or sweep. cause: str = Field(default="api", max_length=32) #: Re-execute every node, whatever the stage cache holds for it. diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index fd817a2..4254477 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -20,9 +20,12 @@ from contextlib import contextmanager from pathlib import Path from typing import Any +import httpx + from fluksio.sdk import FLOWS, Flow, SyncError from fluksio.sdk.client import ( GLOBAL_DATA_DIR, + WAIT_TOLERANCE, ApiError, Client, RunHandle, @@ -56,6 +59,14 @@ def _fail(message: str) -> int: return 1 +def _unreachable(exc: Exception, note: str = "") -> int: + """The engine did not answer. Say so as a sentence, not a traceback.""" + return _fail( + f"engine not answering ({type(exc).__name__}: {exc})" + + (f". {note}" if note else "") + ) + + def _status(text: str, width: int = 0) -> str: """A status, coloured when a terminal is reading it. @@ -267,6 +278,8 @@ def cmd_sync(args: argparse.Namespace) -> int: ) except (SyncError, ApiError) as exc: return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc, "Nothing was published; sync again when it is back.") for report in reports: if report.unchanged: @@ -402,9 +415,23 @@ def _follow(client: Client, handle: RunHandle, poll: float = 1.0) -> None: before the numbers, so the last batch is never the one that gets missed. """ seen: set[tuple[str, int]] = set() + failures = 0 while True: - done = handle.refresh().done - for point in client.metrics(handle.id): + try: + done = handle.refresh().done + points = client.metrics(handle.id) + except (httpx.HTTPError, ApiError) as exc: + # Following an eight-hour run must not end because one poll of it + # did. The run is still going; only this side lost sight of it. + if isinstance(exc, ApiError) and exc.status < 500: + raise + failures += 1 + if failures >= WAIT_TOLERANCE: + raise + time.sleep(poll) + continue + failures = 0 + for point in points: mark = (str(point.get("name", "")), int(point.get("step", -1))) if mark in seen: continue @@ -419,7 +446,7 @@ def _cancel(client: Client, handle: RunHandle) -> int: """Ctrl-C means stop the run, not just stop watching it.""" try: client.cancel(handle.id) - except (SyncError, ApiError) as exc: + except (SyncError, ApiError, httpx.HTTPError) as exc: return _fail(f"could not cancel {handle.id}: {exc}") _say(f"{handle.id} {_status('cancelled')}") return 130 @@ -429,7 +456,7 @@ def _cached_note(client: Client, handle: RunHandle) -> str: """How much of the run earlier ones had already answered.""" try: nodes = client.run(handle.id).get("nodes") or [] - except (SyncError, ApiError): + except (SyncError, ApiError, httpx.HTTPError): return "" cached = sum(1 for node in nodes if node.get("status") == "cached") return f" ({cached}/{len(nodes)} {_status('cached')})" if cached else "" @@ -439,6 +466,7 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: # An in-process engine lives exactly as long as this command, so a run # nobody waits for would be thrown away with the queue holding it. wait = args.wait or args.follow or args.local + handle: RunHandle | None = None try: with _client_for(args) as client: if not args.no_sync: @@ -482,6 +510,12 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: return 0 if handle.status == "ok" else 1 except (SyncError, ApiError) as exc: return _fail(str(exc)) + except httpx.HTTPError as exc: + # The run is the engine's, not this command's: it carries on, and its + # id is how to find it again. + return _unreachable( + exc, f"Run {handle.id} is still on the engine." if handle else "" + ) # --------------------------------------------------------------------------- @@ -651,6 +685,14 @@ def cmd_status(args: argparse.Namespace) -> int: return 130 except (SyncError, ApiError) as exc: return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc) + + +#: How much of a run's inputs the list shows. A flow taking a few kB of JSON +#: would otherwise make the table unreadable; `client.runs()` is where the +#: whole value is read. +PARAMS_WIDTH = 80 def cmd_runs(args: argparse.Namespace) -> int: @@ -659,11 +701,16 @@ def cmd_runs(args: argparse.Namespace) -> int: rows = client.runs(flow=args.flow, limit=args.limit) except (SyncError, ApiError) as exc: return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc) for row in rows: commit = (row.get("origin_commit") or "")[:7] + params = json.dumps(row["params"]) + if len(params) > PARAMS_WIDTH: + params = params[: PARAMS_WIDTH - 3] + "..." _say( f"{row['id']} {_status(row['status'], 9)} {row['flow']:<16} " - f"{row['duration_ms'] / 1000:7.1f}s {commit:<8} {json.dumps(row['params'])}" + f"{row['duration_ms'] / 1000:7.1f}s {commit:<8} {params}" ) return 0 @@ -725,6 +772,8 @@ def cmd_sweep(args: argparse.Namespace) -> int: return 1 if failed else 0 except (SyncError, ApiError) as exc: return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc, "The runs are still on the engine; `fluksio runs`.") # --------------------------------------------------------------------------- diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 95672b9..357c676 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -11,6 +11,7 @@ import json import os import subprocess import time +import uuid from collections.abc import Iterable from pathlib import Path from typing import Any @@ -34,6 +35,27 @@ API = "/api/v1" #: A run is over when it reaches one of these. DONE = frozenset({"ok", "error", "cancelled", "abandoned"}) +#: How long to wait for an answer. Split, because the two halves mean +#: different things: an engine that is not there refuses the connection at +#: once, so waiting 30 s for it only ever delays a typo in the URL, while an +#: engine that *is* there can be busy — a rebuild waits 15 s before it answers +#: 503, and compiling a node is given 60 s — so the read has to outlast the +#: slowest thing the engine does on purpose. +CONNECT_TIMEOUT = 5.0 +READ_TIMEOUT = 120.0 + +#: How many times an idempotent call is tried again before it gives up, and +#: what a busy engine answers with. A read timeout used to kill a driver +#: script outright, which is a poor trade for a request that costs nothing to +#: repeat. 503 is the engine's own "ask again": it is what a rebuild answers. +RETRIES = 3 +RETRY_STATUS = frozenset({502, 503, 504}) + +#: How many refreshes in a row may fail before waiting gives up. Each one is +#: already several attempts, so this is minutes of a missing engine, not a +#: blip. +WAIT_TOLERANCE = 5 + #: What a project-local installation is called, beside `.venv` and `.git`. DATA_DIR_NAME = ".fluksio" @@ -124,13 +146,19 @@ class Client: """An authenticated engine, addressed over its HTTP API.""" def __init__( - self, url: str = "", token: str = "", http: Any = None, timeout: float = 30.0 + self, + url: str = "", + token: str = "", + http: Any = None, + timeout: Any = None, + retries: int = RETRIES, ) -> None: stored = _stored() self.url = url or os.environ.get("FLUKSIO_URL") or stored.get("url") or "" self.token = ( token or os.environ.get("FLUKSIO_TOKEN") or stored.get("token") or "" ) + self.retries = retries if http is None: if not self.url: raise SyncError( @@ -139,6 +167,10 @@ class Client: ) import httpx + if timeout is None: + timeout = httpx.Timeout( + 30.0, connect=CONNECT_TIMEOUT, read=READ_TIMEOUT + ) http = httpx.Client(base_url=self.url, timeout=timeout) self.http = http if self.token: @@ -146,8 +178,38 @@ class Client: # -- plumbing ---------------------------------------------------------- - def _call(self, method: str, path: str, **kwargs: Any) -> Any: - response = self.http.request(method, f"{API}{path}", **kwargs) + def _request( + self, method: str, path: str, idempotent: bool = False, **kwargs: Any + ) -> Any: + """One request, tried again while it is safe and worth it. + + Only a call that may be repeated without meaning something different + is retried — every GET, and the writes that carry a key or are already + a no-op the second time. The last attempt is made outside the loop, so + whatever it raises is what the caller sees. + """ + import httpx + + attempts = self.retries + 1 if idempotent else 1 + for attempt in range(attempts - 1): + try: + response = self.http.request(method, f"{API}{path}", **kwargs) + if response.status_code not in RETRY_STATUS: + return response + except httpx.TransportError: + pass + time.sleep(2**attempt) + return self.http.request(method, f"{API}{path}", **kwargs) + + def _call( + self, method: str, path: str, idempotent: bool | None = None, **kwargs: Any + ) -> Any: + response = self._request( + method, + path, + idempotent=(method == "GET") if idempotent is None else idempotent, + **kwargs, + ) if response.status_code == 409: raise Conflict(_detail(response)) if response.status_code >= 400: @@ -192,7 +254,9 @@ class Client: def refresh_modules(self) -> None: """Retire the engine's workers, so the next run imports the code as it is.""" - self._call("POST", "/modules/refresh") + # Retiring workers twice is retiring workers, and this is the call most + # likely to meet a rebuild's 503. + self._call("POST", "/modules/refresh", idempotent=True) def flows(self) -> list[dict[str, Any]]: """Every flow with its node and error counts, as the home screen lists.""" @@ -211,11 +275,29 @@ class Client: result: dict[str, Any] = self._call("GET", "/cloud/status") return result - def events(self, kind: str = "failure", limit: int = 10) -> list[dict[str, Any]]: - """What went wrong, or who changed what. Newest first.""" - result = self._call( - "GET", "/observability/events", params={"kind": kind, "limit": limit} - ) + def events( + self, + kind: str = "failure", + limit: int = 10, + flow: str = "", + since: Any = None, + until: Any = None, + ) -> list[dict[str, Any]]: + """What went wrong, or who changed what. Newest first. + + Engine-wide, and narrowed by flow or by time. What *one run* did is a + question about that run: :attr:`RunHandle.failures` answers it from the + run's own node rows, which carry the traceback anyway. + """ + query: dict[str, Any] = {"kind": kind, "limit": limit} + if flow: + query["flow"] = flow + for name, value in (("since", since), ("until", until)): + if value is not None: + query[name] = ( + value.isoformat() if hasattr(value, "isoformat") else value + ) + result = self._call("GET", "/observability/events", params=query) return list(result or []) # -- runs -------------------------------------------------------------- @@ -229,15 +311,22 @@ class Client: cause: str = "sdk", ) -> RunHandle: """Queue a run. ``cause`` is what the history records it as coming - from — a script is the default, `fluksio run` says so itself.""" + from — a script is the default, `fluksio run` says so itself. + + Safe to retry: the key is minted here, once per call, so an attempt + that timed out on the way back is answered with the run it made rather + than starting a second one. + """ row = self._call( "POST", f"/runs/flows/{flow}", + idempotent=True, json={ "params": params or {}, "seed": seed, "no_cache": no_cache, "cause": cause, + "idempotency_key": uuid.uuid4().hex, }, ) return RunHandle(self, row["id"], row) @@ -248,11 +337,17 @@ class Client: entries: list[dict[str, Any]], no_cache: bool = False, ) -> list[RunHandle]: - """Many runs of one flow at once. The caller decides what varies.""" + """Many runs of one flow at once. The caller decides what varies. + + A key per entry rather than one for the sweep: a retry then recreates + only the runs whose rows never landed. + """ + keyed = [{**entry, "idempotency_key": uuid.uuid4().hex} for entry in entries] rows = self._call( "POST", f"/runs/flows/{flow}/sweep", - json={"runs": entries, "no_cache": no_cache}, + idempotent=True, + json={"runs": keyed, "no_cache": no_cache}, ) return [RunHandle(self, row["id"], row) for row in rows] @@ -280,10 +375,11 @@ class Client: ) def cancel(self, run_id: str) -> Any: - return self._call("POST", f"/runs/{run_id}/cancel") + # Cancelling a cancelled run is cancelled. + return self._call("POST", f"/runs/{run_id}/cancel", idempotent=True) def download(self, digest: str) -> bytes: - response = self.http.request("GET", f"{API}/artifacts/{digest}") + response = self._request("GET", f"/artifacts/{digest}", idempotent=True) if response.status_code >= 400: raise ApiError(response.status_code, _detail(response)) return bytes(response.content) @@ -328,13 +424,45 @@ class RunHandle: rows = self._row.get("artifacts") return rows if isinstance(rows, list) else [] - def wait(self, timeout: float = 0.0, poll: float = 1.0) -> RunHandle: - """Block until the run is over, or ``timeout`` seconds have passed.""" - deadline = time.monotonic() + timeout if timeout else 0.0 - while True: + @property + def failures(self) -> list[dict[str, Any]]: + """The nodes that failed, each with its error and its logs.""" + if "nodes" not in self._row: self.refresh() - if self.done: - return self + rows = self._row.get("nodes") + rows = rows if isinstance(rows, list) else [] + return [row for row in rows if row.get("status") == "error"] + + def wait(self, timeout: float = 0.0, poll: float = 1.0) -> RunHandle: + """Block until the run is over, or ``timeout`` seconds have passed. + + A run outlives the engine being briefly unreachable, so waiting for one + does too: a few failed refreshes in a row are a busy engine, not a + finished run. What is not tolerated is the engine answering — a 404 + means the run is gone, and repeating the question will not bring it + back. + """ + import httpx + + deadline = time.monotonic() + timeout if timeout else 0.0 + failures = 0 + while True: + try: + self.refresh() + except ApiError as exc: + if exc.status < 500: + raise + failures += 1 + if failures >= WAIT_TOLERANCE: + raise + except httpx.HTTPError: + failures += 1 + if failures >= WAIT_TOLERANCE: + raise + else: + failures = 0 + if self.done: + return self if deadline and time.monotonic() > deadline: raise TimeoutError(f"run {self.id} is still {self.status}") time.sleep(poll) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 64e0163..61811dd 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "fluksio" -version = "0.1.3" +version = "0.1.4" description = "Node-based automation engine: flows, dashboards, batch runs" readme = "README.md" license = "AGPL-3.0-or-later" diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 4e016cc..01d0481 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -19,6 +19,7 @@ from fluksio.flow.runs import ( OUTPUT_CAP, RunCache, RunRejected, + RunService, _cacheable, new_run_id, resolve_references, @@ -349,6 +350,59 @@ def test_a_run_records_which_caller_asked_for_it( assert refused.status_code == 422 +class _Unusable: + """Anything reaching this is something a deduplicated submit should not do.""" + + def __getattr__(self, name): + raise AssertionError(f"a repeated submit must not reach {name}") + + +def test_a_repeated_submit_returns_the_run_it_already_made(): + """The key is the answer to "did my first attempt land?". + + Answered before the flow is even read: a caller retrying a submit it never + got a reply to is owed that run, whatever has been published since. + """ + service = RunService(controller=_Unusable(), queue=_Unusable()) + with Session(db_engine) as session: + session.add( + Run( + id="dedup-1", + flow="study", + status="running", + idempotency_key="key-abc", + created_at=datetime.now(UTC), + ) + ) + session.commit() + + again = service.submit("study", {"lr": 0.1}, idempotency_key="key-abc") + + assert again.id == "dedup-1" + + +def test_a_key_nobody_used_submits_normally( + client, superuser_token_headers, monkeypatch +): + """The route carries the key through; without one nothing changes.""" + seen: dict[str, object] = {} + + class Recorder: + def submit(self, name, **kwargs): + seen.update(kwargs) + return Run(id="keyed-1", flow=name, created_at=datetime.now(UTC)) + + monkeypatch.setattr(client.app.state, "run_service", Recorder()) + answer = client.post( + f"{settings.API_V1_STR}/runs/flows/demo", + headers=superuser_token_headers, + json={"idempotency_key": "key-xyz"}, + ) + + assert answer.status_code == 202 + assert seen["idempotency_key"] == "key-xyz" + + def test_overview_is_not_read_as_a_run_id(client, superuser_token_headers): """`/overview` is declared before `/{run_id}`, which would swallow it.""" answer = client.get( diff --git a/backend/tests/sdk/test_client.py b/backend/tests/sdk/test_client.py new file mode 100644 index 0000000..71b58d5 --- /dev/null +++ b/backend/tests/sdk/test_client.py @@ -0,0 +1,132 @@ +"""What the client does when the engine is slow, busy, or briefly gone.""" + +import httpx +import pytest + +from fluksio.sdk.client import WAIT_TOLERANCE, ApiError, Client, RunHandle + + +def a_client(handler, monkeypatch, **kwargs): + """A client whose transport is a function, and whose backoff costs nothing.""" + monkeypatch.setattr("fluksio.sdk.client.time.sleep", lambda _seconds: None) + transport = httpx.MockTransport(handler) + http = httpx.Client(transport=transport, base_url="http://engine") + return Client(url="http://engine", token="t", http=http, **kwargs) + + +def test_idempotent_get_is_tried_again(monkeypatch): + calls = [] + + def handler(request): + calls.append(request) + if len(calls) < 3: + raise httpx.ReadTimeout("too slow", request=request) + return httpx.Response(200, json={"id": "r1", "status": "ok"}) + + client = a_client(handler, monkeypatch) + assert client.run("r1")["status"] == "ok" + assert len(calls) == 3 + + +def test_a_busy_engine_is_asked_again(monkeypatch): + codes = iter([503, 503, 200]) + + def handler(request): + code = next(codes) + return httpx.Response(code, json={"id": "r1"} if code == 200 else {}) + + client = a_client(handler, monkeypatch) + assert client.run("r1")["id"] == "r1" + + +def test_giving_up_raises_what_it_last_saw(monkeypatch): + def handler(request): + raise httpx.ReadTimeout("too slow", request=request) + + client = a_client(handler, monkeypatch, retries=2) + with pytest.raises(httpx.ReadTimeout): + client.run("r1") + + +def test_a_write_is_not_repeated(monkeypatch): + calls = [] + + def handler(request): + calls.append(request) + raise httpx.ReadTimeout("too slow", request=request) + + client = a_client(handler, monkeypatch) + with pytest.raises(httpx.ReadTimeout): + client.put_source("train", "fit", "code") + assert len(calls) == 1, "a source write means something different twice" + + +def test_submit_carries_one_key_across_its_retries(monkeypatch): + bodies = [] + + def handler(request): + bodies.append(httpx.Response(200, content=request.content).json()) + if len(bodies) < 3: + raise httpx.ConnectError("no route", request=request) + return httpx.Response(202, json={"id": "r1", "status": "queued"}) + + client = a_client(handler, monkeypatch) + assert client.submit("train", {"lr": 0.1}).id == "r1" + keys = {body["idempotency_key"] for body in bodies} + assert len(keys) == 1, "a retry must not read as a second run" + assert len(next(iter(keys))) == 32 + + +def test_a_sweep_keys_every_entry(monkeypatch): + seen = {} + + def handler(request): + body = httpx.Response(200, content=request.content).json() + seen["runs"] = body["runs"] + return httpx.Response(202, json=[{"id": "r1"}, {"id": "r2"}]) + + client = a_client(handler, monkeypatch) + client.sweep("train", [{"params": {"lr": 0.1}}, {"params": {"lr": 0.2}}]) + keys = [entry["idempotency_key"] for entry in seen["runs"]] + assert len(set(keys)) == 2 + + +def test_waiting_survives_a_few_bad_answers(monkeypatch): + answers = iter( + [503] * (WAIT_TOLERANCE - 1) + [200] # then the run is finished + ) + + def handler(request): + code = next(answers) + if code != 200: + return httpx.Response(code, json={}) + return httpx.Response(200, json={"id": "r1", "status": "ok"}) + + client = a_client(handler, monkeypatch, retries=0) + handle = RunHandle(client, "r1", {"status": "running"}) + assert handle.wait(poll=0).status == "ok" + + +def test_waiting_gives_up_eventually(monkeypatch): + def handler(request): + return httpx.Response(503, json={}) + + client = a_client(handler, monkeypatch, retries=0) + handle = RunHandle(client, "r1", {"status": "running"}) + with pytest.raises(ApiError): + handle.wait(poll=0) + + +def test_a_run_that_is_gone_stops_the_wait_at_once(monkeypatch): + calls = [] + + def handler(request): + calls.append(request) + return httpx.Response(404, json={"detail": "no such run"}) + + client = a_client(handler, monkeypatch, retries=0) + handle = RunHandle(client, "r1", {"status": "running"}) + with pytest.raises(ApiError) as caught: + handle.wait(poll=0) + assert caught.value.status == 404 + assert len(calls) == 1, "a 404 is an answer, not a blip" diff --git a/docs/code/cli.md b/docs/code/cli.md index 260ee61..17f838f 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -273,8 +273,10 @@ fluksio runs [--flow train] [--limit 20] The runs an engine has recorded, newest first: id, status, flow, duration, the commit of the repository it came from, and its parameters. Statuses are coloured when a terminal is reading the output — `ok` green, `error` red, -`cached` cyan. `--local` reads the same history from an in-process engine, -without one having to be served. +`cached` cyan. Parameters are clamped to 80 characters so a flow taking a few +kB of JSON still lists as a table; `Client.runs()` is where the whole value is +read. `--local` reads the same history from an in-process engine, without one +having to be served. ### `fluksio sweep` diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md index b177cbf..6f94dfa 100644 --- a/docs/getting-started/data-science.md +++ b/docs/getting-started/data-science.md @@ -449,7 +449,33 @@ curl -s "$FLUKSIO/runs//metrics?name=train.loss" -H "Authorization: Bearer $ The run carries its parameters, a digest of them, the seed, its result, how long each node took, what it logged, every artifact it produced, and both commits — the flow store's and your repository's. That is the answer to "what -was the learning rate on the run that got 94%?". +was the learning rate on the run that got 94%?". When a run failed, +`run.failures` is the node that did it, with its traceback and its logs. + +### When the engine is busy + +A driver script outlives the engine being slow, because a sweep is hours long +and a blip is seconds: + +- **Reads are repeated.** Every GET, and the writes that mean the same thing + twice, are tried up to three more times on a transport error or a `502`, + `503` or `504`, waiting 1 s, 2 s, then 4 s. `Client(retries=0)` turns that + off. +- **Submitting is safe to repeat.** Each `submit()` mints a key the engine + stores with the run, so an attempt that timed out on the way back is + answered with the run it already made rather than starting a second one. A + sweep keys every entry, so retrying a half-created one creates only what is + missing. +- **Waiting tolerates a gap.** `wait()` and `--follow` survive five failed + polls in a row before giving up. An answer is not a gap: a `404` for a run + that is gone stops immediately. +- **Timeouts are split** — 5 s to connect, 120 s to read, so a wrong URL fails + at once while a busy engine is given longer than the slowest thing it does + on purpose. `Client(timeout=…)` takes a number or an `httpx.Timeout`. + +A command that still cannot reach the engine says +`fluksio: engine not answering (…)` and names the run it started, which is +running on the engine regardless — `fluksio runs` finds it again. ## Sweep it diff --git a/worker/pyproject.toml b/worker/pyproject.toml index 0ec3410..e5f2979 100644 --- a/worker/pyproject.toml +++ b/worker/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "fluksio-worker" -version = "0.1.3" +version = "0.1.4" description = "Runs Fluksio nodes on a machine the engine cannot reach" readme = "README.md" license = "AGPL-3.0-or-later"