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/diagnostics.rs b/crates/emulebb-ed2k/src/ed2k_server/diagnostics.rs index 860de1e2..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" - ) -)] - -use std::{ - fs, - net::SocketAddr, - sync::{ - Mutex as StdMutex, OnceLock, - atomic::{AtomicU64, Ordering}, - }, -}; +// 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; + +#[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, @@ -57,6 +40,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 +72,154 @@ pub(super) struct Ed2kServerPacketCapture<'a> { pub(super) note: Option, } +#[cfg(any(feature = "packet-diagnostics", test))] +struct Ed2kServerSendContext { + family: &'static str, + 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 Ed2kServerSendAttempt { + context: Ed2kServerSendContext, + completed: bool, +} + +#[cfg(feature = "packet-diagnostics")] +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), + 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, + } + } + + 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, + 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 Ed2kServerSendAttempt { + fn drop(&mut self) { + if !self.completed { + self.settle( + Ed2kServerSendOutcome::Cancelled, + false, + Some(format!( + "server {} send future cancelled before completion", + if self.context.role == "udp" { + "UDP" + } else { + "TCP" + } + )), + ); + } + } +} + #[cfg(not(feature = "packet-diagnostics"))] pub(super) fn dump_ed2k_server_meta(_session: &ServerSession, _note: impl Into) {} @@ -102,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), @@ -115,6 +243,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}"), @@ -158,6 +283,7 @@ pub(super) fn dump_ed2k_server_loop_meta( phase, direction: "meta", outcome: "meta", + send_attempt_id: None, remote_addr: endpoint.to_string(), transport_mode: "unknown", obfuscated: None, @@ -174,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"))] @@ -203,6 +329,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", @@ -213,10 +349,6 @@ pub(super) fn dump_ed2k_server_udp_packet( .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}"), @@ -226,6 +358,7 @@ pub(super) fn dump_ed2k_server_udp_packet( phase: "udp", direction, outcome: capture.outcome, + send_attempt_id, remote_addr: remote_addr.to_string(), transport_mode: capture.transport_mode, obfuscated: capture.obfuscated, @@ -242,13 +375,22 @@ pub(super) fn dump_ed2k_server_udp_packet( 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")] pub(super) fn dump_ed2k_server_packet( session: &ServerSession, capture: Ed2kServerPacketCapture<'_>, +) { + 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. @@ -258,10 +400,6 @@ pub(super) fn dump_ed2k_server_packet( 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), @@ -271,6 +409,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, @@ -287,9 +426,10 @@ pub(super) fn dump_ed2k_server_packet( 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", @@ -298,15 +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:{}:{}:{}", @@ -314,81 +457,42 @@ 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, 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)); + 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)); } @@ -399,6 +503,9 @@ fn ed2k_server_diag_event_parts(record: &Ed2kServerDumpRecord<'_>) -> (&'static 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)); } @@ -440,24 +547,92 @@ 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")] -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(); +fn emit_ed2k_server_send_outcome( + context: &Ed2kServerSendContext, + outcome: Ed2kServerSendOutcome, + delivery_known: bool, + note: Option<&str>, +) { + let (family, severity, keys, body) = + ed2k_server_send_outcome_parts(context, outcome, delivery_known, note); + crate::diag_event::emit(family, "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, &'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)); + } + + ( + context.family, + outcome.severity(), + Value::Object(keys), + Value::Object(body), + ) } +#[cfg(any(feature = "packet-diagnostics", test))] fn server_opcode_name(opcode: u8) -> &'static str { match opcode { OP_LOGINREQUEST => "OP_LOGINREQUEST", @@ -501,19 +676,89 @@ fn server_opcode_name(opcode: u8) -> &'static str { #[cfg(test)] mod tests { use super::{ - ED2K_SERVER_DUMP_FILE_PREFIX, Ed2kServerDumpRecord, OP_CALLBACKREQUEST, + 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, exact_packet_hex, - server_opcode_name, server_protocol_name, + 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-" + 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(), + 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 (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"); + 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_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] @@ -522,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(), @@ -533,8 +774,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), @@ -552,10 +794,17 @@ 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!(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"); assert_eq!(body["obfuscated"], false); assert_eq!(body["rawLen"], raw.len()); assert_eq!(body["rawHex"].as_str().unwrap().len(), raw.len() * 2); @@ -575,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(), @@ -588,6 +833,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), @@ -605,18 +851,16 @@ 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); } #[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(), @@ -626,6 +870,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), @@ -643,14 +888,56 @@ 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 { + 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", + send_attempt_id: None, + 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"); diff --git a/crates/emulebb-ed2k/src/ed2k_server/session.rs b/crates/emulebb-ed2k/src/ed2k_server/session.rs index 42b6faee..2236d4e4 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, 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")] - dump_ed2k_server_packet( + let send_attempt = Ed2kServerSendAttempt::tcp( self, Ed2kServerPacketCapture { direction: "tx", @@ -695,30 +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(()) - } - - 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(()) + 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) + } + } } fn server_supports_compression(&self) -> bool { @@ -727,8 +726,10 @@ impl ServerSession { 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); @@ -825,40 +826,43 @@ 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 ) - })?; + }); + 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); + } + } 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(); 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] + ); + } }