fix(dlq): stop DLQ reconcilers failing Landing batches

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",

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
 

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
 	}

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
 	}

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
+}

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)
 }
 

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
 	}

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