[coordinator] streaming pull: bound lease renewal to one batch per tick - #69
Merged
Conversation
handle_renew_leases drained every renewable message in an unbounded loop on each lease-renewal tick. Inside the manager's biased select! loop this lets a modack burst monopolize the loop and starve the lowest-priority streaming-pull receive branch. The starvation is self-reinforcing: fewer messages received -> in-flight/on-hold messages age past their lease -> more need renewal -> more modack -> even less receive, a spiral that pins per-stream throughput regardless of pod count and only unwinds when upstream publish volume drops. Renew at most one batch (ACK_IDS_MAX_BATCH_SIZE) per tick and yield back to the loop. collect_ack_ids_that_need_to_be_renewed re-schedules each collected message's next renewal forward, so successive ticks cycle through the rest of the expired set; the per-stream working set is bounded by flow control, so one batch per LEASE_RENEWAL_INTERVAL keeps leases alive.
…arvation Adversarial tests for the property the one-batch-per-tick fix relies on: collect_ack_ids_that_need_to_be_renewed pops expired items and reschedules each forward, so renewing one batch per LEASE_RENEWAL_INTERVAL covers the whole working set across ticks without starving the tail or double-renewing. Covers: single-tick cap == max_chunk_size; rescheduled items aren't recollected at the same instant; 5001 msgs / 2500-batch drain in exactly 3 ticks, each renewed once (no starvation, no duplication); renewal recurs after the lease window; acked messages are never renewed; empty queue renews nothing. Verified the suite bites: mutating the reschedule to reinsert at `now` fails 4 of the 6 tests. Adds tokio test-util dev-dependency for paused-time control.
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.
Overview
The coordinator's rules-sink consumer could wedge into a multi-million-message backlog that neither more pods nor restarts would drain — only falling traffic did. Root cause is a self-reinforcing lease-renewal vs. receive starvation spiral. This change bounds lease renewal so a renewal burst can't starve message reception.
The bug
handle_renew_leasesdrained every renewable message in an unbounded loop each tick. In thebiasedselect!, renewal outranks the streaming-pull receive branch, so a modack burst monopolizes the loop and starves receive → messages age past their lease → more renewals → even less receive. It spirals, pins per-stream throughput regardless of pod count, and is restart-immune.streaming_pull_response_counthalved, with no reconnects and no pod change.Fix
ACK_IDS_MAX_BATCH_SIZE) per tick, then yield. Safe becausecollect_ack_ids_that_need_to_be_renewedpops and reschedules each item forward, so successive ticks cover the rest; the working set is flow-control-bounded. Renewal logic is otherwise unchanged.Validation
cargo check+rustfmtclean.nowfails 4 of 6).handle_renew_leasesitself (would need a pubsub-client mock); its only change is the loop removal, exercised live e2e.Follow-ups (not in this PR)
PUBSUB_MIN_LEASE_EXTENSION_SECS30 → 120 to cut renewal frequency.select!path entirely.