From 891d12c08a48b06abfde46744aeec77ba4e6df1a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 02:34:07 -0500 Subject: [PATCH] netplay: add state-message types to reliable channel protocol (#1) Add message-name constants and reliable-channel helpers for netplay session state transfer: - Message-name constants: netplay_state, netplay_state_ready, resync_state, resync_ready - Chunked message codec (encode_message_chunks / decode_message / decode_packets) reusing the existing 16 KiB reliable-channel transport; send_message/process_reliable_messages refactored onto it - send_netplay_state / receive_netplay_state for full session state - send_ready / wait_for_ready ack helpers (10s timeout, same pattern as receive_message); receive_message extracted to wait_for_message with preserve of input_delay/disconnect side effects - Large payloads (>1MB) accumulate into incoming_message until a full postcard message parses - Unit tests: 5MB payload round-trip through the chunked codec, ack send/receive within timeout, missing ack times out with the standard panic pattern Closes #1 --- src/netplay.rs | 292 +++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 258 insertions(+), 34 deletions(-) diff --git a/src/netplay.rs b/src/netplay.rs index dc56b20..4885286 100644 --- a/src/netplay.rs +++ b/src/netplay.rs @@ -69,15 +69,64 @@ struct NetplayMessage { data: Vec, } +/// Reliable-channel message name for a full session state transfer. +pub const NETPLAY_STATE_MESSAGE: &str = "netplay_state"; +/// Reliable-channel ack name confirming a state transfer was applied. +#[allow(dead_code)] // consumed by the upcoming session save/resume lanes +pub const NETPLAY_STATE_READY_MESSAGE: &str = "netplay_state_ready"; +/// Reliable-channel message name for a desync resync state transfer. +#[allow(dead_code)] // consumed by the upcoming desync resync lanes +pub const RESYNC_STATE_MESSAGE: &str = "resync_state"; +/// Reliable-channel ack name confirming a resync state was applied. +#[allow(dead_code)] // consumed by the upcoming desync resync lanes +pub const RESYNC_READY_MESSAGE: &str = "resync_ready"; + +/// Timeout for waiting on reliable-channel messages. +const MESSAGE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +/// Serialize a message and split it into 16 KiB transport chunks. +fn encode_message_chunks(name: &str, data: &[u8]) -> Vec> { + let message = NetplayMessage { + name: name.to_string(), + data: data.to_vec(), + }; + let encoded = postcard::to_stdvec(&message).unwrap(); + encoded.chunks(16384).map(|chunk| chunk.to_vec()).collect() +} + +/// Decode a complete message from accumulated incoming bytes, returning +/// `None` while the message is still arriving in chunks. +fn decode_message(buffer: &[u8]) -> Option<(String, Vec)> { + postcard::from_bytes::(buffer) + .ok() + .map(|message| (message.name, message.data)) +} + +/// Accumulate received packets into `incoming_message` until complete messages +/// parse, returning every message decoded from the batch. +fn decode_packets( + incoming_message: &mut Vec, + received_data: &mut std::collections::VecDeque>, + packets: Vec>, +) -> Vec<(String, Vec)> { + let mut decoded = Vec::new(); + received_data.extend(packets); + while !received_data.is_empty() { + if let Some(data) = received_data.pop_front() { + incoming_message.extend(data); + if let Some(message) = decode_message(incoming_message) { + decoded.push(message); + incoming_message.clear(); + } + } + } + decoded +} + fn send_message(netplay: &mut Netplay, message: NetplayMessage) { - let data = postcard::to_stdvec(&message).unwrap(); - let chunks = data.chunks(16384).collect::>(); - for peer in netplay.peers.iter() { - for chunk in chunks.iter() { - if let Err(e) = netplay - .reliable_channel - .try_send(chunk.to_vec().into(), *peer) - { + for chunk in encode_message_chunks(&message.name, &message.data) { + for peer in netplay.peers.iter() { + if let Err(e) = netplay.reliable_channel.try_send(chunk.clone().into(), *peer) { eprintln!("Failed to send message: {}", e); } } @@ -85,39 +134,39 @@ fn send_message(netplay: &mut Netplay, message: NetplayMessage) { } fn process_reliable_messages(netplay: &mut Netplay) { - netplay.received_data.extend( - netplay - .reliable_channel - .receive() - .iter() - .map(|(_, data)| data.to_vec()), - ); + let packets = netplay + .reliable_channel + .receive() + .iter() + .map(|(_, data)| data.to_vec()) + .collect::>>(); - while !netplay.received_data.is_empty() { - if let Some(data) = netplay.received_data.pop_front() { - netplay.incoming_message.extend(data); - - if let Ok(decoded_message) = - postcard::from_bytes::(&netplay.incoming_message) - { - netplay - .messages - .insert(decoded_message.name, decoded_message.data); - netplay.incoming_message.clear(); - check_input_delay(netplay); - check_disconnect(netplay); - } - } + for (name, data) in decode_packets( + &mut netplay.incoming_message, + &mut netplay.received_data, + packets, + ) { + netplay.messages.insert(name, data); + check_input_delay(netplay); + check_disconnect(netplay); } } -fn receive_message(netplay: &mut Netplay, name: &str) -> Vec { - let timeout = std::time::Duration::from_secs(10); +/// Poll `messages` for `name` until it arrives or `timeout` elapses, calling +/// `poll` on every iteration to route newly received data into `messages`. +/// Panics with the standard reliable-channel pattern when the message never +/// arrives. +fn wait_for_message( + name: &str, + messages: &mut std::collections::HashMap>, + timeout: std::time::Duration, + mut poll: impl FnMut(&mut std::collections::HashMap>), +) -> Vec { let now = std::time::Instant::now(); loop { - process_reliable_messages(netplay); - if let Some(data) = netplay.messages.remove(name) { + poll(messages); + if let Some(data) = messages.remove(name) { return data; } @@ -128,6 +177,60 @@ fn receive_message(netplay: &mut Netplay, name: &str) -> Vec { } } +fn receive_message(netplay: &mut Netplay, name: &str) -> Vec { + let Netplay { + messages, + reliable_channel, + incoming_message, + received_data, + input_delay, + session, + disconnected, + .. + } = netplay; + + wait_for_message(name, messages, MESSAGE_TIMEOUT, |messages| { + let packets = reliable_channel + .receive() + .iter() + .map(|(_, data)| data.to_vec()) + .collect::>>(); + + for (message_name, data) in decode_packets(incoming_message, received_data, packets) { + // Preserve the side effects the original receive path applied to + // every decoded message (input-delay changes and disconnects). + // These control messages are consumed, never stored in `messages` + // (mirroring check_input_delay/check_disconnect removing them). + if message_name == "input_delay" { + let new_delay = u64::from_be_bytes(data[..].try_into().unwrap()) as usize; + if new_delay != *input_delay { + *input_delay = new_delay; + for handle in session.local_player_handles() { + if let Err(e) = session.set_input_delay(handle, new_delay) { + eprintln!("Error setting input delay: {}", e); + } else { + ui::video::onscreen_message( + &format!("Input delay: {}", new_delay), + ui::video::MESSAGE_LENGTH_MESSAGE_VERY_SHORT, + ); + } + } + } + } else if message_name == "disconnect" { + if !*disconnected { + *disconnected = true; + ui::video::onscreen_message( + "Player disconnected, session has ended", + ui::video::MESSAGE_LENGTH_MESSAGE_LONG, + ); + } + } else { + messages.insert(message_name, data); + } + } + }) +} + fn send_player_number( channel: &mut matchbox_socket::WebRtcChannel, peers: &[matchbox_socket::PeerId], @@ -200,6 +303,38 @@ pub fn receive_save(netplay: &mut Netplay, save_type: &str, save_data: &mut Vec< *save_data = message; } +/// Send a full session state over the reliable channel as chunked messages. +#[allow(dead_code)] // consumed by the upcoming session save/resume lanes +pub fn send_netplay_state(netplay: &mut Netplay, data: &[u8]) { + let message = NetplayMessage { + name: NETPLAY_STATE_MESSAGE.to_string(), + data: data.to_vec(), + }; + send_message(netplay, message); +} + +/// Receive a full session state from the reliable channel. +#[allow(dead_code)] // consumed by the upcoming session save/resume lanes +pub fn receive_netplay_state(netplay: &mut Netplay) -> Vec { + receive_message(netplay, NETPLAY_STATE_MESSAGE) +} + +/// Send a ready-ack message with `name` over the reliable channel. +#[allow(dead_code)] // consumed by the upcoming session save/resume lanes +pub fn send_ready(netplay: &mut Netplay, name: &str) { + let message = NetplayMessage { + name: name.to_string(), + data: vec![], + }; + send_message(netplay, message); +} + +/// Wait for the ready-ack message with `name` (standard 10s timeout). +#[allow(dead_code)] // consumed by the upcoming session save/resume lanes +pub fn wait_for_ready(netplay: &mut Netplay, name: &str) { + receive_message(netplay, name); +} + pub fn send_input_delay(netplay: &mut Netplay, input_delay: usize) { if input_delay < 1 { return; @@ -584,3 +719,92 @@ pub fn close(netplay: &mut Netplay) { let _ = std::fs::remove_file(&netplay.ice_config_path); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn state_message_constants_match_protocol() { + assert_eq!(NETPLAY_STATE_MESSAGE, "netplay_state"); + assert_eq!(NETPLAY_STATE_READY_MESSAGE, "netplay_state_ready"); + assert_eq!(RESYNC_STATE_MESSAGE, "resync_state"); + assert_eq!(RESYNC_READY_MESSAGE, "resync_ready"); + } + + #[test] + fn large_payload_round_trips_through_chunked_codec() { + let payload: Vec = (0..5 * 1024 * 1024).map(|i| (i % 251) as u8).collect(); + + let chunks = encode_message_chunks(NETPLAY_STATE_MESSAGE, &payload); + assert!(chunks.len() > 1, "5MB payload must span multiple chunks"); + + // Mirror what process_reliable_messages does in production: accumulate + // chunks until the full postcard message parses. + let mut incoming_message = Vec::new(); + let mut received_data = std::collections::VecDeque::new(); + let decoded = decode_packets(&mut incoming_message, &mut received_data, chunks); + + assert_eq!( + decoded.len(), + 1, + "exactly one complete message should decode after all chunks" + ); + let (name, data) = &decoded[0]; + assert_eq!(name, NETPLAY_STATE_MESSAGE); + assert_eq!(data, &payload); + } + + #[test] + fn ready_ack_send_receive_within_timeout() { + let timeout = std::time::Duration::from_secs(10); + let start = std::time::Instant::now(); + + // Send side: chunked encoding of the ack message, exactly like send_ready. + let chunks = encode_message_chunks(NETPLAY_STATE_READY_MESSAGE, &[]); + + // Receive side: accumulate + decode, exactly like the production path. + let mut incoming_message = Vec::new(); + let mut received_data = std::collections::VecDeque::new(); + let decoded = decode_packets(&mut incoming_message, &mut received_data, chunks); + + assert!( + start.elapsed() < timeout, + "ack must arrive within the timeout window" + ); + assert_eq!(decoded.len(), 1); + let (name, data) = &decoded[0]; + assert_eq!(name, NETPLAY_STATE_READY_MESSAGE); + assert!(data.is_empty()); + + // The wait API itself must return promptly when the ack is already queued. + let mut messages = std::collections::HashMap::from([( + NETPLAY_STATE_READY_MESSAGE.to_string(), + Vec::::new(), + )]); + let ack = wait_for_message(NETPLAY_STATE_READY_MESSAGE, &mut messages, timeout, |_| {}); + assert!(ack.is_empty()); + assert!(start.elapsed() < timeout); + } + + #[test] + fn missing_ack_times_out_with_standard_panic() { + let mut messages = std::collections::HashMap::new(); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + wait_for_message( + NETPLAY_STATE_READY_MESSAGE, + &mut messages, + std::time::Duration::from_millis(50), + |_| {}, + ); + })); + + let payload = result.expect_err("waiting for a missing ack must panic"); + let message = payload + .downcast_ref::() + .map(String::as_str) + .or_else(|| payload.downcast_ref::<&str>().copied()) + .unwrap_or(""); + assert_eq!(message, "Could not receive message for netplay_state_ready"); + } +}