From d42b7fd578146e8e4e73e09423257188a00ad929 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 Request stored its terminal state but not the reason, so buildsignal relabeled every failed request build_failed while the DLQ reconcilers logged their own stage reason. Two writers of one state/ log disagreed, the materializer rejected the second, and the DLQ retried that rejection forever. Store OutcomeReason on Request in the same compare-and-swap as the terminal state (buildsignal, the DLQ reconcilers, process supersede), and build every state log from the stored reason. Only rows made terminal before this change fall back to the writing stage's reason. Co-Authored-By: Claude Opus 5.5 (1M context) --- doc/rfc/stovepipe/request-log.md | 2 +- .../controller/buildsignal/buildsignal.go | 39 ++++++++++++------- .../buildsignal/buildsignal_test.go | 12 +++--- stovepipe/controller/dlq/build_test.go | 1 + stovepipe/controller/dlq/buildsignal_test.go | 1 + stovepipe/controller/dlq/dlq.go | 9 +++-- 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 | 19 +++++---- .../storage/mysql/request_store_test.go | 30 +++++++------- .../storage/mysql/schema/request.sql | 1 + .../extension/storage/mysql/storage_test.go | 10 +++++ 16 files changed, 111 insertions(+), 50 deletions(-) diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index a4b3dc2ea..48400455c 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -223,7 +223,7 @@ The `project_facts_recorded` event also projects its retained timestamp into `Re 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 c6d3e0f30..5adf101f2 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -255,7 +255,8 @@ func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, r return err } - if err := c.markOutcome(ctx, store, request, buildID, outcomeState(status)); err != nil { + state := outcomeState(status) + if err := c.markOutcome(ctx, store, request, buildID, state, buildOutcomeReason(state)); err != nil { metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) return err } @@ -263,20 +264,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,11 +294,25 @@ 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 +// buildOutcomeReason is the outcome reason a build-established state records. +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 + } +} + +// markOutcome CAS-transitions request from processing to state and reason, 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 (c *Controller) markOutcome(ctx context.Context, store storage.Storage, request *entity.Request, buildID string, state entity.RequestState) error { +func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, request *entity.Request, buildID string, state entity.RequestState, reason entity.RequestOutcomeReason) error { reqStore := store.GetRequestStore() for { @@ -315,6 +323,7 @@ func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, req updated := *request updated.State = state updated.TerminalBuildID = buildID + updated.OutcomeReason = reason 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 5731831d7..597cab9ee 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(), diff --git a/stovepipe/controller/dlq/build_test.go b/stovepipe/controller/dlq/build_test.go index c97686202..ad5772fc3 100644 --- a/stovepipe/controller/dlq/build_test.go +++ b/stovepipe/controller/dlq/build_test.go @@ -71,6 +71,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 08c0f4aab..62a316ad2 100644 --- a/stovepipe/controller/dlq/buildsignal_test.go +++ b/stovepipe/controller/dlq/buildsignal_test.go @@ -110,6 +110,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 cbde8e207..4b4f4154e 100644 --- a/stovepipe/controller/dlq/dlq.go +++ b/stovepipe/controller/dlq/dlq.go @@ -106,8 +106,8 @@ func failRequest( if request.State.IsTerminal() { if request.State == entity.RequestStateFailed { - // The originating DLQ remains the retry trigger after the state write. Its stage-specific - // reason repairs that write's missing log; PersistLog rejects an already-retained conflict. + // The originating DLQ remains the retry trigger after the state write. The repaired log + // carries the reason persisted with that write, so it agrees with any log its writer retained. return persistFailureLog(ctx, store, materializer, request, reason) } logger.Infow("dlq reconcile: request already terminal, skipping", @@ -125,6 +125,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) @@ -145,9 +146,9 @@ func persistFailureLog( store storage.Storage, materializer requestlog.Materializer, request entity.Request, - reason entity.RequestOutcomeReason, + fallbackReason entity.RequestOutcomeReason, ) error { - log := requestlog.NewRequestStateLog(request, reason) + log := requestlog.NewRequestStateLog(request, fallbackReason) if err := materializer.PersistLog(ctx, store, log); err != nil { return fmt.Errorf("failed to record failed state for request %s: %w", request.ID, err) } diff --git a/stovepipe/controller/dlq/dlq_test.go b/stovepipe/controller/dlq/dlq_test.go index eeb7c838c..8f536f67f 100644 --- a/stovepipe/controller/dlq/dlq_test.go +++ b/stovepipe/controller/dlq/dlq_test.go @@ -141,6 +141,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) }, @@ -157,6 +158,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) }, @@ -168,12 +170,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.RequestOutcomeReasonBuildFailed + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + expectFailureLog(m, entity.RequestOutcomeReasonBuildFailed, 2) + }, + }, { name: "request not found is a no-op", setup: func(m dlqMocks) { @@ -186,6 +197,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, @@ -208,6 +220,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) }, @@ -221,6 +234,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) }, @@ -232,6 +246,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 f334f70bd..53e58a279 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -456,6 +456,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 cd8ef4c61..63bbf0f02 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -801,6 +801,7 @@ func TestProcess(t *testing.T) { }, nil) superseded := acceptedRequest(testID) superseded.State = entity.RequestStateSuperseded + superseded.OutcomeReason = entity.RequestOutcomeReasonSupersededByNewerHead updateCall := m.reqStore.EXPECT().Update(gomock.Any(), superseded, int32(1), int32(2)).Return(nil) loggedRequest := superseded loggedRequest.Version = 2 @@ -906,6 +907,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 @@ -925,6 +927,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, @@ -948,6 +951,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, diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go index 2eb424ae7..bc1392785 100644 --- a/stovepipe/core/requestlog/materializer.go +++ b/stovepipe/core/requestlog/materializer.go @@ -68,7 +68,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 850e751d2..3886d3763 100644 --- a/stovepipe/entity/request.go +++ b/stovepipe/entity/request.go @@ -118,6 +118,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 4e0de28ea..d79b2ec33 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -26,6 +26,10 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte `ProjectFactByDegreeStore` is the ordered lookup mapping for named project facts. Its immutable `(queue, URI, degree DESC, project ASC)` primary key stores the original degree and orders by descending degree and ascending bytewise project ID, so listing is a bounded key-range read. It holds no duplicate validation payload; readers hydrate facts by their original keys and verify the mapping. The record controller ensures mappings from persisted winning facts before recording project completion, including retries after partial writes. Create the table before deploying writers. The table starts empty: only results processed by index-aware writers receive mappings, with no historical population or read-time fallback. +## Requests + +Existing MySQL deployments must add `outcome_reason VARCHAR(64) NOT NULL DEFAULT ''` to `request` before deploying the new readers and writers. + ## 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 e9a0e0194..a40464720 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, outcome_reason, version) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, + request.OutcomeReason, 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, COALESCE(terminal_build_id, ''), outcome_reason, version FROM request WHERE queue = ? AND id = ?`, r.queue, id, ).Scan( @@ -93,6 +94,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, &req.BuildStrategy, &req.BaseURI, &req.TerminalBuildID, + &req.OutcomeReason, &req.Version, ) @@ -107,10 +109,10 @@ 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 -// 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. +// base URI, terminal build ID, outcome reason) 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. func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVersion, newVersion int32) (retErr error) { op := metrics.Begin(r.scope, "update", metrics.StorageLatencyBuckets) defer func() { op.Complete(retErr) }() @@ -121,13 +123,14 @@ 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 = ?, terminal_build_id = NULLIF(?, ''), outcome_reason = ?, version = ? WHERE queue = ? AND id = ? AND version = ?`, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, + request.OutcomeReason, newVersion, request.Queue, request.ID, diff --git a/stovepipe/extension/storage/mysql/request_store_test.go b/stovepipe/extension/storage/mysql/request_store_test.go index 190ac279b..9e8065f19 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.OutcomeReason, 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.OutcomeReason, 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.OutcomeReason, request.Version). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -113,10 +113,11 @@ 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: "", TerminalBuildID: "bk-1001", + OutcomeReason: entity.RequestOutcomeReasonBuildFailed, 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", "terminal_build_id", "outcome_reason", "version"}). + AddRow(want.ID, want.Queue, want.URI, string(want.State), string(want.BuildStrategy), want.BaseURI, want.TerminalBuildID, string(want.OutcomeReason), want.Version) + mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE\\(terminal_build_id, ''\\), outcome_reason, 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, COALESCE\\(terminal_build_id, ''\\), outcome_reason, 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, COALESCE\\(terminal_build_id, ''\\), outcome_reason, version"). WithArgs("monorepo/main", "bad"). WillReturnError(fmt.Errorf("connection reset")) }, @@ -190,10 +191,11 @@ 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: "", TerminalBuildID: "bk-1001", + OutcomeReason: entity.RequestOutcomeReasonBuildFailed, } 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.TerminalBuildID, request.OutcomeReason, 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.TerminalBuildID, request.OutcomeReason, 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.TerminalBuildID, request.OutcomeReason, 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.TerminalBuildID, request.OutcomeReason, 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..0552b2a6c 100644 --- a/stovepipe/extension/storage/mysql/schema/request.sql +++ b/stovepipe/extension/storage/mysql/schema/request.sql @@ -10,5 +10,6 @@ CREATE TABLE IF NOT EXISTS request ( base_uri VARCHAR(255) NOT NULL DEFAULT '', version INT NOT NULL, terminal_build_id VARCHAR(255) NULL, + outcome_reason VARCHAR(64) NOT NULL DEFAULT '', 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 514bf5b06..6c20e9f6a 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -315,6 +315,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() {