From 78f226cc4cd9576fd67289c8ee4cbdc713b5842d Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Thu, 22 Oct 2026 21:23:26 +0000 Subject: [PATCH] feat(stovepipe): Add issue lifecycle history --- stovepipe/entity/BUILD.bazel | 2 + stovepipe/entity/issue_log.go | 60 ++++++ stovepipe/entity/issue_log_test.go | 52 +++++ stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 14 +- .../extension/storage/issue_log_store.go | 34 ++++ stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../storage/mock/issue_log_store_mock.go | 71 +++++++ .../extension/storage/mock/storage_mock.go | 14 ++ stovepipe/extension/storage/mysql/BUILD.bazel | 2 + .../storage/mysql/issue_log_store.go | 106 ++++++++++ .../storage/mysql/issue_log_store_test.go | 175 +++++++++++++++++ .../storage/mysql/schema/issue_log.sql | 12 ++ stovepipe/extension/storage/mysql/storage.go | 6 + .../extension/storage/mysql/storage_test.go | 10 + stovepipe/extension/storage/storage.go | 3 + .../extension/storage/mysql/BUILD.bazel | 1 + .../extension/storage/mysql/issue_log_test.go | 181 ++++++++++++++++++ .../extension/storage/mysql/storage_test.go | 6 + 19 files changed, 749 insertions(+), 2 deletions(-) create mode 100644 stovepipe/entity/issue_log.go create mode 100644 stovepipe/entity/issue_log_test.go create mode 100644 stovepipe/extension/storage/issue_log_store.go create mode 100644 stovepipe/extension/storage/mock/issue_log_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/issue_log_store.go create mode 100644 stovepipe/extension/storage/mysql/issue_log_store_test.go create mode 100644 stovepipe/extension/storage/mysql/schema/issue_log.sql create mode 100644 test/integration/stovepipe/extension/storage/mysql/issue_log_test.go diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 16d1bbfd1..326d11ab1 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build.go", "build_issue.go", "ingest.go", + "issue_log.go", "list.go", "open_issue.go", "project_status.go", @@ -32,6 +33,7 @@ go_test( name = "go_default_test", srcs = [ "build_test.go", + "issue_log_test.go", "queue_state_test.go", "request_id_test.go", "request_log_test.go", diff --git a/stovepipe/entity/issue_log.go b/stovepipe/entity/issue_log.go new file mode 100644 index 000000000..99619eac9 --- /dev/null +++ b/stovepipe/entity/issue_log.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 "fmt" + +// IssueEvent identifies an issue lifecycle transition. +type IssueEvent string + +const ( + // IssueEventUnknown is the unset event value. + IssueEventUnknown IssueEvent = "" + // IssueEventOpened records that an issue became open. + IssueEventOpened IssueEvent = "opened" + // IssueEventResolved records that an issue was resolved. + IssueEventResolved IssueEvent = "resolved" +) + +// IssueLog is an immutable issue lifecycle transition within a queue. +// Task, unit, build, and event identify one transition. +type IssueLog struct { + // TaskID identifies the logical validation task. + TaskID string `json:"task_id"` + // UnitID identifies the validation unit; empty for a task-wide issue. + UnitID string `json:"unit_id"` + // Event identifies the opening or resolution. + Event IssueEvent `json:"event"` + // BuildID identifies the build where the issue was first seen or confirmed resolved. + BuildID string `json:"build_id"` + // TimestampMs is when the transition was first recorded, in Unix milliseconds. + TimestampMs int64 `json:"timestamp_ms"` +} + +// Validate verifies the invariants required for a newly persisted issue log. +func (l IssueLog) Validate() error { + if l.BuildID == "" { + return fmt.Errorf("issue log build ID must not be empty") + } + if l.TimestampMs <= 0 { + return fmt.Errorf("issue log timestamp must be positive") + } + switch l.Event { + case IssueEventOpened, IssueEventResolved: + return nil + default: + return fmt.Errorf("invalid issue event %q", l.Event) + } +} diff --git a/stovepipe/entity/issue_log_test.go b/stovepipe/entity/issue_log_test.go new file mode 100644 index 000000000..aa6387149 --- /dev/null +++ b/stovepipe/entity/issue_log_test.go @@ -0,0 +1,52 @@ +// 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 TestIssueLogValidate(t *testing.T) { + for _, tt := range []struct { + name string + change func(*IssueLog) + invalid bool + }{ + {name: "opened"}, + {name: "resolved", change: func(l *IssueLog) { l.Event = IssueEventResolved }}, + {name: "task-wide issue", change: func(l *IssueLog) { l.UnitID = "" }}, + {name: "missing build", change: func(l *IssueLog) { l.BuildID = "" }, invalid: true}, + {name: "zero timestamp", change: func(l *IssueLog) { l.TimestampMs = 0 }, invalid: true}, + {name: "negative timestamp", change: func(l *IssueLog) { l.TimestampMs = -1 }, invalid: true}, + {name: "unknown event", change: func(l *IssueLog) { l.Event = IssueEventUnknown }, invalid: true}, + {name: "unsupported event", change: func(l *IssueLog) { l.Event = "other" }, invalid: true}, + } { + t.Run(tt.name, func(t *testing.T) { + log := IssueLog{TaskID: "task", UnitID: "unit", + Event: IssueEventOpened, BuildID: "build/1", TimestampMs: 1000} + if tt.change != nil { + tt.change(&log) + } + + if tt.invalid { + require.Error(t, log.Validate()) + } else { + require.NoError(t, log.Validate()) + } + }) + } +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index a90b46c32..cdfaa32a7 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_issue_store.go", "build_store.go", + "issue_log_store.go", "open_issue_store.go", "queue_log_store.go", "queue_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 5e9072477..9f21474dd 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`). Implementations live under `extension/storage//`. +Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`, `OpenIssueStore`, `IssueLogStore`). 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. @@ -40,4 +40,14 @@ An open row is immutable. Its first-seen build identifies the current opening, n Deletion matches the task, unit, and first-seen build within the bound queue. It is safe to repeat: an absent opening is a no-op, and a stale deletion never removes a different opening of the same condition. Database failures are returned for retry. -MySQL identifiers use the same binary byte limits as per-build observations. Build IDs are opaque context, not ordering tokens. The store does not choose resolution, enumerate issues, write lifecycle history, or reconcile analysis responses. History and source-order reconciliation belong to later controllers, not this membership store. +MySQL identifiers use the same binary byte limits as per-build observations. Build IDs are opaque context, not ordering tokens. The store does not choose resolution, enumerate issues, write lifecycle history, or reconcile analysis responses. Lifecycle occurrences are retained separately by `IssueLogStore`. Source-order reconciliation belongs to later controllers, not this membership store. + +## Issue lifecycle history + +`IssueLogStore` retains immutable opened and resolved transitions independently of current open-issue membership and per-build observations. Each transition is keyed by task, unit, build, and event within the bound queue. An empty unit identifies a task-wide issue. History remains after an open row is removed or recreated; transitions do not carry a separate reference to the opening they belong to. + +The build ID identifies the build where the issue was first seen or confirmed resolved. The application supplies a positive Unix-millisecond timestamp for when the transition was first recorded, not when the build finished. A duplicate identity returns `ErrAlreadyExists` without replacing the original timestamp. Callers can treat replay as success or read the original event. Build IDs are opaque and do not establish transition ordering. + +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. diff --git a/stovepipe/extension/storage/issue_log_store.go b/stovepipe/extension/storage/issue_log_store.go new file mode 100644 index 000000000..9ea03eafe --- /dev/null +++ b/stovepipe/extension/storage/issue_log_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_log_store.go -destination=mock/issue_log_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// IssueLogStore retains immutable issue transitions within the bound queue. +// Task, unit, build, and event identify one transition. +type IssueLogStore interface { + // Create returns ErrAlreadyExists for a previously recorded transition. + // A duplicate never replaces the original timestamp. + Create(ctx context.Context, log entity.IssueLog) error + + // Get returns ErrNotFound when the transition has not been recorded. + Get(ctx context.Context, taskID, unitID, buildID string, event entity.IssueEvent) (entity.IssueLog, error) +} diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index e43236134..b0ff99808 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_issue_store_mock.go", "build_store_mock.go", + "issue_log_store_mock.go", "open_issue_store_mock.go", "queue_log_store_mock.go", "queue_store_mock.go", diff --git a/stovepipe/extension/storage/mock/issue_log_store_mock.go b/stovepipe/extension/storage/mock/issue_log_store_mock.go new file mode 100644 index 000000000..54453faba --- /dev/null +++ b/stovepipe/extension/storage/mock/issue_log_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: issue_log_store.go +// +// Generated by this command: +// +// mockgen -source=issue_log_store.go -destination=mock/issue_log_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" +) + +// MockIssueLogStore is a mock of IssueLogStore interface. +type MockIssueLogStore struct { + ctrl *gomock.Controller + recorder *MockIssueLogStoreMockRecorder + isgomock struct{} +} + +// MockIssueLogStoreMockRecorder is the mock recorder for MockIssueLogStore. +type MockIssueLogStoreMockRecorder struct { + mock *MockIssueLogStore +} + +// NewMockIssueLogStore creates a new mock instance. +func NewMockIssueLogStore(ctrl *gomock.Controller) *MockIssueLogStore { + mock := &MockIssueLogStore{ctrl: ctrl} + mock.recorder = &MockIssueLogStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockIssueLogStore) EXPECT() *MockIssueLogStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockIssueLogStore) Create(ctx context.Context, log entity.IssueLog) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, log) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockIssueLogStoreMockRecorder) Create(ctx, log any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockIssueLogStore)(nil).Create), ctx, log) +} + +// Get mocks base method. +func (m *MockIssueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, event entity.IssueEvent) (entity.IssueLog, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID, buildID, event) + ret0, _ := ret[0].(entity.IssueLog) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockIssueLogStoreMockRecorder) Get(ctx, taskID, unitID, buildID, event any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockIssueLogStore)(nil).Get), ctx, taskID, unitID, buildID, event) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index 4158fe1ca..c35b5584c 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -107,6 +107,20 @@ func (mr *MockStorageMockRecorder) GetBuildStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBuildStore", reflect.TypeOf((*MockStorage)(nil).GetBuildStore)) } +// GetIssueLogStore mocks base method. +func (m *MockStorage) GetIssueLogStore() storage.IssueLogStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetIssueLogStore") + ret0, _ := ret[0].(storage.IssueLogStore) + return ret0 +} + +// GetIssueLogStore indicates an expected call of GetIssueLogStore. +func (mr *MockStorageMockRecorder) GetIssueLogStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueLogStore", reflect.TypeOf((*MockStorage)(nil).GetIssueLogStore)) +} + // 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 c3adf368f..fcfb72818 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_issue_store.go", "build_store.go", + "issue_log_store.go", "open_issue_store.go", "queue_log_store.go", "queue_store.go", @@ -33,6 +34,7 @@ go_test( srcs = [ "build_issue_store_test.go", "build_store_test.go", + "issue_log_store_test.go", "open_issue_store_test.go", "queue_store_test.go", "queue_summary_store_test.go", diff --git a/stovepipe/extension/storage/mysql/issue_log_store.go b/stovepipe/extension/storage/mysql/issue_log_store.go new file mode 100644 index 000000000..8ef1d1089 --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_log_store.go @@ -0,0 +1,106 @@ +// 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 issueLogStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewIssueLogStore binds issue lifecycle history to queue over the shared database. +func NewIssueLogStore(db *sql.DB, scope tally.Scope, queue string) storage.IssueLogStore { + return &issueLogStore{db: db, scope: scope, queue: queue} +} + +func (s *issueLogStore) Create(ctx context.Context, log entity.IssueLog) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := log.Validate(); err != nil { + return err + } + if err := s.validateKey(log.TaskID, log.UnitID, log.BuildID, log.Event); err != nil { + return err + } + + _, err := s.db.ExecContext(ctx, + `INSERT INTO issue_log (queue, task_id, unit_id, build_id, event, timestamp_ms) + VALUES (?, ?, ?, ?, ?, ?)`, + s.queue, log.TaskID, log.UnitID, log.BuildID, log.Event, log.TimestampMs) + if err != nil { + if isDuplicateEntry(err) { + return fmt.Errorf("issue log task=%q unit=%q build=%q event=%q: %w", + log.TaskID, log.UnitID, log.BuildID, log.Event, storage.ErrAlreadyExists) + } + + return fmt.Errorf("failed to create issue log: %w", err) + } + + return nil +} + +func (s *issueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, event entity.IssueEvent) (ret entity.IssueLog, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := s.validateKey(taskID, unitID, buildID, event); err != nil { + return entity.IssueLog{}, err + } + + var log entity.IssueLog + err := s.db.QueryRowContext(ctx, + `SELECT task_id, unit_id, build_id, event, timestamp_ms + FROM issue_log WHERE queue = ? AND task_id = ? AND unit_id = ? AND build_id = ? AND event = ?`, + s.queue, taskID, unitID, buildID, event, + ).Scan(&log.TaskID, &log.UnitID, &log.BuildID, &log.Event, &log.TimestampMs) + if errors.Is(err, sql.ErrNoRows) { + return entity.IssueLog{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.IssueLog{}, fmt.Errorf("failed to get issue log: %w", err) + } + + return log, nil +} + +func (s *issueLogStore) validateKey(taskID, unitID, buildID string, event entity.IssueEvent) error { + if err := validateBuildIssueKey(s.queue, buildID, taskID, unitID); err != nil { + return err + } + if buildID == "" { + return fmt.Errorf("issue log build ID must not be empty") + } + switch event { + case entity.IssueEventOpened, entity.IssueEventResolved: + return nil + default: + return fmt.Errorf("invalid issue event %q", event) + } +} + +var _ storage.IssueLogStore = (*issueLogStore)(nil) diff --git a/stovepipe/extension/storage/mysql/issue_log_store_test.go b/stovepipe/extension/storage/mysql/issue_log_store_test.go new file mode 100644 index 000000000..ebeec0375 --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_log_store_test.go @@ -0,0 +1,175 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "context" + "database/sql" + "regexp" + "strings" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func issueLogFixture() entity.IssueLog { + return entity.IssueLog{TaskID: "task", UnitID: "unit", + Event: entity.IssueEventResolved, BuildID: "build/2", TimestampMs: 2000} +} + +func TestIssueLogCreate(t *testing.T) { + for _, tt := range []struct { + name string + dbErr error + wantErr error + }{ + {name: "created"}, + {name: "duplicate", dbErr: &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, wantErr: storage.ErrAlreadyExists}, + {name: "database error", dbErr: assert.AnError, wantErr: assert.AnError}, + {name: "cancelled", dbErr: context.Canceled, wantErr: context.Canceled}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewIssueLogStore(db, testMetrics(), "queue") + log := issueLogFixture() + query := mock.ExpectExec(regexp.QuoteMeta( + `INSERT INTO issue_log (queue, task_id, unit_id, build_id, event, timestamp_ms) + VALUES (?, ?, ?, ?, ?, ?)`)). + WithArgs("queue", log.TaskID, log.UnitID, log.BuildID, log.Event, log.TimestampMs) + if tt.dbErr != nil { + query.WillReturnError(tt.dbErr) + } else { + query.WillReturnResult(sqlmock.NewResult(0, 1)) + } + + err = store.Create(context.Background(), log) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueLogGet(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() + store := NewIssueLogStore(db, testMetrics(), "queue") + log := issueLogFixture() + query := mock.ExpectQuery(regexp.QuoteMeta( + `SELECT task_id, unit_id, build_id, event, timestamp_ms + FROM issue_log WHERE queue = ? AND task_id = ? AND unit_id = ? AND build_id = ? AND event = ?`)). + WithArgs("queue", log.TaskID, log.UnitID, log.BuildID, log.Event) + switch kind { + case "missing": + query.WillReturnError(sql.ErrNoRows) + case "database error": + query.WillReturnError(assert.AnError) + default: + var timestamp any = log.TimestampMs + if kind == "scan error" { + timestamp = "invalid" + } + query.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "build_id", "event", "timestamp_ms"}). + AddRow(log.TaskID, log.UnitID, log.BuildID, log.Event, timestamp)) + } + + got, err := store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Event) + switch kind { + case "found": + require.NoError(t, err) + require.Equal(t, log, 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.IssueLog{}, got) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueLogRejectsInvalidKeys(t *testing.T) { + for _, tt := range []struct { + name string + change func(*entity.IssueLog, *string) + }{ + {name: "oversized queue", change: func(_ *entity.IssueLog, q *string) { *q = strings.Repeat("q", 256) }}, + {name: "oversized task", change: func(l *entity.IssueLog, _ *string) { l.TaskID = strings.Repeat("t", 256) }}, + {name: "oversized unit", change: func(l *entity.IssueLog, _ *string) { l.UnitID = strings.Repeat("😀", 257) }}, + {name: "oversized build", change: func(l *entity.IssueLog, _ *string) { l.BuildID = strings.Repeat("b", 1021) }}, + {name: "empty build", change: func(l *entity.IssueLog, _ *string) { l.BuildID = "" }}, + {name: "unknown event", change: func(l *entity.IssueLog, _ *string) { l.Event = entity.IssueEventUnknown }}, + {name: "unsupported event", change: func(l *entity.IssueLog, _ *string) { l.Event = "other" }}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + log, queue := issueLogFixture(), "queue" + tt.change(&log, &queue) + store := NewIssueLogStore(db, testMetrics(), queue) + + mock.MatchExpectationsInOrder(false) + mock.ExpectExec("INSERT INTO issue_log").WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows( + []string{"task_id", "unit_id", "build_id", "event", "timestamp_ms"}). + AddRow(log.TaskID, log.UnitID, log.BuildID, log.Event, log.TimestampMs)) + + require.Error(t, store.Create(context.Background(), log)) + _, err = store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Event) + require.Error(t, err) + }) + } +} + +func TestIssueLogRejectsInvalidValues(t *testing.T) { + for _, tt := range []struct { + name string + change func(*entity.IssueLog) + }{ + {name: "zero timestamp", change: func(l *entity.IssueLog) { l.TimestampMs = 0 }}, + {name: "negative timestamp", change: func(l *entity.IssueLog) { l.TimestampMs = -1 }}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + log := issueLogFixture() + tt.change(&log) + store := NewIssueLogStore(db, testMetrics(), "queue") + + mock.ExpectExec("INSERT INTO issue_log").WillReturnResult(sqlmock.NewResult(0, 1)) + require.Error(t, store.Create(context.Background(), log)) + }) + } +} diff --git a/stovepipe/extension/storage/mysql/schema/issue_log.sql b/stovepipe/extension/storage/mysql/schema/issue_log.sql new file mode 100644 index 000000000..3c26086d0 --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/issue_log.sql @@ -0,0 +1,12 @@ +-- Each issue transition is retained once per build and event. +CREATE TABLE IF NOT EXISTS issue_log ( + queue VARBINARY(255) NOT NULL, + task_id VARBINARY(255) NOT NULL, + unit_id VARBINARY(1024) NOT NULL, + build_id VARBINARY(1020) NOT NULL, -- Accommodates existing VARCHAR(255) utf8mb4 build IDs. + event VARBINARY(32) NOT NULL, + + timestamp_ms BIGINT NOT NULL, + + PRIMARY KEY (queue, task_id, unit_id, build_id, event) +) ENGINE=InnoDB; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index 8b17e5203..cc0d31920 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -57,6 +57,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { buildStore: NewBuildStore(s.db, queueScope.SubScope("build_store"), queueName), buildIssueStore: NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName), openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_store"), queueName), + issueLogStore: NewIssueLogStore(s.db, queueScope.SubScope("issue_log_store"), queueName), validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), }, nil } @@ -79,6 +80,7 @@ type mysqlStorage struct { buildStore storage.BuildStore buildIssueStore storage.BuildIssueStore openIssueStore storage.OpenIssueStore + issueLogStore storage.IssueLogStore validationFactStore storage.ValidationFactStore } @@ -138,3 +140,7 @@ func (f *mysqlStorage) GetBuildIssueStore() storage.BuildIssueStore { func (f *mysqlStorage) GetOpenIssueStore() storage.OpenIssueStore { return f.openIssueStore } + +func (f *mysqlStorage) GetIssueLogStore() storage.IssueLogStore { + return f.issueLogStore +} diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index 81af1f193..81be9e764 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -23,6 +23,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" ) @@ -49,6 +50,7 @@ func TestNewStorage(t *testing.T) { assert.NotNil(t, bound.GetBuildStore()) assert.NotNil(t, bound.GetBuildIssueStore()) assert.NotNil(t, bound.GetOpenIssueStore()) + assert.NotNil(t, bound.GetIssueLogStore()) _, err = s.For("") assert.Error(t, err, "resolving an empty queue name must fail") @@ -145,6 +147,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "issue_log_store", + op: "get", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetIssueLogStore().Get(context.Background(), "task", "unit", "build-id", entity.IssueEventOpened) + return err + }, + }, { name: "validation_fact_store", op: "get", diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 18100c280..c115ed93e 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -100,6 +100,9 @@ type Storage interface { // GetOpenIssueStore returns the OpenIssueStore instance. GetOpenIssueStore() OpenIssueStore + // GetIssueLogStore returns the IssueLogStore instance. + GetIssueLogStore() IssueLogStore + // 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 97a6fe1a2..e78610ffa 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -4,6 +4,7 @@ go_test( name = "go_default_test", srcs = [ "build_issue_test.go", + "issue_log_test.go", "open_issue_test.go", "queue_state_transitions_test.go", "queue_summary_test.go", diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go new file mode 100644 index 000000000..50dbf0b59 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go @@ -0,0 +1,181 @@ +// 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 testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factory) { + t.Helper() + + bound, err := factory.For(storage.Config{QueueName: "issue-history/main"}) + require.NoError(t, err) + store := bound.GetIssueLogStore() + opened := entity.IssueLog{TaskID: "validation", UnitID: "unit", + Event: entity.IssueEventOpened, BuildID: "build/1", TimestampMs: 1000} + + t.Run("history survives resolution and reopening", func(t *testing.T) { + issue := entity.OpenIssue{TaskID: opened.TaskID, UnitID: opened.UnitID, FirstSeenBuildID: opened.BuildID} + require.NoError(t, bound.GetOpenIssueStore().Create(ctx, issue)) + require.NoError(t, bound.GetBuildIssueStore().Create(ctx, opened.BuildID, entity.BuildIssue{TaskID: opened.TaskID, UnitID: opened.UnitID})) + require.NoError(t, store.Create(ctx, opened)) + + resolved := opened + resolved.Event, resolved.BuildID, resolved.TimestampMs = entity.IssueEventResolved, "build/2", 2000 + require.NoError(t, store.Create(ctx, resolved)) + require.NoError(t, bound.GetOpenIssueStore().Delete(ctx, issue)) + _, err := bound.GetOpenIssueStore().Get(ctx, issue.TaskID, issue.UnitID) + require.ErrorIs(t, err, storage.ErrNotFound) + + reopened := opened + reopened.BuildID, reopened.TimestampMs = "build/3", 3000 + issue.FirstSeenBuildID = reopened.BuildID + require.NoError(t, bound.GetOpenIssueStore().Create(ctx, issue)) + require.NoError(t, store.Create(ctx, reopened)) + + for _, want := range []entity.IssueLog{opened, resolved, reopened} { + got, err := store.Get(ctx, want.TaskID, want.UnitID, want.BuildID, want.Event) + require.NoError(t, err) + require.Equal(t, want, got) + } + _, err = store.Get(ctx, reopened.TaskID, reopened.UnitID, reopened.BuildID, entity.IssueEventResolved) + require.ErrorIs(t, err, storage.ErrNotFound) + _, err = bound.GetBuildIssueStore().Get(ctx, opened.BuildID, opened.TaskID, opened.UnitID) + require.NoError(t, err) + }) + + t.Run("duplicate preserves original timestamp", func(t *testing.T) { + duplicate := opened + duplicate.TimestampMs = 4000 + require.ErrorIs(t, store.Create(ctx, duplicate), storage.ErrAlreadyExists) + got, err := store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) + require.NoError(t, err) + require.Equal(t, opened, got) + }) + + t.Run("event distinguishes transitions within one build", func(t *testing.T) { + for _, event := range []entity.IssueEvent{entity.IssueEventOpened, entity.IssueEventResolved} { + log := opened + log.UnitID, log.Event = "same-build", event + require.NoError(t, store.Create(ctx, log)) + + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) + require.NoError(t, err) + require.Equal(t, log, got) + } + }) + + t.Run("queue and byte-exact identity isolation", func(t *testing.T) { + other, err := factory.For(storage.Config{QueueName: "Issue-history/main"}) + require.NoError(t, err) + _, err = other.GetIssueLogStore().Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) + require.ErrorIs(t, err, storage.ErrNotFound) + otherLog := opened + otherLog.TimestampMs++ + require.NoError(t, other.GetIssueLogStore().Create(ctx, otherLog)) + got, err := other.GetIssueLogStore().Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) + require.NoError(t, err) + require.Equal(t, otherLog, got) + + for _, tt := range []struct { + name string + taskID, unitID, buildID string + }{ + {"task case", "Validation", "unit", "build/1"}, + {"task trailing space", "validation ", "unit", "build/1"}, + {"unit case", "validation", "Unit", "build/1"}, + {"unit trailing space", "validation", "unit ", "build/1"}, + {"task-wide issue", "validation", "", "build/1"}, + {"build case", "validation", "unit", "Build/1"}, + {"build trailing space", "validation", "unit", "build/1 "}, + {"embedded separator in task", "a\x00b", "c", "build/1"}, + {"embedded separator in unit", "a", "b\x00c", "build/1"}, + } { + t.Run(tt.name, func(t *testing.T) { + log := opened + log.TaskID, log.UnitID, log.BuildID = tt.taskID, tt.unitID, tt.buildID + require.NoError(t, store.Create(ctx, log)) + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) + require.NoError(t, err) + require.Equal(t, log, got) + }) + } + got, err = store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) + require.NoError(t, err) + require.Equal(t, opened, got) + }) + + t.Run("concurrent replay retains one event", func(t *testing.T) { + log := opened + log.UnitID = "concurrent" + results := make(chan error, 2) + for i := 0; i < 2; i++ { + go func() { results <- store.Create(ctx, log) }() + } + successes, duplicates := 0, 0 + for i := 0; i < 2; i++ { + err := <-results + switch { + case err == nil: + successes++ + case errors.Is(err, storage.ErrAlreadyExists): + duplicates++ + default: + require.NoError(t, err) + } + } + require.Equal(t, 1, successes) + require.Equal(t, 1, duplicates) + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) + require.NoError(t, err) + require.Equal(t, log, got) + }) + + t.Run("binary identifier limits", func(t *testing.T) { + maxBound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) + require.NoError(t, err) + maxStore := maxBound.GetIssueLogStore() + log := entity.IssueLog{TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("😀", 256), + Event: entity.IssueEventResolved, + BuildID: strings.Repeat("😀", 255), TimestampMs: 1000} + require.NoError(t, maxStore.Create(ctx, log)) + got, err := maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) + require.NoError(t, err) + require.Equal(t, log, got) + + oversized := log + oversized.BuildID += "x" + require.Error(t, maxStore.Create(ctx, oversized)) + _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID, oversized.BuildID, oversized.Event) + require.Error(t, err) + oversized = log + oversized.UnitID += "x" + require.Error(t, maxStore.Create(ctx, oversized)) + _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID, oversized.BuildID, oversized.Event) + require.Error(t, err) + + got, err = maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) + require.NoError(t, err) + require.Equal(t, log, got) + }) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 7349c4c17..7dcdd406e 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -100,6 +100,11 @@ func TestMySQLStorage(t *testing.T) { testOpenIssueStore(t, ctx, factory) }) + t.Run("IssueLogStore", func(t *testing.T) { + resetStorage(t, db) + testIssueLogStore(t, ctx, factory) + }) + t.Run("BuildStore", func(t *testing.T) { resetStorage(t, db) testSuite := new(MySQLBuildStoreSuite) @@ -154,6 +159,7 @@ func resetStorage(t *testing.T, db *sql.DB) { for _, statement := range []string{ "TRUNCATE TABLE open_issue", + "TRUNCATE TABLE issue_log", "TRUNCATE TABLE queue_log", "TRUNCATE TABLE queue_summary", "TRUNCATE TABLE request_log",