feat(stovepipe): extend fixer DLQ recovery to every issue and backend

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.
 

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

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

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.

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

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

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

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