From 1b0e51456029ddad891fa5f0fbc08cf0f2c9f19f Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Wed, 11 Feb 2026 21:01:20 +0100 Subject: [PATCH] feat(bridge): add TCP transport for Godot client connection Godot has no Unix socket API, so TCP localhost is required for client-server IPC. TcpBridge implements SimBridge with the same framing protocol as LocalBridge. Includes accept/connect methods and three integration tests over TCP. Co-Authored-By: Claude Opus 4.6 --- server/src/bridge/tcp.rs | 114 +++++++++++++++++++++ server/tests/bridge_tcp.rs | 201 +++++++++++++++++++++++++++++++++++++ 2 files changed, 315 insertions(+) create mode 100644 server/src/bridge/tcp.rs create mode 100644 server/tests/bridge_tcp.rs diff --git a/server/src/bridge/tcp.rs b/server/src/bridge/tcp.rs new file mode 100644 index 000000000..02c4b5a83 --- /dev/null +++ b/server/src/bridge/tcp.rs @@ -0,0 +1,114 @@ +// TcpBridge - TCP localhost IPC implementation +// Implements D-020 subprocess/IPC architecture +// Deterministic client-server communication via TCP sockets +// Used for Godot client which lacks Unix socket support + +use super::{BridgeError, ObserverSnapshot, PlayerInput, SimBridge}; +use crate::bridge::framing::{read_framed, write_framed}; +use std::io::{BufReader, BufWriter}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::sync::Mutex; + +/// TcpBridge: TCP transport for client-server IPC +/// Same semantics as LocalBridge but over TCP localhost +pub struct TcpBridge { + reader: Mutex>, + writer: Mutex>, + local_addr: SocketAddr, +} + +impl TcpBridge { + /// Server-side: bind TCP listener and accept one connection. + /// Binds to the specified address (e.g., "127.0.0.1:0" for OS-assigned port). + /// Returns the bridge with the actual bound address available via local_addr(). + pub fn accept(addr: &str) -> Result { + tracing::info!("TcpBridge binding to {}", addr); + + let listener = TcpListener::bind(addr) + .map_err(|e| BridgeError::Transport(format!("failed to bind TCP socket: {}", e)))?; + + let local_addr = listener + .local_addr() + .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?; + + tracing::info!("TcpBridge listening on {}", local_addr); + + // Accept one connection + let (stream, peer_addr) = listener + .accept() + .map_err(|e| BridgeError::Transport(format!("failed to accept connection: {}", e)))?; + + tracing::info!( + "TcpBridge accepted connection from {} on {}", + peer_addr, + local_addr + ); + + // Clone stream for reader and writer + let reader_stream = stream.try_clone().map_err(|e| { + BridgeError::Transport(format!("failed to clone stream for reader: {}", e)) + })?; + + Ok(Self { + reader: Mutex::new(BufReader::new(reader_stream)), + writer: Mutex::new(BufWriter::new(stream)), + local_addr, + }) + } + + /// Client-side: connect to TCP address (for tests). + pub fn connect(addr: &str) -> Result { + tracing::info!("TcpBridge connecting to {}", addr); + + let stream = TcpStream::connect(addr).map_err(|e| { + BridgeError::Transport(format!("failed to connect to TCP socket: {}", e)) + })?; + + let local_addr = stream + .local_addr() + .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?; + + tracing::trace!("TcpBridge connected to {}", addr); + + // Clone stream for reader and writer + let reader_stream = stream.try_clone().map_err(|e| { + BridgeError::Transport(format!("failed to clone stream for reader: {}", e)) + })?; + + Ok(Self { + reader: Mutex::new(BufReader::new(reader_stream)), + writer: Mutex::new(BufWriter::new(stream)), + local_addr, + }) + } + + /// Get the local address (useful for OS-assigned port discovery in tests). + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } +} + +impl SimBridge for TcpBridge { + fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError> { + let payload = rmp_serde::to_vec_named(snapshot)?; + + let mut writer = self.writer.lock().expect("writer mutex poisoned"); + write_framed(writer.get_mut(), &payload)?; + + tracing::trace!("sent snapshot: tick={}", snapshot.tick); + Ok(()) + } + + fn receive_inputs(&self) -> Result, BridgeError> { + let mut reader = self.reader.lock().expect("reader mutex poisoned"); + + match read_framed(reader.get_mut())? { + Some(payload) => { + let inputs: Vec = rmp_serde::from_slice(&payload)?; + tracing::trace!("received {} inputs", inputs.len()); + Ok(inputs) + } + None => Err(BridgeError::Transport("client disconnected (EOF)".into())), + } + } +} diff --git a/server/tests/bridge_tcp.rs b/server/tests/bridge_tcp.rs new file mode 100644 index 000000000..ae1927b9a --- /dev/null +++ b/server/tests/bridge_tcp.rs @@ -0,0 +1,201 @@ +//! Integration tests for TcpBridge over TCP localhost (D-030 Layer 2: IPC roundtrip). + +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 std::net::{TcpListener, TcpStream}; +use std::sync::mpsc; +use std::thread; +use std::time::Duration; + +/// Helper to bind a TCP listener and return its address. +/// This allows tests to discover the OS-assigned port before calling TcpBridge::accept(). +fn bind_listener() -> (TcpListener, std::net::SocketAddr) { + let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); + let addr = listener.local_addr().expect("failed to get local address"); + (listener, addr) +} + +#[test] +fn snapshot_roundtrip_over_tcp() { + // Pre-bind listener to discover the port + let (listener, server_addr) = bind_listener(); + let addr_str = server_addr.to_string(); + + // Channel to signal when server is ready to accept + let (ready_tx, ready_rx) = mpsc::channel(); + + // Server thread: accept connection and send snapshot + let server_handle = thread::spawn(move || { + // Drop the pre-bound listener since TcpBridge::accept will bind its own + drop(listener); + + // Signal we're about to accept + ready_tx.send(()).expect("failed to send ready signal"); + + let bridge = TcpBridge::accept(&addr_str).expect("failed to bind and accept"); + + let snapshot = ObserverSnapshot { + tick: 42, + entities: vec![VisibleEntity { + entity_id: 100, + x: 10.5, + y: 20.3, + z: 0, + kind: EntityKind::Npc, + }], + }; + + bridge + .send_snapshot(&snapshot) + .expect("failed to send snapshot"); + }); + + // Wait for server thread to start + ready_rx + .recv_timeout(Duration::from_secs(2)) + .expect("server did not become ready"); + + // Give server time to bind and call accept() + thread::sleep(Duration::from_millis(100)); + + // Client: connect and receive snapshot + let stream = TcpStream::connect(server_addr).expect("failed to connect"); + let mut reader = std::io::BufReader::new(stream); + + let payload = read_framed(&mut reader) + .expect("failed to read frame") + .expect("unexpected EOF"); + + let snapshot: ObserverSnapshot = + rmp_serde::from_slice(&payload).expect("failed to deserialize"); + + assert_eq!(snapshot.tick, 42); + assert_eq!(snapshot.entities.len(), 1); + assert_eq!(snapshot.entities[0].entity_id, 100); + assert_eq!(snapshot.entities[0].x, 10.5); + assert_eq!(snapshot.entities[0].y, 20.3); + + server_handle.join().expect("server thread panicked"); +} + +#[test] +fn input_roundtrip_over_tcp() { + // Pre-bind listener to discover the port + let (listener, server_addr) = bind_listener(); + let addr_str = server_addr.to_string(); + + // Channel to signal when server is ready to accept + let (ready_tx, ready_rx) = mpsc::channel(); + + // Server thread: accept connection and receive inputs + let server_handle = thread::spawn(move || { + // Drop the pre-bound listener since TcpBridge::accept will bind its own + drop(listener); + + // Signal we're about to accept + ready_tx.send(()).expect("failed to send ready signal"); + + let bridge = TcpBridge::accept(&addr_str).expect("failed to bind and accept"); + + let inputs = bridge.receive_inputs().expect("failed to receive inputs"); + + assert_eq!(inputs.len(), 2); + assert_eq!(inputs[0].tick, 10); + assert_eq!(inputs[1].tick, 11); + + inputs + }); + + // Wait for server thread to start + ready_rx + .recv_timeout(Duration::from_secs(2)) + .expect("server did not become ready"); + + // Give server time to bind and call accept() + thread::sleep(Duration::from_millis(100)); + + // Client: connect and send inputs + let stream = TcpStream::connect(server_addr).expect("failed to connect"); + let mut writer = std::io::BufWriter::new(stream); + + let inputs = vec![ + PlayerInput { + tick: 10, + action: PlayerAction::MoveNorth, + }, + PlayerInput { + tick: 11, + action: PlayerAction::Interact, + }, + ]; + + let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize"); + write_framed(&mut writer, &payload).expect("failed to write frame"); + + // Drop writer to close connection and signal EOF to server + drop(writer); + + let received_inputs = server_handle.join().expect("server thread panicked"); + + // Verify actions survived the round-trip + match &received_inputs[0].action { + PlayerAction::MoveNorth => {} + _ => panic!("expected MoveNorth action"), + } + match &received_inputs[1].action { + PlayerAction::Interact => {} + _ => panic!("expected Interact action"), + } +} + +#[test] +fn tcp_bridge_eof_returns_error() { + // Pre-bind listener to discover the port + let (listener, server_addr) = bind_listener(); + let addr_str = server_addr.to_string(); + + // Channel to signal when server is ready to accept + let (ready_tx, ready_rx) = mpsc::channel(); + + // Server thread: accept connection and receive EOF + let server_handle = thread::spawn(move || { + // Drop the pre-bound listener since TcpBridge::accept will bind its own + drop(listener); + + // Signal we're about to accept + ready_tx.send(()).expect("failed to send ready signal"); + + let bridge = TcpBridge::accept(&addr_str).expect("failed to bind and accept"); + + // Attempt to receive inputs - should get EOF error + let result = bridge.receive_inputs(); + + assert!(result.is_err(), "expected error on EOF"); + match result { + Err(settled_reach_server::bridge::BridgeError::Transport(msg)) => { + assert!( + msg.contains("disconnected") || msg.contains("EOF"), + "expected disconnect/EOF error, got: {}", + msg + ); + } + _ => panic!("expected Transport error with disconnect/EOF message"), + } + }); + + // Wait for server thread to start + ready_rx + .recv_timeout(Duration::from_secs(2)) + .expect("server did not become ready"); + + // Give server time to bind and call accept() + thread::sleep(Duration::from_millis(100)); + + // Client: connect and immediately disconnect without sending data + let stream = TcpStream::connect(server_addr).expect("failed to connect"); + drop(stream); // Close connection immediately + + server_handle.join().expect("server thread panicked"); +}