fix(submitqueue): replay persisted terminal request logs

submitqueue/entity/request.go

@@ -66,6 +66,17 @@ func IsRequestStateHalted(s RequestState) bool {
 	return IsRequestStateTerminal(s) || s == RequestStateCancelling
 }
 
+// RequestTerminalDetails contains the immutable facts of a terminal request outcome.
+// A zero TimestampMs denotes unavailable details on a legacy snapshot.
+type RequestTerminalDetails struct {
+	// TimestampMs is the terminal outcome time in milliseconds since Unix epoch.
+	TimestampMs int64 `json:"timestamp_ms"`
+	// LastError is the diagnostic associated with the terminal outcome, or empty.
+	LastError string `json:"last_error"`
+	// Metadata contains additional diagnostic facts, or nil when absent.
+	Metadata map[string]string `json:"metadata"`
+}
+
 // Request defines a request to land a set of code changes on the source control repository's target branch.
 // The object is immutable after creation.
 type Request struct {
@@ -89,6 +100,8 @@ type Request struct {
 
 	// State is the current state of the land request.
 	State RequestState `json:"state"`
+	// TerminalDetails contains the facts of the terminal outcome, or its zero value when unavailable.
+	TerminalDetails RequestTerminalDetails `json:"terminal_details"`
 	// Version is the version of the object. It is used for optimistic locking.
 	// Versioning starts at 1 and is incremented for each change to the object.
 	Version int32 `json:"version"`

submitqueue/entity/request_log.go

@@ -84,6 +84,9 @@ const (
 	RequestStatusCancelled RequestStatus = "cancelled"
 )
 
+// TerminalDetailsUnavailable marks a legacy terminal log whose original diagnostics are unavailable.
+const TerminalDetailsUnavailable = "terminal_details_unavailable"
+
 // RequestEvent is something that happened to a request while it sat at a status,
 // rather than a status of its own.
 //

submitqueue/extension/storage/README.md

@@ -72,3 +72,7 @@ Take the first branch that applies:
 3. **Map** — a pipeline controller needs the lookup at runtime → a dedicated mapping store keyed by the attribute.
 
 The bar for 3 is a hot-path need: one mapping per access path, never per attribute, and never for ops/debug queries — run those against SQL replicas directly. But don't contort 1–2 to dodge a legitimate 3; a primary key that hashes half the entity's fields, or an aggregate bloated into listing everything, is the same duplication hidden in a worse place.
+
+### Terminal request snapshots
+
+A terminal Request replacement includes its outcome timestamp, last error, and metadata in the same conditional write as its state. Same-state replay publishes those persisted facts without another state transition; incoming retry diagnostics cannot replace the winning outcome. Legacy snapshots without details emit an explicitly incomplete log, which can repair a non-terminal summary but cannot erase a known same-status, same-request-version terminal diagnostic. The gateway protection must be deployed before enabling legacy fallback publication.

submitqueue/gateway/core/request/materializer.go

@@ -188,6 +188,9 @@ func logWins(log entity.RequestLog, summary entity.RequestSummary) bool {
 		if log.RequestVersion != summary.RequestVersion {
 			return log.RequestVersion > summary.RequestVersion
 		}
+		if log.Status == summary.Status && log.Metadata[entity.TerminalDetailsUnavailable] == "true" && summary.Metadata[entity.TerminalDetailsUnavailable] != "true" {
+			return false
+		}
 	}
 	return log.TimestampMs > summary.StatusTimestampMs
 }

submitqueue/gateway/core/request/materializer_test.go

@@ -63,6 +63,20 @@ func TestMaterializer_PersistLog(t *testing.T) {
 		)
 		require.NoError(t, f.materializer.PersistLog(context.Background(), log))
 	})
+
+	t.Run("legacy repair appends history and repairs mappings without erasing diagnostics", func(t *testing.T) {
+		f := newMaterializerReceiptFixture(gomock.NewController(t))
+		known := base
+		known.Status, known.RequestVersion, known.StatusTimestampMs = entity.RequestStatusError, 2, 20
+		known.LastError, known.Metadata = "original failure", map[string]string{"source": "validator"}
+		fallback := entity.NewRequestStatusLog(base.Queue, base.RequestID, known.Status, known.RequestVersion, "", map[string]string{entity.TerminalDetailsUnavailable: "true"})
+		f.logs.EXPECT().Insert(gomock.Any(), fallback).Return(nil)
+		f.summaries.EXPECT().Get(gomock.Any(), base.RequestID).Return(known, nil)
+		f.expectURIMappings(known, nil)
+		f.receipts.EXPECT().Create(gomock.Any(), receiptFromTestSummary(known)).Return(nil)
+		require.NoError(t, f.materializer.PersistLog(context.Background(), fallback))
+	})
+
 }
 
 func TestMaterializer_RecordsLifecycleMetrics(t *testing.T) {
@@ -313,6 +327,48 @@ func TestLogWins(t *testing.T) {
 				Status: entity.RequestStatusLanded, RequestVersion: 2, TimestampMs: 200,
 			},
 		},
+		{
+			name:     "incomplete legacy repair preserves known same terminal outcome",
+			current:  entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 2, StatusTimestampMs: 100, Metadata: map[string]string{"source": "validator"}},
+			incoming: entity.RequestLog{Status: entity.RequestStatusError, RequestVersion: 2, TimestampMs: 200, Metadata: map[string]string{entity.TerminalDetailsUnavailable: "true"}},
+			want:     false,
+		},
+		{
+			name:     "legacy repair promotes non-terminal",
+			current:  entity.RequestSummary{Status: entity.RequestStatusStarted, RequestVersion: 0, StatusTimestampMs: 300, Metadata: nil},
+			incoming: entity.RequestLog{Status: entity.RequestStatusError, RequestVersion: 2, TimestampMs: 200, Metadata: map[string]string{entity.TerminalDetailsUnavailable: "true"}},
+			want:     true,
+		},
+		{
+			name:     "legacy repair respects higher version",
+			current:  entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 3, StatusTimestampMs: 100, Metadata: nil},
+			incoming: entity.RequestLog{Status: entity.RequestStatusError, RequestVersion: 2, TimestampMs: 200, Metadata: map[string]string{entity.TerminalDetailsUnavailable: "true"}},
+			want:     false,
+		},
+		{
+			name:     "legacy repair permits newer version",
+			current:  entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 1, StatusTimestampMs: 100, Metadata: nil},
+			incoming: entity.RequestLog{Status: entity.RequestStatusError, RequestVersion: 2, TimestampMs: 200, Metadata: map[string]string{entity.TerminalDetailsUnavailable: "true"}},
+			want:     true,
+		},
+		{
+			name:     "legacy repair retains different-status ordering",
+			current:  entity.RequestSummary{Status: entity.RequestStatusLanded, RequestVersion: 2, StatusTimestampMs: 100, Metadata: nil},
+			incoming: entity.RequestLog{Status: entity.RequestStatusError, RequestVersion: 2, TimestampMs: 200, Metadata: map[string]string{entity.TerminalDetailsUnavailable: "true"}},
+			want:     true,
+		},
+		{
+			name:     "legacy repair may replace incomplete diagnostics",
+			current:  entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 2, StatusTimestampMs: 100, Metadata: map[string]string{entity.TerminalDetailsUnavailable: "true"}},
+			incoming: entity.RequestLog{Status: entity.RequestStatusError, RequestVersion: 2, TimestampMs: 200, Metadata: map[string]string{entity.TerminalDetailsUnavailable: "true"}},
+			want:     true,
+		},
+		{
+			name:     "ordinary same-terminal log retains timestamp ordering",
+			current:  entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 2, StatusTimestampMs: 100, Metadata: nil},
+			incoming: entity.RequestLog{Status: entity.RequestStatusError, RequestVersion: 2, TimestampMs: 200, Metadata: nil},
+			want:     true,
+		},
 	}
 
 	for _, tt := range tests {

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

@@ -17,6 +17,7 @@ package cancel
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -64,6 +65,16 @@ func requestWithState(request entity.Request, state entity.RequestState) entity.
 	return request
 }
 
+func terminalRequestWithState(request entity.Request, state entity.RequestState) gomock.Matcher {
+	return gomock.Cond(func(got entity.Request) bool {
+		if got.TerminalDetails.TimestampMs <= 0 {
+			return false
+		}
+		got.TerminalDetails = request.TerminalDetails
+		return reflect.DeepEqual(got, requestWithState(request, state))
+	})
+}
+
 // cancelPayload serializes a CancelRequest to JSON bytes for test message payloads.
 func cancelPayload(t *testing.T, id, reason string) []byte {
 	payload, err := sqmq.Marshal(sqmq.CancelFromEntity(entity.CancelRequest{ID: id, Queue: "q", Reason: reason}))
@@ -161,7 +172,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 +205,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 +224,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(), terminalRequestWithState(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 +247,15 @@ 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], terminalLogs[1])
+	assert.Equal(t, "user changed mind", terminalLogs[1].Metadata["reason"])
+	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
@@ -244,7 +273,7 @@ func TestProcess_AlreadyCancelling_SkipsMarkCancelling(t *testing.T) {
 	// see Cancelling (the prior pass already recorded intent).
 	reqStore.EXPECT().Get(gomock.Any(), "1").Return(cancelling, nil).Times(2)
 	// Only the terminal CAS — the mark-cancelling step is a no-op.
-	reqStore.EXPECT().Update(gomock.Any(), requestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).Return(nil)
+	reqStore.EXPECT().Update(gomock.Any(), terminalRequestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).Return(nil)
 
 	batchStore := storagemock.NewMockBatchStore(ctrl)
 
@@ -301,7 +330,7 @@ func TestProcess_UnbatchedVersionMismatch_Retryable(t *testing.T) {
 		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)).
+		reqStore.EXPECT().Update(gomock.Any(), terminalRequestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).
 			Return(storage.ErrVersionMismatch),
 	)
 
@@ -579,7 +608,7 @@ func TestProcess_BatchedWithoutMatchCancelsRequest(t *testing.T) {
 		requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil),
 		requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateCancelling), int32(2), int32(3)).Return(nil),
 		requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(cancelling, nil),
-		requestStore.EXPECT().Update(gomock.Any(), requestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).Return(nil),
+		requestStore.EXPECT().Update(gomock.Any(), terminalRequestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).Return(nil),
 	)
 
 	batchStore := storagemock.NewMockBatchStore(ctrl)
@@ -615,7 +644,7 @@ func TestProcess_CreatingBatchDoesNotSuppressRequestCancellation(t *testing.T) {
 		requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil),
 		requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateCancelling), int32(2), int32(3)).Return(nil),
 		requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(cancelling, nil),
-		requestStore.EXPECT().Update(gomock.Any(), requestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).Return(nil),
+		requestStore.EXPECT().Update(gomock.Any(), terminalRequestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).Return(nil),
 	)
 
 	batchStore := storagemock.NewMockBatchStore(ctrl)

submitqueue/orchestrator/controller/conclude/conclude_test.go

@@ -17,6 +17,7 @@ package conclude
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -46,9 +47,15 @@ func (f staticStorageFactory) For(orchstorage.Config) (orchstorage.Storage, erro
 	return f.store, nil
 }
 
-func requestWithState(request entity.Request, state entity.RequestState) entity.Request {
+func requestWithState(request entity.Request, state entity.RequestState) gomock.Matcher {
 	request.State = state
-	return request
+	return gomock.Cond(func(got entity.Request) bool {
+		if got.TerminalDetails.TimestampMs <= 0 {
+			return false
+		}
+		got.TerminalDetails = request.TerminalDetails
+		return reflect.DeepEqual(got, request)
+	})
 }
 
 // batchIDPayload serializes a BatchID to JSON bytes for test message payloads.

submitqueue/orchestrator/controller/dlq/dlq_test.go

@@ -17,6 +17,7 @@ package dlq
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"testing"
 
 	orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"
@@ -64,9 +65,15 @@ func batchWithState(batch entity.Batch, state entity.BatchState) entity.Batch {
 	return batch
 }
 
-func requestWithState(request entity.Request, state entity.RequestState) entity.Request {
+func requestWithState(request entity.Request, state entity.RequestState) gomock.Matcher {
 	request.State = state
-	return request
+	return gomock.Cond(func(got entity.Request) bool {
+		if got.TerminalDetails.TimestampMs <= 0 {
+			return false
+		}
+		got.TerminalDetails = request.TerminalDetails
+		return reflect.DeepEqual(got, request)
+	})
 }
 
 // failRequest

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