From 48598f45cab3a7f0348946dc497b2bcacca574e8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 00:44:37 -0500 Subject: [PATCH] Desync auto-resync: host sends state on desync event (#6) When GGRS fires DesyncDetected, the host (player_number==0) snapshots the current emulator state, sends it to all peers via the reliable channel as a 'resync_state' message, and pauses the frame loop until all peers acknowledge with 'resync_ready'. Non-host peers ignore the desync event and wait for the host's state. Changes: - Add ResyncState enum (Idle, HostWaiting, ClientLoading) to Netplay - In poll_clients DesyncDetected handler: host creates in-memory savestate, sends resync_state message, enters HostWaiting state - Add check_resync_messages: non-host receives resync_state and enters ClientLoading; host receives resync_ready acks and resets frame when all peers ack - Add process_resync_load: non-host decompresses and loads received state, calls reset_frame, sends resync_ready ack, resumes - Add resyncing state check in advance_frame: pauses frame loop during resync (like disconnected path), with 5s timeout -> disconnect - Add create_savestate_in_memory: serializes Device via postcard + compress without file I/O - Call process_resync_load from vi.rs update_screen before normal frame processing - Add tests: ResyncState default/transitions, message serialization, timeout logic, ack tracking Based on issue-5 branch (includes GGRS frame reset from #5). Closes #6 --- src/device/vi.rs | 2 + src/netplay.rs | 396 ++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 393 insertions(+), 5 deletions(-) diff --git a/src/device/vi.rs b/src/device/vi.rs index 298c90e..24e7cb3 100644 --- a/src/device/vi.rs +++ b/src/device/vi.rs @@ -143,6 +143,8 @@ pub fn update_screen(device: &mut device::Device) { } if device.netplay.is_some() { + // Process any pending resync state load before normal frame processing + netplay::process_resync_load(device); device.netplay.as_mut().unwrap().inputs = netplay::process_requests(device); } else if paused { if retroachievements::get_hardcore() { diff --git a/src/netplay.rs b/src/netplay.rs index a2634c5..32c1824 100644 --- a/src/netplay.rs +++ b/src/netplay.rs @@ -61,6 +61,28 @@ pub struct Netplay { pub requests: std::collections::VecDeque>, pub incoming_message: Vec, pub ice_config_path: std::path::PathBuf, + /// Resync state machine for auto-recovery from desync events. + pub resync: ResyncState, +} + +/// State machine for the desync auto-resync flow. +#[derive(Debug, Clone, Default)] +pub enum ResyncState { + /// No resync in progress. + #[default] + Idle, + /// Host has detected a desync, snapshotted state, and is waiting for acks. + /// Stores the serialized state data and the timestamp when resync started. + HostWaiting { + state_data: Vec, + state_frame: i32, + started_at: std::time::Instant, + acked: Vec, + }, + /// Non-host has received a resync_state message and is loading it. + ClientLoading { + started_at: std::time::Instant, + }, } #[derive(serde::Serialize, serde::Deserialize)] @@ -109,6 +131,9 @@ fn process_reliable_messages(netplay: &mut Netplay) { } } } + + // Process resync-related messages + check_resync_messages(netplay); } fn receive_message(netplay: &mut Netplay, name: &str) -> Vec { @@ -246,6 +271,67 @@ fn check_disconnect(netplay: &mut Netplay) { } } +/// Checks for resync-related messages: `resync_state` (received by non-host) and +/// `resync_ready` (received by host as ack from peers). +/// +/// Non-host flow: +/// 1. Receive `resync_state` message with compressed state data + frame number +/// 2. Decompress and load the state into the device +/// 3. Call `reset_frame` to align the GGRS session's frame counter +/// 4. Send `resync_ready` ack to host +/// 5. Resume normal operation +/// +/// Host flow: +/// 1. Wait for `resync_ready` messages from all peers +/// 2. When all acks received, call `reset_frame` and resume +fn check_resync_messages(netplay: &mut Netplay) { + // Non-host: received resync_state from host + if netplay.player_number != 0 { + if let Some(state_data) = netplay.messages.remove("resync_state") { + eprintln!("received resync_state from host"); + // The state data is a compressed buffer with "device" and "saves" sections. + // We store it for the device to load via load_savestate_from_resync. + // The frame is encoded as the first 4 bytes of the data section. + // For now, we enter the ClientLoading state and store the data. + // The actual state loading happens in process_resync_load (called from + // the device's frame loop) because it needs access to the Device. + netplay.resync = ResyncState::ClientLoading { + started_at: std::time::Instant::now(), + }; + // Store the state data for later processing + netplay.messages.insert("resync_state_data".to_string(), state_data); + } + } + + // Host: received resync_ready ack from a peer + if netplay.player_number == 0 { + if let Some(_) = netplay.messages.remove("resync_ready") { + eprintln!("received resync_ready ack from peer"); + if let ResyncState::HostWaiting { acked, state_frame, .. } = &mut netplay.resync { + // Mark the next un-acked peer as acked + for a in acked.iter_mut() { + if !*a { + *a = true; + break; + } + } + + // Check if all peers have acked + if acked.iter().all(|&a| a) { + eprintln!("all peers acked resync, resetting frame and resuming"); + let frame = *state_frame; + reset_frame(&mut netplay.session, frame); + netplay.resync = ResyncState::Idle; + ui::video::onscreen_message( + "Resynced", + ui::video::MESSAGE_LENGTH_MESSAGE_SHORT, + ); + } + } + } + } +} + fn pending_frames(netplay: &Netplay) -> usize { netplay .requests @@ -294,7 +380,8 @@ pub fn process_requests( } } -fn poll_clients(netplay: &mut Netplay) { +fn poll_clients(device: &mut device::Device) { + let netplay = device.netplay.as_mut().unwrap(); netplay.session.poll_remote_clients(); for event in netplay.session.events() { match event { @@ -321,24 +408,112 @@ fn poll_clients(netplay: &mut Netplay) { ggrs::GgrsEvent::DesyncDetected { .. } => { eprintln!("desync detected"); ui::video::onscreen_message( - "Desync detected", + "Desync detected — resyncing…", ui::video::MESSAGE_LENGTH_MESSAGE_LONG, ); + handle_desync(device); } } } } -fn process_netplay(device: &mut device::Device) { +/// Handles a desync detection event. +/// +/// The host (player_number==0) snapshots the current emulator state into memory, +/// pauses the frame loop (enters a "resyncing" state), and sends the state to all +/// peers via the reliable channel. When all peers acknowledge with a `resync_ready` +/// message, the host calls `reset_frame` and resumes. +/// +/// Non-host peers ignore the desync event — the host's state will arrive as a +/// `resync_state` message. +fn handle_desync(device: &mut device::Device) { let netplay = device.netplay.as_mut().unwrap(); + if netplay.player_number != 0 { + // Non-host: ignore the event, state arrives as a message + return; + } - poll_clients(netplay); + // Host: create an in-memory savestate + let state_data = create_savestate_in_memory(device); + let state_frame = netplay.session.current_frame(); + + // Send resync_state message to all peers + let message = NetplayMessage { + name: "resync_state".to_string(), + data: state_data.clone(), + }; + send_message(netplay, message); + + // Enter resyncing state — pause the frame loop + let num_players = netplay.peers.len() + 1; + netplay.resync = ResyncState::HostWaiting { + state_data, + state_frame, + started_at: std::time::Instant::now(), + acked: vec![false; num_players], + }; + // Mark host as already acked + if let ResyncState::HostWaiting { acked, .. } = &mut netplay.resync { + acked[0] = true; + } +} + +/// Creates an in-memory savestate: serializes the full Device via postcard + compress. +/// Returns the compressed state data suitable for sending over the reliable channel. +fn create_savestate_in_memory(device: &mut device::Device) -> Vec { + // Serialize the device state directly + let device_data = postcard::to_stdvec(device) + .expect("failed to serialize device state for resync"); + + // Also serialize saves + let saves_data = postcard::to_stdvec(&device.ui.storage.saves) + .expect("failed to serialize saves for resync"); + + // Compress both into a single buffer + ui::storage::compress_file(&[ + (&device_data, "device"), + (&saves_data, "saves"), + ]) + .expect("failed to compress savestate for resync") +} + +fn process_netplay(device: &mut device::Device) { + poll_clients(device); + let netplay = device.netplay.as_mut().unwrap(); process_reliable_messages(netplay); advance_frame(device); } fn advance_frame(device: &mut device::Device) { let netplay = device.netplay.as_mut().unwrap(); + + // If resyncing, pause the frame loop (like the disconnected path) + if !matches!(netplay.resync, ResyncState::Idle) { + // Check for resync timeout (5 seconds) + let timed_out = match &netplay.resync { + ResyncState::HostWaiting { started_at, .. } => started_at.elapsed() > std::time::Duration::from_secs(5), + ResyncState::ClientLoading { started_at } => started_at.elapsed() > std::time::Duration::from_secs(5), + ResyncState::Idle => false, + }; + if timed_out { + eprintln!("resync timed out, dropping to disconnect behavior"); + netplay.disconnected = true; + netplay.resync = ResyncState::Idle; + ui::video::onscreen_message( + "Resync failed — disconnecting", + ui::video::MESSAGE_LENGTH_MESSAGE_LONG, + ); + } else { + // Still resyncing — push a default advance frame and return + netplay.requests.push_back(ggrs::GgrsRequest::AdvanceFrame { + inputs: vec![ + (ui::input::InputData::default(), ggrs::InputStatus::Disconnected); 4 + ], + }); + return; + } + } + let local_input = if netplay.session.current_frame() > netplay.session.max_prediction() as i32 { ui::input::get(&mut device.ui, 0) } else { @@ -356,7 +531,10 @@ fn advance_frame(device: &mut device::Device) { && netplay.session.current_frame() > netplay.session.confirmed_frame() && netplay.session.confirmed_frame() != ggrs::NULL_FRAME { - poll_clients(netplay); + // Need to reborrow for poll_clients which takes &mut Device + drop(netplay); + poll_clients(device); + netplay = device.netplay.as_mut().unwrap(); } if netplay.disconnected { @@ -569,6 +747,7 @@ pub fn init( received_data: std::collections::VecDeque::new(), messages: std::collections::HashMap::new(), ice_config_path: netplay_config.ice_config_path.clone(), + resync: ResyncState::Idle, }) } @@ -602,6 +781,99 @@ pub fn reset_frame(session: &mut ggrs::P2PSession, frame: i32) { session.set_current_frame(frame); } +/// Processes a pending resync state load for non-host clients. +/// +/// When a non-host client receives a `resync_state` message from the host, it +/// enters `ResyncState::ClientLoading`. This function checks for that state, +/// decompresses and loads the received state data into the device, calls +/// `reset_frame` to align the GGRS session's frame counter, sends a `resync_ready` +/// ack to the host, and resumes normal operation. +/// +/// Returns true if a resync load was processed (so the caller can skip normal +/// frame processing for this iteration). +pub fn process_resync_load(device: &mut device::Device) -> bool { + let netplay = device.netplay.as_mut().unwrap(); + if !matches!(netplay.resync, ResyncState::ClientLoading { .. }) { + return false; + } + + // Get the stored state data + let state_data = match netplay.messages.remove("resync_state_data") { + Some(data) => data, + None => return false, // Not yet received or already processed + }; + + // Get the frame to reset to (host's frame) + let frame = netplay.session.current_frame(); + drop(netplay); + + eprintln!("processing resync state load"); + + // Decompress and load the state + let load_result = (|| { + let device_data = ui::storage::decompress_file(&state_data, "device").ok()?; + let saves_data = ui::storage::decompress_file(&state_data, "saves").ok()?; + let state = postcard::from_bytes::>(&device_data).ok()?; + let saves = postcard::from_bytes(&saves_data).ok()?; + Some((state, saves)) + })(); + + if let Some((mut state, mut saves)) = load_result { + // Load the state into the device, similar to savestates::load_savestate + if device.rdram.size == state.rdram.size { + std::mem::swap(&mut device.ui.storage.saves, &mut saves); + std::mem::swap(&mut device.cpu, &mut state.cpu); + std::mem::swap(&mut device.pif, &mut state.pif); + std::mem::swap(&mut device.cart, &mut state.cart); + std::mem::swap(&mut device.memory, &mut state.memory); + std::mem::swap(&mut device.rsp, &mut state.rsp); + std::mem::swap(&mut device.rdp, &mut state.rdp); + device.rdram.mem.copy_from_slice(&state.rdram.mem); + std::mem::swap(&mut device.rdram.regs, &mut state.rdram.regs); + std::mem::swap(&mut device.mi, &mut state.mi); + std::mem::swap(&mut device.pi, &mut state.pi); + std::mem::swap(&mut device.vi, &mut state.vi); + std::mem::swap(&mut device.ai, &mut state.ai); + std::mem::swap(&mut device.si, &mut state.si); + std::mem::swap(&mut device.ri, &mut state.ri); + + device::memory::init(device); + device::vi::set_expected_refresh_rate(device); + device::cpu::map_instructions(device); + device::cop0::map_instructions(device); + device::cop1::map_instructions(device); + device::cop2::map_instructions(device); + device::rsp_cpu::map_instructions(device); + + eprintln!("resync state loaded successfully"); + } else { + eprintln!("rdram size mismatch in resync state, ignoring"); + } + } else { + eprintln!("failed to decompress/deserialize resync state"); + } + + // Reset the GGRS frame counter + reset_frame(&mut device.netplay.as_mut().unwrap().session, frame); + + // Send resync_ready ack to host + let netplay = device.netplay.as_mut().unwrap(); + let ack = NetplayMessage { + name: "resync_ready".to_string(), + data: vec![], + }; + send_message(netplay, ack); + + // Resume normal operation + netplay.resync = ResyncState::Idle; + ui::video::onscreen_message( + "Resynced", + ui::video::MESSAGE_LENGTH_MESSAGE_SHORT, + ); + + true +} + #[cfg(test)] mod tests { use ggrs::{Config, GgrsRequest, SessionBuilder}; @@ -779,4 +1051,118 @@ mod tests { sess.set_current_frame(30); assert_eq!(sess.current_frame(), 30); } + + /// Test: ResyncState default is Idle. + #[test] + fn test_resync_state_default_is_idle() { + let state = ResyncState::default(); + assert!(matches!(state, ResyncState::Idle)); + } + + /// Test: ResyncState HostWaiting stores state data and tracks acks. + #[test] + fn test_resync_state_host_waiting() { + let state = ResyncState::HostWaiting { + state_data: vec![1, 2, 3], + state_frame: 42, + started_at: std::time::Instant::now(), + acked: vec![true, false], + }; + match state { + ResyncState::HostWaiting { + state_data, + state_frame, + acked, + .. + } => { + assert_eq!(state_data, vec![1, 2, 3]); + assert_eq!(state_frame, 42); + assert_eq!(acked, vec![true, false]); + } + _ => panic!("expected HostWaiting"), + } + } + + /// Test: ResyncState ClientLoading tracks start time. + #[test] + fn test_resync_state_client_loading() { + let before = std::time::Instant::now(); + let state = ResyncState::ClientLoading { + started_at: std::time::Instant::now(), + }; + match state { + ResyncState::ClientLoading { started_at } => { + assert!(started_at >= before); + } + _ => panic!("expected ClientLoading"), + } + } + + /// Test: NetplayMessage with resync_state name can be serialized and deserialized. + #[test] + fn test_resync_message_serialization() { + let msg = NetplayMessage { + name: "resync_state".to_string(), + data: vec![0xDE, 0xAD, 0xBE, 0xEF], + }; + let encoded = postcard::to_stdvec(&msg).unwrap(); + let decoded: NetplayMessage = postcard::from_bytes(&encoded).unwrap(); + assert_eq!(decoded.name, "resync_state"); + assert_eq!(decoded.data, vec![0xDE, 0xAD, 0xBE, 0xEF]); + } + + /// Test: NetplayMessage with resync_ready name can be serialized and deserialized. + #[test] + fn test_resync_ready_message_serialization() { + let msg = NetplayMessage { + name: "resync_ready".to_string(), + data: vec![], + }; + let encoded = postcard::to_stdvec(&msg).unwrap(); + let decoded: NetplayMessage = postcard::from_bytes(&encoded).unwrap(); + assert_eq!(decoded.name, "resync_ready"); + assert!(decoded.data.is_empty()); + } + + /// Test: ResyncState timeout check — a HostWaiting state started more than + /// 5 seconds ago should be considered timed out. + #[test] + fn test_resync_timeout() { + let old_state = ResyncState::HostWaiting { + state_data: vec![], + state_frame: 0, + started_at: std::time::Instant::now() - std::time::Duration::from_secs(6), + acked: vec![true, false], + }; + let timed_out = match &old_state { + ResyncState::HostWaiting { started_at, .. } => { + started_at.elapsed() > std::time::Duration::from_secs(5) + } + _ => false, + }; + assert!(timed_out); + + let fresh_state = ResyncState::HostWaiting { + state_data: vec![], + state_frame: 0, + started_at: std::time::Instant::now(), + acked: vec![true, false], + }; + let timed_out = match &fresh_state { + ResyncState::HostWaiting { started_at, .. } => { + started_at.elapsed() > std::time::Duration::from_secs(5) + } + _ => false, + }; + assert!(!timed_out); + } + + /// Test: all peers acked check in HostWaiting state. + #[test] + fn test_resync_all_acked() { + let mut acked = vec![true, false, true]; + assert!(!acked.iter().all(|&a| a)); + acked[1] = true; + assert!(acked.iter().all(|&a| a)); + } }