From fdecb29ffccda7af86981934899608e3c1acc1af Mon Sep 17 00:00:00 2001 From: sergeyb Date: Tue, 27 Oct 2026 23:22:26 +0000 Subject: [PATCH] fix(stovepipe): persist the terminal outcome reason on Request A DLQ can save a failed request before its first outcome log is retained. A late build poll then infers build_failed from the state alone, losing the polling or stage failure that actually ended the request. Store the typed outcome reason in the same terminal compare-and-swap as the state and reuse it when constructing state logs. Keep the existing retained-log replay and terminal-build safeguards; legacy snapshots fall back to the writer's reason. Add the MySQL column before deploying. --- doc/rfc/stovepipe/request-log.md | 2 +- .../controller/buildsignal/buildsignal.go | 35 ++++++------ .../buildsignal/buildsignal_test.go | 53 ++++++++++++++++--- stovepipe/controller/dlq/build_test.go | 1 + stovepipe/controller/dlq/buildsignal_test.go | 1 + stovepipe/controller/dlq/dlq.go | 1 + stovepipe/controller/dlq/dlq_test.go | 17 +++++- stovepipe/controller/process/process.go | 1 + stovepipe/controller/process/process_test.go | 4 ++ stovepipe/core/requestlog/materializer.go | 8 ++- stovepipe/entity/request.go | 3 ++ stovepipe/extension/storage/README.md | 4 ++ .../extension/storage/mysql/request_store.go | 14 +++-- .../storage/mysql/request_store_test.go | 34 ++++++------ .../storage/mysql/schema/request.sql | 1 + .../extension/storage/mysql/storage_test.go | 10 ++++ 16 files changed, 144 insertions(+), 45 deletions(-) diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index fbffbc5fc..ccb4059f3 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -244,7 +244,7 @@ The `project_facts_recorded` and `record_abandoned` events project completion of Build running and unchanged polls create no entry. A failed runner trigger that creates no Build creates no event. Cancelled and superseded requests create no validation fact. -No controller infers an outcome reason from `RequestStateFailed` alone or records DLQ as the domain reason. Record DLQ uses that persisted state to recover a missing broken repository fact; it does not infer project verdicts. +A terminal `Request` persists its outcome reason in the same compare-and-swap as its state, and every terminal state entry copies that reason. Only a Request made terminal before reasons were persisted falls back to the reason derived by the stage repairing its entry. No controller otherwise infers an outcome reason from `RequestStateFailed` alone or records DLQ as the domain reason. Record DLQ uses that persisted state to recover a missing broken repository fact; it does not infer project verdicts. ### Repair trigger requirement diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index d4ccadcb9..0c679626e 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -263,20 +263,13 @@ func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, r } func (c *Controller) persistOutcomeLog(ctx context.Context, store storage.Storage, request entity.Request) error { - var reason entity.RequestOutcomeReason - // The durable request is authoritative when duplicate builds race to record different outcomes. - switch request.State { - case entity.RequestStateSucceeded: - reason = entity.RequestOutcomeReasonBuildSucceeded - case entity.RequestStateFailed: - reason = entity.RequestOutcomeReasonBuildFailed - case entity.RequestStateCancelled: - reason = entity.RequestOutcomeReasonBuildCancelled - default: + if !request.State.HasBuildOutcome() { return fmt.Errorf("request %s has no build outcome to record", request.ID) } - log := requestlog.NewRequestStateLog(request, reason) + // The durable request is authoritative: a DLQ may have forced this outcome, or a duplicate + // build won it. The build-derived reason applies only to a request stored without one. + log := requestlog.NewRequestStateLog(request, buildOutcomeReason(request.State)) if request.TerminalBuildID != "" { log.Metadata[requestlog.MetadataKeyBuildID] = request.TerminalBuildID } @@ -300,10 +293,21 @@ func outcomeState(status entity.BuildStatus) entity.RequestState { } } -// markOutcome CAS-transitions request from processing to state, recording the -// winning build ID and 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 buildOutcomeReason(state entity.RequestState) entity.RequestOutcomeReason { + switch state { + case entity.RequestStateSucceeded: + return entity.RequestOutcomeReasonBuildSucceeded + case entity.RequestStateFailed: + return entity.RequestOutcomeReasonBuildFailed + case entity.RequestStateCancelled: + return entity.RequestOutcomeReasonBuildCancelled + default: + return entity.RequestOutcomeReasonUnknown + } +} + +// The first terminal CAS owns the state, build ID and outcome reason; retries +// reload the winner instead of replacing any of them. func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, request *entity.Request, buildID string, state entity.RequestState) error { reqStore := store.GetRequestStore() @@ -315,6 +319,7 @@ func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, req updated := *request updated.State = state updated.TerminalBuildID = buildID + updated.OutcomeReason = buildOutcomeReason(state) newVersion := request.Version + 1 if err := reqStore.Update(ctx, updated, request.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go index 6676889fb..86477fb9a 100644 --- a/stovepipe/controller/buildsignal/buildsignal_test.go +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -161,6 +161,11 @@ func requestWithState(state entity.RequestState) entity.Request { } if state.HasBuildOutcome() { request.TerminalBuildID = testBuildID + request.OutcomeReason = map[entity.RequestState]entity.RequestOutcomeReason{ + entity.RequestStateSucceeded: entity.RequestOutcomeReasonBuildSucceeded, + entity.RequestStateFailed: entity.RequestOutcomeReasonBuildFailed, + entity.RequestStateCancelled: entity.RequestOutcomeReasonBuildCancelled, + }[state] } return request } @@ -225,14 +230,9 @@ func expectFinish(m buildsignalMocks, state entity.RequestState) *gomock.Call { } func expectOutcomeLog(m buildsignalMocks, state entity.RequestState, version int32) *gomock.Call { - reason := map[entity.RequestState]entity.RequestOutcomeReason{ - entity.RequestStateSucceeded: entity.RequestOutcomeReasonBuildSucceeded, - entity.RequestStateFailed: entity.RequestOutcomeReasonBuildFailed, - entity.RequestStateCancelled: entity.RequestOutcomeReasonBuildCancelled, - }[state] request := requestWithState(state) request.Version = version - log := requestlog.NewRequestStateLog(request, reason) + log := requestlog.NewRequestStateLog(request, request.OutcomeReason) log.Metadata[requestlog.MetadataKeyBuildID] = request.TerminalBuildID return m.materializer.EXPECT().PersistLog( gomock.Any(), @@ -403,6 +403,47 @@ func TestProcess(t *testing.T) { m.publisher.EXPECT().Publish(gomock.Any(), "buildanalysis", gomock.Any()).Return(nil).After(logCall) }, }, + { + name: "late poll preserves the saved polling failure reason", + setup: func(m buildsignalMocks) { + request := requestWithState(entity.RequestStateProcessing) + request.State, request.OutcomeReason, request.Version = entity.RequestStateFailed, "build_polling_exhausted", 2 + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusSucceeded, 3), nil) + 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().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) + eventCall := expectBuildFinished(m) + logCall := m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, entity.RequestLog{ + ID: publish.IntentID("state", "2"), Queue: testQueue, RequestID: testID, + State: "failed", RequestVersion: 2, OutcomeReason: "build_polling_exhausted", Metadata: map[string]string{}, + }).Return(nil).After(eventCall) + m.publisher.EXPECT().Publish(gomock.Any(), "buildanalysis", gomock.Any()).Return(nil).After(logCall) + }, + }, + { + name: "lost terminal CAS preserves the winning polling failure reason", + setup: func(m buildsignalMocks) { + request := requestWithState(entity.RequestStateProcessing) + winner := request + winner.State, winner.OutcomeReason, winner.Version = entity.RequestStateFailed, "build_polling_exhausted", 2 + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusSucceeded, 3), nil) + 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().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) + eventCall := expectBuildFinished(m) + gomock.InOrder( + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(1, 4), nil).After(eventCall), + m.queueStore.EXPECT().Update(gomock.Any(), queueRow(0, 4), int32(4), int32(5)).Return(nil), + m.reqStore.EXPECT().Update(gomock.Any(), requestWithState(entity.RequestStateSucceeded), int32(1), int32(2)).Return(storage.ErrVersionMismatch), + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(winner, nil), + ) + logCall := m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, entity.RequestLog{ + ID: publish.IntentID("state", "2"), Queue: testQueue, RequestID: testID, + State: "failed", RequestVersion: 2, OutcomeReason: "build_polling_exhausted", Metadata: map[string]string{}, + }).Return(nil) + m.publisher.EXPECT().Publish(gomock.Any(), "buildanalysis", gomock.Any()).Return(nil).After(logCall) + }, + }, { name: "build event failure stops before request outcome", wantErr: true, diff --git a/stovepipe/controller/dlq/build_test.go b/stovepipe/controller/dlq/build_test.go index d18d166ba..0fb07f032 100644 --- a/stovepipe/controller/dlq/build_test.go +++ b/stovepipe/controller/dlq/build_test.go @@ -76,6 +76,7 @@ func TestBuildProcess(t *testing.T) { m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{Name: testQueue, Version: 5}, int32(5), int32(6)).Return(nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed + updated.OutcomeReason = entity.RequestOutcomeReasonProcessingFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) expectFailureLog(m, entity.RequestOutcomeReasonProcessingFailed, 3).After(updateCall) }, diff --git a/stovepipe/controller/dlq/buildsignal_test.go b/stovepipe/controller/dlq/buildsignal_test.go index 37b6f7228..5da83e740 100644 --- a/stovepipe/controller/dlq/buildsignal_test.go +++ b/stovepipe/controller/dlq/buildsignal_test.go @@ -123,6 +123,7 @@ func TestBuildSignalProcess(t *testing.T) { }, int32(5), int32(6)).Return(nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed + updated.OutcomeReason = entity.RequestOutcomeReasonBuildPollingExhausted updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) request := requestWithState(entity.RequestStateFailed) request.Version = 3 diff --git a/stovepipe/controller/dlq/dlq.go b/stovepipe/controller/dlq/dlq.go index b170c5bcc..26a4e6436 100644 --- a/stovepipe/controller/dlq/dlq.go +++ b/stovepipe/controller/dlq/dlq.go @@ -128,6 +128,7 @@ func failRequest( updated := request updated.State = entity.RequestStateFailed + updated.OutcomeReason = reason newVersion := request.Version + 1 if err := store.GetRequestStore().Update(ctx, updated, request.Version, newVersion); err != nil { return fmt.Errorf("failed to update request %s state to failed: %w", requestID, err) diff --git a/stovepipe/controller/dlq/dlq_test.go b/stovepipe/controller/dlq/dlq_test.go index 75643db1c..6354ba8dd 100644 --- a/stovepipe/controller/dlq/dlq_test.go +++ b/stovepipe/controller/dlq/dlq_test.go @@ -145,6 +145,7 @@ func TestProcess(t *testing.T) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) updated := requestWithState(entity.RequestStateAccepted) updated.State = entity.RequestStateFailed + updated.OutcomeReason = entity.RequestOutcomeReasonProcessingFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) expectFailureLog(m, entity.RequestOutcomeReasonProcessingFailed, 3).After(updateCall) }, @@ -161,6 +162,7 @@ func TestProcess(t *testing.T) { }, int32(5), int32(6)).Return(nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed + updated.OutcomeReason = entity.RequestOutcomeReasonProcessingFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) expectFailureLog(m, entity.RequestOutcomeReasonProcessingFailed, 3).After(updateCall) }, @@ -172,12 +174,21 @@ func TestProcess(t *testing.T) { }, }, { - name: "already failed repairs its missing log", + name: "already failed without a stored reason repairs its missing log with the stage reason", setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateFailed), nil) expectFailureLog(m, entity.RequestOutcomeReasonProcessingFailed, 2) }, }, + { + name: "already failed repairs its missing log with the stored reason, not the stage reason", + setup: func(m dlqMocks) { + request := requestWithState(entity.RequestStateFailed) + request.OutcomeReason = entity.RequestOutcomeReasonBuildPollingExhausted + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + expectFailureLog(m, entity.RequestOutcomeReasonBuildPollingExhausted, 2) + }, + }, { name: "request not found is a no-op", setup: func(m dlqMocks) { @@ -190,6 +201,7 @@ func TestProcess(t *testing.T) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) updated := requestWithState(entity.RequestStateAccepted) updated.State = entity.RequestStateFailed + updated.OutcomeReason = entity.RequestOutcomeReasonProcessingFailed m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(storage.ErrVersionMismatch) }, wantErr: true, @@ -212,6 +224,7 @@ func TestProcess(t *testing.T) { }, int32(6), int32(7)).Return(nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed + updated.OutcomeReason = entity.RequestOutcomeReasonProcessingFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) expectFailureLog(m, entity.RequestOutcomeReasonProcessingFailed, 3).After(updateCall) }, @@ -225,6 +238,7 @@ func TestProcess(t *testing.T) { }, nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed + updated.OutcomeReason = entity.RequestOutcomeReasonProcessingFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) expectFailureLog(m, entity.RequestOutcomeReasonProcessingFailed, 3).After(updateCall) }, @@ -236,6 +250,7 @@ func TestProcess(t *testing.T) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) updated := requestWithState(entity.RequestStateAccepted) updated.State = entity.RequestStateFailed + updated.OutcomeReason = entity.RequestOutcomeReasonProcessingFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) request := requestWithState(entity.RequestStateFailed) request.Version = 3 diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 10edcf83c..1b6619c75 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -409,6 +409,7 @@ func (c *Controller) supersedeRequest(ctx context.Context, store storage.Storage updated := request updated.State = entity.RequestStateSuperseded + updated.OutcomeReason = entity.RequestOutcomeReasonSupersededByNewerHead newVersion := request.Version + 1 if err := reqStore.Update(ctx, updated, request.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index ef82e7b14..834d3ab44 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -772,6 +772,7 @@ func TestProcess(t *testing.T) { }, nil) updated := acceptedRequest(testOlderID) updated.State = entity.RequestStateSuperseded + updated.OutcomeReason = entity.RequestOutcomeReasonSupersededByNewerHead updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil) loggedRequest := updated loggedRequest.Version = 2 @@ -791,6 +792,7 @@ func TestProcess(t *testing.T) { }, nil) updated := acceptedRequest(testOlderID) updated.State = entity.RequestStateSuperseded + updated.OutcomeReason = entity.RequestOutcomeReasonSupersededByNewerHead updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil) request := entity.Request{ ID: testOlderID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2, @@ -814,6 +816,7 @@ func TestProcess(t *testing.T) { }, nil) updated := acceptedRequest(testOlderID) updated.State = entity.RequestStateSuperseded + updated.OutcomeReason = entity.RequestOutcomeReasonSupersededByNewerHead m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(storage.ErrVersionMismatch) reloaded := entity.Request{ ID: testOlderID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2, @@ -938,6 +941,7 @@ func TestDelayedRegistrationDoesNotAdmitAnOlderRequest(t *testing.T) { m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, LatestRequestID: "8", Version: 2}, nil) request := acceptedRequest(testID) request.State = entity.RequestStateSuperseded + request.OutcomeReason = entity.RequestOutcomeReasonSupersededByNewerHead m.reqStore.EXPECT().Update(gomock.Any(), request, int32(1), int32(2)).Return(nil) request.Version = 2 expectStateLog(t, m, request, entity.RequestOutcomeReasonSupersededByNewerHead) diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go index 4b4751858..9d77ff067 100644 --- a/stovepipe/core/requestlog/materializer.go +++ b/stovepipe/core/requestlog/materializer.go @@ -84,7 +84,13 @@ func NewMaterializer(scope tally.Scope) Materializer { } // NewRequestStateLog constructs the occurrence representing the request's current durable state. -func NewRequestStateLog(request entity.Request, outcomeReason entity.RequestOutcomeReason) entity.RequestLog { +// It records the request's persisted outcome reason. fallbackReason applies only when none is +// persisted: on a nonterminal request, or on one made terminal before reasons were persisted. +func NewRequestStateLog(request entity.Request, fallbackReason entity.RequestOutcomeReason) entity.RequestLog { + outcomeReason := request.OutcomeReason + if outcomeReason == entity.RequestOutcomeReasonUnknown { + outcomeReason = fallbackReason + } metadata := make(map[string]string) switch request.BuildStrategy { case entity.BuildStrategyFull: diff --git a/stovepipe/entity/request.go b/stovepipe/entity/request.go index f4e2fa451..dfc13fbd7 100644 --- a/stovepipe/entity/request.go +++ b/stovepipe/entity/request.go @@ -122,6 +122,9 @@ type Request struct { // 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"` + // OutcomeReason is the durable reason for the current terminal state. It is empty while + // the request is nonterminal and on requests made terminal before reasons were persisted. + OutcomeReason RequestOutcomeReason `json:"outcome_reason"` // Version is the version of the object. It is used for optimistic locking. // Versioning starts at 1 and is incremented for each change to the object. Version int32 `json:"version"` diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 85e5c4261..1a7d9b94b 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -40,6 +40,10 @@ Listing fixes the queue/request prefix and returns bounded pages in ascending by Create returns `ErrAlreadyExists` for a repeated entry. Listing fixes the queue/request/project prefix and returns bounded pages in ascending bytewise task/unit order. The zero cursor starts the range; other cursors must match the project and exclude their task/unit key. An empty page ends the range, not evidence that attribution recording is complete. There are no update or delete operations. +## Requests + +Existing MySQL deployments must add `outcome_reason VARCHAR(64) NOT NULL DEFAULT ''` to `request` before deploying its readers and writers. Empty reasons retain the legacy fallback; existing immutable logs are not rewritten. + ## Build issue observations `BuildIssueStore` accumulates observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Build and task IDs must be nonempty; empty unit IDs represent task-wide observations. diff --git a/stovepipe/extension/storage/mysql/request_store.go b/stovepipe/extension/storage/mysql/request_store.go index 5dc21ceb9..607c1b675 100644 --- a/stovepipe/extension/storage/mysql/request_store.go +++ b/stovepipe/extension/storage/mysql/request_store.go @@ -60,8 +60,8 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE return err } _, err = r.db.ExecContext(ctx, - `INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, build_id, version, metadata) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + `INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, build_id, outcome_reason, version, metadata) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, request.ID, request.Queue, request.URI, @@ -69,6 +69,7 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE request.BuildStrategy, request.BaseURI, request.BuildID, + request.OutcomeReason, request.Version, metadata, ) @@ -90,7 +91,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, var req entity.Request var metadata []byte err := r.db.QueryRowContext(ctx, - `SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE(terminal_build_id, ''), version, metadata + `SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE(terminal_build_id, ''), outcome_reason, version, metadata FROM request WHERE queue = ? AND id = ?`, r.queue, id, ).Scan( @@ -102,6 +103,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, &req.BaseURI, &req.BuildID, &req.TerminalBuildID, + &req.OutcomeReason, &req.Version, &metadata, ) @@ -121,7 +123,8 @@ 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, associated build ID, terminal build ID, metadata) if the oldVersion, writing newVersion. Returns +// base URI, associated and terminal build IDs, outcome reason, metadata) if the +// oldVersion matches, 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. @@ -139,7 +142,7 @@ 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 = ?, build_id = ?, terminal_build_id = NULLIF(?, ''), version = ?, metadata = ? + SET uri = ?, state = ?, build_strategy = ?, base_uri = ?, build_id = ?, terminal_build_id = NULLIF(?, ''), outcome_reason = ?, version = ?, metadata = ? WHERE queue = ? AND id = ? AND version = ?`, request.URI, request.State, @@ -147,6 +150,7 @@ func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVe request.BaseURI, request.BuildID, request.TerminalBuildID, + request.OutcomeReason, newVersion, metadata, request.Queue, diff --git a/stovepipe/extension/storage/mysql/request_store_test.go b/stovepipe/extension/storage/mysql/request_store_test.go index 4fea3f837..f087f3af6 100644 --- a/stovepipe/extension/storage/mysql/request_store_test.go +++ b/stovepipe/extension/storage/mysql/request_store_test.go @@ -63,7 +63,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.BuildID, request.Version, []byte("{}")). + WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.OutcomeReason, request.Version, []byte("{}")). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -71,7 +71,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.BuildID, request.Version, []byte("{}")). + WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.OutcomeReason, request.Version, []byte("{}")). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -81,7 +81,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.BuildID, request.Version, []byte("{}")). + WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.OutcomeReason, request.Version, []byte("{}")). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -114,11 +114,12 @@ func TestRequestStore_Get(t *testing.T) { ID: "1", Queue: "monorepo/main", URI: "git://remote/monorepo/main/deadbeef", - State: entity.RequestStateProcessing, + State: entity.RequestStateFailed, BuildStrategy: entity.BuildStrategyFull, BaseURI: "", BuildID: "bk-1001", TerminalBuildID: "bk-1001", + OutcomeReason: entity.RequestOutcomeReasonBuildFailed, Version: 2, Metadata: map[string]json.RawMessage{"extension": json.RawMessage(`{"thread":"123.456"}`)}, } @@ -135,9 +136,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", "build_id", "terminal_build_id", "version", "metadata"}). - AddRow(want.ID, want.Queue, want.URI, string(want.State), string(want.BuildStrategy), want.BaseURI, want.BuildID, want.TerminalBuildID, want.Version, []byte(`{"extension":{"thread":"123.456"}}`)) - mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), version, metadata"). + rows := sqlmock.NewRows([]string{"id", "queue", "uri", "state", "build_strategy", "base_uri", "build_id", "terminal_build_id", "outcome_reason", "version", "metadata"}). + AddRow(want.ID, want.Queue, want.URI, string(want.State), string(want.BuildStrategy), want.BaseURI, want.BuildID, want.TerminalBuildID, string(want.OutcomeReason), want.Version, []byte(`{"extension":{"thread":"123.456"}}`)) + mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), outcome_reason, version, metadata"). WithArgs("monorepo/main", want.ID). WillReturnRows(rows) }, @@ -147,7 +148,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, build_id, COALESCE\\(terminal_build_id, ''\\), version, metadata"). + mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), outcome_reason, version, metadata"). WithArgs("monorepo/main", "missing"). WillReturnError(sql.ErrNoRows) }, @@ -158,7 +159,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, build_id, COALESCE\\(terminal_build_id, ''\\), version, metadata"). + mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), outcome_reason, version, metadata"). WithArgs("monorepo/main", "bad"). WillReturnError(fmt.Errorf("connection reset")) }, @@ -193,11 +194,12 @@ func TestRequestStore_Update(t *testing.T) { ID: "1", Queue: "monorepo/main", URI: "git://remote/monorepo/main/deadbeef", - State: entity.RequestStateProcessing, + State: entity.RequestStateFailed, BuildStrategy: entity.BuildStrategyFull, BaseURI: "", BuildID: "bk-1001", TerminalBuildID: "bk-1001", + OutcomeReason: entity.RequestOutcomeReasonBuildFailed, Metadata: map[string]json.RawMessage{"extension": json.RawMessage(`{"thread":"123.456"}`)}, } const oldVersion, newVersion = int32(1), int32(2) @@ -212,7 +214,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.BuildID, request.TerminalBuildID, newVersion, []byte(`{"extension":{"thread":"123.456"}}`), request.Queue, request.ID, oldVersion). + WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, request.OutcomeReason, newVersion, []byte(`{"extension":{"thread":"123.456"}}`), request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -220,7 +222,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.BuildID, request.TerminalBuildID, newVersion, []byte(`{"extension":{"thread":"123.456"}}`), request.Queue, request.ID, oldVersion). + WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, request.OutcomeReason, newVersion, []byte(`{"extension":{"thread":"123.456"}}`), request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -230,7 +232,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.BuildID, request.TerminalBuildID, newVersion, []byte(`{"extension":{"thread":"123.456"}}`), request.Queue, request.ID, oldVersion). + WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, request.OutcomeReason, newVersion, []byte(`{"extension":{"thread":"123.456"}}`), request.Queue, request.ID, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -239,7 +241,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.BuildID, request.TerminalBuildID, newVersion, []byte(`{"extension":{"thread":"123.456"}}`), request.Queue, request.ID, oldVersion). + WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, request.OutcomeReason, newVersion, []byte(`{"extension":{"thread":"123.456"}}`), request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, @@ -323,8 +325,8 @@ func TestRequestMetadataRead(t *testing.T) { if scenario.readErr != nil { query.WillReturnError(scenario.readErr) } else { - query.WillReturnRows(sqlmock.NewRows([]string{"id", "queue", "uri", "state", "build_strategy", "base_uri", "build_id", "terminal_build_id", "version", "metadata"}). - AddRow("1", "monorepo/main", "", "accepted", "", "", "", "", 1, scenario.data)) + query.WillReturnRows(sqlmock.NewRows([]string{"id", "queue", "uri", "state", "build_strategy", "base_uri", "build_id", "terminal_build_id", "outcome_reason", "version", "metadata"}). + AddRow("1", "monorepo/main", "", "accepted", "", "", "", "", "", 1, scenario.data)) } got, err := store.Get(context.Background(), "1") if scenario.wantErr { diff --git a/stovepipe/extension/storage/mysql/schema/request.sql b/stovepipe/extension/storage/mysql/schema/request.sql index 3bce13415..4fce8bcf2 100644 --- a/stovepipe/extension/storage/mysql/schema/request.sql +++ b/stovepipe/extension/storage/mysql/schema/request.sql @@ -11,6 +11,7 @@ CREATE TABLE IF NOT EXISTS request ( version INT NOT NULL, terminal_build_id VARCHAR(255) NULL, build_id VARCHAR(255) NOT NULL DEFAULT '', + outcome_reason VARCHAR(64) NOT NULL DEFAULT '', metadata JSON NULL, PRIMARY KEY (queue, id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 086e4a7e8..10a1af340 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -378,6 +378,16 @@ func (s *MySQLRequestStoreSuite) TestUpdateCAS() { // Stale CAS: oldVersion 1 no longer matches the stored version (2). err = s.store.Update(s.ctx, updated, 1, 2) require.ErrorIs(s.T(), err, storage.ErrVersionMismatch) + + failed := updated + failed.State = entity.RequestStateFailed + failed.OutcomeReason = entity.RequestOutcomeReasonBuildPollingExhausted + require.NoError(s.T(), s.store.Update(s.ctx, failed, 2, 3)) + failed.Version = 3 + + got, err = s.store.Get(s.ctx, req.ID) + require.NoError(s.T(), err) + require.Equal(s.T(), failed, got) } func (s *MySQLRequestStoreSuite) TestUpdateNotFoundIsVersionMismatch() {