feat(runway): dlq reconciler for merge topics

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.

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.

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",
+    ],
+)

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(<primary topic>)).
+	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
+}

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)
+}