feat(stovepipe): Share project completion tracking

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) {

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) {

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()

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 {

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
+	}
+}

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)

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

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)