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>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
ffdc206ac2
commit
f8493febb0
@@ -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 => {
|
||||
@@ -1478,8 +1475,6 @@ fn writer_loop(
|
||||
warned += 1;
|
||||
}
|
||||
}
|
||||
// What reached the socket, by name (tag `out`); one command can be several packets.
|
||||
trace_sends(w);
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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(())
|
||||
}
|
||||
|
||||
@@ -778,7 +778,6 @@ impl WorldSession {
|
||||
stream: self.stream,
|
||||
encrypter,
|
||||
chat_language: self.chat_language,
|
||||
sent: None,
|
||||
},
|
||||
))
|
||||
}
|
||||
|
||||
@@ -60,9 +60,6 @@ mod vendor;
|
||||
pub struct WorldWriter {
|
||||
pub(super) stream: TcpStream,
|
||||
pub(super) encrypter: EncrypterHalf,
|
||||
/// 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, which a chat send naming no
|
||||
/// language carries. vmangos drops the whole message, dot-commands included, when the
|
||||
/// character does not know the language.
|
||||
@@ -72,26 +69,6 @@ pub struct WorldWriter {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user