From c754b2ec72b001d4d4e7831674d7ea92c0185326 Mon Sep 17 00:00:00 2001 From: Joshua Higgins Date: Thu, 19 Feb 2026 15:55:56 -0500 Subject: [PATCH] feat: tolerate disconnects/allow for soft reconnects --- README.md | 1 + gameloop.py | 69 ++++++++-- src/main.rs | 28 +++- src/server.rs | 234 ++++++++++++++++++++++++++------- src/tournaments/mod.rs | 2 + src/tournaments/round_robin.rs | 18 +++ src/types.rs | 10 ++ 7 files changed, 299 insertions(+), 63 deletions(-) diff --git a/README.md b/README.md index fd02f8e..b16a276 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ In order to run your AI, you'll need: - Python 3 - `pip install websockets` (Windows) or `pip3 install websockets` (Linux/macOS) - `pip install pip-system-certs` (Windows) or `pip3 install pip-system-certs` (Linux/macOS) +- `pip install wakepy` (Windows) or `pip3 install wakepy` (Linux/macOS) To run the example, run `python gameloop.py` (Windows) or `python3 gameloop.py` (Linux/macOS). diff --git a/gameloop.py b/gameloop.py index 0a60ab0..2dcb642 100644 --- a/gameloop.py +++ b/gameloop.py @@ -1,14 +1,22 @@ import asyncio import websockets +from websockets.exceptions import ConnectionClosed +from wakepy import keep DEFAULT_SERVER_URL = "wss://connect4.abunchofknowitalls.com" +RECONNECT_INTERVAL_SECONDS = 5 +RECONNECT_TIMEOUT_SECONDS = 60 +MAX_RECONNECT_ATTEMPTS = ( + (RECONNECT_TIMEOUT_SECONDS + RECONNECT_INTERVAL_SECONDS - 1) + // RECONNECT_INTERVAL_SECONDS +) from agent import Agent async def gameloop(socket): player = Agent() - while True: # While game is active, continually anticipate messages + while not socket.closed: # While game is active, continually anticipate messages message = (await socket.recv()).split(":") # Receive message from server match message[0]: @@ -37,19 +45,54 @@ async def gameloop(socket): case "ERROR": print(f"{message[0]}: {':'.join(message[1:])}") - await socket.close() - - async def join_server(username, server_url): - async with websockets.connect(server_url, ping_interval=30, ping_timeout=30) as socket: - await socket.send(f"CONNECT:{username}") - await gameloop(socket) + reconnecting = False + reconnect_deadline = None + reconnect_attempt = 0 + + while True: + try: + async with websockets.connect(server_url, ping_interval=30, ping_timeout=30) as socket: + if reconnecting: + await socket.send(f"RECONNECT:{username}") + print("Reconnected to server.") + else: + await socket.send(f"CONNECT:{username}") + + reconnect_deadline = None + reconnect_attempt = 0 + await gameloop(socket) + except (ConnectionClosed, OSError) as error: + print(f"Connection lost ({error}).") + else: + print("Connection closed.") + + now = asyncio.get_running_loop().time() + if reconnect_deadline is None: + reconnect_deadline = now + RECONNECT_TIMEOUT_SECONDS + print(f"Attempting to reconnect every {RECONNECT_INTERVAL_SECONDS} seconds for up to {RECONNECT_TIMEOUT_SECONDS} seconds...") + + remaining = reconnect_deadline - now + if remaining <= 0: + print("Failed to reconnect within 60 seconds. Exiting.") + return + + reconnecting = True + reconnect_attempt += 1 + wait_time = min(RECONNECT_INTERVAL_SECONDS, remaining) + print( + f"Reconnect attempt {reconnect_attempt}/{MAX_RECONNECT_ATTEMPTS}: " + f"retrying in {wait_time:.0f}s " + f"({remaining:.0f}s remaining)." + ) + await asyncio.sleep(wait_time) if __name__ == "__main__": - server_url = ( - input(f"Enter server address [{DEFAULT_SERVER_URL}]: ").strip() - or DEFAULT_SERVER_URL - ) - username = input("Enter username: ") - asyncio.run(join_server(username, server_url)) + with keep.presenting(): + server_url = ( + input(f"Enter server address [{DEFAULT_SERVER_URL}]: ").strip() + or DEFAULT_SERVER_URL + ) + username = input("Enter username: ") + asyncio.run(join_server(username, server_url)) diff --git a/src/main.rs b/src/main.rs index 055cdba..027b74e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -84,6 +84,19 @@ async fn handle_connection( let _ = send(&tx, "ERROR:INVALID:ID:"); } } + "RECONNECT" => { + if parts.len() > 1 { + let requested_username = parts[1].to_string(); + if let Err(e) = + sd.handle_reconnect_cmd(addr, tx.clone(), requested_username).await + { + error!("handle_reconnect: {}", e); + let _ = send(&tx, e.to_string().as_str()); + } + } else { + let _ = send(&tx, "ERROR:INVALID:RECONNECT:"); + } + } "DISCONNECT" => { if let Err(e) = sd.handle_disconnect_cmd(addr, tx.clone()).await { error!("handle_disconnect: {}", e); @@ -288,14 +301,17 @@ async fn handle_connection( if clients_guard.get(&addr).is_some() { let client = clients_guard.get(&addr).unwrap().read().await; let username = client.username.clone(); - if let Some(match_id) = client.current_match { - drop(client); - // TOOD: do a Technical Timeout instead - sd.terminate_match(match_id).await; - } else { - drop(client); + let tournament_guard = sd.tournament.read().await; + if client.current_match.is_some() { + sd.disconnected_clients.write().await.push(username.clone()); + } else if tournament_guard.is_some() { + let tourney = tournament_guard.clone().unwrap(); + if tourney.read().await.contains_player(addr) { + sd.disconnected_clients.write().await.push(username.clone()); + } } + drop(client); drop(clients_guard); sd.clients.write().await.remove(&addr); diff --git a/src/server.rs b/src/server.rs index b4cdb48..b8cc9c9 100644 --- a/src/server.rs +++ b/src/server.rs @@ -4,6 +4,7 @@ use crate::{tournaments::*, types::*, *}; pub struct Server { pub clients: Clients, + pub disconnected_clients: Arc>>, pub usernames: Usernames, pub observers: Observers, pub matches: Matches, @@ -20,6 +21,7 @@ impl Server { pub fn new(admin_password: String, demo_mode: bool) -> Server { Server { clients: Arc::new(RwLock::new(HashMap::new())), + disconnected_clients: Arc::new(RwLock::new(Vec::new())), usernames: Arc::new(RwLock::new(HashMap::new())), observers: Arc::new(RwLock::new(HashMap::new())), matches: Arc::new(RwLock::new(HashMap::new())), @@ -53,9 +55,21 @@ impl Server { ))); } + let mut reconnecting = false; + let disconnected_guard = self.disconnected_clients.read().await; + if disconnected_guard.contains(&requested_username) { + reconnecting = true; + } + let clients_guard = self.clients.read().await; + let mut reconnecting_client = None; for client in clients_guard.values() { if requested_username == client.read().await.username { + if reconnecting { + reconnecting_client = Some(client.clone()); + break; + } + return Err(anyhow::anyhow!(format!( "ERROR:INVALID:ID:{}", requested_username @@ -66,20 +80,141 @@ impl Server { drop(clients_guard); self.remove_observer_from_all_matches(addr).await; - - // not taken self.observers.write().await.remove(&addr); self.usernames.write().await.insert(requested_username.clone(), addr); - self.clients.write().await.insert( - addr.to_string().parse()?, - Arc::new(RwLock::new(Client::new( - requested_username, - tx.clone(), - addr.to_string().parse()?, - ))), - ); - let _ = send(&tx, "CONNECT:ACK"); + + if !reconnecting { + self.clients.write().await.insert( + addr.to_string().parse()?, + Arc::new(RwLock::new(Client::new( + requested_username, + tx.clone(), + addr.to_string().parse()?, + ))), + ); + + return Ok(()); + } + + // reconnecting + self.disconnected_clients.write().await.retain(|name| name != &requested_username); + let client_guard = reconnecting_client.unwrap(); + let mut client = client_guard.write().await; + let old_addr = client.addr; + client.addr = addr; + client.connection = tx.clone(); + // I don't think this will fail + let match_id = client.current_match.unwrap(); + let client_color = client.color; + + drop(client); + + let mut clients_guard = self.clients.write().await; + clients_guard.remove(&old_addr); + clients_guard.insert(addr, client_guard.clone()); + drop(clients_guard); + + let tournament_guard = self.tournament.read().await; + if tournament_guard.is_some() { + let tourney = tournament_guard.clone().unwrap(); + tourney.write().await.inform_reconnect(old_addr, addr); + } + drop(tournament_guard); + + let matches_guard = self.matches.read().await; + let mut the_match = matches_guard.get(&match_id).unwrap().write().await; + if the_match.demo_mode { + drop(the_match); + drop(matches_guard); + self.terminate_match(match_id).await; + return Ok(()); + } else { + the_match.ledger.clear(); + the_match.board = vec![vec![Color::None; 6]; 7]; + let opponent_addr = if the_match.player1 == addr { + the_match.player2 + } else { + the_match.player1 + }; + + if the_match.wait_thread.is_some() { + the_match.wait_thread.as_ref().unwrap().abort(); + } + + if the_match.timeout_thread.is_some() { + the_match.timeout_thread.as_ref().unwrap().abort(); + } + + let clients_guard = self.clients.read().await; + let opponent = clients_guard.get(&opponent_addr).unwrap().read().await; + let _ = send(&opponent.connection, "GAME:TERMINATED"); + let _ = send( + &tx, + &format!("GAME:START:{}", bool::from(client_color) as u8), + ); + let _ = send( + &opponent.connection, + &format!("GAME:START:{}", bool::from(opponent.color) as u8), + ); + } + + Ok(()) + } + + pub async fn handle_reconnect_cmd( + &self, + addr: SocketAddr, + tx: UnboundedSender, + requested_username: String, + ) -> Result<(), anyhow::Error> { + let clients_guard = self.clients.read().await; + let disconnected_guard = self.disconnected_clients.read().await; + let mut found_client = None; + + for client in clients_guard.values() { + if requested_username == client.read().await.username { + if disconnected_guard.contains(&requested_username) { + found_client = Some(client.clone()); + } + break; + } + } + + drop(clients_guard); + drop(disconnected_guard); + + if let Some(client_guard) = found_client { + self.disconnected_clients.write().await.retain(|name| name != &requested_username); + let mut client = client_guard.write().await; + let old_addr = client.addr; + client.addr = addr; + client.connection = tx.clone(); + + let mut clients_guard = self.clients.write().await; + clients_guard.remove(&old_addr); + clients_guard.insert(addr, client_guard.clone()); + drop(clients_guard); + + let _ = send(&tx, "RECONNECT:ACK"); + + let matches_guard = self.matches.read().await; + let the_match = matches_guard.get(&client.current_match.unwrap()).unwrap().read().await; + + let last = the_match.ledger.last(); + if last.is_some() && last.unwrap().0 != client.color { + let _ = send( + &tx, + &format!("OPPONENT:{}", the_match.ledger.last().unwrap().1), + ); + } + } else { + return Err(anyhow::anyhow!(format!( + "ERROR:INVALID:RECONNECT:{}", + requested_username + ))); + } + Ok(()) } @@ -219,26 +354,20 @@ impl Server { let matches_guard = self.matches.read().await; let current_match = matches_guard.get(&client.current_match.unwrap()).unwrap().read().await; - let opponent_addr = if addr == current_match.player1 { - current_match.player2 - } else { - current_match.player1 - }; + let opponent = { + let mut result = None; - let opponent_connection = if current_match.demo_mode { - None - } else if addr == current_match.player1 { - Some(clients_guard.get(¤t_match.player2).unwrap().read().await.connection.clone()) - } else { - Some(clients_guard.get(¤t_match.player1).unwrap().read().await.connection.clone()) - }; + if !current_match.demo_mode { + let opponent_addr = if addr == current_match.player1 { + current_match.player2 + } else { + current_match.player1 + }; - let opponent_username = if current_match.demo_mode { - SERVER_PLAYER_USERNAME.to_string() - } else if addr == current_match.player1 { - clients_guard.get(¤t_match.player2).unwrap().read().await.username.clone() - } else { - clients_guard.get(¤t_match.player1).unwrap().read().await.username.clone() + result = Some(clients_guard.get(&opponent_addr).cloned().unwrap()); + } + + result }; // Check if it's their move @@ -284,9 +413,17 @@ impl Server { self.terminate_match(current_match_id).await; tx.send(Message::Close(None))?; } else { + let opponent = opponent.unwrap(); + let mut opponent = opponent.write().await; + let _ = send(&tx, "GAME:LOSS"); - let _ = send(&opponent_connection.unwrap(), "GAME:WINS"); - self.broadcast_message(&viewers, &format!("GAME:WIN:{}", opponent_username)).await; + let _ = send(&opponent.connection, "GAME:WINS"); + self.broadcast_message(&viewers, &format!("GAME:WIN:{}", opponent.username)).await; + + opponent.current_match = None; + opponent.color = Color::None; + let opponent_addr = opponent.addr; + drop(opponent); let mut clients_guard = self.clients.write().await; let mut client = clients_guard.get_mut(&addr).unwrap().write().await; @@ -294,11 +431,6 @@ impl Server { client.color = Color::None; drop(client); - let mut opponent = clients_guard.get_mut(&opponent_addr).unwrap().write().await; - opponent.current_match = None; - opponent.color = Color::None; - drop(opponent); - let mut tournament_guard = self.tournament.write().await; let tourney = tournament_guard.as_mut().unwrap(); tourney.write().await.inform_winner(opponent_addr, false); @@ -327,13 +459,17 @@ impl Server { if winner == client.color { let _ = send(&tx, "GAME:WINS"); if !current_match.demo_mode { - let _ = send(&opponent_connection.as_ref().unwrap(), "GAME:LOSS"); + let opponent = opponent.clone().unwrap(); + let opponent = opponent.read().await; + let _ = send(&opponent.connection, "GAME:LOSS"); } viewer_messages.push(format!("GAME:WIN:{}", client.username)); } else if filled { let _ = send(&tx, "GAME:DRAW"); if !current_match.demo_mode { - let _ = send(&opponent_connection.as_ref().unwrap(), "GAME:DRAW"); + let opponent = opponent.clone().unwrap(); + let opponent = opponent.read().await; + let _ = send(&opponent.connection, "GAME:DRAW"); } viewer_messages.push("GAME:DRAW".to_string()); } @@ -354,7 +490,8 @@ impl Server { drop(client); if !is_demo_mode { - let mut opponent = clients_guard.get(&opponent_addr).unwrap().write().await; + let opponent = opponent.clone().unwrap(); + let mut opponent = opponent.write().await; opponent.current_match = None; opponent.color = Color::None; drop(opponent); @@ -376,7 +513,9 @@ impl Server { } else if self.tournament.read().await.is_none() { let _ = send(&tx, "TOURNAMENT:END"); if !is_demo_mode { - let _ = send(&opponent_connection.unwrap(), "TOURNAMENT:END"); + let opponent = opponent.clone().unwrap(); + let opponent = opponent.read().await; + let _ = send(&opponent.connection, "TOURNAMENT:END"); } } @@ -405,7 +544,7 @@ impl Server { let demo_move = random_move(¤t_match.board); let no_winner = winner == Color::None && !filled; let observers = self.observers.clone(); - let opp_connection_move = opponent_connection.clone(); + let opponent_move = opponent.clone(); let client_tx = tx.clone(); if current_match.demo_mode { current_match.ledger.push((!client.color, demo_move, Instant::now())); @@ -416,8 +555,10 @@ impl Server { tokio::time::sleep(tokio::time::Duration::from_millis(adjusted_waiting as u64)).await; if !demo_mode && no_winner { + let opponent = opponent_move.unwrap(); + let opponent = opponent.read().await; let _ = send( - &opp_connection_move.as_ref().unwrap(), + &opponent.connection.clone(), &format!("OPPONENT:{}", column), ); } @@ -449,6 +590,7 @@ impl Server { let client_addr = addr.clone(); let observers = self.observers.clone(); let viewers = current_match.viewers.clone(); + let opponent_move = opponent.clone(); current_match.timeout_thread = Some(tokio::spawn(async move { if demo_mode { return; @@ -463,7 +605,10 @@ impl Server { if the_match.ledger.len() == ledger_size { // forfeit the match let _ = send(&client_tx, "GAME:WINS"); - let _ = send(&opponent_connection.unwrap(), "GAME:LOSS"); + let opponent = opponent_move.clone().unwrap(); + let opponent = opponent.read().await; + let _ = send(&opponent.connection, "GAME:LOSS"); + drop(opponent); broadcast_message( &observers, &viewers, @@ -477,7 +622,8 @@ impl Server { client.color = Color::None; drop(client); - let mut opponent = clients_guard.get_mut(&opponent_addr).unwrap().write().await; + let opponent = opponent_move.unwrap(); + let mut opponent = opponent.write().await; opponent.current_match = None; opponent.color = Color::None; drop(opponent); diff --git a/src/tournaments/mod.rs b/src/tournaments/mod.rs index d76d1d7..62d95f4 100644 --- a/src/tournaments/mod.rs +++ b/src/tournaments/mod.rs @@ -16,6 +16,8 @@ pub trait Tournament { async fn start(&mut self, server: &Server); async fn cancel(&mut self, server: &Server); fn inform_winner(&mut self, winner: SocketAddr, is_tie: bool); + fn inform_reconnect(&mut self, old_addr: SocketAddr, new_addr: SocketAddr); + fn contains_player(&self, addr: SocketAddr) -> bool; fn is_completed(&self) -> bool; fn get_type(&self) -> String; } diff --git a/src/tournaments/round_robin.rs b/src/tournaments/round_robin.rs index 8a4d508..18c814f 100644 --- a/src/tournaments/round_robin.rs +++ b/src/tournaments/round_robin.rs @@ -111,6 +111,24 @@ impl Tournament for RoundRobin { } } + fn inform_reconnect(&mut self, old_addr: SocketAddr, new_addr: SocketAddr) { + for (_, (player_addr, _)) in self.players.iter_mut() { + if *player_addr == old_addr { + *player_addr = new_addr; + break; + } + } + } + + fn contains_player(&self, addr: SocketAddr) -> bool { + for (_, (player_addr, _)) in self.players.iter() { + if *player_addr == addr { + return true; + } + } + false + } + async fn next(&mut self, server: &Server) { if self.is_completed { return; diff --git a/src/types.rs b/src/types.rs index 86dbcfb..fef36f7 100644 --- a/src/types.rs +++ b/src/types.rs @@ -24,6 +24,16 @@ impl ops::Not for Color { } } +impl From for bool { + fn from(color: Color) -> bool { + match color { + Color::Red => true, + Color::Yellow => false, + Color::None => panic!("Cannot convert Color::None to bool"), + } + } +} + #[derive(Clone)] pub struct Client { pub username: String,