diff --git a/submitqueue/orchestrator/controller/cancel/BUILD.bazel b/submitqueue/orchestrator/controller/cancel/BUILD.bazel index c3e5f24d..4c7cf3f4 100644 --- a/submitqueue/orchestrator/controller/cancel/BUILD.bazel +++ b/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", diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index 6af35525..a6ff8f5f 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/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. diff --git a/submitqueue/orchestrator/controller/cancel/cancel_test.go b/submitqueue/orchestrator/controller/cancel/cancel_test.go index 887b9acd..328cfada 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel_test.go +++ b/submitqueue/orchestrator/controller/cancel/cancel_test.go @@ -334,11 +334,14 @@ 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) @@ -346,11 +349,16 @@ func TestProcess_BatchPath_HandsOffToSpeculate(t *testing.T) { 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 }, )