diff --git a/server/src/bridge/local.rs b/server/src/bridge/local.rs index 30d38a657..ae3f24afa 100644 --- a/server/src/bridge/local.rs +++ b/server/src/bridge/local.rs @@ -2,7 +2,8 @@ // Implements D-020 subprocess/IPC architecture // Deterministic client-server communication via Unix domain sockets -use super::{BridgeError, ObserverSnapshot, PlayerInput, SimBridge}; +use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge}; +use crate::atlas::layer_proxy::AtlasLayerResponse; use crate::bridge::framing::{read_framed, write_framed}; use std::fs; use std::io::{BufReader, BufWriter}; @@ -123,37 +124,27 @@ impl SimBridge for LocalBridge { Ok(()) } - fn receive_inputs(&self) -> Result, BridgeError> { + fn receive(&self) -> Result, BridgeError> { let mut reader = self .reader .lock() .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?; match read_framed(reader.get_mut())? { - Some(payload) => match rmp_serde::from_slice::>(&payload) { - Ok(inputs) => { - tracing::trace!("received {} inputs", inputs.len()); - Ok(inputs) - } - Err(e) => { - let dump_len = payload.len().min(256); - tracing::error!( - "deserialization failed: {}. Raw bytes ({} of {} total): {:02x?}", - e, - dump_len, - payload.len(), - &payload[..dump_len] - ); - Err(BridgeError::DeserializationWithDump(format!( - "{} (payload {} bytes)", - e, - payload.len() - ))) - } - }, + Some(payload) => decode_inbound(&payload).map(Some), None => Err(BridgeError::Disconnected), } } + + fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> { + let payload = rmp_serde::to_vec_named(resp)?; + let mut writer = self + .writer + .lock() + .map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?; + write_framed(writer.get_mut(), &payload)?; + Ok(()) + } } impl Drop for LocalBridge { diff --git a/server/src/bridge/mod.rs b/server/src/bridge/mod.rs index d26099ccc..def1aee5d 100644 --- a/server/src/bridge/mod.rs +++ b/server/src/bridge/mod.rs @@ -6,6 +6,8 @@ use bevy_app::prelude::*; use bevy_ecs::prelude::*; use bevy_ecs::schedule::IntoScheduleConfigs; +use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse}; + pub mod debug; pub mod framing; pub mod local; @@ -33,6 +35,44 @@ pub enum BridgeError { 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 an +/// `AtlasLayerRequest` frame is a *map*, so they are distinguishable without a +/// wire-level type tag (existing frames are byte-unchanged — additive). +#[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), +} + +/// Demux a received frame payload into an [`Inbound`] (D-225). Tries +/// `Vec` (array), then `AtlasLayerRequest` (map); a frame that is +/// neither 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)); + } + match rmp_serde::from_slice::(payload) { + Ok(req) => Ok(Inbound::AtlasRequest(req)), + Err(e) => { + let dump_len = payload.len().min(256); + tracing::error!( + "inbound decode failed (neither inputs nor atlas request): {}. 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 { @@ -49,8 +89,12 @@ pub trait SimBridge: Send + Sync { /// Send an observer snapshot to the client fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError>; - /// Receive player inputs from the client - fn receive_inputs(&self) -> Result, BridgeError>; + /// Receive one inbound message, or `None` if no frame is ready this tick. + /// The single client→server stream is demuxed by frame shape (D-225). + 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>; } /// BridgeResource: Bevy Resource wrapper for SimBridge trait object @@ -78,8 +122,12 @@ impl BridgeResource { self.inner.send_snapshot(snapshot) } - pub fn receive_inputs(&self) -> Result, BridgeError> { - self.inner.receive_inputs() + pub fn receive(&self) -> Result, BridgeError> { + self.inner.receive() + } + + pub fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> { + self.inner.send_atlas_response(resp) } } @@ -105,13 +153,14 @@ pub fn receive_bridge_inputs( mut running: ResMut, handshake: Res, mut error_buffer: ResMut, + mut atlas_requests: ResMut, time: Option>, ) { let Some(bridge) = bridge else { return }; let current_tick = time.as_ref().map(|t| t.tick).unwrap_or(0); - match bridge.receive_inputs() { - Ok(inputs) => { + match 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)", @@ -129,6 +178,10 @@ pub fn receive_bridge_inputs( input_queue.push(input); } } + Ok(Some(Inbound::AtlasRequest(req))) => { + atlas_requests.0.push(req); + } + Ok(None) => {} Err(BridgeError::Disconnected) => { tracing::info!("Client disconnected, shutting down"); running.0 = false; @@ -207,6 +260,11 @@ impl Default for ServerRunning { } } +/// Inbound atlas layer requests routed off the bridge (#969, D-225), drained by +/// the proxy serve system in `PreInput`. +#[derive(Resource, Default)] +pub struct AtlasRequestBuffer(pub Vec); + /// Bridge plugin for client-server communication /// Abstracts transport layer (LocalBridge/NetworkBridge) pub struct BridgePlugin; @@ -222,6 +280,7 @@ impl Plugin for BridgePlugin { .init_resource::() .init_resource::() .init_resource::() + .init_resource::() // 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)) @@ -274,3 +333,30 @@ impl Plugin for BridgePlugin { 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, + }; + 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()); + } +} diff --git a/server/src/bridge/tcp.rs b/server/src/bridge/tcp.rs index a3df7c985..64f850cd8 100644 --- a/server/src/bridge/tcp.rs +++ b/server/src/bridge/tcp.rs @@ -3,7 +3,8 @@ // Deterministic client-server communication via TCP sockets // Used for Godot client which lacks Unix socket support -use super::{BridgeError, ObserverSnapshot, PlayerInput, SimBridge}; +use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge}; +use crate::atlas::layer_proxy::AtlasLayerResponse; use crate::bridge::framing::{read_framed, write_framed}; use std::io::{BufReader, BufWriter}; use std::net::{SocketAddr, TcpListener, TcpStream}; @@ -197,38 +198,33 @@ impl SimBridge for TcpBridge { Ok(()) } - fn receive_inputs(&self) -> Result, BridgeError> { + fn receive(&self) -> Result, BridgeError> { let mut reader = self .reader .lock() .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?; match read_framed(reader.get_mut()) { - Ok(Some(payload)) => match rmp_serde::from_slice::>(&payload) { - Ok(inputs) => { - tracing::trace!("received {} inputs", inputs.len()); - Ok(inputs) - } - Err(e) => { - let dump_len = payload.len().min(256); - tracing::error!( - "deserialization failed: {}. Raw bytes ({} of {} total): {:02x?}", - e, - dump_len, - payload.len(), - &payload[..dump_len] - ); - Err(BridgeError::DeserializationWithDump(format!( - "{} (payload {} bytes)", - e, - payload.len() - ))) - } - }, + Ok(Some(payload)) => decode_inbound(&payload).map(Some), Ok(None) => Err(BridgeError::Disconnected), // Non-blocking socket: no data available this tick — not an error. - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(vec![]), + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(None), Err(e) => Err(BridgeError::Io(e)), } } + + fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> { + let payload = rmp_serde::to_vec_named(resp)?; + let mut writer = self + .writer + .lock() + .map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?; + // Toggle to blocking for reliable write delivery (same as send_snapshot). + let stream = writer.get_mut(); + stream.set_nonblocking(false).map_err(BridgeError::Io)?; + let result = write_framed(stream, &payload); + stream.set_nonblocking(true).map_err(BridgeError::Io)?; + result?; + Ok(()) + } } diff --git a/server/tests/bridge_ipc.rs b/server/tests/bridge_ipc.rs index a9182648c..a94d8a004 100644 --- a/server/tests/bridge_ipc.rs +++ b/server/tests/bridge_ipc.rs @@ -3,7 +3,7 @@ use settled_reach_server::bridge::framing::{read_framed, write_framed}; use settled_reach_server::bridge::local::LocalBridge; use settled_reach_server::bridge::types::*; -use settled_reach_server::bridge::SimBridge; +use settled_reach_server::bridge::{Inbound, SimBridge}; use settled_reach_server::simulation::time::{DayPhase, TickRate}; use std::os::unix::net::UnixStream; use std::path::PathBuf; @@ -116,7 +116,10 @@ fn input_roundtrip_over_unix_socket() { let server_handle = thread::spawn(move || { let bridge = LocalBridge::accept(&server_path).expect("failed to accept"); - let inputs = bridge.receive_inputs().expect("failed to receive inputs"); + let inputs = match bridge.receive().expect("failed to receive") { + Some(Inbound::Inputs(inputs)) => inputs, + other => panic!("expected inputs, got {other:?}"), + }; assert_eq!(inputs.len(), 2); assert_eq!(inputs[0].tick, 10); diff --git a/server/tests/bridge_tcp.rs b/server/tests/bridge_tcp.rs index 8dac6af4b..d278925ad 100644 --- a/server/tests/bridge_tcp.rs +++ b/server/tests/bridge_tcp.rs @@ -3,7 +3,7 @@ use settled_reach_server::bridge::framing::{read_framed, write_framed}; use settled_reach_server::bridge::tcp::TcpBridge; use settled_reach_server::bridge::types::*; -use settled_reach_server::bridge::SimBridge; +use settled_reach_server::bridge::{Inbound, SimBridge}; use settled_reach_server::simulation::time::{DayPhase, TickRate}; use std::net::{TcpListener, TcpStream}; use std::thread; @@ -101,8 +101,8 @@ fn input_roundtrip_over_tcp() { // Non-blocking socket: retry until data arrives or timeout. let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let inputs = loop { - match bridge.receive_inputs() { - Ok(inputs) if !inputs.is_empty() => break inputs, + match bridge.receive() { + Ok(Some(Inbound::Inputs(inputs))) if !inputs.is_empty() => break inputs, Ok(_) => { assert!( std::time::Instant::now() < deadline, @@ -168,16 +168,16 @@ fn tcp_bridge_eof_returns_error() { // Non-blocking socket: retry until we get Disconnected or timeout. let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); loop { - match bridge.receive_inputs() { - Ok(inputs) if inputs.is_empty() => { - // WouldBlock — client hasn't disconnected yet, retry + match bridge.receive() { + Ok(None) => { + // No data yet — client hasn't disconnected, retry assert!( std::time::Instant::now() < deadline, "timed out waiting for EOF" ); thread::sleep(std::time::Duration::from_millis(1)); } - Ok(inputs) => panic!("expected Disconnected error, got {} inputs", inputs.len()), + Ok(Some(_)) => panic!("expected Disconnected error, got a message"), Err(settled_reach_server::bridge::BridgeError::Disconnected) => break, Err(e) => panic!("expected Disconnected error, got: {}", e), }