8000 feat(task_queue_redis): extend Redis message claim during resumption replay by andrzejfudro · Pull Request #1007 · google/xmc · GitHub
[go: up one dir, main page]
Skip to content

feat(task_queue_redis): extend Redis message claim during resumption replay - #1007

Merged
andrzejfudro merged 1 commit into
mainfrom
feat/redis-extend-claim
Sep 30, 2026
Merged

feat(task_queue_redis): extend Redis message claim during resumption replay#1007
andrzejfudro merged 1 commit into
mainfrom
feat/redis-extend-claim

Conversation

@andrzejfudro
@andrzejfudro andrzejfudro commented Sep 28, 2026 •
edited
Loading
Copy link
Copy Markdown
Collaborator

Author note

This is coming from the design doc. The resumption process can take a long time in theory so to support that I extend redis client with option to extend the claim on the message.

Summary

Implements ExtendClaim for the Redis task queue consumer and calls it from the ESS resumption worker, so a long batch replay keeps its trigger message leased (b/559548459, Task 17).

Key Changes

  • redisclient.ExtendClaim(ctx)
    • Resets the message's PEL idle time with XCLAIM (MinIdle: 0).
    • Signals processMessage over a buffered channel carried in the task context, which resets the local ProcessTimeout timer.
    • Adds debug logs for both the extension and the timer reset.
  • Consumer interface / FakeClient
    • ExtendClaim is added to the interface.
    • The fake records the number of calls and can return an injected error.
  • redisclient.Client interface
    • baseManager now embeds a Client interface listing the 15 go-redis commands the package uses, instead of the concrete *redis.Client.
    • *redis.Client still satisfies it, so no caller changes; the Client field name is unchanged.
    • This lets consumer_test.go use an in-memory fake instead of a Redis server.
  • Resumption worker
    • New ClaimExtender dependency.
    • ExtendClaim is called after every replayed batch and before each backlog throttle pause (DD §3).
    • This replaces the two TODO(pausescheduling) markers.
  • Prod wiring
    • resumption.NewRedisConsumer builds a single Redis consumer that serves as both the trigger-stream consumer and the worker's ClaimExtender.
    • The wiring is moved into startResumptionConsumer so NewProd stays under the gocyclo limit.

Assumptions

  • Private claim context value. DD §2 names exported helpers (WithMsgID, WithWorkerID, WithExtendChan, …). This PR stores one private claim{msgID, workerID, extendChan} value in the context instead. Nothing outside redisclient needs those values, so the exported surface isn't needed.
  • No lost-claim detection. If another consumer took the message, XCLAIM still succeeds. We rely on ResumptionWorkerID fencing in Spanner to stop duplicate replay rather than checking the claim result.
  • A failed ExtendClaim is logged (Warn), not fatal, for the same fencing reason.

Verification

  • bazel test //cmd/experiment_state_server/... //internal/task_queue_redis/redisclient:redisclient_test: all pass.
  • New tests (in-memory fake, no Redis needed) cover:
    • XCLAIM sent with MinIdle: 0;
    • an extension preventing the local timeout;
    • Redis errors;
    • a missing claim in the context;
    • worker extension after each batch and before each throttle pause;
    • a failed extension not being fatal.
  • ./.githooks/pre-commit passes.

TAG=agy
CONV=e5e10511-3aa3-4e3d-9c89-4cf62fcb98f5

gemini-code-assist[bot]
gemini-code-assist Bot reviewed Sep 28, 2026
View reviewed changes
@gemini-code-assist gemini-code-assist Bot left a comment
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request implements message lease extension for the scheduling resumption stream by introducing a ClaimExtender interface and integrating it into the ResumptionWorker and Redis consumer client. Feedback on these changes highlights a high-severity risk of a type assertion panic and goroutine leak in the message processing loop, as well as a potential nil pointer dereference if the claimExtender is not initialized.

Comment thread internal/task_queue_redis/redisclient/consumer.go Resolved
Comment thread cmd/experiment_state_server/resumption/worker.go Resolved
@andrzejfudro
andrzejfudro force-pushed the feat/redis-extend-claim branch from 395a799 to 5d6e385 Compare September 29, 2026 10:03
@andrzejfudro
andrzejfudro changed the base branch from feat/ess-replay-backlog-throttling to main September 29, 2026 10:03
@andrzejfudro
feat(task_queue_redis): extend Redis message claim during resumption …
8012bfa 
…replay

Implement ExtendClaim on the Redis consumer. It resets the message's PEL
idle time via XCLAIM (MinIdle 0) and signals processMessage, through a
channel carried in the task context, to reset the local execution timer.
Long-running handlers are then neither reclaimed by Redis nor timed out
locally.

The ESS resumption worker now calls ExtendClaim after every replayed
batch and before each backlog throttle pause, replacing the related
TODOs. Failures are logged but not fatal, because ResumptionWorkerID
fencing already prevents duplicate replay if another worker takes over.

TAG=agy
CONV=e5e10511-3aa3-4e3d-9c89-4cf62fcb98f5
@andrzejfudro
andrzejfudro force-pushed the feat/redis-extend-claim branch from 5d6e385 to 8012bfa Compare September 29, 2026 12:56
@andrzejfudro
andrzejfudro marked this pull request as ready for review September 29, 2026 13:11
janzak-dev
janzak-dev approved these changes Sep 29, 2026
View reviewed changes
@janzak-dev janzak-dev left a comment •
edited
Loading
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please add a corresponding test case for the extend claim to validate on a real Redis whether it works properly https://github.com/google/xmc/blob/main/internal/task_queue_redis/validator/validator.go, can be done in a follow up PR

@andrzejfudro andrzejfudro mentioned this pull request Sep 30, 2026
Merged
@andrzejfudro
andrzejfudro added this pull request to the merge queue Sep 30, 2026
Merged via the queue into main with commit 5dd3741 Sep 30, 2026
12 checks passed
andrzejfudro added a commit that referenced this pull request Sep 30, 2026
@andrzejfudro
test(task_queue_redis): add ExtendClaim case to Redis validator
6d3b06e 
Adds TestExtendClaim to the real-Redis validator. The handler runs longer
than both ProcessTimeout and AbandonedMessageTimeout while calling
ExtendClaim more often than ProcessTimeout. The message must be processed
exactly once and ACKed: no timeout retry, no cleanup reclaim, empty DLQ.

Follow-up to #1007 requested in review.

TAG=agy
CONV=e5e10511-3aa3-4e3d-9c89-4cf62fcb98f5
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@janzak-dev janzak-dev janzak-dev approved these changes
+1 more reviewer
@gemini-code-assist gemini-code-assist[bot] gemini-code-assist[bot] left review comments
Reviewers whose approvals may not affect merge requirements

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@andrzejfudro @janzak-dev
0