From b6c4ecb3023d584438ce5c85878c4cb261a0e171 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Wed, 25 Feb 2026 12:57:59 +0100 Subject: [PATCH] feat(ci): protocol version handshake client + IPC benchmark (#556, #342) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #556: HANDSHAKING state in sim_bridge.gd — decodes first framed message as HandshakeMessage, validates vs Protocol.PROTOCOL_VERSION, 5s timeout, handshake_complete/handshake_failed signals. #342: IPC benchmark now reads and validates HandshakeMessage before starting the timing loop. Co-Authored-By: Claude Opus 4.6 --- client/scripts/autoloads/sim_bridge.gd | 67 +++++++++++++++++++++++++- server/tests/ipc_bench.rs | 32 ++++++------ 2 files changed, 83 insertions(+), 16 deletions(-) diff --git a/client/scripts/autoloads/sim_bridge.gd b/client/scripts/autoloads/sim_bridge.gd index fa91fb3c3..b765cc62a 100644 --- a/client/scripts/autoloads/sim_bridge.gd +++ b/client/scripts/autoloads/sim_bridge.gd @@ -1,7 +1,7 @@ extends Node # Connection states -enum ConnectionState { DISCONNECTED, CONNECTING, CONNECTED, ERROR } +enum ConnectionState { DISCONNECTED, CONNECTING, HANDSHAKING, CONNECTED, ERROR } var state: ConnectionState = ConnectionState.DISCONNECTED var test_mode: bool = OS.get_environment("SR_LIVE") != "1" # SR_LIVE=1 connects to real server @@ -27,9 +27,15 @@ const CONNECT_RETRY_INTERVAL: float = 0.1 # Seconds between retry attempts var _connect_retries: int = 0 var _retry_timer: float = 0.0 +# Handshake state (#556) +const HANDSHAKE_TIMEOUT_USEC: int = 5_000_000 # 5 seconds +var _handshake_start_usec: int = 0 + # Signals signal connection_state_changed(old_state: ConnectionState, new_state: ConnectionState) signal snapshot_received(snapshot: Dictionary) +signal handshake_complete(protocol_version: int) +signal handshake_failed(reason: String) func _ready() -> void: if test_mode: @@ -121,7 +127,8 @@ func _process(delta: float) -> void: _bridge.poll() match _bridge.get_status(): StreamPeerTCP.STATUS_CONNECTED: - _set_state(ConnectionState.CONNECTED) + _handshake_start_usec = Time.get_ticks_usec() + _set_state(ConnectionState.HANDSHAKING) StreamPeerTCP.STATUS_CONNECTING: pass # Still connecting, wait StreamPeerTCP.STATUS_ERROR: @@ -134,6 +141,62 @@ func _process(delta: float) -> void: _bridge = null # Reset and retry return + # HANDSHAKING state: read first framed message, validate HandshakeMessage (#556) + if state == ConnectionState.HANDSHAKING: + if _bridge == null: + _set_state(ConnectionState.ERROR) + return + _bridge.poll() + + # Check connection dropped during handshake + var bridge_status := _bridge.get_status() + if bridge_status == StreamPeerTCP.STATUS_ERROR or bridge_status == StreamPeerTCP.STATUS_NONE: + var reason := "Connection dropped during handshake" + push_error("SimBridge: %s" % reason) + handshake_failed.emit(reason) + _bridge = null + _set_state(ConnectionState.ERROR) + return + + # Check timeout + if Time.get_ticks_usec() - _handshake_start_usec > HANDSHAKE_TIMEOUT_USEC: + var reason := "Handshake timeout: no message received within 5 seconds" + push_error("SimBridge: %s" % reason) + handshake_failed.emit(reason) + _bridge.disconnect_from_server() + _set_state(ConnectionState.ERROR) + return + + # Try to read first message + var msg := _bridge.poll_message() + if msg.is_empty(): + return # Not ready yet, continue polling + + # Decode HandshakeMessage: { "protocol_version": N } + var decoded: Variant = Messagepack.decode(msg) + if decoded.status != null or not (decoded.value is Dictionary) \ + or not decoded.value.has("protocol_version"): + var reason := "Handshake decode failed: malformed HandshakeMessage" + push_error("SimBridge: %s" % reason) + handshake_failed.emit(reason) + _bridge.disconnect_from_server() + _set_state(ConnectionState.ERROR) + return + + var server_version: int = decoded.value["protocol_version"] + if server_version != Protocol.PROTOCOL_VERSION: + var reason := "Protocol version mismatch: server=%d, client=%d" % [ + server_version, Protocol.PROTOCOL_VERSION] + push_error("SimBridge: %s" % reason) + handshake_failed.emit(reason) + _bridge.disconnect_from_server() + _set_state(ConnectionState.ERROR) + return + + handshake_complete.emit(server_version) + _set_state(ConnectionState.CONNECTED) + return + if _bridge == null: return diff --git a/server/tests/ipc_bench.rs b/server/tests/ipc_bench.rs index dc8776f6e..4f9ffee6d 100644 --- a/server/tests/ipc_bench.rs +++ b/server/tests/ipc_bench.rs @@ -10,9 +10,6 @@ //! //! Output: IPC_BENCH_RESULT:{json} on a single line for tooling to parse. //! -//! BLOCKED (#342): Handshake step is stubbed pending #555 (server) + #556 (client). -//! The test currently skips the HandshakeMessage exchange and starts timing -//! immediately after TCP connection is established. //! //! Spec references: D-020 (subprocess IPC, 5ms budget), D-030 (Layer 3) @@ -97,13 +94,20 @@ fn ipc_round_trip_latency() { let mut reader = BufReader::new(stream.try_clone().expect("clone stream")); let mut writer = BufWriter::new(stream); - // TODO (#342, #555/#556): Wait for HandshakeMessage here before starting timing. - // When the server sends HandshakeMessage { protocol_version: 14 } as the first - // framed message, read and validate it. If version != PROTOCOL_VERSION, abort. - // The timing loop below starts after a successful handshake. - // - // For now, connect and proceed directly — the timing loop handles whatever - // the server sends as its first message. + // 4. Handshake: read and validate HandshakeMessage before timing (#555/#556). + // Server sends HandshakeMessage { protocol_version } as the very first framed message. + let handshake_bytes = read_framed(&mut reader) + .expect("read handshake") + .expect("server closed before sending HandshakeMessage"); + let handshake: HandshakeMessage = + rmp_serde::from_slice(&handshake_bytes).expect("deserialize HandshakeMessage"); + assert_eq!( + handshake.protocol_version, + PROTOCOL_VERSION, + "handshake version mismatch: server={}, client={}", + handshake.protocol_version, + PROTOCOL_VERSION + ); let make_input = |tick: u64| PlayerInput { tick, @@ -112,7 +116,7 @@ fn ipc_round_trip_latency() { let mut round_trip_ms: Vec = Vec::with_capacity(WARMUP_ROUNDS + MEASURE_ROUNDS); - // 4. Warmup rounds (not timed) + // 5. Warmup rounds (not timed) for tick in 0..WARMUP_ROUNDS as u64 { let payload = rmp_serde::to_vec_named(&vec![make_input(tick)]).expect("serialize PlayerInput"); @@ -122,7 +126,7 @@ fn ipc_round_trip_latency() { .expect("server closed during warmup"); } - // 5. Timed measurement rounds + // 6. Timed measurement rounds for tick in WARMUP_ROUNDS as u64..(WARMUP_ROUNDS + MEASURE_ROUNDS) as u64 { let payload = rmp_serde::to_vec_named(&vec![make_input(tick)]).expect("serialize PlayerInput"); @@ -141,7 +145,7 @@ fn ipc_round_trip_latency() { round_trip_ms.push(elapsed_ms); } - // 6. Clean up + // 7. Clean up drop(reader); drop(writer); let exit_deadline = Instant::now() + Duration::from_secs(5); @@ -163,7 +167,7 @@ fn ipc_round_trip_latency() { } } - // 7. Compute percentiles + // 8. Compute percentiles let mut sorted = round_trip_ms.clone(); sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());