feat(orchestrator): carry the queue explicitly on every internal payload (#527)

## Summary

### Why?

Consumers today learn a message's queue only by loading the referenced
entity from storage, and entity IDs are opaque — nothing may parse a
queue out of an ID prefix. For storage to become queue-scoped (resolved
per queue like every other extension), every consumer must hold the
queue before its first storage read, so the queue has to ride on the
wire explicitly.

### What?

The internal payload types (`RequestID`, `BatchID`, `BuildID`,
`CancelRequest`) gain a `queue` field; the change is additive JSON, and
payloads written before the field existed decode with an empty queue.
Publishers stamp it from the entity they already hold: start,
mergeconflictsignal, batch, speculate, mergesignal, build, buildsignal,
and the orchestrator cancel controller; the gateway cancel controller
stamps the authoritative queue from the stored request summary,
overriding caller input. Consumers guard that a non-empty payload queue
matches the loaded entity's queue and reject mismatches as malformed
(non-retryable, straight to DLQ) — the guard becomes the routing input
once storage is queue-resolved. Buildsignal's re-publish to speculate
now partitions by the batch's queue instead of the inherited batch-ID
partition key, matching every other speculate publisher and restoring
the per-queue serial-processing guarantee that stage relies on.

## Test Plan

✅ `go test ./...` ✅ `make fmt` ✅ `make lint` — new unit tests cover the
queue stamp on published payloads and the mismatch rejection guard.

Author: behinddwalls

submitqueue/entity/batch.go

@@ -187,6 +187,8 @@ func BatchFromBytes(data []byte) (Batch, error) {
 type BatchID struct {
 	// ID is the globally unique identifier for the batch.
 	ID string `json:"id"`
+	// Queue is the name of the queue processing the batch. Empty on payloads written before the field existed.
+	Queue string `json:"queue"`
 }
 
 // ToBytes serializes the BatchID to JSON bytes for queue message payload.

submitqueue/entity/build.go

@@ -81,6 +81,8 @@ func BuildFromBytes(data []byte) (Build, error) {
 type BuildID struct {
 	// ID is the globally unique identifier for the build.
 	ID string `json:"id"`
+	// Queue is the name of the queue processing the batch this build verifies. Empty on payloads written before the field existed.
+	Queue string `json:"queue"`
 }
 
 // ToBytes serializes the BuildID to JSON bytes for queue message payload.

submitqueue/entity/cancel.go

@@ -21,6 +21,8 @@ import "encoding/json"
 type CancelRequest struct {
 	// ID is the globally unique identifier of the request to cancel. Format: "<queue>/<counter_value>".
 	ID string `json:"id"`
+	// Queue is the name of the queue processing the request to cancel. Empty on payloads written before the field existed.
+	Queue string `json:"queue"`
 	// Reason is an optional free-form explanation of why the cancellation was requested.
 	Reason string `json:"reason"`
 }

submitqueue/entity/request.go

@@ -113,6 +113,8 @@ func RequestFromBytes(data []byte) (Request, error) {
 type RequestID struct {
 	// ID is the globally unique identifier for the land request.
 	ID string `json:"id"`
+	// Queue is the name of the queue processing the land request. Empty on payloads written before the field existed.
+	Queue string `json:"queue"`
 }
 
 // ToBytes serializes the RequestID to JSON bytes for queue message payload.

submitqueue/extension/storage/mock/request_batch_store_mock.go

@@ -86,14 +86,19 @@ func (c *cancelController) Cancel(ctx context.Context, req entity.CancelRequest)
 	)
 
 	// Verify the sqid exists before recording intent or publishing.
-	if _, err := c.requestSummaryStore.Get(ctx, req.ID); err != nil {
+	summary, err := c.requestSummaryStore.Get(ctx, req.ID)
+	if err != nil {
 		if storage.IsNotFound(err) {
 			metrics.NamedCounter(c.metricsScope, opName, "not_found", 1)
 			return errs.NewUserError(&RequestNotFoundError{Sqid: req.ID})
 		}
 		return fmt.Errorf("failed to look up request summary for sqid=%s: %w", req.ID, err)
 	}
 
+	// Stamp the authoritative queue from the stored summary onto the payload,
+	// overriding whatever the caller supplied.
+	req.Queue = summary.Queue
+
 	// Record the user's intent in the request log before publishing. Writing direct to the
 	// store (rather than via the log topic) keeps the gateway-emitted entry consistent with
 	// the Land "accepted" entry and guarantees the entry is visible the moment Cancel returns.

submitqueue/gateway/controller/cancel.go

@@ -95,6 +95,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get request %s: %w", rid.ID, err)
 	}
 
+	// The payload's queue must match the request's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if rid.Queue != "" && rid.Queue != request.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of request %s", rid.Queue, request.Queue, request.ID)
+	}
+
 	c.logger.Infow("received batch event",
 		"request_id", request.ID,
 		"queue", request.Queue,
@@ -355,15 +362,16 @@ func (c *Controller) populateBatch(ctx context.Context, batch entity.Batch) (ent
 	return batch, nil
 }
 
-// publish publishes a batch ID to the specified topic key.
-func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
-	bid := entity.BatchID{ID: batchID}
+// publish publishes a batch ID to the specified topic key, stamped with and
+// partitioned by the batch's queue.
+func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error {
+	bid := entity.BatchID{ID: batchID, Queue: queue}
 	payload, err := bid.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {

submitqueue/orchestrator/controller/batch/batch.go

@@ -195,6 +195,100 @@ func TestController_Process_Success(t *testing.T) {
 	require.NoError(t, err)
 }
 
+// TestController_Process_QueueMismatchRejected asserts a payload whose queue
+// disagrees with the request's authoritative queue is rejected without
+// touching the counter, the batch store, or the publisher.
+func TestController_Process_QueueMismatchRejected(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	request := testRequest()
+
+	mockReqStore := storagemock.NewMockRequestStore(ctrl)
+	mockReqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil)
+
+	mockStorage := storagemock.NewMockStorage(ctrl)
+	mockStorage.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
+
+	// Counter with no EXPECTs — must not be called.
+	cnt := countermock.NewMockCounter(ctrl)
+	controller := newTestController(t, ctrl, cnt, mockStorage, nil, fmt.Errorf("should not publish"))
+
+	payload, err := entity.RequestID{ID: request.ID, Queue: "some-other-queue"}.ToBytes()
+	require.NoError(t, err)
+	msg := entityqueue.NewMessage(request.ID, payload, request.Queue, nil)
+	delivery := consumermock.NewMockDelivery(ctrl)
+	delivery.EXPECT().Message().Return(msg).AnyTimes()
+	delivery.EXPECT().Attempt().Return(1).AnyTimes()
+
+	require.Error(t, controller.Process(context.Background(), delivery))
+}
+
+// TestController_Process_StampsQueueOnSpeculatePayload asserts the batch ID
+// published to speculate carries the batch's queue.
+func TestController_Process_StampsQueueOnSpeculatePayload(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	request := testRequest()
+
+	var speculateMsgs []entityqueue.Message
+	mockPub := queuemock.NewMockPublisher(ctrl)
+	mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, topic string, msg entityqueue.Message) error {
+			if topic == "speculate" {
+				speculateMsgs = append(speculateMsgs, msg)
+			}
+			return nil
+		},
+	).AnyTimes()
+	mockQ := queuemock.NewMockQueue(ctrl)
+	mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes()
+
+	registry, err := consumer.NewTopicRegistry(
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: mockQ},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
+		},
+	)
+	require.NoError(t, err)
+
+	mockBatchStore := storagemock.NewMockBatchStore(ctrl)
+	mockBatchStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockBatchStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockReqStore := storagemock.NewMockRequestStore(ctrl)
+	mockReqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil).AnyTimes()
+	mockReqStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockBatchDependentStore := storagemock.NewMockBatchDependentStore(ctrl)
+	mockBatchDependentStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockRequestBatchStore := storagemock.NewMockRequestBatchStore(ctrl)
+	mockRequestBatchStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+
+	mockStorage := storagemock.NewMockStorage(ctrl)
+	mockStorage.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
+	mockStorage.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes()
+	mockStorage.EXPECT().GetBatchDependentStore().Return(mockBatchDependentStore).AnyTimes()
+	mockStorage.EXPECT().GetRequestBatchStore().Return(mockRequestBatchStore).AnyTimes()
+	mockStorage.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
+
+	analyzerFactory := conflictmock.NewMockFactory(ctrl)
+	analyzerFactory.EXPECT().For(gomock.Any()).Return(all.New(), nil).AnyTimes()
+	controller := NewController(
+		zaptest.NewLogger(t).Sugar(), tally.NoopScope, registry, newSequentialCounter(ctrl),
+		mockStorage, analyzerFactory, topickey.TopicKeyBatch, "orchestrator-batch",
+	)
+
+	msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil)
+	delivery := consumermock.NewMockDelivery(ctrl)
+	delivery.EXPECT().Message().Return(msg).AnyTimes()
+	delivery.EXPECT().Attempt().Return(1).AnyTimes()
+
+	require.NoError(t, controller.Process(context.Background(), delivery))
+	require.Len(t, speculateMsgs, 1)
+	bid, err := entity.BatchIDFromBytes(speculateMsgs[0].Payload)
+	require.NoError(t, err)
+	assert.Equal(t, request.Queue, bid.Queue)
+	assert.Equal(t, request.Queue, speculateMsgs[0].PartitionKey)
+}
+
 // TestController_Process_PublishesBatchedLog asserts the controller emits a
 // "batched" request log carrying the request ID, the post-CAS request version,
 // and the batch ID it was placed into.

submitqueue/orchestrator/controller/batch/batch_test.go

@@ -89,6 +89,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get batch %s: %w", bid.ID, err)
 	}
 
+	// The payload's queue must match the batch's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if bid.Queue != "" && bid.Queue != batch.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID)
+	}
+
 	c.logger.Infow("received build event",
 		"batch_id", batch.ID,
 		"queue", batch.Queue,
@@ -151,7 +158,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	// Hand off to the buildsignal poll loop; it calls Status, updates the
 	// persisted Build, publishes to speculate, and holds its delivery
 	// between polls until terminal.
-	if err := c.publish(ctx, topickey.TopicKeyBuildSignal, build); err != nil {
+	if err := c.publish(ctx, topickey.TopicKeyBuildSignal, build, batch.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to buildsignal: %w", err)
 	}
@@ -184,11 +191,12 @@ func (c *Controller) loadBatches(ctx context.Context, batchIDs []string) ([]enti
 	return batches, nil
 }
 
-// publish publishes a build's ID to the specified topic key. Only the
-// identifier travels on the queue; the consumer loads the full Build from
-// storage, keeping the message small and the store the single source of truth.
-func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, build entity.Build) error {
-	payload, err := entity.BuildID{ID: build.ID}.ToBytes()
+// publish publishes a build's ID to the specified topic key, stamped with the
+// batch's queue and partitioned by the batch ID. Only the identifier and its
+// queue travel on the queue; the consumer loads the full Build from storage,
+// keeping the message small and the store the single source of truth.
+func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, build entity.Build, queue string) error {
+	payload, err := entity.BuildID{ID: build.ID, Queue: queue}.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize build ID: %w", err)
 	}

submitqueue/orchestrator/controller/build/build.go

@@ -135,6 +135,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get batch %s: %w", build.BatchID, err)
 	}
 
+	// The payload's queue must match the batch's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if buildID.Queue != "" && buildID.Queue != batch.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of batch %s", buildID.Queue, batch.Queue, batch.ID)
+	}
+
 	buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: batch.Queue})
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "status_errors", 1)
@@ -169,8 +176,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to update status for build %s: %w", build.ID, err)
 	}
 
-	// Re-evaluate the batch state machine with the latest build status.
-	if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, updatedBuild.BatchID, msg.PartitionKey); err != nil {
+	// Re-evaluate the batch state machine with the latest build status. The
+	// speculate topic is partitioned by queue like every other speculate
+	// publisher, so a queue's batches keep their serial processing guarantee.
+	if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, updatedBuild.BatchID, batch.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to speculate: %w", err)
 	}
@@ -210,15 +219,16 @@ func pollDelay(status entity.BuildStatus) int64 {
 	}
 }
 
-// publishBatchID publishes a batch ID to the topic identified by key.
-func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
-	bid := entity.BatchID{ID: batchID}
+// publishBatchID publishes a batch ID to the topic identified by key, stamped
+// with and partitioned by the batch's queue.
+func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error {
+	bid := entity.BatchID{ID: batchID, Queue: queue}
 	payload, err := bid.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {