feat(stovepipe): enable fix consumer with no-op repairs

service/stovepipe/README.md

@@ -11,6 +11,7 @@ Runnable wiring for the **Stovepipe** domain — a single-service domain (the do
 - **buildsignal consumer** (`TopicKeyBuildSignal`) — polls/records the build's terminal status and releases the queue's in-flight slot, then publishes to `record`.
 - **buildanalysis consumer** (`TopicKeyBuildAnalysis`) — periodically detects and persists build issues, then publishes queue-partitioned issue-tracking notifications.
 - **issuetrack consumer** (`TopicKeyIssueTrack`) — applies reported build issues to current issue state and history, closing absent issues only when final reconciliation permits it.
+- **fix consumer** (`TopicKeyFix`) — selects an eligible issue and records one repair association per build using the reference host's no-op executor.
 - **record consumer** (`TopicKeyRecord`) — writes the whole-repository validation fact and, for a green fact, advances the queue's last-green bookmark and promotion ref.
 - **hook consumer** (`TopicKeyHook`) — receives `validation.repository.started` from process and `validation.repository.recorded` or `validation.repository.cancelled` from record. The current resolver invokes only `noop`, so these events have no external side effect. Its topic name is domain-qualified (`stovepipe-hook`) because the key is shared across domains.
 - **DLQ consumers** — registered for process, build, buildsignal, buildanalysis, issuetrack, record, and hook, with stage-specific recovery behavior. Issue-tracking recovery abandons incomplete passes while preserving accepted changes; it does not rerun their remaining batches.
@@ -21,10 +22,11 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `queu
 
 ## Wiring notes
 
-`server/main.go` is the composition root and supplies the concrete extension implementations. Two are deliberately demo-only and must be replaced for any real deployment:
+`server/main.go` is the composition root and supplies the concrete extension implementations. Demo implementations must be replaced for any real deployment:
 
 - **`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/<buildID>`. 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.
 
 ## Layout
 

service/stovepipe/server/BUILD.bazel

@@ -29,6 +29,7 @@ go_library(
         "//stovepipe/controller/buildanalysis:go_default_library",
         "//stovepipe/controller/buildsignal:go_default_library",
         "//stovepipe/controller/dlq:go_default_library",
+        "//stovepipe/controller/fix:go_default_library",
         "//stovepipe/controller/issuetrack:go_default_library",
         "//stovepipe/controller/process:go_default_library",
         "//stovepipe/controller/record:go_default_library",
@@ -38,6 +39,8 @@ go_library(
         "//stovepipe/extension/buildissue/noop:go_default_library",
         "//stovepipe/extension/buildrunner:go_default_library",
         "//stovepipe/extension/buildrunner/fake:go_default_library",
+        "//stovepipe/extension/fixer:go_default_library",
+        "//stovepipe/extension/fixer/noop:go_default_library",
         "//stovepipe/extension/projectstatus/noop:go_default_library",
         "//stovepipe/extension/queueconfig/default:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
@@ -86,15 +89,21 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//api/base/hook:go_default_library",
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
+        "//platform/consumer/mock:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//stovepipe/controller/dlq:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
+        "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/buildissue/noop:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
         "@org_uber_go_zap//zaptest:go_default_library",
     ],
 )

service/stovepipe/server/main.go

@@ -51,6 +51,7 @@ import (
 	"github.com/uber/submitqueue/stovepipe/controller/buildanalysis"
 	"github.com/uber/submitqueue/stovepipe/controller/buildsignal"
 	"github.com/uber/submitqueue/stovepipe/controller/dlq"
+	"github.com/uber/submitqueue/stovepipe/controller/fix"
 	"github.com/uber/submitqueue/stovepipe/controller/issuetrack"
 	"github.com/uber/submitqueue/stovepipe/controller/process"
 	"github.com/uber/submitqueue/stovepipe/controller/record"
@@ -60,6 +61,8 @@ import (
 	buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop"
 	"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
 	buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake"
+	"github.com/uber/submitqueue/stovepipe/extension/fixer"
+	fixernoop "github.com/uber/submitqueue/stovepipe/extension/fixer/noop"
 	projectstatusnoop "github.com/uber/submitqueue/stovepipe/extension/projectstatus/noop"
 	queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
 	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
@@ -92,6 +95,12 @@ func (fakeBuildRunnerFactory) For(cfg buildrunner.Config) (buildrunner.BuildRunn
 	return buildrunnerfake.New(cfg), nil
 }
 
+type noopRepairExecutorFactory struct{}
+
+func (noopRepairExecutorFactory) For(fixer.Config) (fixer.RepairExecutor, error) {
+	return fixernoop.New(), nil
+}
+
 // hookResolver sends every event to the no-op hook. Which hooks an event goes to is host
 // policy, so the resolver lives here rather than in the extension package. A deployment
 // with real integrations swaps this for one that selects on the event's source and type.
@@ -248,10 +257,11 @@ func run() error {
 	// silently duplicated across controllers.
 	sourceControl := fakeSourceControlFactory{}
 	brf := fakeBuildRunnerFactory{}
+	repairExecutors := noopRepairExecutorFactory{}
 
 	storageFty := storageFactory{backend: store}
 	materializer := requestlog.NewMaterializer(scope)
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{}, detectorFactory)
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{}, detectorFactory, repairExecutors)
 	if err != nil {
 		return err
 	}
@@ -371,6 +381,7 @@ func registerPrimaryControllers(
 	brf buildrunner.Factory,
 	hooks hookext.Hooks,
 	detectorFactory buildissue.DetectorFactory,
+	repairExecutors fixer.Factory,
 ) (int, error) {
 	var count int
 
@@ -414,6 +425,12 @@ func registerPrimaryControllers(
 	}
 	count++
 
+	fixController := fix.NewController(logger, scope, store, repairExecutors, stovepipemq.TopicKeyFix, "stovepipe-fix")
+	if err := c.Register(fixController); err != nil {
+		return count, fmt.Errorf("failed to register fix controller: %w", err)
+	}
+	count++
+
 	recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.New(), sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record")
 	if err := c.Register(recordController); err != nil {
 		return count, fmt.Errorf("failed to register record controller: %w", err)
@@ -580,9 +597,10 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-issuetrack-dlq"),
 		},
 		{
-			Key:   stovepipemq.TopicKeyFix,
-			Name:  "fix",
-			Queue: q,
+			Key:          stovepipemq.TopicKeyFix,
+			Name:         "fix",
+			Queue:        q,
+			Subscription: extqueue.ExtendedSubscriptionConfig(subscriberName, "stovepipe-fix"),
 		},
 		{
 			Key:          stovepipemq.TopicKeyBuildAnalysis,

service/stovepipe/server/main_test.go

@@ -23,12 +23,18 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
 	basehook "github.com/uber/submitqueue/api/base/hook"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
+	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/controller/dlq"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
+	"github.com/uber/submitqueue/stovepipe/entity"
 	buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+	storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock"
+	"go.uber.org/mock/gomock"
 	"go.uber.org/zap/zaptest"
 )
 
@@ -61,7 +67,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con
 	deadLetter := &recordingConsumer{}
 
 	_, err = registerPrimaryControllers(primary, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry,
-		fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New())
+		fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New(), noopRepairExecutorFactory{})
 	require.NoError(t, err)
 
 	_, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry)
@@ -110,22 +116,86 @@ func TestIssueTrackingRegistration(t *testing.T) {
 	assert.Equal(t, "_dlq", cfg.DLQ.TopicSuffix)
 }
 
-func TestEveryPrimaryStageHasRecovery(t *testing.T) {
+func TestValidationStagesHaveRecovery(t *testing.T) {
 	_, primary, deadLetter := registeredControllers(t)
 	recoveryKeys := topicKeys(deadLetter)
 	for _, controller := range primary {
+		if controller.TopicKey() == stovepipemq.TopicKeyFix {
+			continue
+		}
 		t.Run(controller.Name(), func(t *testing.T) {
 			assert.Contains(t, recoveryKeys, dlq.TopicKey(controller.TopicKey()))
 		})
 	}
 }
 
-func TestFixHasPublicationRouteWithoutConsumption(t *testing.T) {
-	registry, primary, _ := registeredControllers(t)
+func TestFixRegistration(t *testing.T) {
+	registry, primary, deadLetter := registeredControllers(t)
 	name, ok := registry.TopicName(stovepipemq.TopicKeyFix)
 	require.True(t, ok)
 	assert.Equal(t, "fix", name)
-	assert.NotContains(t, topicKeys(primary), stovepipemq.TopicKeyFix)
+	assert.Contains(t, topicKeys(primary), stovepipemq.TopicKeyFix)
+	assert.NotContains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyFix))
+}
+
+func TestRegisteredFixControllerRecordsNoopRepairOncePerBuild(t *testing.T) {
+	const queueName = "validation/main"
+	const buildID = "build/1"
+	ctrl := gomock.NewController(t)
+	stores := storagemock.NewMockFactory(ctrl)
+	store := storagemock.NewMockStorage(ctrl)
+	observations := storagemock.NewMockBuildIssueStore(ctrl)
+	issues := storagemock.NewMockIssueStateStore(ctrl)
+	stores.EXPECT().For(storage.Config{QueueName: queueName}).Return(store, nil).AnyTimes()
+	store.EXPECT().GetBuildIssueStore().Return(observations).AnyTimes()
+	store.EXPECT().GetIssueStateStore().Return(issues).AnyTimes()
+
+	observation := entity.BuildIssue{TaskID: "test", UnitID: "unit", Version: 1}
+	observations.EXPECT().List(gomock.Any(), buildID, gomock.Any()).DoAndReturn(
+		func(_ context.Context, _ string, bounds storage.BuildIssueRange) ([]entity.BuildIssue, error) {
+			if bounds.After.TaskID != "" {
+				return nil, nil
+			}
+			return []entity.BuildIssue{observation}, nil
+		}).AnyTimes()
+	issues.EXPECT().Get(gomock.Any(), observation.TaskID, observation.UnitID).Return(entity.IssueState{
+		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
+		})
+
+	registry, err := newTopicRegistry(nil, "subscriber")
+	require.NoError(t, err)
+	primary := &recordingConsumer{}
+	_, err = registerPrimaryControllers(primary, zaptest.NewLogger(t).Sugar(), tally.NoopScope, stores,
+		requestlog.NewMaterializer(tally.NoopScope), registry, fakeSourceControlFactory{}, fakeBuildRunnerFactory{},
+		hookResolver{}, buildissuenoop.New(), noopRepairExecutorFactory{})
+	require.NoError(t, err)
+	var fixController consumer.Controller
+	for _, controller := range primary.controllers {
+		if controller.TopicKey() == stovepipemq.TopicKeyFix {
+			fixController = controller
+			break
+		}
+	}
+	require.NotNil(t, fixController)
+
+	payload, err := stovepipemq.Marshal(&stovepipemq.FixRequest{Id: buildID, QueueName: queueName})
+	require.NoError(t, err)
+	for _, pass := range []string{"pass/1", "pass/2"} {
+		message := entityqueue.NewMessage(buildID+"/"+pass, payload, buildID, nil)
+		message.Tenant = queueName
+		delivery := consumermock.NewMockDelivery(ctrl)
+		delivery.EXPECT().Message().Return(message)
+		require.NoError(t, fixController.Process(context.Background(), delivery))
+	}
 }
 
 func TestHookStage(t *testing.T) {
@@ -168,6 +238,7 @@ func TestPipelineSubscriptionRetryPolicies(t *testing.T) {
 		{name: "buildsignal", topicKey: stovepipemq.TopicKeyBuildSignal, consumerGroup: "stovepipe-buildsignal", wantRetry: extendedRetry},
 		{name: "buildanalysis", topicKey: stovepipemq.TopicKeyBuildAnalysis, consumerGroup: "stovepipe-buildanalysis", wantRetry: extendedRetry},
 		{name: "issuetrack", topicKey: stovepipemq.TopicKeyIssueTrack, consumerGroup: "stovepipe-issuetrack", wantRetry: extendedRetry},
+		{name: "fix", topicKey: stovepipemq.TopicKeyFix, consumerGroup: "stovepipe-fix", wantRetry: extendedRetry},
 		{name: "record", topicKey: stovepipemq.TopicKeyRecord, consumerGroup: "stovepipe-record", wantRetry: extendedRetry},
 		{name: "hook", topicKey: basehook.TopicKeyHook, consumerGroup: "stovepipe-hook", wantRetry: defaultRetry},
 	}

stovepipe/controller/issuetrack/README.md

@@ -59,7 +59,7 @@ A notification can fail while helping an older pass. Recovery abandons that save
 
 Storage and history failures return for retry; hosts must pair recovery with an always-retryable, unlimited-retry DLQ consumer. Only recovery work repeats. Malformed message identities, unresolvable queues, and missing source records are logged and acknowledged.
 
-The reference service registers queue-partitioned `issuetrack` and `issuetrack_dlq` subscriptions. Normal processing uses backend error classification and the extended bounded retry policy, while recovery uses the existing always-retryable DLQ consumer. The DLQ consumer starts before normal consumption and stops after it. Deploy recovery-aware controllers and provision the issue tables before enabling consumption; older controllers do not recognize the abandonment phases. Completed opening passes publish repair notifications; recovery does not. The fix publication route is registered, but fix consumption and executor configuration remain separate.
+The reference service registers queue-partitioned `issuetrack` and `issuetrack_dlq` subscriptions. Normal processing uses backend error classification and the extended bounded retry policy, while recovery uses the existing always-retryable DLQ consumer. The DLQ consumer starts before normal consumption and stops after it. Deploy recovery-aware controllers and provision the issue tables before enabling consumption; older controllers do not recognize the abandonment phases. Completed opening passes publish repair notifications; recovery does not. The reference host also consumes repair notifications with a no-op executor; fix dead-letter recovery remains separate.
 
 ## Analysis handoff
 

stovepipe/extension/fixer/noop/BUILD.bazel

@@ -0,0 +1,12 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["noop.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/extension/fixer/noop",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/fixer:go_default_library",
+    ],
+)

stovepipe/extension/fixer/noop/noop.go

@@ -0,0 +1,34 @@
+// 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 noop provides a repair executor that accepts submissions without launching work.
+package noop
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/fixer"
+)
+
+type executor struct{}
+
+// New returns a repair executor that accepts submissions without launching work.
+func New() fixer.RepairExecutor {
+	return executor{}
+}
+
+func (executor) Submit(_ context.Context, issue entity.OpenIssue) (string, error) {
+	return "noop/" + issue.FirstSeenBuildID, nil
+}