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