sbalabanov · Nov 13, 2026
@@ -4,7 +4,7 @@ The orchestrator processes land requests through the queue-driven pipeline decla
 
 Two cycles re-enter speculation. `speculate` dispatches funded paths to `build`; `build` starts each pending path and publishes the build id to `buildsignal`; `buildsignal` polls with `delivery.Hold` and, when the status changes, publishes the batch id back to `speculate`. A batch that can land is published to `submitqueue-land`, which sends the merge request to Runway; `landsignal` correlates the result, marks the batch Succeeded or Failed, and fans out to `conclude` and `speculate`. `speculate` also publishes Failed and Cancelled batches straight to `conclude`.
 
-Terminal request states are not `conclude`'s alone. `cancel` completes a request that has not been enrolled in a batch. `landconflictsignal` fails a request Runway reports as conflicted. `conclude` maps a terminal batch onto its member requests. The DLQ reconcilers force a failed terminal state when a consumed stage cannot finish.
+Terminal request states are not `conclude`'s alone. `cancel` completes a request that has not been enrolled in a batch. `landconflictsignal` fails a request Runway reports as conflicted. `conclude` maps a terminal batch onto its member requests. Most DLQ reconcilers force a failed terminal state when a consumed stage cannot finish; conclude’s final reconciler preserves the batch’s authoritative outcome.
 
 The same `Stages` table registers a `submitqueue-hook` stage, outside the order of the flow above. It consumes hook events and invokes the wired hooks, and it publishes nothing onward. No orchestrator controller publishes a hook event; `service/submitqueue/orchestrator/server` resolves every event to noop. The comment on that row in `pipeline.go` is the contract: any stage can publish there.
 
@@ -162,6 +162,8 @@ Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscriptio
 
 DLQ handlers do not resume the failed request's normal workflow. They reconcile terminal state and its outputs, preserving authoritative terminal winners. The generic request reconciler also preserves valid downstream batch ownership. The [DLQ package guide](../../../submitqueue/orchestrator/controller/dlq/README.md) defines the reconcile-only boundary, handler-specific dispositions, and named limits; the [stage table](../../../submitqueue/orchestrator/pipeline.go) defines the bindings. Unlimited final retries protect reconciliation, not another business retry budget.
 
+The conclude DLQ repairs terminal membership and uses primary conclusion’s shared member fan-out: Succeeded becomes Landed, Failed becomes Error, and Cancelled becomes Cancelled. It never infers a merge failure from a failed conclusion. Non-terminal batches (including Landing), missing batches or members, malformed payload identities, and partial writes/publishes remain in the final queue until resolved. Newly failed members receive the original message’s failure reason, never an incidental DLQ infrastructure error; matching terminal members repair their status logs, and different terminal winners remain unchanged.
+
 The narrow speculate nudge concerns other live batches after culprit failure; it never reopens the failed batch. Hook DLQ is report-and-stop, with side-effect recovery left to a person. Neither authorizes replay of the failed business command.
 
 The orchestrator publishes terminal status logs and the gateway alone materializes them; replaying a log into gateway read models is not orchestration restart. Runway owns failure-result publication for dead letters on its inbound merge queues, without rerunning the merge.

@@ -45,12 +45,24 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//api/base/hook:go_default_library",
+        "//platform/base/failure:go_default_library",
+        "//platform/base/messagequeue:go_default_library",
         "//platform/extension/hook:go_default_library",
         "//platform/extension/hook/noop:go_default_library",
+        "//platform/extension/messagequeue:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
         "//platform/pipeline:go_default_library",
+        "//submitqueue/core/messagequeue:go_default_library",
+        "//submitqueue/core/topickey:go_default_library",
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/storage:go_default_library",
+        "//submitqueue/extension/storage/mock:go_default_library",
+        "//submitqueue/orchestrator/extension/storage:go_default_library",
+        "//submitqueue/orchestrator/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
         "@org_uber_go_zap//zaptest:go_default_library",
     ],
 )
@@ -11,9 +11,7 @@ go_library(
         "//platform/metrics:go_default_library",
         "//submitqueue/core/messagequeue:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
-        "//submitqueue/entity:go_default_library",
-        "//submitqueue/extension/storage:go_default_library",
-        "//submitqueue/orchestrator/core/request:go_default_library",
+        "//submitqueue/orchestrator/core/batch:go_default_library",
         "//submitqueue/orchestrator/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",

@@ -24,9 +24,7 @@ import (
 	"github.com/uber/submitqueue/platform/metrics"
 	sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
-	"github.com/uber/submitqueue/submitqueue/entity"
-	storage "github.com/uber/submitqueue/submitqueue/extension/storage"
-	corerequest "github.com/uber/submitqueue/submitqueue/orchestrator/core/request"
+	corebatch "github.com/uber/submitqueue/submitqueue/orchestrator/core/batch"
 	orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"
 	"go.uber.org/zap"
 )
@@ -110,61 +108,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		"partition_key", msg.PartitionKey,
 	)
 
-	// Map batch terminal state to request state.
-	// We expect the batch to be in a terminal state as written by the land
-	// controller (Succeeded) or the speculate controller (Failed via
-	// failOnDependency, Cancelled via cancelBatch).
-	requestState, err := batchStateToRequestState(batch.State)
-	if err != nil {
-		metrics.NamedCounter(c.metricsScope, "process", "unexpected_state_errors", 1, metrics.TagsFromContext(ctx)...)
-		return fmt.Errorf("unexpected batch state %q for batch %s: %w", batch.State, batch.ID, err)
-	}
-
-	// Reconcile each request to the batch's terminal state and emit a terminal
-	// log entry. TerminateRequest is idempotent under at-least-once delivery: a
-	// prior attempt may have completed the CAS but failed before publishing the
-	// log, so the log publish still runs when the request is already in the
-	// target terminal state, and a request that a concurrent path already drove
-	// to a different terminal state is left untouched. A request referenced by
-	// the batch but missing from the store is a hard error — the whole batch is
-	// retried (and eventually dead-lettered) rather than silently skipped. We
-	// translate the result into per-outcome logs and metrics.
-	// The failure reason rides the conclude message, not the batch: the failing
-	// stage stamps it here and it is empty on the landed and cancelled paths.
-	failureReason := msg.Metadata[topickey.MetadataKeyFailureReason]
-	for _, requestID := range batch.Contains {
-		res, err := corerequest.TerminateRequest(ctx, store, c.registry, requestID, requestState, failureReason, map[string]string{
-			"batch_id": batch.ID,
-		})
-		if err != nil {
-			metrics.NamedCounter(c.metricsScope, "process", "terminate_errors", 1, metrics.TagsFromContext(ctx)...)
-			return fmt.Errorf("failed to terminate request %s for batch %s: %w", requestID, batch.ID, err)
-		}
-
-		switch res.Outcome {
-		case corerequest.TerminationOutcomeSuccess:
-			c.logger.Infow("updated request state",
-				"batch_id", batch.ID,
-				"request_id", requestID,
-				"new_state", string(res.AfterState),
-			)
-		case corerequest.TerminationOutcomeAlreadyInTargetState:
-			metrics.NamedCounter(c.metricsScope, "process", "already_reconciled", 1, metrics.TagsFromContext(ctx)...)
-		case corerequest.TerminationOutcomeDiverged:
-			c.logger.Warnw("request already in different terminal state, skipping reconcile",
-				"batch_id", batch.ID,
-				"request_id", requestID,
-				"expected_state", string(requestState),
-				"actual_state", string(res.BeforeState),
-			)
-			metrics.NamedCounter(c.metricsScope, "process", "terminal_state_divergence", 1, metrics.TagsFromContext(ctx)...)
-		case corerequest.TerminationOutcomeNotFound:
-			metrics.NamedCounter(c.metricsScope, "process", "request_store_errors", 1, metrics.TagsFromContext(ctx)...)
-			return fmt.Errorf("failed to get request %s for batch %s: %w", requestID, batch.ID, storage.ErrNotFound)
-		}
-	}
-
-	return nil // Success - message will be acked
+	return corebatch.ConcludeBatchRequests(ctx, store, c.registry, c.logger, c.metricsScope, batch, msg.Metadata[topickey.MetadataKeyFailureReason])
 }
 
 // Name returns the controller name for logging and metrics.
@@ -181,17 +125,3 @@ func (c *Controller) TopicKey() consumer.TopicKey {
 func (c *Controller) ConsumerGroup() string {
 	return c.consumerGroup
 }
-
-// batchStateToRequestState maps a terminal batch state to the corresponding request state.
-func batchStateToRequestState(state entity.BatchState) (entity.RequestState, error) {
-	switch state {
-	case entity.BatchStateSucceeded:
-		return entity.RequestStateLanded, nil
-	case entity.BatchStateFailed:
-		return entity.RequestStateError, nil
-	case entity.BatchStateCancelled:
-		return entity.RequestStateCancelled, nil
-	default:
-		return entity.RequestStateUnknown, fmt.Errorf("non-terminal batch state: %s", state)
-	}
-}
@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "batch.go",
         "buildsignal.go",
+        "conclude.go",
         "dlq.go",
         "landconflictsignal.go",
         "landsignal.go",
@@ -37,6 +38,7 @@ go_test(
     srcs = [
         "batch_test.go",
         "buildsignal_test.go",
+        "conclude_test.go",
         "dlq_test.go",
         "landconflictsignal_test.go",
         "landsignal_test.go",

@@ -14,7 +14,13 @@ Do not bind a normal orchestration controller to a DLQ, create a missing request
 
 Dead letters have exhausted the primary retry budget or returned a non-retryable error. Final consumers use `errs.AlwaysRetryableProcessor`, unlimited attempts (`Retry.MaxAttempts = 0`), and `DLQ.Enabled = false`. Every non-nil error returned by a DLQ controller, including a version mismatch or publish failure, is redelivered until reconciliation succeeds or an operator intervenes. Unlimited retries protect reconciliation and permitted outputs, not another business retry budget.
 
-Disposition remains handler-specific: the generic request/batch reconcilers acknowledge missing entities; buildsignal also acknowledges a missing build or empty batch ID. Decode errors returned by these handlers remain retryable and may need operator removal. Payload/tenant mismatches are acknowledged, while invalid envelope tenant metadata is rejected before controller error processing. The shared hook handler reports and acknowledges even undecodable events. Final retry policy therefore does not promise retention of every poison message or recreation of every missing entity.
+Disposition remains handler-specific: the generic request/batch reconcilers acknowledge missing entities; buildsignal also acknowledges a missing build or empty batch ID. Decode errors returned by these handlers remain retryable and may need operator removal. Generic request/batch payload/tenant mismatches are acknowledged, while invalid envelope tenant metadata is rejected before controller error processing. The shared hook handler reports and acknowledges even undecodable events. Final retry policy therefore does not promise retention of every poison message or recreation of every missing entity.
+
+## Conclude final policy
+
+Conclude repairs terminal queue membership before fan-out and never changes the batch state. Every non-terminal state, including Landing, returns an error, as do missing records and invalid payload identities. The same final delivery retries until member writes and log publication complete; it does not republish a fresh conclude message.
+
+Newly failed members use the original message’s `failure_reason` metadata, not a later DLQ infrastructure error. Newly landed or cancelled members carry no LastError. Already-matching terminal members repair their status logs; a different terminal winner is logged and preserved without overwriting it. Queue-wide dependent reconsideration is outside conclusion’s fan-out contract.
 
 ## Terminal outputs and ownership
 
@@ -30,7 +36,8 @@ Request logs are published to `log`; the gateway alone materializes them into it
 The [stage table](../../pipeline.go) is the source of truth rather than a second topology table here:
 
 - `start`, `cancel`, `validate`, and `batch` use generic request reconciliation. Their decoders read LandRequest, CancelRequest, or the topic's request-ID payload.
-- `dependency-analysis`, `build`, `submitqueue-land`, and `conclude` use generic batch reconciliation.
+- `dependency-analysis`, `build`, and `submitqueue-land` use generic batch reconciliation.
+- `conclude` reconciles a terminal batch to its matching request outcomes and status logs.
 - `buildsignal` carries a build-ID payload, not a full Build. Its handler loads the build to find the batch and leaves build status untouched. The two Runway signal handlers read a request or batch correlation ID from `MergeResult`.
 - **Speculation queue-isolation nudge:** after failing attributed culprit work (or the labelled payload fallback), a delivery that transitioned a batch may nudge `speculate` with another live batch. The primary re-derives current queue state; the failed batch is not reopened. This narrow protection for other work does not authorize replay of the dead-lettered command or guarantee a nudge after every partial reconciliation failure.
 - **Hook report-and-stop:** the shared hook DLQ logs the raw event, counts it as dropped, and acknowledges it. A person decides whether to republish; the DLQ does not invoke hooks.

@@ -29,7 +29,7 @@ import (
 )
 
 // batchController is the DLQ reconciler for batch-scoped pipeline stages
-// (build, land, conclude). All three topics carry a BatchID payload, so this
+// (dependencyanalysis, build, land). 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.
 //
@@ -41,7 +41,7 @@ import (
 //
 // 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,
-// land, or conclusion. The speculate stage is not like that — it re-plans a
+// dependency analysis, build, or land. 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 {

@@ -0,0 +1,84 @@
+// 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"
+	"fmt"
+
+	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+	sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue"
+	"github.com/uber/submitqueue/submitqueue/core/topickey"
+	corebatch "github.com/uber/submitqueue/submitqueue/orchestrator/core/batch"
+	orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"
+	"go.uber.org/zap"
+)
+
+type concludeController struct {
+	logger        *zap.SugaredLogger
+	metricsScope  tally.Scope
+	stores        orchstorage.Factory
+	registry      consumer.TopicRegistry
+	topicKey      consumer.TopicKey
+	consumerGroup string
+}
+
+// NewDLQConcludeController repairs terminal membership and request fan-out without changing the batch outcome.
+func NewDLQConcludeController(logger *zap.SugaredLogger, scope tally.Scope, stores orchstorage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string) consumer.Controller {
+	name := string(topicKey) + "_controller"
+	return &concludeController{
+		logger: logger.Named(name), metricsScope: scope.SubScope(name), stores: stores,
+		registry: registry, topicKey: topicKey, consumerGroup: consumerGroup,
+	}
+}
+
+func (c *concludeController) Process(ctx context.Context, delivery consumer.Delivery) error {
+	msg := delivery.Message()
+	bid, err := sqmq.UnmarshalBatchID(primaryTopicKey(c.topicKey), msg.Payload)
+	if err != nil {
+		return fmt.Errorf("failed to decode conclude batch ID: %w", err)
+	}
+	if err := entityqueue.ValidatePayloadQueue(msg, bid.Queue); err != nil {
+		return fmt.Errorf("invalid conclude message identity: %w", err)
+	}
+	if bid.ID == "" {
+		return fmt.Errorf("conclude dlq payload decoded to empty batch ID")
+	}
+	store, err := c.stores.For(orchstorage.Config{QueueName: bid.Queue})
+	if err != nil {
+		return fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err)
+	}
+	batch, err := store.GetBatchStore().Get(ctx, bid.ID)
+	if err != nil {
+		return fmt.Errorf("failed to get conclude batch %s: %w", bid.ID, err)
+	}
+	if batch.Queue != bid.Queue {
+		return fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID)
+	}
+	// A failed conclusion is not evidence of a failed merge.
+	if !batch.State.IsTerminal() {
+		return fmt.Errorf("cannot conclude non-terminal batch %s in state %q", batch.ID, batch.State)
+	}
+	if err := corebatch.EnsureRecord(ctx, store, batch); err != nil {
+		return err
+	}
+	return corebatch.ConcludeBatchRequests(ctx, store, c.registry, c.logger, c.metricsScope, batch, msg.Metadata[topickey.MetadataKeyFailureReason])
+}
+
+func (c *concludeController) Name() string                { return string(c.topicKey) }
+func (c *concludeController) TopicKey() consumer.TopicKey { return c.topicKey }
+func (c *concludeController) ConsumerGroup() string       { return c.consumerGroup }