From 9a5d7a0d0cb0c5af2bdcd7bcbe2f0186ffc5137c Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 14 Oct 2026 23:49:07 +0200 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx --- backend/scripts/soak.py | 46 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/backend/scripts/soak.py b/backend/scripts/soak.py index a22a29d..813e2c3 100644 --- a/backend/scripts/soak.py +++ b/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",