diff --git a/doc/rfc/stovepipe/project-status-api.md b/doc/rfc/stovepipe/project-status-api.md index 94088c81f..30d13632a 100644 --- a/doc/rfc/stovepipe/project-status-api.md +++ b/doc/rfc/stovepipe/project-status-api.md @@ -41,7 +41,7 @@ Repository and project breakage degrees are independent projections. A degree is | Request identity, baseline, and lifecycle | Durable request summary | | Repository breakage degree | Repository validation fact | | Project results | Implementation-defined project-result records | -| Project completion and pagination | Project-result completion record and cursor | +| Project completion and pagination | Completion timestamp in the durable request summary and cursor | ## Request Selection @@ -72,7 +72,7 @@ Callers use `project_results_complete`, rather than the presence or absence of a ## Implementation -Both endpoints return durable lifecycle summaries, the whole-repository fact when present, and recorded facts for requested projects in paginated request order, or recorded project results by descending breakage degree then ascending project ID when none are requested. `project_results_complete` remains false until a durable completion projection is available. Missing facts are omitted; an explicitly recorded zero degree remains present. `updated_at_ms` is the maximum timestamp of the summary and the facts represented on the page. +Both endpoints return durable lifecycle summaries, the whole-repository fact when present, and recorded facts for requested projects in paginated request order, or recorded project results by descending breakage degree then ascending project ID when none are requested. `project_results_complete` is true when the durable request summary has a positive `project_results_completed_at_ms`, projected from the retained `project_facts_recorded` event. The record stage writes that event after all project facts and ordering mappings, including a successfully resolved empty set, and before publishing its terminal hook. The materializer persists the completion timestamp before the record stage continues, without changing lifecycle state or request version. Duplicate delivery repairs a failed projection using the original event timestamp, and subsequent state projections preserve completion. The API reads the summary before loading facts and includes its completion timestamp in `updated_at_ms`, without an additional event lookup. Missing completion remains false. Missing facts are omitted; an explicitly recorded zero degree remains present. `updated_at_ms` is the maximum timestamp of the summary and the facts represented on the page. A dedicated ordering mapping makes listing a bounded primary-key range read, not a sort over all facts. Listing reads at most `page_size + 1` ordering mappings and hydrates only the returned facts, plus the cursor fact on continuations. The continuation token identifies the last project; its immutable fact supplies the degree needed to resume this ordering. diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index 83d7c04d5..a16820a3b 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -206,6 +206,8 @@ For a Request transition, the controller: Request creation, Build changes, and fact creation use the same source-write, log-write, dependent-publish ordering. A request-log outage can leave a source update visible, but it cannot allow dependent processing to move past an unrecorded transition. +The `project_facts_recorded` event also projects its retained timestamp into `RequestSummary.ProjectResultsCompletedAtMs`. This immutable fact is independent of lifecycle state and request version; state updates preserve it, and optimistic locking retries concurrent projection updates. Completion requires an existing lifecycle summary rather than fabricating state from the event. A failed projection fails persistence, and redelivery reconciles the retained occurrence before retrying the summary update. Other non-state events do not update the summary. + ### Controller ownership | Owner | Source operation and retained log | Retry repair | diff --git a/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index 1a7da8267..c8170d049 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -213,7 +213,7 @@ Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describe Phase 1 uses the states in [stovepipe/entity/request.go](../../../../stovepipe/entity/request.go). `record` runs *after* the Request is terminal: `buildsignal` projects the build's terminal status onto it as `succeeded`, `failed`, or `cancelled` (`RequestState.HasBuildOutcome()`), and only then publishes. So `record` reads an outcome and writes no state. `superseded` is terminal without an outcome. -Phase 2 broadens "complete" to "all planned facts recorded", which needs a marker this stage does not own; see [Completion marker: open](#completion-marker-open). +The `project_facts_recorded` event marks completion of the resolver’s result batch. A future pipeline with independent project builds would need to cover the entire planned set; see [Completion across independent project builds](#completion-across-independent-project-builds). ## Storage and queue contract @@ -299,9 +299,9 @@ The rough idea is to track this with additional stores written during this stage Neither fits `ValidationFactStore`. Facts are looked up by exact identity and URIs do not sort, while previous/next-green needs an ordered seek over positions, a different key shape. A genuinely needed reverse lookup getting its own first-class store is the established pattern here; `RequestURIStore` is the existing example (see [storage README](../../../../stovepipe/extension/storage/README.md#key-value-contract)). -### Completion marker: open +### Completion across independent project builds -`record` makes no `Request` write in Phase 1, so the stage has no durable marker saying it finished. `buildsignal` drives the Request terminal before `record` ever runs, leaving no non-terminal window to occupy. +`record` retains `project_facts_recorded` after all resolved project facts and ordering mappings are durable, including an empty batch, and before publishing its terminal hook. The materializer projects its original timestamp into the request summary before the stage continues, and project status APIs read completion from that summary. A projection failure fails the delivery; replay repairs the summary before the terminal hook. If a future pipeline fans a request out into independent project builds, completion must cover that entire planned set before emitting the marker. Request terminality alone is insufficient because it precedes project recording. ### Supporting re-run of the same URI @@ -309,7 +309,7 @@ Widen the key with the `RequestID` the facts already record, so each attempt is ### Phase 2 -- **Pipeline**: `record` publishes the request id onward to `analyze`, for green and not-green facts alike, because a failed build is when project attribution matters most. An earlier draft had `record` retarget the Request from `processing` to `analyzing`, with `analyze` owning the terminal transition. That no longer fits, because the Request is already terminal before `record` runs, so tracking "all facts recorded" belongs to the `analyze` design (see [Completion marker: open](#completion-marker-open)). The message onward carries the request id and queue name, exactly as `record`'s own does, and the consumer stays idempotent. +- **Pipeline**: `record` publishes the request id onward to `analyze`, for green and not-green facts alike, because a failed build is when project attribution matters most. An earlier draft had `record` retarget the Request from `processing` to `analyzing`, with `analyze` owning the terminal transition. That no longer fits, because the Request is already terminal before `record` runs, so tracking "all facts recorded" belongs to the `analyze` design (see [Completion across independent project builds](#completion-across-independent-project-builds)). The message onward carries the request id and queue name, exactly as `record`'s own does, and the consumer stays idempotent. - **Project facts**: `record` runs the same create-fact-then-notify flow. The fact is keyed by the stable project id carried on the per-project signal, and one `validation.project.recorded` event is published per project identity (see [What the](#what-the-type-carries) `type` [carries](#what-the-type-carries)). Each per-project signal needs its own message id (see [Storage and queue contract](#storage-and-queue-contract)). The Queue's `last_green_uri` describes the whole repository and stays untouched. Deciding project identity, retrieving the target graph, tracking completion, and defining intermediate degrees belong to `analyze.md`. diff --git a/stovepipe/controller/project_status.go b/stovepipe/controller/project_status.go index 21d11de34..24b2c074b 100644 --- a/stovepipe/controller/project_status.go +++ b/stovepipe/controller/project_status.go @@ -143,7 +143,8 @@ func (c *projectStatusController) GetProjectStatus(ctx context.Context, req enti return entity.GetProjectStatusResult{}, &ProjectStatusConsistencyError{message: "project status selector disagrees with stored request summary"} } result.RequestSummary = requestSummary - result.UpdatedAtMs = requestSummary.StateTimestampMs + result.ProjectResultsComplete = requestSummary.ProjectResultsCompletedAtMs > 0 + result.UpdatedAtMs = max(requestSummary.StateTimestampMs, requestSummary.ProjectResultsCompletedAtMs) factStore := store.GetValidationFactStore() fact, err := factStore.Get(ctx, requestSummary.URI, "") if err != nil { diff --git a/stovepipe/controller/project_status_test.go b/stovepipe/controller/project_status_test.go index b94aac66d..227991fb7 100644 --- a/stovepipe/controller/project_status_test.go +++ b/stovepipe/controller/project_status_test.go @@ -590,3 +590,66 @@ func TestGetProjectStatusRejectsInconsistentRequest(t *testing.T) { assert.True(t, IsProjectStatusConsistency(err)) assert.False(t, errors.Is(err, storage.ErrNotFound)) } + +func TestGetProjectStatusCompletion(t *testing.T) { + for _, tt := range []struct { + name string + completedAt int64 + stateAt int64 + empty bool + updatedAt int64 + }{ + {name: "terminal request still recording", stateAt: 10, updatedAt: 20}, + {name: "completed projects", completedAt: 30, stateAt: 10, updatedAt: 30}, + {name: "completed empty set", completedAt: 30, stateAt: 10, empty: true, updatedAt: 30}, + {name: "newer lifecycle timestamp", completedAt: 30, stateAt: 40, updatedAt: 40}, + {name: "newer fact timestamp", completedAt: 15, stateAt: 10, updatedAt: 20}, + } { + for _, byURI := range []bool{false, true} { + selector := "ID" + if byURI { + selector = "URI" + } + t.Run(tt.name+"/"+selector, func(t *testing.T) { + ctrl := gomock.NewController(t) + factory := storagemock.NewMockFactory(ctrl) + store := storagemock.NewMockStorage(ctrl) + summaries := storagemock.NewMockRequestSummaryStore(ctrl) + facts := storagemock.NewMockValidationFactStore(ctrl) + summary := entity.RequestSummary{ + Queue: projectStatusQueue, RequestID: projectStatusID, URI: projectStatusURI, + State: entity.RequestStateFailed, StateTimestampMs: tt.stateAt, + ProjectResultsCompletedAtMs: tt.completedAt, + } + factory.EXPECT().For(storage.Config{QueueName: projectStatusQueue}).Return(store, nil) + req := entity.GetProjectStatusRequest{Queue: projectStatusQueue, RequestID: projectStatusID, Projects: []string{"p", "missing"}, PageSize: 1} + if byURI { + req.RequestID = "" + req.ChangeURI = projectStatusURI + uris := storagemock.NewMockRequestURIStore(ctrl) + store.EXPECT().GetRequestURIStore().Return(uris) + uris.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil) + } + store.EXPECT().GetRequestSummaryStore().Return(summaries) + readSummary := summaries.EXPECT().Get(gomock.Any(), projectStatusID).Return(summary, nil) + store.EXPECT().GetValidationFactStore().Return(facts).After(readSummary) + facts.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) + if tt.empty { + facts.EXPECT().Get(gomock.Any(), projectStatusURI, "p").Return(entity.ValidationFact{}, storage.ErrNotFound) + } else { + facts.EXPECT().Get(gomock.Any(), projectStatusURI, "p").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "p", Degree: 1, CreatedAt: 20}, nil) + } + result, err := NewProjectStatusController(zap.NewNop().Sugar(), tally.NoopScope, factory).GetProjectStatus(context.Background(), req) + require.NoError(t, err) + assert.Equal(t, tt.completedAt > 0, result.ProjectResultsComplete) + assert.Equal(t, tt.updatedAt, result.UpdatedAtMs) + if tt.empty { + assert.Empty(t, result.ProjectValidationFacts) + } else { + assert.Len(t, result.ProjectValidationFacts, 1) + } + assert.NotEmpty(t, result.NextPageToken) + }) + } + } +} diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 29b3fe501..07ab866f9 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -226,9 +226,6 @@ func (c *Controller) persistProjectFacts(ctx context.Context, store storage.Stor return fmt.Errorf("failed to index project fact for request %q and project %q: %w", request.ID, fact.Project, err) } } - if len(results) == 0 { - return nil - } return c.persistProjectFactsRecordedLog(ctx, store, request, len(results)) } diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 03311aa2f..95dc05b93 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -163,11 +163,11 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu 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.Equal(t, entity.RequestEventValidationFactRecorded, log.Event) + assert.Contains(t, []entity.RequestEvent{entity.RequestEventValidationFactRecorded, entity.RequestEventProjectFactsRecorded}, log.Event) assert.Empty(t, log.State) assert.Zero(t, log.RequestVersion) assert.Empty(t, log.OutcomeReason) - assert.NotEmpty(t, log.Metadata[requestlog.MetadataKeyFactDegree]) + assert.NotEmpty(t, log.Metadata) return nil }, ).AnyTimes() @@ -405,6 +405,64 @@ func TestRecordProjectFacts_PassesEmptyTerminalBuild(t *testing.T) { require.NoError(t, err) } +func TestProcessRecordsEmptyProjectCompletion(t *testing.T) { + resolveErr := errors.New("project resolution unavailable") + writeErr := errors.New("completion write unavailable") + for _, tt := range []struct { + name string + results []projectstatus.Result + resolveErr error + writeErr error + }{ + {name: "nil results"}, + {name: "empty results", results: []projectstatus.Result{}}, + {name: "resolution failed", resolveErr: resolveErr}, + {name: "completion write failed", writeErr: writeErr}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := requestWithState(entity.RequestStateFailed) + m.reqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil) + factory := projectstatusmock.NewMockResolverFactory(ctrl) + resolver := projectstatusmock.NewMockResolver(ctrl) + c.projectStatusResolverFactory = factory + factory.EXPECT().For(projectstatus.ResolverConfig{QueueName: testQueue}).Return(resolver, nil) + resolver.EXPECT().Resolve(gomock.Any(), request, request.TerminalBuildID).Return(tt.results, tt.resolveErr) + materializer := requestlogmock.NewMockMaterializer(ctrl) + c.materializer = materializer + if tt.resolveErr == nil { + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + gomock.InOrder( + materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, entity.RequestEventValidationFactRecorded, log.Event) + return nil + }), + materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, "event/project_facts_recorded/all", log.ID) + assert.Equal(t, "0", log.Metadata[requestlog.MetadataKeyProjectFactCount]) + return tt.writeErr + }), + ) + } + err := c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, request.ID))) + switch { + case tt.resolveErr != nil: + require.ErrorIs(t, err, tt.resolveErr) + assert.Empty(t, m.hooks.events) + case tt.writeErr != nil: + require.ErrorIs(t, err, tt.writeErr) + assert.Empty(t, m.hooks.events) + default: + require.NoError(t, err) + assert.Len(t, m.hooks.events, 1) + } + }) + } +} + func TestProcess_RejectsInvalidProjectStatusResultsBeforeWritingFacts(t *testing.T) { ctrl := gomock.NewController(t) c, m := newController(t, ctrl) @@ -538,6 +596,7 @@ func TestProcess_RecordsFactEventBeforeDerivedWork(t *testing.T) { m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI). Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) m.sourceControl.EXPECT().Promote(gomock.Any(), testURI).Return(nil) + eventMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil) require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) } diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go index a572450f3..2eb424ae7 100644 --- a/stovepipe/core/requestlog/materializer.go +++ b/stovepipe/core/requestlog/materializer.go @@ -117,7 +117,7 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l return err } - if retained.State == entity.RequestStateUnknown { + if retained.State == entity.RequestStateUnknown && retained.Event != entity.RequestEventProjectFactsRecorded { return nil } if err := materializeRequestProjections(ctx, stores, retained); err != nil { diff --git a/stovepipe/core/requestlog/materializer_test.go b/stovepipe/core/requestlog/materializer_test.go index 6f3ed0921..037fd344a 100644 --- a/stovepipe/core/requestlog/materializer_test.go +++ b/stovepipe/core/requestlog/materializer_test.go @@ -383,3 +383,20 @@ func TestMaterializerMetricsIncludeContextTags(t *testing.T) { require.True(t, ok) assert.EqualValues(t, 1, counter.Value()) } + +func TestMaterializerRetainsOtherEventsWithoutProjectingSummary(t *testing.T) { + for _, event := range []entity.RequestEvent{ + entity.RequestEventBuildTriggered, + entity.RequestEventBuildFinished, + entity.RequestEventValidationFactRecorded, + entity.RequestEventRecordAbandoned, + entity.RequestEventBuildAnalysisAbandoned, + } { + t.Run(string(event), func(t *testing.T) { + materializer, stores, logs, _, _ := newTestMaterializer(t) + log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, event, "test", nil) + logs.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + require.NoError(t, materializer.PersistLog(context.Background(), stores, log)) + }) + } +} diff --git a/stovepipe/core/requestlog/request_summary.go b/stovepipe/core/requestlog/request_summary.go index 2353768e3..43d8551e1 100644 --- a/stovepipe/core/requestlog/request_summary.go +++ b/stovepipe/core/requestlog/request_summary.go @@ -29,6 +29,9 @@ func materializeRequestSummary(ctx context.Context, stores storage.Storage, log for { currentSummary, err := summaryStore.Get(ctx, log.RequestID) if storage.IsNotFound(err) { + if log.State == entity.RequestStateUnknown { + return entity.RequestSummary{}, fmt.Errorf("request summary required to project event request_id=%q: %w", log.RequestID, err) + } initial, createErr := createInitialRequestSummary(ctx, stores.GetRequestStore(), summaryStore, log) if errors.Is(createErr, storage.ErrAlreadyExists) { continue @@ -113,17 +116,25 @@ func projectRequestSummaryFacts(summary entity.RequestSummary, log entity.Reques log.RequestID, summary.RequestID, log.Queue, summary.Queue, ) } - if log.State != entity.RequestStateAccepted { - return summary, nil + if log.Event == entity.RequestEventProjectFactsRecorded { + if log.TimestampMs <= 0 || (summary.ProjectResultsCompletedAtMs != 0 && summary.ProjectResultsCompletedAtMs != log.TimestampMs) { + return entity.RequestSummary{}, fmt.Errorf("request summary completion time conflicts with retained log request_id=%q", log.RequestID) + } + summary.ProjectResultsCompletedAtMs = log.TimestampMs } - if summary.AcceptedAtMs != 0 && summary.AcceptedAtMs != log.TimestampMs { - return entity.RequestSummary{}, fmt.Errorf("request summary acceptance time conflicts with retained log request_id=%q", log.RequestID) + if log.State == entity.RequestStateAccepted { + if summary.AcceptedAtMs != 0 && summary.AcceptedAtMs != log.TimestampMs { + return entity.RequestSummary{}, fmt.Errorf("request summary acceptance time conflicts with retained log request_id=%q", log.RequestID) + } + summary.AcceptedAtMs = log.TimestampMs } - summary.AcceptedAtMs = log.TimestampMs return summary, nil } func projectRequestSummaryState(summary entity.RequestSummary, log entity.RequestLog) (entity.RequestSummary, error) { + if log.State == entity.RequestStateUnknown { + return summary, nil + } if log.RequestVersion < summary.RequestVersion { return summary, nil } diff --git a/stovepipe/core/requestlog/request_summary_test.go b/stovepipe/core/requestlog/request_summary_test.go index b5097eccc..eecbafa77 100644 --- a/stovepipe/core/requestlog/request_summary_test.go +++ b/stovepipe/core/requestlog/request_summary_test.go @@ -210,12 +210,14 @@ func TestMaterializerRetriesRequestSummaryVersionMismatch(t *testing.T) { current.Version = 4 winner := current winner.Version = 5 + winner.ProjectResultsCompletedAtMs = testNowMs - 500 updated := winner updated.State = log.State updated.RequestVersion = log.RequestVersion updated.StateTimestampMs = testNowMs firstAttempt := updated firstAttempt.Version = current.Version + firstAttempt.ProjectResultsCompletedAtMs = 0 gomock.InOrder( requestLogs.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil), summaries.EXPECT().Get(gomock.Any(), testRequestID).Return(current, nil), @@ -409,6 +411,7 @@ func TestUpdateRequestSummaryNewerState(t *testing.T) { current.RequestVersion = 1 current.StateTimestampMs = testNowMs - 1000 current.AcceptedAtMs = testNowMs - 1000 + current.ProjectResultsCompletedAtMs = testNowMs - 500 current.Version = 4 tests := []struct { @@ -503,3 +506,111 @@ func TestMaterializerRetriesAcceptanceRepairAfterPartialWrite(t *testing.T) { ) require.NoError(t, materializer.PersistLog(context.Background(), stores, log)) } + +func TestMaterializerProjectsProjectCompletion(t *testing.T) { + for _, tt := range []struct { + name string + completedAt int64 + count string + wantErr bool + }{ + {name: "completed projects", count: "3"}, + {name: "completed empty set", count: "0"}, + {name: "duplicate completion", completedAt: testNowMs, count: "3"}, + {name: "conflicting timestamp", completedAt: testNowMs - 1, count: "3", wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + materializer, stores, logs, summaries, _ := newTestMaterializer(t) + log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, entity.RequestEventProjectFactsRecorded, "all", map[string]string{MetadataKeyProjectFactCount: tt.count}) + log.TimestampMs = testNowMs + current := seededRequestSummary() + current.State = entity.RequestStateFailed + current.RequestVersion = 3 + current.StateTimestampMs = testNowMs - 1000 + current.BaseURI = testBaseURI + current.AcceptedAtMs = testNowMs - 2000 + current.OutcomeReason = entity.RequestOutcomeReasonBuildFailed + current.ProjectResultsCompletedAtMs = tt.completedAt + want := current + want.ProjectResultsCompletedAtMs = log.TimestampMs + logs.EXPECT().Create(gomock.Any(), log).Return(storage.ErrAlreadyExists) + logs.EXPECT().Get(gomock.Any(), testRequestID, log.ID).Return(log, nil) + summaries.EXPECT().Get(gomock.Any(), testRequestID).Return(current, nil) + if !tt.wantErr && want != current { + summaries.EXPECT().Update(gomock.Any(), want, current.Version, current.Version+1).Return(nil) + } + err := materializer.PersistLog(context.Background(), stores, log) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + }) + } +} + +func TestMaterializerRepairsCompletionAfterProjectionFailure(t *testing.T) { + materializer, stores, logs, summaries, _ := newTestMaterializer(t) + log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, entity.RequestEventProjectFactsRecorded, "all", nil) + retained := log + retained.TimestampMs = testNowMs + current := seededRequestSummary() + current.State = entity.RequestStateFailed + current.RequestVersion = 3 + current.StateTimestampMs = testNowMs - 1000 + updated := current + updated.ProjectResultsCompletedAtMs = retained.TimestampMs + writeErr := errors.New("summary unavailable") + gomock.InOrder( + logs.EXPECT().Create(gomock.Any(), retained).Return(nil), + summaries.EXPECT().Get(gomock.Any(), testRequestID).Return(current, nil), + summaries.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(writeErr), + ) + require.ErrorIs(t, materializer.PersistLog(context.Background(), stores, log), writeErr) + + materializer.now = func() time.Time { return time.UnixMilli(testNowMs + 1000) } + candidate := log + candidate.TimestampMs = testNowMs + 1000 + gomock.InOrder( + logs.EXPECT().Create(gomock.Any(), candidate).Return(storage.ErrAlreadyExists), + logs.EXPECT().Get(gomock.Any(), testRequestID, log.ID).Return(retained, nil), + summaries.EXPECT().Get(gomock.Any(), testRequestID).Return(current, nil), + summaries.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil), + ) + require.NoError(t, materializer.PersistLog(context.Background(), stores, log)) +} + +func TestMaterializerRetriesCompletionAfterStateProjectionWins(t *testing.T) { + materializer, stores, logs, summaries, _ := newTestMaterializer(t) + log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, entity.RequestEventProjectFactsRecorded, "all", nil) + log.TimestampMs = testNowMs + current := seededRequestSummary() + current.State = entity.RequestStateProcessing + current.RequestVersion = 2 + updated := current + updated.ProjectResultsCompletedAtMs = log.TimestampMs + winner := current + winner.State = entity.RequestStateFailed + winner.RequestVersion = 3 + winner.StateTimestampMs = testNowMs + 1 + winner.OutcomeReason = entity.RequestOutcomeReasonBuildFailed + winner.Version++ + updatedWinner := winner + updatedWinner.ProjectResultsCompletedAtMs = log.TimestampMs + gomock.InOrder( + logs.EXPECT().Create(gomock.Any(), log).Return(nil), + summaries.EXPECT().Get(gomock.Any(), testRequestID).Return(current, nil), + summaries.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(storage.ErrVersionMismatch), + summaries.EXPECT().Get(gomock.Any(), testRequestID).Return(winner, nil), + summaries.EXPECT().Update(gomock.Any(), updatedWinner, int32(2), int32(3)).Return(nil), + ) + require.NoError(t, materializer.PersistLog(context.Background(), stores, log)) +} + +func TestMaterializerCompletionRequiresExistingSummary(t *testing.T) { + materializer, stores, logs, summaries, _ := newTestMaterializer(t) + log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, entity.RequestEventProjectFactsRecorded, "all", nil) + logs.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + summaries.EXPECT().Get(gomock.Any(), testRequestID).Return(entity.RequestSummary{}, storage.ErrNotFound) + require.ErrorIs(t, materializer.PersistLog(context.Background(), stores, log), storage.ErrNotFound) +} diff --git a/stovepipe/entity/request_summary.go b/stovepipe/entity/request_summary.go index 18d2e5d44..b40ac23f1 100644 --- a/stovepipe/entity/request_summary.go +++ b/stovepipe/entity/request_summary.go @@ -32,6 +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 int64 // OutcomeReason is the reason for the represented state; empty means unavailable or inapplicable. OutcomeReason RequestOutcomeReason // Version is the optimistic-lock version of this materialized view. diff --git a/stovepipe/extension/storage/mysql/request_summary_store.go b/stovepipe/extension/storage/mysql/request_summary_store.go index 965dfc297..aa6932434 100644 --- a/stovepipe/extension/storage/mysql/request_summary_store.go +++ b/stovepipe/extension/storage/mysql/request_summary_store.go @@ -48,8 +48,8 @@ func (s *requestSummaryStore) Create(ctx context.Context, summary entity.Request _, err := s.db.ExecContext(ctx, ` INSERT INTO request_summary ( - queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, accepted_at_ms, outcome_reason, version - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, accepted_at_ms, outcome_reason, project_results_completed_at_ms, version + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, summary.Queue, summary.RequestID, summary.URI, @@ -59,6 +59,7 @@ func (s *requestSummaryStore) Create(ctx context.Context, summary entity.Request summary.StateTimestampMs, summary.AcceptedAtMs, summary.OutcomeReason, + summary.ProjectResultsCompletedAtMs, summary.Version, ) if err != nil { @@ -75,7 +76,7 @@ func (s *requestSummaryStore) Get(ctx context.Context, requestID string) (ret en defer func() { op.Complete(retErr) }() err := s.db.QueryRowContext(ctx, ` - SELECT queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, accepted_at_ms, outcome_reason, version + SELECT queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, accepted_at_ms, outcome_reason, project_results_completed_at_ms, version FROM request_summary WHERE queue = ? AND request_id = ?`, s.queue, requestID, @@ -89,6 +90,7 @@ func (s *requestSummaryStore) Get(ctx context.Context, requestID string) (ret en &ret.StateTimestampMs, &ret.AcceptedAtMs, &ret.OutcomeReason, + &ret.ProjectResultsCompletedAtMs, &ret.Version, ) if errors.Is(err, sql.ErrNoRows) { @@ -110,7 +112,7 @@ func (s *requestSummaryStore) Update(ctx context.Context, summary entity.Request result, err := s.db.ExecContext(ctx, ` UPDATE request_summary - SET base_uri = ?, state = ?, request_version = ?, state_timestamp_ms = ?, accepted_at_ms = ?, outcome_reason = ?, version = ? + SET base_uri = ?, state = ?, request_version = ?, state_timestamp_ms = ?, accepted_at_ms = ?, outcome_reason = ?, project_results_completed_at_ms = ?, version = ? WHERE queue = ? AND request_id = ? AND version = ?`, summary.BaseURI, summary.State, @@ -118,6 +120,7 @@ func (s *requestSummaryStore) Update(ctx context.Context, summary entity.Request summary.StateTimestampMs, summary.AcceptedAtMs, summary.OutcomeReason, + summary.ProjectResultsCompletedAtMs, newVersion, summary.Queue, summary.RequestID, diff --git a/stovepipe/extension/storage/mysql/request_summary_store_test.go b/stovepipe/extension/storage/mysql/request_summary_store_test.go index c79669048..54eaf33dc 100644 --- a/stovepipe/extension/storage/mysql/request_summary_store_test.go +++ b/stovepipe/extension/storage/mysql/request_summary_store_test.go @@ -33,16 +33,17 @@ const testRequestSummaryQueue = "monorepo/main" func testRequestSummary() entity.RequestSummary { return entity.RequestSummary{ - RequestID: "1", - Queue: testRequestSummaryQueue, - URI: "git://repo/head", - BaseURI: "git://repo/base", - State: entity.RequestStateFailed, - RequestVersion: 3, - StateTimestampMs: 1735689600000, - AcceptedAtMs: 1735689599000, - OutcomeReason: entity.RequestOutcomeReasonProcessingFailed, - Version: 1, + RequestID: "1", + Queue: testRequestSummaryQueue, + URI: "git://repo/head", + BaseURI: "git://repo/base", + State: entity.RequestStateFailed, + RequestVersion: 3, + StateTimestampMs: 1735689600000, + AcceptedAtMs: 1735689599000, + OutcomeReason: entity.RequestOutcomeReasonProcessingFailed, + ProjectResultsCompletedAtMs: 1735689601000, + Version: 1, } } @@ -67,7 +68,7 @@ func TestRequestSummaryStoreCreate(t *testing.T) { summary: summary, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request_summary"). - WithArgs(summary.Queue, summary.RequestID, summary.URI, summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, summary.AcceptedAtMs, summary.OutcomeReason, summary.Version). + WithArgs(summary.Queue, summary.RequestID, summary.URI, summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, summary.AcceptedAtMs, summary.OutcomeReason, summary.ProjectResultsCompletedAtMs, summary.Version). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -76,7 +77,7 @@ func TestRequestSummaryStoreCreate(t *testing.T) { summary: summary, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request_summary"). - WithArgs(summary.Queue, summary.RequestID, summary.URI, summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, summary.AcceptedAtMs, summary.OutcomeReason, summary.Version). + WithArgs(summary.Queue, summary.RequestID, summary.URI, summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, summary.AcceptedAtMs, summary.OutcomeReason, summary.ProjectResultsCompletedAtMs, summary.Version). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -126,8 +127,8 @@ func TestRequestSummaryStoreGet(t *testing.T) { { name: "found", setup: func(mock sqlmock.Sqlmock) { - rows := sqlmock.NewRows([]string{"queue", "request_id", "uri", "base_uri", "state", "request_version", "state_timestamp_ms", "accepted_at_ms", "outcome_reason", "version"}). - AddRow(want.Queue, want.RequestID, want.URI, want.BaseURI, want.State, want.RequestVersion, want.StateTimestampMs, want.AcceptedAtMs, want.OutcomeReason, want.Version) + rows := sqlmock.NewRows([]string{"queue", "request_id", "uri", "base_uri", "state", "request_version", "state_timestamp_ms", "accepted_at_ms", "outcome_reason", "project_results_completed_at_ms", "version"}). + AddRow(want.Queue, want.RequestID, want.URI, want.BaseURI, want.State, want.RequestVersion, want.StateTimestampMs, want.AcceptedAtMs, want.OutcomeReason, want.ProjectResultsCompletedAtMs, want.Version) mock.ExpectQuery("SELECT queue, request_id, uri, base_uri, state").WithArgs(testRequestSummaryQueue, want.RequestID).WillReturnRows(rows) }, }, @@ -180,7 +181,7 @@ func TestRequestSummaryStoreUpdate(t *testing.T) { db, mock, store := setupRequestSummaryStoreTest(t) defer db.Close() expectation := mock.ExpectExec("UPDATE request_summary"). - WithArgs(summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, summary.AcceptedAtMs, summary.OutcomeReason, newVersion, summary.Queue, summary.RequestID, oldVersion) + WithArgs(summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, summary.AcceptedAtMs, summary.OutcomeReason, summary.ProjectResultsCompletedAtMs, newVersion, summary.Queue, summary.RequestID, oldVersion) if tt.execErr != nil { expectation.WillReturnError(tt.execErr) } else { diff --git a/stovepipe/extension/storage/mysql/schema/request_summary.sql b/stovepipe/extension/storage/mysql/schema/request_summary.sql index 0b10438c5..572e33264 100644 --- a/stovepipe/extension/storage/mysql/schema/request_summary.sql +++ b/stovepipe/extension/storage/mysql/schema/request_summary.sql @@ -12,5 +12,6 @@ CREATE TABLE IF NOT EXISTS request_summary ( version INT NOT NULL, accepted_at_ms BIGINT NOT NULL DEFAULT 0, outcome_reason VARCHAR(64) NOT NULL DEFAULT '', + project_results_completed_at_ms BIGINT NOT NULL DEFAULT 0, PRIMARY KEY (queue, request_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/test/e2e/stovepipe/ingest_test.go b/test/e2e/stovepipe/ingest_test.go index 4fd564d5d..dce88e080 100644 --- a/test/e2e/stovepipe/ingest_test.go +++ b/test/e2e/stovepipe/ingest_test.go @@ -25,6 +25,15 @@ func (s *StovepipeE2ESuite) TestIngest_HappyPath_Processes() { id := s.ingest(queue) s.assertIngestPersisted(queue, id) s.awaitProcessed(queue) + pollUntil(convergencePollInterval, func() bool { + status, err := s.client.GetProjectStatusByID(s.ctx, &pb.GetProjectStatusByIDRequest{Queue: queue, RequestId: id}) + require.NoError(s.T(), err) + return status.ProjectResultsComplete + }) + status, err := s.client.GetProjectStatusByURI(s.ctx, &pb.GetProjectStatusByURIRequest{Queue: queue, ChangeUri: "git://" + queue + "/HEAD"}) + require.NoError(s.T(), err) + assert.True(s.T(), status.ProjectResultsComplete) + assert.Empty(s.T(), status.Projects) } func (s *StovepipeE2ESuite) TestIngest_Idempotent() { diff --git a/test/integration/stovepipe/BUILD.bazel b/test/integration/stovepipe/BUILD.bazel index a3bc59593..3d1397836 100644 --- a/test/integration/stovepipe/BUILD.bazel +++ b/test/integration/stovepipe/BUILD.bazel @@ -16,6 +16,7 @@ go_test( ], deps = [ "//api/stovepipe/protopb:go_default_library", + "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage/mysql:go_default_library", "//test/testutil:go_default_library", diff --git a/test/integration/stovepipe/suite_test.go b/test/integration/stovepipe/suite_test.go index a0e44b712..7f483e9b5 100644 --- a/test/integration/stovepipe/suite_test.go +++ b/test/integration/stovepipe/suite_test.go @@ -38,6 +38,7 @@ import ( "github.com/stretchr/testify/suite" "github.com/uber-go/tally" pb "github.com/uber/submitqueue/api/stovepipe/protopb" + "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" mysqlstorage "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" "github.com/uber/submitqueue/test/testutil" @@ -381,6 +382,27 @@ func (s *StovepipeIntegrationSuite) TestProjectStatusAPI() { assert.Equal(t, int64(4000), remaining.GetUpdatedAtMs()) assert.Empty(t, remaining.NextPageToken) assert.False(t, remaining.ProjectResultsComplete) + completion := requestlog.NewRequestEventLog(entity.Request{Queue: queue, ID: id}, entity.RequestEventProjectFactsRecorded, "all", map[string]string{requestlog.MetadataKeyProjectFactCount: "3"}) + completion.TimestampMs = 6000 + backend, err := mysqlstorage.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err) + stores, err := backend.For(queue) + require.NoError(t, err) + require.NoError(t, requestlog.NewMaterializer(tally.NoopScope).PersistLog(s.ctx, stores, completion)) + completed, err := s.client.GetProjectStatusByURI(s.ctx, &pb.GetProjectStatusByURIRequest{ + Queue: queue, ChangeUri: uri, PageSize: 1, + }) + require.NoError(t, err) + assert.True(t, completed.ProjectResultsComplete) + assert.Equal(t, int64(6000), completed.UpdatedAtMs) + assert.NotEmpty(t, completed.NextPageToken) + completedRemainder, err := s.client.GetProjectStatusByID(s.ctx, &pb.GetProjectStatusByIDRequest{ + Queue: queue, RequestId: id, PageToken: completed.NextPageToken, + }) + require.NoError(t, err) + assert.True(t, completedRemainder.ProjectResultsComplete) + assert.Equal(t, int64(6000), completedRemainder.UpdatedAtMs) + assert.Empty(t, completedRemainder.NextPageToken) for _, selector := range []struct{ queue, id string }{{queue, "999"}, {"project-status-api/other", id}} { _, err := s.client.GetProjectStatusByID(s.ctx, &pb.GetProjectStatusByIDRequest{ diff --git a/web/test/e2e/stovepipe/run_e2e.py b/web/test/e2e/stovepipe/run_e2e.py index be98b146c..04b6f3686 100644 --- a/web/test/e2e/stovepipe/run_e2e.py +++ b/web/test/e2e/stovepipe/run_e2e.py @@ -32,7 +32,10 @@ def _pagination_fixture_sql() -> str: ) state = "failed" if i == 51 else "succeeded" statements.append( - f"INSERT INTO request_summary VALUES ('web/pagination','{i}','{uri}','','{state}',1,{1000+i},1,{1000+i},'');" + "INSERT INTO request_summary " + "(queue,request_id,uri,base_uri,state,request_version," + "state_timestamp_ms,version,accepted_at_ms,outcome_reason) " + f"VALUES ('web/pagination','{i}','{uri}','','{state}',1,{1000+i},1,{1000+i},'');" ) statements.append( f"INSERT INTO request_acceptance VALUES ('web/pagination',{1000+i},'{i}');"