// Bridge module - Client-server communication // Implements D-020 subprocess/IPC architecture // MessagePack serialization for Rust<->Godot communication use bevy_app::prelude::*; use bevy_ecs::prelude::*; use bevy_ecs::schedule::IntoScheduleConfigs; use crate::atlas::atlas_data_proxy::{ CityNamesRequest, CityNamesResponse, StarMapRequest, StarMapResponse, }; use crate::atlas::browse_proxy::{BrowseRequest, BrowseResponse}; use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse}; use crate::bridge::tcp::TcpBridge; pub mod debug; pub mod framing; pub mod local; pub mod tcp; pub mod text_renderer; pub mod types; pub use types::*; /// Error type for bridge operations #[derive(Debug, thiserror::Error)] pub enum BridgeError { #[error("serialization error: {0}")] Serialization(#[from] rmp_serde::encode::Error), #[error("deserialization error: {0}")] Deserialization(#[from] rmp_serde::decode::Error), #[error("deserialization error (raw bytes logged): {0}")] DeserializationWithDump(String), #[error("io error: {0}")] Io(#[from] std::io::Error), #[error("transport error: {0}")] Transport(String), #[error("client disconnected")] Disconnected, #[error("internal mutex poisoned: {0}")] MutexPoisoned(String), } /// One decoded inbound message. The client→server stream is a single demuxed /// channel (D-225): a `Vec` frame is a MessagePack *array* and /// every request type below is a *map*, so array vs. map alone separates /// inputs from everything else without a wire-level type tag (existing frames /// are byte-unchanged — additive). /// /// **Disambiguating the three map shapes (D-225 extension, T-949):** /// `AtlasLayerRequest{body_id, up_to}` was the only map shape until T-949 /// added `StarMapRequest`/`CityNamesRequest` alongside it. serde's derived /// `Deserialize` silently ignores unknown fields by default, so "does this /// struct parse at all" is not a safe discriminator once more than one map /// shape can share a field name (`CityNamesRequest` and `AtlasLayerRequest` /// both key on `body_id`) — a payload carrying every field either shape wants /// would ambiguously satisfy both. Rather than retrofit /// `#[serde(deny_unknown_fields)]` onto the existing `AtlasLayerRequest` (risking /// breakage if any already-deployed client encoder harmlessly sends extra /// fields), the two *new* map shapes each carry a mandatory boolean /// discriminator field the others don't have at all (`star_map` / /// `city_names`): a missing required field is a hard deserialize failure, not /// a silent ignore, so no *minimal well-formed* instance of one shape /// satisfies another — and [`decode_inbound`] additionally REJECTS union /// frames that carry more than one shape's discriminators outright (PR #176 /// review H1). `AtlasLayerRequest` itself is untouched byte-for-byte. /// /// **Ceiling (D-225 trajectory):** [`BrowseRequest`] (T-1131) is the FIFTH /// map shape and, per the ceiling this doc already called at four, the last /// one this hand-rolled scheme should ever carry — it stays at five only /// because six entity kinds x two forms were folded into ONE new shape /// (`browse`'s own internal `kind`/`query` enums pick the sub-behavior, /// exactly as `AtlasLayerRequest.up_to: CascadeLayer` already does) rather /// than added as twelve more top-level shapes. The next genuinely NEW /// inbound shape (a sixth) must migrate the channel to the tagged-envelope /// framing D-225 deferred — do not add a sixth probe. #[derive(Debug)] pub enum Inbound { /// A batch of player inputs (the gameplay path). Inputs(Vec), /// An atlas layer-stream request (#969, D-225). AtlasRequest(AtlasLayerRequest), /// A star-map dataset request (T-949a). StarMapRequest(StarMapRequest), /// A per-body city-names request (T-949b). CityNamesRequest(CityNamesRequest), /// A data-browser request — one of the six D-254 §4 v1 entity kinds /// (T-1131). BrowseRequest(BrowseRequest), } /// Key-presence probe for the defensive multi-shape check in /// [`decode_inbound`]: `Option` records whether a key exists /// without caring about its value or type, so a union frame is detected even /// when the individual values wouldn't parse as their target types. #[derive(serde::Deserialize)] struct ShapeProbe { body_id: Option, up_to: Option, star_map: Option, city_names: Option, browse: Option, } /// Demux a received frame payload into an [`Inbound`] (D-225, T-949, T-1131). /// Tries, in order: `Vec` (array) → `AtlasLayerRequest` (map, /// `body_id`+`up_to`) → `StarMapRequest` (map, `star_map` discriminator) → /// `CityNamesRequest` (map, `city_names` discriminator + `body_id`) → /// `BrowseRequest` (map, `browse` discriminator). /// /// Mutual exclusivity is enforced, not assumed: no minimal well-formed /// instance of one shape satisfies another (see the [`Inbound`] doc), and a /// defensive pre-check rejects any map frame carrying the discriminators of /// more than one shape — e.g. a buggy encoder emitting /// `{"star_map": true, "city_names": true, ...}` — instead of silently /// routing it to whichever shape is tried first (PR #176 review H1). A frame /// satisfying none of the five shapes is a genuinely malformed input frame. pub fn decode_inbound(payload: &[u8]) -> Result { if let Ok(inputs) = rmp_serde::from_slice::>(payload) { return Ok(Inbound::Inputs(inputs)); } // Defensive multi-shape rejection: serde ignores unknown fields, so a // union frame would otherwise route silently by try-order. Unreachable // from the shipped client encoders (each sends one minimal shape) — this // guards buggy or adversarial frames. if let Ok(probe) = rmp_serde::from_slice::(payload) { let atlas = probe.body_id.is_some() && probe.up_to.is_some(); let star_map = probe.star_map.is_some(); let city_names = probe.city_names.is_some(); let browse = probe.browse.is_some(); let shapes = usize::from(atlas) + usize::from(star_map) + usize::from(city_names) + usize::from(browse); if shapes > 1 { let dump_len = payload.len().min(256); tracing::error!( "inbound frame matches {} request shapes at once (atlas={}, star_map={}, city_names={}, browse={}) — rejecting ambiguous frame. Raw ({} of {} bytes): {:02x?}", shapes, atlas, star_map, city_names, browse, dump_len, payload.len(), &payload[..dump_len] ); return Err(BridgeError::DeserializationWithDump(format!( "ambiguous inbound frame matches {shapes} request shapes (payload {} bytes)", payload.len() ))); } } if let Ok(req) = rmp_serde::from_slice::(payload) { return Ok(Inbound::AtlasRequest(req)); } if let Ok(req) = rmp_serde::from_slice::(payload) { return Ok(Inbound::StarMapRequest(req)); } if let Ok(req) = rmp_serde::from_slice::(payload) { return Ok(Inbound::CityNamesRequest(req)); } match rmp_serde::from_slice::(payload) { Ok(req) => Ok(Inbound::BrowseRequest(req)), Err(e) => { let dump_len = payload.len().min(256); tracing::error!( "inbound decode failed (matches no known frame shape): {}. Raw ({} of {} bytes): {:02x?}", e, dump_len, payload.len(), &payload[..dump_len] ); Err(BridgeError::DeserializationWithDump(format!( "{e} (payload {} bytes)", payload.len() ))) } } } /// Abstracts transport layer (D-020) /// Implemented by LocalBridge (stdio) and future NetworkBridge pub trait SimBridge: Send + Sync { /// Send the protocol handshake as the first framed message (#555). /// Must be called exactly once, immediately after connection, before /// any ObserverSnapshot is sent. fn send_handshake(&self) -> Result<(), BridgeError>; /// Receive the client's startup message containing the world seed (#175). /// Called exactly once, after send_handshake(), before entering the tick loop. /// Blocks until the client sends the message. fn receive_startup(&self) -> Result; /// Send an observer snapshot to the client fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError>; /// Receive one inbound message, or `None` if no complete frame is ready. /// The single client→server stream is demuxed by frame shape (D-225). /// `receive_bridge_inputs` loops this until `None` (T-1045), so the /// transport behind `BridgeResource` must not block when no frame is /// buffered (TcpBridge is non-blocking; LocalBridge blocks — test-only). fn receive(&self) -> Result, BridgeError>; /// Send an atlas layer-stream response to the client (#969, D-225). fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError>; /// Send a star-map response to the client (T-949a). fn send_star_map_response(&self, resp: &StarMapResponse) -> Result<(), BridgeError>; /// Send a city-names response to the client (T-949b). fn send_city_names_response(&self, resp: &CityNamesResponse) -> Result<(), BridgeError>; /// Send a browse response to the client (T-1131). fn send_browse_response(&self, resp: &BrowseResponse) -> Result<(), BridgeError>; } /// 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 { 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 { 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())), } } /// 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 { 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> { match &self.player { Some(c) => c.bridge.send_snapshot(snapshot), None => Err(BridgeError::Disconnected), } } /// 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) } /// 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(()) } } } /// 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(()) } } } /// 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(()) } } } /// Send a browse response to exactly the connection that requested it /// (T-1131 — same per-connection routing D-254 §2 established for /// atlas/star-map/city-names). pub fn send_browse_response_to( &self, id: ConnectionId, resp: &BrowseResponse, ) -> Result<(), BridgeError> { match self.connection(id) { Some(c) => c.bridge.send_browse_response(resp), None => { tracing::debug!( "browse response for {:?} dropped — connection {:?} no longer present", resp.kind, id ); Ok(()) } } } } /// Tracks whether the protocol handshake has been sent (#555). /// Inserted by BridgePlugin as Pending. Set to Complete in main.rs after /// `send_handshake()` succeeds. `receive_bridge_inputs` logs a warning /// if inputs arrive while still Pending. #[derive(Resource, Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum HandshakeState { /// Handshake not yet sent. Inputs arriving in this state trigger a warning. #[default] Pending, /// Handshake sent. Normal operation. Complete, } /// Per-tick cap on drained inbound frames (T-1045) — a safety valve so a /// client flooding the stream cannot starve the simulation tick. Generous: /// normal traffic is one input batch plus the occasional atlas request. const MAX_INBOUND_FRAMES_PER_TICK: usize = 64; /// 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>, mut input_queue: ResMut, mut running: ResMut, handshake: Res, mut error_buffer: ResMut, mut atlas_requests: ResMut, mut star_map_requests: ResMut, mut city_names_requests: ResMut, mut browse_requests: ResMut, time: Option>, ) { let Some(mut bridge) = bridge else { return }; let current_tick = time.as_ref().map(|t| t.tick).unwrap_or(0); // -- 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)); } Ok(Some(Inbound::BrowseRequest(req))) => { browse_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!( "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; } } 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(Some(Inbound::BrowseRequest(req))) => { browse_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; } 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; } } } } for id in disconnected_readers .into_iter() .chain(to_disconnect_for_violations) { bridge.remove_reader(id); } } /// 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>, mut buffer: ResMut, mut running: ResMut, ) { 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!("Player disconnected during send, shutting down"); } BridgeError::MutexPoisoned(msg) => { tracing::error!("Player bridge mutex poisoned during send: {}", msg); } _ => { tracing::error!("Player bridge send error: {}", e); } } running.0 = false; bridge.player = None; } } } /// Server running flag resource #[derive(Resource, Debug, Clone)] pub struct ServerRunning(pub bool); impl Default for ServerRunning { fn default() -> Self { Self(true) } } /// Inbound atlas layer requests routed off the bridge (#969, D-225), drained by /// 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<(ConnectionId, AtlasLayerRequest)>); /// Outbound atlas layer responses, filled by the proxy serve system and flushed /// to the client in `PostSnapshot` (#969, D-225). Connection-tagged (D-254 §2). #[derive(Resource, Default)] pub struct AtlasResponseBuffer(pub Vec<(ConnectionId, AtlasLayerResponse)>); /// 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 (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`. Connection-tagged (D-254 §2). #[derive(Resource, Default)] 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). Connection-tagged (D-254 §2). #[derive(Resource, Default)] pub struct StarMapResponseBuffer(pub Vec<(ConnectionId, StarMapResponse)>); /// 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 (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`. Connection-tagged (D-254 §2). #[derive(Resource, Default)] 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). Connection-tagged /// (D-254 §2). #[derive(Resource, Default)] pub struct CityNamesResponseBuffer(pub Vec<(ConnectionId, CityNamesResponse)>); /// 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 (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 {} to {:?}: {}", resp.body_id, id, e ); } } } /// Inbound data-browser requests routed off the bridge (T-1131), drained by /// the proxy serve system in `PreInput`. Connection-tagged (D-254 §2). #[derive(Resource, Default)] pub struct BrowseRequestBuffer(pub Vec<(ConnectionId, BrowseRequest)>); /// Outbound browse responses, filled by the proxy serve system and flushed /// to the client in `PostSnapshot` (T-1131). Connection-tagged (D-254 §2). #[derive(Resource, Default)] pub struct BrowseResponseBuffer(pub Vec<(ConnectionId, BrowseResponse)>); /// Flush buffered browse responses to their requesting connections (T-1131 — /// same per-connection routing D-254 §2 established for /// atlas/star-map/city-names). A failed send is logged but not fatal. pub fn send_browse_responses( bridge: Option>, mut buffer: ResMut, ) { let Some(bridge) = bridge else { return }; for (id, resp) in buffer.0.drain(..) { if let Err(e) = bridge.send_browse_response_to(id, &resp) { tracing::warn!( "failed to send browse response for {:?} to {:?}: {}", resp.kind, 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; impl Plugin for BridgePlugin { fn build(&self, app: &mut App) { use crate::simulation::time::sim_not_paused; use crate::tick_phases::TickPhase; app.init_resource::() .init_resource::() .init_resource::() .init_resource::() .init_resource::() .init_resource::() .init_resource::() .init_resource::() .init_resource::() .init_resource::() .init_resource::() .init_resource::() .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)) .add_systems(Update, send_atlas_responses.in_set(TickPhase::PostSnapshot)) .add_systems( Update, send_star_map_responses.in_set(TickPhase::PostSnapshot), ) .add_systems( Update, send_city_names_responses.in_set(TickPhase::PostSnapshot), ) .add_systems( Update, send_browse_responses.in_set(TickPhase::PostSnapshot), ) // Debug commands — Snapshot phase .add_systems( Update, debug::handle_debug_commands.in_set(TickPhase::Snapshot), ) // Monologue chain — Simulation phase, strict intra-phase sequence. // trigger_event_monologue must run after conversations + sound (also Simulation). // T-970: TickPhase::Simulation is not set-gated (see // social_plugin.rs's collect_sound_events exemption) — this whole // chain is genuine world-advancing dialogue/monologue logic, so // it gates safely on its own. trigger_event_monologue's // .after(collect_sound_events) still holds while gated: // collect_sound_events itself is never gated, and an ordering // edge onto a skipped predecessor is trivially satisfied. .add_systems( Update, ( crate::simulation::monologue::trigger_monologue, crate::simulation::monologue::trigger_recognition_monologue .after(crate::simulation::monologue::trigger_monologue) .after(crate::perception::anomaly::detect_anomalies), crate::simulation::monologue::process_sprint_anomaly_monologue .after(crate::simulation::monologue::trigger_recognition_monologue), crate::simulation::monologue::trigger_event_monologue .after(crate::simulation::monologue::process_sprint_anomaly_monologue) .after(crate::simulation::sound::collect_sound_events) .after(crate::simulation::dialogue::process_walk_away), crate::simulation::monologue::process_contradiction_monologue .after(crate::simulation::monologue::trigger_event_monologue), ) .run_if(sim_not_paused) .in_set(TickPhase::Simulation), ) // Observation systems — Simulation phase (reads positions, feeds snapshot). // T-970: gates safely on its own (see note above) — these compute // "current state" (visibility, nearby interactions) that's valid // as long as nothing moved, which holds while paused since // Movement is frozen too; unlike SoundEventQueue, nothing here // depends on being refreshed on a tick where nothing changed. .add_systems( Update, ( crate::perception::observer::compute_visibility_geometry, crate::simulation::interaction::compute_nearby_interactions, ) .run_if(sim_not_paused) .in_set(TickPhase::Simulation), ) // Observer snapshot assembly — Snapshot phase .add_systems( Update, crate::perception::observer::compute_observer_snapshot.in_set(TickPhase::Snapshot), ) // Post-snapshot: emit observation events .add_systems( Update, crate::perception::observation::emit_observation_events .in_set(TickPhase::PostSnapshot), ); tracing::debug!("BridgePlugin initialized"); } } #[cfg(test)] mod inbound_tests { use super::*; use crate::atlas::cascade::CascadeLayer; #[test] fn demux_routes_inputs_and_atlas_requests() { // A Vec frame (msgpack array) → Inbound::Inputs. let inputs: Vec = vec![]; let frame = rmp_serde::to_vec_named(&inputs).unwrap(); assert!(matches!(decode_inbound(&frame), Ok(Inbound::Inputs(v)) if v.is_empty())); // An AtlasLayerRequest frame (msgpack map) → Inbound::AtlasRequest. let req = AtlasLayerRequest { body_id: "GJ1c".into(), up_to: CascadeLayer::Topography, window_center: None, window_n: 0, }; let frame = rmp_serde::to_vec_named(&req).unwrap(); assert!( matches!(decode_inbound(&frame), Ok(Inbound::AtlasRequest(r)) if r.body_id == "GJ1c") ); // Neither shape → a malformed-frame error. assert!(decode_inbound(&[0xff, 0xff]).is_err()); } #[test] fn demux_routes_star_map_requests() { let req = StarMapRequest { star_map: true }; let frame = rmp_serde::to_vec_named(&req).unwrap(); assert!(matches!( decode_inbound(&frame), Ok(Inbound::StarMapRequest(r)) if r.star_map )); } #[test] fn demux_routes_city_names_requests() { let req = CityNamesRequest { city_names: true, body_id: "GJ1c".into(), }; let frame = rmp_serde::to_vec_named(&req).unwrap(); assert!(matches!( decode_inbound(&frame), Ok(Inbound::CityNamesRequest(r)) if r.body_id == "GJ1c" )); } /// T-949: the array-vs-map trick (D-225) still separates `Inputs` from /// everything else, and the three map shapes' discriminator fields keep /// them mutually exclusive — each of the four frame shapes decodes to /// exactly its own `Inbound` variant, never a neighbor's. #[test] fn inbound_disambiguation_is_unambiguous_across_all_four_shapes() { let inputs_frame = rmp_serde::to_vec_named(&Vec::::new()).unwrap(); let atlas_frame = rmp_serde::to_vec_named(&AtlasLayerRequest { body_id: "GJ1c".into(), up_to: CascadeLayer::Topography, window_center: None, window_n: 0, }) .unwrap(); let star_map_frame = rmp_serde::to_vec_named(&StarMapRequest { star_map: true }).unwrap(); let city_names_frame = rmp_serde::to_vec_named(&CityNamesRequest { city_names: true, body_id: "GJ1c".into(), }) .unwrap(); assert!(matches!( decode_inbound(&inputs_frame), Ok(Inbound::Inputs(_)) )); assert!(matches!( decode_inbound(&atlas_frame), Ok(Inbound::AtlasRequest(_)) )); assert!(matches!( decode_inbound(&star_map_frame), Ok(Inbound::StarMapRequest(_)) )); assert!(matches!( decode_inbound(&city_names_frame), Ok(Inbound::CityNamesRequest(_)) )); // Cross-check: an AtlasLayerRequest frame must NOT decode as // CityNamesRequest even though both key on `body_id` — the missing // `city_names` discriminator makes that a hard failure, not a silent // "extra field ignored" success either shape could show without it. assert!(rmp_serde::from_slice::(&atlas_frame).is_err()); // And a CityNamesRequest frame must NOT decode as AtlasLayerRequest — // it's missing the required `up_to` field. assert!(rmp_serde::from_slice::(&city_names_frame).is_err()); } /// PR #176 review H1: a union frame carrying more than one shape's /// discriminators must be REJECTED, not silently routed to whichever /// shape `decode_inbound` happens to try first. #[test] fn ambiguous_union_frame_is_rejected() { #[derive(serde::Serialize)] struct StarAndCity { star_map: bool, city_names: bool, body_id: String, } let frame = rmp_serde::to_vec_named(&StarAndCity { star_map: true, city_names: true, body_id: "GJ1c".into(), }) .unwrap(); assert!( decode_inbound(&frame).is_err(), "star_map+city_names union frame must be rejected" ); #[derive(serde::Serialize)] struct AtlasAndStar { body_id: String, up_to: CascadeLayer, star_map: bool, } let frame = rmp_serde::to_vec_named(&AtlasAndStar { body_id: "GJ1c".into(), up_to: CascadeLayer::Topography, star_map: true, }) .unwrap(); assert!( decode_inbound(&frame).is_err(), "atlas+star_map union frame must be rejected" ); // T-1131 (PR #184 review): the FIFTH shape's discriminator (`browse`) // must participate in the same union rejection — a well-formed // BrowseRequest smuggling another shape's discriminator alongside it // is rejected, not routed to whichever probe wins. #[derive(serde::Serialize)] struct BrowseAndStar { browse: bool, kind: crate::atlas::browse_proxy::BrowseEntityKind, query: crate::atlas::browse_proxy::BrowseQuery, star_map: bool, } let frame = rmp_serde::to_vec_named(&BrowseAndStar { browse: true, kind: crate::atlas::browse_proxy::BrowseEntityKind::StarSystem, query: crate::atlas::browse_proxy::BrowseQuery::Index { filter_system_id: None, }, star_map: true, }) .unwrap(); assert!( decode_inbound(&frame).is_err(), "browse+star_map union frame must be rejected" ); #[derive(serde::Serialize)] struct BrowseAndCity { browse: bool, kind: crate::atlas::browse_proxy::BrowseEntityKind, query: crate::atlas::browse_proxy::BrowseQuery, city_names: bool, body_id: String, } let frame = rmp_serde::to_vec_named(&BrowseAndCity { browse: true, kind: crate::atlas::browse_proxy::BrowseEntityKind::Body, query: crate::atlas::browse_proxy::BrowseQuery::Detail { id: "GJ1c".into() }, city_names: true, body_id: "GJ1c".into(), }) .unwrap(); assert!( decode_inbound(&frame).is_err(), "browse+city_names union frame must be rejected" ); } }