diff --git a/doc/rfc/stovepipe/steps/buildsignal.md b/doc/rfc/stovepipe/steps/buildsignal.md index 000454a3..7ce0dfca 100644 --- a/doc/rfc/stovepipe/steps/buildsignal.md +++ b/doc/rfc/stovepipe/steps/buildsignal.md @@ -122,13 +122,24 @@ Per `platform/errs`'s non-retryable-by-default rule (see [platform/errs/README.m | Failure | Disposition | Why | |---|---|---| -| `Status` call | raw error; classifier decides | Deliberately left open rather than fixed either way — runner timeout/connection is transient, "runner not deployed for this queue" is not, and only a backend classifier can tell them apart. | +| `Status` call | raw error; classifier decides | Deliberately left open rather than fixed either way — runner timeout/connection is transient, "runner not deployed for this queue" is not, and only a backend classifier can tell them apart. **This means the `BuildRunner` backend has to classify**: an unclassified transport or HTTP error gets the non-retryable default, so one proxy blip ends the poll chain (see below). | | `Update` CAS conflict (`ErrVersionMismatch`) | declaration-level retryable | A concurrent (redelivered) writer moved the row; reload and re-check converges. | `Build`/`Request` not found (`storage.ErrNotFound`) are **not** in this table: storage is required to be read-after-write consistent (see [storage README](stovepipe/extension/storage/README.md)), so a miss here is already the correct default (non-retryable, straight to DLQ) rather than a departure worth overriding. Everything else — factory lookup, an `Update` store error other than a CAS conflict, and the `record` publish — is returned raw with no override, because the default is already correct: a queue with no registered runner is a config error, and storage/queue failures dead-letter and let DLQ reconciliation recover. The poll loop itself no longer has a publish to fail: holding is a local outcome, and a failed postpone write in the framework lapses into a normal visibility-timeout redelivery, so the loop's liveness never rides on an enqueue succeeding. +### What it costs when a backend does not classify `Status` errors + +Leaving `Status` to the classifier only works if the backend classifies. A `BuildRunner` whose transport returns plain `fmt.Errorf` values gets the non-retryable default, and here that default is expensive: dead-lettering ends the *only* poll chain for a build that is still running, and the request keeps holding one of the queue's `in_flight_count` build slots until reconciliation gives it back. A single `502` from a proxy in front of the build API then looks exactly like "this build can never be polled". + +Two things keep a blip from stalling a queue, and a backend needs both: + +- **The backend classifies its own failures.** Transport errors and 5xx/429/408 responses are `errs.NewRetryableDependencyError`. A 4xx about the request itself — unknown build, forbidden — is `errs.NewDependencyError`. Only the layer that sees the status code can tell these apart, which is why the table above leaves the call to it. +- **The retry budget is worth something.** Retryable means nack, and a nacked message comes back on the next poll, so `Retry.MaxAttempts` counts attempts rather than time — the default three are spent in a few hundred milliseconds. Raising `MaxAttempts` on this subscription buys a little more, but each attempt is another request at a dependency that is already failing, so it does not stretch to cover a proxy restart. Until nacks are spaced by the configured retry backoff, it is the reconciler below rather than the retry budget that keeps a longer outage from costing the queue a slot. + +When the budget does run out the message dead-letters, and the buildsignal DLQ reconciler (`stovepipe/controller/dlq/buildsignal.go`) is what makes that recoverable: it maps the build back to its request, releases the slot, and marks the request `failed`. A deployment that registers the primary consumers but not that reconciler has no fail-closed path for this stage, and loses a slot for good every time this happens. + ## Idempotency Every branch is safe under at-least-once redelivery: diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 3b44500b..524a015c 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -450,6 +450,12 @@ func registerDLQControllers( } 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) + } + count++ + return count, nil } @@ -499,6 +505,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Queue: q, Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-process-dlq"), }, + { + Key: dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), + Name: "buildsignal_dlq", + Queue: q, + Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-buildsignal-dlq"), + }, }) } diff --git a/stovepipe/controller/dlq/BUILD.bazel b/stovepipe/controller/dlq/BUILD.bazel index 4e8f713b..ba4498a4 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 = [ + "buildsignal.go", "dlq.go", "request.go", ], @@ -21,7 +22,10 @@ go_library( go_test( name = "go_default_test", - srcs = ["dlq_test.go"], + srcs = [ + "buildsignal_test.go", + "dlq_test.go", + ], embed = [":go_default_library"], deps = [ "//platform/base/messagequeue:go_default_library", diff --git a/stovepipe/controller/dlq/buildsignal.go b/stovepipe/controller/dlq/buildsignal.go new file mode 100644 index 00000000..5de1ba29 --- /dev/null +++ b/stovepipe/controller/dlq/buildsignal.go @@ -0,0 +1,167 @@ +// 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" + "errors" + "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" +) + +// _buildSignalOpName is the metric operation name shared by every emit in this file. +const _buildSignalOpName = "buildsignal_dlq" + +// BuildSignalController is the DLQ reconciler for the buildsignal stage. The +// payload names a build, not a request, so it takes one more step than the +// process reconciler: read the build to get its RequestID, then fail that +// request via failRequest. +// +// This DLQ is the one that matters most. A request only reaches buildsignal +// after process admitted it, so it holds one of the queue's in_flight_count +// build slots, and buildsignal's terminal path is the only thing that gives that +// slot back. Once a poll message dead-letters — a Status call that stayed broken +// through every retry, an unknown build id, a storage write that kept failing — +// nothing else in the pipeline will look at that build again. Without this +// reconciler the request stays processing for good and the slot is never +// returned, so the queue loses one slot per incident until it has none left and +// stops admitting work. +// +// The Build row keeps whatever non-terminal status the runner last reported. +// There is nothing useful to fix: record decides greenness from Request.State, +// not Build.Status, and writing a terminal status here would claim we saw an +// outcome we never saw. +type BuildSignalController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + topicKey consumer.TopicKey + consumerGroup string +} + +// Verify BuildSignalController implements consumer.Controller at compile time. +var _ consumer.Controller = (*BuildSignalController)(nil) + +// NewBuildSignalController creates a DLQ controller for the buildsignal stage's +// dead-letter topic. topicKey is typically +// dlq.TopicKey(stovepipemq.TopicKeyBuildSignal). +func NewBuildSignalController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + topicKey consumer.TopicKey, + consumerGroup string, +) *BuildSignalController { + return &BuildSignalController{ + logger: logger.Named("buildsignal_dlq_controller"), + metricsScope: scope.SubScope("buildsignal_dlq_controller"), + stores: stores, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process reconciles a single DLQ delivery for the buildsignal 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 *BuildSignalController) Process(ctx context.Context, delivery consumer.Delivery) error { + msg := delivery.Message() + + sig := &stovepipemq.BuildSignal{} + if err := stovepipemq.Unmarshal(msg.Payload, sig); err != nil { + metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "deserialize_errors", 1) + // Retried rather than acked, for the same deployment-skew reason the + // process reconciler gives: a newer producer's payload decodes fine once + // the rollout finishes, and acking here would skip the slot release + // without saying so. + return fmt.Errorf("failed to decode dlq payload: %w", err) + } + if sig.Id == "" { + metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "empty_id_errors", 1) + return fmt.Errorf("dlq payload decoded to empty build id") + } + + store, err := c.stores.For(storage.Config{QueueName: sig.GetQueueName()}) + if err != nil { + metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "storage_resolve_errors", 1) + // Non-retryable: a missing or unresolvable queue is a malformed message. + return fmt.Errorf("failed to resolve storage for queue %q: %w", sig.GetQueueName(), err) + } + + dmeta := delivery.Metadata() + c.logger.Warnw("dlq message received", + "build_id", sig.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"], + ) + + build, err := store.GetBuildStore().Get(ctx, sig.Id) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + // The build row was never written — a crash between Trigger and + // Create. There is no request to recover from this payload; the build + // stage's own DLQ handles the request that triggered it. + c.logger.Warnw("dlq reconcile: build not found, skipping", "build_id", sig.Id) + metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "build_not_found", 1) + return nil + } + metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "build_store_errors", 1) + return fmt.Errorf("failed to get build %s: %w", sig.Id, err) + } + + if build.RequestID == "" { + // Defensive: a build with no request has nothing to reconcile and no slot + // to release. Ack it so the DLQ does not grow forever. + c.logger.Errorw("dlq reconcile: build has empty request id, skipping", "build_id", sig.Id) + metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "build_missing_request", 1) + return nil + } + + // Every request reachable from a build row is either still processing, and holding + // the slot failRequest releases, or already terminal, and past releasing it: build + // triggers only once process has written the strategy, which lands in the same CAS + // as accepted→processing, and processing exits only to a terminal outcome. + if err := failRequest(ctx, store, c.logger, build.RequestID); err != nil { + metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "reconcile_errors", 1) + return err + } + + metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "reconciled", 1) + return nil +} + +// Name returns the controller name for logging and metrics. +func (c *BuildSignalController) Name() string { + return "buildsignal_dlq" +} + +// TopicKey returns the topic key this controller subscribes to. +func (c *BuildSignalController) TopicKey() consumer.TopicKey { + return c.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (c *BuildSignalController) ConsumerGroup() string { + return c.consumerGroup +} diff --git a/stovepipe/controller/dlq/buildsignal_test.go b/stovepipe/controller/dlq/buildsignal_test.go new file mode 100644 index 00000000..1ae08af1 --- /dev/null +++ b/stovepipe/controller/dlq/buildsignal_test.go @@ -0,0 +1,186 @@ +// 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" + "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 testBuildID = "go-code-on-odin-submitqueue/builds/2867068" + +type buildSignalDLQMocks struct { + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore + buildStore *storagemock.MockBuildStore +} + +func newBuildSignalController(t *testing.T, ctrl *gomock.Controller) (*BuildSignalController, buildSignalDLQMocks) { + t.Helper() + + m := buildSignalDLQMocks{ + reqStore: storagemock.NewMockRequestStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + buildStore: storagemock.NewMockBuildStore(ctrl), + } + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() + store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() + store.EXPECT().GetBuildStore().Return(m.buildStore).AnyTimes() + + c := NewBuildSignalController( + zap.NewNop().Sugar(), + tally.NewTestScope("test", nil), + staticStorageFactory{store: store}, + TopicKey(stovepipemq.TopicKeyBuildSignal), + "stovepipe-buildsignal-dlq", + ) + return c, m +} + +func buildSignalPayload(t *testing.T, id string) []byte { + t.Helper() + b, err := stovepipemq.Marshal(&stovepipemq.BuildSignal{Id: id, QueueName: testQueue}) + require.NoError(t, err) + return b +} + +func build() entity.Build { + return entity.Build{ + ID: testBuildID, + RequestID: testID, + Status: entity.BuildStatusRunning, + Version: 3, + } +} + +func TestBuildSignalProcess(t *testing.T) { + tests := []struct { + name string + payload []byte + setup func(m buildSignalDLQMocks) + wantErr bool + }{ + { + // The case this reconciler exists for: a poll message that + // dead-lettered while its request was holding a build slot. + name: "processing request releases the queue slot before marking failed", + setup: func(m buildSignalDLQMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(), nil) + 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.RequestStateFailed + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + }, + }, + { + name: "already terminal request is a no-op", + setup: func(m buildSignalDLQMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) + }, + }, + { + name: "build not found is a no-op", + setup: func(m buildSignalDLQMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{}, storage.ErrNotFound) + }, + }, + { + name: "build store error is returned", + setup: func(m buildSignalDLQMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{}, assert.AnError) + }, + wantErr: true, + }, + { + name: "build without a request id is a no-op", + setup: func(m buildSignalDLQMocks) { + b := build() + b.RequestID = "" + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(b, nil) + }, + }, + { + name: "slot release failure aborts the terminal write", + setup: func(m buildSignalDLQMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{}, assert.AnError) + }, + wantErr: true, + }, + { + name: "malformed payload is returned as an error", + payload: []byte("not-a-proto"), + setup: func(buildSignalDLQMocks) {}, + wantErr: true, + }, + { + name: "empty build id is returned as an error", + payload: buildSignalPayload(t, ""), + setup: func(buildSignalDLQMocks) {}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newBuildSignalController(t, ctrl) + tt.setup(m) + + payload := tt.payload + if payload == nil { + payload = buildSignalPayload(t, testBuildID) + } + + err := c.Process(context.Background(), delivery(t, ctrl, payload)) + + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + }) + } +} + +func TestBuildSignalTopicKey(t *testing.T) { + c, _ := newBuildSignalController(t, gomock.NewController(t)) + + assert.Equal(t, consumer.TopicKey("buildsignal_dlq"), TopicKey(stovepipemq.TopicKeyBuildSignal)) + assert.Equal(t, consumer.TopicKey("buildsignal_dlq"), c.TopicKey()) + assert.Equal(t, "buildsignal_dlq", c.Name()) + assert.Equal(t, "stovepipe-buildsignal-dlq", c.ConsumerGroup()) +} diff --git a/stovepipe/controller/dlq/dlq.go b/stovepipe/controller/dlq/dlq.go index ad1bf54f..73931795 100644 --- a/stovepipe/controller/dlq/dlq.go +++ b/stovepipe/controller/dlq/dlq.go @@ -62,7 +62,19 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey { // 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 and no terminal // outcome has released it yet — the queue's -// slot is released first. Queue and Request are separate entities with no cross-entity +// slot is released first. +// +// Processing is the only non-terminal state that can own a slot, so the condition is not a +// narrowing of some broader set: process claims the slot and CAS-marks accepted→processing, +// releasing its own claim if that CAS never lands, and the exits from processing are the +// terminal outcomes, which release the slot themselves. Widening the release to accepted +// would decrement for the far more common request that never claimed a slot, over-admitting +// against MaxConcurrent. The one case that escapes both this reconciler and process's +// compensation is a hard crash between the two admit writes, which leaves an accepted +// request holding a slot that nothing here can tell apart from a request that never +// claimed one; distinguishing them needs per-request slot ownership on the row. +// +// 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 diff --git a/stovepipe/controller/dlq/dlq_test.go b/stovepipe/controller/dlq/dlq_test.go index 93e1503e..a9971235 100644 --- a/stovepipe/controller/dlq/dlq_test.go +++ b/stovepipe/controller/dlq/dlq_test.go @@ -101,7 +101,9 @@ func TestProcess(t *testing.T) { wantErr bool }{ { - name: "accepted request is marked failed", + // No queue expectations: an accepted request never claimed a slot, + // so releasing one here would over-admit against MaxConcurrent. + name: "accepted request is marked failed without releasing a slot", setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) updated := requestWithState(entity.RequestStateAccepted)