feat(stovepipe): publish build repair requests to the fix queue

stovepipe/controller/fix/BUILD.bazel

@@ -2,12 +2,17 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["fix.go"],
+    srcs = [
+        "delivery.go",
+        "fix.go",
+    ],
     importpath = "github.com/uber/submitqueue/stovepipe/controller/fix",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/fixer:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
@@ -18,11 +23,17 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["fix_test.go"],
+    srcs = [
+        "delivery_test.go",
+        "fix_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
+        "//platform/consumer/mock:go_default_library",
         "//platform/errs:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/fixer:go_default_library",
         "//stovepipe/extension/fixer/mock:go_default_library",

stovepipe/controller/fix/README.md

@@ -1,7 +1,9 @@
 # Fix submission
 
-The controller 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 `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`.
 
 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.
 
+Issue tracking calls [`fixrequest.Publish`](../../core/fixrequest/publish.go) after persisting both the build observation and open issue. Publications for the same build share a message identity and partition.
+
 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.

stovepipe/controller/fix/delivery.go

@@ -0,0 +1,41 @@
+// 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"
+	"fmt"
+
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+	mq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+)
+
+var _ consumer.Controller = (*Controller)(nil)
+
+func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error {
+	message := delivery.Message()
+	request := &mq.FixRequest{}
+	if err := mq.Unmarshal(message.Payload, request); err != nil {
+		return fmt.Errorf("failed to deserialize repair request: %w", err)
+	}
+	if err := entityqueue.ValidatePayloadQueue(message, request.GetQueueName()); err != nil {
+		return fmt.Errorf("invalid message identity: %w", err)
+	}
+	if request.GetId() == "" {
+		return fmt.Errorf("repair request requires a build ID")
+	}
+	return c.processBuild(ctx, request.GetQueueName(), request.GetId())
+}

stovepipe/controller/fix/delivery_test.go

@@ -0,0 +1,80 @@
+// 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"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
+	"github.com/uber/submitqueue/platform/errs"
+	mq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"go.uber.org/mock/gomock"
+)
+
+func repairMessage(t *testing.T) entityqueue.Message {
+	t.Helper()
+	payload, err := mq.Marshal(&mq.FixRequest{Id: testBuildID, QueueName: testQueue})
+	require.NoError(t, err)
+	message := entityqueue.NewMessage(testBuildID, payload, testBuildID, nil)
+	message.Tenant = testQueue
+	return message
+}
+
+func (m controllerMocks) delivery(message entityqueue.Message) *consumermock.MockDelivery {
+	delivery := consumermock.NewMockDelivery(m.ctrl)
+	delivery.EXPECT().Message().Return(message)
+	return delivery
+}
+
+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)
+
+	require.NoError(t, controller.Process(context.Background(), m.delivery(repairMessage(t))))
+}
+
+func TestProcessRejectsInvalidMessagesBeforeListingIssues(t *testing.T) {
+	for _, kind := range []string{"invalid JSON", "empty build", "empty queue", "empty tenant", "wrong tenant", "wrong metadata"} {
+		t.Run(kind, func(t *testing.T) {
+			controller, m := newTestController(t)
+			message := repairMessage(t)
+			switch kind {
+			case "invalid JSON":
+				message.Payload = []byte("{")
+			case "empty build":
+				message.Payload = []byte(`{"queue_name":"validation/main"}`)
+			case "empty queue":
+				message.Payload = []byte(`{"id":"Build/42 "}`)
+			case "empty tenant":
+				message.Tenant = ""
+			case "wrong tenant":
+				message.Tenant = "other/main"
+			case "wrong metadata":
+				message.Metadata[entityqueue.MetadataKeyQueueName] = "other/main"
+			}
+
+			err := controller.Process(context.Background(), m.delivery(message))
+			require.Error(t, err)
+			assert.False(t, errs.IsRetryable(err))
+		})
+	}
+}

stovepipe/core/fixrequest/BUILD.bazel

@@ -0,0 +1,28 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["publish.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/core/fixrequest",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/consumer:go_default_library",
+        "//platform/publish:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["publish_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)

stovepipe/core/fixrequest/publish.go

@@ -0,0 +1,46 @@
+// 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 fixrequest publishes builds for repair submission.
+package fixrequest
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/publish"
+	mq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+)
+
+// Publish requires the build and its open issues to be persisted first.
+// Replays after publication errors must call Publish again.
+func Publish(ctx context.Context, registry consumer.TopicRegistry, queueName, buildID string) error {
+	if queueName == "" || buildID == "" {
+		return fmt.Errorf("repair request requires a queue and build ID")
+	}
+
+	payload, err := mq.Marshal(&mq.FixRequest{Id: buildID, QueueName: queueName})
+	if err != nil {
+		return fmt.Errorf("failed to serialize repair request: %w", err)
+	}
+
+	// Repair submission is a one-shot handoff for this build.
+	return publish.Message(ctx, registry, mq.TopicKeyFix, publish.MessageParams{
+		Tenant:       queueName,
+		ID:           publish.IntentID(buildID),
+		Payload:      payload,
+		PartitionKey: buildID,
+	})
+}

stovepipe/core/fixrequest/publish_test.go

@@ -0,0 +1,95 @@
+// 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 fixrequest
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+	queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
+	mq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"go.uber.org/mock/gomock"
+)
+
+func newTestRegistry(t *testing.T) (consumer.TopicRegistry, *queuemock.MockPublisher) {
+	t.Helper()
+	ctrl := gomock.NewController(t)
+	queue := queuemock.NewMockQueue(ctrl)
+	publisher := queuemock.NewMockPublisher(ctrl)
+	queue.EXPECT().Publisher().Return(publisher).AnyTimes()
+	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{Key: mq.TopicKeyFix, Name: "test-fix", Queue: queue},
+	})
+	require.NoError(t, err)
+	return registry, publisher
+}
+
+func TestPublishBuildIdentityAndRetry(t *testing.T) {
+	registry, publisher := newTestRegistry(t)
+	var messages []entityqueue.Message
+	publisher.EXPECT().Publish(gomock.Any(), "test-fix", gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ string, message entityqueue.Message) error {
+			messages = append(messages, message)
+			if len(messages) == 1 {
+				return assert.AnError
+			}
+			return nil
+		}).Times(4)
+
+	ctx := context.Background()
+	require.ErrorIs(t, Publish(ctx, registry, "validation/main", "Build/42 "), assert.AnError)
+	require.NoError(t, Publish(ctx, registry, "validation/main", "Build/42 "))
+	require.NoError(t, Publish(ctx, registry, "validation/main", "Build/43"))
+	require.NoError(t, Publish(ctx, registry, "validation/other", "Build/42 "))
+
+	assert.Equal(t, messages[0].ID, messages[1].ID)
+	assert.Equal(t, messages[0].PartitionKey, messages[1].PartitionKey)
+	assert.NotEqual(t, messages[1].ID, messages[2].ID)
+	assert.NotEqual(t, messages[1].Tenant, messages[3].Tenant)
+	for _, message := range messages {
+		payload := &mq.FixRequest{}
+		require.NoError(t, mq.Unmarshal(message.Payload, payload))
+		assert.Equal(t, payload.GetId(), message.PartitionKey)
+		assert.Equal(t, payload.GetQueueName(), message.Tenant)
+		assert.Equal(t, message.Tenant, message.Metadata[entityqueue.MetadataKeyQueueName])
+	}
+	assert.Equal(t, "Build/42 ", messages[0].PartitionKey)
+}
+
+func TestPublishRejectsMissingIdentity(t *testing.T) {
+	for _, test := range []struct {
+		name  string
+		queue string
+		build string
+	}{
+		{name: "missing queue", build: "build/42"},
+		{name: "missing build", queue: "validation/main"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			registry, _ := newTestRegistry(t)
+			require.Error(t, Publish(context.Background(), registry, test.queue, test.build))
+		})
+	}
+}
+
+func TestPublishReturnsConfigurationAndSerializationErrors(t *testing.T) {
+	require.Error(t, Publish(context.Background(), consumer.TopicRegistry{}, "validation/main", "build/42"))
+	registry, _ := newTestRegistry(t)
+	require.Error(t, Publish(context.Background(), registry, "validation/main", string([]byte{0xff})))
+}

stovepipe/core/messagequeue/messagequeue.go

@@ -56,6 +56,9 @@ type (
 	// IssueTrack identifies persisted build observations for queue-scoped reconciliation.
 	IssueTrack = protopb.IssueTrack
 
+	// FixRequest identifies the build whose issues should be considered for repair.
+	FixRequest = protopb.FixRequest
+
 	// Record is the payload buildsignal publishes to the record stage once a
 	// build reaches a terminal status: the build id to record.
 	Record = protopb.Record

stovepipe/core/messagequeue/messagequeue_test.go

@@ -80,7 +80,7 @@ func TestWireFormat(t *testing.T) {
 // no topic_keys option names an unknown key.
 func TestTopicKeysBindEveryTopicKey(t *testing.T) {
 	bound := map[string]int{}
-	for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &BuildAnalysis{}, &IssueTrack{}, &Record{}} {
+	for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &BuildAnalysis{}, &IssueTrack{}, &FixRequest{}, &Record{}} {
 		keys := TopicKeys(m)
 		require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option")
 		for _, key := range keys {
@@ -94,6 +94,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) {
 		TopicKeyBuildSignal,
 		TopicKeyBuildAnalysis,
 		TopicKeyIssueTrack,
+		TopicKeyFix,
 		TopicKeyRecord,
 	}
 
@@ -146,3 +147,15 @@ func TestIssueTrackToleratesUnknownFields(t *testing.T) {
 	require.NoError(t, Unmarshal([]byte(`{"build_id":"build/42","queue_name":"validation/main","finalize":true,"future_field":42}`), got))
 	assert.True(t, proto.Equal(&IssueTrack{BuildId: "build/42", QueueName: "validation/main", Finalize: true}, got))
 }
+
+func TestFixRequestRoundTrip(t *testing.T) {
+	request := &FixRequest{Id: "Build/42 ", QueueName: "validation/main"}
+	data, err := Marshal(request)
+	require.NoError(t, err)
+
+	got := &FixRequest{}
+	require.NoError(t, Unmarshal(data, got))
+
+	assert.True(t, proto.Equal(request, got))
+	assert.JSONEq(t, `{"id":"Build/42 ","queue_name":"validation/main"}`, string(data))
+}

stovepipe/core/messagequeue/proto/BUILD.bazel

@@ -3,6 +3,7 @@ exports_files(
         "build.proto",
         "buildanalysis.proto",
         "buildsignal.proto",
+        "fix.proto",
         "issuetrack.proto",
         "process.proto",
         "record.proto",