From 7ccc4fffa5fd004122d12cfef38657df3efa6862 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Fri, 16 Oct 2026 05:55:10 -0700 Subject: [PATCH] fix(orchestrator): stop dependency-analysis DLQ failing requests a live batch owns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The batch stage mints a batch per delivery, so a lost ack leaves two batches for one request, and dependency analysis is where one of them is chosen. Its DLQ reused the generic batch reconciler, which fails every member request. A dead-lettered duplicate therefore wrote `Error` over a request whose chosen batch was still live; that batch went on to land the change while the request stayed reported as failed. The same reconciler would also fail a batch that speculate had already admitted, racing land. ### What? - Add `NewDLQDependencyAnalysisController`, which decides by the batch's state: - `Creating` with a request already carried by another batch past `Creating`: a duplicate, left inert (same test the primary controller uses to abandon it). - `Creating` otherwise, `Created`, or `Failed`: the batch owns its requests and nothing downstream will advance it, so it is failed and fanned out as before. - Any later state: a downstream stage owns it; acked untouched. - Wire it as the dependency-analysis stage's DLQ and update the DLQ README's controller mapping (which also still listed `speculate` under the generic controller). ## Test Plan ✅ `bazel test //submitqueue/orchestrator/...` (new table-driven DLQ test covers every batch state, duplicate vs owner, and lookup failure) ✅ `make lint-fmt`, `make check-gazelle` `TestConflictAnalyzerFailure_ReconcilesFromDLQ` (e2e) exercises the owner path and should still see the batch `Failed` and the request `Error`. --- .../orchestrator/controller/dlq/BUILD.bazel | 2 + .../orchestrator/controller/dlq/README.md | 4 +- .../controller/dlq/dependencyanalysis.go | 197 ++++++++++++++++ .../controller/dlq/dependencyanalysis_test.go | 223 ++++++++++++++++++ submitqueue/orchestrator/pipeline.go | 2 +- 5 files changed, 425 insertions(+), 3 deletions(-) create mode 100644 submitqueue/orchestrator/controller/dlq/dependencyanalysis.go create mode 100644 submitqueue/orchestrator/controller/dlq/dependencyanalysis_test.go diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index 68374f9ad..7f098b1af 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "batch.go", "buildsignal.go", + "dependencyanalysis.go", "dlq.go", "landconflictsignal.go", "landsignal.go", @@ -37,6 +38,7 @@ go_test( srcs = [ "batch_test.go", "buildsignal_test.go", + "dependencyanalysis_test.go", "dlq_test.go", "landconflictsignal_test.go", "landsignal_test.go", diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index 0ebbb2556..88bf3b121 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -31,9 +31,9 @@ Two controller shapes cover the eleven primary pipeline topics: | Controller | Topics | Decoded ID | Terminal state | |---|---|---|---| | `NewDLQRequestController` | `start`, `validate`, `batch`, `cancel`, `log` | `RequestID` | `RequestStateError` | -| `NewDLQBatchController` | `speculate`, `build`, `land`, `conclude` | `BatchID` | `BatchStateFailed` + fan-out to member requests as `RequestStateError` | +| `NewDLQBatchController` | `build`, `land`, `conclude` | `BatchID` | `BatchStateFailed` + fan-out to member requests as `RequestStateError` | -`buildsignal` carries a `Build` payload and has its own small dedicated controller. The split exists because the DLQ message payload shape mirrors the primary topic's payload shape (the queue framework preserves bytes verbatim under the `_dlq` topic name), so the decoder is what changes per topic — not the reconciliation step. The package-level `RequestIDDecoder` interface plus `DecodeLandRequestID` / `DecodeCancelRequestID` / `DecodeRequestID` cover the three payload shapes used by request-scoped topics. +`buildsignal` carries a `Build` payload and has its own small dedicated controller. `speculate` and `dependency-analysis` carry a `BatchID` but have dedicated controllers, because the batch on their message is not necessarily the one at fault: a speculate run re-plans the whole queue, and a dependency-analysis batch may be a duplicate whose requests another batch already carries. See the doc comments in `speculate.go` and `dependencyanalysis.go`. The split exists because the DLQ message payload shape mirrors the primary topic's payload shape (the queue framework preserves bytes verbatim under the `_dlq` topic name), so the decoder is what changes per topic — not the reconciliation step. The package-level `RequestIDDecoder` interface plus `DecodeLandRequestID` / `DecodeCancelRequestID` / `DecodeRequestID` cover the three payload shapes used by request-scoped topics. ## Idempotency and concurrent activity diff --git a/submitqueue/orchestrator/controller/dlq/dependencyanalysis.go b/submitqueue/orchestrator/controller/dlq/dependencyanalysis.go new file mode 100644 index 000000000..b243a692e --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/dependencyanalysis.go @@ -0,0 +1,197 @@ +// Copyright (c) 2026 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" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + corebatch "github.com/uber/submitqueue/submitqueue/orchestrator/core/batch" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" + "go.uber.org/zap" +) + +// dependencyAnalysisController is the DLQ reconciler for the dependency-analysis +// topic. +// +// It does not share the generic batch reconciler because the batch on this +// message need not own its requests. The batch stage mints a batch per +// delivery, so one request can have several batches, and dependency analysis is +// where one of them is chosen. The generic reconciler fails every member +// request, so a dead-lettered duplicate would write Error over a request whose +// chosen batch is still live, and that batch would go on to land it. +// +// So the batch's state decides: +// +// - Creating, with a request already carried by another batch past Creating: +// a duplicate. Nothing is touched; it stays inert in Creating, as the +// primary controller leaves it. +// - Creating otherwise, Created, or Failed: this batch owns its requests and +// nothing downstream will advance it, so it is failed and fanned out. +// - Any later state: a downstream stage took the batch over after this message +// was written, and owns its outcome. +type dependencyAnalysisController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores orchstorage.Factory + registry consumer.TopicRegistry + topicKey consumer.TopicKey + consumerGroup string +} + +// Verify dependencyAnalysisController implements consumer.Controller at compile time. +var _ consumer.Controller = (*dependencyAnalysisController)(nil) + +// NewDLQDependencyAnalysisController builds the DLQ controller for the +// dependency-analysis topic. topicKey must be the DLQ topic key (typically +// TopicKey(primary)). +func NewDLQDependencyAnalysisController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores orchstorage.Factory, + registry consumer.TopicRegistry, + topicKey consumer.TopicKey, + consumerGroup string, +) consumer.Controller { + name := string(topicKey) + "_controller" + return &dependencyAnalysisController{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + stores: stores, + registry: registry, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process reconciles a single dead-lettered dependency-analysis message. +func (c *dependencyAnalysisController) Process(ctx context.Context, delivery consumer.Delivery) error { + const opName = "process" + + msg := delivery.Message() + + bid, err := sqmq.UnmarshalBatchID(primaryTopicKey(c.topicKey), 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 err := entityqueue.ValidatePayloadQueue(msg, bid.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "queue_identity_errors", 1) + return nil + } + 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(orchstorage.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) + dmeta := delivery.Metadata() + c.logger.Warnw("dlq message received", + "batch_id", bid.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"], + ) + + batch, err := store.GetBatchStore().Get(ctx, bid.ID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + c.logger.Warnw("dlq reconcile: batch not found, skipping", "batch_id", bid.ID) + return nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get batch %s: %w", bid.ID, err) + } + + switch batch.State { + case entity.BatchStateCreating: + carrier, err := c.findOtherBatchCarryingRequests(ctx, store, batch) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "batch_lookup_errors", 1) + return err + } + if carrier != "" { + metrics.NamedCounter(c.metricsScope, opName, "skipped_duplicate", 1) + c.logger.Infow("dlq reconcile: batch is a duplicate, its requests are carried by another", + "batch_id", batch.ID, + "carried_by", carrier, + ) + return nil + } + case entity.BatchStateCreated, entity.BatchStateFailed: + default: + metrics.NamedCounter(c.metricsScope, opName, "skipped_owned_downstream", 1) + c.logger.Infow("dlq reconcile: batch has moved past dependency analysis, leaving it to its current stage", + "batch_id", batch.ID, + "state", string(batch.State), + ) + return nil + } + + if _, err := failBatch(ctx, store, c.registry, c.logger, batch.ID, lastError, failureMeta); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) + return err + } + return nil +} + +// findOtherBatchCarryingRequests returns a batch other than this one that got +// past Creating with one of this batch's requests, or the empty string. It is +// the same test the primary controller uses to abandon a duplicate. +func (c *dependencyAnalysisController) findOtherBatchCarryingRequests(ctx context.Context, store orchstorage.Storage, batch entity.Batch) (string, error) { + for _, requestID := range batch.Contains { + batches, _, err := corebatch.FindByRequestID(ctx, store, requestID) + if err != nil { + return "", err + } + for _, other := range batches { + if other.ID != batch.ID && other.State != entity.BatchStateCreating { + return other.ID, nil + } + } + } + return "", nil +} + +// Name returns the controller name for logging and metrics. +func (c *dependencyAnalysisController) Name() string { + return string(c.topicKey) +} + +// TopicKey returns the topic key this controller subscribes to. +func (c *dependencyAnalysisController) TopicKey() consumer.TopicKey { + return c.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (c *dependencyAnalysisController) ConsumerGroup() string { + return c.consumerGroup +} diff --git a/submitqueue/orchestrator/controller/dlq/dependencyanalysis_test.go b/submitqueue/orchestrator/controller/dlq/dependencyanalysis_test.go new file mode 100644 index 000000000..f73fb012e --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/dependencyanalysis_test.go @@ -0,0 +1,223 @@ +// Copyright (c) 2026 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" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/platform/consumer" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap/zaptest" +) + +const ( + dependencyAnalysisDLQGroup = "orchestrator-dependency-analysis-dlq" + deadLetteredBatchID = "8" + carriedRequestID = "1" +) + +func newDependencyAnalysisController(t *testing.T, store *orchstoragemock.MockStorage, registry consumer.TopicRegistry) consumer.Controller { + return NewDLQDependencyAnalysisController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store: store}, registry, + TopicKey(topickey.TopicKeyDependencyAnalysis), dependencyAnalysisDLQGroup) +} + +func dependencyAnalysisPayload(t *testing.T, batchID string) []byte { + payload, err := sqmq.MarshalID(sqmq.TopicKeyDependencyAnalysis, batchID, "q") + require.NoError(t, err) + return payload +} + +func TestDLQDependencyAnalysisController_InterfaceAndAccessors(t *testing.T) { + ctrl := gomock.NewController(t) + c := newDependencyAnalysisController(t, orchstoragemock.NewMockStorage(ctrl), consumer.TopicRegistry{}) + + assert.Equal(t, "dependency-analysis_dlq", c.Name()) + assert.Equal(t, consumer.TopicKey("dependency-analysis_dlq"), c.TopicKey()) + assert.Equal(t, dependencyAnalysisDLQGroup, c.ConsumerGroup()) +} + +func TestDLQDependencyAnalysisController_Process_ReconcilesByBatchState(t *testing.T) { + tests := []struct { + name string + state entity.BatchState + // otherBatches are further batches associated with the same request. + otherBatches []entity.Batch + wantBatchFailed bool + wantRequestError bool + }{ + { + name: "creating batch that owns its request is failed", + state: entity.BatchStateCreating, + wantBatchFailed: true, + wantRequestError: true, + }, + { + // A sibling still in Creating has not been chosen either, so it + // does not make this one a duplicate. + name: "creating batch beside another creating batch is failed", + state: entity.BatchStateCreating, + otherBatches: []entity.Batch{{ID: "7", Queue: "q", State: entity.BatchStateCreating}}, + wantBatchFailed: true, + wantRequestError: true, + }, + { + name: "creating duplicate of a live batch is left alone", + state: entity.BatchStateCreating, + otherBatches: []entity.Batch{{ID: "7", Queue: "q", State: entity.BatchStateSpeculating}}, + }, + { + name: "creating duplicate of a landed batch is left alone", + state: entity.BatchStateCreating, + otherBatches: []entity.Batch{{ID: "7", Queue: "q", State: entity.BatchStateSucceeded}}, + }, + { + name: "created batch whose announcement was lost is failed", + state: entity.BatchStateCreated, + wantBatchFailed: true, + wantRequestError: true, + }, + { + // A previous attempt failed the batch and crashed before the fan-out. + name: "failed batch repeats the fan-out", + state: entity.BatchStateFailed, + wantRequestError: true, + }, + {name: "speculating batch is left to speculate", state: entity.BatchStateSpeculating}, + {name: "landing batch is left to land", state: entity.BatchStateLanding}, + {name: "cancelling batch is left to cancel", state: entity.BatchStateCancelling}, + {name: "succeeded batch is left alone", state: entity.BatchStateSucceeded}, + {name: "cancelled batch is left alone", state: entity.BatchStateCancelled}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := entity.Batch{ + ID: deadLetteredBatchID, Queue: "q", Contains: []string{carriedRequestID}, + State: tt.state, Version: 2, + } + batches := map[string]entity.Batch{batch.ID: batch} + associations := []entity.RequestBatch{{RequestID: carriedRequestID, BatchID: batch.ID, Version: 1}} + for _, other := range tt.otherBatches { + batches[other.ID] = other + associations = append(associations, entity.RequestBatch{RequestID: carriedRequestID, BatchID: other.ID, Version: 1}) + } + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, id string) (entity.Batch, error) { + b, ok := batches[id] + if !ok { + return entity.Batch{}, storage.ErrNotFound + } + return b, nil + }, + ).AnyTimes() + if tt.wantBatchFailed { + batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateFailed), int32(2), int32(3)).Return(nil) + } + + requestBatchStore := storagemock.NewMockRequestBatchStore(ctrl) + requestBatchStore.EXPECT().GetByRequestID(gomock.Any(), carriedRequestID).Return(associations, nil).AnyTimes() + + request := entity.Request{ID: carriedRequestID, Queue: "q", Version: 3, State: entity.RequestStateBatched} + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), carriedRequestID).Return(request, nil).AnyTimes() + logPublishes := 0 + if tt.wantRequestError { + requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(3), int32(4)).Return(nil) + logPublishes = 1 + } + + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(requestBatchStore).AnyTimes() + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() + + registry := newTestLogRegistry(t, ctrl, logPublishes, func(entity.RequestLog) error { return nil }) + c := newDependencyAnalysisController(t, store, registry) + + require.NoError(t, c.Process(context.Background(), newMockDelivery(ctrl, dependencyAnalysisPayload(t, batch.ID)))) + }) + } +} + +func TestDLQDependencyAnalysisController_Process_BatchNotFoundAcks(t *testing.T) { + ctrl := gomock.NewController(t) + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), deadLetteredBatchID).Return(entity.Batch{}, storage.ErrNotFound) + + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + c := newDependencyAnalysisController(t, store, consumer.TopicRegistry{}) + + require.NoError(t, c.Process(context.Background(), newMockDelivery(ctrl, dependencyAnalysisPayload(t, deadLetteredBatchID)))) +} + +// Without the association lookup the controller cannot tell a duplicate from +// the owner, so it must retry rather than guess. +func TestDLQDependencyAnalysisController_Process_AssociationLookupErrorRetries(t *testing.T) { + ctrl := gomock.NewController(t) + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), deadLetteredBatchID).Return(entity.Batch{ + ID: deadLetteredBatchID, Queue: "q", Contains: []string{carriedRequestID}, + State: entity.BatchStateCreating, Version: 1, + }, nil) + + requestBatchStore := storagemock.NewMockRequestBatchStore(ctrl) + requestBatchStore.EXPECT().GetByRequestID(gomock.Any(), carriedRequestID).Return(nil, errors.New("store unavailable")) + + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(requestBatchStore).AnyTimes() + c := newDependencyAnalysisController(t, store, consumer.TopicRegistry{}) + + require.Error(t, c.Process(context.Background(), newMockDelivery(ctrl, dependencyAnalysisPayload(t, deadLetteredBatchID)))) +} + +func TestDLQDependencyAnalysisController_Process_MalformedPayloadFails(t *testing.T) { + ctrl := gomock.NewController(t) + c := newDependencyAnalysisController(t, orchstoragemock.NewMockStorage(ctrl), consumer.TopicRegistry{}) + + require.Error(t, c.Process(context.Background(), newMockDelivery(ctrl, []byte("garbage")))) +} + +func TestDLQDependencyAnalysisController_Process_EmptyIDFails(t *testing.T) { + ctrl := gomock.NewController(t) + c := newDependencyAnalysisController(t, orchstoragemock.NewMockStorage(ctrl), consumer.TopicRegistry{}) + + require.Error(t, c.Process(context.Background(), newMockDelivery(ctrl, dependencyAnalysisPayload(t, "")))) +} + +func TestDLQDependencyAnalysisController_Process_TenantPayloadQueueMismatchAcks(t *testing.T) { + ctrl := gomock.NewController(t) + c := newDependencyAnalysisController(t, orchstoragemock.NewMockStorage(ctrl), consumer.TopicRegistry{}) + + require.NoError(t, c.Process(context.Background(), newMockDeliveryWithTenant(ctrl, dependencyAnalysisPayload(t, deadLetteredBatchID), "other-queue"))) +} diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index ed955125a..58b130c29 100644 --- a/submitqueue/orchestrator/pipeline.go +++ b/submitqueue/orchestrator/pipeline.go @@ -164,7 +164,7 @@ var Stages = []pipeline.Stage[Deps]{ return dependencyanalysis.NewController(d.Logger, d.Scope, d.Storage, d.Analyzer, 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.NewDLQDependencyAnalysisController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, }, {