From d2c68bf7dd5f04fe11a40dce03e84e79487daf79 Mon Sep 17 00:00:00 2001 From: MaxSignal Date: Thu, 8 Jan 2026 02:17:34 +0000 Subject: [PATCH] Implement queue timeout #45 (#71) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Description This PR implements a queue timeout in the lobby server. * After the time configured in `timeout` has elapsed (default: 3 minutes) since the first player joins the queue, matchmaking will start even if the queue has not reached the `players_per_game` target. Any missing slots will be filled with fake players. * If the first queued player leaves and other players remain, the timer is recalculated so that matchmaking starts after the `timeout` duration (default: 3 minutes) from the next player’s join time. (After that, the timer continues to be adjusted in the same way whenever the head of the queue changes.) ### Game Robocraft ### Please confirm - [x] I am the legal owner or represent the owner of all work submitted - [x] I consent to my submission being added to this FOSS project - [x] This PR used LLMs to generate some or all of the code changes Reviewed-on: https://git.ngram.ca/OpenJam/rc-servers/pulls/71 Reviewed-by: NGnius Co-authored-by: MaxSignal Co-committed-by: MaxSignal --- rc_core/src/persist/combat.rs | 1 + rc_core/src/persist/config/cubes_json.rs | 4 + rc_core/src/persist/config/traits.rs | 1 + rc_core/src/persist/multiplayer.rs | 1 + rc_lobby_room/src/lobby.rs | 154 +++++++++++++++++++-- rc_lobby_room/src/main.rs | 2 +- rc_lobby_room/src/operations/join_queue.rs | 2 +- 7 files changed, 149 insertions(+), 16 deletions(-) diff --git a/rc_core/src/persist/combat.rs b/rc_core/src/persist/combat.rs index 585d9c5..0c0a978 100644 --- a/rc_core/src/persist/combat.rs +++ b/rc_core/src/persist/combat.rs @@ -619,6 +619,7 @@ fn default_multiplayer() -> super::MultiplayerConfig { super::MultiplayerConfig { players_per_game: 2, enabled: true, + autostart_after_s: 180, network: super::multiplayer::default_net_conf(), fakes: super::multiplayer::default_fake_users(), battle_arena: super::multiplayer::default_ba_conf(), diff --git a/rc_core/src/persist/config/cubes_json.rs b/rc_core/src/persist/config/cubes_json.rs index 7b66198..deb6f6e 100644 --- a/rc_core/src/persist/config/cubes_json.rs +++ b/rc_core/src/persist/config/cubes_json.rs @@ -403,6 +403,10 @@ impl super::ConfigProvider for CubeConfig { fn is_multiplayer_enabled(&self) -> bool { self.battle.multiplayer.enabled } + + fn multiplayer_autostart_after(&self) -> std::time::Duration { + std::time::Duration::from_secs(self.battle.multiplayer.autostart_after_s) + } fn network_config(&self) -> crate::persist::NetworkConf { self.battle.multiplayer.network.clone() diff --git a/rc_core/src/persist/config/traits.rs b/rc_core/src/persist/config/traits.rs index ae32b28..1d43430 100644 --- a/rc_core/src/persist/config/traits.rs +++ b/rc_core/src/persist/config/traits.rs @@ -27,6 +27,7 @@ pub trait ConfigProvider { fn singleplayer_details(&self) -> SingleplayerConfig; fn players_per_game(&self) -> usize; fn is_multiplayer_enabled(&self) -> bool; + fn multiplayer_autostart_after(&self) -> std::time::Duration; // FIXME don't use serializable types in traits fn network_config(&self) -> crate::persist::NetworkConf; fn maps(&self) -> std::collections::HashMap; diff --git a/rc_core/src/persist/multiplayer.rs b/rc_core/src/persist/multiplayer.rs index ee71339..71465c2 100644 --- a/rc_core/src/persist/multiplayer.rs +++ b/rc_core/src/persist/multiplayer.rs @@ -4,6 +4,7 @@ use serde::{Serialize, Deserialize}; pub struct MultiplayerConfig { pub players_per_game: usize, pub enabled: bool, + pub autostart_after_s: u64, #[serde(default = "default_net_conf")] pub network: NetworkConf, #[serde(default = "default_fake_users")] diff --git a/rc_lobby_room/src/lobby.rs b/rc_lobby_room/src/lobby.rs index 69e3bfb..8702b56 100644 --- a/rc_lobby_room/src/lobby.rs +++ b/rc_lobby_room/src/lobby.rs @@ -40,10 +40,12 @@ struct QueueUser { emitter: polariton_server::events::EventEmitter, player: oj_rc_core::data::player_data::PlayerData, user_id: i32, + enqueued_at: chrono::DateTime, + user: std::sync::Arc + Send + Sync>>, } pub struct QueueHandler { - users_in_queue: tokio::sync::Mutex>>, + users_in_queue: std::sync::Arc>>>, users_per_game: usize, is_enabled: bool, hostname: String, @@ -53,13 +55,15 @@ pub struct QueueHandler { cpu_counter: std::sync::Arc, weapon_guesser: std::sync::Arc, change_strategy: GamemodeChangeStrategy, + autostart_after: std::time::Duration, + autostart_task_started: std::sync::atomic::AtomicBool, } impl QueueHandler { pub fn new(conf: &oj_rc_core::ConfigImpl, game_host: &str, factory: std::sync::Arc, cpu_counter: std::sync::Arc, weapon_guesser: std::sync::Arc,) -> Self { let (domain, port_str) = game_host.split_once(':').expect("Invalid redirect address (must be domain:port)"); Self { - users_in_queue: tokio::sync::Mutex::new(HashMap::new()), + users_in_queue: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())), users_per_game: oj_rc_core::ConfigProvider::<()>::players_per_game(conf), is_enabled: oj_rc_core::ConfigProvider::<()>::is_multiplayer_enabled(conf), hostname: domain.to_owned(), @@ -69,10 +73,122 @@ impl QueueHandler { cpu_counter, weapon_guesser, change_strategy: GamemodeChangeStrategy::from_core(>::server_config(conf).queue_mode), + autostart_after: oj_rc_core::ConfigProvider::<()>::multiplayer_autostart_after(conf), + autostart_task_started: std::sync::atomic::AtomicBool::new(false), } } - async fn enter_match(&self, key: QueueKey, mut players: Vec, user: &(dyn oj_rc_core::persist::user::LobbyUser + Send + Sync)) { + fn ensure_autostart_task_running(&self) { + if self.autostart_task_started.swap(true, std::sync::atomic::Ordering::AcqRel) { + return; + } + + let users_in_queue = self.users_in_queue.clone(); + let users_per_game = self.users_per_game; + let hostname = self.hostname.clone(); + let hostport = self.hostport; + let network_conf = self.network_conf.clone(); + let factory = self.factory.clone(); + let cpu_counter = self.cpu_counter.clone(); + let weapon_guesser = self.weapon_guesser.clone(); + let autostart_after = self.autostart_after; + + tokio::spawn(async move { + loop { + let now = chrono::Utc::now(); + let mut to_start: Vec<(QueueKey, Vec)> = Vec::new(); + let mut next_deadline: Option> = None; + + { + let mut lock = users_in_queue.lock().await; + + let mut empty_keys: Vec = Vec::new(); + let mut expired_keys: Vec = Vec::new(); + + for (key, users) in lock.iter() { + if users.is_empty() { + empty_keys.push(key.clone()); + continue; + } + + // first user is always oldest within a queue + let deadline = users[0].enqueued_at + autostart_after; + if now >= deadline { + expired_keys.push(key.clone()); + } else { + next_deadline = Some(match next_deadline { + Some(d) => if deadline < d { deadline } else { d }, + None => deadline, + }); + } + } + + for k in empty_keys { + lock.remove(&k); + } + + for k in expired_keys { + if let Some(players) = lock.remove(&k) { + if !players.is_empty() { + to_start.push((k, players)); + } + } + } + } + + // start expired queues outside the lock + for (key, players) in to_start { + // choose the oldest queued user's LobbyUser handle + let starter = match players.first() { + Some(p) => p.user.clone(), + None => continue, + }; + + QueueHandler::enter_match_static( + hostname.clone(), + hostport, + network_conf.clone(), + factory.clone(), + cpu_counter.clone(), + weapon_guesser.clone(), + users_per_game, + key, + players, + starter.as_ref().as_ref(), + ).await; + } + + let sleep_dur = if let Some(deadline) = next_deadline { + let now2 = chrono::Utc::now(); + match (deadline - now2).to_std() { + Ok(d) => d, + Err(_) => std::time::Duration::from_secs(0), + } + } else { + std::time::Duration::from_secs(1) + }; + + tokio::time::sleep(sleep_dur).await; + } + }); + } + + async fn enter_match(&self, key: QueueKey, players: Vec, user: &(dyn oj_rc_core::persist::user::LobbyUser + Send + Sync)) { + Self::enter_match_static( + self.hostname.clone(), + self.hostport, + self.network_conf.clone(), + self.factory.clone(), + self.cpu_counter.clone(), + self.weapon_guesser.clone(), + self.users_per_game, + key, + players, + user, + ).await + } + + async fn enter_match_static(hostname: String, hostport: u16, network_conf: crate::data::network::NetworkConfigData, factory: std::sync::Arc, cpu_counter: std::sync::Arc, weapon_guesser: std::sync::Arc, users_per_game: usize, key: QueueKey, mut players: Vec, user: &(dyn oj_rc_core::persist::user::LobbyUser + Send + Sync)) { let guid_str = key.guid(); let game_desc = oj_rc_core::persist::user::GameDescriptor { guid: guid_str.clone(), @@ -100,14 +216,15 @@ impl QueueHandler { display_name: x.player.display_name.clone(), }).collect(); - match user.start_game(game_desc, player_descs, self.factory.as_ref(), &self.cpu_counter, &self.weapon_guesser, &team_picker).await { + match user.start_game(game_desc, player_descs, factory.as_ref(), &cpu_counter, &weapon_guesser, &team_picker).await { Ok(fakes) => { + let missing = users_per_game.saturating_sub(players.len()); let player_datas = players.iter().map(|x| x.player.clone()) - .chain(fakes.players.into_iter().map(|(desc, _emu)| desc)) + .chain(fakes.players.into_iter().take(missing).map(|(desc, _emu)| desc),) .collect(); let enter_battle_ev = crate::events::battle_enter::BattleEnter { - host: self.hostname.clone(), - port: self.hostport, + host: hostname.clone(), + port: hostport, map: key.map.clone(), mode: key.mode, guid: guid_str.clone(), @@ -116,7 +233,7 @@ impl QueueHandler { visibility: Some(key.visibility), auto_heal: key.auto_heal, player_datas, - network_config: self.network_conf.clone(), + network_config: network_conf.clone(), }; let arc_event = std::sync::Arc::new(enter_battle_ev); for player in players.iter() { @@ -144,7 +261,7 @@ impl QueueHandler { } } - pub async fn join_queue(&self, map: String, mode: oj_rc_core::data::game_mode::GameMode, visibility: oj_rc_core::data::game_mode::MapVisibility, auto_heal: bool, user: &(dyn oj_rc_core::persist::user::LobbyUser + Send + Sync), event_emitter: polariton_server::events::EventEmitter) { + pub async fn join_queue(&self, map: String, mode: oj_rc_core::data::game_mode::GameMode, visibility: oj_rc_core::data::game_mode::MapVisibility, auto_heal: bool, user: std::sync::Arc + Send + Sync>>, event_emitter: polariton_server::events::EventEmitter) { if !self.is_enabled { event_emitter.emit(crate::events::enqueue_error::QueueJoinError { code: oj_rc_core::data::error_codes::LobbyReasonCode::NoSuitableLobbyFound as i16, @@ -152,15 +269,21 @@ impl QueueHandler { }); return; } + + self.ensure_autostart_task_running(); + let key = QueueKey { map, mode, visibility, auto_heal, }; - match user.player_data(&self.cpu_counter).await { + let lobby_user = user.as_ref().as_ref(); + match lobby_user.player_data(&self.cpu_counter).await { Ok(player_data) => { let new_player = QueueUser { emitter: event_emitter, player: player_data, - user_id: user.user_id(), + user_id: oj_rc_core::persist::user::LobbyUser::user_id(lobby_user), + enqueued_at: chrono::Utc::now(), + user: user.clone(), }; let mut lock = self.users_in_queue.lock().await; // handle game event change @@ -177,7 +300,10 @@ impl QueueHandler { } else { new_queue_map.insert(key.clone(), users); } + } + for users in new_queue_map.values_mut() { + users.sort_by_key(|u| u.enqueued_at); } *lock = new_queue_map; if count != 0 { @@ -217,7 +343,7 @@ impl QueueHandler { let players = if game_ready { lock.remove(&key) } else { None }; drop(lock); if let Some(players) = players { - self.enter_match(key, players, user).await; + self.enter_match(key, players, lobby_user).await; } }, Err(e) => { @@ -229,8 +355,8 @@ impl QueueHandler { } } - pub async fn leave_queue(&self, user: &(dyn oj_rc_core::persist::user::LobbyUser + Send + Sync)) { - let user_id = user.user_id(); + pub async fn leave_queue(&self, user: std::sync::Arc + Send + Sync>>) { + let user_id = oj_rc_core::persist::user::LobbyUser::user_id(user.as_ref().as_ref()); for queue in self.users_in_queue.lock().await.values_mut() { if let Some((i, _)) = queue.iter().enumerate().find(|(_, user)| user.user_id == user_id) { queue.remove(i); diff --git a/rc_lobby_room/src/main.rs b/rc_lobby_room/src/main.rs index afd4bf4..53143ce 100644 --- a/rc_lobby_room/src/main.rs +++ b/rc_lobby_room/src/main.rs @@ -85,7 +85,7 @@ async fn process_socket(mut socket: net::TcpStream, address: std::net::SocketAdd let ctx = polariton::packet::SerdesContext::from_boxed(Default::default(), enc); server.handle_async_with_channel_join(socket_r, socket_w, user_state.clone(), ctx, chann_tx, chann_rx).await; if let Ok(user_info) = user_state.user() { - queue.leave_queue(user_info.as_ref().as_ref()).await; + queue.leave_queue(user_info.clone()).await; } else { log::debug!("Unauthenticated user disconnected"); } diff --git a/rc_lobby_room/src/operations/join_queue.rs b/rc_lobby_room/src/operations/join_queue.rs index d7dd004..2a9bc23 100644 --- a/rc_lobby_room/src/operations/join_queue.rs +++ b/rc_lobby_room/src/operations/join_queue.rs @@ -42,7 +42,7 @@ impl SimpleOperation for QueueJoinProvider { current_lobby.mode, current_lobby.visibility, current_lobby.auto_heal, - user_info.as_ref().as_ref(), + user_info.clone(), events.to_owned(), ).await; }