diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 2b6ef37df..9f22e4c5f 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.NewProjectFactByDegree(fact) + if err != nil { + return err + } + + if err := store.GetProjectFactByDegreeStore().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..09060eda2 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.MockProjectFactByDegreeStore 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.NewMockProjectFactByDegreeStore(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().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) @@ -402,6 +405,13 @@ func TestProcess_RecordsNamedProjectStatusResults(t *testing.T) { }). Times(3) + var indexes []entity.ProjectFactByDegree + m.indexStore.EXPECT().Create(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, entry entity.ProjectFactByDegree) 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.NewProjectFactByDegree(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.NewProjectFactByDegree(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 16d1bbfd1..33f70d6c6 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -8,6 +8,7 @@ go_library( "ingest.go", "list.go", "open_issue.go", + "project_fact_by_degree.go", "project_status.go", "queue.go", "queue_config.go", @@ -32,6 +33,7 @@ go_test( name = "go_default_test", srcs = [ "build_test.go", + "project_fact_by_degree_test.go", "queue_state_test.go", "request_id_test.go", "request_log_test.go", diff --git a/stovepipe/entity/project_fact_by_degree.go b/stovepipe/entity/project_fact_by_degree.go new file mode 100644 index 000000000..867c73b93 --- /dev/null +++ b/stovepipe/entity/project_fact_by_degree.go @@ -0,0 +1,39 @@ +// 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" +) + +// ProjectFactByDegree maps an ordered key to an immutable project fact within a queue. +type ProjectFactByDegree struct { + // URI is the commit containing the project result. + URI string + // Degree is the breakage degree in [0, 1], ordered highest first. + Degree float64 + // Project identifies the fact and breaks equal-degree ties in ascending bytewise order. + Project string +} + +// NewProjectFactByDegree derives a degree lookup entry from a named project's fact. +// Invalid degrees, empty URIs and whole-repository facts return an error. +func NewProjectFactByDegree(fact ValidationFact) (ProjectFactByDegree, error) { + if fact.URI == "" || fact.Project == "" || math.IsNaN(fact.Degree) || fact.Degree < DegreeGreen || fact.Degree > DegreeBroken { + return ProjectFactByDegree{}, fmt.Errorf("project fact degree lookup requires a URI, project and degree in [0, 1]") + } + return ProjectFactByDegree{URI: fact.URI, Degree: fact.Degree, Project: fact.Project}, nil +} diff --git a/stovepipe/entity/project_fact_by_degree_test.go b/stovepipe/entity/project_fact_by_degree_test.go new file mode 100644 index 000000000..69454555f --- /dev/null +++ b/stovepipe/entity/project_fact_by_degree_test.go @@ -0,0 +1,44 @@ +// 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 TestProjectFactByDegreePreservesDegree(t *testing.T) { + for _, degree := range []float64{1, math.Nextafter(1, 0), 0.75, math.Nextafter(0.5, 1), 0.5, math.SmallestNonzeroFloat64, 0, math.Copysign(0, -1)} { + entry, err := NewProjectFactByDegree(ValidationFact{URI: "commit", Project: "project", Degree: degree}) + require.NoError(t, err) + assert.Equal(t, ProjectFactByDegree{URI: "commit", Project: "project", Degree: degree}, entry) + } +} + +func TestProjectFactByDegreeRejectsInvalidFacts(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 := NewProjectFactByDegree(fact) + require.Error(t, err) + } +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index a90b46c32..4b9f50fde 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build_issue_store.go", "build_store.go", "open_issue_store.go", + "project_fact_by_degree_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 5e9072477..8339f0108 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -24,6 +24,8 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte `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). +`ProjectFactByDegreeStore` is the ordered lookup mapping for named project facts. Its immutable `(queue, URI, degree DESC, project ASC)` primary key stores the original degree and orders by 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 before recording project completion, including retries after partial writes. Create the table before deploying writers. The table starts empty: only results processed by index-aware writers receive mappings, with no historical population or read-time fallback. + ## Build issue observations `BuildIssueStore` accumulates immutable observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Empty unit IDs represent task-wide observations. diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index e43236134..febc3d631 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build_issue_store_mock.go", "build_store_mock.go", "open_issue_store_mock.go", + "project_fact_by_degree_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_fact_by_degree_store_mock.go b/stovepipe/extension/storage/mock/project_fact_by_degree_store_mock.go new file mode 100644 index 000000000..d204eea7b --- /dev/null +++ b/stovepipe/extension/storage/mock/project_fact_by_degree_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: project_fact_by_degree_store.go +// +// Generated by this command: +// +// mockgen -source=project_fact_by_degree_store.go -destination=mock/project_fact_by_degree_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" +) + +// MockProjectFactByDegreeStore is a mock of ProjectFactByDegreeStore interface. +type MockProjectFactByDegreeStore struct { + ctrl *gomock.Controller + recorder *MockProjectFactByDegreeStoreMockRecorder + isgomock struct{} +} + +// MockProjectFactByDegreeStoreMockRecorder is the mock recorder for MockProjectFactByDegreeStore. +type MockProjectFactByDegreeStoreMockRecorder struct { + mock *MockProjectFactByDegreeStore +} + +// NewMockProjectFactByDegreeStore creates a new mock instance. +func NewMockProjectFactByDegreeStore(ctrl *gomock.Controller) *MockProjectFactByDegreeStore { + mock := &MockProjectFactByDegreeStore{ctrl: ctrl} + mock.recorder = &MockProjectFactByDegreeStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockProjectFactByDegreeStore) EXPECT() *MockProjectFactByDegreeStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockProjectFactByDegreeStore) Create(ctx context.Context, entry entity.ProjectFactByDegree) 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 *MockProjectFactByDegreeStoreMockRecorder) Create(ctx, entry any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockProjectFactByDegreeStore)(nil).Create), ctx, entry) +} + +// List mocks base method. +func (m *MockProjectFactByDegreeStore) List(ctx context.Context, uri string, after entity.ProjectFactByDegree, limit int) ([]entity.ProjectFactByDegree, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, uri, after, limit) + ret0, _ := ret[0].([]entity.ProjectFactByDegree) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockProjectFactByDegreeStoreMockRecorder) List(ctx, uri, after, limit any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockProjectFactByDegreeStore)(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 4158fe1ca..c11b501c9 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -121,6 +121,20 @@ func (mr *MockStorageMockRecorder) GetOpenIssueStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetOpenIssueStore", reflect.TypeOf((*MockStorage)(nil).GetOpenIssueStore)) } +// GetProjectFactByDegreeStore mocks base method. +func (m *MockStorage) GetProjectFactByDegreeStore() storage.ProjectFactByDegreeStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetProjectFactByDegreeStore") + ret0, _ := ret[0].(storage.ProjectFactByDegreeStore) + return ret0 +} + +// GetProjectFactByDegreeStore indicates an expected call of GetProjectFactByDegreeStore. +func (mr *MockStorageMockRecorder) GetProjectFactByDegreeStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetProjectFactByDegreeStore", reflect.TypeOf((*MockStorage)(nil).GetProjectFactByDegreeStore)) +} + // 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 c3adf368f..9bbd539ee 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build_issue_store.go", "build_store.go", "open_issue_store.go", + "project_fact_by_degree_store.go", "queue_log_store.go", "queue_store.go", "queue_summary_store.go", @@ -34,6 +35,7 @@ go_test( "build_issue_store_test.go", "build_store_test.go", "open_issue_store_test.go", + "project_fact_by_degree_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_fact_by_degree_store.go b/stovepipe/extension/storage/mysql/project_fact_by_degree_store.go new file mode 100644 index 000000000..d3afee730 --- /dev/null +++ b/stovepipe/extension/storage/mysql/project_fact_by_degree_store.go @@ -0,0 +1,109 @@ +// 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" + "math" + + "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 projectFactByDegreeStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewProjectFactByDegreeStore creates a queue-scoped, primary-key-ordered mapping store. +func NewProjectFactByDegreeStore(db *sql.DB, scope tally.Scope, queue string) storage.ProjectFactByDegreeStore { + return &projectFactByDegreeStore{db: db, scope: scope, queue: queue} +} + +func (s *projectFactByDegreeStore) Create(ctx context.Context, entry entity.ProjectFactByDegree) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if entry.URI == "" || entry.Project == "" || math.IsNaN(entry.Degree) || entry.Degree < entity.DegreeGreen || entry.Degree > entity.DegreeBroken { + return fmt.Errorf("project fact degree lookup requires a URI, project and degree in [0, 1]") + } + + _, err := s.db.ExecContext(ctx, ` + INSERT INTO project_fact_by_degree (queue, uri, degree, project) + VALUES (?, ?, ?, ?)`, + s.queue, entry.URI, entry.Degree, entry.Project, + ) + if err != nil { + if isDuplicateEntry(err) { + return fmt.Errorf("project fact degree lookup uri=%q project=%q: %w", entry.URI, entry.Project, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to create project fact degree lookup uri=%q project=%q: %w", entry.URI, entry.Project, err) + } + + return nil +} + +func (s *projectFactByDegreeStore) List(ctx context.Context, uri string, after entity.ProjectFactByDegree, limit int) (ret []entity.ProjectFactByDegree, retErr error) { + op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if uri == "" || limit <= 0 { + return nil, fmt.Errorf("project fact degree lookup requires a URI and positive limit") + } + hasCursor := after != entity.ProjectFactByDegree{} + if hasCursor { + if after.URI != uri { + return nil, fmt.Errorf("project fact degree lookup cursor must match the requested URI") + } + if after.Project == "" { + return nil, fmt.Errorf("project fact degree lookup cursor requires a project") + } + if math.IsNaN(after.Degree) || after.Degree < entity.DegreeGreen || after.Degree > entity.DegreeBroken { + return nil, fmt.Errorf("project fact degree lookup cursor requires a degree in [0, 1]") + } + } + + query := `SELECT uri, degree, project FROM project_fact_by_degree WHERE queue = ? AND uri = ?` + args := []any{s.queue, uri} + if hasCursor { + query += ` AND (degree < ? OR (degree = ? AND project > ?))` + args = append(args, after.Degree, after.Degree, after.Project) + } + query += ` ORDER BY degree DESC, project ASC LIMIT ?` + args = append(args, limit) + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to list project fact degree lookup queue=%q uri=%q: %w", s.queue, uri, err) + } + defer rows.Close() + + for rows.Next() { + var entry entity.ProjectFactByDegree + if err := rows.Scan(&entry.URI, &entry.Degree, &entry.Project); err != nil { + return nil, fmt.Errorf("failed to scan project fact degree lookup: %w", err) + } + ret = append(ret, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate project fact degree lookup: %w", err) + } + + return ret, nil +} diff --git a/stovepipe/extension/storage/mysql/project_fact_by_degree_store_test.go b/stovepipe/extension/storage/mysql/project_fact_by_degree_store_test.go new file mode 100644 index 000000000..0fc7e58d6 --- /dev/null +++ b/stovepipe/extension/storage/mysql/project_fact_by_degree_store_test.go @@ -0,0 +1,162 @@ +// 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" + "math" + "regexp" + "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 TestProjectFactByDegreeStoreCreate(t *testing.T) { + entry := entity.ProjectFactByDegree{URI: "commit", Degree: 0.5, 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_fact_by_degree").WithArgs("queue", entry.URI, entry.Degree, entry.Project) + if tt.dbErr != nil { + query.WillReturnError(tt.dbErr) + } else { + query.WillReturnResult(sqlmock.NewResult(0, 1)) + } + store := NewProjectFactByDegreeStore(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 TestProjectFactByDegreeStoreList(t *testing.T) { + columns := []string{"uri", "degree", "project"} + want := entity.ProjectFactByDegree{URI: "commit", Degree: 0.5, Project: "p"} + readErr := errors.New("connection reset") + for _, after := range []entity.ProjectFactByDegree{ + {}, + {URI: "commit", Degree: 1, Project: "z"}, + {URI: "commit", Degree: 0.75, Project: "z"}, + {URI: "commit", Degree: 0, Project: "z"}, + } { + tests := []struct { + name string + rows *sqlmock.Rows + queryErr error + wantErr bool + want []entity.ProjectFactByDegree + }{ + {name: "found", rows: sqlmock.NewRows(columns).AddRow(want.URI, want.Degree, want.Project), want: []entity.ProjectFactByDegree{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.Degree, 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() + querySQL := "SELECT uri, degree, project FROM project_fact_by_degree WHERE queue = ? AND uri = ?" + if after != (entity.ProjectFactByDegree{}) { + querySQL += " AND (degree < ? OR (degree = ? AND project > ?))" + } + querySQL += " ORDER BY degree DESC, project ASC LIMIT ?" + query := mock.ExpectQuery(regexp.QuoteMeta(querySQL)) + if after == (entity.ProjectFactByDegree{}) { + query.WithArgs("queue", "commit", 3) + } else { + query.WithArgs("queue", "commit", after.Degree, after.Degree, after.Project, 3) + } + if tt.queryErr != nil { + query.WillReturnError(tt.queryErr) + } else { + query.WillReturnRows(tt.rows) + } + store := NewProjectFactByDegreeStore(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 TestProjectFactByDegreeStoreRejectsInvalidInput(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewProjectFactByDegreeStore(db, testMetrics(), "queue") + for _, entry := range []entity.ProjectFactByDegree{ + {Project: "p"}, + {URI: "commit"}, + {URI: "commit", Project: "p", Degree: -1}, + {URI: "commit", Project: "p", Degree: 1.1}, + {URI: "commit", Project: "p", Degree: math.NaN()}, + {URI: "commit", Project: "p", Degree: math.Inf(1)}, + } { + require.Error(t, store.Create(context.Background(), entry)) + } + for _, after := range []entity.ProjectFactByDegree{ + {Degree: -1}, + {Degree: 1}, + {URI: "commit"}, + {URI: "other", Project: "p"}, + {URI: "commit", Project: "p", Degree: -0.1}, + {URI: "commit", Project: "p", Degree: 1.1}, + {URI: "commit", Project: "p", Degree: math.NaN()}, + {URI: "commit", Project: "p", Degree: math.Inf(1)}, + } { + _, 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.ProjectFactByDegree{}, limit) + require.Error(t, err) + } + _, err = store.List(context.Background(), "", entity.ProjectFactByDegree{}, 3) + require.Error(t, err) + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/stovepipe/extension/storage/mysql/schema/project_fact_by_degree.sql b/stovepipe/extension/storage/mysql/schema/project_fact_by_degree.sql new file mode 100644 index 000000000..986001f9e --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/project_fact_by_degree.sql @@ -0,0 +1,9 @@ +-- project_fact_by_degree is a rebuildable ordering mapping for named project facts. +-- The mixed-direction primary key supports bounded, most-broken-first reads. +CREATE TABLE IF NOT EXISTS project_fact_by_degree ( + queue VARBINARY(255) NOT NULL, + uri VARBINARY(255) NOT NULL, + degree DOUBLE NOT NULL, + project VARBINARY(255) NOT NULL, + PRIMARY KEY (queue, uri, degree DESC, project ASC) +) 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 8b17e5203..8ad73700d 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -46,18 +46,19 @@ 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), - buildIssueStore: NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName), - openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_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), + buildIssueStore: NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName), + openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_store"), queueName), + validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), + projectFactByDegreeStore: NewProjectFactByDegreeStore(s.db, queueScope.SubScope("project_fact_by_degree_store"), queueName), }, nil } @@ -68,18 +69,19 @@ 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 - buildIssueStore storage.BuildIssueStore - openIssueStore storage.OpenIssueStore - 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 + buildIssueStore storage.BuildIssueStore + openIssueStore storage.OpenIssueStore + validationFactStore storage.ValidationFactStore + projectFactByDegreeStore storage.ProjectFactByDegreeStore } // Verify mysqlStorage implements the queue-scoped aggregate at compile time. @@ -125,6 +127,10 @@ func (f *mysqlStorage) GetValidationFactStore() storage.ValidationFactStore { return f.validationFactStore } +func (f *mysqlStorage) GetProjectFactByDegreeStore() storage.ProjectFactByDegreeStore { + return f.projectFactByDegreeStore +} + 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 81af1f193..80b63a7f8 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" ) @@ -153,6 +154,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "project_fact_by_degree_store", + op: "list", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetProjectFactByDegreeStore().List(context.Background(), "change-uri", entity.ProjectFactByDegree{}, 1) + return err + }, + }, } for _, tt := range tests { diff --git a/stovepipe/extension/storage/project_fact_by_degree_store.go b/stovepipe/extension/storage/project_fact_by_degree_store.go new file mode 100644 index 000000000..01f4ce079 --- /dev/null +++ b/stovepipe/extension/storage/project_fact_by_degree_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_fact_by_degree_store.go -destination=mock/project_fact_by_degree_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// ProjectFactByDegreeStore holds immutable ordering mappings in its bound queue. +type ProjectFactByDegreeStore interface { + // Create puts one mapping by (URI, degree, project); duplicates return ErrAlreadyExists. + Create(ctx context.Context, entry entity.ProjectFactByDegree) error + + // List scans a primary-key range by descending degree, then ascending bytewise project. + // after is exclusive; its zero value starts the range. limit must be positive. + List(ctx context.Context, uri string, after entity.ProjectFactByDegree, limit int) ([]entity.ProjectFactByDegree, error) +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 18100c280..2ca60ee17 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -102,4 +102,6 @@ type Storage interface { // GetValidationFactStore returns the ValidationFactStore instance. GetValidationFactStore() ValidationFactStore + + GetProjectFactByDegreeStore() ProjectFactByDegreeStore } diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index 97a6fe1a2..c09a744c3 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -5,6 +5,7 @@ go_test( srcs = [ "build_issue_test.go", "open_issue_test.go", + "project_fact_by_degree_test.go", "queue_state_transitions_test.go", "queue_summary_test.go", "request_acceptance_test.go", @@ -30,6 +31,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_fact_by_degree_test.go b/test/integration/stovepipe/extension/storage/mysql/project_fact_by_degree_test.go new file mode 100644 index 000000000..bf14b1130 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/project_fact_by_degree_test.go @@ -0,0 +1,88 @@ +// 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 testProjectFactByDegree(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.GetProjectFactByDegreeStore() + 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}, + {URI: uri, Project: "another-green", Degree: 0}, + } + var entries []entity.ProjectFactByDegree + for _, fact := range facts { + fact.RequestID = "1" + require.NoError(t, bound.GetValidationFactStore().Create(ctx, fact)) + entry, err := entity.NewProjectFactByDegree(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.ProjectFactByDegree{}, 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.GetProjectFactByDegreeStore().Create(ctx, entries[0])) + otherURI := entries[0] + otherURI.URI = "other" + require.NoError(t, index.Create(ctx, otherURI)) + want := []entity.ProjectFactByDegree{entries[1], entries[0], entries[3], entries[2], entries[5], entries[4]} + for _, pageSize := range []int{1, 2, 10} { + var got []entity.ProjectFactByDegree + var after entity.ProjectFactByDegree + 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.ProjectFactByDegree{}, 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 7349c4c17..9b15cae39 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -120,6 +120,11 @@ func TestMySQLStorage(t *testing.T) { suite.Run(t, testSuite) }) + t.Run("ProjectFactByDegree", func(t *testing.T) { + resetStorage(t, db) + testProjectFactByDegree(t, ctx, backend) + }) + t.Run("RequestLogStore", func(t *testing.T) { resetStorage(t, db) testSuite := new(MySQLRequestLogStoreSuite) @@ -165,6 +170,7 @@ func resetStorage(t *testing.T, db *sql.DB) { "TRUNCATE TABLE build_issue", "TRUNCATE TABLE queue", "TRUNCATE TABLE validation_fact", + "TRUNCATE TABLE project_fact_by_degree", } { _, 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..9de49dbdf 100644 --- a/web/test/e2e/stovepipe/run_e2e.py +++ b/web/test/e2e/stovepipe/run_e2e.py @@ -49,6 +49,9 @@ def _pagination_fixture_sql() -> str: statements.append( f"INSERT INTO validation_fact VALUES ('web/pagination','{uri}','project-{i:02d}',{degree},'51',{2000+i});" ) + statements.append( + f"INSERT INTO project_fact_by_degree VALUES ('web/pagination','{uri}',{degree},'project-{i:02d}');" + ) return "\n".join(statements)