sbalabanov · Nov 6, 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,7 +162,7 @@ Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscriptio
 
 Most DLQ controllers do not re-attempt the failed work. They decode the payload to a `RequestID` or `BatchID` and drive that entity to a terminal failed state — `RequestStateError` for requests, `BatchStateFailed` for batches, with fan-out to the member requests. Topics that carry a full payload recover the id from it: the `landconflictsignal` DLQ reads the request id from Runway's `MergeResult`, and the `landsignal` DLQ reads the batch id. The `buildsignal` DLQ decodes a build id, loads that build, and fails the batch that owns it. A missing build or an empty batch id is acked: there is no batch to fail from this signal. State writes use the same optimistic-locking CAS as the primary pipeline, so a late primary-pipeline update wins cleanly and a version mismatch is asked back for redelivery.
 
-Two consumed stages do not follow that shape. The speculate DLQ fails the batches the failure attributes — often not the batch named on the message, because a run re-plans the whole queue — and then republishes to `speculate`, so one dead letter does not strand every other batch. The hook DLQ records the dropped event and acks. A hook never writes pipeline state, so there is no request or batch to fail; republishing the logged event is how an operator recovers the side effect. A genuinely unprocessable request or batch DLQ message, typically a malformed payload, must be removed by an operator, because those consumers retry until reconciliation succeeds.
+Three consumed stages do not follow that shape. 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 identities, and partial writes/publishes remain in the final queue until resolved. Failed members receive the original message’s failure reason, never an incidental DLQ infrastructure error; matching terminal members replay persisted diagnostics and different terminal winners remain unchanged. The speculate DLQ fails the batches the failure attributes — often not the batch named on the message, because a run re-plans the whole queue — and then republishes to `speculate`, so one dead letter does not strand every other batch. The hook DLQ records the dropped event and acks. A hook never writes pipeline state, so there is no request or batch to fail; republishing the logged event is how an operator recovers the side effect. A genuinely unprocessable request or batch DLQ message, typically a malformed payload, must be removed by an operator, because those consumers retry until reconciliation succeeds.
 
 See `submitqueue/orchestrator/controller/dlq/README.md` for the reconcile-only constraints. The speculate and hook controllers in that wiring are the exceptions to the generic request/batch mapping that README's table still summarizes.
 

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

@@ -1,39 +1,46 @@
 # DLQ Reconciliation Controllers
 
-This package contains the controllers that drain each primary pipeline topic's `{topic}_dlq` companion and reconcile the affected request or batch into a terminal failed state. They are wired alongside the primary controllers in `service/submitqueue/orchestrator/server/main.go`.
+This package contains the controllers that drain each primary pipeline topic's `{topic}_dlq` companion and reconcile the affected request or batch. Most stages fail unfinished work; conclude preserves the authoritative terminal batch outcome and completes its request fan-out. They are wired alongside the primary controllers in `service/submitqueue/orchestrator/server/main.go`.
 
 ## Design principles
 
 **The DLQ is the final destination.** A message in `{topic}_dlq` has already failed the primary controller's retry budget (`Retry.MaxAttempts` for a retryable error) or surfaced as non-retryable on the first attempt. There is no further dead-letter beyond this point — every DLQ subscription is configured with `DLQ.Enabled = false`, so a DLQ controller's outcome is either "reconciled" or "the message keeps coming back". A genuinely unprocessable DLQ message (e.g. a malformed payload that no controller can ever decode) must be removed by an operator from the queue manually.
 
-**The implementation is deliberately the simplest and most reliable thing that can work.** Each controller does the same three steps: decode the payload to recover the affected `RequestID` or `BatchID`, fetch the entity, transition it to its terminal failed state with the same optimistic-locking CAS the primary pipeline uses. No business logic, no branching on the original error, no per-stage cleanup. Adding behaviour here trades off the one property the DLQ has to provide — convergence.
+**The implementation is deliberately the simplest and most reliable thing that can work.** Most controllers follow three steps: decode the payload to recover the affected `RequestID` or `BatchID`, fetch the entity, transition it to its terminal failed state with the same optimistic-locking CAS the primary pipeline uses. Conclude instead requires a terminal batch and reuses primary conclusion’s outcome mapping; a failed conclusion is not evidence of a failed merge. Adding behaviour here trades off the one property the DLQ has to provide — convergence.
 
-**Reconcile only; do not attempt to recover.** A DLQ controller never tries to re-run the failed stage, retry the original action against an external dependency, or repair a partially-completed transition. The original controller already failed; the DLQ controller's only job is to make sure the request and batch state stop saying "in progress" and start saying "failed", so the gateway reports an honest answer and downstream tooling can move on. Recovery, when it is appropriate, is a separate concern handled by an operator or a future reconciliation job — not by this code path.
+**Reconcile only; do not attempt to recover.** A DLQ controller never tries to re-run the failed stage, retry the original action against an external dependency, or repeat an external action. Conclude does repair terminal membership and partial request/log fan-out; this is completion of the stored outcome, not recovery of the merge. The original controller already failed; reconciliation makes the request and batch state reflect a justified terminal outcome, so the gateway reports an honest answer and downstream tooling can move on. Recovery, when it is appropriate, is a separate concern handled by an operator or a future reconciliation job — not by this code path.
 
 ## Convergence guarantee
 
 DLQ consumers are wired with `errs.AlwaysRetryableProcessor`, unlimited attempts (`Retry.MaxAttempts = 0`), and `DLQ.Enabled = false` on the DLQ subscription itself. Any non-nil error returned from a DLQ controller — including a plain unclassified infra error — is therefore forced retryable and redelivered rather than silently dropped. This is the convergence guarantee the package relies on.
 
 The recognised error condition is handled explicitly in `dlq.go`:
 
-- `storage.ErrNotFound` → logged at warn and treated as success. The request or batch never persisted; there is nothing to reconcile.
+- `storage.ErrNotFound` is generally logged and treated as success. Conclude is stricter: a missing batch or member leaves a known conclusion obligation incomplete, so it returns an error and retains the delivery.
 
-Everything else — including `storage.ErrVersionMismatch` on the CAS — is returned without controller-level classification and redelivered until it either succeeds or hits the attempt cap. `ErrVersionMismatch` is already retryable at its declaration, while the always-retryable processor makes every other non-nil reconciliation error retryable too.
+Everything else — including `storage.ErrVersionMismatch` on the CAS — is returned without controller-level classification and redelivered until reconciliation succeeds. `ErrVersionMismatch` is already retryable at its declaration, while the always-retryable processor makes every other non-nil reconciliation error retryable too.
 
 ## Request log entries are published to Gateway
 
 When a DLQ controller transitions a request to `RequestStateError`, it publishes the terminal request log to the `log` topic. Gateway consumes that topic and remains the sole writer of the request log and public projections. A publish failure leaves the DLQ delivery unacknowledged so redelivery retries the same idempotent log event.
 
 ## Controller mapping
 
-Two controller shapes cover the eleven primary pipeline topics:
+The request and batch controllers cover most primary topics; conclude has its own final reconciler:
 
 | 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` |
+| `NewDLQRequestController` | `start`, `validate`, `batch`, `cancel` | `RequestID` | `RequestStateError` |
+| `NewDLQBatchController` | `dependencyanalysis`, `build`, `land` | `BatchID` | `BatchStateFailed` + fan-out to member requests as `RequestStateError` |
+| `NewDLQConcludeController` | `conclude` | `BatchID` | Preserve Succeeded → Landed, Failed → Error, Cancelled → Cancelled |
 
-`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. 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 decoders use the original topic key, not the final binding. The package-level `RequestIDDecoder` interface plus `DecodeLandRequestID` / `DecodeCancelRequestID` / `DecodeRequestID` cover the three payload shapes used by request-scoped topics.
+
+## 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 message identities. The same final delivery retries until member writes and log publication complete; it does not republish a fresh conclude message.
+
+Failed batches use the original message’s `failure_reason` metadata, not a later DLQ infrastructure error. An unavailable historical reason remains empty. Newly landed or cancelled members carry no LastError. Already-matching terminal members replay their persisted winning diagnostics; a different terminal winner is logged and preserved without overwriting it. Queue-wide dependent reconsideration is outside conclusion’s fan-out contract.
 
 ## Idempotency and concurrent activity