From c8f16da8b8ec2c36b13b830b83871311e9ba136e Mon Sep 17 00:00:00 2001 From: abettigole Date: Sat, 25 Jul 2026 00:55:43 +0000 Subject: [PATCH 1/2] refactor: extract shared request-termination helper Extract the duplicated "transition a request to a terminal state and publish its terminal log" logic into a single TerminateRequest helper in submitqueue/core/request, and route conclude, dlq, cancel, and validate through it. Changes: - Add corerequest.TerminateRequest(ctx, store, registry, requestID, targetState, lastError, metadata) that performs the idempotent 3-way CAS (reconcile / already-terminal / diverged / not-found), publishes the terminal RequestLog, and returns a TerminationResult{Outcome, BeforeState, AfterState} plus a separate error. Version arithmetic stays caller-side and storage.ErrVersionMismatch is passed through as retryable. - Replace terminalStateToStatus's switch with a terminalStatusByState map lookup. - conclude: reconcile each batch member through the helper, keeping per-outcome logs and metrics at the call site. A request referenced by the batch but missing from the store stays a hard error (retry, eventually DLQ). - dlq: failRequest now delegates to the helper, keeping its NotFound tolerance and its reconcile/divergence log messages. - cancel: cancelRequest delegates to the helper (Cancelling to Cancelled); the helper's re-read reports divergence cleanly instead of a spurious version-mismatch retry. - validate: on an expected rejection (duplicate detected or custom validator failure), terminate the request to Error and ack instead of dead-lettering it; genuine infra failures still return errors. - Add table-driven tests for TerminateRequest and update the conclude, cancel, and validate tests. Co-Authored-By: Claude Opus 4.8 (1M context) --- submitqueue/core/request/BUILD.bazel | 2 + submitqueue/core/request/terminate.go | 168 ++++++++++++++ submitqueue/core/request/terminate_test.go | 208 ++++++++++++++++++ .../orchestrator/controller/cancel/cancel.go | 57 +++-- .../controller/cancel/cancel_test.go | 76 ++++++- .../controller/conclude/conclude.go | 85 +++---- .../controller/conclude/conclude_test.go | 32 +++ .../orchestrator/controller/dlq/dlq.go | 47 +--- .../controller/validate/BUILD.bazel | 2 +- .../controller/validate/validate.go | 30 ++- .../controller/validate/validate_test.go | 105 +++++++-- 11 files changed, 668 insertions(+), 144 deletions(-) create mode 100644 submitqueue/core/request/terminate.go create mode 100644 submitqueue/core/request/terminate_test.go diff --git a/submitqueue/core/request/BUILD.bazel b/submitqueue/core/request/BUILD.bazel index 63c149bc..c47be892 100644 --- a/submitqueue/core/request/BUILD.bazel +++ b/submitqueue/core/request/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "log.go", "materializer.go", "request.go", + "terminate.go", ], importpath = "github.com/uber/submitqueue/submitqueue/core/request", visibility = ["//visibility:public"], @@ -24,6 +25,7 @@ go_test( "log_test.go", "materializer_test.go", "request_test.go", + "terminate_test.go", ], embed = [":go_default_library"], deps = [ diff --git a/submitqueue/core/request/terminate.go b/submitqueue/core/request/terminate.go new file mode 100644 index 00000000..736cad51 --- /dev/null +++ b/submitqueue/core/request/terminate.go @@ -0,0 +1,168 @@ +// Copyright (c) 2025 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 request + +import ( + "context" + "errors" + "fmt" + + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +// TerminationOutcome describes what TerminateRequest did to the request. +// The zero value (TerminationOutcomeUnknown) is only produced alongside a non-nil error. +type TerminationOutcome int + +const ( + // TerminationOutcomeUnknown is the zero value, produced only when TerminateRequest also returns a non-nil error. + TerminationOutcomeUnknown TerminationOutcome = iota + // TerminationOutcomeSuccess means the request was transitioned from a non-terminal + // state to the target terminal state and a terminal log entry was published. + TerminationOutcomeSuccess + // TerminationOutcomeAlreadyInTargetState means the request was already in the target terminal + // state. No state write occurred, but the terminal log entry was re-published to + // repair a possible prior attempt that wrote the state but failed before publishing. + TerminationOutcomeAlreadyInTargetState + // TerminationOutcomeDiverged means the request had already reached a different terminal state - + // a concurrent path won the race and owns the terminal log for the state it wrote. + // Nothing was written or published. + TerminationOutcomeDiverged + // TerminationOutcomeNotFound means the request does not exist. Nothing was done. + TerminationOutcomeNotFound +) + +// TerminationResult reports what TerminateRequest observed and did, so callers can +// log and emit metrics for the outcome at their own level and granularity without re-fetching the request. +// It is returned alongside a separate error: +// TerminationResult describes the expected variation (reconciled / already-terminal / diverged / not-found). +// Error signals an infrastructure failure. +type TerminationResult struct { + // Outcome is what happened to the request. + Outcome TerminationOutcome + // BeforeState is the request's state as observed before any write. + // On a divergence it is the (different) terminal state the request actually reached. + // On a success it is the prior non-terminal state. + // Empty when the request was not found or the call failed. + BeforeState entity.RequestState + // AfterState is the request's state after the operation. + // Equal to the target state on success and already-terminal. + // Equal to BeforeState on divergence. + // Empty when the request was not found or the call failed. + AfterState entity.RequestState +} + +// TerminateRequest transitions a request to the given terminal state and publishes +// the corresponding terminal log entry, idempotently under at-least-once delivery. +// It is the shared primitive behind concluding a batch's requests, dead-letter +// reconciliation, and rejecting an invalid request at validation time. +// +// targetState must be one of the terminal request states (Landed, Error, Cancelled). +// The write follows the immutability / optimistic-locking contract: +// caller-owned version arithmetic (newVersion = version+1) guards a pure +// conditional store write, and the in-memory version is only advanced after the store call succeeds. +// +// Idempotency has three shapes, reported via TerminationResult.Outcome: +// - the request is already in targetState: the state write is skipped but the +// terminal log is re-published (TerminationOutcomeAlreadyInTargetState); +// - the request is in a different terminal state: nothing is written or +// published, since the other writer owns that state's terminal log +// (TerminationOutcomeDiverged); +// - the request is not found: nothing is done (TerminationOutcomeNotFound). +// +// The caller owns all logging and metrics — TerminationResult carries the state context needed to do so. +// lastError and metadata are attached to the published RequestLog for diagnosis. +// +// A storage.ErrVersionMismatch from the conditional write is returned as-is (it is +// intrinsically retryable) so the caller's next attempt re-reads and re-evaluates. +func TerminateRequest( + ctx context.Context, + store storage.Storage, + registry consumer.TopicRegistry, + requestID string, + targetState entity.RequestState, + lastError string, + metadata map[string]string, +) (TerminationResult, error) { + status, err := terminalStateToStatus(targetState) + if err != nil { + return TerminationResult{}, err + } + + request, err := store.GetRequestStore().Get(ctx, requestID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + return TerminationResult{Outcome: TerminationOutcomeNotFound}, nil + } + return TerminationResult{}, fmt.Errorf("failed to get request %s: %w", requestID, err) + } + + // logVersion is the request version reflected in the published terminal log. + // It stays at the current version on the idempotent same-state path and + // advances to the new version only after a successful reconciling write. + logVersion := request.Version + outcome := TerminationOutcomeSuccess + switch { + case request.State == targetState: + // Idempotent retry: a prior attempt already wrote the terminal state. + // Skip the CAS and fall through to re-publish the terminal log. + outcome = TerminationOutcomeAlreadyInTargetState + case entity.IsRequestStateTerminal(request.State): + // Divergent terminal state — a concurrent path reached terminal first + // and owns the terminal log entry for the state it actually wrote. + return TerminationResult{ + Outcome: TerminationOutcomeDiverged, + BeforeState: request.State, + AfterState: request.State, + }, nil + default: + newVersion := request.Version + 1 + if err := store.GetRequestStore().UpdateState(ctx, requestID, request.Version, newVersion, targetState); err != nil { + return TerminationResult{}, fmt.Errorf("failed to update request %s state to %s: %w", requestID, targetState, err) + } + logVersion = newVersion + } + + logEntry := entity.NewRequestLog(requestID, status, logVersion, lastError, metadata) + if err := PublishLog(ctx, registry, logEntry, requestID); err != nil { + return TerminationResult{}, fmt.Errorf("failed to publish request log for %s: %w", requestID, err) + } + + return TerminationResult{ + Outcome: outcome, + BeforeState: request.State, + AfterState: targetState, + }, nil +} + +// terminalStatusByState maps each terminal request state to the customer-facing status published on the terminal log. +// A state absent from this map is not a valid termination target. +var terminalStatusByState = map[entity.RequestState]entity.RequestStatus{ + entity.RequestStateLanded: entity.RequestStatusLanded, + entity.RequestStateError: entity.RequestStatusError, + entity.RequestStateCancelled: entity.RequestStatusCancelled, +} + +// terminalStateToStatus maps a terminal request state to the corresponding customer-facing log status. +// It rejects non-terminal states so callers cannot terminate a request into a state that is not final. +func terminalStateToStatus(state entity.RequestState) (entity.RequestStatus, error) { + status, ok := terminalStatusByState[state] + if !ok { + return entity.RequestStatusUnknown, fmt.Errorf("non-terminal request state: %s", state) + } + return status, nil +} diff --git a/submitqueue/core/request/terminate_test.go b/submitqueue/core/request/terminate_test.go new file mode 100644 index 00000000..a9b1b9b7 --- /dev/null +++ b/submitqueue/core/request/terminate_test.go @@ -0,0 +1,208 @@ +// Copyright (c) 2025 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 request + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "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" + "go.uber.org/mock/gomock" +) + +// recordingRegistry returns a registry whose publisher appends every published +// request log to *logs and returns publishErr. It lets tests assert both that a +// terminal log was (or was not) published and what version/status it carried. +func recordingRegistry(t *testing.T, ctrl *gomock.Controller, publishErr error) (consumer.TopicRegistry, *[]entity.RequestLog) { + t.Helper() + logs := &[]entity.RequestLog{} + mockPub := queuemock.NewMockPublisher(ctrl) + mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + log, err := entity.RequestLogFromBytes(msg.Payload) + require.NoError(t, err) + *logs = append(*logs, log) + return publishErr + }, + ).AnyTimes() + + mockQ := queuemock.NewMockQueue(ctrl) + mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() + + registry, err := consumer.NewTopicRegistry( + []consumer.TopicConfig{{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}}, + ) + require.NoError(t, err) + return registry, logs +} + +func TestTerminateRequest(t *testing.T) { + const requestID = "q/1" + + validated := entity.Request{ID: requestID, Queue: "q", State: entity.RequestStateValidated, Version: 3} + + testCases := map[string]struct { + targetState entity.RequestState + lastError string + metadata map[string]string + mockFunc func(rs *storagemock.MockRequestStore) + publishErr error + wantResult TerminationResult + errMsg string + errIs error + // wantLog, when non-nil, asserts the single published terminal log. + wantLog func(t *testing.T, log entity.RequestLog) + }{ + "non-terminal target rejected": { + targetState: entity.RequestStateValidated, + mockFunc: func(rs *storagemock.MockRequestStore) {}, + wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown}, + errMsg: "non-terminal request state: validated", + }, + "request not found": { + targetState: entity.RequestStateError, + mockFunc: func(rs *storagemock.MockRequestStore) { + rs.EXPECT().Get(gomock.Any(), requestID).Return(entity.Request{}, storage.ErrNotFound) + }, + wantResult: TerminationResult{Outcome: TerminationOutcomeNotFound}, + }, + "get infra error": { + targetState: entity.RequestStateError, + mockFunc: func(rs *storagemock.MockRequestStore) { + rs.EXPECT().Get(gomock.Any(), requestID).Return(entity.Request{}, fmt.Errorf("db down")) + }, + wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown}, + errMsg: "failed to get request q/1: db down", + }, + "reconciled from non-terminal state": { + targetState: entity.RequestStateError, + lastError: "boom", + metadata: map[string]string{"source": "validate"}, + mockFunc: func(rs *storagemock.MockRequestStore) { + rs.EXPECT().Get(gomock.Any(), requestID).Return(validated, nil) + rs.EXPECT().UpdateState(gomock.Any(), requestID, int32(3), int32(4), entity.RequestStateError).Return(nil) + }, + wantResult: TerminationResult{ + Outcome: TerminationOutcomeSuccess, + BeforeState: entity.RequestStateValidated, + AfterState: entity.RequestStateError, + }, + wantLog: func(t *testing.T, log entity.RequestLog) { + assert.Equal(t, entity.RequestStatusError, log.Status) + assert.Equal(t, int32(4), log.RequestVersion) + assert.Equal(t, "boom", log.LastError) + assert.Equal(t, "validate", log.Metadata["source"]) + }, + }, + "already in target terminal state republishes log": { + targetState: entity.RequestStateError, + mockFunc: func(rs *storagemock.MockRequestStore) { + already := entity.Request{ID: requestID, State: entity.RequestStateError, Version: 5} + rs.EXPECT().Get(gomock.Any(), requestID).Return(already, nil) + }, + wantResult: TerminationResult{ + Outcome: TerminationOutcomeAlreadyInTargetState, + BeforeState: entity.RequestStateError, + AfterState: entity.RequestStateError, + }, + wantLog: func(t *testing.T, log entity.RequestLog) { + assert.Equal(t, entity.RequestStatusError, log.Status) + assert.Equal(t, int32(5), log.RequestVersion) + }, + }, + "already in target terminal state returns republish error": { + targetState: entity.RequestStateError, + mockFunc: func(rs *storagemock.MockRequestStore) { + already := entity.Request{ID: requestID, State: entity.RequestStateError, Version: 5} + rs.EXPECT().Get(gomock.Any(), requestID).Return(already, nil) + }, + publishErr: fmt.Errorf("connection refused"), + wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown}, + errMsg: "failed to publish request log for q/1", + }, + "diverged terminal state is left untouched": { + targetState: entity.RequestStateError, + mockFunc: func(rs *storagemock.MockRequestStore) { + landed := entity.Request{ID: requestID, State: entity.RequestStateLanded, Version: 7} + rs.EXPECT().Get(gomock.Any(), requestID).Return(landed, nil) + }, + wantResult: TerminationResult{ + Outcome: TerminationOutcomeDiverged, + BeforeState: entity.RequestStateLanded, + AfterState: entity.RequestStateLanded, + }, + }, + "update version mismatch returned as-is": { + targetState: entity.RequestStateError, + mockFunc: func(rs *storagemock.MockRequestStore) { + rs.EXPECT().Get(gomock.Any(), requestID).Return(validated, nil) + rs.EXPECT().UpdateState(gomock.Any(), requestID, int32(3), int32(4), entity.RequestStateError).Return(storage.ErrVersionMismatch) + }, + wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown}, + errMsg: "version mismatch", + errIs: storage.ErrVersionMismatch, + }, + "publish error after reconcile": { + targetState: entity.RequestStateError, + mockFunc: func(rs *storagemock.MockRequestStore) { + rs.EXPECT().Get(gomock.Any(), requestID).Return(validated, nil) + rs.EXPECT().UpdateState(gomock.Any(), requestID, int32(3), int32(4), entity.RequestStateError).Return(nil) + }, + publishErr: fmt.Errorf("connection refused"), + wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown}, + errMsg: "failed to publish request log for q/1", + }, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + ctrl := gomock.NewController(t) + requestStore := storagemock.NewMockRequestStore(ctrl) + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + tc.mockFunc(requestStore) + + registry, logs := recordingRegistry(t, ctrl, tc.publishErr) + + res, err := TerminateRequest(context.Background(), store, registry, requestID, tc.targetState, tc.lastError, tc.metadata) + + assert.Equal(t, tc.wantResult, res) + if tc.errMsg != "" { + assert.ErrorContains(t, err, tc.errMsg) + } else { + assert.NoError(t, err) + } + if tc.errIs != nil { + assert.ErrorIs(t, err, tc.errIs) + } + + if tc.wantLog != nil { + require.Len(t, *logs, 1) + tc.wantLog(t, (*logs)[0]) + } else if tc.publishErr == nil { + assert.Empty(t, *logs) + } + }) + } +} diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index 5f78689f..aed3cb62 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/submitqueue/orchestrator/controller/cancel/cancel.go @@ -209,35 +209,48 @@ func (c *Controller) findActiveBatch(ctx context.Context, request entity.Request return entity.Batch{}, false, nil } -// cancelRequest performs the terminal CAS (Cancelling → Cancelled) for a request -// that is not part of any active batch, and emits the RequestStatusCancelled log -// entry. storage.ErrVersionMismatch here means a concurrent writer (typically -// conclude after a racing batch terminal transition) advanced the request between -// our mark-cancelling CAS and this terminal CAS — returned as-is because the -// sentinel is intrinsically retryable; the next pass will observe the new state -// (likely terminal) and ack via the top-level terminal-check. +// cancelRequest drives the terminal transition (Cancelling → Cancelled) for a +// request that is not part of any active batch, and emits the RequestStatusCancelled log entry. +// It delegates the CAS-plus-log to the shared TerminateRequest helper, +// which re-reads the request before the terminal write. +// +// A storage.ErrVersionMismatch surfaced by the helper means a concurrent writer +// (typically conclude after a racing batch terminal transition) advanced the +// request between our mark-cancelling CAS and this terminal CAS; it is returned +// as-is because the sentinel is intrinsically retryable, and the next pass will +// observe the new state and ack via the top-level terminal-check. If that +// concurrent writer already reached a *different* terminal state, the helper +// reports TerminationDiverged and we simply ack — the other writer owns the +// terminal log for the state it wrote. func (c *Controller) cancelRequest(ctx context.Context, request entity.Request, reason string) error { - newVersion := request.Version + 1 - if err := c.store.GetRequestStore().UpdateState(ctx, request.ID, request.Version, newVersion, entity.RequestStateCancelled); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "request_update_errors", 1) - return fmt.Errorf("failed to cancel request %s: %w", request.ID, err) - } - metadata := map[string]string{} if reason != "" { metadata["reason"] = reason } - logEntry := entity.NewRequestLog(request.ID, entity.RequestStatusCancelled, newVersion, "", metadata) - if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "log_publish_errors", 1) - return fmt.Errorf("failed to publish cancel log for request %s: %w", request.ID, err) + res, err := corerequest.TerminateRequest(ctx, c.store, c.registry, request.ID, entity.RequestStateCancelled, "", metadata) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "request_terminate_errors", 1) + return fmt.Errorf("failed to cancel request %s: %w", request.ID, err) } - c.logger.Infow("request cancelled (not batched)", - "request_id", request.ID, - "queue", request.Queue, - ) - metrics.NamedCounter(c.metricsScope, opName, "request_cancelled", 1) + switch res.Outcome { + case corerequest.TerminationOutcomeSuccess, corerequest.TerminationOutcomeAlreadyInTargetState: + c.logger.Infow("request cancelled (not batched)", + "request_id", request.ID, + "queue", request.Queue, + ) + metrics.NamedCounter(c.metricsScope, opName, "request_cancelled", 1) + case corerequest.TerminationOutcomeDiverged: + c.logger.Infow("request reached a different terminal state before cancel, skipping", + "request_id", request.ID, + "queue", request.Queue, + "actual_state", string(res.BeforeState), + ) + metrics.NamedCounter(c.metricsScope, opName, "request_terminal_divergence", 1) + case corerequest.TerminationOutcomeNotFound: + metrics.NamedCounter(c.metricsScope, opName, "request_store_errors", 1) + return fmt.Errorf("request %s not found during cancel: %w", request.ID, storage.ErrNotFound) + } return nil } diff --git a/submitqueue/orchestrator/controller/cancel/cancel_test.go b/submitqueue/orchestrator/controller/cancel/cancel_test.go index 162b78d9..79724a99 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel_test.go +++ b/submitqueue/orchestrator/controller/cancel/cancel_test.go @@ -134,12 +134,14 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) { }).AnyTimes() reqStore := storagemock.NewMockRequestStore(ctrl) - reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.Request{ - ID: "q/1", Queue: "q", State: entity.RequestStateStarted, Version: 2, - }, nil) - // Two-step transition: first mark Cancelling, then Cancelled. + started := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateStarted, Version: 2} + cancelling := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateCancelling, Version: 3} + // Two-step transition: mark Cancelling, then — after TerminateRequest re-reads + // the request — the terminal Cancelled CAS. gomock.InOrder( + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(started, nil), reqStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(2), int32(3), entity.RequestStateCancelling).Return(nil), + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(cancelling, nil), reqStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(3), int32(4), entity.RequestStateCancelled).Return(nil), ) @@ -168,9 +170,10 @@ func TestProcess_AlreadyCancelling_SkipsMarkCancelling(t *testing.T) { pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() reqStore := storagemock.NewMockRequestStore(ctrl) - reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.Request{ - ID: "q/1", Queue: "q", State: entity.RequestStateCancelling, Version: 3, - }, nil) + cancelling := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateCancelling, Version: 3} + // Get is called twice: the initial load and TerminateRequest's re-read. Both + // see Cancelling (the prior pass already recorded intent). + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(cancelling, nil).Times(2) // Only the terminal CAS — the mark-cancelling step is a no-op. reqStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(3), int32(4), entity.RequestStateCancelled).Return(nil) @@ -218,11 +221,12 @@ func TestProcess_UnbatchedVersionMismatch_Retryable(t *testing.T) { _ = pub reqStore := storagemock.NewMockRequestStore(ctrl) - reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.Request{ - ID: "q/1", Queue: "q", State: entity.RequestStateStarted, Version: 2, - }, nil) + started := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateStarted, Version: 2} + cancelling := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateCancelling, Version: 3} gomock.InOrder( + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(started, nil), reqStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(2), int32(3), entity.RequestStateCancelling).Return(nil), + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(cancelling, nil), reqStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(3), int32(4), entity.RequestStateCancelled). Return(storage.ErrVersionMismatch), ) @@ -240,6 +244,58 @@ func TestProcess_UnbatchedVersionMismatch_Retryable(t *testing.T) { assert.ErrorIs(t, err, storage.ErrVersionMismatch) } +func TestProcess_UnbatchedRequestDiverged_Acks(t *testing.T) { + ctrl := gomock.NewController(t) + registry, pub := newRegistry(t, ctrl) + _ = pub + + started := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateStarted, Version: 2} + landed := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateLanded, Version: 4} + reqStore := storagemock.NewMockRequestStore(ctrl) + gomock.InOrder( + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(started, nil), + reqStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(2), int32(3), entity.RequestStateCancelling).Return(nil), + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(landed, nil), + ) + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().GetByQueueAndStates(gomock.Any(), "q", gomock.Any()).Return(nil, nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + + controller := newController(t, store, registry) + err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "q/1", ""), "q/1")) + require.NoError(t, err) +} + +func TestProcess_UnbatchedRequestDisappears_Retryable(t *testing.T) { + ctrl := gomock.NewController(t) + registry, pub := newRegistry(t, ctrl) + _ = pub + + started := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateStarted, Version: 2} + reqStore := storagemock.NewMockRequestStore(ctrl) + gomock.InOrder( + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(started, nil), + reqStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(2), int32(3), entity.RequestStateCancelling).Return(nil), + reqStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.Request{}, storage.ErrNotFound), + ) + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().GetByQueueAndStates(gomock.Any(), "q", gomock.Any()).Return(nil, nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + + controller := newController(t, store, registry) + err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "q/1", ""), "q/1")) + require.Error(t, err) + assert.ErrorIs(t, err, storage.ErrNotFound) +} + // TestProcess_BatchPath_HandsOffToSpeculate asserts the entire batch path: // the request intent CAS runs, the batch intent CAS to Cancelling runs, and // exactly one publish lands on the speculate topic with the batch ID as the diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go index 4e451a5b..8e86f19e 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude.go +++ b/submitqueue/orchestrator/controller/conclude/conclude.go @@ -99,64 +99,45 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er metrics.NamedCounter(c.metricsScope, "process", "unexpected_state_errors", 1) return fmt.Errorf("unexpected batch state %q for batch %s: %w", batch.State, batch.ID, err) } - requestStatus, err := requestStateToStatus(requestState) - if err != nil { - // Unreachable: batchStateToRequestState only returns terminal request states. - return fmt.Errorf("failed to map request state %s to status: %w", requestState, err) - } // Reconcile each request to the batch's terminal state and emit a terminal - // log entry. The flow is idempotent under at-least-once delivery: a prior - // attempt may have completed the CAS but failed before publishing the log, - // so the log publish must still run when the request is already in the - // target terminal state. + // log entry. TerminateRequest is idempotent under at-least-once delivery: a + // prior attempt may have completed the CAS but failed before publishing the + // log, so the log publish still runs when the request is already in the + // target terminal state, and a request that a concurrent path already drove + // to a different terminal state is left untouched. A request referenced by + // the batch but missing from the store is a hard error — the whole batch is + // retried (and eventually dead-lettered) rather than silently skipped. We + // translate the result into per-outcome logs and metrics. for _, requestID := range batch.Contains { - request, err := c.store.GetRequestStore().Get(ctx, requestID) + res, err := corerequest.TerminateRequest(ctx, c.store, c.registry, requestID, requestState, "", map[string]string{ + "batch_id": batch.ID, + }) if err != nil { - metrics.NamedCounter(c.metricsScope, "process", "request_store_errors", 1) - return fmt.Errorf("failed to get request %s: %w", requestID, err) + metrics.NamedCounter(c.metricsScope, "process", "terminate_errors", 1) + return fmt.Errorf("failed to terminate request %s for batch %s: %w", requestID, batch.ID, err) } - switch { - case request.State == requestState: - // Idempotent retry: a prior delivery already wrote the terminal - // state. Skip the CAS and fall through to the log publish. + switch res.Outcome { + case corerequest.TerminationReconciled: + c.logger.Infow("updated request state", + "batch_id", batch.ID, + "request_id", requestID, + "new_state", string(res.AfterState), + ) + case corerequest.TerminationAlreadyTerminal: metrics.NamedCounter(c.metricsScope, "process", "already_reconciled", 1) - case entity.IsRequestStateTerminal(request.State): - // Divergent terminal state — a concurrent path (e.g. a racing - // cancel-not-yet-batched transition) reached terminal first. Skip - // the reconcile and the log publish; the other writer owns the - // terminal log entry for the state it actually wrote. + case corerequest.TerminationDiverged: c.logger.Warnw("request already in different terminal state, skipping reconcile", "batch_id", batch.ID, "request_id", requestID, - "actual_state", string(request.State), "expected_state", string(requestState), + "actual_state", string(res.BeforeState), ) metrics.NamedCounter(c.metricsScope, "process", "terminal_state_divergence", 1) - continue - default: - newVersion := request.Version + 1 - if err := c.store.GetRequestStore().UpdateState(ctx, requestID, request.Version, newVersion, requestState); err != nil { - metrics.NamedCounter(c.metricsScope, "process", "request_update_errors", 1) - return fmt.Errorf("failed to update request %s state to %s: %w", requestID, requestState, err) - } - request.Version = newVersion - request.State = requestState - - c.logger.Infow("updated request state", - "batch_id", batch.ID, - "request_id", requestID, - "new_state", string(requestState), - ) - } - - logEntry := entity.NewRequestLog(requestID, requestStatus, request.Version, "", map[string]string{ - "batch_id": batch.ID, - }) - if err := corerequest.PublishLog(ctx, c.registry, logEntry, requestID); err != nil { - metrics.NamedCounter(c.metricsScope, "process", "log_publish_errors", 1) - return fmt.Errorf("failed to publish request log for %s: %w", requestID, err) + case corerequest.TerminationNotFound: + metrics.NamedCounter(c.metricsScope, "process", "request_store_errors", 1) + return fmt.Errorf("failed to get request %s for batch %s: %w", requestID, batch.ID, storage.ErrNotFound) } } @@ -191,17 +172,3 @@ func batchStateToRequestState(state entity.BatchState) (entity.RequestState, err return entity.RequestStateUnknown, fmt.Errorf("non-terminal batch state: %s", state) } } - -// requestStateToStatus maps a terminal request state to the corresponding log status. -func requestStateToStatus(state entity.RequestState) (entity.RequestStatus, error) { - switch state { - case entity.RequestStateLanded: - return entity.RequestStatusLanded, nil - case entity.RequestStateError: - return entity.RequestStatusError, nil - case entity.RequestStateCancelled: - return entity.RequestStatusCancelled, nil - default: - return entity.RequestStatusUnknown, fmt.Errorf("non-terminal request state: %s", state) - } -} diff --git a/submitqueue/orchestrator/controller/conclude/conclude_test.go b/submitqueue/orchestrator/controller/conclude/conclude_test.go index 1e2ebe4c..d2d8c0e7 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude_test.go +++ b/submitqueue/orchestrator/controller/conclude/conclude_test.go @@ -260,6 +260,38 @@ func TestController_Process(t *testing.T) { }, expectLogPublish: false, }, + { + name: "missing request returns error", + batch: entity.Batch{ + ID: "test-queue/batch/11", + Queue: "test-queue", + Contains: []string{"test-queue/40"}, + State: entity.BatchStateSucceeded, + Version: 2, + }, + setupStore: func(ctrl *gomock.Controller) *storagemock.MockStorage { + mockBatchStore := storagemock.NewMockBatchStore(ctrl) + mockBatchStore.EXPECT().Get(gomock.Any(), "test-queue/batch/11").Return(entity.Batch{ + ID: "test-queue/batch/11", + Queue: "test-queue", + Contains: []string{"test-queue/40"}, + State: entity.BatchStateSucceeded, + Version: 2, + }, nil) + + // A request referenced by the batch but missing from the store is a + // hard error (nack/retry, eventually DLQ) — not a silent skip. + mockRequestStore := storagemock.NewMockRequestStore(ctrl) + mockRequestStore.EXPECT().Get(gomock.Any(), "test-queue/40").Return(entity.Request{}, storage.ErrNotFound) + + mockStorage := storagemock.NewMockStorage(ctrl) + mockStorage.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() + mockStorage.EXPECT().GetRequestStore().Return(mockRequestStore).AnyTimes() + return mockStorage + }, + wantErr: true, + retryable: false, + }, { name: "non-terminal batch state returns error", batch: entity.Batch{ diff --git a/submitqueue/orchestrator/controller/dlq/dlq.go b/submitqueue/orchestrator/controller/dlq/dlq.go index 5de49810..8ccf73e3 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq.go +++ b/submitqueue/orchestrator/controller/dlq/dlq.go @@ -73,50 +73,27 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey { // confirm the cancel completed cleanly. Writing Error is the honest signal and // keeps the request from being stuck in a non-terminal state forever. func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string) error { - request, err := store.GetRequestStore().Get(ctx, requestID) + res, err := requestcore.TerminateRequest(ctx, store, registry, requestID, entity.RequestStateError, lastError, nil) if err != nil { - if errors.Is(err, storage.ErrNotFound) { - logger.Warnw("dlq reconcile: request not found, skipping", - "request_id", requestID, - ) - return nil - } - return fmt.Errorf("failed to get request %s: %w", requestID, err) + return fmt.Errorf("dlq reconcile request %s failed: %w", requestID, err) } - logVersion := request.Version - switch request.State { - case entity.RequestStateError: - logger.Infow("dlq reconcile: request already failed, republishing terminal log", + switch res.Outcome { + case requestcore.TerminationOutcomeSuccess: + logger.Infow("dlq reconcile: request marked terminal error", "request_id", requestID, + "previous_state", string(res.BeforeState), ) - case entity.RequestStateLanded, entity.RequestStateCancelled: + case requestcore.TerminationOutcomeAlreadyInTargetState: + logger.Infow("dlq reconcile: request already failed, republished terminal log", "request_id", requestID) + case requestcore.TerminationOutcomeDiverged: logger.Infow("dlq reconcile: request has a different terminal outcome, skipping", "request_id", requestID, - "state", string(request.State), - ) - return nil - default: - newVersion := request.Version + 1 - if err := store.GetRequestStore().UpdateState(ctx, requestID, request.Version, newVersion, entity.RequestStateError); err != nil { - return fmt.Errorf("failed to update request %s state to error: %w", requestID, err) - } - logVersion = newVersion - logger.Infow("dlq reconcile: request marked terminal error", - "request_id", requestID, - "previous_state", string(request.State), + "state", string(res.BeforeState), ) + case requestcore.TerminationOutcomeNotFound: + logger.Warnw("dlq reconcile: request not found, skipping", "request_id", requestID) } - - // Publish the terminal Error status through the log topic so Gateway remains - // the sole writer of request logs and public projections. An existing Error - // state republishes the same logical event so a previous attempt that changed - // the entity but failed to publish can be repaired. - logEntry := entity.NewRequestLog(requestID, entity.RequestStatusError, logVersion, lastError, nil) - if err := requestcore.PublishLog(ctx, registry, logEntry, requestID); err != nil { - return fmt.Errorf("failed to publish request log for %s: %w", requestID, err) - } - return nil } diff --git a/submitqueue/orchestrator/controller/validate/BUILD.bazel b/submitqueue/orchestrator/controller/validate/BUILD.bazel index 0b0beff7..3a0eb238 100644 --- a/submitqueue/orchestrator/controller/validate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/validate/BUILD.bazel @@ -12,8 +12,8 @@ go_library( "//platform/base/mergestrategy:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", - "//platform/errs:go_default_library", "//platform/metrics:go_default_library", + "//submitqueue/core/request:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/changeprovider:go_default_library", "//submitqueue/extension/storage:go_default_library", diff --git a/submitqueue/orchestrator/controller/validate/validate.go b/submitqueue/orchestrator/controller/validate/validate.go index bb3ce96a..eb2e0655 100644 --- a/submitqueue/orchestrator/controller/validate/validate.go +++ b/submitqueue/orchestrator/controller/validate/validate.go @@ -27,8 +27,8 @@ import ( "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" coremetrics "github.com/uber/submitqueue/platform/metrics" + corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -140,7 +140,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "duplicate_id", dupID, ) coremetrics.NamedCounter(c.metricsScope, "process", "duplicate_requests", 1) - return errs.NewUserError(fmt.Errorf("request %s is a duplicate of in-flight request %s", request.ID, dupID)) + return c.reject(ctx, request.ID, fmt.Sprintf("request %s is a duplicate of in-flight request %s", request.ID, dupID)) } // Fetch change metadata @@ -174,7 +174,12 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er if v != nil { if err := v.Validate(ctx, request); err != nil { coremetrics.NamedCounter(c.metricsScope, "process", "custom_validation_failures", 1) - return fmt.Errorf("custom validation failed for request %s: %w", request.ID, err) + c.logger.Infow("custom validation rejected request", + "request_id", request.ID, + "queue", request.Queue, + "error", err.Error(), + ) + return c.reject(ctx, request.ID, fmt.Sprintf("custom validation failed: %v", err)) } } } @@ -218,6 +223,25 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return nil // Success - message will be acked } +// reject terminates a request as an expected validation failure. It transitions +// the request to RequestStateError (with reason preserved as the terminal log's +// last error) and acks the delivery by returning nil. This deliberately does not +// dead-letter the message: the DLQ is for unexpected failures, whereas an +// invalid request (a duplicate, or one a custom validator rejected) is an +// expected terminal outcome that should be reported to the user directly. +// +// Only an infra failure while terminating (storage/publish) is returned as an +// error so the delivery is retried; the request itself is never re-queued for +// validation once rejected. +func (c *Controller) reject(ctx context.Context, requestID, reason string) error { + if _, err := corerequest.TerminateRequest(ctx, c.store, c.registry, requestID, entity.RequestStateError, reason, nil); err != nil { + coremetrics.NamedCounter(c.metricsScope, "process", "terminate_errors", 1) + return fmt.Errorf("failed to terminate rejected request %s: %w", requestID, err) + } + coremetrics.NamedCounter(c.metricsScope, "process", "rejected_requests", 1) + return nil +} + // checkDuplicate looks for any other in-flight request whose URIs overlap with this // request's. It reads the change store before this request claims its own URIs // (claimChanges runs later in Process), so it only sees rows written by other diff --git a/submitqueue/orchestrator/controller/validate/validate_test.go b/submitqueue/orchestrator/controller/validate/validate_test.go index 78691dfc..615638a0 100644 --- a/submitqueue/orchestrator/controller/validate/validate_test.go +++ b/submitqueue/orchestrator/controller/validate/validate_test.go @@ -113,7 +113,10 @@ func newTestController( mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}}, + []consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}, + }, ) require.NoError(t, err) @@ -344,7 +347,7 @@ func TestController_Process_DuplicateDetection(t *testing.T) { ownerLookup map[string]entity.Request ownerNotFound map[string]bool ownerErr map[string]error - wantUserErr bool + wantRejected bool wantUnexpected bool }{ { @@ -352,14 +355,14 @@ func TestController_Process_DuplicateDetection(t *testing.T) { byURI: map[string][]entity.ChangeRecord{uriA: nil}, }, { - name: "overlap with live in-flight request returns user error", + name: "overlap with live in-flight request rejects the request", byURI: map[string][]entity.ChangeRecord{ uriA: {{URI: uriA, RequestID: dupRequestID, Queue: queueName}}, }, ownerLookup: map[string]entity.Request{ dupRequestID: {ID: dupRequestID, Queue: queueName, State: entity.RequestStateStarted, Version: 1}, }, - wantUserErr: true, + wantRejected: true, }, { name: "overlap with terminal owner is skipped", @@ -387,7 +390,7 @@ func TestController_Process_DuplicateDetection(t *testing.T) { ownerLookup: map[string]entity.Request{ dupRequestID: {ID: dupRequestID, Queue: queueName, State: entity.RequestStateValidated, Version: 2}, }, - wantUserErr: true, + wantRejected: true, }, { name: "first URI's owner is terminal, second URI's owner is live", @@ -400,7 +403,7 @@ func TestController_Process_DuplicateDetection(t *testing.T) { terminalReqID: {ID: terminalReqID, State: entity.RequestStateError, Version: 3}, anotherReqID: {ID: anotherReqID, State: entity.RequestStateProcessing, Version: 4}, }, - wantUserErr: true, + wantRejected: true, }, { // Store doesn't exclude self; controller filters by RequestID and must not look up its own row. @@ -420,7 +423,7 @@ func TestController_Process_DuplicateDetection(t *testing.T) { ownerLookup: map[string]entity.Request{ dupRequestID: {ID: dupRequestID, Queue: queueName, State: entity.RequestStateStarted, Version: 1}, }, - wantUserErr: true, + wantRejected: true, }, { name: "owner lookup unexpected error propagates", @@ -453,7 +456,9 @@ func TestController_Process_DuplicateDetection(t *testing.T) { } mockReqStore := storagemock.NewMockRequestStore(ctrl) - mockReqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil) + // Get is called once for the initial load and, on a duplicate, once more + // inside TerminateRequest before the terminal CAS. + mockReqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil).AnyTimes() for id, req := range tt.ownerLookup { mockReqStore.EXPECT().Get(gomock.Any(), id).Return(req, nil) } @@ -463,6 +468,11 @@ func TestController_Process_DuplicateDetection(t *testing.T) { for id, e := range tt.ownerErr { mockReqStore.EXPECT().Get(gomock.Any(), id).Return(entity.Request{}, e) } + if tt.wantRejected { + // A detected duplicate is terminated (Started → Error) rather than + // dead-lettered, then the delivery is acked. + mockReqStore.EXPECT().UpdateState(gomock.Any(), request.ID, int32(1), int32(2), entity.RequestStateError).Return(nil) + } store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes() @@ -489,9 +499,10 @@ func TestController_Process_DuplicateDetection(t *testing.T) { case tt.wantUnexpected: require.Error(t, err) assert.False(t, errs.IsUserError(err), "owner lookup failure should not be a user error") - case tt.wantUserErr: - require.Error(t, err) - assert.True(t, errs.IsUserError(err), "duplicate detection should be a user error") + case tt.wantRejected: + // If we get an expected failure, the request is terminated, and the delivery acked (no error, no DLQ). + // The terminal CAS is asserted via the UpdateState expectation above. + require.NoError(t, err) default: require.NoError(t, err) } @@ -623,16 +634,30 @@ func TestController_Process_CustomValidatorFails(t *testing.T) { State: entity.RequestStateStarted, Version: 1, } - store, _ := newMockStorage(ctrl, request) + store, mockReqStore := newMockStorage(ctrl, request) store.EXPECT().GetChangeStore().Return(newMockChangeStore(ctrl)).AnyTimes() + // A validator rejection terminates the request (Started → Error) rather than dead-lettering it. + mockReqStore.EXPECT().UpdateState(gomock.Any(), request.ID, int32(1), int32(2), entity.RequestStateError).Return(nil) logger := zaptest.NewLogger(t).Sugar() + var gotLog entity.RequestLog mockPub := queuemock.NewMockPublisher(ctrl) + mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + log, err := entity.RequestLogFromBytes(msg.Payload) + require.NoError(t, err) + gotLog = log + return nil + }, + ).AnyTimes() mockQ := queuemock.NewMockQueue(ctrl) mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}}, + []consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}, + }, ) require.NoError(t, err) @@ -654,6 +679,58 @@ func TestController_Process_CustomValidatorFails(t *testing.T) { delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() + // The delivery is acked (no error): an invalid request is an expected terminal + // outcome, not a dead-letter, and the reason is preserved on the terminal log. + require.NoError(t, controller.Process(context.Background(), delivery)) + assert.Equal(t, entity.RequestStatusError, gotLog.Status) + assert.Equal(t, int32(2), gotLog.RequestVersion) + assert.Contains(t, gotLog.LastError, "some validation error") +} + +func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *testing.T) { + ctrl := gomock.NewController(t) + + request := entity.Request{ + ID: "test-queue/123", + Queue: "test-queue", + Change: change.Change{URIs: []string{"github://uber/service/pull/456/abcdef0123456789abcdef0123456789abcdef01"}}, + LandStrategy: mergestrategy.MergeStrategyRebase, + State: entity.RequestStateStarted, + Version: 1, + } + store, mockReqStore := newMockStorage(ctrl, request) + store.EXPECT().GetChangeStore().Return(newMockChangeStore(ctrl)).AnyTimes() + mockReqStore.EXPECT().UpdateState(gomock.Any(), request.ID, int32(1), int32(2), entity.RequestStateError).Return(nil) + + mockPub := queuemock.NewMockPublisher(ctrl) + mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("publish boom")).AnyTimes() + mockQ := queuemock.NewMockQueue(ctrl) + mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() + registry, err := consumer.NewTopicRegistry( + []consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}, + }, + ) + require.NoError(t, err) + + cpFactory := changeprovidermock.NewMockFactory(ctrl) + cpFactory.EXPECT().For(gomock.Any()).Return(&mockChangeProvider{}, nil).AnyTimes() + + mockValidator := validatormock.NewMockValidator(ctrl) + mockValidator.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(fmt.Errorf("some validation error")) + mockValidatorFactory := validatormock.NewMockFactory(ctrl) + mockValidatorFactory.EXPECT().For(validator.Config{ + QueueName: request.Queue, + }).Return(mockValidator, nil) + + controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, registry, cpFactory, mockValidatorFactory, runwaymq.TopicKeyMergeConflictCheck, topickey.TopicKeyValidate, "orchestrator-validate") + msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) + delivery := queuemock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg).AnyTimes() + delivery.EXPECT().Attempt().Return(1).AnyTimes() + err = controller.Process(context.Background(), delivery) - require.ErrorContains(t, err, "some validation error") + require.Error(t, err) + assert.ErrorContains(t, err, "failed to terminate rejected request") } From 576ca9d23639047fea1522604796e1dea4f538d0 Mon Sep 17 00:00:00 2001 From: Adam Bettigole Date: Mon, 27 Jul 2026 16:46:11 -0700 Subject: [PATCH 2/2] fix: use correct TerminationOutcome constant names in conclude controller Co-Authored-By: Claude Opus 4.6 (1M context) --- submitqueue/orchestrator/controller/conclude/conclude.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go index 8e86f19e..5a1810a4 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude.go +++ b/submitqueue/orchestrator/controller/conclude/conclude.go @@ -119,15 +119,15 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } switch res.Outcome { - case corerequest.TerminationReconciled: + case corerequest.TerminationOutcomeSuccess: c.logger.Infow("updated request state", "batch_id", batch.ID, "request_id", requestID, "new_state", string(res.AfterState), ) - case corerequest.TerminationAlreadyTerminal: + case corerequest.TerminationOutcomeAlreadyInTargetState: metrics.NamedCounter(c.metricsScope, "process", "already_reconciled", 1) - case corerequest.TerminationDiverged: + case corerequest.TerminationOutcomeDiverged: c.logger.Warnw("request already in different terminal state, skipping reconcile", "batch_id", batch.ID, "request_id", requestID, @@ -135,7 +135,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "actual_state", string(res.BeforeState), ) metrics.NamedCounter(c.metricsScope, "process", "terminal_state_divergence", 1) - case corerequest.TerminationNotFound: + case corerequest.TerminationOutcomeNotFound: metrics.NamedCounter(c.metricsScope, "process", "request_store_errors", 1) return fmt.Errorf("failed to get request %s for batch %s: %w", requestID, batch.ID, storage.ErrNotFound) }