From f5574b243824f486c017f0fe28b808e30aed3b57 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 6 Nov 2026 22:10:42 +0000 Subject: [PATCH] feat(stovepipe): recover per-executor repair history in the DLQ Summary: Recover every executor's durable submission or skip without submitting backend work. Pending or missing eligible repair records retain an acceptance-unknown event. Wire the executor registry into recovery and remove the superseded build-level selection and history helpers. This is the final layer of the repair-executor stack, building on normal batch dispatch. Test Plan: `TestFixDLQRetainsAllDispositionsWithoutSubmitting`, `TestFixDLQRetainsUnknownAcceptance`, and `TestFixDLQDiscardsInvalidMessages` cover retained outcomes, unknown acceptance, invalid deliveries, and the absence of executor submissions. The 28-target Stovepipe suite also verifies reference-server wiring. Lint, tidy, and Gazelle checks verify the repository metadata. Revert Plan: Revert this PR together with the dependent dispatch and executor-contract changes before deploying the earlier build-level repair flow. --- doc/rfc/stovepipe/request-log.md | 2 +- service/stovepipe/server/main.go | 5 +- service/stovepipe/server/main_test.go | 2 +- stovepipe/controller/fix/README.md | 2 +- stovepipe/controller/fix/dlq.go | 54 ++++++++++++++----- stovepipe/controller/fix/fix.go | 35 ------------- stovepipe/controller/fix/history.go | 29 ---------- stovepipe/controller/fix/history_test.go | 67 ++++++++---------------- 8 files changed, 70 insertions(+), 126 deletions(-) diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index 04b0ed8e3..e9df8e561 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -143,7 +143,7 @@ State entries retain the selected baseline URI when known, including an empty UR Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording. Its `project_fact_count` and `broken_project_fact_count` metadata contain the total and the number with degree above zero, counted from persisted facts after their indexes are durable. Project completion projection requires both counts and assigns them together in `RequestSummary` from the retained event. -`build_analysis_completed` and `issues_reconciled` use the build ID as their occurrence key. `fix_submission_abandoned` uses the build ID. Per-item `fix_submitted` and `fix_submission_skipped` events use a digest of opening build, task, unit, and executor identity, retaining the verbatim fix ID or skip reason and executor ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp. +`build_analysis_completed` and `issues_reconciled` use the build ID as their occurrence key. Per-item `fix_submitted`, `fix_submission_skipped`, and `fix_submission_abandoned` events use a digest of opening build, task, unit, and executor identity, retaining the verbatim fix ID or skip reason and executor ID in metadata. Existing issue-tracking abandonment uses a digest of the failed notification ID to fit the storage key limit. The materializer prefixes each key with its event kind, and request identity scopes all entries. Replays retain the first stored timestamp. Analysis and repair events do not advance Request state or alter validation facts. Analysis completion is retained only when the Request records the analyzed terminal build as its winning outcome, immediately before the final issue-tracking handoff. Its issue count and timestamp are an immutable snapshot of the first retained completion, including zero for an empty result. Retries still persist detected issues and publish the handoff, but reuse that snapshot if later detector results differ; count discrepancies are logged rather than treated as history conflicts. Other semantic conflicts remain errors. `issues_reconciled` records closure mode `reconciled`, `suppressed_cancelled`, or `suppressed_task_wide_issue`; suppression does not mean reconciliation failed. Skipped repair uses reason `no_eligible_issue`; abandoned repair uses `acceptance_unknown`. Raw dependency errors remain in operational logs. diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 3d9696c4f..f17d4b858 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -265,7 +265,7 @@ func run() error { if err != nil { return err } - dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry) + dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry, repairExecutors) if err != nil { return err } @@ -459,6 +459,7 @@ func registerDLQControllers( store storage.Factory, materializer requestlog.Materializer, registry consumer.TopicRegistry, + repairExecutors fixer.Factory, ) (int, error) { var count int @@ -492,7 +493,7 @@ func registerDLQControllers( } count++ - fixDLQController := fix.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyFix), "stovepipe-fix-dlq") + fixDLQController := fix.NewDLQController(logger, scope, store, materializer, repairExecutors, dlq.TopicKey(stovepipemq.TopicKeyFix), "stovepipe-fix-dlq") if err := c.Register(fixDLQController); err != nil { return count, fmt.Errorf("failed to register fix dlq controller: %w", err) } diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index 72d5852ad..57a11b331 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -70,7 +70,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New(), noopRepairExecutorFactory{}) require.NoError(t, err) - _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry) + _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry, noopRepairExecutorFactory{}) require.NoError(t, err) return registry, primary.controllers, deadLetter.controllers diff --git a/stovepipe/controller/fix/README.md b/stovepipe/controller/fix/README.md index 8de074c34..b1f4cfafb 100644 --- a/stovepipe/controller/fix/README.md +++ b/stovepipe/controller/fix/README.md @@ -14,4 +14,4 @@ Each durable disposition restores its request-history event before being skipped Issue tracking publishes another nudge after each nonempty opening pass, so late issues are offered independently of earlier submissions. Queue redelivery rederives pending work and may offer a smaller list. Reopening in another build creates a new identity. A fix ID must reflect actual backend acceptance for every item carrying it; returning an old ID alone does not deliver new work. -The DLQ controller reads the same durable decision without calling the executor. It repairs submitted or skipped history when possible; otherwise, an eligible issue without a fix ID retains `fix_submission_abandoned` once per build with reason `acceptance_unknown`. An executor error without an ID can follow acceptance, so this event does not claim rejection. Source and history errors retry. Unusable payloads, unresolvable queues, and missing or mismatched sources are logged and acknowledged. The reference host registers recovery with unlimited, always-retryable DLQ processing to preserve history repair through a storage outage. +The DLQ controller restores all available disposition history without submitting work. Pending rows and still-open items without repair rows retain `fix_submission_abandoned` with reason `acceptance_unknown`. Hosts register this reconciler with unlimited, always-retryable DLQ processing. Executor IDs must remain registered until outstanding history repair completes. diff --git a/stovepipe/controller/fix/dlq.go b/stovepipe/controller/fix/dlq.go index 65130a8fb..dbe0a5792 100644 --- a/stovepipe/controller/fix/dlq.go +++ b/stovepipe/controller/fix/dlq.go @@ -18,6 +18,8 @@ import ( "context" "fmt" + "errors" + "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" @@ -26,6 +28,7 @@ import ( mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/fixer" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) @@ -36,16 +39,17 @@ type DLQController struct { metricsScope tally.Scope stores storage.Factory materializer requestlog.Materializer + executors fixer.Factory topicKey consumer.TopicKey consumerGroup string } // NewDLQController creates repair history recovery. Hosts must use unlimited, // always-retryable DLQ processing so a history outage cannot discard its retry trigger. -func NewDLQController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, topicKey consumer.TopicKey, consumerGroup string) *DLQController { +func NewDLQController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, executors fixer.Factory, topicKey consumer.TopicKey, consumerGroup string) *DLQController { return &DLQController{ logger: logger.Named("fix_dlq_controller"), metricsScope: scope.SubScope("fix_dlq_controller"), - stores: stores, materializer: materializer, topicKey: topicKey, consumerGroup: consumerGroup, + stores: stores, materializer: materializer, executors: executors, topicKey: topicKey, consumerGroup: consumerGroup, } } @@ -75,21 +79,45 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) } return err } - observation, eligible, err := findIssueToFix(ctx, store.GetBuildIssueStore(), store.GetIssueStateStore(), fix.GetId()) + registrations, err := c.executors.For(fixer.Config{QueueName: fix.GetQueueName()}) if err != nil { - return fmt.Errorf("load repair decision for recovery: %w", err) + return err } - if observation.FixID != "" { - return persistFixSubmittedLog(ctx, c.materializer, store, request, fix.GetId(), observation) + if err := validateRegistrations(registrations); err != nil { + return err } - if !eligible { - return persistFixSubmissionDecision(ctx, c.materializer, store, request, fix.GetId(), - entity.RequestEventFixSubmissionSkipped, "no_eligible_issue") + observations, eligible, err := collectIssues(ctx, store, fix.GetId()) + if err != nil { + return fmt.Errorf("load repair decisions: %w", err) + } + open := map[entity.OpenIssue]bool{} + for _, issue := range eligible { + open[issue] = true + } + for _, registration := range registrations { + for _, observation := range observations { + issue := entity.OpenIssue{TaskID: observation.TaskID, UnitID: observation.UnitID, FirstSeenBuildID: fix.GetId()} + result, err := store.GetIssueRepairStore().Get(ctx, issue, registration.ID) + if errors.Is(err, storage.ErrNotFound) { + if !open[issue] { + continue + } + result = entity.IssueRepair{Issue: issue, ExecutorID: registration.ID} + } else if err != nil { + return err + } + if result.FixID == "" && result.SkipReason == "" { + if err := persistRepairEventLog(ctx, c.materializer, store, request, result, entity.RequestEventFixSubmissionAbandoned, "acceptance_unknown"); err != nil { + return err + } + continue + } + if err := persistRepairSubmissionLog(ctx, c.materializer, store, request, result); err != nil { + return err + } + } } - // A provider can have accepted work even when no ID was returned. Preserve - // that uncertainty and leave repair associations and issue status alone. - return persistFixSubmissionDecision(ctx, c.materializer, store, request, fix.GetId(), - entity.RequestEventFixSubmissionAbandoned, "acceptance_unknown") + return nil } func (c *DLQController) Name() string { return string(c.topicKey) } diff --git a/stovepipe/controller/fix/fix.go b/stovepipe/controller/fix/fix.go index cc1ba8af1..117148438 100644 --- a/stovepipe/controller/fix/fix.go +++ b/stovepipe/controller/fix/fix.go @@ -138,38 +138,3 @@ func collectIssues(ctx context.Context, store storage.Storage, buildID string) ( bounds.After = storage.BuildIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} } } - -func findIssueToFix(ctx context.Context, observations storage.BuildIssueStore, issues storage.IssueStateStore, buildID string) (entity.BuildIssue, bool, error) { - bounds := storage.BuildIssueRange{Limit: buildIssuePageSize} - var candidate entity.BuildIssue - found := false - for { - page, err := observations.List(ctx, buildID, bounds) - if err != nil { - return entity.BuildIssue{}, false, err - } - if len(page) == 0 { - return candidate, found, nil - } - for _, record := range page { - if record.FixID != "" { - return record, false, nil - } - if found { - continue - } - issue, err := issues.Get(ctx, record.TaskID, record.UnitID) - if errors.Is(err, storage.ErrNotFound) { - continue - } - if err != nil { - return entity.BuildIssue{}, false, err - } - if issue.Status == entity.IssueStatusOpen && issue.FirstSeenBuildID == buildID { - candidate, found = record, true - } - } - last := page[len(page)-1] - bounds.After = storage.BuildIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} - } -} diff --git a/stovepipe/controller/fix/history.go b/stovepipe/controller/fix/history.go index 0614ae9bd..d3553c0de 100644 --- a/stovepipe/controller/fix/history.go +++ b/stovepipe/controller/fix/history.go @@ -69,32 +69,3 @@ func persistRepairEventLog(ctx context.Context, materializer requestlog.Material }) return materializer.PersistLog(ctx, store, log) } - -func persistFixSubmittedLog(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, buildID string, observation entity.BuildIssue) error { - // The queue-scoped executor's opaque fix ID identifies the accepted attempt. - // Restoring this from the durable association must precede any replay skip. - // Opaque fix IDs can fill the storage column; the event prefix must still fit its own key. - identity := sha256.Sum256([]byte(observation.FixID)) - log := requestlog.NewRequestEventLog(request, entity.RequestEventFixSubmitted, fmt.Sprintf("%x", identity), - map[string]string{ - requestlog.MetadataKeyBuildID: buildID, - requestlog.MetadataKeyTaskID: observation.TaskID, - requestlog.MetadataKeyUnitID: observation.UnitID, - requestlog.MetadataKeyFixID: observation.FixID, - }) - if err := materializer.PersistLog(ctx, store, log); err != nil { - return fmt.Errorf("retain submitted repair %q for build %q: %w", observation.FixID, buildID, err) - } - return nil -} - -func persistFixSubmissionDecision(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, request entity.Request, buildID string, event entity.RequestEvent, reason string) error { - log := requestlog.NewRequestEventLog(request, event, buildID, map[string]string{ - requestlog.MetadataKeyBuildID: buildID, - requestlog.MetadataKeyReason: reason, - }) - if err := materializer.PersistLog(ctx, store, log); err != nil { - return fmt.Errorf("retain repair decision for build %q: %w", buildID, err) - } - return nil -} diff --git a/stovepipe/controller/fix/history_test.go b/stovepipe/controller/fix/history_test.go index e124de291..c9f83524d 100644 --- a/stovepipe/controller/fix/history_test.go +++ b/stovepipe/controller/fix/history_test.go @@ -16,65 +16,44 @@ package fix import ( "context" + "fmt" "testing" - "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" - consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/stovepipe/core/messagequeue" - "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/mock/gomock" "go.uber.org/zap" ) -func TestFixDLQRetainsDecisionWithoutSubmitting(t *testing.T) { - for _, tt := range []struct { - name string - fixID string - issueStatus entity.IssueStatus - wantEvent entity.RequestEvent - wantReason string - }{ - {name: "associated", fixID: "fix/1", wantEvent: entity.RequestEventFixSubmitted}, - {name: "eligible", issueStatus: entity.IssueStatusOpen, wantEvent: entity.RequestEventFixSubmissionAbandoned, wantReason: "acceptance_unknown"}, - {name: "ineligible", issueStatus: entity.IssueStatusClosed, wantEvent: entity.RequestEventFixSubmissionSkipped, wantReason: "no_eligible_issue"}, - } { - t.Run(tt.name, func(t *testing.T) { - _, m := newTestController(t) - c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, messagequeue.TopicKeyFix, "recovery") +func TestFixDLQRetainsAllDispositionsWithoutSubmitting(t *testing.T) { + c, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + m.repairs[issue] = map[string]entity.IssueRepair{"first": {Issue: issue, ExecutorID: "first", FixID: "fix"}} + dlq := NewDLQController(c.logger, tally.NoopScope, m.stores, m.materializer, m.executors, c.topicKey, c.consumerGroup) + require.NoError(t, dlq.Process(context.Background(), m.delivery(repairMessage(t)))) + require.Len(t, m.history, 1) +} + +func TestFixDLQRetainsUnknownAcceptance(t *testing.T) { + for _, pending := range []bool{false, true} { + t.Run(fmt.Sprint(pending), func(t *testing.T) { + c, m := newTestController(t) issue := testIssue() - observation := issueIdentity(issue) - observation.FixID = tt.fixID - for i := 0; i < 3; i++ { - m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).Return([]entity.BuildIssue{observation}, nil) - if tt.fixID == "" { - current := storedIssue(issue) - current.Status = tt.issueStatus - m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(current, nil) - m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(observation)).Return(nil, nil) - } - d := consumermock.NewMockDelivery(m.ctrl) - d.EXPECT().Message().Return(repairMessage(t)) - if i == 0 { - *m.historyFailure = assert.AnError - require.ErrorIs(t, c.Process(context.Background(), d), assert.AnError) - require.Empty(t, m.history) - *m.historyFailure = nil - } else { - require.NoError(t, c.Process(context.Background(), d)) - } + m.expectIssue(issue) + if pending { + m.repairs[issue] = map[string]entity.IssueRepair{"first": {Issue: issue, ExecutorID: "first", Version: 1}} } - require.Len(t, m.history, 1) + dlq := NewDLQController(c.logger, tally.NoopScope, m.stores, m.materializer, m.executors, c.topicKey, c.consumerGroup) + require.NoError(t, dlq.Process(context.Background(), m.delivery(repairMessage(t)))) for _, log := range m.history { - require.Equal(t, tt.wantEvent, log.Event) - require.Equal(t, testBuildID, log.Metadata[requestlog.MetadataKeyBuildID]) - require.Equal(t, tt.wantReason, log.Metadata[requestlog.MetadataKeyReason]) + require.Equal(t, entity.RequestEventFixSubmissionAbandoned, log.Event) } + require.Len(t, m.history, 1) }) } } @@ -106,7 +85,7 @@ func TestFixDLQDiscardsInvalidMessages(t *testing.T) { } { t.Run(tt.name, func(t *testing.T) { _, m := newTestController(t) - c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, messagequeue.TopicKeyFix, "recovery") + c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.stores, m.materializer, m.executors, messagequeue.TopicKeyFix, "recovery") require.NoError(t, c.Process(context.Background(), m.delivery(tt.configure(t, m)))) require.Empty(t, m.history) })