287 lines
10 KiB
Ruby
287 lines
10 KiB
Ruby
# Matchmaking via Redis.
|
|
#
|
|
# Data structures (all volatile, TTL-backed):
|
|
# mm:queue sorted set — waiting players, score = rating
|
|
# mm:since hash — user_id => enqueue timestamp (for dynamic rating range)
|
|
# mm:rating hash — user_id => rating (to requeue stale pending players)
|
|
# mm:pending sorted set — paired, awaiting finalize, score = pairing timestamp
|
|
# mm:match:<id> string — match payload (JSON)
|
|
# mm:user:<id> string — per-user pointer to the match payload (JSON)
|
|
#
|
|
# All state transitions happen inside Lua scripts, so every operation is
|
|
# atomic with respect to other Redis clients. Delivery to clients happens
|
|
# over Action Cable; the per-user pointer key guarantees at-most-once
|
|
# delivery even if the broadcast is missed (client reconnects and claims it).
|
|
class MatchmakingService
|
|
QUEUE_KEY = "mm:queue"
|
|
SINCE_KEY = "mm:since"
|
|
RATING_KEY = "mm:rating"
|
|
PENDING_KEY = "mm:pending"
|
|
MATCH_KEY_PREFIX = "mm:match:"
|
|
USER_KEY_PREFIX = "mm:user:"
|
|
|
|
# Enqueues a player and tries to find an opponent immediately; in "sweep"
|
|
# mode pairs up everyone in the queue, expires timed-out waiters and
|
|
# requeues stale pending players. Returns an array of result strings:
|
|
# "P:<id1>:<id2>" — a pair was formed
|
|
# "E:<id>" — a player timed out waiting
|
|
PAIR_SCRIPT = <<~LUA
|
|
local queue_key = KEYS[1]
|
|
local since_key = KEYS[2]
|
|
local rating_key = KEYS[3]
|
|
local pending_key = KEYS[4]
|
|
local user_prefix = KEYS[5]
|
|
|
|
-- All timestamps are integer milliseconds: Lua turns large floats into
|
|
-- scientific notation strings which Redis rejects in ZADD/ZRANGEBYSCORE.
|
|
local mode = ARGV[1]
|
|
local now = tonumber(ARGV[2])
|
|
local queue_ttl = tonumber(ARGV[3]) * 1000
|
|
local pending_ttl = tonumber(ARGV[4]) * 1000
|
|
local base_range = tonumber(ARGV[5])
|
|
local range_per_sec = tonumber(ARGV[6])
|
|
local max_range = tonumber(ARGV[7])
|
|
|
|
local function tolerance(ts)
|
|
local waited = (now - ts) / 1000
|
|
local r = base_range + waited * range_per_sec
|
|
return r > max_range and max_range or r
|
|
end
|
|
|
|
local results = {}
|
|
|
|
if mode == "enqueue" then
|
|
local uid = ARGV[8]
|
|
local rating = tonumber(ARGV[9])
|
|
|
|
-- restart a fresh waiting session for this user
|
|
redis.call("ZREM", queue_key, uid)
|
|
redis.call("HDEL", since_key, uid)
|
|
redis.call("HDEL", rating_key, uid)
|
|
|
|
local candidates = redis.call("ZRANGEBYSCORE", queue_key,
|
|
rating - max_range, rating + max_range)
|
|
local best = nil
|
|
local best_diff = nil
|
|
|
|
for _, other in ipairs(candidates) do
|
|
if other ~= uid then
|
|
local other_ts = tonumber(redis.call("HGET", since_key, other) or now)
|
|
local other_rating = tonumber(redis.call("ZSCORE", queue_key, other))
|
|
local diff = math.abs(other_rating - rating)
|
|
if diff <= tolerance(other_ts) and (best_diff == nil or diff < best_diff) then
|
|
best = other
|
|
best_diff = diff
|
|
end
|
|
end
|
|
end
|
|
|
|
if best then
|
|
redis.call("ZREM", queue_key, uid, best)
|
|
redis.call("HDEL", since_key, uid, best)
|
|
redis.call("HDEL", rating_key, uid, best)
|
|
redis.call("ZADD", pending_key, now, uid, now, best)
|
|
results[#results + 1] = "P:" .. uid .. ":" .. best
|
|
else
|
|
redis.call("ZADD", queue_key, rating, uid)
|
|
redis.call("HSET", since_key, uid, now)
|
|
redis.call("HSET", rating_key, uid, rating)
|
|
end
|
|
else -- "sweep"
|
|
local all = redis.call("ZRANGE", queue_key, 0, -1, "WITHSCORES")
|
|
local items = {}
|
|
for i = 1, #all, 2 do
|
|
items[#items + 1] = {
|
|
id = all[i],
|
|
rating = tonumber(all[i + 1]),
|
|
ts = tonumber(redis.call("HGET", since_key, all[i]) or now)
|
|
}
|
|
end
|
|
table.sort(items, function(a, b) return a.rating < b.rating end)
|
|
|
|
local used = {}
|
|
for i = 1, #items do
|
|
local a = items[i]
|
|
if not used[a.id] then
|
|
if now - a.ts > queue_ttl then
|
|
used[a.id] = true
|
|
results[#results + 1] = "E:" .. a.id
|
|
else
|
|
local best = nil
|
|
for j = i + 1, #items do
|
|
local b = items[j]
|
|
-- tolerance is based on the longer-waiting player of the pair
|
|
if not used[b.id] and (b.rating - a.rating) <= tolerance(math.min(a.ts, b.ts)) then
|
|
best = b
|
|
break
|
|
end
|
|
end
|
|
if best then
|
|
used[a.id] = true
|
|
used[best.id] = true
|
|
results[#results + 1] = "P:" .. a.id .. ":" .. best.id
|
|
end
|
|
end
|
|
end
|
|
end
|
|
|
|
for _, res in ipairs(results) do
|
|
local parts = {}
|
|
for part in string.gmatch(res, "[^:]+") do parts[#parts + 1] = part end
|
|
if parts[1] == "P" then
|
|
redis.call("ZREM", queue_key, parts[2], parts[3])
|
|
redis.call("HDEL", since_key, parts[2], parts[3])
|
|
redis.call("HDEL", rating_key, parts[2], parts[3])
|
|
redis.call("ZADD", pending_key, now, parts[2], now, parts[3])
|
|
else
|
|
redis.call("ZREM", queue_key, parts[2])
|
|
redis.call("HDEL", since_key, parts[2])
|
|
redis.call("HDEL", rating_key, parts[2])
|
|
end
|
|
end
|
|
|
|
-- requeue players whose pairing was never finalized (e.g. the Rails
|
|
-- process died between pairing and match creation)
|
|
local stale = redis.call("ZRANGEBYSCORE", pending_key, 0, now - pending_ttl)
|
|
for _, uid in ipairs(stale) do
|
|
if redis.call("EXISTS", user_prefix .. uid) == 0 then
|
|
local r = redis.call("HGET", rating_key, uid)
|
|
if r then
|
|
redis.call("ZADD", queue_key, tonumber(r), uid)
|
|
redis.call("HSET", since_key, uid, now)
|
|
end
|
|
end
|
|
redis.call("ZREM", pending_key, uid)
|
|
end
|
|
end
|
|
|
|
redis.call("EXPIRE", queue_key, math.floor(queue_ttl / 1000) * 4)
|
|
redis.call("EXPIRE", pending_key, math.floor(queue_ttl / 1000) * 4)
|
|
redis.call("EXPIRE", since_key, math.floor(queue_ttl / 1000) * 4)
|
|
redis.call("EXPIRE", rating_key, math.floor(queue_ttl / 1000) * 4)
|
|
return results
|
|
LUA
|
|
|
|
# Atomically stores the match payload and clears the players' pending state.
|
|
FINALIZE_SCRIPT = <<~LUA
|
|
redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2]) -- match payload
|
|
redis.call("SET", KEYS[2], ARGV[1], "EX", ARGV[2]) -- pointer: player 1
|
|
redis.call("SET", KEYS[3], ARGV[1], "EX", ARGV[2]) -- pointer: player 2
|
|
redis.call("ZREM", KEYS[4], ARGV[3], ARGV[4]) -- pending
|
|
redis.call("HDEL", KEYS[5], ARGV[3], ARGV[4]) -- rating
|
|
redis.call("HDEL", KEYS[6], ARGV[3], ARGV[4]) -- since
|
|
return 1
|
|
LUA
|
|
|
|
CANCEL_SCRIPT = <<~LUA
|
|
redis.call("ZREM", KEYS[1], ARGV[1])
|
|
redis.call("ZREM", KEYS[2], ARGV[1])
|
|
redis.call("HDEL", KEYS[3], ARGV[1])
|
|
redis.call("HDEL", KEYS[4], ARGV[1])
|
|
redis.call("DEL", KEYS[5] .. ARGV[1])
|
|
return 1
|
|
LUA
|
|
|
|
class << self
|
|
# Adds the user to the queue and immediately tries to pair them.
|
|
# The result (match or timeout) is delivered asynchronously via
|
|
# MatchmakingChannel broadcasts.
|
|
def enqueue(user_id, rating:)
|
|
results = call_pair_script("enqueue", extra: [ user_id, rating ])
|
|
process_results(results)
|
|
end
|
|
|
|
# Pairs up everyone possible, expires timed-out waiters, requeues stale
|
|
# pending players. Called periodically by MatchmakingSweepJob.
|
|
def sweep
|
|
results = call_pair_script("sweep")
|
|
process_results(results)
|
|
end
|
|
|
|
# Removes the user from the queue / pending state.
|
|
def cancel(user_id)
|
|
RedisConnection.with do |redis|
|
|
redis.eval(CANCEL_SCRIPT, keys: [ QUEUE_KEY, PENDING_KEY, SINCE_KEY, RATING_KEY, USER_KEY_PREFIX ],
|
|
argv: [ user_id ])
|
|
end
|
|
end
|
|
|
|
# Atomically reads and clears the match pointer for the user.
|
|
# Used on channel subscribe to deliver a match that was found while
|
|
# the user had no active subscription (at-most-once delivery).
|
|
def claim(user_id)
|
|
payload = RedisConnection.with { |redis| redis.getdel(user_key(user_id)) }
|
|
JSON.parse(payload, symbolize_names: true) if payload
|
|
end
|
|
|
|
def config
|
|
Rails.application.config.x.matchmaking
|
|
end
|
|
|
|
private
|
|
|
|
def call_pair_script(mode, extra: [])
|
|
RedisConnection.with do |redis|
|
|
redis.eval(PAIR_SCRIPT,
|
|
keys: [ QUEUE_KEY, SINCE_KEY, RATING_KEY, PENDING_KEY, USER_KEY_PREFIX ],
|
|
argv: [ mode, (Time.current.to_f * 1000).to_i, config.queue_ttl, config.pending_ttl,
|
|
config.base_range, config.range_per_second, config.max_range, *extra ])
|
|
end
|
|
end
|
|
|
|
def process_results(results)
|
|
results.each do |result|
|
|
kind, id1, id2 = result.split(":")
|
|
if kind == "P"
|
|
create_match(id1.to_i, id2.to_i)
|
|
else
|
|
broadcast(id1.to_i, { type: "timeout" })
|
|
end
|
|
end
|
|
end
|
|
|
|
# JWT generation happens in Ruby, so pairing (done in Lua) and match
|
|
# creation are two steps. Players sit in the "pending" set between them;
|
|
# the sweep requeues anyone stuck there longer than pending_ttl.
|
|
def create_match(player1_id, player2_id)
|
|
match_id = SecureRandom.uuid
|
|
game_token = JwtService.encode(
|
|
match_id: match_id,
|
|
player1_id: player1_id,
|
|
player2_id: player2_id,
|
|
game_server: config.game_server_url,
|
|
exp: config.token_ttl.seconds.from_now.to_i
|
|
)
|
|
|
|
payload = {
|
|
type: "matched",
|
|
match_id: match_id,
|
|
game_token: game_token,
|
|
game_server: config.game_server_url,
|
|
players: [ player1_id, player2_id ]
|
|
}.to_json
|
|
|
|
RedisConnection.with do |redis|
|
|
redis.eval(FINALIZE_SCRIPT,
|
|
keys: [ match_key(match_id), user_key(player1_id), user_key(player2_id),
|
|
PENDING_KEY, RATING_KEY, SINCE_KEY ],
|
|
argv: [ payload, config.match_ttl, player1_id, player2_id ])
|
|
end
|
|
|
|
[ player1_id, player2_id ].each do |uid|
|
|
# Deliver and clear the pointer so a later subscribe-claim
|
|
# cannot duplicate the match.
|
|
pointer = RedisConnection.with { |redis| redis.getdel(user_key(uid)) }
|
|
broadcast(uid, JSON.parse(pointer, symbolize_names: true)) if pointer
|
|
end
|
|
end
|
|
|
|
def broadcast(user_id, payload)
|
|
MatchmakingChannel.broadcast_to(user_id, payload)
|
|
end
|
|
|
|
def match_key(match_id) = "#{MATCH_KEY_PREFIX}#{match_id}"
|
|
def user_key(user_id) = "#{USER_KEY_PREFIX}#{user_id}"
|
|
end
|
|
end
|