From 21a5ac6a8d1a621905e134b46ac5b00d92bee00b Mon Sep 17 00:00:00 2001 From: sergeyb Date: Sat, 14 Nov 2026 02:24:49 +0000 Subject: [PATCH] fix: retain cancellations received before start Summary: Hold primary cancellation deliveries while the Request is absent without expiry or failure-budget consumption, then apply best-effort cancellation with the original reason. Final cancellation deliveries wait for the authoritative Request and reconcile only Error, preserving owning batches, terminal outcomes and the observed-version guard. Missing IDs remain held for operator investigation. Test Plan: All 21 orchestrator, gateway materialization, message contract and orchestrator integration targets pass uncached, including Docker integration in 25.0s. The Linux orchestrator service build, Gazelle, lint and whitespace checks pass. Revert Plan: Preserve held cancellation deliveries before reverting; no schema migration is involved. Final queues must continue to reconcile status rather than execute normal work. --- doc/rfc/submitqueue/workflow.md | 4 +- .../orchestrator/controller/cancel/cancel.go | 20 ++- .../controller/cancel/cancel_test.go | 31 ++-- .../orchestrator/controller/dlq/README.md | 4 +- .../orchestrator/controller/dlq/request.go | 121 ++++++++----- .../controller/dlq/request_test.go | 162 +++++++++++++----- submitqueue/orchestrator/pipeline.go | 2 +- .../submitqueue/orchestrator/BUILD.bazel | 18 ++ .../submitqueue/orchestrator/suite_test.go | 153 +++++++++++++++++ 9 files changed, 409 insertions(+), 106 deletions(-) diff --git a/doc/rfc/submitqueue/workflow.md b/doc/rfc/submitqueue/workflow.md index a050b2357..b04251a26 100644 --- a/doc/rfc/submitqueue/workflow.md +++ b/doc/rfc/submitqueue/workflow.md @@ -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 | Persist the Request and publish it to 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 | @@ -164,6 +164,8 @@ DLQ handlers do not resume the failed request's normal workflow. They reconcile The start DLQ reports a failed submission through an Error request log with version 0 when no Request was saved. Existing rows retain their ownership, terminal-outcome, and conditional-write protections. +The primary cancel stage holds a cancellation whose Request is missing without expiry or failure-budget consumption, then performs ordinary best-effort cancellation with the original reason. The cancel DLQ holds the same missing identity until an authoritative Request exists, then reconciles only Error; it does not cancel builds or publish normal-work commands. Reconciliation preserves owning batches and other terminal outcomes, and its observed-version write rejects a concurrent batch claim. Missing IDs remain durable for operator investigation. Primary cancellation remains best-effort once an attempt reaches Landing. + 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/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index 330e6a068..8f272b6dd 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 a995fcb08..8d640e69f 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel_test.go +++ b/submitqueue/orchestrator/controller/cancel/cancel_test.go @@ -94,7 +94,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) @@ -173,21 +173,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) { @@ -207,6 +209,7 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) { // Two-step transition: mark Cancelling, then — after TerminateRequest re-reads // the request — the terminal Cancelled CAS. 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), @@ -222,6 +225,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.NoError(t, err) @@ -724,7 +730,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() @@ -732,7 +739,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 04a2167e6..03ab724d3 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -18,6 +18,8 @@ Disposition remains handler-specific: the generic request/batch reconcilers ackn Start reports the failed submission through an Error request log with version 0, because there is no saved Request version. No normal work is published. Reconciliation validates the request identity and queue; existing rows retain authoritative-identity, batch-ownership, terminal-winner, and conditional-write guards. +Cancellation deliveries in the primary and final queues use Hold while the Request is absent, without expiry or failure-budget consumption. The primary controller then applies ordinary best-effort cancellation with the original reason. The cancel DLQ instead reconciles only Error under the batch-ownership and terminal-winner rules; it does not cancel builds or publish normal-work commands. Missing identities remain held for operator investigation. + ## Terminal outputs and ownership - An already-Error request skips the state CAS but republishes its terminal status log. A different terminal request outcome is left unchanged. This repairs status delivery, not the original diagnostic content or timestamp. @@ -31,7 +33,7 @@ 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` uses dedicated request reconciliation; `cancel`, `validate`, and `batch` use generic request reconciliation. Their decoders read LandRequest, CancelRequest, or the topic's request-ID payload. +- `start` and `cancel` use dedicated request reconciliation; `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. - `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. diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go index 9e6d5e709..6517a5e72 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -29,15 +29,12 @@ import ( "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" corebatch "github.com/uber/submitqueue/submitqueue/orchestrator/core/batch" + requestcore "github.com/uber/submitqueue/submitqueue/orchestrator/core/request" "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 (LandRequest on start, 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) // DecodeLandRequestID extracts the request ID from a LandRequest payload @@ -50,14 +47,13 @@ func DecodeLandRequestID(payload []byte) (entity.RequestID, error) { return entity.RequestID{ID: lr.ID, Queue: lr.Queue}, nil } -// DecodeCancelRequestID extracts the request ID from a CancelRequest payload -// (the shape used by the cancel topic). +// DecodeCancelRequestID extracts the request identity from a cancellation payload. func DecodeCancelRequestID(payload []byte) (entity.RequestID, error) { - cr, err := sqmq.UnmarshalCancelRequest(payload) + request, err := sqmq.UnmarshalCancelRequest(payload) if err != nil { return entity.RequestID{}, err } - return entity.RequestID{ID: cr.ID, Queue: cr.Queue}, nil + return entity.RequestID{ID: request.ID, Queue: request.Queue}, nil } // DecodeRequestID extracts the request ID from a validate or batch payload. @@ -69,18 +65,17 @@ func DecodeRequestID(primary consumer.TopicKey) RequestIDDecoder { } // requestController is the DLQ reconciler for request-scoped pipeline stages. -// It is registered once per primary request-scoped topic (cancel, validate, -// batch) with the matching decoder. On each delivery it decodes the -// request ID and transitions the request to RequestStateError, unless a live -// batch already owns the request's outcome. +// It uses the matching decoder for cancel, validate, or batch and fails the +// request unless a batch already owns its outcome. type requestController struct { - logger *zap.SugaredLogger - metricsScope tally.Scope - stores orchstorage.Factory - registry consumer.TopicRegistry - decode RequestIDDecoder - topicKey consumer.TopicKey - consumerGroup string + logger *zap.SugaredLogger + metricsScope tally.Scope + stores orchstorage.Factory + registry consumer.TopicRegistry + decode RequestIDDecoder + topicKey consumer.TopicKey + consumerGroup string + retainMissingRequest bool } // Verify requestController implements consumer.Controller at compile time. @@ -98,6 +93,17 @@ func NewDLQRequestController( topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller { + return newRequestController(logger, scope, stores, registry, decode, topicKey, consumerGroup) +} + +// NewDLQCancelController reconciles only Error; missing Requests remain durable until they exist. +func NewDLQCancelController(logger *zap.SugaredLogger, scope tally.Scope, stores orchstorage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string) consumer.Controller { + controller := newRequestController(logger, scope, stores, registry, DecodeCancelRequestID, topicKey, consumerGroup) + controller.retainMissingRequest = true + return controller +} + +func newRequestController(logger *zap.SugaredLogger, scope tally.Scope, stores orchstorage.Factory, registry consumer.TopicRegistry, decode RequestIDDecoder, topicKey consumer.TopicKey, consumerGroup string) *requestController { name := string(topicKey) + "_controller" return &requestController{ logger: logger.Named(name), @@ -151,34 +157,71 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv "dlq_last_error", dmeta["dlq.last_error"], ) - // A batch that got as far as enrolling the request owns its outcome, and - // conclude writes that outcome when the batch finishes. This dead letter can - // be a redundant attempt whose predecessor is already speculating — the batch - // stage mints one batch per delivery — and TerminateRequest does not guard - // against Batched, so failing the request here would overwrite a live batch's - // claim and leave it building for a request reported as failed. - owner, err := owningBatch(ctx, store, rid.ID) - if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "batch_lookup_errors", 1, metrics.TagsFromContext(ctx)...) + if c.retainMissingRequest { + err := c.reconcileCancelRequest(ctx, delivery, store, rid.ID, lastError, failureMeta) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...) + } return err } - if owner != "" { - metrics.NamedCounter(c.metricsScope, opName, "skipped_owned_by_batch", 1, metrics.TagsFromContext(ctx)...) - c.logger.Infow("dlq reconcile: batch owns the request outcome, leaving completion to conclude", - "request_id", rid.ID, - "batch_id", owner, - ) - return nil + owned, err := c.requestHasOwningBatch(ctx, store, rid.ID) + if err != nil || owned { + return err } - if err := failRequest(ctx, store, c.registry, c.logger, rid.ID, lastError, failureMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...) return err } - return nil } +func (c *requestController) reconcileCancelRequest(ctx context.Context, delivery consumer.Delivery, store orchstorage.Storage, requestID, lastError string, metadata map[string]string) error { + request, err := store.GetRequestStore().Get(ctx, requestID) + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, "process", "waiting_for_request", 1, metrics.TagsFromContext(ctx)...) + delivery.Hold(waitForRequestDelayMs) + return nil + } + if err != nil { + return fmt.Errorf("failed to load request for cancel reconciliation: %w", err) + } + if request.State == entity.RequestStateError { + return requestcore.ResendTerminalRequestLog(ctx, c.registry, request) + } + if entity.IsRequestStateTerminal(request.State) { + return nil + } + owned, err := c.requestHasOwningBatch(ctx, store, requestID) + if err != nil || owned { + return err + } + request.State = entity.RequestStateError + newVersion := request.Version + 1 + // The version observed before owner lookup fences a later batch claim. + if err := store.GetRequestStore().Update(ctx, request, request.Version, newVersion); err != nil { + return fmt.Errorf("failed to fail cancelled request %s: %w", requestID, err) + } + request.Version = newVersion + entry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusError, request.Version, lastError, metadata) + return requestcore.PublishLog(ctx, c.registry, entry, request.ID, "") +} + +func (c *requestController) requestHasOwningBatch(ctx context.Context, store orchstorage.Storage, requestID string) (bool, error) { + owner, err := owningBatch(ctx, store, requestID) + if err != nil { + metrics.NamedCounter(c.metricsScope, "process", "batch_lookup_errors", 1, metrics.TagsFromContext(ctx)...) + return false, err + } + if owner == "" { + return false, nil + } + metrics.NamedCounter(c.metricsScope, "process", "skipped_owned_by_batch", 1, metrics.TagsFromContext(ctx)...) + c.logger.Infow("dlq reconcile: batch owns the request outcome, leaving completion to conclude", "request_id", requestID, "batch_id", owner) + return true, nil +} + +const waitForRequestDelayMs int64 = 1000 + // owningBatch returns a batch that is still running or whose result has not been recorded on the request yet. // Creating batches must have claimed the request; failed batches do not stop it from being marked Error. func owningBatch(ctx context.Context, store orchstorage.Storage, requestID string) (string, error) { diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go index a6773150f..eaa2801a3 100644 --- a/submitqueue/orchestrator/controller/dlq/request_test.go +++ b/submitqueue/orchestrator/controller/dlq/request_test.go @@ -16,6 +16,7 @@ package dlq import ( "context" + "errors" "testing" "github.com/stretchr/testify/assert" @@ -27,6 +28,7 @@ import ( 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" orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock" "go.uber.org/mock/gomock" @@ -85,34 +87,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) @@ -142,26 +116,120 @@ func TestDLQRequestController_Process_RequestIDPayload(t *testing.T) { require.NoError(t, c.Process(context.Background(), delivery)) } -func TestDLQRequestController_Process_DifferentTerminalOutcomeSkips(t *testing.T) { - ctrl := gomock.NewController(t) - - requestStore := storagemock.NewMockRequestStore(ctrl) - requestStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{ - ID: "1", Queue: "q", Version: 5, State: entity.RequestStateLanded, - }, 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}, consumer.TopicRegistry{}, DecodeRequestID(topickey.TopicKeyValidate), TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") - - payload, err := sqmq.MarshalID(sqmq.TopicKeyValidate, "1", "q") - require.NoError(t, err) +func TestDLQCancelController_WaitsThenReconcilesError(t *testing.T) { + for _, claim := range []bool{false, true} { + name := "unowned request fails" + if claim { + name = "batch claim after ownership lookup" + } + t.Run(name, func(t *testing.T) { + ctrl := gomock.NewController(t) + var persisted entity.Request + requests := storagemock.NewMockRequestStore(ctrl) + requests.EXPECT().Get(gomock.Any(), "1").DoAndReturn(func(context.Context, string) (entity.Request, error) { + if persisted.ID == "" { + return entity.Request{}, storage.ErrNotFound + } + return persisted, nil + }).AnyTimes() + requests.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, request entity.Request, oldVersion, newVersion int32) error { + if oldVersion != persisted.Version { + return storage.ErrVersionMismatch + } + assert.Equal(t, oldVersion, request.Version) + assert.Equal(t, oldVersion+1, newVersion) + persisted = request + persisted.Version = newVersion + return nil + }).Times(1) + associations := storagemock.NewMockRequestBatchStore(ctrl) + associations.EXPECT().GetByRequestID(gomock.Any(), "1").DoAndReturn(func(context.Context, string) ([]entity.RequestBatch, error) { + if claim && persisted.State == entity.RequestStateValidated { + persisted.State = entity.RequestStateBatched + persisted.Version++ + return nil, nil + } + if claim && persisted.State == entity.RequestStateBatched { + return []entity.RequestBatch{{RequestID: "1", BatchID: "batch"}}, nil + } + return nil, nil + }).AnyTimes() + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(associations).AnyTimes() + logCount := 1 + if claim { + logCount = 0 + batches := storagemock.NewMockBatchStore(ctrl) + batches.EXPECT().Get(gomock.Any(), "batch").Return(entity.Batch{ID: "batch", State: entity.BatchStateCreated}, nil) + store.EXPECT().GetBatchStore().Return(batches) + } + registry := newTestLogRegistry(t, ctrl, logCount, func(log entity.RequestLog) error { + assert.Equal(t, entity.RequestStatusError, log.Status) + assert.Equal(t, "1", log.RequestID) + assert.Equal(t, int32(8), log.RequestVersion) + assert.Equal(t, "boom", log.LastError) + return nil + }) + controller := NewDLQCancelController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store}, registry, TopicKey(topickey.TopicKeyCancel), "orchestrator-dlq") + payload, err := sqmq.Marshal(sqmq.CancelFromEntity(entity.CancelRequest{ID: "1", Queue: "q", Reason: "user changed mind"})) + require.NoError(t, err) + delivery := newMockDelivery(ctrl, payload) + delivery.EXPECT().Hold(waitForRequestDelayMs).Times(2) + for range 2 { + require.NoError(t, controller.Process(context.Background(), delivery)) + } + persisted = entity.Request{ID: "1", Queue: "q", State: entity.RequestStateValidated, Version: 7} + err = controller.Process(context.Background(), delivery) + if claim { + require.ErrorIs(t, err, storage.ErrVersionMismatch) + require.NoError(t, controller.Process(context.Background(), delivery)) + assert.Equal(t, entity.RequestStateBatched, persisted.State) + } else { + require.NoError(t, err) + assert.Equal(t, entity.RequestStateError, persisted.State) + } + assert.Equal(t, int32(8), persisted.Version) + }) + } +} - delivery := newMockDelivery(ctrl, payload) - require.NoError(t, c.Process(context.Background(), delivery)) +func TestDLQCancelController_Process_TerminalOutcomes(t *testing.T) { + for _, state := range []entity.RequestState{entity.RequestStateLanded, entity.RequestStateCancelled, entity.RequestStateError} { + t.Run(string(state), func(t *testing.T) { + ctrl := gomock.NewController(t) + request := entity.Request{ID: "1", Queue: "q", Version: 5, State: state} + calls := 1 + logCount := 0 + if state == entity.RequestStateError { + calls, logCount = 2, 2 + } + requests := storagemock.NewMockRequestStore(ctrl) + requests.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil).Times(calls) + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() + publishErr := errors.New("publisher unavailable") + published := 0 + registry := newTestLogRegistry(t, ctrl, logCount, func(log entity.RequestLog) error { + assert.Equal(t, entity.RequestStatusError, log.Status) + assert.Equal(t, request.Version, log.RequestVersion) + published++ + if published == 1 { + return publishErr + } + return nil + }) + controller := NewDLQCancelController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store}, registry, TopicKey(topickey.TopicKeyCancel), "orchestrator-cancel-dlq") + payload, err := sqmq.Marshal(sqmq.CancelFromEntity(entity.CancelRequest{ID: request.ID, Queue: request.Queue})) + require.NoError(t, err) + delivery := newMockDelivery(ctrl, payload) + if state == entity.RequestStateError { + require.ErrorIs(t, controller.Process(context.Background(), delivery), publishErr) + } + require.NoError(t, controller.Process(context.Background(), delivery)) + }) + } } func TestDLQRequestController_Process_MalformedPayloadFails(t *testing.T) { diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index f5bbc94b6..470db304b 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 dlq.NewDLQCancelController(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 ca5158a79..454aa347a 100644 --- a/test/integration/submitqueue/orchestrator/BUILD.bazel +++ b/test/integration/submitqueue/orchestrator/BUILD.bazel @@ -16,10 +16,28 @@ go_test( ], deps = [ "//api/submitqueue/orchestrator/protopb:go_default_library", + "//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", "@com_github_stretchr_testify//require:go_default_library", "@com_github_stretchr_testify//suite:go_default_library", + "@com_github_uber_go_tally//:go_default_library", "@org_golang_google_grpc//:go_default_library", + "@org_uber_go_zap//:go_default_library", ], ) diff --git a/test/integration/submitqueue/orchestrator/suite_test.go b/test/integration/submitqueue/orchestrator/suite_test.go index e04952747..9a67121fd 100644 --- a/test/integration/submitqueue/orchestrator/suite_test.go +++ b/test/integration/submitqueue/orchestrator/suite_test.go @@ -28,12 +28,31 @@ import ( "context" "database/sql" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" + "github.com/uber-go/tally" pb "github.com/uber/submitqueue/api/submitqueue/orchestrator/protopb" + "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" "google.golang.org/grpc" ) @@ -115,3 +134,137 @@ func (s *OrchestratorIntegrationSuite) TestPingAPI() { assert.NotEmpty(t, resp.Message) assert.NotZero(t, resp.Timestamp) } + +// 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 + state entity.RequestState + status entity.RequestStatus + }{ + {name: "primary", id: "912", state: entity.RequestStateCancelled, status: entity.RequestStatusCancelled}, + {name: "retained-dlq", id: "913", dlq: true, state: entity.RequestStateError, status: entity.RequestStatusError}, + } { + 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 := entity.LandRequest{ID: tc.id, Queue: "test-queue", Change: change.Change{URIs: []string{"github://github.example.com/org/repo/pull/11/abcdef0123456789abcdef0123456789abcdef01"}}, LandStrategy: mergestrategy.MergeStrategyRebase} + 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 terminalLog entity.RequestLog + for terminalLog.Status != tc.status { + 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 == tc.status { + terminalLog = entry + } + } + require.NoError(t, rows.Err()) + require.NoError(t, rows.Close()) + <-ticker.C + } + if !tc.dlq { + assert.Equal(t, cancelRequest.Reason, terminalLog.Metadata["reason"]) + } + request, err := requestStore.Get(s.ctx, tc.id) + require.NoError(t, err) + assert.Equal(t, tc.state, request.State) + }) + } +} + +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 + } +}