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 <noreply@anthropic.com>
2026-02-11 21:01:20 +01:00
parent f6ce7d4046
commit 1b0e514560
2 changed files with 315 additions and 0 deletions
@@ -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<BufReader<TcpStream>>,
+    writer: Mutex<BufWriter<TcpStream>>,
+    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<Self, BridgeError> {
+        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<Self, BridgeError> {
+        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<Vec<PlayerInput>, BridgeError> {
+        let mut reader = self.reader.lock().expect("reader mutex poisoned");
+
+        match read_framed(reader.get_mut())? {
+            Some(payload) => {
+                let inputs: Vec<PlayerInput> = rmp_serde::from_slice(&payload)?;
+                tracing::trace!("received {} inputs", inputs.len());
+                Ok(inputs)
+            }
+            None => Err(BridgeError::Transport("client disconnected (EOF)".into())),
+        }
+    }
+}
@@ -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");
+}