From 5240c412a8b7175ba1c0e823854cbc618df9c36a Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 10 Aug 2026 15:09:23 -0700 Subject: [PATCH] fix(speculate): blame the batch that failed, and stop the queue stranding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Two things go wrong when a speculate message dead-letters, and together they leave a queue quietly stuck. The first is misattribution. The speculate stage takes a message naming one batch but does a job covering the whole queue: a run lists every in-flight batch, hands them all to the Speculator, and commits outcomes for any of them. Only the handful of errors before the run even starts are about the batch on the message. Everything after is about some other batch, or about the queue itself. Yet the stage shared `dlq.NewDLQBatchController` with build, merge, and conclude — which marks *the named* batch `Failed` and errors all its requests. So a Speculator error, or a storage failure writing another head's path set, terminated a batch that was never at fault while the real culprit carried on. The second is that the reconcile restored nothing. `failBatch` published no message at all. Speculation is driven only by messages, and a batch admitted to `Speculating` produces no build to signal and no merge to conclude — so once the message that would have funded it is gone, nothing is left to look at it again. The dead letter consumed the queue's last edge and left none behind, stranding every other admitted batch with no error recorded against it and its requests still reading `batched`. From the outside the queue looked like it was working. ### What? Speculate now says what its failures are about, and its dead letters act on that. Every error return is attributed with `errs.Attribute`: the message's batch for the errors raised before the run, the specific batch for errors raised while looping over the queue, and `{Type: queue}` for listing the queue and for both Speculator calls, where no batch is at fault. Attribution is added through `entity.BatchSubject` / `entity.QueueSubject`, and a counter tagged by subject type makes queue-scoped failures graphable. Retryability is untouched — the classifiers still read the cause underneath. The stage gets its own reconciler, `dlq/speculate.go`, instead of the shared batch one: - Batch subjects are taken at their word and those batches are failed, which may not be the batch on the message. - A queue subject, or no subjects at all, falls back to the message's batch — a guess, but DLQ reconciliation exists so requests cannot sit non-terminal forever, and that guarantee has to hold even when nothing can say which batch was at fault. Which of the three happened is recorded as `dlq.attribution`, so a fallback is never mistaken for a confident answer. - Afterwards it republishes one speculate message naming a still-live batch, restoring the edge it consumed. Two guards make it terminate: it runs only after a reconcile that actually transitioned a batch, so a redelivery cannot loop, and it names a *live* batch rather than the one just failed, so each pass fails one more. A genuinely broken queue therefore drains to empty with a reason recorded against every batch, bounded by the batch count, while a queue whose failure was transient or queue-wide recovers on the next run having lost one batch instead of stranding all of them. Separately, the shared `failRequest` passed `nil` where `RequestLog.Metadata` goes, discarding the failure count, originating topic, and timestamp the reconciler already had in hand and logged. It now carries them, along with the failure's subjects and detail. No schema or proto change is needed: `RequestLog.Metadata` is already persisted and already exposed on the gateway's status and history, so this lands in front of users directly. All five reconcilers benefit. ## Test Plan ✅ `bazel test //...` ✅ `bazel test //test/... --sandbox_writable_path=$HOME/.docker --jobs=1` (11 container suites) ✅ `make lint check-tidy check-gazelle check-mocks` New coverage in `dlq/speculate_test.go`: - Attribution as a table: a batch subject fails *that* batch rather than the message's; a queue subject and a subject-less failure each fall back, and each records which it was. - The re-trigger publishes for a queue that still holds live batches, choosing deterministically (`ListByStates` promises no order). - The loop guard: an already-failed batch publishes nothing even with live batches remaining, which is what stops a permanently failing queue from re-triggering forever. ## Issue Closes CODEM-428 --- submitqueue/entity/BUILD.bazel | 2 + submitqueue/entity/subject.go | 46 +++ .../orchestrator/controller/dlq/BUILD.bazel | 5 + .../orchestrator/controller/dlq/batch.go | 15 +- .../controller/dlq/buildsignal.go | 3 +- .../orchestrator/controller/dlq/dlq.go | 98 +++++- .../orchestrator/controller/dlq/dlq_test.go | 22 +- .../controller/dlq/mergeconflictsignal.go | 3 +- .../controller/dlq/mergesignal.go | 3 +- .../orchestrator/controller/dlq/request.go | 3 +- .../controller/dlq/request_test.go | 8 + .../orchestrator/controller/dlq/speculate.go | 233 ++++++++++++++ .../controller/dlq/speculate_test.go | 284 ++++++++++++++++++ .../controller/speculate/BUILD.bazel | 2 + .../controller/speculate/dispatch.go | 7 +- .../controller/speculate/finalize.go | 11 +- .../orchestrator/controller/speculate/run.go | 21 +- .../controller/speculate/speculate.go | 35 ++- submitqueue/orchestrator/pipeline.go | 2 +- 19 files changed, 754 insertions(+), 49 deletions(-) create mode 100644 submitqueue/entity/subject.go create mode 100644 submitqueue/orchestrator/controller/dlq/speculate.go create mode 100644 submitqueue/orchestrator/controller/dlq/speculate_test.go diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index b7578c1dd..a2dc0a684 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -24,11 +24,13 @@ go_library( "request_log.go", "request_summary.go", "speculation.go", + "subject.go", ], importpath = "github.com/uber/submitqueue/submitqueue/entity", visibility = ["//visibility:public"], deps = [ "//platform/base/change:go_default_library", + "//platform/base/failure:go_default_library", "//platform/base/mergestrategy:go_default_library", ], ) diff --git a/submitqueue/entity/subject.go b/submitqueue/entity/subject.go new file mode 100644 index 000000000..7aea6f602 --- /dev/null +++ b/submitqueue/entity/subject.go @@ -0,0 +1,46 @@ +// 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 entity + +import "github.com/uber/submitqueue/platform/base/failure" + +// Subject types name what a failure is about. The queue layer treats the type +// as opaque; these are SubmitQueue's own vocabulary for it. +const ( + // SubjectTypeBatch identifies a subject by batch ID. + SubjectTypeBatch = "batch" + // SubjectTypeQueue identifies a subject by queue name. Used where no single + // batch is at fault — a failure reading or planning the queue as a whole. + SubjectTypeQueue = "queue" + // SubjectTypeRequest identifies a subject by request ID. + SubjectTypeRequest = "request" +) + +// BatchSubject names a batch as what a failure is about. +func BatchSubject(batchID string) failure.Subject { + return failure.Subject{Type: SubjectTypeBatch, ID: batchID} +} + +// QueueSubject names a queue as what a failure is about. It is the honest +// subject for work that spans the queue — listing it, planning it — where +// blaming any one batch would be a guess. +func QueueSubject(queue string) failure.Subject { + return failure.Subject{Type: SubjectTypeQueue, ID: queue} +} + +// RequestSubject names a request as what a failure is about. +func RequestSubject(requestID string) failure.Subject { + return failure.Subject{Type: SubjectTypeRequest, ID: requestID} +} diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index 85768f22a..c8840e5ff 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "mergeconflictsignal.go", "mergesignal.go", "request.go", + "speculate.go", ], importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/dlq", visibility = ["//visibility:public"], @@ -18,7 +19,9 @@ go_library( "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", "//submitqueue/core/batch:go_default_library", + "//submitqueue/core/publish:go_default_library", "//submitqueue/core/request:go_default_library", + "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", @@ -37,11 +40,13 @@ go_test( "mergesignal_test.go", "publisher_test.go", "request_test.go", + "speculate_test.go", ], embed = [":go_default_library"], deps = [ "//api/runway/messagequeue:go_default_library", "//api/runway/messagequeue/protopb:go_default_library", + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go index 7fe0b5db1..5111926bf 100644 --- a/submitqueue/orchestrator/controller/dlq/batch.go +++ b/submitqueue/orchestrator/controller/dlq/batch.go @@ -27,15 +27,21 @@ import ( ) // batchController is the DLQ reconciler for batch-scoped pipeline stages -// (speculate, build, merge, conclude). All four topics carry a -// BatchID payload, so this controller is registered four times — one per -// topic, each with the matching DLQ topic key and consumer group. +// (build, merge, conclude). All three topics carry a BatchID payload, so this +// controller is registered three times — one per topic, each with the matching +// DLQ topic key and consumer group. // // On each delivery the controller decodes the BatchID, transitions the batch // to BatchStateFailed (idempotent if already halted), and fans out by // transitioning each member request to RequestStateError. The fan-out exists // because conclude — which normally drives request state from batch state — // will not run for a DLQ'd batch. +// +// Blaming the batch on the message is right for these stages because their +// work is that batch: whatever failed, it failed doing this batch's build, +// merge, or conclusion. The speculate stage is not like that — it re-plans a +// whole queue from a message that names one batch — so it has its own +// reconciler; see speculate.go. type batchController struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -92,6 +98,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver return fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err) } + lastError, failureMeta := failureContext(delivery) dmeta := delivery.Metadata() c.logger.Warnw("dlq message received", "batch_id", bid.ID, @@ -101,7 +108,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failBatch(ctx, store, c.registry, c.logger, bid.ID, dmeta["dlq.last_error"]); err != nil { + if _, err := failBatch(ctx, store, c.registry, c.logger, bid.ID, lastError, failureMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go index 07dc69c55..f6b61c44c 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go @@ -90,6 +90,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D return fmt.Errorf("failed to resolve storage for queue %q: %w", buildID.Queue, err) } + lastError, failureMeta := failureContext(delivery) dmeta := delivery.Metadata() c.logger.Warnw("dlq message received", "build_id", buildID.ID, @@ -121,7 +122,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D return nil } - if err := failBatch(ctx, store, c.registry, c.logger, build.BatchID, dmeta["dlq.last_error"]); err != nil { + if _, err := failBatch(ctx, store, c.registry, c.logger, build.BatchID, lastError, failureMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/dlq.go b/submitqueue/orchestrator/controller/dlq/dlq.go index 689360de7..ee9c6b387 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq.go +++ b/submitqueue/orchestrator/controller/dlq/dlq.go @@ -37,6 +37,7 @@ import ( "context" "errors" "fmt" + "strings" "github.com/uber/submitqueue/platform/consumer" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" @@ -62,19 +63,81 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey { return consumer.TopicKey(string(main) + topicSuffix) } +// failureContext reads everything the queue recorded about a dead-lettered +// message: the human-readable reason, and a metadata map to carry alongside it +// on the terminal request log. +// +// The map is what makes a dead letter diagnosable after the fact. The queue +// already hands the reconciler the failure count, the topic it failed on, and +// when — and, when the producer attributed it, which entities it was about. +// All of that used to be logged and then dropped; the request log is where a +// user can actually see it, through the gateway's status and history. +// +// Values are flattened to strings because RequestLog.Metadata and the gateway's +// wire contract are both string maps. Nested detail becomes dotted keys, which +// read well in a display surface where a JSON blob would not. +func failureContext(delivery consumer.Delivery) (string, map[string]string) { + dmeta := delivery.Metadata() + metadata := make(map[string]string, len(dmeta)) + for _, key := range []string{"dlq.original_topic", "dlq.failure_count", "dlq.failed_at"} { + if v, ok := dmeta[key]; ok && v != "" { + metadata[key] = v + } + } + + lastError := dmeta["dlq.last_error"] + + f, failed := delivery.Failure() + if !failed { + return lastError, metadata + } + if f.Message != "" { + lastError = f.Message + } + + // One key per subject type, so several batches at fault read as a list + // rather than overwriting each other. + byType := make(map[string][]string, len(f.Subjects)) + for _, s := range f.Subjects { + byType[s.Type] = append(byType[s.Type], s.ID) + } + for subjectType, ids := range byType { + metadata["dlq.subject."+subjectType] = strings.Join(ids, ",") + } + + flattenDetail("dlq.detail", f.Detail, metadata) + + return lastError, metadata +} + +// flattenDetail writes a JSON-shaped document into a string map, joining nested +// keys with dots. Values are rendered with %v: this is a display surface, not a +// contract, so a readable rendering beats a faithful one. +func flattenDetail(prefix string, detail map[string]any, out map[string]string) { + for k, v := range detail { + key := prefix + "." + k + if nested, ok := v.(map[string]any); ok { + flattenDetail(key, nested, out) + continue + } + out[key] = fmt.Sprintf("%v", v) + } +} + // failRequest transitions a non-terminal request to RequestStateError and // appends the matching RequestStatusError log. Redelivery for an existing Error // state repeats materialization to repair a previous partial attempt. A // different terminal outcome is left unchanged. // lastError is the failure reason preserved by the queue in DLQ delivery -// metadata and is exposed through Status and History for diagnosis. +// metadata and is exposed through Status and History for diagnosis, alongside +// metadata carrying the rest of the failure context. // // A request in RequestStateCancelling is reconciled to RequestStateError, not // left in place: DLQ means the pipeline failed to converge, so we cannot // confirm the cancel completed cleanly. Writing Error is the honest signal and // keeps the request from being stuck in a non-terminal state forever. -func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string) error { - res, err := requestcore.TerminateRequest(ctx, store, registry, requestID, entity.RequestStateError, lastError, nil) +func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string, metadata map[string]string) error { + res, err := requestcore.TerminateRequest(ctx, store, registry, requestID, entity.RequestStateError, lastError, metadata) if err != nil { return fmt.Errorf("dlq reconcile request %s failed: %w", requestID, err) } @@ -112,19 +175,27 @@ func failRequest(ctx context.Context, store storage.Storage, registry consumer.T // Idempotency: an existing Failed batch repeats fan-out because a previous // attempt may have crashed after updating the batch. Succeeded and Cancelled // are different terminal outcomes and do not fan out errors. -// lastError is propagated to each member request's terminal Error log. -func failBatch(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string) error { +// lastError and metadata are propagated to each member request's terminal +// Error log. +// +// It reports whether it transitioned the batch. A caller that only wants to act +// on real progress — republishing to wake the queue, say — can then tell a +// first reconcile from a redelivery of one already done, and avoid doing it +// again forever. +func failBatch(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string, metadata map[string]string) (bool, error) { batch, err := store.GetBatchStore().Get(ctx, batchID) if err != nil { if errors.Is(err, storage.ErrNotFound) { logger.Warnw("dlq reconcile: batch not found, skipping", "batch_id", batchID, ) - return nil + return false, nil } - return fmt.Errorf("failed to get batch %s: %w", batchID, err) + return false, fmt.Errorf("failed to get batch %s: %w", batchID, err) } + transitioned := false + switch batch.State { case entity.BatchStateFailed: logger.Infow("dlq reconcile: batch already failed, repairing request fan-out", @@ -133,21 +204,22 @@ func failBatch(ctx context.Context, store storage.Storage, registry consumer.Top // A prior attempt may have CAS'd to Failed without completing the // membership record move; repair it alongside the fan-out. if err := corebatch.EnsureRecord(ctx, store, batch); err != nil { - return err + return false, err } case entity.BatchStateSucceeded, entity.BatchStateCancelled: logger.Infow("dlq reconcile: batch has a different terminal outcome, skipping", "batch_id", batchID, "state", string(batch.State), ) - return nil + return false, nil default: previousState := batch.State updated, err := corebatch.Transition(ctx, store, batch, entity.BatchStateFailed) if err != nil { - return err + return false, err } batch = updated + transitioned = true logger.Infow("dlq reconcile: batch marked failed", "batch_id", batchID, "previous_state", string(previousState), @@ -155,9 +227,9 @@ func failBatch(ctx context.Context, store storage.Storage, registry consumer.Top } for _, requestID := range batch.Contains { - if err := failRequest(ctx, store, registry, logger, requestID, lastError); err != nil { - return fmt.Errorf("fan-out for batch %s: %w", batchID, err) + if err := failRequest(ctx, store, registry, logger, requestID, lastError, metadata); err != nil { + return transitioned, fmt.Errorf("fan-out for batch %s: %w", batchID, err) } } - return nil + return transitioned, nil } diff --git a/submitqueue/orchestrator/controller/dlq/dlq_test.go b/submitqueue/orchestrator/controller/dlq/dlq_test.go index e673d7cc8..c77a804af 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq_test.go +++ b/submitqueue/orchestrator/controller/dlq/dlq_test.go @@ -90,7 +90,7 @@ func TestFailRequest_TerminalStates(t *testing.T) { }) } - err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "", nil) require.NoError(t, err) }) } @@ -122,7 +122,7 @@ func TestFailRequest_CancellingTransitionsToError(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "", nil) require.NoError(t, err) } @@ -147,7 +147,7 @@ func TestFailRequest_TransitionsToError(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "", nil) require.NoError(t, err) } @@ -172,7 +172,7 @@ func TestFailRequest_LogPublishErrorPropagates(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "", nil) require.Error(t, err) } @@ -186,7 +186,7 @@ func TestFailRequest_NotFoundIsNoOp(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "", nil) require.NoError(t, err) } @@ -200,7 +200,7 @@ func TestFailRequest_GenericGetErrorIsNonRetryable(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "", nil) require.Error(t, err) assert.False(t, errs.IsRetryable(err)) } @@ -239,7 +239,7 @@ func TestFailBatch_TransitionsAndFansOut(t *testing.T) { store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil) require.NoError(t, err) } @@ -269,7 +269,7 @@ func TestFailBatch_FailedFansOutForRepair(t *testing.T) { store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil) require.NoError(t, err) } @@ -286,7 +286,7 @@ func TestFailBatch_DifferentTerminalOutcomeSkipsFanOut(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + _, err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil) require.NoError(t, err) }) } @@ -325,7 +325,7 @@ func TestFailBatch_CancellingTransitionsToFailed(t *testing.T) { store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil) require.NoError(t, err) } @@ -339,7 +339,7 @@ func TestFailBatch_NotFoundIsNoOp(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + _, err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil) require.NoError(t, err) } diff --git a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go index e57f14311..c51561288 100644 --- a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go @@ -82,6 +82,7 @@ func (c *mergeConflictSignalController) Process(ctx context.Context, delivery co return fmt.Errorf("failed to resolve storage for queue %q: %w", result.GetQueueName(), err) } + lastError, failureMeta := failureContext(delivery) dmeta := delivery.Metadata() c.logger.Warnw("dlq message received", "request_id", result.Id, @@ -91,7 +92,7 @@ func (c *mergeConflictSignalController) Process(ctx context.Context, delivery co "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failRequest(ctx, store, c.registry, c.logger, result.Id, dmeta["dlq.last_error"]); err != nil { + if err := failRequest(ctx, store, c.registry, c.logger, result.Id, lastError, failureMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/mergesignal.go b/submitqueue/orchestrator/controller/dlq/mergesignal.go index 4243df8bb..8eb3d9e4f 100644 --- a/submitqueue/orchestrator/controller/dlq/mergesignal.go +++ b/submitqueue/orchestrator/controller/dlq/mergesignal.go @@ -81,6 +81,7 @@ func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.D return fmt.Errorf("failed to resolve storage for queue %q: %w", result.GetQueueName(), err) } + lastError, failureMeta := failureContext(delivery) dmeta := delivery.Metadata() c.logger.Warnw("dlq message received", "batch_id", result.Id, @@ -90,7 +91,7 @@ func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.D "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failBatch(ctx, store, c.registry, c.logger, result.Id, dmeta["dlq.last_error"]); err != nil { + if _, err := failBatch(ctx, store, c.registry, c.logger, result.Id, lastError, failureMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go index f5fafec9c..46b68743b 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -129,6 +129,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv return fmt.Errorf("failed to resolve storage for queue %q: %w", rid.Queue, err) } + lastError, failureMeta := failureContext(delivery) dmeta := delivery.Metadata() c.logger.Warnw("dlq message received", "request_id", rid.ID, @@ -138,7 +139,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failRequest(ctx, store, c.registry, c.logger, rid.ID, dmeta["dlq.last_error"]); err != nil { + if err := failRequest(ctx, store, c.registry, c.logger, rid.ID, lastError, failureMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go index 4b317d972..5f88c6bcc 100644 --- a/submitqueue/orchestrator/controller/dlq/request_test.go +++ b/submitqueue/orchestrator/controller/dlq/request_test.go @@ -20,6 +20,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/platform/base/failure" queue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" consumermock "github.com/uber/submitqueue/platform/consumer/mock" @@ -179,6 +180,12 @@ func TestDLQRequestController_Process_EmptyIDFails(t *testing.T) { // newMockDelivery returns a MockDelivery wired up enough to be passed through // the DLQ controller Process flow. func newMockDelivery(ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery { + return newMockDeliveryWithFailure(ctrl, payload, failure.Failure{}, false) +} + +// newMockDeliveryWithFailure builds a delivery that also reports a recorded +// failure, as a redelivery from a DLQ topic does. +func newMockDeliveryWithFailure(ctrl *gomock.Controller, payload []byte, f failure.Failure, failed bool) *consumermock.MockDelivery { msg := queue.NewMessage("dlq-msg-1", payload, "", nil) d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() @@ -188,5 +195,6 @@ func newMockDelivery(ctrl *gomock.Controller, payload []byte) *consumermock.Mock "dlq.failure_count": "3", "dlq.last_error": "boom", }).AnyTimes() + d.EXPECT().Failure().Return(f, failed).AnyTimes() return d } diff --git a/submitqueue/orchestrator/controller/dlq/speculate.go b/submitqueue/orchestrator/controller/dlq/speculate.go new file mode 100644 index 000000000..3c41c6b5f --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/speculate.go @@ -0,0 +1,233 @@ +// 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" + corebatch "github.com/uber/submitqueue/submitqueue/core/batch" + "github.com/uber/submitqueue/submitqueue/core/publish" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + "go.uber.org/zap" +) + +// speculateController is the DLQ reconciler for the speculate topic. +// +// Speculate does not share the generic batch reconciler because it is not a +// batch-scoped stage. Its message names one batch, but a run re-plans the +// entire queue: it lists every in-flight batch, hands them all to the +// Speculator, and commits outcomes for any of them. A failure is therefore +// often about a different batch than the message names, or about the queue +// itself — and failing the named batch regardless would terminate one that was +// never at fault while leaving the real culprit running. +// +// So this reconciler reads the failure's subjects and acts on those. It also +// republishes to speculate afterwards, because a dead letter here consumes an +// edge the queue needed. Speculation is driven only by messages; a batch +// admitted to Speculating produces no build to signal and no merge to conclude, +// so once the message that would have funded it is gone, nothing is left to +// look at it again. Without the republish the failure of one batch silently +// strands every other batch in the queue. +type speculateController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + registry consumer.TopicRegistry + topicKey consumer.TopicKey + consumerGroup string +} + +// Verify speculateController implements consumer.Controller at compile time. +var _ consumer.Controller = (*speculateController)(nil) + +// NewDLQSpeculateController builds the DLQ controller for the speculate topic. +// topicKey must be the DLQ topic key (typically TopicKey(primary)). +func NewDLQSpeculateController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + registry consumer.TopicRegistry, + topicKey consumer.TopicKey, + consumerGroup string, +) consumer.Controller { + name := string(topicKey) + "_controller" + return &speculateController{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + stores: stores, + registry: registry, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process reconciles a single dead-lettered speculate message. +func (c *speculateController) Process(ctx context.Context, delivery consumer.Delivery) error { + const opName = "process" + + msg := delivery.Message() + + bid, err := entity.BatchIDFromBytes(msg.Payload) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "deserialize_errors", 1) + return fmt.Errorf("failed to decode batch id from dlq payload: %w", err) + } + if bid.ID == "" { + metrics.NamedCounter(c.metricsScope, opName, "empty_id_errors", 1) + return fmt.Errorf("dlq payload decoded to empty batch id") + } + + store, err := c.stores.For(storage.Config{QueueName: bid.Queue}) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "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", bid.Queue, err) + } + + lastError, failureMeta := failureContext(delivery) + culprits, attribution := c.blame(delivery, bid.ID) + failureMeta["dlq.attribution"] = attribution + + c.logger.Warnw("dlq message received", + "batch_id", bid.ID, + "queue", bid.Queue, + "attempt", delivery.Attempt(), + "attribution", attribution, + "culprits", culprits, + "dlq_last_error", lastError, + ) + metrics.NamedCounter(c.metricsScope, opName, "reconciled", 1, + metrics.NewTag("attribution", attribution)) + + progressed := false + for _, batchID := range culprits { + transitioned, err := failBatch(ctx, store, c.registry, c.logger, batchID, lastError, failureMeta) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) + return err + } + progressed = progressed || transitioned + } + + if !progressed { + // A redelivery of a reconcile already done. Republishing here would + // hand the queue the same message again on every redelivery, forever. + return nil + } + + return c.retrigger(ctx, store, bid.Queue) +} + +// blame decides which batches to fail, and reports how that was decided so the +// answer is visible on the request afterwards rather than inferred. +// +// A failure that names batches is taken at its word — those are the batches the +// run was actually working on when it failed. Anything else falls back to the +// batch on the message: that is a guess, but DLQ reconciliation exists so +// requests cannot sit non-terminal forever, and that guarantee has to hold even +// when nothing can say which batch was at fault. +func (c *speculateController) blame(delivery consumer.Delivery, payloadBatchID string) ([]string, string) { + f, failed := delivery.Failure() + if !failed { + return []string{payloadBatchID}, "unattributed" + } + + if batches := f.IDsOfType(entity.SubjectTypeBatch); len(batches) > 0 { + return batches, "batch" + } + + // A queue subject says no batch was at fault — a failure listing or + // planning the queue. There is still no other way to release the message's + // requests, so the named batch is failed, labelled for what it is. + if len(f.IDsOfType(entity.SubjectTypeQueue)) > 0 { + return []string{payloadBatchID}, "queue" + } + + return []string{payloadBatchID}, "unattributed" +} + +// retrigger wakes the queue if it still holds live batches, restoring the edge +// this dead letter consumed. +// +// It names a batch that is still live rather than the one just failed, and runs +// only after a reconcile that actually transitioned something. Together those +// make it terminate: each pass fails one more batch and hands the queue another +// run, so a queue that is genuinely broken drains to empty — every batch +// recorded with a reason — instead of stranding, while a queue whose failure +// was transient or queue-wide simply recovers on the next run. +func (c *speculateController) retrigger(ctx context.Context, store storage.Storage, queue string) error { + const opName = "process" + + live, err := corebatch.ListByStates(ctx, store, entity.ActiveBatchStates()) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to list live batches of queue %s: %w", queue, err) + } + if len(live) == 0 { + return nil + } + + // ListByStates does not promise an order, so pick deterministically. Any + // live batch would wake the queue — the run re-plans all of it — but a + // stable choice keeps the behaviour reproducible. + next := live[0].ID + for _, b := range live[1:] { + if b.ID < next { + next = b.ID + } + } + + payload, err := entity.BatchID{ID: next, Queue: queue}.ToBytes() + if err != nil { + return fmt.Errorf("failed to serialize batch ID: %w", err) + } + + // A distinct message ID every time: the queue deduplicates on + // (topic, partition, ID) against rows it has not collected yet, so reusing + // the batch ID would make this wake-up a silent no-op. + if err := publish.Message(ctx, c.registry, topickey.TopicKeySpeculate, publish.UniqueID(next), payload, queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to re-trigger speculation for queue %s: %w", queue, err) + } + + metrics.NamedCounter(c.metricsScope, opName, "retriggered", 1) + c.logger.Infow("re-triggered speculation after dlq reconcile", + "queue", queue, + "batch_id", next, + "live_batches", len(live), + ) + return nil +} + +// Name returns the controller name for logging and metrics. +func (c *speculateController) Name() string { + return string(c.topicKey) +} + +// TopicKey returns the topic key this controller subscribes to. +func (c *speculateController) TopicKey() consumer.TopicKey { + return c.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (c *speculateController) ConsumerGroup() string { + return c.consumerGroup +} diff --git a/submitqueue/orchestrator/controller/dlq/speculate_test.go b/submitqueue/orchestrator/controller/dlq/speculate_test.go new file mode 100644 index 000000000..3cb627fbe --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/speculate_test.go @@ -0,0 +1,284 @@ +// 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/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap/zaptest" +) + +func newSpeculateController(registry consumer.TopicRegistry, store *storagemock.MockStorage, t *testing.T) consumer.Controller { + return NewDLQSpeculateController( + zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store: store}, + registry, TopicKey(topickey.TopicKeySpeculate), "orchestrator-speculate-dlq", + ) +} + +// speculateRegistry captures the batch IDs republished to the speculate topic, +// and counts log publishes so the fan-out stays exercised. +func speculateRegistry(t *testing.T, ctrl *gomock.Controller, logPublishes int, captured *[]string, logs *[]entity.RequestLog) consumer.TopicRegistry { + logPublisher := queuemock.NewMockPublisher(ctrl) + logPublisher.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, message entityqueue.Message) error { + entry, err := entity.RequestLogFromBytes(message.Payload) + require.NoError(t, err) + if logs != nil { + *logs = append(*logs, entry) + } + return nil + }, + ).Times(logPublishes) + logQueue := queuemock.NewMockQueue(ctrl) + logQueue.EXPECT().Publisher().Return(logPublisher).Times(logPublishes) + + specPublisher := queuemock.NewMockPublisher(ctrl) + specPublisher.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, message entityqueue.Message) error { + bid, err := entity.BatchIDFromBytes(message.Payload) + require.NoError(t, err) + *captured = append(*captured, bid.ID) + return nil + }, + ).AnyTimes() + specQueue := queuemock.NewMockQueue(ctrl) + specQueue.EXPECT().Publisher().Return(specPublisher).AnyTimes() + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyLog, Name: "log", Queue: logQueue}, + {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: specQueue}, + }) + require.NoError(t, err) + return registry +} + +// A speculate run covers the whole queue, so the batch that failed is often not +// the batch the message names. The subjects on the failure are what say which, +// and failing the named one regardless would terminate an innocent batch while +// leaving the real culprit running. +func TestDLQSpeculateController_Process_Attribution(t *testing.T) { + tests := []struct { + name string + recordedFailure failure.Failure + failed bool + wantFailedBatch string + wantAttribution string + }{ + { + name: "batch subject blames that batch, not the message's", + recordedFailure: failure.New("read failed", entity.BatchSubject("q/batch/other")), + failed: true, + wantFailedBatch: "q/batch/other", + wantAttribution: "batch", + }, + { + name: "queue subject falls back to the message's batch", + recordedFailure: failure.New("speculator failed", entity.QueueSubject("q")), + failed: true, + wantFailedBatch: "q/batch/named", + wantAttribution: "queue", + }, + { + name: "no subjects falls back and says so", + recordedFailure: failure.New("exceeded retry limit"), + failed: true, + wantFailedBatch: "q/batch/named", + wantAttribution: "unattributed", + }, + { + name: "no recorded failure at all falls back", + wantFailedBatch: "q/batch/named", + wantAttribution: "unattributed", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + + blamed := entity.Batch{ + ID: tt.wantFailedBatch, Queue: "q", Contains: []string{"q/1"}, + State: entity.BatchStateSpeculating, Version: 2, + } + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), tt.wantFailedBatch).Return(blamed, nil) + batchStore.EXPECT().Update(gomock.Any(), batchWithState(blamed, entity.BatchStateFailed), int32(2), int32(3)).Return(nil) + + request := entity.Request{ID: "q/1", Version: 1, State: entity.RequestStateProcessing} + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), "q/1").Return(request, nil) + requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil) + + // The queue drained with the blamed batch, so no re-trigger. + queueBatchState := storagemock.NewMockQueueBatchStateStore(ctrl) + queueBatchState.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + queueBatchState.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + queueBatchState.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetQueueBatchStateStore().Return(queueBatchState).AnyTimes() + + var republished []string + var logs []entity.RequestLog + registry := speculateRegistry(t, ctrl, 1, &republished, &logs) + c := newSpeculateController(registry, store, t) + + payload, err := entity.BatchID{ID: "q/batch/named", Queue: "q"}.ToBytes() + require.NoError(t, err) + + delivery := newMockDeliveryWithFailure(ctrl, payload, tt.recordedFailure, tt.failed) + require.NoError(t, c.Process(context.Background(), delivery)) + + // The attribution is recorded where a user can see it, so a + // fallback is never mistaken for a confident answer. + require.Len(t, logs, 1) + assert.Equal(t, tt.wantAttribution, logs[0].Metadata["dlq.attribution"]) + assert.Equal(t, "validate", logs[0].Metadata["dlq.original_topic"]) + }) + } +} + +// The dead letter that consumed the queue's last edge has to leave one behind, +// or every other batch admitted to speculating sits there with nothing to drive +// it — the stranding this reconciler exists to stop. +func TestDLQSpeculateController_Process_RetriggersQueue(t *testing.T) { + ctrl := gomock.NewController(t) + + failedBatch := entity.Batch{ + ID: "q/batch/1", Queue: "q", Contains: nil, + State: entity.BatchStateSpeculating, Version: 1, + } + // Two still live afterwards; the lower ID is chosen so the wake-up is + // reproducible despite ListByStates promising no order. + liveA := entity.Batch{ID: "q/batch/2", Queue: "q", State: entity.BatchStateSpeculating, Version: 1} + liveB := entity.Batch{ID: "q/batch/3", Queue: "q", State: entity.BatchStateSpeculating, Version: 1} + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), "q/batch/1").Return(failedBatch, nil) + batchStore.EXPECT().Update(gomock.Any(), batchWithState(failedBatch, entity.BatchStateFailed), int32(1), int32(2)).Return(nil) + batchStore.EXPECT().Get(gomock.Any(), "q/batch/3").Return(liveB, nil).AnyTimes() + batchStore.EXPECT().Get(gomock.Any(), "q/batch/2").Return(liveA, nil).AnyTimes() + + queueBatchState := storagemock.NewMockQueueBatchStateStore(ctrl) + queueBatchState.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + queueBatchState.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + queueBatchState.EXPECT().List(gomock.Any(), entity.BatchStateSpeculating).Return([]entity.QueueBatchState{ + {Queue: "q", State: entity.BatchStateSpeculating, BatchID: "q/batch/3"}, + {Queue: "q", State: entity.BatchStateSpeculating, BatchID: "q/batch/2"}, + }, nil).AnyTimes() + queueBatchState.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetQueueBatchStateStore().Return(queueBatchState).AnyTimes() + + var republished []string + registry := speculateRegistry(t, ctrl, 0, &republished, nil) + c := newSpeculateController(registry, store, t) + + payload, err := entity.BatchID{ID: "q/batch/1", Queue: "q"}.ToBytes() + require.NoError(t, err) + + delivery := newMockDeliveryWithFailure(ctrl, payload, failure.New("boom"), true) + require.NoError(t, c.Process(context.Background(), delivery)) + + assert.Equal(t, []string{"q/batch/2"}, republished) +} + +// Redelivery of a reconcile already done must publish nothing. Republishing +// unconditionally would hand the queue a fresh message every time this message +// came back, and a permanently failing queue would never stop. +func TestDLQSpeculateController_Process_NoRetriggerWithoutProgress(t *testing.T) { + ctrl := gomock.NewController(t) + + already := entity.Batch{ + ID: "q/batch/1", Queue: "q", Contains: nil, + State: entity.BatchStateFailed, Version: 4, + } + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), "q/batch/1").Return(already, nil) + + queueBatchState := storagemock.NewMockQueueBatchStateStore(ctrl) + queueBatchState.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + queueBatchState.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + // Live batches remain, so only the progress guard can be what stops it. + queueBatchState.EXPECT().List(gomock.Any(), entity.BatchStateSpeculating).Return([]entity.QueueBatchState{ + {Queue: "q", State: entity.BatchStateSpeculating, BatchID: "q/batch/2"}, + }, nil).AnyTimes() + queueBatchState.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetQueueBatchStateStore().Return(queueBatchState).AnyTimes() + + var republished []string + registry := speculateRegistry(t, ctrl, 0, &republished, nil) + c := newSpeculateController(registry, store, t) + + payload, err := entity.BatchID{ID: "q/batch/1", Queue: "q"}.ToBytes() + require.NoError(t, err) + + delivery := newMockDeliveryWithFailure(ctrl, payload, failure.New("boom"), true) + require.NoError(t, c.Process(context.Background(), delivery)) + + assert.Empty(t, republished, "an already-failed batch is not progress") +} + +func TestDLQSpeculateController_InterfaceAndAccessors(t *testing.T) { + ctrl := gomock.NewController(t) + store := storagemock.NewMockStorage(ctrl) + + c := newSpeculateController(consumer.TopicRegistry{}, store, t) + + assert.Equal(t, "speculate_dlq", c.Name()) + assert.Equal(t, consumer.TopicKey("speculate_dlq"), c.TopicKey()) + assert.Equal(t, "orchestrator-speculate-dlq", c.ConsumerGroup()) +} + +func TestDLQSpeculateController_Process_MalformedPayloadFails(t *testing.T) { + ctrl := gomock.NewController(t) + store := storagemock.NewMockStorage(ctrl) + + c := newSpeculateController(consumer.TopicRegistry{}, store, t) + + delivery := newMockDelivery(ctrl, []byte("garbage")) + require.Error(t, c.Process(context.Background(), delivery)) +} + +func TestDLQSpeculateController_Process_EmptyIDFails(t *testing.T) { + ctrl := gomock.NewController(t) + store := storagemock.NewMockStorage(ctrl) + + c := newSpeculateController(consumer.TopicRegistry{}, store, t) + + payload, err := entity.BatchID{ID: ""}.ToBytes() + require.NoError(t, err) + + delivery := newMockDelivery(ctrl, payload) + require.Error(t, c.Process(context.Background(), delivery)) +} diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel index f4eafb864..fdcf53146 100644 --- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel @@ -15,7 +15,9 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate", visibility = ["//visibility:public"], deps = [ + "//platform/base/failure:go_default_library", "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", "//platform/metrics:go_default_library", "//submitqueue/core/batch:go_default_library", "//submitqueue/core/publish:go_default_library", diff --git a/submitqueue/orchestrator/controller/speculate/dispatch.go b/submitqueue/orchestrator/controller/speculate/dispatch.go index 24f991058..1732eca59 100644 --- a/submitqueue/orchestrator/controller/speculate/dispatch.go +++ b/submitqueue/orchestrator/controller/speculate/dispatch.go @@ -87,7 +87,9 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot, ) continue } - return err + // The loop covers every in-flight batch, so the head whose + // write failed is usually not the one named on the message. + return c.attributed(err, entity.BatchSubject(batch.ID)) } } @@ -100,7 +102,8 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot, if hasActionablePaths(set) { if err := c.publishBatchID(ctx, topickey.TopicKeyBuild, batch.ID, queue, batch.ID); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish batch %s to build: %w", batch.ID, err) + return c.attributed(fmt.Errorf("failed to publish batch %s to build: %w", batch.ID, err), + entity.BatchSubject(batch.ID)) } } } diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index 668f7f0de..206be5455 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -197,13 +197,20 @@ func (c *Controller) commitOutcome(ctx context.Context, snap *snapshot, batch en snap.markClean(batch.ID) return false, nil } - return false, err + return false, c.attributed(err, entity.BatchSubject(batch.ID)) } snap.pathSets[batch.ID] = stored snap.markClean(batch.ID) } - return c.applyOutcome(ctx, snap.store, batch, decision, snap.isTrigger(batch.ID)) + // Attributed here rather than at each leaf: this is the one place that + // knows which batch the whole commit is for, and the finalize loop runs it + // over every decided batch in the queue — rarely the one on the message. + landed, err := c.applyOutcome(ctx, snap.store, batch, decision, snap.isTrigger(batch.ID)) + if err != nil { + return landed, c.attributed(err, entity.BatchSubject(batch.ID)) + } + return landed, nil } // recordOutcome writes an outcome's terminal state back into the snapshot so diff --git a/submitqueue/orchestrator/controller/speculate/run.go b/submitqueue/orchestrator/controller/speculate/run.go index 044df5455..b5c108998 100644 --- a/submitqueue/orchestrator/controller/speculate/run.go +++ b/submitqueue/orchestrator/controller/speculate/run.go @@ -77,7 +77,8 @@ func (c *Controller) read(ctx context.Context, store storage.Storage, queue stri inFlight, err := corebatch.ListByStates(ctx, store, entity.ActiveBatchStates()) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return snapshot{}, fmt.Errorf("failed to list in-flight batches of queue %s: %w", queue, err) + return snapshot{}, c.attributed(fmt.Errorf("failed to list in-flight batches of queue %s: %w", queue, err), + entity.QueueSubject(queue)) } snap := snapshot{ @@ -106,7 +107,8 @@ func (c *Controller) read(ctx context.Context, store storage.Storage, queue stri dep, err := store.GetBatchStore().Get(ctx, depID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return snapshot{}, fmt.Errorf("failed to get dependency batch %s of %s: %w", depID, batch.ID, err) + return snapshot{}, c.attributed(fmt.Errorf("failed to get dependency batch %s of %s: %w", depID, batch.ID, err), + entity.BatchSubject(depID)) } snap.batches[depID] = dep } @@ -120,7 +122,8 @@ func (c *Controller) read(ctx context.Context, store storage.Storage, queue stri continue } metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return snapshot{}, fmt.Errorf("failed to get path set for batch %s: %w", batch.ID, err) + return snapshot{}, c.attributed(fmt.Errorf("failed to get path set for batch %s: %w", batch.ID, err), + entity.BatchSubject(batch.ID)) } changed, err := c.updatePathsFromBuilds(ctx, store, &set) if err != nil { @@ -180,7 +183,8 @@ func (c *Controller) updatePathsFromBuilds(ctx context.Context, store storage.St } if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return false, fmt.Errorf("failed to look up build for path %s attempt %d: %w", entry.ID, entry.Attempt, err) + return false, c.attributed(fmt.Errorf("failed to look up build for path %s attempt %d: %w", entry.ID, entry.Attempt, err), + entity.BatchSubject(set.Head)) } build, err := store.GetBuildStore().Get(ctx, link.BuildID) @@ -193,7 +197,8 @@ func (c *Controller) updatePathsFromBuilds(ctx context.Context, store storage.St continue } metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return false, fmt.Errorf("failed to get build %s for path %s: %w", link.BuildID, entry.ID, err) + return false, c.attributed(fmt.Errorf("failed to get build %s for path %s: %w", link.BuildID, entry.ID, err), + entity.BatchSubject(set.Head)) } if !build.Status.IsTerminal() { @@ -260,7 +265,8 @@ func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]en spec, err := c.speculators.For(speculator.Config{QueueName: queue}) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "speculator_errors", 1) - return nil, fmt.Errorf("failed to build speculator for queue %s: %w", queue, err) + return nil, c.attributed(fmt.Errorf("failed to build speculator for queue %s: %w", queue, err), + entity.QueueSubject(queue)) } sets := make([]entity.SpeculationPathSet, 0, len(snap.pathSets)) @@ -273,7 +279,8 @@ func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]en proposals, err := spec.Speculate(ctx, slices.Collect(maps.Values(snap.batches)), sets) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "speculator_errors", 1) - return nil, fmt.Errorf("speculator failed for queue %s: %w", queue, err) + return nil, c.attributed(fmt.Errorf("speculator failed for queue %s: %w", queue, err), + entity.QueueSubject(queue)) } return proposals, nil } diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index a9720d952..95540770c 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -19,7 +19,9 @@ import ( "fmt" "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/failure" "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" "github.com/uber/submitqueue/submitqueue/core/publish" @@ -101,20 +103,23 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er if err != nil { metrics.NamedCounter(c.metricsScope, opName, "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", bid.Queue, err) + return c.attributed(fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err), + entity.QueueSubject(bid.Queue)) } batch, err := store.GetBatchStore().Get(ctx, bid.ID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to get batch %s: %w", bid.ID, err) + return c.attributed(fmt.Errorf("failed to get batch %s: %w", bid.ID, err), + entity.BatchSubject(bid.ID)) } // The payload's queue must match the batch's authoritative queue; a // mismatch is a malformed message. Non-retryable — reject to the DLQ. if bid.Queue != "" && bid.Queue != batch.Queue { metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1) - return fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID) + return c.attributed(fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID), + entity.BatchSubject(batch.ID)) } if batch.State.IsTerminal() { @@ -145,7 +150,7 @@ func (c *Controller) admit(ctx context.Context, store storage.Storage, batch ent updated, err := corebatch.Transition(ctx, store, batch, entity.BatchStateSpeculating) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return batch, err + return batch, c.attributed(err, entity.BatchSubject(batch.ID)) } metrics.NamedCounter(c.metricsScope, opName, "admitted", 1) @@ -163,7 +168,8 @@ func (c *Controller) admit(ctx context.Context, store storage.Storage, batch ent func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batchID, queue, queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to conclude: %w", err) + return c.attributed(fmt.Errorf("failed to publish to conclude: %w", err), + entity.BatchSubject(batchID)) } return nil } @@ -190,6 +196,25 @@ func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, partitionKey) } +// attributed records what a failure was about and counts it by subject type. +// +// It exists because this stage's message names one batch but its work covers +// the whole queue: a run reads every in-flight batch, hands them all to the +// Speculator, and commits outcomes for any of them. So most failures here are +// not about the batch on the message — they are about some other batch, or +// about the queue itself — and a reconciler that assumed otherwise would +// terminate a batch that was never at fault. +// +// Only the attribution is added. Whether the error is retryable stays with the +// classifiers, which read the cause underneath this wrapper unchanged. +func (c *Controller) attributed(err error, subjects ...failure.Subject) error { + for _, s := range subjects { + metrics.NamedCounter(c.metricsScope, opName, "attributed_failure", 1, + metrics.NewTag("subject_type", s.Type)) + } + return errs.Attribute(err, subjects...) +} + // Name returns the controller name for logging and metrics. func (c *Controller) Name() string { return "speculate" diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index 0f4e55008..60cfef5e2 100644 --- a/submitqueue/orchestrator/pipeline.go +++ b/submitqueue/orchestrator/pipeline.go @@ -153,7 +153,7 @@ var Stages = []pipeline.Stage[Deps]{ return speculate.NewController(d.Logger, d.Scope, d.Storage, d.Speculator, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { - return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + return dlq.NewDLQSpeculateController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, }, {