Feat: Implement WebSocket Scaling with Horizontal Load Balancing and Regional Discovery (#568) #680
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,3 +3,4 @@ pub mod game; | |
| pub mod handlers; | ||
| pub mod models; | ||
| pub mod websocket; | ||
| pub mod pubsub; | ||
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,52 @@ | ||||||
| use redis::{AsyncCommands, Client}; | ||||||
| use std::sync::Arc; | ||||||
| use tokio::sync::Mutex; | ||||||
| use crate::models::ServerMessage; | ||||||
| use lazy_static::lazy_static; | ||||||
|
|
||||||
| #[derive(Clone)] | ||||||
| pub struct RedisPubSub { | ||||||
| client: Client, | ||||||
| connection: Arc<Mutex<redis::aio::Connection>>, | ||||||
|
Contributor
There was a problem hiding this comment. 🧩 Analysis chain🌐 Web query:
💡 Result: In redis-rs 0.25, MultiplexedConnection (from redis::aio) is the recommended async connection type, replacing the deprecated aio::Connection. It supports concurrency and pipelining on a single underlying TCP socket. MultiplexedConnection is cheaply cloneable and allows sending requests concurrently from multiple clones/tasks without waiting for responses (responses return in order). This enables high concurrency: clone the connection and use it across async tasks/threads for non-blocking commands. It natively supports pipelining via send_packed_commands (used internally by Pipeline::query_async). A bug affecting single-command pipelines on MultiplexedConnection was fixed in 0.25.2 (PR Citations:
🏁 Script executed: fd -e rs -p "pubsub" | head -20Repository: NOVUS-X/XLMate Length of output: 87 🏁 Script executed: cat -n backend/src/socket/pubsub.rsRepository: NOVUS-X/XLMate Length of output: 2278 Use
♻️ Proposed refactor-use redis::{AsyncCommands, Client};
-use std::sync::Arc;
-use tokio::sync::Mutex;
+use redis::{aio::MultiplexedConnection, AsyncCommands, Client};
@@
#[derive(Clone)]
pub struct RedisPubSub {
client: Client,
- connection: Arc<Mutex<redis::aio::Connection>>,
+ connection: MultiplexedConnection,
}
@@
- pub async fn new() -> Result<Self, Box<dyn std::error::Error>> {
+ pub async fn new() -> Result<Self, Box<dyn std::error::Error>> {
let client = Client::open(REDIS_URL.as_str())?;
- let connection = client.get_async_connection().await?;
+ let connection = client.get_multiplexed_async_connection().await?;
- Ok(Self {
- client,
- connection: Arc::new(Mutex::new(connection)),
- })
+ Ok(Self { client, connection })
}
@@
- pub async fn publish_move(&self, room_id: &str, message: &ServerMessage) -> Result<(), Box<dyn std::error::Error>> {
+ pub async fn publish_move(&self, room_id: &str, message: &ServerMessage) -> Result<(), Box<dyn std::error::Error>> {
let payload = serde_json::to_string(message)?;
- let mut conn = self.connection.lock().await;
+ let mut conn = self.connection.clone();
let _: () = conn.publish(format!("game:{}", room_id), payload).await?;📝 Committable suggestion
Suggested change
🤖 Prompt for AI Agents |
||||||
| } | ||||||
|
|
||||||
| lazy_static! { | ||||||
| static ref REDIS_URL: String = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); | ||||||
| } | ||||||
|
Comment on lines
+13
to
+15
Contributor
There was a problem hiding this comment.
Combined with the missing 🤖 Prompt for AI Agents |
||||||
|
|
||||||
| impl RedisPubSub { | ||||||
| pub async fn new() -> Result<Self, Box<dyn std::error::Error>> { | ||||||
| let client = Client::open(REDIS_URL.as_str())?; | ||||||
| let connection = client.get_async_connection().await?; | ||||||
|
|
||||||
| Ok(Self { | ||||||
| client, | ||||||
| connection: Arc::new(Mutex::new(connection)), | ||||||
| }) | ||||||
| } | ||||||
|
Comment on lines
+18
to
+26
Contributor
There was a problem hiding this comment. 🧩 Analysis chain🌐 Web query:
💡 Result: Box is problematic for tokio::spawn because dyn std::error::Error does not implement Send. The tokio::spawn function requires the future passed to it to implement Future + Send + 'static, with its Output also being Send + 'static. When an async function returns Result<T, Box>, the generated future captures this return type. Since dyn Error lacks Send, the future cannot be Send, causing a compiler error like "(dyn std::error::Error + 'static) cannot be sent between threads safely". The recommended bound is Box<dyn std::error::Error + Send>. This ensures the trait object implements Send, making the future Send and compatible with tokio::spawn. Some sources also suggest + Send + Sync for additional safety (e.g., if the error needs to be shared across threads), but + Send is the minimal requirement for spawning. Citations:
🏁 Script executed: find . -name "pubsub.rs" -type fRepository: NOVUS-X/XLMate Length of output: 89 🏁 Script executed: find . -name "game.rs" -type fRepository: NOVUS-X/XLMate Length of output: 236 🏁 Script executed: wc -l ./backend/src/socket/pubsub.rsRepository: NOVUS-X/XLMate Length of output: 92 🏁 Script executed: cat -n ./backend/src/socket/pubsub.rs | head -60Repository: NOVUS-X/XLMate Length of output: 2278 🏁 Script executed: cat -n ./backend/src/socket/game.rs | head -100Repository: NOVUS-X/XLMate Length of output: 3854 🏁 Script executed: rg "publish_move|subscribe_to_room" ./backend/src/socket/game.rs -A 3 -B 3Repository: NOVUS-X/XLMate Length of output: 40 🏁 Script executed: rg "publish_move|subscribe_to_room" ./backend/ -r --type rust -A 2Repository: NOVUS-X/XLMate Length of output: 765 🏁 Script executed: rg "publish_move|subscribe_to_room" ./backend/ --type-list | head -20Repository: NOVUS-X/XLMate Length of output: 448 🏁 Script executed: rg "publish_move|subscribe_to_room" ./backend/Repository: NOVUS-X/XLMate Length of output: 358 🏁 Script executed: rg "tokio::spawn|tokio::task::spawn" ./backend/ -A 5Repository: NOVUS-X/XLMate Length of output: 1561
The trait object must be 🛠️ Suggested fix- pub async fn new() -> Result<Self, Box<dyn std::error::Error>> {
+ pub async fn new() -> Result<Self, Box<dyn std::error::Error + Send>> {
@@
- pub async fn publish_move(&self, room_id: &str, message: &ServerMessage) -> Result<(), Box<dyn std::error::Error>> {
+ pub async fn publish_move(&self, room_id: &str, message: &ServerMessage) -> Result<(), Box<dyn std::error::Error + Send>> {
@@
- pub async fn subscribe_to_room(&self, room_id: &str) -> Result<redis::aio::PubSub, Box<dyn std::error::Error>> {
+ pub async fn subscribe_to_room(&self, room_id: &str) -> Result<redis::aio::PubSub, Box<dyn std::error::Error + Send>> {🤖 Prompt for AI Agents |
||||||
|
|
||||||
| /** | ||||||
| * Publishes a message to a regional Redis channel to sync horizontally scaled nodes. | ||||||
| * This ensures that players on different server instances stay in sync. | ||||||
| */ | ||||||
| pub async fn publish_move(&self, room_id: &str, message: &ServerMessage) -> Result<(), Box<dyn std::error::Error>> { | ||||||
| let payload = serde_json::to_string(message)?; | ||||||
| let mut conn = self.connection.lock().await; | ||||||
| let _: () = conn.publish(format!("game:{}", room_id), payload).await?; | ||||||
|
|
||||||
| log::debug!("[Redis] Published move for room {}", room_id); | ||||||
| Ok(()) | ||||||
| } | ||||||
|
|
||||||
| /** | ||||||
| * Subscribes to a specific room's updates. | ||||||
| * In a production horizontal setup, this would be used by each node to listen for events from others. | ||||||
| */ | ||||||
| pub async fn subscribe_to_room(&self, room_id: &str) -> Result<redis::aio::PubSub, Box<dyn std::error::Error>> { | ||||||
| let mut pubsub = self.client.get_async_pubsub().await?; | ||||||
| pubsub.subscribe(format!("game:{}", room_id)).await?; | ||||||
|
|
||||||
| log::info!("[Redis] Subscribed to horizontal updates for room {}", room_id); | ||||||
| Ok(pubsub) | ||||||
| } | ||||||
|
Comment on lines
+45
to
+51
Contributor
There was a problem hiding this comment.
Two concerns to consider before this is wired up:
Worth addressing as part of the integration step (which is currently missing — see comment on 🤖 Prompt for AI Agents |
||||||
| } | ||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| "use client"; | ||
|
|
||
| import React, { createContext, useContext, useState, useEffect, useCallback, useRef } from "react"; | ||
|
|
||
| export type WebSocketNode = { | ||
| id: string; | ||
| url: string; | ||
| region: string; | ||
| load: number; // 0 to 1 | ||
| }; | ||
|
|
||
| type WebSocketScalingContextType = { | ||
| currentNode: WebSocketNode | null; | ||
| availableNodes: WebSocketNode[]; | ||
| isScaling: boolean; | ||
| getOptimalNode: (gameId: string) => Promise<WebSocketNode>; | ||
| reportLoad: (nodeId: string, load: number) => void; | ||
| }; | ||
|
|
||
| const WebSocketScalingContext = createContext<WebSocketScalingContextType | undefined>(undefined); | ||
|
|
||
| // Mock discovery service URL - in production this would be a real endpoint | ||
| const DISCOVERY_ENDPOINT = process.env.NEXT_PUBLIC_DISCOVERY_URL || "http://localhost:8000/v1/nodes/discovery"; | ||
|
|
||
| export function WebSocketScalingProvider({ children }: { children: React.ReactNode }) { | ||
| const [currentNode, setCurrentNode] = useState<WebSocketNode | null>(null); | ||
| const [availableNodes, setAvailableNodes] = useState<WebSocketNode[]>([]); | ||
| const [isScaling, setIsScaling] = useState(false); | ||
|
|
||
| // Cache for game-to-node mapping to ensure session stickiness within horizontal scaling | ||
| const gameNodeMap = useRef<Map<string, WebSocketNode>>(new Map()); | ||
|
|
||
| /** | ||
| * Fetches the most optimal WebSocket node based on regional proximity and server load. | ||
| * Implements horizontal load balancing logic on the client side. | ||
| */ | ||
| const getOptimalNode = useCallback(async (gameId: string): Promise<WebSocketNode> => { | ||
| // Check if we already have a sticky session for this game | ||
| if (gameNodeMap.current.has(gameId)) { | ||
| return gameNodeMap.current.get(gameId)!; | ||
| } | ||
|
|
||
| setIsScaling(true); | ||
| try { | ||
| // In a real implementation, we would fetch from DISCOVERY_ENDPOINT | ||
| // For now, we simulate a discovery process that returns available horizontal nodes | ||
| const mockNodes: WebSocketNode[] = [ | ||
| { id: "node-us-east-1", url: "ws://localhost:8080", region: "us-east", load: 0.2 }, | ||
| { id: "node-eu-west-1", url: "ws://localhost:8081", region: "eu-west", load: 0.4 }, | ||
| { id: "node-ap-south-1", url: "ws://localhost:8082", region: "ap-south", load: 0.1 }, | ||
| ]; | ||
|
|
||
| setAvailableNodes(mockNodes); | ||
|
|
||
| // Sorting logic: prioritize lowest load | ||
| const sortedNodes = [...mockNodes].sort((a, b) => a.load - b.load); | ||
| const optimal = sortedNodes[0]; | ||
|
Comment on lines
+33
to
+57
Contributor
There was a problem hiding this comment. Docstring claims "regional proximity" but implementation only sorts by load. Lines 33–35 describe selection "based on regional proximity and server load", but the only criterion at line 56 is 🤖 Prompt for AI Agents |
||
|
|
||
| gameNodeMap.current.set(gameId, optimal); | ||
| setCurrentNode(optimal); | ||
| return optimal; | ||
| } finally { | ||
| setIsScaling(false); | ||
| } | ||
| }, []); | ||
|
Comment on lines
+22
to
+65
Contributor
There was a problem hiding this comment. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
# Confirm DISCOVERY_ENDPOINT / NEXT_PUBLIC_DISCOVERY_URL really has no other consumer.
rg -nP -C2 'DISCOVERY_ENDPOINT|NEXT_PUBLIC_DISCOVERY_URL'Repository: NOVUS-X/XLMate Length of output: 1060 Critical:
Wrap the mock behind 🤖 Prompt for AI Agents
Comment on lines
+30
to
+65
Contributor
There was a problem hiding this comment. Sticky session cache has no invalidation — failover is silently broken.
Add an eviction path: on WebSocket close-with-error / health-check failure, expose an 🤖 Prompt for AI Agents |
||
|
|
||
| const reportLoad = useCallback((nodeId: string, load: number) => { | ||
| setAvailableNodes(prev => | ||
| prev.map(node => node.id === nodeId ? { ...node, load } : node) | ||
| ); | ||
| }, []); | ||
|
|
||
| return ( | ||
| <WebSocketScalingContext.Provider value={{ | ||
| currentNode, | ||
| availableNodes, | ||
| isScaling, | ||
| getOptimalNode, | ||
| reportLoad | ||
| }}> | ||
| {children} | ||
| </WebSocketScalingContext.Provider> | ||
| ); | ||
| } | ||
|
|
||
| export const useWebSocketScaling = () => { | ||
| const context = useContext(WebSocketScalingContext); | ||
| if (!context) { | ||
| throw new Error("useWebSocketScaling must be used within a WebSocketScalingProvider"); | ||
| } | ||
| return context; | ||
| }; | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🧩 Analysis chain
🏁 Script executed:
Repository: NOVUS-X/XLMate
Length of output: 382
🏁 Script executed:
Repository: NOVUS-X/XLMate
Length of output: 90
🏁 Script executed:
Repository: NOVUS-X/XLMate
Length of output: 107
🏁 Script executed:
Repository: NOVUS-X/XLMate
Length of output: 115
🏁 Script executed:
Repository: NOVUS-X/XLMate
Length of output: 40
🏁 Script executed:
# Read main.rs to see the actual code wc -l ./backend/src/socket/main.rsRepository: NOVUS-X/XLMate
Length of output: 90
🏁 Script executed:
# Read the main.rs file cat -n ./backend/src/socket/main.rsRepository: NOVUS-X/XLMate
Length of output: 1904
🏁 Script executed:
Repository: NOVUS-X/XLMate
Length of output: 89
🏁 Script executed:
# Read pubsub.rs to check for env::var usage wc -l ./backend/src/socket/pubsub.rsRepository: NOVUS-X/XLMate
Length of output: 92
🏁 Script executed:
# Read pubsub.rs cat -n ./backend/src/socket/pubsub.rsRepository: NOVUS-X/XLMate
Length of output: 2278
🏁 Script executed:
# Check the full Cargo.toml for context cat -n ./backend/src/socket/Cargo.tomlRepository: NOVUS-X/XLMate
Length of output: 775
dotenvyis added but never initialized —.envwill not be loaded.std::env::var("REDIS_URL")inpubsub.rs(line 14) andstd::env::var("BIND_ADDRESS")inmain.rs(line 16) are read directly without ever callingdotenvy::dotenv(). As a result, this dependency is dead code and operators who placeREDIS_URLorBIND_ADDRESSin a.envfile will silently fall back to the hardcoded defaults (redis://127.0.0.1:6379and127.0.0.1:8080). Either invokedotenvy::dotenv().ok();once at the top ofmain(), or drop the dependency.🛠️ Suggested fix in `main.rs`
#[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { + // Load environment variables from .env if present (no-op in production). + dotenvy::dotenv().ok(); + // Initialize the logger env_logger::init_from_env(env_logger::Env::default().default_filter_or("info"));🤖 Prompt for AI Agents