From 46fd1258d94373f55a845035a7c453a2367a68e7 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Fri, 6 Nov 2026 22:22:48 +0000 Subject: [PATCH 1/2] fix(gateway): recover dead-lettered request logs Summary: A terminal request log that exhausts primary retries can leave the public status permanently stale. Replay the gateway-owned final log backlog through the existing materializer, retain malformed payloads and envelopes, and clean up both consumers on partial startup or shutdown. Test Plan: 1. Seed a terminal log in log_dlq before the gateway starts, while its receipt summary is unavailable. 2. Observe recovery retrying without acknowledgment, then restore the summary without publishing another event. 3. TestFinalLogBacklogRecovery verifies the repaired terminal summary, URI mapping, List receipt, and retained history. TestLogConsumers_RetainFailedRecovery, TestServeGateway_ConsumerCleanup, and TestConsumer_ProcessDelivery_InvalidTenantMetadata cover retention and lifecycle failures. Revert Plan: Revert this change and leave the log_dlq backlog intact. --- doc/rfc/submitqueue/workflow.md | 4 +- platform/consumer/consumer.go | 48 +++-- platform/consumer/consumer_test.go | 86 +++++---- .../submitqueue/gateway/server/BUILD.bazel | 23 ++- .../gateway/server/log_consumer.go | 87 +++++++++ .../gateway/server/log_consumer_test.go | 176 ++++++++++++++++++ service/submitqueue/gateway/server/main.go | 102 ++-------- submitqueue/gateway/README.md | 6 + submitqueue/gateway/controller/log/log.go | 17 +- .../submitqueue/gateway/BUILD.bazel | 1 + .../submitqueue/gateway/suite_test.go | 85 ++++++++- 11 files changed, 480 insertions(+), 155 deletions(-) create mode 100644 service/submitqueue/gateway/server/log_consumer.go create mode 100644 service/submitqueue/gateway/server/log_consumer_test.go diff --git a/doc/rfc/submitqueue/workflow.md b/doc/rfc/submitqueue/workflow.md index 2bd408937..f07e8f191 100644 --- a/doc/rfc/submitqueue/workflow.md +++ b/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. diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 7aab1b906..4dcb4b4b6 100644 --- a/platform/consumer/consumer.go +++ b/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 diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 61de23d91..b95d70373 100644 --- a/platform/consumer/consumer_test.go +++ b/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) { diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index 0a2d9c205..5e1709930 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/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", diff --git a/service/submitqueue/gateway/server/log_consumer.go b/service/submitqueue/gateway/server/log_consumer.go new file mode 100644 index 000000000..572e35661 --- /dev/null +++ b/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()) +} diff --git a/service/submitqueue/gateway/server/log_consumer_test.go b/service/submitqueue/gateway/server/log_consumer_test.go new file mode 100644 index 000000000..d8f7c2883 --- /dev/null +++ b/service/submitqueue/gateway/server/log_consumer_test.go @@ -0,0 +1,176 @@ +// 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"}}, + } { + 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) + } + if tt.cancelled { + assert.ErrorIs(t, err, context.Canceled) + } + assert.Equal(t, tt.want, calls) + }) + } +} diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 27a56497a..4c5b2d411 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -29,24 +29,18 @@ import ( _ "github.com/go-sql-driver/mysql" "github.com/uber-go/tally" pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" - "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" consumergatefile "github.com/uber/submitqueue/platform/extension/consumergate/file" consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" "github.com/uber/submitqueue/platform/extension/counter" mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql" - extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/platform/lifecycle" servicemq "github.com/uber/submitqueue/service/messagequeue" "github.com/uber/submitqueue/service/submitqueue/core" "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper" - "github.com/uber/submitqueue/submitqueue/core/topickey" yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml" "github.com/uber/submitqueue/submitqueue/gateway/controller" - logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" mysqlstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mysql" @@ -319,24 +313,7 @@ func run() error { } subscriberName := fmt.Sprintf("%s-%d", hostname, os.Getpid()) - // Build the topic registry. The gateway publishes to the start of the - // orchestrator pipeline (TopicKeyStart) and the cancel topic (TopicKeyCancel) — - // both publish-only. It additionally consumes the log topic (TopicKeyLog): - // the gateway persists normal pipeline entries that the orchestrator - // publishes there. Orchestrator DLQ reconciliation may repair terminal - // request-log projections directly. - registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ - {Key: topickey.TopicKeyStart, Name: "start", Queue: mysqlQueue}, - {Key: topickey.TopicKeyCancel, Name: "cancel", Queue: mysqlQueue}, - { - Key: topickey.TopicKeyLog, - Name: "log", - Queue: mysqlQueue, - Subscription: extqueue.DefaultSubscriptionConfig( - subscriberName, "gateway-log", - ), - }, - }) + registry, err := newGatewayTopicRegistry(mysqlQueue, subscriberName) if err != nil { return fmt.Errorf("failed to create topic registry: %w", err) } @@ -389,32 +366,20 @@ func run() error { // Register reflection service for debugging with grpcurl reflection.Register(grpcServer) - // Create the queue consumer and register the log controller. The orchestrator - // publishes normal pipeline entries to the log topic and this consumer writes - // them to storage. Orchestrator DLQ reconciliation repairs terminal entries - // directly so reconciliation does not depend on another asynchronous path. - logConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, - errs.NewClassifierProcessor( - // Storage (submitqueue/gateway/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql) - // both run on the same MySQL driver, so a single classifier covers - // errors surfaced from either backend. - genericerrs.Classifier, - mysqlerrs.Classifier, - ), - newConsumerGate(logger), - ) - - logController := logctrl.NewController(logger.Sugar(), scope, materializer, topickey.TopicKeyLog, "gateway-log") - if err := logConsumer.Register(logController); err != nil { - return fmt.Errorf("failed to register log controller: %w", err) + logConsumers, err := newLogConsumers(logger.Sugar(), scope, registry, materializer, newConsumerGate(logger)) + if err != nil { + return err } + return serveGateway(ctx, cancel, grpcServer, logConsumers) +} - if err := logConsumer.Start(ctx); err != nil { - // The error can also be a result of a context cancellation due to SIGINT or SIGTERM. - // This is expected, just propagate it. - return fmt.Errorf("failed to start log consumer: %w", err) +func serveGateway(ctx context.Context, cancel context.CancelFunc, grpcServer *grpc.Server, logConsumers lifecycle.Component) (runErr error) { + if err := logConsumers.Start(ctx); err != nil { + return fmt.Errorf("failed to start log consumers: %w", err) } - logger.Info("log consumer started") + defer func() { + runErr = errors.Join(runErr, logConsumers.Stop(context.Background())) + }() // Listen on configurable port port := os.Getenv("PORT") @@ -426,6 +391,8 @@ func run() error { return fmt.Errorf("failed to listen on port %s: %w", port, err) } + defer listener.Close() + fmt.Printf("Gateway gRPC server is running on %s\n", port) fmt.Println("Press Ctrl+C to stop, or send a SIGTERM.") @@ -435,50 +402,21 @@ func run() error { serverErrCh <- grpcServer.Serve(listener) }() - // Wait for interrupt signal or server critical error - // If interruption is signaled, gracefully stop the server - // If the server exits with an error, cancel the context to signal the consumer - // After this, stop the consumer - // If an error happens during shutdown, return the actual error, not the context cancellation error - var serverErr error + var serverErr, shutdownErr error select { case <-ctx.Done(): - fmt.Println("Shutting down gateway server due to interruption signal...") - - // Set the error to the context cancellation error to be surfaced as a desired exit code by the main function - // to indicate that the server was stopped as intended - // It may be overridden by the server error if any - err = ctx.Err() - - // stop GRPC server and wait for it to exit + shutdownErr = ctx.Err() grpcServer.GracefulStop() serverErr = <-serverErrCh case serverErr = <-serverErrCh: - fmt.Println("Shutting down gateway server due to critical GRPC server error...") - - // Cancel the context to signal cancellation to the queue consumer cancel() } - if serverErr != nil { - serverErr = fmt.Errorf("GRPC server exited with error: %w", serverErr) - } - - // Stop the consumer with a 30s timeout; by this time the context should be - // cancelled and the processing threads may already be exiting; recollect them. - errStop := logConsumer.Stop(30000) - if errStop != nil { - errStop = fmt.Errorf("failed to stop consumer: %w", errStop) - } - - if errStop != nil || serverErr != nil { - // Override context cancellation error with the shutdown error. The server - // error is the primary/root failure, so it leads; the consumer-stop error - // is secondary cleanup. - err = errors.Join(serverErr, errStop) + if serverErr != nil && !errors.Is(serverErr, grpc.ErrServerStopped) { + return fmt.Errorf("GRPC server exited with error: %w", serverErr) } - return err + return shutdownErr } func validateConfiguredQueueTenants(tenants, configuredQueueNames []string) error { diff --git a/submitqueue/gateway/README.md b/submitqueue/gateway/README.md index cd0695524..a7bd5bd2d 100644 --- a/submitqueue/gateway/README.md +++ b/submitqueue/gateway/README.md @@ -24,3 +24,9 @@ The gateway owns the request log read model and is the only service that reads i - `GetRequestHistoryByID` and `GetRequestHistoryByChangeURI` read retained request-log rows directly. The materializer appends every audit event, selects the current authoritative winner, and repairs the queue projection. The normal orchestrator pipeline does not read or write the request-log store directly. + +## Final log recovery + +The primary `gateway-log` consumer sends failed deliveries to `log_dlq` after a non-retryable error or exhaustion of its finite retry budget. The gateway's `gateway-log-dlq` consumer replays retained entries through the same log controller and materializer, with unlimited attempts, capped backoff, and no second-level DLQ. It acknowledges only after history, summary, URI mappings, and receipt projection succeed. Replays may append equivalent history rows; terminal-summary precedence and mapping repair are unchanged. + +Malformed payloads and inconsistent payload or envelope identities stay in `log_dlq`, with topic/group-tagged diagnostics and deserialize, identity, or envelope-error counters. This is retention, not isolated quarantine: poison records can occupy the fetch window and delay later recovery. Operators must durably export a bad record and record its disposition before deleting it; corrected payloads must be replayed through gateway materialization. No export/replay CLI is provided. Rolling back the gateway stops automatic recovery but must leave the backlog intact. diff --git a/submitqueue/gateway/controller/log/log.go b/submitqueue/gateway/controller/log/log.go index d5c87be78..1d8e9fef3 100644 --- a/submitqueue/gateway/controller/log/log.go +++ b/submitqueue/gateway/controller/log/log.go @@ -27,13 +27,7 @@ import ( "go.uber.org/zap" ) -// Controller handles log queue messages. -// It consumes request log entries and persists them to storage. -// Implements consumer.Controller interface for integration with the consumer. -// -// The request log is written exclusively by the gateway: other services -// (e.g. the orchestrator) only publish log entries to the log topic, and this -// controller is the single consumer that persists them to storage. +// Controller materializes gateway-owned request logs from primary or recovery deliveries. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -54,8 +48,8 @@ func NewController( consumerGroup string, ) *Controller { return &Controller{ - logger: logger.Named("log_controller"), - metricsScope: scope.SubScope("log_controller"), + logger: logger.Named("log_controller").With("topic_key", topicKey, "consumer_group", consumerGroup), + metricsScope: scope.SubScope("log_controller").Tagged(map[string]string{"topic_key": topicKey.String(), "consumer_group": consumerGroup}), materializer: materializer, topicKey: topicKey, consumerGroup: consumerGroup, @@ -73,10 +67,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er logEntry, err := sqmq.UnmarshalRequestLog(msg.Payload) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "deserialize_errors", 1) - // Non-retryable: malformed messages will never succeed regardless of retry count return fmt.Errorf("failed to deserialize request log: %w", err) } if err := entityqueue.ValidatePayloadQueue(msg, logEntry.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "identity_errors", 1) return fmt.Errorf("invalid message identity: %w", err) } @@ -87,13 +81,12 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "attempt", delivery.Attempt(), ) - // Persist the audit log and materialized views as one retryable application operation. if err := c.materializer.PersistLog(ctx, logEntry); err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) return fmt.Errorf("failed to persist request log: %w", err) } - return nil // Success - message will be acked + return nil } // Name returns the controller name for logging and metrics. diff --git a/test/integration/submitqueue/gateway/BUILD.bazel b/test/integration/submitqueue/gateway/BUILD.bazel index 678fbc231..1cbc8be4f 100644 --- a/test/integration/submitqueue/gateway/BUILD.bazel +++ b/test/integration/submitqueue/gateway/BUILD.bazel @@ -21,6 +21,7 @@ go_test( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", + "//submitqueue/core/messagequeue:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/gateway/core/request:go_default_library", diff --git a/test/integration/submitqueue/gateway/suite_test.go b/test/integration/submitqueue/gateway/suite_test.go index 46a27865e..222368b2f 100644 --- a/test/integration/submitqueue/gateway/suite_test.go +++ b/test/integration/submitqueue/gateway/suite_test.go @@ -41,6 +41,7 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" corerequest "github.com/uber/submitqueue/submitqueue/gateway/core/request" @@ -91,7 +92,7 @@ func (s *GatewayIntegrationSuite) SetupSuite() { })) // Start the compose stack (Gateway + 2 MySQL DBs) - err := s.stack.Up() + err := s.stack.Up("mysql-app", "mysql-queue") require.NoError(t, err, "failed to start compose stack") s.log.Logf("Compose stack started successfully") @@ -113,6 +114,20 @@ func (s *GatewayIntegrationSuite) SetupSuite() { s.log.Logf("Schemas applied successfully") + // Seed retained work before the gateway starts; its missing summary forces a durable retry. + queue, err := queueMySQL.NewQueue(queueMySQL.Params{ + DB: s.queueDB, Logger: zap.NewNop(), MetricsScope: tally.NoopScope, Tenants: []string{"test-queue"}, + }) + require.NoError(t, err) + log := entity.NewRequestStatusLog("test-queue", "910", entity.RequestStatusLanded, 3, "", nil) + payload, err := sqmq.Marshal(sqmq.LogFromEntity(log)) + require.NoError(t, err) + msg := entityqueue.NewMessage("retained-terminal-log", payload, "test-queue", nil) + msg.Tenant = "test-queue" + require.NoError(t, queue.Publisher().Publish(s.ctx, "log_dlq", msg)) + require.NoError(t, queue.Close()) + require.NoError(t, s.stack.Up("gateway-service")) + // Connect to Gateway gRPC service var conn *grpc.ClientConn conn, err = s.stack.ConnectGRPC("gateway-service", 8080) @@ -345,6 +360,74 @@ func (s *GatewayIntegrationSuite) TestReadAPIErrorCodes() { assert.Equal(t, codes.Internal, status.Code(err)) } +func (s *GatewayIntegrationSuite) TestFinalLogBacklogRecovery() { + t := s.T() + const sqid = "910" + const queueName = "test-queue" + const uri = "uri/final-log-backlog" + ticker := time.NewTicker(persistPollInterval) + defer ticker.Stop() + for { + var retries int + var acked bool + err := s.queueDB.QueryRowContext(s.ctx, `SELECT retry_count, acked FROM queue_delivery_state + WHERE tenant = ? AND topic = 'log_dlq' AND consumer_group = 'gateway-log-dlq'`, queueName).Scan(&retries, &acked) + if err != sql.ErrNoRows { + require.NoError(t, err) + assert.False(t, acked, "failed recovery must leave the terminal log durable") + if retries > 0 { + break + } + } + <-ticker.C + } + + store, err := mysqlstorage.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err) + queueStore, err := store.For(queueName) + require.NoError(t, err) + require.NoError(t, queueStore.GetRequestSummaryStore().Create(s.ctx, entity.RequestSummary{ + RequestID: sqid, Queue: queueName, ChangeURIs: []string{uri}, ReceivedAtMs: 910, + Status: entity.RequestStatusAccepting, StatusTimestampMs: 910, Version: 1, + })) + for { + var acked bool + err := s.queueDB.QueryRowContext(s.ctx, `SELECT acked FROM queue_delivery_state + WHERE tenant = ? AND topic = 'log_dlq' AND consumer_group = 'gateway-log-dlq'`, queueName).Scan(&acked) + // Acknowledged delivery state can already have been reclaimed by the poll loop. + if err == sql.ErrNoRows { + break + } + require.NoError(t, err) + if acked { + break + } + <-ticker.C + } + + summary, err := s.client.GetRequestSummaryByID(s.ctx, &pb.GetRequestSummaryByIDRequest{Sqid: sqid, Queue: queueName}) + require.NoError(t, err) + assert.Equal(t, string(entity.RequestStatusLanded), summary.Request.Status) + byURI, err := s.client.GetRequestSummaryByChangeURI(s.ctx, &pb.GetRequestSummaryByChangeURIRequest{ChangeUri: uri, Queue: queueName}) + require.NoError(t, err) + require.Len(t, byURI.Requests, 1) + assert.Equal(t, summary.Request, byURI.Requests[0]) + listed, err := s.client.List(s.ctx, &pb.ListRequest{Queue: queueName, + ReceivedLowerBound: &pb.ListRequest_ReceivedAtOrAfterMs{ReceivedAtOrAfterMs: 910}, + ReceivedUpperBound: &pb.ListRequest_ReceivedBeforeMs{ReceivedBeforeMs: 911}, + }) + require.NoError(t, err) + assert.Equal(t, []*pb.RequestSummary{summary.Request}, listed.Requests) + history, err := s.client.GetRequestHistoryByChangeURI(s.ctx, &pb.GetRequestHistoryByChangeURIRequest{ChangeUri: uri, Queue: queueName}) + require.NoError(t, err) + require.Len(t, history.Histories, 1) + assert.Equal(t, sqid, history.Histories[0].Sqid) + assert.GreaterOrEqual(t, len(history.Histories[0].Events), 2, "replayed audit rows remain in history") + for _, event := range history.Histories[0].Events { + assert.Equal(t, string(entity.RequestStatusLanded), event.Status) + } +} + // TestRequestLogConsumer verifies the gateway's log-topic consumer in isolation: // no orchestrator runs in this stack, so the test itself publishes a request log // entry to the log topic exactly as the orchestrator does in production (via From 2669cfbc769a216b23acae64d58879c19daeb9ef Mon Sep 17 00:00:00 2001 From: sergeyb Date: Fri, 6 Nov 2026 23:11:35 +0000 Subject: [PATCH 2/2] fix(gateway): surface consumer shutdown failures after cancellation Keep expected signal cancellation from masking a failed consumer stop. Preserve joined cleanup errors for listener and server failures, and continue stopping both consumers. Test Plan: Extend TestServeGateway_ConsumerCleanup with cancellation plus a stop failure; verify both consumers stop, the stop error survives, and context.Canceled does not. Revert Plan: Revert this follow-up commit. --- service/submitqueue/gateway/server/log_consumer_test.go | 5 ++--- service/submitqueue/gateway/server/main.go | 7 ++++++- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/service/submitqueue/gateway/server/log_consumer_test.go b/service/submitqueue/gateway/server/log_consumer_test.go index d8f7c2883..ad52e957e 100644 --- a/service/submitqueue/gateway/server/log_consumer_test.go +++ b/service/submitqueue/gateway/server/log_consumer_test.go @@ -145,6 +145,7 @@ func TestServeGateway_ConsumerCleanup(t *testing.T) { {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 @@ -167,9 +168,7 @@ func TestServeGateway_ConsumerCleanup(t *testing.T) { if tt.stopErr != nil { assert.ErrorIs(t, err, stopErr) } - if tt.cancelled { - assert.ErrorIs(t, err, context.Canceled) - } + assert.Equal(t, tt.cancelled && tt.stopErr == nil, errors.Is(err, context.Canceled)) assert.Equal(t, tt.want, calls) }) } diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 4c5b2d411..eca086794 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -378,7 +378,12 @@ func serveGateway(ctx context.Context, cancel context.CancelFunc, grpcServer *gr return fmt.Errorf("failed to start log consumers: %w", err) } defer func() { - runErr = errors.Join(runErr, logConsumers.Stop(context.Background())) + stopErr := logConsumers.Stop(context.Background()) + if stopErr != nil && runErr == context.Canceled { + runErr = stopErr + return + } + runErr = errors.Join(runErr, stopErr) }() // Listen on configurable port