Skip to content

feat(stovepipe): reschedule process when concurrency gate is closed#338

Open
mnoah1 wants to merge 2 commits into
mainfrom
mnoah1/stovepipe-process-gate-defer
Open

feat(stovepipe): reschedule process when concurrency gate is closed#338
mnoah1 wants to merge 2 commits into
mainfrom
mnoah1/stovepipe-process-gate-defer

Conversation

@mnoah1

@mnoah1 mnoah1 commented Jul 13, 2026

Copy link
Copy Markdown
Contributor

What?

When the gate is closed (on first check or after losing a slot-claim race), the controller now re-enqueues the same ProcessRequest to the process topic via PublishAfter(GateWaitDelayMs) and acks the current delivery.

  • The rescheduled message gets a fresh id ({request_id}/reschedule/{unix_ms}) rather than reusing the in-flight delivery's id: the message store dedupes on (topic, partition_key, id) and the original row is still present until the ack, so a same-id republish would be silently swallowed and then deleted by the ack — permanently stranding the queue head.
  • GateWaitDelayMs must be configured positive; a zero/negative value returns an error (with a config_errors counter) rather than silently publishing with no delay. Publish failures emit publish_errors and surface the raw error so the consumer's classifier decides retryability. The controller now takes the TopicRegistry for publishing, wired through in the stovepipe server main.

Why?

Implementing proposed PublishAfter behavior from RFC: https://github.com/roychying/submitqueue/blob/chenghan.ying/stovepipe-build/buildsignal-doc/doc/rfc/stovepipe/steps/process.md#option-2-ack-and-publishafter

This allows us to reschedule the delivery for later processing, rather than having publish pause the goroutine to wait for the gate to open.

Test Plan

  • make local-stovepipe-start
  • First ingest: grpcurl -plaintext -d '{"queue":"monorepo/main"}' localhost:32784 uber.submitqueue.stovepipe.Stovepipe/Ingest
  • Manually clear from DB, but leave the build slot claimed: DELETE FROM request_uri WHERE queue='monorepo/main';
  • Ingest again: grpcurl -plaintext -d '{"queue":"monorepo/main"}' localhost:32784 uber.submitqueue.stovepipe.Stovepipe/Ingest
  • Log: "rescheduled latest head awaiting build slot" request_id=request/monorepo/main/2 in_flight_count=1 delay_ms=5000 - appears every 5 seconds

@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-admit branch from 2dc04df to 75abca6 Compare July 14, 2026 14:14
@behinddwalls

Copy link
Copy Markdown
Collaborator

⚠️ Automatic stack rebase failed

This PR could not be automatically rebased after its base PR was merged. The rebase hit conflicts that need manual resolution.

To fix manually:

git fetch origin
git checkout mnoah1/stovepipe-process-gate-defer
git rebase --onto origin/main 75abca64a764251e8cd57316ebd6731ca24390b1 mnoah1/stovepipe-process-gate-defer
# resolve conflicts, then:
git push --force-with-lease

Then update this PR's base branch:

gh pr edit 338 --base main

Base automatically changed from mnoah1/stovepipe-process-admit to main July 14, 2026 15:25
@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-gate-defer branch from aa2ccae to e738ad3 Compare July 14, 2026 15:56
Ack the current delivery and PublishAfter the same ProcessRequest when
the latest head cannot claim a build slot, so gate waits do not burn
MaxAttempts or block the partition.
@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-gate-defer branch from e738ad3 to 5b651c5 Compare July 14, 2026 16:07
@mnoah1 mnoah1 changed the title [DRAFT/WIP] feat(stovepipe): reschedule process when concurrency gate is closed feat(stovepipe): reschedule process when concurrency gate is closed Jul 14, 2026

// Suffix the message id with the publish time so the reschedule can't collide with
// the in-flight delivery's still-present message-store row.
msgID := fmt.Sprintf("%s/reschedule/%d", request.ID, time.Now().UnixMilli())

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Alternative considered: add a value on the payload that keeps track of reschedules. I think timestamp provides sufficient uniqueness here for now but we could consider that option as well.

@mnoah1 mnoah1 marked this pull request as ready for review July 14, 2026 18:26
@mnoah1 mnoah1 requested review from a team, behinddwalls and sbalabanov as code owners July 14, 2026 18:26
@mnoah1 mnoah1 enabled auto-merge July 14, 2026 18:26
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants