diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index 0b9075070..e9df8e561 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -143,7 +143,7 @@ State entries retain the selected baseline URI when known, including an empty UR Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording. Its `project_fact_count` and `broken_project_fact_count` metadata contain the total and the number with degree above zero, counted from persisted facts after their indexes are durable. Project completion projection requires both counts and assigns them together in `RequestSummary` from the retained event. -`build_analysis_completed`, `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. Per-item `fix_submitted`, `fix_submission_skipped`, and `fix_submission_abandoned` events use a digest of opening build, task, unit, and executor identity, retaining the verbatim fix ID or skip reason and executor ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp. Analysis and repair events do not advance Request state or alter validation facts. Analysis completion is retained only when the Request records the analyzed terminal build as its winning outcome, immediately before the final issue-tracking handoff. Its issue count and timestamp are an immutable snapshot of the first retained completion, including zero for an empty result. Retries still persist detected issues and publish the handoff, but reuse that snapshot if later detector results differ; count discrepancies are logged rather than treated as history conflicts. Other semantic conflicts remain errors. `issues_reconciled` records closure mode `reconciled`, `suppressed_cancelled`, or `suppressed_task_wide_issue`; suppression does not mean reconciliation failed. Skipped repair uses reason `no_eligible_issue`; abandoned repair uses `acceptance_unknown`. Raw dependency errors remain in operational logs. @@ -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 96802c72a..267145772 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 @@ -265,7 +265,7 @@ func run() error { if err != nil { return err } - dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry) + dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry, repairExecutors) if err != nil { return err } @@ -462,6 +462,7 @@ func registerDLQControllers( store storage.Factory, materializer requestlog.Materializer, registry consumer.TopicRegistry, + repairExecutors fixer.Factory, ) (int, error) { var count int @@ -495,7 +496,7 @@ func registerDLQControllers( } count++ - fixDLQController := fix.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyFix), "stovepipe-fix-dlq") + fixDLQController := fix.NewDLQController(logger, scope, store, materializer, repairExecutors, dlq.TopicKey(stovepipemq.TopicKeyFix), "stovepipe-fix-dlq") if err := c.Register(fixDLQController); err != nil { return count, fmt.Errorf("failed to register fix dlq controller: %w", err) } diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index abea3bca0..57a11b331 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -70,7 +70,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New(), noopRepairExecutorFactory{}) require.NoError(t, err) - _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry) + _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry, noopRepairExecutorFactory{}) require.NoError(t, err) return registry, primary.controllers, deadLetter.controllers @@ -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..b1f4cfafb 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. -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. +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 restores all available disposition history without submitting work. Pending rows and still-open items without repair rows retain `fix_submission_abandoned` with reason `acceptance_unknown`. Hosts register this reconciler with unlimited, always-retryable DLQ processing. Executor IDs must remain registered until outstanding history repair completes. diff --git a/stovepipe/controller/fix/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/dlq.go b/stovepipe/controller/fix/dlq.go index 65130a8fb..dbe0a5792 100644 --- a/stovepipe/controller/fix/dlq.go +++ b/stovepipe/controller/fix/dlq.go @@ -18,6 +18,8 @@ import ( "context" "fmt" + "errors" + "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" @@ -26,6 +28,7 @@ import ( mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/fixer" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) @@ -36,16 +39,17 @@ type DLQController struct { metricsScope tally.Scope stores storage.Factory materializer requestlog.Materializer + executors fixer.Factory topicKey consumer.TopicKey consumerGroup string } // NewDLQController creates repair history recovery. Hosts must use unlimited, // always-retryable DLQ processing so a history outage cannot discard its retry trigger. -func NewDLQController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, topicKey consumer.TopicKey, consumerGroup string) *DLQController { +func NewDLQController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, executors fixer.Factory, topicKey consumer.TopicKey, consumerGroup string) *DLQController { return &DLQController{ logger: logger.Named("fix_dlq_controller"), metricsScope: scope.SubScope("fix_dlq_controller"), - stores: stores, materializer: materializer, topicKey: topicKey, consumerGroup: consumerGroup, + stores: stores, materializer: materializer, executors: executors, topicKey: topicKey, consumerGroup: consumerGroup, } } @@ -75,21 +79,45 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) } return err } - observation, eligible, err := findIssueToFix(ctx, store.GetBuildIssueStore(), store.GetIssueStateStore(), fix.GetId()) + registrations, err := c.executors.For(fixer.Config{QueueName: fix.GetQueueName()}) if err != nil { - return fmt.Errorf("load repair decision for recovery: %w", err) + return err } - if observation.FixID != "" { - return persistFixSubmittedLog(ctx, c.materializer, store, request, fix.GetId(), observation) + if err := validateRegistrations(registrations); err != nil { + return err } - if !eligible { - return persistFixSubmissionDecision(ctx, c.materializer, store, request, fix.GetId(), - entity.RequestEventFixSubmissionSkipped, "no_eligible_issue") + observations, eligible, err := collectIssues(ctx, store, fix.GetId()) + if err != nil { + return fmt.Errorf("load repair decisions: %w", err) + } + open := map[entity.OpenIssue]bool{} + for _, issue := range eligible { + open[issue] = true + } + for _, registration := range registrations { + for _, observation := range observations { + issue := entity.OpenIssue{TaskID: observation.TaskID, UnitID: observation.UnitID, FirstSeenBuildID: fix.GetId()} + result, err := store.GetIssueRepairStore().Get(ctx, issue, registration.ID) + if errors.Is(err, storage.ErrNotFound) { + if !open[issue] { + continue + } + result = entity.IssueRepair{Issue: issue, ExecutorID: registration.ID} + } else if err != nil { + return err + } + if result.FixID == "" && result.SkipReason == "" { + if err := persistRepairEventLog(ctx, c.materializer, store, request, result, entity.RequestEventFixSubmissionAbandoned, "acceptance_unknown"); err != nil { + return err + } + continue + } + if err := persistRepairSubmissionLog(ctx, c.materializer, store, request, result); err != nil { + return err + } + } } - // A provider can have accepted work even when no ID was returned. Preserve - // that uncertainty and leave repair associations and issue status alone. - return persistFixSubmissionDecision(ctx, c.materializer, store, request, fix.GetId(), - entity.RequestEventFixSubmissionAbandoned, "acceptance_unknown") + return nil } func (c *DLQController) Name() string { return string(c.topicKey) } diff --git a/stovepipe/controller/fix/fix.go b/stovepipe/controller/fix/fix.go index cca9abc54..117148438 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,34 +109,29 @@ 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, issues storage.IssueStateStore, buildID string) (entity.BuildIssue, bool, error) { +func collectIssues(ctx context.Context, store storage.Storage, buildID string) ([]entity.BuildIssue, []entity.OpenIssue, error) { bounds := storage.BuildIssueRange{Limit: buildIssuePageSize} - var candidate entity.BuildIssue - found := false + var observations []entity.BuildIssue + var eligible []entity.OpenIssue for { - page, err := observations.List(ctx, buildID, bounds) + page, err := store.GetBuildIssueStore().List(ctx, buildID, bounds) if err != nil { - return entity.BuildIssue{}, false, err + return nil, nil, err } if len(page) == 0 { - return candidate, found, nil + return observations, eligible, nil } for _, record := range page { - if record.FixID != "" { - return record, false, nil - } - if found { - continue - } - issue, err := issues.Get(ctx, record.TaskID, record.UnitID) + 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 entity.BuildIssue{}, false, err + return nil, nil, err } if issue.Status == entity.IssueStatusOpen && issue.FirstSeenBuildID == buildID { - candidate, found = record, true + eligible = append(eligible, entity.OpenIssue{TaskID: record.TaskID, UnitID: record.UnitID, FirstSeenBuildID: buildID}) } } last := page[len(page)-1] 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..d3553c0de 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,31 +45,27 @@ func loadRequestForBuild(ctx context.Context, store storage.Storage, queue, buil return request, nil } -func persistFixSubmittedLog(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, buildID string, observation entity.BuildIssue) error { - // The queue-scoped executor's opaque fix ID identifies the accepted attempt. - // Restoring this from the durable association must precede any replay skip. - // Opaque fix IDs can fill the storage column; the event prefix must still fit its own key. - identity := sha256.Sum256([]byte(observation.FixID)) - log := requestlog.NewRequestEventLog(request, entity.RequestEventFixSubmitted, fmt.Sprintf("%x", identity), - map[string]string{ - requestlog.MetadataKeyBuildID: buildID, - requestlog.MetadataKeyTaskID: observation.TaskID, - requestlog.MetadataKeyUnitID: observation.UnitID, - requestlog.MetadataKeyFixID: observation.FixID, - }) - if err := materializer.PersistLog(ctx, store, log); err != nil { - return fmt.Errorf("retain submitted repair %q for build %q: %w", observation.FixID, buildID, err) +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 nil + return persistRepairEventLog(ctx, materializer, store, request, result, event, result.SkipReason) } -func persistFixSubmissionDecision(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, buildID string, event entity.RequestEvent, reason string) error { - log := requestlog.NewRequestEventLog(request, event, buildID, map[string]string{ - requestlog.MetadataKeyBuildID: buildID, - requestlog.MetadataKeyReason: reason, - }) - if err := materializer.PersistLog(ctx, store, log); err != nil { - return fmt.Errorf("retain repair decision for build %q: %w", buildID, err) +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 } - return nil + 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) } diff --git a/stovepipe/controller/fix/history_test.go b/stovepipe/controller/fix/history_test.go index 0097a3cc4..c9f83524d 100644 --- a/stovepipe/controller/fix/history_test.go +++ b/stovepipe/controller/fix/history_test.go @@ -16,105 +16,44 @@ package fix import ( "context" - "strings" + "fmt" "testing" - "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" - consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/stovepipe/core/messagequeue" - "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/mock/gomock" "go.uber.org/zap" ) -func 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 TestFixDLQRetainsAllDispositionsWithoutSubmitting(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + m.repairs[issue] = map[string]entity.IssueRepair{"first": {Issue: issue, ExecutorID: "first", FixID: "fix"}} + dlq := NewDLQController(c.logger, tally.NoopScope, m.stores, m.materializer, m.executors, c.topicKey, c.consumerGroup) + require.NoError(t, dlq.Process(context.Background(), m.delivery(repairMessage(t)))) + require.Len(t, m.history, 1) } -func TestFixDLQRetainsDecisionWithoutSubmitting(t *testing.T) { - for _, tt := range []struct { - name string - fixID string - issueStatus entity.IssueStatus - wantEvent entity.RequestEvent - wantReason string - }{ - {name: "associated", fixID: "fix/1", wantEvent: entity.RequestEventFixSubmitted}, - {name: "eligible", issueStatus: entity.IssueStatusOpen, wantEvent: entity.RequestEventFixSubmissionAbandoned, wantReason: "acceptance_unknown"}, - {name: "ineligible", issueStatus: entity.IssueStatusClosed, wantEvent: entity.RequestEventFixSubmissionSkipped, wantReason: "no_eligible_issue"}, - } { - t.Run(tt.name, func(t *testing.T) { - _, m := newTestController(t) - c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, messagequeue.TopicKeyFix, "recovery") +func TestFixDLQRetainsUnknownAcceptance(t *testing.T) { + for _, pending := range []bool{false, true} { + t.Run(fmt.Sprint(pending), func(t *testing.T) { + c, m := newTestController(t) issue := testIssue() - observation := issueIdentity(issue) - observation.FixID = tt.fixID - for i := 0; i < 3; i++ { - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).Return([]entity.BuildIssue{observation}, nil) - if tt.fixID == "" { - current := storedIssue(issue) - current.Status = tt.issueStatus - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(current, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(observation)).Return(nil, nil) - } - d := consumermock.NewMockDelivery(m.ctrl) - d.EXPECT().Message().Return(repairMessage(t)) - if i == 0 { - *m.historyFailure = assert.AnError - require.ErrorIs(t, c.Process(context.Background(), d), assert.AnError) - require.Empty(t, m.history) - *m.historyFailure = nil - } else { - require.NoError(t, c.Process(context.Background(), d)) - } + m.expectIssue(issue) + if pending { + m.repairs[issue] = map[string]entity.IssueRepair{"first": {Issue: issue, ExecutorID: "first", Version: 1}} } - require.Len(t, m.history, 1) + dlq := NewDLQController(c.logger, tally.NoopScope, m.stores, m.materializer, m.executors, c.topicKey, c.consumerGroup) + require.NoError(t, dlq.Process(context.Background(), m.delivery(repairMessage(t)))) for _, log := range m.history { - require.Equal(t, tt.wantEvent, log.Event) - require.Equal(t, testBuildID, log.Metadata[requestlog.MetadataKeyBuildID]) - require.Equal(t, tt.wantReason, log.Metadata[requestlog.MetadataKeyReason]) + require.Equal(t, entity.RequestEventFixSubmissionAbandoned, log.Event) } + require.Len(t, m.history, 1) }) } } @@ -146,33 +85,9 @@ func TestFixDLQDiscardsInvalidMessages(t *testing.T) { } { t.Run(tt.name, func(t *testing.T) { _, m := newTestController(t) - c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, messagequeue.TopicKeyFix, "recovery") + c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, m.executors, messagequeue.TopicKeyFix, "recovery") require.NoError(t, c.Process(context.Background(), m.delivery(tt.configure(t, m)))) require.Empty(t, m.history) }) } } - -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/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index c50e915c3..f08bb614e 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -36,7 +36,7 @@ The post-read cursor write and processed-build updates, including unchanged stat Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal pass retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh. Encountering a task-wide issue (empty unit ID) sets `HasTaskWideIssues` in that same versioned record. This fact remains set across batches and retries, including when the issue was already applied by an earlier pass. -Once the entire build-issue range is exhausted, an opening pass with a nonempty task cursor publishes a build-scoped repair request before saving its next phase. Empty passes and closing sweeps do not publish. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. The repair message identity includes the persisted checkpoint version: retries before phase completion share an identity, while later refreshes of one long-lived job can trigger another evaluation without depending on queue garbage collection. The fix controller selects an eligible issue and submits at most one repair per build. +Once the entire build-issue range is exhausted, an opening pass with a nonempty task cursor publishes a build-scoped repair request before saving its next phase. Empty passes and closing sweeps do not publish. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. The repair message identity includes the persisted checkpoint version: retries before phase completion share an identity, while later refreshes of one long-lived job can trigger another evaluation without depending on queue garbage collection. The fix controller offers each registered executor all pending issue openings first seen in that build; each executor chooses how to group its repairs. Once the entire build-issue range is exhausted: 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. diff --git a/stovepipe/extension/fixer/fixer.go b/stovepipe/extension/fixer/fixer.go index 533849367..cf4a79a77 100644 --- a/stovepipe/extension/fixer/fixer.go +++ b/stovepipe/extension/fixer/fixer.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package fixer defines repair submission for an open issue. +// Package fixer defines repair submission for open issues. package fixer //go:generate mockgen -source=fixer.go -destination=mock/fixer_mock.go -package=mock @@ -23,22 +23,42 @@ import ( "github.com/uber/submitqueue/stovepipe/entity" ) -// RepairExecutor submits repair work for one open issue within a configured queue. -// Implementations must be concurrency-safe and return promptly after submission. +// RepairExecutor chooses how to group repair work. Implementations must be concurrency-safe. type RepairExecutor interface { - // Submit returns a nonempty opaque fix ID on acceptance, without establishing recovery. - // An ID returned alongside an error still identifies the repair attempt. - // Errors without an ID may follow acceptance; repeated calls can start additional work. - Submit(context.Context, entity.OpenIssue) (string, error) + // Submit returns per-item backend acceptance or terminal skips, not repair completion. + // Valid submissions returned alongside errors are persisted; omitted items remain pending. + // Retries may include subsets or additional items; implementations deduplicate by opening. + // Pending rows exist before Submit. Executors manage metadata through injected storage, + // preserving submission fields and reloading on version conflicts. + Submit(context.Context, []entity.OpenIssue) ([]Submission, error) } -// Config identifies the queue whose repair executor is requested. +// Submission associates one input item with accepted repair work or a deliberate skip. +// Acceptance does not indicate repair completion or success. +type Submission struct { + // Issue identifies the input opening, including its first-seen build. + Issue entity.OpenIssue + // FixID is the backend request or run ID for accepted work; multiple items may share it. + FixID string + // SkipReason is nonempty for a terminal skip. Exactly one of FixID and SkipReason is required. + SkipReason string +} + +// Registration binds a stable queue-local identity to an executor. +type Registration struct { + // ID is durable across configuration changes and must not be reused for another backend. + ID string + // Executor implements the registered repair policy. + Executor RepairExecutor +} + +// Config identifies a validation queue. type Config struct { // QueueName identifies the validation queue. QueueName string } -// Factory resolves queue-bound executors; routing belongs to service wiring. +// Factory resolves all registered executors; an empty slice disables dispatch. type Factory interface { - For(Config) (RepairExecutor, error) + For(Config) ([]Registration, error) } diff --git a/stovepipe/extension/fixer/mock/fixer_mock.go b/stovepipe/extension/fixer/mock/fixer_mock.go index 25e4779f5..9e9912274 100644 --- a/stovepipe/extension/fixer/mock/fixer_mock.go +++ b/stovepipe/extension/fixer/mock/fixer_mock.go @@ -43,10 +43,10 @@ func (m *MockRepairExecutor) EXPECT() *MockRepairExecutorMockRecorder { } // Submit mocks base method. -func (m *MockRepairExecutor) Submit(arg0 context.Context, arg1 entity.OpenIssue) (string, error) { +func (m *MockRepairExecutor) Submit(arg0 context.Context, arg1 []entity.OpenIssue) ([]fixer.Submission, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "Submit", arg0, arg1) - ret0, _ := ret[0].(string) + ret0, _ := ret[0].([]fixer.Submission) ret1, _ := ret[1].(error) return ret0, ret1 } @@ -82,10 +82,10 @@ func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { } // For mocks base method. -func (m *MockFactory) For(arg0 fixer.Config) (fixer.RepairExecutor, error) { +func (m *MockFactory) For(arg0 fixer.Config) ([]fixer.Registration, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "For", arg0) - ret0, _ := ret[0].(fixer.RepairExecutor) + ret0, _ := ret[0].([]fixer.Registration) ret1, _ := ret[1].(error) return ret0, ret1 } diff --git a/stovepipe/extension/fixer/noop/noop.go b/stovepipe/extension/fixer/noop/noop.go index 7ab458c96..88e128c06 100644 --- a/stovepipe/extension/fixer/noop/noop.go +++ b/stovepipe/extension/fixer/noop/noop.go @@ -29,6 +29,10 @@ func New() fixer.RepairExecutor { return executor{} } -func (executor) Submit(_ context.Context, issue entity.OpenIssue) (string, error) { - return "noop/" + issue.FirstSeenBuildID, nil +func (executor) Submit(_ context.Context, issues []entity.OpenIssue) ([]fixer.Submission, error) { + submissions := make([]fixer.Submission, 0, len(issues)) + for _, issue := range issues { + submissions = append(submissions, fixer.Submission{Issue: issue, FixID: "noop/" + issue.FirstSeenBuildID}) + } + return submissions, nil }