From a7075378020e59827a3779a6b0d1eb43b8bc9bdd Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 18 Aug 2026 14:26:33 +0000 Subject: [PATCH 1/3] fix(stovepipe): reconcile build dead letters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Intent: - Prevent build-stage failures from leaving requests processing and consuming queue capacity indefinitely. Changes: - Add a build-stage DLQ controller that resolves the request from the dead-lettered build payload and drives it to a conservative failed outcome. - Register the build DLQ topic, subscription, and controller with the always-retryable reconciliation consumer. - Cover successful reconciliation, malformed payloads, and empty request identifiers. --- Generated by the 🪄 [pr-create](https://sg.uberinternal.com/code.uber.internal/uber-code/devexp-agent-marketplace/-/blob/claude-code/plugins/dev/uber-dev/skills/pr-create/SKILL.md) skill in devexp-agent-marketplace --- service/stovepipe/server/main.go | 12 +++ stovepipe/controller/dlq/BUILD.bazel | 2 + stovepipe/controller/dlq/build.go | 102 +++++++++++++++++++++++++ stovepipe/controller/dlq/build_test.go | 84 ++++++++++++++++++++ 4 files changed, 200 insertions(+) create mode 100644 stovepipe/controller/dlq/build.go create mode 100644 stovepipe/controller/dlq/build_test.go diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 60176d33b..990bd028e 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -456,6 +456,12 @@ func registerDLQControllers( } count++ + buildDLQController := dlq.NewBuildController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq") + if err := c.Register(buildDLQController); err != nil { + return count, fmt.Errorf("failed to register build dlq controller: %w", err) + } + count++ + buildSignalDLQController := dlq.NewBuildSignalController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), "stovepipe-buildsignal-dlq") if err := c.Register(buildSignalDLQController); err != nil { return count, fmt.Errorf("failed to register buildsignal dlq controller: %w", err) @@ -511,6 +517,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Queue: q, Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-process-dlq"), }, + { + Key: dlq.TopicKey(stovepipemq.TopicKeyBuild), + Name: "build_dlq", + Queue: q, + Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-build-dlq"), + }, { Key: dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), Name: "buildsignal_dlq", diff --git a/stovepipe/controller/dlq/BUILD.bazel b/stovepipe/controller/dlq/BUILD.bazel index ba4498a40..d921b5e02 100644 --- a/stovepipe/controller/dlq/BUILD.bazel +++ b/stovepipe/controller/dlq/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "build.go", "buildsignal.go", "dlq.go", "request.go", @@ -23,6 +24,7 @@ go_library( go_test( name = "go_default_test", srcs = [ + "build_test.go", "buildsignal_test.go", "dlq_test.go", ], diff --git a/stovepipe/controller/dlq/build.go b/stovepipe/controller/dlq/build.go new file mode 100644 index 000000000..6f260b80a --- /dev/null +++ b/stovepipe/controller/dlq/build.go @@ -0,0 +1,102 @@ +// 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" +) + +// BuildController reconciles build-stage dead letters by failing the request +// whose build could not be triggered, persisted, or handed to the poll loop. +type BuildController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + topicKey consumer.TopicKey + consumerGroup string +} + +var _ consumer.Controller = (*BuildController)(nil) + +const _buildOpName = "build_dlq" + +// NewBuildController creates a reconciler for the build dead-letter topic. +func NewBuildController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + topicKey consumer.TopicKey, + consumerGroup string, +) *BuildController { + return &BuildController{ + logger: logger.Named("build_dlq_controller"), + metricsScope: scope.SubScope("build_dlq_controller"), + stores: stores, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process drives the request named by a dead-lettered BuildRequest to failed. +func (c *BuildController) Process(ctx context.Context, delivery consumer.Delivery) error { + request := &stovepipemq.BuildRequest{} + if err := stovepipemq.Unmarshal(delivery.Message().Payload, request); err != nil { + metrics.NamedCounter(c.metricsScope, _buildOpName, "deserialize_errors", 1) + return fmt.Errorf("failed to decode build dlq payload: %w", err) + } + if request.Id == "" { + metrics.NamedCounter(c.metricsScope, _buildOpName, "empty_id_errors", 1) + return fmt.Errorf("build dlq payload decoded to empty request id") + } + + store, err := c.stores.For(storage.Config{QueueName: request.GetQueueName()}) + if err != nil { + metrics.NamedCounter(c.metricsScope, _buildOpName, "storage_resolve_errors", 1) + return fmt.Errorf("failed to resolve storage for queue %q: %w", request.GetQueueName(), err) + } + + metadata := delivery.Metadata() + c.logger.Warnw("build dlq message received", + "request_id", request.Id, + "attempt", delivery.Attempt(), + "dlq_original_topic", metadata["dlq.original_topic"], + "dlq_failure_count", metadata["dlq.failure_count"], + "dlq_last_error", metadata["dlq.last_error"], + ) + + if err := failRequest(ctx, store, c.logger, request.Id); err != nil { + metrics.NamedCounter(c.metricsScope, _buildOpName, "reconcile_errors", 1) + return err + } + metrics.NamedCounter(c.metricsScope, _buildOpName, "reconciled", 1) + return nil +} + +// Name returns the controller name. +func (c *BuildController) Name() string { return "build_dlq" } + +// TopicKey returns the dead-letter topic key. +func (c *BuildController) TopicKey() consumer.TopicKey { return c.topicKey } + +// ConsumerGroup returns the offset-tracking group. +func (c *BuildController) ConsumerGroup() string { return c.consumerGroup } diff --git a/stovepipe/controller/dlq/build_test.go b/stovepipe/controller/dlq/build_test.go new file mode 100644 index 000000000..cfd612c43 --- /dev/null +++ b/stovepipe/controller/dlq/build_test.go @@ -0,0 +1,84 @@ +// 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" + "github.com/uber/submitqueue/platform/consumer" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +func buildPayload(t *testing.T, id string) []byte { + t.Helper() + payload, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id, QueueName: testQueue}) + require.NoError(t, err) + return payload +} + +func TestBuildProcess(t *testing.T) { + tests := []struct { + name string + payload []byte + wantErr bool + }{ + {name: "processing request is failed"}, + {name: "malformed payload is returned", payload: []byte("not-a-proto"), wantErr: true}, + {name: "empty request id is returned", payload: buildPayload(t, ""), wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + reqStore := storagemock.NewMockRequestStore(ctrl) + queueStore := storagemock.NewMockQueueStore(ctrl) + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() + store.EXPECT().GetQueueStore().Return(queueStore).AnyTimes() + + controller := NewBuildController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq") + if !tt.wantErr { + reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, InFlightCount: 1, Version: 5}, nil) + queueStore.EXPECT().Update(gomock.Any(), entity.Queue{Name: testQueue, Version: 5}, int32(5), int32(6)).Return(nil) + updated := requestWithState(entity.RequestStateProcessing) + updated.State = entity.RequestStateFailed + reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + } + + payload := tt.payload + if payload == nil { + payload = buildPayload(t, testID) + } + err := controller.Process(context.Background(), delivery(t, ctrl, payload)) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, consumer.TopicKey("build_dlq"), controller.TopicKey()) + assert.Equal(t, "build_dlq", controller.Name()) + assert.Equal(t, "stovepipe-build-dlq", controller.ConsumerGroup()) + }) + } +} From d2373ef0be1ff92ccac942b0483cdc41818a37ec Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Wed, 19 Aug 2026 18:07:16 +0000 Subject: [PATCH 2/3] refactor(stovepipe): follow DLQ controller conventions --- service/stovepipe/server/main.go | 2 +- stovepipe/controller/dlq/build.go | 45 +++++++++++++------------- stovepipe/controller/dlq/build_test.go | 2 +- 3 files changed, 25 insertions(+), 24 deletions(-) diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 990bd028e..2ddb3dbbb 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -456,7 +456,7 @@ func registerDLQControllers( } count++ - buildDLQController := dlq.NewBuildController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq") + buildDLQController := dlq.NewDLQBuildController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq") if err := c.Register(buildDLQController); err != nil { return count, fmt.Errorf("failed to register build dlq controller: %w", err) } diff --git a/stovepipe/controller/dlq/build.go b/stovepipe/controller/dlq/build.go index 6f260b80a..3afed6815 100644 --- a/stovepipe/controller/dlq/build.go +++ b/stovepipe/controller/dlq/build.go @@ -26,9 +26,9 @@ import ( "go.uber.org/zap" ) -// BuildController reconciles build-stage dead letters by failing the request +// buildController reconciles build-stage dead letters by failing the request // whose build could not be triggered, persisted, or handed to the poll loop. -type BuildController struct { +type buildController struct { logger *zap.SugaredLogger metricsScope tally.Scope stores storage.Factory @@ -36,21 +36,22 @@ type BuildController struct { consumerGroup string } -var _ consumer.Controller = (*BuildController)(nil) +var _ consumer.Controller = (*buildController)(nil) const _buildOpName = "build_dlq" -// NewBuildController creates a reconciler for the build dead-letter topic. -func NewBuildController( +// NewDLQBuildController creates a reconciler for the build dead-letter topic. +func NewDLQBuildController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, topicKey consumer.TopicKey, consumerGroup string, -) *BuildController { - return &BuildController{ - logger: logger.Named("build_dlq_controller"), - metricsScope: scope.SubScope("build_dlq_controller"), +) consumer.Controller { + name := string(topicKey) + "_controller" + return &buildController{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), stores: stores, topicKey: topicKey, consumerGroup: consumerGroup, @@ -58,33 +59,33 @@ func NewBuildController( } // Process drives the request named by a dead-lettered BuildRequest to failed. -func (c *BuildController) Process(ctx context.Context, delivery consumer.Delivery) error { - request := &stovepipemq.BuildRequest{} - if err := stovepipemq.Unmarshal(delivery.Message().Payload, request); err != nil { +func (c *buildController) Process(ctx context.Context, delivery consumer.Delivery) error { + buildRequest := &stovepipemq.BuildRequest{} + if err := stovepipemq.Unmarshal(delivery.Message().Payload, buildRequest); err != nil { metrics.NamedCounter(c.metricsScope, _buildOpName, "deserialize_errors", 1) - return fmt.Errorf("failed to decode build dlq payload: %w", err) + return fmt.Errorf("failed to decode dlq payload: %w", err) } - if request.Id == "" { + if buildRequest.Id == "" { metrics.NamedCounter(c.metricsScope, _buildOpName, "empty_id_errors", 1) return fmt.Errorf("build dlq payload decoded to empty request id") } - store, err := c.stores.For(storage.Config{QueueName: request.GetQueueName()}) + store, err := c.stores.For(storage.Config{QueueName: buildRequest.GetQueueName()}) if err != nil { metrics.NamedCounter(c.metricsScope, _buildOpName, "storage_resolve_errors", 1) - return fmt.Errorf("failed to resolve storage for queue %q: %w", request.GetQueueName(), err) + return fmt.Errorf("failed to resolve storage for queue %q: %w", buildRequest.GetQueueName(), err) } metadata := delivery.Metadata() - c.logger.Warnw("build dlq message received", - "request_id", request.Id, + c.logger.Warnw("dlq message received", + "request_id", buildRequest.Id, "attempt", delivery.Attempt(), "dlq_original_topic", metadata["dlq.original_topic"], "dlq_failure_count", metadata["dlq.failure_count"], "dlq_last_error", metadata["dlq.last_error"], ) - if err := failRequest(ctx, store, c.logger, request.Id); err != nil { + if err := failRequest(ctx, store, c.logger, buildRequest.Id); err != nil { metrics.NamedCounter(c.metricsScope, _buildOpName, "reconcile_errors", 1) return err } @@ -93,10 +94,10 @@ func (c *BuildController) Process(ctx context.Context, delivery consumer.Deliver } // Name returns the controller name. -func (c *BuildController) Name() string { return "build_dlq" } +func (c *buildController) Name() string { return string(c.topicKey) } // TopicKey returns the dead-letter topic key. -func (c *BuildController) TopicKey() consumer.TopicKey { return c.topicKey } +func (c *buildController) TopicKey() consumer.TopicKey { return c.topicKey } // ConsumerGroup returns the offset-tracking group. -func (c *BuildController) ConsumerGroup() string { return c.consumerGroup } +func (c *buildController) ConsumerGroup() string { return c.consumerGroup } diff --git a/stovepipe/controller/dlq/build_test.go b/stovepipe/controller/dlq/build_test.go index cfd612c43..c9c48a674 100644 --- a/stovepipe/controller/dlq/build_test.go +++ b/stovepipe/controller/dlq/build_test.go @@ -56,7 +56,7 @@ func TestBuildProcess(t *testing.T) { store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() store.EXPECT().GetQueueStore().Return(queueStore).AnyTimes() - controller := NewBuildController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq") + controller := NewDLQBuildController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq") if !tt.wantErr { reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, InFlightCount: 1, Version: 5}, nil) From 52c119172dee81820524f7b5d2a923ef1ad0c16b Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Wed, 19 Aug 2026 18:16:58 +0000 Subject: [PATCH 3/3] test(stovepipe): align build DLQ fixtures --- stovepipe/controller/dlq/build_test.go | 45 +++++++++++++++++--------- 1 file changed, 30 insertions(+), 15 deletions(-) diff --git a/stovepipe/controller/dlq/build_test.go b/stovepipe/controller/dlq/build_test.go index c9c48a674..a725b521d 100644 --- a/stovepipe/controller/dlq/build_test.go +++ b/stovepipe/controller/dlq/build_test.go @@ -29,6 +29,21 @@ import ( "go.uber.org/zap" ) +func newBuildController(t *testing.T, ctrl *gomock.Controller) (consumer.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 := NewDLQBuildController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq") + return c, m +} + func buildPayload(t *testing.T, id string) []byte { t.Helper() payload, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id, QueueName: testQueue}) @@ -40,9 +55,20 @@ func TestBuildProcess(t *testing.T) { tests := []struct { name string payload []byte + setup func(m dlqMocks) wantErr bool }{ - {name: "processing request is failed"}, + { + name: "processing request is 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, Version: 5}, int32(5), int32(6)).Return(nil) + updated := requestWithState(entity.RequestStateProcessing) + updated.State = entity.RequestStateFailed + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + }, + }, {name: "malformed payload is returned", payload: []byte("not-a-proto"), wantErr: true}, {name: "empty request id is returned", payload: buildPayload(t, ""), wantErr: true}, } @@ -50,20 +76,9 @@ func TestBuildProcess(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { ctrl := gomock.NewController(t) - reqStore := storagemock.NewMockRequestStore(ctrl) - queueStore := storagemock.NewMockQueueStore(ctrl) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() - store.EXPECT().GetQueueStore().Return(queueStore).AnyTimes() - - controller := NewDLQBuildController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq") - if !tt.wantErr { - reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) - queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, InFlightCount: 1, Version: 5}, nil) - queueStore.EXPECT().Update(gomock.Any(), entity.Queue{Name: testQueue, Version: 5}, int32(5), int32(6)).Return(nil) - updated := requestWithState(entity.RequestStateProcessing) - updated.State = entity.RequestStateFailed - reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + controller, mocks := newBuildController(t, ctrl) + if tt.setup != nil { + tt.setup(mocks) } payload := tt.payload