fix(submitqueue): repair terminal request status on retry

submitqueue/orchestrator/controller/cancel/cancel.go

@@ -147,6 +147,12 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		"partition_key", msg.PartitionKey,
 	)
 
+	// The cancellation may be terminal even though its final log was never published.
+	if request.State == entity.RequestStateCancelled {
+		return corerequest.ResendTerminalRequestLog(ctx, c.registry, request)
+	}
+
+	// Other terminal outcomes belong to their winning writer and are left untouched.
 	if entity.IsRequestStateTerminal(request.State) {
 		metrics.NamedCounter(c.metricsScope, opName, "already_terminal", 1, metrics.TagsFromContext(ctx)...)
 		return nil

submitqueue/orchestrator/controller/cancel/cancel_test.go

@@ -161,7 +161,7 @@ func TestProcess_AlreadyTerminal_NoOp(t *testing.T) {
 
 	reqStore := storagemock.NewMockRequestStore(ctrl)
 	reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{
-		ID: "1", Queue: "q", State: entity.RequestStateCancelled, Version: 5,
+		ID: "1", Queue: "q", State: entity.RequestStateLanded, Version: 5,
 	}, nil)
 
 	store := orchstoragemock.NewMockStorage(ctrl)

submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go

@@ -114,7 +114,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		"partition_key", msg.PartitionKey,
 	)
 
-	// Short-circuit halted requests: the cancel path owns driving them terminal.
+	// The failed conflict check may be terminal even though its final log was never published.
+	if request.State == entity.RequestStateError {
+		return corerequest.ResendTerminalRequestLog(ctx, c.registry, request)
+	}
+
 	if entity.IsRequestStateHalted(request.State) {
 		metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1, metrics.TagsFromContext(ctx)...)
 		c.logger.Infow("skipping landconflict signal for halted request",
@@ -165,39 +169,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	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-landable 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, store orchstorage.Storage, 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
-		request.State = entity.RequestStateError
-		if err := store.GetRequestStore().Update(ctx, request, request.Version, newVersion); err != nil {
-			return fmt.Errorf("failed to update request %s state to error: %w", request.ID, err)
-		}
-		request.Version = newVersion
-	}
-
-	logEntry := entity.NewRequestStatusLog(request.Queue, 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
+	_, err := corerequest.TerminateRequest(ctx, store, c.registry, request.ID, entity.RequestStateError, reason, nil)
+	return err
 }
 
 // publishRequestID publishes a request ID to the given topic key, stamped with

submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal_test.go

@@ -146,7 +146,7 @@ func TestProcess_NotLandableMarksRequestError(t *testing.T) {
 
 	reqStore := storagemock.NewMockRequestStore(ctrl)
 	request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateStarted, Version: 1}
-	reqStore.EXPECT().Get(gomock.Any(), testRequestID).Return(request, nil)
+	reqStore.EXPECT().Get(gomock.Any(), testRequestID).Return(request, nil).Times(2)
 	// The request is driven to terminal Error inline (version 1 -> 2).
 	reqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil)
 
@@ -197,10 +197,11 @@ func TestFailRequest_UpdateFailureLeavesRequestUnchanged(t *testing.T) {
 
 	request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateStarted, Version: 1}
 	reqStore := storagemock.NewMockRequestStore(ctrl)
+	reqStore.EXPECT().Get(gomock.Any(), testRequestID).Return(request, nil)
 	reqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(fmt.Errorf("db down"))
 
 	store := orchstoragemock.NewMockStorage(ctrl)
-	store.EXPECT().GetRequestStore().Return(reqStore)
+	store.EXPECT().GetRequestStore().Return(reqStore).Times(2)
 
 	controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, staticStorageFactory{store: store}, consumer.TopicRegistry{},
 		runwaymq.TopicKeyMergeConflictCheckSignal, "orchestrator-landconflictsignal")

submitqueue/orchestrator/controller/validate/validate.go

@@ -133,10 +133,12 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		"partition_key", msg.PartitionKey,
 	)
 
-	// Short-circuit if the request has been halted — either it already reached a
-	// terminal state, or the cancel controller has recorded a cancellation intent
-	// (RequestStateCancelling). Without this guard we would still publish to batch
-	// and spawn a batch for a request that should never proceed.
+	// The rejection may be terminal even though its final log was never published.
+	if request.State == entity.RequestStateError {
+		return corerequest.ResendTerminalRequestLog(ctx, c.registry, request)
+	}
+
+	// Cancellation intent halts forward progress even before the terminal outcome.
 	if entity.IsRequestStateHalted(request.State) {
 		coremetrics.NamedCounter(c.metricsScope, "process", "skipped_halted", 1, coremetrics.TagsFromContext(ctx)...)
 		c.logger.Infow("skipping validate for halted request",

submitqueue/orchestrator/controller/validate/validate_test.go

@@ -614,7 +614,7 @@ func TestController_Process_TerminalShortCircuit(t *testing.T) {
 	for _, state := range []entity.RequestState{
 		entity.RequestStateCancelled,
 		entity.RequestStateLanded,
-		entity.RequestStateError,
+		entity.RequestStateCancelling,
 	} {
 		t.Run(string(state), func(t *testing.T) {
 			ctrl := gomock.NewController(t)

submitqueue/orchestrator/core/request/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "log.go",
+        "terminal_log.go",
         "terminate.go",
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/core/request",

submitqueue/orchestrator/core/request/terminal_log.go

@@ -0,0 +1,39 @@
+// 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"
+
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/submitqueue/entity"
+)
+
+// ResendTerminalRequestLog resends the request's terminal status without another state write.
+// The log identifies the resend; original diagnostics may be unavailable.
+// For example, validation can store Error and crash before publishing its status log.
+// Redelivery must resend the status without another request-state write.
+func ResendTerminalRequestLog(ctx context.Context, registry consumer.TopicRegistry, request entity.Request) error {
+	status, err := terminalStateToStatus(request.State)
+	if err != nil {
+		return err
+	}
+	logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, status, request.Version, "terminal status resent for consistency; original details may not have been retained", nil)
+	if err := PublishLog(ctx, registry, logEntry, request.ID, ""); err != nil {
+		return fmt.Errorf("failed to publish request log for %s: %w", request.ID, err)
+	}
+	return nil
+}

submitqueue/orchestrator/core/request/terminate.go

@@ -86,7 +86,7 @@ type TerminationResult struct {
 //   - 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.
+// lastError and metadata describe the first transition; same-state repair reports unavailable original details.
 //
 // 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.
@@ -123,6 +123,7 @@ func TerminateRequest(
 		// Idempotent retry: a prior attempt already wrote the terminal state.
 		// Skip the CAS and fall through to re-publish the terminal log.
 		outcome = TerminationOutcomeAlreadyInTargetState
+		lastError, metadata = "terminal status resent for consistency; original details may not have been retained", nil
 	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.