From 7efa75e2421c8397c6c79c07c3e11a5f235d2923 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 20:00:08 +0200 Subject: [PATCH] Refuse a zero concurrency limit instead of reading it as the default MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FLOW_MAX_WORKERS, FLOW_MAX_CASCADES and FLOW_MAX_RUNS are all pool sizes, so 0 says neither "none" nor "unlimited" — it is a pool that cannot be built. They are PositiveInt now, so a 0 fails at startup naming the setting rather than being swallowed by `max_cascades or MAX_CASCADES`. The consuming fallbacks take only None as "nobody said": explicit `is None` in the executor, and no clamp on RunService.parallel. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- backend/fluksio/core/config.py | 11 ++++++++--- backend/fluksio/flow/executor.py | 10 ++++++++-- backend/fluksio/flow/runs.py | 4 +++- docs/reference/configuration.md | 4 +++- 4 files changed, 22 insertions(+), 7 deletions(-) diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 1e8729b..91497fc 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -9,6 +9,7 @@ from pydantic import ( BeforeValidator, EmailStr, HttpUrl, + PositiveInt, computed_field, model_validator, ) @@ -97,15 +98,19 @@ class Settings(BaseSettings): # rather than a person, and it can refresh unattended. MCP_TOKEN_EXPIRE_MINUTES: int = 60 MCP_REFRESH_EXPIRE_DAYS: int = 30 - FLOW_MAX_WORKERS: int = 4 + # The three below are all pool sizes, so 0 says neither "none" nor + # "unlimited" — it is a pool that cannot be built and an engine that would + # accept no work. Rejected here rather than quietly read as the default, + # because a limit somebody set and did not get is the worse surprise. + FLOW_MAX_WORKERS: PositiveInt = 4 # How many cascades may be in flight at once. Sustained throughput is this # over the mean cascade time, so an installation whose nodes wait on the # network rather than on a CPU wants it higher than the core count. - FLOW_MAX_CASCADES: int = 4 + FLOW_MAX_CASCADES: PositiveInt = 4 # How many batch runs are driven at once. A different limit from the one # above: a run drives a whole graph, and its nodes are bounded by the worker # pool rather than by cascade slots. A sweep is what this governs. - FLOW_MAX_RUNS: int = 4 + FLOW_MAX_RUNS: PositiveInt = 4 # How long a python node may be silent before its worker is killed, unless # the node sets its own. 0, the default, disables it: a dead worker still # fails fast, and a slow one is left to finish. Set it where silence means diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index 8711df1..cbbef7a 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -46,6 +46,11 @@ DELAYED_INTERVAL_S = 1.0 #: whose nodes wait on a network rather than a CPU may want more of them — #: `FLOW_MAX_CASCADES` is where that is said. MAX_CASCADES = 4 +#: Node threads, unless the service is given a number. Both this and the one +#: above are taken as written: only ``None`` means "nobody said", so a number +#: that reached here is one somebody chose, and an unusable one is the pool's +#: ``ValueError`` rather than a silent 4. +MAX_WORKERS = 4 # How long a reload waits for claimed work to finish before rebuilding anyway. DRAIN_TIMEOUT_S = 10.0 # Work waiting in the stream, undelivered. A burst is normal — the pool claims @@ -68,7 +73,7 @@ class ExecutionService: ) -> None: self.queue = queue self._events = events - self.max_cascades = max_cascades or MAX_CASCADES + self.max_cascades = MAX_CASCADES if max_cascades is None else max_cascades self._pipeline: Pipeline | None = None self._stop = threading.Event() # Set when a deadline moves closer, so the timer thread stops waiting @@ -82,7 +87,8 @@ class ExecutionService: # Entry ids claimed and still running, under _inflight_lock. self._active: set[str] = set() self.node_pool = ThreadPoolExecutor( - max_workers=max_workers or 4, thread_name_prefix="node" + max_workers=MAX_WORKERS if max_workers is None else max_workers, + thread_name_prefix="node", ) self._cascade_pool = ThreadPoolExecutor( max_workers=self.max_cascades, thread_name_prefix="cascade" diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 39f4260..c3bd6b8 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -769,7 +769,9 @@ class RunService: # the isolation it wants, minus surviving the process. self._state_factory = state_factory or (lambda _ns: MemoryState()) self.engine_name = f"{socket.gethostname()}-{os.getpid()}"[:64] - self.parallel = max(1, parallel) + # Taken as written: clamping a 0 up to 1 would hide a limit somebody + # set, and the pool below rejects an unusable one loudly anyway. + self.parallel = parallel self._pool = ThreadPoolExecutor( max_workers=self.parallel, thread_name_prefix="run" ) diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 914efce..ace2df9 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -135,7 +135,9 @@ warning into a refusal to start. The three concurrency limits are also flags on `fluksio serve` — `--max-workers`, `--max-cascades`, `--max-runs` — which outrank the file, and the engine says -which numbers it started with in its first lines. +which numbers it started with in its first lines. Each is a pool size, so each +must be at least 1: a `0` is refused at startup by name rather than read as the +default. Leave one empty (or unset) to get the default. An artifact is referred to by a run that recorded it or by a message currently holding it; anything else is what a camera published four hours ago, and the