diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 80405d335..a8c5cd632 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -8,6 +8,7 @@ go_library( "ingest.go", "issue_log.go", "issue_state.go", + "issue_summary.go", "issue_tracking_state.go", "list.go", "open_issue.go", @@ -38,6 +39,7 @@ go_test( "build_test.go", "issue_log_test.go", "issue_state_test.go", + "issue_summary_test.go", "issue_tracking_state_test.go", "project_fact_by_degree_test.go", "queue_state_test.go", diff --git a/stovepipe/entity/issue_summary.go b/stovepipe/entity/issue_summary.go new file mode 100644 index 000000000..89e59aecb --- /dev/null +++ b/stovepipe/entity/issue_summary.go @@ -0,0 +1,62 @@ +// 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" + + "github.com/uber/submitqueue/platform/base/id" +) + +// IssueSummary is the latest read model of an issue within a queue. +// Closed summaries retain the source of their latest transition. +type IssueSummary struct { + // TaskID identifies the logical validation task and must be nonempty. + TaskID string + // UnitID identifies the validation unit; empty means a task-wide issue. + UnitID string + // Status identifies whether the issue is open or closed. + Status IssueStatus + // FirstSeenBuildID identifies the current or most recent opening. + FirstSeenBuildID string + // RequestID is the canonical decimal source request ID of the latest transition. + RequestID string + // BuildID identifies the source build of the latest transition. + BuildID string + // ChangedAtMs is when the latest transition was first recorded, in Unix milliseconds. + ChangedAtMs int64 + // Version is the positive optimistic-lock version of this summary. + Version int32 +} + +// Validate checks the identity, status, timestamp, and version of a summary. +func (s IssueSummary) Validate() error { + if s.TaskID == "" || s.FirstSeenBuildID == "" || s.BuildID == "" { + return fmt.Errorf("issue summary requires task, first-seen build, and source build IDs") + } + if err := id.Validate(s.RequestID); err != nil { + return fmt.Errorf("invalid issue summary request ID: %w", err) + } + if s.Status != IssueStatusOpen && s.Status != IssueStatusClosed { + return fmt.Errorf("invalid issue summary status %q", s.Status) + } + if s.ChangedAtMs <= 0 { + return fmt.Errorf("issue summary requires a positive change timestamp") + } + if s.Version <= 0 { + return fmt.Errorf("issue summary requires a positive version") + } + return nil +} diff --git a/stovepipe/entity/issue_summary_test.go b/stovepipe/entity/issue_summary_test.go new file mode 100644 index 000000000..72a7795a3 --- /dev/null +++ b/stovepipe/entity/issue_summary_test.go @@ -0,0 +1,60 @@ +// 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 ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestIssueSummaryValidate(t *testing.T) { + valid := IssueSummary{TaskID: "validation", UnitID: "unit", Status: IssueStatusOpen, + FirstSeenBuildID: "build/1", RequestID: "1", BuildID: "build/1", ChangedAtMs: 1000, Version: 1} + tests := []struct { + name string + change func(*IssueSummary) + wantErr bool + }{ + {name: "open"}, + {name: "closed", change: func(s *IssueSummary) { s.Status = IssueStatusClosed }}, + {name: "task wide", change: func(s *IssueSummary) { s.UnitID = "" }}, + {"missing task", func(s *IssueSummary) { s.TaskID = "" }, true}, + {"missing opening", func(s *IssueSummary) { s.FirstSeenBuildID = "" }, true}, + {"missing source build", func(s *IssueSummary) { s.BuildID = "" }, true}, + {"missing source request", func(s *IssueSummary) { s.RequestID = "" }, true}, + {"noncanonical source request", func(s *IssueSummary) { s.RequestID = "01" }, true}, + {"invalid source request", func(s *IssueSummary) { s.RequestID = "build/1" }, true}, + {"unknown status", func(s *IssueSummary) { s.Status = IssueStatusUnknown }, true}, + {"invalid status", func(s *IssueSummary) { s.Status = "other" }, true}, + {"missing timestamp", func(s *IssueSummary) { s.ChangedAtMs = 0 }, true}, + {"negative timestamp", func(s *IssueSummary) { s.ChangedAtMs = -1 }, true}, + {"missing version", func(s *IssueSummary) { s.Version = 0 }, true}, + {"negative version", func(s *IssueSummary) { s.Version = -1 }, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + summary := valid + if tt.change != nil { + tt.change(&summary) + } + if tt.wantErr { + require.Error(t, summary.Validate()) + } else { + require.NoError(t, summary.Validate()) + } + }) + } +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index ebdcb1015..f9a584535 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "build_store.go", "issue_log_store.go", "issue_state_store.go", + "issue_summary_store.go", "issue_tracking_store.go", "project_fact_by_degree_store.go", "queue_log_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 7eb97d1be..37bc68699 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -50,6 +50,14 @@ Listing reads a bounded primary-key range in bytewise task/unit order and includ The MySQL backend uses the new `issue_state` table. Create it before deploying the updated readers. The prior `open_issue` table is not read or automatically copied; no in-repository runtime writer was enabled for it. Deployments with externally populated rows need an explicit backfill before switching readers. The repair executor still receives an `OpenIssue`, which describes an open occurrence rather than the storage record. +## Issue summaries + +`IssueSummaryStore` retains one read model per task/unit within its queue, independently of operational `IssueState`. It supports only complete-key create, get, and CAS replacement. Closed summaries remain present; removing them would also remove the source-order information needed to reject an older opening. There is no deletion or listing operation. + +`RequestID` and `BuildID` identify the latest represented transition, while `FirstSeenBuildID` identifies its current or most recent opening. Source order follows request ingest order, not opaque build IDs or timestamps. Within the same request, a closing transition supersedes an opening. `Version` is only the summary's CAS guard, not the source order. The materializer must compare incoming transitions before replacing a summary; the store does not enforce lifecycle ordering. + +The MySQL primary key is `(queue, task_id, unit_id)` with byte-exact identity and the same identifier limits as the issue-state store. Source request IDs follow the existing canonical positive decimal ID contract. The new `issue_summary` table must exist before deploying readers or writers. This storage layer does not yet populate summaries or expose a public open-issues list. + ## Issue lifecycle history `IssueLogStore` retains immutable open/closed status transitions independently of current issue state and per-build observations. Each transition is keyed by task, unit, build, and status within the bound queue. An empty unit identifies a task-wide issue. History remains when an issue closes or reopens; transitions do not carry a separate reference to the opening they belong to. diff --git a/stovepipe/extension/storage/issue_summary_store.go b/stovepipe/extension/storage/issue_summary_store.go new file mode 100644 index 000000000..d7f35662b --- /dev/null +++ b/stovepipe/extension/storage/issue_summary_store.go @@ -0,0 +1,37 @@ +// 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=issue_summary_store.go -destination=mock/issue_summary_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// IssueSummaryStore retains queue-scoped issue read models, including closed issues. +// Source ordering is the caller's responsibility; Version guards concurrent replacements. +type IssueSummaryStore interface { + // Create returns ErrAlreadyExists if the task/unit key exists, even when closed. + Create(ctx context.Context, summary entity.IssueSummary) error + + // Get returns ErrNotFound if the task/unit key has no summary. + Get(ctx context.Context, taskID, unitID string) (entity.IssueSummary, error) + + // Update replaces summary only at oldVersion, writing the supplied newVersion. + // A missing key or stale version returns ErrVersionMismatch. + Update(ctx context.Context, summary entity.IssueSummary, oldVersion, newVersion int32) error +} diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index 549e2df41..c9f45072b 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "build_store_mock.go", "issue_log_store_mock.go", "issue_state_store_mock.go", + "issue_summary_store_mock.go", "issue_tracking_store_mock.go", "project_fact_by_degree_store_mock.go", "queue_log_store_mock.go", diff --git a/stovepipe/extension/storage/mock/issue_summary_store_mock.go b/stovepipe/extension/storage/mock/issue_summary_store_mock.go new file mode 100644 index 000000000..06c6d7d67 --- /dev/null +++ b/stovepipe/extension/storage/mock/issue_summary_store_mock.go @@ -0,0 +1,85 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: issue_summary_store.go +// +// Generated by this command: +// +// mockgen -source=issue_summary_store.go -destination=mock/issue_summary_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" +) + +// MockIssueSummaryStore is a mock of IssueSummaryStore interface. +type MockIssueSummaryStore struct { + ctrl *gomock.Controller + recorder *MockIssueSummaryStoreMockRecorder + isgomock struct{} +} + +// MockIssueSummaryStoreMockRecorder is the mock recorder for MockIssueSummaryStore. +type MockIssueSummaryStoreMockRecorder struct { + mock *MockIssueSummaryStore +} + +// NewMockIssueSummaryStore creates a new mock instance. +func NewMockIssueSummaryStore(ctrl *gomock.Controller) *MockIssueSummaryStore { + mock := &MockIssueSummaryStore{ctrl: ctrl} + mock.recorder = &MockIssueSummaryStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockIssueSummaryStore) EXPECT() *MockIssueSummaryStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockIssueSummaryStore) Create(ctx context.Context, summary entity.IssueSummary) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, summary) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockIssueSummaryStoreMockRecorder) Create(ctx, summary any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockIssueSummaryStore)(nil).Create), ctx, summary) +} + +// Get mocks base method. +func (m *MockIssueSummaryStore) Get(ctx context.Context, taskID, unitID string) (entity.IssueSummary, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID) + ret0, _ := ret[0].(entity.IssueSummary) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockIssueSummaryStoreMockRecorder) Get(ctx, taskID, unitID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockIssueSummaryStore)(nil).Get), ctx, taskID, unitID) +} + +// Update mocks base method. +func (m *MockIssueSummaryStore) Update(ctx context.Context, summary entity.IssueSummary, oldVersion, newVersion int32) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Update", ctx, summary, oldVersion, newVersion) + ret0, _ := ret[0].(error) + return ret0 +} + +// Update indicates an expected call of Update. +func (mr *MockIssueSummaryStoreMockRecorder) Update(ctx, summary, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockIssueSummaryStore)(nil).Update), ctx, summary, oldVersion, newVersion) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index ec86a0158..7b45df72f 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -135,6 +135,20 @@ func (mr *MockStorageMockRecorder) GetIssueStateStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueStateStore", reflect.TypeOf((*MockStorage)(nil).GetIssueStateStore)) } +// GetIssueSummaryStore mocks base method. +func (m *MockStorage) GetIssueSummaryStore() storage.IssueSummaryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetIssueSummaryStore") + ret0, _ := ret[0].(storage.IssueSummaryStore) + return ret0 +} + +// GetIssueSummaryStore indicates an expected call of GetIssueSummaryStore. +func (mr *MockStorageMockRecorder) GetIssueSummaryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetIssueSummaryStore)) +} + // GetIssueTrackingStore mocks base method. func (m *MockStorage) GetIssueTrackingStore() storage.IssueTrackingStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index f5660c3fc..c681a7e63 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "build_store.go", "issue_log_store.go", "issue_state_store.go", + "issue_summary_store.go", "issue_tracking_store.go", "project_fact_by_degree_store.go", "queue_log_store.go", @@ -41,6 +42,7 @@ go_test( "issue_log_store_test.go", "issue_state_list_test.go", "issue_state_store_test.go", + "issue_summary_store_test.go", "issue_tracking_store_test.go", "project_fact_by_degree_store_test.go", "queue_store_test.go", diff --git a/stovepipe/extension/storage/mysql/issue_summary_store.go b/stovepipe/extension/storage/mysql/issue_summary_store.go new file mode 100644 index 000000000..09b536f80 --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_summary_store.go @@ -0,0 +1,132 @@ +// 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 issueSummaryStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewIssueSummaryStore binds issue summaries to one queue. +func NewIssueSummaryStore(db *sql.DB, scope tally.Scope, queue string) storage.IssueSummaryStore { + return &issueSummaryStore{db: db, scope: scope, queue: queue} +} + +func (s *issueSummaryStore) Create(ctx context.Context, summary entity.IssueSummary) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := s.validateSummary(summary); err != nil { + return err + } + + _, err := s.db.ExecContext(ctx, + `INSERT INTO issue_summary (queue, task_id, unit_id, status, first_seen_build_id, request_id, build_id, changed_at_ms, version) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + s.queue, summary.TaskID, summary.UnitID, summary.Status, summary.FirstSeenBuildID, + summary.RequestID, summary.BuildID, summary.ChangedAtMs, summary.Version) + if isDuplicateEntry(err) { + return storage.ErrAlreadyExists + } + if err != nil { + return fmt.Errorf("create issue summary: %w", err) + } + return nil +} + +func (s *issueSummaryStore) Get(ctx context.Context, taskID, unitID string) (summary entity.IssueSummary, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if taskID == "" { + return entity.IssueSummary{}, fmt.Errorf("issue summary task ID must not be empty") + } + if err := validateBuildIssueKey(s.queue, "", taskID, unitID); err != nil { + return entity.IssueSummary{}, err + } + + err := s.db.QueryRowContext(ctx, + `SELECT task_id, unit_id, status, first_seen_build_id, request_id, build_id, changed_at_ms, version + FROM issue_summary WHERE queue = ? AND task_id = ? AND unit_id = ?`, + s.queue, taskID, unitID, + ).Scan(&summary.TaskID, &summary.UnitID, &summary.Status, &summary.FirstSeenBuildID, + &summary.RequestID, &summary.BuildID, &summary.ChangedAtMs, &summary.Version) + if errors.Is(err, sql.ErrNoRows) { + return entity.IssueSummary{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.IssueSummary{}, fmt.Errorf("get issue summary: %w", err) + } + return summary, nil +} + +func (s *issueSummaryStore) Update(ctx context.Context, summary entity.IssueSummary, oldVersion, newVersion int32) (retErr error) { + op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + replacement := summary + replacement.Version = newVersion + if err := s.validateSummary(replacement); err != nil { + return err + } + if oldVersion <= 0 { + return fmt.Errorf("issue summary guard version must be positive") + } + + result, err := s.db.ExecContext(ctx, + `UPDATE issue_summary SET status = ?, first_seen_build_id = ?, request_id = ?, build_id = ?, changed_at_ms = ?, version = ? + WHERE queue = ? AND task_id = ? AND unit_id = ? AND version = ?`, + summary.Status, summary.FirstSeenBuildID, summary.RequestID, summary.BuildID, summary.ChangedAtMs, + newVersion, s.queue, summary.TaskID, summary.UnitID, oldVersion) + if err != nil { + return fmt.Errorf("update issue summary: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("read issue summary update count: %w", err) + } + if rows != 1 { + return storage.ErrVersionMismatch + } + return nil +} + +func (s *issueSummaryStore) validateSummary(summary entity.IssueSummary) error { + if err := summary.Validate(); err != nil { + return err + } + if err := validateBuildIssueKey(s.queue, summary.BuildID, summary.TaskID, summary.UnitID); err != nil { + return err + } + if err := validateBuildIssueKey(s.queue, summary.FirstSeenBuildID, summary.TaskID, summary.UnitID); err != nil { + return err + } + return nil +} + +var _ storage.IssueSummaryStore = (*issueSummaryStore)(nil) diff --git a/stovepipe/extension/storage/mysql/issue_summary_store_test.go b/stovepipe/extension/storage/mysql/issue_summary_store_test.go new file mode 100644 index 000000000..f93a6761a --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_summary_store_test.go @@ -0,0 +1,205 @@ +// 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" + "strings" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +const ( + insertIssueSummarySQL = `INSERT INTO issue_summary (queue, task_id, unit_id, status, first_seen_build_id, request_id, build_id, changed_at_ms, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)` + getIssueSummarySQL = `SELECT task_id, unit_id, status, first_seen_build_id, request_id, build_id, changed_at_ms, version FROM issue_summary WHERE queue = ? AND task_id = ? AND unit_id = ?` + updateIssueSummarySQL = `UPDATE issue_summary SET status = ?, first_seen_build_id = ?, request_id = ?, build_id = ?, changed_at_ms = ?, version = ? WHERE queue = ? AND task_id = ? AND unit_id = ? AND version = ?` +) + +func testIssueSummary() entity.IssueSummary { + return entity.IssueSummary{TaskID: "validation", UnitID: "unit", Status: entity.IssueStatusOpen, + FirstSeenBuildID: "build/1", RequestID: "1", BuildID: "build/1", ChangedAtMs: 1000, Version: 1} +} + +func newIssueSummaryTestStore(t *testing.T) (sqlmock.Sqlmock, storage.IssueSummaryStore) { + t.Helper() + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual)) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, mock.ExpectationsWereMet()) + _ = db.Close() + }) + return mock, NewIssueSummaryStore(db, testMetrics(), "queue/main") +} + +func TestIssueSummaryStoreCreate(t *testing.T) { + dependencyErr := errors.New("database unavailable") + for _, tt := range []struct { + name string + dbErr error + wantErr error + }{ + {name: "created"}, + {"duplicate", &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, storage.ErrAlreadyExists}, + {"database error", dependencyErr, dependencyErr}, + } { + t.Run(tt.name, func(t *testing.T) { + mock, store := newIssueSummaryTestStore(t) + summary := testIssueSummary() + expected := mock.ExpectExec(insertIssueSummarySQL).WithArgs("queue/main", summary.TaskID, summary.UnitID, + summary.Status, summary.FirstSeenBuildID, summary.RequestID, summary.BuildID, summary.ChangedAtMs, summary.Version) + if tt.dbErr != nil { + expected.WillReturnError(tt.dbErr) + } else { + expected.WillReturnResult(sqlmock.NewResult(0, 1)) + } + err := store.Create(context.Background(), summary) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + }) + } +} + +func TestIssueSummaryStoreGet(t *testing.T) { + dependencyErr := errors.New("database unavailable") + for _, tt := range []struct { + name string + status entity.IssueStatus + dbErr error + wantErr error + }{ + {name: "open", status: entity.IssueStatusOpen}, + {name: "closed", status: entity.IssueStatusClosed}, + {name: "not found", dbErr: sql.ErrNoRows, wantErr: storage.ErrNotFound}, + {name: "database error", dbErr: dependencyErr, wantErr: dependencyErr}, + } { + t.Run(tt.name, func(t *testing.T) { + mock, store := newIssueSummaryTestStore(t) + want := testIssueSummary() + want.Status = tt.status + expected := mock.ExpectQuery(getIssueSummarySQL).WithArgs("queue/main", want.TaskID, want.UnitID) + if tt.dbErr != nil { + expected.WillReturnError(tt.dbErr) + } else { + expected.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "status", "first_seen_build_id", "request_id", "build_id", "changed_at_ms", "version"}). + AddRow(want.TaskID, want.UnitID, want.Status, want.FirstSeenBuildID, want.RequestID, want.BuildID, want.ChangedAtMs, want.Version)) + } + got, err := store.Get(context.Background(), want.TaskID, want.UnitID) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + require.Empty(t, got) + } else { + require.NoError(t, err) + require.Equal(t, want, got) + } + }) + } +} + +func TestIssueSummaryStoreUpdate(t *testing.T) { + dependencyErr := errors.New("database unavailable") + for _, tt := range []struct { + name string + dbErr error + result sql.Result + wantErr error + }{ + {name: "replaced", result: sqlmock.NewResult(0, 1)}, + {name: "stale or missing", result: sqlmock.NewResult(0, 0), wantErr: storage.ErrVersionMismatch}, + {name: "database error", dbErr: dependencyErr, wantErr: dependencyErr}, + {name: "row count error", result: sqlmock.NewErrorResult(dependencyErr), wantErr: dependencyErr}, + } { + t.Run(tt.name, func(t *testing.T) { + mock, store := newIssueSummaryTestStore(t) + summary := testIssueSummary() + summary.Status, summary.RequestID, summary.BuildID = entity.IssueStatusClosed, "2", "build/2" + summary.ChangedAtMs = 2000 + original := summary + expected := mock.ExpectExec(updateIssueSummarySQL).WithArgs(summary.Status, summary.FirstSeenBuildID, + summary.RequestID, summary.BuildID, summary.ChangedAtMs, 2, "queue/main", summary.TaskID, summary.UnitID, 1) + if tt.dbErr != nil { + expected.WillReturnError(tt.dbErr) + } else { + expected.WillReturnResult(tt.result) + } + err := store.Update(context.Background(), summary, 1, 2) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + require.Equal(t, original, summary) + }) + } +} + +func TestIssueSummaryStoreRejectsInvalidWrites(t *testing.T) { + for _, tt := range []struct { + name string + queue string + change func(*entity.IssueSummary) + oldVersion int32 + newVersion int32 + }{ + {"empty task", "queue", func(s *entity.IssueSummary) { s.TaskID = "" }, 1, 2}, + {"invalid status", "queue", func(s *entity.IssueSummary) { s.Status = "invalid" }, 1, 2}, + {"long queue", strings.Repeat("q", 256), nil, 1, 2}, + {"long task", "queue", func(s *entity.IssueSummary) { s.TaskID = strings.Repeat("t", 256) }, 1, 2}, + {"long unit", "queue", func(s *entity.IssueSummary) { s.UnitID = strings.Repeat("u", 1025) }, 1, 2}, + {"long opening build", "queue", func(s *entity.IssueSummary) { s.FirstSeenBuildID = strings.Repeat("b", 1021) }, 1, 2}, + {"long source build", "queue", func(s *entity.IssueSummary) { s.BuildID = strings.Repeat("b", 1021) }, 1, 2}, + {"long source request", "queue", func(s *entity.IssueSummary) { s.RequestID = strings.Repeat("1", 256) }, 1, 2}, + {"invalid guard", "queue", nil, 0, 2}, + {"invalid replacement version", "queue", nil, 1, 0}, + } { + t.Run(tt.name, func(t *testing.T) { + mock, store := newIssueSummaryTestStore(t) + concrete := store.(*issueSummaryStore) + concrete.queue = tt.queue + summary := testIssueSummary() + if tt.change != nil { + tt.change(&summary) + } + if tt.oldVersion == 1 && tt.newVersion == 2 { + require.Error(t, store.Create(context.Background(), summary)) + } + require.Error(t, store.Update(context.Background(), summary, tt.oldVersion, tt.newVersion)) + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueSummaryStoreRejectsInvalidReadKeys(t *testing.T) { + for _, tt := range []struct{ name, task, unit string }{ + {"empty task", "", "unit"}, + {"long task", strings.Repeat("t", 256), "unit"}, + {"long unit", "task", strings.Repeat("u", 1025)}, + } { + t.Run(tt.name, func(t *testing.T) { + _, store := newIssueSummaryTestStore(t) + _, err := store.Get(context.Background(), tt.task, tt.unit) + require.Error(t, err) + }) + } +} diff --git a/stovepipe/extension/storage/mysql/schema/issue_summary.sql b/stovepipe/extension/storage/mysql/schema/issue_summary.sql new file mode 100644 index 000000000..9caa4169c --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/issue_summary.sql @@ -0,0 +1,15 @@ +-- Closed summaries retain their source order independently of the projection's CAS version. +CREATE TABLE IF NOT EXISTS issue_summary ( + queue VARBINARY(255) NOT NULL, + task_id VARBINARY(255) NOT NULL, + unit_id VARBINARY(1024) NOT NULL, + + status VARBINARY(32) NOT NULL, + first_seen_build_id VARBINARY(1020) NOT NULL, + request_id VARBINARY(255) NOT NULL, + build_id VARBINARY(1020) NOT NULL, + changed_at_ms BIGINT NOT NULL, + version INT NOT NULL, + + 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 afb28d28e..c7408770c 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -58,6 +58,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { buildIssueStore: NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName), issueStateStore: NewIssueStateStore(s.db, queueScope.SubScope("issue_state_store"), queueName), issueLogStore: NewIssueLogStore(s.db, queueScope.SubScope("issue_log_store"), queueName), + issueSummaryStore: NewIssueSummaryStore(s.db, queueScope.SubScope("issue_summary_store"), queueName), issueTrackingStore: NewIssueTrackingStore(s.db, queueScope.SubScope("issue_tracking_store"), queueName), validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), projectFactByDegreeStore: NewProjectFactByDegreeStore(s.db, queueScope.SubScope("project_fact_by_degree_store"), queueName), @@ -83,6 +84,7 @@ type mysqlStorage struct { buildIssueStore storage.BuildIssueStore issueStateStore storage.IssueStateStore issueLogStore storage.IssueLogStore + issueSummaryStore storage.IssueSummaryStore issueTrackingStore storage.IssueTrackingStore validationFactStore storage.ValidationFactStore projectFactByDegreeStore storage.ProjectFactByDegreeStore @@ -153,6 +155,10 @@ func (f *mysqlStorage) GetIssueLogStore() storage.IssueLogStore { return f.issueLogStore } +func (f *mysqlStorage) GetIssueSummaryStore() storage.IssueSummaryStore { + return f.issueSummaryStore +} + func (f *mysqlStorage) GetIssueTrackingStore() storage.IssueTrackingStore { return f.issueTrackingStore } diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index 2d2fda26b..f9232d8ee 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -51,6 +51,7 @@ func TestNewStorage(t *testing.T) { assert.NotNil(t, bound.GetBuildIssueStore()) assert.NotNil(t, bound.GetIssueStateStore()) assert.NotNil(t, bound.GetIssueLogStore()) + assert.NotNil(t, bound.GetIssueSummaryStore()) assert.NotNil(t, bound.GetIssueTrackingStore()) _, err = s.For("") @@ -108,6 +109,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "issue_summary_store", + op: "get", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetIssueSummaryStore().Get(context.Background(), "task", "unit") + return err + }, + }, { name: "request_acceptance_store", op: "list", diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 68132fd04..c63e94d0e 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -103,6 +103,9 @@ type Storage interface { // GetIssueLogStore returns the IssueLogStore instance. GetIssueLogStore() IssueLogStore + // GetIssueSummaryStore returns the IssueSummaryStore instance. + GetIssueSummaryStore() IssueSummaryStore + // GetIssueTrackingStore returns the operational reconciliation progress store. GetIssueTrackingStore() IssueTrackingStore diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index e4a69b5aa..47db77393 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -9,6 +9,7 @@ go_test( "issue_log_test.go", "issue_state_list_test.go", "issue_state_test.go", + "issue_summary_test.go", "issue_tracking_test.go", "project_fact_by_degree_test.go", "queue_state_transitions_test.go", diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_summary_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_summary_test.go new file mode 100644 index 000000000..16d14635b --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/issue_summary_test.go @@ -0,0 +1,133 @@ +// 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 testIssueSummaryStore(t *testing.T, ctx context.Context, factory storage.Factory) { + t.Helper() + bound, err := factory.For(storage.Config{QueueName: "issue-summary/main"}) + require.NoError(t, err) + store := bound.GetIssueSummaryStore() + summary := entity.IssueSummary{TaskID: "validation", UnitID: "unit", Status: entity.IssueStatusOpen, + FirstSeenBuildID: "build/1", RequestID: "1", BuildID: "build/1", ChangedAtMs: 1000, Version: 1} + + _, err = store.Get(ctx, summary.TaskID, summary.UnitID) + require.ErrorIs(t, err, storage.ErrNotFound) + require.ErrorIs(t, store.Update(ctx, summary, 1, 2), storage.ErrVersionMismatch) + require.NoError(t, store.Create(ctx, summary)) + require.ErrorIs(t, store.Create(ctx, summary), storage.ErrAlreadyExists) + got, err := store.Get(ctx, summary.TaskID, summary.UnitID) + require.NoError(t, err) + require.Equal(t, summary, got) + + closed := summary + closed.Status, closed.RequestID, closed.BuildID = entity.IssueStatusClosed, "2", "build/2" + closed.ChangedAtMs = 2000 + require.NoError(t, store.Update(ctx, closed, 1, 2)) + closed.Version = 2 + got, err = store.Get(ctx, summary.TaskID, summary.UnitID) + require.NoError(t, err) + require.Equal(t, closed, got) + require.ErrorIs(t, store.Create(ctx, summary), storage.ErrAlreadyExists) + require.ErrorIs(t, store.Update(ctx, summary, 1, 2), storage.ErrVersionMismatch) + + reopened := closed + reopened.Status, reopened.RequestID, reopened.BuildID = entity.IssueStatusOpen, "3", "build/3" + reopened.FirstSeenBuildID, reopened.ChangedAtMs = "build/3", 3000 + require.NoError(t, store.Update(ctx, reopened, 2, 3)) + reopened.Version = 3 + require.ErrorIs(t, store.Update(ctx, closed, 2, 3), storage.ErrVersionMismatch) + got, err = store.Get(ctx, summary.TaskID, summary.UnitID) + require.NoError(t, err) + require.Equal(t, reopened, got) + + t.Run("byte exact keys and queue isolation", func(t *testing.T) { + other, err := factory.For(storage.Config{QueueName: "Issue-summary/main"}) + require.NoError(t, err) + _, err = other.GetIssueSummaryStore().Get(ctx, summary.TaskID, summary.UnitID) + require.ErrorIs(t, err, storage.ErrNotFound) + require.NoError(t, other.GetIssueSummaryStore().Create(ctx, summary)) + require.NoError(t, other.GetIssueSummaryStore().Update(ctx, closed, 1, 2)) + got, err := store.Get(ctx, summary.TaskID, summary.UnitID) + require.NoError(t, err) + require.Equal(t, reopened, got) + + for _, key := range []entity.BuildIssue{ + {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 := summary + 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) + } + }) + + t.Run("one competing replacement wins", func(t *testing.T) { + results := make(chan error, 2) + for _, status := range []entity.IssueStatus{entity.IssueStatusOpen, entity.IssueStatusClosed} { + replacement := reopened + replacement.Status = status + go func() { results <- store.Update(ctx, replacement, 3, 4) }() + } + successes, conflicts := 0, 0 + for range 2 { + err := <-results + switch { + case err == nil: + successes++ + case errors.Is(err, storage.ErrVersionMismatch): + conflicts++ + default: + require.NoError(t, err) + } + } + require.Equal(t, 1, successes) + require.Equal(t, 1, conflicts) + }) + + t.Run("identifier boundaries", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) + require.NoError(t, err) + value := summary + value.TaskID, value.UnitID = strings.Repeat("t", 255), strings.Repeat("😀", 256) + value.FirstSeenBuildID, value.BuildID = strings.Repeat("😀", 255), strings.Repeat("😀", 255) + value.RequestID = "9223372036854775807" + store := bound.GetIssueSummaryStore() + require.NoError(t, store.Create(ctx, value)) + got, err := store.Get(ctx, value.TaskID, value.UnitID) + require.NoError(t, err) + require.Equal(t, value, got) + value.UnitID += "x" + require.Error(t, store.Create(ctx, value)) + require.Error(t, store.Update(ctx, value, 1, 2)) + }) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index fb8d05dcb..0c4f8654e 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) { testIssueStateStore(t, ctx, factory) }) + t.Run("IssueSummaryStore", func(t *testing.T) { + resetStorage(t, db) + testIssueSummaryStore(t, ctx, factory) + }) + t.Run("IssueStateList", func(t *testing.T) { resetStorage(t, db) testIssueStateList(t, ctx, factory) @@ -184,6 +189,7 @@ func resetStorage(t *testing.T, db *sql.DB) { for _, statement := range []string{ "TRUNCATE TABLE issue_state", + "TRUNCATE TABLE issue_summary", "TRUNCATE TABLE issue_log", "TRUNCATE TABLE issue_tracking", "TRUNCATE TABLE queue_log",