feat: use redis locks

Author: tituschewxj

apps/matching-service/databases/redis.go

@@ -2,38 +2,67 @@ package databases
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"os"
+	"time"
 
+	"github.com/bsm/redislock"
 	"github.com/redis/go-redis/v9"
 )
 
 const MatchmakingQueueRedisKey = "matchmaking_queue"
+const matchmakingRedisLock = "matchmaking_lock"
 
 var redisClient *redis.Client
+var redisLock *redislock.Client
 
 // SetupRedisClient sets-up the Redis client, and assigns it to a global variable
-func SetupRedisClient() {
+func SetupRedisClient() *redis.Client {
 	// Retrieve redis url env variable and setup the redis client
 	redisAddr := os.Getenv("REDIS_URL")
-	client := redis.NewClient(&redis.Options{
+	redisClient = redis.NewClient(&redis.Options{
 		Addr:     redisAddr,
 		Password: "", // no password set
 		DB:       0,  // use default DB
 	})
 
 	// Ping the redis server
-	_, err := client.Ping(context.Background()).Result()
+	_, err := redisClient.Ping(context.Background()).Result()
 	if err != nil {
 		log.Fatalf("Could not connect to Redis: %v", err)
 	} else {
 		log.Println("Connected to Redis at the following address: " + redisAddr)
 	}
 
-	redisClient = client
+	// Create a new lock client.
+	redisLock = redislock.New(redisClient)
+
+	return redisClient
 }
 
-// Get redisclient
 func GetRedisClient() *redis.Client {
 	return redisClient
 }
+
+func GetRedisLock() *redislock.Client {
+	return redisLock
+}
+
+func ObtainRedisLock(ctx context.Context) (*redislock.Lock, error) {
+	// Retry every 100ms, for up-to 10x
+	backoff := redislock.LimitRetry(redislock.LinearBackoff(100*time.Millisecond), 10)
+
+	// Obtain lock with retry
+	lock, err := redisLock.Obtain(ctx, matchmakingRedisLock, time.Second, &redislock.Options{
+		RetryStrategy: backoff,
+	})
+	if err == redislock.ErrNotObtained {
+		fmt.Println("Could not obtain lock!")
+		return nil, err
+	} else if err != nil {
+		log.Fatalln(err)
+		return nil, err
+	}
+	return lock, err
+}

apps/matching-service/go.mod

@@ -3,6 +3,7 @@ module matching-service
 go 1.23.1
 
 require (
+	github.com/bsm/redislock v0.9.4
 	github.com/gorilla/websocket v1.5.3
 	github.com/joho/godotenv v1.5.1
 	github.com/redis/go-redis/v9 v9.6.2

apps/matching-service/go.sum

@@ -2,6 +2,8 @@ github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/bsm/redislock v0.9.4 h1:X/Wse1DPpiQgHbVYRE9zv6m070UcKoOGekgvpNhiSvw=
+github.com/bsm/redislock v0.9.4/go.mod h1:Epf7AJLiSFwLCiZcfi6pWFO/8eAYrYpQXFxEDPoDeAk=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=

apps/matching-service/handlers/websocket.go

@@ -96,12 +96,18 @@ func cleanUpUser(username string) {
 	redisClient := databases.GetRedisClient()
 	ctx := context.Background()
 
-	err := redisClient.Watch(ctx, func(tx *redis.Tx) error {
+	// Obtain lock with retry
+	lock, err := databases.ObtainRedisLock(ctx)
+	if err != nil {
+		return
+	}
+	defer lock.Release(ctx)
+
+	if err := redisClient.Watch(ctx, func(tx *redis.Tx) error {
 		// Cleanup Redis
 		databases.CleanUpUser(tx, username, ctx)
 		return nil
-	})
-	if err != nil {
+	}); err != nil {
 		return
 	}
 }

apps/matching-service/main.go

@@ -13,7 +13,8 @@ import (
 
 func main() {
 	setUpEnvironment()
-	databases.SetupRedisClient()
+	client := databases.SetupRedisClient()
+	defer client.Close()
 	setupRoutes()
 	startServer()
 }

apps/matching-service/processes/performmatches.go

@@ -18,78 +18,72 @@ import (
 // until a match is found or no match and the user is enqueued to the queue.
 func PerformMatching(rdb *redis.Client, matchRequest models.MatchRequest, ctx context.Context, errorChan chan error) {
 	currentUsername := matchRequest.Username
-	keys := []string{databases.MatchmakingQueueRedisKey, currentUsername}
-	for _, topic := range matchRequest.Topics {
-		keys = append(keys, topic)
+
+	// Obtain lock with retry
+	lock, err := databases.ObtainRedisLock(ctx)
+	if err != nil {
+		return
 	}
-	for true {
+	defer lock.Release(ctx)
 
-		err := rdb.Watch(ctx, func(tx *redis.Tx) error {
-			queuedUsernames, err := databases.GetAllQueuedUsers(tx, ctx)
-			if err != nil {
-				return err
-			}
+	if err := rdb.Watch(ctx, func(tx *redis.Tx) error {
+		queuedUsernames, err := databases.GetAllQueuedUsers(tx, ctx)
+		if err != nil {
+			return err
+		}
 
-			// Check that user is not part of the existing queue
-			for _, username := range queuedUsernames {
-				if username == currentUsername {
-					return models.ExistingUserError
-				}
+		// Check that user is not part of the existing queue
+		for _, username := range queuedUsernames {
+			if username == currentUsername {
+				return models.ExistingUserError
 			}
+		}
 
-			databases.AddUser(tx, matchRequest, ctx)
+		databases.AddUser(tx, matchRequest, ctx)
 
-			// Log queue before and after matchmaking
-			databases.PrintMatchingQueue(tx, "Before Matchmaking", ctx)
-			defer databases.PrintMatchingQueue(tx, "After Matchmaking", ctx)
+		// Log queue before and after matchmaking
+		databases.PrintMatchingQueue(tx, "Before Matchmaking", ctx)
+		defer databases.PrintMatchingQueue(tx, "After Matchmaking", ctx)
+		// Find a matching user if any
+		matchFound, err := databases.FindMatchingUser(tx, currentUsername, ctx)
+		if err != nil {
+			log.Println("Error finding matching user:", err)
+			return err
+		}
 
-			// Find a matching user if any
-			matchFound, err := databases.FindMatchingUser(tx, currentUsername, ctx)
+		if matchFound != nil {
+			matchedUsername := matchFound.MatchedUser
+			matchedTopic := matchFound.Topic
+			matchedDifficulty := matchFound.Difficulty
+
+			// Generate a random match ID
+			matchId, err := utils.GenerateMatchID()
 			if err != nil {
-				log.Println("Error finding matching user:", err)
-				return err
+				log.Println("Unable to randomly generate matchID")
 			}
 
-			if matchFound != nil {
-				matchedUsername := matchFound.MatchedUser
-				matchedTopic := matchFound.Topic
-				matchedDifficulty := matchFound.Difficulty
-
-				// Generate a random match ID
-				matchId, err := utils.GenerateMatchID()
-				if err != nil {
-					log.Println("Unable to randomly generate matchID")
-				}
+			// Log down which users got matched
+			matchFound.MatchID = matchId
+			log.Printf("Users %s and %s matched on the topic: %s with difficulty: %s", currentUsername, matchedUsername, matchedTopic, matchedDifficulty)
 
-				// Log down which users got matched
-				matchFound.MatchID = matchId
-				log.Printf("Users %s and %s matched on the topic: %s with difficulty: %s", currentUsername, matchedUsername, matchedTopic, matchedDifficulty)
+			// Clean up redis for this match
+			databases.CleanUpUser(tx, currentUsername, ctx)
+			databases.CleanUpUser(tx, matchedUsername, ctx)
 
-				// Clean up redis for this match
-				databases.CleanUpUser(tx, currentUsername, ctx)
-				databases.CleanUpUser(tx, matchedUsername, ctx)
-
-				publishMatch(tx, ctx, currentUsername, matchedUsername, matchFound)
-				publishMatch(tx, ctx, matchedUsername, currentUsername, matchFound)
-			}
+			publishMatch(tx, ctx, currentUsername, matchedUsername, matchFound)
+			publishMatch(tx, ctx, matchedUsername, currentUsername, matchFound)
+		}
 
-			time.Sleep(time.Duration(time.Second * 1))
+		time.Sleep(time.Duration(time.Second * 1))
 
-			return nil
-		}, keys...)
-		if err != nil {
-			// return
-			if errors.Is(err, models.ExistingUserError) {
-				errorChan <- err
-				break
-			} else {
-				// transaction failed, so will retry
-				println(fmt.Errorf("transaction execution failed: %v", err))
-			}
-		}
-		if err == nil {
-			// transaction succeeded
-			break
+		return nil
+	}); err != nil {
+		// return
+		if errors.Is(err, models.ExistingUserError) {
+			errorChan <- err
+		} else {
+			// transaction failed, no retry
+			println(fmt.Errorf("transaction execution failed: %v", err))
 		}
 	}
 }