fix: recover from total-cluster outages longer than delivery.timeout.ms

[pirumu]

The bug (P1)

A producer whose entire cluster becomes unreachable (TCP broken — docker stop, not pause) for longer than delivery.timeout.ms wedges permanently: produced/acked freeze and the delivery futures for buffered records hang forever, even after the cluster fully recovers with ready batches still buffered.

Ruled out (all recover fine): idempotence on/off; outages ≤ delivery.timeout.ms; TCP-preserving outages of any length (docker pause).

Root cause

The background sender loop treated any error from its drive pass as fatal and parked the loop (SenderLoopWait::Parked). During a total outage the synchronous pre-dispatch step (metadata fetch) raises a wire Timeout; that parked the loop. A parked loop only wakes on an append or a dispatch completion — and once the producer's appends dry up (buffer full / bounded in-flight window) with no in-flight dispatch, nothing ever wakes it again.

Captured directly with a wedge trace:

wait=Err(Wire(Timeout)) inflight=0 buffered=1179648 acc_buffered=1179648 next_ready=true

→ loop parks with ready batches buffered, brokers already back, and never runs again.

The fix

• background_loop retries on a transient drive error (backoff via retry.backoff.ms) instead of parking forever — mirroring Kafka Sender.runOnce, which swallows a per-iteration failure and loops. The decision is extracted into resolve_background_loop_wait and unit-tested.
• drive_wake_until_waiting's per-iteration reap now uses the pump policy that swallows a per-batch Delivered(Err) (e.g. a post-outage DeliveryTimeout the dispatch task already surfaced on the futures) instead of propagating it and parking; it also no longer drops later Requeue results in the same reap batch. (Complementary correctness fix; the same class as fix: recover from broker outages instead of wedging the client #47 but on the loop's top-of-iteration reap.)

No per-record or per-iteration scan is added, so no expiry-sweep-style hot-path cost — the earlier hypothesis of needing a Java-style expiredBatches sweep was wrong; the dispatch task's own delivery-timeout checks are sufficient once the loop stops parking.

Validation

Deterministic soak repro (total 3-broker stop > delivery.timeout.ms under buffer pressure):

metric @ t=105	baseline (wedged)	fixed
produced	17251 (frozen)	105250 (climbing)
acked	15200 (frozen)	103104 (climbing)
delivery_errors	50 (frozen)	2146 (batches that genuinely expired)

103104 acked + 2146 expired = 105250 produced — every record resolved; production resumes at the target rate after recovery.

• make fmt-check, make clippy (-D warnings), make test (--workspace --all-features), make deny — all green
• New unit tests: background_loop_retries_on_transient_drive_error_instead_of_parking, background_loop_reap_swallows_per_batch_delivery_error
• producer_mock_bench: 5.31M msg/s (10 B), 708 MiB/s (10 KiB) — no regression to the shared dispatch path

Second commit adds env knobs to soak_bench (delivery/request timeout, buffer.memory, in-flight cap, idempotence toggle) so this class of outage is reproducible in ~2 min; all default to prior values.