diff --git a/doc/rfc/submitqueue/workflow.md b/doc/rfc/submitqueue/workflow.md index 6706cdcdc..f82601586 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. Most DLQ reconcilers force a failed terminal state when a consumed stage cannot finish; start resumes the original full-payload work instead. +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; start and cancel resume their original full-payload work instead. 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. @@ -140,7 +140,7 @@ class node_submitter toneIndigo | **gateway/Land** | RPC | start | Mint the request id, persist an accepting receipt, publish the full land request, then persist Accepted | | **gateway/Cancel** | RPC | cancel | Persist `cancelling` for an existing land, then publish the cancel request | | **start** | LandRequest | validate, log | Create/load the authoritative Request; only Started replays log and validate | -| **cancel** | CancelRequest | log, or speculate | Record Cancelling. Finish a request that has no applicable batch. Hand each cancellable batch attempt to speculate. Leave a Landing or already-terminal batch for conclude | +| **cancel** | CancelRequest | log, or speculate | Hold a missing Request without spending failure budget or expiring, then record Cancelling. Finish a request that has no applicable batch. Hand each cancellable batch attempt to speculate. Leave a Landing or already-terminal batch for conclude | | **validate** | RequestID | merge-conflict-check (Runway), log | Dedup, fetch change metadata, claim changes, then publish the full `MergeRequest` to Runway keyed by the request id | | **landconflictsignal** | MergeResult | batch, or log | Correlate Runway's check; advance a landable request to batch, or fail a conflicted request | | **batch** | RequestID | dependency-analysis, log | Mint a Creating batch for the request and hand that batch id onward | @@ -162,9 +162,9 @@ 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. -Three consumed stages do not follow that shape. The start DLQ uses the start controller to create or resume the original full-payload request in place, without returning to the finite-budget primary subscription. After Create or AlreadyExists it loads the authoritative snapshot and verifies the queue, ID, ordered URIs, and strategy. Started replays its log and normal validate hand-off; later states and Cancelling preserve their current owner, and terminal states replay their stored winning terminal log. Conflicting inputs, unknown states, failed reads, and malformed payloads remain unacknowledged under the final retry policy. This reconciliation read is not a fence against already-running later stages. 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. +Four consumed stages do not follow that shape. The start DLQ uses the start controller to create or resume the original full-payload request in place, without returning to the finite-budget primary subscription. After Create or AlreadyExists it loads the authoritative snapshot and verifies the queue, ID, ordered URIs, and strategy. Started replays its log and normal validate hand-off; later states and Cancelling preserve their current owner, and terminal states replay their stored winning terminal log. Conflicting inputs, unknown states, failed reads, and malformed payloads remain unacknowledged under the final retry policy. This reconciliation read is not a fence against already-running later stages. The cancel DLQ uses the cancel controller with the original reason and cancellation semantics. Start and cancel use independent topics, so cancellation can arrive before the Request exists; either cancel subscription holds that same delivery until the row appears, without a synthetic Error or an age-based expiry. An ID that never appears remains a visible durable obligation for operator investigation. Cancellation remains best-effort once an attempt reaches Landing and does not fence batch claims or promotions. 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. +See `submitqueue/orchestrator/controller/dlq/README.md` for the reconcile-only constraints. The start, cancel, speculate, and hook controllers in that wiring are the exceptions to the generic request/batch mapping that README's table still summarizes. ## Ownership by service diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index c867f575e..d3a2191f8 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/submitqueue/orchestrator/controller/cancel/cancel.go @@ -45,13 +45,15 @@ // Concurrent producers surface as the intrinsically retryable // storage.ErrVersionMismatch; the controller returns the wrapped error as-is // so the next attempt sees the new state and takes the other branch. -// storage.ErrNotFound on the initial Get (the start -// controller has not yet persisted the request) is returned as-is for the -// same reason. +// A missing request on the initial read holds the original delivery without +// spending failure budget or expiring: independent cancel and start topics can +// overtake one another. An ID that never appears remains an operator-visible +// obligation, not a synthetic terminal outcome. package cancel import ( "context" + "errors" "fmt" "github.com/uber-go/tally" @@ -82,7 +84,10 @@ type Controller struct { // Verify Controller implements consumer.Controller interface at compile time. var _ consumer.Controller = (*Controller)(nil) -const opName = "process" +const ( + opName = "process" + waitForRequestDelayMs int64 = 1000 +) // NewController creates a new cancel controller for the orchestrator. func NewController( @@ -104,7 +109,7 @@ func NewController( } // Process processes a cancel delivery from the queue. -// Returns nil to ack (success), or error to nack (retry). +// Returns nil to ack or honor a recorded Hold; errors follow the consumer retry policy. func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() @@ -125,6 +130,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } request, err := store.GetRequestStore().Get(ctx, cancelReq.ID) + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "waiting_for_request", 1, metrics.TagsFromContext(ctx)...) + delivery.Hold(waitForRequestDelayMs) + return nil + } if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) return fmt.Errorf("failed to get request %s: %w", cancelReq.ID, err) diff --git a/submitqueue/orchestrator/controller/cancel/cancel_test.go b/submitqueue/orchestrator/controller/cancel/cancel_test.go index 136a6ec96..cc4dcfd14 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel_test.go +++ b/submitqueue/orchestrator/controller/cancel/cancel_test.go @@ -105,7 +105,7 @@ func newController(t *testing.T, store orchstorage.Storage, registry consumer.To return NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, staticStorageFactory{store: store}, registry, topickey.TopicKeyCancel, "orchestrator-cancel") } -func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte, partitionKey string) consumer.Delivery { +func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte, partitionKey string) *consumermock.MockDelivery { msg := entityqueue.NewMessage("cancel-msg", payload, partitionKey, nil) msg.Tenant = "q" d := consumermock.NewMockDelivery(ctrl) @@ -184,21 +184,23 @@ func TestProcess_AlreadyTerminal_NoOp(t *testing.T) { require.NoError(t, err) } -func TestProcess_RequestNotFound_Retryable(t *testing.T) { +func TestProcess_RequestNotFound_HoldsOriginalDelivery(t *testing.T) { ctrl := gomock.NewController(t) registry, _ := newRegistry(t, ctrl) - reqStore := storagemock.NewMockRequestStore(ctrl) - reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, storage.ErrNotFound) - + reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, fmt.Errorf("not visible yet: %w", storage.ErrNotFound)).Times(4) store := orchstoragemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() - controller := newController(t, store, registry) - err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", ""), "1")) - require.Error(t, err) - assert.ErrorIs(t, err, storage.ErrNotFound) + payload := cancelPayload(t, "1", "user changed mind") + msg := entityqueue.NewMessage("cancel-msg", payload, "1", nil) + msg.Tenant = "q" + delivery := consumermock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg).Times(4) + delivery.EXPECT().Hold(waitForRequestDelayMs).Times(4) + for range 4 { + require.NoError(t, controller.Process(context.Background(), delivery)) + } } func TestProcess_CancelsUnbatchedRequest(t *testing.T) { @@ -226,6 +228,7 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) { // the request — the terminal Cancelled CAS. var terminal entity.Request gomock.InOrder( + reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, storage.ErrNotFound), reqStore.EXPECT().Get(gomock.Any(), "1").Return(started, nil), reqStore.EXPECT().Update(gomock.Any(), requestWithState(started, entity.RequestStateCancelling), int32(2), int32(3)).Return(nil), reqStore.EXPECT().Get(gomock.Any(), "1").Return(cancelling, nil), @@ -246,6 +249,9 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) { expectBatchLookup(ctrl, store, batchStore, "1") controller := newController(t, store, registry) + earlyDelivery := newDelivery(t, ctrl, cancelPayload(t, "1", "user changed mind"), "1") + earlyDelivery.EXPECT().Hold(waitForRequestDelayMs) + require.NoError(t, controller.Process(context.Background(), earlyDelivery)) err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", "user changed mind"), "1")) require.Error(t, err) require.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", "changed reason"), "1"))) @@ -753,7 +759,8 @@ func TestProcess_RequestStoreError(t *testing.T) { registry, _ := newRegistry(t, ctrl) reqStore := storagemock.NewMockRequestStore(ctrl) - reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, fmt.Errorf("db down")) + storeErr := fmt.Errorf("db down") + reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, storeErr) store := orchstoragemock.NewMockStorage(ctrl) store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() @@ -761,7 +768,7 @@ func TestProcess_RequestStoreError(t *testing.T) { controller := newController(t, store, registry) err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", ""), "1")) - require.Error(t, err) + require.ErrorIs(t, err, storeErr) } func TestFindBatches(t *testing.T) { diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index 329ff178c..a35b942e7 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -1,22 +1,24 @@ # DLQ Reconciliation Controllers -This package contains the controllers that drain primary pipeline topics' `{topic}_dlq` companions and reconcile the affected request or batch into a terminal failed state. The pipeline topology in `submitqueue/orchestrator/pipeline.go` wires these beside the primary controllers. Start is the exception: its owning controller resumes the full original request rather than manufacturing a failed outcome. +This package contains the controllers that drain primary pipeline topics' `{topic}_dlq` companions and reconcile the affected request or batch into a terminal failed state. The pipeline topology in `submitqueue/orchestrator/pipeline.go` wires these beside the primary controllers. Start and cancel use their owning controllers to resume the full original request rather than manufacturing a failed outcome. ## 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.** Each generic reconciler 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 failed stages; recover start in place.** Except for start, 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 these generic reconcilers. +**Reconcile failed stages; recover start and cancel in place.** A generic reconciler 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 these generic reconcilers. The `start_dlq` subscription uses the start controller under the same unlimited-retry final policy. It creates the original Request when absent, then loads the authoritative snapshot after Create or AlreadyExists. Matching Started snapshots replay the Started log and normal `validate` hand-off. Later states and Cancelling retain their current owner; terminal snapshots replay their stored winning terminal log. Conflicting immutable inputs, unknown states, missing snapshots, and malformed messages return errors and remain durable. No tombstone, new request ID, synthetic Error, or return to the finite-budget start subscription is needed. +The `cancel_dlq` subscription uses the cancel controller and the full original cancellation payload, including its reason. A missing Request holds the same delivery until start persists it, with no expiry or failure-budget consumption; the primary cancel subscription follows the same wait policy because the independent topics can overtake one another. Once present, ordinary cancellation resumes, including matching Cancelled-log repair and the existing best-effort Landing outcome. An ID that never appears remains durable for operator investigation. + ## Convergence guarantee DLQ consumers are wired with `errs.AlwaysRetryableProcessor`, unlimited attempts (`Retry.MaxAttempts = 0`), and `DLQ.Enabled = false` on the DLQ subscription itself. Any non-nil error returned from a DLQ controller — including a plain unclassified infra error — is therefore forced retryable and redelivered rather than silently dropped. This is the convergence guarantee the package relies on. -The recognised error condition is handled explicitly in `dlq.go`: +The generic reconcilers handle this recognised error condition explicitly in `dlq.go` (it is not the start/cancel recovery policy): - `storage.ErrNotFound` → logged at warn and treated as success. The request or batch never persisted; there is nothing to reconcile. @@ -28,14 +30,14 @@ When a DLQ controller transitions a request to `RequestStateError`, it publishes ## Controller mapping -Two controller shapes cover the eleven primary pipeline topics: +Two generic controller shapes cover the ID-payload stages: | Controller | Topics | Decoded ID | Terminal state | |---|---|---|---| -| `NewDLQRequestController` | `validate`, `batch`, `cancel` | `RequestID` | `RequestStateError` | +| `NewDLQRequestController` | `validate`, `batch` | `RequestID` | `RequestStateError` | | `NewDLQBatchController` | `speculate`, `build`, `land`, `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 `DecodeCancelRequestID` / `DecodeRequestID` cover the two 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 the decoder is what changes per topic — not the reconciliation step. The generic request reconciler uses `DecodeRequestID`; cancellation preserves its full payload in its owning controller. ## Idempotency and concurrent activity @@ -44,7 +46,7 @@ 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 in `RequestStateCancelling` reached through a generic failed-stage reconciler 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. ## See also diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go index f4bc54288..030df2236 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -32,24 +32,10 @@ import ( "go.uber.org/zap" ) -// RequestIDDecoder extracts the affected request's identity — its ID and its -// queue — from the raw payload bytes of a DLQ message. Different primary -// topics carry different payload shapes (CancelRequest on cancel, RequestID -// on validate / batch), so the caller injects the right -// decoder for the topic being reconciled. Returning an empty ID is treated as -// a decode failure. +// RequestIDDecoder extracts a request's ID and queue from a request-scoped +// DLQ payload. Returning an empty ID is treated as a decode failure. type RequestIDDecoder func(payload []byte) (entity.RequestID, error) -// DecodeCancelRequestID extracts the request ID from a CancelRequest payload -// (the shape used by the cancel topic). -func DecodeCancelRequestID(payload []byte) (entity.RequestID, error) { - cr, err := sqmq.UnmarshalCancelRequest(payload) - if err != nil { - return entity.RequestID{}, err - } - return entity.RequestID{ID: cr.ID, Queue: cr.Queue}, nil -} - // DecodeRequestID extracts the request ID from a validate or batch payload. // primary is the originating topic (TopicKeyValidate or TopicKeyBatch), not the DLQ key. func DecodeRequestID(primary consumer.TopicKey) RequestIDDecoder { @@ -59,7 +45,7 @@ func DecodeRequestID(primary consumer.TopicKey) RequestIDDecoder { } // requestController is the DLQ reconciler for request-scoped pipeline stages. -// It uses the matching decoder for cancel, validate, or batch and fails the +// It uses the matching decoder for validate or batch and fails the // request unless a live batch already owns its outcome. type requestController struct { logger *zap.SugaredLogger diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go index d02b2cae9..82e084f24 100644 --- a/submitqueue/orchestrator/controller/dlq/request_test.go +++ b/submitqueue/orchestrator/controller/dlq/request_test.go @@ -57,34 +57,6 @@ func TestDLQRequestController_Process_TenantPayloadQueueMismatchAcks(t *testing. require.NoError(t, c.Process(context.Background(), newMockDeliveryWithTenant(ctrl, payload, "other-queue"))) } -func TestDLQRequestController_Process_CancelRequestPayload(t *testing.T) { - ctrl := gomock.NewController(t) - - requestStore := storagemock.NewMockRequestStore(ctrl) - request := entity.Request{ - ID: "7", Queue: "q", Version: 2, State: entity.RequestStateBatched, - } - requestStore.EXPECT().Get(gomock.Any(), "7").Return(request, nil) - requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(2), int32(3)).Return(nil) - - registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { - return nil - }) - - store := orchstoragemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() - store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - - c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store: store}, registry, DecodeCancelRequestID, TopicKey(topickey.TopicKeyCancel), "orchestrator-cancel-dlq") - - payload, err := sqmq.Marshal(sqmq.CancelFromEntity(entity.CancelRequest{ID: "7", Queue: "q", Reason: "user"})) - require.NoError(t, err) - - delivery := newMockDelivery(ctrl, payload) - require.NoError(t, c.Process(context.Background(), delivery)) -} - func TestDLQRequestController_Process_RequestIDPayload(t *testing.T) { ctrl := gomock.NewController(t) diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index cff5301b6..b01447dee 100644 --- a/submitqueue/orchestrator/pipeline.go +++ b/submitqueue/orchestrator/pipeline.go @@ -120,7 +120,7 @@ var Stages = []pipeline.Stage[Deps]{ return cancel.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.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeCancelRequestID, sc.TopicKey, sc.ConsumerGroup), nil + return cancel.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, }, { diff --git a/test/integration/submitqueue/orchestrator/BUILD.bazel b/test/integration/submitqueue/orchestrator/BUILD.bazel index 0d78fd34c..2614392dc 100644 --- a/test/integration/submitqueue/orchestrator/BUILD.bazel +++ b/test/integration/submitqueue/orchestrator/BUILD.bazel @@ -21,10 +21,18 @@ go_test( "//platform/base/change:go_default_library", "//platform/base/mergestrategy:go_default_library", "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", + "//platform/extension/consumergate/noop:go_default_library", + "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql: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/orchestrator:go_default_library", + "//submitqueue/orchestrator/extension/storage:go_default_library", "//submitqueue/orchestrator/extension/storage/mysql:go_default_library", "//test/testutil:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/test/integration/submitqueue/orchestrator/suite_test.go b/test/integration/submitqueue/orchestrator/suite_test.go index 64003a02f..43440ed9a 100644 --- a/test/integration/submitqueue/orchestrator/suite_test.go +++ b/test/integration/submitqueue/orchestrator/suite_test.go @@ -40,10 +40,18 @@ import ( "github.com/uber/submitqueue/platform/base/change" "github.com/uber/submitqueue/platform/base/mergestrategy" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "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" + "github.com/uber/submitqueue/submitqueue/orchestrator" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" mysqlstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mysql" "github.com/uber/submitqueue/test/testutil" "go.uber.org/zap" @@ -186,3 +194,135 @@ func (s *OrchestratorIntegrationSuite) TestRetainedStartCreatesRequestAndHandsOf assert.Equal(t, request.Change.URIs, check.GetSteps()[0].GetChange().GetUris()) assert.Equal(t, mergestrategypb.Strategy_REBASE, check.GetSteps()[0].GetStrategy()) } + +// No validate subscriber runs on these topics: the test controls the pre-enrollment window, not a race against Landing. +func (s *OrchestratorIntegrationSuite) TestCancelBeforeStartRetainsIntent() { + for _, tc := range []struct { + name string + id string + dlq bool + }{ + {name: "primary", id: "912"}, + {name: "retained-dlq", id: "913", dlq: true}, + } { + s.T().Run(tc.name, func(t *testing.T) { + queue, err := queueMySQL.NewQueue(queueMySQL.Params{DB: s.queueDB, Logger: zap.NewNop(), MetricsScope: tally.NoopScope, Tenants: []string{"test-queue"}}) + require.NoError(t, err) + defer queue.Close() + backend, err := mysqlstorage.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err) + stores := cancellationStorageFactory{backend} + subscription := extqueue.SingleMessageSubscriptionConfig("early-cancel", "early-cancel") + subscription.Retry.MaxAttempts = 3 + cancelKey := topickey.TopicKeyCancel + cancelSubscription := subscription + if tc.dlq { + cancelKey = "cancel_dlq" + cancelSubscription = extqueue.DLQSubscriptionConfig("early-cancel", "early-cancel") + } + cancelTopic := "early-cancel-" + tc.name + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyStart, Name: "early-start-" + tc.name, Queue: queue, Subscription: subscription}, + {Key: cancelKey, Name: cancelTopic, Queue: queue, Subscription: cancelSubscription}, + {Key: topickey.TopicKeyLog, Name: "early-log-" + tc.name, Queue: queue}, + {Key: topickey.TopicKeyValidate, Name: "early-validate-" + tc.name, Queue: queue}, + }) + require.NoError(t, err) + deps := orchestrator.Deps{Logger: zap.NewNop().Sugar(), Scope: tally.NoopScope, Storage: stores} + for _, stage := range orchestrator.Stages { + if stage.Key != topickey.TopicKeyStart && stage.Key != topickey.TopicKeyCancel { + continue + } + constructor, key := stage.New, stage.Key + processor := errs.NewClassifierProcessor() + if stage.Key == topickey.TopicKeyCancel && tc.dlq { + constructor, key = stage.DLQ, cancelKey + processor = errs.AlwaysRetryableProcessor + } + controller, err := constructor(deps, pipeline.StageContext{Registry: registry, TopicKey: key, ConsumerGroup: subscription.ConsumerGroup}) + require.NoError(t, err) + worker := consumer.New(zap.NewNop().Sugar(), tally.NoopScope, registry, processor, consumergatenoop.New()) + require.NoError(t, worker.Register(controller)) + require.NoError(t, worker.Start(s.ctx)) + defer func() { require.NoError(t, worker.Stop(10000)) }() + } + cancelRequest := entity.CancelRequest{ID: tc.id, Queue: "test-queue", Reason: "cancel before admission"} + payload, err := sqmq.Marshal(sqmq.CancelFromEntity(cancelRequest)) + require.NoError(t, err) + msg := entityqueue.NewMessage("cancel-"+tc.id, payload, tc.id, nil) + msg.Tenant = cancelRequest.Queue + require.NoError(t, queue.Publisher().Publish(s.ctx, cancelTopic, msg)) + + // Observe committed postponements beyond the finite primary budget before releasing start. + s.waitForCancelPostponements(t, cancelTopic, subscription.Retry.MaxAttempts+1) + requestStore := mysqlstorage.NewRequestStore(s.db, tally.NoopScope, cancelRequest.Queue) + _, err = requestStore.Get(s.ctx, tc.id) + require.ErrorIs(t, err, storage.ErrNotFound) + var terminalCount int + require.NoError(t, s.queueDB.QueryRowContext(s.ctx, "SELECT COUNT(*) FROM queue_messages WHERE topic IN (?, ?)", "early-log-"+tc.name, cancelTopic+"_dlq").Scan(&terminalCount)) + assert.Zero(t, terminalCount, "waiting must neither publish a terminal log nor spend the primary failure budget") + + landRequest := retainedStartRequest() + landRequest.ID = tc.id + payload, err = sqmq.Marshal(sqmq.StartFromLandRequest(landRequest)) + require.NoError(t, err) + msg = entityqueue.NewMessage("start-"+tc.id, payload, landRequest.Queue, nil) + msg.Tenant = landRequest.Queue + require.NoError(t, queue.Publisher().Publish(s.ctx, "early-start-"+tc.name, msg)) + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + var cancelledLog entity.RequestLog + for cancelledLog.Status != entity.RequestStatusCancelled { + rows, err := s.queueDB.QueryContext(s.ctx, "SELECT payload FROM queue_messages WHERE topic = ?", "early-log-"+tc.name) + require.NoError(t, err) + for rows.Next() { + var payload []byte + require.NoError(t, rows.Scan(&payload)) + entry, err := sqmq.UnmarshalRequestLog(payload) + require.NoError(t, err) + if entry.Status == entity.RequestStatusCancelled { + cancelledLog = entry + } + } + require.NoError(t, rows.Err()) + require.NoError(t, rows.Close()) + <-ticker.C + } + assert.Equal(t, cancelRequest.Reason, cancelledLog.Metadata["reason"]) + request, err := requestStore.Get(s.ctx, tc.id) + require.NoError(t, err) + assert.Equal(t, entity.RequestStateCancelled, request.State) + assert.Equal(t, cancelRequest.Reason, request.TerminalDetails.Metadata["reason"]) + }) + } +} + +type cancellationStorageFactory struct{ backend *mysqlstorage.Storage } + +func (f cancellationStorageFactory) For(config orchstorage.Config) (orchstorage.Storage, error) { + return f.backend.For(config.QueueName) +} + +func (s *OrchestratorIntegrationSuite) waitForCancelPostponements(t *testing.T, topic string, count int) { + t.Helper() + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + var previousDelay int64 + for observed := 0; observed < count; { + var acked, postponed bool + var retryCount int + var invisibleUntil int64 + err := s.queueDB.QueryRowContext(s.ctx, `SELECT acked, postponed, retry_count, invisible_until FROM queue_delivery_state + WHERE topic = ? AND consumer_group = 'early-cancel'`, topic).Scan(&acked, &postponed, &retryCount, &invisibleUntil) + if err != sql.ErrNoRows { + require.NoError(t, err) + require.False(t, acked, "the missing request must retain its original delivery") + if postponed && invisibleUntil > previousDelay { + assert.Zero(t, retryCount) + previousDelay = invisibleUntil + observed++ + } + } + <-ticker.C + } +}