jpmschweitzer/settled-reach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(bridge): add Disconnected error variant, fix test race condition

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-02-11 21:15:56 +01:00
co-authored by Claude Opus 4.6
parent 058d0352ab
commit 2cd8786036
4 changed files with 65 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/bridge/local.rs
+1 -1
View File
@@ -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),
}
}
}
server/src/bridge/mod.rs
+13 -2
View File
@@ -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<Res<BridgeResource>>,
mut buffer: ResMut<SnapshotBuffer>,
mut running: ResMut<ServerRunning>,
) {
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),
),
);
server/src/bridge/tcp.rs
+31 -1
View File
@@ -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<Self, BridgeError> {
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<Self, BridgeError> {
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),
}
}
}
server/tests/bridge_tcp.rs
+20 -90
View File
@@ -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");
}