From c109d1b569544e2dbb0982b9e4e828aa2ea03075 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Fri, 6 Nov 2026 22:34:12 +0000 Subject: [PATCH 1/2] fix(start): recover retained requests in place Summary: Intent: - Resume retained start deliveries whose request creation never completed, without requiring another user submission or inventing a failed outcome. Changes: - Reuse start for final recovery and classify the authoritative snapshot after Create or AlreadyExists. - Preserve immutable inputs and later-stage ownership; replay stored terminal facts through the shared terminal-log publisher. - Replace the unused start ID-only decoder and cover recovery through the actual pipeline binding and running service. Test Plan: - TestRetainedStartCreatesRequestAndHandsOffToValidate seeds a full start_dlq payload before service startup with no request row, then verifies acknowledgment, the original immutable inputs, and the normal validate command. - TestController_Process_ReplaysFailedHandOff exercises dependency failures and lost acknowledgment on both input topics; TestController_Process_AuthoritativeState proves later/cancelling ownership and stored terminal diagnostics are preserved. Revert Plan: Revert this change while retaining final-DLQ records and existing request rows. Already-admitted work continues under its current owner. --- doc/rfc/submitqueue/modular-queue-wiring.md | 5 +- doc/rfc/submitqueue/workflow.md | 6 +- submitqueue/orchestrator/BUILD.bazel | 12 + .../orchestrator/controller/dlq/README.md | 12 +- .../orchestrator/controller/dlq/request.go | 20 +- .../controller/dlq/request_test.go | 28 --- .../orchestrator/controller/start/start.go | 67 ++++-- .../controller/start/start_test.go | 220 ++++++++++++++---- submitqueue/orchestrator/pipeline.go | 2 +- submitqueue/orchestrator/pipeline_test.go | 72 ++++++ .../submitqueue/orchestrator/BUILD.bazel | 11 + .../submitqueue/orchestrator/suite_test.go | 63 ++++- 12 files changed, 391 insertions(+), 127 deletions(-) diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md index 7e95e29d4..f360c1ea8 100644 --- a/doc/rfc/submitqueue/modular-queue-wiring.md +++ b/doc/rfc/submitqueue/modular-queue-wiring.md @@ -157,9 +157,8 @@ var Stages = []pipeline.Stage[Deps]{ return start.NewController(d.Logger, d.Scope, d.Storage, d.Queues), nil }, DLQ: func(d Deps) (consumer.Controller, error) { - return dlq.NewRequestController(d.Logger, d.Scope, d.Storage, - dlq.DecodeLandRequestID, dlq.TopicKey(topickey.Start), - "orchestrator-start-dlq"), nil + return start.NewController(d.Logger, d.Scope, d.Storage, d.Queues, + dlq.TopicKey(topickey.Start), "orchestrator-start-dlq"), nil }, }, { Key: topickey.Validate, /* same shape */ }, diff --git a/doc/rfc/submitqueue/workflow.md b/doc/rfc/submitqueue/workflow.md index f07e8f191..6706cdcdc 100644 --- a/doc/rfc/submitqueue/workflow.md +++ b/doc/rfc/submitqueue/workflow.md @@ -4,7 +4,7 @@ The orchestrator processes land requests through the queue-driven pipeline decla Two cycles re-enter speculation. `speculate` dispatches funded paths to `build`; `build` starts each pending path and publishes the build id to `buildsignal`; `buildsignal` polls with `delivery.Hold` and, when the status changes, publishes the batch id back to `speculate`. A batch that can land is published to `submitqueue-land`, which sends the merge request to Runway; `landsignal` correlates the result, marks the batch Succeeded or Failed, and fans out to `conclude` and `speculate`. `speculate` also publishes Failed and Cancelled batches straight to `conclude`. -Terminal request states are not `conclude`'s alone. `cancel` completes a request that has not been enrolled in a batch. `landconflictsignal` fails a request Runway reports as conflicted. `conclude` maps a terminal batch onto its member requests. The DLQ reconcilers force a failed terminal state when a consumed stage cannot finish. +Terminal request states are not `conclude`'s alone. `cancel` completes a request that has not been enrolled in a batch. `landconflictsignal` fails a request Runway reports as conflicted. `conclude` maps a terminal batch onto its member requests. Most DLQ reconcilers force a failed terminal state when a consumed stage cannot finish; start resumes the 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. @@ -139,7 +139,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 | +| **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 | | **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 | @@ -162,7 +162,7 @@ Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscriptio Most DLQ controllers do not re-attempt the failed work. They decode the payload to a `RequestID` or `BatchID` and drive that entity to a terminal failed state — `RequestStateError` for requests, `BatchStateFailed` for batches, with fan-out to the member requests. Topics that carry a full payload recover the id from it: the `landconflictsignal` DLQ reads the request id from Runway's `MergeResult`, and the `landsignal` DLQ reads the batch id. The `buildsignal` DLQ decodes a build id, loads that build, and fails the batch that owns it. A missing build or an empty batch id is acked: there is no batch to fail from this signal. State writes use the same optimistic-locking CAS as the primary pipeline, so a late primary-pipeline update wins cleanly and a version mismatch is asked back for redelivery. -Two consumed stages do not follow that shape. The speculate DLQ fails the batches the failure attributes — often not the batch named on the message, because a run re-plans the whole queue — and then republishes to `speculate`, so one dead letter does not strand every other batch. The hook DLQ records the dropped event and acks. A hook never writes pipeline state, so there is no request or batch to fail; republishing the logged event is how an operator recovers the side effect. A genuinely unprocessable request or batch DLQ message, typically a malformed payload, must be removed by an operator, because those consumers retry until reconciliation succeeds. +Three consumed stages do not follow that shape. The 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. See `submitqueue/orchestrator/controller/dlq/README.md` for the reconcile-only constraints. The speculate and hook controllers in that wiring are the exceptions to the generic request/batch mapping that README's table still summarizes. diff --git a/submitqueue/orchestrator/BUILD.bazel b/submitqueue/orchestrator/BUILD.bazel index fff290a6b..ec6a896e2 100644 --- a/submitqueue/orchestrator/BUILD.bazel +++ b/submitqueue/orchestrator/BUILD.bazel @@ -45,12 +45,24 @@ go_test( embed = [":go_default_library"], deps = [ "//api/base/hook:go_default_library", + "//platform/base/change:go_default_library", + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/hook:go_default_library", "//platform/extension/hook/noop:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", "//platform/pipeline:go_default_library", + "//submitqueue/core/messagequeue:go_default_library", + "//submitqueue/core/topickey:go_default_library", + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/storage/mock:go_default_library", + "//submitqueue/orchestrator/extension/storage:go_default_library", + "//submitqueue/orchestrator/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", "@org_uber_go_zap//zaptest:go_default_library", ], ) diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index 0ebbb2556..329ff178c 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -1,6 +1,6 @@ # 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 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. ## Design principles @@ -8,7 +8,9 @@ This package contains the controllers that drain each primary pipeline topic's ` **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 only; do not attempt to recover.** A DLQ controller never tries to re-run the failed stage, retry the original action against an external dependency, or repair a partially-completed transition. The original controller already failed; the DLQ controller's only job is to make sure the request and batch state stop saying "in progress" and start saying "failed", so the gateway reports an honest answer and downstream tooling can move on. Recovery, when it is appropriate, is a separate concern handled by an operator or a future reconciliation job — not by this code path. +**Reconcile 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. + +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. ## Convergence guarantee @@ -18,7 +20,7 @@ The recognised error condition is handled explicitly in `dlq.go`: - `storage.ErrNotFound` → logged at warn and treated as success. The request or batch never persisted; there is nothing to reconcile. -Everything else — including `storage.ErrVersionMismatch` on the CAS — is returned without controller-level classification and redelivered until it either succeeds or hits the attempt cap. `ErrVersionMismatch` is already retryable at its declaration, while the always-retryable processor makes every other non-nil reconciliation error retryable too. +Everything else — including `storage.ErrVersionMismatch` on the CAS — is returned without controller-level classification and redelivered until reconciliation succeeds. `ErrVersionMismatch` is already retryable at its declaration, while the always-retryable processor makes every other non-nil reconciliation error retryable too. ## Request log entries are published to Gateway @@ -30,10 +32,10 @@ Two controller shapes cover the eleven primary pipeline topics: | Controller | Topics | Decoded ID | Terminal state | |---|---|---|---| -| `NewDLQRequestController` | `start`, `validate`, `batch`, `cancel`, `log` | `RequestID` | `RequestStateError` | +| `NewDLQRequestController` | `validate`, `batch`, `cancel` | `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 `DecodeLandRequestID` / `DecodeCancelRequestID` / `DecodeRequestID` cover the three payload shapes used by request-scoped topics. +`buildsignal` carries a `Build` payload and has its own small dedicated controller. The split exists because the DLQ message payload shape mirrors the primary topic's payload shape (the queue framework preserves bytes verbatim under the `_dlq` topic name), so 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. ## Idempotency and concurrent activity diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go index 6e6fc51f1..f4bc54288 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -34,22 +34,12 @@ import ( // 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 +// 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. type RequestIDDecoder func(payload []byte) (entity.RequestID, error) -// DecodeLandRequestID extracts the request ID from a LandRequest payload -// (the shape used by the start topic). -func DecodeLandRequestID(payload []byte) (entity.RequestID, error) { - lr, err := sqmq.UnmarshalLandRequest(payload) - if err != nil { - return entity.RequestID{}, err - } - 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). func DecodeCancelRequestID(payload []byte) (entity.RequestID, error) { @@ -69,10 +59,8 @@ 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 (start, 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 live batch already owns its outcome. type requestController struct { logger *zap.SugaredLogger metricsScope tally.Scope diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go index 5cc6c6f3a..d02b2cae9 100644 --- a/submitqueue/orchestrator/controller/dlq/request_test.go +++ b/submitqueue/orchestrator/controller/dlq/request_test.go @@ -46,34 +46,6 @@ func TestDLQRequestController_InterfaceAndAccessors(t *testing.T) { assert.Equal(t, "orchestrator-validate-dlq", c.ConsumerGroup()) } -func TestDLQRequestController_Process_LandRequestPayload(t *testing.T) { - ctrl := gomock.NewController(t) - - requestStore := storagemock.NewMockRequestStore(ctrl) - request := entity.Request{ - ID: "1", Queue: "q", Version: 1, State: entity.RequestStateStarted, - } - requestStore.EXPECT().Get(gomock.Any(), "1").Return(request, nil) - requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).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, DecodeLandRequestID, TopicKey(topickey.TopicKeyStart), "orchestrator-start-dlq") - - payload, err := sqmq.Marshal(sqmq.StartFromLandRequest(entity.LandRequest{ID: "1", Queue: "q"})) - require.NoError(t, err) - - delivery := newMockDelivery(ctrl, payload) - require.NoError(t, c.Process(context.Background(), delivery)) -} - func TestDLQRequestController_Process_TenantPayloadQueueMismatchAcks(t *testing.T) { ctrl := gomock.NewController(t) store := orchstoragemock.NewMockStorage(ctrl) diff --git a/submitqueue/orchestrator/controller/start/start.go b/submitqueue/orchestrator/controller/start/start.go index 85c415bf9..9582715cf 100644 --- a/submitqueue/orchestrator/controller/start/start.go +++ b/submitqueue/orchestrator/controller/start/start.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "slices" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -33,12 +34,8 @@ import ( "go.uber.org/zap" ) -// Controller handles start queue messages. -// It consumes requests, persists them to the request store, and publishes to the -// validate stage. The request write is idempotent on retries. URI claiming and -// duplicate detection are performed downstream by the validate controller (which -// claims each URI in the change store with its provider details and consults the -// request store for liveness). Implements consumer.Controller. +// Controller persists start requests and repairs their log and validate hand-off. +// Primary and final-DLQ deliveries preserve the persisted request's current owner. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -88,6 +85,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("invalid message identity: %w", err) } + if landRequest.ID == "" { + return fmt.Errorf("land request id is required") + } + store, err := c.stores.For(orchstorage.Config{QueueName: landRequest.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_resolve_errors", 1, metrics.TagsFromContext(ctx)...) @@ -116,21 +117,29 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "partition_key", msg.PartitionKey, ) - // Persist request to storage. ErrAlreadyExists means a queue redelivery of the same - // request_id (an at-least-once retry of THIS message), not a cross-request collision. - if err := store.GetRequestStore().Create(ctx, request); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + request, err = createOrLoadRequest(ctx, store.GetRequestStore(), request) + if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) - return fmt.Errorf("failed to create request: %w", err) + return err + } + + switch request.State { + case entity.RequestStateStarted: + case entity.RequestStateValidated, entity.RequestStateBatched, entity.RequestStateProcessing, entity.RequestStateCancelling: + return nil + case entity.RequestStateLanded, entity.RequestStateError, entity.RequestStateCancelled: + return corerequest.PublishTerminalRequestLog(ctx, c.registry, request) + default: + return fmt.Errorf("unsupported request state %q for request %s", request.State, request.ID) } - // Record the "new" status in the request log. logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusStarted, request.Version, "", nil) if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID, ""); err != nil { metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1, metrics.TagsFromContext(ctx)...) return fmt.Errorf("failed to publish request log: %w", err) } - if err := c.publish(ctx, topickey.TopicKeyValidate, request.ID, request.Queue); err != nil { + if err := c.publishToValidate(ctx, request); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1, metrics.TagsFromContext(ctx)...) return fmt.Errorf("failed to publish to validate: %w", err) } @@ -143,22 +152,34 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return nil } -// publish publishes a request ID to the specified topic key, stamped with and -// partitioned by the request's queue. -// -// The request ID is the message ID with no cause: a request is handed to the -// next stage once, so a redelivery that re-hands it is meant to dedup away. -func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, requestID string, queue string) error { - payload, err := sqmq.MarshalID(key, requestID, queue) +func createOrLoadRequest(ctx context.Context, store storage.RequestStore, initial entity.Request) (entity.Request, error) { + if err := store.Create(ctx, initial); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return entity.Request{}, fmt.Errorf("failed to create request: %w", err) + } + // Create can lose to a delayed attempt or a later stage; only the persisted snapshot owns replay. + request, err := store.Get(ctx, initial.ID) + if err != nil { + return entity.Request{}, fmt.Errorf("failed to load request after create: %w", err) + } + if request.ID != initial.ID || request.Queue != initial.Queue || + request.LandStrategy != initial.LandStrategy || !slices.Equal(request.Change.URIs, initial.Change.URIs) { + return entity.Request{}, fmt.Errorf("conflicting immutable inputs for request %s in queue %q", initial.ID, initial.Queue) + } + return request, nil +} + +// publishToValidate uses one stable intent ID so replay deduplicates the hand-off. +func (c *Controller) publishToValidate(ctx context.Context, request entity.Request) error { + payload, err := sqmq.MarshalID(topickey.TopicKeyValidate, request.ID, request.Queue) if err != nil { return fmt.Errorf("failed to serialize request ID: %w", err) } - if err := publish.Message(ctx, c.registry, key, publish.MessageParams{ - Tenant: queue, - ID: publish.IntentID(requestID), + if err := publish.Message(ctx, c.registry, topickey.TopicKeyValidate, publish.MessageParams{ + Tenant: request.Queue, + ID: publish.IntentID(request.ID), Payload: payload, - PartitionKey: queue, + PartitionKey: request.Queue, }); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/orchestrator/controller/start/start_test.go b/submitqueue/orchestrator/controller/start/start_test.go index 0b7f027be..2fc6403c3 100644 --- a/submitqueue/orchestrator/controller/start/start_test.go +++ b/submitqueue/orchestrator/controller/start/start_test.go @@ -53,7 +53,7 @@ func newTestController( t *testing.T, ctrl *gomock.Controller, store *orchstoragemock.MockStorage, - publishErr error, + publishMessage func(string, entityqueue.Message) error, ) *Controller { logger := zaptest.NewLogger(t).Sugar() scope := tally.NoopScope @@ -61,7 +61,10 @@ func newTestController( mockPub := queuemock.NewMockPublisher(ctrl) mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( func(ctx context.Context, topic string, msg entityqueue.Message) error { - return publishErr + if publishMessage != nil { + return publishMessage(topic, msg) + } + return nil }, ).AnyTimes() @@ -79,10 +82,17 @@ func newTestController( return NewController(logger, scope, staticStorageFactory{store: store}, registry, topickey.TopicKeyStart, "orchestrator-start") } -// newMockStorage creates a MockStorage with a MockRequestStore that succeeds on Create. +// newMockStorage retains the created snapshot for reconciliation reads. func newMockStorage(ctrl *gomock.Controller) *orchstoragemock.MockStorage { mockReqStore := storagemock.NewMockRequestStore(ctrl) - mockReqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + var request entity.Request + mockReqStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, req entity.Request) error { + request = req + return nil + }).AnyTimes() + mockReqStore.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, _ string) (entity.Request, error) { + return request, nil + }).AnyTimes() store := orchstoragemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes() @@ -143,19 +153,20 @@ func TestController_Process_RejectsTenantPayloadQueueMismatch(t *testing.T) { assert.False(t, errs.IsRetryable(err)) } -func TestController_Process_InvalidJSON(t *testing.T) { - ctrl := gomock.NewController(t) - controller := newTestController(t, ctrl, newMockStorage(ctrl), nil) - - invalidPayload := []byte(`{"invalid": json"}`) - msg := entityqueue.NewMessage("invalid-msg", invalidPayload, "partition1", nil) - delivery := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() - - err := controller.Process(context.Background(), delivery) - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) +func TestController_Process_InvalidPayload(t *testing.T) { + for _, payload := range []string{`{"invalid": json"}`, `{"queue": "test-queue"}`} { + t.Run(payload, func(t *testing.T) { + ctrl := gomock.NewController(t) + controller := newTestController(t, ctrl, newMockStorage(ctrl), nil) + msg := entityqueue.NewMessage("invalid-msg", []byte(payload), "test-queue", nil) + msg.Tenant = "test-queue" + delivery := consumermock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg).AnyTimes() + err := controller.Process(context.Background(), delivery) + require.Error(t, err) + assert.False(t, errs.IsRetryable(err)) + }) + } } func TestController_Process_ConstructsRequestWithStateAndVersion(t *testing.T) { @@ -169,6 +180,9 @@ func TestController_Process_ConstructsRequestWithStateAndVersion(t *testing.T) { return nil }, ) + mockReqStore.EXPECT().Get(gomock.Any(), "42").DoAndReturn(func(_ context.Context, _ string) (entity.Request, error) { + return capturedRequest, nil + }) store := orchstoragemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes() @@ -219,18 +233,65 @@ func TestController_Process_AllStrategies(t *testing.T) { } } -func TestController_Process_PublishFailure(t *testing.T) { - ctrl := gomock.NewController(t) - controller := newTestController(t, ctrl, newMockStorage(ctrl), fmt.Errorf("publish failed")) - - delivery := makeDelivery(t, ctrl, entity.LandRequest{ - ID: "123", - Queue: "test-queue", - Change: change.Change{URIs: []string{"github://github.example.com/uber/service/pull/1/789abc1234567890abcdef1234567890abcdef12"}}, - LandStrategy: mergestrategy.MergeStrategyRebase, - }) - - assert.Error(t, controller.Process(context.Background(), delivery)) +func TestController_Process_ReplaysFailedHandOff(t *testing.T) { + for _, topic := range []consumer.TopicKey{topickey.TopicKeyStart, "start_dlq"} { + for _, failure := range []string{"create", "get", "log", "validate", "lost acknowledgment"} { + t.Run(string(topic)+"/"+failure, func(t *testing.T) { + ctrl := gomock.NewController(t) + landRequest := entity.LandRequest{ID: "123", Queue: "test-queue", Change: change.Change{URIs: []string{"uri/one", "uri/two"}}, LandStrategy: mergestrategy.MergeStrategyRebase} + var persisted entity.Request + firstAttempt := true + failureErr := fmt.Errorf("dependency failed") + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, request entity.Request) error { + if persisted.ID != "" { + return storage.ErrAlreadyExists + } + persisted = request + if firstAttempt && failure == "create" { + return failureErr // The insert succeeded but its response was lost. + } + return nil + }).AnyTimes() + requestStore.EXPECT().Get(gomock.Any(), landRequest.ID).DoAndReturn(func(_ context.Context, _ string) (entity.Request, error) { + if firstAttempt && failure == "get" { + return entity.Request{}, failureErr + } + return persisted, nil + }).AnyTimes() + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + var published []string + controller := newTestController(t, ctrl, store, func(topic string, msg entityqueue.Message) error { + published = append(published, topic) + if topic == "validate" { + id, err := sqmq.UnmarshalRequestID(topickey.TopicKeyValidate, msg.Payload) + require.NoError(t, err) + assert.Equal(t, entity.RequestID{ID: landRequest.ID, Queue: landRequest.Queue}, id) + } + if firstAttempt && topic == failure { + return failureErr + } + return nil + }) + controller.topicKey = topic + delivery := makeDelivery(t, ctrl, landRequest) + err := controller.Process(context.Background(), delivery) + if failure == "lost acknowledgment" { + require.NoError(t, err) + } else { + require.ErrorIs(t, err, failureErr) + } + firstAttempt = false + published = nil + require.NoError(t, controller.Process(context.Background(), delivery)) + assert.Equal(t, []string{"log", "validate"}, published) + assert.Equal(t, landRequest.Change, persisted.Change) + assert.Equal(t, landRequest.LandStrategy, persisted.LandStrategy) + assert.Equal(t, int32(1), persisted.Version) + }) + } + } } func TestController_Process_StorageFailure(t *testing.T) { @@ -255,24 +316,91 @@ func TestController_Process_StorageFailure(t *testing.T) { assert.False(t, errs.IsRetryable(err)) } -func TestController_Process_AlreadyExistsSucceeds(t *testing.T) { - ctrl := gomock.NewController(t) - - mockReqStore := storagemock.NewMockRequestStore(ctrl) - mockReqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(fmt.Errorf("duplicate: %w", storage.ErrAlreadyExists)) - store := orchstoragemock.NewMockStorage(ctrl) - store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes() - - controller := newTestController(t, ctrl, store, nil) - - delivery := makeDelivery(t, ctrl, entity.LandRequest{ - ID: "123", - Queue: "test-queue", - Change: change.Change{URIs: []string{"github://github.example.com/uber/service/pull/1/789abc1234567890abcdef1234567890abcdef12"}}, - LandStrategy: mergestrategy.MergeStrategyRebase, - }) +func TestController_Process_AuthoritativeState(t *testing.T) { + for _, topic := range []consumer.TopicKey{topickey.TopicKeyStart, "start_dlq"} { + for _, state := range []entity.RequestState{entity.RequestStateStarted, entity.RequestStateValidated, entity.RequestStateBatched, entity.RequestStateProcessing, entity.RequestStateCancelling, entity.RequestStateLanded, entity.RequestStateError, entity.RequestStateCancelled, entity.RequestStateUnknown} { + t.Run(string(topic)+"/"+string(state), func(t *testing.T) { + ctrl := gomock.NewController(t) + landRequest := entity.LandRequest{ID: "123", Queue: "test-queue", Change: change.Change{URIs: []string{"uri/one", "uri/two"}}, LandStrategy: mergestrategy.MergeStrategyRebase} + persisted := entity.Request{ID: landRequest.ID, Queue: landRequest.Queue, Change: landRequest.Change, LandStrategy: landRequest.LandStrategy, State: state, Version: 9, TerminalDetails: entity.RequestTerminalDetails{TimestampMs: 12345, LastError: "winning diagnostic", Metadata: map[string]string{"cause": "stored"}}} + requestStore := storagemock.NewMockRequestStore(ctrl) + var createErr error + if topic != topickey.TopicKeyStart { + createErr = fmt.Errorf("duplicate: %w", storage.ErrAlreadyExists) + } + requestStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(createErr) + requestStore.EXPECT().Get(gomock.Any(), landRequest.ID).Return(persisted, nil) + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + var published []string + controller := newTestController(t, ctrl, store, func(topic string, msg entityqueue.Message) error { + published = append(published, topic) + if topic == "log" { + log, err := sqmq.UnmarshalRequestLog(msg.Payload) + require.NoError(t, err) + assert.Equal(t, persisted.Version, log.RequestVersion) + if entity.IsRequestStateTerminal(state) { + assert.Equal(t, entity.RequestStatus(state), log.Status) + assert.Equal(t, persisted.TerminalDetails.TimestampMs, log.TimestampMs) + assert.Equal(t, persisted.TerminalDetails.LastError, log.LastError) + assert.Equal(t, persisted.TerminalDetails.Metadata, log.Metadata) + } else { + assert.Equal(t, entity.RequestStatusStarted, log.Status) + } + } + return nil + }) + controller.topicKey = topic + err := controller.Process(context.Background(), makeDelivery(t, ctrl, landRequest)) + switch { + case state == entity.RequestStateUnknown: + require.Error(t, err) + case state == entity.RequestStateStarted: + require.NoError(t, err) + assert.Equal(t, []string{"log", "validate"}, published) + case entity.IsRequestStateTerminal(state): + require.NoError(t, err) + assert.Equal(t, []string{"log"}, published) + default: + require.NoError(t, err) + assert.Empty(t, published) + } + }) + } + } +} - require.NoError(t, controller.Process(context.Background(), delivery)) +func TestController_Process_RejectsInconsistentSnapshot(t *testing.T) { + landRequest := entity.LandRequest{ID: "123", Queue: "test-queue", Change: change.Change{URIs: []string{"uri/one", "uri/two"}}, LandStrategy: mergestrategy.MergeStrategyRebase} + for _, conflict := range []string{"id", "queue", "uri order", "strategy", "missing snapshot"} { + t.Run(conflict, func(t *testing.T) { + ctrl := gomock.NewController(t) + persisted := entity.Request{ID: landRequest.ID, Queue: landRequest.Queue, Change: landRequest.Change, LandStrategy: landRequest.LandStrategy, State: entity.RequestStateStarted, Version: 1} + var getErr error + switch conflict { + case "id": + persisted.ID = "456" + case "queue": + persisted.Queue = "other" + case "uri order": + persisted.Change.URIs = []string{"uri/two", "uri/one"} + case "strategy": + persisted.LandStrategy = mergestrategy.MergeStrategyMerge + case "missing snapshot": + getErr = storage.ErrNotFound + } + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + requestStore.EXPECT().Get(gomock.Any(), landRequest.ID).Return(persisted, getErr) + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + controller := newTestController(t, ctrl, store, func(string, entityqueue.Message) error { + require.Fail(t, "inconsistent snapshot must not publish") + return nil + }) + require.Error(t, controller.Process(context.Background(), makeDelivery(t, ctrl, landRequest))) + }) + } } func TestController_InterfaceImplementation(t *testing.T) { diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index ed955125a..cff5301b6 100644 --- a/submitqueue/orchestrator/pipeline.go +++ b/submitqueue/orchestrator/pipeline.go @@ -109,7 +109,7 @@ var Stages = []pipeline.Stage[Deps]{ return start.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.DecodeLandRequestID, sc.TopicKey, sc.ConsumerGroup), nil + return start.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, }, { diff --git a/submitqueue/orchestrator/pipeline_test.go b/submitqueue/orchestrator/pipeline_test.go index 780586674..c0e34745b 100644 --- a/submitqueue/orchestrator/pipeline_test.go +++ b/submitqueue/orchestrator/pipeline_test.go @@ -15,15 +15,28 @@ package orchestrator import ( + "context" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/platform/base/change" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" hookext "github.com/uber/submitqueue/platform/extension/hook" hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/platform/pipeline" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" + orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock" + "go.uber.org/mock/gomock" "go.uber.org/zap/zaptest" ) @@ -78,3 +91,62 @@ func TestHookStage(t *testing.T) { require.Error(t, err) }) } + +type startStorageFactory struct{ store orchstorage.Storage } + +func (f startStorageFactory) For(orchstorage.Config) (orchstorage.Storage, error) { + return f.store, nil +} + +func TestStartDLQRecoversOriginalRequest(t *testing.T) { + ctrl := gomock.NewController(t) + landRequest := entity.LandRequest{ID: "123", Queue: "test-queue", Change: change.Change{URIs: []string{"uri/one", "uri/two"}}} + var persisted entity.Request + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, request entity.Request) error { + persisted = request + return nil + }) + requestStore.EXPECT().Get(gomock.Any(), landRequest.ID).DoAndReturn(func(_ context.Context, _ string) (entity.Request, error) { + return persisted, nil + }) + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + publisher := queuemock.NewMockPublisher(ctrl) + gomock.InOrder( + publisher.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).Return(nil), + publisher.EXPECT().Publish(gomock.Any(), "validate", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + id, err := sqmq.UnmarshalRequestID(topickey.TopicKeyValidate, msg.Payload) + require.NoError(t, err) + assert.Equal(t, entity.RequestID{ID: landRequest.ID, Queue: landRequest.Queue}, id) + return nil + }), + ) + queue := queuemock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyLog, Name: "log", Queue: queue}, + {Key: topickey.TopicKeyValidate, Name: "validate", Queue: queue}, + }) + require.NoError(t, err) + for _, stage := range Stages { + if stage.Key != topickey.TopicKeyStart { + continue + } + controller, err := stage.DLQ(Deps{Logger: zaptest.NewLogger(t).Sugar(), Scope: tally.NoopScope, Storage: startStorageFactory{store}}, pipeline.StageContext{Registry: registry, TopicKey: "start_dlq", ConsumerGroup: "orchestrator-dlq"}) + require.NoError(t, err) + assert.Equal(t, consumer.TopicKey("start_dlq"), controller.TopicKey()) + payload, err := sqmq.Marshal(sqmq.StartFromLandRequest(landRequest)) + require.NoError(t, err) + msg := entityqueue.NewMessage(landRequest.ID, payload, landRequest.Queue, nil) + msg.Tenant = landRequest.Queue + delivery := consumermock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg) + delivery.EXPECT().Attempt().Return(1) + require.NoError(t, controller.Process(context.Background(), delivery)) + assert.Equal(t, landRequest.Change, persisted.Change) + assert.Equal(t, entity.RequestStateStarted, persisted.State) + return + } + require.Fail(t, "start stage must be registered") +} diff --git a/test/integration/submitqueue/orchestrator/BUILD.bazel b/test/integration/submitqueue/orchestrator/BUILD.bazel index ca5158a79..7757ecc20 100644 --- a/test/integration/submitqueue/orchestrator/BUILD.bazel +++ b/test/integration/submitqueue/orchestrator/BUILD.bazel @@ -16,10 +16,21 @@ 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/extension/messagequeue/mysql: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/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..6f82bd287 100644 --- a/test/integration/submitqueue/orchestrator/suite_test.go +++ b/test/integration/submitqueue/orchestrator/suite_test.go @@ -28,12 +28,24 @@ 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" + queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + 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" + mysqlstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mysql" "github.com/uber/submitqueue/test/testutil" + "go.uber.org/zap" "google.golang.org/grpc" ) @@ -68,8 +80,8 @@ func (s *OrchestratorIntegrationSuite) SetupSuite() { "service/submitqueue/orchestrator/server/Dockerfile": "service/submitqueue/orchestrator/server/Dockerfile", })) - // Start the compose stack (Orchestrator + 2 MySQL DBs) - err := s.stack.Up() + // Schemas and retained work must exist before the service starts. + err := s.stack.Up("mysql-app", "mysql-queue") require.NoError(t, err, "failed to start compose stack") s.log.Logf("Compose stack started successfully") @@ -91,6 +103,19 @@ func (s *OrchestratorIntegrationSuite) SetupSuite() { s.log.Logf("Schemas applied successfully") + requestStore := mysqlstorage.NewRequestStore(s.db, tally.NoopScope, "test-queue") + _, err = requestStore.Get(s.ctx, "911") + require.ErrorIs(t, err, storage.ErrNotFound) + queue, err := queueMySQL.NewQueue(queueMySQL.Params{DB: s.queueDB, Logger: zap.NewNop(), MetricsScope: tally.NoopScope}) + require.NoError(t, err) + payload, err := sqmq.Marshal(sqmq.StartFromLandRequest(retainedStartRequest())) + require.NoError(t, err) + msg := entityqueue.NewMessage("retained-start", payload, "test-queue", nil) + msg.Tenant = "test-queue" + require.NoError(t, queue.Publisher().Publish(s.ctx, "start_dlq", msg)) + require.NoError(t, queue.Close()) + require.NoError(t, s.stack.Up("orchestrator-service")) + // Connect to Orchestrator gRPC service var conn *grpc.ClientConn conn, err = s.stack.ConnectGRPC("orchestrator-service", 8080) @@ -115,3 +140,37 @@ func (s *OrchestratorIntegrationSuite) TestPingAPI() { assert.NotEmpty(t, resp.Message) assert.NotZero(t, resp.Timestamp) } + +func retainedStartRequest() entity.LandRequest { + return entity.LandRequest{ID: "911", Queue: "test-queue", Change: change.Change{URIs: []string{"github://github.example.com/org/repo/pull/11/abcdef0123456789abcdef0123456789abcdef01"}}, LandStrategy: mergestrategy.MergeStrategyRebase} +} + +func (s *OrchestratorIntegrationSuite) TestRetainedStartCreatesRequestAndHandsOffToValidate() { + t := s.T() + landRequest := retainedStartRequest() + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + for { + var acked bool + err := s.queueDB.QueryRowContext(s.ctx, `SELECT acked FROM queue_delivery_state + WHERE tenant = ? AND topic = 'start_dlq' AND consumer_group = 'orchestrator-dlq'`, landRequest.Queue).Scan(&acked) + if err != sql.ErrNoRows { + require.NoError(t, err) + if acked { + break + } + } + <-ticker.C + } + request, err := mysqlstorage.NewRequestStore(s.db, tally.NoopScope, landRequest.Queue).Get(s.ctx, landRequest.ID) + require.NoError(t, err) + assert.Equal(t, landRequest.ID, request.ID) + assert.Equal(t, landRequest.Queue, request.Queue) + assert.Equal(t, landRequest.Change, request.Change) + assert.Equal(t, landRequest.LandStrategy, request.LandStrategy) + var payload []byte + require.NoError(t, s.queueDB.QueryRowContext(s.ctx, `SELECT payload FROM queue_messages WHERE tenant = ? AND topic = 'validate'`, landRequest.Queue).Scan(&payload)) + id, err := sqmq.UnmarshalRequestID(topickey.TopicKeyValidate, payload) + require.NoError(t, err) + assert.Equal(t, entity.RequestID{ID: request.ID, Queue: request.Queue}, id) +} From 0d365e1e195f8e6fc0e7f5c35550ec9b0e05082d Mon Sep 17 00:00:00 2001 From: sergeyb Date: Fri, 6 Nov 2026 22:54:03 +0000 Subject: [PATCH 2/2] test(start): observe recovery after consumed-message collection Summary: Replace transient queue receipt and validate-message assertions with the authoritative request and unconsumed cross-service hand-off. Complete the consumers using durable acknowledgment watermarks, then explicitly remove consumed evidence to exercise garbage-collected queues without timing or test-order dependencies. Test Plan: TestRetainedStartCreatesRequestAndHandsOffToValidate passes with consumed start and validate evidence removed, preserving the request ID, queue, ordered changes, and merge strategy in the downstream request. All 19 orchestrator and service integration targets pass. Revert Plan: Revert this test-only change; production recovery behavior is unchanged. --- .../submitqueue/orchestrator/BUILD.bazel | 3 +- .../submitqueue/orchestrator/suite_test.go | 38 ++++++++++++------- 2 files changed, 27 insertions(+), 14 deletions(-) diff --git a/test/integration/submitqueue/orchestrator/BUILD.bazel b/test/integration/submitqueue/orchestrator/BUILD.bazel index 7757ecc20..0d78fd34c 100644 --- a/test/integration/submitqueue/orchestrator/BUILD.bazel +++ b/test/integration/submitqueue/orchestrator/BUILD.bazel @@ -15,13 +15,14 @@ go_test( "requires-network", ], deps = [ + "//api/base/mergestrategy/protopb:go_default_library", + "//api/runway/messagequeue:go_default_library", "//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/extension/messagequeue/mysql: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/extension/storage/mysql:go_default_library", diff --git a/test/integration/submitqueue/orchestrator/suite_test.go b/test/integration/submitqueue/orchestrator/suite_test.go index 6f82bd287..64003a02f 100644 --- a/test/integration/submitqueue/orchestrator/suite_test.go +++ b/test/integration/submitqueue/orchestrator/suite_test.go @@ -34,13 +34,14 @@ import ( "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "github.com/uber-go/tally" + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" 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" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" 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" mysqlstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mysql" @@ -150,18 +151,23 @@ func (s *OrchestratorIntegrationSuite) TestRetainedStartCreatesRequestAndHandsOf landRequest := retainedStartRequest() ticker := time.NewTicker(100 * time.Millisecond) defer ticker.Stop() + // Watermarks survive GC; complete both consumers before removing their transient evidence. for { - var acked bool - err := s.queueDB.QueryRowContext(s.ctx, `SELECT acked FROM queue_delivery_state - WHERE tenant = ? AND topic = 'start_dlq' AND consumer_group = 'orchestrator-dlq'`, landRequest.Queue).Scan(&acked) - if err != sql.ErrNoRows { - require.NoError(t, err) - if acked { - break - } + var completed int + err := s.queueDB.QueryRowContext(s.ctx, `SELECT COUNT(*) FROM queue_offsets + WHERE tenant = ? AND partition_key = ? AND offset_acked > 0 + AND ((topic = 'start_dlq' AND consumer_group = 'orchestrator-dlq') + OR (topic = 'validate' AND consumer_group = 'orchestrator'))`, landRequest.Queue, landRequest.Queue).Scan(&completed) + require.NoError(t, err) + if completed == 2 { + break } <-ticker.C } + for _, table := range []string{"queue_delivery_state", "queue_messages"} { + _, err := s.queueDB.ExecContext(s.ctx, "DELETE FROM "+table+" WHERE tenant = ? AND partition_key = ? AND topic IN ('start_dlq', 'validate')", landRequest.Queue, landRequest.Queue) + require.NoError(t, err) + } request, err := mysqlstorage.NewRequestStore(s.db, tally.NoopScope, landRequest.Queue).Get(s.ctx, landRequest.ID) require.NoError(t, err) assert.Equal(t, landRequest.ID, request.ID) @@ -169,8 +175,14 @@ func (s *OrchestratorIntegrationSuite) TestRetainedStartCreatesRequestAndHandsOf assert.Equal(t, landRequest.Change, request.Change) assert.Equal(t, landRequest.LandStrategy, request.LandStrategy) var payload []byte - require.NoError(t, s.queueDB.QueryRowContext(s.ctx, `SELECT payload FROM queue_messages WHERE tenant = ? AND topic = 'validate'`, landRequest.Queue).Scan(&payload)) - id, err := sqmq.UnmarshalRequestID(topickey.TopicKeyValidate, payload) - require.NoError(t, err) - assert.Equal(t, entity.RequestID{ID: request.ID, Queue: request.Queue}, id) + // This fixture has no Runway consumer, so the cross-service hand-off remains observable. + require.NoError(t, s.queueDB.QueryRowContext(s.ctx, `SELECT payload FROM queue_messages WHERE tenant = ? AND topic = 'merge-conflict-check' AND id = ?`, landRequest.Queue, landRequest.ID).Scan(&payload)) + check := &runwaymq.MergeRequest{} + require.NoError(t, runwaymq.Unmarshal(payload, check)) + assert.Equal(t, request.ID, check.GetId()) + assert.Equal(t, request.Queue, check.GetQueueName()) + require.Len(t, check.GetSteps(), 1) + assert.Equal(t, request.ID, check.GetSteps()[0].GetStepId()) + assert.Equal(t, request.Change.URIs, check.GetSteps()[0].GetChange().GetUris()) + assert.Equal(t, mergestrategypb.Strategy_REBASE, check.GetSteps()[0].GetStrategy()) }