The out trace logs every world send, the login handshake's included (#806)

The `out` trace tag logged only what the long-lived writer sent, drained after each command. Everything the session sends before the reader and writer split never reached it: `CMSG_AUTH_SESSION`, `CMSG_CHAR_ENUM`, `CMSG_PLAYER_LOGIN`, character create and delete, the cinematic ack and logout. So a login trace could not show where `CMSG_PLAYER_LOGIN` sits against `SMSG_LOGIN_VERIFY_WORLD`, which wire-timing work needs.

The world socket has one write path, `send_packet`. It now reports each packet to an observer right after the write, and the app installs the trace there only when the `out` tag is on. The writer's sent-log and its drains are gone; the line format is unchanged.

A protocol test over the fake server checks that the handshake, a pre-split send and a writer send are all observed, in order. A live login trace now opens with `CMSG_AUTH_SESSION`, `CMSG_CHAR_ENUM` and `CMSG_PLAYER_LOGIN`.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/benilla-app/src/net/io.rs

@@ -190,6 +190,10 @@ pub(super) struct NetHandles {
 
 /// Spawns the read thread with its park and cycle loop, and the one long-lived write thread.
 pub(super) fn spawn_net(connect: bool) -> NetHandles {
+    // The outbound opcode trace (tag `out`), armed before any thread can send.
+    if benilla_assets::trace::enabled_for("out") {
+        benilla_protocol::observe_sends(trace_out);
+    }
     let (events_tx, events_rx) = crossbeam_channel::unbounded();
     let (cmd_tx, cmd_rx) = crossbeam_channel::unbounded();
     let (pick_tx, pick_rx) = crossbeam_channel::unbounded::<CharRequest>();
@@ -763,18 +767,16 @@ fn refuse_once(guid: u64) -> u64 {
     }
 }
 
-/// Drains the writer's sent-packet log into the trace as `out` lines, one per packet that reached
-/// the socket; a no-op unless the `out` tag armed it.
-fn trace_sends(w: &mut WorldWriter) {
-    w.drain_sent(|opcode, len| {
-        benilla_assets::trace::line(
-            "out",
-            &format!(
-                "{opcode:#06x} {} len={len}",
-                benilla_protocol::messages::opcode_name(opcode).unwrap_or("?")
-            ),
-        );
-    });
+/// One `out` trace line per client packet written to the world socket, whichever session or writer
+/// sent it: the hook [`spawn_net`] installs when the `out` tag is on.
+fn trace_out(opcode: u16, len: usize) {
+    benilla_assets::trace::line(
+        "out",
+        &format!(
+            "{opcode:#06x} {} len={len}",
+            benilla_protocol::messages::opcode_name(opcode).unwrap_or("?")
+        ),
+    );
 }
 
 /// The write thread: app commands, writer swaps and the [`PING_INTERVAL`] keepalive. With no live
@@ -792,11 +794,7 @@ fn writer_loop(
     loop {
         crossbeam_channel::select! {
             recv(writer_rx) -> w => match w {
-                Ok(mut w) => {
-                    // Arm the outbound opcode trace (tag `out`); a fresh socket starts a fresh log.
-                    if benilla_assets::trace::enabled_for("out") {
-                        w.watch_sends();
-                    }
+                Ok(w) => {
                     writer = Some(w);
                     warned = 0;
                     // Sequence 1 is the new socket's first ping, so an old socket's pong cannot
@@ -833,7 +831,6 @@ fn writer_loop(
                             warned += 1;
                         }
                     }
-                    trace_sends(w);
                 }
             },
             recv(cmd_rx) -> cmd => {
@@ -1491,8 +1488,6 @@ fn writer_loop(
                         warned += 1;
                     }
                 }
-                // What reached the socket, by name (tag `out`); one command can be several packets.
-                trace_sends(w);
             },
         }
     }

crates/benilla-protocol/src/lib.rs

@@ -20,7 +20,8 @@ pub use messages::{
     CHARACTER_FLAG_HIDE_CLOAK, CHARACTER_FLAG_HIDE_HELM, CHARACTER_FLAG_RENAME,
 };
 pub use world::{
-    WardenRequired, WorldAuthReject, WorldReader, WorldSession, WorldWriter, WORLD_PORT,
+    observe_sends, SendObserver, WardenRequired, WorldAuthReject, WorldReader, WorldSession,
+    WorldWriter, WORLD_PORT,
 };
 
 use std::net::TcpStream;

crates/benilla-protocol/src/world/mod.rs

@@ -6,6 +6,7 @@
 
 use std::io::{Read, Write};
 use std::net::TcpStream;
+use std::sync::OnceLock;
 
 use anyhow::{anyhow, Result};
 use benilla_srp::vanilla_header::{DecrypterHalf, EncrypterHalf};
@@ -47,7 +48,21 @@ pub(super) fn recv_packet(
     messages::parse_server(opcode, &body).map_err(|e| anyhow!("parsing opcode {opcode:#x}: {e}"))
 }
 
+/// What [`observe_sends`] takes: called with a sent packet's opcode and body length.
+pub type SendObserver = fn(opcode: u16, body_len: usize);
+
+static SEND_OBSERVER: OnceLock<SendObserver> = OnceLock::new();
+
+/// Report every client packet that reaches a world socket, from the `CMSG_AUTH_SESSION` of the
+/// handshake through each [`WorldSession`] and [`WorldWriter`] send, on the thread that wrote it
+/// and right after the write. The first observer set is kept for the process; with none set a send
+/// pays one atomic load.
+pub fn observe_sends(observer: SendObserver) {
+    let _ = SEND_OBSERVER.set(observer);
+}
+
 /// Write one client packet: a 6-byte header, its size counting opcode and body, then the body.
+/// The sole write path of the world socket, so the one place [`observe_sends`] reports from.
 pub(super) fn send_packet(
     stream: &mut TcpStream,
     encrypter: Option<&mut EncrypterHalf>,
@@ -70,5 +85,9 @@ pub(super) fn send_packet(
     packet.extend_from_slice(body);
     stream
         .write_all(&packet)
-        .map_err(|e| anyhow!("sending opcode {opcode:#x}: {e}"))
+        .map_err(|e| anyhow!("sending opcode {opcode:#x}: {e}"))?;
+    if let Some(observer) = SEND_OBSERVER.get() {
+        observer(opcode, body.len());
+    }
+    Ok(())
 }

crates/benilla-protocol/src/world/session.rs

@@ -778,7 +778,6 @@ impl WorldSession {
                 stream: self.stream,
                 encrypter,
                 chat_language: self.chat_language,
-                sent: None,
             },
         ))
     }

crates/benilla-protocol/src/world/writer/mod.rs

@@ -63,34 +63,15 @@ pub struct WorldWriter {
     /// Every packet that reached the socket since the last drain, as `(opcode, body length)`;
     /// `None` until [`Self::watch_sends`]. Pushed only after a successful write.
     pub(super) sent: Option<Vec<(u16, usize)>>,
-    /// The character's faction language, set at login from its race; every chat send carries it.
-    /// vmangos drops the whole message, dot-commands included, when the character does not know it.
+    /// The character's faction language, set at login from its race, which a chat send naming no
+    /// language carries. vmangos drops the whole message, dot-commands included, when the
+    /// character does not know the language.
     pub(super) chat_language: u32,
 }
 
 impl WorldWriter {
     /// Frame, encrypt and write one packet: the sole write path every verb goes through.
     fn send(&mut self, opcode: u16, body: &[u8]) -> Result<()> {
-        let sent = send_packet(&mut self.stream, Some(&mut self.encrypter), opcode, body);
-        if sent.is_ok() {
-            if let Some(log) = &mut self.sent {
-                log.push((opcode, body.len()));
-            }
-        }
-        sent
-    }
-
-    /// Start recording what reaches the socket; a second call keeps what is not yet drained.
-    pub fn watch_sends(&mut self) {
-        self.sent.get_or_insert_with(Vec::new);
-    }
-
-    /// Hand over and clear the recorded `(opcode, body length)` pairs, in send order.
-    pub fn drain_sent(&mut self, mut each: impl FnMut(u16, usize)) {
-        if let Some(log) = &mut self.sent {
-            for (opcode, len) in log.drain(..) {
-                each(opcode, len);
-            }
-        }
+        send_packet(&mut self.stream, Some(&mut self.encrypter), opcode, body)
     }
 }

crates/benilla-protocol/tests/handshake.rs

@@ -94,3 +94,50 @@ fn a_warden_server_is_refused_at_the_handshake() {
         "expected WardenRequired, got: {err:#}"
     );
 }
+
+/// Every packet the observer saw, with the thread that wrote it: the tests of this file share one
+/// process, so each filters for its own thread.
+static SEEN: std::sync::Mutex<Vec<(thread::ThreadId, u16, usize)>> =
+    std::sync::Mutex::new(Vec::new());
+
+fn record(opcode: u16, len: usize) {
+    SEEN.lock()
+        .unwrap()
+        .push((thread::current().id(), opcode, len));
+}
+
+/// A packet sent before the session splits (the handshake's `CMSG_AUTH_SESSION`, the world entry's
+/// `CMSG_PLAYER_LOGIN`) reaches the observer in write order, ahead of the writer's own.
+#[test]
+fn sends_before_the_split_are_observed_in_order() {
+    benilla_protocol::observe_sends(record);
+    let addr = fake_server(vec![]);
+    let mut session = WorldSession::connect(&addr, "one", SESSION_KEY).unwrap();
+    session.player_login(0x2a).unwrap();
+    let (_reader, mut writer) = session.into_split().unwrap();
+    writer.complete_cinematic().unwrap();
+    let me = thread::current().id();
+    let mine: Vec<(u16, usize)> = SEEN
+        .lock()
+        .unwrap()
+        .iter()
+        .filter(|(t, ..)| *t == me)
+        .map(|&(_, op, len)| (op, len))
+        .collect();
+    let auth_len = messages::auth_session(
+        u32::from(benilla_protocol::CLIENT_BUILD),
+        "ONE",
+        0,
+        &[0; 20],
+        &messages::STOCK_SECURE_ADDONS,
+    )
+    .len();
+    assert_eq!(
+        mine,
+        [
+            (opcode::CMSG_AUTH_SESSION, auth_len),
+            (opcode::CMSG_PLAYER_LOGIN, messages::full_guid(0x2a).len()),
+            (opcode::CMSG_COMPLETE_CINEMATIC, 0),
+        ]
+    );
+}