From 2cd8786036f48701178a7f0427048aee2a66591e Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Wed, 11 Feb 2026 21:15:56 +0100 Subject: [PATCH] fix(bridge): add Disconnected error variant, fix test race condition Replace string-matching disconnect detection with explicit BridgeError::Disconnected variant. Add TcpBridge::accept_on(listener) that takes a pre-bound TcpListener, eliminating the 100ms sleep hack in TCP tests. Send errors now also trigger ServerRunning=false. Add trace logging to generate_snapshot for entity count visibility. Co-Authored-By: Claude Opus 4.6 --- server/src/bridge/local.rs | 2 +- server/src/bridge/mod.rs | 15 ++++- server/src/bridge/tcp.rs | 32 ++++++++++- server/tests/bridge_tcp.rs | 110 +++++++------------------------------ 4 files changed, 65 insertions(+), 94 deletions(-) diff --git a/server/src/bridge/local.rs b/server/src/bridge/local.rs index 62c36f873..5e73649a6 100644 --- a/server/src/bridge/local.rs +++ b/server/src/bridge/local.rs @@ -99,7 +99,7 @@ impl SimBridge for LocalBridge { tracing::trace!("received {} inputs", inputs.len()); Ok(inputs) } - None => Err(BridgeError::Transport("client disconnected (EOF)".into())), + None => Err(BridgeError::Disconnected), } } } diff --git a/server/src/bridge/mod.rs b/server/src/bridge/mod.rs index 44e6a7fbd..702bd608d 100644 --- a/server/src/bridge/mod.rs +++ b/server/src/bridge/mod.rs @@ -23,6 +23,8 @@ pub enum BridgeError { Io(#[from] std::io::Error), #[error("transport error: {0}")] Transport(String), + #[error("client disconnected")] + Disconnected, } /// Abstracts transport layer (D-020) @@ -86,6 +88,11 @@ pub fn generate_snapshot( kind, }); } + tracing::trace!( + "generate_snapshot: tick={}, entities={}", + time.tick, + visible.len() + ); buffer.snapshot = Some(ObserverSnapshot { tick: time.tick, entities: visible, @@ -105,7 +112,7 @@ pub fn receive_bridge_inputs( input_queue.push(input); } } - Err(BridgeError::Transport(ref msg)) if msg.contains("disconnected") => { + Err(BridgeError::Disconnected) => { tracing::info!("Client disconnected, shutting down"); running.0 = false; } @@ -119,11 +126,13 @@ pub fn receive_bridge_inputs( pub fn send_bridge_snapshot( bridge: Option>, mut buffer: ResMut, + mut running: ResMut, ) { let Some(bridge) = bridge else { return }; if let Some(snapshot) = buffer.snapshot.take() { if let Err(e) = bridge.send_snapshot(&snapshot) { tracing::error!("Bridge send error: {}", e); + running.0 = false; } } } @@ -150,7 +159,9 @@ impl Plugin for BridgePlugin { Update, ( receive_bridge_inputs.before(crate::simulation::input::process_player_input), - generate_snapshot.after(crate::simulation::movement::validate_movement), + generate_snapshot + .after(crate::simulation::movement::validate_movement) + .before(crate::simulation::time::advance_tick), send_bridge_snapshot.after(generate_snapshot), ), ); diff --git a/server/src/bridge/tcp.rs b/server/src/bridge/tcp.rs index 02c4b5a83..72781e96e 100644 --- a/server/src/bridge/tcp.rs +++ b/server/src/bridge/tcp.rs @@ -56,6 +56,36 @@ impl TcpBridge { }) } + /// Server-side: accept one connection on an existing TcpListener. + /// Avoids race conditions in tests by separating bind from accept. + pub fn accept_on(listener: TcpListener) -> Result { + let local_addr = listener + .local_addr() + .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?; + + tracing::info!("TcpBridge accepting on {}", local_addr); + + let (stream, peer_addr) = listener + .accept() + .map_err(|e| BridgeError::Transport(format!("failed to accept connection: {}", e)))?; + + tracing::info!( + "TcpBridge accepted connection from {} on {}", + peer_addr, + local_addr + ); + + let reader_stream = stream.try_clone().map_err(|e| { + BridgeError::Transport(format!("failed to clone stream for reader: {}", e)) + })?; + + Ok(Self { + reader: Mutex::new(BufReader::new(reader_stream)), + writer: Mutex::new(BufWriter::new(stream)), + local_addr, + }) + } + /// Client-side: connect to TCP address (for tests). pub fn connect(addr: &str) -> Result { tracing::info!("TcpBridge connecting to {}", addr); @@ -108,7 +138,7 @@ impl SimBridge for TcpBridge { tracing::trace!("received {} inputs", inputs.len()); Ok(inputs) } - None => Err(BridgeError::Transport("client disconnected (EOF)".into())), + None => Err(BridgeError::Disconnected), } } } diff --git a/server/tests/bridge_tcp.rs b/server/tests/bridge_tcp.rs index ae1927b9a..9f5bdf97a 100644 --- a/server/tests/bridge_tcp.rs +++ b/server/tests/bridge_tcp.rs @@ -5,36 +5,17 @@ use settled_reach_server::bridge::tcp::TcpBridge; use settled_reach_server::bridge::types::*; use settled_reach_server::bridge::SimBridge; use std::net::{TcpListener, TcpStream}; -use std::sync::mpsc; use std::thread; -use std::time::Duration; - -/// Helper to bind a TCP listener and return its address. -/// This allows tests to discover the OS-assigned port before calling TcpBridge::accept(). -fn bind_listener() -> (TcpListener, std::net::SocketAddr) { - let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); - let addr = listener.local_addr().expect("failed to get local address"); - (listener, addr) -} #[test] fn snapshot_roundtrip_over_tcp() { - // Pre-bind listener to discover the port - let (listener, server_addr) = bind_listener(); - let addr_str = server_addr.to_string(); + // Bind listener first — port is guaranteed ready before spawning threads + let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); + let server_addr = listener.local_addr().expect("failed to get local address"); - // Channel to signal when server is ready to accept - let (ready_tx, ready_rx) = mpsc::channel(); - - // Server thread: accept connection and send snapshot + // Server thread: accept on pre-bound listener and send snapshot let server_handle = thread::spawn(move || { - // Drop the pre-bound listener since TcpBridge::accept will bind its own - drop(listener); - - // Signal we're about to accept - ready_tx.send(()).expect("failed to send ready signal"); - - let bridge = TcpBridge::accept(&addr_str).expect("failed to bind and accept"); + let bridge = TcpBridge::accept_on(listener).expect("failed to accept"); let snapshot = ObserverSnapshot { tick: 42, @@ -52,15 +33,7 @@ fn snapshot_roundtrip_over_tcp() { .expect("failed to send snapshot"); }); - // Wait for server thread to start - ready_rx - .recv_timeout(Duration::from_secs(2)) - .expect("server did not become ready"); - - // Give server time to bind and call accept() - thread::sleep(Duration::from_millis(100)); - - // Client: connect and receive snapshot + // Client: connect and receive snapshot (no sleep needed — listener already bound) let stream = TcpStream::connect(server_addr).expect("failed to connect"); let mut reader = std::io::BufReader::new(stream); @@ -82,22 +55,11 @@ fn snapshot_roundtrip_over_tcp() { #[test] fn input_roundtrip_over_tcp() { - // Pre-bind listener to discover the port - let (listener, server_addr) = bind_listener(); - let addr_str = server_addr.to_string(); + let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); + let server_addr = listener.local_addr().expect("failed to get local address"); - // Channel to signal when server is ready to accept - let (ready_tx, ready_rx) = mpsc::channel(); - - // Server thread: accept connection and receive inputs let server_handle = thread::spawn(move || { - // Drop the pre-bound listener since TcpBridge::accept will bind its own - drop(listener); - - // Signal we're about to accept - ready_tx.send(()).expect("failed to send ready signal"); - - let bridge = TcpBridge::accept(&addr_str).expect("failed to bind and accept"); + let bridge = TcpBridge::accept_on(listener).expect("failed to accept"); let inputs = bridge.receive_inputs().expect("failed to receive inputs"); @@ -108,14 +70,6 @@ fn input_roundtrip_over_tcp() { inputs }); - // Wait for server thread to start - ready_rx - .recv_timeout(Duration::from_secs(2)) - .expect("server did not become ready"); - - // Give server time to bind and call accept() - thread::sleep(Duration::from_millis(100)); - // Client: connect and send inputs let stream = TcpStream::connect(server_addr).expect("failed to connect"); let mut writer = std::io::BufWriter::new(stream); @@ -139,7 +93,6 @@ fn input_roundtrip_over_tcp() { let received_inputs = server_handle.join().expect("server thread panicked"); - // Verify actions survived the round-trip match &received_inputs[0].action { PlayerAction::MoveNorth => {} _ => panic!("expected MoveNorth action"), @@ -152,50 +105,27 @@ fn input_roundtrip_over_tcp() { #[test] fn tcp_bridge_eof_returns_error() { - // Pre-bind listener to discover the port - let (listener, server_addr) = bind_listener(); - let addr_str = server_addr.to_string(); + let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); + let server_addr = listener.local_addr().expect("failed to get local address"); - // Channel to signal when server is ready to accept - let (ready_tx, ready_rx) = mpsc::channel(); - - // Server thread: accept connection and receive EOF let server_handle = thread::spawn(move || { - // Drop the pre-bound listener since TcpBridge::accept will bind its own - drop(listener); + let bridge = TcpBridge::accept_on(listener).expect("failed to accept"); - // Signal we're about to accept - ready_tx.send(()).expect("failed to send ready signal"); - - let bridge = TcpBridge::accept(&addr_str).expect("failed to bind and accept"); - - // Attempt to receive inputs - should get EOF error let result = bridge.receive_inputs(); assert!(result.is_err(), "expected error on EOF"); - match result { - Err(settled_reach_server::bridge::BridgeError::Transport(msg)) => { - assert!( - msg.contains("disconnected") || msg.contains("EOF"), - "expected disconnect/EOF error, got: {}", - msg - ); - } - _ => panic!("expected Transport error with disconnect/EOF message"), - } + assert!( + matches!( + result, + Err(settled_reach_server::bridge::BridgeError::Disconnected) + ), + "expected Disconnected error" + ); }); - // Wait for server thread to start - ready_rx - .recv_timeout(Duration::from_secs(2)) - .expect("server did not become ready"); - - // Give server time to bind and call accept() - thread::sleep(Duration::from_millis(100)); - // Client: connect and immediately disconnect without sending data let stream = TcpStream::connect(server_addr).expect("failed to connect"); - drop(stream); // Close connection immediately + drop(stream); server_handle.join().expect("server thread panicked"); }