From 087124c1ac6613d95786c55c7b12dad0e3509203 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Tue, 14 Jul 2026 01:09:24 +0000 Subject: [PATCH] fix(submitqueue): keep internal merge stage topic distinct from runway merge queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orchestrator registers its internal merge pipeline stage and the runway-owned merge queue in one topic registry over one shared queue backend, and both used the topic key "merge" (topickey.TopicKeyMerge == runwaymq.TopicKeyMerge). The publish-only runway entry silently shadowed the internal stage entry (last write wins), collapsing the two topics into one: - runway consumed the internal batch-ID messages as zero-step MergeRequests and reported SUCCEEDED without ever seeing the real steps; - the orchestrator merge controller's real MergeRequest publish was then silently dropped by the queue's (topic, partition_key, id) dedupe, because it reuses the batch ID already published to the same topic. The pipeline still reached "landed" only because the zero-step result correlates by batch ID — with a real merger nothing would ever have been merged. Found by the e2e harness asserting runway's merge signal carries the request's step IDs. Namespace the merge topic keys by owner so the two can never collide: the runway-owned queue becomes "runway-merge" and the internal orchestrator stage becomes "submitqueue-merge" (both key and wire name). Also make consumer.NewTopicRegistry reject duplicate topic keys so a future collision fails at startup instead of silently misrouting messages. Co-Authored-By: Claude Opus 4.8 (1M context) --- api/runway/messagequeue/proto/merge.proto | 2 +- api/runway/messagequeue/protopb/merge.pb.go | 4 ++-- api/runway/messagequeue/topics.go | 2 +- platform/consumer/registry.go | 11 ++++++++++- service/runway/server/main.go | 2 +- service/submitqueue/orchestrator/server/main.go | 4 ++-- .../extension/storage/mock/queue_store_mock.go | 14 ++++++++++---- submitqueue/core/topickey/topickey.go | 2 +- .../orchestrator/controller/dlq/batch_test.go | 4 ++-- .../orchestrator/controller/merge/merge_test.go | 12 ++++++------ .../controller/speculate/speculate_test.go | 2 +- 11 files changed, 37 insertions(+), 22 deletions(-) diff --git a/api/runway/messagequeue/proto/merge.proto b/api/runway/messagequeue/proto/merge.proto index b2805f03..72210667 100644 --- a/api/runway/messagequeue/proto/merge.proto +++ b/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. diff --git a/api/runway/messagequeue/protopb/merge.pb.go b/api/runway/messagequeue/protopb/merge.pb.go index 80785e48..5409c180 100644 --- a/api/runway/messagequeue/protopb/merge.pb.go +++ b/api/runway/messagequeue/protopb/merge.pb.go @@ -443,12 +443,12 @@ const file_merge_proto_rawDesc = "" + "\tMergeStep\x12\x17\n" + "\astep_id\x18\x01 \x01(\tR\x06stepId\x122\n" + "\achanges\x18\x02 \x03(\v2\x18.uber.base.change.ChangeR\achanges\x12=\n" + - "\bstrategy\x18\x03 \x01(\x0e2!.uber.base.mergestrategy.StrategyR\bstrategy\"\x9b\x01\n" + + "\bstrategy\x18\x03 \x01(\x0e2!.uber.base.mergestrategy.StrategyR\bstrategy\"\xa2\x01\n" + "\fMergeRequest\x12\x0e\n" + "\x02id\x18\x01 \x01(\tR\x02id\x12\x1d\n" + "\n" + "queue_name\x18\x02 \x01(\tR\tqueueName\x129\n" + - "\x05steps\x18\x03 \x03(\v2#.uber.runway.messagequeue.MergeStepR\x05steps:!\x8a\xb5\x18\x14merge-conflict-check\x8a\xb5\x18\x05merge\"\x1c\n" + + "\x05steps\x18\x03 \x03(\v2#.uber.runway.messagequeue.MergeStepR\x05steps:(\x8a\xb5\x18\x14merge-conflict-check\x8a\xb5\x18\frunway-merge\"\x1c\n" + "\n" + "StepOutput\x12\x0e\n" + "\x02id\x18\x01 \x01(\tR\x02id\"}\n" + diff --git a/api/runway/messagequeue/topics.go b/api/runway/messagequeue/topics.go index 062fca43..e80e7f9b 100644 --- a/api/runway/messagequeue/topics.go +++ b/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. diff --git a/platform/consumer/registry.go b/platform/consumer/registry.go index fd3f8b30..0cfe52bf 100644 --- a/platform/consumer/registry.go +++ b/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 diff --git a/service/runway/server/main.go b/service/runway/server/main.go index d694f705..ee6a0936 100644 --- a/service/runway/server/main.go +++ b/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", diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 4ddda8cf..6ed707ca 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/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, }) diff --git a/stovepipe/extension/storage/mock/queue_store_mock.go b/stovepipe/extension/storage/mock/queue_store_mock.go index 023f829b..60b7168b 100644 --- a/stovepipe/extension/storage/mock/queue_store_mock.go +++ b/stovepipe/extension/storage/mock/queue_store_mock.go @@ -1,5 +1,10 @@ // Code generated by MockGen. DO NOT EDIT. -// Source: stovepipe/extension/storage/queue_store.go +// Source: queue_store.go +// +// Generated by this command: +// +// mockgen -source=queue_store.go -destination=mock/queue_store_mock.go -package=mock +// // Package mock is a generated GoMock package. package mock @@ -16,6 +21,7 @@ import ( type MockQueueStore struct { ctrl *gomock.Controller recorder *MockQueueStoreMockRecorder + isgomock struct{} } // MockQueueStoreMockRecorder is the mock recorder for MockQueueStore. @@ -44,7 +50,7 @@ func (m *MockQueueStore) Create(ctx context.Context, queue entity.Queue) error { } // Create indicates an expected call of Create. -func (mr *MockQueueStoreMockRecorder) Create(ctx, queue interface{}) *gomock.Call { +func (mr *MockQueueStoreMockRecorder) Create(ctx, queue any) *gomock.Call { mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockQueueStore)(nil).Create), ctx, queue) } @@ -59,7 +65,7 @@ func (m *MockQueueStore) Get(ctx context.Context, name string) (entity.Queue, er } // Get indicates an expected call of Get. -func (mr *MockQueueStoreMockRecorder) Get(ctx, name interface{}) *gomock.Call { +func (mr *MockQueueStoreMockRecorder) Get(ctx, name any) *gomock.Call { mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockQueueStore)(nil).Get), ctx, name) } @@ -73,7 +79,7 @@ func (m *MockQueueStore) Update(ctx context.Context, queue entity.Queue, oldVers } // Update indicates an expected call of Update. -func (mr *MockQueueStoreMockRecorder) Update(ctx, queue, oldVersion, newVersion interface{}) *gomock.Call { +func (mr *MockQueueStoreMockRecorder) Update(ctx, queue, oldVersion, newVersion any) *gomock.Call { mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockQueueStore)(nil).Update), ctx, queue, oldVersion, newVersion) } diff --git a/submitqueue/core/topickey/topickey.go b/submitqueue/core/topickey/topickey.go index 66b23bf3..a7d93c3b 100644 --- a/submitqueue/core/topickey/topickey.go +++ b/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. diff --git a/submitqueue/orchestrator/controller/dlq/batch_test.go b/submitqueue/orchestrator/controller/dlq/batch_test.go index 69c8e317..8fc79edf 100644 --- a/submitqueue/orchestrator/controller/dlq/batch_test.go +++ b/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()) } diff --git a/submitqueue/orchestrator/controller/merge/merge_test.go b/submitqueue/orchestrator/controller/merge/merge_test.go index dd9eabaa..8207cb44 100644 --- a/submitqueue/orchestrator/controller/merge/merge_test.go +++ b/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,7 +130,7 @@ 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) @@ -138,7 +138,7 @@ func TestProcess_PublishesFullPayloadToRunway(t *testing.T) { 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) diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index 7eff01b5..3cdbd7ce 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/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}, },