fix(submitqueue): keep internal merge stage topic distinct from runway merge queue

api/runway/messagequeue/proto/merge.proto

@@ -48,7 +48,7 @@ message MergeStep {
 // Runway echoes it back unchanged.
 message MergeRequest {
     option (uber.base.messagequeue.topic_keys) = "merge-conflict-check";
-    option (uber.base.messagequeue.topic_keys) = "merge";
+    option (uber.base.messagequeue.topic_keys) = "runway-merge";
 
     // id is the client-owned correlation id for this request (one per request).
     // Runway echoes it back on the result unchanged.

api/runway/messagequeue/topics.go

@@ -34,7 +34,7 @@ const (
 	// TopicKeyMerge carries committing merge requests. A client publishes a
 	// MergeRequest here; Runway applies the steps, commits the result, and
 	// reports the revisions it produced.
-	TopicKeyMerge TopicKey = "merge"
+	TopicKeyMerge TopicKey = "runway-merge"
 	// TopicKeyMergeSignal carries committing merge results. Runway publishes a
 	// MergeResult here (with the produced revisions populated); the requesting
 	// client consumes it.

platform/consumer/registry.go

@@ -62,7 +62,10 @@ type topicGroup struct {
 }
 
 // NewTopicRegistry creates a new TopicRegistry from a list of TopicConfigs.
-// Returns an error if any topic name is invalid.
+// Returns an error if any topic name is invalid, or if two configs share a
+// topic key — a duplicate key would silently shadow the earlier entry (last
+// write wins on the key→queue/name maps), routing publishes and subscriptions
+// registered against one topic onto another.
 func NewTopicRegistry(configs []TopicConfig) (TopicRegistry, error) {
 	queues := make(map[TopicKey]extqueue.Queue, len(configs))
 	topicNames := make(map[TopicKey]string, len(configs))
@@ -73,6 +76,12 @@ func NewTopicRegistry(configs []TopicConfig) (TopicRegistry, error) {
 			return TopicRegistry{}, fmt.Errorf("invalid topic name for key %s: %w", cfg.Key, err)
 		}
 
+		if existing, ok := topicNames[cfg.Key]; ok {
+			return TopicRegistry{}, fmt.Errorf(
+				"duplicate topic key %s: already registered with name %q, cannot also register name %q",
+				cfg.Key, existing, cfg.Name)
+		}
+
 		queues[cfg.Key] = cfg.Queue
 		topicNames[cfg.Key] = cfg.Name
 

service/runway/server/main.go

@@ -275,7 +275,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 		},
 		{
 			Key:   runwaymq.TopicKeyMerge,
-			Name:  "merge",
+			Name:  "runway-merge",
 			Queue: q,
 			Subscription: extqueue.DefaultSubscriptionConfig(
 				subscriberName, "runway-merge",

service/submitqueue/orchestrator/server/main.go

@@ -388,7 +388,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 		{topickey.TopicKeyPrioritize, "prioritize", "orchestrator-prioritize"},
 		{topickey.TopicKeyBuild, "build", "orchestrator-build"},
 		{topickey.TopicKeyBuildSignal, "buildsignal", "orchestrator-buildsignal"},
-		{topickey.TopicKeyMerge, "merge", "orchestrator-merge"},
+		{topickey.TopicKeyMerge, "submitqueue-merge", "orchestrator-merge"},
 		{runwaymq.TopicKeyMergeSignal, "merge-signal", "orchestrator-mergesignal"},
 		{topickey.TopicKeyConclude, "conclude", "orchestrator-conclude"},
 	}
@@ -456,7 +456,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 	// consumed primary topic above.
 	configs = append(configs, consumer.TopicConfig{
 		Key:   runwaymq.TopicKeyMerge,
-		Name:  "merge",
+		Name:  "runway-merge",
 		Queue: q,
 	})
 

submitqueue/core/topickey/topickey.go

@@ -50,7 +50,7 @@ const (
 	// PublishAfter when the build has not yet reached a terminal state.
 	TopicKeyBuildSignal TopicKey = "buildsignal"
 	// TopicKeyMerge is the pipeline stage where speculated batches are published for merging.
-	TopicKeyMerge TopicKey = "merge"
+	TopicKeyMerge TopicKey = "submitqueue-merge"
 	// TopicKeyConclude is the pipeline stage where merged requests are published for conclusion.
 	TopicKeyConclude TopicKey = "conclude"
 	// TopicKeyLog is the pipeline stage where per-request logs are written.

submitqueue/orchestrator/controller/dlq/batch_test.go

@@ -34,8 +34,8 @@ func TestDLQBatchController_InterfaceAndAccessors(t *testing.T) {
 
 	c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq")
 
-	assert.Equal(t, "merge_dlq", c.Name())
-	assert.Equal(t, consumer.TopicKey("merge_dlq"), c.TopicKey())
+	assert.Equal(t, "submitqueue-merge_dlq", c.Name())
+	assert.Equal(t, consumer.TopicKey("submitqueue-merge_dlq"), c.TopicKey())
 	assert.Equal(t, "orchestrator-merge-dlq", c.ConsumerGroup())
 }
 

submitqueue/orchestrator/controller/merge/merge_test.go

@@ -70,7 +70,7 @@ func TestNewController(t *testing.T) {
 	store := storagemock.NewMockStorage(ctrl)
 	q := queuemock.NewMockQueue(ctrl)
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "merge", Queue: q}},
+		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}},
 	)
 	require.NoError(t, err)
 
@@ -130,15 +130,15 @@ func TestProcess_PublishesFullPayloadToRunway(t *testing.T) {
 	q := queuemock.NewMockQueue(ctrl)
 	q.EXPECT().Publisher().Return(pub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "merge", Queue: q}},
+		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}},
 	)
 	require.NoError(t, err)
 
 	c := newController(t, store, registry)
 	require.NoError(t, c.Process(context.Background(), newDelivery(t, ctrl, batchID, batch.Queue)))
 
 	// Full payload published to runway, keyed by the batch id (the correlation id).
-	assert.Equal(t, "merge", gotTopic)
+	assert.Equal(t, "runway-merge", gotTopic)
 	got := &runwaymq.MergeRequest{}
 	require.NoError(t, runwaymq.Unmarshal(gotPayload, got))
 	assert.Equal(t, batch.ID, got.Id)
@@ -180,7 +180,7 @@ func TestProcess_HaltedBatchSkips(t *testing.T) {
 			q := queuemock.NewMockQueue(ctrl)
 			q.EXPECT().Publisher().Return(pub).AnyTimes()
 			registry, err := consumer.NewTopicRegistry(
-				[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "merge", Queue: q}},
+				[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}},
 			)
 			require.NoError(t, err)
 
@@ -211,7 +211,7 @@ func TestProcess_PublishFailureReturnsError(t *testing.T) {
 	q := queuemock.NewMockQueue(ctrl)
 	q.EXPECT().Publisher().Return(pub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "merge", Queue: q}},
+		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}},
 	)
 	require.NoError(t, err)
 
@@ -233,7 +233,7 @@ func TestProcess_BatchStoreGetFailureNotRetryable(t *testing.T) {
 
 	q := queuemock.NewMockQueue(ctrl)
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "merge", Queue: q}},
+		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}},
 	)
 	require.NoError(t, err)
 

submitqueue/orchestrator/controller/speculate/speculate_test.go

@@ -71,7 +71,7 @@ func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock
 	registry, err := consumer.NewTopicRegistry(
 		[]consumer.TopicConfig{
 			{Key: topickey.TopicKeyBuild, Name: "build", Queue: mockQ},
-			{Key: topickey.TopicKeyMerge, Name: "merge", Queue: mockQ},
+			{Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: mockQ},
 			{Key: topickey.TopicKeyConclude, Name: "conclude", Queue: mockQ},
 			{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
 		},