fix: production-readiness fixes for the Kafka and NATS backends#51
Merged
Conversation
The shutdown branch of the concurrent receive loop fired the last offset commit with CommitMode::Async and dropped the result, so a clean deploy could redeliver the last drained batch if the process exited before the async commit completed. Commit synchronously and log (not fail) on error so shutdown still proceeds.
run_with_reconnect never reset its attempt counter or backoff after a successful stretch of connectivity, so max_reconnect_attempts acted as a lifetime budget instead of bounding consecutive failures: a consumer that reconnected occasionally over hours or days would eventually exhaust the budget and give up even though each individual outage was brief and isolated. Track how long the last invocation ran before failing and reset attempts/backoff once it exceeds a 60s threshold.
On a mid-batch submission error, publish_batch abandoned every ack future collected so far instead of awaiting them, always reporting succeeded=0 even though NATS had genuinely stored some messages — callers that re-drove the batch on that signal produced duplicates. The ack loop also stopped draining on its own first failure. Always drain every submitted ack; let a submission error take precedence in the reported error, but let an ack error fill it in only if nothing has failed yet.
…tted groups The per-partition lag arithmetic treated a never-committed group (Offset::Invalid or partition absent from the committed TPL) as committed-at-zero, reporting the full high watermark as lag. That is wrong for latest-reset groups (real lag is ~0, causing a spurious scale-to-max at startup) and wrong for earliest groups once retention truncates the log (lag is high - low, not high). Extract the arithmetic into a pure partition_lag(committed, low, high, reset) function and thread the group's auto.offset.reset policy through KafkaQueueStatsProvider::get_queue_stats (breaking change to the pub trait, pre-1.0). fetch_metrics resolves the policy from the group config, defaulting to Earliest exactly as the consumer does. The snapshot path addresses the default group and passes Earliest. Also re-export KafkaAutoOffsetReset from the shove::kafka facade: the type already appears in public signatures and the docs import it, but it was unreachable outside the crate, which would have made the new trait signature unimplementable externally.
KafkaConfig, KafkaTls, KafkaSasl, NatsConfig, TopologyBuilder, and TopologyDeclarer all have consuming -> Self builder methods, so a dropped result like `config.with_tls(tls);` silently shipped without TLS. #[must_use] on the type makes any unused function return of that type warn, covering all current and future builder methods. No existing call site tripped the lint.
KafkaConfig::with_ssl does not exist; the method is with_tls. The rebalance gotcha told users to tune session.timeout.ms and max.poll.interval.ms in KafkaConfig, but neither is settable — they are fixed constants (10s / 5min in kafka::constants). Point slow- handler advice at with_handler_timeout / with_default_handler_timeout instead.
The concurrent consumer used manual commits with a per-partition OffsetTracker but installed no rebalance callback and never reset tracker state when partitions moved. Two failures followed: 1. Commit stall: partition P revoked from consumer A, committed on by consumer B, then reassigned to A left A's stale next_to_commit waiting for a contiguous run B already consumed — P stopped committing for the life of the connection. 2. Stale-partition commits: completions for messages in flight at revocation were still committed even though the member no longer owned the partition. Install a RebalanceContext (ConsumerContext) that forwards incremental assign/revoke deltas from pre_rebalance to the receive loop over a std::sync::mpsc channel; the loop drops tracker entries for the listed partitions on BOTH revoke (stop committing unowned partitions) and assign (re-seed next_to_commit from the first delivered offset). Events are also applied right before track_received: the callback runs inside recv()'s poll, so an Assign may already be queued when the first post-reassignment message arrives from the same poll, and applying it after tracking would wipe the freshly seeded entry. Testing against a real broker exposed a second, subtler loss: async commits submitted between the revoke and assign phases of a cooperative rebalance can be dropped by librdkafka without any error — no commit_callback fires — and since drain_committable only yields on new completions, the lost commit was never retried (a partition with no follow-up traffic stayed uncommitted indefinitely). Countermeasures: - commit_callback is now implemented: a rejected commit marks its partitions dirty via the same channel and the tracker re-offers their current position on the next drain; - every drained rebalance event re-offers all retained partitions' positions (re-committing an already-committed offset is a broker-side no-op), closing the silent-drop window because the rebalance always ends with an assign round; - a 5s housekeeping tick wakes the loop so events and retries drain even when no messages or completions arrive. The FIFO and DLQ paths commit per message and keep no tracker; they receive the wrapper but deliberately drop the event receiver. MskIamContext's ClientContext overrides (ENABLE_REFRESH_OAUTH_TOKEN, generate_oauth_token) are forwarded through the wrapper, as are log/stats/stats_raw/error.
New tests/kafka_rebalance.rs: consumer A owns all 8 partitions and drains a batch; consumer B joins the group (cooperative rebalance moves partitions to B, which processes and commits on them); B leaves; a final batch must be both processed AND committed on every partition (polled via KafkaLagStatsProvider until lag reaches zero). On pre-fix code the test fails: partitions with commits in flight around the rebalance stall below the high watermark and never converge. On failure the test dumps per-partition low/high/committed state so a flake report shows exactly which partitions stalled and where.
A consumer task that exited with a non-retryable error (or exhausted max_reconnect_attempts) was only error-counted and logged; its JoinHandle stayed in the consumers vec forever. active_consumers() returned consumers.len(), so a group whose members had all died still reported full strength: the autoscaler computed capacity from a fiction and would not scale up to compensate, scale_up refused at "max capacity" with zero consumers running, and scale_down could "cancel" an already-dead consumer while live ones kept working. - active_consumers() now counts handles where !is_finished(). It counts on read (with &self) rather than relying on prior pruning, because the autoscaler's fetch_metrics reads it through immutable registry access. - scale_up/scale_down first call a new prune_finished() that removes finished handles from both the consumers and retiring lists, so the min/max gates and the idle-pick operate on live members only. - Pruned handles are polled once with a no-op waker to harvest their result: a panic (non-cancelled JoinError) is counted into panic_count, which drain_into already snapshots into ShutdownTally — preserving the panic-accounting invariant that previously relied on every handle surviving until the final drain. No double counting: a harvested handle is removed from the vec and never awaited again. With truthful counts the existing lag-driven autoscaler self-heals groups under load (capacity shrinks, lag exceeds the threshold, scale up spawns replacements). Respawn-to-min for idle groups remains a deliberately deferred supervision-policy decision.
…imeout Shove created JetStream pull consumers without setting ack_wait, leaving the 30s server default — exactly equal to the default handler timeout. Zero safety margin: a handler legitimately running near its limit, or a message waiting behind a saturated prefetch buffer before its handler even starts (the ack_wait clock ticks from delivery), exceeded ack_wait and was redelivered while still in flight — duplicate processing under normal slow-handler load. An operator raising with_handler_timeout(60s) silently made every such handler a guaranteed duplicate. Derive ack_wait = max(3 x effective handler timeout, 30s) via a single derive_ack_wait helper at every consumer-creation site: - the group durable (declare_pull_consumer gains an ack_wait parameter; the registry passes the group's resolved handler timeout, registry defaults already folded in); - the NotFound bootstrap fallback in the concurrent consumer path; - FIFO shard durables (get_or_create_consumer returns existing durables verbatim, so only newly created shards pick up the derived value — pre-existing ones keep 30s until recreated); - the DLQ durable (no handler-timeout knob on run_dlq; uses the default-timeout derivation). The hold-heartbeat pacing already reads the effective ack_wait from cached_info() at runtime and adapts automatically — unchanged. Integration coverage asserts the derived value on the group durable (explicit 20s -> 60s and default -> 90s), on a FIFO shard durable, and that re-registering against an existing durable (the deployment upgrade path) successfully updates its ack_wait via the upsert.
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
Deploying with
|
| Status | Name | Latest Commit | Preview URL | Updated (UTC) |
|---|---|---|---|---|
| ✅ Deployment successful! View logs |
shove-docs | 7cce999 | Commit Preview URL Branch Preview URL |
Jul 02 2026, 01:16 PM |
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.
Production-readiness fixes for the Kafka and NATS backends, from a focused audit of both. Every fix was verified against real brokers (testcontainers); each commit is one self-contained fix.
Kafka
86671e0,6523978): the consumer installed no rebalance callback, so a partition revoked and later reassigned kept a stalenext_to_commit— that partition stopped committing for the life of the connection (routine with the autoscaler; reproduced in the newkafka_rebalanceintegration test, which fails on the old code). ARebalanceContextnow forwards assign/revoke deltas to the receive loop, which drops affected tracker entries so reassigned partitions re-seed from actual delivery. While building the test, a second stall surfaced with hard evidence: librdkafka silently drops async commits submitted mid-rebalance (no error, nocommit_callback). Fixed via acommit_callbackfailure hook plus re-offering retained partitions' positions after every rebalance event (re-commits are broker-side no-ops), and a 5s housekeeping tick so retries drain on idle topics. Test passed 8/8 consecutive runs.4ad824f): the drain-then-commit on graceful shutdown usedCommitMode::Async+.ok()and immediately dropped the consumer — the just-drained batch could be redelivered on every deploy. NowSync, with the error logged.auto.offset.reset(a82ac25): a group that had never committed was assigned the full high-watermark as lag — forlatest-reset groups that meant a spurious scale-toward-max at startup, and forearliestgroups it over-counted once retention truncated the log. Lag now derives from the reset policy (earliest→ high−low,latest→ 0). Breaking:KafkaQueueStatsProvider::get_queue_statsgains a requiredreset: KafkaAutoOffsetResetparameter (also newly re-exported fromshove::kafka, which the docs already claimed).NATS
ack_waitmargin above the handler timeout (7cce999): pull consumers leftack_waitat JetStream's 30s default — exactly equal to the default handler timeout, so a handler running near its limit (or a message queued behind a full prefetch buffer) was redelivered while still in flight. All four consumer-creation sites now setack_wait = max(3 × handler timeout, 30s); existing group durables are upgraded on redeclare (asserted in a test). Rollout note: FIFO-shard and DLQ durables created before this change keep 30s until recreated (get_or_create_consumerreturns existing config verbatim).publish_batchpartial-failure accounting (3f74aa1): a mid-batch submission error abandoned every already-submitted ack and reportedsucceeded=0, so callers re-driving the batch produced duplicates. All submitted acks are now drained and counted; a fault-injection test asserts the reported count equals what JetStream actually stored.Both backends
6faadf5):max_reconnect_attemptsnever reset after a healthy run, so occasional blips over hours would eventually kill a healthy consumer. The budget and backoff now reset after 60s of healthy connection.b5ecbcc): dead consumer tasks stayed in the group vec forever —active_consumers()reported corpses as capacity, the autoscaler scaled against fiction, andscale_downcould "cancel" an already-dead member. Counts now filter onJoinHandle::is_finished(); scale paths prune finished handles, harvesting panics into the shutdown tally so no accounting is lost.#[must_use]on builder types (0cbf109):config.with_tls(tls);with a dropped result compiled silently and shipped without TLS.KafkaConfig/KafkaTls/KafkaSasl/NatsConfig/TopologyBuilder/TopologyDeclarerare now#[must_use](covers the new topic-config builder methods for free).3f79f34):KafkaConfig::with_ssl→with_tls(copy-paste code didn't compile); the tuning gotcha no longer tells users to set config keys that aren't settable.Verification
All on this branch, rebased onto
e902d7c:cargo fmt --check/clippy --all-features/clippy --no-default-features/clippy --features kafka-msk-iamall clean;nextestno-default 140/140, kafka lib 257/257, nats lib 235/235,kafka_integration49/49,kafka_rebalance1/1,nats_integration48/48.🤖 Generated with Claude Code