netplay: state-message types for reliable channel protocol (#1)

src/netplay.rs

@@ -69,15 +69,64 @@ struct NetplayMessage {
     data: Vec<u8>,
 }
 
+/// 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<Vec<u8>> {
+    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<u8>)> {
+    postcard::from_bytes::<NetplayMessage>(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<u8>,
+    received_data: &mut std::collections::VecDeque<Vec<u8>>,
+    packets: Vec<Vec<u8>>,
+) -> Vec<(String, Vec<u8>)> {
+    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::<Vec<&[u8]>>();
-    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::<Vec<Vec<u8>>>();
 
-    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::<NetplayMessage>(&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<u8> {
-    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<String, Vec<u8>>,
+    timeout: std::time::Duration,
+    mut poll: impl FnMut(&mut std::collections::HashMap<String, Vec<u8>>),
+) -> Vec<u8> {
     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<u8> {
     }
 }
 
+fn receive_message(netplay: &mut Netplay, name: &str) -> Vec<u8> {
+    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::<Vec<Vec<u8>>>();
+
+        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<u8> {
+    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<u8> = (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::<u8>::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::<String>()
+            .map(String::as_str)
+            .or_else(|| payload.downcast_ref::<&str>().copied())
+            .unwrap_or("");
+        assert_eq!(message, "Could not receive message for netplay_state_ready");
+    }
+}