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,7 @@ 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. Valid log recovery acknowledges only after all public read-model writes succeed; unlimited retries and capped backoff keep failed writes durable. With valid envelope metadata, malformed payloads and payload-queue conflicts remain in the final queue, with structured diagnostics and counters; retrying unchanged bad data does not repair it. 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.
+
+Invalid envelope metadata is instead logged and rejected before controller processing. It is a permanent error while unchanged; without another DLQ, rejection may acknowledge and discard the delivery rather than retry it.

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,91 @@
+// 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) {
+	// Both lanes share the controller/materializer. One always-retry primary with
+	// unlimited attempts and no DLQ is an alternative. Separate recovery keeps failed
+	// logs off normal traffic; in the SQL backend, the primary acknowledgment/cleanup
+	// checkpoint can then advance past them.
+	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())
+}

service/submitqueue/gateway/server/log_consumer_test.go

@@ -0,0 +1,175 @@
+// 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"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/base/failure"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+	consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
+	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
+	queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
+	"github.com/uber/submitqueue/platform/lifecycle"
+	sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue"
+	"github.com/uber/submitqueue/submitqueue/entity"
+	requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request"
+	gwstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage"
+	gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock"
+	"go.uber.org/mock/gomock"
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+)
+
+func TestLogConsumers_RetainFailedRecovery(t *testing.T) {
+	for _, tt := range []struct {
+		name       string
+		topic      string
+		malformed  bool
+		wrongQueue bool
+	}{
+		{name: "primary rejects unclassified failure", topic: "log"},
+		{name: "final retries unclassified failure", topic: "log_dlq"},
+		{name: "final retains malformed payload", topic: "log_dlq", malformed: true},
+		{name: "final retains payload identity mismatch", topic: "log_dlq", wrongQueue: true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			primary := make(chan extqueue.Delivery, 1)
+			final := make(chan extqueue.Delivery, 1)
+			subscriber := queuemock.NewMockSubscriber(ctrl)
+			subscriber.EXPECT().Subscribe(gomock.Any(), "log", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, config extqueue.SubscriptionConfig) (<-chan extqueue.Delivery, error) {
+				assert.Equal(t, "gateway-log", config.ConsumerGroup)
+				assert.True(t, config.DLQ.Enabled)
+				assert.Equal(t, "_dlq", config.DLQ.TopicSuffix)
+				assert.Positive(t, config.Retry.MaxAttempts)
+				return primary, nil
+			})
+			subscriber.EXPECT().Subscribe(gomock.Any(), "log_dlq", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, config extqueue.SubscriptionConfig) (<-chan extqueue.Delivery, error) {
+				assert.Equal(t, "gateway-log-dlq", config.ConsumerGroup)
+				assert.False(t, config.DLQ.Enabled)
+				assert.Zero(t, config.Retry.MaxAttempts)
+				assert.Positive(t, config.Retry.MaxBackoffMs)
+				return final, nil
+			})
+			queue := queuemock.NewMockQueue(ctrl)
+			queue.EXPECT().Subscriber().Return(subscriber).Times(2)
+			registry, err := newGatewayTopicRegistry(queue, "worker")
+			require.NoError(t, err)
+			factory := gwstoragemock.NewMockFactory(ctrl)
+			if !tt.malformed && !tt.wrongQueue {
+				factory.EXPECT().For(gwstorage.Config{QueueName: "queue-a"}).Return(nil, errors.New("storage unavailable"))
+			}
+			consumers, err := newLogConsumers(zap.NewNop().Sugar(), tally.NoopScope, registry, requestcore.NewMaterializer(factory), consumergatenoop.New())
+			require.NoError(t, err)
+			require.NoError(t, consumers.Start(context.Background()))
+			t.Cleanup(func() { require.NoError(t, consumers.Stop(context.Background())) })
+
+			log := entity.NewRequestStatusLog("queue-a", "1", entity.RequestStatusLanded, 3, "", nil)
+			payload, err := sqmq.Marshal(sqmq.LogFromEntity(log))
+			require.NoError(t, err)
+			if tt.malformed {
+				payload = []byte("not JSON")
+			}
+			msg := entityqueue.NewMessage("terminal-log", payload, "queue-a", nil)
+			msg.Tenant = "queue-a"
+			if tt.wrongQueue {
+				msg.Tenant = "queue-b"
+			}
+			delivery := queuemock.NewMockDelivery(ctrl)
+			delivery.EXPECT().Message().Return(msg).AnyTimes()
+			delivery.EXPECT().Attempt().Return(100).AnyTimes()
+			delivery.EXPECT().Metadata().Return(nil).AnyTimes()
+			settled := make(chan struct{})
+			settle := func(context.Context, failure.Failure) error { close(settled); return nil }
+			if tt.topic == "log" {
+				delivery.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(settle)
+				primary <- delivery
+			} else {
+				delivery.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(settle)
+				final <- delivery
+			}
+			<-settled
+		})
+	}
+}
+
+type lifecycleTestConsumer struct {
+	consumer.Consumer
+	name     string
+	calls    *[]string
+	startErr error
+	stopErr  error
+}
+
+func (c lifecycleTestConsumer) Start(context.Context) error {
+	*c.calls = append(*c.calls, "start:"+c.name)
+	return c.startErr
+}
+
+func (c lifecycleTestConsumer) Stop(timeoutMs int64) error {
+	*c.calls = append(*c.calls, "stop:"+c.name)
+	return c.stopErr
+}
+
+func TestServeGateway_ConsumerCleanup(t *testing.T) {
+	startErr := errors.New("subscribe unavailable")
+	stopErr := errors.New("stop unavailable")
+	for _, tt := range []struct {
+		name        string
+		primaryErr  error
+		recoveryErr error
+		stopErr     error
+		cancelled   bool
+		want        []string
+	}{
+		{name: "primary fails", primaryErr: startErr, want: []string{"start:primary"}},
+		{name: "recovery fails and rollback fails", recoveryErr: startErr, stopErr: stopErr, want: []string{"start:primary", "start:recovery", "stop:primary"}},
+		{name: "listener fails and both stop despite cleanup failure", stopErr: stopErr, want: []string{"start:primary", "start:recovery", "stop:recovery", "stop:primary"}},
+		{name: "signal drains both", cancelled: true, want: []string{"start:primary", "start:recovery", "stop:recovery", "stop:primary"}},
+		{name: "signal does not mask shutdown failure", cancelled: true, stopErr: stopErr, want: []string{"start:primary", "start:recovery", "stop:recovery", "stop:primary"}},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			var calls []string
+			group := lifecycle.NewGroup(
+				logConsumerLifecycle{Consumer: lifecycleTestConsumer{name: "primary", calls: &calls, startErr: tt.primaryErr, stopErr: tt.stopErr}},
+				logConsumerLifecycle{Consumer: lifecycleTestConsumer{name: "recovery", calls: &calls, startErr: tt.recoveryErr, stopErr: tt.stopErr}},
+			)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			t.Setenv("PORT", "invalid-port")
+			if tt.cancelled {
+				t.Setenv("PORT", "127.0.0.1:0")
+				cancel()
+			}
+			err := serveGateway(ctx, cancel, grpc.NewServer(), group)
+			require.Error(t, err)
+			if tt.primaryErr != nil || tt.recoveryErr != nil {
+				assert.ErrorIs(t, err, startErr)
+			}
+			if tt.stopErr != nil {
+				assert.ErrorIs(t, err, stopErr)
+			}
+			assert.Equal(t, tt.cancelled && tt.stopErr == nil, errors.Is(err, context.Canceled))
+			assert.Equal(t, tt.want, calls)
+		})
+	}
+}