From 79dc99f089c717b74547907e7509f6e883e3ba35 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Thu, 29 Oct 2026 03:59:13 +0000 Subject: [PATCH] feat(stovepipe): Add open issue lookup storage --- stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 12 +- stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../storage/mock/open_issue_store_mock.go | 117 ++++++++ .../extension/storage/mock/storage_mock.go | 14 + stovepipe/extension/storage/mysql/BUILD.bazel | 2 + .../storage/mysql/open_issue_store.go | 194 ++++++++++++ .../storage/mysql/open_issue_store_test.go | 276 ++++++++++++++++++ .../storage/mysql/schema/open_issue.sql | 2 +- stovepipe/extension/storage/mysql/storage.go | 6 + .../extension/storage/mysql/storage_test.go | 9 + .../extension/storage/open_issue_store.go | 61 ++++ stovepipe/extension/storage/storage.go | 3 + .../extension/storage/mysql/BUILD.bazel | 1 + .../storage/mysql/open_issue_test.go | 175 +++++++++++ .../extension/storage/mysql/storage_test.go | 6 + 16 files changed, 878 insertions(+), 2 deletions(-) create mode 100644 stovepipe/extension/storage/mock/open_issue_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/open_issue_store.go create mode 100644 stovepipe/extension/storage/mysql/open_issue_store_test.go create mode 100644 stovepipe/extension/storage/open_issue_store.go create mode 100644 test/integration/stovepipe/extension/storage/mysql/open_issue_test.go diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index f9a584535..de5782f47 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "issue_state_store.go", "issue_summary_store.go", "issue_tracking_store.go", + "open_issue_store.go", "project_fact_by_degree_store.go", "queue_log_store.go", "queue_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 646a7adea..e896596bb 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -48,7 +48,7 @@ Create is conditional on absence, including closed records. Update checks the su Listing reads a bounded primary-key range in bytewise task/unit order and includes both statuses. Callers filter closed records in memory and advance after the last scanned key, including closed records. A page containing only closed records is not the end of the range. No status index or server-side filtering is required. Task IDs must be nonempty; empty units identify task-wide issues. Identifier byte limits match the build-issue store. -The MySQL backend uses the new `issue_state` table. Create it before deploying the updated readers. The prior `open_issue` table is not read or automatically copied; no in-repository runtime writer was enabled for it. Deployments with externally populated rows need an explicit backfill before switching readers. The repair executor still receives an `OpenIssue`, which describes an open occurrence rather than the storage record. +The MySQL backend uses `issue_state` for operational tracking. The separate `open_issue` table is a derived lookup, not a replacement for retained issue state. The repair executor still receives an `OpenIssue`, which describes an open occurrence rather than the operational storage record. ## Issue summaries @@ -58,6 +58,16 @@ The MySQL backend uses the new `issue_state` table. Create it before deploying t The MySQL primary key is `(queue, task_id, unit_id)` with byte-exact identity and the same identifier limits as the issue-state store. Source request IDs follow the existing canonical positive decimal ID contract. The new `issue_summary` table must exist before deploying readers or writers. The [issue-log materializer](../../core/issuelog/README.md) updates summaries during normal issue tracking and recovery. No public open-issues list is exposed yet. +## Open-issue lookups + +`OpenIssueStore` exposes the existing `open_issue` table as a queue-scoped lookup keyed by task/unit. Each open-issue row contains only the opening's `FirstSeenBuildID`, using the existing `OpenIssue` entity and identifier byte limits. No columns, indexes, or operational issue-state fields are added. + +Create returns `ErrAlreadyExists` for any existing task/unit key. Replace compares the expected opening build ID before changing it; Delete compares the opening in the supplied open-issue row before removing it. Both operations return whether they changed the open-issue row, with false for a missing key or a different opening. Replacing an opening with itself is also a no-op returning false. These are atomic conditional writes to one complete key, not build-order checks. An old close that still expects a previous opening cannot delete a newer opening. + +Listing traverses a bounded queue-prefix primary-key range in ascending bytewise task/unit order. The zero cursor starts the range; otherwise the cursor excludes its key and all earlier keys, even if that open-issue row has been deleted. Empty units identify task-wide issues, while task IDs must be nonempty. Membership can change between reads. A consumer that skips stale references must still advance after the last returned open-issue row, not the last accepted issue. + +Open-issue rows are not authoritative issue status. A delayed create can leave a reference to a closed issue, so readers must verify that the summary is open and its opening matches before returning a result. The store does not read summaries, enforce source ordering, backfill existing rows, or populate open-issue rows automatically. No runtime writer or public list API uses this store yet. + ## Issue lifecycle history `IssueLogStore` retains immutable open/closed status transitions independently of current issue state and per-build observations. Each transition is keyed by task, unit, build, and status within the bound queue. An empty unit identifies a task-wide issue. History remains when an issue closes or reopens; transitions do not carry a separate reference to the opening they belong to. diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index c9f45072b..7e7545c10 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "issue_state_store_mock.go", "issue_summary_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", "queue_store_mock.go", diff --git a/stovepipe/extension/storage/mock/open_issue_store_mock.go b/stovepipe/extension/storage/mock/open_issue_store_mock.go new file mode 100644 index 000000000..af140bfa7 --- /dev/null +++ b/stovepipe/extension/storage/mock/open_issue_store_mock.go @@ -0,0 +1,117 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: open_issue_store.go +// +// Generated by this command: +// +// mockgen -source=open_issue_store.go -destination=mock/open_issue_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + storage "github.com/uber/submitqueue/stovepipe/extension/storage" + gomock "go.uber.org/mock/gomock" +) + +// MockOpenIssueStore is a mock of OpenIssueStore interface. +type MockOpenIssueStore struct { + ctrl *gomock.Controller + recorder *MockOpenIssueStoreMockRecorder + isgomock struct{} +} + +// MockOpenIssueStoreMockRecorder is the mock recorder for MockOpenIssueStore. +type MockOpenIssueStoreMockRecorder struct { + mock *MockOpenIssueStore +} + +// NewMockOpenIssueStore creates a new mock instance. +func NewMockOpenIssueStore(ctrl *gomock.Controller) *MockOpenIssueStore { + mock := &MockOpenIssueStore{ctrl: ctrl} + mock.recorder = &MockOpenIssueStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockOpenIssueStore) EXPECT() *MockOpenIssueStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockOpenIssueStore) Create(ctx context.Context, issue entity.OpenIssue) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, issue) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockOpenIssueStoreMockRecorder) Create(ctx, issue any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockOpenIssueStore)(nil).Create), ctx, issue) +} + +// Delete mocks base method. +func (m *MockOpenIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Delete", ctx, issue) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Delete indicates an expected call of Delete. +func (mr *MockOpenIssueStoreMockRecorder) Delete(ctx, issue any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockOpenIssueStore)(nil).Delete), ctx, issue) +} + +// Get mocks base method. +func (m *MockOpenIssueStore) Get(ctx context.Context, taskID, unitID string) (entity.OpenIssue, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID) + ret0, _ := ret[0].(entity.OpenIssue) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockOpenIssueStoreMockRecorder) Get(ctx, taskID, unitID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockOpenIssueStore)(nil).Get), ctx, taskID, unitID) +} + +// List mocks base method. +func (m *MockOpenIssueStore) List(ctx context.Context, bounds storage.OpenIssueRange) ([]entity.OpenIssue, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, bounds) + ret0, _ := ret[0].([]entity.OpenIssue) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockOpenIssueStoreMockRecorder) List(ctx, bounds any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockOpenIssueStore)(nil).List), ctx, bounds) +} + +// Replace mocks base method. +func (m *MockOpenIssueStore) Replace(ctx context.Context, issue entity.OpenIssue, expectedFirstSeenBuildID string) (bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Replace", ctx, issue, expectedFirstSeenBuildID) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Replace indicates an expected call of Replace. +func (mr *MockOpenIssueStoreMockRecorder) Replace(ctx, issue, expectedFirstSeenBuildID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Replace", reflect.TypeOf((*MockOpenIssueStore)(nil).Replace), ctx, issue, expectedFirstSeenBuildID) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index 7b45df72f..2b0e19c93 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -163,6 +163,20 @@ func (mr *MockStorageMockRecorder) GetIssueTrackingStore() *gomock.Call { 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() + ret := m.ctrl.Call(m, "GetOpenIssueStore") + ret0, _ := ret[0].(storage.OpenIssueStore) + return ret0 +} + +// GetOpenIssueStore indicates an expected call of GetOpenIssueStore. +func (mr *MockStorageMockRecorder) GetOpenIssueStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetOpenIssueStore", reflect.TypeOf((*MockStorage)(nil).GetOpenIssueStore)) +} + // GetProjectFactByDegreeStore mocks base method. func (m *MockStorage) GetProjectFactByDegreeStore() storage.ProjectFactByDegreeStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index c681a7e63..45b86d387 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "issue_state_store.go", "issue_summary_store.go", "issue_tracking_store.go", + "open_issue_store.go", "project_fact_by_degree_store.go", "queue_log_store.go", "queue_store.go", @@ -44,6 +45,7 @@ go_test( "issue_state_store_test.go", "issue_summary_store_test.go", "issue_tracking_store_test.go", + "open_issue_store_test.go", "project_fact_by_degree_store_test.go", "queue_store_test.go", "queue_summary_store_test.go", diff --git a/stovepipe/extension/storage/mysql/open_issue_store.go b/stovepipe/extension/storage/mysql/open_issue_store.go new file mode 100644 index 000000000..4eec563df --- /dev/null +++ b/stovepipe/extension/storage/mysql/open_issue_store.go @@ -0,0 +1,194 @@ +// 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" +) + +const listOpenIssuesQuery = ` + SELECT task_id, unit_id, first_seen_build_id + FROM open_issue WHERE queue = ? + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +const listOpenIssuesAfterCursorQuery = ` + SELECT task_id, unit_id, first_seen_build_id + FROM open_issue WHERE queue = ? + AND (task_id > ? OR (task_id = ? AND unit_id > ?)) + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +type openIssueStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewOpenIssueStore binds derived open-issue open issues to one queue. +func NewOpenIssueStore(db *sql.DB, scope tally.Scope, queue string) storage.OpenIssueStore { + return &openIssueStore{db: db, scope: scope, queue: queue} +} + +func (s *openIssueStore) Create(ctx context.Context, issue entity.OpenIssue) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := s.validateIssue(issue); err != nil { + return err + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO open_issue (queue, task_id, unit_id, first_seen_build_id) VALUES (?, ?, ?, ?)`, + s.queue, issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) + if isDuplicateEntry(err) { + return storage.ErrAlreadyExists + } + if err != nil { + return fmt.Errorf("create open issue open issue: %w", err) + } + return nil +} + +func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (issue entity.OpenIssue, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if taskID == "" { + return entity.OpenIssue{}, fmt.Errorf("open issue task ID must not be empty") + } + if err := validateBuildIssueKey(s.queue, "", taskID, unitID); err != nil { + return entity.OpenIssue{}, err + } + err := s.db.QueryRowContext(ctx, + `SELECT task_id, unit_id, first_seen_build_id FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ?`, + s.queue, taskID, unitID, + ).Scan(&issue.TaskID, &issue.UnitID, &issue.FirstSeenBuildID) + if errors.Is(err, sql.ErrNoRows) { + return entity.OpenIssue{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.OpenIssue{}, fmt.Errorf("get open issue open issue: %w", err) + } + return issue, nil +} + +func (s *openIssueStore) Replace(ctx context.Context, issue entity.OpenIssue, expectedFirstSeenBuildID string) (replaced bool, retErr error) { + op := metrics.Begin(s.scope, "replace", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := s.validateIssue(issue); err != nil { + return false, err + } + if expectedFirstSeenBuildID == "" { + return false, fmt.Errorf("open issue replacement requires an expected opening build ID") + } + if err := validateBuildIssueKey(s.queue, expectedFirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil { + return false, err + } + if issue.FirstSeenBuildID == expectedFirstSeenBuildID { + return false, nil + } + + result, err := s.db.ExecContext(ctx, + `UPDATE open_issue SET first_seen_build_id = ? WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?`, + issue.FirstSeenBuildID, s.queue, issue.TaskID, issue.UnitID, expectedFirstSeenBuildID) + if err != nil { + return false, fmt.Errorf("replace open issue open issue: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("read open issue replacement count: %w", err) + } + return rows == 1, nil +} + +func (s *openIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (deleted bool, retErr error) { + op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := s.validateIssue(issue); err != nil { + return false, err + } + result, err := s.db.ExecContext(ctx, + `DELETE FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?`, + s.queue, issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) + if err != nil { + return false, fmt.Errorf("delete open issue open issue: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("read open issue deletion count: %w", err) + } + return rows == 1, nil +} + +func (s *openIssueStore) List(ctx context.Context, bounds storage.OpenIssueRange) (ret []entity.OpenIssue, retErr error) { + op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + cursor := bounds.After + if bounds.Limit <= 0 { + return nil, fmt.Errorf("open issue range requires a positive limit") + } + if cursor.TaskID == "" && cursor.UnitID != "" { + return nil, fmt.Errorf("open issue cursor requires a task ID when a unit ID is supplied") + } + if err := validateBuildIssueKey(s.queue, "", cursor.TaskID, cursor.UnitID); err != nil { + return nil, err + } + + var ( + rows *sql.Rows + err error + ) + if cursor.TaskID != "" { + rows, err = s.db.QueryContext(ctx, listOpenIssuesAfterCursorQuery, + s.queue, cursor.TaskID, cursor.TaskID, cursor.UnitID, bounds.Limit) + } else { + rows, err = s.db.QueryContext(ctx, listOpenIssuesQuery, s.queue, bounds.Limit) + } + if err != nil { + return nil, fmt.Errorf("list open issue open issues: %w", err) + } + defer rows.Close() + + issues := make([]entity.OpenIssue, 0) + for rows.Next() { + var issue entity.OpenIssue + if err := rows.Scan(&issue.TaskID, &issue.UnitID, &issue.FirstSeenBuildID); err != nil { + return nil, fmt.Errorf("scan open issue open issue: %w", err) + } + issues = append(issues, issue) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate open issue open issues: %w", err) + } + return issues, nil +} + +func (s *openIssueStore) validateIssue(issue entity.OpenIssue) error { + if issue.TaskID == "" || issue.FirstSeenBuildID == "" { + return fmt.Errorf("open issue open issue requires task and opening build IDs") + } + return validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID) +} + +var _ storage.OpenIssueStore = (*openIssueStore)(nil) diff --git a/stovepipe/extension/storage/mysql/open_issue_store_test.go b/stovepipe/extension/storage/mysql/open_issue_store_test.go new file mode 100644 index 000000000..ba88eb2fe --- /dev/null +++ b/stovepipe/extension/storage/mysql/open_issue_store_test.go @@ -0,0 +1,276 @@ +// 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" + "database/sql/driver" + "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" +) + +const ( + insertOpenIssueSQL = `INSERT INTO open_issue (queue, task_id, unit_id, first_seen_build_id) VALUES (?, ?, ?, ?)` + getOpenIssueSQL = `SELECT task_id, unit_id, first_seen_build_id FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ?` + replaceOpenIssueSQL = `UPDATE open_issue SET first_seen_build_id = ? WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?` + deleteOpenIssueSQL = `DELETE FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?` +) + +func newOpenIssueTestStore(t *testing.T) (sqlmock.Sqlmock, storage.OpenIssueStore) { + t.Helper() + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual)) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, mock.ExpectationsWereMet()) + _ = db.Close() + }) + return mock, NewOpenIssueStore(db, testMetrics(), "queue/main") +} + +func TestOpenIssueStoreCreate(t *testing.T) { + for _, tt := range []struct { + name string + dbErr, wantErr error + }{ + {name: "created"}, + {"duplicate", &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, storage.ErrAlreadyExists}, + {"database error", assert.AnError, assert.AnError}, + } { + t.Run(tt.name, func(t *testing.T) { + mock, store := newOpenIssueTestStore(t) + issue := entity.OpenIssue{TaskID: "task", UnitID: "unit", FirstSeenBuildID: "build/1"} + expected := mock.ExpectExec(insertOpenIssueSQL).WithArgs("queue/main", issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) + if tt.dbErr != nil { + expected.WillReturnError(tt.dbErr) + } else { + expected.WillReturnResult(sqlmock.NewResult(0, 1)) + } + err := store.Create(context.Background(), issue) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + }) + } +} + +func TestOpenIssueStoreGet(t *testing.T) { + for _, tt := range []struct { + name, unit string + dbErr, wantErr error + }{ + {name: "found", unit: "unit"}, + {name: "task wide"}, + {name: "missing", dbErr: sql.ErrNoRows, wantErr: storage.ErrNotFound}, + {name: "database error", dbErr: assert.AnError, wantErr: assert.AnError}, + } { + t.Run(tt.name, func(t *testing.T) { + mock, store := newOpenIssueTestStore(t) + want := entity.OpenIssue{TaskID: "task", UnitID: tt.unit, FirstSeenBuildID: "build/1"} + expected := mock.ExpectQuery(getOpenIssueSQL).WithArgs("queue/main", want.TaskID, want.UnitID) + if tt.dbErr != nil { + expected.WillReturnError(tt.dbErr) + } else { + expected.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}).AddRow(want.TaskID, want.UnitID, want.FirstSeenBuildID)) + } + got, err := store.Get(context.Background(), want.TaskID, want.UnitID) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + require.Empty(t, got) + } else { + require.NoError(t, err) + require.Equal(t, want, got) + } + }) + } +} + +func TestOpenIssueStoreConditionalWrites(t *testing.T) { + issue := entity.OpenIssue{TaskID: "task", UnitID: "unit", FirstSeenBuildID: "build/2"} + for _, operation := range []struct { + name, query string + args []driver.Value + call func(storage.OpenIssueStore) (bool, error) + }{ + {"replace", replaceOpenIssueSQL, []driver.Value{issue.FirstSeenBuildID, "queue/main", issue.TaskID, issue.UnitID, "build/1"}, + func(s storage.OpenIssueStore) (bool, error) { return s.Replace(context.Background(), issue, "build/1") }}, + {"delete", deleteOpenIssueSQL, []driver.Value{"queue/main", issue.TaskID, issue.UnitID, issue.FirstSeenBuildID}, + func(s storage.OpenIssueStore) (bool, error) { return s.Delete(context.Background(), issue) }}, + } { + t.Run(operation.name, func(t *testing.T) { + for _, tt := range []struct { + name string + result sql.Result + dbErr error + changed bool + wantErr error + }{ + {name: "matched", result: sqlmock.NewResult(0, 1), changed: true}, + {name: "missing or different opening", result: sqlmock.NewResult(0, 0)}, + {name: "database error", dbErr: assert.AnError, wantErr: assert.AnError}, + {name: "row count error", result: sqlmock.NewErrorResult(assert.AnError), wantErr: assert.AnError}, + } { + t.Run(tt.name, func(t *testing.T) { + mock, store := newOpenIssueTestStore(t) + expected := mock.ExpectExec(operation.query).WithArgs(operation.args...) + if tt.dbErr != nil { + expected.WillReturnError(tt.dbErr) + } else { + expected.WillReturnResult(tt.result) + } + changed, err := operation.call(store) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + require.Equal(t, tt.changed, changed) + }) + } + }) + } +} + +func TestOpenIssueStoreUnchangedReplacement(t *testing.T) { + _, store := newOpenIssueTestStore(t) + issue := entity.OpenIssue{TaskID: "task", UnitID: "unit", FirstSeenBuildID: "build/1"} + replaced, err := store.Replace(context.Background(), issue, issue.FirstSeenBuildID) + require.NoError(t, err) + require.False(t, replaced) +} + +func TestOpenIssueStoreList(t *testing.T) { + for _, tt := range []struct { + name string + after storage.OpenIssueCursor + empty, scanError, rowError bool + dbErr error + }{ + {name: "start"}, + {name: "after task wide", after: storage.OpenIssueCursor{TaskID: "task"}}, + {name: "after unit", after: storage.OpenIssueCursor{TaskID: "task", UnitID: "earlier"}}, + {name: "empty", empty: true}, + {name: "query error", dbErr: assert.AnError}, + {name: "scan error", scanError: true}, + {name: "iteration error", rowError: true}, + } { + t.Run(tt.name, func(t *testing.T) { + mock, store := newOpenIssueTestStore(t) + want := entity.OpenIssue{TaskID: "task", UnitID: "unit", FirstSeenBuildID: "build/1"} + query := `SELECT task_id, unit_id, first_seen_build_id FROM open_issue WHERE queue = ? ORDER BY task_id ASC, unit_id ASC LIMIT ?` + args := []driver.Value{"queue/main", int64(2)} + if tt.after.TaskID != "" { + query = `SELECT task_id, unit_id, first_seen_build_id FROM open_issue WHERE queue = ? AND (task_id > ? OR (task_id = ? AND unit_id > ?)) ORDER BY task_id ASC, unit_id ASC LIMIT ?` + args = []driver.Value{"queue/main", tt.after.TaskID, tt.after.TaskID, tt.after.UnitID, int64(2)} + } + expected := mock.ExpectQuery(query).WithArgs(args...) + if tt.dbErr != nil { + expected.WillReturnError(tt.dbErr) + } else { + rows := sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}) + if !tt.empty { + var opening any = want.FirstSeenBuildID + if tt.scanError { + opening = nil + } + rows.AddRow(want.TaskID, want.UnitID, opening) + if tt.rowError { + rows.AddRow("task", "unit-2", "build/2").RowError(1, assert.AnError) + } + } + expected.WillReturnRows(rows).RowsWillBeClosed() + } + got, err := store.List(context.Background(), storage.OpenIssueRange{After: tt.after, Limit: 2}) + switch { + case tt.dbErr != nil: + require.ErrorIs(t, err, tt.dbErr) + require.Nil(t, got) + case tt.rowError: + require.ErrorIs(t, err, assert.AnError) + require.Nil(t, got) + case tt.scanError: + require.Error(t, err) + require.Nil(t, got) + case tt.empty: + require.NoError(t, err) + require.NotNil(t, got) + require.Empty(t, got) + default: + require.NoError(t, err) + require.Equal(t, []entity.OpenIssue{want}, got) + } + }) + } +} + +func TestOpenIssueStoreRejectsInvalidOpenIssues(t *testing.T) { + for _, tt := range []struct { + name string + issue entity.OpenIssue + }{ + {"missing task", entity.OpenIssue{FirstSeenBuildID: "build/1"}}, + {"missing opening", entity.OpenIssue{TaskID: "task"}}, + {"long task", entity.OpenIssue{TaskID: strings.Repeat("t", 256), FirstSeenBuildID: "build/1"}}, + {"long unit", entity.OpenIssue{TaskID: "task", UnitID: strings.Repeat("u", 1025), FirstSeenBuildID: "build/1"}}, + {"long opening", entity.OpenIssue{TaskID: "task", FirstSeenBuildID: strings.Repeat("b", 1021)}}, + } { + t.Run(tt.name, func(t *testing.T) { + _, store := newOpenIssueTestStore(t) + require.Error(t, store.Create(context.Background(), tt.issue)) + replaced, err := store.Replace(context.Background(), tt.issue, "build/previous") + require.Error(t, err) + require.False(t, replaced) + deleted, err := store.Delete(context.Background(), tt.issue) + require.Error(t, err) + require.False(t, deleted) + }) + } + for _, expected := range []string{"", strings.Repeat("b", 1021)} { + t.Run("invalid replacement guard", func(t *testing.T) { + _, store := newOpenIssueTestStore(t) + replaced, err := store.Replace(context.Background(), entity.OpenIssue{TaskID: "task", FirstSeenBuildID: "build/1"}, expected) + require.Error(t, err) + require.False(t, replaced) + }) + } +} + +func TestOpenIssueStoreRejectsInvalidReads(t *testing.T) { + for _, cursor := range []storage.OpenIssueCursor{ + {UnitID: "unit"}, + {TaskID: strings.Repeat("t", 256)}, + {TaskID: "task", UnitID: strings.Repeat("u", 1025)}, + } { + _, store := newOpenIssueTestStore(t) + _, err := store.Get(context.Background(), cursor.TaskID, cursor.UnitID) + require.Error(t, err) + _, err = store.List(context.Background(), storage.OpenIssueRange{After: cursor, Limit: 1}) + require.Error(t, err) + } + for _, limit := range []int{0, -1} { + _, store := newOpenIssueTestStore(t) + _, err := store.List(context.Background(), storage.OpenIssueRange{Limit: limit}) + require.Error(t, err) + } +} diff --git a/stovepipe/extension/storage/mysql/schema/open_issue.sql b/stovepipe/extension/storage/mysql/schema/open_issue.sql index 8112e4260..901adfc61 100644 --- a/stovepipe/extension/storage/mysql/schema/open_issue.sql +++ b/stovepipe/extension/storage/mysql/schema/open_issue.sql @@ -1,4 +1,4 @@ --- Retained for schema compatibility with existing deployments; active issue state uses issue_state. +-- Derived opening references for open-only lookups; issue_summary retains authoritative read state. CREATE TABLE IF NOT EXISTS open_issue ( queue VARBINARY(255) NOT NULL, task_id VARBINARY(255) NOT NULL, diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index c7408770c..ea6207fe3 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -59,6 +59,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { 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), + openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_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), @@ -85,6 +86,7 @@ type mysqlStorage struct { issueStateStore storage.IssueStateStore issueLogStore storage.IssueLogStore issueSummaryStore storage.IssueSummaryStore + openIssueStore storage.OpenIssueStore issueTrackingStore storage.IssueTrackingStore validationFactStore storage.ValidationFactStore projectFactByDegreeStore storage.ProjectFactByDegreeStore @@ -159,6 +161,10 @@ func (f *mysqlStorage) GetIssueSummaryStore() storage.IssueSummaryStore { return f.issueSummaryStore } +func (f *mysqlStorage) GetOpenIssueStore() storage.OpenIssueStore { + return f.openIssueStore +} + 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 f9232d8ee..a96407ea6 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -52,6 +52,7 @@ func TestNewStorage(t *testing.T) { assert.NotNil(t, bound.GetIssueStateStore()) assert.NotNil(t, bound.GetIssueLogStore()) assert.NotNil(t, bound.GetIssueSummaryStore()) + assert.NotNil(t, bound.GetOpenIssueStore()) assert.NotNil(t, bound.GetIssueTrackingStore()) _, err = s.For("") @@ -117,6 +118,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "open_issue_store", + op: "get", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetOpenIssueStore().Get(context.Background(), "task", "unit") + return err + }, + }, { name: "request_acceptance_store", op: "list", diff --git a/stovepipe/extension/storage/open_issue_store.go b/stovepipe/extension/storage/open_issue_store.go new file mode 100644 index 000000000..0e79a1362 --- /dev/null +++ b/stovepipe/extension/storage/open_issue_store.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 storage + +//go:generate mockgen -source=open_issue_store.go -destination=mock/open_issue_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// OpenIssueCursor is an exclusive position in bytewise task/unit order. +type OpenIssueCursor struct { + // TaskID is the task component; empty with an empty UnitID starts the range. + TaskID string + // UnitID is the unit component; empty identifies a task-wide issue. + UnitID string +} + +// OpenIssueRange bounds a primary-key scan within the store's queue. +type OpenIssueRange struct { + // After excludes this key and earlier keys; the zero cursor starts the range. + After OpenIssueCursor + // Limit is the positive maximum number of open issues to return. + Limit int +} + +// OpenIssueStore retains queue-scoped opening references keyed by task/unit. +// Open-issue rows are derived lookups; consumers must validate them against issue summaries. +type OpenIssueStore interface { + // Create returns ErrAlreadyExists if the task/unit key already exists. + Create(ctx context.Context, issue entity.OpenIssue) error + + // Get returns ErrNotFound when the task/unit key has no open issue. + Get(ctx context.Context, taskID, unitID string) (entity.OpenIssue, error) + + // Replace changes the opening only when it matches expectedFirstSeenBuildID. + // It returns false for a missing key, a mismatch, or an unchanged opening. + Replace(ctx context.Context, issue entity.OpenIssue, expectedFirstSeenBuildID string) (bool, error) + + // Delete removes the open issue only when its opening matches issue.FirstSeenBuildID. + // It returns false for a missing key or a different opening. + Delete(ctx context.Context, issue entity.OpenIssue) (bool, error) + + // List returns at most Limit open issues in bytewise task/unit order. + // Membership may change between reads; a cursor need not still exist. + List(ctx context.Context, bounds OpenIssueRange) ([]entity.OpenIssue, error) +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index c63e94d0e..b69297ae9 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -106,6 +106,9 @@ type Storage interface { // GetIssueSummaryStore returns the IssueSummaryStore instance. GetIssueSummaryStore() IssueSummaryStore + // GetOpenIssueStore returns the derived open-issue lookup store. + GetOpenIssueStore() OpenIssueStore + // GetIssueTrackingStore returns the operational reconciliation progress store. GetIssueTrackingStore() IssueTrackingStore diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index 47db77393..1f82fa4f6 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -11,6 +11,7 @@ go_test( "issue_state_test.go", "issue_summary_test.go", "issue_tracking_test.go", + "open_issue_test.go", "project_fact_by_degree_test.go", "queue_state_transitions_test.go", "queue_summary_test.go", diff --git a/test/integration/stovepipe/extension/storage/mysql/open_issue_test.go b/test/integration/stovepipe/extension/storage/mysql/open_issue_test.go new file mode 100644 index 000000000..f8769a148 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/open_issue_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" + "fmt" + "sort" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func testOpenIssueStore(t *testing.T, ctx context.Context, factory storage.Factory) { + t.Helper() + bound, err := factory.For(storage.Config{QueueName: "open-issue/main"}) + require.NoError(t, err) + store := bound.GetOpenIssueStore() + first := entity.OpenIssue{TaskID: "task", UnitID: "unit", FirstSeenBuildID: "build/1"} + _, err = store.Get(ctx, first.TaskID, first.UnitID) + require.ErrorIs(t, err, storage.ErrNotFound) + changed, err := store.Delete(ctx, first) + require.NoError(t, err) + require.False(t, changed) + changed, err = store.Replace(ctx, first, "missing") + require.NoError(t, err) + require.False(t, changed) + require.NoError(t, store.Create(ctx, first)) + require.ErrorIs(t, store.Create(ctx, first), storage.ErrAlreadyExists) + got, err := store.Get(ctx, first.TaskID, first.UnitID) + require.NoError(t, err) + require.Equal(t, first, got) + + second := first + second.FirstSeenBuildID = "build/2" + require.ErrorIs(t, store.Create(ctx, second), storage.ErrAlreadyExists) + changed, err = store.Replace(ctx, second, first.FirstSeenBuildID) + require.NoError(t, err) + require.True(t, changed) + changed, err = store.Replace(ctx, second, second.FirstSeenBuildID) + require.NoError(t, err) + require.False(t, changed) + changed, err = store.Delete(ctx, first) + require.NoError(t, err) + require.False(t, changed) + reopened := first + reopened.FirstSeenBuildID = "build/3" + changed, err = store.Replace(ctx, reopened, first.FirstSeenBuildID) + require.NoError(t, err) + require.False(t, changed) + got, err = store.Get(ctx, first.TaskID, first.UnitID) + require.NoError(t, err) + require.Equal(t, second, got) + + changed, err = store.Delete(ctx, second) + require.NoError(t, err) + require.True(t, changed) + changed, err = store.Delete(ctx, second) + require.NoError(t, err) + require.False(t, changed) + require.NoError(t, store.Create(ctx, reopened)) + for _, staleOpening := range []string{first.FirstSeenBuildID, second.FirstSeenBuildID, "Build/3", "build/3 "} { + stale := first + stale.FirstSeenBuildID = staleOpening + changed, err = store.Delete(ctx, stale) + require.NoError(t, err) + require.False(t, changed) + changed, err = store.Replace(ctx, first, staleOpening) + require.NoError(t, err) + require.False(t, changed) + } + got, err = store.Get(ctx, first.TaskID, first.UnitID) + require.NoError(t, err) + require.Equal(t, reopened, got) + + t.Run("queue isolation", func(t *testing.T) { + other, err := factory.For(storage.Config{QueueName: "Open-issue/main"}) + require.NoError(t, err) + otherStore := other.GetOpenIssueStore() + _, err = otherStore.Get(ctx, first.TaskID, first.UnitID) + require.ErrorIs(t, err, storage.ErrNotFound) + require.NoError(t, otherStore.Create(ctx, first)) + changed, err := otherStore.Replace(ctx, second, first.FirstSeenBuildID) + require.NoError(t, err) + require.True(t, changed) + changed, err = otherStore.Delete(ctx, second) + require.NoError(t, err) + require.True(t, changed) + got, err := store.Get(ctx, first.TaskID, first.UnitID) + require.NoError(t, err) + require.Equal(t, reopened, got) + }) + + t.Run("bounded bytewise listing", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: "open-issue/list"}) + require.NoError(t, err) + store := bound.GetOpenIssueStore() + empty, err := store.List(ctx, storage.OpenIssueRange{Limit: 1}) + require.NoError(t, err) + require.Empty(t, empty) + expected := []entity.OpenIssue{ + {TaskID: "A"}, {TaskID: "a"}, + {TaskID: "a", UnitID: "\x00"}, {TaskID: "a", UnitID: "Unit"}, + {TaskID: "a", UnitID: "unit"}, {TaskID: "a", UnitID: "unit "}, + {TaskID: "a ", UnitID: "unit"}, {TaskID: "a\x00b", UnitID: "c"}, + {TaskID: "a", UnitID: "b\x00c"}, {TaskID: "é", UnitID: "unit"}, + } + for i := range expected { + expected[i].FirstSeenBuildID = fmt.Sprintf("build/%d", i) + } + for i := len(expected) - 1; i >= 0; i-- { + require.NoError(t, store.Create(ctx, expected[i])) + } + sort.Slice(expected, func(i, j int) bool { + if expected[i].TaskID != expected[j].TaskID { + return expected[i].TaskID < expected[j].TaskID + } + return expected[i].UnitID < expected[j].UnitID + }) + for _, limit := range []int{1, 3, len(expected) + 1} { + var all []entity.OpenIssue + bounds := storage.OpenIssueRange{Limit: limit} + for { + batch, err := store.List(ctx, bounds) + require.NoError(t, err) + require.LessOrEqual(t, len(batch), limit) + if len(batch) == 0 { + break + } + all = append(all, batch...) + require.LessOrEqual(t, len(all), len(expected)) + last := batch[len(batch)-1] + bounds.After = storage.OpenIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} + } + require.Equal(t, expected, all) + } + changed, err := store.Delete(ctx, expected[0]) + require.NoError(t, err) + require.True(t, changed) + rest, err := store.List(ctx, storage.OpenIssueRange{Limit: len(expected), After: storage.OpenIssueCursor{TaskID: expected[0].TaskID, UnitID: expected[0].UnitID}}) + require.NoError(t, err) + require.Equal(t, expected[1:], rest) + }) + + t.Run("identifier boundaries", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) + require.NoError(t, err) + store := bound.GetOpenIssueStore() + issue := entity.OpenIssue{TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("😀", 256), FirstSeenBuildID: strings.Repeat("😀", 255)} + require.NoError(t, store.Create(ctx, issue)) + got, err := store.Get(ctx, issue.TaskID, issue.UnitID) + require.NoError(t, err) + require.Equal(t, issue, got) + batch, err := store.List(ctx, storage.OpenIssueRange{Limit: 1}) + require.NoError(t, err) + require.Equal(t, []entity.OpenIssue{issue}, batch) + issue.UnitID += "x" + require.Error(t, store.Create(ctx, issue)) + }) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 0c4f8654e..259326d9a 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -115,6 +115,11 @@ func TestMySQLStorage(t *testing.T) { testIssueSummaryStore(t, ctx, factory) }) + t.Run("OpenIssueStore", func(t *testing.T) { + resetStorage(t, db) + testOpenIssueStore(t, ctx, factory) + }) + t.Run("IssueStateList", func(t *testing.T) { resetStorage(t, db) testIssueStateList(t, ctx, factory) @@ -190,6 +195,7 @@ func resetStorage(t *testing.T, db *sql.DB) { for _, statement := range []string{ "TRUNCATE TABLE issue_state", "TRUNCATE TABLE issue_summary", + "TRUNCATE TABLE open_issue", "TRUNCATE TABLE issue_log", "TRUNCATE TABLE issue_tracking", "TRUNCATE TABLE queue_log",