implement alternative match-making approach with Lua scripts in Redis

2026-10-25 15:57:24 +03:00
parent 582ecb7c74
commit acdaebd491
19 changed files with 796 additions and 139 deletions
@@ -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=<JWT>`

-**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}"}
 ```
@@ -0,0 +1,4 @@
+module ApplicationCable
+  class Channel < ActionCable::Channel::Base
+  end
+end
@@ -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
@@ -0,0 +1,31 @@
+# Real-time matchmaking channel.
+#
+# Client flow:
+#   1. Connect: ws(s)://host/cable?token=<JWT>
+#   2. Subscribe to "MatchmakingChannel"
+#   3. Send { action: "start", rating: <optional int> } 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
@@ -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" }
@@ -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
@@ -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:<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 = "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:<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]

-  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
@@ -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
@@ -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
@@ -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

@@ -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.
@@ -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
@@ -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
@@ -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.
@@ -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
@@ -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
@@ -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.
@@ -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
@@ -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