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.