From 894deead2281dfb3195dccb91ce466131cbac775 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 23 Oct 2026 16:06:09 +0000 Subject: [PATCH 1/4] fix(stovepipe): report completed project results Summary: Intent: - Let callers distinguish a finished project result set from results that are still being recorded. Changes: - Read the durable project_facts_recorded event before loading project facts, expose completion through both status lookups, and include its timestamp in updated_at_ms. - Record completion for successfully resolved empty result sets before publishing the terminal hook. - Cover completion, missing markers, storage failures, pagination, index repair on retry, and the real pipeline in unit, integration, and end-to-end tests. Revert Plan: Revert the application change. Existing facts and request events remain compatible; no schema migration is required. --- doc/rfc/stovepipe/project-status-api.md | 2 +- doc/rfc/stovepipe/steps/record.md | 8 +- stovepipe/controller/project_status.go | 15 ++++ stovepipe/controller/project_status_test.go | 96 +++++++++++++++++++++ stovepipe/controller/record/record.go | 3 - stovepipe/controller/record/record_test.go | 51 ++++++++++- stovepipe/core/requestlog/materializer.go | 7 +- test/e2e/stovepipe/ingest_test.go | 9 ++ test/integration/stovepipe/BUILD.bazel | 1 + test/integration/stovepipe/suite_test.go | 18 ++++ 10 files changed, 199 insertions(+), 11 deletions(-) diff --git a/doc/rfc/stovepipe/project-status-api.md b/doc/rfc/stovepipe/project-status-api.md index 94088c81f..151fd4580 100644 --- a/doc/rfc/stovepipe/project-status-api.md +++ b/doc/rfc/stovepipe/project-status-api.md @@ -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 request’s durable `project_facts_recorded` event exists. 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 API reads completion before loading facts and includes its timestamp in `updated_at_ms`. Missing completion remains false, including for older requests without the event. 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/steps/record.md b/doc/rfc/stovepipe/steps/record.md index 5d0140027..4f335820a 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 project status APIs use this event as their completion marker. 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..75ec95b08 100644 --- a/stovepipe/controller/project_status.go +++ b/stovepipe/controller/project_status.go @@ -28,6 +28,7 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" @@ -144,6 +145,20 @@ func (c *projectStatusController) GetProjectStatus(ctx context.Context, req enti } result.RequestSummary = requestSummary result.UpdatedAtMs = requestSummary.StateTimestampMs + // Completion must be observed before facts so a concurrent writer cannot complete a partial page. + completionID := requestlog.RequestEventLogID(entity.RequestEventProjectFactsRecorded, "all") + completion, err := store.GetRequestLogStore().Get(ctx, requestID, completionID) + if err != nil { + if !storage.IsNotFound(err) { + return entity.GetProjectStatusResult{}, fmt.Errorf("GetProjectStatus failed to load project completion for request %q: %w", requestID, err) + } + } else { + if completion.ID != completionID || completion.Queue != req.Queue || completion.RequestID != requestID || completion.Event != entity.RequestEventProjectFactsRecorded || completion.TimestampMs <= 0 { + return entity.GetProjectStatusResult{}, &ProjectStatusConsistencyError{message: "project completion disagrees with stored request"} + } + result.ProjectResultsComplete = true + result.UpdatedAtMs = max(result.UpdatedAtMs, completion.TimestampMs) + } 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 ee7d766a7..fc24a74ed 100644 --- a/stovepipe/controller/project_status_test.go +++ b/stovepipe/controller/project_status_test.go @@ -25,6 +25,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" @@ -120,6 +121,7 @@ func TestGetProjectStatus(t *testing.T) { store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, tt.summaryErr) if tt.summaryErr == nil { + expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(tt.fact, tt.factErr) if !tt.noProjectLookup && !(tt.factErr == nil && tt.wantConsistency) { @@ -205,6 +207,7 @@ func TestGetProjectStatusByID(t *testing.T) { store.EXPECT().GetRequestSummaryStore().Return(summaries) summaries.EXPECT().Get(gomock.Any(), projectStatusID).Return(tt.summary, tt.summaryErr) if tt.summaryErr == nil && tt.summary == summary { + expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(facts) facts.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(tt.fact, tt.factErr) if !tt.wantConsistency && tt.wantErr == nil { @@ -283,6 +286,7 @@ func TestGetProjectStatusUsesRequestSummaryTimestamp(t *testing.T) { uriStore.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil) store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, nil) + expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, CreatedAt: 20}, nil) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "project-a").Return(entity.ValidationFact{}, storage.ErrNotFound) @@ -306,6 +310,7 @@ func TestGetProjectStatusReturnsProjectFactsInRequestedOrder(t *testing.T) { uriStore.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil) store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, nil) + expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "project-b").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "project-b", Degree: entity.DegreeGreen, CreatedAt: 20}, nil) @@ -338,6 +343,7 @@ func TestGetProjectStatusPaginatesProjectFacts(t *testing.T) { uriStore.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil) store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, nil) + expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "project-a").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "project-a", Degree: entity.DegreeGreen}, nil) @@ -358,6 +364,7 @@ func TestGetProjectStatusPaginatesProjectFacts(t *testing.T) { factory.EXPECT().For(storage.Config{QueueName: projectStatusQueue}).Return(store, nil) store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, nil) + expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "project-b").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "project-b", Degree: entity.DegreeBroken}, nil) @@ -476,6 +483,7 @@ func TestGetProjectStatusPaginatesStoredProjects(t *testing.T) { factory.EXPECT().For(storage.Config{QueueName: projectStatusQueue}).Return(store, nil).Times(2) store.EXPECT().GetRequestSummaryStore().Return(summaries).Times(2) summaries.EXPECT().Get(gomock.Any(), projectStatusID).Return(summary, nil).Times(2) + expectNoProjectCompletion(ctrl, store, 2) store.EXPECT().GetValidationFactStore().Return(facts).Times(2) store.EXPECT().GetProjectFactByDegreeStore().Return(indexStore).Times(2) facts.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound).Times(2) @@ -589,3 +597,91 @@ func TestGetProjectStatusRejectsInconsistentRequest(t *testing.T) { assert.True(t, IsProjectStatusConsistency(err)) assert.False(t, errors.Is(err, storage.ErrNotFound)) } + +func expectNoProjectCompletion(ctrl *gomock.Controller, store *storagemock.MockStorage, count int) { + logs := storagemock.NewMockRequestLogStore(ctrl) + store.EXPECT().GetRequestLogStore().Return(logs).Times(count) + logs.EXPECT().Get(gomock.Any(), projectStatusID, requestlog.RequestEventLogID(entity.RequestEventProjectFactsRecorded, "all")). + Return(entity.RequestLog{}, storage.ErrNotFound).Times(count) +} + +func TestGetProjectStatusCompletion(t *testing.T) { + readErr := errors.New("completion read unavailable") + completion := requestlog.NewRequestEventLog(entity.Request{Queue: projectStatusQueue, ID: projectStatusID}, entity.RequestEventProjectFactsRecorded, "all", nil) + completion.TimestampMs = 30 + for _, tt := range []struct { + name string + completion entity.RequestLog + readErr error + complete bool + empty bool + updatedAt int64 + inconsistent bool + }{ + {name: "terminal request still recording", readErr: storage.ErrNotFound, updatedAt: 20}, + {name: "completed projects", completion: completion, complete: true, updatedAt: 30}, + {name: "completed empty set", completion: completion, complete: true, empty: true, updatedAt: 30}, + {name: "completion read failed", readErr: readErr}, + {name: "completion belongs to another request", completion: func() entity.RequestLog { log := completion; log.RequestID = "8"; return log }(), inconsistent: true}, + {name: "newer lifecycle timestamp", completion: completion, complete: true, updatedAt: 40}, + } { + 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) + logs := storagemock.NewMockRequestLogStore(ctrl) + facts := storagemock.NewMockValidationFactStore(ctrl) + summary := entity.RequestSummary{Queue: projectStatusQueue, RequestID: projectStatusID, URI: projectStatusURI, State: entity.RequestStateFailed, StateTimestampMs: 10} + if tt.updatedAt == 40 { + summary.StateTimestampMs = 40 + } + 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) + summaries.EXPECT().Get(gomock.Any(), projectStatusID).Return(summary, nil) + store.EXPECT().GetRequestLogStore().Return(logs) + readCompletion := logs.EXPECT().Get(gomock.Any(), projectStatusID, completion.ID).Return(tt.completion, tt.readErr) + if (tt.readErr == nil || storage.IsNotFound(tt.readErr)) && !tt.inconsistent { + store.EXPECT().GetValidationFactStore().Return(facts).After(readCompletion) + facts.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) + fact := entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "p", Degree: 1, CreatedAt: 20} + if tt.empty { + facts.EXPECT().Get(gomock.Any(), projectStatusURI, "p").Return(entity.ValidationFact{}, storage.ErrNotFound) + } else { + facts.EXPECT().Get(gomock.Any(), projectStatusURI, "p").Return(fact, nil) + } + } + result, err := NewProjectStatusController(zap.NewNop().Sugar(), tally.NoopScope, factory).GetProjectStatus(context.Background(), req) + if tt.inconsistent { + require.Error(t, err) + assert.True(t, IsProjectStatusConsistency(err)) + } else if tt.readErr != nil && !storage.IsNotFound(tt.readErr) { + require.ErrorIs(t, err, tt.readErr) + } else { + require.NoError(t, err) + assert.Equal(t, tt.complete, 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 9f22e4c5f..d110b773b 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -217,9 +217,6 @@ func (c *Controller) recordProjectFacts(ctx context.Context, store storage.Stora 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 09060eda2..461452f89 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() @@ -457,6 +457,52 @@ func TestRecordProjectFacts_PassesEmptyTerminalBuild(t *testing.T) { require.NoError(t, err) } +func TestRecordProjectFactsEmptyCompletion(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) + 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 { + materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, requestlog.RequestEventLogID(entity.RequestEventProjectFactsRecorded, "all"), log.ID) + assert.Equal(t, "0", log.Metadata[requestlog.MetadataKeyProjectFactCount]) + return tt.writeErr + }) + } + err := c.recordProjectFacts(queueContext(), m.store, request) + switch { + case tt.resolveErr != nil: + require.ErrorIs(t, err, tt.resolveErr) + case tt.writeErr != nil: + require.ErrorIs(t, err, tt.writeErr) + default: + require.NoError(t, err) + } + }) + } +} + func TestProcess_RejectsInvalidProjectStatusResultsBeforeWritingFacts(t *testing.T) { ctrl := gomock.NewController(t) c, m := newController(t, ctrl) @@ -596,6 +642,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..29dd8c658 100644 --- a/stovepipe/core/requestlog/materializer.go +++ b/stovepipe/core/requestlog/materializer.go @@ -95,7 +95,7 @@ func NewRequestEventLog( metadata map[string]string, ) entity.RequestLog { return entity.RequestLog{ - ID: publish.IntentID(_occurrenceKindEvent, string(event), occurrence), + ID: RequestEventLogID(event, occurrence), Queue: request.Queue, RequestID: request.ID, Event: event, @@ -103,6 +103,11 @@ func NewRequestEventLog( } } +// RequestEventLogID returns the stable identity of an event occurrence within a request. +func RequestEventLogID(event entity.RequestEvent, occurrence string) string { + return publish.IntentID(_occurrenceKindEvent, string(event), occurrence) +} + func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, log entity.RequestLog) error { if log.TimestampMs == 0 { log.TimestampMs = m.now().UnixMilli() 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..37314fa60 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,23 @@ 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 + require.NoError(t, mysqlstorage.NewRequestLogStore(s.db, tally.NoopScope, queue).Create(s.ctx, 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{ From a2c995ddbd00189574a1ff767ae9c2e4f4f8df3a Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 23 Oct 2026 17:50:01 +0000 Subject: [PATCH 2/4] fix(stovepipe): project result completion into request summaries --- doc/rfc/stovepipe/project-status-api.md | 4 +- doc/rfc/stovepipe/request-log.md | 2 + doc/rfc/stovepipe/steps/record.md | 2 +- stovepipe/controller/project_status.go | 18 +-- stovepipe/controller/project_status_test.go | 89 ++++------- stovepipe/controller/record/record_test.go | 2 +- stovepipe/core/requestlog/materializer.go | 9 +- .../core/requestlog/materializer_test.go | 17 +++ stovepipe/core/requestlog/request_summary.go | 21 ++- .../core/requestlog/request_summary_test.go | 111 ++++++++++++++ stovepipe/entity/request_summary.go | 2 + .../storage/mysql/migrations/BUILD.bazel | 12 ++ .../storage/mysql/migrations/README.md | 16 ++ .../backfill_project_results_completion.sql | 15 ++ .../storage/mysql/migrations/migrations.go | 23 +++ .../migrations/project_results_completion.sql | 2 + .../storage/mysql/request_summary_store.go | 11 +- .../mysql/request_summary_store_test.go | 31 ++-- .../storage/mysql/schema/request_summary.sql | 1 + test/integration/stovepipe/BUILD.bazel | 1 + test/integration/stovepipe/suite_test.go | 138 +++++++++++++++++- 21 files changed, 414 insertions(+), 113 deletions(-) create mode 100644 stovepipe/extension/storage/mysql/migrations/BUILD.bazel create mode 100644 stovepipe/extension/storage/mysql/migrations/README.md create mode 100644 stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql create mode 100644 stovepipe/extension/storage/mysql/migrations/migrations.go create mode 100644 stovepipe/extension/storage/mysql/migrations/project_results_completion.sql diff --git a/doc/rfc/stovepipe/project-status-api.md b/doc/rfc/stovepipe/project-status-api.md index 151fd4580..5d85f305c 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` is true when the request’s durable `project_facts_recorded` event exists. 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 API reads completion before loading facts and includes its timestamp in `updated_at_ms`. Missing completion remains false, including for older requests without the event. 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. Existing events require the [summary backfill](../../../stovepipe/extension/storage/mysql/migrations/README.md); older empty batches without an event remain incomplete. 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 4f335820a..e5b95596b 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -301,7 +301,7 @@ Neither fits `ValidationFactStore`. Facts are looked up by exact identity and UR ### Completion across independent project builds -`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 project status APIs use this event as their completion marker. 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. +`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 diff --git a/stovepipe/controller/project_status.go b/stovepipe/controller/project_status.go index 75ec95b08..24b2c074b 100644 --- a/stovepipe/controller/project_status.go +++ b/stovepipe/controller/project_status.go @@ -28,7 +28,6 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" - "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" @@ -144,21 +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 - // Completion must be observed before facts so a concurrent writer cannot complete a partial page. - completionID := requestlog.RequestEventLogID(entity.RequestEventProjectFactsRecorded, "all") - completion, err := store.GetRequestLogStore().Get(ctx, requestID, completionID) - if err != nil { - if !storage.IsNotFound(err) { - return entity.GetProjectStatusResult{}, fmt.Errorf("GetProjectStatus failed to load project completion for request %q: %w", requestID, err) - } - } else { - if completion.ID != completionID || completion.Queue != req.Queue || completion.RequestID != requestID || completion.Event != entity.RequestEventProjectFactsRecorded || completion.TimestampMs <= 0 { - return entity.GetProjectStatusResult{}, &ProjectStatusConsistencyError{message: "project completion disagrees with stored request"} - } - result.ProjectResultsComplete = true - result.UpdatedAtMs = max(result.UpdatedAtMs, completion.TimestampMs) - } + 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 fc24a74ed..6af20f207 100644 --- a/stovepipe/controller/project_status_test.go +++ b/stovepipe/controller/project_status_test.go @@ -25,7 +25,6 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/errs" - "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" @@ -121,7 +120,6 @@ func TestGetProjectStatus(t *testing.T) { store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, tt.summaryErr) if tt.summaryErr == nil { - expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(tt.fact, tt.factErr) if !tt.noProjectLookup && !(tt.factErr == nil && tt.wantConsistency) { @@ -207,7 +205,6 @@ func TestGetProjectStatusByID(t *testing.T) { store.EXPECT().GetRequestSummaryStore().Return(summaries) summaries.EXPECT().Get(gomock.Any(), projectStatusID).Return(tt.summary, tt.summaryErr) if tt.summaryErr == nil && tt.summary == summary { - expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(facts) facts.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(tt.fact, tt.factErr) if !tt.wantConsistency && tt.wantErr == nil { @@ -286,7 +283,6 @@ func TestGetProjectStatusUsesRequestSummaryTimestamp(t *testing.T) { uriStore.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil) store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, nil) - expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, CreatedAt: 20}, nil) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "project-a").Return(entity.ValidationFact{}, storage.ErrNotFound) @@ -310,7 +306,6 @@ func TestGetProjectStatusReturnsProjectFactsInRequestedOrder(t *testing.T) { uriStore.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil) store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, nil) - expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "project-b").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "project-b", Degree: entity.DegreeGreen, CreatedAt: 20}, nil) @@ -343,7 +338,6 @@ func TestGetProjectStatusPaginatesProjectFacts(t *testing.T) { uriStore.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil) store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, nil) - expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "project-a").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "project-a", Degree: entity.DegreeGreen}, nil) @@ -364,7 +358,6 @@ func TestGetProjectStatusPaginatesProjectFacts(t *testing.T) { factory.EXPECT().For(storage.Config{QueueName: projectStatusQueue}).Return(store, nil) store.EXPECT().GetRequestSummaryStore().Return(summaryStore) summaryStore.EXPECT().Get(gomock.Any(), projectStatusID).Return(requestSummary, nil) - expectNoProjectCompletion(mockCtrl, store, 1) store.EXPECT().GetValidationFactStore().Return(factStore) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) factStore.EXPECT().Get(gomock.Any(), projectStatusURI, "project-b").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "project-b", Degree: entity.DegreeBroken}, nil) @@ -483,7 +476,6 @@ func TestGetProjectStatusPaginatesStoredProjects(t *testing.T) { factory.EXPECT().For(storage.Config{QueueName: projectStatusQueue}).Return(store, nil).Times(2) store.EXPECT().GetRequestSummaryStore().Return(summaries).Times(2) summaries.EXPECT().Get(gomock.Any(), projectStatusID).Return(summary, nil).Times(2) - expectNoProjectCompletion(ctrl, store, 2) store.EXPECT().GetValidationFactStore().Return(facts).Times(2) store.EXPECT().GetProjectFactByDegreeStore().Return(indexStore).Times(2) facts.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound).Times(2) @@ -598,32 +590,19 @@ func TestGetProjectStatusRejectsInconsistentRequest(t *testing.T) { assert.False(t, errors.Is(err, storage.ErrNotFound)) } -func expectNoProjectCompletion(ctrl *gomock.Controller, store *storagemock.MockStorage, count int) { - logs := storagemock.NewMockRequestLogStore(ctrl) - store.EXPECT().GetRequestLogStore().Return(logs).Times(count) - logs.EXPECT().Get(gomock.Any(), projectStatusID, requestlog.RequestEventLogID(entity.RequestEventProjectFactsRecorded, "all")). - Return(entity.RequestLog{}, storage.ErrNotFound).Times(count) -} - func TestGetProjectStatusCompletion(t *testing.T) { - readErr := errors.New("completion read unavailable") - completion := requestlog.NewRequestEventLog(entity.Request{Queue: projectStatusQueue, ID: projectStatusID}, entity.RequestEventProjectFactsRecorded, "all", nil) - completion.TimestampMs = 30 for _, tt := range []struct { - name string - completion entity.RequestLog - readErr error - complete bool - empty bool - updatedAt int64 - inconsistent bool + name string + completedAt int64 + stateAt int64 + empty bool + updatedAt int64 }{ - {name: "terminal request still recording", readErr: storage.ErrNotFound, updatedAt: 20}, - {name: "completed projects", completion: completion, complete: true, updatedAt: 30}, - {name: "completed empty set", completion: completion, complete: true, empty: true, updatedAt: 30}, - {name: "completion read failed", readErr: readErr}, - {name: "completion belongs to another request", completion: func() entity.RequestLog { log := completion; log.RequestID = "8"; return log }(), inconsistent: true}, - {name: "newer lifecycle timestamp", completion: completion, complete: true, updatedAt: 40}, + {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" @@ -635,11 +614,11 @@ func TestGetProjectStatusCompletion(t *testing.T) { factory := storagemock.NewMockFactory(ctrl) store := storagemock.NewMockStorage(ctrl) summaries := storagemock.NewMockRequestSummaryStore(ctrl) - logs := storagemock.NewMockRequestLogStore(ctrl) facts := storagemock.NewMockValidationFactStore(ctrl) - summary := entity.RequestSummary{Queue: projectStatusQueue, RequestID: projectStatusID, URI: projectStatusURI, State: entity.RequestStateFailed, StateTimestampMs: 10} - if tt.updatedAt == 40 { - summary.StateTimestampMs = 40 + 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} @@ -651,36 +630,24 @@ func TestGetProjectStatusCompletion(t *testing.T) { uris.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil) } store.EXPECT().GetRequestSummaryStore().Return(summaries) - summaries.EXPECT().Get(gomock.Any(), projectStatusID).Return(summary, nil) - store.EXPECT().GetRequestLogStore().Return(logs) - readCompletion := logs.EXPECT().Get(gomock.Any(), projectStatusID, completion.ID).Return(tt.completion, tt.readErr) - if (tt.readErr == nil || storage.IsNotFound(tt.readErr)) && !tt.inconsistent { - store.EXPECT().GetValidationFactStore().Return(facts).After(readCompletion) - facts.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound) - fact := entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "p", Degree: 1, CreatedAt: 20} - if tt.empty { - facts.EXPECT().Get(gomock.Any(), projectStatusURI, "p").Return(entity.ValidationFact{}, storage.ErrNotFound) - } else { - facts.EXPECT().Get(gomock.Any(), projectStatusURI, "p").Return(fact, nil) - } + 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) - if tt.inconsistent { - require.Error(t, err) - assert.True(t, IsProjectStatusConsistency(err)) - } else if tt.readErr != nil && !storage.IsNotFound(tt.readErr) { - require.ErrorIs(t, err, tt.readErr) + 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 { - require.NoError(t, err) - assert.Equal(t, tt.complete, 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) + assert.Len(t, result.ProjectValidationFacts, 1) } + assert.NotEmpty(t, result.NextPageToken) }) } } diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 461452f89..d7cf45086 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -485,7 +485,7 @@ func TestRecordProjectFactsEmptyCompletion(t *testing.T) { if tt.resolveErr == nil { materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { - assert.Equal(t, requestlog.RequestEventLogID(entity.RequestEventProjectFactsRecorded, "all"), log.ID) + assert.Equal(t, "event/project_facts_recorded/all", log.ID) assert.Equal(t, "0", log.Metadata[requestlog.MetadataKeyProjectFactCount]) return tt.writeErr }) diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go index 29dd8c658..2eb424ae7 100644 --- a/stovepipe/core/requestlog/materializer.go +++ b/stovepipe/core/requestlog/materializer.go @@ -95,7 +95,7 @@ func NewRequestEventLog( metadata map[string]string, ) entity.RequestLog { return entity.RequestLog{ - ID: RequestEventLogID(event, occurrence), + ID: publish.IntentID(_occurrenceKindEvent, string(event), occurrence), Queue: request.Queue, RequestID: request.ID, Event: event, @@ -103,11 +103,6 @@ func NewRequestEventLog( } } -// RequestEventLogID returns the stable identity of an event occurrence within a request. -func RequestEventLogID(event entity.RequestEvent, occurrence string) string { - return publish.IntentID(_occurrenceKindEvent, string(event), occurrence) -} - func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, log entity.RequestLog) error { if log.TimestampMs == 0 { log.TimestampMs = m.now().UnixMilli() @@ -122,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/migrations/BUILD.bazel b/stovepipe/extension/storage/mysql/migrations/BUILD.bazel new file mode 100644 index 000000000..7ff6fff20 --- /dev/null +++ b/stovepipe/extension/storage/mysql/migrations/BUILD.bazel @@ -0,0 +1,12 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["migrations.go"], + embedsrcs = [ + "backfill_project_results_completion.sql", + "project_results_completion.sql", + ], + importpath = "github.com/uber/submitqueue/stovepipe/extension/storage/mysql/migrations", + visibility = ["//visibility:public"], +) diff --git a/stovepipe/extension/storage/mysql/migrations/README.md b/stovepipe/extension/storage/mysql/migrations/README.md new file mode 100644 index 000000000..32ec64773 --- /dev/null +++ b/stovepipe/extension/storage/mysql/migrations/README.md @@ -0,0 +1,16 @@ +# Project-result completion rollout + +Fresh databases use `../schema/request_summary.sql`. Existing databases must add the defaulted `project_results_completed_at_ms` column with `project_results_completion.sql` before running the new application version. Apply that schema migration once; keep the column when reverting the application. + +The backfill copies only the retained `event/project_facts_recorded/all` timestamp into existing request summaries whose completion timestamp is zero. It preserves lifecycle state and timestamps, leaves known completion unchanged, and increments the summary version so a concurrent projection must reload rather than overwrite completion. Requests without a completion event remain incomplete, including historical empty result sets. + +Run the backfill before relying on historical completion, and run it again after old application instances drain to cover events they recorded during rollout. Use a MySQL session connected to the Stovepipe storage database and select a queue and bounded request-ID range: + +```sql +SET @queue = 'example/main'; +SET @first_request_id = '1'; +SET @last_request_id = '1'; +SOURCE stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql; +``` + +Paths are relative to the repository root. Range bounds are inclusive and use the `VARCHAR` primary-key ordering, not numeric request-ID order. Repeat for each queue and range requiring historical coverage. Repeating a completed batch updates no rows. The scripts are packaged for integration tests, which exercise the schema upgrade, backfill idempotency, queue and range isolation, version changes, and both public status APIs. diff --git a/stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql b/stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql new file mode 100644 index 000000000..485815ee8 --- /dev/null +++ b/stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql @@ -0,0 +1,15 @@ +UPDATE request_summary AS summary +JOIN request_log AS completion + ON completion.queue = summary.queue + AND completion.request_id = summary.request_id + AND completion.log_id = 'event/project_facts_recorded/all' +SET summary.project_results_completed_at_ms = completion.timestamp_ms, + summary.version = summary.version + 1 +WHERE summary.queue = CAST(@queue AS BINARY) + AND summary.request_id >= CAST(@first_request_id AS BINARY) + AND summary.request_id <= CAST(@last_request_id AS BINARY) + AND summary.project_results_completed_at_ms = 0 + AND completion.event = 'project_facts_recorded' + AND completion.state = '' + AND completion.request_version = 0 + AND completion.timestamp_ms > 0; diff --git a/stovepipe/extension/storage/mysql/migrations/migrations.go b/stovepipe/extension/storage/mysql/migrations/migrations.go new file mode 100644 index 000000000..cd3da269b --- /dev/null +++ b/stovepipe/extension/storage/mysql/migrations/migrations.go @@ -0,0 +1,23 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package migrations provides SQL for upgrading existing Stovepipe storage. +package migrations + +import "embed" + +// Files contains migration and backfill scripts; fresh databases use the schema directory. +// +//go:embed *.sql +var Files embed.FS diff --git a/stovepipe/extension/storage/mysql/migrations/project_results_completion.sql b/stovepipe/extension/storage/mysql/migrations/project_results_completion.sql new file mode 100644 index 000000000..1be77786a --- /dev/null +++ b/stovepipe/extension/storage/mysql/migrations/project_results_completion.sql @@ -0,0 +1,2 @@ +ALTER TABLE request_summary + ADD COLUMN project_results_completed_at_ms BIGINT NOT NULL DEFAULT 0; 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/integration/stovepipe/BUILD.bazel b/test/integration/stovepipe/BUILD.bazel index 3d1397836..d787cd37f 100644 --- a/test/integration/stovepipe/BUILD.bazel +++ b/test/integration/stovepipe/BUILD.bazel @@ -19,6 +19,7 @@ go_test( "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage/mysql:go_default_library", + "//stovepipe/extension/storage/mysql/migrations:go_default_library", "//test/testutil:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/test/integration/stovepipe/suite_test.go b/test/integration/stovepipe/suite_test.go index 37314fa60..09ffe0627 100644 --- a/test/integration/stovepipe/suite_test.go +++ b/test/integration/stovepipe/suite_test.go @@ -41,6 +41,7 @@ import ( "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/stovepipe/extension/storage/mysql/migrations" "github.com/uber/submitqueue/test/testutil" "google.golang.org/grpc" "google.golang.org/protobuf/proto" @@ -384,7 +385,11 @@ func (s *StovepipeIntegrationSuite) TestProjectStatusAPI() { 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 - require.NoError(t, mysqlstorage.NewRequestLogStore(s.db, tally.NoopScope, queue).Create(s.ctx, completion)) + 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, }) @@ -454,3 +459,134 @@ func assertHistoryEvents(t *testing.T, events []*pb.HistoryEvent) { assert.Equal(t, "succeeded", events[2].GetRequestState()) assert.Equal(t, "build_succeeded", events[2].OutcomeReason) } + +func (s *StovepipeIntegrationSuite) TestProjectCompletionMigrationAndBackfill() { + t := s.T() + conn, err := s.db.Conn(s.ctx) + require.NoError(t, err) + defer conn.Close() + defer func() { + _, err := conn.ExecContext(s.ctx, "DROP TEMPORARY TABLE IF EXISTS request_summary, request_log, migration_request_summary, migration_request_log") + require.NoError(t, err) + }() + for _, query := range []string{ + "CREATE TEMPORARY TABLE migration_request_summary LIKE request_summary", + "ALTER TABLE migration_request_summary RENAME TO request_summary", + "CREATE TEMPORARY TABLE migration_request_log LIKE request_log", + "ALTER TABLE migration_request_log RENAME TO request_log", + "ALTER TABLE request_summary DROP COLUMN project_results_completed_at_ms", + `INSERT INTO request_summary (queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, version) + VALUES ('completion/main', '1', 'git://repo/1', '', 'failed', 3, 1000, 10), + ('completion/main', '2', 'git://repo/2', '', 'failed', 3, 1000, 10), + ('completion/main', '3', 'git://repo/3', '', 'failed', 3, 1000, 10), + ('completion/main', '4', 'git://repo/4', '', 'failed', 3, 1000, 10), + ('completion/main', '5', 'git://repo/5', '', 'failed', 3, 1000, 10), + ('completion/other', '1', 'git://repo/other', '', 'failed', 3, 1000, 10)`, + } { + _, err = conn.ExecContext(s.ctx, query) + require.NoError(t, err) + } + migration, err := migrations.Files.ReadFile("project_results_completion.sql") + require.NoError(t, err) + _, err = conn.ExecContext(s.ctx, string(migration)) + require.NoError(t, err) + var completedAt int64 + require.NoError(t, conn.QueryRowContext(s.ctx, "SELECT project_results_completed_at_ms FROM request_summary WHERE queue = 'completion/main' AND request_id = '1'").Scan(&completedAt)) + assert.Zero(t, completedAt) + _, err = conn.ExecContext(s.ctx, "UPDATE request_summary SET project_results_completed_at_ms = 7000 WHERE queue = 'completion/main' AND request_id = '2'") + require.NoError(t, err) + for _, record := range []struct { + queue, id string + timestamp int64 + }{ + {queue: "completion/main", id: "1", timestamp: 5000}, + {queue: "completion/main", id: "2", timestamp: 6000}, + {queue: "completion/main", id: "4", timestamp: 0}, + {queue: "completion/main", id: "5", timestamp: 5000}, + {queue: "completion/other", id: "1", timestamp: 5000}, + } { + _, err = conn.ExecContext(s.ctx, `INSERT INTO request_log + (queue, request_id, log_id, timestamp_ms, state, event, request_version, outcome_reason, metadata) + VALUES (?, ?, 'event/project_facts_recorded/all', ?, '', 'project_facts_recorded', 0, '', '{}')`, record.queue, record.id, record.timestamp) + require.NoError(t, err) + } + for _, query := range []string{"SET @queue = 'completion/main'", "SET @first_request_id = '1'", "SET @last_request_id = '4'"} { + _, err = conn.ExecContext(s.ctx, query) + require.NoError(t, err) + } + backfill, err := migrations.Files.ReadFile("backfill_project_results_completion.sql") + require.NoError(t, err) + for _, wantAffected := range []int64{1, 0} { + result, err := conn.ExecContext(s.ctx, string(backfill)) + require.NoError(t, err) + affected, err := result.RowsAffected() + require.NoError(t, err) + assert.Equal(t, wantAffected, affected) + } + rows, err := conn.QueryContext(s.ctx, "SELECT queue, request_id, project_results_completed_at_ms, version, state, request_version, state_timestamp_ms FROM request_summary ORDER BY queue, request_id") + require.NoError(t, err) + defer rows.Close() + count := 0 + for rows.Next() { + var queue, id, state string + var version, requestVersion int32 + var stateAt int64 + require.NoError(t, rows.Scan(&queue, &id, &completedAt, &version, &state, &requestVersion, &stateAt)) + wantCompletion, wantVersion := int64(0), int32(10) + if queue == "completion/main" && id == "1" { + wantCompletion, wantVersion = 5000, 11 + } else if queue == "completion/main" && id == "2" { + wantCompletion = 7000 + } + assert.Equal(t, wantCompletion, completedAt) + assert.Equal(t, wantVersion, version) + assert.Equal(t, "failed", state) + assert.EqualValues(t, 3, requestVersion) + assert.EqualValues(t, 1000, stateAt) + count++ + } + require.NoError(t, rows.Err()) + assert.Equal(t, 6, count) + result, err := conn.ExecContext(s.ctx, "UPDATE request_summary SET project_results_completed_at_ms = 0 WHERE queue = 'completion/main' AND request_id = '1' AND version = 10") + require.NoError(t, err) + affected, err := result.RowsAffected() + require.NoError(t, err) + assert.Zero(t, affected) +} + +func (s *StovepipeIntegrationSuite) TestHistoricalProjectCompletionBackfill() { + t := s.T() + const queue, id, uri = "completion-history/main", "1", "git://completion-history/main/head" + _, err := s.db.ExecContext(s.ctx, `INSERT INTO request_summary + (queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, version) + VALUES (?, ?, ?, '', 'failed', 3, 1000, 1)`, queue, id, uri) + require.NoError(t, err) + _, err = s.db.ExecContext(s.ctx, "INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, 1)", queue, uri, id) + require.NoError(t, err) + log := requestlog.NewRequestEventLog(entity.Request{Queue: queue, ID: id}, entity.RequestEventProjectFactsRecorded, "all", nil) + log.TimestampMs = 6000 + require.NoError(t, mysqlstorage.NewRequestLogStore(s.db, tally.NoopScope, queue).Create(s.ctx, log)) + before, err := s.client.GetProjectStatusByID(s.ctx, &pb.GetProjectStatusByIDRequest{Queue: queue, RequestId: id}) + require.NoError(t, err) + assert.False(t, before.ProjectResultsComplete) + + conn, err := s.db.Conn(s.ctx) + require.NoError(t, err) + defer conn.Close() + for name, value := range map[string]string{"queue": queue, "first_request_id": id, "last_request_id": id} { + _, err = conn.ExecContext(s.ctx, "SET @"+name+" = ?", value) + require.NoError(t, err) + } + backfill, err := migrations.Files.ReadFile("backfill_project_results_completion.sql") + require.NoError(t, err) + _, err = conn.ExecContext(s.ctx, string(backfill)) + require.NoError(t, err) + byID, err := s.client.GetProjectStatusByID(s.ctx, &pb.GetProjectStatusByIDRequest{Queue: queue, RequestId: id}) + require.NoError(t, err) + byURI, err := s.client.GetProjectStatusByURI(s.ctx, &pb.GetProjectStatusByURIRequest{Queue: queue, ChangeUri: uri}) + require.NoError(t, err) + assert.True(t, byID.GetProjectResultsComplete()) + assert.EqualValues(t, 6000, byID.GetUpdatedAtMs()) + assert.True(t, byURI.GetProjectResultsComplete()) + assert.EqualValues(t, 6000, byURI.GetUpdatedAtMs()) +} From ff4c6e2c12b8c34f8962f285d4f7275d074e5bce Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 23 Oct 2026 17:53:11 +0000 Subject: [PATCH 3/4] chore(stovepipe): remove completion migration and backfill tooling --- doc/rfc/stovepipe/project-status-api.md | 2 +- .../storage/mysql/migrations/BUILD.bazel | 12 -- .../storage/mysql/migrations/README.md | 16 --- .../backfill_project_results_completion.sql | 15 -- .../storage/mysql/migrations/migrations.go | 23 --- .../migrations/project_results_completion.sql | 2 - test/integration/stovepipe/BUILD.bazel | 1 - test/integration/stovepipe/suite_test.go | 132 ------------------ 8 files changed, 1 insertion(+), 202 deletions(-) delete mode 100644 stovepipe/extension/storage/mysql/migrations/BUILD.bazel delete mode 100644 stovepipe/extension/storage/mysql/migrations/README.md delete mode 100644 stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql delete mode 100644 stovepipe/extension/storage/mysql/migrations/migrations.go delete mode 100644 stovepipe/extension/storage/mysql/migrations/project_results_completion.sql diff --git a/doc/rfc/stovepipe/project-status-api.md b/doc/rfc/stovepipe/project-status-api.md index 5d85f305c..30d13632a 100644 --- a/doc/rfc/stovepipe/project-status-api.md +++ b/doc/rfc/stovepipe/project-status-api.md @@ -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` 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. Existing events require the [summary backfill](../../../stovepipe/extension/storage/mysql/migrations/README.md); older empty batches without an event remain incomplete. 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/stovepipe/extension/storage/mysql/migrations/BUILD.bazel b/stovepipe/extension/storage/mysql/migrations/BUILD.bazel deleted file mode 100644 index 7ff6fff20..000000000 --- a/stovepipe/extension/storage/mysql/migrations/BUILD.bazel +++ /dev/null @@ -1,12 +0,0 @@ -load("@rules_go//go:def.bzl", "go_library") - -go_library( - name = "go_default_library", - srcs = ["migrations.go"], - embedsrcs = [ - "backfill_project_results_completion.sql", - "project_results_completion.sql", - ], - importpath = "github.com/uber/submitqueue/stovepipe/extension/storage/mysql/migrations", - visibility = ["//visibility:public"], -) diff --git a/stovepipe/extension/storage/mysql/migrations/README.md b/stovepipe/extension/storage/mysql/migrations/README.md deleted file mode 100644 index 32ec64773..000000000 --- a/stovepipe/extension/storage/mysql/migrations/README.md +++ /dev/null @@ -1,16 +0,0 @@ -# Project-result completion rollout - -Fresh databases use `../schema/request_summary.sql`. Existing databases must add the defaulted `project_results_completed_at_ms` column with `project_results_completion.sql` before running the new application version. Apply that schema migration once; keep the column when reverting the application. - -The backfill copies only the retained `event/project_facts_recorded/all` timestamp into existing request summaries whose completion timestamp is zero. It preserves lifecycle state and timestamps, leaves known completion unchanged, and increments the summary version so a concurrent projection must reload rather than overwrite completion. Requests without a completion event remain incomplete, including historical empty result sets. - -Run the backfill before relying on historical completion, and run it again after old application instances drain to cover events they recorded during rollout. Use a MySQL session connected to the Stovepipe storage database and select a queue and bounded request-ID range: - -```sql -SET @queue = 'example/main'; -SET @first_request_id = '1'; -SET @last_request_id = '1'; -SOURCE stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql; -``` - -Paths are relative to the repository root. Range bounds are inclusive and use the `VARCHAR` primary-key ordering, not numeric request-ID order. Repeat for each queue and range requiring historical coverage. Repeating a completed batch updates no rows. The scripts are packaged for integration tests, which exercise the schema upgrade, backfill idempotency, queue and range isolation, version changes, and both public status APIs. diff --git a/stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql b/stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql deleted file mode 100644 index 485815ee8..000000000 --- a/stovepipe/extension/storage/mysql/migrations/backfill_project_results_completion.sql +++ /dev/null @@ -1,15 +0,0 @@ -UPDATE request_summary AS summary -JOIN request_log AS completion - ON completion.queue = summary.queue - AND completion.request_id = summary.request_id - AND completion.log_id = 'event/project_facts_recorded/all' -SET summary.project_results_completed_at_ms = completion.timestamp_ms, - summary.version = summary.version + 1 -WHERE summary.queue = CAST(@queue AS BINARY) - AND summary.request_id >= CAST(@first_request_id AS BINARY) - AND summary.request_id <= CAST(@last_request_id AS BINARY) - AND summary.project_results_completed_at_ms = 0 - AND completion.event = 'project_facts_recorded' - AND completion.state = '' - AND completion.request_version = 0 - AND completion.timestamp_ms > 0; diff --git a/stovepipe/extension/storage/mysql/migrations/migrations.go b/stovepipe/extension/storage/mysql/migrations/migrations.go deleted file mode 100644 index cd3da269b..000000000 --- a/stovepipe/extension/storage/mysql/migrations/migrations.go +++ /dev/null @@ -1,23 +0,0 @@ -// Copyright (c) 2026 Uber Technologies, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -// Package migrations provides SQL for upgrading existing Stovepipe storage. -package migrations - -import "embed" - -// Files contains migration and backfill scripts; fresh databases use the schema directory. -// -//go:embed *.sql -var Files embed.FS diff --git a/stovepipe/extension/storage/mysql/migrations/project_results_completion.sql b/stovepipe/extension/storage/mysql/migrations/project_results_completion.sql deleted file mode 100644 index 1be77786a..000000000 --- a/stovepipe/extension/storage/mysql/migrations/project_results_completion.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE request_summary - ADD COLUMN project_results_completed_at_ms BIGINT NOT NULL DEFAULT 0; diff --git a/test/integration/stovepipe/BUILD.bazel b/test/integration/stovepipe/BUILD.bazel index d787cd37f..3d1397836 100644 --- a/test/integration/stovepipe/BUILD.bazel +++ b/test/integration/stovepipe/BUILD.bazel @@ -19,7 +19,6 @@ go_test( "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage/mysql:go_default_library", - "//stovepipe/extension/storage/mysql/migrations:go_default_library", "//test/testutil:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/test/integration/stovepipe/suite_test.go b/test/integration/stovepipe/suite_test.go index 09ffe0627..7f483e9b5 100644 --- a/test/integration/stovepipe/suite_test.go +++ b/test/integration/stovepipe/suite_test.go @@ -41,7 +41,6 @@ import ( "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/stovepipe/extension/storage/mysql/migrations" "github.com/uber/submitqueue/test/testutil" "google.golang.org/grpc" "google.golang.org/protobuf/proto" @@ -459,134 +458,3 @@ func assertHistoryEvents(t *testing.T, events []*pb.HistoryEvent) { assert.Equal(t, "succeeded", events[2].GetRequestState()) assert.Equal(t, "build_succeeded", events[2].OutcomeReason) } - -func (s *StovepipeIntegrationSuite) TestProjectCompletionMigrationAndBackfill() { - t := s.T() - conn, err := s.db.Conn(s.ctx) - require.NoError(t, err) - defer conn.Close() - defer func() { - _, err := conn.ExecContext(s.ctx, "DROP TEMPORARY TABLE IF EXISTS request_summary, request_log, migration_request_summary, migration_request_log") - require.NoError(t, err) - }() - for _, query := range []string{ - "CREATE TEMPORARY TABLE migration_request_summary LIKE request_summary", - "ALTER TABLE migration_request_summary RENAME TO request_summary", - "CREATE TEMPORARY TABLE migration_request_log LIKE request_log", - "ALTER TABLE migration_request_log RENAME TO request_log", - "ALTER TABLE request_summary DROP COLUMN project_results_completed_at_ms", - `INSERT INTO request_summary (queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, version) - VALUES ('completion/main', '1', 'git://repo/1', '', 'failed', 3, 1000, 10), - ('completion/main', '2', 'git://repo/2', '', 'failed', 3, 1000, 10), - ('completion/main', '3', 'git://repo/3', '', 'failed', 3, 1000, 10), - ('completion/main', '4', 'git://repo/4', '', 'failed', 3, 1000, 10), - ('completion/main', '5', 'git://repo/5', '', 'failed', 3, 1000, 10), - ('completion/other', '1', 'git://repo/other', '', 'failed', 3, 1000, 10)`, - } { - _, err = conn.ExecContext(s.ctx, query) - require.NoError(t, err) - } - migration, err := migrations.Files.ReadFile("project_results_completion.sql") - require.NoError(t, err) - _, err = conn.ExecContext(s.ctx, string(migration)) - require.NoError(t, err) - var completedAt int64 - require.NoError(t, conn.QueryRowContext(s.ctx, "SELECT project_results_completed_at_ms FROM request_summary WHERE queue = 'completion/main' AND request_id = '1'").Scan(&completedAt)) - assert.Zero(t, completedAt) - _, err = conn.ExecContext(s.ctx, "UPDATE request_summary SET project_results_completed_at_ms = 7000 WHERE queue = 'completion/main' AND request_id = '2'") - require.NoError(t, err) - for _, record := range []struct { - queue, id string - timestamp int64 - }{ - {queue: "completion/main", id: "1", timestamp: 5000}, - {queue: "completion/main", id: "2", timestamp: 6000}, - {queue: "completion/main", id: "4", timestamp: 0}, - {queue: "completion/main", id: "5", timestamp: 5000}, - {queue: "completion/other", id: "1", timestamp: 5000}, - } { - _, err = conn.ExecContext(s.ctx, `INSERT INTO request_log - (queue, request_id, log_id, timestamp_ms, state, event, request_version, outcome_reason, metadata) - VALUES (?, ?, 'event/project_facts_recorded/all', ?, '', 'project_facts_recorded', 0, '', '{}')`, record.queue, record.id, record.timestamp) - require.NoError(t, err) - } - for _, query := range []string{"SET @queue = 'completion/main'", "SET @first_request_id = '1'", "SET @last_request_id = '4'"} { - _, err = conn.ExecContext(s.ctx, query) - require.NoError(t, err) - } - backfill, err := migrations.Files.ReadFile("backfill_project_results_completion.sql") - require.NoError(t, err) - for _, wantAffected := range []int64{1, 0} { - result, err := conn.ExecContext(s.ctx, string(backfill)) - require.NoError(t, err) - affected, err := result.RowsAffected() - require.NoError(t, err) - assert.Equal(t, wantAffected, affected) - } - rows, err := conn.QueryContext(s.ctx, "SELECT queue, request_id, project_results_completed_at_ms, version, state, request_version, state_timestamp_ms FROM request_summary ORDER BY queue, request_id") - require.NoError(t, err) - defer rows.Close() - count := 0 - for rows.Next() { - var queue, id, state string - var version, requestVersion int32 - var stateAt int64 - require.NoError(t, rows.Scan(&queue, &id, &completedAt, &version, &state, &requestVersion, &stateAt)) - wantCompletion, wantVersion := int64(0), int32(10) - if queue == "completion/main" && id == "1" { - wantCompletion, wantVersion = 5000, 11 - } else if queue == "completion/main" && id == "2" { - wantCompletion = 7000 - } - assert.Equal(t, wantCompletion, completedAt) - assert.Equal(t, wantVersion, version) - assert.Equal(t, "failed", state) - assert.EqualValues(t, 3, requestVersion) - assert.EqualValues(t, 1000, stateAt) - count++ - } - require.NoError(t, rows.Err()) - assert.Equal(t, 6, count) - result, err := conn.ExecContext(s.ctx, "UPDATE request_summary SET project_results_completed_at_ms = 0 WHERE queue = 'completion/main' AND request_id = '1' AND version = 10") - require.NoError(t, err) - affected, err := result.RowsAffected() - require.NoError(t, err) - assert.Zero(t, affected) -} - -func (s *StovepipeIntegrationSuite) TestHistoricalProjectCompletionBackfill() { - t := s.T() - const queue, id, uri = "completion-history/main", "1", "git://completion-history/main/head" - _, err := s.db.ExecContext(s.ctx, `INSERT INTO request_summary - (queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, version) - VALUES (?, ?, ?, '', 'failed', 3, 1000, 1)`, queue, id, uri) - require.NoError(t, err) - _, err = s.db.ExecContext(s.ctx, "INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, 1)", queue, uri, id) - require.NoError(t, err) - log := requestlog.NewRequestEventLog(entity.Request{Queue: queue, ID: id}, entity.RequestEventProjectFactsRecorded, "all", nil) - log.TimestampMs = 6000 - require.NoError(t, mysqlstorage.NewRequestLogStore(s.db, tally.NoopScope, queue).Create(s.ctx, log)) - before, err := s.client.GetProjectStatusByID(s.ctx, &pb.GetProjectStatusByIDRequest{Queue: queue, RequestId: id}) - require.NoError(t, err) - assert.False(t, before.ProjectResultsComplete) - - conn, err := s.db.Conn(s.ctx) - require.NoError(t, err) - defer conn.Close() - for name, value := range map[string]string{"queue": queue, "first_request_id": id, "last_request_id": id} { - _, err = conn.ExecContext(s.ctx, "SET @"+name+" = ?", value) - require.NoError(t, err) - } - backfill, err := migrations.Files.ReadFile("backfill_project_results_completion.sql") - require.NoError(t, err) - _, err = conn.ExecContext(s.ctx, string(backfill)) - require.NoError(t, err) - byID, err := s.client.GetProjectStatusByID(s.ctx, &pb.GetProjectStatusByIDRequest{Queue: queue, RequestId: id}) - require.NoError(t, err) - byURI, err := s.client.GetProjectStatusByURI(s.ctx, &pb.GetProjectStatusByURIRequest{Queue: queue, ChangeUri: uri}) - require.NoError(t, err) - assert.True(t, byID.GetProjectResultsComplete()) - assert.EqualValues(t, 6000, byID.GetUpdatedAtMs()) - assert.True(t, byURI.GetProjectResultsComplete()) - assert.EqualValues(t, 6000, byURI.GetUpdatedAtMs()) -} From 55e6b9367e1f7f9fff808ff7b185549993433e7c Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 23 Oct 2026 18:50:16 +0000 Subject: [PATCH 4/4] fix(stovepipe): update web fixtures for completion summary schema --- stovepipe/controller/record/record_test.go | 28 +++++++++++++++------- web/test/e2e/stovepipe/run_e2e.py | 5 +++- 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index bc79039d9..95dc05b93 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -405,7 +405,7 @@ func TestRecordProjectFacts_PassesEmptyTerminalBuild(t *testing.T) { require.NoError(t, err) } -func TestRecordProjectFactsEmptyCompletion(t *testing.T) { +func TestProcessRecordsEmptyProjectCompletion(t *testing.T) { resolveErr := errors.New("project resolution unavailable") writeErr := errors.New("completion write unavailable") for _, tt := range []struct { @@ -423,6 +423,7 @@ func TestRecordProjectFactsEmptyCompletion(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 @@ -431,21 +432,32 @@ func TestRecordProjectFactsEmptyCompletion(t *testing.T) { materializer := requestlogmock.NewMockMaterializer(ctrl) c.materializer = materializer if tt.resolveErr == 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 - }) + 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.recordProjectFacts(queueContext(), m.store, request) + 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) } }) } 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}');"