netplay: count same-batch peer acks in session-resume handshake

wait_for_ready_all counted acks by removing them from the by-name
messages map, which collapses every ack that arrives in the same
channel poll into one entry: with 3+ players the host could
undercount and time out a session whose peers all synced. Count the
matching messages of each decoded batch instead (process_reliable_messages
now returns the batch it decoded).

src/netplay.rs

@@ -149,7 +149,12 @@ fn send_message(netplay: &mut Netplay, message: NetplayMessage) {
     }
 }
 
-fn process_reliable_messages(netplay: &mut Netplay) {
+/// Drain the reliable channel and route every complete message into
+/// `netplay.messages`, returning the decoded batch so callers can count
+/// messages that share a name: several peers may ack within a single
+/// channel poll, and the by-name `messages` map would collapse them into
+/// one entry.
+fn process_reliable_messages(netplay: &mut Netplay) -> Vec<(String, Vec<u8>)> {
     let packets = netplay
         .reliable_channel
         .receive()
@@ -157,15 +162,18 @@ fn process_reliable_messages(netplay: &mut Netplay) {
         .map(|(_, data)| data.to_vec())
         .collect::<Vec<Vec<u8>>>();
 
+    let mut decoded = Vec::new();
     for (name, data) in decode_packets(
         &mut netplay.incoming_message,
         &mut netplay.received_data,
         packets,
     ) {
-        netplay.messages.insert(name, data);
+        netplay.messages.insert(name.clone(), data.clone());
         check_input_delay(netplay);
         check_disconnect(netplay);
+        decoded.push((name, data));
     }
+    decoded
 }
 
 /// Poll `messages` for `name` until it arrives or `timeout` elapses, calling
@@ -469,11 +477,20 @@ pub fn save_netplay_session(device: &mut device::Device, path: &std::path::Path)
 /// 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);
 
+/// Count how many messages in a decoded batch carry `name`. Used for
+/// ready-acks, where one ack per peer is expected and several peers can
+/// acknowledge within a single channel poll.
+fn count_messages_named(decoded: &[(String, Vec<u8>)], name: &str) -> usize {
+    decoded.iter().filter(|(decoded_name, _)| decoded_name == name).count()
+}
+
 /// 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.
+/// broadcasts to every peer, so the host receives one copy per peer. The
+/// count comes from the decoded batch of every poll — not from the
+/// by-name `messages` map — so acks arriving together are all counted.
 fn wait_for_ready_all(
     netplay: &mut Netplay,
     name: &str,
@@ -483,10 +500,9 @@ fn wait_for_ready_all(
     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;
+        received += count_messages_named(&process_reliable_messages(netplay), name);
+        if received >= expected {
+            return true;
         }
         if now.elapsed() > timeout {
             return false;
@@ -1239,7 +1255,7 @@ mod tests {
     /// there, not at 0.
     #[test]
     fn resume_state_reaches_peer_with_same_frame_metadata() {
-        use crate::savestates::{create_savestate_in_memory, load_savestate_from_data};
+        use crate::savestates::create_savestate_in_memory;
 
         // Host: a session with distinctive emulator state, captured in
         // memory (issue #2 API) and wrapped in the issue #3 on-disk
@@ -1318,6 +1334,7 @@ mod tests {
     #[test]
     fn resume_host_loads_state_from_disk_and_resets_frame() {
         use crate::device::Device;
+        use crate::savestates::create_savestate_in_memory;
 
         let mut host_device = Device::new(false);
         host_device.cpu.gpr[5] = 0xDEAD_BEEF;
@@ -1368,23 +1385,29 @@ mod tests {
         let _ = std::fs::remove_file(&path);
     }
 
-    /// The host's 30s ack wait succeeds once every peer acknowledged and
-    /// times out (returning false) when acks are missing.
+    /// Ack counting is per decoded batch: several peers acknowledging
+    /// within a single channel poll are all counted, and unrelated
+    /// messages are ignored.
+    #[test]
+    fn ack_batch_counts_every_ack() {
+        let batch = vec![
+            (NETPLAY_STATE_READY_MESSAGE.to_string(), vec![]),
+            (NETPLAY_STATE_READY_MESSAGE.to_string(), vec![]),
+            ("unrelated".to_string(), vec![1]),
+        ];
+        assert_eq!(
+            count_messages_named(&batch, NETPLAY_STATE_READY_MESSAGE),
+            2,
+            "both acks in one poll are counted; unrelated messages are not"
+        );
+    }
+
+    /// The host's 30s ack wait times out (returning false) when acks are
+    /// missing instead of hanging forever.
     #[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.
+        // Missing acks (the mock channel never delivers): gives up after
+        // the timeout.
         let mut netplay = mock_netplay(0);
         assert!(!wait_for_ready_all(
             &mut netplay,