bench: fail fast on lost produce; kafka: survive host stalls

.gitignore

@@ -85,6 +85,11 @@ mutants/
 /more_help.md
 /About_DE_project.md
 /audit_*.md
+# Guard the common typo/variant so a misnamed audit can't dodge the rule above
+# (e.g. autdit_*.md). Same class as security_preaudit_* slipping past /audit_*.
+/autdit_*.md
+# Root-level planning notes are session scratch, not product docs.
+/plan_*.md
 # Session handoff for the next agent: working notes, not product docs. Untracked
 # but unignored is one `git add -A` away from a public repo, so name it.
 /_NEXT_SESSION.md

docker-compose.yml

@@ -18,6 +18,11 @@ services:
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
       KAFKA_LOG_RETENTION_HOURS: 24
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
+      # Single-node KRaft heartbeats broker->controller over TCP inside one JVM;
+      # when the host stalls longer than the default 9s session the broker fences
+      # ITSELF and every partition goes leaderless. 45s absorbs the stalls a
+      # loaded dev host actually produces.
+      KAFKA_BROKER_SESSION_TIMEOUT_MS: "45000"
       CLUSTER_ID: "AgentFlowKafkaCluster01"
       KAFKA_JMX_PORT: 9101
     volumes:

docs/perf/throughput-realpath-paced100-4h-2026-07-18.md

@@ -0,0 +1,81 @@
+# S10 paced produce @ 100 eps — 4-hour soak attempt (FAIL, stand disk exhaustion)
+
+> Measured: `2026-07-18T00:23:20Z` → `05:03:25Z` on `deproject-mac`
+> (Colima vz 6 GiB / 4 CPU / 60 GiB docker volume). Code: `main` @ `05afb32`.
+> Path: Kafka → Flink stream_processor (**1 TM**) → `events.validated` →
+> host `bridge_consumer` → ClickHouse.
+> JSON: `/tmp/s10_soak4h.json` on the stand.
+
+## Result
+
+| Metric | Value |
+|--------|------:|
+| Produce | **100.0 eps** (1 440 000 events in 14 400.0 s, paced) |
+| Flink hop | **84.8 eps** (1 424 309 validated msgs, incl. replays) |
+| Bridge apply | **67.4 eps** (+1 132 164 unique applied over 16 800 s incl. catch-up timeout) |
+| Failures / duplicates | **0 / 292 145** |
+| Lag start → end | 0 → **0** (peak 1939) |
+
+**Gate:** produce ≥ 95 ✓, apply ≥ 90 ✗, flink ≥ 90 ✗, failures = 0 ✓,
+duplicates < 5 000 ✗, lag_end ≤ 100 ✓ → **FAIL**.
+
+## Root cause — stand infrastructure, not the apply path
+
+The colima docker volume (59 GiB) entered the run ~50 GiB full
+(37.85 GiB accumulated images ×55, ~8 GiB stale anonymous volumes,
+3.4 GiB unrelated stacks). The soak's own footprint (~8 GiB: Kafka topic
+data 1.5 GiB, CH 0.5 GiB, TM/JM container logs 3.8+1.7 GiB — json-file
+driver, no rotation) filled it to 99–100 % by ~02:58Z.
+
+Causal chain, all evidenced in JM/TM logs and `df`:
+
+1. ~02:58Z first S3 upload retries from MinIO (same volume) on checkpoint
+   `chk-315`; from 03:02Z checkpoint failures continuous.
+2. CheckpointFailureManager fails the job → restart from last successful
+   checkpoint → re-emission of already-validated events. 116 job restarts
+   total; **292 145 duplicate deliveries** into `events.validated`.
+3. 03:33:03Z the TM **container** dies (exit 127 on a full disk; no
+   restart policy) → job stuck in `NoResourceAvailableException` forever;
+   unique validated frozen at 1 132 164 (≈ events produced up to 03:33).
+4. Bench finishes produce (produce lane unaffected: exactly 100.0 eps for
+   4 h), waits out the 2 400 s catch-up timeout, reports FAIL.
+
+Steady state before the disk filled was healthy for ~2.5 h:
+at 01:30Z dup=0, lag ≤ 260; apply tracked produce ≈ 101 eps.
+
+## What the failure proved anyway
+
+- **Journal dedup held under a 292 k replay storm**: consumed = applied +
+  duplicates exactly (1 424 309 = 1 132 164 + 292 145), apply_failures = 0.
+  No event applied twice, no false failures.
+- Produce pacing and the bridge stayed stable for the full window; the
+  bridge drained its topic to lag 0 even while Flink crash-looped.
+
+Not proved: multi-hour sustained ≥100 eps through Flink. The claim table
+of the 1 h doc stands — **multi-hour remains open**.
+
+## Stand remediation required before re-run
+
+1. Free the docker volume: `docker image prune` (≈14.5 GiB reclaimable),
+   remove stale anonymous volumes (≈8.7 GiB), keep unrelated stacks intact.
+2. Cap container logs on the stand: colima `/etc/docker/daemon.json`
+   `{"log-driver":"json-file","log-opts":{"max-size":"50m","max-file":"3"}}`
+   (TM/JM wrote 5.5 GiB in one run; restart loops amplify this).
+3. Pre-flight gate: **≥ 12 GiB free** on `/mnt/lima-colima` before starting
+   a multi-hour arm (soak footprint ~8 GiB + margin).
+4. Optional hardening exposed by this run: TM `restart: unless-stopped` in
+   the compose flink overlay, so a container death degrades to a recoverable
+   restart instead of a permanent stall.
+
+## Reproduce
+
+As in `throughput-realpath-paced100-1h-2026-07-17.md`, with
+`--count 1440000 --pace-eps 100 --catchup-timeout-seconds 2400`, after the
+pre-flight disk gate above.
+
+## Status
+
+**4 h paced @ 100 eps: FAIL — stand disk exhaustion at ~2.5 h** (checkpoint
+storage + TM shared a 99 %-full docker volume). Apply-path integrity held
+(0 failures, exact dedup of 292 k replays). Multi-hour gate remains open;
+re-run after stand remediation.

docs/perf/throughput-realpath-paced100-4h-r3-2026-07-19.md

@@ -0,0 +1,82 @@
+# S10 paced produce @ 100 eps — 4-hour soak, run r3 (FAIL: harness silently lost produce)
+
+> Measured: `2026-07-19T01:26Z` → `06:07Z` on `deproject-mac`
+> (Colima vz 6 GiB / 4 CPU / 60 GiB docker volume, fresh VM restart, log-rotation
+> 50m×3 active). Code: `main` @ `05afb32`. Path: Kafka → Flink stream_processor
+> (**1 TM**) → `events.validated` → host `bridge_consumer` → ClickHouse.
+> JSON: `/tmp/s10_soak4h_r3.json`; logs `/tmp/r3_jm.log`, `/tmp/r3_tm.log` on the stand.
+
+## Result
+
+| Metric | Value |
+|--------|------:|
+| Produce (as counted by bench) | **100.0 eps** paced, 1 440 000 in 14 430 s |
+| Produce (actually delivered to Kafka) | **1 031 462** — 408 538 lost client-side |
+| Flink hop | 1 031 462 validated = **100 % of delivered** |
+| Bridge apply | +1 031 462 unique = **100 % of validated** |
+| Failures / duplicates | **0 / 0** |
+| Lag start → end | 0 → **0** (peak 2015) |
+| Flink restarts / checkpoints | **0 restarts**, 578/578 checkpoints COMPLETED |
+
+**Gate:** produce ✓ / flink 61.3 ✗ / apply 61.3 ✗ → formally **FAIL**, but the
+shortfall is entirely upstream of the product: the pipeline processed every
+event that reached Kafka, exactly once, with zero failures over 4.7 h.
+**Multi-hour sustained ≥100 eps remains open** — the path only saw real load
+for 2 h 52 m.
+
+## Root cause — two stacked defects, neither in the serving path
+
+**1. Stand: single-node KRaft broker fenced itself for 16 seconds.**
+Broker and controller share one JVM yet heartbeat over TCP. From 04:15:29Z
+heartbeats timed out (4 500 ms limit) under host load; at **04:18:13Z** the
+controller fenced the broker (`Fencing broker 1 because its session has timed
+out`), all **84 partitions went leaderless**; at **04:18:29Z** it unfenced and
+leaders were restored. Total outage: **16 s**. Identical blips recurred at
+06:04 and 06:13 — this is chronic on a 4-core VM, not a one-off.
+
+**2. Harness: fire-and-forget producer wedged permanently and lied about it.**
+The bench produced with no `on_delivery` callback and an unchecked `flush(30)`.
+confluent-kafka 2.15.0 (librdkafka 2.15.0) received the leaderless metadata
+during the 16 s window (`new leader-1 not found in cache`, 04:18:56Z) and
+**never re-resolved the leaders** — although a fresh producer on the same host
+port produced fine (verified live post-run, delivery err=None). From 04:18:56Z
+to the end every queued message silently expired after `message.timeout.ms`
+(300 s): steady-state queue depth ≈ 300 s × 100 eps = 30 000 stayed below the
+100 k buffer, so `produce()` never raised and the pace loop counted phantom
+events to 1 440 000. At termination librdkafka reported
+`26 983 messages still in queue or transit` — the un-expired tail
+(≈ 270 s × 100 eps), unchecked. Net client-side loss: **408 538 events**.
+
+Evidence chain: broker log (fence 04:18:13 → unfence 04:18:29); bench log
+(rdkafka METADATAUPDATE burst 04:18:56, then silence until TERMINATE);
+`orders.raw` end-offsets sum **exactly 1 031 462** with Flink group lag 0 on
+all six partitions; `events.validated` end-offsets sum exactly the same.
+
+## What r3 proved anyway
+
+- The full streaming path ran **4.7 h with zero loss, zero duplicates, zero
+  failures** on everything actually delivered (consumed = applied exactly).
+- Flink rode out multiple broker fencing blips with **no job restarts** and a
+  perfect checkpoint record (578/578, ~30 s cadence, sub-second latency).
+- Fresh-VM + log-rotation remediation from r1 held: disk peaked at 73 %.
+
+## Remediation (implemented in the same change that adds this report)
+
+1. **Bench fails fast and loud**: every `produce()` carries an `on_delivery`
+   hook; the paced loop aborts (exit 3) on the first surfaced delivery failure;
+   the final `flush(120)` must drain clean or the run aborts. A 16 s broker
+   blip that librdkafka rides out (retries within 300 s) does NOT abort; only
+   genuinely lost events do.
+2. **Broker session timeout 9 s → 45 s** (`KAFKA_BROKER_SESSION_TIMEOUT_MS`,
+   base compose): self-fencing now requires a 45 s host stall.
+
+## Run-r2 (context, still undiagnosed)
+
+r2's TM death (exit=1, 10 min in) remains unexplained — its logs were lost.
+r3 adds evidence that this stand chronically degrades under host contention
+(repeated broker self-fencing), which is consistent with a stand-level cause
+for r2 but does not prove it.
+
+## Next
+
+r4 with guarded bench + hardened broker on a fresh stack (`down -v`).

scripts/benchmark_throughput_realpath.py

@@ -418,6 +418,19 @@ def main() -> int:
     lag_peak = baseline["lag"]
     pending_latency: dict[str, tuple[float, str]] = {}  # event_id -> (t0, order_id)
 
+    # Delivery errors surface ONLY via callback: without one librdkafka expires
+    # undeliverable messages silently (message.timeout.ms) and `produced` counts
+    # events that never reached the broker — the gate math is then meaningless.
+    delivery_failures = 0
+    first_delivery_error: str | None = None
+
+    def _on_delivery(err: object, _msg: object) -> None:
+        nonlocal delivery_failures, first_delivery_error
+        if err is not None:
+            delivery_failures += 1
+            if first_delivery_error is None:
+                first_delivery_error = str(err)
+
     t_produce0 = time.perf_counter()
     next_pace = t_produce0
     for i in range(args.count):
@@ -428,10 +441,18 @@ def main() -> int:
         if want_lat:
             pending_latency[event["event_id"]] = (time.perf_counter(), event["order_id"])
 
-        producer.produce(args.source_topic, value=payload)
+        producer.produce(args.source_topic, value=payload, on_delivery=_on_delivery)
         produced += 1
         if produced % 50 == 0:
             producer.poll(0)
+            if delivery_failures:
+                print(
+                    f"FATAL: {delivery_failures} events failed delivery after retries "
+                    f"(first: {first_delivery_error}); aborting — a run with lost "
+                    "produce cannot judge the gate",
+                    file=sys.stderr,
+                )
+                return 3
 
         if args.pace_eps > 0:
             next_pace += 1.0 / args.pace_eps
@@ -458,7 +479,14 @@ def main() -> int:
             except Exception as exc:  # noqa: BLE001
                 print(f"  produced={produced} (metrics warn: {exc})", flush=True)
 
-    producer.flush(30)
+    unflushed = producer.flush(120)
+    if unflushed or delivery_failures:
+        print(
+            f"FATAL: produce lane lost events (delivery_failures={delivery_failures}, "
+            f"still_in_queue={unflushed}, first: {first_delivery_error}); aborting",
+            file=sys.stderr,
+        )
+        return 3
     t_produce1 = time.perf_counter()
     produce_wall = t_produce1 - t_produce0