refactor: extract shared request-termination helper (#448)

## Summary
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
(success / already-target-state / 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 so they go
to the DLQ.
- Add table-driven tests for TerminateRequest and update the conclude,
cancel, and validate tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

## Test Plan
Unit tests

## Issues

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ubettigole

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

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

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