fix(start): recover retained requests in place

doc/rfc/submitqueue/modular-queue-wiring.md

@@ -157,9 +157,8 @@ var Stages = []pipeline.Stage[Deps]{
             return start.NewController(d.Logger, d.Scope, d.Storage, d.Queues), nil
         },
         DLQ: func(d Deps) (consumer.Controller, error) {
-            return dlq.NewRequestController(d.Logger, d.Scope, d.Storage,
-                dlq.DecodeLandRequestID, dlq.TopicKey(topickey.Start),
-                "orchestrator-start-dlq"), nil
+            return start.NewController(d.Logger, d.Scope, d.Storage, d.Queues,
+                dlq.TopicKey(topickey.Start), "orchestrator-start-dlq"), nil
         },
     },
     { Key: topickey.Validate, /* same shape */ },

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. The DLQ reconcilers force a failed terminal state when a consumed stage cannot finish.
+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.
 
 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.
 
@@ -139,7 +139,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 |
+| **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 |
 | **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 |
@@ -162,7 +162,7 @@ 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.
 
-Two consumed stages do not follow that shape. 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.
+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.
 
 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.
 

submitqueue/orchestrator/BUILD.bazel

@@ -45,12 +45,24 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//api/base/hook:go_default_library",
+        "//platform/base/change:go_default_library",
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/consumer/mock:go_default_library",
         "//platform/extension/hook:go_default_library",
         "//platform/extension/hook/noop:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
         "//platform/pipeline:go_default_library",
+        "//submitqueue/core/messagequeue:go_default_library",
+        "//submitqueue/core/topickey:go_default_library",
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/storage/mock:go_default_library",
+        "//submitqueue/orchestrator/extension/storage:go_default_library",
+        "//submitqueue/orchestrator/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
         "@org_uber_go_zap//zaptest:go_default_library",
     ],
 )

submitqueue/orchestrator/controller/dlq/README.md

@@ -1,14 +1,16 @@
 # DLQ Reconciliation Controllers
 
-This package contains the controllers that drain each primary pipeline topic's `{topic}_dlq` companion and reconcile the affected request or batch into a terminal failed state. They are wired alongside the primary controllers in `service/submitqueue/orchestrator/server/main.go`.
+This package contains the controllers that drain primary pipeline topics' `{topic}_dlq` companions and reconcile the affected request or batch into a terminal failed state. The pipeline topology in `submitqueue/orchestrator/pipeline.go` wires these beside the primary controllers. Start is the exception: its owning controller resumes the full original request rather than manufacturing a failed outcome.
 
 ## Design principles
 
 **The DLQ is the final destination.** A message in `{topic}_dlq` has already failed the primary controller's retry budget (`Retry.MaxAttempts` for a retryable error) or surfaced as non-retryable on the first attempt. There is no further dead-letter beyond this point — every DLQ subscription is configured with `DLQ.Enabled = false`, so a DLQ controller's outcome is either "reconciled" or "the message keeps coming back". A genuinely unprocessable DLQ message (e.g. a malformed payload that no controller can ever decode) must be removed by an operator from the queue manually.
 
 **The implementation is deliberately the simplest and most reliable thing that can work.** Each controller does the same three steps: decode the payload to recover the affected `RequestID` or `BatchID`, fetch the entity, transition it to its terminal failed state with the same optimistic-locking CAS the primary pipeline uses. No business logic, no branching on the original error, no per-stage cleanup. Adding behaviour here trades off the one property the DLQ has to provide — convergence.
 
-**Reconcile only; do not attempt to recover.** A DLQ controller never tries to re-run the failed stage, retry the original action against an external dependency, or repair a partially-completed transition. The original controller already failed; the DLQ controller's only job is to make sure the request and batch state stop saying "in progress" and start saying "failed", so the gateway reports an honest answer and downstream tooling can move on. Recovery, when it is appropriate, is a separate concern handled by an operator or a future reconciliation job — not by this code path.
+**Reconcile failed stages; recover start in place.** Except for start, a DLQ controller never tries to re-run the failed stage, retry the original action against an external dependency, or repair a partially-completed transition. The original controller already failed; the DLQ controller's only job is to make sure the request and batch state stop saying "in progress" and start saying "failed", so the gateway reports an honest answer and downstream tooling can move on. Recovery, when it is appropriate, is a separate concern handled by an operator or a future reconciliation job — not by these generic reconcilers.
+
+The `start_dlq` subscription uses the start controller under the same unlimited-retry final policy. It creates the original Request when absent, then loads the authoritative snapshot after Create or AlreadyExists. Matching Started snapshots replay the Started log and normal `validate` hand-off. Later states and Cancelling retain their current owner; terminal snapshots replay their stored winning terminal log. Conflicting immutable inputs, unknown states, missing snapshots, and malformed messages return errors and remain durable. No tombstone, new request ID, synthetic Error, or return to the finite-budget start subscription is needed.
 
 ## Convergence guarantee
 
@@ -18,7 +20,7 @@ The recognised error condition is handled explicitly in `dlq.go`:
 
 - `storage.ErrNotFound` → logged at warn and treated as success. The request or batch never persisted; there is nothing to reconcile.
 
-Everything else — including `storage.ErrVersionMismatch` on the CAS — is returned without controller-level classification and redelivered until it either succeeds or hits the attempt cap. `ErrVersionMismatch` is already retryable at its declaration, while the always-retryable processor makes every other non-nil reconciliation error retryable too.
+Everything else — including `storage.ErrVersionMismatch` on the CAS — is returned without controller-level classification and redelivered until reconciliation succeeds. `ErrVersionMismatch` is already retryable at its declaration, while the always-retryable processor makes every other non-nil reconciliation error retryable too.
 
 ## Request log entries are published to Gateway
 
@@ -30,10 +32,10 @@ Two controller shapes cover the eleven primary pipeline topics:
 
 | Controller | Topics | Decoded ID | Terminal state |
 |---|---|---|---|
-| `NewDLQRequestController` | `start`, `validate`, `batch`, `cancel`, `log` | `RequestID` | `RequestStateError` |
+| `NewDLQRequestController` | `validate`, `batch`, `cancel` | `RequestID` | `RequestStateError` |
 | `NewDLQBatchController` | `speculate`, `build`, `land`, `conclude` | `BatchID` | `BatchStateFailed` + fan-out to member requests as `RequestStateError` |
 
-`buildsignal` carries a `Build` payload and has its own small dedicated controller. The split exists because the DLQ message payload shape mirrors the primary topic's payload shape (the queue framework preserves bytes verbatim under the `_dlq` topic name), so the decoder is what changes per topic — not the reconciliation step. The package-level `RequestIDDecoder` interface plus `DecodeLandRequestID` / `DecodeCancelRequestID` / `DecodeRequestID` cover the three payload shapes used by request-scoped topics.
+`buildsignal` carries a `Build` payload and has its own small dedicated controller. The split exists because the DLQ message payload shape mirrors the primary topic's payload shape (the queue framework preserves bytes verbatim under the `_dlq` topic name), so the decoder is what changes per topic — not the reconciliation step. The package-level `RequestIDDecoder` interface plus `DecodeCancelRequestID` / `DecodeRequestID` cover the two payload shapes used by request-scoped topics.
 
 ## Idempotency and concurrent activity
 

submitqueue/orchestrator/controller/dlq/request.go

@@ -34,22 +34,12 @@ import (
 
 // 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
+// topics carry different payload shapes (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.
 type RequestIDDecoder func(payload []byte) (entity.RequestID, error)
 
-// DecodeLandRequestID extracts the request ID from a LandRequest payload
-// (the shape used by the start topic).
-func DecodeLandRequestID(payload []byte) (entity.RequestID, error) {
-	lr, err := sqmq.UnmarshalLandRequest(payload)
-	if err != nil {
-		return entity.RequestID{}, err
-	}
-	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).
 func DecodeCancelRequestID(payload []byte) (entity.RequestID, error) {
@@ -69,10 +59,8 @@ 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 (start, 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 live batch already owns its outcome.
 type requestController struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope

submitqueue/orchestrator/controller/dlq/request_test.go

@@ -46,34 +46,6 @@ func TestDLQRequestController_InterfaceAndAccessors(t *testing.T) {
 	assert.Equal(t, "orchestrator-validate-dlq", c.ConsumerGroup())
 }
 
-func TestDLQRequestController_Process_LandRequestPayload(t *testing.T) {
-	ctrl := gomock.NewController(t)
-
-	requestStore := storagemock.NewMockRequestStore(ctrl)
-	request := entity.Request{
-		ID: "1", Queue: "q", Version: 1, State: entity.RequestStateStarted,
-	}
-	requestStore.EXPECT().Get(gomock.Any(), "1").Return(request, nil)
-	requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil)
-
-	registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error {
-		return nil
-	})
-
-	store := orchstoragemock.NewMockStorage(ctrl)
-	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
-	store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes()
-	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
-
-	c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store: store}, registry, DecodeLandRequestID, TopicKey(topickey.TopicKeyStart), "orchestrator-start-dlq")
-
-	payload, err := sqmq.Marshal(sqmq.StartFromLandRequest(entity.LandRequest{ID: "1", Queue: "q"}))
-	require.NoError(t, err)
-
-	delivery := newMockDelivery(ctrl, payload)
-	require.NoError(t, c.Process(context.Background(), delivery))
-}
-
 func TestDLQRequestController_Process_TenantPayloadQueueMismatchAcks(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	store := orchstoragemock.NewMockStorage(ctrl)