From 24cc5f2bf877c6b87b7171d31038d478125d665c Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Fri, 6 Nov 2026 18:35:00 +0000 Subject: [PATCH] feat(stovepipe): Store request issue summaries --- stovepipe/entity/BUILD.bazel | 2 + stovepipe/entity/request_issue_summary.go | 61 +++++++ .../entity/request_issue_summary_test.go | 39 ++++ stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 4 + stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../mock/request_issue_summary_store_mock.go | 101 ++++++++++ .../extension/storage/mock/storage_mock.go | 14 ++ stovepipe/extension/storage/mysql/BUILD.bazel | 3 + .../mysql/request_issue_summary_store.go | 172 ++++++++++++++++++ .../mysql/request_issue_summary_store_test.go | 134 ++++++++++++++ .../mysql/schema/request_issue_summary.sql | 15 ++ stovepipe/extension/storage/mysql/storage.go | 6 + .../storage/request_issue_summary_store.go | 54 ++++++ stovepipe/extension/storage/storage.go | 3 + .../extension/storage/mysql/BUILD.bazel | 1 + .../mysql/request_issue_summary_test.go | 72 ++++++++ .../extension/storage/mysql/storage_test.go | 4 + 18 files changed, 687 insertions(+) create mode 100644 stovepipe/entity/request_issue_summary.go create mode 100644 stovepipe/entity/request_issue_summary_test.go create mode 100644 stovepipe/extension/storage/mock/request_issue_summary_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/request_issue_summary_store.go create mode 100644 stovepipe/extension/storage/mysql/request_issue_summary_store_test.go create mode 100644 stovepipe/extension/storage/mysql/schema/request_issue_summary.sql create mode 100644 stovepipe/extension/storage/request_issue_summary_store.go create mode 100644 test/integration/stovepipe/extension/storage/mysql/request_issue_summary_test.go diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index ea53fdf11..cca9bc322 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -25,6 +25,7 @@ go_library( "request_acceptance.go", "request_history.go", "request_id.go", + "request_issue_summary.go", "request_log.go", "request_summary.go", "validation_fact.go", @@ -45,6 +46,7 @@ go_test( "project_fact_by_degree_test.go", "queue_state_test.go", "request_id_test.go", + "request_issue_summary_test.go", "request_log_test.go", "request_test.go", "validation_fact_test.go", diff --git a/stovepipe/entity/request_issue_summary.go b/stovepipe/entity/request_issue_summary.go new file mode 100644 index 000000000..a2aa3fa1f --- /dev/null +++ b/stovepipe/entity/request_issue_summary.go @@ -0,0 +1,61 @@ +// 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" + +// RequestIssueKind classifies an accepted issue update within one request. +type RequestIssueKind string + +const ( + RequestIssueKindNew RequestIssueKind = "new" + RequestIssueKindExisting RequestIssueKind = "existing" + RequestIssueKindClosed RequestIssueKind = "closed" +) + +// RequestIssueSummary is a materialized result of one request's issue reconciliation. +// Its identity is request, task, and unit within a queue. +type RequestIssueSummary struct { + // RequestID identifies the request that processed the issue. + RequestID string `json:"request_id"` + // TaskID identifies the logical validation task. + TaskID string `json:"task_id"` + // UnitID identifies the validation unit; empty for a task-wide failure. + UnitID string `json:"unit_id"` + // Kind distinguishes newly opened, already open, and closed issues. + Kind RequestIssueKind `json:"kind"` + // BuildID identifies the build whose accepted update produced this result. + BuildID string `json:"build_id"` + // FirstSeenBuildID identifies the opening, including for an existing or closed issue. + FirstSeenBuildID string `json:"first_seen_build_id"` + // Version is the optimistic-lock version, initially zero. + Version int32 `json:"version"` +} + +// Validate checks the identity and classification of a request issue. +func (i RequestIssueSummary) Validate() error { + if i.Version < 0 { + return fmt.Errorf("request issue summary version must not be negative") + } + if i.RequestID == "" || i.TaskID == "" || i.BuildID == "" || i.FirstSeenBuildID == "" { + return fmt.Errorf("request issue requires request, task, build, and first-seen build IDs") + } + switch i.Kind { + case RequestIssueKindNew, RequestIssueKindExisting, RequestIssueKindClosed: + return nil + default: + return fmt.Errorf("invalid request issue kind %q", i.Kind) + } +} diff --git a/stovepipe/entity/request_issue_summary_test.go b/stovepipe/entity/request_issue_summary_test.go new file mode 100644 index 000000000..fa8126ad6 --- /dev/null +++ b/stovepipe/entity/request_issue_summary_test.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 ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRequestIssueValidate(t *testing.T) { + valid := RequestIssueSummary{RequestID: "12", TaskID: "test", BuildID: "build/12", FirstSeenBuildID: "build/10", Kind: RequestIssueKindExisting} + for _, kind := range []RequestIssueKind{RequestIssueKindNew, RequestIssueKindExisting, RequestIssueKindClosed} { + issue := valid + issue.Kind = kind + require.NoError(t, issue.Validate()) + } + for _, change := range []func(*RequestIssueSummary){ + func(i *RequestIssueSummary) { i.RequestID = "" }, func(i *RequestIssueSummary) { i.TaskID = "" }, + func(i *RequestIssueSummary) { i.BuildID = "" }, func(i *RequestIssueSummary) { i.FirstSeenBuildID = "" }, + func(i *RequestIssueSummary) { i.Kind = "open" }, + } { + issue := valid + change(&issue) + require.Error(t, issue.Validate()) + } +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index de5782f47..00542e3f3 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -15,6 +15,7 @@ go_library( "queue_store.go", "queue_summary_store.go", "request_acceptance_store.go", + "request_issue_summary_store.go", "request_log_store.go", "request_store.go", "request_summary_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index c194e43bb..31b103e7c 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -91,3 +91,7 @@ The issue-tracking table must include the notification, cursor, `finalize`, and ### Applied transition facts `IssueState.LastProcessedBuildID` identifies the most recent build applied to an issue, including unchanged results. `LastChangedBuildID` and `ChangedAtMs` identify the actual opening or closing and supply its history identity and timestamp. They do not advance when only the processed build changes. `FirstSeenBuildID` remains the opening identity across closure and changes on reopening. These references must remain readable while used for source ordering. The current-issue cursor and the row's processed source provide the storage needed for retry-safe application without retaining a future write payload. + +## Request issue results + +The request-issue summary store retains one materialized result per request, task, and unit in its bound queue. New includes reopened issues; existing means this request reported an issue that was already open; closed means this request actually closed it. Issues merely left open because closure was suppressed are not existing results. Reads use a bounded request-prefix scan in task/unit order. These results describe the request's accepted updates, not the queue's full open-issue set. diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index 7e7545c10..15380ccf8 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -15,6 +15,7 @@ go_library( "queue_store_mock.go", "queue_summary_store_mock.go", "request_acceptance_store_mock.go", + "request_issue_summary_store_mock.go", "request_log_store_mock.go", "request_store_mock.go", "request_summary_store_mock.go", diff --git a/stovepipe/extension/storage/mock/request_issue_summary_store_mock.go b/stovepipe/extension/storage/mock/request_issue_summary_store_mock.go new file mode 100644 index 000000000..6c774ff38 --- /dev/null +++ b/stovepipe/extension/storage/mock/request_issue_summary_store_mock.go @@ -0,0 +1,101 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_issue_summary_store.go +// +// Generated by this command: +// +// mockgen -source=request_issue_summary_store.go -destination=mock/request_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" + storage "github.com/uber/submitqueue/stovepipe/extension/storage" + gomock "go.uber.org/mock/gomock" +) + +// MockRequestIssueSummaryStore is a mock of RequestIssueSummaryStore interface. +type MockRequestIssueSummaryStore struct { + ctrl *gomock.Controller + recorder *MockRequestIssueSummaryStoreMockRecorder + isgomock struct{} +} + +// MockRequestIssueSummaryStoreMockRecorder is the mock recorder for MockRequestIssueSummaryStore. +type MockRequestIssueSummaryStoreMockRecorder struct { + mock *MockRequestIssueSummaryStore +} + +// NewMockRequestIssueSummaryStore creates a new mock instance. +func NewMockRequestIssueSummaryStore(ctrl *gomock.Controller) *MockRequestIssueSummaryStore { + mock := &MockRequestIssueSummaryStore{ctrl: ctrl} + mock.recorder = &MockRequestIssueSummaryStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestIssueSummaryStore) EXPECT() *MockRequestIssueSummaryStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestIssueSummaryStore) Create(arg0 context.Context, arg1 entity.RequestIssueSummary) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", arg0, arg1) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockRequestIssueSummaryStoreMockRecorder) Create(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestIssueSummaryStore)(nil).Create), arg0, arg1) +} + +// Get mocks base method. +func (m *MockRequestIssueSummaryStore) Get(ctx context.Context, requestID, taskID, unitID string) (entity.RequestIssueSummary, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, requestID, taskID, unitID) + ret0, _ := ret[0].(entity.RequestIssueSummary) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockRequestIssueSummaryStoreMockRecorder) Get(ctx, requestID, taskID, unitID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockRequestIssueSummaryStore)(nil).Get), ctx, requestID, taskID, unitID) +} + +// List mocks base method. +func (m *MockRequestIssueSummaryStore) List(ctx context.Context, requestID string, bounds storage.RequestIssueSummaryRange) ([]entity.RequestIssueSummary, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, requestID, bounds) + ret0, _ := ret[0].([]entity.RequestIssueSummary) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockRequestIssueSummaryStoreMockRecorder) List(ctx, requestID, bounds any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockRequestIssueSummaryStore)(nil).List), ctx, requestID, bounds) +} + +// Update mocks base method. +func (m *MockRequestIssueSummaryStore) Update(ctx context.Context, summary entity.RequestIssueSummary, 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 *MockRequestIssueSummaryStoreMockRecorder) Update(ctx, summary, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockRequestIssueSummaryStore)(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 2b0e19c93..1594c3929 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -247,6 +247,20 @@ func (mr *MockStorageMockRecorder) GetRequestAcceptanceStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestAcceptanceStore", reflect.TypeOf((*MockStorage)(nil).GetRequestAcceptanceStore)) } +// GetRequestIssueSummaryStore mocks base method. +func (m *MockStorage) GetRequestIssueSummaryStore() storage.RequestIssueSummaryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestIssueSummaryStore") + ret0, _ := ret[0].(storage.RequestIssueSummaryStore) + return ret0 +} + +// GetRequestIssueSummaryStore indicates an expected call of GetRequestIssueSummaryStore. +func (mr *MockStorageMockRecorder) GetRequestIssueSummaryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestIssueSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestIssueSummaryStore)) +} + // GetRequestLogStore mocks base method. func (m *MockStorage) GetRequestLogStore() storage.RequestLogStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 45b86d387..258cc5464 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -15,6 +15,7 @@ go_library( "queue_store.go", "queue_summary_store.go", "request_acceptance_store.go", + "request_issue_summary_store.go", "request_log_store.go", "request_store.go", "request_summary_store.go", @@ -25,6 +26,7 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/extension/storage/mysql", visibility = ["//visibility:public"], deps = [ + "//platform/base/id:go_default_library", "//platform/metrics:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage:go_default_library", @@ -50,6 +52,7 @@ go_test( "queue_store_test.go", "queue_summary_store_test.go", "request_acceptance_store_test.go", + "request_issue_summary_store_test.go", "request_log_store_test.go", "request_store_test.go", "request_summary_store_test.go", diff --git a/stovepipe/extension/storage/mysql/request_issue_summary_store.go b/stovepipe/extension/storage/mysql/request_issue_summary_store.go new file mode 100644 index 000000000..628b9eed0 --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_issue_summary_store.go @@ -0,0 +1,172 @@ +// 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/base/id" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +const listRequestIssueSummariesQuery = ` + SELECT request_id, task_id, unit_id, kind, build_id, first_seen_build_id, version + FROM request_issue_summary WHERE queue = ? AND request_id = ? + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +const listRequestIssueSummariesAfterCursorQuery = ` + SELECT request_id, task_id, unit_id, kind, build_id, first_seen_build_id, version + FROM request_issue_summary WHERE queue = ? AND request_id = ? + AND (task_id > ? OR (task_id = ? AND unit_id > ?)) + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +type requestIssueSummaryStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewRequestIssueSummaryStore binds per-request issue results to one queue. +func NewRequestIssueSummaryStore(db *sql.DB, scope tally.Scope, queue string) storage.RequestIssueSummaryStore { + return &requestIssueSummaryStore{db: db, scope: scope, queue: queue} +} + +func (s *requestIssueSummaryStore) Create(ctx context.Context, issue entity.RequestIssueSummary) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + if err := issue.Validate(); err != nil { + return err + } + if err := s.validateKey(issue.RequestID, issue.TaskID, issue.UnitID); err != nil { + return err + } + if err := validateBuildIssueKey(s.queue, issue.BuildID, issue.TaskID, issue.UnitID); err != nil { + return err + } + if err := validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil { + return err + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO request_issue_summary (queue, request_id, task_id, unit_id, kind, build_id, first_seen_build_id, version) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, s.queue, issue.RequestID, issue.TaskID, issue.UnitID, issue.Kind, issue.BuildID, issue.FirstSeenBuildID, issue.Version) + if isDuplicateEntry(err) { + return storage.ErrAlreadyExists + } + if err != nil { + return fmt.Errorf("create request issue: %w", err) + } + return nil +} + +func (s *requestIssueSummaryStore) Get(ctx context.Context, requestID, taskID, unitID string) (ret entity.RequestIssueSummary, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + if taskID == "" { + return ret, fmt.Errorf("request issue task ID must not be empty") + } + if err := s.validateKey(requestID, taskID, unitID); err != nil { + return ret, err + } + err := s.db.QueryRowContext(ctx, + `SELECT request_id, task_id, unit_id, kind, build_id, first_seen_build_id, version + FROM request_issue_summary WHERE queue = ? AND request_id = ? AND task_id = ? AND unit_id = ?`, + s.queue, requestID, taskID, unitID).Scan(&ret.RequestID, &ret.TaskID, &ret.UnitID, &ret.Kind, &ret.BuildID, &ret.FirstSeenBuildID, &ret.Version) + if errors.Is(err, sql.ErrNoRows) { + return entity.RequestIssueSummary{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.RequestIssueSummary{}, fmt.Errorf("get request issue: %w", err) + } + return ret, nil +} + +func (s *requestIssueSummaryStore) List(ctx context.Context, requestID string, bounds storage.RequestIssueSummaryRange) (ret []entity.RequestIssueSummary, retErr error) { + op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + if bounds.Limit <= 0 { + return nil, fmt.Errorf("request issue range requires a positive limit") + } + cursor := bounds.After + if cursor.TaskID == "" && cursor.UnitID != "" { + return nil, fmt.Errorf("request issue cursor requires a task ID") + } + if err := s.validateKey(requestID, cursor.TaskID, cursor.UnitID); err != nil { + return nil, err + } + var rows *sql.Rows + var err error + if cursor.TaskID == "" { + rows, err = s.db.QueryContext(ctx, listRequestIssueSummariesQuery, s.queue, requestID, bounds.Limit) + } else { + rows, err = s.db.QueryContext(ctx, listRequestIssueSummariesAfterCursorQuery, s.queue, requestID, cursor.TaskID, cursor.TaskID, cursor.UnitID, bounds.Limit) + } + if err != nil { + return nil, fmt.Errorf("list request issues: %w", err) + } + defer rows.Close() + issues := make([]entity.RequestIssueSummary, 0) + for rows.Next() { + var issue entity.RequestIssueSummary + if err := rows.Scan(&issue.RequestID, &issue.TaskID, &issue.UnitID, &issue.Kind, &issue.BuildID, &issue.FirstSeenBuildID, &issue.Version); err != nil { + return nil, fmt.Errorf("scan request issue: %w", err) + } + issues = append(issues, issue) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate request issues: %w", err) + } + return issues, nil +} + +func (s *requestIssueSummaryStore) validateKey(requestID, taskID, unitID string) error { + if err := id.Validate(requestID); err != nil { + return err + } + return validateBuildIssueKey(s.queue, "", taskID, unitID) +} + +var _ storage.RequestIssueSummaryStore = (*requestIssueSummaryStore)(nil) + +func (s *requestIssueSummaryStore) Update(ctx context.Context, summary entity.RequestIssueSummary, oldVersion, newVersion int32) (retErr error) { + op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + if err := summary.Validate(); err != nil { + return err + } + if err := s.validateKey(summary.RequestID, summary.TaskID, summary.UnitID); err != nil { + return err + } + result, err := s.db.ExecContext(ctx, + `UPDATE request_issue_summary SET kind = ?, build_id = ?, first_seen_build_id = ?, version = ? + WHERE queue = ? AND request_id = ? AND task_id = ? AND unit_id = ? AND version = ?`, + summary.Kind, summary.BuildID, summary.FirstSeenBuildID, newVersion, s.queue, summary.RequestID, summary.TaskID, summary.UnitID, oldVersion) + if err != nil { + return fmt.Errorf("update request issue summary: %w", err) + } + affected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("read request issue summary update count: %w", err) + } + if affected != 1 { + return storage.ErrVersionMismatch + } + return nil +} diff --git a/stovepipe/extension/storage/mysql/request_issue_summary_store_test.go b/stovepipe/extension/storage/mysql/request_issue_summary_store_test.go new file mode 100644 index 000000000..0de563c75 --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_issue_summary_store_test.go @@ -0,0 +1,134 @@ +// 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" + "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 TestRequestIssueCreate(t *testing.T) { + for _, dbErr := range []error{nil, &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, assert.AnError} { + db, mock, err := sqlmock.New() + require.NoError(t, err) + store := NewRequestIssueSummaryStore(db, testMetrics(), "queue") + issue := entity.RequestIssueSummary{RequestID: "12", TaskID: "test", Kind: entity.RequestIssueKindNew, BuildID: "b/12", FirstSeenBuildID: "b/12"} + query := mock.ExpectExec("INSERT INTO request_issue_summary").WithArgs("queue", "12", "test", "", "new", "b/12", "b/12", int64(0)) + if dbErr == nil { + query.WillReturnResult(sqlmock.NewResult(0, 1)) + } else { + query.WillReturnError(dbErr) + } + err = store.Create(context.Background(), issue) + switch dbErr { + case nil: + require.NoError(t, err) + case assert.AnError: + require.ErrorIs(t, err, assert.AnError) + default: + require.ErrorIs(t, err, storage.ErrAlreadyExists) + } + require.NoError(t, mock.ExpectationsWereMet()) + db.Close() + } +} + +func TestRequestIssueGet(t *testing.T) { + for _, dbErr := range []error{nil, sql.ErrNoRows, assert.AnError} { + db, mock, err := sqlmock.New() + require.NoError(t, err) + store := NewRequestIssueSummaryStore(db, testMetrics(), "queue") + query := mock.ExpectQuery("SELECT .* FROM request_issue_summary WHERE queue = \\? AND request_id = \\? AND task_id = \\? AND unit_id = \\?").WithArgs("queue", "12", "test", "") + if dbErr == nil { + query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "task_id", "unit_id", "kind", "build_id", "first_seen_build_id", "version"}).AddRow("12", "test", "", "existing", "b/12", "b/10", int64(0))) + } else { + query.WillReturnError(dbErr) + } + got, err := store.Get(context.Background(), "12", "test", "") + switch dbErr { + case nil: + require.NoError(t, err) + require.Equal(t, entity.RequestIssueKindExisting, got.Kind) + case sql.ErrNoRows: + require.ErrorIs(t, err, storage.ErrNotFound) + default: + require.ErrorIs(t, err, dbErr) + } + require.NoError(t, mock.ExpectationsWereMet()) + db.Close() + } +} + +func TestRequestIssueList(t *testing.T) { + for _, after := range []storage.RequestIssueSummaryCursor{{}, {TaskID: "test", UnitID: "unit"}} { + db, mock, err := sqlmock.New() + require.NoError(t, err) + store := NewRequestIssueSummaryStore(db, testMetrics(), "queue") + var query *sqlmock.ExpectedQuery + if after.TaskID == "" { + query = mock.ExpectQuery(regexp.QuoteMeta(listRequestIssueSummariesQuery)).WithArgs("queue", "12", 2) + } else { + query = mock.ExpectQuery(regexp.QuoteMeta(listRequestIssueSummariesAfterCursorQuery)).WithArgs("queue", "12", "test", "test", "unit", 2) + } + query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "task_id", "unit_id", "kind", "build_id", "first_seen_build_id", "version"}).AddRow("12", "test", "z", "closed", "b/12", "b/10", int64(0))) + got, err := store.List(context.Background(), "12", storage.RequestIssueSummaryRange{After: after, Limit: 2}) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, entity.RequestIssueKindClosed, got[0].Kind) + require.NoError(t, mock.ExpectationsWereMet()) + db.Close() + } +} + +func TestRequestIssueInvalidBounds(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewRequestIssueSummaryStore(db, testMetrics(), "queue") + for _, bounds := range []storage.RequestIssueSummaryRange{{}, {Limit: -1}, {Limit: 1, After: storage.RequestIssueSummaryCursor{UnitID: "u"}}} { + _, err := store.List(context.Background(), "12", bounds) + require.Error(t, err) + } + _, err = store.List(context.Background(), "not-a-request", storage.RequestIssueSummaryRange{Limit: 1}) + require.Error(t, err) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestRequestIssueSummaryUpdate(t *testing.T) { + for _, affected := range []int64{0, 1} { + db, mock, err := sqlmock.New() + require.NoError(t, err) + summary := entity.RequestIssueSummary{RequestID: "12", TaskID: "test", Kind: entity.RequestIssueKindClosed, BuildID: "b/12", FirstSeenBuildID: "b/10", Version: 3} + mock.ExpectExec("UPDATE request_issue_summary").WithArgs("closed", "b/12", "b/10", int64(4), "queue", "12", "test", "", int64(3)).WillReturnResult(sqlmock.NewResult(0, affected)) + err = NewRequestIssueSummaryStore(db, testMetrics(), "queue").Update(context.Background(), summary, 3, 4) + if affected == 0 { + require.ErrorIs(t, err, storage.ErrVersionMismatch) + } else { + require.NoError(t, err) + } + require.Equal(t, int32(3), summary.Version) + require.NoError(t, mock.ExpectationsWereMet()) + db.Close() + } +} diff --git a/stovepipe/extension/storage/mysql/schema/request_issue_summary.sql b/stovepipe/extension/storage/mysql/schema/request_issue_summary.sql new file mode 100644 index 000000000..6433fc2f3 --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/request_issue_summary.sql @@ -0,0 +1,15 @@ +-- Accepted issue results within each request, ordered by task and unit. +CREATE TABLE IF NOT EXISTS request_issue_summary ( + queue VARBINARY(255) NOT NULL, + request_id VARBINARY(255) NOT NULL, + task_id VARBINARY(255) NOT NULL, + unit_id VARBINARY(1024) NOT NULL, + + kind VARCHAR(16) NOT NULL, + build_id VARBINARY(1020) NOT NULL, + first_seen_build_id VARBINARY(1020) NOT NULL, + + version INT NOT NULL DEFAULT 0, + + PRIMARY KEY (queue, request_id, task_id, unit_id) +) ENGINE=InnoDB; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index ea6207fe3..2b5f28b0a 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), + requestIssueSummaryStore: NewRequestIssueSummaryStore(s.db, queueScope.SubScope("request_issue_summary_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), @@ -83,6 +84,7 @@ type mysqlStorage struct { queueLogStore storage.QueueLogStore buildStore storage.BuildStore buildIssueStore storage.BuildIssueStore + requestIssueSummaryStore storage.RequestIssueSummaryStore issueStateStore storage.IssueStateStore issueLogStore storage.IssueLogStore issueSummaryStore storage.IssueSummaryStore @@ -168,3 +170,7 @@ func (f *mysqlStorage) GetOpenIssueStore() storage.OpenIssueStore { func (f *mysqlStorage) GetIssueTrackingStore() storage.IssueTrackingStore { return f.issueTrackingStore } + +func (f *mysqlStorage) GetRequestIssueSummaryStore() storage.RequestIssueSummaryStore { + return f.requestIssueSummaryStore +} diff --git a/stovepipe/extension/storage/request_issue_summary_store.go b/stovepipe/extension/storage/request_issue_summary_store.go new file mode 100644 index 000000000..e31ba21b8 --- /dev/null +++ b/stovepipe/extension/storage/request_issue_summary_store.go @@ -0,0 +1,54 @@ +// 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=request_issue_summary_store.go -destination=mock/request_issue_summary_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// RequestIssueSummaryCursor is an exclusive position in bytewise task/unit order. +type RequestIssueSummaryCursor struct { + // TaskID is the task component; empty starts the range. + TaskID string + // UnitID is the unit component; empty identifies a task-wide issue. + UnitID string +} + +// RequestIssueSummaryRange bounds a scan within one request's primary-key prefix. +type RequestIssueSummaryRange struct { + // After excludes this key and earlier keys. + After RequestIssueSummaryCursor + // Limit is the positive maximum number of results to return. + Limit int +} + +// RequestIssueSummaryStore retains materialized per-request results within the bound queue. +// Existing means reported and already open, not every issue left open in the queue. +type RequestIssueSummaryStore interface { + // Create returns ErrAlreadyExists without changing a previously recorded result. + Create(context.Context, entity.RequestIssueSummary) error + // Get returns ErrNotFound when the complete request/task/unit key is absent. + Get(ctx context.Context, requestID, taskID, unitID string) (entity.RequestIssueSummary, error) + // List returns at most Limit results in ascending bytewise task/unit order. + // Results may be added while reconciliation is in progress; completed results are stable. + List(ctx context.Context, requestID string, bounds RequestIssueSummaryRange) ([]entity.RequestIssueSummary, error) + // Update replaces a result if its version matches oldVersion, writing newVersion. + // Returns ErrVersionMismatch for a stale or missing result; callers own version arithmetic. + Update(ctx context.Context, summary entity.RequestIssueSummary, oldVersion, newVersion int32) error +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index b69297ae9..933e35e75 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 + // GetRequestIssueSummaryStore returns immutable per-request issue results. + GetRequestIssueSummaryStore() RequestIssueSummaryStore + // GetIssueStateStore returns the IssueStateStore instance. GetIssueStateStore() IssueStateStore diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index b0351076a..80833ee2f 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -17,6 +17,7 @@ go_test( "queue_state_transitions_test.go", "queue_summary_test.go", "request_acceptance_test.go", + "request_issue_summary_test.go", "storage_test.go", ], data = [ diff --git a/test/integration/stovepipe/extension/storage/mysql/request_issue_summary_test.go b/test/integration/stovepipe/extension/storage/mysql/request_issue_summary_test.go new file mode 100644 index 000000000..11e5015f5 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/request_issue_summary_test.go @@ -0,0 +1,72 @@ +// 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" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func testRequestIssueSummaryStore(t *testing.T, ctx context.Context, factory storage.Factory) { + bound, err := factory.For(storage.Config{QueueName: "request-issues"}) + require.NoError(t, err) + store := bound.GetRequestIssueSummaryStore() + for _, issue := range []entity.RequestIssueSummary{ + {RequestID: "12", TaskID: "test", UnitID: "z", Kind: entity.RequestIssueKindClosed, BuildID: "b/12", FirstSeenBuildID: "b/10"}, + {RequestID: "12", TaskID: "test", Kind: entity.RequestIssueKindNew, BuildID: "b/12", FirstSeenBuildID: "b/12"}, + {RequestID: "12", TaskID: "test", UnitID: "a", Kind: entity.RequestIssueKindExisting, BuildID: "b/12", FirstSeenBuildID: "b/10"}, + } { + require.NoError(t, store.Create(ctx, issue)) + } + first, err := store.List(ctx, "12", storage.RequestIssueSummaryRange{Limit: 2}) + require.NoError(t, err) + require.Len(t, first, 2) + require.Equal(t, "", first[0].UnitID) + require.Equal(t, "a", first[1].UnitID) + last, err := store.List(ctx, "12", storage.RequestIssueSummaryRange{After: storage.RequestIssueSummaryCursor{TaskID: "test", UnitID: "a"}, Limit: 2}) + require.NoError(t, err) + require.Len(t, last, 1) + require.Equal(t, "z", last[0].UnitID) + duplicate := first[0] + duplicate.Kind = entity.RequestIssueKindExisting + require.ErrorIs(t, store.Create(ctx, duplicate), storage.ErrAlreadyExists) + retained, err := store.Get(ctx, "12", "test", "") + require.NoError(t, err) + require.Equal(t, first[0], retained) + other, err := factory.For(storage.Config{QueueName: "other-request-issues"}) + require.NoError(t, err) + _, err = other.GetRequestIssueSummaryStore().Get(ctx, "12", "test", "") + require.ErrorIs(t, err, storage.ErrNotFound) + empty, err := store.List(ctx, "13", storage.RequestIssueSummaryRange{Limit: 2}) + require.NoError(t, err) + require.Empty(t, empty) + require.NoError(t, store.Create(ctx, entity.RequestIssueSummary{RequestID: "13", TaskID: "test", Kind: entity.RequestIssueKindExisting, BuildID: "b/13", FirstSeenBuildID: "b/12"})) + retained, err = store.Get(ctx, "12", "test", "") + require.NoError(t, err) + require.Equal(t, first[0], retained) + + closed := retained + closed.Kind = entity.RequestIssueKindClosed + require.NoError(t, store.Update(ctx, closed, 0, 1)) + require.ErrorIs(t, store.Update(ctx, retained, 0, 1), storage.ErrVersionMismatch) + got, err := store.Get(ctx, "12", "test", "") + require.NoError(t, err) + require.Equal(t, entity.RequestIssueKindClosed, got.Kind) + require.Equal(t, int32(1), got.Version) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index de593bd9c..3b3ab46c9 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -135,6 +135,10 @@ func TestMySQLStorage(t *testing.T) { testIssueTrackingStore(t, ctx, factory) }) + t.Run("RequestIssueSummaryStore", func(t *testing.T) { + testRequestIssueSummaryStore(t, ctx, factory) + }) + t.Run("IssueLogStore", func(t *testing.T) { resetStorage(t, db) testIssueLogStore(t, ctx, factory)