Merge pull request #3 from EmbarkStudios/do/batch-ticket-deletion

feat: batch ticket deletion api for open-match

Author: odouglas-es

api/frontend.proto

@@ -123,14 +123,18 @@ message GetBackfillRequest {
 }
 
 // UpdateBackfillRequest - update searchFields, extensions and set assignment.
-// 
+//
 // BETA FEATURE WARNING: This Request message is not finalized and still subject
 // to possible change or removal.
 message UpdateBackfillRequest {
   // A Backfill object with ID set and fields to update.
   Backfill backfill = 1;
 }
 
+message DeleteTicketsRequest {
+  // TicketIds of generated Tickets to be deleted.
+  repeated string ticket_ids = 1;
+}
 
 // The FrontendService implements APIs to manage and query status of a Tickets.
 service FrontendService {
@@ -146,7 +150,7 @@ service FrontendService {
   }
 
   // DeleteTicket immediately stops Open Match from using the Ticket for matchmaking and removes the Ticket from state storage.
-  // The client should delete the Ticket when finished matchmaking with it. 
+  // The client should delete the Ticket when finished matchmaking with it.
   rpc DeleteTicket(DeleteTicketRequest) returns (google.protobuf.Empty) {
     option (google.api.http) = {
       delete: "/v1/frontendservice/tickets/{ticket_id}"
@@ -161,7 +165,7 @@ service FrontendService {
   }
 
   // WatchAssignments stream back Assignment of the specified TicketId if it is updated.
-  //   - If the Assignment is not updated, GetAssignment will retry using the configured backoff strategy. 
+  //   - If the Assignment is not updated, GetAssignment will retry using the configured backoff strategy.
   rpc WatchAssignments(WatchAssignmentsRequest)
       returns (stream WatchAssignmentsResponse) {
     option (google.api.http) = {
@@ -208,7 +212,7 @@ service FrontendService {
       get: "/v1/frontendservice/backfills/{backfill_id}"
     };
   }
-  
+
   // UpdateBackfill updates search_fields and extensions for the backfill with the provided id.
   // Any tickets waiting for this backfill will be returned to the active pool, no longer pending.
   // BETA FEATURE WARNING:  This call and the associated Request and Response
@@ -228,4 +232,12 @@ service FrontendService {
       get: "/v1/frontendservice/backfills/{backfill_id}/tickets"
     };
   }
+
+  // DeleteTickets immediately stops Open Match from using the Tickets for matchmaking and removes the Tickets from state storage.
+  // The client should delete the Tickets when finished matchmaking with them.
+  rpc DeleteTickets(DeleteTicketsRequest) returns (google.protobuf.Empty) {
+    option (google.api.http) = {
+      delete: "/v1/frontendservice/tickets"
+    };
+  }
 }

api/frontend.swagger.json

@@ -273,6 +273,48 @@
       }
     },
     "/v1/frontendservice/tickets": {
+      "delete": {
+        "summary": "DeleteTickets immediately stops Open Match from using the Tickets for matchmaking and removes the Tickets from state storage.\nThe client should delete the Tickets when finished matchmaking with them.",
+        "operationId": "FrontendService_DeleteTickets",
+        "responses": {
+          "200": {
+            "description": "A successful response.",
+            "schema": {
+              "type": "object",
+              "properties": {}
+            }
+          },
+          "404": {
+            "description": "Returned when the resource does not exist.",
+            "schema": {
+              "type": "string",
+              "format": "string"
+            }
+          },
+          "default": {
+            "description": "An unexpected error response.",
+            "schema": {
+              "$ref": "#/definitions/rpcStatus"
+            }
+          }
+        },
+        "parameters": [
+          {
+            "name": "ticket_ids",
+            "description": "TicketIds of generated Tickets to be deleted.",
+            "in": "query",
+            "required": false,
+            "type": "array",
+            "items": {
+              "type": "string"
+            },
+            "collectionFormat": "multi"
+          }
+        ],
+        "tags": [
+          "FrontendService"
+        ]
+      },
       "post": {
         "summary": "CreateTicket assigns an unique TicketId to the input Ticket and record it in state storage.\nA ticket is considered as ready for matchmaking once it is created.\n  - If a TicketId exists in a Ticket request, an auto-generated TicketId will override this field.\n  - If SearchFields exist in a Ticket, CreateTicket will also index these fields such that one can query the ticket with query.QueryTickets function.",
         "operationId": "FrontendService_CreateTicket",

internal/app/frontend/frontend_service.go

@@ -427,3 +427,48 @@ func (s *frontendService) GetBackfillTickets(ctx context.Context, req *pb.GetBac
 
 	return resp, nil
 }
+
+// DeleteTickets immediately stops Open Match from using the Tickets for matchmaking and removes the Tickets from state storage.
+// The client must delete the Tickets when finished matchmaking with it.
+//   - If SearchFields exist in a Ticket, DeleteTickets will deindex the fields lazily.
+//
+// Users may still be able to assign/get tickets after calling DeleteTickets on it.
+func (s *frontendService) DeleteTickets(ctx context.Context, req *pb.DeleteTicketsRequest) (*emptypb.Empty, error) {
+	err := doDeleteTickets(ctx, req.GetTicketIds(), s.store)
+	if err != nil {
+		return nil, err
+	}
+	return &emptypb.Empty{}, nil
+}
+
+func doDeleteTickets(ctx context.Context, ids []string, store statestore.Service) error {
+	// Deindex these Tickets to remove it from matchmaking pool.
+	err := store.DeindexTickets(ctx, ids)
+	if err != nil {
+		return err
+	}
+
+	//'lazy' tickets delete that should be called after one or more tickets
+	// have been deindexed.
+	go func() {
+		ctx, span := trace.StartSpan(context.Background(), "open-match/frontend.DeleteTicketsLazy")
+		defer span.End()
+		err := store.DeleteTickets(ctx, ids)
+		if err != nil {
+			logger.WithFields(logrus.Fields{
+				"error": err.Error(),
+				"ids":   ids,
+			}).Error("failed to delete tickets")
+		}
+		err = store.DeleteTicketsFromPendingRelease(ctx, ids)
+		if err != nil {
+			logger.WithFields(logrus.Fields{
+				"error": err.Error(),
+				"ids":   ids,
+			}).Error("failed to delete tickets from pendingRelease")
+		}
+		// TODO: If other redis queues are implemented or we have custom index fields
+		// created by Open Match, those need to be cleaned up here.
+	}()
+	return nil
+}

internal/app/frontend/frontend_service_test.go

@@ -461,8 +461,8 @@ func TestDoDeleteTicket(t *testing.T) {
 		{
 			description: "expect ok code",
 			preAction: func(ctx context.Context, _ context.CancelFunc, store statestore.Service) {
-				store.CreateTicket(ctx, fakeTicket)
-				store.IndexTicket(ctx, fakeTicket)
+				_ = store.CreateTicket(ctx, fakeTicket)
+				_ = store.IndexTicket(ctx, fakeTicket)
 			},
 		},
 	}
@@ -482,6 +482,81 @@ func TestDoDeleteTicket(t *testing.T) {
 	}
 }
 
+func TestDoDeleteTickets(t *testing.T) {
+	fakeTickets := []*pb.Ticket{
+		{
+			Id: "1",
+			SearchFields: &pb.SearchFields{
+				DoubleArgs: map[string]float64{
+					"test-arg": 1,
+				},
+			},
+		},
+		{
+			Id: "2",
+			SearchFields: &pb.SearchFields{
+				DoubleArgs: map[string]float64{
+					"test-arg": 2,
+				},
+			},
+		},
+		{
+			Id: "3",
+			SearchFields: &pb.SearchFields{
+				DoubleArgs: map[string]float64{
+					"test-arg": 3,
+				},
+			},
+		},
+	}
+
+	tests := []struct {
+		description string
+		preAction   func(context.Context, context.CancelFunc, statestore.Service)
+		wantCode    codes.Code
+	}{
+		{
+			description: "expect unavailable code since context is canceled before being called",
+			preAction: func(_ context.Context, cancel context.CancelFunc, _ statestore.Service) {
+				cancel()
+			},
+			wantCode: codes.Unavailable,
+		},
+		{
+			description: "expect ok code since delete tickets does not care about if ticket exists or not",
+			preAction:   func(_ context.Context, _ context.CancelFunc, _ statestore.Service) {},
+			wantCode:    codes.OK,
+		},
+		{
+			description: "expect ok code",
+			preAction: func(ctx context.Context, _ context.CancelFunc, store statestore.Service) {
+				for _, ticket := range fakeTickets {
+					_ = store.CreateTicket(ctx, ticket)
+					_ = store.IndexTicket(ctx, ticket)
+				}
+			},
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.description, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(utilTesting.NewContext(t))
+			store, closer := statestoreTesting.NewStoreServiceForTesting(t, viper.New())
+			defer closer()
+
+			test.preAction(ctx, cancel, store)
+			ids := make([]string, len(fakeTickets))
+			for i := 0; i < len(fakeTickets); i++ {
+				ids[i] = fakeTickets[i].Id
+			}
+
+			err := doDeleteTickets(ctx, ids, store)
+			require.Equal(t, test.wantCode.String(), status.Convert(err).Code().String())
+		})
+	}
+}
+
 func TestDoGetTicket(t *testing.T) {
 	fakeTicket := &pb.Ticket{
 		Id: "1",

internal/statestore/instrumented.go

@@ -53,6 +53,12 @@ func (is *instrumentedService) DeleteTicket(ctx context.Context, id string) erro
 	return is.s.DeleteTicket(ctx, id)
 }
 
+func (is *instrumentedService) DeleteTickets(ctx context.Context, ids []string) error {
+	ctx, span := trace.StartSpan(ctx, "statestore/instrumented.DeleteTickets")
+	defer span.End()
+	return is.s.DeleteTickets(ctx, ids)
+}
+
 func (is *instrumentedService) IndexTicket(ctx context.Context, ticket *pb.Ticket) error {
 	ctx, span := trace.StartSpan(ctx, "statestore/instrumented.IndexTicket")
 	defer span.End()
@@ -65,6 +71,12 @@ func (is *instrumentedService) DeindexTicket(ctx context.Context, id string) err
 	return is.s.DeindexTicket(ctx, id)
 }
 
+func (is *instrumentedService) DeindexTickets(ctx context.Context, ids []string) error {
+	ctx, span := trace.StartSpan(ctx, "statestore/instrumented.DeindexTickets")
+	defer span.End()
+	return is.s.DeindexTickets(ctx, ids)
+}
+
 func (is *instrumentedService) GetTickets(ctx context.Context, ids []string) ([]*pb.Ticket, error) {
 	ctx, span := trace.StartSpan(ctx, "statestore/instrumented.GetTickets")
 	defer span.End()

internal/statestore/public.go

@@ -43,12 +43,19 @@ type Service interface {
 	// This method succeeds if the Ticket does not exist.
 	DeleteTicket(ctx context.Context, id string) error
 
+	// DeleteTickets removes the Tickets with the specified id from state storage.
+	// This method succeeds if any of the Tickets do not exist.
+	DeleteTickets(ctx context.Context, ids []string) error
+
 	// IndexTicket adds the ticket to the index.
 	IndexTicket(ctx context.Context, ticket *pb.Ticket) error
 
 	// DeindexTicket removes specified ticket from the index. The Ticket continues to exist.
 	DeindexTicket(ctx context.Context, id string) error
 
+	// DeindexTickets removes specified tickets from the index. The Tickets continue to exist.
+	DeindexTickets(ctx context.Context, ids []string) error
+
 	// GetIndexedIDSet returns the ids of all tickets currently indexed.
 	GetIndexedIDSet(ctx context.Context) (map[string]struct{}, error)
 

internal/statestore/ticket.go

@@ -118,6 +118,32 @@ func (rb *redisBackend) DeleteTicket(ctx context.Context, id string) error {
 	return nil
 }
 
+// DeleteTickets removes the Tickets with the specified id from state storage.
+func (rb *redisBackend) DeleteTickets(ctx context.Context, ids []string) error {
+	redisConn, err := rb.redisPool.GetContext(ctx)
+	if err != nil {
+		return status.Errorf(codes.Unavailable, "DeleteTickets, id: %v, failed to connect to redis: %v", ids, err)
+	}
+	defer handleConnectionClose(&redisConn)
+
+	args := make([]any, len(ids))
+	for i, id := range ids {
+		args[i] = id
+	}
+
+	value, err := redis.Int(redisConn.Do("DEL", args...))
+	if err != nil {
+		err = errors.Wrapf(err, "failed to delete tickets from state storage, ids: %v", ids)
+		return status.Errorf(codes.Internal, "%v", err)
+	}
+
+	if value == 0 {
+		return status.Errorf(codes.NotFound, "Ticket ids: %s not found", ids)
+	}
+
+	return nil
+}
+
 // IndexTicket indexes the Ticket id for the configured index fields.
 func (rb *redisBackend) IndexTicket(ctx context.Context, ticket *pb.Ticket) error {
 	redisConn, err := rb.redisPool.GetContext(ctx)
@@ -152,6 +178,29 @@ func (rb *redisBackend) DeindexTicket(ctx context.Context, id string) error {
 	return nil
 }
 
+// DeindexTickets removes the indexing for the specified Tickets. Only the indexes are removed but Tickets continues to exist.
+func (rb *redisBackend) DeindexTickets(ctx context.Context, ids []string) error {
+	redisConn, err := rb.redisPool.GetContext(ctx)
+	if err != nil {
+		return status.Errorf(codes.Unavailable, "DeindexTickets, id: %v, failed to connect to redis: %v", ids, err)
+	}
+	defer handleConnectionClose(&redisConn)
+
+	args := make([]any, len(ids)+1)
+	args[0] = allTickets
+	for i, id := range ids {
+		args[i+1] = id
+	}
+
+	err = redisConn.Send("SREM", args...)
+	if err != nil {
+		err = errors.Wrapf(err, "failed to remove ticket from all tickets, id: %v", ids)
+		return status.Errorf(codes.Internal, "%v", err)
+	}
+
+	return nil
+}
+
 // GetIndexedIds returns the ids of all tickets currently indexed.
 func (rb *redisBackend) GetIndexedIDSet(ctx context.Context) (map[string]struct{}, error) {
 	redisConn, err := rb.redisPool.GetContext(ctx)