Repository navigation
feat(resumption): throttle paused-task replay when tasks queue is busy - #986
Merged
Merged
Conversation
Before fetching each batch, the ESS resumption worker checks the backlog on xmc-tasks-stream (pending + lag via XINFO GROUPS and a capped XRANGE). If it reaches max_backlog_threshold, the worker waits 10s and returns to the top of the loop, so pause state and fencing are re-checked first. Backlog calculation lives in the task_queue client so ESS only depends on a small BacklogChecker interface. ExtendClaim is left as a TODO until the lease-extension API lands. TAG=agy CONV=089ef675-f585-4067-9e62-27ef0bad0823
There was a problem hiding this comment.
Code Review
This pull request introduces downstream task queue backlog checking and throttling to the ResumptionWorker. It adds a BacklogChecker interface and implements Redis stream backlog calculation (pending + lag) via CalculateStreamBacklog to pause task replay when the backlog exceeds a specified threshold. Comprehensive unit tests and Bazel build updates are also included. The review feedback suggests two key improvements: replacing time.After with time.NewTimer in the throttling loop to prevent potential short-term memory leaks, and optimizing the backlog calculation to skip or limit XRANGE queries when the pending message count already meets or exceeds the threshold.
- Use time.NewTimer with explicit Stop for the throttle wait and extract it into a reusable waitOrCancel method. - Skip XRANGE when pending alone meets the threshold and otherwise read only threshold - pending + 1 entries. TAG=agy CONV=089ef675-f585-4067-9e62-27ef0bad0823
…ty backlog fallbacks - Merge PayloadPublisher and BacklogChecker into a single TaskQueue interface so the task queue client is passed once. - Warn when backlog inspection silently falls back to an empty backlog (missing/unsupported Redis client, missing consumer group). - Add tests for backlog == threshold and the default threshold fallback. TAG=agy CONV=089ef675-f585-4067-9e62-27ef0bad0823
- Merge throttle decision cases into a single table test. - Share worker setup across stop-while-throttled tests. - Drop waitOrCancel tests already covered by the throttling tests. TAG=agy CONV=089ef675-f585-4067-9e62-27ef0bad0823
andrzejfudro
marked this pull request as ready for review
September 28, 2026 12:25
andrzejfudro
enabled auto-merge
September 28, 2026 12:42
janzak-dev
reviewed
Sep 28, 2026
Move the stream backlog logic into redisclient as PublisherClient.BacklogLength, using the publisher's stream and a caller-provided consumer group. task_queue.Client.CalculateBacklogLength now takes the group and delegates. Address review feedback: - Rename CalculateBacklog to CalculateBacklogLength and document the capped count. - Return 0 when the stream disappears before XRANGE (pending entries go with it). - Log pause_duration when throttling replay. TAG=agy CONV=089ef675-f585-4067-9e62-27ef0bad0823
…sher TAG=agy CONV=089ef675-f585-4067-9e62-27ef0bad0823
Collaborator
Author
|
@janzak-dev While addressing the comments I noticed that the backlog.go was put in the wrong place (in the taskqueue instead of redisclient). That's why more things were changed since your last review, all of it is intentional. |
janzak-dev
approved these changes
Sep 29, 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
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
The ESS resumption worker now pauses replay while
xmc-tasks-streamis busy, so replaying paused tasks can't push Redis toward its memory limit.Key Changes
Client.CalculateBacklogcomputespending + lagusingXINFO GROUPSandXRANGE. XRANGE is skipped whenpendingalone reaches the threshold, and otherwise reads onlythreshold - pending + 1entries, so the result is capped just above the threshold. A missing stream counts as zero backlog. Cases that silently fall back to an empty backlog (missing or unsupported Redis client, missing consumer group) log a warning. A compile-time check makes sure the production publisher supports the needed Redis commands.max_backlog_thresholdfrom the trigger message (default 500). If the backlog is at or above it, the worker waits 10s (cancellable) and goes back to the top of the loop, so pause state and fencing are re-checked before anything is replayed. The worker takes a singleTaskQueuedependency (enqueue + backlog), whichtask_queue.Clientsatisfies.Out of scope
xmc_resumption_throttles_totaletc.) come in a separate PR.Verification
bazel test //internal/task_queue:task_queue_test //cmd/experiment_state_server/resumption:resumption_test(Passed)bazel build //cmd/experiment_state_server/...(Passed)TAG=agy
CONV=089ef675-f585-4067-9e62-27ef0bad0823