From fddd6b25f55d4c2f97d9efad0f1cdbc7a24d5821 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 23 Oct 2026 20:50:11 +0000 Subject: [PATCH] feat(stovepipe): select and submit one repair per build Scan build observations for existing fix IDs, submit an eligible open issue, and associate the returned ID through entity CAS. Return submission errors without an ID for normal queue handling; document caller-owned retry policy. Test Plan: Run ./tool/bazel test //stovepipe/controller/fix:go_default_test and make lint check-tidy check-gazelle. Revert Plan: Revert this commit. --- stovepipe/controller/fix/BUILD.bazel | 38 ++++ stovepipe/controller/fix/README.md | 7 + stovepipe/controller/fix/fix.go | 162 ++++++++++++++ stovepipe/controller/fix/fix_test.go | 322 +++++++++++++++++++++++++++ stovepipe/extension/fixer/README.md | 4 +- stovepipe/extension/fixer/fixer.go | 7 +- 6 files changed, 534 insertions(+), 6 deletions(-) create mode 100644 stovepipe/controller/fix/BUILD.bazel create mode 100644 stovepipe/controller/fix/README.md create mode 100644 stovepipe/controller/fix/fix.go create mode 100644 stovepipe/controller/fix/fix_test.go diff --git a/stovepipe/controller/fix/BUILD.bazel b/stovepipe/controller/fix/BUILD.bazel new file mode 100644 index 000000000..a89b8a27a --- /dev/null +++ b/stovepipe/controller/fix/BUILD.bazel @@ -0,0 +1,38 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["fix.go"], + importpath = "github.com/uber/submitqueue/stovepipe/controller/fix", + visibility = ["//visibility:public"], + deps = [ + "//platform/consumer:go_default_library", + "//platform/metrics:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/fixer:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["fix_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/fixer:go_default_library", + "//stovepipe/extension/fixer/mock:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/extension/storage/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@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 new file mode 100644 index 000000000..4e2cbfba1 --- /dev/null +++ b/stovepipe/controller/fix/README.md @@ -0,0 +1,7 @@ +# Fix submission + +The controller 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`. + +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 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. diff --git a/stovepipe/controller/fix/fix.go b/stovepipe/controller/fix/fix.go new file mode 100644 index 000000000..51b331828 --- /dev/null +++ b/stovepipe/controller/fix/fix.go @@ -0,0 +1,162 @@ +// 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 submits build-scoped repair work. +package fix + +import ( + "context" + "errors" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "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" + "go.uber.org/zap" +) + +const buildIssuePageSize = 100 + +type Controller struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + executors fixer.Factory + topicKey consumer.TopicKey + consumerGroup string +} + +func NewController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + executors fixer.Factory, + topicKey consumer.TopicKey, + consumerGroup string, +) *Controller { + return &Controller{ + logger: logger.Named("fix_controller"), + metricsScope: scope.SubScope("fix_controller"), + stores: stores, + executors: executors, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +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.GetOpenIssueStore(), buildID) + if err != nil { + return fmt.Errorf("failed to select an issue for build %q: %w", buildID, err) + } + if !found { + return nil + } + + executor, err := c.executors.For(fixer.Config{QueueName: queueName}) + if err != nil { + return fmt.Errorf("failed to resolve repair executor: %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") + } + op.Complete(err) + 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) + } + } else { + c.logger.Infow("repair submitted", + "queue", queueName, "build_id", buildID, + "task_id", issue.TaskID, "unit_id", issue.UnitID, "fix_id", fixID) + } + + // 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 + } + 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) + } + return nil +} + +func (c *Controller) Name() string { return "fix" } + +func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey } + +func (c *Controller) ConsumerGroup() string { return c.consumerGroup } + +func findIssueToFix(ctx context.Context, observations storage.BuildIssueStore, openIssues storage.OpenIssueStore, 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 entity.BuildIssue{}, false, nil + } + if found { + continue + } + issue, err := openIssues.Get(ctx, record.TaskID, record.UnitID) + if errors.Is(err, storage.ErrNotFound) { + continue + } + if err != nil { + return entity.BuildIssue{}, false, err + } + if 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/fix_test.go b/stovepipe/controller/fix/fix_test.go new file mode 100644 index 000000000..a1c575245 --- /dev/null +++ b/stovepipe/controller/fix/fix_test.go @@ -0,0 +1,322 @@ +// 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" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/fixer" + fixermock "github.com/uber/submitqueue/stovepipe/extension/fixer/mock" + "github.com/uber/submitqueue/stovepipe/extension/storage" + 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 ( + testQueue = "validation/main" + testBuildID = "Build/42 " +) + +type controllerMocks struct { + ctrl *gomock.Controller + stores *storagemock.MockFactory + store *storagemock.MockStorage + issues *storagemock.MockOpenIssueStore + observations *storagemock.MockBuildIssueStore + executors *fixermock.MockFactory + executor *fixermock.MockRepairExecutor + logs *observer.ObservedLogs +} + +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), + store: storagemock.NewMockStorage(ctrl), + issues: storagemock.NewMockOpenIssueStore(ctrl), + observations: storagemock.NewMockBuildIssueStore(ctrl), + executors: fixermock.NewMockFactory(ctrl), + executor: fixermock.NewMockRepairExecutor(ctrl), + logs: logs, + } + m.store.EXPECT().GetOpenIssueStore().Return(m.issues).AnyTimes() + m.store.EXPECT().GetBuildIssueStore().Return(m.observations).AnyTimes() + return NewController(zap.New(core).Sugar(), tally.NoopScope, m.stores, m.executors, + consumer.TopicKey("fix"), "stovepipe-fix"), m +} + +func testIssue() entity.OpenIssue { + return entity.OpenIssue{TaskID: "test", UnitID: "unit/1 ", FirstSeenBuildID: testBuildID} +} + +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}, + Limit: buildIssuePageSize, + } +} + +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(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 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(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 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.OpenIssue{TaskID: older.TaskID, FirstSeenBuildID: "older-build"}, 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(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 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(issue, nil), + m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))). + Return([]entity.BuildIssue{{TaskID: "resolved", FixID: "fix/1"}}, nil), + ) + require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) +} + +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(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 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.OpenIssue{}, storage.ErrNotFound) + } else { + m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID). + Return(entity.OpenIssue{TaskID: issue.TaskID, FirstSeenBuildID: firstSeenBuildID}, nil) + } + 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 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) + 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)) + } + require.Len(t, m.logs.All(), 1) + assert.Equal(t, zap.ErrorLevel, m.logs.All()[0].Level) + }) + } +} + +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 + }{ + {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}, + } { + t.Run(test.name, func(t *testing.T) { + controller, 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) + } else { + require.NoError(t, err) + assert.Equal(t, "another-fix", m.logs.All()[1].ContextMap()["fix_id"]) + } + }) + } +} + +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.OpenIssue{}, assert.AnError) + return + } + lookup.Return(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) + }) + } +} diff --git a/stovepipe/extension/fixer/README.md b/stovepipe/extension/fixer/README.md index 4486c7c24..996ff6270 100644 --- a/stovepipe/extension/fixer/README.md +++ b/stovepipe/extension/fixer/README.md @@ -2,8 +2,8 @@ [`RepairExecutor`](fixer.go) submits repair work for one [`OpenIssue`](../../entity/open_issue.go). The executor can invoke an in-process implementation or wrap an external provider. -`Submit` returns promptly with a nonempty opaque provider attempt ID on acceptance, or an error. Acceptance does not establish recovery. +`Submit` returns promptly with a nonempty opaque fix ID on acceptance, or an error. Acceptance does not establish recovery. -An error, including a timeout or cancellation, can occur after the provider accepted the work, so it must not cause an automatic resubmission. Implementations return plain dependency errors and must support concurrent calls. +An error, including a timeout or cancellation, can occur after the provider accepted the work. Return a known fix ID even alongside an error. Retry policy belongs to the caller; retrying without an ID can start duplicate work. Implementations return plain dependency errors and must support concurrent calls. `Factory.For` resolves an executor bound to `Config.QueueName`. Generated mocks are available in [`mock`](mock/fixer_mock.go). diff --git a/stovepipe/extension/fixer/fixer.go b/stovepipe/extension/fixer/fixer.go index 0622f0be9..533849367 100644 --- a/stovepipe/extension/fixer/fixer.go +++ b/stovepipe/extension/fixer/fixer.go @@ -26,10 +26,9 @@ import ( // RepairExecutor submits repair work for one open issue within a configured queue. // Implementations must be concurrency-safe and return promptly after submission. type RepairExecutor interface { - // Submit returns a nonempty opaque provider attempt ID on acceptance. - // Acceptance does not establish recovery. - // Repeated calls may start additional attempts. A plain error can follow acceptance, - // including on context cancellation; callers must not automatically repeat Submit. + // 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) }