From 376a313f163a07b0d64a0f34ddf64e70615d21fa Mon Sep 17 00:00:00 2001 From: sergeyb Date: Fri, 6 Nov 2026 22:50:46 +0000 Subject: [PATCH] fix(submitqueue): preserve conclude dead-letter outcomes Summary: Share primary conclusion's terminal member reconciliation with a dedicated conclude final handler. Preserve the batch outcome, repair terminal membership, and retain non-terminal, missing, malformed, or partially completed work. Test Plan: TestConcludeFinalBinding runs the actual pipeline final consumer and verifies log replay on success and Nack retention of incomplete obligations. TestDLQConcludeController_TerminalOutcomes checks all three outcome mappings; TestDLQConcludeController_ReplaysPartialFanout proves persisted diagnostics survive a member log failure and replay without overwriting another winner. Revert Plan: Pause conclude final processing before reverting. Retain final queue records and request logs; do not reinterpret stored terminal outcomes as failures. --- doc/rfc/submitqueue/workflow.md | 4 +- submitqueue/orchestrator/BUILD.bazel | 12 ++ .../controller/conclude/BUILD.bazel | 4 +- .../controller/conclude/conclude.go | 74 +--------- .../orchestrator/controller/dlq/BUILD.bazel | 2 + .../orchestrator/controller/dlq/README.md | 25 ++-- .../orchestrator/controller/dlq/batch.go | 4 +- .../orchestrator/controller/dlq/conclude.go | 84 +++++++++++ .../controller/dlq/conclude_test.go | 129 +++++++++++++++++ .../orchestrator/controller/dlq/dlq.go | 4 +- .../orchestrator/core/batch/BUILD.bazel | 6 + .../orchestrator/core/batch/conclude.go | 92 ++++++++++++ submitqueue/orchestrator/pipeline.go | 2 +- submitqueue/orchestrator/pipeline_test.go | 137 ++++++++++++++++++ 14 files changed, 489 insertions(+), 90 deletions(-) create mode 100644 submitqueue/orchestrator/controller/dlq/conclude.go create mode 100644 submitqueue/orchestrator/controller/dlq/conclude_test.go create mode 100644 submitqueue/orchestrator/core/batch/conclude.go diff --git a/doc/rfc/submitqueue/workflow.md b/doc/rfc/submitqueue/workflow.md index f07e8f191..2741234f9 100644 --- a/doc/rfc/submitqueue/workflow.md +++ b/doc/rfc/submitqueue/workflow.md @@ -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. diff --git a/submitqueue/orchestrator/BUILD.bazel b/submitqueue/orchestrator/BUILD.bazel index fff290a6b..b64b0e489 100644 --- a/submitqueue/orchestrator/BUILD.bazel +++ b/submitqueue/orchestrator/BUILD.bazel @@ -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", ], ) diff --git a/submitqueue/orchestrator/controller/conclude/BUILD.bazel b/submitqueue/orchestrator/controller/conclude/BUILD.bazel index dffa02c5e..5cc270f1f 100644 --- a/submitqueue/orchestrator/controller/conclude/BUILD.bazel +++ b/submitqueue/orchestrator/controller/conclude/BUILD.bazel @@ -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", diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go index 42e041e7f..ba7a91a38 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude.go +++ b/submitqueue/orchestrator/controller/conclude/conclude.go @@ -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) - } -} diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index 68374f9ad..4953293a5 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -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", diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index 0ebbb2556..35291a17f 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -1,14 +1,14 @@ # 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 @@ -16,9 +16,9 @@ DLQ consumers are wired with `errs.AlwaysRetryableProcessor`, unlimited attempts 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 @@ -26,14 +26,21 @@ When a DLQ controller transitions a request to `RequestStateError`, it publishes ## 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 diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go index 591c1669b..9376f7b99 100644 --- a/submitqueue/orchestrator/controller/dlq/batch.go +++ b/submitqueue/orchestrator/controller/dlq/batch.go @@ -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 { diff --git a/submitqueue/orchestrator/controller/dlq/conclude.go b/submitqueue/orchestrator/controller/dlq/conclude.go new file mode 100644 index 000000000..32de63bd3 --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/conclude.go @@ -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 } diff --git a/submitqueue/orchestrator/controller/dlq/conclude_test.go b/submitqueue/orchestrator/controller/dlq/conclude_test.go new file mode 100644 index 000000000..30c396fc0 --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/conclude_test.go @@ -0,0 +1,129 @@ +// 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" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + queue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap/zaptest" +) + +func concludeDelivery(t *testing.T, ctrl *gomock.Controller, reason string) consumer.Delivery { + t.Helper() + payload, err := sqmq.MarshalID(sqmq.TopicKeyConclude, "9", "q") + require.NoError(t, err) + message := queue.NewMessage("conclude", payload, "q", map[string]string{topickey.MetadataKeyFailureReason: reason}) + message.Tenant = "q" + delivery := consumermock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(message).AnyTimes() + delivery.EXPECT().Metadata().Return(map[string]string{"dlq.last_error": "conclude database unavailable"}).AnyTimes() + return delivery +} + +func TestDLQConcludeController_TerminalOutcomes(t *testing.T) { + for _, tt := range []struct { + batchState entity.BatchState + requestState entity.RequestState + status entity.RequestStatus + lastError string + }{ + {entity.BatchStateSucceeded, entity.RequestStateLanded, entity.RequestStatusLanded, ""}, + {entity.BatchStateFailed, entity.RequestStateError, entity.RequestStatusError, "merge conflict"}, + {entity.BatchStateCancelled, entity.RequestStateCancelled, entity.RequestStatusCancelled, ""}, + } { + t.Run(string(tt.batchState), func(t *testing.T) { + ctrl := gomock.NewController(t) + batch := entity.Batch{ID: "9", Queue: "q", State: tt.batchState, Version: 3, Contains: []string{"1"}} + batches := storagemock.NewMockBatchStore(ctrl) + batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + memberships := storagemock.NewMockQueueBatchStateStore(ctrl) + memberships.EXPECT().Put(gomock.Any(), entity.QueueBatchState{Queue: "q", State: tt.batchState, BatchID: "9"}).Return(nil) + request := entity.Request{ID: "1", Queue: "q", State: entity.RequestStateProcessing, Version: 2} + requests := storagemock.NewMockRequestStore(ctrl) + requests.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil) + requests.EXPECT().Update(gomock.Any(), requestWithState(request, tt.requestState), int32(2), int32(3)).Return(nil) + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batches) + store.EXPECT().GetQueueBatchStateStore().Return(memberships) + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + registry := newTestLogRegistry(t, ctrl, 1, func(log entity.RequestLog) error { + assert.Equal(t, tt.status, log.Status) + assert.Equal(t, tt.lastError, log.LastError) + assert.Equal(t, "9", log.Metadata["batch_id"]) + return nil + }) + controller := NewDLQConcludeController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store}, registry, TopicKey(topickey.TopicKeyConclude), "conclude-dlq") + require.NoError(t, controller.Process(context.Background(), concludeDelivery(t, ctrl, "merge conflict"))) + }) + } +} + +func TestDLQConcludeController_ReplaysPartialFanout(t *testing.T) { + ctrl := gomock.NewController(t) + batch := entity.Batch{ID: "9", Queue: "q", State: entity.BatchStateFailed, Version: 3, Contains: []string{"1", "2", "3"}} + batches := storagemock.NewMockBatchStore(ctrl) + batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).Times(2) + memberships := storagemock.NewMockQueueBatchStateStore(ctrl) + memberships.EXPECT().Put(gomock.Any(), entity.QueueBatchState{Queue: "q", State: batch.State, BatchID: "9"}).Return(nil).Times(2) + requests := storagemock.NewMockRequestStore(ctrl) + for _, id := range []string{"1", "2"} { + request := entity.Request{ID: id, Queue: "q", State: entity.RequestStateProcessing, Version: 1} + var persisted entity.Request + gomock.InOrder( + requests.EXPECT().Get(gomock.Any(), id).Return(request, nil), + requests.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).DoAndReturn( + func(_ context.Context, updated entity.Request, _, version int32) error { + persisted = updated + persisted.Version = version + return nil + }), + requests.EXPECT().Get(gomock.Any(), id).DoAndReturn(func(context.Context, string) (entity.Request, error) { return persisted, nil }), + ) + } + requests.EXPECT().Get(gomock.Any(), "3").Return(entity.Request{ID: "3", Queue: "q", State: entity.RequestStateCancelled, Version: 7}, nil) + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batches).AnyTimes() + store.EXPECT().GetQueueBatchStateStore().Return(memberships).AnyTimes() + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + var logs []entity.RequestLog + publishErr := errors.New("log unavailable") + registry := newTestLogRegistry(t, ctrl, 4, func(log entity.RequestLog) error { + logs = append(logs, log) + if len(logs) == 2 { + return publishErr + } + return nil + }) + controller := NewDLQConcludeController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store}, registry, TopicKey(topickey.TopicKeyConclude), "conclude-dlq") + require.ErrorIs(t, controller.Process(context.Background(), concludeDelivery(t, ctrl, "merge conflict")), publishErr) + require.NoError(t, controller.Process(context.Background(), concludeDelivery(t, ctrl, "later infrastructure error"))) + require.Len(t, logs, 4) + assert.Equal(t, logs[0], logs[2]) + assert.Equal(t, logs[1], logs[3]) + assert.Equal(t, "merge conflict", logs[3].LastError) +} diff --git a/submitqueue/orchestrator/controller/dlq/dlq.go b/submitqueue/orchestrator/controller/dlq/dlq.go index 31c0f7825..98ab771e0 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq.go +++ b/submitqueue/orchestrator/controller/dlq/dlq.go @@ -14,7 +14,9 @@ // Package dlq contains controllers that consume messages from per-topic // dead-letter queues and reconcile the affected request and batch entities -// into a terminal failed state. +// into a terminal failed state. Conclude instead preserves the authoritative +// terminal batch outcome and completes its matching member fan-out; non-terminal +// or missing records remain errors rather than evidence of a failed merge. // // Background. The consumer framework moves a message to its DLQ after the // controller for the original topic returns a non-retryable error or exhausts diff --git a/submitqueue/orchestrator/core/batch/BUILD.bazel b/submitqueue/orchestrator/core/batch/BUILD.bazel index 96c8699ed..a8c811921 100644 --- a/submitqueue/orchestrator/core/batch/BUILD.bazel +++ b/submitqueue/orchestrator/core/batch/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "conclude.go", "find.go", "list.go", "transition.go", @@ -10,10 +11,15 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/core/batch", visibility = ["//visibility:public"], deps = [ + "//platform/consumer:go_default_library", + "//platform/metrics:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", + "//submitqueue/orchestrator/core/request:go_default_library", "//submitqueue/orchestrator/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", "@org_golang_x_sync//errgroup:go_default_library", + "@org_uber_go_zap//:go_default_library", ], ) diff --git a/submitqueue/orchestrator/core/batch/conclude.go b/submitqueue/orchestrator/core/batch/conclude.go new file mode 100644 index 000000000..1af9c3298 --- /dev/null +++ b/submitqueue/orchestrator/core/batch/conclude.go @@ -0,0 +1,92 @@ +// Copyright (c) 2025 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 batch + +import ( + "context" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + corerequest "github.com/uber/submitqueue/submitqueue/orchestrator/core/request" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" + "go.uber.org/zap" +) + +// ConcludeBatchRequests reconciles each member with the authoritative terminal batch outcome. +// Matching terminal requests replay their persisted logs; different terminal winners remain unchanged. +// A missing member or failed write/publish is an error so redelivery retains unfinished work. +// failureReason applies only to Failed batches, not successful or cancelled outcomes. +func ConcludeBatchRequests(ctx context.Context, store orchstorage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, scope tally.Scope, batch entity.Batch, failureReason string) error { + requestState, err := batchStateToRequestState(batch.State) + if err != nil { + metrics.NamedCounter(scope, "process", "unexpected_state_errors", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("unexpected batch state %q for batch %s: %w", batch.State, batch.ID, err) + } + + if requestState != entity.RequestStateError { + failureReason = "" + } + for _, requestID := range batch.Contains { + res, err := corerequest.TerminateRequest(ctx, store, registry, requestID, requestState, failureReason, map[string]string{ + "batch_id": batch.ID, + }) + if err != nil { + metrics.NamedCounter(scope, "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: + logger.Infow("updated request state", + "batch_id", batch.ID, + "request_id", requestID, + "new_state", string(res.AfterState), + ) + case corerequest.TerminationOutcomeAlreadyInTargetState: + metrics.NamedCounter(scope, "process", "already_reconciled", 1, metrics.TagsFromContext(ctx)...) + case corerequest.TerminationOutcomeDiverged: + 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(scope, "process", "terminal_state_divergence", 1, metrics.TagsFromContext(ctx)...) + case corerequest.TerminationOutcomeNotFound: + metrics.NamedCounter(scope, "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 +} + +// 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) + } +} diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index ed955125a..aef1861c0 100644 --- a/submitqueue/orchestrator/pipeline.go +++ b/submitqueue/orchestrator/pipeline.go @@ -230,7 +230,7 @@ var Stages = []pipeline.Stage[Deps]{ return conclude.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { - return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + return dlq.NewDLQConcludeController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, }, // Any stage can publish here and this one publishes nothing onward, so the diff --git a/submitqueue/orchestrator/pipeline_test.go b/submitqueue/orchestrator/pipeline_test.go index 780586674..0ec9746c2 100644 --- a/submitqueue/orchestrator/pipeline_test.go +++ b/submitqueue/orchestrator/pipeline_test.go @@ -15,15 +15,30 @@ package orchestrator import ( + "context" + "errors" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/platform/base/failure" + queue "github.com/uber/submitqueue/platform/base/messagequeue" hookext "github.com/uber/submitqueue/platform/extension/hook" hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/platform/pipeline" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" + orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap/zaptest" ) @@ -78,3 +93,125 @@ func TestHookStage(t *testing.T) { require.Error(t, err) }) } + +type concludeStorageFactory struct{ store orchstorage.Storage } + +func (f concludeStorageFactory) For(orchstorage.Config) (orchstorage.Storage, error) { + return f.store, nil +} + +func TestConcludeFinalBinding(t *testing.T) { + cases := []struct { + name string + state entity.BatchState + payload []byte + tenant string + metadataQueue string + batchQueue string + batchError error + recordError error + missingMember bool + wantAck bool + }{ + {name: "succeeded replays member log", state: entity.BatchStateSucceeded, wantAck: true}, + {name: "missing batch retained", batchError: storage.ErrNotFound}, + {name: "missing member retained", state: entity.BatchStateFailed, missingMember: true}, + {name: "membership repair failure retained", state: entity.BatchStateCancelled, recordError: errors.New("membership unavailable")}, + {name: "malformed payload retained", payload: []byte("invalid")}, + {name: "empty batch ID retained", payload: []byte(`{"queue":"q"}`)}, + {name: "payload tenant mismatch retained", tenant: "other"}, + {name: "invalid envelope retained", metadataQueue: "other"}, + {name: "authoritative queue mismatch retained", batchQueue: "other"}, + {name: "unknown outcome retained", state: entity.BatchStateUnknown}, + {name: "creating retained", state: entity.BatchStateCreating}, + {name: "created retained", state: entity.BatchStateCreated}, + {name: "speculating retained", state: entity.BatchStateSpeculating}, + {name: "landing retained", state: entity.BatchStateLanding}, + {name: "cancelling retained", state: entity.BatchStateCancelling}, + } + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + var stage pipeline.Stage[Deps] + for _, candidate := range Stages { + if candidate.Key == topickey.TopicKeyConclude { + stage = candidate + } + } + require.NotNil(t, stage.DLQ) + batches := storagemock.NewMockBatchStore(ctrl) + memberships := storagemock.NewMockQueueBatchStateStore(ctrl) + requests := storagemock.NewMockRequestStore(ctrl) + batchQueue := tt.batchQueue + if batchQueue == "" { + batchQueue = "q" + } + batch := entity.Batch{ID: "9", Queue: batchQueue, State: tt.state, Contains: []string{"1"}, Version: 2} + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batches).AnyTimes() + store.EXPECT().GetQueueBatchStateStore().Return(memberships).AnyTimes() + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + if tt.payload == nil && tt.tenant == "" && tt.metadataQueue == "" { + batches.EXPECT().Get(gomock.Any(), "9").Return(batch, tt.batchError) + if tt.batchError == nil && tt.batchQueue == "" && tt.state.IsTerminal() { + memberships.EXPECT().Put(gomock.Any(), entity.QueueBatchState{Queue: "q", State: tt.state, BatchID: "9"}).Return(tt.recordError) + if tt.missingMember { + requests.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, storage.ErrNotFound) + } + if tt.wantAck { + requests.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{ID: "1", Queue: "q", State: entity.RequestStateLanded, Version: 3}, nil) + } + } + } + deliveries := make(chan extqueue.Delivery, 1) + subscriber := queuemock.NewMockSubscriber(ctrl) + subscriber.EXPECT().Subscribe(gomock.Any(), "conclude", gomock.Any()).Return(make(chan extqueue.Delivery), nil) + subscriber.EXPECT().Subscribe(gomock.Any(), "conclude_dlq", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, config extqueue.SubscriptionConfig) (<-chan extqueue.Delivery, error) { + assert.Zero(t, config.Retry.MaxAttempts) + assert.False(t, config.DLQ.Enabled) + return deliveries, nil + }) + backend := queuemock.NewMockQueue(ctrl) + backend.EXPECT().Subscriber().Return(subscriber).Times(2) + if tt.wantAck { + publisher := queuemock.NewMockPublisher(ctrl) + backend.EXPECT().Publisher().Return(publisher) + publisher.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, message queue.Message) error { + log, err := sqmq.UnmarshalRequestLog(message.Payload) + require.NoError(t, err) + assert.Equal(t, entity.RequestStatusLanded, log.Status) + return nil + }) + } + deps := Deps{Logger: zaptest.NewLogger(t).Sugar(), Scope: tally.NoopScope, Storage: concludeStorageFactory{store}} + component, err := pipeline.Construct(deps.Logger, deps.Scope, backend, "test", deps, []pipeline.Stage[Deps]{stage}, pipeline.PublishOnly(pipeline.PublishOnlyTopic{Key: topickey.TopicKeyLog, Name: "log"})) + require.NoError(t, err) + require.NoError(t, component.Start(context.Background())) + t.Cleanup(func() { require.NoError(t, component.Stop(context.Background())) }) + payload := tt.payload + if payload == nil { + payload, err = sqmq.MarshalID(sqmq.TopicKeyConclude, "9", "q") + require.NoError(t, err) + } + message := queue.NewMessage("conclude", payload, "q", map[string]string{queue.MetadataKeyQueueName: tt.metadataQueue}) + message.Tenant = tt.tenant + if message.Tenant == "" { + message.Tenant = "q" + } + delivery := queuemock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(message).AnyTimes() + delivery.EXPECT().Metadata().Return(nil).AnyTimes() + delivery.EXPECT().Attempt().Return(20).AnyTimes() + settled := make(chan struct{}) + if tt.wantAck { + delivery.EXPECT().Ack(gomock.Any()).DoAndReturn(func(context.Context) error { close(settled); return nil }) + } else { + delivery.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(func(context.Context, failure.Failure) error { close(settled); return nil }) + } + deliveries <- delivery + <-settled + }) + } +}