Skip to content

feat(stovepipe): admit latest head to build#335

Open
mnoah1 wants to merge 4 commits into
mnoah1/stovepipe-process-queueconfigfrom
mnoah1/stovepipe-process-admit
Open

feat(stovepipe): admit latest head to build#335
mnoah1 wants to merge 4 commits into
mnoah1/stovepipe-process-queueconfigfrom
mnoah1/stovepipe-process-admit

Conversation

@mnoah1

@mnoah1 mnoah1 commented Jul 10, 2026

Copy link
Copy Markdown
Contributor

What?

This PR adds in handling to admit a request to a build:

  • admitRequestToBuild handles requests that are ready to be have a build enqueued: when a build slot is available and a request matches the queue's current latest request, the request will now hit this branch
  • admitRequestToBuild performs two actions: claim a build slot by incrementing the queue's in-flight builds, and transition the request to processing

This leaves 3 main codepaths to fill out the rest of the initial implementation for the process step (marked with TODO):

  • Strategy selection logic ("full" placeholder currently)
  • Re-enqueue requests when no build slot is available
  • Publish the actual build step

Why?

Continuing to implement the process step outlined in process.md.

Test Plan

  1. make local-stovepipe-start
  2. grpcurl -plaintext -d '{"queue":"monorepo/main"}' localhost:32773 uber.submitqueue.stovepipe.Stovepipe/Ingest

admitted request to build line below:

stovepipe-service-1  | 2026-07-13T15:41:15.769Z	DEBUG	queue_mysql.message_store	mysql/message_store.go:217	fetched messages	{"topic": "process", "partition_key": "monorepo/main", "count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.773Z	DEBUG	consumer/consumer.go:351	processing delivery	{"controller": "process", "topic_key": "process", "message_id": "request/monorepo/main/1", "partition_key": "monorepo/main", "attempt": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.781Z	INFO	process_controller	process/process.go:189	admitted request to build	{"request_id": "request/monorepo/main/1", "queue": "monorepo/main", "uri": "git://monorepo/main/HEAD", "build_strategy": "full"}
stovepipe-service-1  | 2026-07-13T15:41:15.784Z	DEBUG	consumer/consumer.go:479	message processed successfully	{"controller": "process", "topic_key": "process", "message_id": "request/monorepo/main/1", "partition_key": "monorepo/main", "attempt": 1, "elapsed_ms": 7}
stovepipe-service-1  | 2026-07-13T15:41:15.856Z	DEBUG	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:191	retrieved leased partitions	{"topic": "process", "count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.857Z	DEBUG	queue_mysql.subscriber_heartbeat_store	mysql/subscriber_heartbeat_store.go:95	found active subscribers	{"topic": "process", "count": 0, "subscribers": []}
stovepipe-service-1  | 2026-07-13T15:41:15.857Z	DEBUG	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:232	discovered partitions	{"topic": "process", "count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.858Z	DEBUG	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:83	acquired lease	{"topic": "process", "partition_key": "monorepo/main"}
stovepipe-service-1  | 2026-07-13T15:41:15.858Z	INFO	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:294	completed partition discovery and acquisition	{"topic": "process", "discovered_count": 1, "acquired_count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.858Z	DEBUG	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:191	retrieved leased partitions	{"topic": "process", "count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.864Z	DEBUG	queue_mysql.message_store	mysql/message_store.go:217	fetched messages	{"topic": "process", "partition_key": "monorepo/main", "count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.957Z	DEBUG	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:191	retrieved leased partitions	{"topic": "process", "count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.957Z	DEBUG	queue_mysql.subscriber_heartbeat_store	mysql/subscriber_heartbeat_store.go:95	found active subscribers	{"topic": "process", "count": 0, "subscribers": []}
stovepipe-service-1  | 2026-07-13T15:41:15.958Z	DEBUG	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:232	discovered partitions	{"topic": "process", "count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.959Z	DEBUG	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:83	acquired lease	{"topic": "process", "partition_key": "monorepo/main"}
stovepipe-service-1  | 2026-07-13T15:41:15.959Z	INFO	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:294	completed partition discovery and acquisition	{"topic": "process", "discovered_count": 1, "acquired_count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.959Z	DEBUG	queue_mysql.partition_lease_store	mysql/partition_lease_store.go:191	retrieved leased partitions	{"topic": "process", "count": 1}
stovepipe-service-1  | 2026-07-13T15:41:15.964Z	DEBUG	queue_mysql.message_store	mysql/message_store.go:217	fetched messages	{"topic": "process", "partition_key": "monorepo/main", "count": 0}

@mnoah1 mnoah1 changed the title feat(stovepipe): admit latest head and publish to build [draft/wip] feat(stovepipe): admit latest head and publish to build Jul 10, 2026
@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-admit branch from a384a94 to 86eaa35 Compare July 10, 2026 17:24
@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-queueconfig branch 2 times, most recently from c814927 to acdd570 Compare July 10, 2026 18:52
@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-admit branch from 86eaa35 to 9008831 Compare July 10, 2026 19:47
mnoah1 added 3 commits July 10, 2026 20:07
Compare accepted requests against queue.latest_request_id and mark
older heads superseded. Latest heads remain accepted until admit lands
in a follow-up PR.
Update process.md to describe coalescing via CompareRequestID and
queue.latest_request_id instead of the dropped Sequence field and
latest_request_seq pointer. Also align BuildStrategy naming with code.
Introduce queueconfig.Store and wire process to resolve per-queue
max_concurrent at gate-check time. The default store returns global wiring
constants; admit lands in a follow-up PR.
@behinddwalls behinddwalls force-pushed the mnoah1/stovepipe-process-queueconfig branch from acdd570 to 98cef3f Compare July 10, 2026 20:07
@behinddwalls behinddwalls force-pushed the mnoah1/stovepipe-process-admit branch from 9008831 to 44606ba Compare July 10, 2026 20:07
@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-admit branch from d3049fb to 99434b7 Compare July 13, 2026 15:13
@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-admit branch from 99434b7 to 2dc04df Compare July 13, 2026 15:27
@mnoah1 mnoah1 changed the title [draft/wip] feat(stovepipe): admit latest head and publish to build feat(stovepipe): admit latest head and publish to build Jul 13, 2026
@mnoah1 mnoah1 marked this pull request as ready for review July 13, 2026 15:45
@mnoah1 mnoah1 requested review from a team, behinddwalls and sbalabanov as code owners July 13, 2026 15:45
@mnoah1 mnoah1 changed the title feat(stovepipe): admit latest head and publish to build feat(stovepipe): admit latest head to build Jul 13, 2026

@behinddwalls behinddwalls left a comment

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.

The happy path is correct and the version-arithmetic convention (assign after CAS success) is followed throughout — and nice job picking up the #324 feedback (empty-pointer log message, the TODOs). But the failure paths of the admit workflow have one real defect (the slot leak, first inline comment) and two shape issues that I think should be addressed before this merges.

The three main findings share one fix shape: claim slot → re-check staleness against the reloaded pointer → markProcessing returns whether it actually transitioned → on a lost race or error, compensating-decrement the slot.

request.BuildStrategy = entity.BuildStrategyFull
request.BaseURI = ""

if err := c.markProcessing(ctx, &request); err != nil {

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.

Slot leak / self-deadlock. If markProcessing fails with a retryable infra error (or the process crashes here), the slot stays claimed and the delivery redelivers with the request still accepted. On redelivery the gate sees in_flight_count = 1 >= max_concurrent — the request is blocked by its own claimed slot — and acks "awaiting build slot". Nothing ever decrements the count: no build was published, so record never sees it, and the ack means DLQ never sees it either. Permanent capacity loss for the queue and a permanently stuck head (a newer head later inherits the closed gate too).

Suggest a best-effort compensating decrement on the markProcessing error path before returning. That shrinks the hole to a hard crash between the two CASes — worth deciding how that residue gets reconciled (the RFC's record/DLQ (−1) table implies a janitor story; does it cover this?).

if errors.Is(err, storage.ErrVersionMismatch) {
// claimBuildSlot reloaded queueRow; another admit may have taken the last slot.
if queueRow.InFlightCount >= maxConcurrent {
return fmt.Errorf("ProcessController gate closed for queue %s", queueRow.Name)

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.

This returns "gate closed" as a plain (non-retryable) error → nack → DLQ, while the identical condition at the pre-check (L146) acks with a log. That's an expected outcome encoded as an error — against both the repo's "errors for failures, not control flow" rule and the RFC (gate closed → defer, never fail). It's near-unreachable today since concurrent queue-row writers only open the gate or touch other fields, but the path should ack (and later re-enqueue) like the pre-check does.

if err == nil {
break
}
if errors.Is(err, storage.ErrVersionMismatch) {

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.

A version mismatch on the queue row is most plausibly ingest stamping a new latest_request_id — i.e. a newer head just arrived. The RFC says every wake-up re-runs coalesce-then-gate, but this retry loop re-checks only the gate and then admits the now-stale head — spending the slot on exactly the intermediate the design promises to skip. After the reload, re-run CompareRequestID against the reloaded queueRow.LatestRequestID and supersede instead of admit when it's no longer latest.

reqStore := c.store.GetRequestStore()

for {
if request.State != entity.RequestStateAccepted {

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.

Returning nil when the reloaded state isn't accepted swallows a lost race: the caller then logs "admitted request to build" (and, once TODO(build-publish) lands, would publish a build) even if the request was actually superseded by another writer — while still holding the claimed slot. Suggest returning (transitioned bool, err error) so the caller can release the slot and skip the admit on false — this is also the natural hook for the compensating-decrement fix on the error path.

}
}

func expectAdmit(m processMocks, id string) {

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.

Coverage stops at the happy admit path — no cases for the claim-slot version-mismatch retry, gate-closed-after-reload, or the markProcessing lost-race, which is where all the review findings live. expectAdmit makes these cheap to add once the failure-path semantics are settled.

@mnoah1 mnoah1 force-pushed the mnoah1/stovepipe-process-queueconfig branch from 98cef3f to 0c435d1 Compare July 13, 2026 20:31
@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-admit
git rebase --onto origin/main 0c435d12ab8eeba953ea65e940c9deccd522d5cb mnoah1/stovepipe-process-admit
# resolve conflicts, then:
git push --force-with-lease

Then update this PR's base branch:

gh pr edit 335 --base main

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