diff --git a/platform/extension/messagequeue/subscription_config.go b/platform/extension/messagequeue/subscription_config.go index a17ab637..fad4a7cb 100644 --- a/platform/extension/messagequeue/subscription_config.go +++ b/platform/extension/messagequeue/subscription_config.go @@ -78,6 +78,23 @@ type DLQConfig struct { TopicSuffix string } +// DLQSubscriptionConfig returns a SubscriptionConfig for consuming a dead-letter +// topic (DLQ reconciliation). It starts from DefaultSubscriptionConfig and applies +// the two overrides every DLQ consumer needs: +// +// - DLQ.Enabled is false, so a reconciliation failure retries in place instead of +// cascading to a second-level "_dlq_dlq" topic that nobody consumes. +// - Retry.MaxAttempts is a very high backstop so the per-message retry budget +// effectively never runs out. This pairs with errs.AlwaysRetryableProcessor +// wired into the DLQ consumer: reconciliation converges eventually instead of +// being silently dropped after the default retry count. +func DLQSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig { + config := DefaultSubscriptionConfig(subscriberName, consumerGroup) + config.DLQ.Enabled = false + config.Retry.MaxAttempts = 1000 + return config +} + // DefaultSubscriptionConfig returns a SubscriptionConfig with sensible defaults. func DefaultSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig { return SubscriptionConfig{ diff --git a/platform/extension/messagequeue/subscription_config_test.go b/platform/extension/messagequeue/subscription_config_test.go index f8513db0..efb63d60 100644 --- a/platform/extension/messagequeue/subscription_config_test.go +++ b/platform/extension/messagequeue/subscription_config_test.go @@ -73,6 +73,18 @@ func TestSubscriptionConfig_CustomValues(t *testing.T) { assert.Equal(t, "_dead", config.DLQ.TopicSuffix) } +func TestDLQSubscriptionConfig(t *testing.T) { + config := DLQSubscriptionConfig("worker-1", "consumer-1-dlq") + + assert.Equal(t, "worker-1", config.SubscriberName) + assert.Equal(t, "consumer-1-dlq", config.ConsumerGroup) + + // The DLQ consumer must not dead-letter its own failures (no "_dlq_dlq" + // cascade) and needs a far larger retry budget than a primary consumer. + assert.False(t, config.DLQ.Enabled) + assert.Greater(t, config.Retry.MaxAttempts, DefaultSubscriptionConfig("worker-1", "consumer-1").Retry.MaxAttempts) +} + func TestSubscriptionConfig_DifferentConsumerGroups(t *testing.T) { // Test that different consumer groups get independent configs tests := []struct { diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 559e3222..fa4612ff 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -15,6 +15,7 @@ go_library( "//platform/extension/messagequeue/mysql:go_default_library", "//service/stovepipe/server/mapper:go_default_library", "//stovepipe/controller:go_default_library", + "//stovepipe/controller/dlq:go_default_library", "//stovepipe/controller/process:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/extension/queueconfig/default:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 3c4b5c4c..7b3aae31 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -37,6 +37,7 @@ import ( queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/service/stovepipe/server/mapper" "github.com/uber/submitqueue/stovepipe/controller" + "github.com/uber/submitqueue/stovepipe/controller/dlq" "github.com/uber/submitqueue/stovepipe/controller/process" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" @@ -208,13 +209,21 @@ func run() error { return fmt.Errorf("failed to create topic registry: %w", err) } - // Consumer running the process stage. + // Two consumers share the topic registry but apply different error classification + // policies. The primary consumer runs the standard classifier walk. The DLQ consumer + // uses AlwaysRetryableProcessor so every non-nil error from a DLQ controller is + // forced retryable — reconciliation must redeliver on any failure because the DLQ + // subscription is a final destination (DLQ.Enabled is false on it, so there is no + // further DLQ to fall back on). primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, errs.NewClassifierProcessor( genericerrs.Classifier, mysqlerrs.Classifier, ), ) + dlqConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer-dlq"), registry, + errs.AlwaysRetryableProcessor, + ) processController := process.NewController( logger.Sugar(), @@ -230,10 +239,23 @@ func run() error { return fmt.Errorf("failed to register process controller: %w", err) } + processDLQController := dlq.NewController(logger.Sugar(), scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") + if err := dlqConsumer.Register(processDLQController); err != nil { + return fmt.Errorf("failed to register process dlq controller: %w", err) + } + + // Start consumers. DLQ first because Start begins processing messages + // immediately; if the primary consumer then fails to start, the half we + // already started is the DLQ side, whose work is idempotent reconciliation + // and is safe to interrupt mid-flight for rollback. + if err := dlqConsumer.Start(ctx); err != nil { + return fmt.Errorf("failed to start dlq consumer: %w", err) + } if err := primaryConsumer.Start(ctx); err != nil { - return fmt.Errorf("failed to start consumer: %w", err) + stopErr := dlqConsumer.Stop(30000) + return errors.Join(fmt.Errorf("failed to start consumer: %w", err), stopErr) } - logger.Info("consumer started") + logger.Info("consumers started") // Create gRPC server grpcServer := grpc.NewServer() @@ -298,9 +320,15 @@ func run() error { serverErr = fmt.Errorf("GRPC server exited with error: %w", serverErr) } - consumerStopErr := primaryConsumer.Stop(30000) + // Stop consumers in reverse start order: primary first, then DLQ. The primary + // pipeline writes the state that DLQ reconciliation reads, so draining primary + // first means in-flight DLQ reconciliation finishes against a settled primary + // rather than racing its shutdown. + primaryStopErr := primaryConsumer.Stop(30000) + dlqStopErr := dlqConsumer.Stop(30000) + consumerStopErr := errors.Join(primaryStopErr, dlqStopErr) if consumerStopErr != nil { - consumerStopErr = fmt.Errorf("failed to stop consumer: %w", consumerStopErr) + consumerStopErr = fmt.Errorf("failed to stop consumers: %w", consumerStopErr) } if consumerStopErr != nil || serverErr != nil { @@ -312,6 +340,8 @@ func run() error { // newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest // publishes to process; process publishes admitted requests to the publish-only build topic. +// The process_dlq topic is the dead-letter destination the queue backend routes to (per +// DefaultSubscriptionConfig's DLQ.TopicSuffix) when the process controller exhausts retries. func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) { return consumer.NewTopicRegistry([]consumer.TopicConfig{ { @@ -327,5 +357,11 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Name: "build", Queue: q, }, + { + Key: dlq.TopicKey(stovepipemq.TopicKeyProcess), + Name: "process_dlq", + Queue: q, + Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-process-dlq"), + }, }) } diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 6ed707ca..4a398add 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -403,27 +403,15 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe subscriberName, t.groupSuffix, ), }) - // DLQ subscription for the same primary stage. DLQ is disabled here - // to avoid a "_dlq_dlq" cascade: if DLQ reconciliation itself fails, - // the consumer retries forever and the failure is surfaced via logs - // and metrics rather than being moved to a second-level dead-letter - // topic that nobody consumes. - // - // MaxAttempts is bumped to a very high value so the per-message - // retry budget effectively never runs out — this pairs with the - // AlwaysRetryableProcessor wired into the DLQ consumer to guarantee - // reconciliation eventually converges instead of being silently - // dropped after the default retry count. - dlqSub := extqueue.DefaultSubscriptionConfig( - subscriberName, t.groupSuffix+"-dlq", - ) - dlqSub.DLQ.Enabled = false - dlqSub.Retry.MaxAttempts = 1000 + // DLQ subscription for the same primary stage. DLQSubscriptionConfig + // disables the subscription's own DLQ (no "_dlq_dlq" cascade) and sets + // an effectively unlimited retry budget to pair with the + // AlwaysRetryableProcessor wired into the DLQ consumer. configs = append(configs, consumer.TopicConfig{ Key: dlq.TopicKey(t.key), Name: t.name + "_dlq", Queue: q, - Subscription: dlqSub, + Subscription: extqueue.DLQSubscriptionConfig(subscriberName, t.groupSuffix+"-dlq"), }) } diff --git a/stovepipe/controller/dlq/BUILD.bazel b/stovepipe/controller/dlq/BUILD.bazel new file mode 100644 index 00000000..22054560 --- /dev/null +++ b/stovepipe/controller/dlq/BUILD.bazel @@ -0,0 +1,40 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "dlq.go", + "request.go", + ], + importpath = "github.com/uber/submitqueue/stovepipe/controller/dlq", + visibility = ["//visibility:public"], + deps = [ + "//platform/consumer:go_default_library", + "//platform/metrics:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["dlq_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/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_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/stovepipe/controller/dlq/dlq.go b/stovepipe/controller/dlq/dlq.go new file mode 100644 index 00000000..4912ea3e --- /dev/null +++ b/stovepipe/controller/dlq/dlq.go @@ -0,0 +1,151 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package dlq contains controllers that consume messages from a pipeline stage's +// dead-letter topic and reconcile the affected request into a terminal state. +// +// Background. The consumer framework moves a message to its DLQ after the controller +// for the original topic returns a non-retryable error or exhausts retries on a +// retryable error. Without DLQ reconciliation the affected request would remain stuck +// in a non-terminal state (accepted, processing) forever — a caller gating deployments +// on greenness would see that indistinguishably from "not yet validated" (see +// doc/rfc/stovepipe/workflow.md#fail-closed-on-unprocessable-work). +// +// Reconciliation strategy. Each DLQ topic carries the same payload as its originating +// topic (the queue framework preserves the bytes verbatim under a new `{topic}_dlq` +// name). The DLQ controller decodes that payload to recover the affected request, then +// transitions it to RequestStateRecordedNotGreen — the conservative not-green verdict +// for gating (see entity.RequestState) — with an idempotent optimistic-locking write so +// concurrent activity (a late successful pipeline transition) wins cleanly. If the request had +// already been admitted (processing) and was holding a concurrency slot, the +// reconciler also releases it by CAS-decrementing the queue's in_flight_count, per +// doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity. +package dlq + +import ( + "context" + "errors" + "fmt" + + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +// topicSuffix is appended to a primary topic key to derive the corresponding DLQ topic +// key. The queue extension's DefaultSubscriptionConfig also uses "_dlq" as the DLQ +// topic suffix; keeping both in sync is intentional so a registered DLQ subscription's +// topic name matches the controller's TopicKey(). +const topicSuffix = "_dlq" + +// TopicKey returns the DLQ topic key for the given primary pipeline topic. It is +// exported so the stovepipe wiring layer can build matching pairs without duplicating +// the suffix literal. +func TopicKey(main consumer.TopicKey) consumer.TopicKey { + return consumer.TopicKey(string(main) + topicSuffix) +} + +// failRequest transitions request to RequestStateRecordedNotGreen if it is not already +// in a terminal state. If the request had reached RequestStateProcessing — meaning process's +// admit step already CAS-incremented the queue's in_flight_count for it — the queue's +// slot is released first. Queue and Request are separate entities with no cross-entity +// transaction, so the two writes cannot be atomic and the ordering picks which crash +// failure mode we accept: a crash between the writes leaves the request non-terminal, +// redelivery re-runs reconciliation, and releaseSlot (which tracks no per-request slot +// ownership) decrements again — transiently over-admitting by one slot until the +// under-count re-converges at releaseSlot's zero clamp. The reverse order would leak +// the slot instead: redelivery skips terminal requests, permanently shrinking the +// queue's capacity toward a wedge. Over-admission is the failure mode we prefer. See +// doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity for the broader +// counter-drift story. +func failRequest(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, requestID string) error { + request, err := store.GetRequestStore().Get(ctx, requestID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + logger.Warnw("dlq reconcile: request not found, skipping", + "request_id", requestID, + ) + return nil + } + return fmt.Errorf("failed to get request %s: %w", requestID, err) + } + + if request.State.IsTerminal() { + logger.Infow("dlq reconcile: request already terminal, skipping", + "request_id", requestID, + "state", string(request.State), + ) + return nil + } + + if request.State == entity.RequestStateProcessing { + if err := releaseSlot(ctx, store, logger, request.Queue); err != nil { + return fmt.Errorf("failed to release queue slot for request %s: %w", requestID, err) + } + } + + updated := request + updated.State = entity.RequestStateRecordedNotGreen + newVersion := request.Version + 1 + if err := store.GetRequestStore().Update(ctx, updated, request.Version, newVersion); err != nil { + return fmt.Errorf("failed to update request %s state to recorded_not_green: %w", requestID, err) + } + logger.Infow("dlq reconcile: request forced terminal not-green", + "request_id", requestID, + "previous_state", string(request.State), + ) + return nil +} + +// releaseSlot CAS-decrements the queue's in_flight_count, retrying on version +// conflicts, mirroring process.Controller's own CAS-retry loop for queue updates. +func releaseSlot(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, queueName string) error { + queueStore := store.GetQueueStore() + + for { + queueRow, err := queueStore.Get(ctx, queueName) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + logger.Warnw("dlq reconcile: queue not found, skipping slot release", + "queue", queueName, + ) + return nil + } + return fmt.Errorf("failed to get queue %s: %w", queueName, err) + } + + if queueRow.InFlightCount <= 0 { + logger.Warnw("dlq reconcile: queue in_flight_count already at zero, skipping slot release", + "queue", queueName, + ) + return nil + } + + updated := queueRow + updated.InFlightCount-- + newVersion := queueRow.Version + 1 + if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + continue + } + return fmt.Errorf("failed to release slot for queue %s: %w", queueName, err) + } + logger.Infow("dlq reconcile: released queue slot", + "queue", queueName, + "in_flight_count", updated.InFlightCount, + ) + return nil + } +} diff --git a/stovepipe/controller/dlq/dlq_test.go b/stovepipe/controller/dlq/dlq_test.go new file mode 100644 index 00000000..7e6ed037 --- /dev/null +++ b/stovepipe/controller/dlq/dlq_test.go @@ -0,0 +1,220 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dlq + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +const ( + testQueue = "monorepo/main" + testID = "request/monorepo/main/7" +) + +type dlqMocks struct { + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore +} + +func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, dlqMocks) { + t.Helper() + + m := dlqMocks{ + reqStore: storagemock.NewMockRequestStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + } + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() + store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() + + c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") + return c, m +} + +func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { + t.Helper() + d := queuemock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, testQueue, nil)).AnyTimes() + d.EXPECT().Attempt().Return(4).AnyTimes() + d.EXPECT().Metadata().Return(map[string]string{ + "dlq.original_topic": "process", + "dlq.failure_count": "3", + "dlq.last_error": "boom", + }).AnyTimes() + return d +} + +func processPayload(t *testing.T, id string) []byte { + t.Helper() + b, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: id}) + require.NoError(t, err) + return b +} + +func requestWithState(state entity.RequestState) entity.Request { + return entity.Request{ + ID: testID, + Queue: testQueue, + State: state, + Version: 2, + } +} + +func TestProcess(t *testing.T) { + tests := []struct { + name string + payload []byte + setup func(m dlqMocks) + wantErr bool + }{ + { + name: "accepted request is marked failed", + setup: func(m dlqMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) + updated := requestWithState(entity.RequestStateAccepted) + updated.State = entity.RequestStateRecordedNotGreen + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + }, + }, + { + name: "processing request releases the queue slot before marking failed", + setup: func(m dlqMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, InFlightCount: 1, Version: 5, + }, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, InFlightCount: 0, Version: 5, + }, int32(5), int32(6)).Return(nil) + updated := requestWithState(entity.RequestStateProcessing) + updated.State = entity.RequestStateRecordedNotGreen + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + }, + }, + { + name: "already superseded is a no-op", + setup: func(m dlqMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSuperseded), nil) + }, + }, + { + name: "already failed is a no-op", + setup: func(m dlqMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateRecordedNotGreen), nil) + }, + }, + { + name: "request not found is a no-op", + setup: func(m dlqMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, storage.ErrNotFound) + }, + }, + { + name: "request update retries on version mismatch", + setup: func(m dlqMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) + updated := requestWithState(entity.RequestStateAccepted) + updated.State = entity.RequestStateRecordedNotGreen + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(storage.ErrVersionMismatch) + }, + wantErr: true, + }, + { + name: "queue update retries on version mismatch then succeeds", + setup: func(m dlqMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, InFlightCount: 1, Version: 5, + }, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, InFlightCount: 0, Version: 5, + }, int32(5), int32(6)).Return(storage.ErrVersionMismatch) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, InFlightCount: 1, Version: 6, + }, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, InFlightCount: 0, Version: 6, + }, int32(6), int32(7)).Return(nil) + updated := requestWithState(entity.RequestStateProcessing) + updated.State = entity.RequestStateRecordedNotGreen + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + }, + }, + { + name: "queue already drained is a no-op for slot release", + setup: func(m dlqMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, InFlightCount: 0, Version: 5, + }, nil) + updated := requestWithState(entity.RequestStateProcessing) + updated.State = entity.RequestStateRecordedNotGreen + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + }, + }, + { + name: "malformed payload is not retryable", + payload: []byte("not-json"), + setup: func(m dlqMocks) {}, + wantErr: true, + }, + { + name: "empty request id is not retryable", + payload: processPayload(t, ""), + setup: func(m dlqMocks) {}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + tt.setup(m) + + payload := tt.payload + if payload == nil { + payload = processPayload(t, testID) + } + + err := c.Process(context.Background(), delivery(t, ctrl, payload)) + + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + }) + } +} + +func TestTopicKey(t *testing.T) { + assert.Equal(t, consumer.TopicKey("process_dlq"), TopicKey(stovepipemq.TopicKeyProcess)) +} diff --git a/stovepipe/controller/dlq/request.go b/stovepipe/controller/dlq/request.go new file mode 100644 index 00000000..8e0b420e --- /dev/null +++ b/stovepipe/controller/dlq/request.go @@ -0,0 +1,125 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dlq + +import ( + "context" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +// Controller is the DLQ reconciler for the process stage. It is registered against the +// process topic's DLQ (see TopicKey) and, on each delivery, decodes the request id from +// the same ProcessRequest payload the primary process controller consumes, then drives +// the referenced request to a terminal failed state via failRequest. +type Controller struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + store storage.Storage + topicKey consumer.TopicKey + consumerGroup string +} + +// Verify Controller implements consumer.Controller at compile time. +var _ consumer.Controller = (*Controller)(nil) + +// _opName is the metric operation name shared by every emit in this file. +const _opName = "process_dlq" + +// NewController creates a new DLQ controller for the process stage's dead-letter topic. +// topicKey is typically dlq.TopicKey(stovepipemq.TopicKeyProcess). +func NewController( + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + topicKey consumer.TopicKey, + consumerGroup string, +) *Controller { + return &Controller{ + logger: logger.Named("process_dlq_controller"), + metricsScope: scope.SubScope("process_dlq_controller"), + store: store, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process reconciles a single DLQ delivery for the process topic. Returns nil to ack +// (success) or an error to nack (retry) — pair this controller only with a consumer +// wired with errs.AlwaysRetryableProcessor so a transient reconcile failure retries +// instead of dead-lettering the DLQ message itself. +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { + op := metrics.Begin(c.metricsScope, _opName) + defer func() { op.Complete(retErr) }() + + msg := delivery.Message() + + pr := &stovepipemq.ProcessRequest{} + if err := stovepipemq.Unmarshal(msg.Payload, pr); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "deserialize_errors", 1) + // Decoding the same bytes normally fails deterministically, but this error is + // still retried: the DLQ consumer's AlwaysRetryableProcessor (see Process doc) + // classifies every error as retryable. That is deliberate — the recoverable + // cause is deployment skew, where a newer producer's payload shape reaches a + // not-yet-upgraded consumer and decodes fine once the rollout completes. A + // genuinely malformed payload exhausts the DLQ subscription's MaxAttempts + // backstop and is dropped by the subscriber with a warning log; acking it here + // instead would skip reconciliation silently and leave the referenced request + // non-terminal. + return fmt.Errorf("failed to decode dlq payload: %w", err) + } + if pr.Id == "" { + metrics.NamedCounter(c.metricsScope, _opName, "empty_id_errors", 1) + return fmt.Errorf("dlq payload decoded to empty request id") + } + + dmeta := delivery.Metadata() + c.logger.Warnw("dlq message received", + "request_id", pr.Id, + "attempt", delivery.Attempt(), + "dlq_original_topic", dmeta["dlq.original_topic"], + "dlq_failure_count", dmeta["dlq.failure_count"], + "dlq_last_error", dmeta["dlq.last_error"], + ) + + if err := failRequest(ctx, c.store, c.logger, pr.Id); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "reconcile_errors", 1) + return err + } + + metrics.NamedCounter(c.metricsScope, _opName, "reconciled", 1) + return nil +} + +// Name returns the controller name for logging and metrics. +func (c *Controller) Name() string { + return "process_dlq" +} + +// TopicKey returns the topic key this controller subscribes to. +func (c *Controller) TopicKey() consumer.TopicKey { + return c.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (c *Controller) ConsumerGroup() string { + return c.consumerGroup +}