From f87bf65794ae92ff4a6123e6c8a45545a3061173 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Wed, 21 Oct 2026 19:54:55 +0000 Subject: [PATCH] feat(stovepipe): Add project ordering index --- doc/rfc/stovepipe/project-validation-index.md | 17 +++ stovepipe/controller/record/record.go | 11 +- stovepipe/controller/record/record_test.go | 56 ++++++++ stovepipe/entity/BUILD.bazel | 2 + stovepipe/entity/project_validation_index.go | 45 ++++++ .../entity/project_validation_index_test.go | 52 +++++++ stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 2 + stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../project_validation_index_store_mock.go | 71 ++++++++++ .../extension/storage/mock/storage_mock.go | 14 ++ stovepipe/extension/storage/mysql/BUILD.bazel | 2 + .../mysql/project_validation_index_store.go | 97 +++++++++++++ .../project_validation_index_store_test.go | 129 ++++++++++++++++++ .../mysql/schema/project_validation_index.sql | 10 ++ stovepipe/extension/storage/mysql/storage.go | 46 ++++--- .../extension/storage/mysql/storage_test.go | 9 ++ .../storage/project_validation_index_store.go | 33 +++++ stovepipe/extension/storage/storage.go | 2 + .../extension/storage/mysql/BUILD.bazel | 2 + .../mysql/project_validation_index_test.go | 87 ++++++++++++ .../extension/storage/mysql/storage_test.go | 6 + web/test/e2e/stovepipe/run_e2e.py | 8 ++ 23 files changed, 682 insertions(+), 21 deletions(-) create mode 100644 doc/rfc/stovepipe/project-validation-index.md create mode 100644 stovepipe/entity/project_validation_index.go create mode 100644 stovepipe/entity/project_validation_index_test.go create mode 100644 stovepipe/extension/storage/mock/project_validation_index_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/project_validation_index_store.go create mode 100644 stovepipe/extension/storage/mysql/project_validation_index_store_test.go create mode 100644 stovepipe/extension/storage/mysql/schema/project_validation_index.sql create mode 100644 stovepipe/extension/storage/project_validation_index_store.go create mode 100644 test/integration/stovepipe/extension/storage/mysql/project_validation_index_test.go diff --git a/doc/rfc/stovepipe/project-validation-index.md b/doc/rfc/stovepipe/project-validation-index.md new file mode 100644 index 000000000..9b64c0789 --- /dev/null +++ b/doc/rfc/stovepipe/project-validation-index.md @@ -0,0 +1,17 @@ +# Project validation ordering index + +The project validation index is an immutable mapping from `(queue, URI, degree_order, project)` to the fact keyed by `(queue, URI, project)`. It contains no duplicate validation payload. Its primary key supports bounded range reads in descending breakage degree and ascending bytewise project ID, without sorting a non-key attribute or scanning every fact. + +`degree_order` reverses nonnegative IEEE-754 degree bit patterns into an ascending integer key without rounding. Positive and negative zero share one key; invalid degrees and whole-repository facts cannot produce mappings. Binary identifiers preserve bytewise tie order. + +## Persistence and repair + +The record controller persists the winning immutable fact before ensuring its mapping. Redelivery repairs missing mappings even when the fact already exists, and derives their keys from the stored degree rather than a losing candidate. Project-recording logs and downstream terminal notifications follow successful mapping writes. No cross-store transaction is required. + +## Rollout + +The ordering table is additive; the original fact identity and primary key are unchanged. Install the table before deploying index writers, then adopt index-based readers after the writers are deployed. + +The table starts empty. There is no historical backfill or fallback scan: older requests may have empty project listings, and requests spanning the rollout may have partial listings. Existing facts remain available through explicit project lookups. This historical-coverage limitation is intentional; no data migration or compatibility state is required. + +Keep the additive table when reverting application adoption. Facts recorded while index writers are disabled remain outside index-based listings unless the normal record step processes them again. diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 2b6ef37df..b8af9fe3d 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -197,7 +197,7 @@ func (c *Controller) recordProjectFacts(ctx context.Context, store storage.Stora return err } for _, result := range results { - _, _, err := c.recordValidationFact(ctx, store.GetValidationFactStore(), entity.ValidationFact{ + fact, _, err := c.recordValidationFact(ctx, store.GetValidationFactStore(), entity.ValidationFact{ URI: request.URI, Project: result.Project, Degree: result.Degree, @@ -207,6 +207,15 @@ func (c *Controller) recordProjectFacts(ctx context.Context, store storage.Stora if err != nil { return err } + + entry, err := entity.NewProjectValidationIndex(fact) + if err != nil { + return err + } + + if err := store.GetProjectValidationIndexStore().Create(ctx, entry); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + 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 diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 0743ef77a..257725310 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -74,6 +74,7 @@ type recordMocks struct { reqStore *storagemock.MockRequestStore queueStore *storagemock.MockQueueStore factStore *storagemock.MockValidationFactStore + indexStore *storagemock.MockProjectValidationIndexStore store *storagemock.MockStorage materializer *requestlogmock.MockMaterializer sourceControl *sourcecontrolmock.MockSourceControl @@ -148,6 +149,7 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu reqStore: storagemock.NewMockRequestStore(ctrl), queueStore: storagemock.NewMockQueueStore(ctrl), factStore: storagemock.NewMockValidationFactStore(ctrl), + indexStore: storagemock.NewMockProjectValidationIndexStore(ctrl), store: storagemock.NewMockStorage(ctrl), materializer: requestlogmock.NewMockMaterializer(ctrl), sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), @@ -158,6 +160,7 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu m.store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() m.store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() m.store.EXPECT().GetValidationFactStore().Return(m.factStore).AnyTimes() + m.store.EXPECT().GetProjectValidationIndexStore().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) @@ -402,6 +405,13 @@ func TestProcess_RecordsNamedProjectStatusResults(t *testing.T) { }). Times(3) + var indexes []entity.ProjectValidationIndex + m.indexStore.EXPECT().Create(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, entry entity.ProjectValidationIndex) error { + indexes = append(indexes, entry) + return nil + }).Times(2) + var logs []entity.RequestLog eventMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()). DoAndReturn(func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { @@ -412,6 +422,12 @@ func TestProcess_RecordsNamedProjectStatusResults(t *testing.T) { require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) require.Len(t, facts, 3) + require.Len(t, indexes, 2) + for i, fact := range facts[1:] { + entry, err := entity.NewProjectValidationIndex(fact) + require.NoError(t, err) + assert.Equal(t, entry, indexes[i]) + } assert.Equal(t, []string{"", "project-a", "project-b"}, []string{facts[0].Project, facts[1].Project, facts[2].Project}) for _, fact := range facts { assert.Equal(t, testURI, fact.URI) @@ -1217,3 +1233,43 @@ func TestProcess_QueueMismatchFails(t *testing.T) { require.Error(t, c.Process(queueContext(), d)) } + +func TestRecordProjectFactsRepairsIndexOnRetry(t *testing.T) { + for _, retryErr := range []error{nil, storage.ErrAlreadyExists} { + name := "index repaired" + if retryErr != nil { + name = "index already repaired" + } + t.Run(name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := requestWithState(entity.RequestStateFailed) + resolverFactory := projectstatusmock.NewMockResolverFactory(ctrl) + resolver := projectstatusmock.NewMockResolver(ctrl) + c.projectStatusResolverFactory = resolverFactory + resolverFactory.EXPECT().For(projectstatus.ResolverConfig{QueueName: testQueue}).Return(resolver, nil).Times(2) + stored := entity.ValidationFact{URI: request.URI, Project: "p", Degree: 0.25, RequestID: request.ID, CreatedAt: 10} + entry, err := entity.NewProjectValidationIndex(stored) + require.NoError(t, err) + writeErr := errors.New("index write unavailable") + materializer := requestlogmock.NewMockMaterializer(ctrl) + c.materializer = materializer + + gomock.InOrder( + resolver.EXPECT().Resolve(gomock.Any(), request, request.TerminalBuildID).Return([]projectstatus.Result{{Project: "p", Degree: 0.25}}, nil), + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil), + m.indexStore.EXPECT().Create(gomock.Any(), entry).Return(writeErr), + resolver.EXPECT().Resolve(gomock.Any(), request, request.TerminalBuildID).Return([]projectstatus.Result{{Project: "p", Degree: 0.75}}, nil), + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists), + m.factStore.EXPECT().Get(gomock.Any(), request.URI, "p").Return(stored, nil), + m.indexStore.EXPECT().Create(gomock.Any(), entry).Return(retryErr), + materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, entity.RequestEventProjectFactsRecorded, log.Event) + return nil + }), + ) + require.ErrorIs(t, c.recordProjectFacts(queueContext(), m.store, request), writeErr) + require.NoError(t, c.recordProjectFacts(queueContext(), m.store, request)) + }) + } +} diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index f5269f909..3216aee55 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -8,6 +8,7 @@ go_library( "ingest.go", "list.go", "project_status.go", + "project_validation_index.go", "queue.go", "queue_config.go", "queue_log.go", @@ -31,6 +32,7 @@ go_test( name = "go_default_test", srcs = [ "build_test.go", + "project_validation_index_test.go", "queue_state_test.go", "request_id_test.go", "request_log_test.go", diff --git a/stovepipe/entity/project_validation_index.go b/stovepipe/entity/project_validation_index.go new file mode 100644 index 000000000..19f46f88b --- /dev/null +++ b/stovepipe/entity/project_validation_index.go @@ -0,0 +1,45 @@ +// 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 entity + +import ( + "fmt" + "math" +) + +// ProjectValidationIndex maps an ordered key to an immutable project fact within a queue. +type ProjectValidationIndex struct { + // URI is the commit containing the project result. + URI string + // DegreeOrder is the ascending integer key for descending breakage degree. + DegreeOrder int64 + // Project identifies the fact and breaks equal-degree ties in ascending bytewise order. + Project string +} + +// NewProjectValidationIndex derives an exact ordering key from a named project's fact. +// Invalid degrees, empty URIs and whole-repository facts return an error. +func NewProjectValidationIndex(fact ValidationFact) (ProjectValidationIndex, error) { + if fact.URI == "" || fact.Project == "" || math.IsNaN(fact.Degree) || fact.Degree < DegreeGreen || fact.Degree > DegreeBroken { + return ProjectValidationIndex{}, fmt.Errorf("project validation index requires a URI, project and degree in [0, 1]") + } + degree := fact.Degree + if degree == 0 { + degree = 0 // Canonicalize negative zero to the same key as positive zero. + } + // Nonnegative IEEE-754 bit patterns increase with degree; integer subtraction reverses that order without rounding. + order := int64(math.Float64bits(DegreeBroken) - math.Float64bits(degree)) + return ProjectValidationIndex{URI: fact.URI, DegreeOrder: order, Project: fact.Project}, nil +} diff --git a/stovepipe/entity/project_validation_index_test.go b/stovepipe/entity/project_validation_index_test.go new file mode 100644 index 000000000..b0cea1645 --- /dev/null +++ b/stovepipe/entity/project_validation_index_test.go @@ -0,0 +1,52 @@ +// 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 entity + +import ( + "math" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestProjectValidationIndexOrdersDegreesExactly(t *testing.T) { + degrees := []float64{1, math.Nextafter(1, 0), 0.75, math.Nextafter(0.5, 1), 0.5, math.SmallestNonzeroFloat64, 0} + var previous ProjectValidationIndex + for i, degree := range degrees { + entry, err := NewProjectValidationIndex(ValidationFact{URI: "commit", Project: "project", Degree: degree}) + require.NoError(t, err) + if i > 0 { + assert.Less(t, previous.DegreeOrder, entry.DegreeOrder) + } + previous = entry + } + negativeZero, err := NewProjectValidationIndex(ValidationFact{URI: "commit", Project: "project", Degree: math.Copysign(0, -1)}) + require.NoError(t, err) + assert.Equal(t, previous, negativeZero) +} + +func TestProjectValidationIndexRejectsInvalidFacts(t *testing.T) { + for _, fact := range []ValidationFact{ + {Project: "p"}, {URI: "commit"}, + {URI: "commit", Project: "p", Degree: math.NaN()}, + {URI: "commit", Project: "p", Degree: math.Inf(1)}, + {URI: "commit", Project: "p", Degree: -0.1}, + {URI: "commit", Project: "p", Degree: 1.1}, + } { + _, err := NewProjectValidationIndex(fact) + require.Error(t, err) + } +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index fc12fdf84..99b35bd64 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "build_store.go", + "project_validation_index_store.go", "queue_log_store.go", "queue_store.go", "queue_summary_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index de79c8959..316469d10 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -23,3 +23,5 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte `RequestAcceptanceStore` is the ordered lookup mapping for List: immutable `(queue, accepted_at_ms, request_id)` keys with bounded primary-key range scans, not a secondary index or a second status projection. The request-log materializer ensures each known acceptance has a mapping after summary persistence, including retries where the summary is unchanged. Unknown acceptance times are not mapped; historical backfill is separate. `QueueStore` owns internal enablement alongside pipeline coordination. `QueueSummaryStore` persists the current read model projected from immutable `QueueLogStore` occurrences. Enablement sequences, operational row versions and projection versions are distinct. These stores use primary-key operations; version arithmetic, duplicate reconciliation and projection repair belong to business logic. See [queue state](../../../doc/rfc/stovepipe/queue-state.md). + +`ProjectValidationIndexStore` is the ordered lookup mapping for named project facts. Its immutable `(queue, URI, degree_order, project)` primary key encodes descending degree and ascending bytewise project ID, so listing is a bounded key-range read. It holds no duplicate validation payload; readers hydrate facts by their original keys and verify the mapping. The record controller ensures mappings from persisted winning facts, including retries after partial writes. The table starts empty: only results processed by index-aware writers receive mappings, with no historical population or read-time fallback. diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index dab7cef72..b6090c9f9 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "build_store_mock.go", + "project_validation_index_store_mock.go", "queue_log_store_mock.go", "queue_store_mock.go", "queue_summary_store_mock.go", diff --git a/stovepipe/extension/storage/mock/project_validation_index_store_mock.go b/stovepipe/extension/storage/mock/project_validation_index_store_mock.go new file mode 100644 index 000000000..e9c65add8 --- /dev/null +++ b/stovepipe/extension/storage/mock/project_validation_index_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: project_validation_index_store.go +// +// Generated by this command: +// +// mockgen -source=project_validation_index_store.go -destination=mock/project_validation_index_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockProjectValidationIndexStore is a mock of ProjectValidationIndexStore interface. +type MockProjectValidationIndexStore struct { + ctrl *gomock.Controller + recorder *MockProjectValidationIndexStoreMockRecorder + isgomock struct{} +} + +// MockProjectValidationIndexStoreMockRecorder is the mock recorder for MockProjectValidationIndexStore. +type MockProjectValidationIndexStoreMockRecorder struct { + mock *MockProjectValidationIndexStore +} + +// NewMockProjectValidationIndexStore creates a new mock instance. +func NewMockProjectValidationIndexStore(ctrl *gomock.Controller) *MockProjectValidationIndexStore { + mock := &MockProjectValidationIndexStore{ctrl: ctrl} + mock.recorder = &MockProjectValidationIndexStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockProjectValidationIndexStore) EXPECT() *MockProjectValidationIndexStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockProjectValidationIndexStore) Create(ctx context.Context, entry entity.ProjectValidationIndex) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, entry) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockProjectValidationIndexStoreMockRecorder) Create(ctx, entry any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockProjectValidationIndexStore)(nil).Create), ctx, entry) +} + +// List mocks base method. +func (m *MockProjectValidationIndexStore) List(ctx context.Context, uri string, after entity.ProjectValidationIndex, limit int) ([]entity.ProjectValidationIndex, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, uri, after, limit) + ret0, _ := ret[0].([]entity.ProjectValidationIndex) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockProjectValidationIndexStoreMockRecorder) List(ctx, uri, after, limit any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockProjectValidationIndexStore)(nil).List), ctx, uri, after, limit) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index 11d137b25..1998e8694 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -93,6 +93,20 @@ func (mr *MockStorageMockRecorder) GetBuildStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBuildStore", reflect.TypeOf((*MockStorage)(nil).GetBuildStore)) } +// GetProjectValidationIndexStore mocks base method. +func (m *MockStorage) GetProjectValidationIndexStore() storage.ProjectValidationIndexStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetProjectValidationIndexStore") + ret0, _ := ret[0].(storage.ProjectValidationIndexStore) + return ret0 +} + +// GetProjectValidationIndexStore indicates an expected call of GetProjectValidationIndexStore. +func (mr *MockStorageMockRecorder) GetProjectValidationIndexStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetProjectValidationIndexStore", reflect.TypeOf((*MockStorage)(nil).GetProjectValidationIndexStore)) +} + // GetQueueLogStore mocks base method. func (m *MockStorage) GetQueueLogStore() storage.QueueLogStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 26edd5e8b..b09e89631 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "build_store.go", + "project_validation_index_store.go", "queue_log_store.go", "queue_store.go", "queue_summary_store.go", @@ -30,6 +31,7 @@ go_test( name = "go_default_test", srcs = [ "build_store_test.go", + "project_validation_index_store_test.go", "queue_store_test.go", "queue_summary_store_test.go", "request_acceptance_store_test.go", diff --git a/stovepipe/extension/storage/mysql/project_validation_index_store.go b/stovepipe/extension/storage/mysql/project_validation_index_store.go new file mode 100644 index 000000000..a1912f8fe --- /dev/null +++ b/stovepipe/extension/storage/mysql/project_validation_index_store.go @@ -0,0 +1,97 @@ +// 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 mysql + +import ( + "context" + "database/sql" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +type projectValidationIndexStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewProjectValidationIndexStore creates a queue-scoped, primary-key-ordered mapping store. +func NewProjectValidationIndexStore(db *sql.DB, scope tally.Scope, queue string) storage.ProjectValidationIndexStore { + return &projectValidationIndexStore{db: db, scope: scope, queue: queue} +} + +func (s *projectValidationIndexStore) Create(ctx context.Context, entry entity.ProjectValidationIndex) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if entry.URI == "" || entry.Project == "" || entry.DegreeOrder < 0 { + return fmt.Errorf("project validation index requires a URI, project and nonnegative degree order") + } + + _, err := s.db.ExecContext(ctx, ` + INSERT INTO project_validation_index (queue, uri, degree_order, project) + VALUES (?, ?, ?, ?)`, + s.queue, entry.URI, entry.DegreeOrder, entry.Project, + ) + if err != nil { + if isDuplicateEntry(err) { + return fmt.Errorf("project validation index uri=%q project=%q: %w", entry.URI, entry.Project, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to create project validation index uri=%q project=%q: %w", entry.URI, entry.Project, err) + } + + return nil +} + +func (s *projectValidationIndexStore) List(ctx context.Context, uri string, after entity.ProjectValidationIndex, limit int) (ret []entity.ProjectValidationIndex, retErr error) { + op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if uri == "" || limit <= 0 { + return nil, fmt.Errorf("project validation index requires a URI and positive limit") + } + if after != (entity.ProjectValidationIndex{}) && (after.DegreeOrder < 0 || after.Project == "" || after.URI != uri) { + return nil, fmt.Errorf("project validation index requires a URI, positive limit and matching cursor or zero value") + } + + rows, err := s.db.QueryContext(ctx, ` + SELECT uri, degree_order, project FROM project_validation_index + WHERE queue = ? AND uri = ? + AND (degree_order > ? OR (degree_order = ? AND project > ?)) + ORDER BY degree_order ASC, project ASC LIMIT ?`, + s.queue, uri, after.DegreeOrder, after.DegreeOrder, after.Project, limit, + ) + if err != nil { + return nil, fmt.Errorf("failed to list project validation index queue=%q uri=%q: %w", s.queue, uri, err) + } + defer rows.Close() + + for rows.Next() { + var entry entity.ProjectValidationIndex + if err := rows.Scan(&entry.URI, &entry.DegreeOrder, &entry.Project); err != nil { + return nil, fmt.Errorf("failed to scan project validation index: %w", err) + } + ret = append(ret, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate project validation index: %w", err) + } + + return ret, nil +} diff --git a/stovepipe/extension/storage/mysql/project_validation_index_store_test.go b/stovepipe/extension/storage/mysql/project_validation_index_store_test.go new file mode 100644 index 000000000..7bf8723fc --- /dev/null +++ b/stovepipe/extension/storage/mysql/project_validation_index_store_test.go @@ -0,0 +1,129 @@ +// 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 mysql + +import ( + "context" + "errors" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func TestProjectValidationIndexStoreCreate(t *testing.T) { + entry := entity.ProjectValidationIndex{URI: "commit", DegreeOrder: 7, Project: "p"} + readErr := errors.New("connection reset") + for _, tt := range []struct { + name string + dbErr error + wantErr error + }{ + {name: "created"}, + {name: "duplicate", dbErr: &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, wantErr: storage.ErrAlreadyExists}, + {name: "write failure", dbErr: readErr, wantErr: readErr}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + query := mock.ExpectExec("INSERT INTO project_validation_index").WithArgs("queue", entry.URI, entry.DegreeOrder, entry.Project) + if tt.dbErr != nil { + query.WillReturnError(tt.dbErr) + } else { + query.WillReturnResult(sqlmock.NewResult(0, 1)) + } + store := NewProjectValidationIndexStore(db, testMetrics(), "queue") + err = store.Create(context.Background(), entry) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestProjectValidationIndexStoreList(t *testing.T) { + columns := []string{"uri", "degree_order", "project"} + want := entity.ProjectValidationIndex{URI: "commit", DegreeOrder: 7, Project: "p"} + readErr := errors.New("connection reset") + for _, after := range []entity.ProjectValidationIndex{{}, {URI: "commit", DegreeOrder: 3, Project: "z"}} { + tests := []struct { + name string + rows *sqlmock.Rows + queryErr error + wantErr bool + want []entity.ProjectValidationIndex + }{ + {name: "found", rows: sqlmock.NewRows(columns).AddRow(want.URI, want.DegreeOrder, want.Project), want: []entity.ProjectValidationIndex{want}}, + {name: "empty", rows: sqlmock.NewRows(columns)}, + {name: "query failure", queryErr: readErr, wantErr: true}, + {name: "scan failure", rows: sqlmock.NewRows(columns).AddRow(want.URI, "invalid", want.Project), wantErr: true}, + {name: "iteration failure", rows: sqlmock.NewRows(columns).AddRow(want.URI, want.DegreeOrder, want.Project).RowError(0, readErr), wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name+"/"+after.Project, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + query := mock.ExpectQuery("SELECT uri, degree_order, project FROM project_validation_index").WithArgs("queue", "commit", after.DegreeOrder, after.DegreeOrder, after.Project, 3) + if tt.queryErr != nil { + query.WillReturnError(tt.queryErr) + } else { + query.WillReturnRows(tt.rows) + } + store := NewProjectValidationIndexStore(db, testMetrics(), "queue") + got, err := store.List(context.Background(), "commit", after, 3) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + assert.Equal(t, tt.want, got) + } + if tt.queryErr != nil { + assert.ErrorIs(t, err, tt.queryErr) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } + } +} + +func TestProjectValidationIndexStoreRejectsInvalidInput(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewProjectValidationIndexStore(db, testMetrics(), "queue") + for _, entry := range []entity.ProjectValidationIndex{{Project: "p"}, {URI: "commit"}, {URI: "commit", Project: "p", DegreeOrder: -1}} { + require.Error(t, store.Create(context.Background(), entry)) + } + for _, after := range []entity.ProjectValidationIndex{{DegreeOrder: -1}, {DegreeOrder: 1}, {URI: "commit"}, {URI: "other", Project: "p"}} { + _, err := store.List(context.Background(), "commit", after, 3) + require.Error(t, err) + } + for _, limit := range []int{0, -1} { + _, err := store.List(context.Background(), "commit", entity.ProjectValidationIndex{}, limit) + require.Error(t, err) + } + _, err = store.List(context.Background(), "", entity.ProjectValidationIndex{}, 3) + require.Error(t, err) + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/stovepipe/extension/storage/mysql/schema/project_validation_index.sql b/stovepipe/extension/storage/mysql/schema/project_validation_index.sql new file mode 100644 index 000000000..2d31612b7 --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/project_validation_index.sql @@ -0,0 +1,10 @@ +-- project_validation_index is a rebuildable ordering mapping for named project facts. +-- degree_order is an exact integer encoding of descending degree, so listing uses +-- a bounded primary-key range rather than sorting a non-key attribute. +CREATE TABLE IF NOT EXISTS project_validation_index ( + queue VARBINARY(255) NOT NULL, + uri VARBINARY(255) NOT NULL, + degree_order BIGINT NOT NULL, + project VARBINARY(255) NOT NULL, + PRIMARY KEY (queue, uri, degree_order, project) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index e3961c95a..28504aa97 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -46,16 +46,17 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { } queueScope := s.scope.Tagged(map[string]string{"queue": queueName}) return &mysqlStorage{ - requestStore: NewRequestStore(s.db, queueScope.SubScope("request_store"), queueName), - requestURIStore: NewRequestURIStore(s.db, queueScope.SubScope("request_uri_store"), queueName), - requestLogStore: NewRequestLogStore(s.db, queueScope.SubScope("request_log_store"), queueName), - requestSummaryStore: NewRequestSummaryStore(s.db, queueScope.SubScope("request_summary_store"), queueName), - requestAcceptanceStore: NewRequestAcceptanceStore(s.db, queueScope.SubScope("request_acceptance_store"), queueName), - queueSummaryStore: NewQueueSummaryStore(s.db, queueScope.SubScope("queue_summary_store"), queueName), - queueLogStore: NewQueueLogStore(s.db, queueScope.SubScope("queue_log_store"), queueName), - queueStore: NewQueueStore(s.db, queueScope.SubScope("queue_store"), queueName), - buildStore: NewBuildStore(s.db, queueScope.SubScope("build_store"), queueName), - validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), + requestStore: NewRequestStore(s.db, queueScope.SubScope("request_store"), queueName), + requestURIStore: NewRequestURIStore(s.db, queueScope.SubScope("request_uri_store"), queueName), + requestLogStore: NewRequestLogStore(s.db, queueScope.SubScope("request_log_store"), queueName), + requestSummaryStore: NewRequestSummaryStore(s.db, queueScope.SubScope("request_summary_store"), queueName), + requestAcceptanceStore: NewRequestAcceptanceStore(s.db, queueScope.SubScope("request_acceptance_store"), queueName), + queueSummaryStore: NewQueueSummaryStore(s.db, queueScope.SubScope("queue_summary_store"), queueName), + queueLogStore: NewQueueLogStore(s.db, queueScope.SubScope("queue_log_store"), queueName), + queueStore: NewQueueStore(s.db, queueScope.SubScope("queue_store"), queueName), + buildStore: NewBuildStore(s.db, queueScope.SubScope("build_store"), queueName), + validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), + projectValidationIndexStore: NewProjectValidationIndexStore(s.db, queueScope.SubScope("project_validation_index_store"), queueName), }, nil } @@ -66,16 +67,17 @@ func (s *Storage) Close() error { // mysqlStorage is the queue-scoped store aggregate returned by For. type mysqlStorage struct { - requestStore storage.RequestStore - requestURIStore storage.RequestURIStore - requestLogStore storage.RequestLogStore - requestSummaryStore storage.RequestSummaryStore - requestAcceptanceStore storage.RequestAcceptanceStore - queueStore storage.QueueStore - queueSummaryStore storage.QueueSummaryStore - queueLogStore storage.QueueLogStore - buildStore storage.BuildStore - validationFactStore storage.ValidationFactStore + requestStore storage.RequestStore + requestURIStore storage.RequestURIStore + requestLogStore storage.RequestLogStore + requestSummaryStore storage.RequestSummaryStore + requestAcceptanceStore storage.RequestAcceptanceStore + queueStore storage.QueueStore + queueSummaryStore storage.QueueSummaryStore + queueLogStore storage.QueueLogStore + buildStore storage.BuildStore + validationFactStore storage.ValidationFactStore + projectValidationIndexStore storage.ProjectValidationIndexStore } // Verify mysqlStorage implements the queue-scoped aggregate at compile time. @@ -121,6 +123,10 @@ func (f *mysqlStorage) GetValidationFactStore() storage.ValidationFactStore { return f.validationFactStore } +func (f *mysqlStorage) GetProjectValidationIndexStore() storage.ProjectValidationIndexStore { + return f.projectValidationIndexStore +} + func (f *mysqlStorage) GetQueueSummaryStore() storage.QueueSummaryStore { return f.queueSummaryStore } func (f *mysqlStorage) GetQueueLogStore() storage.QueueLogStore { diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index b80f5017c..12674f98c 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -23,6 +23,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" ) @@ -135,6 +136,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "project_validation_index_store", + op: "list", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetProjectValidationIndexStore().List(context.Background(), "change-uri", entity.ProjectValidationIndex{}, 1) + return err + }, + }, } for _, tt := range tests { diff --git a/stovepipe/extension/storage/project_validation_index_store.go b/stovepipe/extension/storage/project_validation_index_store.go new file mode 100644 index 000000000..f95c9d97a --- /dev/null +++ b/stovepipe/extension/storage/project_validation_index_store.go @@ -0,0 +1,33 @@ +// 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 storage + +//go:generate mockgen -source=project_validation_index_store.go -destination=mock/project_validation_index_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// ProjectValidationIndexStore holds immutable ordering mappings in its bound queue. +type ProjectValidationIndexStore interface { + // Create puts one mapping by (URI, degree order, project); duplicates return ErrAlreadyExists. + Create(ctx context.Context, entry entity.ProjectValidationIndex) error + + // List scans a primary-key range in ascending (degree order, bytewise project) order. + // after is exclusive; its zero value starts the range. limit must be positive. + List(ctx context.Context, uri string, after entity.ProjectValidationIndex, limit int) ([]entity.ProjectValidationIndex, error) +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index f0bb5d079..32333b639 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -96,4 +96,6 @@ type Storage interface { // GetValidationFactStore returns the ValidationFactStore instance. GetValidationFactStore() ValidationFactStore + + GetProjectValidationIndexStore() ProjectValidationIndexStore } diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index 408c73051..e235d3ee3 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_test") go_test( name = "go_default_test", srcs = [ + "project_validation_index_test.go", "queue_state_transitions_test.go", "queue_summary_test.go", "request_acceptance_test.go", @@ -28,6 +29,7 @@ go_test( "//test/integration/stovepipe/extension/storage: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", "@com_github_stretchr_testify//require:go_default_library", "@com_github_stretchr_testify//suite:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/test/integration/stovepipe/extension/storage/mysql/project_validation_index_test.go b/test/integration/stovepipe/extension/storage/mysql/project_validation_index_test.go new file mode 100644 index 000000000..76527da13 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/project_validation_index_test.go @@ -0,0 +1,87 @@ +// 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 mysql + +import ( + "context" + "math" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + mysqlstorage "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" +) + +func testProjectValidationIndex(t *testing.T, ctx context.Context, backend *mysqlstorage.Storage) { + const queue, uri = "index/main", "commit" + bound, err := backend.For(queue) + require.NoError(t, err) + index := bound.GetProjectValidationIndexStore() + facts := []entity.ValidationFact{ + {URI: uri, Project: "z", Degree: 1}, + {URI: uri, Project: "y", Degree: 1}, + {URI: uri, Project: "a", Degree: 0.5}, + {URI: uri, Project: "b", Degree: math.Nextafter(0.5, 1)}, + {URI: uri, Project: "green", Degree: 0}, + } + var entries []entity.ProjectValidationIndex + for _, fact := range facts { + fact.RequestID = "1" + require.NoError(t, bound.GetValidationFactStore().Create(ctx, fact)) + entry, err := entity.NewProjectValidationIndex(fact) + require.NoError(t, err) + require.NoError(t, index.Create(ctx, entry)) + require.ErrorIs(t, index.Create(ctx, entry), storage.ErrAlreadyExists) + entries = append(entries, entry) + } + require.NoError(t, bound.GetValidationFactStore().Create(ctx, entity.ValidationFact{URI: uri, Degree: 1, RequestID: "1"})) + + historical := entity.ValidationFact{URI: "older-commit", Project: "old-project", Degree: 1, RequestID: "2"} + require.NoError(t, bound.GetValidationFactStore().Create(ctx, historical)) + unindexed, err := index.List(ctx, historical.URI, entity.ProjectValidationIndex{}, 2) + require.NoError(t, err) + assert.Empty(t, unindexed) + preserved, err := bound.GetValidationFactStore().Get(ctx, historical.URI, historical.Project) + require.NoError(t, err) + assert.Equal(t, historical, preserved) + + other, err := backend.For("other") + require.NoError(t, err) + require.NoError(t, other.GetProjectValidationIndexStore().Create(ctx, entries[0])) + otherURI := entries[0] + otherURI.URI = "other" + require.NoError(t, index.Create(ctx, otherURI)) + want := []entity.ProjectValidationIndex{entries[1], entries[0], entries[3], entries[2], entries[4]} + for _, pageSize := range []int{1, 2, 10} { + var got []entity.ProjectValidationIndex + var after entity.ProjectValidationIndex + for { + page, err := index.List(ctx, uri, after, pageSize) + require.NoError(t, err) + if len(page) == 0 { + break + } + got = append(got, page...) + require.LessOrEqual(t, len(got), len(want)) + after = page[len(page)-1] + } + assert.Equal(t, want, got) + } + missing, err := index.List(ctx, "missing", entity.ProjectValidationIndex{}, 2) + require.NoError(t, err) + assert.Empty(t, missing) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index fc93afcaf..377aa6297 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -110,6 +110,11 @@ func TestMySQLStorage(t *testing.T) { suite.Run(t, testSuite) }) + t.Run("ProjectValidationIndex", func(t *testing.T) { + resetStorage(t, db) + testProjectValidationIndex(t, ctx, backend) + }) + t.Run("RequestLogStore", func(t *testing.T) { resetStorage(t, db) testSuite := new(MySQLRequestLogStoreSuite) @@ -153,6 +158,7 @@ func resetStorage(t *testing.T, db *sql.DB) { "TRUNCATE TABLE build", "TRUNCATE TABLE queue", "TRUNCATE TABLE validation_fact", + "TRUNCATE TABLE project_validation_index", } { _, err := db.ExecContext(context.Background(), statement) require.NoError(t, err) diff --git a/web/test/e2e/stovepipe/run_e2e.py b/web/test/e2e/stovepipe/run_e2e.py index 32bb69f1d..44be9b801 100644 --- a/web/test/e2e/stovepipe/run_e2e.py +++ b/web/test/e2e/stovepipe/run_e2e.py @@ -3,6 +3,7 @@ from pathlib import Path import shutil import signal +import struct import subprocess import sys import tempfile @@ -49,6 +50,13 @@ def _pagination_fixture_sql() -> str: statements.append( f"INSERT INTO validation_fact VALUES ('web/pagination','{uri}','project-{i:02d}',{degree},'51',{2000+i});" ) + degree_order = ( + struct.unpack(">Q", struct.pack(">d", 1.0))[0] + - struct.unpack(">Q", struct.pack(">d", degree))[0] + ) + statements.append( + f"INSERT INTO project_validation_index VALUES ('web/pagination','{uri}',{degree_order},'project-{i:02d}');" + ) return "\n".join(statements)