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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
This commit is contained in:
2026-10-15 09:50:18 +02:00
co-authored by Claude Opus 5.5
parent 2dc9295c26
commit 025123a219
5 changed files with 39 additions and 5 deletions
+20
View File
@@ -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
+3 -3
View File
@@ -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
+3 -2
View File
@@ -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]]
+8
View File
@@ -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)
+5
View File
@@ -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: