From 2c221c87da962d5e6aa6595f9b05d0c82e70a1f3 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Fri, 13 Nov 2026 21:30:15 +0000 Subject: [PATCH] fix(submitqueue): preserve conclude dead-letter outcomes Summary: Reconcile a terminal batch's member requests and status logs when conclusion cannot finish. Preserve successful, failed, and cancelled outcomes, repair terminal membership, and retain incomplete payload obligations for retry. Test Plan: Existing tests cover all terminal outcome mappings, partial member log publication retries, different terminal winners, and final queue retention of incomplete work. Focused unit tests, affected library builds, formatting, lint, mock, module, and build metadata checks passed. Revert Plan: Pause conclude final processing before reverting. Preserve 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 | 11 +- .../orchestrator/controller/dlq/batch.go | 4 +- .../orchestrator/controller/dlq/conclude.go | 84 +++++++++++ .../controller/dlq/conclude_test.go | 131 +++++++++++++++++ .../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 | 136 ++++++++++++++++++ 14 files changed, 484 insertions(+), 82 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 c681a40f4..6cadf5a99 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,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. 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 523d4fbcf..1fcbae35b 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -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. 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..b09487469 --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/conclude_test.go @@ -0,0 +1,131 @@ +// 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) + for i, requestID := range []string{"1", "2", "1", "2"} { + assert.Equal(t, requestID, logs[i].RequestID) + assert.Equal(t, entity.RequestStatusError, logs[i].Status) + assert.Equal(t, int32(2), logs[i].RequestVersion) + } +} 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..a28768e7f --- /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 repair their status 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..e1882d465 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,124 @@ 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: "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 + }) + } +}