Harness: read past the family entry packets; client queue room for busy maps

integration-harness/src/packets.rs

@@ -275,6 +275,9 @@ pub fn opcode_name(opcode: i16) -> &'static str {
         x if x == SendOpcode::SpawnReactor as i16 => "SpawnReactor",
         x if x == SendOpcode::HitReactor as i16 => "HitReactor",
         x if x == SendOpcode::RemoveReactor as i16 => "RemoveReactor",
+        x if x == SendOpcode::FamilyInfo as i16 => "FamilyInfo",
+        x if x == SendOpcode::FamilyPrivilegeList as i16 => "FamilyPrivilegeList",
+        x if x == SendOpcode::FamilySetPrivilege as i16 => "FamilySetPrivilege",
         _ => "Unknown",
     }
 }

integration-harness/src/protocol.rs

@@ -17,7 +17,7 @@ use crate::packets::{
 use net::packet::op::SendOpcode;
 use std::net::Ipv4Addr;
 use std::net::SocketAddr;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use tokio::time::timeout;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -73,6 +73,10 @@ pub struct WorldSession {
     pub quick_slots: Option<[i32; 8]>,
     /// The name Blessing of the Fairy's tooltip shows, if any.
     pub linked_name: Option<String>,
+    /// The family packets of entering the game the login read past (INFO,
+    /// the entitlements, and a working entitlement's icon), as (opcode,
+    /// body). Any that came after another packet are still to be read.
+    pub family: Vec<(i16, Vec<u8>)>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -258,7 +262,7 @@ pub async fn login_to_world_session(config: &HarnessConfig) -> Result<WorldSessi
         ));
     };
 
-    drain_spawn_npcs(&mut world_conn).await;
+    let family = drain_spawn_npcs(&mut world_conn).await;
 
     Ok(WorldSession {
         connection: world_conn,
@@ -291,6 +295,7 @@ pub async fn login_to_world_session(config: &HarnessConfig) -> Result<WorldSessi
         spawn_point: world_entry.spawn_point,
         login_addr: config.login_addr.to_string(),
         world_addr: format!("{}:{}", redirect.redirect_ip, redirect.redirect_port),
+        family,
     })
 }
 
@@ -492,18 +497,44 @@ async fn login_until_char_select(
     Ok((login_conn, selected))
 }
 
-async fn drain_spawn_npcs(connection: &mut MapleTestConnection) {
-    while let Ok(Ok(envelope)) = timeout(
-        Duration::from_millis(100),
-        connection.read_packet("drain spawn npcs"),
-    )
-    .await
-    {
+/// Reads past map setup and returns the family packets of entering the
+/// game. Every entry ends with the Family window's INFO and entitlements,
+/// so this waits for those; anything else that comes first is left to read,
+/// in order.
+async fn drain_spawn_npcs(connection: &mut MapleTestConnection) -> Vec<(i16, Vec<u8>)> {
+    let mut family = Vec::new();
+    let mut held = Vec::new();
+    let mut listed = false;
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        let wait = if listed {
+            Duration::from_millis(100)
+        } else {
+            deadline.saturating_duration_since(Instant::now())
+        };
+        let Ok(Ok(envelope)) = timeout(wait, connection.read_packet("drain spawn npcs")).await
+        else {
+            break;
+        };
+        let opcode = envelope.opcode();
+        // The Family window's INFO and entitlements, and a working
+        // entitlement's icon.
+        if [
+            SendOpcode::FamilyInfo as i16,
+            SendOpcode::FamilyPrivilegeList as i16,
+            SendOpcode::FamilySetPrivilege as i16,
+        ]
+        .contains(&opcode)
+        {
+            listed |= opcode == SendOpcode::FamilyPrivilegeList as i16;
+            family.push((opcode, envelope.packet.bytes[2..].to_vec()));
+            continue;
+        }
         // Map setup: NPC spawns and control, and the map entry banner.
         // ...and the map's mobs, drops, and reactors, and the skill update
         // that gives Blessing of the Fairy its master level (SET_FIELD
         // already carries its level).
-        if ![
+        let setup = [
             SendOpcode::SpawnNpc as i16,
             SendOpcode::SpawnNpcRequestController as i16,
             SendOpcode::FieldEffect as i16,
@@ -513,12 +544,19 @@ async fn drain_spawn_npcs(connection: &mut MapleTestConnection) {
             SendOpcode::SpawnReactor as i16,
             SendOpcode::UpdateSkills as i16,
         ]
-        .contains(&envelope.opcode())
-        {
-            connection.push_back_packet(envelope);
+        .contains(&opcode);
+        if setup && held.is_empty() {
+            continue;
+        }
+        held.push(envelope);
+        if listed {
             break;
         }
     }
+    for envelope in held.into_iter().rev() {
+        connection.push_back_packet(envelope);
+    }
+    family
 }
 
 /// Log in again right after dropping a session. The server closes the old

integration-harness/tests/families.rs

@@ -125,12 +125,22 @@ async fn new_character(config: &HarnessConfig, level: i16) -> i32 {
 /// Log in and go to `map`; the family packets of entering the game.
 async fn enter(config: &HarnessConfig, map: i32) -> (WorldSession, Trace) {
     let mut session = relogin(config).await;
-    let mut t = Trace::default();
-    until(&mut session.connection, &mut t, |t| t.family.len() >= 2).await;
+    let mut t = Trace {
+        family: entered(&mut session, 2).await,
+        ..Trace::default()
+    };
     assert_eq!(warp(&mut session.connection, map, "", &mut t).await, map);
     (session, t)
 }
 
+/// The family packets of `session` entering the game, at least `count`:
+/// those the login read past, then any still to come.
+async fn entered(session: &mut WorldSession, count: usize) -> Vec<(i16, Vec<u8>)> {
+    let mut packets = std::mem::take(&mut session.family);
+    packets.extend(family(session, count.saturating_sub(packets.len())).await);
+    packets
+}
+
 /// The next `count` family packets `session` gets.
 async fn family(session: &mut WorldSession, count: usize) -> Vec<(i16, Vec<u8>)> {
     let mut t = Trace::default();
@@ -157,9 +167,10 @@ async fn a_senior_takes_a_junior_on_their_map_who_says_yes() {
 
     // Out of a family: no leader and no name, and nothing else on login.
     let mut a = relogin(&a_config).await;
-    let mut at = Trace::default();
-    until(&mut a.connection, &mut at, |t| t.family.len() >= 2).await;
-    read_for(&mut a.connection, &mut at, 300).await;
+    let mut at = Trace {
+        family: entered(&mut a, 2).await,
+        ..Trace::default()
+    };
     assert_eq!(at.family.len(), 2);
     assert_eq!(at.family[0], info(0, 0, ""));
     // The eleven entitlements, the first two to a relative named next.
@@ -255,8 +266,8 @@ async fn a_senior_takes_a_junior_on_their_map_who_says_yes() {
     assert_eq!(family(&mut a, 1).await, [online(0)]);
     let mut b = relogin(&b_config).await;
     assert_eq!(family(&mut a, 1).await, [online(1)]);
-    let entered = family(&mut b, 2).await;
-    assert_eq!(entered[0], info(0, a_id, &an));
+    let packets = entered(&mut b, 2).await;
+    assert_eq!(packets[0], info(0, a_id, &an));
 }
 
 /// `senior` asks `junior`, both on one map, who says yes.
@@ -639,7 +650,10 @@ async fn members_buy_rates_and_see_their_icon_until_they_end() {
     tokio::time::sleep(Duration::from_millis(500)).await;
     end_family_buff_in(b_id, 8).await.expect("buff");
     let mut b = relogin(&b_config).await;
-    let mut t = Trace::default();
+    let mut t = Trace {
+        family: std::mem::take(&mut b.family),
+        ..Trace::default()
+    };
     until(&mut b.connection, &mut t, |t| {
         t.family
             .iter()

runtime/src/message.rs

@@ -30,8 +30,11 @@ use tokio_util::sync::CancellationToken;
 use tracing::warn;
 
 /// Capacity of each client's inbound queue. A client this far behind is
-/// considered dead weight and disconnected.
-pub const CLIENT_QUEUE_CAPACITY: usize = 256;
+/// considered dead weight and disconnected. Entering a map queues a packet
+/// for each of its NPCs, players, mobs (and control of them), reactors and
+/// drops while the client is still busy with the map change, close to 200
+/// in Kerning City's subway, so this leaves room for the busiest maps.
+pub const CLIENT_QUEUE_CAPACITY: usize = 1024;
 
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 pub struct FieldKey {