From 92e17757003332028ce76f5921490e477076a03f Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Tue, 28 Jul 2026 15:03:02 -0700 Subject: [PATCH] feat(stovepipe): record the build outcome on the request and free its slot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Nothing wrote the build outcome states the previous commit introduced, and nothing released a request's build slot on the normal path. The slot was specified as `record`'s to release, but `record` does not exist — so today every admitted request holds its slot forever and only the DLQ reconciler ever frees one. Releasing it in `record` is also the wrong home. The gate `process` claims is a *build* slot: it bounds concurrent builds per Queue, and once the build reaches a terminal status the build is over. Waiting for a later stage to notice keeps the gate closed for no reason, and it breaks a useful invariant — *a terminal request has already released its slot* — which is exactly what makes the DLQ reconciler's early-return on terminal requests safe. ### What? When a build reaches a terminal status, `buildsignal` now releases the queue's build slot and CAS-transitions the request from `processing` to the matching outcome, then publishes to `record` as before. The transition is first-writer-wins, so the duplicate builds that at-least-once delivery can produce cannot flip a request's verdict back and forth. Two ordering rules, both serving one invariant — *the request must not go terminal while still holding a slot*, because a terminal request is skipped by redelivery and by the reconciler alike, so nothing would ever decrement it: - the slot is released *before* the terminal write, and - a failed release aborts the write. Failing this way leaves the request non-terminal, so redelivery re-runs both steps and decrements twice — transiently over-admitting by one until the zero clamp reconverges. That is the same trade the DLQ reconciler already documents and makes, for the same reason. The early-exit guard also narrows. With outcomes now terminal, the old `IsTerminal()` check would return before republishing to `record`, losing the message if the process died between the stamp and the publish. The guard now proceeds when the request is `processing` or already carries an outcome; falling through in the second case is safe, since the re-poll is redundant but harmless, the status and outcome writes both no-op, and the slot is not released twice. It is phrased that way rather than as "skip if superseded" because a superseded request cannot reach the poll loop at all — `process` supersedes solely from `accepted`, and a build only exists once the request was admitted. Naming an unreachable state would have implied a case that does not exist. The guard is still kept rather than dropped: the slot release is conditioned on not already having an outcome, so a terminal-without-an-outcome request arriving here would decrement a slot that superseding never claimed. This also fixes a stale RFC sentence in `process.md` that still described the slot as released by `record` writing green or not-green. ## Test Plan ✅ `bazel test //stovepipe/...` — outcome stamped for each terminal status; the slot released exactly once and *not* on a redelivery of an already-stamped request; a failed release leaves the request non-terminal with no record publish; write-once still holds when a later poll disagrees. ✅ `bazel test //test/e2e/stovepipe/...` — the pipeline runs end to end. --- doc/rfc/stovepipe/steps/build.md | 4 +- doc/rfc/stovepipe/steps/buildsignal.md | 32 +++- doc/rfc/stovepipe/steps/process.md | 6 +- doc/rfc/stovepipe/workflow.md | 8 +- .../controller/buildsignal/buildsignal.go | 137 +++++++++++++++++- .../buildsignal/buildsignal_test.go | 136 ++++++++++++++--- 6 files changed, 282 insertions(+), 41 deletions(-) diff --git a/doc/rfc/stovepipe/steps/build.md b/doc/rfc/stovepipe/steps/build.md index c74de713..9042b3c7 100644 --- a/doc/rfc/stovepipe/steps/build.md +++ b/doc/rfc/stovepipe/steps/build.md @@ -10,7 +10,7 @@ It handles only the trigger: it does not poll for completion, record greenness, `build` consumes a request id, published by `process` in Phase 1 (see [workflow.md](doc/rfc/stovepipe/workflow.md#workflow)). Both phases drive the same `build` → `buildsignal` machinery against the same `Request` row. The `process`/`analyze` → `build` topic is partitioned by **request id**; see [Partitioning](#partitioning) for the full rationale, including why `build` → `buildsignal` partitions finer (by build id) than SubmitQueue's equivalent topic. -**`build` is not the sole writer of the rows it touches, and its own writes are narrow.** On `Request`, `build` never writes at all — it only reads the `BuildStrategy`/`BaseURI`/`URI` fields `process` set at admit, and it leaves `Request.State` untouched at `processing` throughout (`process` and the DLQ reconciler are `Request.State`'s only writers). On `Build`, `build` is the sole creator — it calls `BuildStore.Create` exactly once, at step 6 — and never mutates the row again; `buildsignal` is the sole writer of `Build.Status`/`Build.Version` afterward (see [buildsignal.md](doc/rfc/stovepipe/steps/buildsignal.md#input-and-re-entrancy)). This division is why `build` never needs a CAS/version write of its own: `Create` is the only storage mutation in its algorithm. +**`build` is not the sole writer of the rows it touches, and its own writes are narrow.** On `Request`, `build` never writes at all — it only reads the `BuildStrategy`/`BaseURI`/`URI` fields `process` set at admit, and it leaves `Request.State` untouched at `processing` throughout (`process`, `buildsignal`, and the DLQ reconciler are `Request.State`'s only writers). On `Build`, `build` is the sole creator — it calls `BuildStore.Create` exactly once, at step 6 — and never mutates the row again; `buildsignal` is the sole writer of `Build.Status`/`Build.Version` afterward (see [buildsignal.md](doc/rfc/stovepipe/steps/buildsignal.md#input-and-re-entrancy)). This division is why `build` never needs a CAS/version write of its own: `Create` is the only storage mutation in its algorithm. `build` is phase-agnostic: it never asks "which phase is this?" It reads whatever scope is already persisted and immutable on the `Request` and acts on it. Phase 2's project-scoped invocation is expected to read project-scoped equivalents of the scope fields off the same `Request`; the exact shape of a project-scoped trigger is left to the `analyze` design, consistent with `workflow.md`'s "project mapping contract" open question — see [Project-scoped `Trigger`: reserved, not yet designed](#project-scoped-trigger-reserved-not-yet-designed) for the resulting gap in the `BuildRunner` contract itself. @@ -69,7 +69,7 @@ For a delivery carrying request id `R`: `Build.ID` is **minted by the runner at `Trigger`**, exactly as in SubmitQueue's build controller: the runner returns its native id (a Buildkite build number, a CI-gateway job id), `build` adopts it as the `Build`'s key, and that same id travels on every downstream hop — `build` → `buildsignal` → `record` each carry the build id in the message, so every reader reaches the `Build` by a direct get on identity it was handed. No reader ever *derives* a build id or needs a reverse index; `Build.RequestID` covers the one navigation the pipeline needs in the other direction (`Build` → `Request`). Another approach is deriving the key from the Request (`buildKey(R)`) and/or passed a caller-supplied idempotency key to `Trigger`; see [Alternatives considered](#alternatives-considered-for-the-build-identity) for what each would buy and cost. -`build` writes only the `Build`, and only at creation; it never mutates `Request.State`. The Request stays `processing` (set by `process`) through `build` and `buildsignal` until `record` moves it terminal. `Build.Status` is the fine-grained build lifecycle; `Request.State` is the coarse pipeline lifecycle. This is also what keeps `process.md` step 3 correct: because `build` leaves the Request at `processing`, a redelivered `process` message still matches its "if processing, re-publish to build" guard. +`build` writes only the `Build`, and only at creation; it never mutates `Request.State`. The Request stays `processing` (set by `process`) through `build` until `buildsignal` moves it terminal by recording the build's outcome. `Build.Status` is the fine-grained build lifecycle; `Request.State` is the coarse pipeline lifecycle. This is also what keeps `process.md` step 3 correct: because `build` leaves the Request at `processing`, a redelivered `process` message still matches its "if processing, re-publish to build" guard. ## Idempotency diff --git a/doc/rfc/stovepipe/steps/buildsignal.md b/doc/rfc/stovepipe/steps/buildsignal.md index a05d96ab..20a4c30f 100644 --- a/doc/rfc/stovepipe/steps/buildsignal.md +++ b/doc/rfc/stovepipe/steps/buildsignal.md @@ -1,8 +1,8 @@ # Buildsignal stage -`buildsignal` polls the build-runner until a build reaches a terminal state, records the status, and — once the build is terminal — publishes the build id to `record`. See [workflow.md](doc/rfc/stovepipe/workflow.md) for where it sits in the pipeline and [build.md](doc/rfc/stovepipe/steps/build.md) for how the build it polls was triggered. +`buildsignal` polls the build-runner until a build reaches a terminal state, records the status, and — once the build is terminal — releases the queue's build slot, projects the outcome onto the `Request`, and publishes the build id to `record`. See [workflow.md](doc/rfc/stovepipe/workflow.md) for where it sits in the pipeline and [build.md](doc/rfc/stovepipe/steps/build.md) for how the build it polls was triggered. -It handles only the poll loop: it does not decide build strategy, write greenness, or map targets to projects — those are `process`'s, `record`'s, and `analyze`'s jobs. +It handles only the poll loop: it does not decide build strategy, write greenness, or map targets to projects — those are `process`'s, `record`'s, and `analyze`'s jobs. It reports the *fact* (this request's build ended succeeded, failed, or cancelled); `record` derives what that fact *means* for greenness. `buildsignal` is structurally the same controller as `submitqueue/orchestrator/controller/buildsignal/buildsignal.go`, and the "Flow", "Status delivery", and "Polling primitive" sections of [doc/rfc/submitqueue/build-runner.md](doc/rfc/submitqueue/build-runner.md) are the reference rationale it reuses directly. The entity and storage shapes it depends on are defined in [build.md](doc/rfc/stovepipe/steps/build.md#entity-and-storage-additions-needed); this doc introduces none of its own. @@ -12,7 +12,11 @@ It handles only the poll loop: it does not decide build strategy, write greennes Its logic does not branch on phase: it loads the `Build`, polls it toward terminal, persists the result, and publishes the build id onward to `record`. What differs between phases is what `record` does with that publish (whole-repo vs. per-project greenness) — not anything `buildsignal` decides. -`buildsignal` is the sole writer of `Build.Status`/`Build.Version` after `build` creates the row (see [build.md](doc/rfc/stovepipe/steps/build.md#input-partitioning-and-the-single-writer-property)); it only reads `Request` via `RequestStore.Get` (for `R.Queue`, to resolve the build-runner) and never writes it. +`buildsignal` is the sole writer of `Build.Status`/`Build.Version` after `build` creates the row (see [build.md](doc/rfc/stovepipe/steps/build.md#input-partitioning-and-the-single-writer-property)). It reads `Request` via `RequestStore.Get` (for `R.Queue`, to resolve the build-runner) and writes it exactly once, at the terminal transition, to record the build's outcome — the one `Request.State` write outside `process` and the DLQ reconciler. + +Its early-exit guard is deliberately narrower than `State.IsTerminal()`: it proceeds when the request is `processing` **or** already carries a build outcome. The second case matters because a redelivery after the outcome was stamped but before the `record` publish landed must re-publish rather than drop the signal; everything it re-runs is a no-op (the status is unchanged, the outcome is already recorded, the slot is not released twice) and the `record` publish is idempotent. + +The states the guard actually rejects are terminal-without-an-outcome — today `superseded` alone, which `process` sets only from `accepted`, so it cannot reach the poll loop at all. The guard is kept rather than dropped because the slot release in step 7a is conditioned on *not* already having an outcome, so a superseded request arriving here would decrement a slot that superseding never claimed. ## Algorithm @@ -27,8 +31,13 @@ For a delivery carrying build id `B`: - ErrNotFound -> return raw; non-retryable, same as step 1 — the Build's existence proves the Request write is already committed, so a miss here is a storage defect, not a lagging read. -3. If R.State is terminal (superseded / succeeded / failed / cancelled): ack and return. - - the Request is done (its build outcome was recorded, or the head was superseded); stop polling. +3. If R.State is neither processing nor already an outcome: ack and return. + - a request reaches the poll loop only after process admitted it, so it is processing — + or already carries this build's outcome on a redelivery, which falls through so the + signal is re-published to record instead of dropped. + - the only other terminal-without-an-outcome state is superseded, and process supersedes + solely from accepted, so it cannot occur here. Guarded anyway: step 7a would otherwise + release a build slot the request never claimed, since superseding consumes none. - mirrors submitqueue's halted-batch short-circuit in buildsignal.go. 4. Resolve the build-runner: buildRunner = Factory.For(Config{QueueName: R.Queue}). @@ -51,7 +60,12 @@ For a delivery carrying build id `B`: - with the write-once rule, accepted -> running -> {succeeded|failed|cancelled} is monotonic by mechanism, not by assumption about the backend. -7. If the stored status is terminal: publish B (the build key, Build.ID) to the record topic, +7. If the stored status is terminal, and R does not already carry an outcome: + a. Release the queue's build slot: CAS-decrement Queue.in_flight_count, clamped at zero. + - failure here aborts the step: R must not go terminal while still holding a slot. + b. CAS R from processing to the outcome the stored status projects onto it: + succeeded -> succeeded, failed -> failed, cancelled -> cancelled. First writer wins. + Then publish B (the build key, Build.ID) to the record topic, partitioned by request id; ack, return. No re-publish to buildsignal. - record loads the Build directly by this key (Build->Request gives it R for the per-Queue writes), so no reverse lookup from Request to its builds is ever needed. @@ -68,6 +82,10 @@ For a delivery carrying build id `B`: - ack. ``` +**Why the slot is released before the outcome write, and why a failed release aborts it**: `Queue` and `Request` are separate entities with no cross-entity transaction, so the ordering picks which crash failure mode we accept. Both rules serve one invariant — *the request must not go terminal while still holding a slot* — because a terminal request is skipped by redelivery and by the DLQ reconciler alike, so nothing would ever decrement it. Failing this way leaves the request non-terminal: redelivery re-runs both steps and decrements again, transiently over-admitting by one slot until the zero clamp reconverges. Over-admission is the failure mode this pipeline already prefers, for the same reason and in the same words as the DLQ reconciler (see [process.md](doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity)). + +**Why `buildsignal` releases the slot rather than `record`**: the gate `process` claims is a *build* slot — it exists to bound concurrent builds per Queue — and once the build is terminal the build is over. Releasing here also keeps the invariant *a terminal `Request` has already released its slot*, which is what makes the DLQ reconciler's early-return on terminal requests safe. + **Why `record` hears only terminal signals**: `record` has no non-terminal work — by its own contract a non-terminal signal would be a pure no-op — and step 7 already branches on terminality to decide whether to keep polling, so gating the publish costs nothing and spares `record` a no-op delivery on every poll tick of every running build. Crash-safety is unaffected: a crash between the terminal `Update` and the publish redelivers the message; step 5 re-polls (the runner reports the same terminal status), step 6 no-ops, step 7 publishes. This is a deliberate divergence from SubmitQueue, whose buildsignal republishes to `speculate` on every tick — sound there because speculate is a state machine that may act on any signal; stovepipe has no such consumer. **Why step 6 guards on status and makes terminal write-once**: an unchanged status skips the CAS write entirely, so a long build being polled every couple of seconds doesn't churn `Build.Version` on every tick — the version only advances on a real state transition. The write-once rule exists because CAS alone cannot provide it: optimistic locking defends against *concurrent* writers, but a later delivery that polls a flaky backend and sees a different terminal status would CAS cleanly against the current version and overwrite (see Edge cases). A given `Build` has a single poll partition (see [Partitioning](doc/rfc/stovepipe/steps/build.md#partitioning)), so the only writer racing the CAS is a redelivery of the same message (e.g. after a lapsed visibility lease); `ErrVersionMismatch` carries a retryable classification and converges on redelivery. @@ -136,4 +154,4 @@ One boundary of that posture is worth stating: the `MaxAttempts` path fires only ## Entity, storage, and queue additions -No additions beyond [build.md](doc/rfc/stovepipe/steps/build.md#entity-and-storage-additions-needed): `buildsignal` calls `BuildStore.Get`/`Update` and `RequestStore.Get` against the `Build`/`Request` shapes defined there — `Build.ID` being the runner-assigned id it hands straight back to `Status` — and consumes/re-produces the `BuildSignal` message on `TopicKeyBuildSignal` introduced there. The message it publishes to `record`, and the `record` topic key itself, are owned by the `record` stage and land with `record.md`; `buildsignal` only needs that the **build id** (the build key, so `record` loads the `Build` directly) reaches the record topic once the build is terminal, partitioned by request id. +No additions beyond [build.md](doc/rfc/stovepipe/steps/build.md#entity-and-storage-additions-needed): `buildsignal` calls `BuildStore.Get`/`Update` and `RequestStore.Get`/`Update` against the `Build`/`Request` shapes defined there — `Build.ID` being the runner-assigned id it hands straight back to `Status` — plus `QueueStore.Get`/`Update` to release the build slot, and consumes/re-produces the `BuildSignal` message on `TopicKeyBuildSignal` introduced there. `Request.State` gains the three build outcomes (`succeeded`, `failed`, `cancelled`), all terminal. The message it publishes to `record`, and the `record` topic key itself, are owned by the `record` stage and land with `record.md`; `buildsignal` only needs that the **build id** (the build key, so `record` loads the `Build` directly) reaches the record topic once the build is terminal, partitioned by request id. diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md index 63e3e4f5..f297a395 100644 --- a/doc/rfc/stovepipe/steps/process.md +++ b/doc/rfc/stovepipe/steps/process.md @@ -9,7 +9,7 @@ It handles everything *before* the first build is triggered: it does not run bui Ingest publishes a `ProcessRequest` (request id only — producer and consumer share the store) to the process topic, **partitioned by Queue name**. Two consequences the design relies on: 1. **One partition per Queue**, and `BatchSize = 1` (strict serialization; see [sql-queue-rfc.md](../../sql-queue-rfc.md#strict-serialization-opt-in)) — so at most one `process` invocation runs per Queue at a time. The read-modify-write on the Queue row (gate check, count increment) is therefore race-free **without** a transaction, the same property SubmitQueue's `validate` gets from per-queue partition leasing. -2. **`process` is not the only Queue-row writer.** `ingest` stamps `latest_request_id` (the request id of the newest accepted head, not a URI; see [Backlog coalescing](#backlog-coalescing)) and `record` advances the `last_green_uri` bookmark and releases slots. They touch different fields under optimistic-locking CAS, so concurrent writes converge instead of clobbering. +2. **`process` is not the only Queue-row writer.** `ingest` stamps `latest_request_id` (the request id of the newest accepted head, not a URI; see [Backlog coalescing](#backlog-coalescing)) and `buildsignal` releases slots when a build goes terminal, and `record` advances the `last_green_uri` bookmark. They touch different fields under optimistic-locking CAS, so concurrent writes converge instead of clobbering. ## Process algorithm @@ -62,7 +62,7 @@ Validation is expensive and shares a baseline, so heads arriving while an earlie | Queue row | `in_flight_count` | Requests past `process` and not yet terminal. `process` increments on admit; `record` (or DLQ reconciliation) decrements on terminal. | | Queue config | `max_concurrent` | Cap on concurrent in-flight validations. **Default 1** (global wiring default for MVP; per-queue override when a Stovepipe `queueconfig` extension lands). | -A slot is held for the **entire** Phase 1 cycle (`process → build → buildsignal → record`), not just while `process` runs. It is released when the Request reaches **any** terminal state and `in_flight_count` is decremented — `record` writing green *or* not-green, or the DLQ reconciler forcing a terminal not-green (see [integrity](#in_flight_count-integrity)). A build *failure* frees the slot just like a success; only a Request that never terminates keeps its slot. +A slot is held from admit until the build goes terminal (`process → build → buildsignal`), not just while `process` runs. It is released when the Request reaches **any** terminal state and `in_flight_count` is decremented — `buildsignal` recording the build's outcome, success *or* failure, or the DLQ reconciler forcing a terminal `failed` (see [integrity](#in_flight_count-integrity)). A build *failure* frees the slot just like a success; only a Request that never terminates keeps its slot. **Liveness — a stuck Request wedges the whole Queue.** The slot is shared per-Queue, so a Request admitted but never driven terminal holds the only slot and stalls the whole Queue. The fail-closed path in [workflow.md](../workflow.md#fail-closed-on-unprocessable-work) prevents this: every admitted Request terminates — `build`/`buildsignal` errors retry to `MaxAttempts`, then dead-letter, and the DLQ reconciler forces a conservative terminal `failed`, freeing the slot. Residual risk is operational: a **poison DLQ message** the reconciler can't process loops forever (it runs always-retryable), wedging that Queue until an operator removes it. The gate makes the blast radius the whole Queue, not one Request — monitor and alert on the DLQ. @@ -196,7 +196,7 @@ Per-queue knobs such as `max_concurrent` live outside this row — see [Per-Queu | `accepted` | Ingested, awaiting `process` | no | | `processing` | Admitted; strategy recorded; build in flight | no | | `superseded` | Skipped by coalescing for a newer head | **yes** | -| succeeded / failed / cancelled | Phase 1 build outcome | **yes** | +| *(owned by buildsignal)* succeeded / failed / cancelled | Phase 1 build outcome | **yes** | | *(later)* building, recording, … | Finer states as downstream stages need them | — | Transitions use the repo's optimistic-locking pattern: compute `newVersion = oldVersion + 1`, call `RequestStore.Update(ctx, req, oldVersion, newVersion)`, assign `req.Version = newVersion` only on success (see [storage README](../../../../submitqueue/extension/storage/README.md) and [CLAUDE.md](../../../../CLAUDE.md)). diff --git a/doc/rfc/stovepipe/workflow.md b/doc/rfc/stovepipe/workflow.md index c752321c..5ad4b9b4 100644 --- a/doc/rfc/stovepipe/workflow.md +++ b/doc/rfc/stovepipe/workflow.md @@ -128,8 +128,8 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe 1. **ingest** — invoked by the external poller with a **Queue name**. It asks `SourceControl` for that Queue's current head URI, mints a Request namespaced by the Queue, persists it with no recorded greenness yet, and dedups on `(Queue, head URI)` so a re-reported head is processed once. It publishes the RequestID onward. 2. **process** — decides build strategy (incremental since last-green vs full monorepo), gates concurrent work per Queue, coalesces backlog to the latest head, and publishes to `build`. See [process.md](steps/process.md). 3. **build** — runs the build-runner for the chosen scope. A flag derived from `process` decides whether to build relative to the last-green **baseline URI** (incremental) or from scratch (full). It records a build and publishes the BuildID. -4. **buildsignal** — records the build's status and target graph when the build completes, then publishes the BuildID to `record` (the `Build` row carries its RequestID, so `record` reaches the Request with a direct get). -5. **record** — writes the whole-repo greenness for the head URI (`0` green / `1` broken to start). On green it advances the Queue's **last-green URI** so the next `process` can build incrementally from here. It also decrements the Queue's `in_flight_count`, opening the process concurrency gate for the next head. It fires the **Hooks** extension with the green/not-green event, then fans out into Phase 2. +4. **buildsignal** — records the build's status and target graph when the build completes, then releases the Queue's `in_flight_count` slot, projects the terminal status onto the Request (`succeeded` / `failed` / `cancelled`), and publishes the BuildID to `record` (the `Build` row carries its RequestID, so `record` reaches the Request with a direct get). +5. **record** — writes the whole-repo greenness for the head URI (`0` green / `1` broken to start), derived from the Request's build outcome. On green it advances the Queue's **last-green URI** so the next `process` can build incrementally from here. It fires the **Hooks** extension with the green/not-green event, then fans out into Phase 2. The Queue's `in_flight_count` was already released by `buildsignal` when the build went terminal. ### Phase 2 — project greenness @@ -146,8 +146,8 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe | **ingest** | Queue name (from poller) | process | Resolve head URI via SourceControl, mint Request, persist (no greenness), dedup on `(Queue, head URI)` | | **process** | RequestID | build | Build strategy, concurrency gate, backlog coalescing → [process.md](steps/process.md) | | **build** | RequestID | buildsignal | Run the build-runner for the chosen scope; baseline = last-green URI iff incremental | -| **buildsignal** | BuildID | record (P1), record (P2) | Record build status + target graph; signal completion | -| **record** | BuildID | analyze (P1→P2), Hooks | Write greenness; advance last-green URI on whole-repo green; decrement `in_flight_count`; fire Hooks | +| **buildsignal** | BuildID | record (P1), record (P2) | Record build status + target graph; release `in_flight_count`; project the outcome onto the Request; signal completion | +| **record** | BuildID | analyze (P1→P2), Hooks | Write greenness; advance last-green URI on whole-repo green; fire Hooks | | **analyze** | RequestID | build | Map broken/at-risk targets → projects; decide project-scoped builds | ## Step RFCs diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index 4b3092e4..8915a3ae 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -15,12 +15,14 @@ // Package buildsignal holds the buildsignal-stage queue controller. It consumes // BuildSignal messages (a build id), polls the build-runner until the build // reaches a terminal status, persists that status, and — once terminal — +// releases the queue's build slot, projects the outcome onto the request, and // publishes the build id to record. See // doc/rfc/stovepipe/steps/buildsignal.md. package buildsignal import ( "context" + "errors" "fmt" "strconv" "strings" @@ -54,6 +56,7 @@ var ( // Controller consumes BuildSignal messages, polls the build-runner toward a // terminal status, persists the result, and either reschedules itself or +// releases the queue's build slot, projects the outcome onto the request, and // publishes the build id to record. Implements consumer.Controller. type Controller struct { logger *zap.SugaredLogger @@ -94,8 +97,9 @@ func NewController( // Process reloads the build referenced by the delivery, polls its runner for // the latest status, persists a real transition, and either reschedules a -// poll or, once terminal, publishes the build id to record. Returns nil to -// ack (success) or an error to nack (retry) / reject (DLQ). +// poll or, once terminal, releases the queue's build slot, projects the +// outcome onto the request, and publishes the build id to record. Returns nil +// to ack (success) or an error to nack (retry) / reject (DLQ). func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() @@ -118,9 +122,18 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return err } - // The request is done (record already ran, or the head was superseded); - // stop polling. - if request.State.IsTerminal() { + // A request only reaches the poll loop after process admitted it, so it is + // `processing` — or it already carries this build's outcome, on a redelivery + // after the outcome was stamped but before record was published. Both cases + // proceed: the redelivery re-publishes to record rather than dropping the + // signal, and every write below is a no-op the second time. + // + // Anything else is terminal without a build outcome, which today means only + // `superseded` — and process supersedes solely from `accepted` (see + // supersedeRequest), so a superseded request can never reach this point. It + // is guarded rather than assumed away because finishRequest would otherwise + // release a build slot the request never claimed: superseding consumes none. + if request.State != entity.RequestStateProcessing && !request.State.HasBuildOutcome() { return nil } @@ -141,6 +154,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } if effective.IsTerminal() { + if err := c.finishRequest(ctx, &request, effective); err != nil { + return err + } if err := c.publishRecord(ctx, build.ID, request.ID); err != nil { return fmt.Errorf("failed to publish record for build %s: %w", build.ID, err) } @@ -148,6 +164,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "build_id", build.ID, "request_id", request.ID, "status", string(effective), + "request_state", string(request.State), ) return nil } @@ -164,6 +181,116 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return nil } +// finishRequest releases the queue's build slot and projects the build's +// terminal status onto the request, leaving it terminal. It is a no-op when the +// request already carries an outcome, so a redelivery neither double-releases +// the slot nor rewrites the state. +// +// The slot is released *before* the terminal write, and a failed release aborts +// the write, matching the DLQ reconciler (see stovepipe/controller/dlq/dlq.go). +// Both rules exist for the same reason: the request must not go terminal while +// still holding a slot, because a terminal request is skipped on redelivery and +// by the reconciler, so nothing would ever decrement it. Failing this way leaves +// the request non-terminal, so redelivery re-runs both steps and decrements again +// — transiently over-admitting by one until releaseBuildSlot's zero clamp +// reconverges, which is the failure mode this pipeline prefers. +func (c *Controller) finishRequest(ctx context.Context, request *entity.Request, status entity.BuildStatus) error { + if request.State.HasBuildOutcome() { + return nil + } + + if err := c.releaseBuildSlot(ctx, request.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1) + return err + } + + if err := c.markOutcome(ctx, request, outcomeState(status)); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1) + return err + } + return nil +} + +// outcomeState maps a terminal build status onto the request state that records +// it. Cancelled stays distinct from failed: what a cancellation implies for +// greenness is decided where greenness is recorded. +func outcomeState(status entity.BuildStatus) entity.RequestState { + switch status { + case entity.BuildStatusSucceeded: + return entity.RequestStateSucceeded + case entity.BuildStatusCancelled: + return entity.RequestStateCancelled + default: + return entity.RequestStateFailed + } +} + +// markOutcome CAS-transitions request from processing to state, retrying on version +// conflicts. First writer wins: once any outcome is recorded a later caller leaves it +// alone, so duplicate builds for one request (which build.md accepts) cannot flip the +// verdict back and forth. +func (c *Controller) markOutcome(ctx context.Context, request *entity.Request, state entity.RequestState) error { + reqStore := c.store.GetRequestStore() + + for { + if request.State != entity.RequestStateProcessing { + return nil + } + + updated := *request + updated.State = state + newVersion := request.Version + 1 + if err := reqStore.Update(ctx, updated, request.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + got, getErr := reqStore.Get(ctx, request.ID) + if getErr != nil { + return fmt.Errorf("failed to reload request %s after version mismatch: %w", request.ID, getErr) + } + *request = got + continue + } + return fmt.Errorf("failed to mark request %s %s: %w", request.ID, state, err) + } + updated.Version = newVersion + *request = updated + metrics.NamedCounter(c.metricsScope, _opName, "outcomes", 1, + metrics.NewTag("state", string(state)), + ) + return nil + } +} + +// releaseBuildSlot CAS-decrements the queue's in_flight_count, reopening the process +// concurrency gate now that this request's build is over. It decrements relatively +// (preserving concurrent updates), clamps at zero, and retries on version conflicts. +// Unlike process's unwind-path release this is not best-effort: the caller must not +// mark the request terminal if the slot was not freed, so a hard failure is returned. +func (c *Controller) releaseBuildSlot(ctx context.Context, queueName string) error { + queueStore := c.store.GetQueueStore() + + for { + queueRow, err := queueStore.Get(ctx, queueName) + if err != nil { + return fmt.Errorf("failed to load queue %s to release build slot: %w", queueName, err) + } + if queueRow.InFlightCount <= 0 { + return nil + } + + updated := queueRow + updated.InFlightCount = queueRow.InFlightCount - 1 + newVersion := queueRow.Version + 1 + if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + continue + } + return fmt.Errorf("failed to release build slot for queue %s: %w", queueName, err) + } + metrics.NamedCounter(c.metricsScope, _opName, "slot_released", 1) + return nil + } +} + // reconcile persists a real status transition and returns the status that // should drive the rest of Process: the polled status when persisted (or // already unchanged), or the stored status when a stored terminal status is diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go index beb71469..6face14e 100644 --- a/stovepipe/controller/buildsignal/buildsignal_test.go +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -47,6 +47,7 @@ const ( type buildsignalMocks struct { reqStore *storagemock.MockRequestStore buildStore *storagemock.MockBuildStore + queueStore *storagemock.MockQueueStore runnerFactory *buildrunnermock.MockFactory runner *buildrunnermock.MockBuildRunner publisher *mqmock.MockPublisher @@ -58,6 +59,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsig m := buildsignalMocks{ reqStore: storagemock.NewMockRequestStore(ctrl), buildStore: storagemock.NewMockBuildStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), runnerFactory: buildrunnermock.NewMockFactory(ctrl), runner: buildrunnermock.NewMockBuildRunner(ctrl), publisher: mqmock.NewMockPublisher(ctrl), @@ -66,6 +68,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsig store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() store.EXPECT().GetBuildStore().Return(m.buildStore).AnyTimes() + store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() queue := mqmock.NewMockQueue(ctrl) queue.EXPECT().Publisher().Return(m.publisher).AnyTimes() @@ -115,6 +118,24 @@ func build(status entity.BuildStatus, version int32) entity.Build { } } +// queueRow returns the testQueue's row holding inFlight admitted validations. +func queueRow(inFlight, version int32) entity.Queue { + return entity.Queue{ + Name: testQueue, + InFlightCount: inFlight, + Version: version, + } +} + +// expectFinish wires the slot release and outcome transition a terminal build +// performs: the queue row is CAS-decremented, then the request is CAS-moved from +// processing to state. +func expectFinish(m buildsignalMocks, state entity.RequestState) { + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(1, 4), nil) + m.queueStore.EXPECT().Update(gomock.Any(), queueRow(0, 4), int32(4), int32(5)).Return(nil) + m.reqStore.EXPECT().Update(gomock.Any(), requestWithState(state), int32(1), int32(2)).Return(nil) +} + func TestProcess(t *testing.T) { tests := []struct { name string @@ -165,24 +186,13 @@ func TestProcess(t *testing.T) { }, }, { - name: "succeeded request is a no-op", - setup: func(m buildsignalMocks) { - m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) - m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) - }, - }, - { - name: "failed request is a no-op", + // Neither state can reach the poll loop today — a build only exists once + // process admitted the request. The guard covers them anyway because + // finishRequest would release a build slot they never claimed. + name: "request that never claimed a slot is a no-op", setup: func(m buildsignalMocks) { m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) - m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateFailed), nil) - }, - }, - { - name: "cancelled request is a no-op", - setup: func(m buildsignalMocks) { - m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) - m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateCancelled), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) }, }, { @@ -229,14 +239,50 @@ func TestProcess(t *testing.T) { }, }, { - name: "transition to terminal persists and publishes to record", + name: "succeeded build frees the slot, marks the request and publishes to record", setup: func(m buildsignalMocks) { m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) - updated := build(entity.BuildStatusSucceeded, 2) - m.buildStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + m.buildStore.EXPECT().Update(gomock.Any(), build(entity.BuildStatusSucceeded, 2), int32(2), int32(3)).Return(nil) + expectFinish(m, entity.RequestStateSucceeded) + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil) + }, + }, + { + name: "failed build marks the request failed", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusFailed, nil, nil) + m.buildStore.EXPECT().Update(gomock.Any(), build(entity.BuildStatusFailed, 2), int32(2), int32(3)).Return(nil) + expectFinish(m, entity.RequestStateFailed) + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil) + }, + }, + { + name: "cancelled build marks the request cancelled, not failed", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusCancelled, nil, nil) + m.buildStore.EXPECT().Update(gomock.Any(), build(entity.BuildStatusCancelled, 2), int32(2), int32(3)).Return(nil) + expectFinish(m, entity.RequestStateCancelled) + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil) + }, + }, + { + name: "redelivery of an already-stamped request republishes without touching the slot", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusSucceeded, 3), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) + // No build/queue/request write: the status is unchanged and the outcome + // is already recorded, so the slot must not be released a second time. m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil) }, }, @@ -247,7 +293,9 @@ func TestProcess(t *testing.T) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusFailed, nil, nil) - // No Update call expected: a stored terminal status is write-once. + // No build Update: a stored terminal status is write-once. The stored + // status, not the poll, decides the request's outcome. + expectFinish(m, entity.RequestStateSucceeded) m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil) }, }, @@ -275,6 +323,35 @@ func TestProcess(t *testing.T) { m.buildStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(errors.New("db down")) }, }, + { + name: "outcome write failure is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusSucceeded, 3), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(1, 4), nil) + m.queueStore.EXPECT().Update(gomock.Any(), queueRow(0, 4), int32(4), int32(5)).Return(nil) + m.reqStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(errors.New("db down")) + }, + }, + { + name: "slot release failure leaves the request non-terminal", + wantErr: true, + wantRetry: false, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusSucceeded, 3), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{}, errors.New("db down")) + // No request Update and no record publish: marking the request terminal + // while it still holds a slot would strand the slot, since a terminal + // request is skipped by both redelivery and the DLQ reconciler. + }, + }, { name: "publish to record failure is not retryable", wantErr: true, @@ -285,6 +362,7 @@ func TestProcess(t *testing.T) { m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) m.buildStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)).Return(nil) + expectFinish(m, entity.RequestStateSucceeded) m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(errors.New("queue down")) }, }, @@ -407,3 +485,21 @@ func TestNextPollMessageID(t *testing.T) { }) } } + +func TestOutcomeState(t *testing.T) { + tests := []struct { + name string + status entity.BuildStatus + expected entity.RequestState + }{ + {name: "succeeded", status: entity.BuildStatusSucceeded, expected: entity.RequestStateSucceeded}, + {name: "failed", status: entity.BuildStatusFailed, expected: entity.RequestStateFailed}, + {name: "cancelled", status: entity.BuildStatusCancelled, expected: entity.RequestStateCancelled}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expected, outcomeState(tt.status)) + }) + } +}