feat(stovepipe): admit latest head to build

[mnoah1]

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}

[behinddwalls]

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.

[behinddwalls]

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?).

[behinddwalls]

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.

[behinddwalls]

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.

[behinddwalls]

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.

[behinddwalls]

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.

[behinddwalls]

⚠️ 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