diff --git a/server/src/atlas/plugin.rs b/server/src/atlas/plugin.rs index 2124d8f1c..8ae60e2b1 100644 --- a/server/src/atlas/plugin.rs +++ b/server/src/atlas/plugin.rs @@ -94,7 +94,11 @@ fn serve_atlas_requests( let reader = city_reader.as_ref().map(|r| &r.0); let params_reader = body_params_reader.as_ref().map(|r| &r.0); let pending: Vec<_> = requests.0.drain(..).collect(); - for req in pending { + // D-254 §2: 1:1, in-order request->response — the connection id rides + // alongside the request untouched by handle_atlas_request (which has no + // notion of connections) and is re-attached to the response so the + // bridge's send_atlas_responses routes it back to only that connection. + for (conn_id, req) in pending { let resp = match resolver.as_ref() { Some(r) => handle_atlas_request( &req, @@ -116,7 +120,7 @@ fn serve_atlas_requests( region_grid: None, }, }; - responses.0.push(resp); + responses.0.push((conn_id, resp)); } } @@ -133,7 +137,7 @@ fn serve_star_map_requests( return; } let pending: Vec<_> = requests.0.drain(..).collect(); - for req in pending { + for (conn_id, req) in pending { let resp = match path.as_ref() { Some(p) => handle_star_map_request(&req, &p.0), None => crate::atlas::atlas_data_proxy::StarMapResponse { @@ -143,7 +147,7 @@ fn serve_star_map_requests( data: None, }, }; - responses.0.push(resp); + responses.0.push((conn_id, resp)); } } @@ -159,8 +163,10 @@ fn serve_city_names_requests( } let reader = city_reader.as_ref().map(|r| &r.0); let pending: Vec<_> = requests.0.drain(..).collect(); - for req in pending { - responses.0.push(handle_city_names_request(&req, reader)); + for (conn_id, req) in pending { + responses + .0 + .push((conn_id, handle_city_names_request(&req, reader))); } } @@ -760,6 +766,7 @@ mod tests { use super::*; use crate::atlas::gen_queue::{GenPriority, GenWorkItem}; use crate::atlas::road_graph::{RoadEdge, RoadNode, RoadNodeKind}; + use crate::bridge::ConnectionId; use crate::seed::SeedChain; use crate::simulation::generator::{ ArrangementPattern, AttractorType, FoundingOrientation, MaintenanceAuthority, @@ -834,10 +841,13 @@ mod tests { use crate::atlas::layer_proxy::AtlasLayerRequest; let mut world = World::new(); - world.insert_resource(AtlasRequestBuffer(vec![AtlasLayerRequest { - body_id: "GJ1c".to_string(), - up_to: CascadeLayer::Topography, - }])); + world.insert_resource(AtlasRequestBuffer(vec![( + ConnectionId(0), + AtlasLayerRequest { + body_id: "GJ1c".to_string(), + up_to: CascadeLayer::Topography, + }, + )])); world.insert_resource(AtlasResponseBuffer::default()); world.insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY)); world.insert_resource(GenerationQueue::with_threads(1)); @@ -849,9 +859,13 @@ mod tests { let responses = world.resource::(); assert_eq!(responses.0.len(), 1, "request should produce one response"); - assert_eq!(responses.0[0].body_id, "GJ1c"); + assert_eq!(responses.0[0].0, ConnectionId(0), "connection id preserved"); + assert_eq!(responses.0[0].1.body_id, "GJ1c"); // No resolver wired → Error status (exercises the drain + push path). - assert!(matches!(responses.0[0].status, AtlasLayerStatus::Error(_))); + assert!(matches!( + responses.0[0].1.status, + AtlasLayerStatus::Error(_) + )); // The request buffer was drained. assert!(world.resource::().0.is_empty()); } @@ -869,9 +883,10 @@ mod tests { std::fs::write(&path, r#"{"_meta": {}, "nodes": [], "edges": []}"#).unwrap(); let mut world = World::new(); - world.insert_resource(StarMapRequestBuffer(vec![StarMapRequest { - star_map: true, - }])); + world.insert_resource(StarMapRequestBuffer(vec![( + ConnectionId(0), + StarMapRequest { star_map: true }, + )])); world.insert_resource(StarMapResponseBuffer::default()); world.insert_resource(StarMapDataPath(path.clone())); @@ -881,7 +896,8 @@ mod tests { let responses = world.resource::(); assert_eq!(responses.0.len(), 1); - assert_eq!(responses.0[0].status, StarMapStatus::Ready); + assert_eq!(responses.0[0].0, ConnectionId(0), "connection id preserved"); + assert_eq!(responses.0[0].1.status, StarMapStatus::Ready); assert!(world.resource::().0.is_empty()); let _ = std::fs::remove_file(&path); @@ -894,9 +910,10 @@ mod tests { use crate::atlas::atlas_data_proxy::{StarMapRequest, StarMapStatus}; let mut world = World::new(); - world.insert_resource(StarMapRequestBuffer(vec![StarMapRequest { - star_map: true, - }])); + world.insert_resource(StarMapRequestBuffer(vec![( + ConnectionId(0), + StarMapRequest { star_map: true }, + )])); world.insert_resource(StarMapResponseBuffer::default()); // No StarMapDataPath resource. @@ -906,7 +923,7 @@ mod tests { let responses = world.resource::(); assert_eq!(responses.0.len(), 1); - assert!(matches!(responses.0[0].status, StarMapStatus::Error(_))); + assert!(matches!(responses.0[0].1.status, StarMapStatus::Error(_))); } /// T-949b: without `CityContextReaderResource` wired, the serve system @@ -917,10 +934,13 @@ mod tests { use crate::atlas::atlas_data_proxy::{CityNamesRequest, CityNamesStatus}; let mut world = World::new(); - world.insert_resource(CityNamesRequestBuffer(vec![CityNamesRequest { - city_names: true, - body_id: "GJ1c".to_string(), - }])); + world.insert_resource(CityNamesRequestBuffer(vec![( + ConnectionId(0), + CityNamesRequest { + city_names: true, + body_id: "GJ1c".to_string(), + }, + )])); world.insert_resource(CityNamesResponseBuffer::default()); // No CityContextReaderResource. @@ -930,8 +950,9 @@ mod tests { let responses = world.resource::(); assert_eq!(responses.0.len(), 1); - assert_eq!(responses.0[0].body_id, "GJ1c"); - assert!(matches!(responses.0[0].status, CityNamesStatus::Error(_))); + assert_eq!(responses.0[0].0, ConnectionId(0), "connection id preserved"); + assert_eq!(responses.0[0].1.body_id, "GJ1c"); + assert!(matches!(responses.0[0].1.status, CityNamesStatus::Error(_))); assert!(world.resource::().0.is_empty()); } diff --git a/server/src/bridge/mod.rs b/server/src/bridge/mod.rs index 8bd2e4ef2..d285fba8e 100644 --- a/server/src/bridge/mod.rs +++ b/server/src/bridge/mod.rs @@ -10,6 +10,7 @@ use crate::atlas::atlas_data_proxy::{ CityNamesRequest, CityNamesResponse, StarMapRequest, StarMapResponse, }; use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse}; +use crate::bridge::tcp::TcpBridge; pub mod debug; pub mod framing; @@ -190,45 +191,247 @@ pub trait SimBridge: Send + Sync { fn send_city_names_response(&self, resp: &CityNamesResponse) -> Result<(), BridgeError>; } -/// BridgeResource: Bevy Resource wrapper for SimBridge trait object -#[derive(Resource)] +/// Identifies one connection for response-tagging and role-lookup purposes +/// (D-254 §2, T-1130). Assigned at accept time by `BridgeResource`; never +/// reused within a server process lifetime (monotonic counter), so a stale +/// id from a disconnected connection can never collide with a live one. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ConnectionId(pub u64); + +/// One live connection: its transport, role, and (for readers) a violation +/// strike counter (D-254 §2 — "log + drop on first offense, disconnect on +/// repeated violations"). +struct Connection { + id: ConnectionId, + bridge: Box, + /// Not read today — `player`/`readers` (which list a `Connection` lives + /// in) already fully determines role-gated behavior in this ticket's + /// scope. Kept because D-254 §6's future `TradingReader` widening wants + /// a role-keyed per-verb match (`Player | Reader | TradingReader`) on + /// exactly this field rather than a third top-level `Vec` — carrying it + /// now avoids a struct-shape change alongside that later widening. + #[allow(dead_code)] + role: ConnectionRole, + /// Count of role-violating frames seen from this connection (e.g. a + /// Reader sending `Vec`). Always 0 for `Player` — the + /// violation path only applies to non-Player roles. + violation_strikes: u32, +} + +/// Strikes tolerated before a violating reader is disconnected (D-254 §2). +/// First offense logs + drops the frame; this is the ceiling before the +/// connection itself is torn down. Deliberately small — a well-behaved +/// reader client sends zero disallowed frames ever, so any nonzero count is +/// already a bug or hostile probe, not normal traffic. +const READER_VIOLATION_DISCONNECT_THRESHOLD: u32 = 3; + +/// Per-reader per-tick inbound frame cap (D-254 §2) — lower than the Player +/// cap (`MAX_INBOUND_FRAMES_PER_TICK`, 64): a reader has no legitimate reason +/// to send that volume of atlas/star-map/city-names requests in one 50ms +/// tick. Cheap insurance against a runaway or misbehaving companion client; +/// does not affect determinism either way (reader frames never reach +/// InputQueue/SimRng regardless of how many are drained). +const MAX_READER_INBOUND_FRAMES_PER_TICK: usize = 8; + +/// BridgeResource: Bevy Resource holding the server's connection set +/// (D-254 §2, T-1130). +/// +/// Scoped honestly as **0-1 Player + 0-N Readers** — not general N-player +/// (D-009's separate, larger, out-of-scope ambition). `player` is the +/// original single connection this resource used to wrap directly; +/// `readers` is new. `pending` holds accepted-but-not-yet-handshaken +/// connections (see `tcp::PendingConnection`) — polled non-blockingly each +/// tick by `main.rs`'s accept-loop system until they either promote into +/// `readers` (or, in the anomalous case of a second Player attempt, get +/// cleanly rejected — see `main.rs`) or fail and are dropped. +#[derive(Resource, Default)] pub struct BridgeResource { - inner: Box, + player: Option, + readers: Vec, + next_id: u64, } impl BridgeResource { + /// Construct a `BridgeResource` with `bridge` installed as the Player + /// connection. This is the pre-D-254 constructor signature, preserved + /// byte-for-byte so every existing call site (`main.rs`, and the + /// single-connection test suites in `server/tests/`) needs no change — + /// "install this bridge" always meant "install the Player" before + /// readers existed, and still does when called this way. pub fn new(bridge: impl SimBridge + 'static) -> Self { - Self { - inner: Box::new(bridge), + let mut resource = Self::default(); + resource.insert_player(bridge); + resource + } + + /// Install `bridge` as the Player connection, assigning it the next + /// `ConnectionId`. Overwrites any existing Player connection (there is + /// never more than one — `main.rs`'s accept-loop rejects a second Player + /// attempt before calling this). + pub fn insert_player(&mut self, bridge: impl SimBridge + 'static) -> ConnectionId { + let id = ConnectionId(self.next_id); + self.next_id += 1; + self.player = Some(Connection { + id, + bridge: Box::new(bridge), + role: ConnectionRole::Player, + violation_strikes: 0, + }); + id + } + + /// Install `bridge` as a new Reader connection, assigning it the next + /// `ConnectionId`. + pub fn insert_reader(&mut self, bridge: impl SimBridge + 'static) -> ConnectionId { + let id = ConnectionId(self.next_id); + self.next_id += 1; + self.readers.push(Connection { + id, + bridge: Box::new(bridge), + role: ConnectionRole::Reader, + violation_strikes: 0, + }); + id + } + + /// True if a Player connection is currently installed (D-254 §2: used by + /// the accept-loop to cleanly reject a second Player attempt instead of + /// silently starving it the way the pre-D-254 single-`accept()` did). + pub fn has_player(&self) -> bool { + self.player.is_some() + } + + /// Number of currently-installed Reader connections (test/observability + /// helper, D-254 §2 — the 0-N reader count this ticket's scope is built + /// around). + pub fn reader_count(&self) -> usize { + self.readers.len() + } + + /// `ConnectionId`s of every currently-installed reader, in insertion + /// order (test/observability helper, D-254 §2). + pub fn reader_ids(&self) -> Vec { + self.readers.iter().map(|c| c.id).collect() + } + + /// Remove and drop the Player connection (its `TcpBridge`/socket is + /// dropped, closing the TCP connection). + pub fn remove_player(&mut self) { + self.player = None; + } + + /// Remove and drop the reader with the given id, if present. A no-op if + /// the id doesn't match any current reader (already removed, or was + /// never a reader — e.g. the Player's own id). + pub fn remove_reader(&mut self, id: ConnectionId) { + self.readers.retain(|c| c.id != id); + } + + /// Send the protocol handshake on the Player connection, if any. + /// Used only by `main.rs`'s original single-connection startup path — + /// reader connections send their own handshake as part of + /// `tcp::PendingConnection`'s state machine, not through this resource. + pub fn send_handshake(&self) -> Result<(), BridgeError> { + match &self.player { + Some(c) => c.bridge.send_handshake(), + None => Err(BridgeError::Transport("no player connection".into())), } } - pub fn send_handshake(&self) -> Result<(), BridgeError> { - self.inner.send_handshake() - } - + /// Receive the Player's startup message. Used only by `main.rs`'s + /// original single-connection startup path (the first accept, before + /// the tick loop and its non-blocking accept-loop begin). pub fn receive_startup(&self) -> Result { - self.inner.receive_startup() + match &self.player { + Some(c) => c.bridge.receive_startup(), + None => Err(BridgeError::Transport("no player connection".into())), + } } + /// Send an `ObserverSnapshot` to the Player connection ONLY (D-254 §2 — + /// structural enforcement: this method has no reader-facing counterpart + /// at all, so a future edit cannot accidentally start broadcasting + /// snapshots to readers by forgetting a role check — there is no code + /// path here that could reach a reader's `bridge.send_snapshot`). pub fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError> { - self.inner.send_snapshot(snapshot) + match &self.player { + Some(c) => c.bridge.send_snapshot(snapshot), + None => Err(BridgeError::Disconnected), + } } - pub fn receive(&self) -> Result, BridgeError> { - self.inner.receive() + /// Look up the connection (Player or Reader) matching `id`, for + /// per-connection response routing (D-254 §2: atlas/star-map/city-names + /// responses must go back to only the connection that asked). + fn connection(&self, id: ConnectionId) -> Option<&Connection> { + if let Some(p) = &self.player { + if p.id == id { + return Some(p); + } + } + self.readers.iter().find(|c| c.id == id) } - pub fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> { - self.inner.send_atlas_response(resp) + /// Send an atlas layer-stream response to exactly the connection that + /// requested it (#969, D-225 original; D-254 §2 adds the routing — the + /// connection may be the Player or any Reader). `Ok(())` with a debug + /// log if the connection has since disconnected (the response is simply + /// dropped — not an error condition, the requester is gone). + pub fn send_atlas_response_to( + &self, + id: ConnectionId, + resp: &AtlasLayerResponse, + ) -> Result<(), BridgeError> { + match self.connection(id) { + Some(c) => c.bridge.send_atlas_response(resp), + None => { + tracing::debug!( + "atlas response for {:?} dropped — connection {:?} no longer present", + resp.body_id, + id + ); + Ok(()) + } + } } - pub fn send_star_map_response(&self, resp: &StarMapResponse) -> Result<(), BridgeError> { - self.inner.send_star_map_response(resp) + /// Send a star-map response to exactly the connection that requested it + /// (T-949a original; D-254 §2 adds the routing). + pub fn send_star_map_response_to( + &self, + id: ConnectionId, + resp: &StarMapResponse, + ) -> Result<(), BridgeError> { + match self.connection(id) { + Some(c) => c.bridge.send_star_map_response(resp), + None => { + tracing::debug!( + "star map response dropped — connection {:?} no longer present", + id + ); + Ok(()) + } + } } - pub fn send_city_names_response(&self, resp: &CityNamesResponse) -> Result<(), BridgeError> { - self.inner.send_city_names_response(resp) + /// Send a city-names response to exactly the connection that requested + /// it (T-949b original; D-254 §2 adds the routing). + pub fn send_city_names_response_to( + &self, + id: ConnectionId, + resp: &CityNamesResponse, + ) -> Result<(), BridgeError> { + match self.connection(id) { + Some(c) => c.bridge.send_city_names_response(resp), + None => { + tracing::debug!( + "city names response for {:?} dropped — connection {:?} no longer present", + resp.body_id, + id + ); + Ok(()) + } + } } } @@ -250,14 +453,30 @@ pub enum HandshakeState { /// normal traffic is one input batch plus the occasional atlas request. const MAX_INBOUND_FRAMES_PER_TICK: usize = 64; -/// Receive inputs from bridge and push to InputQueue. -/// Drains every complete frame buffered this tick (T-1045) — a single -/// receive() per tick would backlog mixed input/atlas traffic at one frame -/// per 50 ms. Relies on receive() being non-blocking (Ok(None) = no frame). +/// Receive inputs/requests from every connection and route by role (D-254 +/// §2, T-1130). Drains every complete frame buffered this tick per +/// connection (T-1045) — a single receive() per connection per tick would +/// backlog mixed input/atlas traffic at one frame per 50 ms. Relies on +/// receive() being non-blocking (Ok(None) = no frame). +/// +/// Role gate (D-254 §2 permitted-message matrix): `Vec` is only +/// ever pushed to `InputQueue` from the Player connection. A Reader sending +/// inputs is syntactically valid (the D-225 demux parses it fine) but +/// role-disallowed — logged, dropped, and struck; `READER_VIOLATION_ +/// DISCONNECT_THRESHOLD` repeated violations disconnect that reader (never +/// the Player, never other readers). Atlas/star-map/city-names requests are +/// accepted from ANY connection and tagged with the sender's `ConnectionId` +/// so `send_*_responses` can route the reply back to only that connection. +/// +/// THE CRITICAL FIX (D-254 §2): only the Player connection's disconnect +/// flips `ServerRunning` — a reader disconnecting (or never having +/// connected) must never affect a running player session. Reader +/// disconnects just remove that reader from `BridgeResource` and continue. +/// /// Protocol errors (malformed input) are recoverable: the frame is skipped /// and a SimError is pushed to the SimErrorBuffer for client reporting (#85). pub fn receive_bridge_inputs( - bridge: Option>, + bridge: Option>, mut input_queue: ResMut, mut running: ResMut, handshake: Res, @@ -267,115 +486,266 @@ pub fn receive_bridge_inputs( mut city_names_requests: ResMut, time: Option>, ) { - let Some(bridge) = bridge else { return }; + let Some(mut bridge) = bridge else { return }; let current_tick = time.as_ref().map(|t| t.tick).unwrap_or(0); - for _ in 0..MAX_INBOUND_FRAMES_PER_TICK { - match bridge.receive() { - Ok(Some(Inbound::Inputs(inputs))) => { - if !inputs.is_empty() && *handshake == HandshakeState::Pending { + // -- Player connection ------------------------------------------------ + // Unchanged behavior from before D-254: the Player's Inputs go to + // InputQueue, its atlas/etc. requests get tagged with its ConnectionId, + // and ITS disconnect (and only its disconnect) shuts the server down. + if let Some(player) = bridge.player.as_mut() { + let player_id = player.id; + let mut player_disconnected = false; + + for _ in 0..MAX_INBOUND_FRAMES_PER_TICK { + match player.bridge.receive() { + Ok(Some(Inbound::Inputs(inputs))) => { + if !inputs.is_empty() && *handshake == HandshakeState::Pending { + tracing::warn!( + "Received {} input(s) before handshake completed — processing anyway (forward-compatible)", + inputs.len() + ); + } + for input in &inputs { + tracing::trace!( + "Received input: tick={} action={:?}", + input.tick, + input.action + ); + } + for input in inputs { + input_queue.push(input); + } + } + Ok(Some(Inbound::AtlasRequest(req))) => { + atlas_requests.0.push((player_id, req)); + } + Ok(Some(Inbound::StarMapRequest(req))) => { + star_map_requests.0.push((player_id, req)); + } + Ok(Some(Inbound::CityNamesRequest(req))) => { + city_names_requests.0.push((player_id, req)); + } + // No complete frame ready — the backlog is drained. + Ok(None) => break, + Err(BridgeError::Disconnected) => { + tracing::info!("Player disconnected, shutting down"); + running.0 = false; + player_disconnected = true; + break; + } + Err(BridgeError::Io(ref e)) + if e.kind() == std::io::ErrorKind::BrokenPipe + || e.kind() == std::io::ErrorKind::ConnectionReset => + { + tracing::info!("Player pipe broken, shutting down cleanly"); + running.0 = false; + player_disconnected = true; + break; + } + Err(BridgeError::MutexPoisoned(ref msg)) => { + tracing::error!("Player bridge mutex poisoned: {}. Shutting down.", msg); + running.0 = false; + player_disconnected = true; + break; + } + Err(BridgeError::DeserializationWithDump(ref msg)) => { + // Recoverable: skip this frame's input, report to client (#85), + // keep draining — the frame was consumed, later ones may be fine. + tracing::error!("Skipping malformed input frame: {}", msg); + error_buffer.push(SimError { + kind: SimErrorKind::ProtocolError, + message: format!("Malformed input frame: {}", msg), + tick: current_tick, + }); + } + Err(ref e @ BridgeError::Deserialization(_)) => { + // Recoverable deserialization error without dump + tracing::error!("Skipping malformed input: {}", e); + error_buffer.push(SimError { + kind: SimErrorKind::ProtocolError, + message: format!("Deserialization error: {}", e), + tick: current_tick, + }); + } + Err(e) => { + // Unknown error: log once per tick instead of hammering a + // persistently failing stream within one tick. A permanently + // corrupt stream (e.g. the oversized-prefix poison state) + // therefore logs every tick without escalation — follow-up + // ticket covers shutdown-after-N-consecutive-errors. + tracing::error!("Bridge receive error: {}", e); + break; + } + } + } + + if player_disconnected { + bridge.player = None; + } + } + + // -- Reader connections ------------------------------------------------- + // Role-gated: Inputs are never forwarded to InputQueue from a reader — + // logged, dropped, and struck instead. Atlas/star-map/city-names + // requests ARE forwarded, tagged with the reader's own ConnectionId. + // A reader's disconnect only removes that one reader — ServerRunning is + // untouched, and other connections (Player, other readers) are unaffected. + let mut disconnected_readers: Vec = Vec::new(); + let mut to_disconnect_for_violations: Vec = Vec::new(); + + for reader in bridge.readers.iter_mut() { + let reader_id = reader.id; + for _ in 0..MAX_READER_INBOUND_FRAMES_PER_TICK { + match reader.bridge.receive() { + Ok(Some(Inbound::Inputs(inputs))) => { + // D-254 §2 permitted-message matrix: Reader -> Inputs is + // disallowed. Syntactically valid, role-forbidden — log, + // drop the frame (never reaches InputQueue/SimRng, so + // determinism is unaffected by construction), and strike. + reader.violation_strikes += 1; tracing::warn!( - "Received {} input(s) before handshake completed — processing anyway (forward-compatible)", - inputs.len() + "Reader connection {:?} sent {} disallowed PlayerInput(s) — dropped (strike {}/{})", + reader_id, + inputs.len(), + reader.violation_strikes, + READER_VIOLATION_DISCONNECT_THRESHOLD ); + error_buffer.push(SimError { + kind: SimErrorKind::ProtocolError, + message: format!( + "Reader connection {:?} sent disallowed PlayerInput (role violation)", + reader_id + ), + tick: current_tick, + }); + if reader.violation_strikes >= READER_VIOLATION_DISCONNECT_THRESHOLD { + tracing::warn!( + "Reader connection {:?} exceeded violation threshold — disconnecting", + reader_id + ); + to_disconnect_for_violations.push(reader_id); + break; + } } - for input in &inputs { - tracing::trace!( - "Received input: tick={} action={:?}", - input.tick, - input.action + Ok(Some(Inbound::AtlasRequest(req))) => { + atlas_requests.0.push((reader_id, req)); + } + Ok(Some(Inbound::StarMapRequest(req))) => { + star_map_requests.0.push((reader_id, req)); + } + Ok(Some(Inbound::CityNamesRequest(req))) => { + city_names_requests.0.push((reader_id, req)); + } + Ok(None) => break, + Err(BridgeError::Disconnected) => { + tracing::info!("Reader connection {:?} disconnected", reader_id); + disconnected_readers.push(reader_id); + break; + } + Err(BridgeError::Io(ref e)) + if e.kind() == std::io::ErrorKind::BrokenPipe + || e.kind() == std::io::ErrorKind::ConnectionReset => + { + tracing::info!("Reader connection {:?} pipe broken", reader_id); + disconnected_readers.push(reader_id); + break; + } + Err(BridgeError::MutexPoisoned(ref msg)) => { + tracing::error!( + "Reader connection {:?} bridge mutex poisoned: {}", + reader_id, + msg ); + disconnected_readers.push(reader_id); + break; } - for input in inputs { - input_queue.push(input); + Err(BridgeError::DeserializationWithDump(ref msg)) => { + tracing::error!( + "Reader connection {:?}: skipping malformed frame: {}", + reader_id, + msg + ); + error_buffer.push(SimError { + kind: SimErrorKind::ProtocolError, + message: format!("Malformed reader frame: {}", msg), + tick: current_tick, + }); + } + Err(ref e @ BridgeError::Deserialization(_)) => { + tracing::error!( + "Reader connection {:?}: skipping malformed frame: {}", + reader_id, + e + ); + error_buffer.push(SimError { + kind: SimErrorKind::ProtocolError, + message: format!("Deserialization error: {}", e), + tick: current_tick, + }); + } + Err(e) => { + tracing::error!("Reader connection {:?} receive error: {}", reader_id, e); + break; } - } - Ok(Some(Inbound::AtlasRequest(req))) => { - atlas_requests.0.push(req); - } - Ok(Some(Inbound::StarMapRequest(req))) => { - star_map_requests.0.push(req); - } - Ok(Some(Inbound::CityNamesRequest(req))) => { - city_names_requests.0.push(req); - } - // No complete frame ready — the backlog is drained. - Ok(None) => break, - Err(BridgeError::Disconnected) => { - tracing::info!("Client disconnected, shutting down"); - running.0 = false; - break; - } - Err(BridgeError::Io(ref e)) - if e.kind() == std::io::ErrorKind::BrokenPipe - || e.kind() == std::io::ErrorKind::ConnectionReset => - { - tracing::info!("Pipe broken, shutting down cleanly"); - running.0 = false; - break; - } - Err(BridgeError::MutexPoisoned(ref msg)) => { - tracing::error!("Bridge mutex poisoned: {}. Shutting down.", msg); - running.0 = false; - break; - } - Err(BridgeError::DeserializationWithDump(ref msg)) => { - // Recoverable: skip this frame's input, report to client (#85), - // keep draining — the frame was consumed, later ones may be fine. - tracing::error!("Skipping malformed input frame: {}", msg); - error_buffer.push(SimError { - kind: SimErrorKind::ProtocolError, - message: format!("Malformed input frame: {}", msg), - tick: current_tick, - }); - } - Err(ref e @ BridgeError::Deserialization(_)) => { - // Recoverable deserialization error without dump - tracing::error!("Skipping malformed input: {}", e); - error_buffer.push(SimError { - kind: SimErrorKind::ProtocolError, - message: format!("Deserialization error: {}", e), - tick: current_tick, - }); - } - Err(e) => { - // Unknown error: log once per tick instead of hammering a - // persistently failing stream within one tick. A permanently - // corrupt stream (e.g. the oversized-prefix poison state) - // therefore logs every tick without escalation — follow-up - // ticket covers shutdown-after-N-consecutive-errors. - tracing::error!("Bridge receive error: {}", e); - break; } } } + + for id in disconnected_readers + .into_iter() + .chain(to_disconnect_for_violations) + { + bridge.remove_reader(id); + } } -/// Send snapshot from buffer to bridge. -/// Any send error is fatal — the client cannot proceed without snapshots. +/// Send snapshot from buffer to the Player connection ONLY (D-254 §2 — a +/// Reader receives no `ObserverSnapshot` at all, not even filtered; this +/// function never touches `bridge.readers`, structurally). +/// +/// A send error on an EXISTING Player connection is fatal for that +/// connection — the client cannot proceed without snapshots — and shuts the +/// server down (THE CRITICAL FIX: this is the only way this function ever +/// touches `ServerRunning`, and it is scoped correctly, because a reader was +/// never a snapshot recipient to begin with). +/// +/// Having NO Player connection at all is a DIFFERENT, valid case (D-254 §1: +/// a spawn-mode server whose sole connection is a Reader) — `bridge.player` +/// being `None` here is not an error and must never shut the server down; +/// the snapshot is simply not delivered anywhere (there is currently no +/// character-controlling connection to deliver it to) and stays queued in +/// `buffer` for whenever a Player does connect, if ever. pub fn send_bridge_snapshot( - bridge: Option>, + bridge: Option>, mut buffer: ResMut, mut running: ResMut, ) { - let Some(bridge) = bridge else { + let Some(mut bridge) = bridge else { tracing::error!("send_bridge_snapshot: no BridgeResource"); return; }; + if !bridge.has_player() { + // Reader-only server (D-254 §1 spawn-mode) or a Player that hasn't + // finished its accept-loop handshake yet — neither is an error. + // Leave the snapshot queued; it is simply not deliverable this tick. + return; + } if let Some(snapshot) = buffer.snapshot.take() { if let Err(e) = bridge.send_snapshot(&snapshot) { match &e { BridgeError::Disconnected => { - tracing::info!("Client disconnected during send, shutting down"); + tracing::info!("Player disconnected during send, shutting down"); } BridgeError::MutexPoisoned(msg) => { - tracing::error!("Bridge mutex poisoned during send: {}", msg); + tracing::error!("Player bridge mutex poisoned during send: {}", msg); } _ => { - tracing::error!("Bridge send error: {}", e); + tracing::error!("Player bridge send error: {}", e); } } running.0 = false; + bridge.player = None; } } } @@ -391,81 +761,230 @@ impl Default for ServerRunning { } /// Inbound atlas layer requests routed off the bridge (#969, D-225), drained by -/// the proxy serve system in `PreInput`. +/// the proxy serve system in `PreInput`. Each entry is tagged with the +/// requesting connection's id (D-254 §2) so the matching response — pushed +/// 1:1 and in order by `serve_atlas_requests` in `atlas/plugin.rs` — routes +/// back to only that connection, never a broadcast. #[derive(Resource, Default)] -pub struct AtlasRequestBuffer(pub Vec); +pub struct AtlasRequestBuffer(pub Vec<(ConnectionId, AtlasLayerRequest)>); /// Outbound atlas layer responses, filled by the proxy serve system and flushed -/// to the client in `PostSnapshot` (#969, D-225). +/// to the client in `PostSnapshot` (#969, D-225). Connection-tagged (D-254 §2). #[derive(Resource, Default)] -pub struct AtlasResponseBuffer(pub Vec); +pub struct AtlasResponseBuffer(pub Vec<(ConnectionId, AtlasLayerResponse)>); -/// Flush buffered atlas responses to the client (#969, D-225). A failed send is -/// logged but not fatal — an atlas response is not load-bearing like a snapshot. +/// Flush buffered atlas responses to their requesting connections (#969, +/// D-225 original; D-254 §2 adds per-connection routing). A failed send is +/// logged but not fatal — an atlas response is not load-bearing like a +/// snapshot, and a stale/disconnected recipient is not an error (see +/// `BridgeResource::send_atlas_response_to`). pub fn send_atlas_responses( bridge: Option>, mut buffer: ResMut, ) { let Some(bridge) = bridge else { return }; - for resp in buffer.0.drain(..) { - if let Err(e) = bridge.send_atlas_response(&resp) { - tracing::warn!("failed to send atlas response for {}: {}", resp.body_id, e); + for (id, resp) in buffer.0.drain(..) { + if let Err(e) = bridge.send_atlas_response_to(id, &resp) { + tracing::warn!( + "failed to send atlas response for {} to {:?}: {}", + resp.body_id, + id, + e + ); } } } /// Inbound star-map requests routed off the bridge (T-949a), drained by the -/// proxy serve system in `PreInput`. +/// proxy serve system in `PreInput`. Connection-tagged (D-254 §2). #[derive(Resource, Default)] -pub struct StarMapRequestBuffer(pub Vec); +pub struct StarMapRequestBuffer(pub Vec<(ConnectionId, StarMapRequest)>); /// Outbound star-map responses, filled by the proxy serve system and flushed -/// to the client in `PostSnapshot` (T-949a). +/// to the client in `PostSnapshot` (T-949a). Connection-tagged (D-254 §2). #[derive(Resource, Default)] -pub struct StarMapResponseBuffer(pub Vec); +pub struct StarMapResponseBuffer(pub Vec<(ConnectionId, StarMapResponse)>); -/// Flush buffered star-map responses to the client (T-949a). A failed send is -/// logged but not fatal. +/// Flush buffered star-map responses to their requesting connections +/// (T-949a original; D-254 §2 adds per-connection routing). A failed send +/// is logged but not fatal. pub fn send_star_map_responses( bridge: Option>, mut buffer: ResMut, ) { let Some(bridge) = bridge else { return }; - for resp in buffer.0.drain(..) { - if let Err(e) = bridge.send_star_map_response(&resp) { - tracing::warn!("failed to send star map response: {}", e); + for (id, resp) in buffer.0.drain(..) { + if let Err(e) = bridge.send_star_map_response_to(id, &resp) { + tracing::warn!("failed to send star map response to {:?}: {}", id, e); } } } /// Inbound city-names requests routed off the bridge (T-949b), drained by the -/// proxy serve system in `PreInput`. +/// proxy serve system in `PreInput`. Connection-tagged (D-254 §2). #[derive(Resource, Default)] -pub struct CityNamesRequestBuffer(pub Vec); +pub struct CityNamesRequestBuffer(pub Vec<(ConnectionId, CityNamesRequest)>); /// Outbound city-names responses, filled by the proxy serve system and -/// flushed to the client in `PostSnapshot` (T-949b). +/// flushed to the client in `PostSnapshot` (T-949b). Connection-tagged +/// (D-254 §2). #[derive(Resource, Default)] -pub struct CityNamesResponseBuffer(pub Vec); +pub struct CityNamesResponseBuffer(pub Vec<(ConnectionId, CityNamesResponse)>); -/// Flush buffered city-names responses to the client (T-949b). A failed send +/// Flush buffered city-names responses to their requesting connections +/// (T-949b original; D-254 §2 adds per-connection routing). A failed send /// is logged but not fatal. pub fn send_city_names_responses( bridge: Option>, mut buffer: ResMut, ) { let Some(bridge) = bridge else { return }; - for resp in buffer.0.drain(..) { - if let Err(e) = bridge.send_city_names_response(&resp) { + for (id, resp) in buffer.0.drain(..) { + if let Err(e) = bridge.send_city_names_response_to(id, &resp) { tracing::warn!( - "failed to send city names response for {}: {}", + "failed to send city names response for {} to {:?}: {}", resp.body_id, + id, e ); } } } +/// Holds the server's TCP listener for accepting connections AFTER the +/// first Player connection (D-254 §2, T-1130). +/// +/// The first connection is still accepted by `main.rs`'s original +/// blocking `listener.accept()` before the tick loop begins (unchanged — +/// see `main.rs`), matching a normal game launch exactly byte-for-byte +/// when nobody else ever connects. This resource wraps the SAME listener +/// (moved into it after that first accept) so `accept_new_connections` can +/// keep accepting *additional* connections once the tick loop is running — +/// this is what fixes the original starvation bug (a second client used to +/// hang forever waiting for an `accept()` call that would never come). +/// +/// `None` when no listener is wired (e.g. most existing unit/integration +/// tests that construct a `BridgeResource` directly and never spawn a real +/// listener) — `accept_new_connections` is a no-op in that case, so it is +/// always safe to add to any `App`/`World` without also wiring a listener. +#[derive(Resource, Default)] +pub struct ConnectionListener(pub Option); + +/// Connections that have been TCP-accepted but have not yet completed their +/// handshake/startup exchange (D-254 §2, T-1130). Polled non-blockingly +/// every tick by `accept_new_connections` — see `tcp::PendingConnection`'s +/// doc for why this can never stall the tick loop. +#[derive(Resource, Default)] +pub struct PendingConnections(pub Vec); + +/// Accept new TCP connections and advance in-progress handshakes, without +/// ever blocking the tick loop (D-254 §2, T-1130). +/// +/// Two independent, non-blocking steps each tick: +/// 1. Try to accept any newly-arrived TCP connection on `ConnectionListener` +/// (the listener itself is non-blocking — `main.rs` sets this before +/// wrapping it in the resource). A `WouldBlock`/no-pending-connection +/// result is the overwhelmingly common case (no new client this tick) +/// and is silently ignored, not logged. +/// 2. Poll every connection in `PendingConnections`. `PendingPoll::Waiting` +/// connections stay queued for next tick. `PendingPoll::Ready` +/// connections are promoted based on `startup.role`: +/// - `Player`, and `BridgeResource` has no Player yet → installed as the +/// Player connection. +/// - `Player`, and a Player already exists → THE SECOND-PLAYER CASE +/// (D-254 §2/T-1130 scope: "a second Player attempt gets a clean +/// rejection, not a hang"). The connection is dropped immediately +/// after the handshake completes — no silent starvation (the original +/// bug), and no impact on the existing Player's session (its +/// connection is never touched). A client attempting to connect as a +/// second Player sees a clean disconnect right after startup, which +/// is a well-defined, discoverable failure — the correct behavior for +/// an out-of-scope case (general N-player is explicitly D-009's +/// separate ambition, not this ticket's). +/// - `Reader` → installed as a new Reader connection, always (0-N +/// readers is the whole point of this ticket). +/// `PendingPoll::Failed` connections (EOF before completing handshake, +/// malformed startup, etc.) are simply dropped — never logged as errors +/// at more than `warn` level, since an incomplete handshake from a +/// probing/misbehaving client is an expected occurrence, not a bug. +pub fn accept_new_connections( + listener: Option>, + mut pending: Option>, + bridge: Option>, +) { + let (Some(listener), Some(pending), Some(mut bridge)) = (listener, pending.as_mut(), bridge) + else { + return; + }; + + // Step 1: accept any newly-arrived connection (non-blocking listener). + if let Some(l) = listener.0.as_ref() { + match l.accept() { + Ok((stream, peer_addr)) => { + tracing::info!("accepted new connection from {}", peer_addr); + match crate::bridge::tcp::PendingConnection::new(stream) { + Ok(conn) => pending.0.push(conn), + Err(e) => tracing::warn!("failed to wrap accepted connection: {}", e), + } + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + // No connection pending — the overwhelmingly common case. + } + Err(e) => { + tracing::warn!("accept() failed: {}", e); + } + } + } + + // Step 2: advance every in-progress handshake by one non-blocking poll. + let mut still_pending = Vec::with_capacity(pending.0.len()); + for mut conn in pending.0.drain(..) { + match conn.poll() { + tcp::PendingPoll::Waiting => still_pending.push(conn), + tcp::PendingPoll::Ready { stream, startup } => match startup.role { + ConnectionRole::Player => { + if bridge.has_player() { + tracing::warn!( + "second Player connection attempt rejected (0-1 Player + 0-N Reader scope, D-254 §2) — disconnecting" + ); + // Dropping `stream` closes the TCP connection — a + // clean, immediate disconnect, not a hang. + drop(stream); + } else { + match TcpBridge::from_connected_stream(stream) { + Ok(tcp_bridge) => { + let id = bridge.insert_player(tcp_bridge); + tracing::info!("Player connection established: {:?}", id); + } + Err(e) => { + tracing::warn!( + "failed to promote pending Player connection: {}", + e + ); + } + } + } + } + ConnectionRole::Reader => match TcpBridge::from_connected_stream(stream) { + Ok(tcp_bridge) => { + let id = bridge.insert_reader(tcp_bridge); + tracing::info!("Reader connection established: {:?}", id); + } + Err(e) => { + tracing::warn!("failed to promote pending Reader connection: {}", e); + } + }, + }, + tcp::PendingPoll::Failed => { + // Handshake never completed (EOF, malformed startup, etc.) — + // drop silently at info level. Not a server error. + tracing::info!("pending connection failed to complete handshake"); + } + } + } + pending.0 = still_pending; +} + /// Bridge plugin for client-server communication /// Abstracts transport layer (LocalBridge/NetworkBridge) pub struct BridgePlugin; @@ -488,6 +1007,18 @@ impl Plugin for BridgePlugin { .init_resource::() .init_resource::() .init_resource::() + .init_resource::() + .init_resource::() + // Multi-connection accept-loop (D-254 §2, T-1130) — must run + // before receive_bridge_inputs so a connection whose handshake + // completes this tick has its first frame drained the same + // tick, not next tick. + .add_systems( + Update, + accept_new_connections + .before(receive_bridge_inputs) + .in_set(TickPhase::PreInput), + ) // Bridge I/O — PreInput (receive) and PostSnapshot (send) .add_systems(Update, receive_bridge_inputs.in_set(TickPhase::PreInput)) .add_systems(Update, send_bridge_snapshot.in_set(TickPhase::PostSnapshot)) diff --git a/server/src/bridge/tcp.rs b/server/src/bridge/tcp.rs index 8a8eb2358..ecf3012b4 100644 --- a/server/src/bridge/tcp.rs +++ b/server/src/bridge/tcp.rs @@ -143,6 +143,35 @@ impl TcpBridge { }) } + /// Server-side: promote an already-connected, already-handshaken stream + /// into a full `TcpBridge` (D-254 §2, T-1130). + /// + /// Used by [`PendingConnection`] once its non-blocking handshake/startup + /// exchange completes — unlike `accept`/`accept_on`, this does not call + /// `listener.accept()` itself; the stream is already connected and (per + /// `PendingConnection`'s contract) already sent `HandshakeMessage` and + /// received a valid `StartupMessage`. Sets the stream non-blocking for + /// the steady-state tick loop, same as every other constructor here. + pub fn from_connected_stream(stream: TcpStream) -> Result { + let local_addr = stream + .local_addr() + .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?; + + stream + .set_nonblocking(true) + .map_err(|e| BridgeError::Transport(format!("failed to set non-blocking: {}", e)))?; + + let reader_stream = stream.try_clone().map_err(|e| { + BridgeError::Transport(format!("failed to clone stream for reader: {}", e)) + })?; + + Ok(Self { + reader: Mutex::new(ReadHalf::new(reader_stream)), + writer: Mutex::new(BufWriter::new(stream)), + local_addr, + }) + } + /// Get the local address (useful for OS-assigned port discovery in tests). pub fn local_addr(&self) -> SocketAddr { self.local_addr @@ -285,3 +314,139 @@ impl SimBridge for TcpBridge { Ok(()) } } + +/// A connection that has been TCP-accepted but has not yet completed the +/// handshake/startup exchange (D-254 §2, T-1130). +/// +/// Exists because the multi-connection accept-loop in `main.rs` runs inside +/// the live tick loop — unlike the original single-connection startup +/// sequence (`TcpBridge::accept` + `receive_startup`, which blocks freely +/// because nothing else is running yet), a connection accepted *after* the +/// server is already ticking must never stall other connections while it +/// completes its own handshake. `PendingConnection` is polled once per tick, +/// non-blockingly, exactly like `TcpBridge::receive()` already is — a slow +/// or hostile client sits here indefinitely, consuming no thread and +/// blocking nothing, until it finishes or disconnects. +/// +/// State machine: `AwaitingHandshakeSend` (send the empty `HandshakeMessage` +/// — reused from `send_handshake`'s wire shape) → `AwaitingStartup` (poll for +/// a complete `StartupMessage` frame via the same `FrameAccumulator` the +/// steady-state `receive()` path uses, so a startup message split across TCP +/// segments reassembles correctly here too). +pub struct PendingConnection { + stream: TcpStream, + accum: FrameAccumulator, + state: PendingState, +} + +enum PendingState { + AwaitingHandshakeSend, + AwaitingStartup, +} + +/// Result of one `PendingConnection::poll()` call. +pub enum PendingPoll { + /// Handshake not yet complete — keep polling next tick. + Waiting, + /// Startup message received and decoded. Caller promotes this into a + /// full `TcpBridge` via `TcpBridge::from_connected_stream`. + Ready { + stream: TcpStream, + startup: super::StartupMessage, + }, + /// The connection died before completing its handshake (EOF or a fatal + /// I/O error). Caller drops this pending connection. + Failed, +} + +impl PendingConnection { + /// Wrap a freshly-`accept()`-ed stream. Sets non-blocking immediately — + /// this type never blocks the tick loop, by construction. + pub fn new(stream: TcpStream) -> Result { + stream + .set_nonblocking(true) + .map_err(|e| BridgeError::Transport(format!("failed to set non-blocking: {}", e)))?; + Ok(Self { + stream, + accum: FrameAccumulator::new(), + state: PendingState::AwaitingHandshakeSend, + }) + } + + /// Advance the handshake by one tick's worth of non-blocking I/O. + /// + /// Never blocks: a `WouldBlock` on either the handshake write or the + /// startup read simply returns `PendingPoll::Waiting` for another tick + /// to retry. `HandshakeMessage` is a fixed few bytes (an empty + /// MessagePack map plus the 4-byte length prefix) — in practice it + /// completes in a single non-blocking write, but the retry path exists + /// for the same reason `send_handshake`'s blocking toggle exists on the + /// first connection: TCP send buffers are not guaranteed instantaneous. + pub fn poll(&mut self) -> PendingPoll { + use super::types::HandshakeMessage; + use std::io::Write; + + if matches!(self.state, PendingState::AwaitingHandshakeSend) { + let msg = HandshakeMessage {}; + let payload = match rmp_serde::to_vec_named(&msg) { + Ok(p) => p, + Err(e) => { + tracing::error!("pending connection: failed to encode handshake: {}", e); + return PendingPoll::Failed; + } + }; + match write_framed(&mut self.stream, &payload) { + Ok(()) => { + self.state = PendingState::AwaitingStartup; + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + return PendingPoll::Waiting; + } + Err(e) => { + tracing::warn!("pending connection: handshake send failed: {}", e); + return PendingPoll::Failed; + } + } + // write_framed calls flush() internally — no separate flush needed. + let _ = self.stream.flush(); + } + + match self.accum.poll_frame(&mut self.stream) { + Ok(Some(payload)) => match rmp_serde::from_slice::(&payload) { + Ok(startup) => { + tracing::info!( + "pending connection: startup received, role={:?}", + startup.role + ); + // try_clone so the caller gets an owned stream; self.stream + // is dropped with this PendingConnection once the caller + // promotes the clone into a TcpBridge. + match self.stream.try_clone() { + Ok(stream) => PendingPoll::Ready { stream, startup }, + Err(e) => { + tracing::error!( + "pending connection: failed to clone stream for promotion: {}", + e + ); + PendingPoll::Failed + } + } + } + Err(e) => { + tracing::warn!("pending connection: malformed startup message: {}", e); + PendingPoll::Failed + } + }, + Ok(None) => { + // Clean EOF — the peer disconnected before sending startup. + PendingPoll::Failed + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => PendingPoll::Waiting, + Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => PendingPoll::Failed, + Err(e) => { + tracing::warn!("pending connection: startup read failed: {}", e); + PendingPoll::Failed + } + } + } +} diff --git a/server/src/bridge/types.rs b/server/src/bridge/types.rs index 43ef1edcd..0c2682c2f 100644 --- a/server/src/bridge/types.rs +++ b/server/src/bridge/types.rs @@ -20,6 +20,29 @@ pub use crate::simulation::time::{DayPhase, TickRate}; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct HandshakeMessage {} +/// Connection role, gating what a connection may send/receive (D-254 §2). +/// +/// `Player` is the sole role that may enter the character-spawn path, send +/// `PlayerInput`, or receive `ObserverSnapshot`. `Reader` is a genuinely new +/// class: no character, no inputs, no per-tick snapshot — structurally +/// enforced server-side (`main.rs`, `bridge::mod`), never client courtesy. +/// +/// Shaped for growth per D-254 §6: a future `TradingReader` variant is a +/// strict superset of `Reader` (everything `Reader` gets, plus a narrow +/// per-verb `PlayerInput` allowlist) — not a replacement, and not added by +/// this ticket (T-1130 scope is `Player`/`Reader` only). +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)] +pub enum ConnectionRole { + /// The single character-controlling connection. Only role that existed + /// before D-254. Default for back-compat (see `StartupMessage::role`). + #[default] + Player, + /// Read-only observer: no character, no inputs, no `ObserverSnapshot`. + /// May request install-static/world-public data (atlas/star-map/ + /// city-names) — see the permitted-message matrix in `bridge::mod`. + Reader, +} + /// Startup message sent by the client after receiving HandshakeMessage (#175). /// Contains the world seed for deterministic simulation (D-010, D-029). /// @@ -35,7 +58,22 @@ pub struct StartupMessage { /// World seed for SimRng initialization. /// Generated by SessionManager.new_game() on the client. /// Same seed → same EntanglementConfig → same NPC population (D-029). + /// + /// Ignored server-side for `Reader` connections (D-254 §2): a reader + /// inherits whatever `SimRng` state the server already has (from the + /// Player's own `StartupMessage`, or `--seed` in spawn-mode) — a second + /// StartupMessage must never re-seed `SimRng` after tick 0, or a + /// reader attaching mid-session would silently break determinism for + /// the Player already connected. pub world_seed: u64, + /// Connection role (D-254 §2). `#[serde(default)]` makes this field + /// optional on the wire: an old-format client that only ever sent + /// `{world_seed}` (every client before D-254) still decodes cleanly, + /// with `role` defaulting to `ConnectionRole::Player` — the same + /// connection behavior that client always got. Byte-compatible, + /// forward-compatible; no protocol version bump (D-192 precedent). + #[serde(default)] + pub role: ConnectionRole, } /// The ONLY data structure crossing the client-server boundary (D-020) @@ -1101,6 +1139,7 @@ mod tests { fn startup_message_roundtrip() { let msg = StartupMessage { world_seed: 0xDEADBEEF, + role: ConnectionRole::Player, }; let bytes = rmp_serde::to_vec_named(&msg).expect("serialize"); let decoded: StartupMessage = rmp_serde::from_slice(&bytes).expect("deserialize"); @@ -1110,7 +1149,10 @@ mod tests { #[test] fn startup_message_zero_seed() { - let msg = StartupMessage { world_seed: 0 }; + let msg = StartupMessage { + world_seed: 0, + role: ConnectionRole::Player, + }; let bytes = rmp_serde::to_vec_named(&msg).expect("serialize"); let decoded: StartupMessage = rmp_serde::from_slice(&bytes).expect("deserialize"); assert_eq!(decoded.world_seed, 0); @@ -1120,12 +1162,57 @@ mod tests { fn startup_message_max_seed() { let msg = StartupMessage { world_seed: u64::MAX, + role: ConnectionRole::Player, }; let bytes = rmp_serde::to_vec_named(&msg).expect("serialize"); let decoded: StartupMessage = rmp_serde::from_slice(&bytes).expect("deserialize"); assert_eq!(decoded.world_seed, u64::MAX); } + /// D-254 §2 / T-1130 behavior 1: an old-format `StartupMessage` — the + /// exact wire shape every client before this ticket sent, `{world_seed}` + /// only, no `role` key at all — must still decode cleanly, with `role` + /// defaulting to `Player`. This is the byte-compatibility guarantee: no + /// already-shipped client encoder needs to change for this ticket to be + /// safe to deploy alongside it. + #[test] + fn old_format_startup_message_decodes_as_player() { + // Hand-encode the pre-D-254 shape directly (a single-key map), rather + // than deriving it from a struct literal, so this test can't + // accidentally pass just because both sides changed together. + #[derive(Serialize)] + struct OldStartupMessage { + world_seed: u64, + } + let old_msg = OldStartupMessage { + world_seed: 0x1234_5678, + }; + let bytes = rmp_serde::to_vec_named(&old_msg).expect("serialize old-format message"); + + let decoded: StartupMessage = + rmp_serde::from_slice(&bytes).expect("old-format message must still decode"); + assert_eq!(decoded.world_seed, 0x1234_5678); + assert_eq!( + decoded.role, + ConnectionRole::Player, + "role must default to Player when absent from the wire" + ); + } + + /// D-254 §2: a new-format message explicitly carrying `role: Reader` + /// must decode with that role preserved — the default only applies when + /// the field is genuinely absent, it must not clobber an explicit value. + #[test] + fn startup_message_reader_role_roundtrips() { + let msg = StartupMessage { + world_seed: 99, + role: ConnectionRole::Reader, + }; + let bytes = rmp_serde::to_vec_named(&msg).expect("serialize"); + let decoded: StartupMessage = rmp_serde::from_slice(&bytes).expect("deserialize"); + assert_eq!(decoded.role, ConnectionRole::Reader); + } + #[test] fn handshake_is_distinct_from_snapshot() { // HandshakeMessage and ObserverSnapshot are different types on the wire. diff --git a/server/src/main.rs b/server/src/main.rs index 131c1905b..f547a71e3 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -11,7 +11,9 @@ use bevy_app::prelude::*; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; use settled_reach_server::bridge::tcp::TcpBridge; -use settled_reach_server::bridge::{BridgePlugin, BridgeResource, HandshakeState, ServerRunning}; +use settled_reach_server::bridge::{ + BridgePlugin, BridgeResource, ConnectionListener, ConnectionRole, HandshakeState, ServerRunning, +}; use settled_reach_server::simulation::SimulationPlugin; fn main() { @@ -117,6 +119,25 @@ fn main() { } tracing::info!("Waiting for client connection on port {}", actual_port); + // D-254 §2/T-1130: the FIRST connection is still accepted here, exactly + // as before — one blocking listener.accept() call, byte-identical to + // pre-D-254 behavior when nobody else ever connects. What changes is + // AFTER: the listener is set non-blocking and handed to + // ConnectionListener (inserted below) so accept_new_connections can + // keep accepting additional connections once the tick loop starts, + // instead of the original bug where a second accept() call never + // happened at all and a second client hung forever. + // + // accept_on() consumes the listener; clone it first so both the first + // accept AND the later non-blocking accept-loop have a working handle + // on the same underlying socket (TcpListener::try_clone shares the fd, + // not a new listener — connections queued on either handle are visible + // to both, same as TcpStream::try_clone is already used for read/write + // halves throughout this bridge). + let listener_for_loop = listener.try_clone().unwrap_or_else(|e| { + tracing::error!("Failed to clone listener for accept-loop: {}", e); + std::process::exit(1); + }); let bridge = TcpBridge::accept_on(listener).unwrap_or_else(|e| { tracing::error!("Failed to accept: {}", e); std::process::exit(1); @@ -291,9 +312,42 @@ fn main() { } } - app.insert_resource(BridgeResource::new(bridge)); + // D-254 §2/T-1130: the FIRST connection's role, exactly as it does for + // every later accept-loop connection (main.rs's accept-loop handles + // connections 2+; this handles the honest first-connection case a + // spawn-mode Reader server actually needs — D-254 §1's spawn-mode + // Atlas companion connects as the ONLY connection to a freshly-spawned + // server, so "first connection" and "Reader" are not mutually + // exclusive). `BridgeResource::default()` + explicit insert_player/ + // insert_reader replaces the old unconditional `BridgeResource::new` + // (which always meant "install as Player" — there was no other role + // before this ticket). + let mut bridge_resource = BridgeResource::default(); + match startup.role { + ConnectionRole::Player => { + bridge_resource.insert_player(bridge); + } + ConnectionRole::Reader => { + tracing::info!( + "first connection is a Reader (D-254 §1 spawn-mode) — no character will be spawned for it" + ); + bridge_resource.insert_reader(bridge); + } + } + app.insert_resource(bridge_resource); app.insert_resource(HandshakeState::Complete); + // D-254 §2/T-1130: wire the cloned listener non-blocking so + // accept_new_connections (BridgePlugin, PreInput) can accept additional + // connections every tick without ever blocking the tick loop. This is + // the actual fix for the original starvation bug — before this, there + // was exactly one listener.accept() call in the whole process lifetime. + listener_for_loop.set_nonblocking(true).unwrap_or_else(|e| { + tracing::error!("Failed to set accept-loop listener non-blocking: {}", e); + std::process::exit(1); + }); + app.insert_resource(ConnectionListener(Some(listener_for_loop))); + // SimulationPlugin { seed } already inserts SimRng with the correct seed // during plugin build. We re-insert here as a defensive override for one // specific ordering risk: any future plugin that registers *before* @@ -316,6 +370,23 @@ fn main() { ); // Gauntlet test world for --test-mode, proof room for normal mode. + // + // D-254 §2/T-1130 scope note: this call is NOT gated on the first + // connection's role, even though it spawns a `PlayerCharacter` entity + // unconditionally. A Reader-only spawned server (D-254 §1 spawn-mode) + // therefore has an inert, unpiloted `PlayerCharacter` entity sitting in + // its ECS world — nothing drives it (no Player connection exists to + // send it inputs), and the Reader never learns it exists: `send_ + // bridge_snapshot` routes `ObserverSnapshot` to the Player connection + // ONLY and is a documented no-op with no Player installed (see + // `bridge::send_bridge_snapshot`), so this entity's existence has no + // observable effect on a Reader-only session. Splitting character-spawn + // out of `setup_proof_room`/`setup_gauntlet` (both of which also wire + // NPCs, the walkability map, and the relationship graph — genuinely + // "whole world setup", not just "spawn the player") into an optional + // step is real refactoring work, correctly out of scope for this + // gating ticket; tracked as follow-up, not required for the D-010 + // information-boundary guarantee this ticket exists to establish. if test_mode { #[cfg(feature = "gauntlet")] settled_reach_server::test_world::setup_gauntlet(&mut app); diff --git a/server/tests/bridge_tcp.rs b/server/tests/bridge_tcp.rs index 4c31b700a..359a9efce 100644 --- a/server/tests/bridge_tcp.rs +++ b/server/tests/bridge_tcp.rs @@ -421,9 +421,521 @@ fn single_tick_drains_all_ready_inbound_frames() { 1, "the city-names request must drain in the same tick (H6: real wire path)" ); - assert_eq!(city_names[0].body_id, "GJ1c"); + assert_eq!(city_names[0].1.body_id, "GJ1c"); assert!( world.resource::().0, "draining must not shut the server down" ); } + +// -- D-254 §2 / T-1130: multi-connection bridge + ConnectionRole gate --------- +// +// Six behaviors below (a 7th — old-format StartupMessage decodes as Player — +// is a unit test in server/src/bridge/types.rs, next to the type itself). +// Each test drives the real wire path: a genuine TcpStream client performs +// the handshake/startup exchange a real Godot client (or PendingConnection's +// server-side counterpart) would, against the real non-blocking accept-loop +// and drain-loop systems via `run_system_once` — no mocks of the framing or +// role-gate logic itself. + +/// Client-side test helper: perform one full handshake/startup exchange over +/// an already-connected stream, exactly as `sim_bridge.gd`'s live-mode path +/// does (read HandshakeMessage, write StartupMessage). Returns the stream so +/// the caller can continue driving it (send inputs, read responses, etc.). +fn client_handshake(mut stream: TcpStream, role: ConnectionRole) -> TcpStream { + let handshake_payload = read_framed(&mut stream) + .expect("failed to read handshake") + .expect("unexpected EOF reading handshake"); + let _: HandshakeMessage = + rmp_serde::from_slice(&handshake_payload).expect("failed to decode HandshakeMessage"); + + let startup = StartupMessage { + world_seed: 12345, + role, + }; + let payload = rmp_serde::to_vec_named(&startup).expect("failed to serialize StartupMessage"); + write_framed(&mut stream, &payload).expect("failed to write StartupMessage"); + stream +} + +/// Drive `accept_new_connections` for up to `max_ticks` schedule passes, or +/// until `bridge.readers().len() + bridge.has_player() as usize` (checked via +/// the passed predicate) is satisfied — matches this test file's existing +/// wall-clock-deadline-over-fixed-sleep hardening (see `collect_input_batches` +/// above): connection promotion depends on TCP delivery timing, not a fixed +/// tick count, so poll until true or time out rather than guessing a sleep. +fn drive_accept_loop_until( + world: &mut bevy_ecs::world::World, + condition: impl Fn(&bevy_ecs::world::World) -> bool, +) { + use bevy_ecs::system::RunSystemOnce; + use settled_reach_server::bridge::accept_new_connections; + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !condition(world) { + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for accept-loop condition" + ); + world + .run_system_once(accept_new_connections) + .expect("accept_new_connections failed to run"); + thread::sleep(std::time::Duration::from_millis(1)); + } +} + +/// Build a `bevy_ecs::World` wired exactly like `BridgePlugin` wires it for +/// the systems under test here (accept-loop + drain-loop + response +/// buffers), bound to a fresh OS-assigned port. +fn new_multi_connection_world() -> (bevy_ecs::world::World, std::net::SocketAddr) { + use settled_reach_server::bridge::{ + AtlasRequestBuffer, AtlasResponseBuffer, BridgeResource, CityNamesRequestBuffer, + CityNamesResponseBuffer, ConnectionListener, HandshakeState, PendingConnections, + ServerRunning, SnapshotBuffer, StarMapRequestBuffer, StarMapResponseBuffer, + }; + use settled_reach_server::simulation::input::InputQueue; + + 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 + .set_nonblocking(true) + .expect("failed to set listener non-blocking"); + + let mut world = bevy_ecs::world::World::new(); + world.insert_resource(ConnectionListener(Some(listener))); + world.init_resource::(); + world.insert_resource(BridgeResource::default()); + world.init_resource::(); + world.init_resource::(); + world.insert_resource(HandshakeState::Complete); + world.init_resource::(); + world.init_resource::(); + world.init_resource::(); + world.init_resource::(); + world.init_resource::(); + world.init_resource::(); + world.init_resource::(); + world.init_resource::(); + (world, addr) +} + +/// T-1130 behavior 2: a Reader connection's handshake succeeds — it reaches +/// `BridgeResource`'s reader collection — without any character-spawn +/// concept ever entering the picture. There is no `PlayerCharacter`-query +/// assertion here because that's the correct proof of the D-254 §2 +/// guarantee: character-spawn is a `main.rs`/world-setup concern entirely +/// disjoint from connection acceptance (see `main.rs`'s D-254 §2 scope-note +/// comment at the `setup_proof_room`/`setup_gauntlet` call site) — a Reader +/// reaching `BridgeResource.readers` never touches that code path at all, +/// which this test demonstrates by never invoking it and the connection +/// still working end-to-end. +#[test] +fn reader_handshake_succeeds_and_installs_as_reader() { + use settled_reach_server::bridge::BridgeResource; + + let (mut world, addr) = new_multi_connection_world(); + + let client_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect"); + client_handshake(stream, ConnectionRole::Reader) + }); + + drive_accept_loop_until(&mut world, |w| { + w.resource::().reader_count() == 1 + }); + + let _stream = client_handle.join().expect("client thread panicked"); + + let bridge = world.resource::(); + assert!( + !bridge.has_player(), + "a Reader-only connection must never be installed as Player" + ); + assert_eq!(bridge.reader_count(), 1, "exactly one reader installed"); +} + +/// T-1130 behavior 3: an atlas/star-map/city-names response addressed to a +/// Reader's `ConnectionId` reaches that Reader over the wire — the +/// per-connection response-tagging plumbing (D-254 §2) actually delivers, +/// not just tags in-memory. Drives `send_star_map_responses` directly +/// (bypassing `serve_star_map_requests`/the atlas proxy, which are already +/// covered elsewhere) to isolate exactly the tagging/routing behavior this +/// ticket adds. +#[test] +fn reader_receives_tagged_star_map_response() { + use bevy_ecs::system::RunSystemOnce; + use settled_reach_server::atlas::atlas_data_proxy::{StarMapResponse, StarMapStatus}; + use settled_reach_server::bridge::{ + send_star_map_responses, BridgeResource, StarMapResponseBuffer, + }; + + let (mut world, addr) = new_multi_connection_world(); + + let client_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect"); + client_handshake(stream, ConnectionRole::Reader) + }); + + drive_accept_loop_until(&mut world, |w| { + w.resource::().reader_count() == 1 + }); + let mut stream = client_handle.join().expect("client thread panicked"); + + let reader_id = world + .resource::() + .reader_ids() + .first() + .copied() + .expect("reader must be installed"); + + world.resource_mut::().0.push(( + reader_id, + StarMapResponse { + status: StarMapStatus::Ready, + data: Some(serde_json::json!({"nodes": [], "edges": []})), + }, + )); + + world + .run_system_once(send_star_map_responses) + .expect("send_star_map_responses failed to run"); + + let payload = read_framed(&mut stream) + .expect("failed to read response frame") + .expect("unexpected EOF reading response"); + let resp: StarMapResponse = rmp_serde::from_slice(&payload).expect("failed to decode"); + assert_eq!(resp.status, StarMapStatus::Ready); +} + +/// T-1130 behavior 4: while a Player streams `ObserverSnapshot` every tick, +/// a concurrently-connected Reader receives NOTHING on its socket — not a +/// filtered snapshot, not an empty one, nothing at all (D-254 §2: "not even +/// filtered" is structural, this test proves it holds over the wire with +/// both connections genuinely live at once, not just by code inspection). +#[test] +fn reader_never_receives_observer_snapshot_while_player_streams() { + use bevy_ecs::system::RunSystemOnce; + use settled_reach_server::bridge::{send_bridge_snapshot, BridgeResource, SnapshotBuffer}; + + let (mut world, addr) = new_multi_connection_world(); + + // Player connects first (mirrors main.rs's structural first-connection + // path in spirit, though this test drives it through the SAME + // accept-loop the Reader below uses — the accept-loop must handle a + // Player exactly as well as a Reader, not just readers-after-a- + // pre-existing-Player). + let player_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect (player)"); + client_handshake(stream, ConnectionRole::Player) + }); + drive_accept_loop_until(&mut world, |w| w.resource::().has_player()); + let mut player_stream = player_handle.join().expect("player thread panicked"); + + let reader_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect (reader)"); + client_handshake(stream, ConnectionRole::Reader) + }); + drive_accept_loop_until(&mut world, |w| { + w.resource::().reader_count() == 1 + }); + let mut reader_stream = reader_handle.join().expect("reader thread panicked"); + // Non-blocking so a read that would otherwise hang forever (the whole + // point being tested — nothing ever arrives) returns WouldBlock instead. + reader_stream + .set_nonblocking(true) + .expect("failed to set reader stream non-blocking"); + + world.resource_mut::().snapshot = Some(sample_snapshot(7)); + world + .run_system_once(send_bridge_snapshot) + .expect("send_bridge_snapshot failed to run"); + + // Player DID get the snapshot — establishes the positive control so a + // trivially-broken send path (e.g. send_bridge_snapshot silently + // no-op'ing for everyone) can't masquerade as "reader correctly got + // nothing". + let payload = read_framed(&mut player_stream) + .expect("failed to read player frame") + .expect("unexpected EOF reading player snapshot"); + let snapshot: ObserverSnapshot = rmp_serde::from_slice(&payload).expect("failed to decode"); + assert_eq!(snapshot.tick, 7); + + // Reader got NOTHING — not WouldBlock-then-eventually-something, a + // sustained absence over a real wall-clock window. + let check_until = std::time::Instant::now() + std::time::Duration::from_millis(300); + while std::time::Instant::now() < check_until { + let mut probe = [0u8; 1]; + match std::io::Read::read(&mut reader_stream, &mut probe) { + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {} + Ok(0) => panic!("reader stream unexpectedly closed"), + other => panic!( + "reader received unexpected data/result while player streamed: {:?}", + other + ), + } + thread::sleep(std::time::Duration::from_millis(10)); + } +} + +/// T-1130 behavior 5: a Reader sending the forbidden `Vec` +/// shape is dropped (not forwarded to `InputQueue`) on each offense, and +/// disconnected once it crosses the strike threshold — never on the first +/// offense (D-254 §2: "log + drop on first offense, repeated = disconnect"). +#[test] +fn forbidden_reader_input_is_dropped_then_disconnected() { + use bevy_ecs::system::RunSystemOnce; + use settled_reach_server::bridge::{receive_bridge_inputs, BridgeResource}; + use settled_reach_server::simulation::input::InputQueue; + + let (mut world, addr) = new_multi_connection_world(); + + let client_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect"); + client_handshake(stream, ConnectionRole::Reader) + }); + drive_accept_loop_until(&mut world, |w| { + w.resource::().reader_count() == 1 + }); + let mut stream = client_handle.join().expect("client thread panicked"); + + let send_one_input_frame = |stream: &mut TcpStream| { + let inputs = vec![PlayerInput { + tick: 1, + action: PlayerAction::MoveNorth, + }]; + let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize"); + write_framed(stream, &payload).expect("failed to write input frame"); + }; + + // Strikes 1 and 2: the reader survives, but nothing reaches InputQueue. + for strike in 1..=2 { + send_one_input_frame(&mut stream); + // Wall-clock settle for TCP delivery, then drain — mirrors this + // file's existing pattern (see single_tick_drains_all_ready_inbound_frames). + thread::sleep(std::time::Duration::from_millis(50)); + world + .run_system_once(receive_bridge_inputs) + .expect("receive_bridge_inputs failed to run"); + assert_eq!( + world.resource::().len(), + 0, + "forbidden reader input must never reach InputQueue (strike {})", + strike + ); + assert_eq!( + world.resource::().reader_count(), + 1, + "reader must survive strike {} (below disconnect threshold)", + strike + ); + } + + // Strike 3 crosses READER_VIOLATION_DISCONNECT_THRESHOLD (3) — the + // reader is disconnected. + send_one_input_frame(&mut stream); + thread::sleep(std::time::Duration::from_millis(50)); + world + .run_system_once(receive_bridge_inputs) + .expect("receive_bridge_inputs failed to run"); + assert_eq!( + world.resource::().len(), + 0, + "forbidden reader input must never reach InputQueue, even on the disconnecting strike" + ); + assert_eq!( + world.resource::().reader_count(), + 0, + "reader must be disconnected after crossing the violation threshold" + ); +} + +/// T-1130 behavior 6 (THE CRITICAL FIX): a Reader disconnecting must leave +/// a running Player session completely unaffected — `ServerRunning` stays +/// true, and the Player connection keeps working (proven by successfully +/// sending it a snapshot AFTER the reader is gone, not just by inspecting +/// the flag). +#[test] +fn reader_disconnect_does_not_affect_running_player_session() { + use bevy_ecs::system::RunSystemOnce; + use settled_reach_server::bridge::{ + receive_bridge_inputs, send_bridge_snapshot, BridgeResource, ServerRunning, SnapshotBuffer, + }; + + let (mut world, addr) = new_multi_connection_world(); + + let player_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect (player)"); + client_handshake(stream, ConnectionRole::Player) + }); + drive_accept_loop_until(&mut world, |w| w.resource::().has_player()); + let mut player_stream = player_handle.join().expect("player thread panicked"); + + let reader_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect (reader)"); + client_handshake(stream, ConnectionRole::Reader) + }); + drive_accept_loop_until(&mut world, |w| { + w.resource::().reader_count() == 1 + }); + let reader_stream = reader_handle.join().expect("reader thread panicked"); + + // The reader disconnects (drop closes the TCP connection — clean EOF). + drop(reader_stream); + + // Drain: the server observes the reader's EOF and removes it. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + world + .run_system_once(receive_bridge_inputs) + .expect("receive_bridge_inputs failed to run"); + if world.resource::().reader_count() == 0 { + break; + } + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for reader disconnect to be observed" + ); + thread::sleep(std::time::Duration::from_millis(10)); + } + + assert!( + world.resource::().0, + "a reader's disconnect must never flip ServerRunning" + ); + assert!( + world.resource::().has_player(), + "the player connection must still be installed after the reader disconnects" + ); + + // Prove the player session is still genuinely functional, not just + // structurally present: send it a snapshot and read it back. + world.resource_mut::().snapshot = Some(sample_snapshot(99)); + world + .run_system_once(send_bridge_snapshot) + .expect("send_bridge_snapshot failed to run"); + let payload = read_framed(&mut player_stream) + .expect("failed to read post-disconnect player frame") + .expect("unexpected EOF — player session was affected by reader disconnect"); + let snapshot: ObserverSnapshot = rmp_serde::from_slice(&payload).expect("failed to decode"); + assert_eq!( + snapshot.tick, 99, + "player session must remain fully functional after reader disconnect" + ); + assert!( + world.resource::().0, + "sending to the still-live player must not flip ServerRunning either" + ); +} + +/// T-1130 behavior 7: a second connection attempting `role: Player` while a +/// Player is already connected gets a clean, immediate rejection (the +/// connection closes right after its handshake completes) — NOT the +/// original starvation bug (silent hang forever), and NOT a crash or +/// disruption to the existing Player's session. +#[test] +fn second_player_attempt_is_cleanly_rejected() { + use bevy_ecs::system::RunSystemOnce; + use settled_reach_server::bridge::BridgeResource; + + let (mut world, addr) = new_multi_connection_world(); + + let first_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect (first player)"); + client_handshake(stream, ConnectionRole::Player) + }); + drive_accept_loop_until(&mut world, |w| w.resource::().has_player()); + let _first_stream = first_handle.join().expect("first player thread panicked"); + + let second_handle = thread::spawn(move || { + let mut stream = TcpStream::connect(addr).expect("failed to connect (second player)"); + // Read handshake and send StartupMessage{role: Player} exactly like + // a normal client — the rejection happens AFTER this, not by + // refusing the handshake itself (the connection is genuinely + // accepted and handshaken; it's the ROLE promotion that's refused). + stream = client_handshake(stream, ConnectionRole::Player); + // The server closes the connection right after — prove it's a + // clean disconnect (not the original silent-hang bug) by reading + // until EOF. read_framed blocks here (this stream is never set + // non-blocking), which is deliberate: it's this thread's own proof + // that the disconnect actually happens — if the original starvation + // bug were still present, this call would hang forever with no + // internal deadline of its own. The OUTER test loop below bounds + // total test time via `second_handle.is_finished()` polling against + // ITS OWN 5s deadline, so a regression here still fails the test in + // bounded time rather than hanging the test suite. + match read_framed(&mut stream) { + Ok(None) => {} // clean EOF — the expected rejection signal + Ok(Some(_)) => panic!("second player attempt must never receive a message"), + Err(_) => {} // connection reset also counts as "rejected, not hung" + } + }); + + // Drive the accept-loop until the second connection has been processed + // (it will never appear as a reader OR a second player — reader_count + // stays 0 and has_player stays true throughout, which is exactly the + // rejection this test verifies). + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !second_handle.is_finished() { + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for second player rejection to complete" + ); + world + .run_system_once(settled_reach_server::bridge::accept_new_connections) + .expect("accept_new_connections failed to run"); + thread::sleep(std::time::Duration::from_millis(5)); + } + second_handle.join().expect("second player thread panicked"); + + assert!( + world.resource::().has_player(), + "the original player connection must be untouched by the rejected second attempt" + ); + assert_eq!( + world.resource::().reader_count(), + 0, + "a rejected second-Player attempt must never be silently installed as a reader" + ); +} + +/// Minimal `ObserverSnapshot` for the tests above — same shape as +/// `snapshot_roundtrip_over_tcp`'s at the top of this file, parameterized +/// only by `tick` (the one field these tests assert on). +fn sample_snapshot(tick: u64) -> ObserverSnapshot { + ObserverSnapshot { + tick, + game_time: GameTime { + day: 0, + time_of_day: 0, + day_phase: DayPhase::Morning, + tick_rate: TickRate::Full, + }, + player_facing: FacingDirection::North, + player_stance: MovementStance::default(), + player_inventory: vec![], + entities: vec![], + visible_tiles: vec![], + nearby_interactions: vec![], + current_monologue: None, + pending_recognitions: vec![], + dialogue_response: None, + blocked_entities: vec![], + scan_events: vec![], + sound_events: vec![], + follow_state: None, + character_pressure: None, + rng_seed: None, + poi_list: vec![], + examine_result: None, + player_knowledge: None, + save_result: None, + triangle_crisis_events: vec![], + state_hash: None, + debug_response: None, + sim_errors: vec![], + current_ticker: None, + settings_response: None, + economy_snapshot: None, + bookmark_catalog: None, + } +} diff --git a/server/tests/layer3.rs b/server/tests/layer3.rs index 5fd5d6ca7..d01564e93 100644 --- a/server/tests/layer3.rs +++ b/server/tests/layer3.rs @@ -81,7 +81,10 @@ fn server_subprocess_sends_snapshot_on_connect() { rmp_serde::from_slice(&handshake_frame).expect("deserialize HandshakeMessage"); // 5. Send StartupMessage with world_seed (#175) - let startup = StartupMessage { world_seed: 42 }; + let startup = StartupMessage { + world_seed: 42, + role: ConnectionRole::Player, + }; let startup_payload = rmp_serde::to_vec_named(&startup).expect("serialize StartupMessage"); write_framed(&mut writer, &startup_payload).expect("send StartupMessage to server");