diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 9c3f367b..ec8bec44 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -695,10 +695,8 @@ func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger, return count, nil } -// registerDLQControllers creates one DLQ reconciler per primary stage and -// registers them with the DLQ consumer. Each reconciler drives the affected -// request or batch into a terminal Error/Failed state so the gateway stops -// reporting it as stuck-in-progress. +// registerDLQControllers creates one reconciler per primary stage. Normal DLQs +// fail the affected entity; conclude DLQ repairs its existing terminal outcome. func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, store storage.Storage) (int, error) { dlqScope := scope.SubScope("dlq") dlqRegs := []struct { @@ -716,7 +714,7 @@ func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scop {"buildsignal_dlq", dlq.NewDLQBuildSignalController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq")}, {"merge_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq")}, {"mergesignal_dlq", dlq.NewDLQMergeSignalController(logger, dlqScope, store, registry, dlq.TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq")}, - {"conclude_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyConclude), "orchestrator-conclude-dlq")}, + {"conclude_dlq", dlq.NewDLQConcludeController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyConclude), "orchestrator-conclude-dlq")}, } var count int for _, reg := range dlqRegs { diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index 08a52c22..9d6dd17c 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -45,6 +45,7 @@ go_test( "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", + "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index 300dbf55..23708889 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -1,51 +1,45 @@ # 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 affected request and batch entities into terminal states. 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 DLQ is the final destination.** A message in `{topic}_dlq` has already exhausted the primary controller's retry policy or surfaced as non-retryable. DLQ subscriptions disable their own DLQ, so a controller either reconciles the message or keeps retrying it. An unprocessable message must be removed by an operator. -**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. +**Reconcile state rather than rerun business work.** DLQ controllers decode the original payload, resolve the affected identity, and converge state and public request logs. They do not repeat external actions such as builds or merges. -**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. +**Repair partial materialization.** Request reconciliation republishes the terminal log when the state was already written. Failed batches repeat request fanout, and the dedicated conclude DLQ repairs fanout according to the batch's existing terminal outcome. ## Convergence guarantee -DLQ consumers are wired with `errs.AlwaysRetryableProcessor` and a very high `Retry.MaxAttempts` (currently 1000). Together with `DLQ.Enabled = false` on the DLQ subscription itself, this means any non-nil error returned from a DLQ controller — including a plain unclassified infra error — is forced retryable and redelivered rather than silently dropped. The combination is "always retryable + bounded-but-effectively-infinite attempts" and is the property the package relies on for convergence. +DLQ consumers use `errs.AlwaysRetryableProcessor`, a high retry limit, and `DLQ.Enabled = false`. Any non-nil controller error is therefore redelivered rather than silently dropped. -The recognised error condition is handled explicitly in `dlq.go`: +`storage.ErrNotFound` is logged and treated as success because there is no entity to reconcile. `storage.ErrVersionMismatch` retains its declaration-level retryable classification through reconciliation, while other DLQ errors are made retryable by `AlwaysRetryableProcessor`. -- `storage.ErrNotFound` → logged at warn and treated as success. The request or batch never persisted; there is nothing to reconcile. +## Request log entries -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. - -## 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. +When a DLQ terminates a request as `RequestStateError`, it publishes the matching log to the `log` topic. Gateway remains the sole writer of request logs and public projections. The log preserves `dlq.last_error` in `LastError` and records the original topic, failure count, and failure time as metadata when available. ## Controller mapping -Two controller shapes cover the eleven primary pipeline topics: - -| Controller | Topics | Decoded ID | Terminal state | +| Controller | Topics | Decoded identity | Reconciliation | |---|---|---|---| -| `NewDLQRequestController` | `start`, `validate`, `batch`, `cancel`, `log` | `RequestID` | `RequestStateError` | -| `NewDLQBatchController` | `score`, `speculate`, `build`, `merge`, `conclude` | `BatchID` | `BatchStateFailed` + fan-out to member requests as `RequestStateError` | - -`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. +| `NewDLQRequestController` | `start`, `validate`, `batch`, `cancel` | Request ID from the original payload | Request to `Error` | +| `NewDLQMergeConflictSignalController` | `mergeconflictsignal` | Request ID from Runway `MergeResult` | Request to `Error` | +| `NewDLQBatchController` | `score`, `speculate`, `build`, `merge` | `BatchID` | Batch to `Failed`, members to `Error` | +| `NewDLQBuildSignalController` | `buildsignal` | `BuildID`, then `BatchID` | Batch to `Failed`, members to `Error` | +| `NewDLQMergeSignalController` | `mergesignal` | Batch ID from Runway `MergeResult` | Batch to `Failed`, members to `Error` | +| `NewDLQConcludeController` | `conclude` | `BatchID` | Preserve terminal batch outcome and repair member fanout | ## Idempotency and concurrent activity -Reconciliation is safe to run more than once for the same message: - -- A request already in `RequestStateError` skips the state-transition CAS but republishes its terminal log so redelivery repairs an earlier publish failure. Requests in a different terminal state are left unchanged. -- A batch already in `BatchStateFailed` still fans out to member requests, because a previous attempt may have transitioned the batch but crashed before completing the fan-out. Batches in a different terminal state are left unchanged. -- Per-request fan-out is itself idempotent via `failRequest`. -- A request in `RequestStateCancelling` is reconciled to `RequestStateError`, not left in place: DLQ means the pipeline failed to converge, so we cannot confirm the cancel completed cleanly. Writing Error is the honest signal. +- A request already in the target terminal state skips the CAS and republishes its terminal log. +- A request in a different terminal state is left unchanged. +- A batch already in `BatchStateFailed` repeats member fanout. +- A request in `RequestStateCancelling` is reconciled to `RequestStateError` when its pipeline message reaches a normal DLQ. +- A conclude DLQ maps `Succeeded`, `Failed`, and `Cancelled` batches to `Landed`, `Error`, and `Cancelled` request outcomes respectively. ## See also -- `core/errs/README.md` — the error-processing framework, including `AlwaysRetryableProcessor` and the choice of processor for primary vs. DLQ consumers. -- `core/consumer/README.md` — how the consumer applies the processor to controller errors and decides ack/nack/reject. -- `doc/rfc/submitqueue/workflow.md` — the per-stage primary pipeline that the DLQ companions mirror. +- `platform/errs/README.md` describes error processing and `AlwaysRetryableProcessor`. +- `doc/rfc/submitqueue/workflow.md` describes the primary pipeline mirrored by these DLQ companions. diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go index 41309a48..59c2172b 100644 --- a/submitqueue/orchestrator/controller/dlq/batch.go +++ b/submitqueue/orchestrator/controller/dlq/batch.go @@ -26,25 +26,20 @@ import ( "go.uber.org/zap" ) -// batchController is the DLQ reconciler for batch-scoped pipeline stages -// (score, speculate, build, merge, conclude). All five topics carry a -// BatchID payload, so this controller is registered five times — one per -// topic, each with the matching DLQ topic key and consumer group. -// -// On each delivery the controller decodes the BatchID, transitions the batch -// to BatchStateFailed (idempotent if already halted), and fans out by -// transitioning each member request to RequestStateError. The fan-out exists -// because conclude — which normally drives request state from batch state — -// will not run for a DLQ'd batch. +// batchController handles DLQ messages whose payload is a BatchID. Its +// reconciler selects failure handling or conclude fanout repair. type batchController struct { logger *zap.SugaredLogger metricsScope tally.Scope store storage.Storage registry consumer.TopicRegistry + reconcile batchReconciler topicKey consumer.TopicKey consumerGroup string } +type batchReconciler func(context.Context, storage.Storage, consumer.TopicRegistry, *zap.SugaredLogger, string, map[string]string) error + // Verify batchController implements consumer.Controller at compile time. var _ consumer.Controller = (*batchController)(nil) @@ -57,6 +52,31 @@ func NewDLQBatchController( registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, +) consumer.Controller { + return newBatchController(logger, scope, store, registry, topicKey, consumerGroup, failBatch) +} + +// NewDLQConcludeController builds a controller that preserves the batch's +// terminal outcome while repairing request fanout. +func NewDLQConcludeController( + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + registry consumer.TopicRegistry, + topicKey consumer.TopicKey, + consumerGroup string, +) consumer.Controller { + return newBatchController(logger, scope, store, registry, topicKey, consumerGroup, concludeBatch) +} + +func newBatchController( + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + registry consumer.TopicRegistry, + topicKey consumer.TopicKey, + consumerGroup string, + reconcile batchReconciler, ) consumer.Controller { name := string(topicKey) + "_controller" return &batchController{ @@ -64,6 +84,7 @@ func NewDLQBatchController( metricsScope: scope.SubScope(name), store: store, registry: registry, + reconcile: reconcile, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -97,7 +118,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failBatch(ctx, c.store, c.registry, c.logger, bid.ID, dmeta["dlq.last_error"]); err != nil { + if err := c.reconcile(ctx, c.store, c.registry, c.logger, bid.ID, dmeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go index 58a6c5f2..952664c6 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go @@ -121,7 +121,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D return nil } - if err := failBatch(ctx, c.store, c.registry, c.logger, build.BatchID, dmeta["dlq.last_error"]); err != nil { + if err := failBatch(ctx, c.store, c.registry, c.logger, build.BatchID, dmeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/dlq.go b/submitqueue/orchestrator/controller/dlq/dlq.go index 5de49810..63f1315a 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq.go +++ b/submitqueue/orchestrator/controller/dlq/dlq.go @@ -14,7 +14,7 @@ // 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 terminal states. // // Background. The consumer framework moves a message to its DLQ after the // controller for the original topic returns a non-retryable error or exhausts @@ -26,11 +26,10 @@ // Reconciliation strategy. Each DLQ topic carries the same payload as its // originating topic (the queue framework preserves the bytes verbatim under a // new `{topic}_dlq` name). The DLQ controllers decode that payload to recover -// the affected request or batch, then transition it to a terminal failed -// state — Error for requests, Failed for batches — with an idempotent -// optimistic-locking write so concurrent activity (a late merge, a cancel -// race) wins cleanly. Batch failures also fan out to the member requests so -// the gateway no longer reports them as in-progress. +// the affected request or batch. Normal pipeline DLQs reconcile requests to +// Error and batches to Failed. The conclude DLQ preserves the terminal batch +// outcome while repairing request fanout. Optimistic locking and idempotent log +// publication let concurrent activity win cleanly. package dlq import ( @@ -61,19 +60,37 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey { return consumer.TopicKey(string(main) + topicSuffix) } -// failRequest transitions a non-terminal request to RequestStateError and -// appends the matching RequestStatusError log. Redelivery for an existing Error -// state repeats materialization to repair a previous partial attempt. A -// different terminal outcome is left unchanged. -// lastError is the failure reason preserved by the queue in DLQ delivery -// metadata and is exposed through Status and History for diagnosis. +func dlqFailureOutcome(metadata map[string]string) requestcore.TerminalOutcome { + logMetadata := make(map[string]string, 3) + for _, key := range []string{"dlq.original_topic", "dlq.failure_count", "dlq.failed_at"} { + if value := metadata[key]; value != "" { + logMetadata[key] = value + } + } + return requestcore.TerminalOutcome{ + State: entity.RequestStateError, + LastError: metadata["dlq.last_error"], + Metadata: logMetadata, + } +} + +// reconcileRequest converges a request on the caller-selected terminal outcome. +// Redelivery for the same terminal state republishes the log to repair a +// previous partial attempt. A different terminal outcome is left unchanged. // -// A request in RequestStateCancelling is reconciled to RequestStateError, not -// left in place: DLQ means the pipeline failed to converge, so we cannot -// confirm the cancel completed cleanly. Writing Error is the honest signal and -// keeps the request from being stuck in a non-terminal state forever. -func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string) error { - request, err := store.GetRequestStore().Get(ctx, requestID) +// Normal pipeline DLQs select RequestStateError for a request in +// RequestStateCancelling because the failed pipeline cannot confirm that +// cancellation completed cleanly. +func reconcileRequest( + ctx context.Context, + store storage.Storage, + registry consumer.TopicRegistry, + logger *zap.SugaredLogger, + requestID string, + outcome requestcore.TerminalOutcome, +) error { + requestStore := store.GetRequestStore() + request, err := requestStore.Get(ctx, requestID) if err != nil { if errors.Is(err, storage.ErrNotFound) { logger.Warnw("dlq reconcile: request not found, skipping", @@ -84,39 +101,24 @@ func failRequest(ctx context.Context, store storage.Storage, registry consumer.T return fmt.Errorf("failed to get request %s: %w", requestID, err) } - logVersion := request.Version - switch request.State { - case entity.RequestStateError: - logger.Infow("dlq reconcile: request already failed, republishing terminal log", - "request_id", requestID, - ) - case entity.RequestStateLanded, entity.RequestStateCancelled: + reconciled, err := requestcore.ReconcileTerminalState( + ctx, + requestStore, + registry, + request, + outcome, + ) + if err != nil { + return fmt.Errorf("failed to reconcile request %s: %w", requestID, err) + } + if !reconciled { logger.Infow("dlq reconcile: request has a different terminal outcome, skipping", "request_id", requestID, "state", string(request.State), - ) - return nil - default: - newVersion := request.Version + 1 - if err := store.GetRequestStore().UpdateState(ctx, requestID, request.Version, newVersion, entity.RequestStateError); err != nil { - return fmt.Errorf("failed to update request %s state to error: %w", requestID, err) - } - logVersion = newVersion - logger.Infow("dlq reconcile: request marked terminal error", - "request_id", requestID, - "previous_state", string(request.State), + "target_state", string(outcome.State), ) } - // Publish the terminal Error status through the log topic so Gateway remains - // the sole writer of request logs and public projections. An existing Error - // state republishes the same logical event so a previous attempt that changed - // the entity but failed to publish can be repaired. - logEntry := entity.NewRequestLog(requestID, entity.RequestStatusError, logVersion, lastError, nil) - if err := requestcore.PublishLog(ctx, registry, logEntry, requestID); err != nil { - return fmt.Errorf("failed to publish request log for %s: %w", requestID, err) - } - return nil } @@ -124,18 +126,24 @@ func failRequest(ctx context.Context, store storage.Storage, registry consumer.T // terminal state, then fans out by transitioning each member request to // RequestStateError. The fan-out mirrors what the conclude controller would do // for a normally-completed batch, but skips re-publishing to the conclude -// topic — for DLQ messages there is no guarantee that conclude would ever run, -// so the reconciliation has to drive each request directly. +// topic. For DLQ messages there is no guarantee that conclude would ever run, +// so reconciliation drives each request directly. // -// A batch in BatchStateCancelling is reconciled to BatchStateFailed for the -// same reason failRequest reconciles Cancelling requests: DLQ means we cannot -// confirm the cancel completed, so the batch must reach a terminal state. +// A batch in BatchStateCancelling is reconciled to BatchStateFailed because the +// failed pipeline cannot confirm that cancellation completed cleanly. // // Idempotency: an existing Failed batch repeats fan-out because a previous // attempt may have crashed after updating the batch. Succeeded and Cancelled // are different terminal outcomes and do not fan out errors. -// lastError is propagated to each member request's terminal Error log. -func failBatch(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string) error { +// outcome is propagated to each member request's terminal Error log. +func failBatch( + ctx context.Context, + store storage.Storage, + registry consumer.TopicRegistry, + logger *zap.SugaredLogger, + batchID string, + metadata map[string]string, +) error { batch, err := store.GetBatchStore().Get(ctx, batchID) if err != nil { if errors.Is(err, storage.ErrNotFound) { @@ -169,10 +177,58 @@ func failBatch(ctx context.Context, store storage.Storage, registry consumer.Top ) } + outcome := dlqFailureOutcome(metadata) for _, requestID := range batch.Contains { - if err := failRequest(ctx, store, registry, logger, requestID, lastError); err != nil { + if err := reconcileRequest(ctx, store, registry, logger, requestID, outcome); err != nil { return fmt.Errorf("fan-out for batch %s: %w", batchID, err) } } return nil } + +// concludeBatch preserves a terminal batch outcome and repairs request fanout. +func concludeBatch( + ctx context.Context, + store storage.Storage, + registry consumer.TopicRegistry, + logger *zap.SugaredLogger, + batchID string, + _ map[string]string, +) error { + batch, err := store.GetBatchStore().Get(ctx, batchID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + logger.Warnw("dlq reconcile: batch not found, skipping", "batch_id", batchID) + return nil + } + return fmt.Errorf("failed to get batch %s: %w", batchID, err) + } + + requestState, err := terminalRequestState(batch.State) + if err != nil { + return fmt.Errorf("cannot reconcile conclude dlq for batch %s: %w", batchID, err) + } + outcome := requestcore.TerminalOutcome{ + State: requestState, + Metadata: map[string]string{"batch_id": batch.ID}, + } + for _, requestID := range batch.Contains { + if err := reconcileRequest(ctx, store, registry, logger, requestID, outcome); err != nil { + return fmt.Errorf("conclude fan-out for batch %s: %w", batchID, err) + } + } + return nil +} + +func terminalRequestState(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("batch state %s is not terminal", state) + } +} diff --git a/submitqueue/orchestrator/controller/dlq/dlq_test.go b/submitqueue/orchestrator/controller/dlq/dlq_test.go index ad8f6ff3..9cb40dcb 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq_test.go +++ b/submitqueue/orchestrator/controller/dlq/dlq_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" + requestcore "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" @@ -31,9 +32,9 @@ import ( "go.uber.org/zap/zaptest" ) -// failRequest +// reconcileRequest -func TestFailRequest_TerminalStates(t *testing.T) { +func TestReconcileRequest_TerminalStates(t *testing.T) { tests := []struct { state entity.RequestState wantLog bool @@ -64,18 +65,18 @@ func TestFailRequest_TerminalStates(t *testing.T) { }) } - err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := reconcileRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", testErrorOutcome()) require.NoError(t, err) }) } } -// TestFailRequest_CancellingTransitionsToError verifies that a request stuck in -// the non-terminal Cancelling state is reconciled to Error. If failRequest +// TestReconcileRequest_CancellingTransitionsToError verifies that a request stuck in +// the non-terminal Cancelling state is reconciled to Error. If reconcileRequest // short-circuited on Cancelling the request would remain in-progress forever, // because the cancel pipeline that owns the Cancelling → Cancelled transition // has itself died (that's why we're in the DLQ). -func TestFailRequest_CancellingTransitionsToError(t *testing.T) { +func TestReconcileRequest_CancellingTransitionsToError(t *testing.T) { ctrl := gomock.NewController(t) requestStore := storagemock.NewMockRequestStore(ctrl) @@ -94,11 +95,11 @@ func TestFailRequest_CancellingTransitionsToError(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := reconcileRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", testErrorOutcome()) require.NoError(t, err) } -func TestFailRequest_TransitionsToError(t *testing.T) { +func TestReconcileRequest_TransitionsToError(t *testing.T) { ctrl := gomock.NewController(t) requestStore := storagemock.NewMockRequestStore(ctrl) @@ -117,14 +118,14 @@ func TestFailRequest_TransitionsToError(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := reconcileRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", testErrorOutcome()) require.NoError(t, err) } -// TestFailRequest_LogPublishErrorPropagates verifies that a terminal log +// TestReconcileRequest_LogPublishErrorPropagates verifies that a terminal log // publish failure is surfaced so the always-retryable processor redelivers the // DLQ message. -func TestFailRequest_LogPublishErrorPropagates(t *testing.T) { +func TestReconcileRequest_LogPublishErrorPropagates(t *testing.T) { ctrl := gomock.NewController(t) requestStore := storagemock.NewMockRequestStore(ctrl) @@ -140,11 +141,11 @@ func TestFailRequest_LogPublishErrorPropagates(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := reconcileRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", testErrorOutcome()) require.Error(t, err) } -func TestFailRequest_NotFoundIsNoOp(t *testing.T) { +func TestReconcileRequest_NotFoundIsNoOp(t *testing.T) { ctrl := gomock.NewController(t) requestStore := storagemock.NewMockRequestStore(ctrl) @@ -153,11 +154,11 @@ func TestFailRequest_NotFoundIsNoOp(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := reconcileRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", testErrorOutcome()) require.NoError(t, err) } -func TestFailRequest_GenericGetErrorIsNonRetryable(t *testing.T) { +func TestReconcileRequest_GenericGetErrorIsNonRetryable(t *testing.T) { ctrl := gomock.NewController(t) requestStore := storagemock.NewMockRequestStore(ctrl) @@ -166,7 +167,7 @@ func TestFailRequest_GenericGetErrorIsNonRetryable(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "") + err := reconcileRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", testErrorOutcome()) require.Error(t, err) assert.False(t, errs.IsRetryable(err)) } @@ -201,7 +202,7 @@ func TestFailBatch_TransitionsAndFansOut(t *testing.T) { store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", nil) require.NoError(t, err) } @@ -229,7 +230,7 @@ func TestFailBatch_FailedFansOutForRepair(t *testing.T) { store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", nil) require.NoError(t, err) } @@ -245,7 +246,7 @@ func TestFailBatch_DifferentTerminalOutcomeSkipsFanOut(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", nil) require.NoError(t, err) }) } @@ -281,7 +282,7 @@ func TestFailBatch_CancellingTransitionsToFailed(t *testing.T) { store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", nil) require.NoError(t, err) } @@ -294,10 +295,69 @@ func TestFailBatch_NotFoundIsNoOp(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", nil) require.NoError(t, err) } +func TestConcludeBatch_PreservesOutcome(t *testing.T) { + tests := []struct { + state entity.BatchState + wantState entity.RequestState + wantStatus entity.RequestStatus + wantErr bool + }{ + {state: entity.BatchStateSucceeded, wantState: entity.RequestStateLanded, wantStatus: entity.RequestStatusLanded}, + {state: entity.BatchStateFailed, wantState: entity.RequestStateError, wantStatus: entity.RequestStatusError}, + {state: entity.BatchStateCancelled, wantState: entity.RequestStateCancelled, wantStatus: entity.RequestStatusCancelled}, + {state: entity.BatchStateMerging, wantErr: true}, + } + + for _, tt := range tests { + t.Run(string(tt.state), func(t *testing.T) { + ctrl := gomock.NewController(t) + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), "q/batch/1").Return(entity.Batch{ + ID: "q/batch/1", Contains: []string{"q/1"}, State: tt.state, + }, nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore) + registry := consumer.TopicRegistry{} + if !tt.wantErr { + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.Request{ + ID: "q/1", State: entity.RequestStateProcessing, Version: 2, + }, nil) + requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(2), int32(3), tt.wantState).Return(nil) + store.EXPECT().GetRequestStore().Return(requestStore) + registry = newTestLogRegistry(t, ctrl, 1, func(log entity.RequestLog) error { + assert.Equal(t, tt.wantStatus, log.Status) + assert.Equal(t, map[string]string{"batch_id": "q/batch/1"}, log.Metadata) + return nil + }) + } + + err := concludeBatch( + context.Background(), + store, + registry, + zaptest.NewLogger(t).Sugar(), + "q/batch/1", + nil, + ) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + }) + } +} + +func testErrorOutcome() requestcore.TerminalOutcome { + return requestcore.TerminalOutcome{State: entity.RequestStateError} +} + // TopicKey func TestDLQTopicKey(t *testing.T) { diff --git a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go index 26f0fa7f..6fae783c 100644 --- a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go @@ -29,7 +29,7 @@ import ( // mergeConflictSignalController is the DLQ reconciler for the // mergeconflictsignal topic. Its payload carries a runway // MergeResult whose id is the request id echoed back, so -// reconciliation fails that request directly via failRequest. +// reconciliation fails that request directly via reconcileRequest. type mergeConflictSignalController struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -87,7 +87,7 @@ func (c *mergeConflictSignalController) Process(ctx context.Context, delivery co "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failRequest(ctx, c.store, c.registry, c.logger, result.Id, dmeta["dlq.last_error"]); err != nil { + if err := reconcileRequest(ctx, c.store, c.registry, c.logger, result.Id, dlqFailureOutcome(dmeta)); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/mergesignal.go b/submitqueue/orchestrator/controller/dlq/mergesignal.go index fff67ce1..9fac2bd4 100644 --- a/submitqueue/orchestrator/controller/dlq/mergesignal.go +++ b/submitqueue/orchestrator/controller/dlq/mergesignal.go @@ -86,7 +86,7 @@ func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.D "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failBatch(ctx, c.store, c.registry, c.logger, result.Id, dmeta["dlq.last_error"]); err != nil { + if err := failBatch(ctx, c.store, c.registry, c.logger, result.Id, dmeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go index b00244be..badecbad 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -137,7 +137,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failRequest(ctx, c.store, c.registry, c.logger, requestID, dmeta["dlq.last_error"]); err != nil { + if err := reconcileRequest(ctx, c.store, c.registry, c.logger, requestID, dlqFailureOutcome(dmeta)); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go index 17fc0eab..4e5dc38f 100644 --- a/submitqueue/orchestrator/controller/dlq/request_test.go +++ b/submitqueue/orchestrator/controller/dlq/request_test.go @@ -102,6 +102,10 @@ func TestDLQRequestController_Process_RequestIDPayload(t *testing.T) { registry := newTestLogRegistry(t, ctrl, 1, func(log entity.RequestLog) error { assert.Equal(t, "boom", log.LastError) + assert.Equal(t, map[string]string{ + "dlq.original_topic": "validate", + "dlq.failure_count": "3", + }, log.Metadata) return nil })