diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 764d91f..b1ca365 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -93,6 +93,10 @@ class Settings(BaseSettings): MCP_TOKEN_EXPIRE_MINUTES: int = 60 MCP_REFRESH_EXPIRE_DAYS: int = 30 FLOW_MAX_WORKERS: int = 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 # 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 510766f..77ab476 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -28,7 +28,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -CLAIM_COUNT = 4 CLAIM_BLOCK_MS = 1000 # Long enough that a busy cascade is not mistaken for a dead one. RECLAIM_IDLE_MS = 60_000 @@ -38,6 +37,10 @@ RECLAIM_INTERVAL_S = 30.0 # stopping — which is what an engine that died does. TOUCH_INTERVAL_S = 20.0 DELAYED_INTERVAL_S = 1.0 +#: How many cascades may be in flight, unless the service is given a number. +#: Sustained throughput is this over the mean cascade time, so an installation +#: 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 # How long a reload waits for claimed work to finish before rebuilding anyway. DRAIN_TIMEOUT_S = 10.0 @@ -57,9 +60,11 @@ class ExecutionService: queue: WorkQueue, max_workers: int | None = None, events: EventBus | None = None, + max_cascades: int | None = None, ) -> None: self.queue = queue self._events = events + self.max_cascades = max_cascades or MAX_CASCADES self._pipeline: Pipeline | None = None self._stop = threading.Event() self._intake = threading.Event() @@ -72,7 +77,7 @@ class ExecutionService: max_workers=max_workers or 4, thread_name_prefix="node" ) self._cascade_pool = ThreadPoolExecutor( - max_workers=MAX_CASCADES, thread_name_prefix="cascade" + max_workers=self.max_cascades, thread_name_prefix="cascade" ) self._consumer: threading.Thread | None = None self._timers: threading.Thread | None = None @@ -136,6 +141,11 @@ class ExecutionService: def alive(self) -> bool: return self._consumer is not None and self._consumer.is_alive() + @property + def inflight(self) -> int: + """Cascades claimed and still running.""" + return self._inflight + # ------------------------------------------------------------------------- # Threads # ------------------------------------------------------------------------- @@ -150,7 +160,7 @@ class ExecutionService: if not free: continue try: - items = self.queue.claim(min(CLAIM_COUNT, free), CLAIM_BLOCK_MS) + items = self.queue.claim(free, CLAIM_BLOCK_MS) failures = 0 except Exception as exc: failures += 1 @@ -251,9 +261,9 @@ class ExecutionService: work that is actually being run. """ with self._inflight_lock: - while self._inflight >= MAX_CASCADES and not self._stop.is_set(): + while self._inflight >= self.max_cascades and not self._stop.is_set(): self._inflight_lock.wait(0.5) - return 0 if self._stop.is_set() else MAX_CASCADES - self._inflight + return 0 if self._stop.is_set() else self.max_cascades - self._inflight def _dispatch(self, item: WorkItem) -> None: with self._inflight_lock: diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py index d2297a3..4adf66c 100644 --- a/backend/fluksio/flow/metrics.py +++ b/backend/fluksio/flow/metrics.py @@ -67,8 +67,16 @@ RECORDED = { } -def _minute(ts: float) -> datetime: - return datetime.fromtimestamp(ts, UTC).replace(second=0, microsecond=0) +def _minute(ts: float) -> int: + """The epoch second the minute containing ``ts`` starts at. + + An integer rather than a datetime: this runs once per event on the API's + event loop, and building a tz-aware datetime to key a dict with cost more + than everything the collector does with the event afterwards. The row + still wants one, so `_write` builds it — once per minute per node rather + than once per event. + """ + return int(ts // 60) * 60 def _detail(event: dict[str, Any]) -> str: @@ -84,7 +92,7 @@ class MetricsCollector: def __init__(self, events: EventBus, flush_s: float = FLUSH_INTERVAL_S) -> None: self._events = events self._flush_s = flush_s - self._buckets: dict[tuple[str, str, datetime], dict[str, float]] = {} + self._buckets: dict[tuple[str, str, int], dict[str, float]] = {} self._runs: dict[str, dict[str, Any]] = {} self._pending: list[EngineEvent] = [] # The traceback arrives one event before the failure it belongs to, @@ -302,7 +310,7 @@ class MetricsCollector: def _hold( self, - buckets: dict[tuple[str, str, datetime], dict[str, float]], + buckets: dict[tuple[str, str, int], dict[str, float]], pending: list[EngineEvent], ) -> None: """Take a batch the database refused back, rather than losing it. @@ -332,7 +340,7 @@ class MetricsCollector: def _write( self, - buckets: dict[tuple[str, str, datetime], dict[str, float]], + buckets: dict[tuple[str, str, int], dict[str, float]], pending: list[EngineEvent], runs: list[dict[str, Any]], prune: bool, @@ -340,7 +348,10 @@ class MetricsCollector: with Session(engine) as session: for (flow, node, minute), agg in buckets.items(): statement = insert(MetricBucket).values( - flow=flow, node=node, bucket=minute, **agg + flow=flow, + node=node, + bucket=datetime.fromtimestamp(minute, UTC), + **agg, ) # The same minute is written several times, so the counters add # and the maxima take whichever is larger. Columns are read by diff --git a/backend/fluksio/flow/nodes/http.py b/backend/fluksio/flow/nodes/http.py index 80d6d3f..b5879b2 100644 --- a/backend/fluksio/flow/nodes/http.py +++ b/backend/fluksio/flow/nodes/http.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import hmac import logging import threading @@ -388,8 +389,11 @@ class HttpNode(Node): if spec.port in data: typed_data[spec.port] = spec.coerce(data[spec.port]) - # Inject data into the pipeline (trigger mode nodes inject via provides) - result = self.inject(typed_data) + # Inject data into the pipeline (trigger mode nodes inject via + # provides). Off the loop: journalling is a blocking Redis + # round trip, and every webhook was making it on the thread + # the whole API answers from. + result = await asyncio.to_thread(self.inject, typed_data) return JSONResponse( content={ diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 39fa0e4..4cfb331 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -678,8 +678,7 @@ class Pipeline: ) def _increment_message_versions(self, outputs: dict[str, Any]) -> None: - for msg_name in outputs: - self._state.increment(self._version_key(msg_name)) + self._state.increment_multi([self._version_key(name) for name in outputs]) def _check_synchronous_ready(self, node: Node) -> tuple[bool, dict[str, int]]: """A synchronous node runs once every input is newer than last time. @@ -761,19 +760,17 @@ class Pipeline: ) return False - with self._state.lock(): - for name in due: - self._state[self._delivered_key(node.id, name)] = now + self._state.update({self._delivered_key(node.id, name): now for name in due}) return True def _is_node_ready(self, node: Node, state: StateBackend) -> bool: - with state.lock(): - for msg_name, spec in node.requires.items(): - # A non-triggering input is read if it happens to be there; - # waiting for it would make an accumulator's first run - # impossible, since it is what the node is about to write. - if spec.trigger and msg_name not in state: - return False + # A non-triggering input is read if it happens to be there; waiting for + # it would make an accumulator's first run impossible, since it is what + # the node is about to write. + waited_on = [msg for msg, spec in node.requires.items() if spec.trigger] + # One MGET rather than an EXISTS per input under the global state lock. + if len(state.get_present(waited_on)) != len(waited_on): + return False if not self._input_is_due(node): return False @@ -928,8 +925,10 @@ class Pipeline: started = time.perf_counter() collected = logs.Collector() try: - with state.lock(): - inputs = {k: state[k] for k in node.requires if k in state} + # One MGET. The lock this used to be read under bought nothing a + # single bulk read does not, and it was the engine's one global + # mutex — every node of every cascade queued behind it. + inputs = state.get_present(list(node.requires)) key = "" if self.run_cache is not None and node.fingerprint: @@ -977,37 +976,42 @@ class Pipeline: "ts": time.time(), } ) - self._observe( - NodeOutcome( - node=node.id, - ok=True, - duration_ms=duration_ms, - outputs=len(result or {}), - logs=collected.text, - artifacts={ - name: value - for name, value in (result or {}).items() - if is_reference(value) - }, - cache_key=key, - # Post-throttle: what went into state is what a later run - # restoring this node has to find. - output_values=result, + # Guarded rather than left to `_observe`: a live cascade has no + # observer, and building this model to drop it was one pydantic + # validation per node per message on the path that runs most. + if self.observer is not None: + self._observe( + NodeOutcome( + node=node.id, + ok=True, + duration_ms=duration_ms, + outputs=len(result or {}), + logs=collected.text, + artifacts={ + name: value + for name, value in (result or {}).items() + if is_reference(value) + }, + cache_key=key, + # Post-throttle: what went into state is what a later + # run restoring this node has to find. + output_values=result, + ) ) - ) return result except Exception as exc: # One failing node must not take the rest of the graph down. error = self.publish_error(node, exc, collected, entry_id) - self._observe( - NodeOutcome( - node=node.id, - ok=False, - duration_ms=round((time.perf_counter() - started) * 1000, 2), - error=error, - logs=collected.text, + if self.observer is not None: + self._observe( + NodeOutcome( + node=node.id, + ok=False, + duration_ms=round((time.perf_counter() - started) * 1000, 2), + error=error, + logs=collected.text, + ) ) - ) return None def _record_outputs( @@ -1021,13 +1025,14 @@ class Pipeline: node produced, leaving through a port it declared. """ ts = time.time() - with state.lock(): - state.update(outputs) - state.update({self._timestamp_key(name): ts for name in outputs}) + # Value and timestamp in one write, which is what the lock around two + # of them was for — a pipeline is a transaction, so they still land + # together and nobody waits on a mutex to do it. + state.update({**outputs, **{self._timestamp_key(name): ts for name in outputs}}) # Append-only, so it needs no lock of its own. state.append_history(outputs, ts, self.history_limits) self._increment_message_versions(outputs) - origin = node_source(node) + origin = node_source(node).model_dump() for name, value in outputs.items(): self._publish( { @@ -1036,7 +1041,7 @@ class Pipeline: "name": name, "value": value, "ts": ts, - "source": origin.model_dump(), + "source": origin, } ) @@ -1273,8 +1278,9 @@ class Pipeline: ) -> StateBackend: """Execute the graph (or one flow's nodes) against the shared state.""" if inputs: - with self._state.lock(): - self._state.update(inputs) + # `update` is already one transaction; the lock around it was not + # holding anything else still. + self._state.update(inputs) self._increment_message_versions(inputs) return self._execute_parallel(nodes, self._state, check_ready=False) @@ -1298,12 +1304,10 @@ class Pipeline: state = self._state ts = time.time() - with state.lock(): - state.update(outputs) - state.update({self._timestamp_key(name): ts for name in outputs}) + state.update({**outputs, **{self._timestamp_key(name): ts for name in outputs}}) state.append_history(outputs, ts, self.history_limits) self._increment_message_versions(outputs) - origin = node_source(node) + origin = node_source(node).model_dump() for name, value in outputs.items(): self._publish( { @@ -1312,7 +1316,7 @@ class Pipeline: "name": name, "value": value, "ts": ts, - "source": origin.model_dump(), + "source": origin, } ) # An injecting node — an MQTT subscriber, a webhook — publishes @@ -1487,11 +1491,12 @@ class Pipeline: origin = source or ValueSource(kind="api", label="API") ts = time.time() - with self._state.lock(): - self._state.update(values) - self._state.update({self._timestamp_key(name): ts for name in values}) + self._state.update( + {**values, **{self._timestamp_key(name): ts for name in values}} + ) self._state.append_history(values, ts, self.history_limits) self._increment_message_versions(values) + source_dump = origin.model_dump() for name, value in values.items(): self._publish( { @@ -1500,7 +1505,7 @@ class Pipeline: "name": name, "value": value, "ts": ts, - "source": origin.model_dump(), + "source": source_dump, } ) @@ -1580,18 +1585,25 @@ class Pipeline: self._run_here(node, outputs, cause="external") def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]: - """Last value and timestamp of every message, optionally one flow's.""" - out: dict[str, dict[str, Any]] = {} - with self._state.lock(): - keys = [k for k in self._state.keys() if not k.startswith("__")] - for key in keys: - if flow and flow_of(key) != flow: - continue - out[key] = { - "value": self._state.get(key), - "ts": self._state.get(self._timestamp_key(key)), - } - return out + """Last value and timestamp of every message, optionally one flow's. + + Two reads whatever the message count: this is what every websocket + snapshot calls, and it used to be a round trip per value and another + per timestamp, one at a time under the global state lock. + """ + keys = [ + k + for k in self._state.keys() + if not k.startswith("__") and (not flow or flow_of(k) == flow) + ] + if not keys: + return {} + stamps = [self._timestamp_key(k) for k in keys] + found = self._state.get_multi(keys + stamps) + return { + key: {"value": found.get(key), "ts": found.get(ts)} + for key, ts in zip(keys, stamps, strict=True) + } def reset(self) -> None: self._state.clear() diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index 39baa61..1953db4 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -13,7 +13,6 @@ honestly to "not". from __future__ import annotations import heapq -import json import logging import threading import time @@ -22,6 +21,7 @@ from collections import deque from dataclasses import dataclass, field from typing import Any, cast +import orjson import redis logger = logging.getLogger(__name__) @@ -78,7 +78,7 @@ class WorkItem: "kind": self.kind, "node": self.node, "flow": self.flow, - "outputs": json.dumps(self.outputs), + "outputs": orjson.dumps(self.outputs).decode(), "cause": self.cause, "not_before": str(self.not_before), "guard_key": self.guard_key, @@ -95,7 +95,7 @@ class WorkItem: kind=fields.get("kind", "cascade"), node=fields.get("node", ""), flow=fields.get("flow", ""), - outputs=json.loads(fields.get("outputs") or "{}"), + outputs=orjson.loads(fields.get("outputs") or "{}"), cause=fields.get("cause", "system"), not_before=float(fields.get("not_before") or 0.0), guard_key=fields.get("guard_key", ""), @@ -371,7 +371,9 @@ class RedisWorkQueue(WorkQueue): def add_delayed(self, item: WorkItem, not_before: float) -> None: item.not_before = not_before - self._redis.zadd(self._delayed_key, {json.dumps(item.to_fields()): not_before}) + self._redis.zadd( + self._delayed_key, {orjson.dumps(item.to_fields()): not_before} + ) def claim(self, count: int, block_ms: int) -> list[WorkItem]: response = cast( @@ -445,17 +447,29 @@ class RedisWorkQueue(WorkQueue): list[str], self._redis.zrangebyscore(self._delayed_key, "-inf", now, start=0, num=100), ) - moved = 0 + if not due: + return 0 + # Whoever removes it owns it: a second engine gets 0 for that member and + # leaves it alone. One round trip for the batch rather than one each, + # which matters because this runs every second. + pipe = self._redis.pipeline() for raw in due: - # Whoever removes it owns it: a second engine would get 0 here. - if self._redis.zrem(self._delayed_key, raw): - self._redis.xadd( + pipe.zrem(self._delayed_key, raw) + claimed = cast(list[int], pipe.execute()) + + pipe = self._redis.pipeline() + moved = 0 + for raw, owned in zip(due, claimed, strict=True): + if owned: + pipe.xadd( self._stream, - cast(Any, json.loads(raw)), + cast(Any, orjson.loads(raw)), maxlen=STREAM_MAXLEN, approximate=True, ) moved += 1 + if moved: + pipe.execute() return moved def dead_letter(self, item: WorkItem, reason: str) -> None: @@ -468,7 +482,7 @@ class RedisWorkQueue(WorkQueue): def park(self, flow: str, item: WorkItem) -> None: pipe = self._redis.pipeline() - pipe.rpush(self._parked_key(flow), json.dumps(item.to_fields())) + pipe.rpush(self._parked_key(flow), orjson.dumps(item.to_fields())) pipe.sadd(self._parked_flows_key, flow) pipe.execute() @@ -479,7 +493,7 @@ class RedisWorkQueue(WorkQueue): pipe.delete(key) pipe.srem(self._parked_flows_key, flow) pipe.execute() - return [WorkItem.from_fields(json.loads(r), "") for r in raw] + return [WorkItem.from_fields(orjson.loads(r), "") for r in raw] def unpark_one(self, flow: str) -> WorkItem | None: key = self._parked_key(flow) @@ -489,7 +503,7 @@ class RedisWorkQueue(WorkQueue): raw, remaining = cast(tuple["str | None", int], pipe.execute()) if not remaining: self._redis.srem(self._parked_flows_key, flow) - return WorkItem.from_fields(json.loads(raw), "") if raw else None + return WorkItem.from_fields(orjson.loads(raw), "") if raw else None def clear_flow(self, flow: str) -> None: pipe = self._redis.pipeline() diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py index 9f8f158..9e10b00 100644 --- a/backend/fluksio/flow/state.py +++ b/backend/fluksio/flow/state.py @@ -7,7 +7,6 @@ supporting both in-memory storage and Redis for distributed execution. from __future__ import annotations -import json from abc import ABC, abstractmethod from collections import deque from collections.abc import Iterator @@ -15,8 +14,16 @@ from contextlib import contextmanager from threading import RLock from typing import Any, cast +import orjson import redis +#: A node may return a dict keyed by something other than a string, and the +#: stdlib encoder this replaced turned those keys into strings rather than +#: refusing them. Two differences remain and are both improvements: a +#: non-finite number is stored as ``null`` instead of the bare ``NaN`` that is +#: not JSON at all, and a datetime serialises rather than raising. +_JSON_OPTS = orjson.OPT_NON_STR_KEYS + # A sparkline only means something for numbers, so the history keeps the values # it can plot and nothing else. 120 points fill a panel-wide chart while leaving # Redis a cache rather than a time-series database. @@ -147,6 +154,26 @@ class StateBackend(ABC): """ ... + @abstractmethod + def get_present(self, keys: list[str]) -> dict[str, Any]: + """The values of those keys that exist, in one round trip. + + Unlike :meth:`get_multi`, a key that is absent is left out rather than + mapped to None — which is what reading a node's inputs needs, since a + message holding ``null`` is a message the node has, and one that was + never published is not. + """ + ... + + @abstractmethod + def increment_multi(self, keys: list[str]) -> None: + """Bump several counters at once. + + The new values are not returned: the only reader compares them against + what it saw last time and reads them back itself. + """ + ... + @abstractmethod def compare_and_swap_multi( self, @@ -294,6 +321,16 @@ class MemoryState(StateBackend): with self._lock: return {k: self._data.get(k) for k in keys} + def get_present(self, keys: list[str]) -> dict[str, Any]: + """The values of those keys that are there.""" + with self._lock: + return {k: self._data[k] for k in keys if k in self._data} + + def increment_multi(self, keys: list[str]) -> None: + with self._lock: + for key in keys: + self._data[key] = self._data.get(key, 0) + 1 + def compare_and_swap_multi( self, expected: dict[str, Any], @@ -389,11 +426,11 @@ class RedisState(StateBackend): def _serialize(self, value: Any) -> bytes: """Serialize value for storage.""" - return json.dumps(value).encode() + return orjson.dumps(value, option=_JSON_OPTS) def _deserialize(self, data: bytes | None) -> Any: """Deserialize value from storage.""" - return json.loads(data) if data else None + return orjson.loads(data) if data else None # redis-py types every command as a sync/async union; this is the # synchronous client, so the results are narrowed where they are consumed. @@ -503,6 +540,15 @@ class RedisState(StateBackend): """Atomically increment a counter using Redis INCR.""" return cast(int, self._client.incr(self._key(key))) + def increment_multi(self, keys: list[str]) -> None: + """Bump every counter in one round trip rather than one INCR each.""" + if not keys: + return + pipe = self._client.pipeline() + for key in keys: + pipe.incr(self._key(key)) + pipe.execute() + def get_multi(self, keys: list[str]) -> dict[str, Any]: """Get multiple values atomically using Redis MGET.""" if not keys: @@ -517,6 +563,33 @@ class RedisState(StateBackend): return result + def get_present(self, keys: list[str]) -> dict[str, Any]: + """The keys that are there, from one MGET. + + A missing key comes back as ``None`` from Redis and a stored ``null`` + comes back as the four bytes; only the raw reply tells the two apart, + which is why this is not ``get_multi`` with the Nones filtered out. + """ + if not keys: + return {} + values = cast( + list[bytes | None], self._client.mget([self._key(k) for k in keys]) + ) + return { + key: self._deserialize(raw) + for key, raw in zip(keys, values, strict=True) + if raw is not None + } + + def __getitem__(self, key: str) -> Any: + # One GET rather than the base class's EXISTS-then-GET: a missing key + # and a stored `null` differ in the raw reply, so nothing has to ask + # twice. + raw = cast(bytes | None, self._client.get(self._key(key))) + if raw is None: + raise KeyError(key) + return self._deserialize(raw) + def compare_and_swap_multi( self, expected: dict[str, Any], @@ -586,7 +659,7 @@ class RedisState(StateBackend): continue history_key = self._history_key(key) cap = (limits or {}).get(key, HISTORY_LIMIT) - pipe.lpush(history_key, json.dumps([ts, number])) + pipe.lpush(history_key, orjson.dumps([ts, number])) pipe.ltrim(history_key, 0, cap - 1) if self._ttl: pipe.expire(history_key, self._ttl) @@ -600,6 +673,6 @@ class RedisState(StateBackend): # LPUSH puts the newest first, a chart reads the other way round. points: list[tuple[float, float]] = [] for entry in reversed(entries): - ts, value = json.loads(entry) + ts, value = orjson.loads(entry) points.append((float(ts), float(value))) return points diff --git a/backend/fluksio/flow/workers.py b/backend/fluksio/flow/workers.py index d52e66b..985b2a7 100644 --- a/backend/fluksio/flow/workers.py +++ b/backend/fluksio/flow/workers.py @@ -14,7 +14,6 @@ from __future__ import annotations import hashlib import itertools -import json import logging import os import queue @@ -28,6 +27,7 @@ from collections.abc import Callable from pathlib import Path from typing import Any +import orjson from fluksio_worker import worker_main as _worker_main from fluksio.flow.events import EventBus @@ -155,7 +155,12 @@ class _Worker: def send(self, request: dict[str, Any]) -> None: assert self.proc.stdin is not None - self.proc.stdin.write((json.dumps(request) + "\n").encode()) + # Node inputs may be keyed by something other than a string, which + # the stdlib encoder this replaced turned into strings rather than + # refusing. + self.proc.stdin.write( + orjson.dumps(request, option=orjson.OPT_NON_STR_KEYS) + b"\n" + ) self.proc.stdin.flush() def read_line(self, deadline: float) -> str | None: @@ -414,7 +419,7 @@ class PythonWorkerPool: ) if line: try: - message = dict(json.loads(line)) + message = dict(orjson.loads(line)) except (TypeError, ValueError) as exc: # A reply we cannot read leaves this worker out of step: # whatever is still in its pipe would be taken by the diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index d440a93..587ceaa 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -107,6 +107,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: queue=_work_queue(), max_workers=settings.FLOW_MAX_WORKERS, events=event_bus, + max_cascades=settings.FLOW_MAX_CASCADES, ) store = FlowStore(settings.FLOWS_DIR) # The packages node code imports, before anything tries to import them. diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 65c5341..ef34de7 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -38,6 +38,10 @@ dependencies = [ "pwdlib[argon2,bcrypt]>=0.3.0", "numpy>=2.2.6", "redis>=7.1.0", + # Every message costs a round trip's worth of encode and decode — state, + # the journal, the worker pipe. Engine-side only: `fluksio-worker` is + # copied onto other people's machines and stays dependency-free. + "orjson>=3.10", "cryptography>=44.0.0", "aiomqtt>=2.0.0", "influxdb-client[async]>=1.40.0", diff --git a/backend/scripts/bench_engine.py b/backend/scripts/bench_engine.py index 6ed1164..3274366 100644 --- a/backend/scripts/bench_engine.py +++ b/backend/scripts/bench_engine.py @@ -198,14 +198,18 @@ class Engine: def drain(self, timeout: float = 60.0) -> bool: """Wait for the queue to go quiet. False means it never did. + Asks the backlog rather than the whole of ``stats()``: the health + summary's version costs four round trips, and polling it every twenty + milliseconds competes with the engine for the connection it is + supposedly measuring. + Delayed items are not waited for: a rate-limit flush is due a whole window from now, and the scenario is over long before that. """ deadline = time.monotonic() + timeout quiet_since = 0.0 while time.monotonic() < deadline: - stats = self.service.stats() - busy = stats.get("cascades_busy", 0) or stats.get("backlog", 0) + busy = self.service.inflight or self.queue.backlog() if not busy: # Two consecutive quiet reads: a cascade between claim and ack # shows as neither. diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index 6d7620f..25c3a99 100644 --- a/backend/tests/flow/test_queue.py +++ b/backend/tests/flow/test_queue.py @@ -423,9 +423,9 @@ def test_no_more_is_claimed_than_the_pool_can_run(): time.sleep(0.5) stats = service.stats() - assert stats["cascades_busy"] <= executor.MAX_CASCADES + assert stats["cascades_busy"] <= service.max_cascades # And the journal entries of what is only waiting are still free. - assert stats["pending"] <= executor.MAX_CASCADES + assert stats["pending"] <= service.max_cascades # Waiting is not idle: the rest of the forty is the backlog. assert stats["backlog"] >= 30 finally: diff --git a/uv.lock b/uv.lock index 0f44785..118d288 100644 --- a/uv.lock +++ b/uv.lock @@ -885,6 +885,7 @@ dependencies = [ { name = "jinja2" }, { name = "mcp" }, { name = "numpy" }, + { name = "orjson" }, { name = "pwdlib", extra = ["argon2", "bcrypt"] }, { name = "pydantic" }, { name = "pydantic-settings" }, @@ -922,6 +923,7 @@ requires-dist = [ { name = "jinja2", specifier = ">=3.1.4,<4.0.0" }, { name = "mcp", specifier = ">=1.29,<2" }, { name = "numpy", specifier = ">=2.2.6" }, + { name = "orjson", specifier = ">=3.10" }, { name = "pwdlib", extras = ["argon2", "bcrypt"], specifier = ">=0.3.0" }, { name = "pydantic", specifier = ">2.0" }, { name = "pydantic-settings", specifier = ">=2.2.1,<3.0.0" }, @@ -1697,6 +1699,58 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/32/0a/2ec5deea6dcd158f254a7b372fb09cfba5719419c8d66343bab35237b3fb/numpy-2.4.2-cp314-cp314t-win_arm64.whl", hash = "sha256:1f92f53998a17265194018d1cc321b2e96e900ca52d54c7c77837b71b9465181", size = 10565379, upload-time = "2026-01-31T23:12:51.345Z" }, ] +[[package]] +name = "orjson" +version = "3.12.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/0f/f3/742fb1f62b825f2c010697eaf4e828004bc2a81e7e806666989c132c7c42/orjson-3.12.0.tar.gz", hash = "sha256:d14203fb1aae2ad9b3d52f8a0e82aeb10197ef1c9bc61da7f358bd70b00123d5", size = 4142915, upload-time = "2026-08-14T16:13:30.607Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/be/4a/295da39c651c2faac8bd351a2a346f0fdedd9d50b847ee9dfc27d2207ef6/orjson-3.12.0-cp312-cp312-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:aa3e43a6846e91d7bde3d5a9c66090fcd8744f569a9b6cffc5e1ca38f6a461c0", size = 223427, upload-time = "2026-08-14T16:12:28.525Z" }, + { url = "https://files.pythonhosted.org/packages/29/98/758cf90fbeaaafb7f8141bfac75a432099959f3a2f5db93a412e876415d8/orjson-3.12.0-cp312-cp312-macosx_15_0_arm64.whl", hash = "sha256:11edb4660a6680abee9788a3a9072208a2c96538cc1322bd79542065229d8e54", size = 123725, upload-time = "2026-08-14T16:12:30.013Z" }, + { url = "https://files.pythonhosted.org/packages/32/b5/5b934d251f8651f7e41df180ad0c57a6e1cabe15c7bd331638413a50ebc9/orjson-3.12.0-cp312-cp312-manylinux2014_armv7l.manylinux_2_17_armv7l.whl", hash = "sha256:2d3a9da945a4d96ae758fdaaca56742e6b73b6fd554c5d8876f252a6dad70b83", size = 113375, upload-time = "2026-08-14T16:12:31.209Z" }, + { url = "https://files.pythonhosted.org/packages/cd/d2/37efb5b12a176ce3ced29f4144f20da57d02757f78ce549637dc1b4e1fc8/orjson-3.12.0-cp312-cp312-manylinux2014_i686.manylinux_2_17_i686.whl", hash = "sha256:92ffc09e07233a6ab6d4e067f7841edcbcc134cb4812155cf171ea5255a421d7", size = 129983, upload-time = "2026-08-14T16:12:32.721Z" }, + { url = "https://files.pythonhosted.org/packages/50/22/0644b87c73f13e0092df8f35a1fe280d991e5e90072087411e0dd7e44e0c/orjson-3.12.0-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:bf44e374aadde77b1f6109f1030be51433eb61984379852766b6f4e187db7b1e", size = 130629, upload-time = "2026-08-14T16:12:34.084Z" }, + { url = "https://files.pythonhosted.org/packages/8c/57/80b986ebfecd9c6a177ddf1c2319717f0cd8feffb2b78946595a18a2fc88/orjson-3.12.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:1192a7021b6d071aaf909864f6e924d6a2675ca360485b972b8401749311750b", size = 131245, upload-time = "2026-08-14T16:12:35.713Z" }, + { url = "https://files.pythonhosted.org/packages/80/3d/75c5ac5a69161f44492a68fbdde66f4cc4ce48cd5e1fb05918e46f0c8848/orjson-3.12.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:53c0c474a9d9aff9aebfc0c88de1f28f843d940e6e3a80729abdf6a20274356f", size = 135397, upload-time = "2026-08-14T16:12:37.128Z" }, + { url = "https://files.pythonhosted.org/packages/71/93/4d71f2df314a97ff0d27a4559bf5888fc8406e3c6dec90e92291e3511215/orjson-3.12.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:532ff8cd4bd59a327a953a7dcde922c7fc25b85e29721bb8633265430d3a3873", size = 127693, upload-time = "2026-08-14T16:12:38.627Z" }, + { url = "https://files.pythonhosted.org/packages/bc/1d/0dbc6be5adfd1730491072fb60beb6bcdf5d7b2596ee41b7fc2e298bfc09/orjson-3.12.0-cp312-cp312-win32.whl", hash = "sha256:a6cf4b18e7de173f209f2084ffbd736dd72389a396326ee80a7022168be232e5", size = 128000, upload-time = "2026-08-14T16:12:39.954Z" }, + { url = "https://files.pythonhosted.org/packages/2d/c9/97b1ce0112ebf5e949c775ed5b1755e562233179f3584579673cc24d6378/orjson-3.12.0-cp312-cp312-win_amd64.whl", hash = "sha256:010811c1b69773450a01cef97727a67b223242f350b77d4ca000e59a9ef2155a", size = 122106, upload-time = "2026-08-14T16:12:41.324Z" }, + { url = "https://files.pythonhosted.org/packages/a8/6a/facd8b312e4a0d3a7fa978c7e15821f74a336adf1d65529faec33b48e18b/orjson-3.12.0-cp312-cp312-win_arm64.whl", hash = "sha256:ad29eece0c601737f2a60edc2752a84e7a0785df3efb62e3012834700a5afe0d", size = 126869, upload-time = "2026-08-14T16:12:42.651Z" }, + { url = "https://files.pythonhosted.org/packages/54/cb/d7b78218a987eb8a8ce4eeae0286b1bb679333eb631ea0eeaf6371680bfc/orjson-3.12.0-cp313-cp313-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:9a36ec60f1796f9a3f13e3b98390295e17a1c7c10155b448d264098bf9ee5900", size = 223397, upload-time = "2026-08-14T16:12:44.003Z" }, + { url = "https://files.pythonhosted.org/packages/f8/4a/bc87c45e7ec639d35ebefd62618e01939531ac8e171426606a01bda05914/orjson-3.12.0-cp313-cp313-macosx_15_0_arm64.whl", hash = "sha256:ad0422b92d5195443a39f80c3bcf731cc2e00f153bd32063a47b73b057bd0f03", size = 123662, upload-time = "2026-08-14T16:12:45.433Z" }, + { url = "https://files.pythonhosted.org/packages/94/ee/c9a4ff3f2dbedbbe9e635d0fa72c8866adede09b6335ef9644f53752f0d8/orjson-3.12.0-cp313-cp313-manylinux2014_armv7l.manylinux_2_17_armv7l.whl", hash = "sha256:5a0fdbc216388f653d3752ff310e710f59253bd4ed6a2bfb3f4f06b84714bbd8", size = 113374, upload-time = "2026-08-14T16:12:46.755Z" }, + { url = "https://files.pythonhosted.org/packages/75/09/3f330a026a796c8b4c97a6f429652a5e912e7065039bf96ed25e42aa7b25/orjson-3.12.0-cp313-cp313-manylinux2014_i686.manylinux_2_17_i686.whl", hash = "sha256:2eb5c56e534127b2b8fa38d2363c8b1b8190367ee0d1d16c041517d880843b94", size = 130029, upload-time = "2026-08-14T16:12:48.06Z" }, + { url = "https://files.pythonhosted.org/packages/7d/40/094cc53126a3d22f76cdf83b6ea67338bed01d774037621a785aa8e6e5ea/orjson-3.12.0-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:784106539f4b9d4b930e0b4eb8d45168507dae001945e71b4675a367f1e5e806", size = 130528, upload-time = "2026-08-14T16:12:49.362Z" }, + { url = "https://files.pythonhosted.org/packages/bc/74/89bb236deb9565f99434b13052bb40ddfcce4adf3afbfa3132ee7e421468/orjson-3.12.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:1c680706fc8396d95e7c4c1f9482563f552137aef91b57237a3ad5aaf64629df", size = 131075, upload-time = "2026-08-14T16:12:50.692Z" }, + { url = "https://files.pythonhosted.org/packages/0c/ac/1176360d762c01b5bd34acd56fc098e936c491363d8b6b397ad4aa475547/orjson-3.12.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:83445adc40cba26d6d621185a45128ce455b766af368cad2ab64b970603a7978", size = 135321, upload-time = "2026-08-14T16:12:52.114Z" }, + { url = "https://files.pythonhosted.org/packages/7a/02/bbd881c8b9276d50b998de38b4e97de8ace1aac940b0ee545aedbf65ed00/orjson-3.12.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:644d005bc82f917337a95ce270c9f6f92f9834c2bed7b1477572f8db00784222", size = 127472, upload-time = "2026-08-14T16:12:53.517Z" }, + { url = "https://files.pythonhosted.org/packages/8e/02/a0934d7503e6dcbedd6afac3e7f3f8597fd09389949ad94d0f7540e9dbca/orjson-3.12.0-cp313-cp313-win32.whl", hash = "sha256:d8e78d3d93705e3d27cc17cdb209e44d7a8ea203010cac6ce9c7ffc1ae1996f1", size = 128000, upload-time = "2026-08-14T16:12:55.14Z" }, + { url = "https://files.pythonhosted.org/packages/52/87/69f98f8d40faff103a965a5fbb83f08241b01beaf92badb5413fbc9358cc/orjson-3.12.0-cp313-cp313-win_amd64.whl", hash = "sha256:b85931be5b6763c31283805c9bdaae1ca03ad9f6f12a15f1cbf6745b907932c2", size = 121841, upload-time = "2026-08-14T16:12:56.507Z" }, + { url = "https://files.pythonhosted.org/packages/e6/07/b83046a4e3cadcc0987d0f160696107c4af706a619b56e4ad01940cadadf/orjson-3.12.0-cp313-cp313-win_arm64.whl", hash = "sha256:6a31348d7dfa64cd9c78bd1f510ff44c48fe64d71094e6b90e364dba3b55949e", size = 126765, upload-time = "2026-08-14T16:12:57.806Z" }, + { url = "https://files.pythonhosted.org/packages/12/9d/3931253e6f3148abf2cbe14830367042a4806b362ea520df2303db188fb9/orjson-3.12.0-cp314-cp314-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:9e6fee342a48760e854d743e7a81534d8e2925a6f46e09f750cf56b50fd1de5d", size = 223391, upload-time = "2026-08-14T16:12:59.184Z" }, + { url = "https://files.pythonhosted.org/packages/8a/0e/b4a4f1e305367245877b967a0bad70fcf001d77c54ac4339a120b66fdae4/orjson-3.12.0-cp314-cp314-macosx_15_0_arm64.whl", hash = "sha256:8c3bb86dd10f39b3fbf434b7d5dc7cac77d6fc8ac572ae30a10731ede2c4b647", size = 123659, upload-time = "2026-08-14T16:13:00.548Z" }, + { url = "https://files.pythonhosted.org/packages/96/f3/6782c6fa85e2702bc66be183c3b421486167dcf266ee4dc1403fe3824870/orjson-3.12.0-cp314-cp314-manylinux2014_armv7l.manylinux_2_17_armv7l.whl", hash = "sha256:2bb3ce43203936072dd8b4917b01d3aecfc02329bfb42510cb7cfb24708adc9c", size = 113337, upload-time = "2026-08-14T16:13:02.009Z" }, + { url = "https://files.pythonhosted.org/packages/bf/79/b32ab64bacda9d0fa4942ef483bd03cabf0eaf2be819ca9fb7ff610c559d/orjson-3.12.0-cp314-cp314-manylinux2014_i686.manylinux_2_17_i686.whl", hash = "sha256:6a2a79c89984dc719817d388c8709e0efc2a2795a934eaa746b4882eb6045adc", size = 130112, upload-time = "2026-08-14T16:13:03.404Z" }, + { url = "https://files.pythonhosted.org/packages/ee/49/6e6142999ca01509219be5e5a9c338a3e5ea011f63e91ff473fbbf3734ed/orjson-3.12.0-cp314-cp314-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f06dd838d1e07d9b1de0932ec0485ec92c4d5f5d1ad4817a656268c3e88be1e1", size = 130520, upload-time = "2026-08-14T16:13:04.798Z" }, + { url = "https://files.pythonhosted.org/packages/49/d0/3745af0a4cc9867784f29722929cec4d10bd1c877cd754b01ba6d96eb21a/orjson-3.12.0-cp314-cp314-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c6b11be792c3d2c6a4be2af4ebf97a68d0bf5f580aca6e86a418a354f6cc846a", size = 131053, upload-time = "2026-08-14T16:13:06.14Z" }, + { url = "https://files.pythonhosted.org/packages/c3/f4/6fe5a22fa478fffb190e65c338c84df5c311ef597b363150a17cc57063c0/orjson-3.12.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:477ecaf6b9f88f873341b91fcc736119ca81b5e002a9f7f308ff5b4f2ce2a70e", size = 135321, upload-time = "2026-08-14T16:13:07.544Z" }, + { url = "https://files.pythonhosted.org/packages/ff/41/b1b0ec30289646a81a76e2dbaae2686b96fcccb7cb0323dc1dd78cbc7875/orjson-3.12.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:f3c0683136acdc29afdf88a5bc2f7d3d0e34087788d1d63c0144b805a87a196f", size = 127485, upload-time = "2026-08-14T16:13:08.88Z" }, + { url = "https://files.pythonhosted.org/packages/bf/2b/277404bdcc21c93b112b963655b76443ebfe828f8a3ff1de7d90f8850eb3/orjson-3.12.0-cp314-cp314-win32.whl", hash = "sha256:d39f3f5c3927e2dc0913fe5bbc1a2f6b1b9d1bba1de6358340d0ad0d0c00ca92", size = 128048, upload-time = "2026-08-14T16:13:10.305Z" }, + { url = "https://files.pythonhosted.org/packages/41/2b/395b36fa2b4ce7af70b651d715e88f80d884b2c2b14a6b53e84d554fb5f0/orjson-3.12.0-cp314-cp314-win_amd64.whl", hash = "sha256:0b1ac5bf6609b2716c7954011c5fef6254922df029f45d032ee4ebf5d363cbed", size = 121858, upload-time = "2026-08-14T16:13:11.634Z" }, + { url = "https://files.pythonhosted.org/packages/ea/a3/833e895ff452859eebe75093d26691fe9108f1a7a6a08435d7a5780ea652/orjson-3.12.0-cp314-cp314-win_arm64.whl", hash = "sha256:50fae885cb073eac7556353ff3df93312b0d5137b0a5056b2bb63f97ed9a93c7", size = 126749, upload-time = "2026-08-14T16:13:13.117Z" }, + { url = "https://files.pythonhosted.org/packages/58/64/99c8947ece10c17176af9aae85c4948f1d109da77440ec14d87239efaf73/orjson-3.12.0-cp315-cp315-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:01efac2074fffb4cb1ea3fab7861e9d0f2a26913854a972f5ac760525dbdaf6e", size = 223398, upload-time = "2026-08-14T16:13:14.694Z" }, + { url = "https://files.pythonhosted.org/packages/3e/30/cf983fe09f2731420fda097a9f7ef4343f47fa216c228961ad8f6da44f3d/orjson-3.12.0-cp315-cp315-macosx_15_0_arm64.whl", hash = "sha256:ed4ca42bd55955aa34deedcfdfd0e0c31abf51143aae158ae2bc3520b626e517", size = 123655, upload-time = "2026-08-14T16:13:16.221Z" }, + { url = "https://files.pythonhosted.org/packages/11/50/9cb8ae73fa4749dbbc20f617004213b5ff01c20aaeec34c3f31124f2c1d8/orjson-3.12.0-cp315-cp315-manylinux_2_39_aarch64.whl", hash = "sha256:40f92192227505acca4e2533ce565f8e6b9535f7d0d09b0968452f18b7376b38", size = 130515, upload-time = "2026-08-14T16:13:17.601Z" }, + { url = "https://files.pythonhosted.org/packages/9f/0a/adb6ce1a5b5fbf9cb1790f9961bb668a0dd5429aadaf6cee044724681795/orjson-3.12.0-cp315-cp315-manylinux_2_39_armv7l.whl", hash = "sha256:33efefcf5d88eaf400b47e2eba02f91f319bb9951be61ca500b7d536d3f2079d", size = 113327, upload-time = "2026-08-14T16:13:18.927Z" }, + { url = "https://files.pythonhosted.org/packages/51/5c/d17f61581d8dbdde7048f87a330fa24915edec38db4d72b381fec14fbb56/orjson-3.12.0-cp315-cp315-manylinux_2_39_i686.whl", hash = "sha256:8e386b0bc0ddd7cd2056f884b5a0af33592bd01ac66a7ca4b42a65a7e7774a13", size = 130105, upload-time = "2026-08-14T16:13:20.317Z" }, + { url = "https://files.pythonhosted.org/packages/9f/b7/938befcf33bee4704a92ecec6a2731224c539d939bf9429fd39396d28931/orjson-3.12.0-cp315-cp315-manylinux_2_39_x86_64.whl", hash = "sha256:58c58e1de0006ffb580368d6793c36c7b0b021db066479cf281bf5061e732328", size = 131049, upload-time = "2026-08-14T16:13:21.719Z" }, + { url = "https://files.pythonhosted.org/packages/b0/15/cfa2021d64d5aa8bb5c9f604ef375e00ec8b657651b5dd650b1b7ad13df1/orjson-3.12.0-cp315-cp315-musllinux_1_2_aarch64.whl", hash = "sha256:08231552159be266a7269555bd9f7c016aee7d9ad6dab06eb58796c5ccb7101c", size = 135320, upload-time = "2026-08-14T16:13:23.415Z" }, + { url = "https://files.pythonhosted.org/packages/1a/50/3e75dfe357c1e8f9e287c7a5740260ef15bd23a5299eae8d0835dcad5375/orjson-3.12.0-cp315-cp315-musllinux_1_2_x86_64.whl", hash = "sha256:a15f9a891bce5f5cc5d210e3ad8614d4d1b489a56448c099d6d2a7168b2d954a", size = 127488, upload-time = "2026-08-14T16:13:24.791Z" }, + { url = "https://files.pythonhosted.org/packages/11/a6/79aed402eb3ab284dc5b4791a7ad62c5875127de01b8e3f04bd92d551298/orjson-3.12.0-cp315-cp315-win32.whl", hash = "sha256:03091c8a64db4be38746597ceea68f33c238e27acd9bfe99fb59420224ae7a55", size = 128048, upload-time = "2026-08-14T16:13:26.217Z" }, + { url = "https://files.pythonhosted.org/packages/64/f7/2723e264aab7248c1ed6ecaad8e5d0cb866c0cffde75442102ffa7491aba/orjson-3.12.0-cp315-cp315-win_amd64.whl", hash = "sha256:2b7bcefb9f40fa242fa6b06377232c048e655747790829609168c01162f60578", size = 121860, upload-time = "2026-08-14T16:13:27.577Z" }, + { url = "https://files.pythonhosted.org/packages/82/56/630c9113ec8996778f1f0304b364b091b9a9db5fef5fdc17cca622f5ea24/orjson-3.12.0-cp315-cp315-win_arm64.whl", hash = "sha256:859fc4196855890150bb08e649b30d2c93b249b3e3edd0d3bb2231abf8aa8adc", size = 126754, upload-time = "2026-08-14T16:13:28.962Z" }, +] + [[package]] name = "packaging" version = "25.0"