sbalabanov · Nov 14, 2026
@@ -140,7 +140,7 @@ class node_submitter toneIndigo
 | **gateway/Land** | RPC | start | Mint the request id, persist an accepting receipt, publish the full land request, then persist Accepted |
 | **gateway/Cancel** | RPC | cancel | Persist `cancelling` for an existing land, then publish the cancel request |
 | **start** | LandRequest | validate, log | Persist the Request and publish it to validate |
-| **cancel** | CancelRequest | log, or speculate | Record Cancelling. Finish a request that has no applicable batch. Hand each cancellable batch attempt to speculate. Leave a Landing or already-terminal batch for conclude |
+| **cancel** | CancelRequest | log, or speculate | Hold a missing Request without spending failure budget or expiring, then record Cancelling. Finish a request that has no applicable batch. Hand each cancellable batch attempt to speculate. Leave a Landing or already-terminal batch for conclude |
 | **validate** | RequestID | merge-conflict-check (Runway), log | Dedup, fetch change metadata, claim changes, then publish the full `MergeRequest` to Runway keyed by the request id |
 | **landconflictsignal** | MergeResult | batch, or log | Correlate Runway's check; advance a landable request to batch, or fail a conflicted request |
 | **batch** | RequestID | dependency-analysis, log | Mint a Creating batch for the request and hand that batch id onward |
@@ -164,6 +164,8 @@ DLQ handlers do not resume the failed request's normal workflow. They reconcile
 
 The start DLQ reports a failed submission through an Error request log with version 0 when no Request was saved. Existing rows retain their ownership, terminal-outcome, and conditional-write protections.
 
+The primary cancel stage holds a cancellation whose Request is missing without expiry or failure-budget consumption, then performs ordinary best-effort cancellation with the original reason. The cancel DLQ holds the same missing identity until an authoritative Request exists, then reconciles only Error; it does not cancel builds or publish normal-work commands. Reconciliation preserves owning batches and other terminal outcomes, and its observed-version write rejects a concurrent batch claim. Missing IDs remain durable for operator investigation. Primary cancellation remains best-effort once an attempt reaches Landing.
+
 The narrow speculate nudge concerns other live batches after culprit failure; it never reopens the failed batch. Hook DLQ is report-and-stop, with side-effect recovery left to a person. Neither authorizes replay of the failed business command.
 
 The orchestrator publishes terminal status logs and the gateway alone materializes them; replaying a log into gateway read models is not orchestration restart. Runway owns failure-result publication for dead letters on its inbound merge queues, without rerunning the merge.

@@ -45,13 +45,15 @@
 // 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.
+// A missing request on the initial read holds the original delivery without
+// spending failure budget or expiring: independent cancel and start topics can
+// overtake one another. An ID that never appears remains an operator-visible
+// obligation, not a synthetic terminal outcome.
 package cancel
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/uber-go/tally"
@@ -82,7 +84,10 @@ type Controller struct {
 // Verify Controller implements consumer.Controller interface at compile time.
 var _ consumer.Controller = (*Controller)(nil)
 
-const opName = "process"
+const (
+	opName                      = "process"
+	waitForRequestDelayMs int64 = 1000
+)
 
 // NewController creates a new cancel controller for the orchestrator.
 func NewController(
@@ -104,7 +109,7 @@ func NewController(
 }
 
 // Process processes a cancel delivery from the queue.
-// Returns nil to ack (success), or error to nack (retry).
+// Returns nil to ack or honor a recorded Hold; errors follow the consumer retry policy.
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error {
 	msg := delivery.Message()
 
@@ -125,6 +130,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	}
 
 	request, err := store.GetRequestStore().Get(ctx, cancelReq.ID)
+	if errors.Is(err, storage.ErrNotFound) {
+		metrics.NamedCounter(c.metricsScope, opName, "waiting_for_request", 1, metrics.TagsFromContext(ctx)...)
+		delivery.Hold(waitForRequestDelayMs)
+		return nil
+	}
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...)
 		return fmt.Errorf("failed to get request %s: %w", cancelReq.ID, err)

@@ -94,7 +94,7 @@ func newController(t *testing.T, store orchstorage.Storage, registry consumer.To
 	return NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, staticStorageFactory{store: store}, registry, topickey.TopicKeyCancel, "orchestrator-cancel")
 }
 
-func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte, partitionKey string) consumer.Delivery {
+func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte, partitionKey string) *consumermock.MockDelivery {
 	msg := entityqueue.NewMessage("cancel-msg", payload, partitionKey, nil)
 	msg.Tenant = "q"
 	d := consumermock.NewMockDelivery(ctrl)
@@ -173,21 +173,23 @@ func TestProcess_AlreadyTerminal_NoOp(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestProcess_RequestNotFound_Retryable(t *testing.T) {
+func TestProcess_RequestNotFound_HoldsOriginalDelivery(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	registry, _ := newRegistry(t, ctrl)
-
 	reqStore := storagemock.NewMockRequestStore(ctrl)
-	reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, storage.ErrNotFound)
-
+	reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, fmt.Errorf("not visible yet: %w", storage.ErrNotFound)).Times(4)
 	store := orchstoragemock.NewMockStorage(ctrl)
-	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes()
-
 	controller := newController(t, store, registry)
-	err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", ""), "1"))
-	require.Error(t, err)
-	assert.ErrorIs(t, err, storage.ErrNotFound)
+	payload := cancelPayload(t, "1", "user changed mind")
+	msg := entityqueue.NewMessage("cancel-msg", payload, "1", nil)
+	msg.Tenant = "q"
+	delivery := consumermock.NewMockDelivery(ctrl)
+	delivery.EXPECT().Message().Return(msg).Times(4)
+	delivery.EXPECT().Hold(waitForRequestDelayMs).Times(4)
+	for range 4 {
+		require.NoError(t, controller.Process(context.Background(), delivery))
+	}
 }
 
 func TestProcess_CancelsUnbatchedRequest(t *testing.T) {
@@ -207,6 +209,7 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) {
 	// Two-step transition: mark Cancelling, then — after TerminateRequest re-reads
 	// the request — the terminal Cancelled CAS.
 	gomock.InOrder(
+		reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, storage.ErrNotFound),
 		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),
@@ -222,6 +225,9 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) {
 	expectBatchLookup(ctrl, store, batchStore, "1")
 
 	controller := newController(t, store, registry)
+	earlyDelivery := newDelivery(t, ctrl, cancelPayload(t, "1", "user changed mind"), "1")
+	earlyDelivery.EXPECT().Hold(waitForRequestDelayMs)
+	require.NoError(t, controller.Process(context.Background(), earlyDelivery))
 	err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", "user changed mind"), "1"))
 	require.NoError(t, err)
 
@@ -724,15 +730,16 @@ func TestProcess_RequestStoreError(t *testing.T) {
 	registry, _ := newRegistry(t, ctrl)
 
 	reqStore := storagemock.NewMockRequestStore(ctrl)
-	reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, fmt.Errorf("db down"))
+	storeErr := fmt.Errorf("db down")
+	reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{}, storeErr)
 
 	store := orchstoragemock.NewMockStorage(ctrl)
 	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes()
 
 	controller := newController(t, store, registry)
 	err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", ""), "1"))
-	require.Error(t, err)
+	require.ErrorIs(t, err, storeErr)
 }
 
 func TestFindBatches(t *testing.T) {

@@ -18,6 +18,8 @@ Disposition remains handler-specific: the generic request/batch reconcilers ackn
 
 Start reports the failed submission through an Error request log with version 0, because there is no saved Request version. No normal work is published. Reconciliation validates the request identity and queue; existing rows retain authoritative-identity, batch-ownership, terminal-winner, and conditional-write guards.
 
+Cancellation deliveries in the primary and final queues use Hold while the Request is absent, without expiry or failure-budget consumption. The primary controller then applies ordinary best-effort cancellation with the original reason. The cancel DLQ instead reconciles only Error under the batch-ownership and terminal-winner rules; it does not cancel builds or publish normal-work commands. Missing identities remain held for operator investigation.
+
 ## Terminal outputs and ownership
 
 - An already-Error request skips the state CAS but republishes its terminal status log. A different terminal request outcome is left unchanged. This repairs status delivery, not the original diagnostic content or timestamp.
@@ -31,7 +33,7 @@ Request logs are published to `log`; the gateway alone materializes them into it
 
 The [stage table](../../pipeline.go) is the source of truth rather than a second topology table here:
 
-- `start` uses dedicated request reconciliation; `cancel`, `validate`, and `batch` use generic request reconciliation. Their decoders read LandRequest, CancelRequest, or the topic's request-ID payload.
+- `start` and `cancel` use dedicated request reconciliation; `validate` and `batch` use generic request reconciliation. Their decoders read LandRequest, CancelRequest, or the topic's request-ID payload.
 - `dependency-analysis`, `build`, `submitqueue-land`, and `conclude` use generic batch reconciliation.
 - `buildsignal` carries a build-ID payload, not a full Build. Its handler loads the build to find the batch and leaves build status untouched. The two Runway signal handlers read a request or batch correlation ID from `MergeResult`.
 - **Speculation queue-isolation nudge:** after failing attributed culprit work (or the labelled payload fallback), a delivery that transitioned a batch may nudge `speculate` with another live batch. The primary re-derives current queue state; the failed batch is not reopened. This narrow protection for other work does not authorize replay of the dead-lettered command or guarantee a nudge after every partial reconciliation failure.

@@ -29,15 +29,12 @@ import (
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	corebatch "github.com/uber/submitqueue/submitqueue/orchestrator/core/batch"
+	requestcore "github.com/uber/submitqueue/submitqueue/orchestrator/core/request"
 	"go.uber.org/zap"
 )
 
-// RequestIDDecoder extracts the affected request's identity — its ID and its
-// queue — from the raw payload bytes of a DLQ message. Different primary
-// topics carry different payload shapes (LandRequest on start, CancelRequest
-// on cancel, RequestID on validate / batch), so the caller injects the right
-// decoder for the topic being reconciled. Returning an empty ID is treated as
-// a decode failure.
+// RequestIDDecoder extracts a request's ID and queue from a request-scoped
+// DLQ payload. Returning an empty ID is treated as a decode failure.
 type RequestIDDecoder func(payload []byte) (entity.RequestID, error)
 
 // DecodeLandRequestID extracts the request ID from a LandRequest payload
@@ -50,14 +47,13 @@ func DecodeLandRequestID(payload []byte) (entity.RequestID, error) {
 	return entity.RequestID{ID: lr.ID, Queue: lr.Queue}, nil
 }
 
-// DecodeCancelRequestID extracts the request ID from a CancelRequest payload
-// (the shape used by the cancel topic).
+// DecodeCancelRequestID extracts the request identity from a cancellation payload.
 func DecodeCancelRequestID(payload []byte) (entity.RequestID, error) {
-	cr, err := sqmq.UnmarshalCancelRequest(payload)
+	request, err := sqmq.UnmarshalCancelRequest(payload)
 	if err != nil {
 		return entity.RequestID{}, err
 	}
-	return entity.RequestID{ID: cr.ID, Queue: cr.Queue}, nil
+	return entity.RequestID{ID: request.ID, Queue: request.Queue}, nil
 }
 
 // DecodeRequestID extracts the request ID from a validate or batch payload.
@@ -69,18 +65,17 @@ func DecodeRequestID(primary consumer.TopicKey) RequestIDDecoder {
 }
 
 // requestController is the DLQ reconciler for request-scoped pipeline stages.
-// It is registered once per primary request-scoped topic (cancel, validate,
-// batch) with the matching decoder. On each delivery it decodes the
-// request ID and transitions the request to RequestStateError, unless a live
-// batch already owns the request's outcome.
+// It uses the matching decoder for cancel, validate, or batch and fails the
+// request unless a batch already owns its outcome.
 type requestController struct {
-	logger        *zap.SugaredLogger
-	metricsScope  tally.Scope
-	stores        orchstorage.Factory
-	registry      consumer.TopicRegistry
-	decode        RequestIDDecoder
-	topicKey      consumer.TopicKey
-	consumerGroup string
+	logger               *zap.SugaredLogger
+	metricsScope         tally.Scope
+	stores               orchstorage.Factory
+	registry             consumer.TopicRegistry
+	decode               RequestIDDecoder
+	topicKey             consumer.TopicKey
+	consumerGroup        string
+	retainMissingRequest bool
 }
 
 // Verify requestController implements consumer.Controller at compile time.
@@ -98,6 +93,17 @@ func NewDLQRequestController(
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) consumer.Controller {
+	return newRequestController(logger, scope, stores, registry, decode, topicKey, consumerGroup)
+}
+
+// NewDLQCancelController reconciles only Error; missing Requests remain durable until they exist.
+func NewDLQCancelController(logger *zap.SugaredLogger, scope tally.Scope, stores orchstorage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string) consumer.Controller {
+	controller := newRequestController(logger, scope, stores, registry, DecodeCancelRequestID, topicKey, consumerGroup)
+	controller.retainMissingRequest = true
+	return controller
+}
+
+func newRequestController(logger *zap.SugaredLogger, scope tally.Scope, stores orchstorage.Factory, registry consumer.TopicRegistry, decode RequestIDDecoder, topicKey consumer.TopicKey, consumerGroup string) *requestController {
 	name := string(topicKey) + "_controller"
 	return &requestController{
 		logger:        logger.Named(name),
@@ -151,34 +157,71 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv
 		"dlq_last_error", dmeta["dlq.last_error"],
 	)
 
-	// A batch that got as far as enrolling the request owns its outcome, and
-	// conclude writes that outcome when the batch finishes. This dead letter can
-	// be a redundant attempt whose predecessor is already speculating — the batch
-	// stage mints one batch per delivery — and TerminateRequest does not guard
-	// against Batched, so failing the request here would overwrite a live batch's
-	// claim and leave it building for a request reported as failed.
-	owner, err := owningBatch(ctx, store, rid.ID)
-	if err != nil {
-		metrics.NamedCounter(c.metricsScope, opName, "batch_lookup_errors", 1, metrics.TagsFromContext(ctx)...)
+	if c.retainMissingRequest {
+		err := c.reconcileCancelRequest(ctx, delivery, store, rid.ID, lastError, failureMeta)
+		if err != nil {
+			metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...)
+		}
 		return err
 	}
-	if owner != "" {
-		metrics.NamedCounter(c.metricsScope, opName, "skipped_owned_by_batch", 1, metrics.TagsFromContext(ctx)...)
-		c.logger.Infow("dlq reconcile: batch owns the request outcome, leaving completion to conclude",
-			"request_id", rid.ID,
-			"batch_id", owner,
-		)
-		return nil
+	owned, err := c.requestHasOwningBatch(ctx, store, rid.ID)
+	if err != nil || owned {
+		return err
 	}
-
 	if err := failRequest(ctx, store, c.registry, c.logger, rid.ID, lastError, failureMeta); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...)
 		return err
 	}
-
 	return nil
 }
 
+func (c *requestController) reconcileCancelRequest(ctx context.Context, delivery consumer.Delivery, store orchstorage.Storage, requestID, lastError string, metadata map[string]string) error {
+	request, err := store.GetRequestStore().Get(ctx, requestID)
+	if errors.Is(err, storage.ErrNotFound) {
+		metrics.NamedCounter(c.metricsScope, "process", "waiting_for_request", 1, metrics.TagsFromContext(ctx)...)
+		delivery.Hold(waitForRequestDelayMs)
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("failed to load request for cancel reconciliation: %w", err)
+	}
+	if request.State == entity.RequestStateError {
+		return requestcore.ResendTerminalRequestLog(ctx, c.registry, request)
+	}
+	if entity.IsRequestStateTerminal(request.State) {
+		return nil
+	}
+	owned, err := c.requestHasOwningBatch(ctx, store, requestID)
+	if err != nil || owned {
+		return err
+	}
+	request.State = entity.RequestStateError
+	newVersion := request.Version + 1
+	// The version observed before owner lookup fences a later batch claim.
+	if err := store.GetRequestStore().Update(ctx, request, request.Version, newVersion); err != nil {
+		return fmt.Errorf("failed to fail cancelled request %s: %w", requestID, err)
+	}
+	request.Version = newVersion
+	entry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusError, request.Version, lastError, metadata)
+	return requestcore.PublishLog(ctx, c.registry, entry, request.ID, "")
+}
+
+func (c *requestController) requestHasOwningBatch(ctx context.Context, store orchstorage.Storage, requestID string) (bool, error) {
+	owner, err := owningBatch(ctx, store, requestID)
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, "process", "batch_lookup_errors", 1, metrics.TagsFromContext(ctx)...)
+		return false, err
+	}
+	if owner == "" {
+		return false, nil
+	}
+	metrics.NamedCounter(c.metricsScope, "process", "skipped_owned_by_batch", 1, metrics.TagsFromContext(ctx)...)
+	c.logger.Infow("dlq reconcile: batch owns the request outcome, leaving completion to conclude", "request_id", requestID, "batch_id", owner)
+	return true, nil
+}
+
+const waitForRequestDelayMs int64 = 1000
+
 // owningBatch returns a batch that is still running or whose result has not been recorded on the request yet.
 // Creating batches must have claimed the request; failed batches do not stop it from being marked Error.
 func owningBatch(ctx context.Context, store orchstorage.Storage, requestID string) (string, error) {