fix: recover from broker outages instead of wedging the client#47
Merged
Conversation
Four fixes for the client-wedge class the 2026-07-07 soak run exposed (a prolonged broker outage wedged producer and consumers permanently, surviving even full cluster restoration): - wire: bound connect_and_negotiate with request_timeout. A freshly restarted (still fenced) KRaft broker accepts TCP but answers nothing; the unbounded ApiVersions read parked the per-broker task forever — no reconnect, no expire_pending_commands, queued requests unfailable, and the cached handle is never replaced. An elapse now retries like any setup error, so the loop keeps cycling and recovery after the broker unfences is automatic. (Root cause of both the producer wedge and the established-but-silent sockets.) - wire: tie the per-connection reader task to serve_connection's scope (AbortOnDrop). The detached reader held the socket's read half — the generic io::split keeps the fd alive — so every teardown leaked an ESTABLISHED connection until the broker idle-closed it (~10 min); observed as 26 broker connections where ~11 belong. Also abort in_flight_fetch in Consumer::drop (close() already did) so a dropped consumer cannot detach a fetch holding a WireClient clone. - consumer: wire-level timeouts/connection failures now clear the cached coordinator (is_coordinator_moved). A dead coordinator can only time out — it never sends NOT_COORDINATOR — so the Kafka-code- only check pinned a dead incarnation forever. Java marks the coordinator unknown on any coordinator request failure. - consumer: JoinGroup/SyncGroup are bounded by rebalance timeout + 5s (Java parity) instead of request.timeout.ms, via a new per-request timeout threaded through the wire (RequestCommand.timeout, pipeline reserve_with_timeout, expire_pending_commands). The coordinator legitimately holds these responses up to the rebalance window (session.timeout default 45s > request.timeout 30s), so every member timed out mid-join and its retry restarted the round — a group-wide livelock matching the soak's ~45s Wire(Timeout) cycle, wedging fresh consumers too. New tests: silent-listener negotiate bound (requests fail Timeout, no hang), per-command expiry override, pipeline per-request deadline, coordinator-cache clearing on wire errors. 632 lib tests, clippy workspace all-targets clean, rustdoc clean.
The second half of the producer outage wedge: when every leader is unroutable (metadata refresh fails — e.g. all brokers down), dispatch returns the batches to the accumulator and the sender loop parked until the NEXT APPEND woke it. Any caller applying backpressure — flush(), close(), a bounded in-flight window — never appends again, so the requeued batches were never re-drained, the drain-time delivery-timeout check never ran, and their delivery futures hung forever, surviving even full cluster restoration (nothing re-attempted metadata without traffic). The pause is now a timed retry at retry.backoff.ms: each pass re-attempts metadata + dispatch, expiring batches past delivery.timeout.ms, so a sustained outage fails loudly and a restored cluster resumes without new traffic. The hot path is untouched — the branch only runs after a requeue. In-VM E2E (pause all brokers 2.5min then unpause, producer-only soak, 300 rec/s x 300s): 10,275 aged batches failed with DeliveryTimeout in one sweep at restore, produce resumed at full rate immediately, and the run balanced exactly — produced 90,000 = acked 79,725 + errored 10,275, zero hung futures. The same scenario on master wedges permanently with 25/20,000 futures resolved.
The final layer of the outage wedge. When an outage outlasts the in-flight retries, a dispatch completes with Delivered(Err(Delivery Timeout)) — its records already failed via fail_deliveries inside the task. handle_completed_dispatch propagated that per-batch error up as the drive's own Result::Err, which the background sender loop treated as a park condition. But the offending batch was already gone (delivered), so the loop parked and every OTHER partition's buffered records starved — a permanent producer hang after a long total-cluster outage even once brokers returned (traced: drive_wake_until_waiting looping on Err(DeliveryTimeout partition N), buffered ~10MB never draining). drive_ready_dispatch_until_blocked gains a policy: the background pump (drive_wake_step / drive_wake_until_waiting) passes surface_delivered_errors=false and reaps completions leniently — a Delivered(Err) is swallowed (records already on their futures) so the pump keeps draining; structural failures (requeue accounting, join panic) still propagate. Flush and append-capacity keep surface_delivered_errors=true so their awaiting callers still see the failure (unchanged; existing tests green). Validated end to end: a 50s total-cluster outage (docker stop all three brokers, restart) now recovers fully with zero lost records (produce resumes the instant brokers return); before this the producer stayed frozen. Two unit tests pin both policies.
Three green-gate fixes on this branch: - pipeline reserve() is now test-only (the wire write path always uses reserve_with_timeout), so gate it #[cfg(test)] to satisfy dead_code under -D warnings. - Sender requeue-retry deadline uses checked_add instead of Instant + Duration, satisfying arithmetic_side_effects. - Bump crossbeam-epoch 0.9.18 -> 0.9.20 for RUSTSEC-2026-0204 (this branch predates the same lockfile bump on master), fixing cargo deny. make clippy / test / deny / doc-check all green.
pirumu
added a commit
that referenced
this pull request
Jul 7, 2026
The overnight soak report's F3 finding guessed "delivery.timeout.ms not enforced" — wrong. Add a Resolution section: F3 fixed + verified (PR #48 core, #47 partial) with the traced root cause (background loop parked on a transient metadata Wire(Timeout) and never woke once appends dried up); F1/F4 fixed in code by #47 pending a fresh compound soak; F2 still open. Preserves the historical run data; only appends the post-run resolution.
This was referenced Jul 7, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Six fixes for the client-wedge class the 2026-07-07 soak run exposed (see
SOAK-REPORT.mdfindings F1–F4). A prolonged broker outage could wedge the producer and consumer group permanently, surviving even full cluster restoration. Each fix is unit-tested; the set is validated end-to-end against the 3-broker compose withdocker stop/pauseoutage injection.Fixes
Wire
connect_and_negotiateunderrequest_timeout). A freshly restarted, still-fenced KRaft broker accepts TCP but answers nothing; the unbounded ApiVersions read parked the per-broker task forever (no reconnect, queued requests unfailable, handle never replaced). Now an elapse retries like any setup error.AbortOnDropon the per-connection reader;Consumer::dropaborts the in-flight fetch). The detached reader held the socket's read half open until the broker idle-closed it (~10 min) — the F4 connection leak across consumer close/recreate.Consumer
Wire(Timeout)/ConnectionClosed/Ionow clear the cached coordinator. A dead coordinator can only time out — it never sendsNOT_COORDINATOR— so the Kafka-code-only check pinned a dead incarnation forever (F1).Wire(Timeout)cadence, wedging fresh consumers too (F1).Producer
retry.backoff.ms, so batches expire pastdelivery.timeout.msand a restored cluster resumes without new traffic (F3).Delivered(Err(DeliveryTimeout))(records already failed viafail_deliveries) was propagated up as the drive's own error, parking the loop and starving every other partition. The background pump now reaps leniently; flush/append-capacity still surface the error to their callers (F3).Validation
--all-features --all-targetsclean; rustdoc clean.docker pause2.5 min) recovers fully — 90,000 produced = 79,725 acked + 10,275 expired, zero hung futures. A TCP-breaking outage up todelivery.timeout.ms(docker stop/startall three, 50s) recovers fully with zero lost records. Before these fixes both scenarios wedged permanently.Known remaining limitation (follow-up)
A TCP-breaking total-cluster outage LONGER than
delivery.timeout.ms(default 2 min) still wedges: records drained into in-flight dispatch retries end up in a limbo that neither the retry-loop nor an accumulator sweep expires, so they hang instead of failing cleanly at the deadline. This is the deferred P1 (Java'sRecordAccumulator.expiredBatchesindependent sweep) — it touches the perf-critical accumulator/sender hot path and warrants its own change with a perf A/B, so it is intentionally not in this PR. Precise root-cause notes are inSOAK-REPORT.md.