fix: retain cancellations received before start

doc/rfc/submitqueue/workflow.md

@@ -4,7 +4,7 @@ The orchestrator processes land requests through the queue-driven pipeline decla
 
 Two cycles re-enter speculation. `speculate` dispatches funded paths to `build`; `build` starts each pending path and publishes the build id to `buildsignal`; `buildsignal` polls with `delivery.Hold` and, when the status changes, publishes the batch id back to `speculate`. A batch that can land is published to `submitqueue-land`, which sends the merge request to Runway; `landsignal` correlates the result, marks the batch Succeeded or Failed, and fans out to `conclude` and `speculate`. `speculate` also publishes Failed and Cancelled batches straight to `conclude`.
 
-Terminal request states are not `conclude`'s alone. `cancel` completes a request that has not been enrolled in a batch. `landconflictsignal` fails a request Runway reports as conflicted. `conclude` maps a terminal batch onto its member requests. Most DLQ reconcilers force a failed terminal state when a consumed stage cannot finish; start resumes the original full-payload work instead.
+Terminal request states are not `conclude`'s alone. `cancel` completes a request that has not been enrolled in a batch. `landconflictsignal` fails a request Runway reports as conflicted. `conclude` maps a terminal batch onto its member requests. Most DLQ reconcilers force a failed terminal state when a consumed stage cannot finish; start and cancel resume their original full-payload work instead.
 
 The same `Stages` table registers a `submitqueue-hook` stage, outside the order of the flow above. It consumes hook events and invokes the wired hooks, and it publishes nothing onward. No orchestrator controller publishes a hook event; `service/submitqueue/orchestrator/server` resolves every event to noop. The comment on that row in `pipeline.go` is the contract: any stage can publish there.
 
@@ -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 | Create/load the authoritative Request; only Started replays log and 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 |
@@ -162,9 +162,9 @@ Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscriptio
 
 Most DLQ controllers do not re-attempt the failed work. They decode the payload to a `RequestID` or `BatchID` and drive that entity to a terminal failed state — `RequestStateError` for requests, `BatchStateFailed` for batches, with fan-out to the member requests. Topics that carry a full payload recover the id from it: the `landconflictsignal` DLQ reads the request id from Runway's `MergeResult`, and the `landsignal` DLQ reads the batch id. The `buildsignal` DLQ decodes a build id, loads that build, and fails the batch that owns it. A missing build or an empty batch id is acked: there is no batch to fail from this signal. State writes use the same optimistic-locking CAS as the primary pipeline, so a late primary-pipeline update wins cleanly and a version mismatch is asked back for redelivery.
 
-Three consumed stages do not follow that shape. The start DLQ uses the start controller to create or resume the original full-payload request in place, without returning to the finite-budget primary subscription. After Create or AlreadyExists it loads the authoritative snapshot and verifies the queue, ID, ordered URIs, and strategy. Started replays its log and normal validate hand-off; later states and Cancelling preserve their current owner, and terminal states replay their stored winning terminal log. Conflicting inputs, unknown states, failed reads, and malformed payloads remain unacknowledged under the final retry policy. This reconciliation read is not a fence against already-running later stages. The speculate DLQ fails the batches the failure attributes — often not the batch named on the message, because a run re-plans the whole queue — and then republishes to `speculate`, so one dead letter does not strand every other batch. The hook DLQ records the dropped event and acks. A hook never writes pipeline state, so there is no request or batch to fail; republishing the logged event is how an operator recovers the side effect. A genuinely unprocessable request or batch DLQ message, typically a malformed payload, must be removed by an operator, because those consumers retry until reconciliation succeeds.
+Four consumed stages do not follow that shape. The start DLQ uses the start controller to create or resume the original full-payload request in place, without returning to the finite-budget primary subscription. After Create or AlreadyExists it loads the authoritative snapshot and verifies the queue, ID, ordered URIs, and strategy. Started replays its log and normal validate hand-off; later states and Cancelling preserve their current owner, and terminal states replay their stored winning terminal log. Conflicting inputs, unknown states, failed reads, and malformed payloads remain unacknowledged under the final retry policy. This reconciliation read is not a fence against already-running later stages. The cancel DLQ uses the cancel controller with the original reason and cancellation semantics. Start and cancel use independent topics, so cancellation can arrive before the Request exists; either cancel subscription holds that same delivery until the row appears, without a synthetic Error or an age-based expiry. An ID that never appears remains a visible durable obligation for operator investigation. Cancellation remains best-effort once an attempt reaches Landing and does not fence batch claims or promotions. The speculate DLQ fails the batches the failure attributes — often not the batch named on the message, because a run re-plans the whole queue — and then republishes to `speculate`, so one dead letter does not strand every other batch. The hook DLQ records the dropped event and acks. A hook never writes pipeline state, so there is no request or batch to fail; republishing the logged event is how an operator recovers the side effect. A genuinely unprocessable request or batch DLQ message, typically a malformed payload, must be removed by an operator, because those consumers retry until reconciliation succeeds.
 
-See `submitqueue/orchestrator/controller/dlq/README.md` for the reconcile-only constraints. The speculate and hook controllers in that wiring are the exceptions to the generic request/batch mapping that README's table still summarizes.
+See `submitqueue/orchestrator/controller/dlq/README.md` for the reconcile-only constraints. The start, cancel, speculate, and hook controllers in that wiring are the exceptions to the generic request/batch mapping that README's table still summarizes.
 
 ## Ownership by service
 

submitqueue/orchestrator/controller/cancel/cancel.go

@@ -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)

submitqueue/orchestrator/controller/cancel/cancel_test.go

@@ -105,7 +105,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)
@@ -184,21 +184,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) {
@@ -226,6 +228,7 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) {
 	// the request — the terminal Cancelled CAS.
 	var terminal entity.Request
 	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),
@@ -246,6 +249,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.Error(t, err)
 	require.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", "changed reason"), "1")))
@@ -753,15 +759,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) {