From 9b4145c3cea1e7ae69e6d30d6f1607fbae9b784c Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Wed, 22 Jul 2026 11:45:31 -0700 Subject: [PATCH] refactor(orchestrator): centralize request termination Summary: Intent: - Remove duplicated terminal request state and log publication logic. - Keep existing controller behavior while standardizing CAS and idempotency handling. Changes: - Add a shared request terminal-state reconciliation helper with metadata support. - Migrate cancel, conclude, and merge-conflict failure paths to the helper. - Preserve declaration-level retryability for storage version conflicts. --- submitqueue/core/request/BUILD.bazel | 2 + submitqueue/core/request/terminal.go | 87 +++++++++++ submitqueue/core/request/terminal_test.go | 144 ++++++++++++++++++ .../orchestrator/controller/cancel/cancel.go | 48 +++--- .../controller/conclude/conclude.go | 71 +++------ .../mergeconflictsignal.go | 46 ++---- 6 files changed, 290 insertions(+), 108 deletions(-) create mode 100644 submitqueue/core/request/terminal.go create mode 100644 submitqueue/core/request/terminal_test.go diff --git a/submitqueue/core/request/BUILD.bazel b/submitqueue/core/request/BUILD.bazel index 63c149bc..3281db91 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", + "terminal.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", + "terminal_test.go", ], embed = [":go_default_library"], deps = [ diff --git a/submitqueue/core/request/terminal.go b/submitqueue/core/request/terminal.go new file mode 100644 index 00000000..d9bf7227 --- /dev/null +++ b/submitqueue/core/request/terminal.go @@ -0,0 +1,87 @@ +// 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 request + +import ( + "context" + "fmt" + + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +// TerminalOutcome describes the terminal request state and public log context +// selected by the controller that owns the business or reconciliation decision. +type TerminalOutcome struct { + // State is the desired terminal request state. + State entity.RequestState + // LastError is the failure context exposed through request status and history. + LastError string + // Metadata is display and debugging context persisted with the terminal log. + Metadata map[string]string +} + +// ReconcileTerminalState converges a request entity and its public log on the +// caller-selected terminal outcome. It returns false without writing when the +// request already has a different terminal state. +func ReconcileTerminalState( + ctx context.Context, + requestStore storage.RequestStore, + registry consumer.TopicRegistry, + request entity.Request, + outcome TerminalOutcome, +) (bool, error) { + status, err := terminalStatus(outcome.State) + if err != nil { + return false, err + } + + switch { + case request.State == outcome.State: + case entity.IsRequestStateTerminal(request.State): + return false, nil + default: + newVersion := request.Version + 1 + if err := requestStore.UpdateState(ctx, request.ID, request.Version, newVersion, outcome.State); err != nil { + return false, fmt.Errorf( + "failed to update request %s to terminal state %s: %w", + request.ID, + outcome.State, + err, + ) + } + request.Version = newVersion + } + + logEntry := entity.NewRequestLog(request.ID, status, request.Version, outcome.LastError, outcome.Metadata) + if err := PublishLog(ctx, registry, logEntry, request.ID); err != nil { + return false, fmt.Errorf("failed to publish terminal request log for %s: %w", request.ID, err) + } + return true, nil +} + +func terminalStatus(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("request state %s is not terminal", state) + } +} diff --git a/submitqueue/core/request/terminal_test.go b/submitqueue/core/request/terminal_test.go new file mode 100644 index 00000000..57946aae --- /dev/null +++ b/submitqueue/core/request/terminal_test.go @@ -0,0 +1,144 @@ +// 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 request + +import ( + "context" + "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" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + "go.uber.org/mock/gomock" +) + +func TestReconcileTerminalState(t *testing.T) { + tests := []struct { + name string + request entity.Request + outcome TerminalOutcome + wantUpdate bool + wantPublish bool + wantReconciled bool + wantStatus entity.RequestStatus + wantVersion int32 + wantErr bool + }{ + { + name: "updates non-terminal request and publishes context", + request: entity.Request{ID: "q/1", State: entity.RequestStateProcessing, Version: 3}, + outcome: TerminalOutcome{ + State: entity.RequestStateError, + LastError: "merge conflict", + Metadata: map[string]string{"reason_code": "merge_conflict"}, + }, + wantUpdate: true, + wantPublish: true, + wantReconciled: true, + wantStatus: entity.RequestStatusError, + wantVersion: 4, + }, + { + name: "same terminal state republishes log without CAS", + request: entity.Request{ID: "q/1", State: entity.RequestStateLanded, Version: 5}, + outcome: TerminalOutcome{State: entity.RequestStateLanded}, + wantPublish: true, + wantReconciled: true, + wantStatus: entity.RequestStatusLanded, + wantVersion: 5, + }, + { + name: "different terminal outcome is preserved", + request: entity.Request{ID: "q/1", State: entity.RequestStateCancelled, Version: 5}, + outcome: TerminalOutcome{State: entity.RequestStateLanded}, + }, + { + name: "non-terminal target is rejected", + request: entity.Request{ID: "q/1", State: entity.RequestStateStarted, Version: 1}, + outcome: TerminalOutcome{State: entity.RequestStateValidated}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + requestStore := storagemock.NewMockRequestStore(ctrl) + if tt.wantUpdate { + requestStore.EXPECT().UpdateState( + gomock.Any(), + tt.request.ID, + tt.request.Version, + tt.request.Version+1, + tt.outcome.State, + ).Return(nil) + } + + registry := consumer.TopicRegistry{} + if tt.wantPublish { + registry = newTerminalTestRegistry(t, ctrl, func(log entity.RequestLog) { + assert.Equal(t, tt.wantStatus, log.Status) + assert.Equal(t, tt.wantVersion, log.RequestVersion) + assert.Equal(t, tt.outcome.LastError, log.LastError) + if tt.outcome.Metadata == nil { + assert.Empty(t, log.Metadata) + } else { + assert.Equal(t, tt.outcome.Metadata, log.Metadata) + } + }) + } + + reconciled, err := ReconcileTerminalState(context.Background(), requestStore, registry, tt.request, tt.outcome) + assert.Equal(t, tt.wantReconciled, reconciled) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + }) + } +} + +func newTerminalTestRegistry( + t *testing.T, + ctrl *gomock.Controller, + checkLog func(entity.RequestLog), +) consumer.TopicRegistry { + publisher := queuemock.NewMockPublisher(ctrl) + publisher.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, message entityqueue.Message) error { + logEntry, err := entity.RequestLogFromBytes(message.Payload) + require.NoError(t, err) + checkLog(logEntry) + return nil + }, + ) + + queue := queuemock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher) + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{ + Key: topickey.TopicKeyLog, + Name: "log", + Queue: queue, + }}) + require.NoError(t, err) + return registry +} diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index c83f1ab2..0331f8d5 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/submitqueue/orchestrator/controller/cancel/cancel.go @@ -40,18 +40,16 @@ // its terminal state. // // The controller is idempotent: re-delivery of the same CancelRequest after -// the terminal request transition is a no-op; re-delivery after the -// Cancelling write skips the mark-cancelling step and proceeds straight to -// the batch lookup. On the batch path, re-delivery against an already -// Cancelling batch re-publishes to TopicKeySpeculate (a cheap no-op nudge -// the speculate controller absorbs). +// the terminal request transition republishes the terminal log without another +// state write; re-delivery after the Cancelling write skips the mark-cancelling +// step and proceeds straight to the batch lookup. On the batch path, +// re-delivery against an already Cancelling batch re-publishes to +// TopicKeySpeculate. // // Concurrent producers surface as the intrinsically retryable // storage.ErrVersionMismatch; the controller returns the wrapped error as-is // so the next attempt sees the new state and takes the other branch. -// storage.ErrNotFound on the initial Get (the start -// controller has not yet persisted the request) is returned as-is for the -// same reason. +// storage.ErrNotFound on the initial Get is returned for classifier handling. package cancel import ( @@ -208,28 +206,28 @@ 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 reconciles an unbatched request to Cancelled and emits the +// matching terminal log. A redelivery that observes an already-Cancelled request +// republishes the log without another state write. 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 { - c.metricsScope.Counter("request_update_errors").Inc(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 { - c.metricsScope.Counter("log_publish_errors").Inc(1) - return fmt.Errorf("failed to publish cancel log for request %s: %w", request.ID, err) + + _, err := corerequest.ReconcileTerminalState( + ctx, + c.store.GetRequestStore(), + c.registry, + request, + corerequest.TerminalOutcome{ + State: entity.RequestStateCancelled, + Metadata: metadata, + }, + ) + if err != nil { + c.metricsScope.Counter("request_reconcile_errors").Inc(1) + return fmt.Errorf("failed to reconcile cancelled request %s: %w", request.ID, err) } c.logger.Infow("request cancelled (not batched)", diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go index 4f7ed3ae..0309d910 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude.go +++ b/submitqueue/orchestrator/controller/conclude/conclude.go @@ -102,11 +102,6 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r 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 @@ -114,22 +109,31 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // so the log publish must still run when the request is already in the // target terminal state. for _, requestID := range batch.Contains { - request, err := c.store.GetRequestStore().Get(ctx, requestID) + requestStore := c.store.GetRequestStore() + request, err := requestStore.Get(ctx, requestID) if err != nil { metrics.NamedCounter(c.metricsScope, "process", "request_store_errors", 1) return fmt.Errorf("failed to get request %s: %w", requestID, 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. - 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. + reconciled, err := corerequest.ReconcileTerminalState( + ctx, + requestStore, + c.registry, + request, + corerequest.TerminalOutcome{ + State: requestState, + Metadata: map[string]string{ + "batch_id": batch.ID, + }, + }, + ) + if err != nil { + metrics.NamedCounter(c.metricsScope, "process", "request_reconcile_errors", 1) + return fmt.Errorf("failed to reconcile terminal state for request %s: %w", requestID, err) + } + + if !reconciled { c.logger.Warnw("request already in different terminal state, skipping reconcile", "batch_id", batch.ID, "request_id", requestID, @@ -138,29 +142,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r ) 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 - + } + if request.State == requestState { + metrics.NamedCounter(c.metricsScope, "process", "already_reconciled", 1) + } else { 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) - } } return nil // Success - message will be acked @@ -194,17 +185,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/mergeconflictsignal/mergeconflictsignal.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go index 66bbc9bd..babd6716 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go @@ -121,7 +121,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r "request_id", request.ID, "reason", result.Reason, ) - if err := c.failRequest(ctx, request, result.Reason); err != nil { + if _, err := corerequest.ReconcileTerminalState( + ctx, + c.store.GetRequestStore(), + c.registry, + request, + corerequest.TerminalOutcome{ + State: entity.RequestStateError, + LastError: result.Reason, + }, + ); err != nil { metrics.NamedCounter(c.metricsScope, opName, "fail_errors", 1) return fmt.Errorf("failed to fail request %s: %w", request.ID, err) } @@ -156,41 +165,6 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return nil // Success - message will be acked } -// failRequest drives the request to terminal RequestStateError and records the -// conflict reason on the request log. A not-mergeable verdict is an expected -// terminal outcome of the check, so the request is concluded here directly. -// -// Idempotent under at-least-once delivery: a redelivery whose request is already -// in Error skips the state CAS but still publishes the log (so a prior attempt -// that flipped the state but failed before logging is repaired); a request that -// reached a different terminal state (e.g. a racing cancel) is left untouched. -func (c *Controller) failRequest(ctx context.Context, request entity.Request, reason string) error { - switch { - case request.State == entity.RequestStateError: - // Idempotent retry: a prior delivery already wrote Error. Fall through to - // the log publish. - case entity.IsRequestStateTerminal(request.State): - c.logger.Warnw("request already in different terminal state, skipping fail", - "request_id", request.ID, - "state", string(request.State), - ) - return nil - default: - newVersion := request.Version + 1 - if err := c.store.GetRequestStore().UpdateState(ctx, request.ID, request.Version, newVersion, entity.RequestStateError); err != nil { - return fmt.Errorf("failed to update request %s state to error: %w", request.ID, err) - } - request.Version = newVersion - request.State = entity.RequestStateError - } - - logEntry := entity.NewRequestLog(request.ID, entity.RequestStatusError, request.Version, reason, nil) - if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil { - return fmt.Errorf("failed to publish request log for %s: %w", request.ID, err) - } - return nil -} - // publishRequestID publishes a request ID to the given topic key, partitioned by queue. func (c *Controller) publishRequestID(ctx context.Context, key consumer.TopicKey, requestID string, partitionKey string) error { payload, err := entity.RequestID{ID: requestID}.ToBytes()