fix(orchestrator): mint distinct message IDs for cancel re-publishes

## Summary

### Why?

The cancel controller published its speculate hand-off with the bare batch ID as the message ID. The queue deduplicates on (topic, partition key, message ID) against every row it has not garbage-collected yet, consumed ones included, so a redelivery's re-publish for the same batch was a silent no-op — leaving a batch stuck Cancelling with nothing driving it to terminal. Same class of bug as b1f5795 in stovepipe.

### What?

`publishBatchID` now goes through `submitqueue/core/publish`: `publish.UniqueID` mints a distinct message ID per publish (and documents the dedup rule once), and `publish.Message` owns the registry-lookup plumbing the controller previously hand-rolled. The batch-path test now asserts the payload still carries the batch ID while the message ID is distinct per publish, and the multi-batch tests assert on the payload's batch ID for the same reason.

## Test Plan

✅ `bazel test //submitqueue/orchestrator/controller/cancel/...`

# Conflicts:
#	submitqueue/orchestrator/controller/cancel/BUILD.bazel
#	submitqueue/orchestrator/controller/cancel/cancel.go

# Please enter the commit message for your changes. Lines starting
# with '#' will be kept; you may remove them yourself if you want to.
# An empty message aborts the commit.
#
# interactive rebase in progress; onto 8548220f
# Last command done (1 command done):
#    pick c167e22 # fix(orchestrator): mint distinct message IDs for cancel re-publishes
# No commands remaining.
# You are currently rebasing branch 'preetam/speculation-cancel-msgid' on '8548220f'.
#
# Changes to be committed:
#	modified:   submitqueue/orchestrator/controller/cancel/BUILD.bazel
#	modified:   submitqueue/orchestrator/controller/cancel/cancel.go
#	modified:   submitqueue/orchestrator/controller/cancel/cancel_test.go
#

Author: behinddwalls

submitqueue/orchestrator/controller/cancel/BUILD.bazel

@@ -6,10 +6,10 @@ go_library(
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/cancel",
     visibility = ["//visibility:public"],
     deps = [
-        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
         "//submitqueue/core/batch:go_default_library",
+        "//submitqueue/core/publish:go_default_library",
         "//submitqueue/core/request:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",

submitqueue/orchestrator/controller/cancel/cancel.go

@@ -57,10 +57,10 @@ import (
 	"sort"
 
 	"github.com/uber-go/tally"
-	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/metrics"
 	corebatch "github.com/uber/submitqueue/submitqueue/core/batch"
+	"github.com/uber/submitqueue/submitqueue/core/publish"
 	corerequest "github.com/uber/submitqueue/submitqueue/core/request"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/entity"
@@ -359,29 +359,18 @@ func (c *Controller) cancelBatch(ctx context.Context, store storage.Storage, bat
 
 // publishBatchID publishes a BatchID-payload message to the specified topic
 // key, stamped with and partitioned by the batch's queue.
+//
+// The message ID is distinct per publish (publish.UniqueID). The queue
+// deduplicates on (topic, partition key, message ID) against every row it has
+// not collected yet, consumed ones included, so a bare batch ID would make the
+// redelivery re-publish documented above a silent no-op — leaving a batch
+// Cancelling with nothing driving it to terminal.
 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()
+	payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
-
-	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
-
-	q, ok := c.registry.Queue(key)
-	if !ok {
-		return fmt.Errorf("no queue registered for topic key %s", key)
-	}
-
-	topicName, ok := c.registry.TopicName(key)
-	if !ok {
-		return fmt.Errorf("no topic name registered for topic key %s", key)
-	}
-
-	if err := q.Publisher().Publish(ctx, topicName, msg); err != nil {
-		return fmt.Errorf("failed to publish message: %w", err)
-	}
-	return nil
+	return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, queue)
 }
 
 // Name returns the controller name for logging and metrics.

submitqueue/orchestrator/controller/cancel/cancel_test.go

@@ -358,23 +358,31 @@ func TestProcess_UnbatchedRequestDisappears_Retryable(t *testing.T) {
 
 // TestProcess_BatchPath_HandsOffToSpeculate asserts the entire batch path:
 // the request intent CAS runs, the batch intent CAS to Cancelling runs, and
-// exactly one publish lands on the speculate topic with the batch ID as the
-// message ID. The controller does NOT perform a terminal batch CAS, does
-// NOT publish to conclude, and does NOT emit a per-request log on this path
-// (the gateway already wrote the Cancelling intent log; conclude writes the
-// terminal log when it reconciles request state).
+// exactly one publish lands on the speculate topic carrying the batch ID. The
+// message ID is not the bare batch ID: the queue deduplicates on it, so a
+// redelivery's re-publish would be silently dropped and the batch left
+// Cancelling with nothing driving it. The controller does NOT perform a
+// terminal batch CAS, does NOT publish to conclude, and does NOT emit a
+// per-request log on this path (the gateway already wrote the Cancelling
+// intent log; conclude writes the terminal log when it reconciles request
+// state).
 func TestProcess_BatchPath_HandsOffToSpeculate(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	registry, pub := newRegistry(t, ctrl)
 
 	type pubRec struct {
 		topic string
 		msgID string
+		// payloadID is the batch ID the message actually carries, which is what
+		// the consumer acts on — the message ID is only the queue's dedup key.
+		payloadID string
 	}
 	var records []pubRec
 	pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 		func(_ context.Context, topic string, msg entityqueue.Message) error {
-			records = append(records, pubRec{topic: topic, msgID: msg.ID})
+			bid, err := entity.BatchIDFromBytes(msg.Payload)
+			require.NoError(t, err)
+			records = append(records, pubRec{topic: topic, msgID: msg.ID, payloadID: bid.ID})
 			return nil
 		}).AnyTimes()
 
@@ -406,7 +414,12 @@ func TestProcess_BatchPath_HandsOffToSpeculate(t *testing.T) {
 	err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "q/1", "stop"), "q/1"))
 	require.NoError(t, err)
 
-	assert.Equal(t, []pubRec{{topic: "speculate", msgID: "q/batch/1"}}, records)
+	require.Len(t, records, 1)
+	assert.Equal(t, "speculate", records[0].topic)
+	assert.Equal(t, batch.ID, records[0].payloadID)
+	assert.NotEqual(t, batch.ID, records[0].msgID,
+		"a bare batch ID as the message ID lets the queue swallow the redelivery re-publish")
+	assert.Contains(t, records[0].msgID, batch.ID)
 }
 
 func TestProcess_CancelsEveryApplicableBatch(t *testing.T) {
@@ -438,7 +451,11 @@ func TestProcess_CancelsEveryApplicableBatch(t *testing.T) {
 	)
 	publisher.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).DoAndReturn(
 		func(_ context.Context, _ string, msg entityqueue.Message) error {
-			operations = append(operations, "publish:"+msg.ID)
+			// The message ID is the queue's dedup key and is distinct per
+			// publish; the payload carries the batch ID the consumer acts on.
+			bid, err := entity.BatchIDFromBytes(msg.Payload)
+			require.NoError(t, err)
+			operations = append(operations, "publish:"+bid.ID)
 			return nil
 		},
 	).Times(2)
@@ -477,7 +494,9 @@ func TestProcess_BatchFailureDoesNotPreventLaterCancellation(t *testing.T) {
 	batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch2, entity.BatchStateCancelling), int32(2), int32(3)).Return(nil)
 	publisher.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).DoAndReturn(
 		func(_ context.Context, _ string, msg entityqueue.Message) error {
-			assert.Equal(t, batch2.ID, msg.ID)
+			bid, err := entity.BatchIDFromBytes(msg.Payload)
+			require.NoError(t, err)
+			assert.Equal(t, batch2.ID, bid.ID)
 			return nil
 		},
 	)