From cc05954068e4bb89e547efa516f1abf15f5c8cba Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 6 Nov 2026 21:36:32 +0000 Subject: [PATCH 1/4] 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 From 6b1397531c290367326fa6e98ee990a0174e18ae Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 6 Nov 2026 22:09:07 +0000 Subject: [PATCH 2/4] feat(stovepipe): define named batch repair executor contracts Summary: Let queues register multiple repair backends and let each backend group its own issue openings. Submission receipts identify accepted work or a terminal skip for each input item; metadata stays in versioned storage. This contract change builds on repair-state persistence. Controller and reference-server callers migrate in the next PR in the stack. Test Plan: `./tool/bazel build //stovepipe/extension/fixer/...` builds the contract, generated mock, and no-op implementation. `make mocks` regenerated the mock, and lint, tidy, and Gazelle checks verify the generated sources and build metadata. Caller compilation requires the next PR. Revert Plan: Revert the dependent dispatch change before reverting this contract change. --- stovepipe/extension/fixer/fixer.go | 40 +++++++++++++++----- stovepipe/extension/fixer/mock/fixer_mock.go | 8 ++-- stovepipe/extension/fixer/noop/noop.go | 8 +++- 3 files changed, 40 insertions(+), 16 deletions(-) diff --git a/stovepipe/extension/fixer/fixer.go b/stovepipe/extension/fixer/fixer.go index 533849367..cf4a79a77 100644 --- a/stovepipe/extension/fixer/fixer.go +++ b/stovepipe/extension/fixer/fixer.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package fixer defines repair submission for an open issue. +// Package fixer defines repair submission for open issues. package fixer //go:generate mockgen -source=fixer.go -destination=mock/fixer_mock.go -package=mock @@ -23,22 +23,42 @@ import ( "github.com/uber/submitqueue/stovepipe/entity" ) -// RepairExecutor submits repair work for one open issue within a configured queue. -// Implementations must be concurrency-safe and return promptly after submission. +// RepairExecutor chooses how to group repair work. Implementations must be concurrency-safe. type RepairExecutor interface { - // Submit returns a nonempty opaque fix ID on acceptance, without establishing recovery. - // An ID returned alongside an error still identifies the repair attempt. - // Errors without an ID may follow acceptance; repeated calls can start additional work. - Submit(context.Context, entity.OpenIssue) (string, error) + // Submit returns per-item backend acceptance or terminal skips, not repair completion. + // Valid submissions returned alongside errors are persisted; omitted items remain pending. + // Retries may include subsets or additional items; implementations deduplicate by opening. + // Pending rows exist before Submit. Executors manage metadata through injected storage, + // preserving submission fields and reloading on version conflicts. + Submit(context.Context, []entity.OpenIssue) ([]Submission, error) } -// Config identifies the queue whose repair executor is requested. +// Submission associates one input item with accepted repair work or a deliberate skip. +// Acceptance does not indicate repair completion or success. +type Submission struct { + // Issue identifies the input opening, including its first-seen build. + Issue entity.OpenIssue + // FixID is the backend request or run ID for accepted work; multiple items may share it. + FixID string + // SkipReason is nonempty for a terminal skip. Exactly one of FixID and SkipReason is required. + SkipReason string +} + +// Registration binds a stable queue-local identity to an executor. +type Registration struct { + // ID is durable across configuration changes and must not be reused for another backend. + ID string + // Executor implements the registered repair policy. + Executor RepairExecutor +} + +// Config identifies a validation queue. type Config struct { // QueueName identifies the validation queue. QueueName string } -// Factory resolves queue-bound executors; routing belongs to service wiring. +// Factory resolves all registered executors; an empty slice disables dispatch. type Factory interface { - For(Config) (RepairExecutor, error) + For(Config) ([]Registration, error) } diff --git a/stovepipe/extension/fixer/mock/fixer_mock.go b/stovepipe/extension/fixer/mock/fixer_mock.go index 25e4779f5..9e9912274 100644 --- a/stovepipe/extension/fixer/mock/fixer_mock.go +++ b/stovepipe/extension/fixer/mock/fixer_mock.go @@ -43,10 +43,10 @@ func (m *MockRepairExecutor) EXPECT() *MockRepairExecutorMockRecorder { } // Submit mocks base method. -func (m *MockRepairExecutor) Submit(arg0 context.Context, arg1 entity.OpenIssue) (string, error) { +func (m *MockRepairExecutor) Submit(arg0 context.Context, arg1 []entity.OpenIssue) ([]fixer.Submission, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "Submit", arg0, arg1) - ret0, _ := ret[0].(string) + ret0, _ := ret[0].([]fixer.Submission) ret1, _ := ret[1].(error) return ret0, ret1 } @@ -82,10 +82,10 @@ func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { } // For mocks base method. -func (m *MockFactory) For(arg0 fixer.Config) (fixer.RepairExecutor, error) { +func (m *MockFactory) For(arg0 fixer.Config) ([]fixer.Registration, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "For", arg0) - ret0, _ := ret[0].(fixer.RepairExecutor) + ret0, _ := ret[0].([]fixer.Registration) ret1, _ := ret[1].(error) return ret0, ret1 } diff --git a/stovepipe/extension/fixer/noop/noop.go b/stovepipe/extension/fixer/noop/noop.go index 7ab458c96..88e128c06 100644 --- a/stovepipe/extension/fixer/noop/noop.go +++ b/stovepipe/extension/fixer/noop/noop.go @@ -29,6 +29,10 @@ func New() fixer.RepairExecutor { return executor{} } -func (executor) Submit(_ context.Context, issue entity.OpenIssue) (string, error) { - return "noop/" + issue.FirstSeenBuildID, nil +func (executor) Submit(_ context.Context, issues []entity.OpenIssue) ([]fixer.Submission, error) { + submissions := make([]fixer.Submission, 0, len(issues)) + for _, issue := range issues { + submissions = append(submissions, fixer.Submission{Issue: issue, FixID: "noop/" + issue.FirstSeenBuildID}) + } + return submissions, nil } From 1148984e6bd154728b97c93f4c2022d5416af8f2 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 6 Nov 2026 22:09:59 +0000 Subject: [PATCH 3/4] feat(stovepipe): dispatch pending issues to named batch repair executors Summary: Offer each registered executor all outstanding issue openings in one batch. Persist pending rows before dispatch and retain per-item acceptance or skips while preserving executor metadata through versioned updates. Restore durable history on replay and preserve independent error causes and valid partial receipts. This builds on the batch executor contract. The existing DLQ behavior remains in place until the next PR migrates recovery to per-executor repair records. Test Plan: `./tool/bazel test //stovepipe/... //service/stovepipe/server:go_default_test --test_output=errors` exercises 28 targets covering batching across pages, multiple executors, partial acceptance, failed writes, version conflicts, history replay, and preservation of retry classification. Lint, tidy, and Gazelle checks verify the repository metadata. Revert Plan: Revert the dependent DLQ change first, then revert this PR together with its executor-contract prerequisite. --- doc/rfc/stovepipe/request-log.md | 4 +- service/stovepipe/README.md | 2 +- service/stovepipe/server/main.go | 4 +- service/stovepipe/server/main_test.go | 27 +- stovepipe/controller/fix/BUILD.bazel | 3 +- stovepipe/controller/fix/README.md | 14 +- stovepipe/controller/fix/delivery_test.go | 4 +- stovepipe/controller/fix/fix.go | 106 ++--- stovepipe/controller/fix/fix_test.go | 536 +++++++++++++--------- stovepipe/controller/fix/history.go | 26 ++ stovepipe/controller/fix/history_test.go | 64 --- stovepipe/controller/fix/submission.go | 167 +++++++ stovepipe/core/requestlog/materializer.go | 2 + 13 files changed, 606 insertions(+), 353 deletions(-) create mode 100644 stovepipe/controller/fix/submission.go diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index 0b9075070..04b0ed8e3 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -143,7 +143,7 @@ State entries retain the selected baseline URI when known, including an empty UR Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording. Its `project_fact_count` and `broken_project_fact_count` metadata contain the total and the number with degree above zero, counted from persisted facts after their indexes are durable. Project completion projection requires both counts and assigns them together in `RequestSummary` from the retained event. -`build_analysis_completed`, `issues_reconciled`, `fix_submission_skipped`, and `fix_submission_abandoned` use the build ID as their occurrence key. `fix_submitted` uses a digest of the executor fix ID to fit the storage key limit, retaining the verbatim fix ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp. +`build_analysis_completed` and `issues_reconciled` use the build ID as their occurrence key. `fix_submission_abandoned` uses the build ID. Per-item `fix_submitted` and `fix_submission_skipped` events use a digest of opening build, task, unit, and executor identity, retaining the verbatim fix ID or skip reason and executor ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp. Analysis and repair events do not advance Request state or alter validation facts. Analysis completion is retained only when the Request records the analyzed terminal build as its winning outcome, immediately before the final issue-tracking handoff. Its issue count and timestamp are an immutable snapshot of the first retained completion, including zero for an empty result. Retries still persist detected issues and publish the handoff, but reuse that snapshot if later detector results differ; count discrepancies are logged rather than treated as history conflicts. Other semantic conflicts remain errors. `issues_reconciled` records closure mode `reconciled`, `suppressed_cancelled`, or `suppressed_task_wide_issue`; suppression does not mean reconciliation failed. Skipped repair uses reason `no_eligible_issue`; abandoned repair uses `acceptance_unknown`. Raw dependency errors remain in operational logs. @@ -238,7 +238,7 @@ The `project_facts_recorded` and `record_abandoned` events project completion of | Build analysis DLQ | Retain `build_analysis_abandoned` without rerunning detection or changing validation state. | One stable event per build deduplicates recovery deliveries; only state reads and history retention retry. | | Issue tracking DLQ | Protect the current cursor from delayed writes, retain its issue history and `issue_tracking_abandoned`, then mark the pass abandoned. | A stable hash of the notification ID deduplicates the event. Recovery retains accepted issue changes and never processes the remaining range. | | Issue tracking | Finalize the durable checkpoint, then retain `issues_reconciled`. | Reconstruct history from the finalized checkpoint before it can be replaced by another pass. | -| Fix | Associate the returned fix ID, then retain `fix_submitted`; retain `fix_submission_skipped` when no issue is eligible. | Ensure submitted history from any existing association before skipping executor work. | +| Fix | Retain each executor disposition, then retain its submitted or skipped event. | Restore every durable disposition event before skipping that item for that executor. | | Fix DLQ | Repair submitted or skipped history from current durable state, or retain `fix_submission_abandoned` when acceptance is unknown. | Only read sources and retain history; never call the executor or change issue state. | | Reconciler | CAS an unrecoverable non-terminal Request to failed, then retain failed. | An existing terminal Request is repaired from its persisted outcome without relabeling it. | diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md index 639aec723..3aace589e 100644 --- a/service/stovepipe/README.md +++ b/service/stovepipe/README.md @@ -26,7 +26,7 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `queu - **`inMemoryCounter`** — a process-local `counter.Counter` for sequence numbers; not durable. A real deployment uses a persistent implementation (e.g. `platform/extension/counter/mysql`). - **`fakeSourceControlFactory`** — seeds each queue with a deterministic single-commit history so ingest resolves a stable head URI (and re-ingesting the same queue exercises the dedup path). A real deployment supplies a VCS-backed `sourcecontrol.Factory`, which is also where a queue's promotion ref is resolved. The fake has no ref to move, so a promotion locally shows up only in the record consumer's logs. -- **`noopRepairExecutorFactory`** — accepts repairs without launching work and returns `noop/`. The fix controller stores that ID, consuming the build's one-repair allowance. Replacing this factory does not retry builds that already have a no-op association. The default no-op issue detector reports no failures, so the reference stack does not normally reach repair submission. +- **`noopRepairExecutorFactory`** — accepts repairs without launching work and returns `noop/`. The fix controller stores per-item submission decisions under registration ID `noop`. A different registration ID has independent eligibility. The default no-op issue detector reports no failures, so the reference stack does not normally reach repair submission. ## Layout diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 9d388c2df..3d9696c4f 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -97,8 +97,8 @@ func (fakeBuildRunnerFactory) For(cfg buildrunner.Config) (buildrunner.BuildRunn type noopRepairExecutorFactory struct{} -func (noopRepairExecutorFactory) For(fixer.Config) (fixer.RepairExecutor, error) { - return fixernoop.New(), nil +func (noopRepairExecutorFactory) For(fixer.Config) ([]fixer.Registration, error) { + return []fixer.Registration{{ID: "noop", Executor: fixernoop.New()}}, nil } // hookResolver sends every event to the no-op hook. Which hooks an event goes to is host diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index abea3bca0..72d5852ad 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -136,7 +136,7 @@ func TestFixRegistration(t *testing.T) { assert.Contains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyFix)) } -func TestRegisteredFixControllerRecordsNoopRepairOncePerBuild(t *testing.T) { +func TestRegisteredFixControllerRetainsPerExecutorDisposition(t *testing.T) { const queueName = "validation/main" const buildID = "build/1" ctrl := gomock.NewController(t) @@ -181,14 +181,23 @@ func TestRegisteredFixControllerRecordsNoopRepairOncePerBuild(t *testing.T) { TaskID: observation.TaskID, UnitID: observation.UnitID, Status: entity.IssueStatusOpen, FirstSeenBuildID: buildID, LastProcessedBuildID: buildID, LastChangedBuildID: buildID, ChangedAtMs: 1000, Version: 1, - }, nil) - observations.EXPECT().Update(gomock.Any(), buildID, gomock.Any(), int32(1), int32(2)).DoAndReturn( - func(_ context.Context, _ string, updated entity.BuildIssue, _, newVersion int32) error { - require.Equal(t, "noop/"+buildID, updated.FixID) - observation = updated - observation.Version = newVersion - return nil - }) + }, nil).AnyTimes() + repairs := storagemock.NewMockIssueRepairStore(ctrl) + store.EXPECT().GetIssueRepairStore().Return(repairs).AnyTimes() + var repair entity.IssueRepair + repairs.EXPECT().Get(gomock.Any(), gomock.Any(), "noop").DoAndReturn(func(context.Context, entity.OpenIssue, string) (entity.IssueRepair, error) { + if repair.Version == 0 { + return entity.IssueRepair{}, storage.ErrNotFound + } + return repair, nil + }).Times(3) + repairs.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, value entity.IssueRepair) error { repair = value; return nil }) + repairs.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn(func(_ context.Context, value entity.IssueRepair, _, newVersion int32) error { + require.Empty(t, repair.FixID) + repair = value + repair.Version = newVersion + return nil + }) registry, err := newTopicRegistry(nil, "subscriber") require.NoError(t, err) diff --git a/stovepipe/controller/fix/BUILD.bazel b/stovepipe/controller/fix/BUILD.bazel index 863978491..dd5527c58 100644 --- a/stovepipe/controller/fix/BUILD.bazel +++ b/stovepipe/controller/fix/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "dlq.go", "fix.go", "history.go", + "submission.go", ], importpath = "github.com/uber/submitqueue/stovepipe/controller/fix", visibility = ["//visibility:public"], @@ -39,6 +40,7 @@ go_test( "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", + "//platform/errs/generic:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", @@ -51,6 +53,5 @@ go_test( "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_mock//gomock:go_default_library", "@org_uber_go_zap//:go_default_library", - "@org_uber_go_zap//zaptest/observer:go_default_library", ], ) diff --git a/stovepipe/controller/fix/README.md b/stovepipe/controller/fix/README.md index 9ed812de0..8de074c34 100644 --- a/stovepipe/controller/fix/README.md +++ b/stovepipe/controller/fix/README.md @@ -1,13 +1,17 @@ # Fix submission -The controller consumes a `FixRequest` containing a queue name and build ID. It pages through a build's durable issue records and skips submission when any record has a fix ID, including recovered issues. Otherwise, it selects one currently open issue whose first-seen build matches this build and calls `RepairExecutor.Submit`. +The controller consumes a build-scoped `FixRequest`. It reads every page of the build's observations, selects still-open issues whose opening build matches, and offers each registered executor all of its outstanding items in one `Submit` call. Different executors may choose different groupings. Their stable, queue-local registration IDs are part of the durable identity; an empty registry disables submission. -A returned fix ID is associated with the issue through `BuildIssueStore.Update`, using the version read during selection. A concurrent association is preserved. A submission error without an ID leaves the issue eligible for normal queue retries; an ID returned alongside an error is retained. +The executor returns per-item `Submission` values once the backend accepts work or the executor deliberately skips it. `FixID` identifies the backend request or run; it does not indicate repair completion or success. Multiple items and successive calls may share a fix ID. Each returned disposition is retained independently in `issue_repair`, keyed by queue, opening build, task, unit, and executor. Accepted submissions are retained even alongside an error. Omitted items remain pending; omitting items without an error violates the contract. Invalid, foreign, or duplicate submissions fail processing. One executor's error does not suppress another executor's submission. -Issue tracking calls [`fixrequest.Publish`](../../core/fixrequest/publish.go) after completing a nonempty opening pass and persisting its issue state and history. Publications share a build partition, while the message identity includes the pass notification ID so a later pass can trigger another evaluation. A pass with no eligible issue completes without submitting a repair; a recorded fix ID prevents later passes from submitting another repair for the build. +Before calling an executor, the controller ensures a pending `issue_repair` row exists for every offered opening and executor. Empty `FixID` and `SkipReason` mean pending; a row's existence does not establish backend acceptance. Returned submissions update those fields after acceptance or a deliberate skip. Existing pending rows are offered again, while submitted and skipped rows restore history without resubmission. -The one-per-build limit is controller policy. Submission precedes persistence, so overlapping deliveries, crashes, and failed writes can still produce duplicate executor calls. Issue recovery remains separate from repair association. +`IssueRepair.Metadata` is an executor-owned `map[string]string` stored in the JSON `metadata` column. Executors read and update it through injected storage; it is absent from `Submission`. String values may contain serialized JSON, and nil and empty maps are equivalent. Storage uses versioned replacement snapshots: both metadata writers and the controller must preserve other fields and reload before reapplying an edit after `ErrVersionMismatch`. The controller reloads the latest row before recording a receipt, preserving metadata written during submission. Recorded submission outcomes are never replaced by conflicting receipts. -A durable association retains `fix_submitted` in the owning Request's history, with build, task, unit, and fix IDs. A digest of the fix ID identifies the occurrence and fits the history key limit; metadata retains the verbatim fix ID. An existing association repairs missing history before skipping submission, including when the issue has since recovered. Without an eligible issue, the controller retains `fix_submission_skipped` once per build with reason `no_eligible_issue`. These entries do not change Request outcome or establish issue recovery. +There is no controller-wide repair cap or batch identity. Executor calls can overlap; pending rows are not exclusive dispatch claims. Implementations must deduplicate by issue opening even after a lost response or a failed local write. Build-level limits belong to host adapters. The legacy `build_issue.fix_id` field is preserved but does not gate this controller. + +Each durable disposition restores its request-history event before being skipped on replay, including after issue closure. Events retain build, task, unit, executor, and accepted fix ID or skip reason. A digest of the opening and executor identifies the occurrence. The legacy request-issue scalar fix field exposes the bytewise least accepted fix ID; complete associations remain in history and `issue_repair`. Repair submission does not establish issue recovery. + +Issue tracking publishes another nudge after each nonempty opening pass, so late issues are offered independently of earlier submissions. Queue redelivery rederives pending work and may offer a smaller list. Reopening in another build creates a new identity. A fix ID must reflect actual backend acceptance for every item carrying it; returning an old ID alone does not deliver new work. The DLQ controller reads the same durable decision without calling the executor. It repairs submitted or skipped history when possible; otherwise, an eligible issue without a fix ID retains `fix_submission_abandoned` once per build with reason `acceptance_unknown`. An executor error without an ID can follow acceptance, so this event does not claim rejection. Source and history errors retry. Unusable payloads, unresolvable queues, and missing or mismatched sources are logged and acknowledged. The reference host registers recovery with unlimited, always-retryable DLQ processing to preserve history repair through a storage outage. diff --git a/stovepipe/controller/fix/delivery_test.go b/stovepipe/controller/fix/delivery_test.go index 43efff41c..b5341a6e5 100644 --- a/stovepipe/controller/fix/delivery_test.go +++ b/stovepipe/controller/fix/delivery_test.go @@ -24,6 +24,7 @@ import ( consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" "go.uber.org/mock/gomock" ) @@ -46,8 +47,7 @@ func TestProcessSubmitsBuildIssue(t *testing.T) { controller, m := newTestController(t) issue := testIssue() m.expectIssue(issue) - m.executor.EXPECT().Submit(gomock.Any(), issue).Return("attempt", nil) - m.observations.EXPECT().Update(gomock.Any(), testBuildID, issueWithFixID(issue, "attempt"), int32(0), int32(1)).Return(nil) + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).Return(accepted([]entity.OpenIssue{issue}, "attempt"), nil) require.NoError(t, controller.Process(context.Background(), m.delivery(repairMessage(t)))) } diff --git a/stovepipe/controller/fix/fix.go b/stovepipe/controller/fix/fix.go index cca9abc54..cc1ba8af1 100644 --- a/stovepipe/controller/fix/fix.go +++ b/stovepipe/controller/fix/fix.go @@ -22,7 +22,7 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" - "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/fixer" @@ -65,72 +65,42 @@ func NewController( func (c *Controller) processBuild(ctx context.Context, queueName, buildID string) error { store, err := c.stores.For(storage.Config{QueueName: queueName}) if err != nil { - return fmt.Errorf("failed to resolve repair storage: %w", err) - } - observation, found, err := findIssueToFix(ctx, store.GetBuildIssueStore(), store.GetIssueStateStore(), buildID) - if err != nil { - return fmt.Errorf("failed to select an issue for build %q: %w", buildID, err) + return fmt.Errorf("resolve repair storage: %w", err) } request, err := loadRequestForBuild(ctx, store, queueName, buildID) if err != nil { return err } - if observation.FixID != "" { - return persistFixSubmittedLog(ctx, c.materializer, store, request, buildID, observation) - } - if !found { - return persistFixSubmissionDecision(ctx, c.materializer, store, request, buildID, - entity.RequestEventFixSubmissionSkipped, "no_eligible_issue") - } - - executor, err := c.executors.For(fixer.Config{QueueName: queueName}) + registrations, err := c.executors.For(fixer.Config{QueueName: queueName}) if err != nil { - return fmt.Errorf("failed to resolve repair executor: %w", err) + return fmt.Errorf("resolve repair executors: %w", err) } - - op := metrics.Begin(c.metricsScope, "submit", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) - issue := entity.OpenIssue{TaskID: observation.TaskID, UnitID: observation.UnitID, FirstSeenBuildID: buildID} - fixID, err := executor.Submit(ctx, issue) - if err == nil && fixID == "" { - err = fmt.Errorf("repair executor returned an empty fix ID") + if err := validateRegistrations(registrations); err != nil { + return err } - op.Complete(err) + observations, issues, err := collectIssues(ctx, store, buildID) if err != nil { - c.logger.Errorw("repair submission failed", - "queue", queueName, "build_id", buildID, - "task_id", issue.TaskID, "unit_id", issue.UnitID, - "fix_id", fixID, "error", err) - if fixID == "" { - return fmt.Errorf("failed to submit repair for build %q: %w", buildID, err) + return err + } + var failures []error + for _, registration := range registrations { + if err := c.submitIssues(ctx, store, request, buildID, registration, observations, issues); err != nil { + c.logger.Errorw("repair executor failed", "queue", queueName, "build_id", buildID, "executor_id", registration.ID, "error", err) + failures = append(failures, fmt.Errorf("executor %q: %w", registration.ID, err)) } - } else { - c.logger.Infow("repair submitted", - "queue", queueName, "build_id", buildID, - "task_id", issue.TaskID, "unit_id", issue.UnitID, "fix_id", fixID) } + return errs.Group(failures...) +} - // Submission precedes persistence; a crash or write failure can allow duplicates. - updated := observation - updated.FixID = fixID - observations := store.GetBuildIssueStore() - if err := observations.Update(ctx, buildID, updated, observation.Version, observation.Version+1); err != nil { - if !errors.Is(err, storage.ErrVersionMismatch) { - return fmt.Errorf("failed to associate repair with build issue: %w", err) - } - current, readErr := observations.Get(ctx, buildID, observation.TaskID, observation.UnitID) - if readErr != nil { - return fmt.Errorf("failed to reload build issue after concurrent update: %w", readErr) - } - if current.FixID == "" { - return err +func validateRegistrations(registrations []fixer.Registration) error { + seen := map[string]bool{} + for _, registration := range registrations { + if registration.ID == "" || len(registration.ID) > 128 || registration.Executor == nil || seen[registration.ID] { + return fmt.Errorf("repair executors require unique nonempty IDs (at most 128 bytes) and implementations") } - c.logger.Warnw("preserving previously recorded repair association", - "queue", queueName, "build_id", buildID, - "task_id", issue.TaskID, "unit_id", issue.UnitID, - "fix_id", current.FixID, "submitted_fix_id", fixID) - updated.FixID = current.FixID + seen[registration.ID] = true } - return persistFixSubmittedLog(ctx, c.materializer, store, request, buildID, updated) + return nil } func (c *Controller) Name() string { return "fix" } @@ -139,6 +109,36 @@ func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey } func (c *Controller) ConsumerGroup() string { return c.consumerGroup } +func collectIssues(ctx context.Context, store storage.Storage, buildID string) ([]entity.BuildIssue, []entity.OpenIssue, error) { + bounds := storage.BuildIssueRange{Limit: buildIssuePageSize} + var observations []entity.BuildIssue + var eligible []entity.OpenIssue + for { + page, err := store.GetBuildIssueStore().List(ctx, buildID, bounds) + if err != nil { + return nil, nil, err + } + if len(page) == 0 { + return observations, eligible, nil + } + for _, record := range page { + observations = append(observations, record) + issue, err := store.GetIssueStateStore().Get(ctx, record.TaskID, record.UnitID) + if errors.Is(err, storage.ErrNotFound) { + continue + } + if err != nil { + return nil, nil, err + } + if issue.Status == entity.IssueStatusOpen && issue.FirstSeenBuildID == buildID { + eligible = append(eligible, entity.OpenIssue{TaskID: record.TaskID, UnitID: record.UnitID, FirstSeenBuildID: buildID}) + } + } + last := page[len(page)-1] + bounds.After = storage.BuildIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} + } +} + func findIssueToFix(ctx context.Context, observations storage.BuildIssueStore, issues storage.IssueStateStore, buildID string) (entity.BuildIssue, bool, error) { bounds := storage.BuildIssueRange{Limit: buildIssuePageSize} var candidate entity.BuildIssue diff --git a/stovepipe/controller/fix/fix_test.go b/stovepipe/controller/fix/fix_test.go index 84d325361..4e2ebd264 100644 --- a/stovepipe/controller/fix/fix_test.go +++ b/stovepipe/controller/fix/fix_test.go @@ -16,6 +16,10 @@ package fix import ( "context" + "fmt" + "maps" + "math" + "sort" "testing" "github.com/stretchr/testify/assert" @@ -23,6 +27,8 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" + genericerrs "github.com/uber/submitqueue/platform/errs/generic" + "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/fixer" @@ -31,7 +37,6 @@ import ( storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" "go.uber.org/mock/gomock" "go.uber.org/zap" - "go.uber.org/zap/zaptest/observer" ) const ( @@ -52,13 +57,15 @@ type controllerMocks struct { observations *storagemock.MockBuildIssueStore executors *fixermock.MockFactory executor *fixermock.MockRepairExecutor - logs *observer.ObservedLogs + repairs map[entity.OpenIssue]map[string]entity.IssueRepair + results *storagemock.MockIssueRepairStore + repairFailure *error + updateFailure *error } func newTestController(t *testing.T) (*Controller, controllerMocks) { t.Helper() ctrl := gomock.NewController(t) - core, logs := observer.New(zap.InfoLevel) m := controllerMocks{ ctrl: ctrl, stores: storagemock.NewMockFactory(ctrl), @@ -67,11 +74,49 @@ func newTestController(t *testing.T) (*Controller, controllerMocks) { observations: storagemock.NewMockBuildIssueStore(ctrl), executors: fixermock.NewMockFactory(ctrl), executor: fixermock.NewMockRepairExecutor(ctrl), - logs: logs, } m.store.EXPECT().GetIssueStateStore().Return(m.issues).AnyTimes() m.store.EXPECT().GetBuildIssueStore().Return(m.observations).AnyTimes() + m.repairs = map[entity.OpenIssue]map[string]entity.IssueRepair{} + m.repairFailure = new(error) + m.results = storagemock.NewMockIssueRepairStore(ctrl) + m.store.EXPECT().GetIssueRepairStore().Return(m.results).AnyTimes() + m.results.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.OpenIssue, id string) (entity.IssueRepair, error) { + if result, ok := m.repairs[issue][id]; ok { + result.Metadata = maps.Clone(result.Metadata) + return result, nil + } + return entity.IssueRepair{}, storage.ErrNotFound + }).AnyTimes() + m.results.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, result entity.IssueRepair) error { + if *m.repairFailure != nil { + return *m.repairFailure + } + if _, ok := m.repairs[result.Issue][result.ExecutorID]; ok { + return storage.ErrAlreadyExists + } + if m.repairs[result.Issue] == nil { + m.repairs[result.Issue] = map[string]entity.IssueRepair{} + } + m.repairs[result.Issue][result.ExecutorID] = result + return nil + }).AnyTimes() + m.updateFailure = new(error) + m.results.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, replacement entity.IssueRepair, oldVersion, newVersion int32) error { + if *m.updateFailure != nil { + return *m.updateFailure + } + current, exists := m.repairs[replacement.Issue][replacement.ExecutorID] + if !exists || current.Version != oldVersion { + return storage.ErrVersionMismatch + } + replacement.Version = newVersion + replacement.Metadata = maps.Clone(replacement.Metadata) + m.repairs[replacement.Issue][replacement.ExecutorID] = replacement + return nil + }).AnyTimes() + m.history = map[string]entity.RequestLog{} m.historyFailure = new(error) m.materializer = requestlog.NewMaterializer(tally.NoopScope) @@ -101,7 +146,7 @@ func newTestController(t *testing.T) (*Controller, controllerMocks) { m.store.EXPECT().GetRequestStore().Return(m.requests).AnyTimes() m.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{ID: testBuildID, RequestID: "1"}, nil).AnyTimes() m.requests.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{ID: "1", Queue: testQueue, State: entity.RequestStateFailed}, nil).AnyTimes() - return NewController(zap.New(core).Sugar(), tally.NoopScope, m.stores, m.materializer, m.executors, + return NewController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, m.executors, consumer.TopicKey("fix"), "stovepipe-fix"), m } @@ -117,12 +162,6 @@ func issueIdentity(issue entity.OpenIssue) entity.BuildIssue { return entity.BuildIssue{TaskID: issue.TaskID, UnitID: issue.UnitID} } -func issueWithFixID(issue entity.OpenIssue, fixID string) entity.BuildIssue { - observation := issueIdentity(issue) - observation.FixID = fixID - return observation -} - func afterIssue(issue entity.BuildIssue) storage.BuildIssueRange { return storage.BuildIssueRange{ After: storage.BuildIssueCursor{TaskID: issue.TaskID, UnitID: issue.UnitID}, @@ -130,245 +169,314 @@ func afterIssue(issue entity.BuildIssue) storage.BuildIssueRange { } } -func (m controllerMocks) expectIssue(issue entity.OpenIssue) { - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}). - Return([]entity.BuildIssue{issueIdentity(issue)}, nil) - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).Return(nil, nil) - m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(m.executor, nil) +func (m controllerMocks) expectIssues(items []entity.OpenIssue) { + m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil).AnyTimes() + observations := make([]entity.BuildIssue, 0, len(items)) + for _, issue := range items { + observations = append(observations, issueIdentity(issue)) + m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil).AnyTimes() + } + sort.Slice(observations, func(i, j int) bool { + if observations[i].TaskID == observations[j].TaskID { + return observations[i].UnitID < observations[j].UnitID + } + return observations[i].TaskID < observations[j].TaskID + }) + m.observations.EXPECT().List(gomock.Any(), testBuildID, gomock.Any()).DoAndReturn(func(_ context.Context, _ string, bounds storage.BuildIssueRange) ([]entity.BuildIssue, error) { + var page []entity.BuildIssue + for _, observation := range observations { + if observation.TaskID > bounds.After.TaskID || observation.TaskID == bounds.After.TaskID && observation.UnitID > bounds.After.UnitID { + page = append(page, observation) + if len(page) == bounds.Limit { + break + } + } + } + return page, nil + }).AnyTimes() } -func TestProcessBuildSubmitsOneIssueAndAssociatesFixAfterCall(t *testing.T) { - controller, m := newTestController(t) - issue := testIssue() - observation := issueIdentity(issue) - observation.Version = 7 - another := entity.BuildIssue{TaskID: "test", UnitID: "unit/2"} - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}). - Return([]entity.BuildIssue{observation, another}, nil) - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(another)).Return(nil, nil) - m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(m.executor, nil) - updated := observation - updated.FixID = "fix/9" - gomock.InOrder( - m.executor.EXPECT().Submit(gomock.Any(), issue).Return(updated.FixID, nil), - m.observations.EXPECT().Update(gomock.Any(), testBuildID, updated, int32(7), int32(8)).Return(nil), - ) - require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) - entries := m.logs.All() - require.Len(t, entries, 1) - assert.Equal(t, testQueue, entries[0].ContextMap()["queue"]) - assert.Equal(t, testBuildID, entries[0].ContextMap()["build_id"]) - assert.Equal(t, issue.TaskID, entries[0].ContextMap()["task_id"]) - assert.Equal(t, issue.UnitID, entries[0].ContextMap()["unit_id"]) - assert.Equal(t, updated.FixID, entries[0].ContextMap()["fix_id"]) +func (m controllerMocks) expectIssue(issue entity.OpenIssue) { + m.expectIssues([]entity.OpenIssue{issue}) + m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return([]fixer.Registration{{ID: "first", Executor: m.executor}}, nil).AnyTimes() } -func TestProcessBuildFindsCandidateOnLaterPage(t *testing.T) { - controller, m := newTestController(t) - issue := entity.OpenIssue{TaskID: "test", FirstSeenBuildID: testBuildID} - older := entity.BuildIssue{TaskID: "build"} - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - gomock.InOrder( - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}). - Return([]entity.BuildIssue{older}, nil), - m.issues.EXPECT().Get(gomock.Any(), older.TaskID, older.UnitID). - Return(entity.IssueState{TaskID: older.TaskID, FirstSeenBuildID: "older-build", Status: entity.IssueStatusOpen, Version: 1, LastProcessedBuildID: "older-build", LastChangedBuildID: "older-build", ChangedAtMs: 1000}, nil), - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(older)). - Return([]entity.BuildIssue{issueIdentity(issue)}, nil), - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil), - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).Return(nil, nil), - ) - m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(m.executor, nil) - m.executor.EXPECT().Submit(gomock.Any(), issue).Return("attempt", nil) - m.observations.EXPECT().Update(gomock.Any(), testBuildID, issueWithFixID(issue, "attempt"), int32(0), int32(1)).Return(nil) - - require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) +func accepted(issues []entity.OpenIssue, id string) []fixer.Submission { + results := make([]fixer.Submission, 0, len(issues)) + for _, issue := range issues { + results = append(results, fixer.Submission{Issue: issue, FixID: id}) + } + return results } -func TestProcessBuildChecksLaterPagesForRecordedFix(t *testing.T) { - controller, m := newTestController(t) - issue := testIssue() - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - gomock.InOrder( - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}). - Return([]entity.BuildIssue{issueIdentity(issue)}, nil), - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil), - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))). - Return([]entity.BuildIssue{{TaskID: "closed", FixID: "fix/1"}}, nil), - ) - require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) +func TestProcessBuildBatchesAcrossPagesAndReplays(t *testing.T) { + c, m := newTestController(t) + issues := make([]entity.OpenIssue, 205) + for i := range issues { + issues[i] = entity.OpenIssue{TaskID: "tests", UnitID: fmt.Sprintf("unit/%03d", i), FirstSeenBuildID: testBuildID} + } + m.expectIssues(issues) + m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return([]fixer.Registration{{ID: "first", Executor: m.executor}}, nil).Times(2) + m.executor.EXPECT().Submit(gomock.Any(), issues).Return(accepted(issues, "shared-fix"), nil) + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + require.Len(t, m.repairs, 205) + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + require.Len(t, m.history, 205) } -func TestProcessBuildDoesNotResubmitAfterRecordedFixID(t *testing.T) { - controller, m := newTestController(t) - issue := testIssue() - record := issueIdentity(issue) - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil).Times(2) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}). - DoAndReturn(func(context.Context, string, storage.BuildIssueRange) ([]entity.BuildIssue, error) { - return []entity.BuildIssue{record}, nil - }).Times(2) - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(record)).Return(nil, nil) - m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(m.executor, nil) - m.executor.EXPECT().Submit(gomock.Any(), issue).Return("fix/1", nil) - m.observations.EXPECT().Update(gomock.Any(), testBuildID, issueWithFixID(issue, "fix/1"), int32(0), int32(1)). - DoAndReturn(func(_ context.Context, _ string, updated entity.BuildIssue, _, newVersion int32) error { - record = updated - record.Version = newVersion - return nil - }) - for i := 0; i < 2; i++ { - require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) - } +func TestIndependentExecutorsRetainPartialResults(t *testing.T) { + c, m := newTestController(t) + first := testIssue() + second := first + second.UnitID = "unit/2" + items := []entity.OpenIssue{first, second} + m.expectIssues(items) + other := fixermock.NewMockRepairExecutor(m.ctrl) + m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return([]fixer.Registration{{ID: "first", Executor: m.executor}, {ID: "other", Executor: other}}, nil).AnyTimes() + m.executor.EXPECT().Submit(gomock.Any(), items).Return([]fixer.Submission{{Issue: first, FixID: "shared-fix"}}, assert.AnError) + other.EXPECT().Submit(gomock.Any(), items).Return([]fixer.Submission{ + {Issue: first, SkipReason: "unsupported"}, + {Issue: second, FixID: "other-fix"}, + }, nil) + require.ErrorIs(t, c.processBuild(context.Background(), testQueue, testBuildID), assert.AnError) + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{second}).Return(accepted([]entity.OpenIssue{second}, "shared-fix"), nil) + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + require.Len(t, m.repairs[first], 2) + require.Len(t, m.repairs[second], 2) + assert.Equal(t, "shared-fix", m.repairs[second]["first"].FixID) } -func TestProcessBuildWithoutEligibleIssue(t *testing.T) { - for _, firstSeenBuildID := range []string{"", "other-build", "build/42 ", "Build/42"} { - t.Run(firstSeenBuildID, func(t *testing.T) { - controller, m := newTestController(t) - issue := entity.BuildIssue{TaskID: "build"} - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}). - Return([]entity.BuildIssue{issue}, nil) - if firstSeenBuildID == "" { - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(entity.IssueState{}, storage.ErrNotFound) - } else { - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID). - Return(entity.IssueState{TaskID: issue.TaskID, FirstSeenBuildID: firstSeenBuildID, Status: entity.IssueStatusOpen, Version: 1, LastProcessedBuildID: firstSeenBuildID, LastChangedBuildID: firstSeenBuildID, ChangedAtMs: 1000}, nil) +func TestInvalidExecutorResultsLeaveItemsPending(t *testing.T) { + for _, kind := range []string{"omitted", "foreign", "both dispositions", "empty disposition", "duplicate"} { + t.Run(kind, func(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + var results []fixer.Submission + switch kind { + case "foreign": + changed := issue + changed.FirstSeenBuildID = "other" + results = accepted([]entity.OpenIssue{changed}, "id") + case "both dispositions": + results = []fixer.Submission{{Issue: issue, FixID: "id", SkipReason: "skip"}} + case "empty disposition": + results = []fixer.Submission{{Issue: issue}} + case "duplicate": + results = append(accepted([]entity.OpenIssue{issue}, "id"), accepted([]entity.OpenIssue{issue}, "id")...) + } + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).Return(results, nil) + require.Error(t, c.processBuild(context.Background(), testQueue, testBuildID)) + if kind != "duplicate" { + require.Empty(t, m.repairs[issue]["first"].FixID) + require.Empty(t, m.repairs[issue]["first"].SkipReason) } - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issue)).Return(nil, nil) - - require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) }) } } -func TestProcessBuildWithoutObservations(t *testing.T) { - controller, m := newTestController(t) - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).Return(nil, nil) - require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) +func TestClosedAndPreviousOpeningsAreNotSubmitted(t *testing.T) { + for _, state := range []entity.IssueState{{Status: entity.IssueStatusClosed, FirstSeenBuildID: testBuildID}, {Status: entity.IssueStatusOpen, FirstSeenBuildID: "old-build"}, {}} { + t.Run(fmt.Sprint(state), func(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.stores.EXPECT().For(gomock.Any()).Return(m.store, nil) + m.executors.EXPECT().For(gomock.Any()).Return([]fixer.Registration{{ID: "first", Executor: m.executor}}, nil) + m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).Return([]entity.BuildIssue{issueIdentity(issue)}, nil) + m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).Return(nil, nil) + m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(state, nil) + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + }) + } } -func TestProcessBuildLeavesSubmissionWithoutAnIDEligibleForRetry(t *testing.T) { - for _, test := range []struct { - name string - err error - }{ - {name: "empty fix ID"}, - {name: "provider error", err: assert.AnError}, - {name: "retryable provider error", err: errs.NewRetryableError(assert.AnError)}, - {name: "cancelled call", err: context.Canceled}, - } { - t.Run(test.name, func(t *testing.T) { - controller, m := newTestController(t) +func TestRepairHistoryRetryDoesNotResubmit(t *testing.T) { + for _, skipped := range []bool{false, true} { + t.Run(fmt.Sprint(skipped), func(t *testing.T) { + c, m := newTestController(t) issue := testIssue() m.expectIssue(issue) - m.executor.EXPECT().Submit(gomock.Any(), issue).Return("", test.err) - err := controller.processBuild(context.Background(), testQueue, testBuildID) - require.Error(t, err) - if test.err != nil { - require.ErrorIs(t, err, test.err) - assert.Equal(t, errs.IsRetryable(test.err), errs.IsRetryable(err)) + submission := fixer.Submission{Issue: issue, FixID: "fix"} + if skipped { + submission.FixID, submission.SkipReason = "", "unsupported" } - require.Len(t, m.logs.All(), 1) - assert.Equal(t, zap.ErrorLevel, m.logs.All()[0].Level) + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).Return([]fixer.Submission{submission}, nil) + *m.historyFailure = assert.AnError + require.Error(t, c.processBuild(context.Background(), testQueue, testBuildID)) + *m.historyFailure = nil + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + require.Len(t, m.history, 1) }) } } -func TestProcessBuildRetainsFixIDReturnedWithAnError(t *testing.T) { - controller, m := newTestController(t) - issue := testIssue() - m.expectIssue(issue) - m.executor.EXPECT().Submit(gomock.Any(), issue).Return("fix/1", assert.AnError) - m.observations.EXPECT().Update(gomock.Any(), testBuildID, issueWithFixID(issue, "fix/1"), int32(0), int32(1)).Return(nil) - require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) -} - -func TestProcessBuildHandlesAssociationPersistenceResults(t *testing.T) { - for _, test := range []struct { - name string - updateErr error - current entity.BuildIssue - readErr error - wantErr error +func TestConcurrentDispositionIsNeverOverwritten(t *testing.T) { + for _, tt := range []struct { + name string + fixID string + skipReason string + conflicts bool }{ - {name: "write error", updateErr: assert.AnError, wantErr: assert.AnError}, - {name: "concurrent association", updateErr: storage.ErrVersionMismatch, current: entity.BuildIssue{FixID: "another-fix"}}, - {name: "concurrent update without an association", updateErr: storage.ErrVersionMismatch, wantErr: storage.ErrVersionMismatch}, - {name: "reload error", updateErr: storage.ErrVersionMismatch, readErr: assert.AnError, wantErr: assert.AnError}, + {name: "same decision", fixID: "fix"}, + {name: "different fix", fixID: "already-recorded", conflicts: true}, + {name: "skip", skipReason: "unsupported", conflicts: true}, } { - t.Run(test.name, func(t *testing.T) { - controller, m := newTestController(t) + t.Run(tt.name, func(t *testing.T) { + c, m := newTestController(t) issue := testIssue() m.expectIssue(issue) - m.executor.EXPECT().Submit(gomock.Any(), issue).Return("fix/1", nil) - m.observations.EXPECT().Update(gomock.Any(), testBuildID, issueWithFixID(issue, "fix/1"), int32(0), int32(1)).Return(test.updateErr) - if test.updateErr == storage.ErrVersionMismatch { - m.observations.EXPECT().Get(gomock.Any(), testBuildID, issue.TaskID, issue.UnitID).Return(test.current, test.readErr) - } - err := controller.processBuild(context.Background(), testQueue, testBuildID) - if test.wantErr != nil { - require.ErrorIs(t, err, test.wantErr) + metadata := map[string]string{"state": "retained"} + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).DoAndReturn(func(ctx context.Context, _ []entity.OpenIssue) ([]fixer.Submission, error) { + current, err := m.results.Get(ctx, issue, "first") + require.NoError(t, err) + current.FixID, current.SkipReason, current.Metadata = tt.fixID, tt.skipReason, metadata + require.NoError(t, m.results.Update(ctx, current, current.Version, current.Version+1)) + return []fixer.Submission{{Issue: issue, FixID: "fix"}}, nil + }) + err := c.processBuild(context.Background(), testQueue, testBuildID) + if tt.conflicts { + require.Error(t, err) } else { require.NoError(t, err) - assert.Equal(t, "another-fix", m.logs.All()[1].ContextMap()["fix_id"]) } + require.Equal(t, metadata, m.repairs[issue]["first"].Metadata) + require.Equal(t, tt.fixID, m.repairs[issue]["first"].FixID) + require.Equal(t, tt.skipReason, m.repairs[issue]["first"].SkipReason) + require.Len(t, m.history, 1) }) } } -func TestProcessBuildReturnsErrorsBeforeSubmission(t *testing.T) { - for _, step := range []string{"storage factory", "issue listing", "open issue lookup", "later page", "executor factory"} { - t.Run(step, func(t *testing.T) { - controller, m := newTestController(t) - issue := testIssue() - configureFailure := func() { - if step == "storage factory" { - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(nil, assert.AnError) - return - } - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - first := m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}) - if step == "issue listing" { - first.Return(nil, assert.AnError) - return - } - first.Return([]entity.BuildIssue{issueIdentity(issue)}, nil) - lookup := m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID) - if step == "open issue lookup" { - lookup.Return(entity.IssueState{}, assert.AnError) - return - } - lookup.Return(storedIssue(issue), nil) - last := m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))) - if step == "later page" { - last.Return(nil, assert.AnError) - return - } - last.Return(nil, nil) - m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(nil, assert.AnError) - } - configureFailure() - require.ErrorIs(t, controller.processBuild(context.Background(), testQueue, testBuildID), assert.AnError) - }) - } +func TestFixerMetadataPersistsAcrossSubmissionAndReplay(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + metadata := map[string]string{"state": `{"attempt":2,"workflow":"run/1"}`} + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).DoAndReturn(func(ctx context.Context, _ []entity.OpenIssue) ([]fixer.Submission, error) { + current, err := m.results.Get(ctx, issue, "first") + require.NoError(t, err) + require.Empty(t, current.FixID) + require.Empty(t, current.SkipReason) + current.Metadata = metadata + require.NoError(t, m.results.Update(ctx, current, current.Version, current.Version+1)) + return accepted([]entity.OpenIssue{issue}, "fix"), nil + }) + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + require.Equal(t, "fix", m.repairs[issue]["first"].FixID) + require.Equal(t, metadata, m.repairs[issue]["first"].Metadata) + require.Len(t, m.history, 1) +} + +func TestConcurrentMetadataUpdateRetriesReceipt(t *testing.T) { + ctrl := gomock.NewController(t) + repairs := storagemock.NewMockIssueRepairStore(ctrl) + issue := testIssue() + pending := entity.IssueRepair{Issue: issue, ExecutorID: "first", Version: 1} + concurrent := pending + concurrent.Version = 2 + concurrent.Metadata = map[string]string{"state": "concurrent edit"} + replacement := concurrent + replacement.FixID = "fix" + gomock.InOrder( + repairs.EXPECT().Get(gomock.Any(), issue, "first").Return(pending, nil), + repairs.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(storage.ErrVersionMismatch), + repairs.EXPECT().Get(gomock.Any(), issue, "first").Return(concurrent, nil), + repairs.EXPECT().Update(gomock.Any(), replacement, int32(2), int32(3)).Return(nil), + ) + actual, err := persistRepairSubmission(context.Background(), repairs, "first", fixer.Submission{Issue: issue, FixID: "fix"}) + require.NoError(t, err) + require.Equal(t, "fix", actual.FixID) + require.Equal(t, concurrent.Metadata, actual.Metadata) + require.Equal(t, int32(3), actual.Version) } -func TestProcessBuildSkipsClosedIssue(t *testing.T) { - controller, m := newTestController(t) +func TestPartialExecutorFailuresRetainRetryClassification(t *testing.T) { + c, m := newTestController(t) issue := testIssue() - closed := storedIssue(issue) - closed.Status = entity.IssueStatusClosed - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).Return([]entity.BuildIssue{issueIdentity(issue)}, nil) - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(closed, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).Return(nil, nil) - require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) + m.expectIssue(issue) + foreign := issue + foreign.FirstSeenBuildID = "other" + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).Return([]fixer.Submission{ + {Issue: issue, FixID: "accepted"}, + {Issue: foreign, FixID: "foreign"}, + }, context.Canceled) + err := c.processBuild(context.Background(), testQueue, testBuildID) + processed := errs.NewClassifierProcessor(genericerrs.Classifier).Process(err) + require.True(t, errs.IsRetryable(processed)) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, "accepted", m.repairs[issue]["first"].FixID) + require.Len(t, m.history, 1) +} + +func TestFailedDispositionWriteRetriesOnlyUnrecordedItems(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).Return(accepted([]entity.OpenIssue{issue}, "fix"), nil).Times(2) + *m.updateFailure = assert.AnError + require.ErrorIs(t, c.processBuild(context.Background(), testQueue, testBuildID), assert.AnError) + require.Empty(t, m.repairs[issue]["first"].FixID) + require.Empty(t, m.history) + *m.updateFailure = nil + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + require.Equal(t, "fix", m.repairs[issue]["first"].FixID) +} + +func TestFailedPendingWriteDoesNotCallExecutor(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + *m.repairFailure = assert.AnError + require.ErrorIs(t, c.processBuild(context.Background(), testQueue, testBuildID), assert.AnError) + require.Empty(t, m.repairs) +} + +func TestExistingPendingRowIsResubmitted(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + m.repairs[issue] = map[string]entity.IssueRepair{"first": {Issue: issue, ExecutorID: "first", Version: 1, Metadata: map[string]string{"state": "before crash"}}} + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).Return(accepted([]entity.OpenIssue{issue}, "fix"), nil) + require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) + require.Equal(t, "fix", m.repairs[issue]["first"].FixID) + require.Equal(t, map[string]string{"state": "before crash"}, m.repairs[issue]["first"].Metadata) +} + +func TestPreparationFailurePreservesHistoryErrors(t *testing.T) { + c, m := newTestController(t) + recorded := testIssue() + pending := recorded + pending.UnitID = "unit/2" + m.expectIssues([]entity.OpenIssue{recorded, pending}) + m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return([]fixer.Registration{{ID: "first", Executor: m.executor}}, nil).AnyTimes() + m.repairs[recorded] = map[string]entity.IssueRepair{"first": {Issue: recorded, ExecutorID: "first", FixID: "accepted", Version: 2}} + *m.repairFailure = assert.AnError + *m.historyFailure = context.Canceled + + err := c.processBuild(context.Background(), testQueue, testBuildID) + + require.ErrorIs(t, err, assert.AnError) + require.ErrorIs(t, err, context.Canceled) + require.True(t, errs.IsRetryable(errs.NewClassifierProcessor(genericerrs.Classifier).Process(err))) + require.Empty(t, m.repairs[pending]) + *m.historyFailure = nil + require.ErrorIs(t, c.processBuild(context.Background(), testQueue, testBuildID), assert.AnError) + require.Len(t, m.history, 1) +} + +func TestExhaustedRepairVersionDoesNotOverwriteState(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + pending := entity.IssueRepair{Issue: issue, ExecutorID: "first", Version: math.MaxInt32, Metadata: map[string]string{"state": "retained"}} + m.repairs[issue] = map[string]entity.IssueRepair{"first": pending} + m.executor.EXPECT().Submit(gomock.Any(), []entity.OpenIssue{issue}).Return(accepted([]entity.OpenIssue{issue}, "fix"), nil) + + err := c.processBuild(context.Background(), testQueue, testBuildID) + + require.Error(t, err) + require.False(t, errs.IsRetryable(errs.NewClassifierProcessor(genericerrs.Classifier).Process(err))) + require.Equal(t, pending, m.repairs[issue]["first"]) + require.Empty(t, m.history) } diff --git a/stovepipe/controller/fix/history.go b/stovepipe/controller/fix/history.go index 6f1817395..0614ae9bd 100644 --- a/stovepipe/controller/fix/history.go +++ b/stovepipe/controller/fix/history.go @@ -17,6 +17,7 @@ package fix import ( "context" "crypto/sha256" + "encoding/json" "fmt" "github.com/uber/submitqueue/platform/errs" @@ -44,6 +45,31 @@ func loadRequestForBuild(ctx context.Context, store storage.Storage, queue, buil return request, nil } +func persistRepairSubmissionLog(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, result entity.IssueRepair) error { + event := entity.RequestEventFixSubmitted + if result.SkipReason != "" { + event = entity.RequestEventFixSubmissionSkipped + } + return persistRepairEventLog(ctx, materializer, store, request, result, event, result.SkipReason) +} + +func persistRepairEventLog(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, result entity.IssueRepair, event entity.RequestEvent, reason string) error { + identity, err := json.Marshal([]string{result.Issue.FirstSeenBuildID, result.Issue.TaskID, result.Issue.UnitID, result.ExecutorID}) + if err != nil { + return err + } + digest := sha256.Sum256(identity) + log := requestlog.NewRequestEventLog(request, event, fmt.Sprintf("%x", digest), map[string]string{ + requestlog.MetadataKeyBuildID: result.Issue.FirstSeenBuildID, + requestlog.MetadataKeyTaskID: result.Issue.TaskID, + requestlog.MetadataKeyUnitID: result.Issue.UnitID, + requestlog.MetadataKeyFixID: result.FixID, + requestlog.MetadataKeyReason: reason, + requestlog.MetadataKeyExecutorID: result.ExecutorID, + }) + return materializer.PersistLog(ctx, store, log) +} + func persistFixSubmittedLog(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, buildID string, observation entity.BuildIssue) error { // The queue-scoped executor's opaque fix ID identifies the accepted attempt. // Restoring this from the durable association must precede any replay skip. diff --git a/stovepipe/controller/fix/history_test.go b/stovepipe/controller/fix/history_test.go index 0097a3cc4..e124de291 100644 --- a/stovepipe/controller/fix/history_test.go +++ b/stovepipe/controller/fix/history_test.go @@ -16,7 +16,6 @@ package fix import ( "context" - "strings" "testing" "github.com/stretchr/testify/assert" @@ -32,45 +31,6 @@ import ( "go.uber.org/zap" ) -func TestSubmittedFixHistoryRepairsWithoutResubmitting(t *testing.T) { - for _, tt := range []struct { - name string - submitErr error - }{ - {name: "successful submission"}, - {name: "identified submission with error", submitErr: assert.AnError}, - } { - t.Run(tt.name, func(t *testing.T) { - c, m := newTestController(t) - issue := testIssue() - m.expectIssue(issue) - m.executor.EXPECT().Submit(gomock.Any(), issue).Return("fix/1", tt.submitErr) - saved := issueWithFixID(issue, "fix/1") - m.observations.EXPECT().Update(gomock.Any(), testBuildID, saved, int32(0), int32(1)).Return(nil) - *m.historyFailure = assert.AnError - require.ErrorIs(t, c.processBuild(context.Background(), testQueue, testBuildID), assert.AnError) - require.Empty(t, m.history) - *m.historyFailure = nil - for i := 0; i < 2; i++ { - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).Return([]entity.BuildIssue{saved}, nil) - require.NoError(t, c.processBuild(context.Background(), testQueue, testBuildID)) - } - require.Len(t, m.history, 1) - var log entity.RequestLog - for _, retained := range m.history { - log = retained - } - require.Equal(t, entity.RequestEventFixSubmitted, log.Event) - require.Equal(t, map[string]string{ - requestlog.MetadataKeyBuildID: testBuildID, requestlog.MetadataKeyFixID: "fix/1", - requestlog.MetadataKeyTaskID: issue.TaskID, requestlog.MetadataKeyUnitID: issue.UnitID, - }, log.Metadata) - require.Empty(t, log.State) - }) - } -} - func TestFixDLQRetainsDecisionWithoutSubmitting(t *testing.T) { for _, tt := range []struct { name string @@ -152,27 +112,3 @@ func TestFixDLQDiscardsInvalidMessages(t *testing.T) { }) } } - -func TestSubmittedFixHistoryPreservesFixID(t *testing.T) { - for _, tt := range []struct { - name string - fixID string - }{ - {name: "opaque ID", fixID: "fix/1 "}, - {name: "maximum length", fixID: strings.Repeat("f", 255)}, - } { - t.Run(tt.name, func(t *testing.T) { - _, m := newTestController(t) - observation := issueWithFixID(testIssue(), tt.fixID) - request := entity.Request{ID: "1", Queue: testQueue} - for i := 0; i < 2; i++ { - require.NoError(t, persistFixSubmittedLog(context.Background(), m.materializer, m.store, request, testBuildID, observation)) - } - require.Len(t, m.history, 1) - for _, log := range m.history { - require.LessOrEqual(t, len(log.ID), 255) - require.Equal(t, tt.fixID, log.Metadata[requestlog.MetadataKeyFixID]) - } - }) - } -} diff --git a/stovepipe/controller/fix/submission.go b/stovepipe/controller/fix/submission.go new file mode 100644 index 000000000..ea51670a0 --- /dev/null +++ b/stovepipe/controller/fix/submission.go @@ -0,0 +1,167 @@ +// 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 fix + +import ( + "context" + "errors" + "fmt" + "math" + + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/fixer" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func (c *Controller) submitIssues(ctx context.Context, store storage.Storage, request entity.Request, buildID string, + registration fixer.Registration, observations []entity.BuildIssue, eligible []entity.OpenIssue) error { + pending, recorded, prepareErr := prepareIssueRepairs(ctx, store.GetIssueRepairStore(), buildID, registration.ID, observations, eligible) + historyErr := c.persistRepairHistory(ctx, store, request, recorded) + if prepareErr != nil || len(pending) == 0 { + return errs.Group(prepareErr, historyErr) + } + + op := metrics.Begin(c.metricsScope, "submit", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) + submissions, submitErr := registration.Executor.Submit(ctx, append([]entity.OpenIssue(nil), pending...)) + op.Complete(submitErr) + recordErr := c.recordRepairSubmissions(ctx, store, request, registration.ID, pending, submissions, submitErr == nil) + if submitErr != nil { + submitErr = fmt.Errorf("submit pending repairs: %w", submitErr) + } + return errs.Group(historyErr, submitErr, recordErr) +} + +func prepareIssueRepairs(ctx context.Context, repairs storage.IssueRepairStore, buildID, executorID string, + observations []entity.BuildIssue, eligible []entity.OpenIssue) ([]entity.OpenIssue, []entity.IssueRepair, error) { + pending := make([]entity.OpenIssue, 0, len(eligible)) + var recorded []entity.IssueRepair + eligibleSet := make(map[entity.OpenIssue]bool, len(eligible)) + for _, issue := range eligible { + eligibleSet[issue] = true + } + // Restore durable history even for issues that have since closed. + for _, observation := range observations { + issue := entity.OpenIssue{TaskID: observation.TaskID, UnitID: observation.UnitID, FirstSeenBuildID: buildID} + result, err := repairs.Get(ctx, issue, executorID) + if errors.Is(err, storage.ErrNotFound) { + if !eligibleSet[issue] { + continue + } + result = entity.IssueRepair{Issue: issue, ExecutorID: executorID, Version: 1} + err = repairs.Create(ctx, result) + if errors.Is(err, storage.ErrAlreadyExists) { + result, err = repairs.Get(ctx, issue, executorID) + } + if err != nil { + return nil, recorded, fmt.Errorf("ensure pending repair for task %q unit %q: %w", issue.TaskID, issue.UnitID, err) + } + } + if err != nil { + return nil, recorded, fmt.Errorf("load repair for task %q unit %q: %w", issue.TaskID, issue.UnitID, err) + } + if result.FixID != "" || result.SkipReason != "" { + recorded = append(recorded, result) + } else if eligibleSet[issue] { + pending = append(pending, issue) + } + } + return pending, recorded, nil +} + +func (c *Controller) recordRepairSubmissions(ctx context.Context, store storage.Storage, request entity.Request, executorID string, + pending []entity.OpenIssue, submissions []fixer.Submission, requireAll bool) error { + allowed := make(map[entity.OpenIssue]bool, len(pending)) + for _, issue := range pending { + allowed[issue] = true + } + var failures []error + var recorded []entity.IssueRepair + seen := make(map[entity.OpenIssue]bool, len(submissions)) + for _, submission := range submissions { + var validationErr error + switch { + case !allowed[submission.Issue]: + validationErr = fmt.Errorf("issue was not submitted") + case seen[submission.Issue]: + validationErr = fmt.Errorf("duplicate receipt") + case (submission.FixID == "") == (submission.SkipReason == ""): + validationErr = fmt.Errorf("receipt must contain exactly one of fix ID or skip reason") + case len(submission.FixID) > 255 || len(submission.SkipReason) > 255: + validationErr = fmt.Errorf("fix ID and skip reason must be at most 255 bytes") + } + if validationErr != nil { + failures = append(failures, fmt.Errorf("invalid repair submission for task %q unit %q: %w", submission.Issue.TaskID, submission.Issue.UnitID, validationErr)) + continue + } + seen[submission.Issue] = true + record, err := persistRepairSubmission(ctx, store.GetIssueRepairStore(), executorID, submission) + if err != nil { + failures = append(failures, fmt.Errorf("record repair submission for task %q unit %q: %w", submission.Issue.TaskID, submission.Issue.UnitID, err)) + } + // A conflicting receipt still returns the durable outcome for history recovery. + if record.FixID != "" || record.SkipReason != "" { + recorded = append(recorded, record) + } + } + if requireAll && len(seen) < len(pending) { + failures = append(failures, fmt.Errorf("repair executor omitted %d pending items", len(pending)-len(seen))) + } + failures = append(failures, c.persistRepairHistory(ctx, store, request, recorded)) + return errs.Group(failures...) +} + +func (c *Controller) persistRepairHistory(ctx context.Context, store storage.Storage, request entity.Request, recorded []entity.IssueRepair) error { + var failures []error + for _, repair := range recorded { + if err := persistRepairSubmissionLog(ctx, c.materializer, store, request, repair); err != nil { + failures = append(failures, fmt.Errorf("persist repair history for task %q unit %q: %w", repair.Issue.TaskID, repair.Issue.UnitID, err)) + } + } + return errs.Group(failures...) +} + +func persistRepairSubmission(ctx context.Context, repairs storage.IssueRepairStore, executorID string, submission fixer.Submission) (entity.IssueRepair, error) { + for { + if err := ctx.Err(); err != nil { + return entity.IssueRepair{}, err + } + current, err := repairs.Get(ctx, submission.Issue, executorID) + if err != nil { + return entity.IssueRepair{}, fmt.Errorf("reload repair: %w", err) + } + if current.FixID != "" || current.SkipReason != "" { + if current.FixID != submission.FixID || current.SkipReason != submission.SkipReason { + return current, fmt.Errorf("executor returned conflicting repair disposition") + } + return current, nil + } + if current.Version == math.MaxInt32 { + return entity.IssueRepair{}, fmt.Errorf("repair version exhausted") + } + replacement := current + replacement.FixID, replacement.SkipReason = submission.FixID, submission.SkipReason + newVersion := current.Version + 1 + if err := repairs.Update(ctx, replacement, current.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + continue + } + return entity.IssueRepair{}, fmt.Errorf("update repair: %w", err) + } + replacement.Version = newVersion + return replacement, nil + } +} diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go index 1034f84f1..e56805ade 100644 --- a/stovepipe/core/requestlog/materializer.go +++ b/stovepipe/core/requestlog/materializer.go @@ -52,6 +52,8 @@ const ( MetadataKeyClosureMode = "closure_mode" // MetadataKeyFixID identifies an opaque repair attempt. MetadataKeyFixID = "fix_id" + // MetadataKeyExecutorID identifies the queue-local registered repair backend. + MetadataKeyExecutorID = "executor_id" // MetadataKeyTaskID identifies the validation task associated with a repair attempt. MetadataKeyTaskID = "task_id" // MetadataKeyUnitID identifies the validation unit; empty means a task-wide issue. From f5574b243824f486c017f0fe28b808e30aed3b57 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 6 Nov 2026 22:10:42 +0000 Subject: [PATCH 4/4] feat(stovepipe): recover per-executor repair history in the DLQ Summary: Recover every executor's durable submission or skip without submitting backend work. Pending or missing eligible repair records retain an acceptance-unknown event. Wire the executor registry into recovery and remove the superseded build-level selection and history helpers. This is the final layer of the repair-executor stack, building on normal batch dispatch. Test Plan: `TestFixDLQRetainsAllDispositionsWithoutSubmitting`, `TestFixDLQRetainsUnknownAcceptance`, and `TestFixDLQDiscardsInvalidMessages` cover retained outcomes, unknown acceptance, invalid deliveries, and the absence of executor submissions. The 28-target Stovepipe suite also verifies reference-server wiring. Lint, tidy, and Gazelle checks verify the repository metadata. Revert Plan: Revert this PR together with the dependent dispatch and executor-contract changes before deploying the earlier build-level repair flow. --- doc/rfc/stovepipe/request-log.md | 2 +- service/stovepipe/server/main.go | 5 +- service/stovepipe/server/main_test.go | 2 +- stovepipe/controller/fix/README.md | 2 +- stovepipe/controller/fix/dlq.go | 54 ++++++++++++++----- stovepipe/controller/fix/fix.go | 35 ------------- stovepipe/controller/fix/history.go | 29 ---------- stovepipe/controller/fix/history_test.go | 67 ++++++++---------------- 8 files changed, 70 insertions(+), 126 deletions(-) diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index 04b0ed8e3..e9df8e561 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -143,7 +143,7 @@ State entries retain the selected baseline URI when known, including an empty UR Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording. Its `project_fact_count` and `broken_project_fact_count` metadata contain the total and the number with degree above zero, counted from persisted facts after their indexes are durable. Project completion projection requires both counts and assigns them together in `RequestSummary` from the retained event. -`build_analysis_completed` and `issues_reconciled` use the build ID as their occurrence key. `fix_submission_abandoned` uses the build ID. Per-item `fix_submitted` and `fix_submission_skipped` events use a digest of opening build, task, unit, and executor identity, retaining the verbatim fix ID or skip reason and executor ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp. +`build_analysis_completed` and `issues_reconciled` use the build ID as their occurrence key. Per-item `fix_submitted`, `fix_submission_skipped`, and `fix_submission_abandoned` events use a digest of opening build, task, unit, and executor identity, retaining the verbatim fix ID or skip reason and executor ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp. Analysis and repair events do not advance Request state or alter validation facts. Analysis completion is retained only when the Request records the analyzed terminal build as its winning outcome, immediately before the final issue-tracking handoff. Its issue count and timestamp are an immutable snapshot of the first retained completion, including zero for an empty result. Retries still persist detected issues and publish the handoff, but reuse that snapshot if later detector results differ; count discrepancies are logged rather than treated as history conflicts. Other semantic conflicts remain errors. `issues_reconciled` records closure mode `reconciled`, `suppressed_cancelled`, or `suppressed_task_wide_issue`; suppression does not mean reconciliation failed. Skipped repair uses reason `no_eligible_issue`; abandoned repair uses `acceptance_unknown`. Raw dependency errors remain in operational logs. diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 3d9696c4f..f17d4b858 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -265,7 +265,7 @@ func run() error { if err != nil { return err } - dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry) + dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry, repairExecutors) if err != nil { return err } @@ -459,6 +459,7 @@ func registerDLQControllers( store storage.Factory, materializer requestlog.Materializer, registry consumer.TopicRegistry, + repairExecutors fixer.Factory, ) (int, error) { var count int @@ -492,7 +493,7 @@ func registerDLQControllers( } count++ - fixDLQController := fix.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyFix), "stovepipe-fix-dlq") + fixDLQController := fix.NewDLQController(logger, scope, store, materializer, repairExecutors, dlq.TopicKey(stovepipemq.TopicKeyFix), "stovepipe-fix-dlq") if err := c.Register(fixDLQController); err != nil { return count, fmt.Errorf("failed to register fix dlq controller: %w", err) } diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index 72d5852ad..57a11b331 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -70,7 +70,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New(), noopRepairExecutorFactory{}) require.NoError(t, err) - _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry) + _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry, noopRepairExecutorFactory{}) require.NoError(t, err) return registry, primary.controllers, deadLetter.controllers diff --git a/stovepipe/controller/fix/README.md b/stovepipe/controller/fix/README.md index 8de074c34..b1f4cfafb 100644 --- a/stovepipe/controller/fix/README.md +++ b/stovepipe/controller/fix/README.md @@ -14,4 +14,4 @@ Each durable disposition restores its request-history event before being skipped Issue tracking publishes another nudge after each nonempty opening pass, so late issues are offered independently of earlier submissions. Queue redelivery rederives pending work and may offer a smaller list. Reopening in another build creates a new identity. A fix ID must reflect actual backend acceptance for every item carrying it; returning an old ID alone does not deliver new work. -The DLQ controller reads the same durable decision without calling the executor. It repairs submitted or skipped history when possible; otherwise, an eligible issue without a fix ID retains `fix_submission_abandoned` once per build with reason `acceptance_unknown`. An executor error without an ID can follow acceptance, so this event does not claim rejection. Source and history errors retry. Unusable payloads, unresolvable queues, and missing or mismatched sources are logged and acknowledged. The reference host registers recovery with unlimited, always-retryable DLQ processing to preserve history repair through a storage outage. +The DLQ controller restores all available disposition history without submitting work. Pending rows and still-open items without repair rows retain `fix_submission_abandoned` with reason `acceptance_unknown`. Hosts register this reconciler with unlimited, always-retryable DLQ processing. Executor IDs must remain registered until outstanding history repair completes. diff --git a/stovepipe/controller/fix/dlq.go b/stovepipe/controller/fix/dlq.go index 65130a8fb..dbe0a5792 100644 --- a/stovepipe/controller/fix/dlq.go +++ b/stovepipe/controller/fix/dlq.go @@ -18,6 +18,8 @@ import ( "context" "fmt" + "errors" + "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" @@ -26,6 +28,7 @@ import ( mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/fixer" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) @@ -36,16 +39,17 @@ type DLQController struct { metricsScope tally.Scope stores storage.Factory materializer requestlog.Materializer + executors fixer.Factory topicKey consumer.TopicKey consumerGroup string } // NewDLQController creates repair history recovery. Hosts must use unlimited, // always-retryable DLQ processing so a history outage cannot discard its retry trigger. -func NewDLQController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, topicKey consumer.TopicKey, consumerGroup string) *DLQController { +func NewDLQController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, executors fixer.Factory, topicKey consumer.TopicKey, consumerGroup string) *DLQController { return &DLQController{ logger: logger.Named("fix_dlq_controller"), metricsScope: scope.SubScope("fix_dlq_controller"), - stores: stores, materializer: materializer, topicKey: topicKey, consumerGroup: consumerGroup, + stores: stores, materializer: materializer, executors: executors, topicKey: topicKey, consumerGroup: consumerGroup, } } @@ -75,21 +79,45 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) } return err } - observation, eligible, err := findIssueToFix(ctx, store.GetBuildIssueStore(), store.GetIssueStateStore(), fix.GetId()) + registrations, err := c.executors.For(fixer.Config{QueueName: fix.GetQueueName()}) if err != nil { - return fmt.Errorf("load repair decision for recovery: %w", err) + return err } - if observation.FixID != "" { - return persistFixSubmittedLog(ctx, c.materializer, store, request, fix.GetId(), observation) + if err := validateRegistrations(registrations); err != nil { + return err } - if !eligible { - return persistFixSubmissionDecision(ctx, c.materializer, store, request, fix.GetId(), - entity.RequestEventFixSubmissionSkipped, "no_eligible_issue") + observations, eligible, err := collectIssues(ctx, store, fix.GetId()) + if err != nil { + return fmt.Errorf("load repair decisions: %w", err) + } + open := map[entity.OpenIssue]bool{} + for _, issue := range eligible { + open[issue] = true + } + for _, registration := range registrations { + for _, observation := range observations { + issue := entity.OpenIssue{TaskID: observation.TaskID, UnitID: observation.UnitID, FirstSeenBuildID: fix.GetId()} + result, err := store.GetIssueRepairStore().Get(ctx, issue, registration.ID) + if errors.Is(err, storage.ErrNotFound) { + if !open[issue] { + continue + } + result = entity.IssueRepair{Issue: issue, ExecutorID: registration.ID} + } else if err != nil { + return err + } + if result.FixID == "" && result.SkipReason == "" { + if err := persistRepairEventLog(ctx, c.materializer, store, request, result, entity.RequestEventFixSubmissionAbandoned, "acceptance_unknown"); err != nil { + return err + } + continue + } + if err := persistRepairSubmissionLog(ctx, c.materializer, store, request, result); err != nil { + return err + } + } } - // A provider can have accepted work even when no ID was returned. Preserve - // that uncertainty and leave repair associations and issue status alone. - return persistFixSubmissionDecision(ctx, c.materializer, store, request, fix.GetId(), - entity.RequestEventFixSubmissionAbandoned, "acceptance_unknown") + return nil } func (c *DLQController) Name() string { return string(c.topicKey) } diff --git a/stovepipe/controller/fix/fix.go b/stovepipe/controller/fix/fix.go index cc1ba8af1..117148438 100644 --- a/stovepipe/controller/fix/fix.go +++ b/stovepipe/controller/fix/fix.go @@ -138,38 +138,3 @@ func collectIssues(ctx context.Context, store storage.Storage, buildID string) ( bounds.After = storage.BuildIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} } } - -func findIssueToFix(ctx context.Context, observations storage.BuildIssueStore, issues storage.IssueStateStore, buildID string) (entity.BuildIssue, bool, error) { - bounds := storage.BuildIssueRange{Limit: buildIssuePageSize} - var candidate entity.BuildIssue - found := false - for { - page, err := observations.List(ctx, buildID, bounds) - if err != nil { - return entity.BuildIssue{}, false, err - } - if len(page) == 0 { - return candidate, found, nil - } - for _, record := range page { - if record.FixID != "" { - return record, false, nil - } - if found { - continue - } - issue, err := issues.Get(ctx, record.TaskID, record.UnitID) - if errors.Is(err, storage.ErrNotFound) { - continue - } - if err != nil { - return entity.BuildIssue{}, false, err - } - if issue.Status == entity.IssueStatusOpen && issue.FirstSeenBuildID == buildID { - candidate, found = record, true - } - } - last := page[len(page)-1] - bounds.After = storage.BuildIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} - } -} diff --git a/stovepipe/controller/fix/history.go b/stovepipe/controller/fix/history.go index 0614ae9bd..d3553c0de 100644 --- a/stovepipe/controller/fix/history.go +++ b/stovepipe/controller/fix/history.go @@ -69,32 +69,3 @@ func persistRepairEventLog(ctx context.Context, materializer requestlog.Material }) return materializer.PersistLog(ctx, store, log) } - -func persistFixSubmittedLog(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, buildID string, observation entity.BuildIssue) error { - // The queue-scoped executor's opaque fix ID identifies the accepted attempt. - // Restoring this from the durable association must precede any replay skip. - // Opaque fix IDs can fill the storage column; the event prefix must still fit its own key. - identity := sha256.Sum256([]byte(observation.FixID)) - log := requestlog.NewRequestEventLog(request, entity.RequestEventFixSubmitted, fmt.Sprintf("%x", identity), - map[string]string{ - requestlog.MetadataKeyBuildID: buildID, - requestlog.MetadataKeyTaskID: observation.TaskID, - requestlog.MetadataKeyUnitID: observation.UnitID, - requestlog.MetadataKeyFixID: observation.FixID, - }) - if err := materializer.PersistLog(ctx, store, log); err != nil { - return fmt.Errorf("retain submitted repair %q for build %q: %w", observation.FixID, buildID, err) - } - return nil -} - -func persistFixSubmissionDecision(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, buildID string, event entity.RequestEvent, reason string) error { - log := requestlog.NewRequestEventLog(request, event, buildID, map[string]string{ - requestlog.MetadataKeyBuildID: buildID, - requestlog.MetadataKeyReason: reason, - }) - if err := materializer.PersistLog(ctx, store, log); err != nil { - return fmt.Errorf("retain repair decision for build %q: %w", buildID, err) - } - return nil -} diff --git a/stovepipe/controller/fix/history_test.go b/stovepipe/controller/fix/history_test.go index e124de291..c9f83524d 100644 --- a/stovepipe/controller/fix/history_test.go +++ b/stovepipe/controller/fix/history_test.go @@ -16,65 +16,44 @@ package fix import ( "context" + "fmt" "testing" - "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" - consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/stovepipe/core/messagequeue" - "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/mock/gomock" "go.uber.org/zap" ) -func TestFixDLQRetainsDecisionWithoutSubmitting(t *testing.T) { - for _, tt := range []struct { - name string - fixID string - issueStatus entity.IssueStatus - wantEvent entity.RequestEvent - wantReason string - }{ - {name: "associated", fixID: "fix/1", wantEvent: entity.RequestEventFixSubmitted}, - {name: "eligible", issueStatus: entity.IssueStatusOpen, wantEvent: entity.RequestEventFixSubmissionAbandoned, wantReason: "acceptance_unknown"}, - {name: "ineligible", issueStatus: entity.IssueStatusClosed, wantEvent: entity.RequestEventFixSubmissionSkipped, wantReason: "no_eligible_issue"}, - } { - t.Run(tt.name, func(t *testing.T) { - _, m := newTestController(t) - c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, messagequeue.TopicKeyFix, "recovery") +func TestFixDLQRetainsAllDispositionsWithoutSubmitting(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + m.repairs[issue] = map[string]entity.IssueRepair{"first": {Issue: issue, ExecutorID: "first", FixID: "fix"}} + dlq := NewDLQController(c.logger, tally.NoopScope, m.stores, m.materializer, m.executors, c.topicKey, c.consumerGroup) + require.NoError(t, dlq.Process(context.Background(), m.delivery(repairMessage(t)))) + require.Len(t, m.history, 1) +} + +func TestFixDLQRetainsUnknownAcceptance(t *testing.T) { + for _, pending := range []bool{false, true} { + t.Run(fmt.Sprint(pending), func(t *testing.T) { + c, m := newTestController(t) issue := testIssue() - observation := issueIdentity(issue) - observation.FixID = tt.fixID - for i := 0; i < 3; i++ { - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).Return([]entity.BuildIssue{observation}, nil) - if tt.fixID == "" { - current := storedIssue(issue) - current.Status = tt.issueStatus - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(current, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(observation)).Return(nil, nil) - } - d := consumermock.NewMockDelivery(m.ctrl) - d.EXPECT().Message().Return(repairMessage(t)) - if i == 0 { - *m.historyFailure = assert.AnError - require.ErrorIs(t, c.Process(context.Background(), d), assert.AnError) - require.Empty(t, m.history) - *m.historyFailure = nil - } else { - require.NoError(t, c.Process(context.Background(), d)) - } + m.expectIssue(issue) + if pending { + m.repairs[issue] = map[string]entity.IssueRepair{"first": {Issue: issue, ExecutorID: "first", Version: 1}} } - require.Len(t, m.history, 1) + dlq := NewDLQController(c.logger, tally.NoopScope, m.stores, m.materializer, m.executors, c.topicKey, c.consumerGroup) + require.NoError(t, dlq.Process(context.Background(), m.delivery(repairMessage(t)))) for _, log := range m.history { - require.Equal(t, tt.wantEvent, log.Event) - require.Equal(t, testBuildID, log.Metadata[requestlog.MetadataKeyBuildID]) - require.Equal(t, tt.wantReason, log.Metadata[requestlog.MetadataKeyReason]) + require.Equal(t, entity.RequestEventFixSubmissionAbandoned, log.Event) } + require.Len(t, m.history, 1) }) } } @@ -106,7 +85,7 @@ func TestFixDLQDiscardsInvalidMessages(t *testing.T) { } { t.Run(tt.name, func(t *testing.T) { _, m := newTestController(t) - c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, messagequeue.TopicKeyFix, "recovery") + c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, m.executors, messagequeue.TopicKeyFix, "recovery") require.NoError(t, c.Process(context.Background(), m.delivery(tt.configure(t, m)))) require.Empty(t, m.history) })