feat(orchestrator): dispatch builds per speculation path

submitqueue/core/publish/BUILD.bazel

@@ -0,0 +1,26 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["publish.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/core/publish",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer: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",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)

submitqueue/core/publish/publish.go

@@ -0,0 +1,67 @@
+// 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.
+
+// Package publish sends a message to the queue behind a topic key. It owns the
+// lookup-and-send plumbing every orchestrator stage otherwise repeats — resolve
+// the key to a queue and a topic name, wrap the payload in a message, publish —
+// and the message-ID convention that controls deduplication (see UniqueID).
+package publish
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+)
+
+// Message publishes payload to the topic registered for key.
+//
+// msgID selects the dedup behavior, so the caller must choose it deliberately.
+// The queue deduplicates on (topic, partition key, message ID) against every
+// row it has not garbage-collected yet, consumed ones included:
+//
+//   - A stable msgID (an entity's own ID) makes a repeat publish a silent
+//     no-op. Right for a hand-off that must happen at most once per entity.
+//   - UniqueID(id) makes every publish distinct. Right for signals that are
+//     re-sent by design — wake-ups, polls, re-dispatches — where a swallowed
+//     repeat would stall the pipeline.
+func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error {
+	q, ok := registry.Queue(key)
+	if !ok {
+		return fmt.Errorf("no queue registered for topic key %s", key)
+	}
+	topicName, ok := registry.TopicName(key)
+	if !ok {
+		return fmt.Errorf("no topic name registered for topic key %s", key)
+	}
+
+	msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil)
+	return q.Publisher().Publish(ctx, topicName, msg)
+}
+
+// sequence breaks ties between UniqueID calls that land on the same clock
+// tick: some platforms quantize time.Now coarsely enough for consecutive calls
+// to read the same nanosecond.
+var sequence atomic.Uint64
+
+// UniqueID returns a message ID no earlier publish for the same entity has
+// used, so the queue's (topic, partition key, message ID) dedup never swallows
+// the repeat. Use it for every publish that is re-sent by design; reusing the
+// bare entity ID instead would make the second publish a silent no-op.
+func UniqueID(id string) string {
+	return fmt.Sprintf("%s@%d-%d", id, time.Now().UnixNano(), sequence.Add(1))
+}

submitqueue/core/publish/publish_test.go

@@ -0,0 +1,80 @@
+// 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.
+
+package publish
+
+import (
+	"context"
+	"strings"
+	"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"
+	"go.uber.org/mock/gomock"
+)
+
+const testKey consumer.TopicKey = "test-topic-key"
+
+func newTestRegistry(t *testing.T, ctrl *gomock.Controller) (consumer.TopicRegistry, *queuemock.MockPublisher) {
+	t.Helper()
+
+	publisher := queuemock.NewMockPublisher(ctrl)
+	q := queuemock.NewMockQueue(ctrl)
+	q.EXPECT().Publisher().Return(publisher).AnyTimes()
+
+	registry, err := consumer.NewTopicRegistry(
+		[]consumer.TopicConfig{{Key: testKey, Name: "test-topic", Queue: q}},
+	)
+	require.NoError(t, err)
+	return registry, publisher
+}
+
+func TestMessage(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	registry, publisher := newTestRegistry(t, ctrl)
+
+	var published entityqueue.Message
+	publisher.EXPECT().
+		Publish(gomock.Any(), "test-topic", gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error {
+			published = msg
+			return nil
+		})
+
+	err := Message(context.Background(), registry, testKey, "msg-1", []byte("payload"), "partition-1")
+	require.NoError(t, err)
+	assert.Equal(t, "msg-1", published.ID)
+	assert.Equal(t, []byte("payload"), published.Payload)
+	assert.Equal(t, "partition-1", published.PartitionKey)
+}
+
+func TestMessage_UnregisteredKey(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	registry, _ := newTestRegistry(t, ctrl)
+
+	err := Message(context.Background(), registry, "unregistered-key", "msg-1", []byte("payload"), "partition-1")
+	require.Error(t, err)
+}
+
+func TestUniqueID(t *testing.T) {
+	a := UniqueID("batch-1")
+	b := UniqueID("batch-1")
+
+	assert.True(t, strings.HasPrefix(a, "batch-1@"))
+	assert.True(t, strings.HasPrefix(b, "batch-1@"))
+	assert.NotEqual(t, a, b)
+}

submitqueue/orchestrator/controller/build/BUILD.bazel

@@ -6,9 +6,9 @@ go_library(
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/build",
     visibility = ["//visibility:public"],
     deps = [
-        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
+        "//submitqueue/core/publish:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/buildrunner:go_default_library",
@@ -26,13 +26,10 @@ go_test(
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
-        "//platform/errs:go_default_library",
         "//platform/extension/messagequeue/mock:go_default_library",
-        "//submitqueue/core/changeset/fake:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/buildrunner:go_default_library",
-        "//submitqueue/extension/buildrunner/fake:go_default_library",
         "//submitqueue/extension/buildrunner/mock:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mock:go_default_library",