Stop paying five Redis round trips and a global lock per message

The engine was I/O-bound on its own state backend. `RedisState.lock()` is one
key — `pipeline:_lock` — for the whole process, taken five times a message at
two round trips each, and every cascade and every node read queued behind it.
Inside it, reading a node's inputs was three round trips per input (an EXISTS
for `in`, then EXISTS and GET for the value), writing was two updates that a
single transaction already gives, and the version counters went one INCR at a
time.

Replaced with the atomic command that was always available: `get_present` is
one MGET and tells a missing key from one holding null, so the lock it used to
be read under bought nothing; value and timestamp land in one `update`, which
is a MULTI/EXEC; `increment_multi` pipelines the counters. `values()` — what
every websocket snapshot calls — is two reads whatever the message count
instead of two per message.

Beside that: every webhook did its blocking XADD on the asyncio event loop
(MQTT already used `to_thread`); the per-execution `NodeOutcome` was built and
validated even with no run watching; `_minute` built a tz-aware datetime per
event on the loop thread to key a dict, and now keys on an int; `move_due`
promoted delayed items one round trip each, every second; `FLOW_MAX_CASCADES`
makes the in-flight ceiling a setting rather than a constant.

`orjson` replaces stdlib json where a message pays for it — state, the
journal, the engine side of the worker pipe. `fluksio-worker` stays
dependency-free, and the run-cache digest stays on stdlib so no stored key is
invalidated. A non-finite number now stores as `null` rather than the bare
`NaN` that was never JSON.

Measured with `scripts/bench_engine.py` against a real Redis, 200 messages:
a five-node chain went from 43.9 to 103.1 msg/s with p50 latency 2110ms →
782ms and p95 3913ms → 1439ms; one source into twenty consumers went from 5.4
to 33.7 msg/s. In memory, twenty consumers went from 187 to 448 msg/s.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BpfSinyCBfjuieikyfMPbf

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()