fix(submitqueue): repair terminal request status on retry

submitqueue/orchestrator/controller/cancel/cancel.go

@@ -147,6 +147,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		"partition_key", msg.PartitionKey,
 	)
 
+	if request.State == entity.RequestStateCancelled {
+		return corerequest.PublishTerminalRequestLog(ctx, c.registry, request)
+	}
+
 	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)
@@ -194,10 +194,17 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	registry, pub := newRegistry(t, ctrl)
 
+	var terminalLogs []entity.RequestLog
 	var publishedTopics []string
 	pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
-		func(_ context.Context, topic string, _ entityqueue.Message) error {
+		func(_ context.Context, topic string, msg entityqueue.Message) error {
 			publishedTopics = append(publishedTopics, topic)
+			entry, err := sqmq.UnmarshalRequestLog(msg.Payload)
+			require.NoError(t, err)
+			terminalLogs = append(terminalLogs, entry)
+			if len(terminalLogs) == 1 {
+				return fmt.Errorf("publish unavailable")
+			}
 			return nil
 		}).AnyTimes()
 
@@ -206,11 +213,17 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) {
 	cancelling := entity.Request{ID: "1", Queue: "q", State: entity.RequestStateCancelling, Version: 3}
 	// Two-step transition: mark Cancelling, then — after TerminateRequest re-reads
 	// the request — the terminal Cancelled CAS.
+	var terminal entity.Request
 	gomock.InOrder(
 		reqStore.EXPECT().Get(gomock.Any(), "1").Return(started, nil),
 		reqStore.EXPECT().Update(gomock.Any(), requestWithState(started, entity.RequestStateCancelling), int32(2), int32(3)).Return(nil),
 		reqStore.EXPECT().Get(gomock.Any(), "1").Return(cancelling, nil),
-		reqStore.EXPECT().Update(gomock.Any(), requestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).Return(nil),
+		reqStore.EXPECT().Update(gomock.Any(), requestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).DoAndReturn(func(_ context.Context, snapshot entity.Request, _, newVersion int32) error {
+			terminal = snapshot
+			terminal.Version = newVersion
+			return nil
+		}),
+		reqStore.EXPECT().Get(gomock.Any(), "1").DoAndReturn(func(context.Context, string) (entity.Request, error) { return terminal, nil }),
 	)
 
 	batchStore := storagemock.NewMockBatchStore(ctrl)
@@ -223,10 +236,18 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) {
 
 	controller := newController(t, store, registry)
 	err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", "user changed mind"), "1"))
-	require.NoError(t, err)
+	require.Error(t, err)
+	require.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", "changed reason"), "1")))
+	require.Len(t, terminalLogs, 2)
+	assert.Equal(t, terminalLogs[0].Status, terminalLogs[1].Status)
+	assert.Equal(t, terminalLogs[0].RequestVersion, terminalLogs[1].RequestVersion)
+	assert.Equal(t, "original terminal details were not retained", terminalLogs[1].LastError)
+	assert.Equal(t, "user changed mind", terminalLogs[0].Metadata["reason"])
+	assert.Empty(t, terminalLogs[1].Metadata)
+	assert.Equal(t, int32(4), terminalLogs[1].RequestVersion)
 
 	// Only the request log entry should have been published.
-	assert.Equal(t, []string{"log"}, publishedTopics)
+	assert.Equal(t, []string{"log", "log"}, publishedTopics)
 }
 
 // TestProcess_AlreadyCancelling_SkipsMarkCancelling exercises the idempotent

submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go

@@ -114,7 +114,10 @@ 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.
+	if request.State == entity.RequestStateError {
+		return corerequest.PublishTerminalRequestLog(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 +168,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,25 +146,34 @@ 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).DoAndReturn(func(context.Context, string) (entity.Request, error) { return request, nil }).Times(3)
 	// 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)
+	reqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).DoAndReturn(func(_ context.Context, snapshot entity.Request, _, newVersion int32) error {
+		request = snapshot
+		request.Version = newVersion
+		return nil
+	})
 
 	store := orchstoragemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes()
 
-	// One publish is expected — the terminal log entry to the log topic. A publish
-	// to the batch topic would be a bug (gomock fails on the unexpected 2nd call).
+	var terminalLogs []entity.RequestLog
 	var gotTopic string
 	var gotPayload []byte
 	pub := queuemock.NewMockPublisher(ctrl)
 	pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 		func(_ context.Context, topic string, msg entityqueue.Message) error {
 			gotTopic = topic
 			gotPayload = msg.Payload
+			entry, err := sqmq.UnmarshalRequestLog(msg.Payload)
+			require.NoError(t, err)
+			terminalLogs = append(terminalLogs, entry)
+			if len(terminalLogs) == 1 {
+				return fmt.Errorf("publish unavailable")
+			}
 			return nil
 		},
-	)
+	).Times(2)
 	q := queuemock.NewMockQueue(ctrl)
 	q.EXPECT().Publisher().Return(pub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
@@ -180,27 +189,33 @@ func TestProcess_NotLandableMarksRequestError(t *testing.T) {
 
 	res := runwaymq.MergeResult{Id: testRequestID, Outcome: runwaypb.Outcome_FAILED, Reason: "conflict in foo.go"}
 	msg := entityqueue.NewMessage(testRequestID, resultPayload(t, res), testQueue, nil)
-	// Not-landable is an expected terminal outcome, so Process acks (no error).
+	require.Error(t, controller.Process(context.Background(), newDelivery(ctrl, msg)))
+	res.Outcome, res.Reason = runwaypb.Outcome_SUCCEEDED, "different retry result"
+	msg = entityqueue.NewMessage(testRequestID, resultPayload(t, res), testQueue, nil)
 	require.NoError(t, controller.Process(context.Background(), newDelivery(ctrl, msg)))
+	require.Len(t, terminalLogs, 2)
+	assert.Equal(t, terminalLogs[0].Status, terminalLogs[1].Status)
+	assert.Equal(t, terminalLogs[0].RequestVersion, terminalLogs[1].RequestVersion)
+	assert.Equal(t, "original terminal details were not retained", terminalLogs[1].LastError)
 
-	// The single publish is the terminal log entry carrying the conflict reason.
 	assert.Equal(t, "log", gotTopic)
 	logEntry, err := sqmq.UnmarshalRequestLog(gotPayload)
 	require.NoError(t, err)
 	assert.Equal(t, entity.RequestStatusError, logEntry.Status)
 	assert.Equal(t, int32(2), logEntry.RequestVersion)
-	assert.Equal(t, "conflict in foo.go", logEntry.LastError)
+	assert.Equal(t, "conflict in foo.go", terminalLogs[0].LastError)
 }
 
 func TestFailRequest_UpdateFailureLeavesRequestUnchanged(t *testing.T) {
 	ctrl := gomock.NewController(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).AnyTimes()
 
 	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,11 @@ 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.
+	if request.State == entity.RequestStateError {
+		return corerequest.PublishTerminalRequestLog(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)
@@ -843,10 +843,18 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te
 		State:        entity.RequestStateStarted,
 		Version:      1,
 	}
-	store, mockReqStore := newMockStorage(ctrl, request)
+	mockReqStore := storagemock.NewMockRequestStore(ctrl)
+	mockReqStore.EXPECT().Get(gomock.Any(), request.ID).DoAndReturn(func(context.Context, string) (entity.Request, error) { return request, nil }).Times(3)
+	store := orchstoragemock.NewMockStorage(ctrl)
+	store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
 	store.EXPECT().GetChangeStore().Return(newMockChangeStore(ctrl)).AnyTimes()
-	mockReqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil)
+	mockReqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).DoAndReturn(func(_ context.Context, snapshot entity.Request, _, newVersion int32) error {
+		request = snapshot
+		request.Version = newVersion
+		return nil
+	})
 
+	var terminalLogs []entity.RequestLog
 	mockPub := queuemock.NewMockPublisher(ctrl)
 	// Fail only the terminal log. The "validating" entry this stage now emits on
 	// the way in goes to the same topic, so failing every publish would abort
@@ -855,7 +863,10 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te
 		func(_ context.Context, _ string, msg entityqueue.Message) error {
 			entry, err := sqmq.UnmarshalRequestLog(msg.Payload)
 			if err == nil && entry.Status == entity.RequestStatusError {
-				return fmt.Errorf("publish boom")
+				terminalLogs = append(terminalLogs, entry)
+				if len(terminalLogs) == 1 {
+					return fmt.Errorf("publish boom")
+				}
 			}
 			return nil
 		},
@@ -871,7 +882,7 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te
 	require.NoError(t, err)
 
 	cpFactory := changeprovidermock.NewMockFactory(ctrl)
-	cpFactory.EXPECT().For(gomock.Any()).Return(&mockChangeProvider{}, nil).AnyTimes()
+	cpFactory.EXPECT().For(gomock.Any()).Return(&mockChangeProvider{}, nil)
 
 	mockValidator := validatormock.NewMockValidator(ctrl)
 	mockValidator.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(fmt.Errorf("%w: some validation error", validator.ErrRejected))
@@ -890,4 +901,11 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te
 	err = controller.Process(context.Background(), delivery)
 	require.Error(t, err)
 	assert.ErrorContains(t, err, "failed to terminate rejected request")
+	require.NoError(t, controller.Process(context.Background(), delivery))
+	require.Len(t, terminalLogs, 2)
+	assert.Equal(t, terminalLogs[0].Status, terminalLogs[1].Status)
+	assert.Equal(t, terminalLogs[0].RequestVersion, terminalLogs[1].RequestVersion)
+	assert.Equal(t, "original terminal details were not retained", terminalLogs[1].LastError)
+	assert.Contains(t, terminalLogs[0].LastError, "some validation error")
+	assert.Equal(t, int32(2), terminalLogs[1].RequestVersion)
 }

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"
+)
+
+const terminalDetailsNotRetained = "original terminal details were not retained"
+
+// PublishTerminalRequestLog repairs a terminal request's status without another state write.
+// Original diagnostics are not retained; the repair explicitly reports their absence.
+func PublishTerminalRequestLog(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, terminalDetailsNotRetained, 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 = terminalDetailsNotRetained, 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.

submitqueue/orchestrator/core/request/terminate_test.go

@@ -125,6 +125,8 @@ func TestTerminateRequest(t *testing.T) {
 		},
 		"already in target terminal state republishes log": {
 			targetState: entity.RequestStateError,
+			lastError:   "not the original reason",
+			metadata:    map[string]string{"source": "retry"},
 			mockFunc: func(rs *storagemock.MockRequestStore) {
 				already := entity.Request{ID: requestID, Queue: "q", State: entity.RequestStateError, Version: 5}
 				rs.EXPECT().Get(gomock.Any(), requestID).Return(already, nil)
@@ -137,6 +139,8 @@ func TestTerminateRequest(t *testing.T) {
 			wantLog: func(t *testing.T, log entity.RequestLog) {
 				assert.Equal(t, entity.RequestStatusError, log.Status)
 				assert.Equal(t, int32(5), log.RequestVersion)
+				assert.Equal(t, "original terminal details were not retained", log.LastError)
+				assert.Empty(t, log.Metadata)
 			},
 		},
 		"already in target terminal state returns republish error": {