diff --git a/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index 8bfa58cef..a0a35716c 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -1,6 +1,6 @@ # Record stage -`record` turns a terminal build outcome into a durable validation fact. +`record` turns a terminal build outcome into durable validation facts and per-request issue/project associations. - Phase 1 records whole-repository greenness and optional project facts. On green it promotes the commit onto the Queue's promotion ref if the Request still holds the last-green bookmark, then announces the outcome as a hook event. This is implemented: see [stovepipe/controller/record/record.go](../../../../stovepipe/controller/record/record.go). - Phase 2 records greenness per project instead of per repository. Sketched here, to be expanded before implementation. @@ -9,7 +9,7 @@ Notifying downstream systems rides the cross-domain hook framework rather than a See [workflow.md](../workflow.md) for the whole pipeline, [build.md](build.md) for how builds are created, and [buildsignal.md](buildsignal.md) for the terminal-only handoff into this stage. -`record` owns the validation facts and the promotion ref. `buildsignal` advances the last-green build baseline before handing off to this stage. `record` does not decide build scope, poll a build, release the Queue's build slot, interpret a target graph, or map targets to projects. Those belong to `process`, `buildsignal`, and `analyze`. +`record` owns validation facts, issue/project attribution recording, and the promotion ref. `buildsignal` advances the last-green build baseline before handing off to this stage. Repository-specific project discovery and dependency traversal belong to the project-status and issue-attribution extensions, not the controller. ## Phase 1 algorithm @@ -39,8 +39,11 @@ For a delivery carrying a `Record` payload: - superseded -> ack, no fact. Unreachable in practice. - accepted / processing / anything else -> return a non-retryable invariant error. -6. Resolve optional project results using R and its TerminalBuildID, and validate - the entire result set before writing facts. A resolver error returns raw. +6. Record issue/project attribution after terminal issue analysis completes + (see "Issue/project attribution"). While analysis is pending, hold the delivery + for 30 seconds without consuming its retry budget. + Then resolve optional project results using R and its TerminalBuildID, and + validate the entire result set before writing facts. Extension errors return raw. 7. Map R.State and the project results to a whole-repository degree and create the ValidationFact keyed by (R.URI, empty project) in the queue-bound fact store. @@ -75,6 +78,16 @@ Every decision after step 7 uses the persisted fact, not the outcome read from t Project-status processing delays fact creation and promotion, while the build baseline is already available from `buildsignal`. All fact writes and history complete before promotion and the recorded notification. The fact writes, bookmark advancement, and ref promotion remain separate operations. +## Issue/project attribution + +The record stage calls the queue's issue attributer with the request. The attributer loads the confirmed issues from its terminal build and determines their affected projects. Recording waits for that build's `build_analysis_completed` occurrence, because the request becomes terminal before final issue analysis necessarily finishes. Completion for a different build does not release the wait. Abandoned analysis returns an error instead of waiting indefinitely; the existing record DLQ path handles recovery. + +Each returned issue/project pair goes through the attribution materializer, which retains `request_projects_by_issue` before creating `request_issues_by_project`. Only after every pair is durable in both directions does recording retain `project_issues_recorded`, identified by the terminal build and carrying its build ID. An empty successful result also records completion, without association rows. An extension error discards all returned results. + +Replays skip attribution when completion is already retained. Before completion, retries may recompute attribution and replay any already-written pairs; attributers must return stable results for the same request. A failed pair or completion write returns an error and cannot mark the request's attribution complete. These are historical per-request associations; closing an issue later does not remove them. + +Cancelled and superseded requests do not run attribution. A failed request without a recorded CI outcome has no terminal build to attribute and skips this step. Record DLQ recovery does not compute attribution or claim it completed; any incomplete association rows remain incomplete, with `record_abandoned` identifying the stopped recording attempt. + ## Validation facts A fact answers "how broken was this scope at this Queue URI?" Its identity is `(queue, uri, project)`, where `project` is empty for whole-repository greenness and a stable project id in Phase 2. diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 27a199acd..16803be6e 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -41,6 +41,7 @@ go_library( "//stovepipe/extension/buildrunner/fake:go_default_library", "//stovepipe/extension/fixer:go_default_library", "//stovepipe/extension/fixer/noop:go_default_library", + "//stovepipe/extension/projectissues/noop:go_default_library", "//stovepipe/extension/projectstatus/noop:go_default_library", "//stovepipe/extension/queueconfig/default:go_default_library", "//stovepipe/extension/sourcecontrol:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 9d388c2df..ffe9bdf69 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -63,6 +63,7 @@ import ( buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake" "github.com/uber/submitqueue/stovepipe/extension/fixer" fixernoop "github.com/uber/submitqueue/stovepipe/extension/fixer/noop" + projectissuesnoop "github.com/uber/submitqueue/stovepipe/extension/projectissues/noop" projectstatusnoop "github.com/uber/submitqueue/stovepipe/extension/projectstatus/noop" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" @@ -435,7 +436,7 @@ func registerPrimaryControllers( } count++ - recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.New(), sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record") + recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.New(), projectissuesnoop.New(), sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record") if err := c.Register(recordController); err != nil { return count, fmt.Errorf("failed to register record controller: %w", err) } diff --git a/stovepipe/controller/record/BUILD.bazel b/stovepipe/controller/record/BUILD.bazel index 1f4ee39f6..1062836e9 100644 --- a/stovepipe/controller/record/BUILD.bazel +++ b/stovepipe/controller/record/BUILD.bazel @@ -18,8 +18,10 @@ go_library( "//stovepipe/core/hookevent:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/core/projectissues:go_default_library", "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", + "//stovepipe/extension/projectissues:go_default_library", "//stovepipe/extension/projectstatus:go_default_library", "//stovepipe/extension/sourcecontrol:go_default_library", "//stovepipe/extension/storage:go_default_library", @@ -48,6 +50,9 @@ go_test( "//stovepipe/core/requestlog:go_default_library", "//stovepipe/core/requestlog/mock:go_default_library", "//stovepipe/entity:go_default_library", + "//stovepipe/extension/projectissues:go_default_library", + "//stovepipe/extension/projectissues/mock:go_default_library", + "//stovepipe/extension/projectissues/noop:go_default_library", "//stovepipe/extension/projectstatus:go_default_library", "//stovepipe/extension/projectstatus/mock:go_default_library", "//stovepipe/extension/projectstatus/noop:go_default_library", diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 55d31af43..a891d4084 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -43,8 +43,10 @@ import ( "github.com/uber/submitqueue/stovepipe/core/hookevent" "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + projectissuescore "github.com/uber/submitqueue/stovepipe/core/projectissues" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/projectissues" "github.com/uber/submitqueue/stovepipe/extension/projectstatus" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" "github.com/uber/submitqueue/stovepipe/extension/storage" @@ -55,21 +57,26 @@ import ( // when that fact is green and holds the queue's last-green bookmark promotes // the commit. Implements consumer.Controller. type Controller struct { - logger *zap.SugaredLogger - metricsScope tally.Scope - materializer requestlog.Materializer - registry consumer.TopicRegistry - stores storage.Factory - projectStatusResolverFactory projectstatus.ResolverFactory - sourceControl sourcecontrol.Factory - topicKey consumer.TopicKey - consumerGroup string + logger *zap.SugaredLogger + metricsScope tally.Scope + materializer requestlog.Materializer + registry consumer.TopicRegistry + stores storage.Factory + projectStatusResolverFactory projectstatus.ResolverFactory + projectIssuesAttributerFactory projectissues.AttributerFactory + projectIssuesMaterializer projectissuescore.Materializer + sourceControl sourcecontrol.Factory + topicKey consumer.TopicKey + consumerGroup string } // Verify Controller implements consumer.Controller interface at compile time. var _ consumer.Controller = (*Controller)(nil) -const _opName = "record" +const ( + _opName = "record" + _analysisWaitDelayMs int64 = 30000 +) // wholeRepositoryProject is the project component of a fact covering the whole // repository rather than one project within it. @@ -82,6 +89,7 @@ func NewController( stores storage.Factory, materializer requestlog.Materializer, projectStatusResolverFactory projectstatus.ResolverFactory, + projectIssuesAttributerFactory projectissues.AttributerFactory, sourceControl sourcecontrol.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, @@ -89,15 +97,17 @@ func NewController( ) *Controller { name := string(topicKey) + "_controller" return &Controller{ - logger: logger.Named(name), - metricsScope: scope.SubScope(name), - materializer: materializer, - registry: registry, - stores: stores, - projectStatusResolverFactory: projectStatusResolverFactory, - sourceControl: sourceControl, - topicKey: topicKey, - consumerGroup: consumerGroup, + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + materializer: materializer, + registry: registry, + stores: stores, + projectStatusResolverFactory: projectStatusResolverFactory, + projectIssuesAttributerFactory: projectIssuesAttributerFactory, + projectIssuesMaterializer: projectissuescore.NewMaterializer(scope), + sourceControl: sourceControl, + topicKey: topicKey, + consumerGroup: consumerGroup, } } @@ -143,6 +153,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er switch request.State { case entity.RequestStateSucceeded, entity.RequestStateFailed: + complete, err := c.recordProjectIssues(ctx, store, request) + if err != nil { + return err + } + if !complete { + delivery.Hold(_analysisWaitDelayMs) + return nil + } results, err := c.resolveProjectResults(ctx, request) if err != nil { return err @@ -183,6 +201,70 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } } +func (c *Controller) recordProjectIssues(ctx context.Context, store storage.Storage, request entity.Request) (complete bool, retErr error) { + op := metrics.Begin(c.metricsScope, "record_project_issues", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + if request.TerminalBuildID == "" { + return true, nil + } + history := store.GetRequestLogStore() + completion := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, + map[string]string{requestlog.MetadataKeyBuildID: request.TerminalBuildID}) + _, err := history.Get(ctx, request.ID, completion.ID) + if err == nil { + return true, nil + } + if !errors.Is(err, storage.ErrNotFound) { + return false, fmt.Errorf("read project issue completion: %w", err) + } + + // The terminal request can be visible before analysis has saved its final issues. + analysis := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisCompleted, request.TerminalBuildID, nil) + _, err = history.Get(ctx, request.ID, analysis.ID) + if errors.Is(err, storage.ErrNotFound) { + abandoned := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisAbandoned, request.TerminalBuildID, nil) + _, err = history.Get(ctx, request.ID, abandoned.ID) + if errors.Is(err, storage.ErrNotFound) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("read abandoned build analysis: %w", err) + } + return false, fmt.Errorf("cannot attribute issues for build %q: build analysis was abandoned", request.TerminalBuildID) + } + if err != nil { + return false, fmt.Errorf("read completed build analysis: %w", err) + } + + attributer, err := c.projectIssuesAttributerFactory.For(projectissues.AttributerConfig{QueueName: request.Queue}) + if err != nil { + return false, fmt.Errorf("resolve project issue attributer for queue %q: %w", request.Queue, err) + } + attributions, err := attributer.Attribute(ctx, request) + if err != nil { + return false, fmt.Errorf("attribute issues for request %q: %w", request.ID, err) + } + for _, attribution := range attributions { + if attribution.TaskID == "" { + return false, fmt.Errorf("project issue attribution has an empty task ID") + } + if attribution.UnitID == "" { + return false, fmt.Errorf("project issue attribution requires a unit-level issue") + } + for _, project := range attribution.Projects { + entry := entity.ProjectIssueAttribution{TaskID: attribution.TaskID, UnitID: attribution.UnitID, Project: project} + if err := c.projectIssuesMaterializer.PersistAttribution(ctx, store, request.ID, entry); err != nil { + return false, err + } + } + } + if err := c.materializer.PersistLog(ctx, store, completion); err != nil { + return false, fmt.Errorf("record project issue completion: %w", err) + } + return true, nil +} + func (c *Controller) resolveProjectResults(ctx context.Context, request entity.Request) (results []projectstatus.Result, retErr error) { op := metrics.Begin(c.metricsScope, "resolve_project_results", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) defer func() { op.Complete(retErr) }() diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 1e99f656b..8355a9bd2 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -35,6 +35,9 @@ import ( "github.com/uber/submitqueue/stovepipe/core/requestlog" requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock" "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/projectissues" + projectissuesmock "github.com/uber/submitqueue/stovepipe/extension/projectissues/mock" + projectissuesnoop "github.com/uber/submitqueue/stovepipe/extension/projectissues/noop" "github.com/uber/submitqueue/stovepipe/extension/projectstatus" projectstatusmock "github.com/uber/submitqueue/stovepipe/extension/projectstatus/mock" projectstatusnoop "github.com/uber/submitqueue/stovepipe/extension/projectstatus/noop" @@ -71,15 +74,16 @@ var testChangeTime = time.Unix(1_700_000_000, 0).UTC() // recordMocks bundles the mocks a record controller test case wires // expectations on. type recordMocks struct { - reqStore *storagemock.MockRequestStore - queueStore *storagemock.MockQueueStore - factStore *storagemock.MockValidationFactStore - indexStore *storagemock.MockProjectFactByDegreeStore - store *storagemock.MockStorage - materializer *requestlogmock.MockMaterializer - sourceControl *sourcecontrolmock.MockSourceControl - hooks *hookRecorder - metricsScope tally.TestScope + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore + factStore *storagemock.MockValidationFactStore + indexStore *storagemock.MockProjectFactByDegreeStore + store *storagemock.MockStorage + materializer *requestlogmock.MockMaterializer + sourceControl *sourcecontrolmock.MockSourceControl + hooks *hookRecorder + metricsScope tally.TestScope + requestHistory map[string]entity.RequestLog } // hookRecorder stands in for the hook topic, decoding whatever the controller @@ -146,28 +150,41 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu scope := tally.NewTestScope("", nil) m := recordMocks{ - reqStore: storagemock.NewMockRequestStore(ctrl), - queueStore: storagemock.NewMockQueueStore(ctrl), - factStore: storagemock.NewMockValidationFactStore(ctrl), - indexStore: storagemock.NewMockProjectFactByDegreeStore(ctrl), - store: storagemock.NewMockStorage(ctrl), - materializer: requestlogmock.NewMockMaterializer(ctrl), - sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), - hooks: &hookRecorder{}, - metricsScope: scope, + reqStore: storagemock.NewMockRequestStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + factStore: storagemock.NewMockValidationFactStore(ctrl), + indexStore: storagemock.NewMockProjectFactByDegreeStore(ctrl), + store: storagemock.NewMockStorage(ctrl), + materializer: requestlogmock.NewMockMaterializer(ctrl), + sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), + hooks: &hookRecorder{}, + metricsScope: scope, + requestHistory: make(map[string]entity.RequestLog), } + completion := requestlog.NewRequestEventLog(requestWithState(entity.RequestStateFailed), entity.RequestEventProjectIssuesRecorded, "bk-1", nil) + m.requestHistory[completion.ID] = completion + history := storagemock.NewMockRequestLogStore(ctrl) + m.store.EXPECT().GetRequestLogStore().Return(history).AnyTimes() + history.EXPECT().Get(gomock.Any(), testID, gomock.Any()).DoAndReturn(func(_ context.Context, _ string, logID string) (entity.RequestLog, error) { + log, ok := m.requestHistory[logID] + if !ok { + return entity.RequestLog{}, storage.ErrNotFound + } + return log, nil + }).AnyTimes() m.store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() m.store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() m.store.EXPECT().GetValidationFactStore().Return(m.factStore).AnyTimes() m.store.EXPECT().GetProjectFactByDegreeStore().Return(m.indexStore).AnyTimes() m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { - assert.Contains(t, []entity.RequestEvent{entity.RequestEventValidationFactRecorded, entity.RequestEventProjectFactsRecorded}, log.Event) + assert.Contains(t, []entity.RequestEvent{entity.RequestEventValidationFactRecorded, entity.RequestEventProjectFactsRecorded, entity.RequestEventProjectIssuesRecorded}, log.Event) assert.Empty(t, log.State) assert.Zero(t, log.RequestVersion) assert.Empty(t, log.OutcomeReason) assert.NotEmpty(t, log.Metadata) + m.requestHistory[log.ID] = log return nil }, ).AnyTimes() @@ -202,6 +219,7 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu staticStorageFactory{store: m.store}, m.materializer, projectstatusnoop.New(), + projectissuesnoop.New(), staticSourceControlFactory{sourceControl: m.sourceControl}, registry, topicKey, @@ -1504,6 +1522,252 @@ func TestProcessRetriesProjectPersistenceWithImmutablePartialFact(t *testing.T) assert.Len(t, m.hooks.events, 1) } +func prepareProjectIssueRecording(m recordMocks, request entity.Request) entity.RequestLog { + completion := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil) + delete(m.requestHistory, completion.ID) + analysis := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisCompleted, request.TerminalBuildID, nil) + m.requestHistory[analysis.ID] = analysis + return completion +} + +func TestProcessWaitsForTerminalIssueAnalysis(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := requestWithState(entity.RequestStateFailed) + clear(m.requestHistory) + otherBuild := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisCompleted, "other-build", nil) + m.requestHistory[otherBuild.ID] = otherBuild + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + d := delivery(t, ctrl, recordPayload(t, testID)) + d.EXPECT().Hold(_analysisWaitDelayMs) + + require.NoError(t, c.Process(queueContext(), d)) + assert.Empty(t, m.hooks.events) +} + +func TestProcessRecordsProjectIssues(t *testing.T) { + for _, tt := range []struct { + name string + attributions []projectissues.Attribution + pairs []entity.ProjectIssueAttribution + }{ + {name: "no issues"}, + {name: "issue affects no projects", attributions: []projectissues.Attribution{{TaskID: "test", UnitID: "//app:test"}}}, + { + name: "multiple issues and projects", + attributions: []projectissues.Attribution{ + {TaskID: "test", UnitID: "//app:test", Projects: []string{"app", "dependent"}}, + {TaskID: "lint", UnitID: "//lib:lint", Projects: []string{"dependent"}}, + }, + pairs: []entity.ProjectIssueAttribution{ + {TaskID: "test", UnitID: "//app:test", Project: "app"}, + {TaskID: "test", UnitID: "//app:test", Project: "dependent"}, + {TaskID: "lint", UnitID: "//lib:lint", Project: "dependent"}, + }, + }, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := requestWithState(entity.RequestStateFailed) + request.BuildID = "superseded-build" + completion := prepareProjectIssueRecording(m, request) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + factory := projectissuesmock.NewMockAttributerFactory(ctrl) + attributer := projectissuesmock.NewMockAttributer(ctrl) + c.projectIssuesAttributerFactory = factory + factory.EXPECT().For(projectissues.AttributerConfig{QueueName: request.Queue}).Return(attributer, nil) + attributer.EXPECT().Attribute(gomock.Any(), request).Return(tt.attributions, nil) + source := storagemock.NewMockRequestProjectsByIssueStore(ctrl) + projection := storagemock.NewMockRequestIssuesByProjectStore(ctrl) + m.store.EXPECT().GetRequestProjectsByIssueStore().Return(source).AnyTimes() + m.store.EXPECT().GetRequestIssuesByProjectStore().Return(projection).AnyTimes() + var calls []any + for _, entry := range tt.pairs { + calls = append(calls, + source.EXPECT().Create(gomock.Any(), request.ID, entry).Return(nil), + projection.EXPECT().Create(gomock.Any(), request.ID, entry).DoAndReturn( + func(context.Context, string, entity.ProjectIssueAttribution) error { + assert.NotContains(t, m.requestHistory, completion.ID) + return nil + }), + ) + } + gomock.InOrder(calls...) + var fact entity.ValidationFact + m.expectFactCreated(&fact) + + require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) + require.Contains(t, m.requestHistory, completion.ID) + recorded := m.requestHistory[completion.ID] + assert.Equal(t, entity.RequestEventProjectIssuesRecorded, recorded.Event) + assert.Equal(t, request.Queue, recorded.Queue) + assert.Equal(t, request.ID, recorded.RequestID) + assert.Equal(t, map[string]string{requestlog.MetadataKeyBuildID: request.TerminalBuildID}, recorded.Metadata) + assert.Len(t, m.hooks.events, 1) + complete, err := c.recordProjectIssues(queueContext(), m.store, request) + require.NoError(t, err) + assert.True(t, complete) + }) + } +} + +func TestRecordProjectIssuesRetriesIncompleteWrites(t *testing.T) { + for _, failProjection := range []bool{true, false} { + name := "completion failure" + if failProjection { + name = "projection failure" + } + t.Run(name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := requestWithState(entity.RequestStateFailed) + completion := prepareProjectIssueRecording(m, request) + factory := projectissuesmock.NewMockAttributerFactory(ctrl) + attributer := projectissuesmock.NewMockAttributer(ctrl) + c.projectIssuesAttributerFactory = factory + factory.EXPECT().For(projectissues.AttributerConfig{QueueName: request.Queue}).Return(attributer, nil).Times(2) + attributer.EXPECT().Attribute(gomock.Any(), request).Return( + []projectissues.Attribution{{TaskID: "test", UnitID: "//app:test", Projects: []string{"app"}}}, nil, + ).Times(2) + entry := entity.ProjectIssueAttribution{TaskID: "test", UnitID: "//app:test", Project: "app"} + source := storagemock.NewMockRequestProjectsByIssueStore(ctrl) + projection := storagemock.NewMockRequestIssuesByProjectStore(ctrl) + m.store.EXPECT().GetRequestProjectsByIssueStore().Return(source).AnyTimes() + m.store.EXPECT().GetRequestIssuesByProjectStore().Return(projection).AnyTimes() + var projectionErr, retryProjectionErr error + if failProjection { + projectionErr = assert.AnError + } else { + retryProjectionErr = storage.ErrAlreadyExists + logs := requestlogmock.NewMockMaterializer(ctrl) + c.materializer = logs + gomock.InOrder( + logs.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(assert.AnError), + logs.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + m.requestHistory[log.ID] = log + return nil + }), + ) + } + gomock.InOrder( + source.EXPECT().Create(gomock.Any(), request.ID, entry).Return(nil), + projection.EXPECT().Create(gomock.Any(), request.ID, entry).Return(projectionErr), + source.EXPECT().Create(gomock.Any(), request.ID, entry).Return(storage.ErrAlreadyExists), + projection.EXPECT().Create(gomock.Any(), request.ID, entry).Return(retryProjectionErr), + ) + + complete, err := c.recordProjectIssues(queueContext(), m.store, request) + require.ErrorIs(t, err, assert.AnError) + assert.False(t, complete) + assert.NotContains(t, m.requestHistory, completion.ID) + complete, err = c.recordProjectIssues(queueContext(), m.store, request) + require.NoError(t, err) + assert.True(t, complete) + assert.Contains(t, m.requestHistory, completion.ID) + }) + } +} + +func TestRecordProjectIssuesRejectsUnusableResults(t *testing.T) { + for _, tt := range []struct { + name string + factoryErr error + attributeErr error + attributions []projectissues.Attribution + }{ + {name: "factory failure", factoryErr: assert.AnError}, + {name: "partial results with error", attributeErr: assert.AnError, + attributions: []projectissues.Attribution{{TaskID: "test", UnitID: "//app:test", Projects: []string{"app"}}}}, + {name: "empty task", attributions: []projectissues.Attribution{{UnitID: "//app:test"}}}, + {name: "task wide issue", attributions: []projectissues.Attribution{{TaskID: "test"}}}, + {name: "empty project", attributions: []projectissues.Attribution{{TaskID: "test", UnitID: "//app:test", Projects: []string{""}}}}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := requestWithState(entity.RequestStateFailed) + completion := prepareProjectIssueRecording(m, request) + factory := projectissuesmock.NewMockAttributerFactory(ctrl) + attributer := projectissuesmock.NewMockAttributer(ctrl) + c.projectIssuesAttributerFactory = factory + factory.EXPECT().For(projectissues.AttributerConfig{QueueName: request.Queue}).Return(attributer, tt.factoryErr) + if tt.factoryErr == nil { + attributer.EXPECT().Attribute(gomock.Any(), request).Return(tt.attributions, tt.attributeErr) + } + + complete, err := c.recordProjectIssues(queueContext(), m.store, request) + require.Error(t, err) + assert.False(t, complete) + assert.NotContains(t, m.requestHistory, completion.ID) + }) + } +} + +func TestRecordProjectIssuesAnalysisAbandoned(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := requestWithState(entity.RequestStateFailed) + clear(m.requestHistory) + abandoned := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisAbandoned, request.TerminalBuildID, nil) + m.requestHistory[abandoned.ID] = abandoned + + complete, err := c.recordProjectIssues(queueContext(), m.store, request) + require.Error(t, err) + assert.False(t, complete) +} + +func TestRecordProjectIssuesWithoutBuildOutcome(t *testing.T) { + ctrl := gomock.NewController(t) + c, _ := newController(t, ctrl) + store := storagemock.NewMockStorage(ctrl) + request := requestWithState(entity.RequestStateFailed) + request.TerminalBuildID = "" + + complete, err := c.recordProjectIssues(queueContext(), store, request) + require.NoError(t, err) + assert.True(t, complete) +} + +func TestRecordProjectIssuesHistoryFailures(t *testing.T) { + for _, event := range []entity.RequestEvent{ + entity.RequestEventProjectIssuesRecorded, + entity.RequestEventBuildAnalysisCompleted, + entity.RequestEventBuildAnalysisAbandoned, + } { + t.Run(string(event), func(t *testing.T) { + ctrl := gomock.NewController(t) + c, _ := newController(t, ctrl) + store := storagemock.NewMockStorage(ctrl) + history := storagemock.NewMockRequestLogStore(ctrl) + store.EXPECT().GetRequestLogStore().Return(history) + request := requestWithState(entity.RequestStateFailed) + var calls []any + for _, step := range []entity.RequestEvent{ + entity.RequestEventProjectIssuesRecorded, + entity.RequestEventBuildAnalysisCompleted, + entity.RequestEventBuildAnalysisAbandoned, + } { + log := requestlog.NewRequestEventLog(request, step, request.TerminalBuildID, nil) + err := storage.ErrNotFound + if step == event { + err = assert.AnError + } + calls = append(calls, history.EXPECT().Get(gomock.Any(), request.ID, log.ID).Return(entity.RequestLog{}, err)) + if step == event { + break + } + } + gomock.InOrder(calls...) + + complete, err := c.recordProjectIssues(queueContext(), store, request) + require.ErrorIs(t, err, assert.AnError) + assert.False(t, complete) + }) + } +} + func TestPersistProjectFactsCountsPersistedOutcomes(t *testing.T) { for _, tt := range []struct { name string diff --git a/stovepipe/extension/projectissues/README.md b/stovepipe/extension/projectissues/README.md index 86e1a4d5c..70ba8a11f 100644 --- a/stovepipe/extension/projectissues/README.md +++ b/stovepipe/extension/projectissues/README.md @@ -14,4 +14,4 @@ A completed assessment returns all results in one call, with one entry per disti Attribution is all-or-nothing: any task-wide failure, incomplete validation evidence, or dependency failure returns an error for the entire validation. Callers must discard results returned alongside an error. -Implementations must support concurrent calls, leave inputs unchanged, and not consume validation evidence. +Implementations must support concurrent calls, leave inputs unchanged, and not consume validation evidence. Recording can repeat attribution before completion is retained, so retries for the same request must return the same associations. diff --git a/stovepipe/extension/projectissues/projectissues.go b/stovepipe/extension/projectissues/projectissues.go index 2b6e909e1..0f8b73deb 100644 --- a/stovepipe/extension/projectissues/projectissues.go +++ b/stovepipe/extension/projectissues/projectissues.go @@ -36,9 +36,9 @@ type Attribution struct { // Attributer attributes confirmed build issues within the request's validation scope. // Return an error for incomplete attribution, including any task-wide failure. -// Returning no results is valid. +// Returning no results is valid; discard any results returned with an error. +// Results must be stable across retries for the same request. // Calls must be concurrency-safe and leave inputs unchanged. -// Results returned with an error must not be used. type Attributer interface { Attribute(ctx context.Context, request entity.Request) ([]Attribution, error) }