From cc05954068e4bb89e547efa516f1abf15f5c8cba Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 6 Nov 2026 21:36:32 +0000 Subject: [PATCH] feat(stovepipe): persist versioned repair submission state --- stovepipe/entity/BUILD.bazel | 1 + stovepipe/entity/issue_repair.go | 32 ++++ stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 6 +- .../extension/storage/issue_repair_store.go | 35 +++++ stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../storage/mock/issue_repair_store_mock.go | 85 +++++++++++ .../extension/storage/mock/storage_mock.go | 14 ++ stovepipe/extension/storage/mysql/BUILD.bazel | 2 + .../storage/mysql/issue_repair_store.go | 137 +++++++++++++++++ .../storage/mysql/issue_repair_store_test.go | 140 ++++++++++++++++++ .../storage/mysql/schema/issue_repair.sql | 13 ++ stovepipe/extension/storage/mysql/storage.go | 6 + stovepipe/extension/storage/storage.go | 3 + 14 files changed, 473 insertions(+), 3 deletions(-) create mode 100644 stovepipe/entity/issue_repair.go create mode 100644 stovepipe/extension/storage/issue_repair_store.go create mode 100644 stovepipe/extension/storage/mock/issue_repair_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/issue_repair_store.go create mode 100644 stovepipe/extension/storage/mysql/issue_repair_store_test.go create mode 100644 stovepipe/extension/storage/mysql/schema/issue_repair.sql diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index ea53fdf11..8f4fea19f 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_repair.go", "issue_state.go", "issue_summary.go", "issue_tracking_state.go", diff --git a/stovepipe/entity/issue_repair.go b/stovepipe/entity/issue_repair.go new file mode 100644 index 000000000..11b748a76 --- /dev/null +++ b/stovepipe/entity/issue_repair.go @@ -0,0 +1,32 @@ +// 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 + +// IssueRepair is one executor's repair submission state for an issue opening. +// Empty FixID and SkipReason identify pending submission, not backend acceptance. +type IssueRepair struct { + // Issue identifies the issue opening within a queue. + Issue OpenIssue + // ExecutorID is the stable queue-local executor identity. + ExecutorID string + // FixID is the backend request or run ID for accepted work, empty while pending or skipped. + FixID string + // SkipReason is the terminal skip reason, empty while pending or accepted. + SkipReason string + // Metadata contains executor-owned string values; nil and empty maps are equivalent. + Metadata map[string]string + // Version is the positive optimistic-lock version. + Version int32 +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index de5782f47..d224139b5 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_repair_store.go", "issue_state_store.go", "issue_summary_store.go", "issue_tracking_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index c194e43bb..9c6986352 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -8,7 +8,7 @@ This is a separate contract from `submitqueue/extension/storage` — same shape ## Optimistic locking contract -Entities that support concurrent mutation (`Request`, `Build`, `IssueState`, `IssueTrackingState`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [AGENTS.md](../../../AGENTS.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here. +Entities that support concurrent mutation (`Request`, `Build`, `IssueState`, `IssueTrackingState`, `IssueRepair`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [AGENTS.md](../../../AGENTS.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here. ## Read-after-write consistency @@ -36,9 +36,9 @@ Each observation is stored separately, not in a serialized collection. Observati The MySQL primary key contains the queue, build ID, task ID, and unit ID directly as binary strings, preserving byte-exact identity. Their limits are 255, 1020, 255, and 1024 bytes respectively, for a combined maximum of 2554 bytes. Creates and reads reject identifiers exceeding these limits before querying the database; identifiers are never truncated. -`BuildIssue.FixID` associates an observation with an opaque executor-provided repair ID. An empty ID means no association is recorded. `Update` uses the existing entity CAS convention: it guards `oldVersion` and writes `newVersion`, returning `ErrVersionMismatch` for a stale or missing record. The caller owns version arithmetic and repair policy; storage does not enforce a per-build submission limit. Observation identity and membership survive recovery. +`BuildIssue.FixID` is the legacy single-executor repair association. An empty ID means no association is recorded. `Update` uses the existing entity CAS convention: it guards `oldVersion` and writes `newVersion`, returning `ErrVersionMismatch` for a stale or missing record. The caller owns version arithmetic and repair policy; storage does not enforce a per-build submission limit. Observation identity and membership survive recovery. -Existing MySQL deployments must add `fix_id VARCHAR(255) NOT NULL DEFAULT ''` and `version INT NOT NULL DEFAULT 0` to `build_issue` before deploying the new readers and writers. The schema's `CREATE TABLE IF NOT EXISTS` does not alter existing tables. +`IssueRepairStore` retains versioned submission state keyed by the complete opening identity and executor ID. Pending rows have empty fix IDs and skip reasons. Create never overwrites existing state; Update replaces a snapshot guarded by its version. The controller creates pending rows and owns submission outcomes; executors own metadata and must preserve other fields when updating. Both writers reload on version conflicts before reapplying their edits. Fix IDs are not unique: an executor can accept multiple issues in one external request. Executor-owned metadata remains a `map[string]string` in Go and is serialized to the non-null JSON `metadata` column. Nil and empty maps are equivalent; empty metadata is stored as `{}`. String values may contain serialized JSON without interpretation by the store. Apply `mysql/schema/issue_repair.sql` before deploying the new readers and writers. Host adapters own backend-specific dispatch limits. The schema's `CREATE TABLE IF NOT EXISTS` does not alter existing tables. ## Issue state diff --git a/stovepipe/extension/storage/issue_repair_store.go b/stovepipe/extension/storage/issue_repair_store.go new file mode 100644 index 000000000..776bf4ecb --- /dev/null +++ b/stovepipe/extension/storage/issue_repair_store.go @@ -0,0 +1,35 @@ +// 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_repair_store.go -destination=mock/issue_repair_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// IssueRepairStore retains versioned submissions keyed by opening and executor. +// The controller owns FixID and SkipReason; executors own Metadata. Callers must +// preserve other fields and reload on version conflicts before reapplying changes. +type IssueRepairStore interface { + // Get returns ErrNotFound if no repair row exists. + Get(context.Context, entity.OpenIssue, string) (entity.IssueRepair, error) + // Create returns ErrAlreadyExists without overwriting existing state. + Create(context.Context, entity.IssueRepair) error + // Update replaces the record only at oldVersion; callers supply newVersion. + Update(ctx context.Context, repair entity.IssueRepair, oldVersion, newVersion int32) error +} diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index 7e7545c10..5008a24a9 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_repair_store_mock.go", "issue_state_store_mock.go", "issue_summary_store_mock.go", "issue_tracking_store_mock.go", diff --git a/stovepipe/extension/storage/mock/issue_repair_store_mock.go b/stovepipe/extension/storage/mock/issue_repair_store_mock.go new file mode 100644 index 000000000..b758bf7b4 --- /dev/null +++ b/stovepipe/extension/storage/mock/issue_repair_store_mock.go @@ -0,0 +1,85 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: issue_repair_store.go +// +// Generated by this command: +// +// mockgen -source=issue_repair_store.go -destination=mock/issue_repair_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" +) + +// MockIssueRepairStore is a mock of IssueRepairStore interface. +type MockIssueRepairStore struct { + ctrl *gomock.Controller + recorder *MockIssueRepairStoreMockRecorder + isgomock struct{} +} + +// MockIssueRepairStoreMockRecorder is the mock recorder for MockIssueRepairStore. +type MockIssueRepairStoreMockRecorder struct { + mock *MockIssueRepairStore +} + +// NewMockIssueRepairStore creates a new mock instance. +func NewMockIssueRepairStore(ctrl *gomock.Controller) *MockIssueRepairStore { + mock := &MockIssueRepairStore{ctrl: ctrl} + mock.recorder = &MockIssueRepairStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockIssueRepairStore) EXPECT() *MockIssueRepairStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockIssueRepairStore) Create(arg0 context.Context, arg1 entity.IssueRepair) 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 *MockIssueRepairStoreMockRecorder) Create(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockIssueRepairStore)(nil).Create), arg0, arg1) +} + +// Get mocks base method. +func (m *MockIssueRepairStore) Get(arg0 context.Context, arg1 entity.OpenIssue, arg2 string) (entity.IssueRepair, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", arg0, arg1, arg2) + ret0, _ := ret[0].(entity.IssueRepair) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockIssueRepairStoreMockRecorder) Get(arg0, arg1, arg2 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockIssueRepairStore)(nil).Get), arg0, arg1, arg2) +} + +// Update mocks base method. +func (m *MockIssueRepairStore) Update(ctx context.Context, repair entity.IssueRepair, oldVersion, newVersion int32) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Update", ctx, repair, oldVersion, newVersion) + ret0, _ := ret[0].(error) + return ret0 +} + +// Update indicates an expected call of Update. +func (mr *MockIssueRepairStoreMockRecorder) Update(ctx, repair, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockIssueRepairStore)(nil).Update), ctx, repair, oldVersion, newVersion) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index 2b0e19c93..dac18dfd9 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)) } +// GetIssueRepairStore mocks base method. +func (m *MockStorage) GetIssueRepairStore() storage.IssueRepairStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetIssueRepairStore") + ret0, _ := ret[0].(storage.IssueRepairStore) + return ret0 +} + +// GetIssueRepairStore indicates an expected call of GetIssueRepairStore. +func (mr *MockStorageMockRecorder) GetIssueRepairStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueRepairStore", reflect.TypeOf((*MockStorage)(nil).GetIssueRepairStore)) +} + // GetIssueStateStore mocks base method. func (m *MockStorage) GetIssueStateStore() storage.IssueStateStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 45b86d387..1325d7788 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_repair_store.go", "issue_state_store.go", "issue_summary_store.go", "issue_tracking_store.go", @@ -41,6 +42,7 @@ go_test( "build_issue_update_test.go", "build_store_test.go", "issue_log_store_test.go", + "issue_repair_store_test.go", "issue_state_list_test.go", "issue_state_store_test.go", "issue_summary_store_test.go", diff --git a/stovepipe/extension/storage/mysql/issue_repair_store.go b/stovepipe/extension/storage/mysql/issue_repair_store.go new file mode 100644 index 000000000..1ecbaf911 --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_repair_store.go @@ -0,0 +1,137 @@ +// 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" + "encoding/json" + "errors" + "fmt" + + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +type issueRepairStore struct { + db *sql.DB + queue string +} + +// NewIssueRepairStore binds repair submission state to a queue. +func NewIssueRepairStore(db *sql.DB, queue string) storage.IssueRepairStore { + return &issueRepairStore{db: db, queue: queue} +} + +func (s *issueRepairStore) validateKey(issue entity.OpenIssue, executorID string) error { + if s.queue == "" || issue.FirstSeenBuildID == "" || issue.TaskID == "" || executorID == "" || len(executorID) > 128 { + return fmt.Errorf("issue repair requires queue, opening, task, and executor IDs (executor at most 128 bytes)") + } + return validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID) +} + +func (s *issueRepairStore) Get(ctx context.Context, issue entity.OpenIssue, executorID string) (entity.IssueRepair, error) { + if err := s.validateKey(issue, executorID); err != nil { + return entity.IssueRepair{}, err + } + result := entity.IssueRepair{Issue: issue, ExecutorID: executorID} + var metadataJSON []byte + err := s.db.QueryRowContext(ctx, `SELECT fix_id, skip_reason, metadata, version FROM issue_repair + WHERE queue=? AND opening_build_id=? AND task_id=? AND unit_id=? AND executor_id=?`, + s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID, executorID).Scan(&result.FixID, &result.SkipReason, &metadataJSON, &result.Version) + if errors.Is(err, sql.ErrNoRows) { + return entity.IssueRepair{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.IssueRepair{}, fmt.Errorf("read issue repair: %w", err) + } + if err := json.Unmarshal(metadataJSON, &result.Metadata); err != nil { + return entity.IssueRepair{}, fmt.Errorf("decode issue repair metadata: %w", err) + } + if result.Metadata == nil { + result.Metadata = map[string]string{} + } + return result, nil +} + +func (s *issueRepairStore) Create(ctx context.Context, result entity.IssueRepair) error { + if err := s.validateRepair(result); err != nil { + return err + } + metadataJSON, err := marshalRepairMetadata(result.Metadata) + if err != nil { + return fmt.Errorf("encode issue repair metadata: %w", err) + } + _, err = s.db.ExecContext(ctx, `INSERT INTO issue_repair + (queue,opening_build_id,task_id,unit_id,executor_id,fix_id,skip_reason,metadata,version) VALUES (?,?,?,?,?,?,?,?,?)`, + s.queue, result.Issue.FirstSeenBuildID, result.Issue.TaskID, result.Issue.UnitID, result.ExecutorID, result.FixID, result.SkipReason, metadataJSON, result.Version) + if isDuplicateEntry(err) { + return storage.ErrAlreadyExists + } + if err != nil { + return fmt.Errorf("create issue repair: %w", err) + } + return nil +} + +func (s *issueRepairStore) Update(ctx context.Context, repair entity.IssueRepair, oldVersion, newVersion int32) error { + replacement := repair + replacement.Version = newVersion + if err := s.validateRepair(replacement); err != nil { + return err + } + if oldVersion <= 0 { + return fmt.Errorf("issue repair guard version must be positive") + } + metadataJSON, err := marshalRepairMetadata(repair.Metadata) + if err != nil { + return err + } + result, err := s.db.ExecContext(ctx, `UPDATE issue_repair SET fix_id=?, skip_reason=?, metadata=?, version=? + WHERE queue=? AND opening_build_id=? AND task_id=? AND unit_id=? AND executor_id=? AND version=?`, + repair.FixID, repair.SkipReason, metadataJSON, newVersion, + s.queue, repair.Issue.FirstSeenBuildID, repair.Issue.TaskID, repair.Issue.UnitID, repair.ExecutorID, oldVersion) + if err != nil { + return fmt.Errorf("update issue repair: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("read issue repair update count: %w", err) + } + if rows != 1 { + return storage.ErrVersionMismatch + } + return nil +} + +func (s *issueRepairStore) validateRepair(repair entity.IssueRepair) error { + if err := s.validateKey(repair.Issue, repair.ExecutorID); err != nil { + return err + } + if repair.FixID != "" && repair.SkipReason != "" || len(repair.FixID) > 255 || len(repair.SkipReason) > 255 { + return fmt.Errorf("issue repair permits at most one fix ID or skip reason (at most 255 bytes)") + } + if repair.Version <= 0 { + return fmt.Errorf("issue repair version must be positive") + } + return nil +} + +func marshalRepairMetadata(metadata map[string]string) ([]byte, error) { + if metadata == nil { + metadata = map[string]string{} + } + return json.Marshal(metadata) +} diff --git a/stovepipe/extension/storage/mysql/issue_repair_store_test.go b/stovepipe/extension/storage/mysql/issue_repair_store_test.go new file mode 100644 index 000000000..858551fbe --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_repair_store_test.go @@ -0,0 +1,140 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "context" + "testing" + + sqlmock "github.com/DATA-DOG/go-sqlmock" + mysql "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func TestIssueRepairPendingLifecycle(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewIssueRepairStore(db, "queue") + issue := entity.OpenIssue{FirstSeenBuildID: "build", TaskID: "task", UnitID: "unit"} + pending := entity.IssueRepair{Issue: issue, ExecutorID: "executor", Version: 1} + mock.ExpectExec("INSERT INTO issue_repair").WithArgs("queue", "build", "task", "unit", "executor", "", "", []byte("{}"), 1).WillReturnResult(sqlmock.NewResult(1, 1)) + require.NoError(t, store.Create(context.Background(), pending)) + mock.ExpectExec("INSERT INTO issue_repair").WillReturnError(&mysql.MySQLError{Number: 1062}) + require.ErrorIs(t, store.Create(context.Background(), pending), storage.ErrAlreadyExists) + + metadata := map[string]string{"state": `{"attempt":2,"workflow":"run/1"}`} + metadataJSON := []byte(`{"state":"{\"attempt\":2,\"workflow\":\"run/1\"}"}`) + pending.Metadata = metadata + mock.ExpectExec("UPDATE issue_repair").WithArgs("", "", metadataJSON, 2, "queue", "build", "task", "unit", "executor", 1).WillReturnResult(sqlmock.NewResult(0, 1)) + require.NoError(t, store.Update(context.Background(), pending, 1, 2)) + require.Equal(t, int32(1), pending.Version) + + mock.ExpectQuery("SELECT fix_id, skip_reason, metadata, version FROM issue_repair").WithArgs("queue", "build", "task", "unit", "executor").WillReturnRows(sqlmock.NewRows([]string{"fix_id", "skip_reason", "metadata", "version"}).AddRow("", "", metadataJSON, 2)) + actual, err := store.Get(context.Background(), issue, "executor") + require.NoError(t, err) + require.Equal(t, metadata, actual.Metadata) + actual.FixID = "shared-fix" + mock.ExpectExec("UPDATE issue_repair").WithArgs("shared-fix", "", metadataJSON, 3, "queue", "build", "task", "unit", "executor", 2).WillReturnResult(sqlmock.NewResult(0, 1)) + require.NoError(t, store.Update(context.Background(), actual, 2, 3)) + + mock.ExpectQuery("SELECT fix_id, skip_reason, metadata, version FROM issue_repair").WithArgs("queue", "build", "task", "unit", "other").WillReturnRows(sqlmock.NewRows([]string{"fix_id", "skip_reason", "metadata", "version"})) + _, err = store.Get(context.Background(), issue, "other") + require.ErrorIs(t, err, storage.ErrNotFound) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestIssueRepairUpdateConflicts(t *testing.T) { + for _, skipped := range []bool{false, true} { + t.Run(map[bool]string{false: "submitted", true: "skipped"}[skipped], func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewIssueRepairStore(db, "queue") + repair := entity.IssueRepair{Issue: entity.OpenIssue{FirstSeenBuildID: "build", TaskID: "task"}, ExecutorID: "executor", Version: 1, FixID: "fix"} + if skipped { + repair.FixID, repair.SkipReason = "", "unsupported" + } + mock.ExpectExec("UPDATE issue_repair").WithArgs(repair.FixID, repair.SkipReason, []byte("{}"), 2, "queue", "build", "task", "", "executor", 1).WillReturnResult(sqlmock.NewResult(0, 1)) + require.NoError(t, store.Update(context.Background(), repair, 1, 2)) + mock.ExpectExec("UPDATE issue_repair").WithArgs(repair.FixID, repair.SkipReason, []byte("{}"), 3, "queue", "build", "task", "", "executor", 2).WillReturnResult(sqlmock.NewResult(0, 0)) + require.ErrorIs(t, store.Update(context.Background(), repair, 2, 3), storage.ErrVersionMismatch) + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueRepairEmptyMetadata(t *testing.T) { + for _, tt := range []struct { + name string + metadata map[string]string + stored string + }{ + {name: "nil", stored: "{}"}, + {name: "empty", metadata: map[string]string{}, stored: "{}"}, + {name: "JSON null", stored: "null"}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewIssueRepairStore(db, "queue") + issue := entity.OpenIssue{FirstSeenBuildID: "build", TaskID: "task"} + repair := entity.IssueRepair{Issue: issue, ExecutorID: "executor", Metadata: tt.metadata, Version: 1} + mock.ExpectExec("INSERT INTO issue_repair").WithArgs("queue", "build", "task", "", "executor", "", "", []byte("{}"), 1).WillReturnResult(sqlmock.NewResult(1, 1)) + require.NoError(t, store.Create(context.Background(), repair)) + require.Equal(t, tt.metadata, repair.Metadata) + mock.ExpectQuery("SELECT fix_id, skip_reason, metadata, version FROM issue_repair").WillReturnRows(sqlmock.NewRows([]string{"fix_id", "skip_reason", "metadata", "version"}).AddRow("", "", tt.stored, 1)) + actual, err := store.Get(context.Background(), issue, "executor") + require.NoError(t, err) + require.Equal(t, map[string]string{}, actual.Metadata) + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueRepairInvalidMetadata(t *testing.T) { + for _, stored := range []string{`{`, `[]`, `{"state":{"attempt":2}}`} { + t.Run(stored, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewIssueRepairStore(db, "queue") + mock.ExpectQuery("SELECT fix_id, skip_reason, metadata, version FROM issue_repair").WillReturnRows(sqlmock.NewRows([]string{"fix_id", "skip_reason", "metadata", "version"}).AddRow("fix", "", stored, 1)) + _, err = store.Get(context.Background(), entity.OpenIssue{FirstSeenBuildID: "build", TaskID: "task"}, "executor") + require.Error(t, err) + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestIssueRepairValidation(t *testing.T) { + db, _, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewIssueRepairStore(db, "queue") + for _, repair := range []entity.IssueRepair{ + {}, + {Issue: entity.OpenIssue{FirstSeenBuildID: "build", TaskID: "task"}, ExecutorID: "executor"}, + {Issue: entity.OpenIssue{FirstSeenBuildID: "build", TaskID: "task"}, ExecutorID: "executor", FixID: "fix", SkipReason: "skip", Version: 1}, + } { + require.Error(t, store.Create(context.Background(), repair)) + } + repair := entity.IssueRepair{Issue: entity.OpenIssue{FirstSeenBuildID: "build", TaskID: "task"}, ExecutorID: "executor", Version: 1} + require.Error(t, store.Update(context.Background(), repair, 0, 2)) + require.Error(t, store.Update(context.Background(), repair, 1, 0)) +} diff --git a/stovepipe/extension/storage/mysql/schema/issue_repair.sql b/stovepipe/extension/storage/mysql/schema/issue_repair.sql new file mode 100644 index 000000000..cbc35d09b --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/issue_repair.sql @@ -0,0 +1,13 @@ +-- Per-opening, per-executor submission state. Empty fix_id and skip_reason mean pending. +CREATE TABLE IF NOT EXISTS issue_repair ( + queue VARBINARY(255) NOT NULL, + opening_build_id VARBINARY(1020) NOT NULL, + task_id VARBINARY(255) NOT NULL, + unit_id VARBINARY(1024) NOT NULL, + executor_id VARBINARY(128) NOT NULL, + fix_id VARCHAR(255) NOT NULL DEFAULT '', + skip_reason VARCHAR(255) NOT NULL DEFAULT '', + metadata JSON NOT NULL, + version INT NOT NULL, + PRIMARY KEY (queue, opening_build_id, task_id, unit_id, executor_id) +) ENGINE=InnoDB; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index ea6207fe3..3bef124d3 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -46,6 +46,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { } queueScope := s.scope.Tagged(map[string]string{"queue": queueName}) return &mysqlStorage{ + issueRepairStore: NewIssueRepairStore(s.db, queueName), requestStore: NewRequestStore(s.db, queueScope.SubScope("request_store"), queueName), requestURIStore: NewRequestURIStore(s.db, queueScope.SubScope("request_uri_store"), queueName), requestLogStore: NewRequestLogStore(s.db, queueScope.SubScope("request_log_store"), queueName), @@ -73,6 +74,7 @@ func (s *Storage) Close() error { // mysqlStorage is the queue-scoped store aggregate returned by For. type mysqlStorage struct { + issueRepairStore storage.IssueRepairStore requestStore storage.RequestStore requestURIStore storage.RequestURIStore requestLogStore storage.RequestLogStore @@ -168,3 +170,7 @@ func (f *mysqlStorage) GetOpenIssueStore() storage.OpenIssueStore { func (f *mysqlStorage) GetIssueTrackingStore() storage.IssueTrackingStore { return f.issueTrackingStore } + +func (f *mysqlStorage) GetIssueRepairStore() storage.IssueRepairStore { + return f.issueRepairStore +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index b69297ae9..95564443d 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -69,6 +69,9 @@ type Factory interface { // to that queue: entity arguments whose queue disagrees with the binding are // rejected, and reads never surface another queue's records. type Storage interface { + // GetIssueRepairStore returns per-executor repair dispositions. + GetIssueRepairStore() IssueRepairStore + // GetRequestStore returns the RequestStore instance. GetRequestStore() RequestStore