address comments

Author: manjari25

entity/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "batch.go",
         "batch_dependent.go",
         "build.go",
+        "cancel.go",
         "change_provider.go",
         "queue_config.go",
         "request.go",
@@ -20,6 +21,7 @@ go_test(
     srcs = [
         "batch_test.go",
         "build_test.go",
+        "cancel_test.go",
         "request_test.go",
     ],
     embed = [":entity"],

entity/cancel.go

@@ -0,0 +1,22 @@
+package entity
+
+import "encoding/json"
+
+// Cancel represents a request to cancel a previously submitted land request.
+// The object is immutable after creation.
+type Cancel struct {
+	// Sqid is the globally unique identifier of the land request to cancel.
+	Sqid string `json:"sqid"`
+}
+
+// ToBytes serializes the Cancel to JSON bytes for queue message payload.
+func (c Cancel) ToBytes() ([]byte, error) {
+	return json.Marshal(c)
+}
+
+// CancelFromBytes deserializes a Cancel from JSON bytes.
+func CancelFromBytes(data []byte) (Cancel, error) {
+	var c Cancel
+	err := json.Unmarshal(data, &c)
+	return c, err
+}

entity/cancel_test.go

@@ -0,0 +1,63 @@
+package entity
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCancel_ToBytes(t *testing.T) {
+	cancel := Cancel{
+		Sqid: "test-queue/123",
+	}
+
+	data, err := cancel.ToBytes()
+	require.NoError(t, err)
+	assert.Contains(t, string(data), "test-queue/123")
+}
+
+func TestCancelFromBytes(t *testing.T) {
+	original := Cancel{
+		Sqid: "my-queue/999",
+	}
+
+	data, err := original.ToBytes()
+	require.NoError(t, err)
+
+	deserialized, err := CancelFromBytes(data)
+	require.NoError(t, err)
+	assert.Equal(t, original, deserialized)
+}
+
+func TestCancelFromBytes_InvalidJSON(t *testing.T) {
+	_, err := CancelFromBytes([]byte(`{"invalid": json"}`))
+	assert.Error(t, err)
+}
+
+func TestCancel_SerializationRoundTrip(t *testing.T) {
+	tests := []struct {
+		name   string
+		cancel Cancel
+	}{
+		{
+			name:   "standard sqid",
+			cancel: Cancel{Sqid: "queue1/100"},
+		},
+		{
+			name:   "long sqid",
+			cancel: Cancel{Sqid: "my-long-queue-name/999999"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			data, err := tt.cancel.ToBytes()
+			require.NoError(t, err)
+
+			deserialized, err := CancelFromBytes(data)
+			require.NoError(t, err)
+			assert.Equal(t, tt.cancel, deserialized)
+		})
+	}
+}

gateway/controller/cancel.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/uber-go/tally/v4"
 	"github.com/uber/submitqueue/core/metrics"
+	"github.com/uber/submitqueue/entity"
 	"github.com/uber/submitqueue/entity/queue"
 	extqueue "github.com/uber/submitqueue/extension/queue"
 	pb "github.com/uber/submitqueue/gateway/protopb"
@@ -40,9 +41,25 @@ func (c *CancelController) Cancel(ctx context.Context, req *pb.CancelRequest) (r
 
 	sqid := req.Sqid
 
-	// TODO: Insert the request to the event store
+	if sqid == "" {
+		return &pb.CancelResponse{
+			Error: &pb.Error{Message: "sqid is required"},
+		}, nil
+	}
+
+	// TODO -
+	// Look up event store to see if sqid exists -
+	//  - if found and request is already in a terminal state; return cancellation failed with appropriate
+	//    error message.
+	//  - if found and request is not in a terminal state; publish a cancel entity onto the cancel topic
+	//    to be picked up by the orchestrator.
+	//  - if not found; publish a cancel entity onto the cancel topic
+	//    to be picked up by the orchestrator.
+	cancel := entity.Cancel{
+		Sqid: sqid,
+	}
 
-	if err := c.publishToQueue(ctx, sqid); err != nil {
+	if err := c.publishToQueue(ctx, cancel); err != nil {
 		c.logger.Errorw("failed to publish cancel request to queue",
 			"sqid", sqid,
 			"error", err,
@@ -56,19 +73,23 @@ func (c *CancelController) Cancel(ctx context.Context, req *pb.CancelRequest) (r
 	)
 
 	return &pb.CancelResponse{
-		Sqid: sqid,
+		Sqid:          sqid,
+		CurrentStatus: pb.RequestStatus_CANCELLATION_ACCEPTED,
 	}, nil
 }
 
-// publishToQueue publishes a cancel request to the cancel queue for async processing.
-func (c *CancelController) publishToQueue(ctx context.Context, sqid string) error {
-	payload := []byte(sqid)
+// publishToQueue publishes a cancel entity to the cancel queue for async processing.
+func (c *CancelController) publishToQueue(ctx context.Context, cancel entity.Cancel) error {
+	payload, err := cancel.ToBytes()
+	if err != nil {
+		return fmt.Errorf("failed to serialize cancel entity: %w", err)
+	}
 
 	// Create queue message
 	// - Message ID: sqid for idempotency
-	// - Payload: sqid as bytes
-	// - Partition key: sqid (ensures ordering per request)
-	msg := queue.NewMessage(sqid, payload, sqid, nil)
+	// - Payload: serialized Cancel entity
+	// - Partition key: empty (no inherent ordering for cancel requests)
+	msg := queue.NewMessage(cancel.Sqid, payload, "", nil)
 
 	if err := c.publisher.Publish(ctx, c.topic, msg); err != nil {
 		return fmt.Errorf("failed to publish message: %w", err)

gateway/controller/cancel_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally/v4"
+	"github.com/uber/submitqueue/entity"
 	"github.com/uber/submitqueue/entity/queue"
 	queuemock "github.com/uber/submitqueue/extension/queue/mock"
 	pb "github.com/uber/submitqueue/gateway/protopb"
@@ -21,6 +22,21 @@ func TestNewCancelController(t *testing.T) {
 	require.NotNil(t, controller)
 }
 
+func TestCancel_ReturnsErrorOnEmptySqid(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, noopPublisher(ctrl), "cancel")
+	ctx := context.Background()
+
+	req := &pb.CancelRequest{Sqid: ""}
+	resp, err := controller.Cancel(ctx, req)
+
+	require.NoError(t, err)
+	assert.Empty(t, resp.Sqid)
+	assert.Equal(t, pb.RequestStatus_UNKNOWN, resp.CurrentStatus)
+	require.NotNil(t, resp.Error)
+	assert.Equal(t, "sqid is required", resp.Error.Message)
+}
+
 func TestCancel_ReturnsSqid(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, noopPublisher(ctrl), "cancel")
@@ -31,6 +47,7 @@ func TestCancel_ReturnsSqid(t *testing.T) {
 
 	require.NoError(t, err)
 	assert.Equal(t, "test-queue/123", resp.Sqid)
+	assert.Equal(t, pb.RequestStatus_CANCELLATION_ACCEPTED, resp.CurrentStatus)
 }
 
 func TestCancel_PublishesToQueue(t *testing.T) {
@@ -55,12 +72,17 @@ func TestCancel_PublishesToQueue(t *testing.T) {
 
 	require.NoError(t, err)
 	assert.Equal(t, "my-queue/42", resp.Sqid)
+	assert.Equal(t, pb.RequestStatus_CANCELLATION_ACCEPTED, resp.CurrentStatus)
 
 	// Verify message was published to the cancel topic
 	assert.Equal(t, "cancel", publishedTopic)
 	assert.Equal(t, "my-queue/42", publishedMessage.ID)
 	assert.Equal(t, "my-queue/42", publishedMessage.PartitionKey)
-	assert.Equal(t, []byte("my-queue/42"), publishedMessage.Payload)
+
+	// Verify payload is a serialized Cancel entity
+	cancel, err := entity.CancelFromBytes(publishedMessage.Payload)
+	require.NoError(t, err)
+	assert.Equal(t, "my-queue/42", cancel.Sqid)
 }
 
 func TestCancel_ReturnsErrorOnPublishFailure(t *testing.T) {

gateway/proto/gateway.proto

@@ -79,10 +79,32 @@ message CancelRequest {
     string sqid = 1;
 }
 
+// RequestStatus defines the possible states of a land request.
+enum RequestStatus {
+    // Unknown status.
+    UNKNOWN = 0;
+    // Initial state of a land request. Confirmed by the system but processing has not started.
+    NEW = 1;
+    // The request has been validated (duplicate check, merge check etc.) successfully.
+    VALIDATED = 2;
+    // The request is being processed.
+    PROCESSING = 3;
+    // The request has been successfully processed and landed. Terminal state.
+    LANDED = 4;
+    // The cancellation request has been accepted and is being processed.
+    CANCELLATION_ACCEPTED = 5;
+    // The request has been cancelled. Terminal state.
+    CANCELLED = 6;
+}
+
 // CancelResponse defines the response to a cancel request.
 message CancelResponse {
-    // Globally unique identifier for the land request that was cancelled.
+    // Globally unique identifier for the land request that was requested to be cancelled.
     string sqid = 1;
+    // Current status of the land request at the time of cancellation request.
+    RequestStatus current_status = 2;
+    // Optional error describing why the cancellation could not be completed.
+    Error error = 3;
 }
 
 // ***************
@@ -116,7 +138,11 @@ service SubmitQueueGateway {
     // The processing is asynchronous and returns a LandResponse immediately. The land request is processed in the background.
     rpc Land(LandRequest) returns (LandResponse) {}
 
-    // Cancel cancels a previously submitted land request. The cancellation is asynchronous and
-    // published to a queue for processing. No validation is performed on whether the request exists.
+    // Cancel requests cancellation of a previously submitted land request.
+    // The cancellation is asynchronous and published to a queue for
+    // processing. Note that cancellation is best effort and not guaranteed.
+    // To check for the status of the request, use Status API (TODO). A
+    // "Cancelled" status indicates that the requested cancellation was
+    // successful.
     rpc Cancel(CancelRequest) returns (CancelResponse) {}
 }