diff --git a/doc/rfc/runway/workflow.md b/doc/rfc/runway/workflow.md index 8a180110..568e8d47 100644 --- a/doc/rfc/runway/workflow.md +++ b/doc/rfc/runway/workflow.md @@ -61,6 +61,20 @@ The merge-conflict-check controller always publishes a result — even when all The merge controller publishes a conflict result (and acks) when the merge detects a conflict; SubmitQueue handles rebatching. On infrastructure error it nacks for retry. On success it publishes per-step outcomes (output IDs of the revisions produced) so SubmitQueue can update its request state. +## Terminal failures and dead-lettering + +Runway is stateless and the sole responder on the client's correlation id: SubmitQueue records the in-flight work before publishing and waits for exactly one `MergeResult` echoing that id. Every request must therefore resolve to a result — success or failure — or the client waits forever. + +Failures split into two classes: + +- **Named terminal outcomes** — a merge conflict or an invalid request (an unknown/unsupported strategy, a malformed change URI, or an invalid `PROMOTE` composition). These can never succeed on retry, so the controller publishes a `FAILED` `MergeResult` (with a reason) and acks, rather than nacking. The merger surfaces them as the `ErrConflict` / `ErrInvalidRequest` sentinels; `IsTerminal` is the single classification point. + +- **Infrastructure faults** — fetch/network/auth failures, a push rejected for a reason other than a moved tip, and so on. These are nacked for retry. + +An infrastructure fault that never recovers would exhaust retries and dead-letter. Because nothing consumed those dead-letter topics, such a request produced no signal and left the client's correlation id unresolved. Runway closes that gap with a **DLQ reconciler**: a dedicated consumer subscribes to the inbound topics' `_dlq` queues and, for each dead-lettered request, republishes a `FAILED` `MergeResult` (echoing the correlation id) to the corresponding signal topic. Unlike the SubmitQueue/Stovepipe DLQ reconcilers it writes no entity state — the signal is the resolution. It runs under an always-retryable error policy so a transient publish failure retries indefinitely rather than dead-lettering again. A payload that cannot even be decoded carries no correlation id and is dropped. + +Together these guarantee the client's correlation id always resolves: the primary controllers handle what they can name, and the reconciler is the backstop for everything else. + ## Idempotency Runway has no persistent state — no request store, no job store, no database. Idempotency is achieved through the VCS contract: merge detects already-pushed changes (revisions reachable from HEAD) and treats them as already-landed. Merge-conflict check is read-only and naturally idempotent. diff --git a/runway/README.md b/runway/README.md index 5113f738..f8810559 100644 --- a/runway/README.md +++ b/runway/README.md @@ -16,3 +16,5 @@ Each controller deserializes the `MergeRequest`, obtains a `Merger` for the requ ## Failure handling A merge outcome the controller can name is published as a `FAILED` result and acked, not retried: a merge conflict (`merger.ErrConflict`) or an invalid request (`merger.ErrInvalidRequest` — unknown strategy, malformed change URI, invalid PROMOTE composition). The `merger.IsTerminal` helper draws that line. Any other error is an infrastructure fault and is nacked for retry. + +Because Runway is stateless and the sole responder on the client's correlation id, a request that exhausts retries (or hits an unexpected fault) must still resolve the client. The inbound topics dead-letter by default; the [`dlq`](controller/dlq) reconciler drains those `_dlq` topics and republishes a `FAILED` `MergeResult` to the signal topic so the correlation id never hangs. diff --git a/runway/controller/dlq/BUILD.bazel b/runway/controller/dlq/BUILD.bazel new file mode 100644 index 00000000..d3b6d5a0 --- /dev/null +++ b/runway/controller/dlq/BUILD.bazel @@ -0,0 +1,35 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["dlq.go"], + importpath = "github.com/uber/submitqueue/runway/controller/dlq", + visibility = ["//visibility:public"], + deps = [ + "//api/runway/messagequeue:go_default_library", + "//api/runway/messagequeue/protopb:go_default_library", + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/metrics:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["dlq_test.go"], + embed = [":go_default_library"], + deps = [ + "//api/runway/messagequeue:go_default_library", + "//api/runway/messagequeue/protopb:go_default_library", + "//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", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//zaptest:go_default_library", + ], +) diff --git a/runway/controller/dlq/dlq.go b/runway/controller/dlq/dlq.go new file mode 100644 index 00000000..64f5832c --- /dev/null +++ b/runway/controller/dlq/dlq.go @@ -0,0 +1,193 @@ +// 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 dlq reconciles dead-lettered merge requests back to the client. +// +// Runway's inbound merge topics dead-letter a message after the primary +// controller returns a non-retryable error or exhausts retries on a retryable +// one. Runway is stateless and the sole responder on the client's correlation +// id, so a dead-lettered request that produced no signal would leave the client +// (SubmitQueue) waiting forever. Expected outcomes — conflicts and invalid +// requests — are already published as FAILED results by the primary controllers; +// this reconciler is the backstop for everything else (unexpected faults, retry +// exhaustion). +// +// On each delivery from a `{topic}_dlq` topic it decodes the same MergeRequest +// payload the primary controller consumes (the queue preserves the bytes +// verbatim), and republishes a FAILED MergeResult — echoing the correlation id +// — to the corresponding signal topic. Unlike the stovepipe/orchestrator DLQ +// reconcilers it writes no entity state (Runway has none); the signal is the +// resolution. A payload that cannot be decoded carries no correlation id to +// resolve, so it is logged and acked (dropped). +// +// Wire this controller on a dedicated consumer built with +// errs.AlwaysRetryableProcessor so a transient publish failure retries forever +// rather than dead-lettering again (the DLQ topic has no DLQ of its own). +package dlq + +import ( + "context" + "fmt" + + "github.com/uber-go/tally" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + "go.uber.org/zap" +) + +// topicSuffix is appended to a primary topic key to derive its DLQ topic key. +// It matches DefaultSubscriptionConfig's DLQ TopicSuffix so a registered DLQ +// subscription's topic name matches the controller's TopicKey(). +const topicSuffix = "_dlq" + +// TopicKey returns the DLQ topic key for a primary merge topic. Exported so the +// wiring layer builds matching pairs without duplicating the suffix literal. +func TopicKey(main consumer.TopicKey) consumer.TopicKey { + return consumer.TopicKey(string(main) + topicSuffix) +} + +// Verify Controller implements consumer.Controller at compile time. +var _ consumer.Controller = (*Controller)(nil) + +// Controller consumes a merge topic's dead-letter queue and republishes a +// terminal FAILED result to the corresponding signal topic. +type Controller struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + registry consumer.TopicRegistry + topicKey consumer.TopicKey + signalTopicKey consumer.TopicKey + consumerGroup string +} + +// Params are the parameters for creating a new DLQ reconciler. +type Params struct { + // TopicKey is the dead-letter topic this controller consumes + // (typically dlq.TopicKey()). + TopicKey consumer.TopicKey + // SignalTopicKey is the signal topic the FAILED result is published to. + SignalTopicKey consumer.TopicKey + ConsumerGroup string + + Registry consumer.TopicRegistry + + Scope tally.Scope + Logger *zap.SugaredLogger +} + +// NewController creates a DLQ reconciler for a merge topic's dead-letter queue. +func NewController(p Params) *Controller { + return &Controller{ + logger: p.Logger.Named("merge_dlq_controller"), + metricsScope: p.Scope.SubScope("merge_dlq_controller"), + registry: p.Registry, + topicKey: p.TopicKey, + signalTopicKey: p.SignalTopicKey, + consumerGroup: p.ConsumerGroup, + } +} + +// Process decodes the dead-lettered merge request and republishes a FAILED +// result to the signal topic so the client's correlation id resolves. Returns +// nil to ack; an error to nack (retried indefinitely under +// AlwaysRetryableProcessor). +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { + const opName = "process" + + msg := delivery.Message() + meta := delivery.Metadata() + + request := &runwaymq.MergeRequest{} + if err := runwaymq.Unmarshal(msg.Payload, request); err != nil { + // No correlation id is recoverable, so there is nothing to resolve for + // the client. Log and ack (drop) rather than retry forever. + metrics.NamedCounter(c.metricsScope, opName, "undecodable", 1) + c.logger.Errorw("dlq reconcile: undecodable merge request, dropping", + "err", err, + "original_topic", meta["dlq.original_topic"], + ) + return nil + } + + reason := meta["dlq.last_error"] + if reason == "" { + reason = "runway failed to process the merge request" + } + + c.logger.Warnw("dlq reconcile: publishing terminal failure", + "id", request.GetId(), + "queue_name", request.GetQueueName(), + "original_topic", meta["dlq.original_topic"], + "failure_count", meta["dlq.failure_count"], + "last_error", reason, + ) + + result := &runwaymq.MergeResult{ + Id: request.GetId(), + Outcome: runwaypb.Outcome_FAILED, + Reason: fmt.Sprintf("dead-lettered: %s", reason), + } + + if err := c.publish(ctx, result, msg.PartitionKey); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to publish dlq failure result for %s: %w", request.GetId(), err) + } + + metrics.NamedCounter(c.metricsScope, opName, "reconciled", 1) + return nil +} + +// publish serializes a MergeResult and publishes it to the signal topic. +func (c *Controller) publish(ctx context.Context, result *runwaymq.MergeResult, partitionKey string) error { + payload, err := runwaymq.Marshal(result) + if err != nil { + return fmt.Errorf("failed to serialize merge result: %w", err) + } + + msg := entityqueue.NewMessage(result.GetId(), payload, partitionKey, nil) + + q, ok := c.registry.Queue(c.signalTopicKey) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", c.signalTopicKey) + } + + topicName, ok := c.registry.TopicName(c.signalTopicKey) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", c.signalTopicKey) + } + + if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + return fmt.Errorf("failed to publish message: %w", err) + } + + return nil +} + +// Name returns the controller name for logging and metrics. +func (c *Controller) Name() string { + return string(c.topicKey) +} + +// TopicKey returns the dead-letter topic key this controller subscribes to. +func (c *Controller) TopicKey() consumer.TopicKey { + return c.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (c *Controller) ConsumerGroup() string { + return c.consumerGroup +} diff --git a/runway/controller/dlq/dlq_test.go b/runway/controller/dlq/dlq_test.go new file mode 100644 index 00000000..42f3a5c5 --- /dev/null +++ b/runway/controller/dlq/dlq_test.go @@ -0,0 +1,131 @@ +// 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 dlq + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" + 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" + "go.uber.org/zap/zaptest" +) + +const ( + testID = "test-queue/1" + testQueue = "test-queue" + testPartitionKey = "test-queue" +) + +// publishedMsg captures a message published to the signal topic. +type publishedMsg struct { + topic string + msg entityqueue.Message +} + +func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte, meta map[string]string) *queuemock.MockDelivery { + t.Helper() + msg := entityqueue.NewMessage(testID, payload, testPartitionKey, nil) + d := queuemock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(msg).AnyTimes() + d.EXPECT().Metadata().Return(meta).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() + return d +} + +// newRegistry builds a registry whose signal-topic publisher records every +// message it receives into the returned slice pointer. +func newRegistry(t *testing.T, ctrl *gomock.Controller) (consumer.TopicRegistry, *[]publishedMsg) { + t.Helper() + var published []publishedMsg + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, topic string, msg entityqueue.Message) error { + published = append(published, publishedMsg{topic: topic, msg: msg}) + return nil + }, + ).AnyTimes() + + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMergeSignal, Name: "merge-signal", Queue: q}, + }) + require.NoError(t, err) + return registry, &published +} + +func newController(t *testing.T, registry consumer.TopicRegistry) *Controller { + t.Helper() + return NewController(Params{ + Logger: zaptest.NewLogger(t).Sugar(), + Scope: tally.NoopScope, + Registry: registry, + TopicKey: TopicKey(runwaymq.TopicKeyMerge), + SignalTopicKey: runwaymq.TopicKeyMergeSignal, + ConsumerGroup: "runway-merge-dlq", + }) +} + +func TestProcess_DecodableRepublishesFailure(t *testing.T) { + ctrl := gomock.NewController(t) + registry, published := newRegistry(t, ctrl) + controller := newController(t, registry) + + req := &runwaymq.MergeRequest{ + Id: testID, + QueueName: testQueue, + Steps: []*runwaymq.MergeStep{{StepId: "step-1"}}, + } + payload, err := runwaymq.Marshal(req) + require.NoError(t, err) + + meta := map[string]string{ + "dlq.last_error": "boom: connection refused", + "dlq.original_topic": "runway-merge", + } + delivery := newDelivery(t, ctrl, payload, meta) + + require.NoError(t, controller.Process(context.Background(), delivery)) + + require.Len(t, *published, 1) + got := (*published)[0] + assert.Equal(t, "merge-signal", got.topic) + + result := &runwaymq.MergeResult{} + require.NoError(t, runwaymq.Unmarshal(got.msg.Payload, result)) + assert.Equal(t, testID, result.Id) + assert.Equal(t, runwaypb.Outcome_FAILED, result.Outcome) + assert.Contains(t, result.Reason, "boom: connection refused") +} + +func TestProcess_UndecodableAcksAndPublishesNothing(t *testing.T) { + ctrl := gomock.NewController(t) + registry, published := newRegistry(t, ctrl) + controller := newController(t, registry) + + delivery := newDelivery(t, ctrl, []byte("{bad"), map[string]string{"dlq.original_topic": "runway-merge"}) + + require.NoError(t, controller.Process(context.Background(), delivery)) + assert.Empty(t, *published) +} diff --git a/service/runway/README.md b/service/runway/README.md index a60348e5..a2a7598d 100644 --- a/service/runway/README.md +++ b/service/runway/README.md @@ -9,7 +9,9 @@ Runway registers two consuming subscriptions against the shared MySQL-backed mes - **merge-conflict-check** (`TopicKeyMergeConflictCheck`) — handled by `runway/controller/mergeconflictcheck`. - **merge** (`TopicKeyMerge`) — handled by `runway/controller/merge`. -These topic keys and their wire contracts are owned by the queue's producer side and published under `api/runway/messagequeue/` (the external, cross-domain contract). The corresponding signal queues where Runway will publish results are not wired yet. +Each controller applies the request's ordered steps via a `Merger` and publishes a `MergeResult` to the corresponding signal queue (`merge-conflict-check-signal` / `merge-signal`). A second, DLQ consumer drains the inbound topics' dead-letter queues (`merge-conflict-check_dlq`, `runway-merge_dlq`) and republishes a `FAILED` result to the matching signal queue, so a dead-lettered request still resolves the client's correlation id. + +These topic keys and their wire contracts are owned by the queue's producer side and published under `api/runway/messagequeue/` (the external, cross-domain contract). Because Runway only consumes queues and serves `Ping`, it needs a **queue** database but no application/storage database. @@ -18,7 +20,7 @@ Because Runway only consumes queues and serves `Ping`, it needs a **queue** data ``` runway/ ├── server/ -│ ├── main.go # gRPC server (Ping) + primary consumer wiring +│ ├── main.go # gRPC server (Ping) + merge/DLQ consumer wiring │ ├── Dockerfile │ └── docker-compose.yml # Runway service + queue MySQL └── client/ @@ -68,5 +70,5 @@ grpcurl -plaintext -d '{"message": "hello"}' localhost:8086 uber.runway.Runway/P ## Shutdown -The server handles `SIGINT` / `SIGTERM` gracefully: it drains in-flight RPCs, then stops the queue consumer (30s timeout). It exits `0` on clean shutdown, `143` (128 + SIGTERM) when stopped by signal, and `1` on startup/runtime errors (details on stderr). Shutdown errors override the signal exit code. +The server handles `SIGINT` / `SIGTERM` gracefully: it drains in-flight RPCs, then stops the queue consumers — primary and DLQ, 30s timeout each. It exits `0` on clean shutdown, `143` (128 + SIGTERM) when stopped by signal, and `1` on startup/runtime errors (details on stderr). Shutdown errors override the signal exit code. diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index d8a19cc3..9cdf9231 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -23,6 +23,7 @@ go_library( "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", "//runway/controller:go_default_library", + "//runway/controller/dlq:go_default_library", "//runway/controller/merge:go_default_library", "//runway/controller/mergeconflictcheck:go_default_library", "//runway/extension/merger:go_default_library", diff --git a/service/runway/server/main.go b/service/runway/server/main.go index 00deacf7..cb121315 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -40,6 +40,7 @@ import ( extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/runway/controller" + "github.com/uber/submitqueue/runway/controller/dlq" "github.com/uber/submitqueue/runway/controller/merge" "github.com/uber/submitqueue/runway/controller/mergeconflictcheck" "github.com/uber/submitqueue/runway/extension/merger" @@ -150,12 +151,17 @@ func run() error { return fmt.Errorf("failed to create topic registry: %w", err) } + // One gate is shared by the primary and DLQ consumers: a subscription is + // gated by its own consumer group, so a DLQ stage is paused by its own + // group name just like a primary stage. + gate := newConsumerGate(logger) + primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, errs.NewClassifierProcessor( genericerrs.Classifier, mysqlerrs.Classifier, ), - newConsumerGate(logger), + gate, ) mergerFactory := newMergerFactory() @@ -185,10 +191,47 @@ func run() error { } logger.Info("controllers registered", zap.Int("primary", 2)) + // The DLQ consumer reconciles dead-lettered merge requests: it republishes a + // terminal FAILED result to the signal topic so the client's correlation id + // always resolves. It uses AlwaysRetryableProcessor so a transient publish + // failure retries forever rather than dead-lettering again. + dlqConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer-dlq"), registry, + errs.AlwaysRetryableProcessor, + gate, + ) + + mergeConflictCheckDLQController := dlq.NewController(dlq.Params{ + Logger: logger.Sugar(), + Scope: scope, + Registry: registry, + TopicKey: dlq.TopicKey(runwaymq.TopicKeyMergeConflictCheck), + SignalTopicKey: runwaymq.TopicKeyMergeConflictCheckSignal, + ConsumerGroup: "runway-mergeconflictcheck-dlq", + }) + if err := dlqConsumer.Register(mergeConflictCheckDLQController); err != nil { + return fmt.Errorf("failed to register merge-conflict-check DLQ controller: %w", err) + } + + mergeDLQController := dlq.NewController(dlq.Params{ + Logger: logger.Sugar(), + Scope: scope, + Registry: registry, + TopicKey: dlq.TopicKey(runwaymq.TopicKeyMerge), + SignalTopicKey: runwaymq.TopicKeyMergeSignal, + ConsumerGroup: "runway-merge-dlq", + }) + if err := dlqConsumer.Register(mergeDLQController); err != nil { + return fmt.Errorf("failed to register merge DLQ controller: %w", err) + } + logger.Info("DLQ controllers registered", zap.Int("dlq", 2)) + if err := primaryConsumer.Start(ctx); err != nil { return fmt.Errorf("failed to start primary consumer: %w", err) } - logger.Info("consumer started") + if err := dlqConsumer.Start(ctx); err != nil { + return fmt.Errorf("failed to start DLQ consumer: %w", err) + } + logger.Info("consumers started") grpcServer := grpc.NewServer() @@ -240,8 +283,13 @@ func run() error { primaryStopErr = fmt.Errorf("failed to stop consumer: %w", primaryStopErr) } - if primaryStopErr != nil || serverErr != nil { - err = errors.Join(primaryStopErr, serverErr) + dlqStopErr := dlqConsumer.Stop(30000) + if dlqStopErr != nil { + dlqStopErr = fmt.Errorf("failed to stop DLQ consumer: %w", dlqStopErr) + } + + if primaryStopErr != nil || dlqStopErr != nil || serverErr != nil { + err = errors.Join(primaryStopErr, dlqStopErr, serverErr) } return err @@ -290,6 +338,26 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Name: "merge-signal", Queue: q, }, + // DLQ topics: the reconciler consumes these and republishes a FAILED + // result to the corresponding signal topic. Names match the primary + // topic name plus the "_dlq" suffix the subscriber uses when + // dead-lettering (see dlq.TopicKey / DefaultSubscriptionConfig). + { + Key: dlq.TopicKey(runwaymq.TopicKeyMergeConflictCheck), + Name: "merge-conflict-check_dlq", + Queue: q, + Subscription: extqueue.DLQSubscriptionConfig( + subscriberName, "runway-mergeconflictcheck-dlq", + ), + }, + { + Key: dlq.TopicKey(runwaymq.TopicKeyMerge), + Name: "runway-merge_dlq", + Queue: q, + Subscription: extqueue.DLQSubscriptionConfig( + subscriberName, "runway-merge-dlq", + ), + }, }) }