From 025123a219c6392ee8ef5e1894f800f9c3b8e105 Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 15 Oct 2026 09:50:18 +0200 Subject: [PATCH] Engine timings move to FLOW_* settings NODE_STOP_TIMEOUT, REBUILD_WAIT, FLOW_REBUILD_WAIT, BACKOFF and CANCEL_GRACE keep their defaults and are read from Settings. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx --- backend/fluksio/core/config.py | 20 ++++++++++++++++++++ backend/fluksio/flow/controller.py | 6 +++--- backend/fluksio/flow/supervision.py | 5 +++-- backend/tests/flow/test_supervision.py | 8 ++++++++ docs/reference/configuration.md | 5 +++++ 5 files changed, 39 insertions(+), 5 deletions(-) diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index eac2267..f88d8af 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -8,6 +8,7 @@ from pydantic import ( AnyUrl, BeforeValidator, EmailStr, + Field, HttpUrl, PositiveInt, computed_field, @@ -163,6 +164,25 @@ class Settings(BaseSettings): # grew it then gives the memory back, and the next call pays its imports # again. 0, the default, keeps every worker warm. FLOW_WORKER_MAX_RSS_MB: int = 0 + # How long a node's `stop` may take before a rebuild moves on without it. + # A broker that has gone away can leave a node waiting on an + # acknowledgement that never comes. + FLOW_NODE_STOP_TIMEOUT: float = 5.0 + # How long a request waits for a rebuild that is already running before it + # is answered as busy: one for the whole pipeline, one for a single flow's. + # Past either, the engine is wedged rather than busy. + FLOW_PIPELINE_REBUILD_WAIT: float = 15.0 + FLOW_REBUILD_WAIT: float = 5.0 + # Seconds between restarts of a node's crashed loop, one step per attempt + # with the last repeating. A JSON list in the environment. + FLOW_RESTART_BACKOFF: Annotated[tuple[float, ...], Field(min_length=1)] = ( + 1.0, + 5.0, + 30.0, + 60.0, + ) + # How long a cancelled node loop gets to stop before it is abandoned. + FLOW_CANCEL_GRACE: float = 5.0 # How long the engine's own metrics, events and run records are kept. #: The largest body `PUT /artifacts` will take, in bytes. A checkpoint or a #: video segment is a legitimate artifact, so this is generous rather than diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 1e1eb12..9665cd9 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -106,7 +106,7 @@ def _is_hook(node: Node | None) -> TypeGuard[HttpNode | WebSocketNode]: # A node's `stop` talks to whatever it connected to, and a broker that has gone # away can leave it waiting for an acknowledgement that never arrives — which # used to hold the rebuild, and everything queued behind it, forever. -NODE_STOP_TIMEOUT = 5.0 +NODE_STOP_TIMEOUT = settings.FLOW_NODE_STOP_TIMEOUT # How long a rebuild asked for by a request waits for one already running. # Generous on purpose: a rebuild of a populated instance reconnects every @@ -114,14 +114,14 @@ NODE_STOP_TIMEOUT = 5.0 # healthy one of those should not be turned away. Past that the controller is # wedged rather than busy, and an error the caller can act on beats a request # that never ends. -REBUILD_WAIT = 15.0 +REBUILD_WAIT = settings.FLOW_PIPELINE_REBUILD_WAIT # How long a caller waits for a *flow* rebuild that is already running. A # per-flow rebuild reconnects one flow's nodes rather than the instance's, # so this is a queueing budget — several of them back to back, which is what # seeding does — not the room a single one needs. Fifteen seconds was sized for # the whole-pipeline rebuild and would let a wedge sit unreported. -FLOW_REBUILD_WAIT = 5.0 +FLOW_REBUILD_WAIT = settings.FLOW_REBUILD_WAIT # How long a node has to have been down before it is shown as down. A device # that misses one poll and comes back is a flake, not a fault, and without this diff --git a/backend/fluksio/flow/supervision.py b/backend/fluksio/flow/supervision.py index ff2d624..b526271 100644 --- a/backend/fluksio/flow/supervision.py +++ b/backend/fluksio/flow/supervision.py @@ -26,11 +26,12 @@ from collections import deque from collections.abc import Callable, Coroutine from typing import Any +from fluksio.core.config import settings from fluksio.flow.events import EventBus logger = logging.getLogger(__name__) -BACKOFF = (1.0, 5.0, 30.0, 60.0) +BACKOFF = settings.FLOW_RESTART_BACKOFF # A task that burns through this many restarts in the window is not going to # recover by being restarted again *now*. # @@ -49,7 +50,7 @@ QUARANTINE_BACKOFF = (300.0, 900.0, 3600.0) # this is not going to stop on its own — a client closing a socket the broker # no longer answers on is the case seen in the wild — and the rebuild asking # for it must not wait on that forever. -CANCEL_GRACE = 5.0 +CANCEL_GRACE = settings.FLOW_CANCEL_GRACE TaskFactory = Callable[[], Coroutine[Any, Any, None]] diff --git a/backend/tests/flow/test_supervision.py b/backend/tests/flow/test_supervision.py index a5a6746..94d214f 100644 --- a/backend/tests/flow/test_supervision.py +++ b/backend/tests/flow/test_supervision.py @@ -234,3 +234,11 @@ def test_cancelling_one_flow_lifts_only_its_quarantine(): await supervisor.cancel_all() asyncio.run(scenario()) + + +def test_the_restart_steps_are_read_from_the_environment(monkeypatch): + """A list in an environment variable, so it arrives as JSON.""" + from fluksio.core.config import Settings + + monkeypatch.setenv("FLOW_RESTART_BACKOFF", "[2, 10]") + assert Settings().FLOW_RESTART_BACKOFF == (2.0, 10.0) diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 3e37778..5932fa3 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -129,6 +129,11 @@ Set `TZ` in `.env` for schedules that should follow local time, such as | `FLOW_GPUS` | `0` | local GPU count; `0` detects NVIDIA device nodes | | `FLOW_RAM` | `0` | memory budget in MB; `0` detects host/container memory and reserves 1 GiB | | `FLOW_WORKER_MAX_RSS_MB` | `0` | retire a local worker above this memory use after a call; `0` disables | +| `FLOW_NODE_STOP_TIMEOUT` | `5` | seconds a node gets to stop during a rebuild before the rebuild moves on | +| `FLOW_PIPELINE_REBUILD_WAIT` | `15` | seconds a request waits for a running whole-pipeline rebuild before it gets a 503 | +| `FLOW_REBUILD_WAIT` | `5` | the same for a single flow's rebuild, such as a deploy | +| `FLOW_RESTART_BACKOFF` | `[1,5,30,60]` | seconds between restarts of a crashed node loop; the last step repeats | +| `FLOW_CANCEL_GRACE` | `5` | seconds a cancelled node loop gets to stop before it is abandoned | | `OBS_RETENTION_DAYS` | `30` | retention for metrics, events and run records | The three concurrency limits must be at least `1`. They also have CLI flags: