refactor(orchestrator): centralize request termination

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 = [

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)
+	}
+}

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
+}

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)",