feat(stovepipe): publish admitted requests to build

service/stovepipe/server/main.go

@@ -222,6 +222,7 @@ func run() error {
 		store,
 		queueconfigdefault.NewStore(),
 		fakeSourceControlFactory{},
+		registry,
 		stovepipemq.TopicKeyProcess,
 		"stovepipe-process",
 	)
@@ -310,7 +311,7 @@ func run() error {
 }
 
 // newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest
-// publishes to the process topic and the process consumer subscribes to it.
+// publishes to process; process publishes admitted requests to the publish-only build topic.
 func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
 	return consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{
@@ -321,5 +322,10 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 				subscriberName, "stovepipe-process",
 			),
 		},
+		{
+			Key:   stovepipemq.TopicKeyBuild,
+			Name:  "build",
+			Queue: q,
+		},
 	})
 }

stovepipe/controller/process/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/uber/submitqueue/stovepipe/controller/process",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/errs:go_default_library",
         "//platform/metrics:go_default_library",

stovepipe/controller/process/process.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 
 	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/errs"
 	"github.com/uber/submitqueue/platform/metrics"
@@ -44,6 +45,7 @@ type Controller struct {
 	store         storage.Storage
 	queueConfigs  queueconfig.Store
 	sourceControl sourcecontrol.Factory
+	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -61,6 +63,7 @@ func NewController(
 	store storage.Storage,
 	queueConfigs queueconfig.Store,
 	sourceControl sourcecontrol.Factory,
+	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
@@ -70,6 +73,7 @@ func NewController(
 		store:         store,
 		queueConfigs:  queueConfigs,
 		sourceControl: sourceControl,
+		registry:      registry,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
@@ -98,7 +102,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 
 	switch request.State {
 	case entity.RequestStateProcessing:
-		// TODO: re-publish to build once the build stage lands (RFC process algorithm, step 3).
+		if err := c.publishBuild(ctx, request.ID); err != nil {
+			metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1)
+			return fmt.Errorf("ProcessController failed to publish request %s to build: %w", request.ID, err)
+		}
 		return nil
 	case entity.RequestStateSuperseded:
 		return nil
@@ -238,7 +245,10 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request
 		return nil
 	}
 
-	// TODO(build-publish): publish BuildRequest to the build stage here.
+	if err := c.publishBuild(ctx, request.ID); err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1)
+		return fmt.Errorf("ProcessController failed to publish request %s to build: %w", request.ID, err)
+	}
 
 	metrics.NamedCounter(c.metricsScope, _opName, "admitted", 1,
 		metrics.NewTag("strategy", string(request.BuildStrategy)),
@@ -432,6 +442,29 @@ func (c *Controller) loadQueue(ctx context.Context, name string) (entity.Queue,
 	return entity.Queue{}, fmt.Errorf("ProcessController failed to load queue %s: %w", name, err)
 }
 
+// publishBuild publishes the admitted request ID to the build stage. The build
+// controller reloads the Request from storage to read its immutable strategy
+// and baseline.
+func (c *Controller) publishBuild(ctx context.Context, id string) error {
+	payload, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id})
+	if err != nil {
+		return fmt.Errorf("failed to serialize build request: %w", err)
+	}
+
+	q, ok := c.registry.Queue(stovepipemq.TopicKeyBuild)
+	if !ok {
+		return fmt.Errorf("no queue registered for topic key %s", stovepipemq.TopicKeyBuild)
+	}
+	topicName, ok := c.registry.TopicName(stovepipemq.TopicKeyBuild)
+	if !ok {
+		return fmt.Errorf("no topic name registered for topic key %s", stovepipemq.TopicKeyBuild)
+	}
+	if err := q.Publisher().Publish(ctx, topicName, entityqueue.NewMessage(id, payload, id, nil)); err != nil {
+		return fmt.Errorf("failed to publish build request: %w", err)
+	}
+	return nil
+}
+
 // Name returns the controller name for logging and metrics.
 func (c *Controller) Name() string {
 	return "process"

stovepipe/controller/process/process_test.go

@@ -49,6 +49,7 @@ type processMocks struct {
 	queueStore    *storagemock.MockQueueStore
 	sourceFactory *sourcecontrolmock.MockFactory
 	sourceControl *sourcecontrolmock.MockSourceControl
+	publisher     *mqmock.MockPublisher
 }
 
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) {
@@ -63,18 +64,27 @@ func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.S
 		queueStore:    storagemock.NewMockQueueStore(ctrl),
 		sourceFactory: sourcecontrolmock.NewMockFactory(ctrl),
 		sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl),
+		publisher:     mqmock.NewMockPublisher(ctrl),
 	}
 
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
 	store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
 
+	queue := mqmock.NewMockQueue(ctrl)
+	queue.EXPECT().Publisher().Return(m.publisher).AnyTimes()
+	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{Key: stovepipemq.TopicKeyBuild, Name: "build", Queue: queue},
+	})
+	require.NoError(t, err)
+
 	c := NewController(
 		zap.NewNop().Sugar(),
 		scope,
 		store,
 		queueconfigdefault.NewStore(),
 		m.sourceFactory,
+		registry,
 		stovepipemq.TopicKeyProcess,
 		"stovepipe-process",
 	)
@@ -106,7 +116,24 @@ func acceptedRequest(id string) entity.Request {
 	}
 }
 
-func expectAdmit(m processMocks, id string) {
+func TestProcessBuildPublishRequiresRegisteredTopic(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, m := newController(t, ctrl)
+	c.registry = consumer.TopicRegistry{}
+
+	m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{
+		ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2,
+	}, nil)
+
+	err := c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))
+
+	require.Error(t, err)
+	assert.False(t, errs.IsRetryable(err))
+}
+
+func expectAdmit(t *testing.T, m processMocks, id string) {
+	t.Helper()
+
 	updatedQueue := entity.Queue{
 		Name:            testQueue,
 		LatestRequestID: id,
@@ -119,6 +146,23 @@ func expectAdmit(m processMocks, id string) {
 	updatedReq.State = entity.RequestStateProcessing
 	updatedReq.BuildStrategy = entity.BuildStrategyFull
 	m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil)
+
+	expectBuildPublish(t, m, id)
+}
+
+func expectBuildPublish(t *testing.T, m processMocks, id string) {
+	t.Helper()
+
+	m.publisher.EXPECT().
+		Publish(gomock.Any(), "build", gomock.AssignableToTypeOf(entityqueue.Message{})).
+		DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error {
+			assert.Equal(t, id, msg.ID)
+			assert.Equal(t, id, msg.PartitionKey)
+			buildReq := &stovepipemq.BuildRequest{}
+			require.NoError(t, stovepipemq.Unmarshal(msg.Payload, buildReq))
+			assert.Equal(t, id, buildReq.Id)
+			return nil
+		})
 }
 
 func TestDeriveBuildStrategy(t *testing.T) {
@@ -257,7 +301,7 @@ func TestProcessEmitsAdmittedStrategyMetric(t *testing.T) {
 		LatestRequestID: testID,
 		Version:         1,
 	}, nil)
-	expectAdmit(m, testID)
+	expectAdmit(t, m, testID)
 
 	require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))))
 
@@ -349,6 +393,7 @@ func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) {
 			m.sourceControl.EXPECT().IsAncestor(gomock.Any(), reloadedLastGreen, testURI).Return(true, nil)
 			m.queueStore.EXPECT().Update(gomock.Any(), claimedQueue, int32(2), int32(3)).Return(nil)
 			m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil)
+			expectBuildPublish(t, m, testID)
 
 			require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))))
 		})
@@ -373,11 +418,12 @@ func TestProcess(t *testing.T) {
 			},
 		},
 		{
-			name: "processing is no-op until build publish lands",
+			name: "processing republishes to build",
 			setup: func(m processMocks) {
 				m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{
 					ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2,
 				}, nil)
+				expectBuildPublish(t, m, testID)
 			},
 		},
 		{
@@ -397,7 +443,33 @@ func TestProcess(t *testing.T) {
 					LatestRequestID: testID,
 					Version:         1,
 				}, nil)
-				expectAdmit(m, testID)
+				expectAdmit(t, m, testID)
+			},
+		},
+		{
+			name:    "build publish failure retains admitted request and claimed slot",
+			wantErr: true,
+			setup: func(m processMocks) {
+				m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil)
+				m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{
+					Name:            testQueue,
+					LatestRequestID: testID,
+					Version:         1,
+				}, nil)
+				updatedQueue := entity.Queue{
+					Name:            testQueue,
+					LatestRequestID: testID,
+					InFlightCount:   1,
+					Version:         1,
+				}
+				m.queueStore.EXPECT().Update(gomock.Any(), updatedQueue, int32(1), int32(2)).Return(nil)
+				updatedRequest := acceptedRequest(testID)
+				updatedRequest.State = entity.RequestStateProcessing
+				updatedRequest.BuildStrategy = entity.BuildStrategyFull
+				m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil)
+				m.publisher.EXPECT().
+					Publish(gomock.Any(), "build", gomock.AssignableToTypeOf(entityqueue.Message{})).
+					Return(errors.New("queue unavailable"))
 			},
 		},
 		{
@@ -462,6 +534,7 @@ func TestProcess(t *testing.T) {
 				updatedReq.State = entity.RequestStateProcessing
 				updatedReq.BuildStrategy = entity.BuildStrategyFull
 				m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil)
+				expectBuildPublish(t, m, testID)
 			},
 		},
 		{
@@ -529,6 +602,7 @@ func TestProcess(t *testing.T) {
 				retry.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen
 				retry.BaseURI = lastGreenURI
 				m.reqStore.EXPECT().Update(gomock.Any(), retry, int32(2), int32(3)).Return(nil)
+				expectBuildPublish(t, m, testID)
 			},
 		},
 		{

stovepipe/core/messagequeue/messagequeue.go

@@ -41,6 +41,10 @@ type (
 	// ProcessRequest is the payload ingest publishes to the process stage: the
 	// minted request id to validate.
 	ProcessRequest = protopb.ProcessRequest
+
+	// BuildRequest is the payload process publishes to the build stage: the
+	// admitted request id whose persisted strategy and baseline build reloads.
+	BuildRequest = protopb.BuildRequest
 )
 
 // marshalOpts keeps the JSON field names identical to the proto field names

stovepipe/core/messagequeue/process_test.go

@@ -33,6 +33,17 @@ func TestProcessRequestRoundTrip(t *testing.T) {
 	assert.True(t, proto.Equal(req, got), "round-tripped ProcessRequest should equal the original")
 }
 
+func TestBuildRequestRoundTrip(t *testing.T) {
+	req := &BuildRequest{Id: "request/monorepo/main/42"}
+
+	data, err := Marshal(req)
+	require.NoError(t, err)
+
+	got := &BuildRequest{}
+	require.NoError(t, Unmarshal(data, got))
+	assert.True(t, proto.Equal(req, got), "round-tripped BuildRequest should equal the original")
+}
+
 // TestWireFormat locks the protojson encoding decision the contract relies on:
 // snake_case field names (UseProtoNames).
 func TestWireFormat(t *testing.T) {
@@ -47,7 +58,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{}} {
+	for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}} {
 		keys := TopicKeys(m)
 		require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option")
 		for _, key := range keys {
@@ -57,6 +68,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) {
 
 	keys := []TopicKey{
 		TopicKeyProcess,
+		TopicKeyBuild,
 	}
 
 	valid := map[string]bool{}

stovepipe/core/messagequeue/proto/BUILD.bazel

@@ -1,4 +1,7 @@
 exports_files(
-    ["process.proto"],
+    [
+        "build.proto",
+        "process.proto",
+    ],
     visibility = ["//tool/proto:__pkg__"],
 )

stovepipe/core/messagequeue/proto/build.proto

@@ -0,0 +1,34 @@
+// Copyright (c) 2025 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.
+
+syntax = "proto3";
+
+package uber.stovepipe.messagequeue;
+
+import "api/base/messagequeue/proto/messagequeue.proto";
+
+option go_package = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb";
+option java_multiple_files = true;
+option java_outer_classname = "BuildProto";
+option java_package = "com.uber.submitqueue.stovepipe.messagequeue";
+
+// BuildRequest is the payload process publishes after admitting a request. Only
+// the request id travels; build reloads the request from storage and uses the
+// strategy and baseline process persisted at admission.
+message BuildRequest {
+    option (uber.base.messagequeue.topic_keys) = "build";
+
+    // id is the admitted request id to build. Format: "request/<queue>/<counter>".
+    string id = 1;
+}

stovepipe/core/messagequeue/protopb/BUILD.bazel

@@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library")
 
 go_library(
     name = "go_default_library",
-    srcs = ["process.pb.go"],
+    srcs = [
+        "build.pb.go",
+        "process.pb.go",
+    ],
     importpath = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb",
     visibility = ["//visibility:public"],
     deps = [