From ac2c9f21326f624b2078c66a7c100fe32f5eb2de Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Tue, 28 Jul 2026 15:05:15 -0700 Subject: [PATCH] feat(stovepipe)!: key the record stage on the request id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? `buildsignal` published the *build* id to `record`, but `record`'s unit of work is a Request: it writes whole-repo greenness for the Request's URI, advances the Queue's last-green bookmark, and fires Hooks. Keying its input on a build artifact forces every future reader to navigate Build→Request, and it bakes a one-build-per-Request assumption into the wire contract. The assumption is also the wrong one to encode. The concurrency knob gates how many *Requests* are in flight per Queue — same base, different head, one build each — so the fan-out axis is Queue→N Requests, not Request→N Builds. And if a Request ever did fan out, a build-keyed input would give `record` N messages with no way to tell "one build failed" from "one failed and three haven't reported yet". A request-keyed input lets `record` answer its own question. ### What? `Record.id` now carries the request id, and `publishRecord` uses it for the payload, the message id, and the partition key. The field name stays `id`, matching `ProcessRequest.id` and `BuildRequest.id`. Partitioning is unchanged — it was already request id — so the only wire change is what the payload means. No reverse index is needed to make this work: the previous commit projects the build's terminal status onto `Request.State`, so `record` reads the Request and never reaches a `Build`. That also leaves the "no reverse index from Request to its builds" property in `build.md` intact, for a better reason than before, and the derived-build-key alternative explicitly untriggered — its condition is a stage that must derive a build's key from a Request, which `record` no longer does. Marked `!` because the `record` topic's payload semantics change. Nothing consumes it yet — the `record` stage lands later — so there is no migration. ## Test Plan ✅ `bazel test //stovepipe/...` — new unit test pins that the record payload, message id, and partition key are all the request id. ✅ `bazel test //test/e2e/stovepipe/...` — the slow-build case now additionally asserts the request reaches `succeeded` and the queue's `in_flight_count` returns to 0. # Conflicts: # stovepipe/controller/buildsignal/buildsignal_test.go # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto b1937b43 # Last command done (1 command done): # pick d3e6221a feat(stovepipe)!: key the record stage on the request id # No commands remaining. # You are currently rebasing branch 'preetam/stovepipe-record-request-id' on 'b1937b43'. # # Changes to be committed: # modified: doc/rfc/stovepipe/steps/build.md # modified: doc/rfc/stovepipe/steps/buildsignal.md # modified: doc/rfc/stovepipe/workflow.md # modified: stovepipe/controller/buildsignal/buildsignal.go # modified: stovepipe/controller/buildsignal/buildsignal_test.go # modified: stovepipe/core/messagequeue/proto/record.proto # modified: stovepipe/core/messagequeue/protopb/record.pb.go # modified: stovepipe/core/messagequeue/topics.go # modified: test/e2e/stovepipe/harness_test.go # modified: test/e2e/stovepipe/suite_test.go # --- doc/rfc/stovepipe/steps/build.md | 8 +++--- doc/rfc/stovepipe/steps/buildsignal.md | 22 ++++++++-------- doc/rfc/stovepipe/workflow.md | 4 +-- .../controller/buildsignal/buildsignal.go | 24 ++++++++++-------- .../buildsignal/buildsignal_test.go | 23 +++++++++++++++++ .../core/messagequeue/proto/record.proto | 15 +++++------ .../core/messagequeue/protopb/record.pb.go | 15 +++++------ stovepipe/core/messagequeue/topics.go | 9 ++++--- test/e2e/stovepipe/harness_test.go | 25 +++++++++++++++++++ test/e2e/stovepipe/suite_test.go | 6 +++++ 10 files changed, 107 insertions(+), 44 deletions(-) diff --git a/doc/rfc/stovepipe/steps/build.md b/doc/rfc/stovepipe/steps/build.md index 9042b3c7..c8b397b8 100644 --- a/doc/rfc/stovepipe/steps/build.md +++ b/doc/rfc/stovepipe/steps/build.md @@ -67,7 +67,7 @@ For a delivery carrying request id `R`: 8. ack. ``` -`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.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 hop that needs a build — `build` → `buildsignal` carries the build id in the message, so the poll loop reaches the `Build` by a direct get on identity it was handed. `buildsignal` → `record` carries the **request id** instead: `record`'s unit of work is a Request, and the build's terminal status is projected onto `Request.State` before the publish, so `record` never reaches a `Build` at all. 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 passing 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` 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. @@ -263,7 +263,9 @@ Key the `Build` by identity derived from the Request — `buildKey(R) = R.ID` fo | `Request` → `Build` navigation with no reverse index, per the KV key-derivation rule in CLAUDE.md | No current reader needs to *derive* a build id — the id travels in every message hop, so each consumer already holds the key it needs | | Enforces (rather than assumes) the direct-navigation property SubmitQueue's speculate takes on faith | Diverges entity shape and controller flow from SubmitQueue, weakening the "structurally the same controller" claim and dual-implementing-backend symmetry | -Trade-offs: the dedup guards a rare event at a permanent modeling cost. The duplicate it prevents arises only from a redelivery inside the trigger window — rare, and already harmless (identical scope; `buildsignal`'s terminal-Request short-circuit and `record`'s CAS make the loser a no-op — see [Idempotency](#idempotency)). The prospective key-derivers — a future canceller, or `analyze` reaching back to the Phase-1 target graph — would need to be handed the id by their producing stage instead, if those designs land. +Trade-offs: the dedup guards a rare event at a permanent modeling cost. The duplicate it prevents arises only from a redelivery inside the trigger window — rare, and already harmless (identical scope; `buildsignal`'s superseded short-circuit and its first-writer-wins outcome CAS make the loser a no-op — see [Idempotency](#idempotency)). The prospective key-derivers — a future canceller, or `analyze` reaching back to the Phase-1 target graph — would need to be handed the id by their producing stage instead, if those designs land. + +Note that moving `record`'s input from the build id to the request id does *not* trigger this alternative, even though it removes the last hop that carried a build id to a Request-scoped consumer. The trigger condition is a stage that must **derive a build's key from a Request**, and `record` does not: the build's terminal status is projected onto `Request.State` before the publish, so `record` reads the Request and never reaches a `Build`. #### Alternative B: caller-supplied idempotency key on `Trigger` @@ -312,7 +314,7 @@ Either could be adopted independently: the idempotency token, if a backend that `IsTerminal()` on `entity.BuildStatus` covers exactly the three terminal rows. Once `buildsignal` persists one of them, that status is **write-once** — a later poll reporting a different terminal value never overwrites it (see [buildsignal.md](doc/rfc/stovepipe/steps/buildsignal.md#algorithm), step 6). -Plus the `BuildID{ID string}` wire type in `stovepipe/entity` (same "id only travels" convention as `RequestID`, shaped like SubmitQueue's own `entity.BuildID` but not the same Go type — see the [contract sketch](#stovepipe-buildrunner-contract-design-sketch)), wrapping the one runner-assigned id everywhere it appears — `Trigger`'s return, the queue payload, `Status`/`Cancel`'s parameter. `buildsignal` and `record` reach a build by the id carried in their messages, so no reverse index from `Request` to its builds is ever needed. +Plus the `BuildID{ID string}` wire type in `stovepipe/entity` (same "id only travels" convention as `RequestID`, shaped like SubmitQueue's own `entity.BuildID` but not the same Go type — see the [contract sketch](#stovepipe-buildrunner-contract-design-sketch)), wrapping the one runner-assigned id everywhere it appears — `Trigger`'s return, the queue payload, `Status`/`Cancel`'s parameter. `buildsignal` reaches a build by the id carried in its message, and `record` reads the `Request` (whose state carries the build's outcome) rather than a `Build`, so no reverse index from `Request` to its builds is ever needed. **`BuildStore`** (new, added to the `Storage` aggregator via `GetBuildStore()`), matching stovepipe's existing `RequestStore` conventions — **generic `Update` with caller-owned version arithmetic, not SubmitQueue's field-specific `UpdateStatus`**: diff --git a/doc/rfc/stovepipe/steps/buildsignal.md b/doc/rfc/stovepipe/steps/buildsignal.md index 20a4c30f..05fa600f 100644 --- a/doc/rfc/stovepipe/steps/buildsignal.md +++ b/doc/rfc/stovepipe/steps/buildsignal.md @@ -1,6 +1,6 @@ # Buildsignal stage -`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. +`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 **request 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 reports the *fact* (this request's build ended succeeded, failed, or cancelled); `record` derives what that fact *means* for greenness. @@ -10,7 +10,7 @@ It handles only the poll loop: it does not decide build strategy, write greennes `buildsignal` consumes a build id, published by `build` — once per phase, since Phase 1 (whole-repo) and Phase 2 (per-project) each run their own `build` → `buildsignal` cycle against builds tied to the same `Request` (see [workflow.md](doc/rfc/stovepipe/workflow.md#workflow)). -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. +Its logic does not branch on phase: it loads the `Build`, polls it toward terminal, persists the result, and publishes the request 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 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. @@ -65,12 +65,14 @@ For a delivery carrying build id `B`: - 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. - - a redelivery republishes the same terminal signal; record is idempotent. - - publish failure -> return raw (non-retryable); status is persisted, operational republish recovers. + Then publish R.ID to the record topic, partitioned by request id; ack, return. + No re-publish to buildsignal. + - record loads the Request directly by this key and derives greenness from its outcome, + so it never reaches a Build and no reverse lookup from Request to its builds is needed. + - the message id is the request id, so a redelivery republishing the same terminal signal + dedups into the original message; record is idempotent regardless. + - publish failure -> return raw (non-retryable); the outcome is persisted, operational + republish recovers. 8. Else PublishAfter(B -> buildsignal, delayMs), partitioned by build id: - delayMs = pollDelay(status): shorter while running, longer while accepted. @@ -135,7 +137,7 @@ Everything else — factory lookup, an `Update` store error other than a CAS con Every branch is safe under at-least-once redelivery: - **Build not found** — non-retryable; storage's read-after-write guarantee means a miss here is a storage defect, not a lag condition to retry through. -- **Status already persisted** — a redelivery re-runs the whole algorithm from step 1, including a redundant `Status` poll (harmless — the runner reports the same thing); step 6 no-ops on the unchanged status, and the delivery proceeds to re-schedule the poll (non-terminal) or republish to `record` (terminal, idempotent). No corruption. +- **Status already persisted** — a redelivery re-runs the whole algorithm from step 1, including a redundant `Status` poll (harmless — the runner reports the same thing); step 6 no-ops on the unchanged status, and the delivery proceeds to re-schedule the poll (non-terminal) or republish the request id to `record` (terminal, idempotent). No corruption. - **Terminal already published** — a redelivery reloads, re-polls, no-ops at step 6, republishes the same terminal signal to `record` (idempotent), and acks. Harmless. - **`PublishAfter` failed, then retried** — the nacked delivery re-runs from step 1; there is no way to resume mid-algorithm, so it re-polls the runner too, but the row already carries the non-terminal status and step 6 no-ops. Only the final enqueue does new work. @@ -154,4 +156,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`/`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. +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 **request id** reaches the record topic once the build is terminal, partitioned by request id. diff --git a/doc/rfc/stovepipe/workflow.md b/doc/rfc/stovepipe/workflow.md index 5ad4b9b4..93e3c132 100644 --- a/doc/rfc/stovepipe/workflow.md +++ b/doc/rfc/stovepipe/workflow.md @@ -128,7 +128,7 @@ 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 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). +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 RequestID to `record`. 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 @@ -147,7 +147,7 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe | **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; 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 | +| **record** | RequestID | 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 8915a3ae..42b6b819 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -16,7 +16,7 @@ // 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 +// publishes the request id to record. See // doc/rfc/stovepipe/steps/buildsignal.md. package buildsignal @@ -57,7 +57,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. +// publishes the request id to record. Implements consumer.Controller. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -98,8 +98,8 @@ 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, 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). +// outcome onto the request, and publishes the request 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() @@ -157,8 +157,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er 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) + if err := c.publishRecord(ctx, request.ID); err != nil { + return fmt.Errorf("failed to publish record for request %s: %w", request.ID, err) } c.logger.Infow("build reached terminal status", "build_id", build.ID, @@ -336,14 +336,16 @@ func pollDelay(status entity.BuildStatus) int64 { } } -// publishRecord publishes buildID to the record stage, partitioned by -// requestID. -func (c *Controller) publishRecord(ctx context.Context, buildID, requestID string) error { - payload, err := stovepipemq.Marshal(&stovepipemq.Record{Id: buildID}) +// publishRecord publishes requestID to the record stage, partitioned by request +// id. The message id is the request id too, so a redelivery republishing the +// same terminal signal dedups into the original message rather than enqueuing a +// second one. +func (c *Controller) publishRecord(ctx context.Context, requestID string) error { + payload, err := stovepipemq.Marshal(&stovepipemq.Record{Id: requestID}) if err != nil { return fmt.Errorf("failed to serialize record: %w", err) } - msg := entityqueue.NewMessage(buildID, payload, requestID, nil) + msg := entityqueue.NewMessage(requestID, payload, requestID, nil) return c.publish(ctx, stovepipemq.TopicKeyRecord, msg, 0) } diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go index 6face14e..f139e861 100644 --- a/stovepipe/controller/buildsignal/buildsignal_test.go +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -412,6 +412,29 @@ func TestProcess(t *testing.T) { } } +// TestPublishRecordCarriesRequestID pins the record contract: the payload and the +// message id are the request id, not the build id, so record never has to reach a +// Build. The message id is stable so a redelivery dedups instead of enqueuing twice. +func TestPublishRecordCarriesRequestID(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + var got entityqueue.Message + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + got = msg + return nil + }) + + require.NoError(t, c.publishRecord(context.Background(), testID)) + + var payload stovepipemq.Record + require.NoError(t, stovepipemq.Unmarshal(got.Payload, &payload)) + assert.Equal(t, testID, payload.Id) + assert.Equal(t, testID, got.ID) + assert.Equal(t, testID, got.PartitionKey) +} + // TestPublishBuildSignalAdvancesPollGeneration is the regression test for the poll // loop stalling. The queue dedups on (topic, partition_key, id) and the delivery that // scheduled a re-poll is still un-acked when the re-poll is published, so a reused diff --git a/stovepipe/core/messagequeue/proto/record.proto b/stovepipe/core/messagequeue/proto/record.proto index 0ca18e58..68204ccb 100644 --- a/stovepipe/core/messagequeue/proto/record.proto +++ b/stovepipe/core/messagequeue/proto/record.proto @@ -23,15 +23,16 @@ option java_multiple_files = true; option java_outer_classname = "RecordProto"; option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; -// Record is the payload buildsignal publishes once a build reaches a terminal -// status: only the build id travels on the queue. record reloads the full -// Build from storage by this id (producer and consumer share the store, so -// the id is enough and redelivery stays idempotent). See -// doc/rfc/stovepipe/steps/buildsignal.md. +// Record is the payload buildsignal publishes once a request's build reaches a +// terminal status: only the request id travels on the queue. record reloads the +// full Request from storage by this id (producer and consumer share the store, +// so the id is enough and redelivery stays idempotent). The build's terminal +// status is projected onto Request.State before this is published, so record +// never reaches a Build. See doc/rfc/stovepipe/steps/buildsignal.md. message Record { option (uber.base.messagequeue.topic_keys) = "record"; - // id is the terminal build's id: the runner-assigned id minted by - // BuildRunner.Trigger (entity.Build.ID / entity.BuildID.ID). + // id is the request id whose build reached a terminal status. Format: + // "request//" (entity.Request.ID). string id = 1; } diff --git a/stovepipe/core/messagequeue/protopb/record.pb.go b/stovepipe/core/messagequeue/protopb/record.pb.go index ca79d9cc..5c85e7c8 100644 --- a/stovepipe/core/messagequeue/protopb/record.pb.go +++ b/stovepipe/core/messagequeue/protopb/record.pb.go @@ -37,15 +37,16 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -// Record is the payload buildsignal publishes once a build reaches a terminal -// status: only the build id travels on the queue. record reloads the full -// Build from storage by this id (producer and consumer share the store, so -// the id is enough and redelivery stays idempotent). See -// doc/rfc/stovepipe/steps/buildsignal.md. +// Record is the payload buildsignal publishes once a request's build reaches a +// terminal status: only the request id travels on the queue. record reloads the +// full Request from storage by this id (producer and consumer share the store, +// so the id is enough and redelivery stays idempotent). The build's terminal +// status is projected onto Request.State before this is published, so record +// never reaches a Build. See doc/rfc/stovepipe/steps/buildsignal.md. type Record struct { state protoimpl.MessageState `protogen:"open.v1"` - // id is the terminal build's id: the runner-assigned id minted by - // BuildRunner.Trigger (entity.Build.ID / entity.BuildID.ID). + // id is the request id whose build reached a terminal status. Format: + // "request//" (entity.Request.ID). Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache diff --git a/stovepipe/core/messagequeue/topics.go b/stovepipe/core/messagequeue/topics.go index 0dac733d..bac3f71d 100644 --- a/stovepipe/core/messagequeue/topics.go +++ b/stovepipe/core/messagequeue/topics.go @@ -42,9 +42,10 @@ const ( // partition. TopicKeyBuildSignal TopicKey = "buildsignal" - // TopicKeyRecord carries a build's terminal status from buildsignal to the - // record stage. The buildsignal controller publishes a Record (the build - // id) here once, and only once, a build reaches a terminal status; - // non-terminal polls never publish here. Partitioned by request id. + // TopicKeyRecord carries a request whose build reached a terminal status from + // buildsignal to the record stage. The buildsignal controller publishes a + // Record (the request id) here once, and only once, a build reaches a + // terminal status; non-terminal polls never publish here. Partitioned by + // request id. TopicKeyRecord TopicKey = "record" ) diff --git a/test/e2e/stovepipe/harness_test.go b/test/e2e/stovepipe/harness_test.go index 70c8604f..918482e5 100644 --- a/test/e2e/stovepipe/harness_test.go +++ b/test/e2e/stovepipe/harness_test.go @@ -131,6 +131,31 @@ func (s *StovepipeE2ESuite) assertIngestPersisted(queue, id string) { assert.Equal(t, 1, s.publishedMessageCount(id), "should have published one process message for %s", id) } +// awaitRequestState blocks until the request row reaches want. buildsignal projects +// the build's terminal status onto the request, so this is the durable, black-box +// signal that the whole ingest→process→build→buildsignal chain converged. +func (s *StovepipeE2ESuite) awaitRequestState(id, want string) { + pollUntil(processPollInterval, func() bool { + var state string + if err := s.db.QueryRow("SELECT state FROM request WHERE id = ?", id).Scan(&state); err != nil { + s.log.Logf("request %s state not readable yet: %v", id, err) + return false + } + s.log.Logf("request %s state = %q (want %q)", id, state, want) + return state == want + }) +} + +// inFlightCount returns the queue row's in_flight_count, the process concurrency +// gate's counter. +func (s *StovepipeE2ESuite) inFlightCount(queue string) int32 { + t := s.T() + var count int32 + require.NoError(t, s.db.QueryRow("SELECT in_flight_count FROM queue WHERE name = ?", queue).Scan(&count), + "failed to read in_flight_count for queue %s", queue) + return count +} + // awaitBuildStatus blocks until the build row for a request reaches want. The // pipeline runs inside the stovepipe-service container, so the build's own status // column is the durable, black-box signal that the poll loop converged: buildsignal diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index 072ae70e..66ade520 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -183,4 +183,10 @@ func (s *StovepipeE2ESuite) TestIngest_SlowBuild_PollsToCompletion() { // Getting here at all means the reschedule produced a deliverable message. s.awaitBuildStatus(id, "succeeded") + + // buildsignal projects the terminal build status onto the request and, in the + // same step, releases the build slot that reopens the process gate. + s.awaitRequestState(id, "succeeded") + assert.Equal(s.T(), int32(0), s.inFlightCount(queue), + "a terminal build should release the queue's build slot") }