diff --git a/stovepipe/controller/projectanalysis/dlq.go b/stovepipe/controller/projectanalysis/dlq.go index 928cd8862..86e73c5d8 100644 --- a/stovepipe/controller/projectanalysis/dlq.go +++ b/stovepipe/controller/projectanalysis/dlq.go @@ -120,8 +120,11 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) if request.TerminalBuildID != "" { projectIssuesRecordedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil) - _, err := store.GetRequestLogStore().Get(ctx, request.ID, projectIssuesRecordedLog.ID) + retained, err := store.GetRequestLogStore().Get(ctx, request.ID, projectIssuesRecordedLog.ID) if err == nil { + if err := c.requestLogMaterializer.PersistLog(ctx, store, retained); err != nil { + return err + } return publishRecord(ctx, c.registry, request) } if !errors.Is(err, storage.ErrNotFound) { diff --git a/stovepipe/controller/projectanalysis/projectanalysis.go b/stovepipe/controller/projectanalysis/projectanalysis.go index a38bf627b..9bb127d8e 100644 --- a/stovepipe/controller/projectanalysis/projectanalysis.go +++ b/stovepipe/controller/projectanalysis/projectanalysis.go @@ -119,8 +119,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r } history := store.GetRequestLogStore() projectAnalysisAbandonedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil) - _, err = history.Get(ctx, request.ID, projectAnalysisAbandonedLog.ID) + retained, err := history.Get(ctx, request.ID, projectAnalysisAbandonedLog.ID) if err == nil { + if err := c.requestLogMaterializer.PersistLog(ctx, store, retained); err != nil { + return err + } return publishRecord(ctx, c.registry, request) } if !errors.Is(err, storage.ErrNotFound) { @@ -131,8 +134,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r } projectIssuesRecordedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil) - _, err = history.Get(ctx, request.ID, projectIssuesRecordedLog.ID) + retained, err = history.Get(ctx, request.ID, projectIssuesRecordedLog.ID) if err == nil { + if err := c.requestLogMaterializer.PersistLog(ctx, store, retained); err != nil { + return err + } return publishRecord(ctx, c.registry, request) } if !errors.Is(err, storage.ErrNotFound) { diff --git a/stovepipe/controller/projectanalysis/projectanalysis_test.go b/stovepipe/controller/projectanalysis/projectanalysis_test.go index 084051c4b..a63f2c80f 100644 --- a/stovepipe/controller/projectanalysis/projectanalysis_test.go +++ b/stovepipe/controller/projectanalysis/projectanalysis_test.go @@ -478,6 +478,46 @@ func TestProcessRejectsInvalidIdentity(t *testing.T) { } } +func TestCompletedAnalysisRepairsProjectionBeforeHandoff(t *testing.T) { + for _, recovery := range []bool{false, true} { + name := "normal delivery" + if recovery { + name = "dead-letter recovery" + } + t.Run(name, func(t *testing.T) { + f := newFixture(t) + msg := f.message(t) + f.failures["record/publish"] = assert.AnError + require.ErrorIs(t, f.deliver(context.Background(), msg), assert.AnError) + f.failures["log/project_issues_recorded"] = assert.AnError + deliver := f.deliver + if recovery { + deliver = f.recover + } + require.ErrorIs(t, deliver(context.Background(), msg), assert.AnError) + require.Empty(t, f.records) + require.Equal(t, 1, f.resolverCalls) + require.Equal(t, 1, f.attributerCalls) + require.NoError(t, deliver(context.Background(), msg)) + require.Len(t, f.records, 1) + }) + } +} + +func TestAbandonedAnalysisRepairsProjectionBeforeHandoff(t *testing.T) { + f := newFixture(t) + msg := f.message(t) + f.failures["record/publish"] = assert.AnError + require.ErrorIs(t, f.recover(context.Background(), msg), assert.AnError) + f.failures["log/project_analysis_abandoned"] = assert.AnError + require.ErrorIs(t, f.deliver(context.Background(), msg), assert.AnError) + require.Empty(t, f.records) + require.Zero(t, f.resolverCalls) + require.Zero(t, f.attributerCalls) + require.NoError(t, f.deliver(context.Background(), msg)) + require.Len(t, f.records, 1) +} + const projectStatusTestQueue = "monorepo/main" var projectStatusTestChangeTime = time.Unix(1_700_000_000, 0).UTC() diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go index 1034f84f1..ee0b1abb5 100644 --- a/stovepipe/core/requestlog/materializer.go +++ b/stovepipe/core/requestlog/materializer.go @@ -131,9 +131,7 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l return err } - if retained.State == entity.RequestStateUnknown && - retained.Event != entity.RequestEventProjectFactsRecorded && - retained.Event != entity.RequestEventRecordAbandoned { + if retained.State == entity.RequestStateUnknown && !completesProjectResults(retained.Event) { return nil } if err := materializeRequestProjections(ctx, stores, retained); err != nil { diff --git a/stovepipe/core/requestlog/request_summary.go b/stovepipe/core/requestlog/request_summary.go index ec6af16a3..3d6a51bda 100644 --- a/stovepipe/core/requestlog/request_summary.go +++ b/stovepipe/core/requestlog/request_summary.go @@ -117,7 +117,7 @@ func projectRequestSummaryFacts(summary entity.RequestSummary, log entity.Reques log.RequestID, summary.RequestID, log.Queue, summary.Queue, ) } - if log.Event == entity.RequestEventProjectFactsRecorded || log.Event == entity.RequestEventRecordAbandoned { + if completesProjectResults(log.Event) { if log.TimestampMs <= 0 { return entity.RequestSummary{}, fmt.Errorf("request summary completion time must be positive request_id=%q", log.RequestID) } @@ -200,3 +200,13 @@ func requestSummaryFromRequestAndLog(request entity.Request, log entity.RequestL } return summary } + +func completesProjectResults(event entity.RequestEvent) bool { + switch event { + case entity.RequestEventProjectFactsRecorded, entity.RequestEventProjectIssuesRecorded, + entity.RequestEventProjectAnalysisAbandoned, entity.RequestEventRecordAbandoned: + return true + default: + return false + } +} diff --git a/stovepipe/core/requestlog/request_summary_test.go b/stovepipe/core/requestlog/request_summary_test.go index 82535b424..f5e1d1b6c 100644 --- a/stovepipe/core/requestlog/request_summary_test.go +++ b/stovepipe/core/requestlog/request_summary_test.go @@ -524,6 +524,10 @@ func TestMaterializerProjectsProjectCompletion(t *testing.T) { {name: "duplicate completion", event: entity.RequestEventProjectFactsRecorded, completedAt: testNowMs, count: "3", projectCount: 3}, {name: "recording after abandonment", event: entity.RequestEventProjectFactsRecorded, completedAt: testNowMs - 1, count: "3", projectCount: 3}, {name: "delayed recording after abandonment", event: entity.RequestEventProjectFactsRecorded, completedAt: testNowMs + 1, count: "3", projectCount: 3}, + {name: "attribution completed", event: entity.RequestEventProjectIssuesRecorded}, + {name: "attribution completed after facts", event: entity.RequestEventProjectIssuesRecorded, completedAt: testNowMs - 1}, + {name: "project analysis abandoned", event: entity.RequestEventProjectAnalysisAbandoned}, + {name: "project analysis abandoned after facts", event: entity.RequestEventProjectAnalysisAbandoned, completedAt: testNowMs - 1}, {name: "abandoned recording", event: entity.RequestEventRecordAbandoned}, {name: "duplicate abandonment", event: entity.RequestEventRecordAbandoned, completedAt: testNowMs}, {name: "abandonment after recording", event: entity.RequestEventRecordAbandoned, completedAt: testNowMs - 1}, @@ -567,7 +571,7 @@ func TestMaterializerProjectsProjectCompletion(t *testing.T) { } func TestMaterializerRepairsCompletionAfterProjectionFailure(t *testing.T) { - for _, event := range []entity.RequestEvent{entity.RequestEventProjectFactsRecorded, entity.RequestEventRecordAbandoned} { + for _, event := range []entity.RequestEvent{entity.RequestEventProjectFactsRecorded, entity.RequestEventProjectIssuesRecorded, entity.RequestEventProjectAnalysisAbandoned, entity.RequestEventRecordAbandoned} { t.Run(string(event), func(t *testing.T) { materializer, stores, logs, summaries, _ := newTestMaterializer(t) log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, event, "test", map[string]string{MetadataKeyProjectFactCount: "3", MetadataKeyBrokenProjectFactCount: "1"}) @@ -606,7 +610,7 @@ func TestMaterializerRepairsCompletionAfterProjectionFailure(t *testing.T) { } func TestMaterializerRetriesCompletionAfterStateProjectionWins(t *testing.T) { - for _, event := range []entity.RequestEvent{entity.RequestEventProjectFactsRecorded, entity.RequestEventRecordAbandoned} { + for _, event := range []entity.RequestEvent{entity.RequestEventProjectFactsRecorded, entity.RequestEventProjectIssuesRecorded, entity.RequestEventProjectAnalysisAbandoned, entity.RequestEventRecordAbandoned} { t.Run(string(event), func(t *testing.T) { materializer, stores, logs, summaries, _ := newTestMaterializer(t) log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, event, "test", map[string]string{MetadataKeyProjectFactCount: "3", MetadataKeyBrokenProjectFactCount: "1"}) @@ -645,7 +649,7 @@ func TestMaterializerRetriesCompletionAfterStateProjectionWins(t *testing.T) { } func TestMaterializerCompletionRequiresExistingSummary(t *testing.T) { - for _, event := range []entity.RequestEvent{entity.RequestEventProjectFactsRecorded, entity.RequestEventRecordAbandoned} { + for _, event := range []entity.RequestEvent{entity.RequestEventProjectFactsRecorded, entity.RequestEventProjectIssuesRecorded, entity.RequestEventProjectAnalysisAbandoned, entity.RequestEventRecordAbandoned} { t.Run(string(event), func(t *testing.T) { materializer, stores, logs, summaries, _ := newTestMaterializer(t) log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, event, "test", nil) diff --git a/stovepipe/entity/request_summary.go b/stovepipe/entity/request_summary.go index 451521e6a..f2c0060e9 100644 --- a/stovepipe/entity/request_summary.go +++ b/stovepipe/entity/request_summary.go @@ -32,7 +32,8 @@ type RequestSummary struct { StateTimestampMs int64 // AcceptedAtMs is the original acceptance timestamp in Unix milliseconds; zero means unknown and a positive value is immutable. AcceptedAtMs int64 - // ProjectResultsCompletedAtMs is the completion timestamp in Unix milliseconds; zero means unknown and a positive value is immutable. + // ProjectResultsCompletedAtMs is when project-result processing finished or was abandoned, in Unix milliseconds. + // Zero means unknown and a positive value is immutable. ProjectResultsCompletedAtMs int64 // ProjectCount is the number of recorded project results; zero means none recorded. ProjectCount int32 diff --git a/test/e2e/stovepipe/project_issues_test.go b/test/e2e/stovepipe/project_issues_test.go index f4bbd90ac..de75d79e3 100644 --- a/test/e2e/stovepipe/project_issues_test.go +++ b/test/e2e/stovepipe/project_issues_test.go @@ -31,8 +31,10 @@ func (s *StovepipeE2ESuite) TestProjectIssues() { t := s.T() stores, err := s.appStorage.For(projectIssuesQueue) require.NoError(t, err) - request := entity.Request{ID: "900", Queue: projectIssuesQueue, TerminalBuildID: "build/900"} + request := entity.Request{ID: "900", Queue: projectIssuesQueue, URI: "git://repository/main/project-issues", State: entity.RequestStateFailed, Version: 1, TerminalBuildID: "build/900"} + require.NoError(t, stores.GetRequestStore().Create(s.ctx, request)) history := requestlog.NewMaterializer(tally.NoopScope) + require.NoError(t, history.PersistLog(s.ctx, stores, requestlog.NewRequestStateLog(request, entity.RequestOutcomeReasonBuildFailed))) metadata := map[string]string{requestlog.MetadataKeyBuildID: request.TerminalBuildID} require.NoError(t, history.PersistLog(s.ctx, stores, requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisCompleted, request.TerminalBuildID, metadata))) materializer := projectissuescore.NewMaterializer(tally.NoopScope)