From 42c35c759e90ec4327233ad826ee80e0683fe10e Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Fri, 23 Oct 2026 20:44:57 +0000 Subject: [PATCH] feat(stovepipe): Store issue-tracking progress --- stovepipe/entity/BUILD.bazel | 2 + stovepipe/entity/issue_tracking_state.go | 86 ++++++++ stovepipe/entity/issue_tracking_state_test.go | 51 +++++ stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 10 +- .../extension/storage/issue_tracking_store.go | 34 +++ stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../storage/mock/issue_tracking_store_mock.go | 85 ++++++++ .../extension/storage/mock/storage_mock.go | 14 ++ stovepipe/extension/storage/mysql/BUILD.bazel | 2 + .../storage/mysql/issue_tracking_store.go | 111 ++++++++++ .../mysql/issue_tracking_store_test.go | 202 ++++++++++++++++++ .../storage/mysql/schema/issue_tracking.sql | 12 ++ stovepipe/extension/storage/mysql/storage.go | 6 + .../extension/storage/mysql/storage_test.go | 9 + stovepipe/extension/storage/storage.go | 3 + .../extension/storage/mysql/BUILD.bazel | 1 + .../storage/mysql/issue_tracking_test.go | 116 ++++++++++ .../extension/storage/mysql/storage_test.go | 6 + 19 files changed, 751 insertions(+), 1 deletion(-) create mode 100644 stovepipe/entity/issue_tracking_state.go create mode 100644 stovepipe/entity/issue_tracking_state_test.go create mode 100644 stovepipe/extension/storage/issue_tracking_store.go create mode 100644 stovepipe/extension/storage/mock/issue_tracking_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/issue_tracking_store.go create mode 100644 stovepipe/extension/storage/mysql/issue_tracking_store_test.go create mode 100644 stovepipe/extension/storage/mysql/schema/issue_tracking.sql create mode 100644 test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index fff333eb6..444eb0a73 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "build_issue.go", "ingest.go", "issue_log.go", + "issue_tracking_state.go", "list.go", "open_issue.go", "project_fact_by_degree.go", @@ -35,6 +36,7 @@ go_test( srcs = [ "build_test.go", "issue_log_test.go", + "issue_tracking_state_test.go", "project_fact_by_degree_test.go", "queue_state_test.go", "request_id_test.go", diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go new file mode 100644 index 000000000..789564f0f --- /dev/null +++ b/stovepipe/entity/issue_tracking_state.go @@ -0,0 +1,86 @@ +// 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" +) + +// IssueTrackingPhase describes whether one source build's issue updates are pending or complete. +type IssueTrackingPhase string + +const ( + // IssueTrackingPhaseUnknown is the unset phase. + IssueTrackingPhaseUnknown IssueTrackingPhase = "" + // IssueTrackingPhaseRefreshing means reported build issues are being opened or kept open. + IssueTrackingPhaseRefreshing IssueTrackingPhase = "refreshing" + // IssueTrackingPhaseRefreshed means a refresh pass completed; later refresh notifications remain eligible. + IssueTrackingPhaseRefreshed IssueTrackingPhase = "refreshed" + // IssueTrackingPhaseFinalizing means final opening and closing updates are pending. + IssueTrackingPhaseFinalizing IssueTrackingPhase = "finalizing" + // IssueTrackingPhaseFinalized means final reconciliation completed; later refresh notifications are ignored. + IssueTrackingPhaseFinalized IssueTrackingPhase = "finalized" +) + +// IssueTrackingState is the queue-scoped source-order boundary and resumable reconciliation state. +type IssueTrackingState struct { + // RequestID is the newest accepted source request, compared by ingest order. + RequestID string `json:"request_id"` + // BuildID identifies the source build whose pass is pending or most recently completed. + BuildID string `json:"build_id"` + // NotificationID identifies the message that owns this reconciliation pass. + NotificationID string `json:"notification_id"` + // AfterTaskID is the exclusive task cursor after fully processed issue keys. + // Empty with an empty AfterUnitID starts a pass. + AfterTaskID string `json:"after_task_id"` + // AfterUnitID is the unit component of the exclusive cursor; empty is task-wide. + AfterUnitID string `json:"after_unit_id"` + // Phase distinguishes pending work from completed refresh or final passes. + Phase IssueTrackingPhase `json:"phase"` + // Version is the positive optimistic-lock version of this state. + Version int32 `json:"version"` +} + +// IsPending reports whether the phase represents unfinished reconciliation. +func (s IssueTrackingState) IsPending() bool { + return s.Phase == IssueTrackingPhaseRefreshing || s.Phase == IssueTrackingPhaseFinalizing +} + +// Validate verifies the identity, phase, and version of a persisted state. +func (s IssueTrackingState) Validate() error { + if err := id.Validate(s.RequestID); err != nil { + return fmt.Errorf("invalid issue tracking request ID: %w", err) + } + if s.BuildID == "" { + return fmt.Errorf("issue tracking build ID must not be empty") + } + if s.NotificationID == "" { + return fmt.Errorf("issue tracking notification ID must not be empty") + } + if s.AfterTaskID == "" && s.AfterUnitID != "" { + return fmt.Errorf("issue tracking cursor unit requires a task") + } + if s.Version <= 0 { + return fmt.Errorf("issue tracking version must be positive") + } + switch s.Phase { + case IssueTrackingPhaseRefreshing, IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalizing, IssueTrackingPhaseFinalized: + return nil + default: + return fmt.Errorf("invalid issue tracking phase %q", s.Phase) + } +} diff --git a/stovepipe/entity/issue_tracking_state_test.go b/stovepipe/entity/issue_tracking_state_test.go new file mode 100644 index 000000000..d0287d57b --- /dev/null +++ b/stovepipe/entity/issue_tracking_state_test.go @@ -0,0 +1,51 @@ +// 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 TestIssueTrackingState(t *testing.T) { + for _, phase := range []IssueTrackingPhase{IssueTrackingPhaseRefreshing, IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalizing, IssueTrackingPhaseFinalized} { + t.Run(string(phase), func(t *testing.T) { + state := IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: phase, Version: 1} + require.NoError(t, state.Validate()) + require.Equal(t, phase == IssueTrackingPhaseRefreshing || phase == IssueTrackingPhaseFinalizing, state.IsPending()) + }) + } + for _, tt := range []struct { + name string + change func(*IssueTrackingState) + }{ + {"missing request", func(s *IssueTrackingState) { s.RequestID = "" }}, + {"noncanonical request", func(s *IssueTrackingState) { s.RequestID = "010" }}, + {"nonnumeric request", func(s *IssueTrackingState) { s.RequestID = "request/10" }}, + {"missing notification", func(s *IssueTrackingState) { s.NotificationID = "" }}, + {"unit cursor without task", func(s *IssueTrackingState) { s.AfterUnitID = "unit" }}, + {"missing build", func(s *IssueTrackingState) { s.BuildID = "" }}, + {"unknown phase", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseUnknown }}, + {"unsupported phase", func(s *IssueTrackingState) { s.Phase = "other" }}, + {"zero version", func(s *IssueTrackingState) { s.Version = 0 }}, + } { + t.Run(tt.name, func(t *testing.T) { + state := IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: IssueTrackingPhaseRefreshing, Version: 1} + tt.change(&state) + require.Error(t, state.Validate()) + }) + } +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index e0e32cf7d..c924bd1c5 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build_issue_store.go", "build_store.go", "issue_log_store.go", + "issue_tracking_store.go", "open_issue_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 18b191e5a..a0a4b450f 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`, `OpenIssueStore`, `IssueLogStore`). Implementations live under `extension/storage//`. +Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`, `OpenIssueStore`, `IssueLogStore`, `IssueTrackingStore`). 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. @@ -59,3 +59,11 @@ The build ID identifies the build where the issue was first seen or confirmed re The MySQL key uses the same binary identifier limits as the issue stores, plus a 32-byte event field, for a maximum of 2586 bytes. Reads and creates use primary-key operations; there is no update or deletion API, metadata collection, or repair state. History persistence does not alter open-issue membership or validate transitions against it. No runtime writer is enabled here. A later reconciliation controller must own opening/resolution decisions, retry-safe writes to both stores, and protection against stale build analysis. History enumeration is outside this storage contract. + +## Issue reconciliation progress + +`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, whether refresh or final writes are pending or complete, the owning notification, an exclusive task/unit cursor, and its optimistic version. The zero cursor starts a pass. A cursor checkpoint follows the writes for every processed key; notification identity lets the controller distinguish its own continuation from helping another pass. Keeping it after the last open row is removed preserves the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. + +The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of pending writes, and completion; no runtime writer is enabled here. + +The issue-tracking table must include the notification and cursor columns before updated readers or writers are deployed. No runtime writer is enabled by the storage layer. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores. diff --git a/stovepipe/extension/storage/issue_tracking_store.go b/stovepipe/extension/storage/issue_tracking_store.go new file mode 100644 index 000000000..2654bc041 --- /dev/null +++ b/stovepipe/extension/storage/issue_tracking_store.go @@ -0,0 +1,34 @@ +// 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_tracking_store.go -destination=mock/issue_tracking_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// IssueTrackingStore retains one operational reconciliation state for the bound queue. +// It remains present when no issues are open so older results cannot reopen closed issues. +type IssueTrackingStore interface { + // Get returns ErrNotFound before the queue's first reconciliation is accepted. + Get(context.Context) (entity.IssueTrackingState, error) + // Create returns ErrAlreadyExists when the queue already has a state. + Create(context.Context, entity.IssueTrackingState) error + // Update replaces the state only at oldVersion; callers supply newVersion without store arithmetic. + Update(ctx context.Context, state entity.IssueTrackingState, oldVersion, newVersion int32) error +} diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index a1187266f..54499c25f 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build_issue_store_mock.go", "build_store_mock.go", "issue_log_store_mock.go", + "issue_tracking_store_mock.go", "open_issue_store_mock.go", "project_fact_by_degree_store_mock.go", "queue_log_store_mock.go", diff --git a/stovepipe/extension/storage/mock/issue_tracking_store_mock.go b/stovepipe/extension/storage/mock/issue_tracking_store_mock.go new file mode 100644 index 000000000..5a79df4e3 --- /dev/null +++ b/stovepipe/extension/storage/mock/issue_tracking_store_mock.go @@ -0,0 +1,85 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: issue_tracking_store.go +// +// Generated by this command: +// +// mockgen -source=issue_tracking_store.go -destination=mock/issue_tracking_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" +) + +// MockIssueTrackingStore is a mock of IssueTrackingStore interface. +type MockIssueTrackingStore struct { + ctrl *gomock.Controller + recorder *MockIssueTrackingStoreMockRecorder + isgomock struct{} +} + +// MockIssueTrackingStoreMockRecorder is the mock recorder for MockIssueTrackingStore. +type MockIssueTrackingStoreMockRecorder struct { + mock *MockIssueTrackingStore +} + +// NewMockIssueTrackingStore creates a new mock instance. +func NewMockIssueTrackingStore(ctrl *gomock.Controller) *MockIssueTrackingStore { + mock := &MockIssueTrackingStore{ctrl: ctrl} + mock.recorder = &MockIssueTrackingStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockIssueTrackingStore) EXPECT() *MockIssueTrackingStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockIssueTrackingStore) Create(arg0 context.Context, arg1 entity.IssueTrackingState) 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 *MockIssueTrackingStoreMockRecorder) Create(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockIssueTrackingStore)(nil).Create), arg0, arg1) +} + +// Get mocks base method. +func (m *MockIssueTrackingStore) Get(arg0 context.Context) (entity.IssueTrackingState, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", arg0) + ret0, _ := ret[0].(entity.IssueTrackingState) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockIssueTrackingStoreMockRecorder) Get(arg0 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockIssueTrackingStore)(nil).Get), arg0) +} + +// Update mocks base method. +func (m *MockIssueTrackingStore) Update(ctx context.Context, state entity.IssueTrackingState, oldVersion, newVersion int32) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Update", ctx, state, oldVersion, newVersion) + ret0, _ := ret[0].(error) + return ret0 +} + +// Update indicates an expected call of Update. +func (mr *MockIssueTrackingStoreMockRecorder) Update(ctx, state, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockIssueTrackingStore)(nil).Update), ctx, state, oldVersion, newVersion) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index 07353c796..7b1e2c61d 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -121,6 +121,20 @@ func (mr *MockStorageMockRecorder) GetIssueLogStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueLogStore", reflect.TypeOf((*MockStorage)(nil).GetIssueLogStore)) } +// GetIssueTrackingStore mocks base method. +func (m *MockStorage) GetIssueTrackingStore() storage.IssueTrackingStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetIssueTrackingStore") + ret0, _ := ret[0].(storage.IssueTrackingStore) + return ret0 +} + +// GetIssueTrackingStore indicates an expected call of GetIssueTrackingStore. +func (mr *MockStorageMockRecorder) GetIssueTrackingStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueTrackingStore", reflect.TypeOf((*MockStorage)(nil).GetIssueTrackingStore)) +} + // GetOpenIssueStore mocks base method. func (m *MockStorage) GetOpenIssueStore() storage.OpenIssueStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 1f3b25ea2..c96caee34 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build_issue_store.go", "build_store.go", "issue_log_store.go", + "issue_tracking_store.go", "open_issue_store.go", "project_fact_by_degree_store.go", "queue_log_store.go", @@ -37,6 +38,7 @@ go_test( "build_issue_store_test.go", "build_store_test.go", "issue_log_store_test.go", + "issue_tracking_store_test.go", "open_issue_list_test.go", "open_issue_store_test.go", "project_fact_by_degree_store_test.go", diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store.go b/stovepipe/extension/storage/mysql/issue_tracking_store.go new file mode 100644 index 000000000..fd3ec329e --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_tracking_store.go @@ -0,0 +1,111 @@ +// 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 issueTrackingStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewIssueTrackingStore binds operational reconciliation progress to one queue. +func NewIssueTrackingStore(db *sql.DB, scope tally.Scope, queue string) storage.IssueTrackingStore { + return &issueTrackingStore{db: db, scope: scope, queue: queue} +} + +func (s *issueTrackingStore) Get(ctx context.Context) (state entity.IssueTrackingState, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + err := s.db.QueryRowContext(ctx, + `SELECT request_id, build_id, notification_id, after_task_id, after_unit_id, phase, version FROM issue_tracking WHERE queue = ?`, s.queue). + Scan(&state.RequestID, &state.BuildID, &state.NotificationID, &state.AfterTaskID, &state.AfterUnitID, &state.Phase, &state.Version) + if errors.Is(err, sql.ErrNoRows) { + return entity.IssueTrackingState{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("read issue tracking state: %w", err) + } + return state, nil +} + +func (s *issueTrackingStore) Create(ctx context.Context, state entity.IssueTrackingState) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + if err := s.validateState(state); err != nil { + return err + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, after_task_id, after_unit_id, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + s.queue, state.RequestID, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, state.Version) + if isDuplicateEntry(err) { + return storage.ErrAlreadyExists + } + if err != nil { + return fmt.Errorf("create issue tracking state: %w", err) + } + return nil +} + +func (s *issueTrackingStore) Update(ctx context.Context, state entity.IssueTrackingState, oldVersion, newVersion int32) (retErr error) { + op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + replacement := state + replacement.Version = newVersion + if err := s.validateState(replacement); err != nil { + return err + } + if oldVersion <= 0 { + return fmt.Errorf("issue tracking guard version must be positive") + } + result, err := s.db.ExecContext(ctx, + `UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, after_task_id = ?, after_unit_id = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`, + state.RequestID, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, newVersion, s.queue, oldVersion) + if err != nil { + return fmt.Errorf("update issue tracking state: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("read issue tracking update count: %w", err) + } + if rows != 1 { + return storage.ErrVersionMismatch + } + return nil +} + +func (s *issueTrackingStore) validateState(state entity.IssueTrackingState) error { + if err := state.Validate(); err != nil { + return err + } + // The shared key validator below covers queue, build, task, and unit lengths. + if len(state.NotificationID) > 1020 { + return fmt.Errorf("issue tracking notification ID exceeds 1020-byte limit") + } + return validateBuildIssueKey(s.queue, state.BuildID, state.AfterTaskID, state.AfterUnitID) +} + +var _ storage.IssueTrackingStore = (*issueTrackingStore)(nil) diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go new file mode 100644 index 000000000..3d7e03825 --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go @@ -0,0 +1,202 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "context" + "database/sql" + "fmt" + "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 trackingFixture() entity.IssueTrackingState { + return entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} +} + +func TestIssueTrackingStoreGet(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() + + state := trackingFixture() + store := NewIssueTrackingStore(db, testMetrics(), "queue") + query := mock.ExpectQuery(regexp.QuoteMeta(`SELECT request_id, build_id, notification_id, after_task_id, after_unit_id, phase, version FROM issue_tracking WHERE queue = ?`)).WithArgs("queue") + switch kind { + case "missing": + query.WillReturnError(sql.ErrNoRows) + case "database error": + query.WillReturnError(assert.AnError) + default: + var request any = state.RequestID + if kind == "scan error" { + request = nil + } + query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "after_task_id", "after_unit_id", "phase", "version"}).AddRow(request, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, state.Version)) + } + + got, err := store.Get(context.Background()) + switch kind { + case "found": + require.NoError(t, err) + require.Equal(t, state, 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" { + require.Equal(t, entity.IssueTrackingState{}, got) + } + + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueTrackingStoreCreate(t *testing.T) { + for _, tt := range []struct { + name string + dbErr, errorIs error + }{ + {name: "created"}, + {name: "duplicate", dbErr: &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, errorIs: storage.ErrAlreadyExists}, + {name: "database error", dbErr: assert.AnError, errorIs: assert.AnError}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + + state := trackingFixture() + store := NewIssueTrackingStore(db, testMetrics(), "queue") + query := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, after_task_id, after_unit_id, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`)). + WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, state.Version) + if tt.dbErr != nil { + query.WillReturnError(tt.dbErr) + } else { + query.WillReturnResult(sqlmock.NewResult(0, 1)) + } + + err = store.Create(context.Background(), state) + if tt.errorIs != nil { + require.ErrorIs(t, err, tt.errorIs) + } else { + require.NoError(t, err) + } + + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueTrackingStoreUpdate(t *testing.T) { + for _, tt := range []struct { + name string + rows int64 + dbErr, errorIs error + resultError bool + }{ + {name: "updated", rows: 1}, + {name: "stale guard", errorIs: storage.ErrVersionMismatch}, + {name: "database error", dbErr: assert.AnError, errorIs: assert.AnError}, + {name: "affected rows error", resultError: true, errorIs: assert.AnError}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + + state := trackingFixture() + store := NewIssueTrackingStore(db, testMetrics(), "queue") + query := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, after_task_id = ?, after_unit_id = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`)). + WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, int32(7), "queue", int32(1)) + switch { + case tt.dbErr != nil: + query.WillReturnError(tt.dbErr) + case tt.resultError: + query.WillReturnResult(sqlmock.NewErrorResult(assert.AnError)) + default: + query.WillReturnResult(sqlmock.NewResult(0, tt.rows)) + } + + err = store.Update(context.Background(), state, 1, 7) + if tt.errorIs != nil { + require.ErrorIs(t, err, tt.errorIs) + } else { + require.NoError(t, err) + } + require.Equal(t, int32(1), state.Version) + + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueTrackingStoreRejectsInvalidState(t *testing.T) { + for _, tt := range []struct { + name string + change func(*entity.IssueTrackingState, *string) + }{ + {"invalid request", func(s *entity.IssueTrackingState, _ *string) { s.RequestID = "invalid" }}, + {"empty notification", func(s *entity.IssueTrackingState, _ *string) { s.NotificationID = "" }}, + {"oversized notification", func(s *entity.IssueTrackingState, _ *string) { s.NotificationID = strings.Repeat("n", 1021) }}, + {"oversized task cursor", func(s *entity.IssueTrackingState, _ *string) { s.AfterTaskID = strings.Repeat("t", 256) }}, + {"oversized unit cursor", func(s *entity.IssueTrackingState, _ *string) { + s.AfterTaskID = "task" + s.AfterUnitID = strings.Repeat("u", 1025) + }}, + {"empty build", func(s *entity.IssueTrackingState, _ *string) { s.BuildID = "" }}, + {"oversized build", func(s *entity.IssueTrackingState, _ *string) { s.BuildID = strings.Repeat("b", 1021) }}, + {"unknown phase", func(s *entity.IssueTrackingState, _ *string) { s.Phase = entity.IssueTrackingPhaseUnknown }}, + {"oversized queue", func(_ *entity.IssueTrackingState, q *string) { *q = strings.Repeat("q", 256) }}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + state, queue := trackingFixture(), "queue" + tt.change(&state, &queue) + store := NewIssueTrackingStore(db, testMetrics(), queue) + mock.MatchExpectationsInOrder(false) + mock.ExpectExec("INSERT").WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("UPDATE").WillReturnResult(sqlmock.NewResult(0, 1)) + require.Error(t, store.Create(context.Background(), state)) + require.Error(t, store.Update(context.Background(), state, 1, 2)) + }) + } + for _, versions := range [][2]int32{{0, 1}, {1, 0}, {1, -1}} { + t.Run(fmt.Sprintf("versions %d/%d", versions[0], versions[1]), func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewIssueTrackingStore(db, testMetrics(), "queue") + mock.ExpectExec("UPDATE").WillReturnResult(sqlmock.NewResult(0, 1)) + require.Error(t, store.Update(context.Background(), trackingFixture(), versions[0], versions[1])) + }) + } +} diff --git a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql new file mode 100644 index 000000000..565630ddc --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql @@ -0,0 +1,12 @@ +-- Persistent source-order boundary; pending passes are resumed before admitting another source. +CREATE TABLE IF NOT EXISTS issue_tracking ( + queue VARBINARY(255) NOT NULL, + request_id VARBINARY(255) NOT NULL, + build_id VARBINARY(1020) NOT NULL, + notification_id VARBINARY(1020) NOT NULL, + after_task_id VARBINARY(255) NOT NULL, + after_unit_id VARBINARY(1024) NOT NULL, + phase VARBINARY(32) NOT NULL, + version INT NOT NULL, + PRIMARY KEY (queue) +) ENGINE=InnoDB; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index 25f0643fb..4a00fa56c 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), openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_store"), queueName), issueLogStore: NewIssueLogStore(s.db, queueScope.SubScope("issue_log_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), }, nil @@ -82,6 +83,7 @@ type mysqlStorage struct { buildIssueStore storage.BuildIssueStore openIssueStore storage.OpenIssueStore issueLogStore storage.IssueLogStore + issueTrackingStore storage.IssueTrackingStore validationFactStore storage.ValidationFactStore projectFactByDegreeStore storage.ProjectFactByDegreeStore } @@ -150,3 +152,7 @@ func (f *mysqlStorage) GetOpenIssueStore() storage.OpenIssueStore { func (f *mysqlStorage) GetIssueLogStore() storage.IssueLogStore { return f.issueLogStore } + +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 d5cf12bdd..291bf9697 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.GetOpenIssueStore()) assert.NotNil(t, bound.GetIssueLogStore()) + assert.NotNil(t, bound.GetIssueTrackingStore()) _, err = s.For("") assert.Error(t, err, "resolving an empty queue name must fail") @@ -155,6 +156,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "issue_tracking_store", + op: "get", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetIssueTrackingStore().Get(context.Background()) + return err + }, + }, { name: "validation_fact_store", op: "get", diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 08fdc281a..90d4d55a5 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 + // GetIssueTrackingStore returns the operational reconciliation progress store. + GetIssueTrackingStore() IssueTrackingStore + // 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 0ba8776b6..ba721c65f 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -6,6 +6,7 @@ go_test( "build_issue_list_test.go", "build_issue_test.go", "issue_log_test.go", + "issue_tracking_test.go", "open_issue_list_test.go", "open_issue_test.go", "project_fact_by_degree_test.go", diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go new file mode 100644 index 000000000..3facd0805 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go @@ -0,0 +1,116 @@ +// 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 testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.Factory) { + t.Helper() + bound, err := factory.For(storage.Config{QueueName: "tracking/main"}) + require.NoError(t, err) + store := bound.GetIssueTrackingStore() + _, err = store.Get(ctx) + require.ErrorIs(t, err, storage.ErrNotFound) + state := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} + require.NoError(t, store.Create(ctx, state)) + require.ErrorIs(t, store.Create(ctx, state), storage.ErrAlreadyExists) + got, err := store.Get(ctx) + require.NoError(t, err) + require.Equal(t, state, got) + + state.AfterTaskID, state.AfterUnitID = "task", "unit" + state.Phase = entity.IssueTrackingPhaseRefreshed + require.NoError(t, store.Update(ctx, state, 1, 2)) + state.Version = 2 + require.ErrorIs(t, store.Update(ctx, state, 1, 3), storage.ErrVersionMismatch) + got, err = store.Get(ctx) + require.NoError(t, err) + require.Equal(t, state, got) + + state.RequestID, state.BuildID, state.Phase = "11", "build/11", entity.IssueTrackingPhaseFinalizing + require.NoError(t, store.Update(ctx, state, 2, 3)) + state.Version = 3 + state.Phase = entity.IssueTrackingPhaseFinalized + require.NoError(t, store.Update(ctx, state, 3, 4)) + state.Version = 4 + got, err = store.Get(ctx) + require.NoError(t, err) + require.Equal(t, state, got) + opens, err := bound.GetOpenIssueStore().List(ctx, storage.OpenIssueRange{Limit: 1}) + require.NoError(t, err) + require.Empty(t, opens) + + t.Run("queue isolation", func(t *testing.T) { + other, err := factory.For(storage.Config{QueueName: "Tracking/main"}) + require.NoError(t, err) + _, err = other.GetIssueTrackingStore().Get(ctx) + require.ErrorIs(t, err, storage.ErrNotFound) + initial := entity.IssueTrackingState{RequestID: "1", BuildID: "other", NotificationID: "other", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} + require.NoError(t, other.GetIssueTrackingStore().Create(ctx, initial)) + got, err := store.Get(ctx) + require.NoError(t, err) + require.Equal(t, state, got) + }) + + t.Run("one conditional update wins", func(t *testing.T) { + results := make(chan error, 2) + for _, requestID := range []string{"12", "13"} { + next := state + next.RequestID = requestID + next.Phase = entity.IssueTrackingPhaseRefreshing + go func() { results <- store.Update(ctx, next, 4, 5) }() + } + successes, conflicts := 0, 0 + for i := 0; i < 2; i++ { + 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) + got, err := store.Get(ctx) + require.NoError(t, err) + require.Equal(t, int32(5), got.Version) + require.Contains(t, []string{"12", "13"}, got.RequestID) + }) + + t.Run("identifier limits", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) + require.NoError(t, err) + store := bound.GetIssueTrackingStore() + value := entity.IssueTrackingState{RequestID: "9223372036854775807", BuildID: strings.Repeat("😀", 255), NotificationID: strings.Repeat("😀", 255), AfterTaskID: strings.Repeat("t", 255), AfterUnitID: strings.Repeat("u", 1024), Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} + require.NoError(t, store.Create(ctx, value)) + got, err := store.Get(ctx) + require.NoError(t, err) + require.Equal(t, value, got) + value.BuildID += "x" + 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 e3b5f97e1..9a6f89562 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) { testOpenIssueList(t, ctx, factory) }) + t.Run("IssueTrackingStore", func(t *testing.T) { + resetStorage(t, db) + testIssueTrackingStore(t, ctx, factory) + }) + t.Run("IssueLogStore", func(t *testing.T) { resetStorage(t, db) testIssueLogStore(t, ctx, factory) @@ -175,6 +180,7 @@ func resetStorage(t *testing.T, db *sql.DB) { for _, statement := range []string{ "TRUNCATE TABLE open_issue", "TRUNCATE TABLE issue_log", + "TRUNCATE TABLE issue_tracking", "TRUNCATE TABLE queue_log", "TRUNCATE TABLE queue_summary", "TRUNCATE TABLE request_log",