Locks discord events in Redis for replicated shard operation

Author: denverquane

discord/bot.go

@@ -144,6 +144,7 @@ func (bot *Bot) GracefulClose() {
 func (bot *Bot) Close() {
 	bot.SessionManager.Close()
 	bot.RedisInterface.Close()
+	bot.StorageInterface.Close()
 }
 
 func (bot *Bot) PurgeConnection(socketID string) {
@@ -200,15 +201,18 @@ func (bot *Bot) newGuild(emojiGuildID string) func(s *discordgo.Session, m *disc
 			bot.addAllMissingEmojis(s, m.Guild.ID, false, allEmojis)
 		}
 
-		games := bot.RedisInterface.LoadAllActiveGamesAndDelete(m.Guild.ID)
+		games := bot.RedisInterface.LoadAllActiveGames(m.Guild.ID)
 
 		for _, connCode := range games {
 			gsr := GameStateRequest{
 				GuildID:     m.Guild.ID,
 				ConnectCode: connCode,
 			}
 			lock, dgs := bot.RedisInterface.GetDiscordGameStateAndLock(gsr)
-			if lock != nil && dgs != nil && !dgs.Subscribed && dgs.ConnectCode != "" {
+			for lock == nil {
+				lock, dgs = bot.RedisInterface.GetDiscordGameStateAndLock(gsr)
+			}
+			if dgs != nil && dgs.ConnectCode != "" {
 				log.Println("Resubscribing to Redis events for an old game: " + connCode)
 				killChan := make(chan EndGameMessage)
 				go bot.SubscribeToGameByConnectCode(gsr.GuildID, dgs.ConnectCode, killChan)
@@ -219,19 +223,9 @@ func (bot *Bot) newGuild(emojiGuildID string) func(s *discordgo.Session, m *disc
 				bot.ChannelsMapLock.Lock()
 				bot.EndGameChannels[dgs.ConnectCode] = killChan
 				bot.ChannelsMapLock.Unlock()
-			} else if lock != nil {
-				//log.Println("UNLOCKING")
-				lock.Release(ctx)
 			}
+			lock.Release(ctx)
 		}
-
-		//if len(games) == 0 {
-		//	dsg := NewDiscordGameState(m.Guild.ID)
-		//
-		//	//put an empty entry in Redis
-		//	bot.RedisInterface.SetDiscordGameState(dsg, nil)
-		//}
-
 	}
 }
 

discord/eventHandler.go

@@ -58,6 +58,7 @@ func (bot *Bot) SubscribeToGameByConnectCode(guildID, connectCode string, endGam
 				}
 				log.Println("Popped job w/ payload " + job.Payload.(string))
 				bot.refreshGameLiveness(connectCode)
+				bot.RedisInterface.RefreshActiveGame(guildID, connectCode)
 
 				switch job.JobType {
 				case broker.Connection:

discord/message_handlers.go

@@ -23,6 +23,13 @@ func (bot *Bot) handleMessageCreate(s *discordgo.Session, m *discordgo.MessageCr
 		return
 	}
 
+	lock := bot.RedisInterface.LockSnowflake(m.ID)
+	//couldn't obtain lock; bail bail bail!
+	if lock == nil {
+		return
+	}
+	defer lock.Release(ctx)
+
 	g, err := s.State.Guild(m.GuildID)
 	if err != nil {
 		log.Println(err)
@@ -89,6 +96,12 @@ func (bot *Bot) handleReactionGameStartAdd(s *discordgo.Session, m *discordgo.Me
 	if m.UserID == s.State.User.ID {
 		return
 	}
+	lock := bot.RedisInterface.LockSnowflake(m.ChannelID + m.UserID + m.MessageID)
+	//couldn't obtain lock; bail bail bail!
+	if lock == nil {
+		return
+	}
+	defer lock.Release(ctx)
 
 	g, err := s.State.Guild(m.GuildID)
 	if err != nil {
@@ -157,6 +170,13 @@ func (bot *Bot) handleReactionGameStartAdd(s *discordgo.Session, m *discordgo.Me
 //relevant discord api requests are fully applied successfully. Otherwise, we can issue multiple requests for
 //the same mute/unmute, erroneously
 func (bot *Bot) handleVoiceStateChange(s *discordgo.Session, m *discordgo.VoiceStateUpdate) {
+	lock := bot.RedisInterface.LockSnowflake(m.ChannelID + m.UserID + m.SessionID)
+	//couldn't obtain lock; bail bail bail!
+	if lock == nil {
+		return
+	}
+	defer lock.Release(ctx)
+
 	sett := bot.StorageInterface.GetGuildSettings(m.GuildID)
 	gsr := GameStateRequest{
 		GuildID:      m.GuildID,
@@ -241,7 +261,7 @@ func (bot *Bot) handleNewGameMessage(s *discordgo.Session, m *discordgo.MessageC
 
 	dgs.ConnectCode = connectCode
 
-	bot.RedisInterface.AppendToActiveGames(m.GuildID, connectCode)
+	bot.RedisInterface.RefreshActiveGame(m.GuildID, connectCode)
 
 	killChan := make(chan EndGameMessage)
 

discord/redis.go

@@ -16,9 +16,13 @@ var ctx = context.Background()
 const LockTimeoutSecs = 3
 const LinearBackoffMs = 200
 const MaxRetries = 10
+const SnowflakeLockMs = 100
 
 const SecsPerHour = 3600
 
+//10 minute threshold for games to not be loaded up
+const StaleGameThresholdSec = 600
+
 type RedisInterface struct {
 	client *redis.Client
 }
@@ -87,6 +91,10 @@ func matchIDKey() string {
 	return "automuteus:match:counter"
 }
 
+func snowflakeLockID(snowflake string) string {
+	return "automuteus:snowflake:" + snowflake + ":lock"
+}
+
 func (redisInterface *RedisInterface) GetAndIncrementMatchID() int64 {
 	num, err := redisInterface.client.Incr(ctx, matchIDKey()).Result()
 	if err != nil {
@@ -297,15 +305,16 @@ func (redisInterface *RedisInterface) SetDiscordGameState(data *DiscordGameState
 	}
 }
 
-func (redisInterface *RedisInterface) AppendToActiveGames(guildID, connectCode string) {
+func (redisInterface *RedisInterface) RefreshActiveGame(guildID, connectCode string) {
 	key := activeGamesKey(guildID)
-
-	count, err := redisInterface.client.SAdd(ctx, key, connectCode).Result()
+	t := time.Now().Unix()
+	_, err := redisInterface.client.ZAdd(ctx, key, &redis.Z{
+		Score:  float64(t),
+		Member: connectCode,
+	}).Result()
 
 	if err != nil {
 		log.Println(err)
-	} else {
-		log.Printf("Active games: %d", count)
 	}
 }
 
@@ -319,19 +328,23 @@ func (redisInterface *RedisInterface) RemoveOldGame(guildID, connectCode string)
 }
 
 //only deletes from the guild's responsibility, NOT the entire guild counter!
-func (redisInterface *RedisInterface) LoadAllActiveGamesAndDelete(guildID string) []string {
+func (redisInterface *RedisInterface) LoadAllActiveGames(guildID string) []string {
 	hash := activeGamesKey(guildID)
 
-	games, err := redisInterface.client.SMembers(ctx, hash).Result()
+	before := time.Now().Add(-time.Second * StaleGameThresholdSec).Unix()
+
+	games, err := redisInterface.client.ZRangeByScore(ctx, hash, &redis.ZRangeBy{
+		Min:    fmt.Sprintf("%d", before),
+		Max:    fmt.Sprintf("%d", time.Now().Unix()),
+		Offset: 0,
+		Count:  0,
+	}).Result()
+
 	if err != nil {
 		log.Println(err)
 		return []string{}
 	}
 
-	_, err = redisInterface.client.Del(ctx, hash).Result()
-	if err != nil {
-		log.Println(err)
-	}
 	return games
 }
 
@@ -449,6 +462,18 @@ func (redisInterface *RedisInterface) setUsernameOrUserIDMappings(guildID, key s
 	return redisInterface.client.HSet(ctx, cacheHash, key, jBytes).Err()
 }
 
+func (redisInterface *RedisInterface) LockSnowflake(snowflake string) *redislock.Lock {
+	locker := redislock.New(redisInterface.client)
+	lock, err := locker.Obtain(ctx, snowflakeLockID(snowflake), time.Millisecond*SnowflakeLockMs, nil)
+	if err == redislock.ErrNotObtained {
+		return nil
+	} else if err != nil {
+		log.Println(err)
+		return nil
+	}
+	return lock
+}
+
 func (redisInterface *RedisInterface) Close() error {
 	return redisInterface.client.Close()
 }

main.go

@@ -170,11 +170,10 @@ func discordMainWrapper() error {
 	bot := discord.MakeAndStartBot(version, commit, discordToken, discordToken2, url, emojiGuildID, numShards, shardID, &redisClient, &storageInterface, logPath, captureTimeout)
 
 	<-sc
-	bot.GracefulClose()
+	//bot.GracefulClose()
 	log.Printf("Received Sigterm or Kill signal. Bot will terminate in 1 second")
 	time.Sleep(time.Second)
 
 	bot.Close()
-	redisClient.Close()
 	return nil
 }