From 6246417ba403242d8f3d6da42c469dc63ee83668 Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Sat, 24 Oct 2026 21:40:41 +0200 Subject: [PATCH 1/6] feat(diagnostics): preserve server trace state (RUST-CI-031) --- .../src/ed2k_server/diagnostics.rs | 92 ++++++++++++++++--- 1 file changed, 80 insertions(+), 12 deletions(-) diff --git a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs index 860de1e2..04b77e09 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs @@ -370,22 +370,29 @@ fn dump_ed2k_server_record(record: &Ed2kServerDumpRecord<'_>) { #[cfg(feature = "packet-diagnostics")] fn emit_ed2k_server_diag_event(record: &Ed2kServerDumpRecord<'_>) { - let (family, keys, body) = ed2k_server_diag_event_parts(record); - let severity = if record.outcome == "dropped" { - "warn" - } else { - "info" - }; - crate::diag_event::emit(family, "packet", severity, keys, body); + let (family, event, severity, keys, body) = ed2k_server_diag_event_parts(record); + crate::diag_event::emit(family, event, severity, keys, body); } #[cfg(any(feature = "packet-diagnostics", test))] -fn ed2k_server_diag_event_parts(record: &Ed2kServerDumpRecord<'_>) -> (&'static str, Value, Value) { +fn ed2k_server_diag_event_parts( + record: &Ed2kServerDumpRecord<'_>, +) -> (&'static str, &'static str, &'static str, Value, Value) { let family = if record.role == "udp" { "ed2k_server_udp" } else { "ed2k_tcp" }; + let event = if record.direction == "meta" { + "state" + } else { + "packet" + }; + let severity = if matches!(record.outcome, "dropped" | "send_failed") { + "warning" + } else { + "info" + }; let mut keys = Map::new(); keys.insert("peer".to_string(), json!(record.remote_addr)); @@ -440,11 +447,20 @@ fn ed2k_server_diag_event_parts(record: &Ed2kServerDumpRecord<'_>) -> (&'static body.insert("phase".to_string(), json!(record.phase)); body.insert("role".to_string(), json!(record.role)); body.insert("traceId".to_string(), json!(record.trace_id)); + body.insert("traceKey".to_string(), json!(record.trace_key)); + body.insert("stateId".to_string(), json!(record.state_id)); + body.insert("stateLabel".to_string(), json!(record.state_label)); if let Some(note) = record.note.as_deref() { body.insert("note".to_string(), json!(note)); } - (family, Value::Object(keys), Value::Object(body)) + ( + family, + event, + severity, + Value::Object(keys), + Value::Object(body), + ) } #[cfg(feature = "packet-diagnostics")] @@ -552,10 +568,15 @@ mod tests { note: None, }; - let (family, keys, body) = ed2k_server_diag_event_parts(&record); + let (family, event, severity, keys, body) = ed2k_server_diag_event_parts(&record); assert_eq!(family, "ed2k_tcp"); + assert_eq!(event, "packet"); + assert_eq!(severity, "info"); assert_eq!(keys["peer"], "198.51.100.1:4661"); assert_eq!(body["outcome"], "accepted"); + assert_eq!(body["traceKey"], "server:background:7:198.51.100.1:4661"); + assert_eq!(body["stateId"], "server.background.connected"); + assert_eq!(body["stateLabel"], "connected"); assert_eq!(body["obfuscated"], false); assert_eq!(body["rawLen"], raw.len()); assert_eq!(body["rawHex"].as_str().unwrap().len(), raw.len() * 2); @@ -605,8 +626,10 @@ mod tests { note: None, }; - let (family, _, body) = ed2k_server_diag_event_parts(&record); + let (family, event, severity, _, body) = ed2k_server_diag_event_parts(&record); assert_eq!(family, "ed2k_server_udp"); + assert_eq!(event, "packet"); + assert_eq!(severity, "info"); assert_eq!(body["obfuscated"], true); } @@ -643,14 +666,59 @@ mod tests { note: Some("decode failed".to_string()), }; - let (family, _, body) = ed2k_server_diag_event_parts(&record); + let (family, event, severity, _, body) = ed2k_server_diag_event_parts(&record); assert_eq!(family, "ed2k_server_udp"); + assert_eq!(event, "packet"); + assert_eq!(severity, "warning"); assert_eq!(body["outcome"], "dropped"); assert_eq!(body["rawHex"], "1234abcd"); assert!(body.get("clearFrameHex").is_none()); assert!(body.get("payloadHex").is_none()); } + #[test] + fn server_meta_record_is_a_state_event_with_correlation() { + let record = Ed2kServerDumpRecord { + schema: "ed2k_packet_v1", + source: "emulebb-rust", + ts_utc: "2026-10-24T00:00:00.000Z".to_string(), + event_seq: 1, + trace_id: 11, + trace_key: "server:background:11:198.51.100.3:4661".to_string(), + state_id: "server.background.connected".to_string(), + state_label: "connected", + role: "background", + flow: "server", + phase: "connected", + direction: "meta", + outcome: "meta", + remote_addr: "198.51.100.3:4661".to_string(), + transport_mode: "plaintext", + obfuscated: Some(false), + protocol: None, + protocol_marker: None, + opcode: None, + opcode_name: None, + raw_len: None, + raw_hex: None, + clear_frame_len: None, + clear_frame_hex: None, + payload_len: None, + payload_hex: None, + payload_hex_truncated: None, + note: Some("connected".to_string()), + }; + + let (family, event, severity, _, body) = ed2k_server_diag_event_parts(&record); + assert_eq!(family, "ed2k_tcp"); + assert_eq!(event, "state"); + assert_eq!(severity, "info"); + assert_eq!(body["traceId"], 11); + assert_eq!(body["traceKey"], "server:background:11:198.51.100.3:4661"); + assert_eq!(body["stateId"], "server.background.connected"); + assert_eq!(body["stateLabel"], "connected"); + } + #[test] fn server_dump_names_global_udp_opcodes() { assert_eq!(server_opcode_name(OP_CALLBACKREQUEST), "OP_CALLBACKREQUEST"); From a8942a4fbf3de628e88ddb1ca23aae330333949b Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Sat, 24 Oct 2026 21:47:19 +0200 Subject: [PATCH 2/6] feat(diagnostics): record terminal server TCP sends (RUST-CI-031) --- .../src/ed2k_server/diagnostics.rs | 260 +++++++++++++++++- .../emulebb-ed2k/src/ed2k_server/session.rs | 28 +- 2 files changed, 272 insertions(+), 16 deletions(-) diff --git a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs index 04b77e09..f0248ed4 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs @@ -57,6 +57,7 @@ struct Ed2kServerDumpRecord<'a> { phase: &'a str, direction: &'a str, outcome: &'a str, + send_attempt_id: Option, remote_addr: String, transport_mode: &'a str, obfuscated: Option, @@ -88,6 +89,113 @@ pub(super) struct Ed2kServerPacketCapture<'a> { pub(super) note: Option, } +#[cfg(any(feature = "packet-diagnostics", test))] +struct Ed2kServerSendContext { + attempt_id: u64, + trace_id: u64, + trace_key: String, + state_id: String, + state_label: &'static str, + role: &'static str, + phase: &'static str, + remote_addr: String, + transport_mode: &'static str, + obfuscated: Option, + protocol_marker: Option, + opcode: Option, + attempted_len: usize, +} + +#[cfg(any(feature = "packet-diagnostics", test))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Ed2kServerSendOutcome { + Succeeded, + Failed, + Cancelled, +} + +#[cfg(any(feature = "packet-diagnostics", test))] +impl Ed2kServerSendOutcome { + const fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "send_succeeded", + Self::Failed => "send_failed", + Self::Cancelled => "send_cancelled", + } + } + + const fn severity(self) -> &'static str { + match self { + Self::Succeeded => "info", + Self::Failed | Self::Cancelled => "warning", + } + } +} + +#[cfg(feature = "packet-diagnostics")] +pub(super) struct Ed2kServerTcpSendAttempt { + context: Ed2kServerSendContext, + completed: bool, +} + +#[cfg(feature = "packet-diagnostics")] +impl Ed2kServerTcpSendAttempt { + pub(super) fn new(session: &ServerSession, capture: Ed2kServerPacketCapture<'_>) -> Self { + let attempt_id = next_ed2k_server_send_attempt_id(); + let context = Ed2kServerSendContext { + attempt_id, + trace_id: session.trace_id, + trace_key: ed2k_server_trace_key(session), + state_id: ed2k_server_state_id(session), + state_label: session.phase.as_str(), + role: session.trace_role, + phase: session.phase.as_str(), + remote_addr: session.endpoint.to_string(), + transport_mode: capture.transport_mode, + obfuscated: capture.obfuscated, + protocol_marker: capture.protocol_marker, + opcode: capture.opcode, + attempted_len: capture.raw.len(), + }; + dump_ed2k_server_packet_with_attempt(session, capture, Some(attempt_id)); + Self { + context, + completed: false, + } + } + + fn settle( + &mut self, + outcome: Ed2kServerSendOutcome, + delivery_known: bool, + note: Option, + ) { + self.completed = true; + emit_ed2k_server_send_outcome(&self.context, outcome, delivery_known, note.as_deref()); + } + + pub(super) fn succeeded(mut self) { + self.settle(Ed2kServerSendOutcome::Succeeded, true, None); + } + + pub(super) fn failed(mut self, error: String) { + self.settle(Ed2kServerSendOutcome::Failed, true, Some(error)); + } +} + +#[cfg(feature = "packet-diagnostics")] +impl Drop for Ed2kServerTcpSendAttempt { + fn drop(&mut self) { + if !self.completed { + self.settle( + Ed2kServerSendOutcome::Cancelled, + false, + Some("server TCP send future cancelled before completion".to_string()), + ); + } + } +} + #[cfg(not(feature = "packet-diagnostics"))] pub(super) fn dump_ed2k_server_meta(_session: &ServerSession, _note: impl Into) {} @@ -115,6 +223,7 @@ pub(super) fn dump_ed2k_server_meta(session: &ServerSession, note: impl Into, +) { + dump_ed2k_server_packet_with_attempt(session, capture, None); +} + +#[cfg(feature = "packet-diagnostics")] +fn dump_ed2k_server_packet_with_attempt( + session: &ServerSession, + capture: Ed2kServerPacketCapture<'_>, + send_attempt_id: Option, ) { // Normalize the wire direction to the shared send/recv vocabulary so the // packet-trace diff harness aligns these with eMuleBB's server packets. @@ -271,6 +391,7 @@ pub(super) fn dump_ed2k_server_packet( phase: session.phase.as_str(), direction, outcome: capture.outcome, + send_attempt_id, remote_addr: session.endpoint.to_string(), transport_mode: capture.transport_mode, obfuscated: capture.obfuscated, @@ -307,6 +428,12 @@ fn next_ed2k_server_dump_event_seq() -> u64 { NEXT_EVENT_SEQ.fetch_add(1, Ordering::Relaxed) } +#[cfg(feature = "packet-diagnostics")] +fn next_ed2k_server_send_attempt_id() -> u64 { + static NEXT_SEND_ATTEMPT_ID: AtomicU64 = AtomicU64::new(1); + NEXT_SEND_ATTEMPT_ID.fetch_add(1, Ordering::Relaxed) +} + fn ed2k_server_trace_key(session: &ServerSession) -> String { format!( "server:{}:{}:{}", @@ -396,6 +523,9 @@ fn ed2k_server_diag_event_parts( let mut keys = Map::new(); keys.insert("peer".to_string(), json!(record.remote_addr)); + if let Some(send_attempt_id) = record.send_attempt_id { + keys.insert("sendAttemptId".to_string(), json!(send_attempt_id)); + } if let Some(protocol_marker) = record.protocol_marker { keys.insert("protocolMarker".to_string(), json!(protocol_marker)); } @@ -406,6 +536,9 @@ fn ed2k_server_diag_event_parts( let mut body = Map::new(); body.insert("direction".to_string(), json!(record.direction)); body.insert("outcome".to_string(), json!(record.outcome)); + if let Some(send_attempt_id) = record.send_attempt_id { + body.insert("sendAttemptId".to_string(), json!(send_attempt_id)); + } if let Some(protocol_marker) = record.protocol_marker { body.insert("protocolMarker".to_string(), json!(protocol_marker)); } @@ -463,6 +596,70 @@ fn ed2k_server_diag_event_parts( ) } +#[cfg(feature = "packet-diagnostics")] +fn emit_ed2k_server_send_outcome( + context: &Ed2kServerSendContext, + outcome: Ed2kServerSendOutcome, + delivery_known: bool, + note: Option<&str>, +) { + let (severity, keys, body) = + ed2k_server_send_outcome_parts(context, outcome, delivery_known, note); + crate::diag_event::emit("ed2k_tcp", "send_outcome", severity, keys, body); +} + +#[cfg(any(feature = "packet-diagnostics", test))] +fn ed2k_server_send_outcome_parts( + context: &Ed2kServerSendContext, + outcome: Ed2kServerSendOutcome, + delivery_known: bool, + note: Option<&str>, +) -> (&'static str, Value, Value) { + let mut keys = Map::new(); + keys.insert("peer".to_string(), json!(context.remote_addr)); + keys.insert("sendAttemptId".to_string(), json!(context.attempt_id)); + if let Some(protocol_marker) = context.protocol_marker { + keys.insert("protocolMarker".to_string(), json!(protocol_marker)); + } + if let Some(opcode) = context.opcode { + keys.insert("opcode".to_string(), json!(opcode)); + } + + let mut body = Map::new(); + body.insert("direction".to_string(), json!("send")); + body.insert("outcome".to_string(), json!(outcome.as_str())); + body.insert("deliveryKnown".to_string(), json!(delivery_known)); + body.insert("sendAttemptId".to_string(), json!(context.attempt_id)); + body.insert("traceId".to_string(), json!(context.trace_id)); + body.insert("traceKey".to_string(), json!(context.trace_key)); + body.insert("stateId".to_string(), json!(context.state_id)); + body.insert("stateLabel".to_string(), json!(context.state_label)); + body.insert("role".to_string(), json!(context.role)); + body.insert("flow".to_string(), json!("server")); + body.insert("phase".to_string(), json!(context.phase)); + body.insert("transportMode".to_string(), json!(context.transport_mode)); + body.insert("attemptedLen".to_string(), json!(context.attempted_len)); + if let Some(obfuscated) = context.obfuscated { + body.insert("obfuscated".to_string(), json!(obfuscated)); + } + if let Some(protocol_marker) = context.protocol_marker { + body.insert("protocolMarker".to_string(), json!(protocol_marker)); + body.insert( + "protocolName".to_string(), + json!(server_protocol_name(protocol_marker)), + ); + } + if let Some(opcode) = context.opcode { + body.insert("opcode".to_string(), json!(opcode)); + body.insert("opcodeName".to_string(), json!(server_opcode_name(opcode))); + } + if let Some(note) = note { + body.insert("note".to_string(), json!(note)); + } + + (outcome.severity(), Value::Object(keys), Value::Object(body)) +} + #[cfg(feature = "packet-diagnostics")] fn maybe_sync_ed2k_server_dump(file: &fs::File) { static SINCE_SYNC: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); @@ -517,11 +714,12 @@ fn server_opcode_name(opcode: u8) -> &'static str { #[cfg(test)] mod tests { use super::{ - ED2K_SERVER_DUMP_FILE_PREFIX, Ed2kServerDumpRecord, OP_CALLBACKREQUEST, - OP_EDONKEYPROT_MARKER, OP_GLOBCALLBACKREQ, OP_GLOBFOUNDSOURCES, OP_GLOBSEARCHREQ, - OP_GLOBSEARCHREQ2, OP_GLOBSEARCHREQ3, OP_GLOBSEARCHRES, OP_PACKEDPROT, OP_SERVER_LIST_REQ, - OP_SERVER_LIST_REQ2, OP_SERVER_LIST_RES, ed2k_server_diag_event_parts, exact_packet_hex, - server_opcode_name, server_protocol_name, + ED2K_SERVER_DUMP_FILE_PREFIX, Ed2kServerDumpRecord, Ed2kServerSendContext, + Ed2kServerSendOutcome, OP_CALLBACKREQUEST, OP_EDONKEYPROT_MARKER, OP_GLOBCALLBACKREQ, + OP_GLOBFOUNDSOURCES, OP_GLOBSEARCHREQ, OP_GLOBSEARCHREQ2, OP_GLOBSEARCHREQ3, + OP_GLOBSEARCHRES, OP_PACKEDPROT, OP_SERVER_LIST_REQ, OP_SERVER_LIST_REQ2, + OP_SERVER_LIST_RES, ed2k_server_diag_event_parts, ed2k_server_send_outcome_parts, + exact_packet_hex, server_opcode_name, server_protocol_name, }; #[test] @@ -532,6 +730,46 @@ mod tests { ); } + #[test] + fn server_tcp_send_outcome_preserves_attempt_correlation_and_terminal_state() { + let context = Ed2kServerSendContext { + attempt_id: 42, + trace_id: 7, + trace_key: "server:background:7:198.51.100.1:4661".to_string(), + state_id: "server.background.connected".to_string(), + state_label: "connected", + role: "background", + phase: "connected", + remote_addr: "198.51.100.1:4661".to_string(), + transport_mode: "plaintext", + obfuscated: Some(false), + protocol_marker: Some(OP_EDONKEYPROT_MARKER), + opcode: Some(OP_CALLBACKREQUEST), + attempted_len: 73, + }; + + let (severity, keys, body) = + ed2k_server_send_outcome_parts(&context, Ed2kServerSendOutcome::Succeeded, true, None); + assert_eq!(severity, "info"); + assert_eq!(keys["sendAttemptId"], 42); + assert_eq!(body["outcome"], "send_succeeded"); + assert_eq!(body["deliveryKnown"], true); + assert_eq!(body["attemptedLen"], 73); + assert_eq!(body["traceKey"], context.trace_key); + assert_eq!(body["stateId"], context.state_id); + + let (severity, _, body) = ed2k_server_send_outcome_parts( + &context, + Ed2kServerSendOutcome::Cancelled, + false, + Some("cancelled"), + ); + assert_eq!(severity, "warning"); + assert_eq!(body["outcome"], "send_cancelled"); + assert_eq!(body["deliveryKnown"], false); + assert_eq!(body["note"], "cancelled"); + } + #[test] fn server_common_event_preserves_payloads_larger_than_the_former_cap() { let payload = vec![0xCD; 4 * 1024 + 1]; @@ -549,8 +787,9 @@ mod tests { role: "background", flow: "server", phase: "connected", - direction: "recv", - outcome: "accepted", + direction: "send", + outcome: "attempted", + send_attempt_id: Some(17), remote_addr: "198.51.100.1:4661".to_string(), transport_mode: "plaintext", obfuscated: Some(false), @@ -573,7 +812,9 @@ mod tests { assert_eq!(event, "packet"); assert_eq!(severity, "info"); assert_eq!(keys["peer"], "198.51.100.1:4661"); - assert_eq!(body["outcome"], "accepted"); + assert_eq!(keys["sendAttemptId"], 17); + assert_eq!(body["outcome"], "attempted"); + assert_eq!(body["sendAttemptId"], 17); assert_eq!(body["traceKey"], "server:background:7:198.51.100.1:4661"); assert_eq!(body["stateId"], "server.background.connected"); assert_eq!(body["stateLabel"], "connected"); @@ -609,6 +850,7 @@ mod tests { phase: "udp", direction: "send", outcome: "accepted", + send_attempt_id: None, remote_addr: "198.51.100.2:4665".to_string(), transport_mode: "obfuscated", obfuscated: Some(true), @@ -649,6 +891,7 @@ mod tests { phase: "udp", direction: "recv", outcome: "dropped", + send_attempt_id: None, remote_addr: "198.51.100.2:4665".to_string(), transport_mode: "obfuscated", obfuscated: Some(true), @@ -692,6 +935,7 @@ mod tests { phase: "connected", direction: "meta", outcome: "meta", + send_attempt_id: None, remote_addr: "198.51.100.3:4661".to_string(), transport_mode: "plaintext", obfuscated: Some(false), diff --git a/crates/emulebb-ed2k/src/ed2k_server/session.rs b/crates/emulebb-ed2k/src/ed2k_server/session.rs index 42b6faee..162f12ed 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/session.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/session.rs @@ -23,7 +23,7 @@ use crate::ed2k_tcp::MAX_ED2K_PACKET_LEN; use super::Ed2kServerFailure; #[cfg(feature = "packet-diagnostics")] -use super::diagnostics::Ed2kServerPacketCapture; +use super::diagnostics::{Ed2kServerPacketCapture, Ed2kServerTcpSendAttempt}; use super::offer_policy::{ OFFER_FILES_LEGACY_MIN_INTERVAL, OFFER_FILES_NEGOTIATION_WAIT, OfferFilesCapabilityAdvertisement, OfferFilesPolicy, @@ -684,7 +684,7 @@ impl ServerSession { cipher.apply(&mut packet); } #[cfg(feature = "packet-diagnostics")] - dump_ed2k_server_packet( + let send_attempt = Ed2kServerTcpSendAttempt::new( self, Ed2kServerPacketCapture { direction: "tx", @@ -695,17 +695,29 @@ impl ServerSession { raw: &packet, clear_frame: Some(&clear_frame), payload: Some(payload), - outcome: "accepted", + outcome: "attempted", note: None, }, ); #[cfg(not(feature = "packet-diagnostics"))] dump_ed2k_server_packet(self, "tx", protocol_marker, opcode, payload); - self.stream.write_all(&packet).await.with_context(|| { - format!("failed to send opcode=0x{opcode:02X} to {}", self.endpoint) - })?; - self.last_tx = Instant::now(); - Ok(()) + let send_result = + self.stream.write_all(&packet).await.with_context(|| { + format!("failed to send opcode=0x{opcode:02X} to {}", self.endpoint) + }); + match send_result { + Ok(()) => { + #[cfg(feature = "packet-diagnostics")] + send_attempt.succeeded(); + self.last_tx = Instant::now(); + Ok(()) + } + Err(error) => { + #[cfg(feature = "packet-diagnostics")] + send_attempt.failed(format!("{error:#}")); + Err(error) + } + } } pub(super) async fn send_encoded_packet( From 2082e4b574ee935a3b9fc556ad34e4ecc24e6429 Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Sat, 24 Oct 2026 21:52:49 +0200 Subject: [PATCH 3/6] feat(diagnostics): record terminal server UDP sends (RUST-CI-031) --- .../src/ed2k_server/diagnostics.rs | 124 +++++++++++++++--- .../emulebb-ed2k/src/ed2k_server/session.rs | 4 +- .../src/ed2k_server/udp_runtime.rs | 102 +++++++++++--- 3 files changed, 196 insertions(+), 34 deletions(-) diff --git a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs index f0248ed4..6f66266b 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs @@ -91,6 +91,7 @@ pub(super) struct Ed2kServerPacketCapture<'a> { #[cfg(any(feature = "packet-diagnostics", test))] struct Ed2kServerSendContext { + family: &'static str, attempt_id: u64, trace_id: u64, trace_key: String, @@ -133,16 +134,17 @@ impl Ed2kServerSendOutcome { } #[cfg(feature = "packet-diagnostics")] -pub(super) struct Ed2kServerTcpSendAttempt { +pub(super) struct Ed2kServerSendAttempt { context: Ed2kServerSendContext, completed: bool, } #[cfg(feature = "packet-diagnostics")] -impl Ed2kServerTcpSendAttempt { - pub(super) fn new(session: &ServerSession, capture: Ed2kServerPacketCapture<'_>) -> Self { +impl Ed2kServerSendAttempt { + pub(super) fn tcp(session: &ServerSession, capture: Ed2kServerPacketCapture<'_>) -> Self { let attempt_id = next_ed2k_server_send_attempt_id(); let context = Ed2kServerSendContext { + family: "ed2k_tcp", attempt_id, trace_id: session.trace_id, trace_key: ed2k_server_trace_key(session), @@ -164,6 +166,38 @@ impl Ed2kServerTcpSendAttempt { } } + pub(super) fn udp( + server: Option<&ResolvedServerEntry>, + remote_addr: SocketAddr, + capture: Ed2kServerPacketCapture<'_>, + ) -> Self { + let attempt_id = next_ed2k_server_send_attempt_id(); + let base_endpoint = server + .map(ResolvedServerEntry::base_endpoint) + .unwrap_or(remote_addr); + let context = Ed2kServerSendContext { + family: "ed2k_server_udp", + attempt_id, + trace_id: 0, + trace_key: format!("server:udp:{base_endpoint}"), + state_id: "server.udp.send".to_string(), + state_label: "udp", + role: "udp", + phase: "udp", + remote_addr: remote_addr.to_string(), + transport_mode: capture.transport_mode, + obfuscated: capture.obfuscated, + protocol_marker: capture.protocol_marker, + opcode: capture.opcode, + attempted_len: capture.raw.len(), + }; + dump_ed2k_server_udp_packet_with_attempt(server, remote_addr, capture, Some(attempt_id)); + Self { + context, + completed: false, + } + } + fn settle( &mut self, outcome: Ed2kServerSendOutcome, @@ -184,13 +218,20 @@ impl Ed2kServerTcpSendAttempt { } #[cfg(feature = "packet-diagnostics")] -impl Drop for Ed2kServerTcpSendAttempt { +impl Drop for Ed2kServerSendAttempt { fn drop(&mut self) { if !self.completed { self.settle( Ed2kServerSendOutcome::Cancelled, false, - Some("server TCP send future cancelled before completion".to_string()), + Some(format!( + "server {} send future cancelled before completion", + if self.context.role == "udp" { + "UDP" + } else { + "TCP" + } + )), ); } } @@ -313,6 +354,16 @@ pub(super) fn dump_ed2k_server_udp_packet( server: Option<&ResolvedServerEntry>, remote_addr: SocketAddr, capture: Ed2kServerPacketCapture<'_>, +) { + dump_ed2k_server_udp_packet_with_attempt(server, remote_addr, capture, None); +} + +#[cfg(feature = "packet-diagnostics")] +fn dump_ed2k_server_udp_packet_with_attempt( + server: Option<&ResolvedServerEntry>, + remote_addr: SocketAddr, + capture: Ed2kServerPacketCapture<'_>, + send_attempt_id: Option, ) { let direction = match capture.direction { "tx" => "send", @@ -336,7 +387,7 @@ pub(super) fn dump_ed2k_server_udp_packet( phase: "udp", direction, outcome: capture.outcome, - send_attempt_id: None, + send_attempt_id, remote_addr: remote_addr.to_string(), transport_mode: capture.transport_mode, obfuscated: capture.obfuscated, @@ -603,9 +654,9 @@ fn emit_ed2k_server_send_outcome( delivery_known: bool, note: Option<&str>, ) { - let (severity, keys, body) = + let (family, severity, keys, body) = ed2k_server_send_outcome_parts(context, outcome, delivery_known, note); - crate::diag_event::emit("ed2k_tcp", "send_outcome", severity, keys, body); + crate::diag_event::emit(family, "send_outcome", severity, keys, body); } #[cfg(any(feature = "packet-diagnostics", test))] @@ -614,7 +665,7 @@ fn ed2k_server_send_outcome_parts( outcome: Ed2kServerSendOutcome, delivery_known: bool, note: Option<&str>, -) -> (&'static str, Value, Value) { +) -> (&'static str, &'static str, Value, Value) { let mut keys = Map::new(); keys.insert("peer".to_string(), json!(context.remote_addr)); keys.insert("sendAttemptId".to_string(), json!(context.attempt_id)); @@ -657,7 +708,12 @@ fn ed2k_server_send_outcome_parts( body.insert("note".to_string(), json!(note)); } - (outcome.severity(), Value::Object(keys), Value::Object(body)) + ( + context.family, + outcome.severity(), + Value::Object(keys), + Value::Object(body), + ) } #[cfg(feature = "packet-diagnostics")] @@ -717,9 +773,9 @@ mod tests { ED2K_SERVER_DUMP_FILE_PREFIX, Ed2kServerDumpRecord, Ed2kServerSendContext, Ed2kServerSendOutcome, OP_CALLBACKREQUEST, OP_EDONKEYPROT_MARKER, OP_GLOBCALLBACKREQ, OP_GLOBFOUNDSOURCES, OP_GLOBSEARCHREQ, OP_GLOBSEARCHREQ2, OP_GLOBSEARCHREQ3, - OP_GLOBSEARCHRES, OP_PACKEDPROT, OP_SERVER_LIST_REQ, OP_SERVER_LIST_REQ2, - OP_SERVER_LIST_RES, ed2k_server_diag_event_parts, ed2k_server_send_outcome_parts, - exact_packet_hex, server_opcode_name, server_protocol_name, + OP_GLOBSEARCHRES, OP_GLOBSERVSTATREQ, OP_PACKEDPROT, OP_SERVER_LIST_REQ, + OP_SERVER_LIST_REQ2, OP_SERVER_LIST_RES, ed2k_server_diag_event_parts, + ed2k_server_send_outcome_parts, exact_packet_hex, server_opcode_name, server_protocol_name, }; #[test] @@ -733,6 +789,7 @@ mod tests { #[test] fn server_tcp_send_outcome_preserves_attempt_correlation_and_terminal_state() { let context = Ed2kServerSendContext { + family: "ed2k_tcp", attempt_id: 42, trace_id: 7, trace_key: "server:background:7:198.51.100.1:4661".to_string(), @@ -748,8 +805,9 @@ mod tests { attempted_len: 73, }; - let (severity, keys, body) = + let (family, severity, keys, body) = ed2k_server_send_outcome_parts(&context, Ed2kServerSendOutcome::Succeeded, true, None); + assert_eq!(family, "ed2k_tcp"); assert_eq!(severity, "info"); assert_eq!(keys["sendAttemptId"], 42); assert_eq!(body["outcome"], "send_succeeded"); @@ -758,7 +816,7 @@ mod tests { assert_eq!(body["traceKey"], context.trace_key); assert_eq!(body["stateId"], context.state_id); - let (severity, _, body) = ed2k_server_send_outcome_parts( + let (_, severity, _, body) = ed2k_server_send_outcome_parts( &context, Ed2kServerSendOutcome::Cancelled, false, @@ -770,6 +828,42 @@ mod tests { assert_eq!(body["note"], "cancelled"); } + #[test] + fn server_udp_send_outcome_uses_distinct_family_and_server_trace() { + let context = Ed2kServerSendContext { + family: "ed2k_server_udp", + attempt_id: 43, + trace_id: 0, + trace_key: "server:udp:198.51.100.2:4661".to_string(), + state_id: "server.udp.send".to_string(), + state_label: "udp", + role: "udp", + phase: "udp", + remote_addr: "198.51.100.2:4665".to_string(), + transport_mode: "obfuscated", + obfuscated: Some(true), + protocol_marker: Some(OP_EDONKEYPROT_MARKER), + opcode: Some(OP_GLOBSERVSTATREQ), + attempted_len: 12, + }; + + let (family, severity, keys, body) = ed2k_server_send_outcome_parts( + &context, + Ed2kServerSendOutcome::Failed, + true, + Some("send failed"), + ); + assert_eq!(family, "ed2k_server_udp"); + assert_eq!(severity, "warning"); + assert_eq!(keys["peer"], "198.51.100.2:4665"); + assert_eq!(keys["sendAttemptId"], 43); + assert_eq!(body["outcome"], "send_failed"); + assert_eq!(body["traceKey"], "server:udp:198.51.100.2:4661"); + assert_eq!(body["stateId"], "server.udp.send"); + assert_eq!(body["transportMode"], "obfuscated"); + assert_eq!(body["note"], "send failed"); + } + #[test] fn server_common_event_preserves_payloads_larger_than_the_former_cap() { let payload = vec![0xCD; 4 * 1024 + 1]; diff --git a/crates/emulebb-ed2k/src/ed2k_server/session.rs b/crates/emulebb-ed2k/src/ed2k_server/session.rs index 162f12ed..96616ddf 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/session.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/session.rs @@ -23,7 +23,7 @@ use crate::ed2k_tcp::MAX_ED2K_PACKET_LEN; use super::Ed2kServerFailure; #[cfg(feature = "packet-diagnostics")] -use super::diagnostics::{Ed2kServerPacketCapture, Ed2kServerTcpSendAttempt}; +use super::diagnostics::{Ed2kServerPacketCapture, Ed2kServerSendAttempt}; use super::offer_policy::{ OFFER_FILES_LEGACY_MIN_INTERVAL, OFFER_FILES_NEGOTIATION_WAIT, OfferFilesCapabilityAdvertisement, OfferFilesPolicy, @@ -684,7 +684,7 @@ impl ServerSession { cipher.apply(&mut packet); } #[cfg(feature = "packet-diagnostics")] - let send_attempt = Ed2kServerTcpSendAttempt::new( + let send_attempt = Ed2kServerSendAttempt::tcp( self, Ed2kServerPacketCapture { direction: "tx", diff --git a/crates/emulebb-ed2k/src/ed2k_server/udp_runtime.rs b/crates/emulebb-ed2k/src/ed2k_server/udp_runtime.rs index 59c2e45b..37b88476 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/udp_runtime.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/udp_runtime.rs @@ -1,13 +1,13 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; -use anyhow::{Context, Result}; +use anyhow::{Context, Result, anyhow}; use rand::RngExt; use tokio::net::UdpSocket; use emulebb_kad_proto::Ed2kHash; #[cfg(feature = "packet-diagnostics")] -use super::diagnostics::Ed2kServerPacketCapture; +use super::diagnostics::{Ed2kServerPacketCapture, Ed2kServerSendAttempt}; use super::{ Ed2kUdpSourceRequestTarget, OP_EDONKEYPROT, OP_GLOBSERVSTATREQ, OP_SERVER_DESC_REQ, ResolvedServerEntry, ServerUdpPacket, decode_server_udp_datagram, @@ -54,11 +54,11 @@ pub(super) async fn send_server_udp_packet( "obfuscated" }; #[cfg(feature = "packet-diagnostics")] - { + let send_attempt = { let mut clear_frame = Vec::with_capacity(payload.len() + 2); clear_frame.extend_from_slice(&[OP_EDONKEYPROT, opcode]); clear_frame.extend_from_slice(payload); - dump_ed2k_server_udp_packet( + Ed2kServerSendAttempt::udp( Some(server), endpoint, Ed2kServerPacketCapture { @@ -70,20 +70,40 @@ pub(super) async fn send_server_udp_packet( raw: &packet, clear_frame: Some(&clear_frame), payload: Some(payload), - outcome: "accepted", + outcome: "attempted", note: None, }, - ); - } + ) + }; #[cfg(not(feature = "packet-diagnostics"))] dump_ed2k_server_udp_packet(server, "tx", endpoint, transport_mode, opcode, payload); - socket.send_to(&packet, endpoint).await.with_context(|| { + let send_result = socket.send_to(&packet, endpoint).await.with_context(|| { format!( "failed to send ED2K server UDP opcode=0x{opcode:02X} to {}", endpoint ) - })?; - Ok(()) + }); + match send_result { + Ok(sent_len) if sent_len == packet.len() => { + #[cfg(feature = "packet-diagnostics")] + send_attempt.succeeded(); + Ok(()) + } + Ok(sent_len) => { + let error = anyhow!( + "short ED2K server UDP send to {endpoint}: wrote {sent_len} of {} bytes", + packet.len() + ); + #[cfg(feature = "packet-diagnostics")] + send_attempt.failed(error.to_string()); + Err(error) + } + Err(error) => { + #[cfg(feature = "packet-diagnostics")] + send_attempt.failed(format!("{error:#}")); + Err(error) + } + } } /// Request the server's peer list using the client-safe `OP_SERVER_LIST_REQ2` @@ -130,7 +150,7 @@ pub(super) async fn send_server_udp_crypt_status_request( let packet = encode_server_udp_crypt_ping(challenge, &padding); let endpoint = server_udp_crypt_ping_endpoint(server); #[cfg(feature = "packet-diagnostics")] - dump_ed2k_server_udp_packet( + let send_attempt = Ed2kServerSendAttempt::udp( Some(server), endpoint, Ed2kServerPacketCapture { @@ -142,17 +162,37 @@ pub(super) async fn send_server_udp_crypt_status_request( raw: &packet, clear_frame: None, payload: None, - outcome: "probe", + outcome: "attempted", note: Some(format!( "server UDP crypt-ping challenge={challenge} padding_len={padding_len}" )), }, ); - socket + let send_result = socket .send_to(&packet, endpoint) .await - .with_context(|| format!("failed to send ED2K server UDP crypt-ping to {endpoint}"))?; - Ok(challenge) + .with_context(|| format!("failed to send ED2K server UDP crypt-ping to {endpoint}")); + match send_result { + Ok(sent_len) if sent_len == packet.len() => { + #[cfg(feature = "packet-diagnostics")] + send_attempt.succeeded(); + Ok(challenge) + } + Ok(sent_len) => { + let error = anyhow!( + "short ED2K server UDP crypt-ping send to {endpoint}: wrote {sent_len} of {} bytes", + packet.len() + ); + #[cfg(feature = "packet-diagnostics")] + send_attempt.failed(error.to_string()); + Err(error) + } + Err(error) => { + #[cfg(feature = "packet-diagnostics")] + send_attempt.failed(format!("{error:#}")); + Err(error) + } + } } pub(super) async fn send_udp_keyword_search( @@ -387,9 +427,14 @@ fn udp_response_candidate_servers( #[cfg(test)] mod tests { use std::net::Ipv4Addr; + use std::time::Duration; - use super::super::{ConfiguredServerEntry, ResolvedServerEntry}; - use super::udp_response_candidate_servers; + use tokio::{net::UdpSocket, time::timeout}; + + use super::super::{ + ConfiguredServerEntry, OP_EDONKEYPROT, OP_SERVER_DESC_REQ, ResolvedServerEntry, + }; + use super::{send_server_udp_packet, udp_response_candidate_servers}; fn resolved(ip: Ipv4Addr, port: u16) -> ResolvedServerEntry { ResolvedServerEntry { @@ -434,4 +479,27 @@ mod tests { .is_none() ); } + + #[tokio::test] + async fn server_udp_send_reaches_the_resolved_udp_endpoint() { + let receiver = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + let receiver_port = receiver.local_addr().unwrap().port(); + let server = resolved(Ipv4Addr::LOCALHOST, receiver_port.checked_sub(4).unwrap()); + let sender = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + let payload = [0xAA, 0xBB]; + + send_server_udp_packet(&sender, &server, OP_SERVER_DESC_REQ, &payload) + .await + .unwrap(); + + let mut datagram = [0u8; 16]; + let (len, _) = timeout(Duration::from_secs(1), receiver.recv_from(&mut datagram)) + .await + .unwrap() + .unwrap(); + assert_eq!( + &datagram[..len], + &[OP_EDONKEYPROT, OP_SERVER_DESC_REQ, 0xAA, 0xBB] + ); + } } From a6c8ae79bdff90ccad7469e62c85da5841d931a1 Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Sat, 24 Oct 2026 21:55:26 +0200 Subject: [PATCH 4/6] refactor(diagnostics): close encoded server send gaps (RUST-CI-031) --- .../src/ed2k_server/active_callback.rs | 5 +- .../src/ed2k_server/active_keyword.rs | 18 ++--- .../src/ed2k_server/active_source.rs | 5 +- .../emulebb-ed2k/src/ed2k_server/session.rs | 74 ++++++++----------- .../src/ed2k_server/session_driver.rs | 5 +- .../src/ed2k_server/tests/background.rs | 18 ++--- 6 files changed, 49 insertions(+), 76 deletions(-) diff --git a/crates/emulebb-ed2k/src/ed2k_server/active_callback.rs b/crates/emulebb-ed2k/src/ed2k_server/active_callback.rs index 0d83895f..42a90b73 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/active_callback.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/active_callback.rs @@ -18,7 +18,7 @@ use crate::{ use super::packet_handler::{decode_id_change_payload, log_uncorrelated_server_reject}; use super::{ Ed2kServerState, OP_CALLBACK_FAIL, OP_CALLBACKREQUEST, OP_IDCHANGE, OP_LOGINREQUEST, OP_REJECT, - ServerSession, ServerSessionPhase, ServerSessionRole, encode_login_request, encode_packet, + ServerSession, ServerSessionPhase, ServerSessionRole, encode_login_request, login_identity_for_server_transport, resolve_callback_server_entry, send_connected_server_startup, should_use_server_obfuscation, wait_for_offer_files_settle, }; @@ -68,9 +68,8 @@ pub async fn request_callback_on_server(options: Ed2kCallbackRequestOptions<'_>) let login_payload = encode_login_request(login_identity, &nickname, ServerSessionRole::Callback); if use_server_obfuscation { - let login_request = encode_packet(OP_LOGINREQUEST, &login_payload, false)?; session - .negotiate_obfuscation_and_send(&login_request) + .negotiate_obfuscation_and_send(OP_LOGINREQUEST, &login_payload) .await?; } else { session.send_packet(OP_LOGINREQUEST, &login_payload).await?; diff --git a/crates/emulebb-ed2k/src/ed2k_server/active_keyword.rs b/crates/emulebb-ed2k/src/ed2k_server/active_keyword.rs index 9f5b2c0f..849c57b6 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/active_keyword.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/active_keyword.rs @@ -22,7 +22,7 @@ use super::{ OP_LOGINREQUEST, OP_QUERY_MORE_RESULT, OP_REJECT, OP_SEARCHREQUEST, OP_SEARCHRESULT, ResolvedServerEntry, SERVER_UDP_FLAG_LARGEFILES, SearchCriteria, ServerSession, ServerSessionPhase, ServerSessionRole, bind_server_udp_socket, configured_server_entries, - decode_search_result_page, decode_udp_search_result_pages, encode_login_request, encode_packet, + decode_search_result_page, decode_udp_search_result_pages, encode_login_request, encode_search_request, encode_search_request_with_criteria, login_identity_for_server_transport, read_server_udp_packet_from_any, resolve_server_entry, retain_live_servers, send_connected_server_startup, send_udp_keyword_search, @@ -454,14 +454,11 @@ async fn search_keyword_on_server( }, search_payload.len() ); - let login_request = encode_packet( - OP_LOGINREQUEST, - &encode_login_request(login_identity, nickname, ServerSessionRole::KeywordSearch), - false, - )?; + let login_payload = + encode_login_request(login_identity, nickname, ServerSessionRole::KeywordSearch); if use_server_obfuscation { session - .negotiate_obfuscation_and_send(&login_request) + .negotiate_obfuscation_and_send(OP_LOGINREQUEST, &login_payload) .await .with_context(|| { format!( @@ -470,12 +467,7 @@ async fn search_keyword_on_server( ) })?; } else { - session - .send_encoded_packet( - &login_request, - format!("failed to send ED2K server login request to {transport_endpoint}"), - ) - .await?; + session.send_packet(OP_LOGINREQUEST, &login_payload).await?; } session.set_phase( ServerSessionPhase::AwaitingIdChange, diff --git a/crates/emulebb-ed2k/src/ed2k_server/active_source.rs b/crates/emulebb-ed2k/src/ed2k_server/active_source.rs index d3265ffd..751e49f8 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/active_source.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/active_source.rs @@ -21,7 +21,7 @@ use super::{ Ed2kFoundSource, Ed2kServerState, OP_FOUNDSOURCES, OP_FOUNDSOURCES_OBFU, OP_IDCHANGE, OP_LOGINREQUEST, OP_REJECT, ResolvedServerEntry, ServerSession, ServerSessionPhase, ServerSessionRole, annotate_found_sources_server, configured_server_entries, - decode_found_sources, encode_login_request, encode_packet, encode_source_request, + decode_found_sources, encode_login_request, encode_source_request, login_identity_for_server_transport, merge_found_sources, resolve_server_entry, send_connected_server_startup, should_use_server_obfuscation, source_request_opcode, validate_found_sources, @@ -210,10 +210,9 @@ async fn search_sources_on_server( .await?; let login_payload = encode_login_request(login_identity, nickname, ServerSessionRole::SourceSearch); - let login_request = encode_packet(OP_LOGINREQUEST, &login_payload, false)?; if use_server_obfuscation { session - .negotiate_obfuscation_and_send(&login_request) + .negotiate_obfuscation_and_send(OP_LOGINREQUEST, &login_payload) .await .with_context(|| { format!( diff --git a/crates/emulebb-ed2k/src/ed2k_server/session.rs b/crates/emulebb-ed2k/src/ed2k_server/session.rs index 96616ddf..821b1d16 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/session.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/session.rs @@ -720,27 +720,16 @@ impl ServerSession { } } - pub(super) async fn send_encoded_packet( - &mut self, - packet: &[u8], - context: impl Into, - ) -> Result<()> { - self.stream - .write_all(packet) - .await - .with_context(|| context.into())?; - self.last_tx = Instant::now(); - Ok(()) - } - fn server_supports_compression(&self) -> bool { self.server_flags.unwrap_or_default() & SERVER_TCP_FLAG_COMPRESSION != 0 } pub(super) async fn negotiate_obfuscation_and_send( &mut self, - first_packet: &[u8], + opcode: u8, + payload: &[u8], ) -> Result<()> { + let first_packet = encode_packet(opcode, payload, false)?; let prime = BigUint::from_bytes_be(&SERVER_OBFUSCATION_PRIME_BYTES); let generator = BigUint::from(2u8); let secret = random_nonzero_biguint(SERVER_OBFUSCATION_RANDOM_EXPONENT_LEN); @@ -837,40 +826,39 @@ impl ServerSession { response.extend_from_slice(&response_padding); #[cfg(feature = "packet-diagnostics")] let first_packet_offset = response.len(); - response.extend_from_slice(first_packet); + response.extend_from_slice(&first_packet); send_cipher.apply(&mut response); - if first_packet.len() >= TCP_PACKET_HEADER_LEN { - #[cfg(feature = "packet-diagnostics")] - dump_ed2k_server_packet( - self, - Ed2kServerPacketCapture { - direction: "tx", - transport_mode: "obfuscated", - obfuscated: Some(true), - protocol_marker: Some(first_packet[0]), - opcode: Some(first_packet[5]), - raw: &response[first_packet_offset..], - clear_frame: Some(first_packet), - payload: Some(&first_packet[TCP_PACKET_HEADER_LEN..]), - outcome: "accepted", - note: None, - }, - ); - #[cfg(not(feature = "packet-diagnostics"))] - dump_ed2k_server_packet( - self, - "tx", - first_packet[0], - first_packet[5], - &first_packet[TCP_PACKET_HEADER_LEN..], - ); - } - self.stream.write_all(&response).await.with_context(|| { + #[cfg(feature = "packet-diagnostics")] + let send_attempt = Ed2kServerSendAttempt::tcp( + self, + Ed2kServerPacketCapture { + direction: "tx", + transport_mode: "obfuscated", + obfuscated: Some(true), + protocol_marker: Some(first_packet[0]), + opcode: Some(opcode), + raw: &response[first_packet_offset..], + clear_frame: Some(&first_packet), + payload: Some(payload), + outcome: "attempted", + note: None, + }, + ); + #[cfg(not(feature = "packet-diagnostics"))] + dump_ed2k_server_packet(self, "tx", first_packet[0], opcode, payload); + let send_result = self.stream.write_all(&response).await.with_context(|| { format!( "failed to send ED2K server obfuscation response to {}", self.endpoint ) - })?; + }); + if let Err(error) = send_result { + #[cfg(feature = "packet-diagnostics")] + send_attempt.failed(format!("{error:#}")); + return Err(error); + } + #[cfg(feature = "packet-diagnostics")] + send_attempt.succeeded(); self.receive_cipher = Some(receive_cipher); self.send_cipher = Some(send_cipher); diff --git a/crates/emulebb-ed2k/src/ed2k_server/session_driver.rs b/crates/emulebb-ed2k/src/ed2k_server/session_driver.rs index 09df41ff..41f02162 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/session_driver.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/session_driver.rs @@ -20,7 +20,7 @@ use super::{ OP_FOUNDSOURCES_OBFU, OP_LOGINREQUEST, OP_OFFERFILES, OP_QUERY_MORE_RESULT, OP_SEARCHRESULT, PendingBackgroundServerSearch, ResolvedServerEntry, ServerSession, ServerSessionPhase, ServerSessionRole, ServerTransportMode, annotate_found_sources_server, decode_found_sources, - decode_search_result_page, dump_ed2k_server_loop_meta, encode_login_request, encode_packet, + decode_search_result_page, dump_ed2k_server_loop_meta, encode_login_request, fail_background_search_request, fail_pending_background_search, format_connect_options, handle_background_udp_packet, handle_server_packet, log_search_result_page, login_identity_for_server_transport, run_server_transport_attempts, @@ -164,10 +164,9 @@ async fn connect_server_transport_attempt( transport_endpoint.port(), ); if use_server_obfuscation { - let login_request = encode_packet(OP_LOGINREQUEST, &login_payload, false)?; match tokio::time::timeout_at( TokioInstant::from_std(pending_login_deadline), - session.negotiate_obfuscation_and_send(&login_request), + session.negotiate_obfuscation_and_send(OP_LOGINREQUEST, &login_payload), ) .await { diff --git a/crates/emulebb-ed2k/src/ed2k_server/tests/background.rs b/crates/emulebb-ed2k/src/ed2k_server/tests/background.rs index 44f9143b..f258a2a6 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/tests/background.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/tests/background.rs @@ -769,16 +769,12 @@ async fn server_obfuscation_handshake_encrypts_login_request() { connect_options: emule_connect_options(true), direct_udp_callback: false, }; - let expected_login = encode_packet( - OP_LOGINREQUEST, - &encode_login_request( - hello_identity, - HELLO_NICKNAME, - ServerSessionRole::KeywordSearch, - ), - false, - ) - .unwrap(); + let login_payload = encode_login_request( + hello_identity, + HELLO_NICKNAME, + ServerSessionRole::KeywordSearch, + ); + let expected_login = encode_packet(OP_LOGINREQUEST, &login_payload, false).unwrap(); let expected_login_for_server = expected_login.clone(); let server = tokio::spawn(async move { @@ -860,7 +856,7 @@ async fn server_obfuscation_handshake_encrypts_login_request() { .await .unwrap(); session - .negotiate_obfuscation_and_send(&expected_login) + .negotiate_obfuscation_and_send(OP_LOGINREQUEST, &login_payload) .await .unwrap(); From 0a58c81510774d0d471fe6087760d8cc6b7e62f6 Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Sat, 24 Oct 2026 21:59:03 +0200 Subject: [PATCH 5/6] refactor(diagnostics): retire server packet side file (RUST-CI-031) --- .../src/ed2k_server/diagnostics.rs | 169 +++--------------- 1 file changed, 25 insertions(+), 144 deletions(-) diff --git a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs index 6f66266b..787941c3 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs @@ -1,28 +1,14 @@ -// Gated by the `packet-diagnostics` Cargo feature (matches the client dump and -// eMuleBB's EMULEBB_ENABLE_PACKET_DIAGNOSTICS): off-by-default builds compile the -// server dump out (emitters become no-ops, record machinery is dead-eliminated). -#![cfg_attr( - not(feature = "packet-diagnostics"), - expect( - dead_code, - unused_imports, - reason = "server diagnostic machinery is inert without its feature" - ) -)] +// Gated by the `packet-diagnostics` Cargo feature: off-by-default builds keep +// only no-op call boundaries and compile all unified diagnostic machinery out. +use std::net::SocketAddr; -use std::{ - fs, - net::SocketAddr, - sync::{ - Mutex as StdMutex, OnceLock, - atomic::{AtomicU64, Ordering}, - }, -}; +#[cfg(feature = "packet-diagnostics")] +use std::sync::atomic::{AtomicU64, Ordering}; -use chrono::SecondsFormat; -use serde::Serialize; +#[cfg(any(feature = "packet-diagnostics", test))] use serde_json::{Map, Value, json}; +#[cfg(any(feature = "packet-diagnostics", test))] use super::{ OP_CALLBACK_FAIL, OP_CALLBACKREQUEST, OP_CALLBACKREQUESTED, OP_FOUNDSOURCES, OP_FOUNDSOURCES_OBFU, OP_GETSERVERLIST, OP_GETSOURCES, OP_GETSOURCES_OBFU, OP_GLOBCALLBACKREQ, @@ -31,23 +17,20 @@ use super::{ OP_IDCHANGE, OP_LOGINREQUEST, OP_OFFERFILES, OP_PACKEDPROT, OP_QUERY_MORE_RESULT, OP_REJECT, OP_SEARCHREQUEST, OP_SEARCHRESULT, OP_SERVER_DESC_REQ, OP_SERVER_DESC_RES, OP_SERVER_LIST_REQ, OP_SERVER_LIST_REQ2, OP_SERVER_LIST_RES, OP_SERVERIDENT, OP_SERVERLIST, OP_SERVERMESSAGE, - OP_SERVERSTATUS, ResolvedServerEntry, ServerSession, + OP_SERVERSTATUS, }; - -const ED2K_SERVER_DUMP_FILE_PREFIX: &str = "emulebb-rust-ed2k-server-dump-"; +use super::{ResolvedServerEntry, ServerSession}; /// eD2k server protocol marker (OP_EDONKEYPROT) — server packets ride the eD2k /// protocol byte unless the server advertised packed payload support. +#[cfg(any(feature = "packet-diagnostics", test))] const OP_EDONKEYPROT_MARKER: u8 = 0xE3; /// One server-flow packet in the converged `ed2k_packet_v1` shape (flow="server"), /// so client<->server traces diff 1:1 against eMuleBB's server-packet emitter. /// Carries server-session extras (role/trace_id) the diff harness ignores. -#[derive(Debug, Serialize)] +#[cfg(any(feature = "packet-diagnostics", test))] +#[derive(Debug)] struct Ed2kServerDumpRecord<'a> { - schema: &'static str, - source: &'static str, - ts_utc: String, - event_seq: u64, trace_id: u64, trace_key: String, state_id: String, @@ -251,10 +234,6 @@ pub(super) fn dump_ed2k_server_loop_meta( #[cfg(feature = "packet-diagnostics")] pub(super) fn dump_ed2k_server_meta(session: &ServerSession, note: impl Into) { let record = Ed2kServerDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: next_ed2k_server_dump_event_seq(), trace_id: session.trace_id, trace_key: ed2k_server_trace_key(session), state_id: ed2k_server_state_id(session), @@ -285,7 +264,7 @@ pub(super) fn dump_ed2k_server_meta(session: &ServerSession, note: impl Into, ) { let record = Ed2kServerDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: next_ed2k_server_dump_event_seq(), trace_id: 0, trace_key: format!("server:background:loop:{endpoint}"), state_id: format!("server.background.{phase}"), @@ -325,7 +300,7 @@ pub(super) fn dump_ed2k_server_loop_meta( payload_hex_truncated: None, note: Some(note.into()), }; - dump_ed2k_server_record(&record); + emit_ed2k_server_diag_event(&record); } #[cfg(not(feature = "packet-diagnostics"))] @@ -374,10 +349,6 @@ fn dump_ed2k_server_udp_packet_with_attempt( .map(ResolvedServerEntry::base_endpoint) .unwrap_or(remote_addr); let record = Ed2kServerDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: next_ed2k_server_dump_event_seq(), trace_id: 0, trace_key: format!("server:udp:{base_endpoint}"), state_id: format!("server.udp.{direction}"), @@ -404,7 +375,7 @@ fn dump_ed2k_server_udp_packet_with_attempt( payload_hex_truncated: capture.payload.map(|_| false), note: capture.note, }; - dump_ed2k_server_record(&record); + emit_ed2k_server_diag_event(&record); } #[cfg(feature = "packet-diagnostics")] @@ -429,10 +400,6 @@ fn dump_ed2k_server_packet_with_attempt( other => other, }; let record = Ed2kServerDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: next_ed2k_server_dump_event_seq(), trace_id: session.trace_id, trace_key: ed2k_server_trace_key(session), state_id: ed2k_server_state_id(session), @@ -459,9 +426,10 @@ fn dump_ed2k_server_packet_with_attempt( payload_hex_truncated: capture.payload.map(|_| false), note: capture.note, }; - dump_ed2k_server_record(&record); + emit_ed2k_server_diag_event(&record); } +#[cfg(any(feature = "packet-diagnostics", test))] fn server_protocol_name(protocol_marker: u8) -> &'static str { match protocol_marker { OP_EDONKEYPROT_MARKER => "ed2k", @@ -470,21 +438,18 @@ fn server_protocol_name(protocol_marker: u8) -> &'static str { } } +#[cfg(any(feature = "packet-diagnostics", test))] fn exact_packet_hex(bytes: &[u8]) -> String { hex::encode(bytes) } -fn next_ed2k_server_dump_event_seq() -> u64 { - static NEXT_EVENT_SEQ: AtomicU64 = AtomicU64::new(1); - NEXT_EVENT_SEQ.fetch_add(1, Ordering::Relaxed) -} - #[cfg(feature = "packet-diagnostics")] fn next_ed2k_server_send_attempt_id() -> u64 { static NEXT_SEND_ATTEMPT_ID: AtomicU64 = AtomicU64::new(1); NEXT_SEND_ATTEMPT_ID.fetch_add(1, Ordering::Relaxed) } +#[cfg(feature = "packet-diagnostics")] fn ed2k_server_trace_key(session: &ServerSession) -> String { format!( "server:{}:{}:{}", @@ -492,60 +457,11 @@ fn ed2k_server_trace_key(session: &ServerSession) -> String { ) } +#[cfg(feature = "packet-diagnostics")] fn ed2k_server_state_id(session: &ServerSession) -> String { format!("server.{}.{}", session.trace_role, session.phase.as_str()) } -/// Open (create/append) the ed2k server dump file under `EMULEBB_RUST_LOG_DIR`. -/// `None` when the env var is unset or the file cannot be created. Kept separate -/// from the `OnceLock` cell so callers RE-ATTEMPT the open while the handle is -/// still `None` (an early first-access must not permanently disable the dump). -fn open_ed2k_server_dump_file() -> Option { - let dir = std::env::var("EMULEBB_RUST_LOG_DIR") - .ok() - .map(std::path::PathBuf::from)?; - fs::create_dir_all(&dir).ok()?; - let path = dir.join(format!( - "{}{}.jsonl", - ED2K_SERVER_DUMP_FILE_PREFIX, - chrono::Utc::now().format("%Y.%m.%d-%H.%M.%S") - )); - fs::OpenOptions::new() - .create(true) - .append(true) - .open(path) - .ok() -} - -fn ed2k_server_dump_file() -> &'static StdMutex> { - static DUMP_FILE: OnceLock>> = OnceLock::new(); - DUMP_FILE.get_or_init(|| StdMutex::new(None)) -} - -// WHY: gated like its (all feature-on) callers because it calls the gated -// `maybe_sync_ed2k_server_dump`; leaving it unconditional broke every -// feature-off build with an unresolved-name error. -#[cfg(feature = "packet-diagnostics")] -fn dump_ed2k_server_record(record: &Ed2kServerDumpRecord<'_>) { - emit_ed2k_server_diag_event(record); - let Ok(line) = serde_json::to_string(record) else { - return; - }; - let Ok(mut guard) = ed2k_server_dump_file().lock() else { - return; - }; - if guard.is_none() { - *guard = open_ed2k_server_dump_file(); - } - let Some(file) = guard.as_mut() else { - return; - }; - let _ = std::io::Write::write_all(file, line.as_bytes()); - let _ = std::io::Write::write_all(file, b"\n"); - let _ = std::io::Write::flush(file); - maybe_sync_ed2k_server_dump(file); -} - #[cfg(feature = "packet-diagnostics")] fn emit_ed2k_server_diag_event(record: &Ed2kServerDumpRecord<'_>) { let (family, event, severity, keys, body) = ed2k_server_diag_event_parts(record); @@ -716,17 +632,7 @@ fn ed2k_server_send_outcome_parts( ) } -#[cfg(feature = "packet-diagnostics")] -fn maybe_sync_ed2k_server_dump(file: &fs::File) { - static SINCE_SYNC: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); - if SINCE_SYNC - .fetch_add(1, std::sync::atomic::Ordering::Relaxed) - .is_multiple_of(128) - { - let _ = file.sync_data(); - } -} - +#[cfg(any(feature = "packet-diagnostics", test))] fn server_opcode_name(opcode: u8) -> &'static str { match opcode { OP_LOGINREQUEST => "OP_LOGINREQUEST", @@ -770,22 +676,13 @@ fn server_opcode_name(opcode: u8) -> &'static str { #[cfg(test)] mod tests { use super::{ - ED2K_SERVER_DUMP_FILE_PREFIX, Ed2kServerDumpRecord, Ed2kServerSendContext, - Ed2kServerSendOutcome, OP_CALLBACKREQUEST, OP_EDONKEYPROT_MARKER, OP_GLOBCALLBACKREQ, - OP_GLOBFOUNDSOURCES, OP_GLOBSEARCHREQ, OP_GLOBSEARCHREQ2, OP_GLOBSEARCHREQ3, - OP_GLOBSEARCHRES, OP_GLOBSERVSTATREQ, OP_PACKEDPROT, OP_SERVER_LIST_REQ, - OP_SERVER_LIST_REQ2, OP_SERVER_LIST_RES, ed2k_server_diag_event_parts, + Ed2kServerDumpRecord, Ed2kServerSendContext, Ed2kServerSendOutcome, OP_CALLBACKREQUEST, + OP_EDONKEYPROT_MARKER, OP_GLOBCALLBACKREQ, OP_GLOBFOUNDSOURCES, OP_GLOBSEARCHREQ, + OP_GLOBSEARCHREQ2, OP_GLOBSEARCHREQ3, OP_GLOBSEARCHRES, OP_GLOBSERVSTATREQ, OP_PACKEDPROT, + OP_SERVER_LIST_REQ, OP_SERVER_LIST_REQ2, OP_SERVER_LIST_RES, ed2k_server_diag_event_parts, ed2k_server_send_outcome_parts, exact_packet_hex, server_opcode_name, server_protocol_name, }; - #[test] - fn server_dump_prefix_uses_emulebb_rust_name() { - assert_eq!( - ED2K_SERVER_DUMP_FILE_PREFIX, - "emulebb-rust-ed2k-server-dump-" - ); - } - #[test] fn server_tcp_send_outcome_preserves_attempt_correlation_and_terminal_state() { let context = Ed2kServerSendContext { @@ -870,10 +767,6 @@ mod tests { let raw = vec![0xAB; payload.len() + 6]; let clear_frame = vec![0xBC; payload.len() + 6]; let record = Ed2kServerDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: "2026-10-24T00:00:00.000Z".to_string(), - event_seq: 1, trace_id: 7, trace_key: "server:background:7:198.51.100.1:4661".to_string(), state_id: "server.background.connected".to_string(), @@ -931,10 +824,6 @@ mod tests { #[test] fn server_udp_common_event_uses_a_distinct_family() { let record = Ed2kServerDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: "2026-10-24T00:00:00.000Z".to_string(), - event_seq: 1, trace_id: 0, trace_key: "server:udp:198.51.100.2:4661".to_string(), state_id: "server.udp.send".to_string(), @@ -972,10 +861,6 @@ mod tests { #[test] fn server_udp_drop_keeps_raw_bytes_without_faking_decoded_stages() { let record = Ed2kServerDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: "2026-10-24T00:00:00.000Z".to_string(), - event_seq: 1, trace_id: 0, trace_key: "server:udp:198.51.100.2:4665".to_string(), state_id: "server.udp.recv".to_string(), @@ -1016,10 +901,6 @@ mod tests { #[test] fn server_meta_record_is_a_state_event_with_correlation() { let record = Ed2kServerDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: "2026-10-24T00:00:00.000Z".to_string(), - event_seq: 1, trace_id: 11, trace_key: "server:background:11:198.51.100.3:4661".to_string(), state_id: "server.background.connected".to_string(), From 2429477943fd69ca042c481c431736a9fe6b435a Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Sat, 24 Oct 2026 22:05:35 +0200 Subject: [PATCH 6/6] fix(diagnostics): settle obfuscated send outcomes explicitly (RUST-CI-031) --- crates/emulebb-ed2k/src/ed2k_server/session.rs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/crates/emulebb-ed2k/src/ed2k_server/session.rs b/crates/emulebb-ed2k/src/ed2k_server/session.rs index 821b1d16..2236d4e4 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/session.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/session.rs @@ -852,13 +852,17 @@ impl ServerSession { self.endpoint ) }); - if let Err(error) = send_result { - #[cfg(feature = "packet-diagnostics")] - send_attempt.failed(format!("{error:#}")); - return Err(error); + match send_result { + Ok(()) => { + #[cfg(feature = "packet-diagnostics")] + send_attempt.succeeded(); + } + Err(error) => { + #[cfg(feature = "packet-diagnostics")] + send_attempt.failed(format!("{error:#}")); + return Err(error); + } } - #[cfg(feature = "packet-diagnostics")] - send_attempt.succeeded(); self.receive_cipher = Some(receive_cipher); self.send_cipher = Some(send_cipher);