diff --git a/crates/emulebb-ed2k/Cargo.toml b/crates/emulebb-ed2k/Cargo.toml index 177ab93f..37353990 100644 --- a/crates/emulebb-ed2k/Cargo.toml +++ b/crates/emulebb-ed2k/Cargo.toml @@ -11,10 +11,10 @@ description = "ED2K transfer runtime extracted for emulebb-rust" workspace = true [features] -# Compile-time gate for the ed2k_packet_v1 packet dump (the Cargo equivalent of +# Compile-time gate for exact protocol diagnostics (the Cargo equivalent of # eMuleBB's EMULEBB_ENABLE_PACKET_DIAGNOSTICS #ifdef). Off by default so release -# builds compile the dump out entirely; enable for diagnostic/live-wire builds. -# When enabled, the dump is still runtime-gated by the EMULEBB_RUST_LOG_DIR env var. +# builds compile capture out entirely; enable for diagnostic/live-wire builds. +# When enabled, emission is still runtime-gated by the EMULEBB_RUST_LOG_DIR env var. packet-diagnostics = [ "emulebb-diagnostics/enabled", "emulebb-kad-dht/packet-diagnostics", diff --git a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs index 787941c3..2b36680f 100644 --- a/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs +++ b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs @@ -25,9 +25,10 @@ use super::{ResolvedServerEntry, ServerSession}; /// 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. +/// One server-flow diagnostic record, normalized into the converged +/// `diag_event_v2` envelope so client<->server traces diff 1:1 against eMuleBB's +/// server-packet emitter. Carries server-session extras (role/trace_id) the diff +/// harness ignores. #[cfg(any(feature = "packet-diagnostics", test))] #[derive(Debug)] struct Ed2kServerDumpRecord<'a> { diff --git a/crates/emulebb-ed2k/src/ed2k_tcp/diag_event.rs b/crates/emulebb-ed2k/src/ed2k_tcp/diag_event.rs index 8e3aabd1..bea5a06f 100644 --- a/crates/emulebb-ed2k/src/ed2k_tcp/diag_event.rs +++ b/crates/emulebb-ed2k/src/ed2k_tcp/diag_event.rs @@ -1,12 +1,9 @@ //! `family:"ed2k_tcp"` `diag_event_v2` mapping. //! -//! Re-shapes an [`Ed2kTcpDumpRecord`] (the legacy `ed2k_packet_v1` record) into -//! the converged `ed2k_tcp` `diag_event_v2` envelope and forwards -//! it to the shared writer. Kept in its own module because diagnostics-envelope -//! mapping is a separate responsibility from packet capture. Both writes are -//! behind the `packet-diagnostics` feature: the only caller -//! (`dump.rs::dump_ed2k_tcp_record`) is invoked solely from the feature-gated -//! send/recv/meta builders. +//! Maps an [`Ed2kTcpDiagnosticRecord`] into the converged `ed2k_tcp` +//! `diag_event_v2` envelope and forwards it to the shared writer. Kept in its +//! own module because diagnostics-envelope mapping is a separate responsibility +//! from packet capture. #![cfg_attr( not(feature = "packet-diagnostics"), @@ -18,13 +15,19 @@ use serde_json::{Map, Value, json}; -use super::dump::Ed2kTcpDumpRecord; +use super::dump::Ed2kTcpDiagnosticRecord; -/// Emit the converged `ed2k_tcp` `diag_event_v2` record from the same data that -/// built the legacy `ed2k_packet_v1` record. Optional `keys`/`body` fields are -/// omitted when absent rather than faked (`peerHash`/`fileHash` are not known at -/// this layer, so they are not emitted). -pub(super) fn emit_ed2k_tcp_diag_event(record: &Ed2kTcpDumpRecord<'_>) { +/// Emit one converged `ed2k_tcp` `diag_event_v2` record. Optional `keys`/`body` +/// fields are omitted when absent rather than faked (`peerHash`/`fileHash` are +/// not known at this layer, so they are not emitted). +pub(super) fn emit_ed2k_tcp_diag_event(record: &Ed2kTcpDiagnosticRecord<'_>) { + let (event, severity, keys, body) = ed2k_tcp_diag_event_parts(record); + crate::diag_event::emit("ed2k_tcp", event, severity, keys, body); +} + +fn ed2k_tcp_diag_event_parts( + record: &Ed2kTcpDiagnosticRecord<'_>, +) -> (&'static str, &'static str, Value, Value) { let mut keys = Map::new(); keys.insert("peer".to_string(), json!(record.remote_addr)); if let Some(protocol_marker) = record.protocol_marker { @@ -82,6 +85,9 @@ pub(super) fn emit_ed2k_tcp_diag_event(record: &Ed2kTcpDumpRecord<'_>) { body.insert("transportMode".to_string(), json!(record.transport_mode)); body.insert("flow".to_string(), json!(record.flow)); body.insert("phase".to_string(), json!(record.phase)); + 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)); } @@ -94,16 +100,71 @@ pub(super) fn emit_ed2k_tcp_diag_event(record: &Ed2kTcpDumpRecord<'_>) { } else { "info" }; - let event = if record.schema == "ed2k_handshake_v1" { - "obfuscation_handshake" - } else { - "packet" - }; - crate::diag_event::emit( - "ed2k_tcp", - event, + ( + record.event, severity, Value::Object(keys), Value::Object(body), - ); + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn record<'a>( + event: &'static str, + direction: &'a str, + outcome: &'a str, + ) -> Ed2kTcpDiagnosticRecord<'a> { + Ed2kTcpDiagnosticRecord { + event, + trace_key: "native_download:198.51.100.2:4662".to_string(), + state_id: "native_download.session".to_string(), + state_label: "session", + flow: "native_download", + phase: "session", + direction, + outcome, + remote_addr: "198.51.100.2:4662".to_string(), + transport_mode: "obfuscated", + obfuscated: Some(true), + protocol: None, + protocol_marker: None, + opcode: None, + opcode_name: None, + raw_len: None, + raw_hex: None, + clear_frame_len: None, + clear_frame_hex: None, + attempted_len: None, + written_len: None, + payload_len: None, + payload_hex: None, + payload_hex_truncated: None, + note: None, + } + } + + #[test] + fn peer_tcp_state_event_preserves_trace_context() { + let record = record("state", "meta", "meta"); + let (event, severity, keys, body) = ed2k_tcp_diag_event_parts(&record); + + assert_eq!(event, "state"); + assert_eq!(severity, "info"); + assert_eq!(keys["peer"], "198.51.100.2:4662"); + assert_eq!(body["traceKey"], "native_download:198.51.100.2:4662"); + assert_eq!(body["stateId"], "native_download.session"); + assert_eq!(body["stateLabel"], "session"); + } + + #[test] + fn peer_tcp_handshake_keeps_distinct_event_classification() { + let record = record("obfuscation_handshake", "send", "send_succeeded"); + let (event, _, _, body) = ed2k_tcp_diag_event_parts(&record); + + assert_eq!(event, "obfuscation_handshake"); + assert_eq!(body["traceKey"], "native_download:198.51.100.2:4662"); + } } diff --git a/crates/emulebb-ed2k/src/ed2k_tcp/dump.rs b/crates/emulebb-ed2k/src/ed2k_tcp/dump.rs index e68643fa..d60aef40 100644 --- a/crates/emulebb-ed2k/src/ed2k_tcp/dump.rs +++ b/crates/emulebb-ed2k/src/ed2k_tcp/dump.rs @@ -1,7 +1,6 @@ -// The packet dump is gated by the `packet-diagnostics` Cargo feature (the -// equivalent of eMuleBB's EMULEBB_ENABLE_PACKET_DIAGNOSTICS #ifdef). When the -// feature is off the public dump wrappers below become no-ops and the record -// machinery is dead-code-eliminated, so release builds carry zero dump cost. +// Exact peer diagnostics are gated by the `packet-diagnostics` Cargo feature. +// When the feature is off the public wrappers below become no-ops and the +// record machinery is dead-code-eliminated, so release builds carry zero cost. #![cfg_attr( not(feature = "packet-diagnostics"), expect( @@ -13,18 +12,10 @@ use std::{ borrow::Cow, - fs, - io::Write, net::SocketAddr, - sync::{ - Mutex as StdMutex, OnceLock, - atomic::{AtomicU64, Ordering}, - }, + sync::atomic::{AtomicU64, Ordering}, }; -use chrono::SecondsFormat; -use serde::Serialize; - use super::{ Ed2kTransportMode, EmuleTcpPacket, OP_ACCEPTUPLOADREQ, OP_AICHANSWER, OP_AICHFILEHASHANS, OP_AICHFILEHASHREQ, OP_AICHREQUEST, OP_ANSWERSOURCES, OP_ANSWERSOURCES2, OP_ASKSHAREDDENIEDANS, @@ -44,18 +35,12 @@ use super::{ OP_STARTUPLOADREQ, TCP_PACKET_HEADER_LEN, }; -const ED2K_TCP_DUMP_FILE_PREFIX: &str = "emulebb-rust-ed2k-tcp-dump-"; -const ED2K_TCP_LIVE_SYNC_EVERY: u64 = 128; - -#[derive(Debug, Serialize)] -pub(super) struct Ed2kTcpDumpRecord<'a> { - pub(super) schema: &'static str, - source: &'static str, - ts_utc: String, - event_seq: u64, - trace_key: String, - state_id: String, - state_label: &'a str, +#[derive(Debug)] +pub(super) struct Ed2kTcpDiagnosticRecord<'a> { + pub(super) event: &'static str, + pub(super) trace_key: String, + pub(super) state_id: String, + pub(super) state_label: &'a str, pub(super) flow: &'static str, pub(super) phase: &'a str, pub(super) direction: &'a str, @@ -83,11 +68,6 @@ fn exact_packet_hex(bytes: &[u8]) -> String { hex::encode(bytes) } -fn ed2k_tcp_dump_event_seq() -> u64 { - static NEXT_EVENT_SEQ: AtomicU64 = AtomicU64::new(1); - NEXT_EVENT_SEQ.fetch_add(1, Ordering::Relaxed) -} - fn ed2k_tcp_trace_key(flow: &'static str, remote_addr: SocketAddr) -> String { format!("{flow}:{remote_addr}") } @@ -96,36 +76,6 @@ fn ed2k_tcp_state_id(flow: &'static str, phase: &str) -> String { format!("{flow}.{phase}") } -/// Open (create/append) the ed2k TCP dump file under `EMULEBB_RUST_LOG_DIR`. -/// Returns `None` when the env var is unset or the file cannot be created. -/// -/// Kept separate from the `OnceLock` cell so callers can RE-ATTEMPT the open -/// while the handle is still `None`: the previous design opened inside -/// `get_or_init`, so a single early first-access (before the daemon made -/// `EMULEBB_RUST_LOG_DIR` visible) permanently cached `None` and silently -/// disabled the dump for the whole run. -fn open_ed2k_tcp_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_TCP_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_tcp_dump_file() -> &'static StdMutex> { - static DUMP_FILE: OnceLock>> = OnceLock::new(); - DUMP_FILE.get_or_init(|| StdMutex::new(None)) -} - fn ed2k_protocol_name(protocol: u8) -> &'static str { match protocol { OP_EDONKEYPROT => "ed2k", @@ -331,42 +281,10 @@ fn should_skip_bulk_packet(protocol: u8, opcode: u8, from_recv: bool) -> bool { !(n < HEAD || n % EVERY == 0) } -fn dump_ed2k_tcp_record(record: &Ed2kTcpDumpRecord<'_>) { - if let Ok(line) = serde_json::to_string(record) - && let Ok(mut guard) = ed2k_tcp_dump_file().lock() - { - // Open on the first record and re-attempt whenever the handle is still - // None, so an early first-access (before EMULEBB_RUST_LOG_DIR is visible) - // does not permanently disable the dump for the run. - if guard.is_none() { - *guard = open_ed2k_tcp_dump_file(); - } - if let Some(file) = guard.as_mut() { - let _ = writeln!(file, "{line}"); - let _ = file.flush(); - maybe_sync_ed2k_tcp_dump(file); - } - } - - // uniform-diagnostics-v2 (lane D2): also emit the converged `ed2k_tcp` - // `diag_event_v2` record from the SAME data. The legacy - // `ed2k_packet_v1` line above is kept during migration. Both writes are - // already behind the `packet-diagnostics` feature (the only callers of this - // fn are the feature-gated send/recv/meta builders). The mapping itself lives - // in the sibling `diag_event` module to keep this file within budget. +fn emit_ed2k_tcp_record(record: &Ed2kTcpDiagnosticRecord<'_>) { super::diag_event::emit_ed2k_tcp_diag_event(record); } -fn maybe_sync_ed2k_tcp_dump(file: &fs::File) { - static ED2K_TCP_LIVE_SYNC_COUNTER: AtomicU64 = AtomicU64::new(0); - if ED2K_TCP_LIVE_SYNC_COUNTER - .fetch_add(1, Ordering::Relaxed) - .is_multiple_of(ED2K_TCP_LIVE_SYNC_EVERY) - { - let _ = file.sync_data(); - } -} - #[cfg(feature = "packet-diagnostics")] fn dump_ed2k_tcp_meta( flow: &'static str, @@ -375,11 +293,8 @@ fn dump_ed2k_tcp_meta( phase: &str, note: impl FnOnce() -> String, ) { - let record = Ed2kTcpDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: ed2k_tcp_dump_event_seq(), + let record = Ed2kTcpDiagnosticRecord { + event: "state", trace_key: ed2k_tcp_trace_key(flow, remote_addr), state_id: ed2k_tcp_state_id(flow, phase), state_label: phase, @@ -405,7 +320,7 @@ fn dump_ed2k_tcp_meta( payload_hex_truncated: None, note: Some(note()), }; - dump_ed2k_tcp_record(&record); + emit_ed2k_tcp_record(&record); } #[cfg(feature = "packet-diagnostics")] @@ -423,11 +338,8 @@ pub(super) fn dump_ed2k_tcp_recv_capture( } let canonical_phase = canonical_ed2k_recv_phase(flow, phase, packet.protocol, packet.opcode); let payload_hex = exact_packet_hex(&packet.payload); - let record = Ed2kTcpDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: ed2k_tcp_dump_event_seq(), + let record = Ed2kTcpDiagnosticRecord { + event: "packet", trace_key: ed2k_tcp_trace_key(flow, remote_addr), state_id: ed2k_tcp_state_id(flow, canonical_phase.as_ref()), state_label: canonical_phase.as_ref(), @@ -453,7 +365,7 @@ pub(super) fn dump_ed2k_tcp_recv_capture( payload_hex_truncated: Some(false), note: None, }; - dump_ed2k_tcp_record(&record); + emit_ed2k_tcp_record(&record); } #[cfg(feature = "packet-diagnostics")] @@ -470,11 +382,8 @@ pub(super) fn dump_ed2k_tcp_handshake_step( raw: &[u8], clear: &[u8], ) { - let record = Ed2kTcpDumpRecord { - schema: "ed2k_handshake_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: ed2k_tcp_dump_event_seq(), + let record = Ed2kTcpDiagnosticRecord { + event: "obfuscation_handshake", trace_key: ed2k_tcp_trace_key(flow, remote_addr), state_id: ed2k_tcp_state_id(flow, phase), state_label: phase, @@ -500,7 +409,7 @@ pub(super) fn dump_ed2k_tcp_handshake_step( payload_hex_truncated: None, note: Some("eD2K TCP obfuscation handshake".to_string()), }; - dump_ed2k_tcp_record(&record); + emit_ed2k_tcp_record(&record); } #[cfg(feature = "packet-diagnostics")] @@ -521,11 +430,8 @@ pub(super) fn dump_ed2k_tcp_recv_terminal( || Cow::Borrowed(phase), |(protocol, opcode)| canonical_ed2k_recv_phase(flow, phase, protocol, opcode), ); - let record = Ed2kTcpDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: ed2k_tcp_dump_event_seq(), + let record = Ed2kTcpDiagnosticRecord { + event: "packet", trace_key: ed2k_tcp_trace_key(flow, remote_addr), state_id: ed2k_tcp_state_id(flow, canonical_phase.as_ref()), state_label: canonical_phase.as_ref(), @@ -553,7 +459,7 @@ pub(super) fn dump_ed2k_tcp_recv_terminal( payload_hex_truncated: None, note: Some(detail), }; - dump_ed2k_tcp_record(&record); + emit_ed2k_tcp_record(&record); } #[cfg(feature = "packet-diagnostics")] @@ -580,11 +486,8 @@ pub(super) fn dump_ed2k_tcp_send_capture( let payload = (clear_frame.len() > TCP_PACKET_HEADER_LEN).then(|| &clear_frame[TCP_PACKET_HEADER_LEN..]); let canonical_phase = canonical_ed2k_send_phase(flow, phase, protocol, opcode); - let record = Ed2kTcpDumpRecord { - schema: "ed2k_packet_v1", - source: "emulebb-rust", - ts_utc: chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - event_seq: ed2k_tcp_dump_event_seq(), + let record = Ed2kTcpDiagnosticRecord { + event: "packet", trace_key: ed2k_tcp_trace_key(flow, remote_addr), state_id: ed2k_tcp_state_id(flow, canonical_phase.as_ref()), state_label: canonical_phase.as_ref(), @@ -612,7 +515,7 @@ pub(super) fn dump_ed2k_tcp_send_capture( payload_hex_truncated: payload.map(|_| false), note: detail, }; - dump_ed2k_tcp_record(&record); + emit_ed2k_tcp_record(&record); } // No-op variants compiled when the `packet-diagnostics` feature is off, so the @@ -674,15 +577,7 @@ pub(crate) fn dump_ed2k_tcp_download_meta( #[cfg(test)] mod tests { - use super::{ - ED2K_TCP_DUMP_FILE_PREFIX, OP_EMULEPROT, OP_SIGNATURE, canonical_ed2k_send_phase, - exact_packet_hex, - }; - - #[test] - fn tcp_dump_prefix_uses_emulebb_rust_name() { - assert_eq!(ED2K_TCP_DUMP_FILE_PREFIX, "emulebb-rust-ed2k-tcp-dump-"); - } + use super::{OP_EMULEPROT, OP_SIGNATURE, canonical_ed2k_send_phase, exact_packet_hex}; #[test] fn packet_hex_preserves_payloads_larger_than_the_former_cap() { @@ -692,45 +587,6 @@ mod tests { assert!(encoded.ends_with("ab")); } - // Regression guard: an early dump access before EMULEBB_RUST_LOG_DIR is - // visible must NOT permanently disable the dump (the OnceLock previously - // opened the file inside get_or_init and cached the resulting None forever, - // silently killing the whole run's ed2k dump). The writer must re-attempt. - #[cfg(feature = "packet-diagnostics")] - #[test] - fn dump_recovers_when_log_dir_appears_after_first_access() { - use std::net::SocketAddr; - let addr: SocketAddr = "127.0.0.1:4662".parse().unwrap(); - // First access with the log dir NOT visible: must be a recoverable no-op. - // SAFETY: this test owns the dump OnceLock's first access in its run. - unsafe { - std::env::remove_var("EMULEBB_RUST_LOG_DIR"); - } - super::dump_ed2k_tcp_download_meta(addr, None, "session", || ("pre-env").into()); - // The log dir now becomes available (mirrors the daemon making it visible - // after an early first-access): the next record must create + write. - let dir = std::env::temp_dir().join(format!("ed2k-dump-probe-{}", std::process::id())); - let _ = std::fs::remove_dir_all(&dir); - unsafe { - std::env::set_var("EMULEBB_RUST_LOG_DIR", &dir); - } - super::dump_ed2k_tcp_download_meta(addr, None, "session", || ("post-env").into()); - let created = std::fs::read_dir(&dir) - .into_iter() - .flatten() - .flatten() - .any(|e| { - e.file_name() - .to_string_lossy() - .starts_with(ED2K_TCP_DUMP_FILE_PREFIX) - }); - let _ = std::fs::remove_dir_all(&dir); - assert!( - created, - "dump must recover once EMULEBB_RUST_LOG_DIR appears after an early access" - ); - } - #[test] fn secure_ident_signature_send_phase_matches_mfc_oracle() { assert_eq!(