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..ad52e957e --- /dev/null +++ b/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) + }) + } +} diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 27a56497a..eca086794 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,25 @@ 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() { + stopErr := logConsumers.Stop(context.Background()) + if stopErr != nil && runErr == context.Canceled { + runErr = stopErr + return + } + runErr = errors.Join(runErr, stopErr) + }() // Listen on configurable port port := os.Getenv("PORT") @@ -426,6 +396,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 +407,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