diff --git a/doc/rfc/submitqueue/workflow.md b/doc/rfc/submitqueue/workflow.md index c681a40f4..a050b2357 100644 --- a/doc/rfc/submitqueue/workflow.md +++ b/doc/rfc/submitqueue/workflow.md @@ -162,6 +162,8 @@ Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscriptio DLQ handlers do not resume the failed request's normal workflow. They reconcile terminal state and its outputs, preserving authoritative terminal winners. The generic request reconciler also preserves valid downstream batch ownership. The [DLQ package guide](../../../submitqueue/orchestrator/controller/dlq/README.md) defines the reconcile-only boundary, handler-specific dispositions, and named limits; the [stage table](../../../submitqueue/orchestrator/pipeline.go) defines the bindings. Unlimited final retries protect reconciliation, not another business retry budget. +The 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 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/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index 68374f9ad..c6dad4544 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -11,6 +11,7 @@ go_library( "log.go", "request.go", "speculate.go", + "start.go", ], importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/dlq", visibility = ["//visibility:public"], @@ -44,12 +45,15 @@ go_test( "publisher_test.go", "request_test.go", "speculate_test.go", + "start_test.go", ], embed = [":go_default_library"], deps = [ "//api/runway/messagequeue:go_default_library", "//api/runway/messagequeue/protopb:go_default_library", + "//platform/base/change:go_default_library", "//platform/base/failure:go_default_library", + "//platform/base/mergestrategy:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index 523d4fbcf..3cadad906 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -16,6 +16,8 @@ Dead letters have exhausted the primary retry budget or returned a non-retryable Disposition remains handler-specific: the generic request/batch reconcilers acknowledge missing entities; buildsignal also acknowledges a missing build or empty batch ID. Decode errors returned by these handlers remain retryable and may need operator removal. Payload/tenant mismatches are acknowledged, while invalid envelope tenant metadata is rejected before controller error processing. The shared hook handler reports and acknowledges even undecodable events. Final retry policy therefore does not promise retention of every poison message or recreation of every missing entity. +When the Request is absent, 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. + ## 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. @@ -29,7 +31,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`, `cancel`, `validate`, and `batch` use generic request reconciliation. Their decoders read LandRequest, CancelRequest, or the topic's request-ID payload. +- `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. - `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 6e6fc51f1..9e6d5e709 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -69,8 +69,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 +// 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. type requestController struct { @@ -164,7 +164,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv } if owner != "" { metrics.NamedCounter(c.metricsScope, opName, "skipped_owned_by_batch", 1, metrics.TagsFromContext(ctx)...) - c.logger.Infow("dlq reconcile: request is carried by a live batch, leaving the outcome to conclude", + c.logger.Infow("dlq reconcile: batch owns the request outcome, leaving completion to conclude", "request_id", rid.ID, "batch_id", owner, ) @@ -179,53 +179,45 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv return nil } -// owningBatch returns a batch that owns the request's outcome, or the empty -// string if none does. -// -// A non-terminal batch past Creating owns the request outright: conclude writes -// its outcome when the batch finishes. Creating is the ambiguous case, because -// the association is written before the claim that enrols the request, so it -// also survives an abandoned attempt — one whose claim lost to cancel, or found -// the request already halted. Such a batch is inert: nothing promotes it and no -// conclude will ever run for it, so letting it answer "owned" would suppress -// the reconcile forever and strand the request. -// -// The request's own state separates the two, because the claim is the only -// writer of Batched: a Creating batch alongside a Batched request is -// mid-promotion and owns it, and any other request state means the claim never -// landed. A request that has since disappeared is left to failRequest, which -// reports the missing row. +// 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) { batches, _, err := corebatch.FindByRequestID(ctx, store, requestID) if err != nil { return "", err } - creating := "" + owner := "" + needsClaim := false for _, batch := range batches { switch { - case batch.State.IsTerminal(): + case batch.State == entity.BatchStateFailed: + case batch.State == entity.BatchStateSucceeded || batch.State == entity.BatchStateCancelled: + owner, needsClaim = batch.ID, false case batch.State != entity.BatchStateCreating: return batch.ID, nil - case creating == "": - creating = batch.ID + case owner == "": + owner, needsClaim = batch.ID, true } } - if creating == "" { + if owner == "" { return "", nil } request, err := store.GetRequestStore().Get(ctx, requestID) if err != nil { - if errors.Is(err, storage.ErrNotFound) { + if errors.Is(err, storage.ErrNotFound) && needsClaim { return "", nil } + if errors.Is(err, storage.ErrNotFound) { + return owner, nil + } return "", err } - if request.State != entity.RequestStateBatched { + if entity.IsRequestStateTerminal(request.State) || needsClaim && request.State != entity.RequestStateBatched { return "", nil } - return creating, nil + return owner, nil } // Name returns the controller name for logging and metrics. diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go index 5cc6c6f3a..a6773150f 100644 --- a/submitqueue/orchestrator/controller/dlq/request_test.go +++ b/submitqueue/orchestrator/controller/dlq/request_test.go @@ -267,9 +267,8 @@ func TestDLQRequestController_Process_SkipsRequestOwnedByLiveBatch(t *testing.T) } } -// A batch that already concluded does not own anything conclude has not written -// yet, so the request is still the dead letter's to fail. -func TestDLQRequestController_Process_FailsWhenEveryBatchIsTerminal(t *testing.T) { +// A failed batch permits the same Error outcome to be reconciled. +func TestDLQRequestController_Process_FailsWhenAssociatedBatchFailed(t *testing.T) { ctrl := gomock.NewController(t) request := entity.Request{ID: "1", Queue: "q", Version: 1, State: entity.RequestStateBatched} diff --git a/submitqueue/orchestrator/controller/dlq/start.go b/submitqueue/orchestrator/controller/dlq/start.go new file mode 100644 index 000000000..0972d41d9 --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/start.go @@ -0,0 +1,134 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dlq + +import ( + "context" + "errors" + "fmt" + + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + corerequest "github.com/uber/submitqueue/submitqueue/orchestrator/core/request" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" + "go.uber.org/zap" +) + +type startController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores orchstorage.Factory + registry consumer.TopicRegistry + topicKey consumer.TopicKey + consumerGroup string +} + +// NewDLQStartController reconciles failed submissions without restarting processing. +func NewDLQStartController(logger *zap.SugaredLogger, scope tally.Scope, stores orchstorage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string) consumer.Controller { + return &startController{ + logger: logger.Named("start_dlq"), + metricsScope: scope.SubScope("start_dlq"), + stores: stores, + registry: registry, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +func (c *startController) Process(ctx context.Context, delivery consumer.Delivery) (err error) { + defer func() { + if err != nil { + metrics.NamedCounter(c.metricsScope, "process", "reconcile_errors", 1, metrics.TagsFromContext(ctx)...) + } + }() + + msg := delivery.Message() + rid, err := DecodeLandRequestID(msg.Payload) + if err != nil { + return fmt.Errorf("failed to decode start dlq payload: %w", err) + } + if err := entityqueue.ValidatePayloadQueue(msg, rid.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, "process", "queue_identity_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Warnw("invalid start dlq queue identity", "error", err) + return nil + } + if rid.ID == "" { + return fmt.Errorf("dlq payload decoded to empty request id") + } + + store, err := c.stores.For(orchstorage.Config{QueueName: msg.Tenant}) + if err != nil { + return fmt.Errorf("failed to resolve storage for queue %q: %w", rid.Queue, err) + } + lastError, metadata := failureContext(delivery) + c.logger.Warnw("reconciling failed start", "request_id", rid.ID, "queue", rid.Queue, "attempt", delivery.Attempt()) + request, err := store.GetRequestStore().Get(ctx, rid.ID) + switch { + case errors.Is(err, storage.ErrNotFound): + err = c.publishStartFailure(ctx, rid.ID, msg.Tenant, 0, lastError, metadata) + case err == nil: + err = c.failExistingStartRequest(ctx, store, request, rid, lastError, metadata) + } + return err +} + +func (c *startController) failExistingStartRequest(ctx context.Context, store orchstorage.Storage, request entity.Request, rid entity.RequestID, lastError string, metadata map[string]string) error { + if request.ID != rid.ID || request.Queue != rid.Queue { + return fmt.Errorf("conflicting request identity for failed start request %s", rid.ID) + } + // A Start retry can fail after an earlier delivery published validation and the request joined a batch. + // Let that batch finish instead of changing its request to Error. + owner, err := owningBatch(ctx, store, request.ID) + if err != nil || owner != "" { + return err + } + switch request.State { + case entity.RequestStateError: + _, err := corerequest.TerminateRequest(ctx, store, c.registry, request.ID, entity.RequestStateError, lastError, metadata) + return err + case entity.RequestStateLanded, entity.RequestStateCancelled: + return nil + } + updated := request + updated.State = entity.RequestStateError + newVersion := request.Version + 1 + // The observed version guards against a batch claim or terminal winner after the owner lookup. + if err := store.GetRequestStore().Update(ctx, updated, request.Version, newVersion); err != nil { + return fmt.Errorf("failed to fail start request: %w", err) + } + updated.Version = newVersion + return c.publishStartFailure(ctx, updated.ID, updated.Queue, updated.Version, lastError, metadata) +} + +func (c *startController) publishStartFailure(ctx context.Context, requestID, queue string, version int32, lastError string, metadata map[string]string) error { + entry := entity.NewRequestStatusLog(queue, requestID, entity.RequestStatusError, version, lastError, metadata) + return corerequest.PublishLog(ctx, c.registry, entry, requestID, "") +} + +func (c *startController) Name() string { + return string(c.topicKey) +} + +func (c *startController) TopicKey() consumer.TopicKey { + return c.topicKey +} + +func (c *startController) ConsumerGroup() string { + return c.consumerGroup +} diff --git a/submitqueue/orchestrator/controller/dlq/start_test.go b/submitqueue/orchestrator/controller/dlq/start_test.go new file mode 100644 index 000000000..49d9ad0ef --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/start_test.go @@ -0,0 +1,291 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dlq + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/change" + "github.com/uber/submitqueue/platform/base/mergestrategy" + "github.com/uber/submitqueue/platform/consumer" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" + orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap/zaptest" +) + +func TestDLQStartController_ReconcilesOnlyError(t *testing.T) { + tests := []struct { + name string + state entity.RequestState + batchState entity.BatchState + queueMismatch bool + tenant string + wantError bool + wantFailed bool + }{ + {name: "missing", wantFailed: true}, + {name: "started", state: entity.RequestStateStarted, wantFailed: true}, + {name: "validated", state: entity.RequestStateValidated, wantFailed: true}, + {name: "cancelling", state: entity.RequestStateCancelling, wantFailed: true}, + {name: "unowned batched", state: entity.RequestStateBatched, wantFailed: true}, + {name: "unowned processing", state: entity.RequestStateProcessing, wantFailed: true}, + {name: "batch owns outcome", state: entity.RequestStateBatched, batchState: entity.BatchStateSpeculating}, + {name: "succeeded batch awaits conclude", state: entity.RequestStateBatched, batchState: entity.BatchStateSucceeded}, + {name: "cancelled batch awaits conclude", state: entity.RequestStateBatched, batchState: entity.BatchStateCancelled}, + {name: "failed batch permits error", state: entity.RequestStateBatched, batchState: entity.BatchStateFailed, wantFailed: true}, + {name: "request error already won", state: entity.RequestStateError, batchState: entity.BatchStateSucceeded, wantFailed: true}, + {name: "already error", state: entity.RequestStateError, wantFailed: true}, + {name: "landed winner", state: entity.RequestStateLanded}, + {name: "cancelled winner", state: entity.RequestStateCancelled}, + {name: "payload tenant mismatch", state: entity.RequestStateStarted, tenant: "other-queue"}, + {name: "different authoritative queue", state: entity.RequestStateStarted, queueMismatch: true, wantError: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + landRequest := entity.LandRequest{ID: "1", Queue: "q", Change: change.Change{URIs: []string{"uri/one", "uri/two"}}, LandStrategy: mergestrategy.MergeStrategyRebase} + var persisted entity.Request + if tt.state != entity.RequestStateUnknown { + persisted = entity.Request{ID: landRequest.ID, Queue: landRequest.Queue, Change: landRequest.Change, LandStrategy: landRequest.LandStrategy, State: tt.state, Version: 7} + } + if tt.queueMismatch { + persisted.Queue = "other-queue" + } + before := persisted + writes := 0 + requests := storagemock.NewMockRequestStore(ctrl) + requests.EXPECT().Get(gomock.Any(), landRequest.ID).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 { + writes++ + assert.Equal(t, persisted.Version, oldVersion) + assert.Equal(t, oldVersion+1, newVersion) + assert.Equal(t, oldVersion, request.Version) + persisted = request + persisted.Version = newVersion + return nil + }).AnyTimes() + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + associations := storagemock.NewMockRequestBatchStore(ctrl) + var memberships []entity.RequestBatch + if tt.batchState != entity.BatchStateUnknown { + memberships = []entity.RequestBatch{{RequestID: landRequest.ID, BatchID: "batch"}} + batches := storagemock.NewMockBatchStore(ctrl) + batches.EXPECT().Get(gomock.Any(), "batch").Return(entity.Batch{ID: "batch", State: tt.batchState}, nil).AnyTimes() + store.EXPECT().GetBatchStore().Return(batches).AnyTimes() + } + associations.EXPECT().GetByRequestID(gomock.Any(), landRequest.ID).Return(memberships, nil).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(associations).AnyTimes() + publishCount := 0 + if tt.wantFailed { + publishCount = 2 + } + var logs []entity.RequestLog + registry := newTestLogRegistry(t, ctrl, publishCount, func(log entity.RequestLog) error { + if tt.state == entity.RequestStateUnknown { + assert.Equal(t, entity.Request{}, persisted) + } else { + assert.Equal(t, entity.RequestStateError, persisted.State) + } + assert.Equal(t, landRequest.Queue, log.Queue) + assert.Equal(t, landRequest.ID, log.RequestID) + assert.Equal(t, persisted.Version, log.RequestVersion) + assert.Equal(t, entity.RequestStatusError, log.Status) + logs = append(logs, log) + return nil + }) + scope := tally.NewTestScope("", nil) + controller := NewDLQStartController(zaptest.NewLogger(t).Sugar(), scope, staticStorageFactory{store}, registry, TopicKey(topickey.TopicKeyStart), "orchestrator-dlq") + payload, err := sqmq.Marshal(sqmq.StartFromLandRequest(landRequest)) + require.NoError(t, err) + tenant := landRequest.Queue + if tt.tenant != "" { + tenant = tt.tenant + } + for range 2 { + err := controller.Process(context.Background(), newMockDeliveryWithTenant(ctrl, payload, tenant)) + if tt.wantError { + require.Error(t, err) + } else { + require.NoError(t, err) + } + } + counters := scope.Snapshot().Counters() + if tt.wantError { + require.Contains(t, counters, "start_dlq.process.reconcile_errors+") + assert.EqualValues(t, 2, counters["start_dlq.process.reconcile_errors+"].Value()) + } else { + assert.NotContains(t, counters, "start_dlq.process.reconcile_errors+") + } + if !tt.wantFailed { + assert.Equal(t, before, persisted) + assert.Zero(t, writes) + return + } + if tt.state == entity.RequestStateUnknown { + assert.Equal(t, entity.Request{}, persisted) + assert.Zero(t, writes) + assert.Zero(t, logs[0].RequestVersion) + assert.Equal(t, "boom", logs[0].LastError) + return + } + assert.Equal(t, landRequest.Change, persisted.Change) + assert.Equal(t, landRequest.LandStrategy, persisted.LandStrategy) + if tt.state == entity.RequestStateError { + assert.Zero(t, writes) + } else { + assert.Equal(t, 1, writes) + assert.Equal(t, "boom", logs[0].LastError) + } + }) + } +} + +func TestDLQStartController_RetryRepairsFailedWritesAndPublication(t *testing.T) { + for _, fault := range []string{"resolve", "read", "update", "publish", "concurrent batch claim", "concurrent terminal winner"} { + t.Run(fault, func(t *testing.T) { + ctrl := gomock.NewController(t) + landRequest := entity.LandRequest{ID: "1", Queue: "q", Change: change.Change{URIs: []string{"uri/one"}}} + initial := entity.Request{ID: "1", Queue: "q", Change: landRequest.Change, State: entity.RequestStateStarted, Version: 1} + var persisted entity.Request + if fault == "update" || fault == "concurrent batch claim" || fault == "concurrent terminal winner" { + persisted = initial + } + dependencyError := errors.New("dependency failed") + firstAttempt := true + requests := storagemock.NewMockRequestStore(ctrl) + requests.EXPECT().Get(gomock.Any(), "1").DoAndReturn(func(context.Context, string) (entity.Request, error) { + if firstAttempt && fault == "read" { + return entity.Request{}, dependencyError + } + 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 firstAttempt && fault == "update" { + return dependencyError + } + if firstAttempt && (fault == "concurrent batch claim" || fault == "concurrent terminal winner") { + persisted.State = entity.RequestStateBatched + if fault == "concurrent terminal winner" { + persisted.State = entity.RequestStateCancelled + } + persisted.Version++ + return storage.ErrVersionMismatch + } + require.Equal(t, persisted.Version, oldVersion) + persisted = request + persisted.Version = newVersion + return nil + }).AnyTimes() + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + associations := storagemock.NewMockRequestBatchStore(ctrl) + associations.EXPECT().GetByRequestID(gomock.Any(), "1").DoAndReturn(func(context.Context, string) ([]entity.RequestBatch, error) { + if persisted.State == entity.RequestStateBatched { + return []entity.RequestBatch{{RequestID: "1", BatchID: "batch"}}, nil + } + return nil, nil + }).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(associations).AnyTimes() + batches := storagemock.NewMockBatchStore(ctrl) + batches.EXPECT().Get(gomock.Any(), "batch").Return(entity.Batch{ID: "batch", State: entity.BatchStateSpeculating}, nil).AnyTimes() + store.EXPECT().GetBatchStore().Return(batches).AnyTimes() + count := 1 + if fault == "publish" { + count = 2 + } + if fault == "concurrent batch claim" || fault == "concurrent terminal winner" { + count = 0 + } + registry := newTestLogRegistry(t, ctrl, count, func(log entity.RequestLog) error { + if persisted.ID != "" { + assert.Equal(t, entity.RequestStateError, persisted.State) + } + assert.Equal(t, entity.RequestStatusError, log.Status) + assert.Equal(t, persisted.Version, log.RequestVersion) + if firstAttempt && fault == "publish" { + return dependencyError + } + return nil + }) + stores := orchstoragemock.NewMockFactory(ctrl) + stores.EXPECT().For(orchstorage.Config{QueueName: "q"}).DoAndReturn(func(orchstorage.Config) (orchstorage.Storage, error) { + if firstAttempt && fault == "resolve" { + return nil, dependencyError + } + return store, nil + }).AnyTimes() + scope := tally.NewTestScope("", nil) + controller := NewDLQStartController(zaptest.NewLogger(t).Sugar(), scope, stores, registry, TopicKey(topickey.TopicKeyStart), "orchestrator-dlq") + payload, err := sqmq.Marshal(sqmq.StartFromLandRequest(landRequest)) + require.NoError(t, err) + err = controller.Process(context.Background(), newMockDelivery(ctrl, payload)) + require.Error(t, err) + require.Contains(t, scope.Snapshot().Counters(), "start_dlq.process.reconcile_errors+") + assert.EqualValues(t, 1, scope.Snapshot().Counters()["start_dlq.process.reconcile_errors+"].Value()) + firstAttempt = false + require.NoError(t, controller.Process(context.Background(), newMockDelivery(ctrl, payload))) + assert.EqualValues(t, 1, scope.Snapshot().Counters()["start_dlq.process.reconcile_errors+"].Value()) + wantState := entity.RequestStateError + if fault == "resolve" || fault == "read" || fault == "publish" { + wantState = entity.RequestStateUnknown + } + if fault == "concurrent batch claim" { + wantState = entity.RequestStateBatched + } + if fault == "concurrent terminal winner" { + wantState = entity.RequestStateCancelled + } + assert.Equal(t, wantState, persisted.State) + if wantState == entity.RequestStateUnknown { + assert.Equal(t, entity.Request{}, persisted) + } else { + assert.Equal(t, landRequest.Change, persisted.Change) + } + }) + } +} + +func TestDLQStartController_RejectsMalformedPayload(t *testing.T) { + for _, payload := range []string{"not json", `{"queue":"q","change":{"uris":["uri/one"]}}`, `{"id":1,"queue":"q"}`, `{"id":"1","id":"2","queue":"q"}`, `{"id":"1","queue":"q","change":false}`} { + t.Run(payload, func(t *testing.T) { + ctrl := gomock.NewController(t) + scope := tally.NewTestScope("", nil) + controller := NewDLQStartController(zaptest.NewLogger(t).Sugar(), scope, staticStorageFactory{orchstoragemock.NewMockStorage(ctrl)}, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyStart), "orchestrator-dlq") + require.Error(t, controller.Process(context.Background(), newMockDelivery(ctrl, []byte(payload)))) + require.Contains(t, scope.Snapshot().Counters(), "start_dlq.process.reconcile_errors+") + assert.EqualValues(t, 1, scope.Snapshot().Counters()["start_dlq.process.reconcile_errors+"].Value()) + }) + } +} diff --git a/submitqueue/orchestrator/core/request/terminal_log.go b/submitqueue/orchestrator/core/request/terminal_log.go index 8b16368f2..0a81b33a0 100644 --- a/submitqueue/orchestrator/core/request/terminal_log.go +++ b/submitqueue/orchestrator/core/request/terminal_log.go @@ -24,8 +24,6 @@ import ( // ResendTerminalRequestLog resends the request's terminal status without another state write. // The log identifies the resend; original diagnostics may be unavailable. -// For example, validation can store Error and crash before publishing its status log. -// Redelivery must resend the status without another request-state write. func ResendTerminalRequestLog(ctx context.Context, registry consumer.TopicRegistry, request entity.Request) error { status, err := terminalStateToStatus(request.State) if err != nil { diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index ed955125a..f5bbc94b6 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 dlq.NewDLQStartController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil }, }, {