netplay: session resume — reload saved state and broadcast to peers (#4)

This commit is contained in:
Claude 2026-08-25 11:22:52 -05:00
parent 6d6b07ae67
commit 9097b455db
9 changed files with 473 additions and 16 deletions

View File

@ -112,6 +112,20 @@ pub fn run_game(
ui::storage::load_saves(&mut device.ui, &mut device.netplay);
ui::storage::format_saves(device);
// issue #4: session resume — host broadcasts the saved session state and
// every peer loads it before the frame loop starts. A failed resume is
// treated like a peer disconnect: the session never plays desynced.
if let Err(e) = netplay::resume_session(device) {
eprintln!("netplay resume failed: {e}");
ui::video::onscreen_message(
&format!("Resume failed: {e}"),
ui::video::MESSAGE_LENGTH_MESSAGE_LONG,
);
if let Some(netplay) = &mut device.netplay {
netplay.disconnected = true;
}
}
if !retroachievements::get_hardcore() {
cheats::init(device, game_settings.cheats);
}

View File

@ -37,6 +37,10 @@ pub struct Args {
pub netplay_number_of_players: Option<usize>,
#[arg(long, value_name = "NETPLAY_INPUT_DELAY", hide = true)]
pub netplay_input_delay: Option<usize>,
#[arg(long, hide = true)]
pub netplay_resume: bool,
#[arg(long, value_name = "NETPLAY_STATE", hide = true)]
pub netplay_state: Option<String>,
#[arg(
long,
value_name = "PROFILE_NAME",
@ -251,6 +255,8 @@ pub fn run(args: Args, arg_count: usize) -> std::io::Result<()> {
number_of_players,
input_delay,
ice_config_path: dirs.cache_dir.join("ice_config.json"),
resume: args.netplay_resume,
state_path: args.netplay_state.map(std::path::PathBuf::from),
})
} else {
None

View File

@ -46,6 +46,11 @@ pub struct NetplayConfig {
pub number_of_players: usize,
pub input_delay: usize,
pub ice_config_path: std::path::PathBuf,
/// Session-resume (issue #4): exchange the saved session state with all
/// peers before the frame loop starts. Host-only payload: the
/// `.netplay-state` file to broadcast.
pub resume: bool,
pub state_path: Option<std::path::PathBuf>,
}
pub struct Netplay {
@ -62,6 +67,16 @@ pub struct Netplay {
pub requests: std::collections::VecDeque<ggrs::GgrsRequest<GgrsConfig>>,
pub incoming_message: Vec<u8>,
pub ice_config_path: std::path::PathBuf,
/// Session-resume (issue #4): whether the session-state exchange runs
/// before the frame loop.
pub resume: bool,
/// Session-resume (issue #4), host only: the `.netplay-state` file to
/// broadcast to peers.
pub state_path: Option<std::path::PathBuf>,
/// Session-resume (issue #4): the GGRS frame the resumed session was
/// saved at, recorded by both sides once the state has been sent
/// (host) or loaded (peer). Issue #5's frame reset will consume this.
pub resume_frame: Option<i32>,
}
#[derive(serde::Serialize, serde::Deserialize)]
@ -448,6 +463,174 @@ pub fn save_netplay_session(device: &mut device::Device, path: &std::path::Path)
}
}
// ---- Session resume (issue #4) ----
/// How long the host waits for every peer to acknowledge receiving and
/// loading the broadcast session state before giving up on that peer.
const NETPLAY_STATE_ACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
/// Wait for `expected` acknowledgements named `name`, polling the reliable
/// channel until all of them arrive or `timeout` elapses.
///
/// Acks are counted, not attributed to a specific peer: `send_ready`
/// broadcasts to every peer, so the host receives one copy per peer.
fn wait_for_ready_all(
netplay: &mut Netplay,
name: &str,
expected: usize,
timeout: std::time::Duration,
) -> bool {
let now = std::time::Instant::now();
let mut received = 0;
while received < expected {
process_reliable_messages(netplay);
if netplay.messages.remove(name).is_some() {
received += 1;
continue;
}
if now.elapsed() > timeout {
return false;
}
std::thread::sleep(std::time::Duration::from_millis(1));
}
true
}
/// Session-resume exchange (issue #4). Runs once, right after the initial
/// RNG/RTC/savestate handshake and before the GGRS frame loop starts:
///
/// - host (player 0): reads the `.netplay-state` file, validates its ROM
/// hash, broadcasts the raw session-save bytes to every peer via
/// `netplay_state` and waits for each peer's `netplay_state_ready` ack
/// (30 s timeout, then the session is disconnected);
/// - peers: receive `netplay_state`, decode it with the same
/// `decode_netplay_session` the host used to read the file, load the
/// embedded savestate via `SavestateData::load_from_memory` +
/// `savestates::load_savestate_from_data`, then acknowledge with
/// `netplay_state_ready`.
///
/// The GGRS session frame counter is intentionally NOT reset here: that is
/// issue #5's `netplay::reset_frame`. Both sides record the saved frame in
/// `netplay.resume_frame` so the frame-reset hook has the value ready; the
/// PR body documents the gap until #5 lands.
///
/// On any failure the session is flagged disconnected (mirroring the
/// existing peer-disconnect path) so the frame loop never starts desynced.
pub fn resume_session(device: &mut device::Device) -> Result<(), String> {
let Some(netplay) = device.netplay.as_ref() else {
return Ok(());
};
if !netplay.resume {
return Ok(());
}
if netplay.player_number == 0 {
resume_session_host(device)
} else {
resume_session_peer(device)
}
}
fn resume_session_host(device: &mut device::Device) -> Result<(), String> {
ui::video::onscreen_message(
"Syncing session state…",
ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
);
let (state_path, expected_peers, game_hash) = {
let netplay = device.netplay.as_ref().unwrap();
(
netplay
.state_path
.clone()
.ok_or_else(|| "Host resume requires a .netplay-state file".to_string())?,
netplay.peers.len(),
device.ui.game_hash.clone(),
)
};
let file_data = std::fs::read(&state_path)
.map_err(|e| format!("Could not read session state {}: {}", state_path.display(), e))?;
let save = decode_netplay_session(&file_data)
.ok_or_else(|| format!("Could not decode session state file {}", state_path.display()))?;
if !save.rom_hash.is_empty() && save.rom_hash != game_hash {
return Err(format!(
"Session state was saved for a different ROM ({} vs {})",
save.rom_hash, game_hash
));
}
// Broadcast the file bytes unmodified: `encode_netplay_session` produces
// exactly the on-disk format, so every peer decodes what the host read.
let broadcast = encode_netplay_session(&save)
.ok_or_else(|| "Could not encode session state for broadcast".to_string())?;
let netplay = device.netplay.as_mut().unwrap();
netplay.resume_frame = Some(save.frame);
send_netplay_state(netplay, &broadcast);
if !wait_for_ready_all(
netplay,
NETPLAY_STATE_READY_MESSAGE,
expected_peers,
NETPLAY_STATE_ACK_TIMEOUT,
) {
// One or more peers did not load the state in time. Acks are not
// peer-attributed, so the lagging peer cannot be singled out: drop
// the whole session rather than start it out of sync.
for handle in 0..expected_peers + 1 {
if handle != netplay.player_number {
let _ = netplay.session.disconnect_player(handle);
}
}
netplay.disconnected = true;
return Err(format!(
"Timed out waiting for {} peer(s) to sync the session state",
expected_peers
));
}
ui::video::onscreen_message(
"Session synced — play resumes",
ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
);
Ok(())
}
fn resume_session_peer(device: &mut device::Device) -> Result<(), String> {
ui::video::onscreen_message(
"Syncing session state…",
ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
);
let state_bytes = {
let netplay = device.netplay.as_mut().unwrap();
// Standard reliable-channel timeout (10 s); the host is the sender.
receive_netplay_state(netplay)
};
let save = decode_netplay_session(&state_bytes)
.ok_or_else(|| "Could not decode session state received from host".to_string())?;
let state_data = savestates::SavestateData::load_from_memory(&save.state)
.ok_or_else(|| "Session state received from host could not be deserialized".to_string())?;
if !savestates::load_savestate_from_data(device, &state_data) {
return Err("Session state received from host could not be applied".to_string());
}
let netplay = device.netplay.as_mut().unwrap();
netplay.resume_frame = Some(save.frame);
send_ready(netplay, NETPLAY_STATE_READY_MESSAGE);
ui::video::onscreen_message(
"Session synced — play resumes",
ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
);
Ok(())
}
pub fn send_input_delay(netplay: &mut Netplay, input_delay: usize) {
if input_delay < 1 {
return;
@ -817,6 +1000,9 @@ pub fn init(
received_data: std::collections::VecDeque::new(),
messages: std::collections::HashMap::new(),
ice_config_path: netplay_config.ice_config_path.clone(),
resume: netplay_config.resume,
state_path: netplay_config.state_path.clone(),
resume_frame: None,
})
}
@ -960,4 +1146,164 @@ mod tests {
let _ = std::fs::remove_file(&path);
}
/// No-op network socket: satisfies ggrs' `NonBlockingSocket` bound so a
/// local-only session can be constructed without any real network.
struct TestSocket;
impl ggrs::NonBlockingSocket<matchbox_socket::PeerId> for TestSocket {
fn send_to(&mut self, _msg: &ggrs::Message, _addr: &matchbox_socket::PeerId) {}
fn receive_all_messages(&mut self) -> Vec<(matchbox_socket::PeerId, ggrs::Message)> {
Vec::new()
}
}
/// Build a real `Netplay` value offline (mirrors the savestates test
/// pattern): local-only ggrs session in `Running` state, WebRTC channels
/// from a builder that never connects.
fn mock_netplay(player_number: usize) -> Netplay {
let (mut socket, _loop_future) = matchbox_socket::WebRtcSocketBuilder::new(
"ws://127.0.0.1:9/room",
)
.add_unreliable_channel()
.add_reliable_channel()
.build();
let reliable_channel = socket.take_channel(1).unwrap();
let session = ggrs::SessionBuilder::<GgrsConfig>::new()
.with_num_players(1)
.unwrap()
.add_player(ggrs::PlayerType::Local, player_number)
.unwrap()
.start_p2p_session(TestSocket)
.unwrap();
Netplay {
disconnected: false,
session,
reliable_channel,
peers: vec![],
player_number,
connected: [true, false, false, false],
input_delay: 0,
messages: Default::default(),
received_data: Default::default(),
inputs: vec![],
requests: Default::default(),
incoming_message: vec![],
ice_config_path: std::path::PathBuf::new(),
resume: true,
state_path: None,
resume_frame: None,
}
}
/// Mock 2-peer resume flow (issue #4): the host's `.netplay-state`
/// envelope travels over the `netplay_state` reliable message, the peer
/// decodes it and loads the embedded savestate, and both sides agree on
/// the frame metadata.
#[test]
fn resume_state_reaches_peer_with_same_frame_metadata() {
use crate::savestates::{create_savestate_in_memory, load_savestate_from_data};
// Host: a session with distinctive emulator state, captured in
// memory (issue #2 API) and wrapped in the issue #3 on-disk
// `.netplay-state` envelope.
let mut host_device = crate::device::Device::new(false);
host_device.cpu.gpr[1] = 0x1234_5678_9abc_def0;
host_device.cpu.gpr[3] = 0x0fed_cba9_8765_4321;
host_device.rdram.mem.resize(0x800, 0);
host_device.rdram.mem[0] = 0xAB;
host_device.rdram.mem[0x3FF] = 0xCD;
host_device.vi.regs[0] = 0x1122_3344;
let snapshot =
create_savestate_in_memory(&mut host_device).expect("create in-memory savestate");
let save = NetplaySessionSave {
state: snapshot.save_to_memory(),
frame: 12_345,
input_delay: 2,
rom_hash: "DEADBEEF-C01DF00D".to_string(),
players: 2,
};
// The host broadcasts exactly the on-disk file bytes.
let broadcast = encode_netplay_session(&save).expect("encode session save");
let chunks = encode_message_chunks(NETPLAY_STATE_MESSAGE, &broadcast);
// Peer accumulates the chunked message, exactly like
// process_reliable_messages does in production.
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, "one complete netplay_state message");
let (name, data) = &decoded[0];
assert_eq!(name, NETPLAY_STATE_MESSAGE);
// Peer decodes with the same envelope the host used.
let peer_save = decode_netplay_session(data).expect("peer decodes session state");
assert_eq!(peer_save.frame, 12_345, "both sides report the saved frame");
assert_eq!(peer_save.input_delay, 2);
assert_eq!(peer_save.players, 2);
assert_eq!(peer_save.rom_hash, "DEADBEEF-C01DF00D");
assert_eq!(peer_save.state, save.state, "state payload travels unmodified");
// Peer loads the embedded savestate into a fresh device (no disk).
let mut peer_device = crate::device::Device::new(false);
peer_device.netplay = Some(mock_netplay(1));
let state_data = crate::savestates::SavestateData::load_from_memory(&peer_save.state)
.expect("peer deserializes the savestate");
assert!(
load_savestate_from_data(&mut peer_device, &state_data),
"peer applies the state"
);
assert_eq!(peer_device.cpu.gpr[1], 0x1234_5678_9abc_def0);
assert_eq!(peer_device.cpu.gpr[3], 0x0fed_cba9_8765_4321);
assert_eq!(peer_device.rdram.mem[0], 0xAB);
assert_eq!(peer_device.rdram.mem[0x3FF], 0xCD);
assert_eq!(peer_device.vi.regs[0], 0x1122_3344);
assert!(
peer_device.netplay.is_some(),
"the live netplay session survives the state load"
);
}
/// The host's 30s ack wait succeeds once every peer acknowledged and
/// times out (returning false) when acks are missing.
#[test]
fn wait_for_ready_all_counts_acks_and_times_out() {
// All acks present: succeeds immediately.
let mut netplay = mock_netplay(0);
netplay
.messages
.insert(NETPLAY_STATE_READY_MESSAGE.to_string(), vec![]);
assert!(wait_for_ready_all(
&mut netplay,
NETPLAY_STATE_READY_MESSAGE,
1,
std::time::Duration::from_millis(200),
));
// Missing acks: gives up after the timeout.
let mut netplay = mock_netplay(0);
assert!(!wait_for_ready_all(
&mut netplay,
NETPLAY_STATE_READY_MESSAGE,
1,
std::time::Duration::from_millis(50),
));
}
/// resume_session is a no-op outside resume sessions, so regular netplay
/// is untouched (host and peer paths).
#[test]
fn resume_session_skips_non_resume_sessions() {
let mut device = crate::device::Device::new(false);
let mut netplay = mock_netplay(0);
netplay.resume = false;
device.netplay = Some(netplay);
assert!(resume_session(&mut device).is_ok());
assert_eq!(device.netplay.as_ref().unwrap().resume_frame, None);
}
}

View File

@ -625,6 +625,9 @@ mod tests {
requests: Default::default(),
incoming_message: vec![],
ice_config_path: std::path::PathBuf::new(),
resume: false,
state_path: None,
resume_frame: None,
}
}

View File

@ -24,6 +24,10 @@ pub struct NetplayDevice {
pub player_number: usize,
pub number_of_players: usize,
pub input_delay: usize,
/// Resume a saved netplay session (issue #4).
pub resume: bool,
/// Host-only path to the `.netplay-state` file to broadcast to peers.
pub state_path: Option<std::path::PathBuf>,
}
fn check_latest_version(weak: slint::Weak<AppWindow>) {
@ -614,6 +618,13 @@ pub fn run_rom(
"--cheats",
cheats_path.to_str().unwrap(),
]);
if netplay_device.resume {
command.arg("--netplay-resume");
}
if let Some(state_path) = &netplay_device.state_path {
command.arg("--netplay-state").arg(state_path);
}
}
let success = command
@ -749,6 +760,28 @@ pub async fn select_rom(rom_dir: slint::SharedString) -> Option<std::path::PathB
}
}
/// File picker for `.netplay-state` session files (issue #4), mirroring
/// `select_rom`. Android SAF content URIs cannot be resolved to a path the
/// CLI subprocess can read, so session resume is a desktop feature for now.
pub async fn select_netplay_state() -> Option<std::path::PathBuf> {
#[cfg(target_os = "android")]
{
// SAF content URIs cannot be resolved to a path the CLI subprocess
// can read; session resume is a desktop feature for now.
None
}
#[cfg(not(target_os = "android"))]
{
rfd::AsyncFileDialog::new()
.set_title("Select Netplay Session State")
.add_filter("Netplay state files", &["netplay-state"])
.pick_file()
.await
.map(|file| file.path().to_path_buf())
}
}
pub async fn select_gb_rom(player: i32) -> Option<std::path::PathBuf> {
#[cfg(target_os = "android")]
return ui::android::select_gb_rom(player).await;

View File

@ -52,8 +52,10 @@ export component AppWindow inherits Window {
callback netplay_refresh_sessions;
callback netplay_create_session(string, string, string, string, string, string, bool, bool);
callback netplay_select_rom(string);
callback netplay_select_state;
callback netplay_join_session(string, string, string, string);
callback netplay_close;
callback netplay_save_session;
callback cheats_select_rom_clicked(string);
callback cheats_clear_clicked;
callback cheat_toggled(string, string, string, bool);
@ -121,6 +123,7 @@ export component AppWindow inherits Window {
in-out property netplay_game_cheats <=> NetplayData.game_cheats;
in-out property netplay_pending_session <=> NetplayData.pending_session;
in-out property netplay_rom_path <=> NetplayData.rom_path;
in-out property netplay_state_path <=> NetplayData.state_path;
in-out property netplay_current_session <=> NetplayData.current_session;
in-out property netplay_sessions <=> NetplayData.sessions;
in-out property netplay_pending_refresh <=> NetplayData.pending_refresh;
@ -134,6 +137,8 @@ export component AppWindow inherits Window {
in-out property netplay_recommended_delay <=> NetplayWaitData.recommended_delay;
in-out property netplay_game_pal <=> NetplayWaitData.game_pal;
in-out property netplay_connected_peers_count <=> NetplayWaitData.connected_peers_count;
in-out property netplay_is_host <=> NetplayWaitData.is_host;
in-out property netplay_resyncing <=> NetplayData.resyncing;
private property <string> popup_message;
private property <bool> popup_has_close;
@ -275,6 +280,9 @@ export component AppWindow inherits Window {
begin_game(a) => {
netplay_begin_game(a);
}
save_session() => {
netplay_save_session();
}
close => {
netplay_close();
}
@ -285,6 +293,9 @@ export component AppWindow inherits Window {
select_rom(rom_dir) => {
netplay_select_rom(rom_dir);
}
select_netplay_state() => {
netplay_select_state();
}
create_session(a, b, c, d, e, f, g, h) => {
netplay_create_session(a,b,c,d,e,f,g,h);
}

View File

@ -16,7 +16,7 @@ export component NetplayCreate inherits Page {
title: @tr("Netplay: Create Session");
callback select_rom(string);
callback create_session(string, string, string, string, string, string, bool, bool);
callback resume_session(string);
callback select_netplay_state;
callback close;
VerticalBox {
@ -118,10 +118,16 @@ export component NetplayCreate inherits Page {
text: @tr("Resume Session…");
enabled: !NetplayData.pending_session;
clicked => {
resume_session(SettingsData.rom_dir);
select_netplay_state();
}
}
if NetplayData.state_path != "" : Text {
text: @tr("Resume session: peers will sync to the saved state when they join.");
horizontal-alignment: center;
wrap: word-wrap;
}
Button {
text: @tr("Cancel");
clicked => {

View File

@ -17,6 +17,7 @@ export global NetplayData {
in-out property <string> game_hash;
in-out property <string> game_cheats;
in-out property <string> rom_path;
in-out property <string> state_path;
in-out property <string> password;
in-out property <bool> pending_session;
in-out property <bool> pending_refresh;

View File

@ -150,6 +150,22 @@ fn setup_callbacks(
netplay_write_receiver_create_session.resubscribe(),
);
// issue #4: a picked `.netplay-state` file marks this session as
// a resume — advertised to joining peers via session features.
let (resume, state_path) = weak
.upgrade()
.map(|handle| {
let path = handle.get_netplay_state_path();
let resume = !path.is_empty();
let state_path = if resume {
Some(std::path::PathBuf::from(path.to_string()))
} else {
None
};
(resume, state_path)
})
.unwrap_or((false, None));
create_session(
netplay_write_sender_create_session.clone(),
netplay_read_receiver_create_session.resubscribe(),
@ -162,6 +178,8 @@ fn setup_callbacks(
password.to_string(),
overclock,
disable_expansion_pak,
resume,
state_path,
weak.clone(),
);
},
@ -272,27 +290,21 @@ fn setup_callbacks(
open_uri("https://github.com/gopher64/gopher64/discussions/453");
});
// Resume Session: open a file picker for .netplay-state files
// Resume Session (issue #4): pick a `.netplay-state` file. Its path
// marks the upcoming session as a resume; the create flow then advertises
// it to joining peers and the host broadcasts the file once they connect.
let weak_app = app.as_weak();
app.on_netplay_resume_session(move |rom_dir| {
app.on_netplay_select_state(move || {
let weak_app = weak_app.clone();
tokio::spawn(async move {
if let Some(file) = rfd::AsyncFilePicker::new()
.add_filter("Netplay State", &["netplay-state"])
.set_directory(rom_dir)
.pick_file()
.await
{
eprintln!("resume session from: {}", file.display());
// TODO: #3 — load the netplay session state from the file and
// re-establish the connection. For now, just show a message.
if let Some(path) = ui::gui::select_netplay_state().await {
weak_app
.upgrade_in_event_loop(|handle| {
.upgrade_in_event_loop(move |handle| {
handle.set_netplay_state_path(path.to_string_lossy().to_string().into());
ui::video::onscreen_message(
"Resume session not yet implemented (#3)",
"Resume session selected — peers will sync to this saved state",
ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
);
let _ = handle;
})
.ok();
}
@ -458,6 +470,8 @@ fn create_session(
password: String,
overclock: bool,
disable_expansion_pak: bool,
resume: bool,
state_path: Option<std::path::PathBuf>,
weak_app: slint::Weak<AppWindow>,
) {
tokio::spawn(async move {
@ -467,6 +481,12 @@ fn create_session(
features.insert("cheats".to_string(), game_cheats);
}
// issue #4: advertise that joining peers must sync to the saved
// session state before play begins.
if resume {
features.insert("resume".to_string(), "1".to_string());
}
features.insert("overclock".to_string(), overclock.to_string());
features.insert(
"disable_expansion_pak".to_string(),
@ -544,6 +564,8 @@ fn create_session(
cheats: serde_json::from_str(cheats).unwrap(),
load_savestate_slot: None,
},
resume,
state_path,
&handle,
);
})
@ -779,6 +801,8 @@ fn setup_wait_window(
game_name: slint::SharedString,
rom_path: slint::SharedString,
game_settings: ui::GameSettings,
resume: bool,
state_path: Option<std::path::PathBuf>,
app: &AppWindow,
) {
update_ice_config(ice_server_config.clone());
@ -871,6 +895,8 @@ fn setup_wait_window(
player_number,
number_of_players: players.row_count(),
input_delay,
resume,
state_path,
}),
weak_app2,
);
@ -962,6 +988,15 @@ fn setup_join_window(
.unwrap()
.get("cheats")
.unwrap_or(&cheats_default);
// issue #4: a session created by a host in resume
// mode advertises it; joining peers sync to the
// state sent by the host once connected.
let resume_feature = session
.features
.as_ref()
.unwrap()
.get("resume")
.is_some_and(|value| value.as_str() == "1");
setup_wait_window(
sender,
receiver,
@ -977,6 +1012,8 @@ fn setup_join_window(
cheats: serde_json::from_str(cheats).unwrap(),
load_savestate_slot: None,
},
resume_feature,
None,
&handle,
);
})