diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index f5269f909..16d1bbfd1 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "build_issue.go", "ingest.go", "list.go", + "open_issue.go", "project_status.go", "queue.go", "queue_config.go", diff --git a/stovepipe/entity/open_issue.go b/stovepipe/entity/open_issue.go new file mode 100644 index 000000000..5119d3380 --- /dev/null +++ b/stovepipe/entity/open_issue.go @@ -0,0 +1,27 @@ +// 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 + +// OpenIssue is a currently open validation issue within a queue. +// Its task and unit identify the condition; the first-seen build identifies +// this opening and is immutable until removal. +type OpenIssue struct { + // TaskID identifies the logical validation task. + TaskID string `json:"task_id"` + // UnitID identifies the validation unit; empty for a task-wide issue. + UnitID string `json:"unit_id"` + // FirstSeenBuildID identifies the build where this opening of the issue was first observed. + FirstSeenBuildID string `json:"first_seen_build_id"` +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index 2ebebccaa..a90b46c32 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_issue_store.go", "build_store.go", + "open_issue_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 e0777fd6e..5e9072477 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -1,6 +1,6 @@ # Storage -Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`). Implementations live under `extension/storage//`. +Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`, `OpenIssueStore`). Implementations live under `extension/storage//`. The aggregate is resolved per queue through a factory keyed by queue name, mirroring the extension contract and the [submitqueue storage seam](../../../submitqueue/extension/storage/README.md#queue-scoped-resolution): a resolved instance is bound to its queue, entity arguments whose queue disagrees with the binding are rejected, and reads never surface another queue's records. Stovepipe has no cross-queue read paths, so every store — including `QueueStore`, whose row key is the queue name itself — lives inside the queue-scoped aggregate; there is no global remainder. @@ -31,3 +31,13 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte Each observation is stored separately, not in a serialized collection. Observations are historical evidence; they do not track whether an issue is currently open or resolved. No update, deletion, or absence-based recovery is implied by this store. The MySQL primary key contains the queue, build ID, task ID, and unit ID directly as binary strings, preserving byte-exact identity. Their limits are 255, 1020, 255, and 1024 bytes respectively, for a combined maximum of 2554 bytes. Creates and reads reject identifiers exceeding these limits before querying the database; identifiers are never truncated. + +## Currently open issues + +`OpenIssueStore` contains only currently open issues, keyed by task and unit within the bound queue. Removing an open row never removes immutable per-build observations. There is no open/resolved flag; a resolved issue has no row here. + +An open row is immutable. Its first-seen build identifies the current opening, not its latest observation. Repeated observations remain in the per-build store. Reopening after removal uses a different first-seen build ID; no last-observed build, version counter, or update operation is stored here. + +Deletion matches the task, unit, and first-seen build within the bound queue. It is safe to repeat: an absent opening is a no-op, and a stale deletion never removes a different opening of the same condition. Database failures are returned for retry. + +MySQL identifiers use the same binary byte limits as per-build observations. Build IDs are opaque context, not ordering tokens. The store does not choose resolution, enumerate issues, write lifecycle history, or reconcile analysis responses. History and source-order reconciliation belong to later controllers, not this membership store. diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index ad9f4504a..e43236134 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_issue_store_mock.go", "build_store_mock.go", + "open_issue_store_mock.go", "queue_log_store_mock.go", "queue_store_mock.go", "queue_summary_store_mock.go", diff --git a/stovepipe/extension/storage/mock/open_issue_store_mock.go b/stovepipe/extension/storage/mock/open_issue_store_mock.go new file mode 100644 index 000000000..2b701b789 --- /dev/null +++ b/stovepipe/extension/storage/mock/open_issue_store_mock.go @@ -0,0 +1,85 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: open_issue_store.go +// +// Generated by this command: +// +// mockgen -source=open_issue_store.go -destination=mock/open_issue_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" +) + +// MockOpenIssueStore is a mock of OpenIssueStore interface. +type MockOpenIssueStore struct { + ctrl *gomock.Controller + recorder *MockOpenIssueStoreMockRecorder + isgomock struct{} +} + +// MockOpenIssueStoreMockRecorder is the mock recorder for MockOpenIssueStore. +type MockOpenIssueStoreMockRecorder struct { + mock *MockOpenIssueStore +} + +// NewMockOpenIssueStore creates a new mock instance. +func NewMockOpenIssueStore(ctrl *gomock.Controller) *MockOpenIssueStore { + mock := &MockOpenIssueStore{ctrl: ctrl} + mock.recorder = &MockOpenIssueStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockOpenIssueStore) EXPECT() *MockOpenIssueStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockOpenIssueStore) Create(ctx context.Context, issue entity.OpenIssue) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, issue) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockOpenIssueStoreMockRecorder) Create(ctx, issue any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockOpenIssueStore)(nil).Create), ctx, issue) +} + +// Delete mocks base method. +func (m *MockOpenIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Delete", ctx, issue) + ret0, _ := ret[0].(error) + return ret0 +} + +// Delete indicates an expected call of Delete. +func (mr *MockOpenIssueStoreMockRecorder) Delete(ctx, issue any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockOpenIssueStore)(nil).Delete), ctx, issue) +} + +// Get mocks base method. +func (m *MockOpenIssueStore) Get(ctx context.Context, taskID, unitID string) (entity.OpenIssue, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID) + ret0, _ := ret[0].(entity.OpenIssue) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockOpenIssueStoreMockRecorder) Get(ctx, taskID, unitID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockOpenIssueStore)(nil).Get), ctx, taskID, unitID) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index b39ce22d7..4158fe1ca 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -107,6 +107,20 @@ func (mr *MockStorageMockRecorder) GetBuildStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBuildStore", reflect.TypeOf((*MockStorage)(nil).GetBuildStore)) } +// GetOpenIssueStore mocks base method. +func (m *MockStorage) GetOpenIssueStore() storage.OpenIssueStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetOpenIssueStore") + ret0, _ := ret[0].(storage.OpenIssueStore) + return ret0 +} + +// GetOpenIssueStore indicates an expected call of GetOpenIssueStore. +func (mr *MockStorageMockRecorder) GetOpenIssueStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetOpenIssueStore", reflect.TypeOf((*MockStorage)(nil).GetOpenIssueStore)) +} + // 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 25a1344f7..c3adf368f 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_issue_store.go", "build_store.go", + "open_issue_store.go", "queue_log_store.go", "queue_store.go", "queue_summary_store.go", @@ -32,6 +33,7 @@ go_test( srcs = [ "build_issue_store_test.go", "build_store_test.go", + "open_issue_store_test.go", "queue_store_test.go", "queue_summary_store_test.go", "request_acceptance_store_test.go", diff --git a/stovepipe/extension/storage/mysql/open_issue_store.go b/stovepipe/extension/storage/mysql/open_issue_store.go new file mode 100644 index 000000000..2c65a79f3 --- /dev/null +++ b/stovepipe/extension/storage/mysql/open_issue_store.go @@ -0,0 +1,114 @@ +// 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" + "errors" + "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 openIssueStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewOpenIssueStore binds currently open issues to queue over the shared database. +func NewOpenIssueStore(db *sql.DB, scope tally.Scope, queue string) storage.OpenIssueStore { + return &openIssueStore{db: db, scope: scope, queue: queue} +} + +func (s *openIssueStore) Create(ctx context.Context, issue entity.OpenIssue) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := s.validateIssue(issue); err != nil { + return err + } + + _, err := s.db.ExecContext(ctx, + `INSERT INTO open_issue (queue, task_id, unit_id, first_seen_build_id) + VALUES (?, ?, ?, ?)`, + s.queue, issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) + if err != nil { + if isDuplicateEntry(err) { + return fmt.Errorf("open issue task=%q unit=%q: %w", issue.TaskID, issue.UnitID, storage.ErrAlreadyExists) + } + + return fmt.Errorf("failed to create open issue: %w", err) + } + + return nil +} + +func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (ret entity.OpenIssue, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := validateBuildIssueKey(s.queue, "", taskID, unitID); err != nil { + return entity.OpenIssue{}, err + } + + var issue entity.OpenIssue + err := s.db.QueryRowContext(ctx, + `SELECT task_id, unit_id, first_seen_build_id + FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ?`, + s.queue, taskID, unitID, + ).Scan(&issue.TaskID, &issue.UnitID, &issue.FirstSeenBuildID) + if errors.Is(err, sql.ErrNoRows) { + return entity.OpenIssue{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.OpenIssue{}, fmt.Errorf("failed to get open issue: %w", err) + } + + return issue, nil +} + +func (s *openIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (retErr error) { + op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := s.validateIssue(issue); err != nil { + return err + } + + _, err := s.db.ExecContext(ctx, + `DELETE FROM open_issue + WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?`, + s.queue, issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) + if err != nil { + return fmt.Errorf("failed to delete open issue: %w", err) + } + + return nil +} + +func (s *openIssueStore) validateIssue(issue entity.OpenIssue) error { + if err := validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil { + return err + } + if issue.FirstSeenBuildID == "" { + return fmt.Errorf("open issue first-seen build ID must not be empty") + } + return nil +} diff --git a/stovepipe/extension/storage/mysql/open_issue_store_test.go b/stovepipe/extension/storage/mysql/open_issue_store_test.go new file mode 100644 index 000000000..75fce58c4 --- /dev/null +++ b/stovepipe/extension/storage/mysql/open_issue_store_test.go @@ -0,0 +1,209 @@ +// 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" + "regexp" + "strings" + "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 openIssueFixture() entity.OpenIssue { + return entity.OpenIssue{ + TaskID: "task", UnitID: "unit", + FirstSeenBuildID: "build/1", + } +} + +func TestOpenIssueCreate(t *testing.T) { + for _, tt := range []struct { + name string + dbErr error + wantErr error + }{ + {name: "created"}, + {name: "duplicate", dbErr: &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, wantErr: storage.ErrAlreadyExists}, + {name: "database error", dbErr: assert.AnError, wantErr: assert.AnError}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + issue := openIssueFixture() + store := NewOpenIssueStore(db, testMetrics(), "queue") + + q := mock.ExpectExec(regexp.QuoteMeta( + `INSERT INTO open_issue (queue, task_id, unit_id, first_seen_build_id) + VALUES (?, ?, ?, ?)`)). + WithArgs("queue", issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) + if tt.dbErr != nil { + q.WillReturnError(tt.dbErr) + } else { + q.WillReturnResult(sqlmock.NewResult(0, 1)) + } + + err = store.Create(context.Background(), issue) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestOpenIssueGet(t *testing.T) { + for _, kind := range []string{"found", "missing", "database error", "scan error"} { + t.Run(kind, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + issue := openIssueFixture() + store := NewOpenIssueStore(db, testMetrics(), "queue") + + q := mock.ExpectQuery(regexp.QuoteMeta( + `SELECT task_id, unit_id, first_seen_build_id + FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ?`)). + WithArgs("queue", issue.TaskID, issue.UnitID) + switch kind { + case "missing": + q.WillReturnError(sql.ErrNoRows) + case "database error": + q.WillReturnError(assert.AnError) + default: + var firstBuild any = issue.FirstSeenBuildID + if kind == "scan error" { + firstBuild = nil + } + + q.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}). + AddRow(issue.TaskID, issue.UnitID, firstBuild)) + } + + got, err := store.Get(context.Background(), issue.TaskID, issue.UnitID) + switch kind { + case "found": + require.NoError(t, err) + assert.Equal(t, issue, got) + case "missing": + require.ErrorIs(t, err, storage.ErrNotFound) + case "database error": + require.ErrorIs(t, err, assert.AnError) + default: + require.Error(t, err) + } + + if kind != "found" { + assert.Equal(t, entity.OpenIssue{}, got) + } + + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestOpenIssueDelete(t *testing.T) { + for _, tt := range []struct { + name string + rows int64 + dbErr error + }{ + {name: "matching opening", rows: 1}, + {name: "already absent"}, + {name: "different opening"}, + {name: "database error", dbErr: assert.AnError}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + issue := openIssueFixture() + store := NewOpenIssueStore(db, testMetrics(), "queue") + + q := mock.ExpectExec(regexp.QuoteMeta( + `DELETE FROM open_issue + WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?`)). + WithArgs("queue", issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) + if tt.dbErr != nil { + q.WillReturnError(tt.dbErr) + } else { + q.WillReturnResult(sqlmock.NewResult(0, tt.rows)) + } + + err = store.Delete(context.Background(), issue) + if tt.dbErr != nil { + require.ErrorIs(t, err, tt.dbErr) + } else { + require.NoError(t, err) + } + + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) { + for _, kind := range []string{"queue", "task", "unit", "first build", "empty first build"} { + t.Run(kind, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + queue := "queue" + issue := openIssueFixture() + + switch kind { + case "queue": + queue = strings.Repeat("q", 256) + case "task": + issue.TaskID = strings.Repeat("t", 256) + case "unit": + issue.UnitID = strings.Repeat("😀", 257) + case "first build": + issue.FirstSeenBuildID = strings.Repeat("b", 1021) + case "empty first build": + issue.FirstSeenBuildID = "" + } + + mock.MatchExpectationsInOrder(false) + mock.ExpectExec("INSERT INTO open_issue").WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("DELETE FROM open_issue").WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectQuery("SELECT task_id, unit_id").WillReturnRows( + sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}). + AddRow(issue.TaskID, issue.UnitID, issue.FirstSeenBuildID), + ) + store := NewOpenIssueStore(db, testMetrics(), queue) + + require.Error(t, store.Create(context.Background(), issue)) + require.Error(t, store.Delete(context.Background(), issue)) + + switch kind { + case "queue", "task", "unit": + _, err = store.Get(context.Background(), issue.TaskID, issue.UnitID) + require.Error(t, err) + } + }) + } +} diff --git a/stovepipe/extension/storage/mysql/schema/open_issue.sql b/stovepipe/extension/storage/mysql/schema/open_issue.sql new file mode 100644 index 000000000..d42436258 --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/open_issue.sql @@ -0,0 +1,10 @@ +-- Only currently open issues; historical observations remain in build_issue. +CREATE TABLE IF NOT EXISTS open_issue ( + queue VARBINARY(255) NOT NULL, + task_id VARBINARY(255) NOT NULL, + unit_id VARBINARY(1024) NOT NULL, + + first_seen_build_id VARBINARY(1020) NOT NULL, -- Accommodates existing VARCHAR(255) utf8mb4 build IDs. + + PRIMARY KEY (queue, task_id, unit_id) +) ENGINE=InnoDB; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index 5acb6e87d..8b17e5203 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -56,6 +56,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { 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), }, nil } @@ -77,6 +78,7 @@ type mysqlStorage struct { queueLogStore storage.QueueLogStore buildStore storage.BuildStore buildIssueStore storage.BuildIssueStore + openIssueStore storage.OpenIssueStore validationFactStore storage.ValidationFactStore } @@ -132,3 +134,7 @@ func (f *mysqlStorage) GetQueueLogStore() storage.QueueLogStore { func (f *mysqlStorage) GetBuildIssueStore() storage.BuildIssueStore { return f.buildIssueStore } + +func (f *mysqlStorage) GetOpenIssueStore() storage.OpenIssueStore { + return f.openIssueStore +} diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index 326070bd1..81af1f193 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -48,6 +48,7 @@ func TestNewStorage(t *testing.T) { assert.NotNil(t, bound.GetQueueStore()) assert.NotNil(t, bound.GetBuildStore()) assert.NotNil(t, bound.GetBuildIssueStore()) + assert.NotNil(t, bound.GetOpenIssueStore()) _, err = s.For("") assert.Error(t, err, "resolving an empty queue name must fail") @@ -136,6 +137,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "open_issue_store", + op: "get", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetOpenIssueStore().Get(context.Background(), "task-id", "unit-id") + return err + }, + }, { name: "validation_fact_store", op: "get", diff --git a/stovepipe/extension/storage/open_issue_store.go b/stovepipe/extension/storage/open_issue_store.go new file mode 100644 index 000000000..63a58cc22 --- /dev/null +++ b/stovepipe/extension/storage/open_issue_store.go @@ -0,0 +1,38 @@ +// 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=open_issue_store.go -destination=mock/open_issue_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// OpenIssueStore contains only currently open issues, keyed by task and unit +// within the bound queue. A reopened issue must use a new first-seen build ID. +type OpenIssueStore interface { + // Create requires a nonempty first-seen build ID. + // An existing task/unit returns ErrAlreadyExists. + Create(ctx context.Context, issue entity.OpenIssue) error + + // Get returns ErrNotFound when the task/unit is not currently open. + Get(ctx context.Context, taskID, unitID string) (entity.OpenIssue, error) + + // Delete removes only the matching opening. It succeeds if that opening is + // already absent and never removes a row with a different first-seen build. + Delete(ctx context.Context, issue entity.OpenIssue) error +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 21b9886b6..18100c280 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -97,6 +97,9 @@ type Storage interface { // GetBuildIssueStore returns the BuildIssueStore instance. GetBuildIssueStore() BuildIssueStore + // GetOpenIssueStore returns the OpenIssueStore instance. + GetOpenIssueStore() OpenIssueStore + // GetValidationFactStore returns the ValidationFactStore instance. GetValidationFactStore() ValidationFactStore } diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index 758f35ad9..97a6fe1a2 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -4,6 +4,7 @@ go_test( name = "go_default_test", srcs = [ "build_issue_test.go", + "open_issue_test.go", "queue_state_transitions_test.go", "queue_summary_test.go", "request_acceptance_test.go", diff --git a/test/integration/stovepipe/extension/storage/mysql/open_issue_test.go b/test/integration/stovepipe/extension/storage/mysql/open_issue_test.go new file mode 100644 index 000000000..e227e0311 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/open_issue_test.go @@ -0,0 +1,156 @@ +// 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" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func testOpenIssueStore(t *testing.T, ctx context.Context, factory storage.Factory) { + t.Helper() + + bound, err := factory.For(storage.Config{QueueName: "open-issues/main"}) + require.NoError(t, err) + store := bound.GetOpenIssueStore() + issue := entity.OpenIssue{ + TaskID: "validation", UnitID: "unit", FirstSeenBuildID: "build/1", + } + + require.NoError(t, store.Create(ctx, issue)) + require.ErrorIs(t, store.Create(ctx, issue), storage.ErrAlreadyExists) + got, err := store.Get(ctx, issue.TaskID, issue.UnitID) + require.NoError(t, err) + require.Equal(t, issue, got) + + require.NoError(t, bound.GetBuildIssueStore().Create(ctx, "build/1", entity.BuildIssue{TaskID: issue.TaskID, UnitID: issue.UnitID})) + require.NoError(t, store.Delete(ctx, got)) + require.NoError(t, store.Delete(ctx, got)) + _, err = store.Get(ctx, issue.TaskID, issue.UnitID) + require.ErrorIs(t, err, storage.ErrNotFound) + _, err = bound.GetBuildIssueStore().Get(ctx, "build/1", issue.TaskID, issue.UnitID) + require.NoError(t, err) + + t.Run("stale deletion cannot remove a reopened issue", func(t *testing.T) { + reopened := issue + reopened.FirstSeenBuildID = "build/3" + require.NoError(t, store.Create(ctx, reopened)) + + require.NoError(t, store.Delete(ctx, issue)) + require.NoError(t, store.Delete(ctx, issue)) + require.ErrorIs(t, store.Create(ctx, issue), storage.ErrAlreadyExists) + got, err := store.Get(ctx, issue.TaskID, issue.UnitID) + require.NoError(t, err) + require.Equal(t, reopened, got) + }) + + t.Run("queue and byte-exact key isolation", func(t *testing.T) { + other, err := factory.For(storage.Config{QueueName: "Open-issues/main"}) + require.NoError(t, err) + _, err = other.GetOpenIssueStore().Get(ctx, issue.TaskID, issue.UnitID) + require.ErrorIs(t, err, storage.ErrNotFound) + require.NoError(t, other.GetOpenIssueStore().Delete(ctx, issue)) + require.NoError(t, other.GetOpenIssueStore().Create(ctx, issue)) + require.NoError(t, other.GetOpenIssueStore().Delete(ctx, issue)) + + got, err := store.Get(ctx, issue.TaskID, issue.UnitID) + require.NoError(t, err) + require.Equal(t, "build/3", got.FirstSeenBuildID) + + for _, key := range []entity.BuildIssue{ + {TaskID: "validation", UnitID: "Unit"}, + {TaskID: "validation", UnitID: "unit "}, + {TaskID: "Validation", UnitID: "unit"}, + {TaskID: "validation ", UnitID: "unit"}, + {TaskID: "validation"}, + {TaskID: "a\x00b", UnitID: "c"}, + {TaskID: "a", UnitID: "b\x00c"}, + } { + value := issue + value.TaskID, value.UnitID = key.TaskID, key.UnitID + require.NoError(t, store.Create(ctx, value)) + + got, err := store.Get(ctx, key.TaskID, key.UnitID) + require.NoError(t, err) + require.Equal(t, value, got) + require.NoError(t, store.Delete(ctx, got)) + } + }) + + t.Run("only one concurrent opening wins", func(t *testing.T) { + results := make(chan error, 2) + for _, buildID := range []string{"build/4", "build/5"} { + value := issue + value.UnitID = "concurrent" + value.FirstSeenBuildID = buildID + go func() { results <- store.Create(ctx, value) }() + } + + successes, duplicates := 0, 0 + for i := 0; i < 2; i++ { + err := <-results + switch { + case err == nil: + successes++ + case errors.Is(err, storage.ErrAlreadyExists): + duplicates++ + default: + require.NoError(t, err) + } + } + require.Equal(t, 1, successes) + require.Equal(t, 1, duplicates) + + got, err := store.Get(ctx, issue.TaskID, "concurrent") + require.NoError(t, err) + require.Contains(t, []string{"build/4", "build/5"}, got.FirstSeenBuildID) + }) + + t.Run("binary identifier limits", func(t *testing.T) { + maxBound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) + require.NoError(t, err) + maxStore := maxBound.GetOpenIssueStore() + value := entity.OpenIssue{ + TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("😀", 256), + FirstSeenBuildID: strings.Repeat("😀", 255), + } + require.NoError(t, maxStore.Create(ctx, value)) + + got, err := maxStore.Get(ctx, value.TaskID, value.UnitID) + require.NoError(t, err) + require.Equal(t, value, got) + + oversized := value + oversized.FirstSeenBuildID += "x" + require.Error(t, maxStore.Create(ctx, oversized)) + require.Error(t, maxStore.Delete(ctx, oversized)) + oversized = value + oversized.UnitID += "x" + require.Error(t, maxStore.Create(ctx, oversized)) + require.Error(t, maxStore.Delete(ctx, oversized)) + _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID) + require.Error(t, err) + + got, err = maxStore.Get(ctx, value.TaskID, value.UnitID) + require.NoError(t, err) + require.Equal(t, value, got) + }) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index cd734ffb5..7349c4c17 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -95,6 +95,11 @@ func TestMySQLStorage(t *testing.T) { testBuildIssueStore(t, ctx, db, factory) }) + t.Run("OpenIssueStore", func(t *testing.T) { + resetStorage(t, db) + testOpenIssueStore(t, ctx, factory) + }) + t.Run("BuildStore", func(t *testing.T) { resetStorage(t, db) testSuite := new(MySQLBuildStoreSuite) @@ -148,6 +153,7 @@ func resetStorage(t *testing.T, db *sql.DB) { t.Helper() for _, statement := range []string{ + "TRUNCATE TABLE open_issue", "TRUNCATE TABLE queue_log", "TRUNCATE TABLE queue_summary", "TRUNCATE TABLE request_log",