Skip to content

Commit 7ccc4ff

Browse files
committed
fix(orchestrator): stop dependency-analysis DLQ failing requests a live batch owns
## 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`.
1 parent 3236a76 commit 7ccc4ff

5 files changed

Lines changed: 425 additions & 3 deletions

File tree

‎submitqueue/orchestrator/controller/dlq/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ go_library(
55
srcs = [
66
"batch.go",
77
"buildsignal.go",
8+
"dependencyanalysis.go",
89
"dlq.go",
910
"landconflictsignal.go",
1011
"landsignal.go",
@@ -37,6 +38,7 @@ go_test(
3738
srcs = [
3839
"batch_test.go",
3940
"buildsignal_test.go",
41+
"dependencyanalysis_test.go",
4042
"dlq_test.go",
4143
"landconflictsignal_test.go",
4244
"landsignal_test.go",

‎submitqueue/orchestrator/controller/dlq/README.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,9 +31,9 @@ Two controller shapes cover the eleven primary pipeline topics:
3131
| Controller | Topics | Decoded ID | Terminal state |
3232
|---|---|---|---|
3333
| `NewDLQRequestController` | `start`, `validate`, `batch`, `cancel`, `log` | `RequestID` | `RequestStateError` |
34-
| `NewDLQBatchController` | `speculate`, `build`, `land`, `conclude` | `BatchID` | `BatchStateFailed` + fan-out to member requests as `RequestStateError` |
34+
| `NewDLQBatchController` | `build`, `land`, `conclude` | `BatchID` | `BatchStateFailed` + fan-out to member requests as `RequestStateError` |
3535

36-
`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.
36+
`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.
3737

3838
## Idempotency and concurrent activity
3939

Lines changed: 197 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,197 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package dlq
16+
17+
import (
18+
"context"
19+
"errors"
20+
"fmt"
21+
22+
"github.com/uber-go/tally"
23+
entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
24+
"github.com/uber/submitqueue/platform/consumer"
25+
"github.com/uber/submitqueue/platform/metrics"
26+
sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue"
27+
"github.com/uber/submitqueue/submitqueue/entity"
28+
"github.com/uber/submitqueue/submitqueue/extension/storage"
29+
corebatch "github.com/uber/submitqueue/submitqueue/orchestrator/core/batch"
30+
orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"
31+
"go.uber.org/zap"
32+
)
33+
34+
// dependencyAnalysisController is the DLQ reconciler for the dependency-analysis
35+
// topic.
36+
//
37+
// It does not share the generic batch reconciler because the batch on this
38+
// message need not own its requests. The batch stage mints a batch per
39+
// delivery, so one request can have several batches, and dependency analysis is
40+
// where one of them is chosen. The generic reconciler fails every member
41+
// request, so a dead-lettered duplicate would write Error over a request whose
42+
// chosen batch is still live, and that batch would go on to land it.
43+
//
44+
// So the batch's state decides:
45+
//
46+
// - Creating, with a request already carried by another batch past Creating:
47+
// a duplicate. Nothing is touched; it stays inert in Creating, as the
48+
// primary controller leaves it.
49+
// - Creating otherwise, Created, or Failed: this batch owns its requests and
50+
// nothing downstream will advance it, so it is failed and fanned out.
51+
// - Any later state: a downstream stage took the batch over after this message
52+
// was written, and owns its outcome.
53+
type dependencyAnalysisController struct {
54+
logger *zap.SugaredLogger
55+
metricsScope tally.Scope
56+
stores orchstorage.Factory
57+
registry consumer.TopicRegistry
58+
topicKey consumer.TopicKey
59+
consumerGroup string
60+
}
61+
62+
// Verify dependencyAnalysisController implements consumer.Controller at compile time.
63+
var _ consumer.Controller = (*dependencyAnalysisController)(nil)
64+
65+
// NewDLQDependencyAnalysisController builds the DLQ controller for the
66+
// dependency-analysis topic. topicKey must be the DLQ topic key (typically
67+
// TopicKey(primary)).
68+
func NewDLQDependencyAnalysisController(
69+
logger *zap.SugaredLogger,
70+
scope tally.Scope,
71+
stores orchstorage.Factory,
72+
registry consumer.TopicRegistry,
73+
topicKey consumer.TopicKey,
74+
consumerGroup string,
75+
) consumer.Controller {
76+
name := string(topicKey) + "_controller"
77+
return &dependencyAnalysisController{
78+
logger: logger.Named(name),
79+
metricsScope: scope.SubScope(name),
80+
stores: stores,
81+
registry: registry,
82+
topicKey: topicKey,
83+
consumerGroup: consumerGroup,
84+
}
85+
}
86+
87+
// Process reconciles a single dead-lettered dependency-analysis message.
88+
func (c *dependencyAnalysisController) Process(ctx context.Context, delivery consumer.Delivery) error {
89+
const opName = "process"
90+
91+
msg := delivery.Message()
92+
93+
bid, err := sqmq.UnmarshalBatchID(primaryTopicKey(c.topicKey), msg.Payload)
94+
if err != nil {
95+
metrics.NamedCounter(c.metricsScope, opName, "deserialize_errors", 1)
96+
return fmt.Errorf("failed to decode batch id from dlq payload: %w", err)
97+
}
98+
if err := entityqueue.ValidatePayloadQueue(msg, bid.Queue); err != nil {
99+
metrics.NamedCounter(c.metricsScope, opName, "queue_identity_errors", 1)
100+
return nil
101+
}
102+
if bid.ID == "" {
103+
metrics.NamedCounter(c.metricsScope, opName, "empty_id_errors", 1)
104+
return fmt.Errorf("dlq payload decoded to empty batch id")
105+
}
106+
107+
store, err := c.stores.For(orchstorage.Config{QueueName: bid.Queue})
108+
if err != nil {
109+
metrics.NamedCounter(c.metricsScope, opName, "storage_resolve_errors", 1)
110+
// Non-retryable: a missing or unresolvable queue is a malformed message.
111+
return fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err)
112+
}
113+
114+
lastError, failureMeta := failureContext(delivery)
115+
dmeta := delivery.Metadata()
116+
c.logger.Warnw("dlq message received",
117+
"batch_id", bid.ID,
118+
"attempt", delivery.Attempt(),
119+
"dlq_original_topic", dmeta["dlq.original_topic"],
120+
"dlq_failure_count", dmeta["dlq.failure_count"],
121+
"dlq_last_error", dmeta["dlq.last_error"],
122+
)
123+
124+
batch, err := store.GetBatchStore().Get(ctx, bid.ID)
125+
if err != nil {
126+
if errors.Is(err, storage.ErrNotFound) {
127+
c.logger.Warnw("dlq reconcile: batch not found, skipping", "batch_id", bid.ID)
128+
return nil
129+
}
130+
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
131+
return fmt.Errorf("failed to get batch %s: %w", bid.ID, err)
132+
}
133+
134+
switch batch.State {
135+
case entity.BatchStateCreating:
136+
carrier, err := c.findOtherBatchCarryingRequests(ctx, store, batch)
137+
if err != nil {
138+
metrics.NamedCounter(c.metricsScope, opName, "batch_lookup_errors", 1)
139+
return err
140+
}
141+
if carrier != "" {
142+
metrics.NamedCounter(c.metricsScope, opName, "skipped_duplicate", 1)
143+
c.logger.Infow("dlq reconcile: batch is a duplicate, its requests are carried by another",
144+
"batch_id", batch.ID,
145+
"carried_by", carrier,
146+
)
147+
return nil
148+
}
149+
case entity.BatchStateCreated, entity.BatchStateFailed:
150+
default:
151+
metrics.NamedCounter(c.metricsScope, opName, "skipped_owned_downstream", 1)
152+
c.logger.Infow("dlq reconcile: batch has moved past dependency analysis, leaving it to its current stage",
153+
"batch_id", batch.ID,
154+
"state", string(batch.State),
155+
)
156+
return nil
157+
}
158+
159+
if _, err := failBatch(ctx, store, c.registry, c.logger, batch.ID, lastError, failureMeta); err != nil {
160+
metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
161+
return err
162+
}
163+
return nil
164+
}
165+
166+
// findOtherBatchCarryingRequests returns a batch other than this one that got
167+
// past Creating with one of this batch's requests, or the empty string. It is
168+
// the same test the primary controller uses to abandon a duplicate.
169+
func (c *dependencyAnalysisController) findOtherBatchCarryingRequests(ctx context.Context, store orchstorage.Storage, batch entity.Batch) (string, error) {
170+
for _, requestID := range batch.Contains {
171+
batches, _, err := corebatch.FindByRequestID(ctx, store, requestID)
172+
if err != nil {
173+
return "", err
174+
}
175+
for _, other := range batches {
176+
if other.ID != batch.ID && other.State != entity.BatchStateCreating {
177+
return other.ID, nil
178+
}
179+
}
180+
}
181+
return "", nil
182+
}
183+
184+
// Name returns the controller name for logging and metrics.
185+
func (c *dependencyAnalysisController) Name() string {
186+
return string(c.topicKey)
187+
}
188+
189+
// TopicKey returns the topic key this controller subscribes to.
190+
func (c *dependencyAnalysisController) TopicKey() consumer.TopicKey {
191+
return c.topicKey
192+
}
193+
194+
// ConsumerGroup returns the consumer group for offset tracking.
195+
func (c *dependencyAnalysisController) ConsumerGroup() string {
196+
return c.consumerGroup
197+
}

0 commit comments

Comments
 (0)