Soak engine scenario: a sleeping fixture node keeps cascades in flight for the kill, and the report says how many were unacknowledged

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx

backend/scripts/soak.py

@@ -149,8 +149,9 @@ PREFIX = "soak_"
 FLOW_LOAD = f"{PREFIX}load"
 FLOW_CHAIN = f"{PREFIX}chain"
 FLOW_MQTT = f"{PREFIX}mqtt"
+FLOW_SLOW = f"{PREFIX}slow"
 #: Deletion order: whatever reads another flow's messages goes first.
-FLOWS = (FLOW_MQTT, FLOW_CHAIN, FLOW_LOAD)
+FLOWS = (FLOW_SLOW, FLOW_MQTT, FLOW_CHAIN, FLOW_LOAD)
 
 COUNT_SOURCE = '''"""Count what made it through, reading the total it also writes."""
 
@@ -159,6 +160,16 @@ def process(tick, total=0, params=None):
     return {"total": int(total) + 1}
 '''
 
+SLOW_SOURCE = '''"""Hold the cascade open, so an engine kill finds work in flight."""
+
+import time
+
+
+def process(total, pause):
+    time.sleep(float(pause))
+    return {"done": total}
+'''
+
 
 def spec(name: str, dtype: str = "int", trigger: bool = True) -> dict[str, Any]:
     return {"name": name, "dtype": dtype, "trigger": trigger}
@@ -214,6 +225,30 @@ def chain_flow() -> dict[str, Any]:
     }
 
 
+def slow_flow(pause: float) -> dict[str, Any]:
+    """A sleep at the end of the load's cascade, for the engine scenario only.
+
+    It reads the total, so it runs after the count and never holds a node
+    worker the count is waiting for.
+    """
+    return {
+        "name": FLOW_SLOW,
+        "title": "Soak slow",
+        "version": 0,
+        "nodes": [
+            {
+                "id": "pause",
+                "type": "python",
+                "title": "Pause",
+                "position": {"x": 40, "y": 40},
+                "params": {"pause": pause},
+                "requires": [spec(f"{FLOW_LOAD}.total")],
+                "provides": [spec("done")],
+            }
+        ],
+    }
+
+
 def mqtt_flow() -> dict[str, Any]:
     """A subscriber on the dev stack's own broker, for the broker scenario."""
     return {
@@ -691,6 +726,12 @@ def scenario_engine(client: httpx.Client, args: argparse.Namespace, url: str) ->
     report = Report("engine")
     dead_before = dead_letters(client)
 
+    # The count alone finishes in a few milliseconds, so a kill used to catch
+    # next to nothing in flight. This pause keeps about three quarters of the
+    # engine's cascade slots busy at this rate, and the engine still keeps up.
+    pause = 0.75 * settings.FLOW_MAX_CASCADES / args.rps
+    create_flow(client, slow_flow(pause), {"pause": SLOW_SOURCE})
+
     driver = Driver(url, args.rps)
     log.info("engine: %d rps, killing %s halfway through", args.rps, API)
     driver.start()
@@ -700,6 +741,8 @@ def scenario_engine(client: httpx.Client, args: argparse.Namespace, url: str) ->
     # Read after the total, so work already accepted but not yet counted makes
     # the claim below easier rather than impossible.
     accepted_at_kill = driver.accepted
+    # Claimed and not acknowledged, so what the reaper has to hand back.
+    unacked = queue_stats(client).get("pending", "?")
 
     docker(API, "kill", API)
     time.sleep(2.0)
@@ -707,6 +750,7 @@ def scenario_engine(client: httpx.Client, args: argparse.Namespace, url: str) ->
     # it back — and it must never be left down.
     docker(API, "start", API)
     recovered = wait_for(lambda: health_code(client) == 200, HEALTH_RECOVER_S, 2.0)
+    report.stat(f"unacknowledged at the kill {unacked}, stack-wide")
     report.stat(f"engine answered healthy again {seconds(recovered)} after restart")
     report.check(
         f"health back to 200 within {HEALTH_RECOVER_S:.0f}s of the restart",