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

submitqueue/orchestrator/controller/cancel/BUILD.bazel

@@ -6,9 +6,9 @@ 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/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,9 +57,9 @@ 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"
+	"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"
@@ -338,29 +338,18 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error
 }
 
 // publishBatchID publishes a BatchID-payload message to the specified topic key.
+//
+// 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, partitionKey string) error {
-	bid := entity.BatchID{ID: batchID}
-	payload, err := bid.ToBytes()
+	payload, err := entity.BatchID{ID: batchID}.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
-
-	msg := entityqueue.NewMessage(batchID, payload, partitionKey, 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, partitionKey)
 }
 
 // Name returns the controller name for logging and metrics.

submitqueue/orchestrator/controller/cancel/cancel_test.go

@@ -334,23 +334,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()
 
@@ -381,7 +389,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) {
@@ -413,7 +426,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)
@@ -451,7 +468,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
 		},
 	)