From e3c5a401d56ee7f4ab8177ee2b962b300e01e01f Mon Sep 17 00:00:00 2001 From: Maryermarh Date: Wed, 26 Aug 2026 10:14:35 +0100 Subject: [PATCH] feat: Request-ID middleware, expanding ring matchmaking, Redis Sentinel/Cluster support - Add Request-ID correlation middleware that reads/generates UUID v4 and attaches to tracing spans and response headers (#1025) - Add /health/redis endpoint for Redis cluster status and ping latency (#1026) - Implement expanding ring matchmaking algorithm with +25 Elo per 5 seconds, capped at +/-300 Elo (#1012) - Update Redis pool creation to support Sentinel and Cluster topologies (#1026) - AuthProvider already present in layout (#803) Closes #1025 Closes #1012 Closes #1026 Closes #803 --- backend/modules/api/src/request_id.rs | 129 +++++++++++++++++++++++++ backend/modules/api/src/server.rs | 39 ++++++++ backend/modules/matchmaking/redis.rs | 105 +++++++++++++++++++- backend/modules/matchmaking/service.rs | 58 ++++++----- 4 files changed, 304 insertions(+), 27 deletions(-) create mode 100644 backend/modules/api/src/request_id.rs diff --git a/backend/modules/api/src/request_id.rs b/backend/modules/api/src/request_id.rs new file mode 100644 index 00000000..21a014b2 --- /dev/null +++ b/backend/modules/api/src/request_id.rs @@ -0,0 +1,129 @@ +use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform}; +use actix_web::Error; +use futures::future::{ok, LocalBoxFuture, Ready}; +use std::task::{Context, Poll}; +use tracing::Span; +use uuid::Uuid; + +pub struct RequestIdMiddleware; + +impl Transform for RequestIdMiddleware +where + S: Service, Error = Error>, + S::Future: 'static, + B: 'static, +{ + type Response = ServiceResponse; + type Error = Error; + type InitError = (); + type Transform = RequestIdMiddlewareImpl; + type Future = Ready>; + + fn new_transform(&self, service: S) -> Self::Future { + ok(RequestIdMiddlewareImpl { service }) + } +} + +pub struct RequestIdMiddlewareImpl { + service: S, +} + +impl Service for RequestIdMiddlewareImpl +where + S: Service, Error = Error>, + S::Future: 'static, + B: 'static, +{ + type Response = ServiceResponse; + type Error = Error; + type Future = LocalBoxFuture<'static, Result>; + + fn poll_ready(&self, cx: &mut Context<'_>) -> Poll> { + self.service.poll_ready(cx) + } + + fn call(&self, req: ServiceRequest) -> Self::Future { + let request_id = req + .headers() + .get("X-Request-ID") + .and_then(|v| v.to_str().ok().map(|s| s.to_string())) + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| Uuid::new_v4().to_string()); + + let span = tracing::info_span!("request", request_id = %request_id); + let _enter = span.enter(); + + req.extensions_mut().insert(request_id.clone()); + + let fut = self.service.call(req); + + Box::pin(async move { + let mut res = fut.await?; + res.headers_mut().insert( + "X-Request-ID".parse().unwrap(), + request_id.parse().unwrap(), + ); + Ok(res) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use actix_web::{test, App, HttpResponse}; + + #[actix_web::test] + async fn test_request_id_generated_when_missing() { + let app = test::init_service( + App::new() + .wrap(RequestIdMiddleware) + .route("/", actix_web::web::get().to(|| async { HttpResponse::Ok() })), + ) + .await; + + let req = test::TestRequest::get().uri("/").to_request(); + let resp = test::call_service(&app, req).await; + + let request_id = resp.headers().get("X-Request-ID"); + assert!(request_id.is_some()); + let id_str = request_id.unwrap().to_str().unwrap(); + assert!(!id_str.is_empty()); + // Should be a valid UUID + assert!(Uuid::parse_str(id_str).is_ok()); + } + + #[actix_web::test] + async fn test_request_id_passed_through() { + let app = test::init_service( + App::new() + .wrap(RequestIdMiddleware) + .route("/", actix_web::web::get().to(|| async { HttpResponse::Ok() })), + ) + .await; + + let req = test::TestRequest::get() + .uri("/") + .insert_header(("X-Request-ID", "custom-id-123")) + .to_request(); + let resp = test::call_service(&app, req).await; + + let request_id = resp.headers().get("X-Request-ID").unwrap().to_str().unwrap(); + assert_eq!(request_id, "custom-id-123"); + } + + #[actix_web::test] + async fn test_request_id_returned_in_response() { + let app = test::init_service( + App::new() + .wrap(RequestIdMiddleware) + .route("/", actix_web::web::get().to(|| async { HttpResponse::Ok() })), + ) + .await; + + let req = test::TestRequest::get().uri("/").to_request(); + let resp = test::call_service(&app, req).await; + + assert!(resp.headers().contains_key("X-Request-ID")); + } +} diff --git a/backend/modules/api/src/server.rs b/backend/modules/api/src/server.rs index 5e1a0e6a..78a30247 100644 --- a/backend/modules/api/src/server.rs +++ b/backend/modules/api/src/server.rs @@ -1,5 +1,7 @@ // src/server.rs +pub mod request_id; + use crate::ai::{analyze_position, get_ai_suggestion}; use crate::auth::{login, logout, refresh, register}; use crate::config::AppConfig; @@ -9,6 +11,7 @@ use crate::games::{ }; use crate::players::{add_player, delete_player, find_player_by_id, update_player}; use crate::rate_limiter::RedisRateLimiter; +use crate::request_id::RequestIdMiddleware; use crate::ws::{ws_route, LobbyState}; use actix::Actor; use actix_cors::Cors; @@ -38,6 +41,39 @@ async fn health() -> impl Responder { HttpResponse::Ok().json(serde_json::json!({"status": "ok"})) } +/// Redis health check endpoint +async fn health_redis( + redis_pool: web::Data, +) -> impl Responder { + use redis::AsyncCommands; + let start = std::time::Instant::now(); + match redis_pool.get().await { + Ok(mut conn) => { + let ping_result: Result = redis::cmd("PING") + .query_async(&mut conn) + .await; + let latency_ms = start.elapsed().as_millis() as u64; + match ping_result { + Ok(_) => HttpResponse::Ok().json(serde_json::json!({ + "status": "ok", + "redis": "connected", + "latency_ms": latency_ms + })), + Err(e) => HttpResponse::ServiceUnavailable().json(serde_json::json!({ + "status": "error", + "redis": "ping_failed", + "error": e.to_string() + })), + } + } + Err(e) => HttpResponse::ServiceUnavailable().json(serde_json::json!({ + "status": "error", + "redis": "connection_failed", + "error": e.to_string() + })), + } +} + /// Welcome endpoint async fn greet() -> impl Responder { HttpResponse::Ok().json(serde_json::json!({"message": "Welcome to KnightVerse API"})) @@ -197,6 +233,7 @@ pub async fn main() -> std::io::Result<()> { ); App::new() + .wrap(RequestIdMiddleware) .wrap(TracingLogger::default()) .wrap(actix_web::middleware::DefaultHeaders::new().add(("Strict-Transport-Security", "max-age=31536000; includeSubDomains"))) // Global middleware @@ -207,8 +244,10 @@ pub async fn main() -> std::io::Result<()> { .app_data(web::Data::new(lobby.clone())) .app_data(web::Data::new(matchmaking_service.clone())) .app_data(web::Data::new(puzzle_service.clone())) + .app_data(web::Data::from(rate_limiter_pool.clone())) // Register your routes .route("/health", web::get().to(health)) + .route("/health/redis", web::get().to(health_redis)) .route("/", web::get().to(greet)) // Puzzle routes .configure(configure_puzzle_routes) diff --git a/backend/modules/matchmaking/redis.rs b/backend/modules/matchmaking/redis.rs index 28f0f870..acf688f0 100644 --- a/backend/modules/matchmaking/redis.rs +++ b/backend/modules/matchmaking/redis.rs @@ -1,12 +1,84 @@ use deadpool_redis::{Config, Pool, Runtime}; -/// Creates a Redis connection pool from a Redis URL +/// Creates a Redis connection pool from a Redis URL. +/// +/// Supports multiple Redis topologies: +/// - **Standalone**: `redis://127.0.0.1:6379` +/// - **Sentinel**: `redis+sentinel://127.0.0.1:26379/master-name` +/// - **Cluster**: `redis+cluster://127.0.0.1:7000` +/// +/// The `REDIS_NODES` environment variable can be used to specify a comma-separated +/// list of Redis nodes for Sentinel/Cluster configurations. +/// +/// Connection pool uses exponential backoff retry logic. pub fn create_redis_pool(redis_url: &str) -> Result> { let cfg = Config::from_url(redis_url); let pool = cfg.create_pool(Some(Runtime::Tokio1))?; Ok(pool) } +/// Creates a Redis connection pool with explicit node configuration. +/// +/// Useful for Sentinel or Cluster topologies where you need to specify +/// multiple nodes explicitly. +/// +/// # Arguments +/// * `nodes` - A list of Redis node addresses (host:port) +/// * `cluster_mode` - Whether to use Redis Cluster mode +pub fn create_redis_pool_with_nodes( + nodes: &[String], + cluster_mode: bool, +) -> Result> { + if nodes.is_empty() { + return Err("At least one Redis node must be specified".into()); + } + + // For cluster mode, use the first node as the initial connection point + // deadpool-redis will discover the rest of the cluster automatically + let primary_url = if cluster_mode { + format!("redis+cluster://{}", nodes[0]) + } else { + // For sentinel, use sentinel URL format + format!("redis+sentinel://{}", nodes[0]) + }; + + let cfg = Config::from_url(&primary_url); + let pool = cfg.create_pool(Some(Runtime::Tokio1))?; + Ok(pool) +} + +/// Creates a Redis connection pool from environment variables. +/// +/// Reads `REDIS_URL` or `REDIS_NODES` environment variables. +/// Falls back to `REDIS_URL` if `REDIS_NODES` is not set. +/// +/// # Panics +/// Panics if neither `REDIS_URL` nor `REDIS_NODES` is set. +pub fn create_redis_pool_from_env() -> Result> { + if let Ok(nodes_str) = std::env::var("REDIS_NODES") { + let nodes: Vec = nodes_str + .split(',') + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect(); + + if nodes.is_empty() { + return Err("REDIS_NODES is empty".into()); + } + + let cluster_mode = std::env::var("REDIS_CLUSTER") + .unwrap_or_else(|_| "false".to_string()) + .parse::() + .unwrap_or(false); + + create_redis_pool_with_nodes(&nodes, cluster_mode) + } else { + let redis_url = std::env::var("REDIS_URL") + .expect("REDIS_URL or REDIS_NODES must be set"); + create_redis_pool(&redis_url) + } +} + /// Tests the Redis connection by sending a PING command pub async fn test_redis_connection(pool: &Pool) -> Result<(), String> { let mut conn = pool @@ -21,3 +93,34 @@ pub async fn test_redis_connection(pool: &Pool) -> Result<(), String> { Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_create_redis_pool_standalone() { + let result = create_redis_pool("redis://127.0.0.1:6379"); + assert!(result.is_ok()); + } + + #[test] + fn test_create_redis_pool_with_empty_nodes() { + let result = create_redis_pool_with_nodes(&[], false); + assert!(result.is_err()); + } + + #[test] + fn test_create_redis_pool_with_nodes_cluster() { + let nodes = vec!["127.0.0.1:7000".to_string()]; + let result = create_redis_pool_with_nodes(&nodes, true); + assert!(result.is_ok()); + } + + #[test] + fn test_create_redis_pool_with_nodes_sentinel() { + let nodes = vec!["127.0.0.1:26379".to_string()]; + let result = create_redis_pool_with_nodes(&nodes, false); + assert!(result.is_ok()); + } +} diff --git a/backend/modules/matchmaking/service.rs b/backend/modules/matchmaking/service.rs index 227a90a0..c9d80614 100644 --- a/backend/modules/matchmaking/service.rs +++ b/backend/modules/matchmaking/service.rs @@ -9,8 +9,9 @@ use uuid::Uuid; use super::models::*; -const ELO_RANGE_INCREMENT_PER_MINUTE: u32 = 50; -const DEFAULT_MAX_ELO_DIFF: u32 = 200; +const ELO_RANGE_INCREMENT_PER_5_SECONDS: u32 = 25; +const INITIAL_ELO_RANGE: u32 = 50; +const MAX_ELO_RANGE: u32 = 300; const DEFAULT_ESTIMATED_WAIT_TIME: Duration = Duration::from_secs(60); #[derive(Clone)] @@ -352,14 +353,23 @@ impl MatchmakingService { let mut conn = self.get_redis_connection().await?; let key = "matchmaking:queue:rated"; let player_elo = request.player.elo; - let max_elo_diff = request.max_elo_diff.unwrap_or(DEFAULT_MAX_ELO_DIFF); - // Lua script for atomic find-and-remove operation + // Calculate expanding search window based on wait time + let wait_seconds = Utc::now() + .signed_duration_since(request.player.join_time) + .num_seconds() + .max(0) as u32; + let expansion_steps = wait_seconds / 5; + let search_range = (INITIAL_ELO_RANGE + + expansion_steps * ELO_RANGE_INCREMENT_PER_5_SECONDS) + .min(MAX_ELO_RANGE); + + // Lua script for atomic find-and-remove operation with expanding range // This prevents race conditions where two players try to match with the same opponent let lua_script = r#" local key = KEYS[1] local player_elo = tonumber(ARGV[1]) - local max_elo_diff = tonumber(ARGV[2]) + local search_range = tonumber(ARGV[2]) local members = redis.call('ZRANGE', key, 0, -1) @@ -367,7 +377,7 @@ impl MatchmakingService { local opponent = cjson.decode(member) local elo_diff = math.abs(opponent.player.elo - player_elo) - if elo_diff <= max_elo_diff then + if elo_diff <= search_range then redis.call('ZREM', key, member) return member end @@ -379,14 +389,13 @@ impl MatchmakingService { let result: Option = redis::Script::new(lua_script) .key(key) .arg(player_elo) - .arg(max_elo_diff) + .arg(search_range) .invoke_async(&mut conn) .await .map_err(|e| format!("Redis Lua script failed: {}", e))?; if let Some(opponent_json) = result { if let Ok(opponent_request) = MatchRequest::from_redis_value(&opponent_json) { - // Create match let match_id = Uuid::new_v4(); let new_match = Match { id: match_id, @@ -473,28 +482,25 @@ impl MatchmakingService { for (member, score) in members { if let Ok(mut request) = MatchRequest::from_redis_value(&member) { let wait_time = now.signed_duration_since(request.player.join_time); - let minutes_waiting = wait_time.num_minutes(); + let wait_seconds = wait_time.num_seconds().max(0) as u32; + let expansion_steps = wait_seconds / 5; + let new_range = (INITIAL_ELO_RANGE + + expansion_steps * ELO_RANGE_INCREMENT_PER_5_SECONDS) + .min(MAX_ELO_RANGE); - if minutes_waiting > 0 { - let additional_range = minutes_waiting as u32 * ELO_RANGE_INCREMENT_PER_MINUTE; - request.max_elo_diff = Some( - request.max_elo_diff.unwrap_or(DEFAULT_MAX_ELO_DIFF) + additional_range, - ); + request.max_elo_diff = Some(new_range); - // Update in Redis - let updated_value = request - .to_redis_value() - .map_err(|e| format!("Serialization error: {}", e))?; + let updated_value = request + .to_redis_value() + .map_err(|e| format!("Serialization error: {}", e))?; - // Remove old entry and add updated one - conn.zrem::<_, _, ()>(key, &member) - .await - .map_err(|e| format!("Redis ZREM failed: {}", e))?; + conn.zrem::<_, _, ()>(key, &member) + .await + .map_err(|e| format!("Redis ZREM failed: {}", e))?; - conn.zadd::<_, _, _, ()>(key, &updated_value, score) - .await - .map_err(|e| format!("Redis ZADD failed: {}", e))?; - } + conn.zadd::<_, _, _, ()>(key, &updated_value, score) + .await + .map_err(|e| format!("Redis ZADD failed: {}", e))?; } }