fix(gateway): recover dead-lettered request logs

doc/rfc/submitqueue/workflow.md

@@ -158,7 +158,7 @@ class node_submitter toneIndigo
 
 ## DLQ reconciliation
 
-Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscription. `pipeline.Construct` registers that companion with `errs.AlwaysRetryableProcessor` and `DLQSubscriptionConfig`, which disables a second-level DLQ and sets `Retry.MaxAttempts` to 0 (unlimited). The consumer moves a message to its DLQ once the primary controller returns a non-retryable error or exhausts retries on a retryable one. The publish-only topics — `log`, Runway `merge-conflict-check`, and Runway `runway-merge` — have no orchestrator subscription and therefore no orchestrator DLQ. The gateway consumes `log`. Runway consumes the two merge request topics.
+Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscription. `pipeline.Construct` registers that companion with `errs.AlwaysRetryableProcessor` and `DLQSubscriptionConfig`, which disables a second-level DLQ and sets `Retry.MaxAttempts` to 0 (unlimited). The consumer moves a message to its DLQ once the primary controller returns a non-retryable error or exhausts retries on a retryable one. The publish-only topics — `log`, Runway `merge-conflict-check`, and Runway `runway-merge` — have no orchestrator subscription and therefore no orchestrator DLQ. The gateway consumes `log` and owns its `log_dlq` recovery subscription. Runway consumes the two merge request topics.
 
 Most DLQ controllers do not re-attempt the failed work. They decode the payload to a `RequestID` or `BatchID` and drive that entity to a terminal failed state — `RequestStateError` for requests, `BatchStateFailed` for batches, with fan-out to the member requests. Topics that carry a full payload recover the id from it: the `landconflictsignal` DLQ reads the request id from Runway's `MergeResult`, and the `landsignal` DLQ reads the batch id. The `buildsignal` DLQ decodes a build id, loads that build, and fails the batch that owns it. A missing build or an empty batch id is acked: there is no batch to fail from this signal. State writes use the same optimistic-locking CAS as the primary pipeline, so a late primary-pipeline update wins cleanly and a version mismatch is asked back for redelivery.
 
@@ -203,3 +203,5 @@ Rejected alternatives:
 - **UUIDs or one global counter.** They buy global uniqueness at the cost of readable IDs or cross-queue coordination.
 - **Integer wire fields.** They would tie the API to the counter's representation without adding meaning.
 - **SQL auto-increment, `MAX(id) + 1`, or process-local counters.** These put allocation inside one store, race under concurrency, or reuse IDs after a restart.
+
+The gateway replays `log_dlq` through its primary log controller and materializer, rather than asking the orchestrator to write gateway projections. Recovery acknowledges only after all public read-model writes succeed; unlimited retries and capped backoff keep failed writes durable. Malformed payloads and conflicting payload/envelope identities remain in the final queue, with structured diagnostics and counters. Retention is not isolated quarantine: poison entries can impede later work. Before deleting a bad record, operators must durably export it and record a disposition; corrected records re-enter gateway materialization. Replays preserve equivalent audit-history rows while summary precedence and incomplete-mapping repair remain unchanged. Rollback leaves final-DLQ rows intact.

platform/consumer/consumer.go

@@ -308,7 +308,7 @@ func (m *consumer) consumeLoop(ctx context.Context, controller Controller, topic
 				m.shutdownPartitions(partitionChs, &wg)
 				return
 			}
-			if m.rejectDeliveryWithInvalidTenant(ctx, controller, delivery) {
+			if m.rejectDeliveryWithInvalidTenant(ctx, controller, delivery, controllerScope) {
 				continue
 			}
 
@@ -340,27 +340,41 @@ func (m *consumer) consumeLoop(ctx context.Context, controller Controller, topic
 	}
 }
 
-func (m *consumer) rejectDeliveryWithInvalidTenant(ctx context.Context, controller Controller, delivery extqueue.Delivery) bool {
+func (m *consumer) rejectDeliveryWithInvalidTenant(ctx context.Context, controller Controller, delivery extqueue.Delivery, scope tally.Scope) bool {
 	msg := delivery.Message()
-	if err := entityqueue.ValidateTenantMetadata(msg); err != nil {
-		m.logger.Errorw("rejecting message with inconsistent queue identity",
-			"controller", controller.Name(),
+	err := entityqueue.ValidateTenantMetadata(msg)
+	if err == nil {
+		return false
+	}
+
+	settle := delivery.Reject
+	action := "reject"
+	config, ok := m.registry.SubscriptionConfig(controller.TopicKey(), controller.ConsumerGroup())
+	if ok && config.Retry.MaxAttempts == 0 && !config.DLQ.Enabled {
+		// Reject without a DLQ acknowledges the message, violating final recovery's retention policy.
+		settle = delivery.Nack
+		action = "nack"
+	}
+	metrics.NamedCounter(scope, "envelope", "invalid_identity", 1)
+	m.logger.Errorw("message has inconsistent queue identity",
+		"controller", controller.Name(),
+		"topic_key", controller.TopicKey(),
+		"consumer_group", controller.ConsumerGroup(),
+		"message_id", msg.ID,
+		"tenant", msg.Tenant,
+		"action", action,
+		"error", err,
+	)
+	if settleErr := settle(ctx, failure.New(err.Error())); settleErr != nil {
+		m.logger.Errorw("failed to settle message with inconsistent queue identity",
 			"topic_key", controller.TopicKey(),
+			"consumer_group", controller.ConsumerGroup(),
 			"message_id", msg.ID,
-			"tenant", msg.Tenant,
-			"error", err,
+			"action", action,
+			"error", settleErr,
 		)
-		if rejectErr := delivery.Reject(ctx, failure.New(err.Error())); rejectErr != nil {
-			m.logger.Errorw("failed to reject message with inconsistent queue identity",
-				"controller", controller.Name(),
-				"topic_key", controller.TopicKey(),
-				"message_id", msg.ID,
-				"error", rejectErr,
-			)
-		}
-		return true
 	}
-	return false
+	return true
 }
 
 // shutdownPartitions closes all partition channels to signal processPartition

platform/consumer/consumer_test.go

@@ -332,46 +332,52 @@ func TestConsumer_ProcessDelivery_Success(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestConsumer_ProcessDelivery_RejectsConflictingTenantMetadata(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	deliveryChan := make(chan extqueue.Delivery, 1)
-	mockSub := queuemock.NewMockSubscriber(ctrl)
-	mockSub.EXPECT().Subscribe(gomock.Any(), gomock.Any(), gomock.Any()).Return(deliveryChan, nil)
-	mockQ := queuemock.NewMockQueue(ctrl)
-	mockQ.EXPECT().Subscriber().Return(mockSub)
-
-	c := New(
-		zaptest.NewLogger(t).Sugar(),
-		tally.NoopScope,
-		newRegistry(t, mockQ, testTopicKeyStart, "test-group"),
-		errs.NewClassifierProcessor(),
-		consumergatenoop.New(),
-	)
-	handler := &testController{}
-	setupController(handler, "test-handler", testTopicKeyStart, "test-group",
-		func(context.Context, Delivery) error {
-			assert.Fail(t, "controller must not receive inconsistent delivery identity")
-			return nil
-		},
-	)
-	require.NoError(t, c.Register(handler))
-	require.NoError(t, c.Start(context.Background()))
-
-	msg := entityqueue.NewMessage("mismatch", []byte("payload"), "partition1", map[string]string{
-		entityqueue.MetadataKeyQueueName: "tenant-b",
-	})
-	msg.Tenant = "tenant-a"
-	rejected := make(chan struct{})
-	delivery := queuemock.NewMockDelivery(ctrl)
-	delivery.EXPECT().Message().Return(msg).AnyTimes()
-	delivery.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(context.Context, failure.Failure) error {
-		close(rejected)
-		return nil
-	})
-
-	deliveryChan <- delivery
-	<-rejected
-	require.NoError(t, c.Stop(30000))
+func TestConsumer_ProcessDelivery_InvalidTenantMetadata(t *testing.T) {
+	for _, final := range []bool{false, true} {
+		for _, settlementErr := range []error{nil, fmt.Errorf("settlement unavailable")} {
+			t.Run(fmt.Sprintf("final=%v settlementError=%v", final, settlementErr != nil), func(t *testing.T) {
+				ctrl := gomock.NewController(t)
+				deliveryChan := make(chan extqueue.Delivery, 1)
+				mockSub := queuemock.NewMockSubscriber(ctrl)
+				mockSub.EXPECT().Subscribe(gomock.Any(), "start", gomock.Any()).Return(deliveryChan, nil)
+				mockQ := queuemock.NewMockQueue(ctrl)
+				mockQ.EXPECT().Subscriber().Return(mockSub)
+				config := extqueue.DefaultSubscriptionConfig("worker", "test-group")
+				if final {
+					config = extqueue.DLQSubscriptionConfig("worker", "test-group")
+				}
+				registry, err := NewTopicRegistry([]TopicConfig{{Key: testTopicKeyStart, Name: "start", Queue: mockQ, Subscription: config}})
+				require.NoError(t, err)
+				c := New(zaptest.NewLogger(t).Sugar(), tally.NoopScope, registry, errs.NewClassifierProcessor(), consumergatenoop.New())
+				handler := &testController{name: "test-handler", topicKey: testTopicKeyStart, consumerGroup: "test-group",
+					processFunc: func(context.Context, Delivery) error {
+						assert.Fail(t, "controller must not receive inconsistent delivery identity")
+						return nil
+					},
+				}
+				require.NoError(t, c.Register(handler))
+				require.NoError(t, c.Start(context.Background()))
+
+				msg := entityqueue.NewMessage("mismatch", []byte("payload"), "partition1", map[string]string{entityqueue.MetadataKeyQueueName: "tenant-b"})
+				msg.Tenant = "tenant-a"
+				settled := make(chan struct{})
+				delivery := queuemock.NewMockDelivery(ctrl)
+				delivery.EXPECT().Message().Return(msg).AnyTimes()
+				settle := func(context.Context, failure.Failure) error {
+					close(settled)
+					return settlementErr
+				}
+				if final {
+					delivery.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(settle)
+				} else {
+					delivery.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(settle)
+				}
+				deliveryChan <- delivery
+				<-settled
+				require.NoError(t, c.Stop(30000))
+			})
+		}
+	}
 }
 
 func TestConsumer_ProcessDelivery_RejectsEmptyTenant(t *testing.T) {

service/submitqueue/gateway/server/BUILD.bazel

@@ -12,7 +12,10 @@ exports_files(
 
 go_library(
     name = "gateway_lib",
-    srcs = ["main.go"],
+    srcs = [
+        "log_consumer.go",
+        "main.go",
+    ],
     importpath = "github.com/uber/submitqueue/service/submitqueue/gateway/server",
     visibility = ["//visibility:private"],
     deps = [
@@ -28,6 +31,7 @@ go_library(
         "//platform/extension/counter/mysql:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
+        "//platform/lifecycle:go_default_library",
         "//service/messagequeue:go_default_library",
         "//service/submitqueue/core:go_default_library",
         "//service/submitqueue/gateway/server/mapper:go_default_library",
@@ -83,17 +87,32 @@ filegroup(
 
 go_test(
     name = "go_default_test",
-    srcs = ["main_test.go"],
+    srcs = [
+        "log_consumer_test.go",
+        "main_test.go",
+    ],
     embed = [":gateway_lib"],  # keep
     deps = [
         "//api/submitqueue/gateway/protopb:go_default_library",
+        "//platform/base/failure:go_default_library",
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
         "//platform/errs:go_default_library",
+        "//platform/extension/consumergate/noop:go_default_library",
+        "//platform/extension/messagequeue:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
+        "//platform/lifecycle:go_default_library",
+        "//submitqueue/core/messagequeue:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/queueconfig/mock:go_default_library",
         "//submitqueue/gateway/controller:go_default_library",
+        "//submitqueue/gateway/core/request:go_default_library",
+        "//submitqueue/gateway/extension/storage:go_default_library",
+        "//submitqueue/gateway/extension/storage/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_golang_google_grpc//:go_default_library",
         "@org_golang_google_grpc//codes:go_default_library",
         "@org_golang_google_grpc//status:go_default_library",
         "@org_uber_go_mock//gomock:go_default_library",

service/submitqueue/gateway/server/log_consumer.go

@@ -0,0 +1,87 @@
+// 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 main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/errs"
+	genericerrs "github.com/uber/submitqueue/platform/errs/generic"
+	mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
+	"github.com/uber/submitqueue/platform/lifecycle"
+	"github.com/uber/submitqueue/submitqueue/core/topickey"
+	logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log"
+	requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request"
+	"go.uber.org/zap"
+)
+
+const (
+	logDLQSuffix        = "_dlq"
+	logDLQTopicKey      = topickey.TopicKeyLog + consumer.TopicKey(logDLQSuffix)
+	logConsumerGroup    = "gateway-log"
+	logDLQConsumerGroup = "gateway-log-dlq"
+)
+
+func newGatewayTopicRegistry(queue extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
+	primary := extqueue.DefaultSubscriptionConfig(subscriberName, logConsumerGroup)
+	primary.DLQ.TopicSuffix = logDLQSuffix
+	return consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{Key: topickey.TopicKeyStart, Name: "start", Queue: queue},
+		{Key: topickey.TopicKeyCancel, Name: "cancel", Queue: queue},
+		{Key: topickey.TopicKeyLog, Name: "log", Queue: queue, Subscription: primary},
+		{
+			Key: logDLQTopicKey, Name: "log" + logDLQSuffix, Queue: queue,
+			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, logDLQConsumerGroup),
+		},
+	})
+}
+
+func newLogConsumers(logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, materializer *requestcore.Materializer, gate consumergate.Gate) (*lifecycle.Group, error) {
+	policies := []struct {
+		topic     consumer.TopicKey
+		group     string
+		processor errs.ErrorProcessor
+	}{
+		{topickey.TopicKeyLog, logConsumerGroup, errs.NewClassifierProcessor(genericerrs.Classifier, mysqlerrs.Classifier)},
+		{logDLQTopicKey, logDLQConsumerGroup, errs.AlwaysRetryableProcessor},
+	}
+	components := make([]lifecycle.Component, 0, len(policies))
+	for _, policy := range policies {
+		c := consumer.New(logger, scope.SubScope("consumer"), registry, policy.processor, gate)
+		if err := c.Register(logctrl.NewController(logger, scope, materializer, policy.topic, policy.group)); err != nil {
+			return nil, fmt.Errorf("failed to register log controller: %w", err)
+		}
+		components = append(components, logConsumerLifecycle{Consumer: c})
+	}
+	return lifecycle.NewGroup(components...), nil
+}
+
+type logConsumerLifecycle struct {
+	consumer.Consumer
+}
+
+func (c logConsumerLifecycle) Stop(ctx context.Context) error {
+	timeout := 30 * time.Second
+	if deadline, ok := ctx.Deadline(); ok {
+		timeout = min(timeout, max(time.Until(deadline), 0))
+	}
+	return c.Consumer.Stop(timeout.Milliseconds())
+}