From acdaebd491df747fbd8ee3b20f260b68e60a6801 Mon Sep 17 00:00:00 2001 From: Slavasil Date: Sun, 25 Oct 2026 15:57:24 +0300 Subject: [PATCH] implement alternative match-making approach with Lua scripts in Redis --- main-server/API.md | 66 ++-- .../app/channels/application_cable/channel.rb | 4 + .../channels/application_cable/connection.rb | 21 ++ .../app/channels/matchmaking_channel.rb | 31 ++ .../controllers/api/matchmaking_controller.rb | 17 +- main-server/app/jobs/matchmaking_sweep_job.rb | 10 + .../app/services/matchmaking_service.rb | 325 +++++++++++++----- main-server/config/cable.yml | 10 + main-server/config/database.yml | 45 ++- .../config/environments/development.rb | 5 + main-server/config/environments/production.rb | 3 +- .../config/initializers/matchmaking.rb | 25 ++ main-server/config/initializers/redis.rb | 6 +- main-server/config/puma.rb | 2 +- main-server/config/queue.yml | 18 + main-server/config/recurring.yml | 18 + main-server/config/routes.rb | 6 +- main-server/db/queue_schema.rb | 160 +++++++++ .../spec/services/matchmaking_service_spec.rb | 163 +++++++++ 19 files changed, 796 insertions(+), 139 deletions(-) create mode 100644 main-server/app/channels/application_cable/channel.rb create mode 100644 main-server/app/channels/application_cable/connection.rb create mode 100644 main-server/app/channels/matchmaking_channel.rb create mode 100644 main-server/app/jobs/matchmaking_sweep_job.rb create mode 100644 main-server/config/cable.yml create mode 100644 main-server/config/initializers/matchmaking.rb create mode 100644 main-server/config/queue.yml create mode 100644 main-server/config/recurring.yml create mode 100644 main-server/db/queue_schema.rb create mode 100644 main-server/spec/services/matchmaking_service_spec.rb diff --git a/main-server/API.md b/main-server/API.md index 7721f90..756eb12 100644 --- a/main-server/API.md +++ b/main-server/API.md @@ -207,31 +207,37 @@ curl -X DELETE http://localhost:3000/api/logout \ --- -### Поиск случайного соперника (Матчмейкинг) +### Поиск соперника (Матчмейкинг) -Начинает поиск соперника. Работает в режиме **long-polling**: запрос зависает до 30 секунд, ожидая второго игрока. +Поиск соперника происходит через **WebSocket** (`Action Cable`), а не через long-polling: HTTP-поток не блокируется, а результат доставляется сервером push-уведомлением, как только найден соперник. -**URL:** `POST /api/matchmaking` +**URL подключения:** `ws://localhost:3000/cable?token=` -**Auth:** требуется +**Auth:** токен передаётся параметром `token` в query string (тот же JWT, что и для HTTP API). -**Параметры:** нет +**Канал:** `MatchmakingChannel` -**Пример запроса:** +#### Поток взаимодействия -```bash -curl -X POST http://localhost:3000/api/matchmaking \ - -H "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9..." \ - --max-time 35 +1. Подключиться к `/cable?token=...` и подписаться на канал `MatchmakingChannel`. +2. Отправить команду начала поиска: + +```json +{ "action": "start", "rating": 1000 } ``` -**Успешный ответ (200 OK)** — когда найден соперник: +Поле `rating` опционально (по умолчанию — `MATCHMAKING_DEFAULT_RATING`). Игроки подбираются по близости рейтинга; допуск расширяется с ростом времени ожидания. + +3. Дождаться одного из сообщений: + +**Матч найден:** ```json { + "type": "matched", "match_id": "145575d6-3a4b-4c5d-9e6f-7a8b9c0d1e2f", "game_token": "eyJhbGciOiJIUzI1NiJ9.eyJtYXRjaF9pZCI6IjE...", - "game_server": "ws://game-server.example.com:8080", + "game_server": "ws://localhost:8080", "players": [1, 2] } ``` @@ -243,23 +249,19 @@ curl -X POST http://localhost:3000/api/matchmaking \ | `game_server` | WebSocket-адрес игрового сервера | | `players` | ID обоих игроков `[player1_id, player2_id]` | -**Таймаут (408 Request Timeout)** — если соперник не найден за 30 секунд: +**Таймаут поиска** (соперник не найден за `MATCHMAKING_QUEUE_TTL` секунд): ```json -{ - "error": "No opponent found. Try again." -} +{ "type": "timeout" } ``` -**Ошибка (401 Unauthorized):** +4. Поиск отменяется автоматически при закрытии WebSocket-подписки. Также можно явно отправить: ```json -{ - "error": "Unauthorized" -} +{ "action": "stop" } ``` -**Примечание:** `game_token` содержит `match_id`, `player1_id`, `player2_id`, `game_server` и действителен **1 час**. +**Примечание:** `game_token` содержит `match_id`, `player1_id`, `player2_id`, `game_server` и действителен `MATCHMAKING_TOKEN_TTL` секунд. --- @@ -311,7 +313,6 @@ curl -X DELETE http://localhost:3000/api/matchmaking \ | `200` | Успешный запрос | | `201` | Ресурс создан (регистрация) | | `401` | Неавторизован (отсутствует/невалиден токен) | -| `408` | Таймаут матчмейкинга (соперник не найден) | | `422` | Ошибка валидации | | `500` | Внутренняя ошибка сервера | @@ -321,8 +322,17 @@ curl -X DELETE http://localhost:3000/api/matchmaking \ | Переменная | По умолчанию | Описание | |------------|--------------|----------| -| `REDIS_URL` | `redis://localhost:6379/0` | Подключение к Redis | -| `GAME_SERVER_URL` | `ws://game-server.example.com:8080` | Адрес игрового сервера для матчмейкинга | +| `REDIS_URL` | `redis://localhost:6380/0` | Подключение к Redis (матчмейкинг и pub/sub Action Cable) | +| `GAME_SERVER_URL` | `ws://localhost:8080` | Адрес игрового сервера для матчмейкинга | +| `MATCHMAKING_DEFAULT_RATING` | `1000` | Рейтинг по умолчанию для новых игроков | +| `MATCHMAKING_QUEUE_TTL` | `60` | Время ожидания соперника до таймаута, сек | +| `MATCHMAKING_PENDING_TTL` | `15` | Время ожидания финализации пары, сек | +| `MATCHMAKING_MATCH_TTL` | `120` | Время жизни данных матча, сек | +| `MATCHMAKING_TOKEN_TTL` | `120` | Время жизни `game_token`, сек | +| `MATCHMAKING_BASE_RANGE` | `100` | Начальный допуск по рейтингу | +| `MATCHMAKING_RANGE_PER_SECOND` | `50` | Рост допуска за секунду ожидания | +| `MATCHMAKING_MAX_RANGE` | `1000` | Максимальный допуск по рейтингу | +| `MATCHMAKING_SWEEP_INTERVAL` | `2` | Периодичность фоновой сверки очереди, сек | | `CORS_ORIGINS` | `*` | Разрешённые origins для CORS (через запятую) | | `SECRET_KEY_BASE` | из `credentials.yml.enc` | Секретный ключ для JWT | @@ -341,8 +351,8 @@ TOKEN=$(curl -s -X POST http://localhost:3000/api/register \ curl -s http://localhost:3000/api/me \ -H "Authorization: Bearer $TOKEN" -# 3. Поиск матча (long-polling) -curl -s -X POST http://localhost:3000/api/matchmaking \ - -H "Authorization: Bearer $TOKEN" \ - --max-time 35 | jq . +# 3. Поиск матча — WebSocket (например, через wscat): +# wscat -c "ws://localhost:3000/cable?token=$TOKEN" +# > {"command":"subscribe","identifier":"{\"channel\":\"MatchmakingChannel\"}"} +# > {"command":"message","identifier":"{\"channel\":\"MatchmakingChannel\"}","data":"{\"action\":\"start\",\"rating\":1000}"} ``` diff --git a/main-server/app/channels/application_cable/channel.rb b/main-server/app/channels/application_cable/channel.rb new file mode 100644 index 0000000..d672697 --- /dev/null +++ b/main-server/app/channels/application_cable/channel.rb @@ -0,0 +1,4 @@ +module ApplicationCable + class Channel < ActionCable::Channel::Base + end +end diff --git a/main-server/app/channels/application_cable/connection.rb b/main-server/app/channels/application_cable/connection.rb new file mode 100644 index 0000000..51a8ee1 --- /dev/null +++ b/main-server/app/channels/application_cable/connection.rb @@ -0,0 +1,21 @@ +module ApplicationCable + class Connection < ActionCable::Connection::Base + identified_by :current_user + + def connect + self.current_user = find_user + reject_unauthorized_connection unless current_user + end + + private + + def find_user + token = request.params[:token].presence || + request.headers["Authorization"]&.split(" ")&.last + return unless token + + decoded = JwtService.decode(token) + User.find_by(id: decoded[:user_id]) if decoded + end + end +end diff --git a/main-server/app/channels/matchmaking_channel.rb b/main-server/app/channels/matchmaking_channel.rb new file mode 100644 index 0000000..d04ddb7 --- /dev/null +++ b/main-server/app/channels/matchmaking_channel.rb @@ -0,0 +1,31 @@ +# Real-time matchmaking channel. +# +# Client flow: +# 1. Connect: ws(s)://host/cable?token= +# 2. Subscribe to "MatchmakingChannel" +# 3. Send { action: "start", rating: } to enter the queue +# 4. Receive { type: "matched", ... } or { type: "timeout" } +# 5. Disconnecting (or sending { action: "stop" }) cancels the search. +class MatchmakingChannel < ApplicationCable::Channel + def subscribed + stream_for current_user + + # If a match was found while this user had no subscription + # (e.g. brief reconnect), deliver it now — exactly once. + if match = MatchmakingService.claim(current_user.id) + transmit(match) + end + end + + def start(data) + MatchmakingService.enqueue(current_user.id, rating: data["rating"]) + end + + def stop + MatchmakingService.cancel(current_user.id) + end + + def unsubscribed + MatchmakingService.cancel(current_user.id) + end +end diff --git a/main-server/app/controllers/api/matchmaking_controller.rb b/main-server/app/controllers/api/matchmaking_controller.rb index 40cf4de..e43a7b6 100644 --- a/main-server/app/controllers/api/matchmaking_controller.rb +++ b/main-server/app/controllers/api/matchmaking_controller.rb @@ -1,19 +1,6 @@ class Api::MatchmakingController < ApplicationController - def create - result = MatchmakingService.find_match(current_user.id) - - if result - render json: { - match_id: result[:match_id], - game_token: result[:game_token], - game_server: result[:game_server], - players: result[:players] - } - else - render json: { error: "No opponent found. Try again." }, status: :request_timeout - end - end - + # Explicit cancellation. Normally the search is cancelled automatically + # when the MatchmakingChannel subscription is closed. def destroy MatchmakingService.cancel(current_user.id) render json: { message: "Matchmaking cancelled" } diff --git a/main-server/app/jobs/matchmaking_sweep_job.rb b/main-server/app/jobs/matchmaking_sweep_job.rb new file mode 100644 index 0000000..3f8dc82 --- /dev/null +++ b/main-server/app/jobs/matchmaking_sweep_job.rb @@ -0,0 +1,10 @@ +# Periodically re-checks the matchmaking queue: pairs players whose rating +# tolerance has widened enough, expires timed-out waiters and requeues +# players stuck in the pending state. Scheduled via config/recurring.yml. +class MatchmakingSweepJob < ApplicationJob + queue_as :default + + def perform + MatchmakingService.sweep + end +end diff --git a/main-server/app/services/matchmaking_service.rb b/main-server/app/services/matchmaking_service.rb index cbc24a2..521414a 100644 --- a/main-server/app/services/matchmaking_service.rb +++ b/main-server/app/services/matchmaking_service.rb @@ -1,122 +1,287 @@ +# 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: string — match payload (JSON) +# mm:user: 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 = "matchmaking:queue" - LOCK_KEY = "matchmaking:lock" - MATCH_PREFIX = "matchmaking:match:" - MATCH_TTL = 60 # seconds - POLL_INTERVAL = 0.5 # seconds - MAX_WAIT = 30 # seconds + 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:" - # Game server to redirect players to (hardcoded for now) - GAME_SERVER = ENV.fetch("GAME_SERVER_URL", "ws://game-server.example.com:8080") + # 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::" — a pair was formed + # "E:" — 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] - class << self - # Main entry point: waits for an opponent and returns match data - def find_match(user_id) - user_id = user_id.to_i + -- 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]) - # Try to pair with an existing waiting player - opponent_id = pop_opponent(user_id) + 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 - if opponent_id - create_match(user_id, opponent_id) + 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 - # Add self to queue and wait - add_to_queue(user_id) - wait_for_match(user_id) + 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 - # Cancel matchmaking (remove from queue) + 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: nil) + rating ||= config.default_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.lrem(QUEUE_KEY, 0, user_id) + 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 - # Atomically pop an opponent from the queue (excluding self) - def pop_opponent(user_id) + def call_pair_script(mode, extra: []) RedisConnection.with do |redis| - redis.watch(QUEUE_KEY) do - queue = redis.lrange(QUEUE_KEY, 0, -1) - opponent = queue.find { |id| id.to_i != user_id } - - if opponent - result = redis.multi do |tx| - tx.lrem(QUEUE_KEY, 0, opponent) - end - return opponent.to_i if result&.first&.positive? - else - redis.unwatch - end - end - end - nil - end - - def add_to_queue(user_id) - RedisConnection.with do |redis| - # Remove any stale entries first - redis.lrem(QUEUE_KEY, 0, user_id) - redis.rpush(QUEUE_KEY, user_id) + 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 wait_for_match(user_id) - start_time = Time.current - match_id = nil - - while Time.current - start_time < MAX_WAIT - RedisConnection.with do |redis| - match_id = redis.get("#{MATCH_PREFIX}#{user_id}") + 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 - break if match_id - - sleep POLL_INTERVAL - end - - if match_id - # Retrieve and delete the match data - data = nil - RedisConnection.with do |redis| - data = redis.get("#{MATCH_PREFIX}match:#{match_id}") - redis.del("#{MATCH_PREFIX}#{user_id}") - redis.del("#{MATCH_PREFIX}match:#{match_id}") - end - JSON.parse(data, symbolize_names: true) if data - else - # Timeout: remove from queue - cancel(user_id) - nil 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: GAME_SERVER, - exp: 2.minute.from_now.to_i + game_server: config.game_server_url, + exp: config.token_ttl.seconds.from_now.to_i ) - match_data = { + payload = { + type: "matched", match_id: match_id, game_token: game_token, - game_server: GAME_SERVER, + game_server: config.game_server_url, players: [ player1_id, player2_id ] - } + }.to_json - # Store match data for both players RedisConnection.with do |redis| - redis.set("#{MATCH_PREFIX}match:#{match_id}", match_data.to_json, ex: MATCH_TTL) - redis.set("#{MATCH_PREFIX}#{player1_id}", match_id, ex: MATCH_TTL) - redis.set("#{MATCH_PREFIX}#{player2_id}", match_id, ex: MATCH_TTL) + 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 - match_data + [ 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 diff --git a/main-server/config/cable.yml b/main-server/config/cable.yml new file mode 100644 index 0000000..b1638e3 --- /dev/null +++ b/main-server/config/cable.yml @@ -0,0 +1,10 @@ +development: + adapter: async + +test: + adapter: test + +production: + adapter: redis + url: <%= ENV.fetch("REDIS_URL", "redis://localhost:6380/0") %> + channel_prefix: ruby_sea_battle_production diff --git a/main-server/config/database.yml b/main-server/config/database.yml index 95234dc..c4aa2f6 100644 --- a/main-server/config/database.yml +++ b/main-server/config/database.yml @@ -10,15 +10,25 @@ default: &default timeout: 5000 development: - <<: *default - database: storage/development.sqlite3 + primary: + <<: *default + database: storage/development.sqlite3 + queue: + <<: *default + database: storage/development_queue.sqlite3 + migrations_paths: db/queue_migrate # Warning: The database defined as "test" will be erased and # re-generated from your development database when you run "rake". # Do not set this db to the same as development or production. test: - <<: *default - database: storage/test.sqlite3 + primary: + <<: *default + database: storage/test.sqlite3 + queue: + <<: *default + database: storage/test_queue.sqlite3 + migrations_paths: db/queue_migrate # PostgreSQL for production. # @@ -29,11 +39,22 @@ test: # DB_NAME, DB_USER, DB_PASSWORD, DB_HOST, DB_PORT # production: - adapter: postgresql - encoding: unicode - pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %> - database: <%= ENV["DB_NAME"] %> - username: <%= ENV["DB_USER"] %> - password: <%= ENV["DB_PASSWORD"] %> - host: <%= ENV["DB_HOST"] %> - port: <%= ENV["DB_PORT"] || 5432 %> + primary: + adapter: postgresql + encoding: unicode + pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %> + database: <%= ENV["DB_NAME"] %> + username: <%= ENV["DB_USER"] %> + password: <%= ENV["DB_PASSWORD"] %> + host: <%= ENV["DB_HOST"] %> + port: <%= ENV["DB_PORT"] || 5432 %> + queue: + adapter: postgresql + encoding: unicode + pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %> + database: <%= ENV["QUEUE_DB_NAME"] || ENV["DB_NAME"] %> + username: <%= ENV["QUEUE_DB_USER"] || ENV["DB_USER"] %> + password: <%= ENV["QUEUE_DB_PASSWORD"] || ENV["DB_PASSWORD"] %> + host: <%= ENV["QUEUE_DB_HOST"] || ENV["DB_HOST"] %> + port: <%= ENV["QUEUE_DB_PORT"] || ENV["DB_PORT"] || 5432 %> + migrations_paths: db/queue_migrate diff --git a/main-server/config/environments/development.rb b/main-server/config/environments/development.rb index f911c68..1b53542 100644 --- a/main-server/config/environments/development.rb +++ b/main-server/config/environments/development.rb @@ -3,6 +3,11 @@ require "active_support/core_ext/integer/time" Rails.application.configure do # Settings specified here will take precedence over those in config/application.rb. + # Run background jobs (e.g. the matchmaking sweeper) through Solid Queue, + # supervised by the puma plugin. + config.active_job.queue_adapter = :solid_queue + config.solid_queue.connects_to = { database: { writing: :queue } } + # Make code changes take effect immediately without server restart. config.enable_reloading = true diff --git a/main-server/config/environments/production.rb b/main-server/config/environments/production.rb index b242327..b91df99 100644 --- a/main-server/config/environments/production.rb +++ b/main-server/config/environments/production.rb @@ -50,7 +50,8 @@ Rails.application.configure do config.cache_store = :memory_store # Replace the default in-process and non-durable queuing backend for Active Job. - config.active_job.queue_adapter = :async + config.active_job.queue_adapter = :solid_queue + config.solid_queue.connects_to = { database: { writing: :queue } } # Ignore bad email addresses and do not raise email delivery errors. # Set this to true and configure the email server for immediate delivery to raise delivery errors. diff --git a/main-server/config/initializers/matchmaking.rb b/main-server/config/initializers/matchmaking.rb new file mode 100644 index 0000000..9c8dc37 --- /dev/null +++ b/main-server/config/initializers/matchmaking.rb @@ -0,0 +1,25 @@ +# Matchmaking configuration. All values can be overridden via environment variables. +mm = Rails.application.config.x.matchmaking = ActiveSupport::OrderedOptions.new + +mm.game_server_url = ENV.fetch("GAME_SERVER_URL", "ws://localhost:8080") +mm.default_rating = ENV.fetch("MATCHMAKING_DEFAULT_RATING", 1000).to_i + +# How long a found match (and its game token) stays valid, in seconds. +mm.match_ttl = ENV.fetch("MATCHMAKING_MATCH_TTL", 120).to_i +mm.token_ttl = ENV.fetch("MATCHMAKING_TOKEN_TTL", 120).to_i + +# How long a player waits in the queue before a timeout, in seconds. +mm.queue_ttl = ENV.fetch("MATCHMAKING_QUEUE_TTL", 60).to_i + +# How long a found (but not yet finalized) pair is kept in the "pending" +# state before the players are put back into the queue, in seconds. +mm.pending_ttl = ENV.fetch("MATCHMAKING_PENDING_TTL", 15).to_i + +# Rating tolerance: players within `base_range` are matched immediately; +# the tolerance grows by `range_per_second` of waiting up to `max_range`. +mm.base_range = ENV.fetch("MATCHMAKING_BASE_RANGE", 100).to_i +mm.range_per_second = ENV.fetch("MATCHMAKING_RANGE_PER_SECOND", 50).to_i +mm.max_range = ENV.fetch("MATCHMAKING_MAX_RANGE", 1000).to_i + +# How often the background sweep job re-checks the queue, in seconds. +mm.sweep_interval = ENV.fetch("MATCHMAKING_SWEEP_INTERVAL", 2).to_i diff --git a/main-server/config/initializers/redis.rb b/main-server/config/initializers/redis.rb index e2d123e..58a1795 100644 --- a/main-server/config/initializers/redis.rb +++ b/main-server/config/initializers/redis.rb @@ -2,6 +2,10 @@ # Development/test: localhost:6379 (default) # Production: set REDIS_URL env variable, e.g. redis://redis.example.com:6379/0 +# Use an isolated Redis logical database in tests to avoid clobbering dev data. +url = ENV.fetch("REDIS_URL", "redis://localhost:6380/0") +url = url.sub(%r{/\d+$}, "/15") if Rails.env.test? + RedisConnection = ConnectionPool.new(size: 5, timeout: 5) do - Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6380/0")) + Redis.new(url: url) end diff --git a/main-server/config/puma.rb b/main-server/config/puma.rb index 2737b92..61cd028 100644 --- a/main-server/config/puma.rb +++ b/main-server/config/puma.rb @@ -35,7 +35,7 @@ port ENV.fetch("PORT", 3000) plugin :tmp_restart # Run the Solid Queue supervisor inside of Puma for single-server deployments. -# plugin :solid_queue if ENV["SOLID_QUEUE_IN_PUMA"] +plugin :solid_queue if ENV["SOLID_QUEUE_IN_PUMA"] || ENV.fetch("RAILS_ENV", "development") == "development" # Specify the PID file. Defaults to tmp/pids/server.pid in development. # In other environments, only set the PID file if requested. diff --git a/main-server/config/queue.yml b/main-server/config/queue.yml new file mode 100644 index 0000000..6b14360 --- /dev/null +++ b/main-server/config/queue.yml @@ -0,0 +1,18 @@ +default: &default + dispatchers: + - polling_interval: 1 + batch_size: 500 + workers: + - queues: "*" + threads: 3 + processes: <%= ENV.fetch("JOB_CONCURRENCY", 1) %> + polling_interval: 1 + +development: + <<: *default + +test: + <<: *default + +production: + <<: *default diff --git a/main-server/config/recurring.yml b/main-server/config/recurring.yml new file mode 100644 index 0000000..5c26f2b --- /dev/null +++ b/main-server/config/recurring.yml @@ -0,0 +1,18 @@ +# examples: +# periodic_cleanup: +# class: CleanSoftDeletedRecordsJob +# queue: background +# args: [ 1000, { batch_size: 500 } ] +# schedule: every hour +# periodic_cleanup_with_command: +# command: "SoftDeletedRecord.due.delete_all" +# priority: 2 +# schedule: at 5am every day + +production: + matchmaking_sweep: + class: MatchmakingSweepJob + schedule: every <%= ENV.fetch("MATCHMAKING_SWEEP_INTERVAL", 2) %> seconds + clear_solid_queue_finished_jobs: + command: "SolidQueue::Job.clear_finished_in_batches(sleep_between_batches: 0.3)" + schedule: every hour at minute 12 diff --git a/main-server/config/routes.rb b/main-server/config/routes.rb index 36434c1..7a5044b 100644 --- a/main-server/config/routes.rb +++ b/main-server/config/routes.rb @@ -1,12 +1,16 @@ Rails.application.routes.draw do + mount ActionCable.server => "/cable" + namespace :api do post "register", to: "registrations#create" post "login", to: "sessions#create" delete "logout", to: "sessions#destroy" get "me", to: "sessions#me" - post "matchmaking", to: "matchmaking#create" delete "matchmaking", to: "matchmaking#destroy" + + # Matchmaking starts over the MatchmakingChannel WebSocket + # (see app/channels/matchmaking_channel.rb). end # Reveal health status on /up that returns 200 if the app boots with no exceptions, otherwise 500. diff --git a/main-server/db/queue_schema.rb b/main-server/db/queue_schema.rb new file mode 100644 index 0000000..f9a71da --- /dev/null +++ b/main-server/db/queue_schema.rb @@ -0,0 +1,160 @@ +ActiveRecord::Schema[7.1].define(version: 1) do + create_table "solid_queue_blocked_executions", force: :cascade do |t| + t.bigint "job_id", null: false + t.string "queue_name", null: false + t.integer "priority", default: 0, null: false + t.string "concurrency_key", null: false + t.datetime "expires_at", null: false + t.datetime "created_at", null: false + t.index [ "concurrency_key", "priority", "job_id" ], name: "index_solid_queue_blocked_executions_for_release" + t.index [ "expires_at", "concurrency_key" ], name: "index_solid_queue_blocked_executions_for_maintenance" + t.index [ "job_id" ], name: "index_solid_queue_blocked_executions_on_job_id", unique: true + end + + create_table "solid_queue_claimed_executions", force: :cascade do |t| + t.bigint "job_id", null: false + t.bigint "process_id" + t.datetime "created_at", null: false + t.index [ "job_id" ], name: "index_solid_queue_claimed_executions_on_job_id", unique: true + t.index [ "process_id", "job_id" ], name: "index_solid_queue_claimed_executions_on_process_id_and_job_id" + end + + create_table "solid_queue_failed_executions", force: :cascade do |t| + t.bigint "job_id", null: false + t.text "error" + t.datetime "created_at", null: false + t.index [ "job_id" ], name: "index_solid_queue_failed_executions_on_job_id", unique: true + end + + create_table "solid_queue_jobs", force: :cascade do |t| + t.string "queue_name", null: false + t.string "class_name", null: false + t.text "arguments" + t.integer "priority", default: 0, null: false + t.string "active_job_id" + t.datetime "scheduled_at" + t.datetime "finished_at" + t.string "concurrency_key" + t.datetime "created_at", null: false + t.datetime "updated_at", null: false + t.bigint "batch_id" + t.index [ "active_job_id" ], name: "index_solid_queue_jobs_on_active_job_id" + t.index [ "batch_id" ], name: "index_solid_queue_jobs_on_batch_id" + t.index [ "class_name" ], name: "index_solid_queue_jobs_on_class_name" + t.index [ "finished_at" ], name: "index_solid_queue_jobs_on_finished_at" + t.index [ "queue_name", "finished_at" ], name: "index_solid_queue_jobs_for_filtering" + t.index [ "scheduled_at", "finished_at" ], name: "index_solid_queue_jobs_for_alerting" + end + + create_table "solid_queue_pauses", force: :cascade do |t| + t.string "queue_name", null: false + t.datetime "created_at", null: false + t.index [ "queue_name" ], name: "index_solid_queue_pauses_on_queue_name", unique: true + end + + create_table "solid_queue_processes", force: :cascade do |t| + t.string "kind", null: false + t.datetime "last_heartbeat_at", null: false + t.bigint "supervisor_id" + t.integer "pid", null: false + t.string "hostname" + t.text "metadata" + t.datetime "created_at", null: false + t.string "name", null: false + t.index [ "last_heartbeat_at" ], name: "index_solid_queue_processes_on_last_heartbeat_at" + t.index [ "name", "supervisor_id" ], name: "index_solid_queue_processes_on_name_and_supervisor_id", unique: true + t.index [ "supervisor_id" ], name: "index_solid_queue_processes_on_supervisor_id" + end + + create_table "solid_queue_ready_executions", force: :cascade do |t| + t.bigint "job_id", null: false + t.string "queue_name", null: false + t.integer "priority", default: 0, null: false + t.datetime "created_at", null: false + t.index [ "job_id" ], name: "index_solid_queue_ready_executions_on_job_id", unique: true + t.index [ "priority", "job_id" ], name: "index_solid_queue_poll_all" + t.index [ "queue_name", "priority", "job_id" ], name: "index_solid_queue_poll_by_queue" + end + + create_table "solid_queue_recurring_executions", force: :cascade do |t| + t.bigint "job_id", null: false + t.string "task_key", null: false + t.datetime "run_at", null: false + t.datetime "created_at", null: false + t.index [ "job_id" ], name: "index_solid_queue_recurring_executions_on_job_id", unique: true + t.index [ "task_key", "run_at" ], name: "index_solid_queue_recurring_executions_on_task_key_and_run_at", unique: true + end + + create_table "solid_queue_recurring_tasks", force: :cascade do |t| + t.string "key", null: false + t.string "schedule", null: false + t.string "command", limit: 2048 + t.string "class_name" + t.text "arguments" + t.string "queue_name" + t.integer "priority", default: 0 + t.boolean "static", default: true, null: false + t.text "description" + t.datetime "created_at", null: false + t.datetime "updated_at", null: false + t.index [ "key" ], name: "index_solid_queue_recurring_tasks_on_key", unique: true + t.index [ "static" ], name: "index_solid_queue_recurring_tasks_on_static" + end + + create_table "solid_queue_scheduled_executions", force: :cascade do |t| + t.bigint "job_id", null: false + t.string "queue_name", null: false + t.integer "priority", default: 0, null: false + t.datetime "scheduled_at", null: false + t.datetime "created_at", null: false + t.index [ "job_id" ], name: "index_solid_queue_scheduled_executions_on_job_id", unique: true + t.index [ "scheduled_at", "priority", "job_id" ], name: "index_solid_queue_dispatch_all" + end + + create_table "solid_queue_semaphores", force: :cascade do |t| + t.string "key", null: false + t.integer "value", default: 1, null: false + t.datetime "expires_at", null: false + t.datetime "created_at", null: false + t.datetime "updated_at", null: false + t.index [ "expires_at" ], name: "index_solid_queue_semaphores_on_expires_at" + t.index [ "key", "value" ], name: "index_solid_queue_semaphores_on_key_and_value" + t.index [ "key" ], name: "index_solid_queue_semaphores_on_key", unique: true + end + + create_table "solid_queue_batches", force: :cascade do |t| + t.string "active_job_batch_id" + t.string "description" + t.text "on_finish" + t.text "on_success" + t.text "on_failure" + t.text "metadata" + t.integer "total_jobs", default: 0, null: false + t.integer "completed_jobs", default: 0, null: false + t.integer "failed_jobs", default: 0, null: false + t.datetime "enqueued_at" + t.datetime "finished_at" + t.datetime "failed_at" + t.datetime "created_at", null: false + t.datetime "updated_at", null: false + t.index [ "active_job_batch_id" ], name: "index_solid_queue_batches_on_active_job_batch_id", unique: true + t.index [ "finished_at" ], name: "index_solid_queue_batches_on_finished_at" + end + + create_table "solid_queue_batch_executions", force: :cascade do |t| + t.bigint "job_id", null: false + t.bigint "batch_id", null: false + t.datetime "created_at", null: false + t.index [ "job_id" ], name: "index_solid_queue_batch_executions_on_job_id", unique: true + t.index [ "batch_id" ], name: "index_solid_queue_batch_executions_on_batch_id" + end + + add_foreign_key "solid_queue_batch_executions", "solid_queue_batches", column: "batch_id", on_delete: :cascade + add_foreign_key "solid_queue_batch_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_ready_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_recurring_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_scheduled_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade +end diff --git a/main-server/spec/services/matchmaking_service_spec.rb b/main-server/spec/services/matchmaking_service_spec.rb new file mode 100644 index 0000000..245a753 --- /dev/null +++ b/main-server/spec/services/matchmaking_service_spec.rb @@ -0,0 +1,163 @@ +require "rails_helper" + +RSpec.describe MatchmakingService do + # The service uses the shared RedisConnection (test env points at db 15). + def redis + RedisConnection.with { |r| return r } + end + + # Unique user ids across the whole spec run (Action Cable test-adapter + # broadcasts persist between examples; example-group instance variables + # are reset per example, so a global counter is used). + def uid(base) + $matchmaking_spec_uid = ($matchmaking_spec_uid || 0) + 1 + $matchmaking_spec_uid * 1000 + base + end + + def broadcasts_for(user_id) + channel = MatchmakingChannel.broadcasting_for(user_id) + ActionCable.server.pubsub.broadcasts(channel) + end + + def broadcast_payloads(user_id) + broadcasts_for(user_id).map { |m| JSON.parse(m, symbolize_names: true) } + end + + before { redis.flushdb } + + describe ".enqueue" do + it "does not match players whose ratings are outside the tolerance" do + a, b = uid(1), uid(2) + described_class.enqueue(a, rating: 1000) + described_class.enqueue(b, rating: 5000) + + expect(redis.zrange(described_class::QUEUE_KEY, 0, -1)).to contain_exactly(a.to_s, b.to_s) + expect(broadcasts_for(a)).to be_empty + expect(broadcasts_for(b)).to be_empty + end + + it "matches two compatible players immediately, nearest rating first" do + a, b, c = uid(1), uid(2), uid(3) + described_class.enqueue(a, rating: 1000) + described_class.enqueue(b, rating: 1400) # outside base range (100) + described_class.enqueue(c, rating: 1050) # nearest to a + + match = broadcast_payloads(a).first + expect(match[:type]).to eq("matched") + expect(match[:players]).to contain_exactly(a, c) + expect(match[:match_id]).to be_present + expect(match[:game_token]).to be_present + expect(match[:game_server]).to eq(Rails.application.config.x.matchmaking.game_server_url) + + expect(broadcasts_for(c)).not_to be_empty + # b stays in the queue + expect(redis.zrange(described_class::QUEUE_KEY, 0, -1)).to eq([ b.to_s ]) + end + + it "restarts the waiting session for a user already in the queue" do + a = uid(1) + described_class.enqueue(a, rating: 1000) + described_class.enqueue(a, rating: 1200) + + expect(redis.zrange(described_class::QUEUE_KEY, 0, -1)).to eq([ a.to_s ]) + expect(redis.zscore(described_class::QUEUE_KEY, a.to_s)).to eq(1200) + end + + it "delivers the match exactly once (broadcast consumes the pointer)" do + a, b = uid(1), uid(2) + described_class.enqueue(a, rating: 1000) + described_class.enqueue(b, rating: 1000) + + expect(broadcasts_for(a).size).to eq(1) + expect(described_class.claim(a)).to be_nil + end + end + + describe ".sweep" do + it "widens the tolerance over waiting time and pairs players" do + a, b = uid(1), uid(2) + described_class.enqueue(a, rating: 1000) + described_class.enqueue(b, rating: 1400) + # backdate b's enqueue time (in ms, as the script stores it) so its + # tolerance has widened enough + redis.hset(described_class::SINCE_KEY, b.to_s, (10.seconds.ago.to_f * 1000).to_i) + + described_class.sweep + + expect(redis.zrange(described_class::QUEUE_KEY, 0, -1)).to be_empty + expect(broadcasts_for(a)).not_to be_empty + expect(broadcasts_for(b)).not_to be_empty + end + + it "expires players who waited longer than the queue TTL" do + a = uid(1) + described_class.enqueue(a, rating: 1000) + redis.hset(described_class::SINCE_KEY, a.to_s, + ((described_class.config.queue_ttl + 5).seconds.ago.to_f * 1000).to_i) + + described_class.sweep + + expect(redis.zrange(described_class::QUEUE_KEY, 0, -1)).to be_empty + expect(broadcast_payloads(a).first[:type]).to eq("timeout") + end + + it "requeues players stuck in the pending state without a match" do + a, b = uid(1), uid(2) + described_class.enqueue(a, rating: 1000) + described_class.enqueue(b, rating: 1000) + + # simulate a crash between pairing and finalize: back into pending + # (scores are pairing timestamps in milliseconds). The pairing step + # already removed the players from the queue; restore their ratings as + # they would exist before finalize deletes them. + redis.hset(described_class::RATING_KEY, a.to_s, 1000, b.to_s, 1000) + past = (described_class.config.pending_ttl + 5).seconds.ago.to_f * 1000 + redis.zadd(described_class::PENDING_KEY, past.to_i, a.to_s) + redis.zadd(described_class::PENDING_KEY, past.to_i, b.to_s) + # pointers must be absent for the requeue to kick in + redis.del("#{described_class::USER_KEY_PREFIX}#{a}", "#{described_class::USER_KEY_PREFIX}#{b}") + + described_class.sweep + + expect(redis.zrange(described_class::QUEUE_KEY, 0, -1)).to contain_exactly(a.to_s, b.to_s) + end + + it "does not requeue pending players whose match was finalized" do + a, b = uid(1), uid(2) + described_class.enqueue(a, rating: 1000) + described_class.enqueue(b, rating: 1000) + + # emulate a leftover stale pending entry for a user with a live pointer + past = (described_class.config.pending_ttl + 5).seconds.ago.to_f * 1000 + redis.zadd(described_class::PENDING_KEY, past.to_i, a.to_s) + + described_class.sweep + + expect(redis.zrange(described_class::QUEUE_KEY, 0, -1)).to be_empty + expect(broadcasts_for(a)).not_to be_empty + end + end + + describe ".cancel" do + it "removes the player from the queue" do + a = uid(1) + described_class.enqueue(a, rating: 1000) + described_class.cancel(a) + + expect(redis.zrange(described_class::QUEUE_KEY, 0, -1)).to be_empty + expect(redis.hget(described_class::SINCE_KEY, a.to_s)).to be_nil + end + end + + describe ".claim" do + it "returns the match payload exactly once" do + a = uid(1) + payload = { type: "matched", match_id: "m-1", players: [ a, a + 1 ] }.to_json + redis.set("#{described_class::USER_KEY_PREFIX}#{a}", payload) + + first = described_class.claim(a) + expect(first[:type]).to eq("matched") + expect(described_class.claim(a)).to be_nil + end + end +end