feat(orchestrator): dispatch builds per speculation path (#503)

## Summary
### Why?

The build stage triggered one build per batch, on the batch's full
dependency list. Speculation needs one build per *path*, on the subset
of dependencies the path assumes will succeed — that subset is what lets
a batch be verified before the batches ahead of it resolve.

### What?

The dispatch controller now reads the head's path set and starts a build
for every pending path, on that path's own base. It **only starts
builds** — stopping them belongs to the poll loop (next commit) — and it
**never writes the path set**, which stays the speculate run's
single-writer state; what this stage knows is the build it started,
recorded in per-build rows of its own.

The write order is Trigger → Build record → write-once link
(`entity.PathBuild`) → signal; each write makes the previous one
reachable, and the link is the idempotency point for redeliveries. A
lost link race publishes both builds' signals and acks — the poll loop
keeps the one the link names. On redeliveries only, every live linked
path gets its signal re-published, closing the crash window between link
and signal without forking duplicate poll chains past the queue's dedup
horizon.

The build signal carries only the runner's build ID and partitions on
it, so one slow build cannot block a head's other paths. Known gap: a
crash between `Trigger` and the link orphans that build; the fix is an
idempotency key on `BuildRunner.Trigger`, marked TODO.

Also adds `submitqueue/core/publish` — the registry-lookup-and-send
plumbing and the message-ID deduplication rule in one place.

## Test Plan
✅ `bazel test //submitqueue/orchestrator/controller/build/...
//submitqueue/core/publish/...` — write order, the base holding only
assumed-success dependencies, redelivery republish without rebuild,
lost-race double signal, halted batch starting nothing, redelivery
healing for live linked paths, and a pin that no path-set write of any
kind happens.

✅ `make fmt`, `make gazelle`, `make mocks`

## Issues

Author: behinddwalls

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",