Claim only what the cascade pool can run

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

backend/fluksio/flow/executor.py

@@ -1,9 +1,11 @@
 """The execution service: what turns journaled work into node runs.
 
 A consumer thread claims items from the work queue and hands each one to a
-dispatch pool, which drives the wave it starts. Node bodies run on a second,
-separate pool: if cascade drivers and node bodies shared one, a wave waiting
-for its own nodes could occupy every thread and deadlock.
+dispatch pool, which drives the wave it starts. It claims only what that pool
+can start, so a backlog waits in the queue rather than inside the process.
+Node bodies run on a second, separate pool: if cascade drivers and node bodies
+shared one, a wave waiting for its own nodes could occupy every thread and
+deadlock.
 
 A reaper takes back items claimed by an engine that died before acknowledging
 them, which is the mechanism that makes a crash mid-cascade recoverable rather
@@ -134,8 +136,11 @@ class ExecutionService:
             if not self._intake.is_set():
                 self._intake.wait(timeout=0.5)
                 continue
+            free = self._await_capacity()
+            if not free:
+                continue
             try:
-                items = self.queue.claim(CLAIM_COUNT, CLAIM_BLOCK_MS)
+                items = self.queue.claim(min(CLAIM_COUNT, free), CLAIM_BLOCK_MS)
                 failures = 0
             except Exception as exc:
                 failures += 1
@@ -185,6 +190,21 @@ class ExecutionService:
             except Exception as exc:
                 logger.error("Could not reclaim stale work: %s", exc)
 
+    def _await_capacity(self) -> int:
+        """How many cascades may be claimed now. Zero means the service stops.
+
+        Claiming past what the pool can run makes nothing faster: the extra
+        items queue up inside the pool, count as in flight and hold their
+        journal entries open the whole time, which is how four cascade threads
+        came to report hundreds busy on a healthy engine. Work left in the
+        stream is work that is still anyone's to take; work that is claimed is
+        work that is actually being run.
+        """
+        with self._inflight_lock:
+            while self._inflight >= 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
+
     def _dispatch(self, item: WorkItem) -> None:
         with self._inflight_lock:
             self._inflight += 1

backend/tests/flow/test_queue.py

@@ -333,3 +333,56 @@ def test_work_in_flight_is_touched_until_it_finishes(monkeypatch):
     finally:
         release.set()
         service.stop()
+
+
+def test_no_more_is_claimed_than_the_pool_can_run():
+    """A backlog belongs in the queue, not inside the process.
+
+    Claiming ahead of the pool used to leave every waiting item counted as a
+    busy cascade and holding its journal entry open, so four cascade threads
+    reported hundreds in flight on an engine that was merely behind.
+    """
+    release = threading.Event()
+
+    def slow(reading, params):
+        release.wait(5)
+        return {"doubled": reading * 2}
+
+    source = Node(
+        f=lambda params: None,
+        provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
+        name="source",
+    )
+    consumer = Node(
+        f=slow,
+        requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
+        provides=[MessageSpec(name="doubled", dtype=DType.FLOAT)],
+        name="consumer",
+    )
+    source.assign_flow("f", "source")
+    consumer.assign_flow("f", "consumer")
+
+    queue = MemoryWorkQueue()
+    pipeline = Pipeline(nodes=[source, consumer], state=MemoryState(), work_queue=queue)
+    service = ExecutionService(queue)
+    service.bind(pipeline)
+    service.start()
+    try:
+        for i in range(40):
+            queue.add(
+                WorkItem(
+                    kind="cascade",
+                    node="f.source",
+                    flow="f",
+                    outputs={"f.reading": float(i)},
+                )
+            )
+        time.sleep(0.5)
+
+        stats = service.stats()
+        assert stats["cascades_busy"] <= executor.MAX_CASCADES
+        # And the journal entries of what is only waiting are still free.
+        assert stats["pending"] <= executor.MAX_CASCADES
+    finally:
+        release.set()
+        service.stop()