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) }