From c6fbd603dfd9e358e08128d2874e36d58fe28e31 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Mon, 2 Nov 2026 03:44:49 +0000 Subject: [PATCH] fix(dlq): stop DLQ reconcilers failing Landing batches A Landing batch has been handed to Runway, so only its merge result can say whether it landed. Every batch DLQ reconciler failed it anyway: a buildsignal, speculate, build, dependency-analysis or conclude dead letter could mark a merged change Error, and the landsignal DLQ failed the batch even when Runway reported the merge succeeded. failBatch now leaves a Landing batch alone unless the caller knows no merge happened: the land DLQ, whose merge was never requested, and the landsignal DLQ for a not-landed result. The landsignal DLQ records a succeeded merge as Succeeded with its requests Landed. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../orchestrator/controller/dlq/BUILD.bazel | 1 + .../orchestrator/controller/dlq/README.md | 1 + .../orchestrator/controller/dlq/batch.go | 6 +- .../controller/dlq/buildsignal.go | 2 +- .../orchestrator/controller/dlq/dlq.go | 56 ++++++++++++++++++- .../orchestrator/controller/dlq/dlq_test.go | 10 ++-- .../orchestrator/controller/dlq/landsignal.go | 12 +++- .../orchestrator/controller/dlq/speculate.go | 2 +- 8 files changed, 78 insertions(+), 12 deletions(-) diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index 68374f9ad..84184343f 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -16,6 +16,7 @@ go_library( visibility = ["//visibility:public"], deps = [ "//api/runway/messagequeue:go_default_library", + "//api/runway/messagequeue/protopb:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index 0ebbb2556..9e9e3260a 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -43,6 +43,7 @@ Reconciliation is safe to run more than once for the same message: - A batch already in `BatchStateFailed` still fans out to member requests, because a previous attempt may have transitioned the batch but crashed before completing the fan-out. Batches in a different terminal state are left unchanged. - Per-request fan-out is itself idempotent via `failRequest`. - 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. +- A batch in `BatchStateLanding` is left to its merge result, because Runway may already have merged it. Only the `land` DLQ, whose merge was never requested, fails it. The `landsignal` DLQ applies the outcome on its message, recording a merged batch as `BatchStateSucceeded` and its requests as `RequestStateLanded`. ## See also diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go index 591c1669b..0223965d9 100644 --- a/submitqueue/orchestrator/controller/dlq/batch.go +++ b/submitqueue/orchestrator/controller/dlq/batch.go @@ -25,6 +25,7 @@ import ( "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/core/topickey" "go.uber.org/zap" ) @@ -114,7 +115,10 @@ 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, lastError, failureMeta); err != nil { + // Land's last step hands the merge to Runway, so its dead letter means the + // merge was never requested and a Landing batch can safely fail. + failLanding := primaryTopicKey(c.topicKey) == topickey.TopicKeyLand + if _, err := failBatch(ctx, store, c.registry, c.logger, bid.ID, failLanding, lastError, failureMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...) return err } diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go index 97e3480e5..084b10afc 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go @@ -129,7 +129,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D return nil } - if _, err := failBatch(ctx, store, c.registry, c.logger, build.BatchID, lastError, failureMeta); err != nil { + if _, err := failBatch(ctx, store, c.registry, c.logger, build.BatchID, false, lastError, failureMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...) return err } diff --git a/submitqueue/orchestrator/controller/dlq/dlq.go b/submitqueue/orchestrator/controller/dlq/dlq.go index 31c0f7825..a250452f4 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq.go +++ b/submitqueue/orchestrator/controller/dlq/dlq.go @@ -186,11 +186,15 @@ func failRequest(ctx context.Context, store orchstorage.Storage, registry consum // lastError and metadata are propagated to each member request's terminal // Error log. // +// A Landing batch is failed only when failLanding is set, by a caller that +// knows no merge happened. Otherwise its merge result decides it: Runway may +// already have put the change on the trunk. +// // 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 orchstorage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string, metadata map[string]string) (bool, error) { +func failBatch(ctx context.Context, store orchstorage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID string, failLanding bool, lastError string, metadata map[string]string) (bool, error) { batch, err := store.GetBatchStore().Get(ctx, batchID) if err != nil { if errors.Is(err, storage.ErrNotFound) { @@ -202,6 +206,13 @@ func failBatch(ctx context.Context, store orchstorage.Storage, registry consumer return false, fmt.Errorf("failed to get batch %s: %w", batchID, err) } + if batch.State == entity.BatchStateLanding && !failLanding { + logger.Infow("dlq reconcile: batch is landing, leaving it to its merge result", + "batch_id", batchID, + ) + return false, nil + } + transitioned := false switch batch.State { @@ -241,3 +252,46 @@ func failBatch(ctx context.Context, store orchstorage.Storage, registry consumer } return transitioned, nil } + +// landBatch records a merge Runway reported as succeeded: a Landing batch moves +// to Succeeded and its member requests to Landed. A Succeeded batch repeats the +// fan-out to repair a partial attempt; any other state is left alone, as the +// primary landsignal stage would leave it. +func landBatch(ctx context.Context, store orchstorage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID string) 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 fmt.Errorf("failed to get batch %s: %w", batchID, err) + } + + switch batch.State { + case entity.BatchStateLanding: + if batch, err = corebatch.Transition(ctx, store, batch, entity.BatchStateSucceeded); err != nil { + return err + } + case entity.BatchStateSucceeded: + if err := corebatch.EnsureRecord(ctx, store, batch); err != nil { + return err + } + default: + logger.Infow("dlq reconcile: merged batch is not landing, skipping", + "batch_id", batchID, + "state", string(batch.State), + ) + return nil + } + + for _, requestID := range batch.Contains { + if _, err := requestcore.TerminateRequest(ctx, store, registry, requestID, entity.RequestStateLanded, "", map[string]string{ + "batch_id": batch.ID, + }); err != nil { + return fmt.Errorf("fan-out for batch %s: %w", batchID, err) + } + } + return nil +} diff --git a/submitqueue/orchestrator/controller/dlq/dlq_test.go b/submitqueue/orchestrator/controller/dlq/dlq_test.go index 02fd82063..c2ed2bf89 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq_test.go +++ b/submitqueue/orchestrator/controller/dlq/dlq_test.go @@ -252,7 +252,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(), "1", "", nil) + _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "1", true, "", nil) require.NoError(t, err) } @@ -282,7 +282,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(), "1", "", nil) + _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "1", false, "", nil) require.NoError(t, err) } @@ -299,7 +299,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(), "1", "", nil) + _, err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "1", false, "", nil) require.NoError(t, err) }) } @@ -338,7 +338,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(), "1", "", nil) + _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "1", false, "", nil) require.NoError(t, err) } @@ -352,7 +352,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(), "1", "", nil) + _, err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "1", false, "", nil) require.NoError(t, err) } diff --git a/submitqueue/orchestrator/controller/dlq/landsignal.go b/submitqueue/orchestrator/controller/dlq/landsignal.go index fc3f9eaf7..e06286f9b 100644 --- a/submitqueue/orchestrator/controller/dlq/landsignal.go +++ b/submitqueue/orchestrator/controller/dlq/landsignal.go @@ -22,6 +22,7 @@ import ( "github.com/uber-go/tally" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" @@ -30,8 +31,8 @@ import ( // landSignalController is the DLQ reconciler for the landsignal topic. Its // payload carries a Runway MergeResult whose id is the batch id echoed back, so -// reconciliation fails that batch directly via failBatch (which also fans out -// to the member requests). +// reconciliation applies that merge's outcome to the batch directly: landBatch +// when it merged, failBatch otherwise. Both fan out to the member requests. type landSignalController struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -97,7 +98,12 @@ func (c *landSignalController) Process(ctx context.Context, delivery consumer.De "dlq_last_error", dmeta["dlq.last_error"], ) - if _, err := failBatch(ctx, store, c.registry, c.logger, result.Id, lastError, failureMeta); err != nil { + if result.Outcome == runwaypb.Outcome_SUCCEEDED { + err = landBatch(ctx, store, c.registry, c.logger, result.Id) + } else { + _, err = failBatch(ctx, store, c.registry, c.logger, result.Id, true, lastError, failureMeta) + } + if err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...) return err } diff --git a/submitqueue/orchestrator/controller/dlq/speculate.go b/submitqueue/orchestrator/controller/dlq/speculate.go index 643646d7b..f467f5274 100644 --- a/submitqueue/orchestrator/controller/dlq/speculate.go +++ b/submitqueue/orchestrator/controller/dlq/speculate.go @@ -126,7 +126,7 @@ func (c *speculateController) Process(ctx context.Context, delivery consumer.Del progressed := false for _, batchID := range culprits { - transitioned, err := failBatch(ctx, store, c.registry, c.logger, batchID, lastError, failureMeta) + transitioned, err := failBatch(ctx, store, c.registry, c.logger, batchID, false, lastError, failureMeta) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...) return err