From dd5223476af5405cd94a1248c8b9865551a68a34 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Wed, 4 Nov 2026 21:05:24 +0000 Subject: [PATCH] fix(stovepipe): Retain request build associations --- doc/rfc/stovepipe/steps/build.md | 37 +++++----- stovepipe/controller/build/build.go | 74 ++++++++++++++----- .../controller/build/build_analysis_test.go | 8 +- stovepipe/controller/build/build_test.go | 45 ++++++++++- stovepipe/entity/request.go | 2 + .../extension/storage/mysql/request_store.go | 13 ++-- .../storage/mysql/request_store_test.go | 26 ++++--- .../storage/mysql/schema/request.sql | 1 + .../extension/storage/mysql/storage_test.go | 3 + 9 files changed, 153 insertions(+), 56 deletions(-) diff --git a/doc/rfc/stovepipe/steps/build.md b/doc/rfc/stovepipe/steps/build.md index 7fb415d74..c5c36349f 100644 --- a/doc/rfc/stovepipe/steps/build.md +++ b/doc/rfc/stovepipe/steps/build.md @@ -1,5 +1,7 @@ # Build stage +Build dispatch retains the request's build association before publishing status and analysis notices. Retries reuse that association instead of triggering again; an uncertain build create is accepted only when its retained request identity matches. + `build` triggers the build-runner for the scope `process` already decided and hands the resulting build id to `buildsignal`. See [workflow.md](../workflow.md) for where it sits in the pipeline, and [process.md](process.md) for how the scope it reads (`BuildStrategy`, `BaseURI`) is chosen. It handles only the trigger: it does not poll for completion, record greenness, or decide incremental-vs-full — those are `buildsignal`'s, `record`'s, and `process`'s jobs respectively. @@ -10,11 +12,11 @@ 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](../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`, `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](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.** It conditionally retains `Request.BuildID` without changing the strategy, baseline, or lifecycle state. The request stays `processing` until its outcome is recorded. On `Build`, this stage creates the row; `buildsignal` remains the sole writer of `Build.Status` and `Build.Version`. Every request association uses the request's version guard and preserves concurrent updates. `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. -In both phases `build` is strictly the trigger — `buildsignal` owns polling — so a crash between `Trigger` and the publish leaves a build no poller ever picks up; the redelivery triggers a replacement and the orphan is accepted waste, mirroring SubmitQueue (see [Idempotency](#idempotency)). +In both phases `buildsignal` owns polling. A crash before retaining the build association can still orphan a triggered build; after association, redelivery reuses the retained build and retries its handoff (see [Idempotency](#idempotency)). ## Algorithm @@ -39,13 +41,12 @@ For a delivery carrying request id `R`: - baseURI = R.BaseURI if R.BuildStrategy == incremental_since_green, else "" (full build). - (headURI = R.URI, baseURI) identify the scope; both are opaque SourceControl tokens. -5. Trigger: buildID, err := buildRunner.Trigger(ctx, baseURI, R.URI, metadata) +5. If R.BuildID is set, reuse it and proceed to step 7. Otherwise trigger: + buildID, err := buildRunner.Trigger(ctx, baseURI, R.URI, metadata) - Trigger takes no caller-supplied id; the runner mints the build's identity, and buildID becomes Build.ID — SubmitQueue's exact convention (see "Alternatives considered" under the contract). - - there is deliberately no "already triggered?" pre-check: with a runner-minted - id there is no key to check by, so a redelivery re-triggers and downstream - idempotency absorbs the duplicate (see Idempotency). + - a retry before association can trigger again; the first retained association wins. - Trigger is async: it returns promptly with the runner-assigned id, not an outcome. - metadata is empty for now; expected to carry real data eventually (e.g. conflict-graph info, or other upstream decisions relevant to the build) once a concrete need lands in @@ -57,11 +58,12 @@ For a delivery carrying request id `R`: - the row carries no scope; it is recoverable from the Request's immutable fields (see the entity table). - a crash between step 5 and this write orphans the triggered build (see Idempotency). - - ErrAlreadyExists -> benign (reachable only with a backend that returns deterministic ids - for retried triggers); continue to step 7. + - ErrAlreadyExists -> reload and verify the retained build belongs to R. + - CAS R.BuildID before publishing. On conflict, reload R and reuse an existing association. - other store error -> return raw; classifier decides. -7. Publish buildID to the buildsignal topic, partitioned by build id. +7. Retain the build-triggered history event and publish the associated build ID to + buildsignal and buildanalysis, partitioned by build id. - each build's poll loop then runs in its own partition (see [Partitioning](#partitioning)). - publish failure -> return raw (non-retryable by default); the Build row exists, and the DLQ path owns recovery (see Error classification). @@ -69,24 +71,20 @@ 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 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 and id are projected onto that Request before the publish, so `record` never reaches a `Build` at all. No reader derives a build id from a Request; the persisted terminal build id supplies the Request → Build navigation, while `Build.RequestID` supplies the other direction. 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.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 and id are projected onto that Request before the publish, so `record` never reaches a `Build` at all. `Request.BuildID` supplies navigation while the build runs, and `TerminalBuildID` identifies the winning terminal build; `Build.RequestID` supplies the other direction. 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. +`build` creates the `Build` and retains its association on the Request; 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 -Every branch is safe under at-least-once redelivery — with SubmitQueue's posture on duplicates adopted wholesale: `build` has no pre-trigger dedup check (there is no caller-derivable key to check by; see [Alternatives considered](#alternatives-considered-for-the-build-identity)), so a redelivery that reaches step 5 starts a second, independent build, and safety comes from downstream idempotency rather than from preventing the duplicate: +A retained `Request.BuildID` prevents a retry from triggering another build after the association is durable. A failed or uncertain create reuses a retained build only after verifying its request identity. The association CAS reloads on conflict and preserves the first associated build; status and analysis publications always use that retained winner. -- **Request 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. -- **Strategy not yet visible** — retryable; the producing stage's write is not visible on this reader yet. -- **Request already terminal** (step 2) — ack, no build. A redelivery after `record` finished, or after `process` superseded the head, never starts a stale build. -- **Redelivery while the Request is still in flight** (crash or failure anywhere in steps 5–8) — the redelivery re-runs from step 1, `Trigger` mints a fresh id, `Create` persists a second `Build` row, and a second poll loop starts. Harmless, in three layers: both builds target the identical `(headURI, baseURI)` scope; each `Build` polls in its own partition and `buildsignal` short-circuits the moment the Request goes terminal (its step 3); and `buildsignal`'s outcome write is first-writer-wins, so the second verdict cannot flip the Request's state or overwrite the create-only validation fact. A build triggered but never persisted (crash between steps 5 and 6) is the same story minus the row: an orphan the runner finishes and nobody ever reads. Wasted CI compute, not a correctness risk — the same accepted trade as SubmitQueue. -- **Trigger / publish / other store failure** — nothing durable is left half-written that a redelivery can't reconcile; the error rejects to `build_dlq`, and `dlq.NewDLQBuildController` drives a still-`processing` Request to `failed` and releases its Queue slot (see [Fail-closed interaction](#fail-closed-interaction)). +The remote trigger and local association are not atomic. A crash before retaining the association can still start another build on retry. Unassociated builds are not announced to polling or analysis; they remain accepted CI waste. A terminal request never triggers again. Publication and storage failures use the normal retry/DLQ policy described below. ## Edge cases - **Head equals last-green** (`R.URI == R.BaseURI`). `process.md` calls this degenerate and leaves it to `build`. Resolution: run a normal incremental build with an empty delta — pass `baseURI = R.BaseURI` unchanged and let the runner apply zero changes on top of it. This needs no branch in `build`: the runner must already handle "no delta" (two adjacent commits) as valid input. Rejected: skipping the build and copying last-green's result forward, because that would force `build` to reason about whether last-green's *recorded greenness* (not just its URI) is still valid — a `record` concern `build` has no business making. -- **Phase 2 triggers another build for the same Request.** No collision by construction: every `Trigger` mints a fresh runner id, so the Phase-1 whole-repo build and each Phase-2 project-scoped build get distinct `Build` rows, all carrying the same `RequestID`. Each build's id travels in its own messages, so downstream navigation stays a direct get; how `analyze` obtains and forwards the ids of the project builds it triggers is part of the deferred project-scoped trigger design. +- **Project-scoped builds.** The current request association identifies one whole-repository validation. Multiple project builds for one Request need separate association identities as part of the still-reserved project-scoped trigger design; they must not reuse this single association. ## Fail-closed interaction @@ -334,6 +332,7 @@ Single-key reads/writes only — no list-by-request, no query-by-attribute — p |---|---|---| | `BuildStrategy` | `incremental_since_green` or `full_monorepo`; immutable once set | `process` | | `BaseURI` | Last-green URI for incremental; empty for full | `process` | +| `BuildID` | Associated CI build; empty before triggering | `build` | | `TerminalBuildID` | The build that established the terminal state; empty when no build established it | `buildsignal` | `URI` already exists; `process` sets `BuildStrategy`/`BaseURI` at admit (process.md step 7c) and `build` reads them (step 4). Both are immutable for the Request's life. @@ -347,4 +346,4 @@ Two topic keys in `stovepipe/core/messagequeue/topics.go` — `TopicKeyBuild` (` ### Partitioning -`process`/`analyze` → `build` is partitioned by **request id**: per-request build work is independent (the per-Queue concurrency gate already ran in `process`), and a single Request's (rare) duplicate deliveries stay ordered — though with no pre-trigger dedup check that ordering is a tidiness property, not a correctness dependency; duplicates are absorbed downstream (see [Idempotency](#idempotency)). `build` → `buildsignal` is partitioned by **build id** (the runner-assigned `Build.ID`), so each build's poll loop is an independent partition. The id is unique per build, so one `Request`'s several Phase-2 project builds land in distinct partitions — the point: a slow poll on one must not block the others. The flip side of request-id partitioning on the `build` topic is that all of one Request's Phase-2 *triggers* serialize through a single partition; that is fine because `build` is trigger-only and cheap, and the concurrency Phase 2 needs comes from the per-build poll partitions, not from parallel triggering. This is a deliberate divergence from SubmitQueue, which partitions its build poll loop by *batch id* (its unit of work); stovepipe goes finer. +`process`/`analyze` → `build` is partitioned by **request id**: per-request build work is independent (the per-Queue concurrency gate already ran in `process`), and a single Request's (rare) duplicate deliveries stay ordered — the retained request association makes duplicate deliveries reuse the same build (see [Idempotency](#idempotency)). `build` → `buildsignal` is partitioned by **build id** (the runner-assigned `Build.ID`), so each build's poll loop is an independent partition. The id is unique per build, so one `Request`'s several Phase-2 project builds land in distinct partitions — the point: a slow poll on one must not block the others. The flip side of request-id partitioning on the `build` topic is that all of one Request's Phase-2 *triggers* serialize through a single partition; that is fine because `build` is trigger-only and cheap, and the concurrency Phase 2 needs comes from the per-build poll partitions, not from parallel triggering. This is a deliberate divergence from SubmitQueue, which partitions its build poll loop by *batch id* (its unit of work); stovepipe goes finer. diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go index eae495190..43e11d970 100644 --- a/stovepipe/controller/build/build.go +++ b/stovepipe/controller/build/build.go @@ -20,7 +20,9 @@ package build import ( "context" + "errors" "fmt" + "math" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -134,43 +136,79 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er baseURI = request.BaseURI } - buildID, err := buildRunner.Trigger(ctx, baseURI, request.URI, nil) - if err != nil { - return fmt.Errorf("failed to trigger build for request %s: %w", request.ID, err) - } - - build := entity.Build{ - ID: buildID.ID, - RequestID: request.ID, - Status: entity.BuildStatusAccepted, - Version: 1, - } - if err := store.GetBuildStore().Create(ctx, build); err != nil { - return fmt.Errorf("failed to persist build %s: %w", build.ID, err) + if request.BuildID == "" { + buildID, err := buildRunner.Trigger(ctx, baseURI, request.URI, nil) + if err != nil { + return fmt.Errorf("failed to trigger build for request %s: %w", request.ID, err) + } + build := entity.Build{ID: buildID.ID, RequestID: request.ID, Status: entity.BuildStatusAccepted, Version: 1} + if err := store.GetBuildStore().Create(ctx, build); errors.Is(err, storage.ErrAlreadyExists) { + retained, err := loader.ByID(ctx, build.ID, store.GetBuildStore().Get, "existing build") + if err != nil { + return err + } + if retained.RequestID != request.ID { + return fmt.Errorf("build %s belongs to another request", build.ID) + } + } else if err != nil { + return fmt.Errorf("failed to persist build %s: %w", build.ID, err) + } + request, err = c.associateBuild(ctx, store, request, build.ID) + if err != nil { + return err + } } - if err := c.persistBuildTriggeredLog(ctx, store, request, build.ID); err != nil { + if err := c.persistBuildTriggeredLog(ctx, store, request, request.BuildID); err != nil { return err } - signalErr := c.publishBuildSignal(ctx, build.ID, request.Queue) + signalErr := c.publishBuildSignal(ctx, request.BuildID, request.Queue) if signalErr != nil { - signalErr = fmt.Errorf("failed to publish build signal for %s: %w", build.ID, signalErr) + signalErr = fmt.Errorf("failed to publish build signal for %s: %w", request.BuildID, signalErr) } - analysisErr := c.publishBuildAnalysis(ctx, build.ID, request.Queue) + analysisErr := c.publishBuildAnalysis(ctx, request.BuildID, request.Queue) if err := errs.Group(signalErr, analysisErr); err != nil { return err } c.logger.Debugw("triggered build", "request_id", request.ID, - "build_id", build.ID, + "build_id", request.BuildID, "queue", request.Queue, "base_uri", baseURI, ) return nil } +func (c *Controller) associateBuild(ctx context.Context, store storage.Storage, request entity.Request, buildID string) (entity.Request, error) { + for request.BuildID == "" { + if err := ctx.Err(); err != nil { + return entity.Request{}, err + } + if request.Version == math.MaxInt32 { + return entity.Request{}, fmt.Errorf("request version exhausted") + } + next := request + next.BuildID = buildID + newVersion := request.Version + 1 + err := store.GetRequestStore().Update(ctx, next, request.Version, newVersion) + if errors.Is(err, storage.ErrVersionMismatch) { + request, err = c.loadRequest(ctx, store, request.ID) + if err != nil { + return entity.Request{}, err + } + continue + } + if err != nil { + return entity.Request{}, fmt.Errorf("associate build with request %s: %w", request.ID, err) + } + next.Version = newVersion + return next, nil + } + return request, nil +} + func (c *Controller) persistBuildTriggeredLog(ctx context.Context, store storage.Storage, request entity.Request, buildID string) error { log := requestlog.NewRequestEventLog( request, diff --git a/stovepipe/controller/build/build_analysis_test.go b/stovepipe/controller/build/build_analysis_test.go index 98642f3a4..c2720c73e 100644 --- a/stovepipe/controller/build/build_analysis_test.go +++ b/stovepipe/controller/build/build_analysis_test.go @@ -45,6 +45,12 @@ func TestBuildAnalysisPipeline(t *testing.T) { var savedBuild entity.Build request := processingRequest(entity.BuildStrategyFull, "") queue := entity.Queue{Name: testQueue, InFlightCount: 1, Version: 1} + m.association.DoAndReturn(func(_ context.Context, value entity.Request, old, next int32) error { + require.Equal(t, request.Version, old) + request = value + request.Version = next + return nil + }) queueStore := storagemock.NewMockQueueStore(ctrl) issueStore := storagemock.NewMockBuildIssueStore(ctrl) @@ -76,7 +82,7 @@ func TestBuildAnalysisPipeline(t *testing.T) { request = value request.Version = next return nil - }) + }).AnyTimes() queueStore.EXPECT().Get(gomock.Any(), testQueue).DoAndReturn(func(context.Context, string) (entity.Queue, error) { return queue, nil diff --git a/stovepipe/controller/build/build_test.go b/stovepipe/controller/build/build_test.go index e98089e8e..2b2d5779f 100644 --- a/stovepipe/controller/build/build_test.go +++ b/stovepipe/controller/build/build_test.go @@ -19,6 +19,7 @@ import ( "database/sql" "database/sql/driver" "errors" + "fmt" "testing" "github.com/stretchr/testify/assert" @@ -66,6 +67,7 @@ type buildMocks struct { runnerFactory *buildrunnermock.MockFactory runner *buildrunnermock.MockBuildRunner publisher *mqmock.MockPublisher + association *gomock.Call metricsScope tally.TestScope } @@ -105,6 +107,9 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMoc require.NoError(t, err) c := NewController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build") + m.association = m.reqStore.EXPECT().Update(gomock.Any(), gomock.Cond(func(request entity.Request) bool { + return request.BuildID == testBuildID && request.Version == 1 + }), int32(1), int32(2)).Return(nil).AnyTimes() return c, m } @@ -332,7 +337,7 @@ func TestProcess(t *testing.T) { }, }, { - name: "duplicate runner build id is rejected", + name: "runner build belonging to another request is rejected", wantErr: true, setup: func(m buildMocks) { req := processingRequest(entity.BuildStrategyFull, "") @@ -340,6 +345,7 @@ func TestProcess(t *testing.T) { m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil) m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{ID: testBuildID, RequestID: "other"}, nil) }, }, { @@ -509,3 +515,40 @@ func TestAnalysisPublishDeduplicatesRetriesOfTheSameOccasion(t *testing.T) { assert.Equal(t, ids[0], ids[1]) assert.NotEqual(t, publish.IntentID(testBuildID, "finished"), ids[0]) } + +func TestBuildAssociationIsRetriedBeforeAnyDownstreamPublication(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := processingRequest(entity.BuildStrategyFull, "") + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil) + m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + m.association.Return(assert.AnError).Times(1) + + require.ErrorIs(t, c.Process(queueContext(), delivery(t, ctrl, buildPayload(t, testID))), assert.AnError) +} + +func TestBuildRetryReusesItsAssociation(t *testing.T) { + for _, associated := range []bool{false, true} { + t.Run(fmt.Sprint(associated), func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := processingRequest(entity.BuildStrategyFull, "") + if associated { + request.BuildID = testBuildID + } + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + if !associated { + m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil) + m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{ID: testBuildID, RequestID: testID}, nil) + } + expectBuildTriggered(m) + m.publisher.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()).Return(nil) + m.publisher.EXPECT().Publish(gomock.Any(), "buildanalysis", gomock.Any()).Return(nil) + require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, buildPayload(t, testID)))) + }) + } +} diff --git a/stovepipe/entity/request.go b/stovepipe/entity/request.go index 850e751d2..163aa5fa4 100644 --- a/stovepipe/entity/request.go +++ b/stovepipe/entity/request.go @@ -115,6 +115,8 @@ type Request struct { // State is the current state of the request in the pipeline. State RequestState `json:"state"` + // BuildID identifies the associated CI build and is empty before it is triggered. + BuildID string `json:"build_id"` // TerminalBuildID is the build that established this request's terminal // state. It is empty when no build established the terminal state. TerminalBuildID string `json:"terminal_build_id"` diff --git a/stovepipe/extension/storage/mysql/request_store.go b/stovepipe/extension/storage/mysql/request_store.go index e9a0e0194..6741f004b 100644 --- a/stovepipe/extension/storage/mysql/request_store.go +++ b/stovepipe/extension/storage/mysql/request_store.go @@ -55,14 +55,15 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE } _, err := r.db.ExecContext(ctx, - `INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version) - VALUES (?, ?, ?, ?, ?, ?, ?)`, + `INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, build_id, version) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, + request.BuildID, request.Version, ) if err != nil { @@ -82,7 +83,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, var req entity.Request err := r.db.QueryRowContext(ctx, - `SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE(terminal_build_id, ''), version + `SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE(terminal_build_id, ''), version FROM request WHERE queue = ? AND id = ?`, r.queue, id, ).Scan( @@ -92,6 +93,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, &req.State, &req.BuildStrategy, &req.BaseURI, + &req.BuildID, &req.TerminalBuildID, &req.Version, ) @@ -107,7 +109,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, } // Update persists the mutable fields of request (URI, state, build strategy, -// base URI, terminal build ID) if the oldVersion, writing newVersion. Returns +// base URI, associated build ID, terminal build ID) if the oldVersion, writing newVersion. Returns // ErrVersionMismatch if the stored version does not match (including when the // request does not exist). This is a pure conditional write; the caller owns // version arithmetic. @@ -121,12 +123,13 @@ func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVe result, err := r.db.ExecContext(ctx, `UPDATE request - SET uri = ?, state = ?, build_strategy = ?, base_uri = ?, terminal_build_id = NULLIF(?, ''), version = ? + SET uri = ?, state = ?, build_strategy = ?, base_uri = ?, build_id = ?, terminal_build_id = NULLIF(?, ''), version = ? WHERE queue = ? AND id = ? AND version = ?`, request.URI, request.State, request.BuildStrategy, request.BaseURI, + request.BuildID, request.TerminalBuildID, newVersion, request.Queue, diff --git a/stovepipe/extension/storage/mysql/request_store_test.go b/stovepipe/extension/storage/mysql/request_store_test.go index 190ac279b..9c48fefe2 100644 --- a/stovepipe/extension/storage/mysql/request_store_test.go +++ b/stovepipe/extension/storage/mysql/request_store_test.go @@ -62,7 +62,7 @@ func TestRequestStore_Create(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request"). - WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.Version). + WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.Version). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -70,7 +70,7 @@ func TestRequestStore_Create(t *testing.T) { name: "duplicate id returns ErrAlreadyExists", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request"). - WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.Version). + WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.Version). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -80,7 +80,7 @@ func TestRequestStore_Create(t *testing.T) { name: "other exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request"). - WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.Version). + WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.Version). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -116,6 +116,7 @@ func TestRequestStore_Get(t *testing.T) { State: entity.RequestStateProcessing, BuildStrategy: entity.BuildStrategyFull, BaseURI: "", + BuildID: "bk-1001", TerminalBuildID: "bk-1001", Version: 2, } @@ -132,9 +133,9 @@ func TestRequestStore_Get(t *testing.T) { name: "found", id: want.ID, setup: func(mock sqlmock.Sqlmock) { - rows := sqlmock.NewRows([]string{"id", "queue", "uri", "state", "build_strategy", "base_uri", "terminal_build_id", "version"}). - AddRow(want.ID, want.Queue, want.URI, string(want.State), string(want.BuildStrategy), want.BaseURI, want.TerminalBuildID, want.Version) - mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE\\(terminal_build_id, ''\\), version"). + rows := sqlmock.NewRows([]string{"id", "queue", "uri", "state", "build_strategy", "base_uri", "build_id", "terminal_build_id", "version"}). + AddRow(want.ID, want.Queue, want.URI, string(want.State), string(want.BuildStrategy), want.BaseURI, want.BuildID, want.TerminalBuildID, want.Version) + mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), version"). WithArgs("monorepo/main", want.ID). WillReturnRows(rows) }, @@ -144,7 +145,7 @@ func TestRequestStore_Get(t *testing.T) { name: "not found", id: "missing", setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE\\(terminal_build_id, ''\\), version"). + mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), version"). WithArgs("monorepo/main", "missing"). WillReturnError(sql.ErrNoRows) }, @@ -155,7 +156,7 @@ func TestRequestStore_Get(t *testing.T) { name: "query error", id: "bad", setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE\\(terminal_build_id, ''\\), version"). + mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), version"). WithArgs("monorepo/main", "bad"). WillReturnError(fmt.Errorf("connection reset")) }, @@ -193,6 +194,7 @@ func TestRequestStore_Update(t *testing.T) { State: entity.RequestStateProcessing, BuildStrategy: entity.BuildStrategyFull, BaseURI: "", + BuildID: "bk-1001", TerminalBuildID: "bk-1001", } const oldVersion, newVersion = int32(1), int32(2) @@ -207,7 +209,7 @@ func TestRequestStore_Update(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion). + WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -215,7 +217,7 @@ func TestRequestStore_Update(t *testing.T) { name: "version mismatch", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion). + WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -225,7 +227,7 @@ func TestRequestStore_Update(t *testing.T) { name: "exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion). + WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -234,7 +236,7 @@ func TestRequestStore_Update(t *testing.T) { name: "rows affected error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion). + WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, diff --git a/stovepipe/extension/storage/mysql/schema/request.sql b/stovepipe/extension/storage/mysql/schema/request.sql index 52ef9c8b6..37d710f26 100644 --- a/stovepipe/extension/storage/mysql/schema/request.sql +++ b/stovepipe/extension/storage/mysql/schema/request.sql @@ -8,6 +8,7 @@ CREATE TABLE IF NOT EXISTS request ( state VARCHAR(64) NOT NULL, build_strategy VARCHAR(64) NOT NULL DEFAULT '', base_uri VARCHAR(255) NOT NULL DEFAULT '', + build_id VARCHAR(255) NOT NULL DEFAULT '', version INT NOT NULL, terminal_build_id VARCHAR(255) NULL, PRIMARY KEY (queue, id) diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 868833b91..de593bd9c 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -295,6 +295,7 @@ func (s *MySQLRequestStoreSuite) TestCreateAndGetWithProcessFields() { State: entity.RequestStateProcessing, BuildStrategy: entity.BuildStrategyIncrementalSinceGreen, BaseURI: "git://remote/monorepo/main/green-bbbb", + BuildID: "build/3", Version: 2, } require.NoError(s.T(), s.store.Create(s.ctx, req)) @@ -323,6 +324,7 @@ func (s *MySQLRequestStoreSuite) TestUpdateCAS() { updated.URI = "git://remote/monorepo/main/resolved" updated.State = entity.RequestStateProcessing updated.BuildStrategy = entity.BuildStrategyFull + updated.BuildID = "build/4" require.NoError(s.T(), s.store.Update(s.ctx, updated, 1, 2)) got, err := s.store.Get(s.ctx, req.ID) @@ -330,6 +332,7 @@ func (s *MySQLRequestStoreSuite) TestUpdateCAS() { require.Equal(s.T(), "git://remote/monorepo/main/resolved", got.URI) require.Equal(s.T(), entity.RequestStateProcessing, got.State) require.Equal(s.T(), entity.BuildStrategyFull, got.BuildStrategy) + require.Equal(s.T(), "build/4", got.BuildID) require.Equal(s.T(), int32(2), got.Version) // Stale CAS: oldVersion 1 no longer matches the stored version (2).