diff --git a/Cargo.lock b/Cargo.lock index b61d5377..f422f3cd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -943,8 +943,7 @@ dependencies = [ [[package]] name = "ant-protocol" version = "2.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3081ee130dd45e8166bc8ac4d789aac17d143e2e7f54f1c3006503dc63cf3edb" +source = "git+https://github.com/WithAutonomi/ant-protocol.git?branch=diagnostics%2Fv2-903-response-transport-metadata#5557a57f010009eac193122d32c5ca0b7f978098" dependencies = [ "blake3", "bytes", @@ -3255,7 +3254,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.57.0", + "windows-core 0.58.0", ] [[package]] @@ -5199,8 +5198,7 @@ dependencies = [ [[package]] name = "saorsa-core" version = "0.26.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "454529f8a72b4cf22f7d9c3b009ad9d4ba78520e11f6444ae460795d55c002da" +source = "git+https://github.com/WithAutonomi/saorsa-core.git?branch=diagnostics%2Fv2-903-peer-route-classification#6be87a81f91a119d0761f9ce6f8ee69dec19edcc" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index f3978b22..1af2f346 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,13 @@ [workspace] members = ["ant-core", "ant-cli"] resolver = "2" + +# Temporary V2-903 diagnostic override. ant-core retains its ONE-pin policy and +# does not depend on saorsa-core directly; ant-protocol resolves this transitively. +# The ant-protocol patch exposes `send_and_await_chunk_response_with_metadata` +# (returning `ChunkProtocolResponse { result, source_peer, transport_source }`) +# and the saorsa-core patch exposes `P2PNode::classify_peer_transport_route` and +# `PeerRouteKind`. Both are needed for schema v2 route classification. +[patch.crates-io] +ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol.git", branch = "diagnostics/v2-903-response-transport-metadata" } +saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core.git", branch = "diagnostics/v2-903-peer-route-classification" } diff --git a/ant-cli/src/commands/data/file.rs b/ant-cli/src/commands/data/file.rs index 1da346dc..46c3a22c 100644 --- a/ant-cli/src/commands/data/file.rs +++ b/ant-cli/src/commands/data/file.rs @@ -9,9 +9,9 @@ use tokio::sync::mpsc; use tracing::info; use ant_core::data::{ - Client, CollisionPolicy, CostEstimateConfidence, DownloadEvent, Error as DataError, - FileChunkPeerReport, FileChunkPeerReportPeer, FileChunkPeerStatus, FileChunkPeerSweepReport, - PaymentMode, UploadEvent, + spawn_download_diagnostics_writer, Client, CollisionPolicy, CostEstimateConfidence, + DownloadEvent, Error as DataError, FileChunkPeerReport, FileChunkPeerReportPeer, + FileChunkPeerStatus, FileChunkPeerSweepReport, PaymentMode, UploadEvent, }; use ant_core::datamap_file::{original_name_from_datamap, read_datamap, write_datamap}; @@ -78,6 +78,9 @@ pub enum FileAction { /// ranked per-peer results after a successful download. #[arg(long, alias = "try-all-peers")] all_peers: bool, + /// Write one JSONL record per normal-path chunk fetch attempt. + #[arg(long, value_name = "PATH", conflicts_with = "all_peers")] + download_diagnostics: Option, }, /// Estimate the cost of uploading a file without uploading. /// @@ -165,6 +168,7 @@ impl FileAction { output, peers, all_peers, + download_diagnostics, } => { let resolved_output = resolve_download_output(output, datamap.as_deref())?; handle_file_download( @@ -175,6 +179,7 @@ impl FileAction { json, peers, all_peers, + download_diagnostics.as_deref(), ) .await } @@ -432,6 +437,7 @@ async fn drive_upload_progress( pb.finish_and_clear(); } +#[allow(clippy::too_many_arguments)] async fn handle_file_download( client: &Client, address: Option<&str>, @@ -440,9 +446,22 @@ async fn handle_file_download( json_output: bool, peer_count: Option, all_peers: bool, + download_diagnostics: Option<&Path>, ) -> anyhow::Result<()> { let output_path = output; let start = Instant::now(); + let (diagnostics, diagnostics_writer) = match download_diagnostics { + Some(path) => { + let (sender, writer) = spawn_download_diagnostics_writer(path).map_err(|e| { + anyhow::anyhow!( + "Failed to open download diagnostics sidecar {}: {e}", + path.display() + ) + })?; + (Some(sender), Some(writer)) + } + None => (None, None), + }; let data_map = if let Some(addr_hex) = address { info!("Downloading public file from address {addr_hex}"); @@ -489,7 +508,18 @@ async fn handle_file_download( .map_err(|e| anyhow::anyhow!("Download failed: {e}"))?; Some(file_peer_check_from_reports(report.chunk_reports)) } else { - let download_result = if let Some(peer_count) = peer_count { + let download_result = if let Some(diagnostics) = diagnostics.clone() { + let peer_count = download_peer_check_count(client, peer_count)?; + client + .file_download_with_progress_and_diagnostics_from_closest_peers( + &data_map, + &output_path, + None, + peer_count, + Some(diagnostics), + ) + .await + } else if let Some(peer_count) = peer_count { client .file_download_from_closest_peers(&data_map, &output_path, peer_count) .await @@ -551,7 +581,18 @@ async fn handle_file_download( .map_err(|e| anyhow::anyhow!("Download failed: {e}"))?; Some(file_peer_check_from_reports(report.chunk_reports)) } else { - let download_result = if let Some(peer_count) = peer_count { + let download_result = if let Some(diagnostics) = diagnostics.clone() { + let peer_count = download_peer_check_count(client, peer_count)?; + client + .file_download_with_progress_and_diagnostics_from_closest_peers( + &data_map, + &output_path, + Some(tx), + peer_count, + Some(diagnostics), + ) + .await + } else if let Some(peer_count) = peer_count { client .file_download_with_progress_from_closest_peers( &data_map, @@ -575,6 +616,13 @@ async fn handle_file_download( chunk_peer_check }; + drop(diagnostics); + if let Some(writer) = diagnostics_writer { + writer + .join() + .map_err(|_| anyhow::anyhow!("Download diagnostics writer thread panicked"))?; + } + let file_size = std::fs::metadata(&output_path)?.len(); let elapsed = start.elapsed(); diff --git a/ant-core/src/data/client/chunk.rs b/ant-core/src/data/client/chunk.rs index 4f9fd338..93c7f3c9 100644 --- a/ant-core/src/data/client/chunk.rs +++ b/ant-core/src/data/client/chunk.rs @@ -5,25 +5,64 @@ use crate::data::client::adaptive::Outcome; use crate::data::client::batch::{finalize_batch_payment, PreparedChunk}; +use crate::data::client::diagnostics::{ + bounded_error, unix_now_ms, DownloadDiagnosticsOutcome, DownloadDiagnosticsRecord, + DownloadDiagnosticsSender, DownloadRequestCorrelation, +}; use crate::data::client::peer_xor_distance; use crate::data::client::Client; use crate::data::error::{Error, Result}; +use crate::data::network::ClosestPeerDiagnostics; use ant_protocol::evm::{QuoteHash, TxHash}; -use ant_protocol::transport::{MultiAddr, PeerId}; +use ant_protocol::transport::{MultiAddr, PeerId, PeerRouteKind}; use ant_protocol::{ - compute_address, detect_proof_type, send_and_await_chunk_response, ChunkGetRequest, - ChunkGetResponse, ChunkMessage, ChunkMessageBody, ChunkPutRequest, ChunkPutResponse, DataChunk, + compute_address, detect_proof_type, send_and_await_chunk_response, + send_and_await_chunk_response_with_metadata, ChunkGetRequest, ChunkGetResponse, ChunkMessage, + ChunkMessageBody, ChunkProtocolResponse, ChunkPutRequest, ChunkPutResponse, DataChunk, ProofType, ProtocolError, XorName, CLOSE_GROUP_MAJORITY, }; use bytes::Bytes; use futures::stream::{self, FuturesUnordered, StreamExt}; use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; use std::time::{Duration, Instant}; use tracing::{debug, info, warn}; /// Data type identifier for chunks (used in quote requests). const CHUNK_DATA_TYPE: u32 = 0; +/// Number of diagnostics-enabled peer requests currently in flight in this +/// process. The counter is untouched when runtime diagnostics are disabled. +static ACTIVE_DIAGNOSTIC_REQUESTS: AtomicUsize = AtomicUsize::new(0); +static NEXT_DIAGNOSTIC_LOOKUP_ID: AtomicUsize = AtomicUsize::new(1); + +struct ActiveDiagnosticRequestGuard; + +impl ActiveDiagnosticRequestGuard { + fn enter() -> (Self, usize) { + let active = ACTIVE_DIAGNOSTIC_REQUESTS.fetch_add(1, AtomicOrdering::Relaxed) + 1; + (Self, active) + } +} + +impl Drop for ActiveDiagnosticRequestGuard { + fn drop(&mut self) { + ACTIVE_DIAGNOSTIC_REQUESTS.fetch_sub(1, AtomicOrdering::Relaxed); + } +} + +fn encode_diagnostic_chunk_get_request( + address: &XorName, + correlation: &DownloadRequestCorrelation, +) -> Result> { + ChunkMessage { + request_id: correlation.request_id, + body: ChunkMessageBody::GetRequest(ChunkGetRequest::new(*address)), + } + .encode() + .map_err(|e| Error::Protocol(format!("Failed to encode GET request: {e}"))) +} + /// Why a single-peer PUT was declined. Drives the surfaced aggregate error /// and keeps the store AIMD limiter honest — only genuine local backpressure /// (a PUT-response `Timeout`) is a "client is sending too fast" signal; a node @@ -195,6 +234,173 @@ struct ChunkPeerGetTarget { xor_distance: [u8; 32], } +/// Shared context for emitting per-chunk download diagnostics records from +/// the normal-path chunk fetch. Constructed only when the caller supplied a +/// [`DownloadDiagnosticsSender`] (i.e. `--download-diagnostics` was passed); +/// otherwise `None` and no records are allocated. +/// +/// `sweep` is set per `chunk_get_try_closest_peers` call: `"initial"` for the +/// first close-group sweep, `"retry"` for the internal retry sweep. +pub(crate) struct ChunkFetchDiagnostics<'a> { + sender: &'a DownloadDiagnosticsSender, + file_attempt: usize, + chunk_index: usize, + chunk_address: [u8; 32], + fetch_cap: usize, +} + +impl<'a> ChunkFetchDiagnostics<'a> { + pub(crate) fn new( + sender: &'a DownloadDiagnosticsSender, + file_attempt: usize, + chunk_index: usize, + chunk_address: [u8; 32], + fetch_cap: usize, + ) -> Self { + Self { + sender, + file_attempt, + chunk_index, + chunk_address, + fetch_cap, + } + } + + /// Emit a per-peer-attempt record. `lookup_duration_ms` is attached only + /// for the first peer attempt of the sweep. + #[allow(clippy::too_many_arguments)] + fn emit_peer_attempt( + &self, + sweep: &'static str, + peer_attempt: usize, + lookup_duration_ms: Option, + lookup_correlation_id: &str, + peer_context: &ClosestPeerDiagnostics, + expected_peer: &PeerId, + source_peer: Option<&PeerId>, + transport_source: Option<&MultiAddr>, + route: PeerRouteKind, + peer_connected_before_request: bool, + active_requests_at_start: usize, + request_started_unix_ms: u64, + request_completed_unix_ms: u64, + correlation: &DownloadRequestCorrelation, + response_elapsed_ms: u64, + bytes: u64, + outcome: DownloadDiagnosticsOutcome, + error: Option, + ) { + self.sender + .try_emit(DownloadDiagnosticsRecord::peer_attempt( + self.file_attempt, + self.chunk_index, + &self.chunk_address, + sweep, + peer_attempt, + lookup_duration_ms, + lookup_correlation_id, + &expected_peer.to_string(), + peer_context + .addresses + .iter() + .map(ToString::to_string) + .collect(), + peer_context.address_types.clone(), + peer_context.local_last_seen_age_ms, + peer_context.publisher_address_set_age_ms, + peer_context.publisher_address_set_unix_ns, + source_peer.map(ToString::to_string).as_deref(), + transport_source.map(ToString::to_string).as_deref(), + route.as_str(), + (route == PeerRouteKind::Unknown) + .then_some(DownloadDiagnosticsRecord::ROUTE_UNKNOWN_NOTE), + Some(peer_connected_before_request), + Some(active_requests_at_start), + Some(self.fetch_cap), + request_started_unix_ms, + request_completed_unix_ms, + correlation, + response_elapsed_ms, + bytes, + outcome, + error, + )); + } + + /// Emit a chunk-level record (cache hit, lookup error, or exhausted). + fn emit_chunk_level( + &self, + sweep: &'static str, + bytes: u64, + outcome: DownloadDiagnosticsOutcome, + error: Option, + ) { + self.sender.try_emit(DownloadDiagnosticsRecord::chunk_level( + self.file_attempt, + self.chunk_index, + &self.chunk_address, + sweep, + Some(self.fetch_cap), + bytes, + outcome, + error, + )); + } +} + +/// Classify a `chunk_get_from_peer` result into a diagnostics outcome plus the +/// returned byte count and whether a response was received (so `source_peer` +/// can be attributed). No secrets: the bounded error string uses only the +/// error's `Display` form. +fn classify_peer_attempt( + result: &Result>, +) -> (DownloadDiagnosticsOutcome, u64, bool, Option) { + match result { + Ok(Some(chunk)) => ( + DownloadDiagnosticsOutcome::Found, + chunk.content.len() as u64, + true, + None, + ), + Ok(None) => (DownloadDiagnosticsOutcome::NotFound, 0, true, None), + Err(Error::Timeout(msg)) => ( + DownloadDiagnosticsOutcome::Timeout, + 0, + false, + Some(bounded_error("timeout", msg)), + ), + Err(Error::Network(msg)) => ( + DownloadDiagnosticsOutcome::NetworkError, + 0, + false, + Some(bounded_error("network", msg)), + ), + // Invalid data can only be constructed after a response body was + // received and validated, so attributing the matched peer is sound. + Err(Error::InvalidData(msg)) => ( + DownloadDiagnosticsOutcome::ProtocolError, + 0, + true, + Some(bounded_error("protocol", msg)), + ), + // `Protocol` includes both a remote GET error and a local request + // encoding failure. Without a distinct provenance bit, conservatively + // avoid claiming a response peer for either case. + Err(Error::Protocol(msg)) => ( + DownloadDiagnosticsOutcome::ProtocolError, + 0, + false, + Some(bounded_error("protocol", msg)), + ), + Err(e) => ( + DownloadDiagnosticsOutcome::ProtocolError, + 0, + false, + Some(bounded_error("protocol", &e.to_string())), + ), + } +} + fn chunk_peer_get_targets( peers: Vec<(PeerId, Vec)>, address: &XorName, @@ -269,7 +475,7 @@ impl Client { /// sustained run of close-group exhaustions correctly drives the /// cap down rather than silently inflating it. pub(crate) async fn chunk_get_observed(&self, address: &XorName) -> Result> { - self.chunk_get_observed_from_closest_peers(address, self.config().close_group_size) + self.chunk_get_observed_from_closest_peers(address, self.config().close_group_size, None) .await } @@ -277,9 +483,12 @@ impl Client { &self, address: &XorName, peer_count: usize, + diag: Option<&ChunkFetchDiagnostics<'_>>, ) -> Result> { let started = Instant::now(); - let result = self.chunk_get_from_closest_peers(address, peer_count).await; + let result = self + .chunk_get_from_closest_peers_with_diagnostics(address, peer_count, diag) + .await; let latency = started.elapsed(); let bytes = result .as_ref() @@ -648,12 +857,30 @@ impl Client { &self, address: &XorName, peer_count: usize, + ) -> Result> { + self.chunk_get_from_closest_peers_with_diagnostics(address, peer_count, None) + .await + } + + async fn chunk_get_from_closest_peers_with_diagnostics( + &self, + address: &XorName, + peer_count: usize, + diag: Option<&ChunkFetchDiagnostics<'_>>, ) -> Result> { // Check cache first, with integrity verification. if let Some(cached) = self.chunk_cache().get(address) { let computed = compute_address(&cached); if computed == *address { debug!("Cache hit for chunk {}", hex::encode(address)); + if let Some(diag) = diag { + diag.emit_chunk_level( + "initial", + cached.len() as u64, + DownloadDiagnosticsOutcome::CacheHit, + None, + ); + } return Ok(Some(DataChunk::new(*address, cached))); } // Cache entry corrupted — evict and fall through to network fetch. @@ -675,7 +902,10 @@ impl Client { // chunk would fail an entire multi-hundred-chunk download. A // zeroed outcome (queried=0) is never authoritative, so it flows // straight to the retry below. - let first = match self.chunk_get_try_closest_peers(address, peer_count).await { + let first = match self + .chunk_get_try_closest_peers(address, peer_count, diag, "initial") + .await + { Ok(outcome) => outcome, Err(e) => { info!("chunk_get first close-group lookup failed for {addr_hex}: {e}; will retry"); @@ -732,7 +962,10 @@ impl Client { // If the retry's DHT lookup itself fails, treat that as "still // couldn't find" rather than escalating the error — matches the // semantics of the first attempt when peers are unreachable. - let retry = match self.chunk_get_try_closest_peers(address, peer_count).await { + let retry = match self + .chunk_get_try_closest_peers(address, peer_count, diag, "retry") + .await + { Ok(o) => o, Err(e) => { info!( @@ -774,21 +1007,152 @@ impl Client { /// One sweep of the requested closest peers: fetch the closest peers /// for `address` from the DHT and ask each for the chunk in turn, /// returning on the first success. + /// + /// `sweep` is `"initial"` for the first close-group attempt and + /// `"retry"` for the internal retry sweep; it is only used as a label + /// on diagnostic records when `diag` is `Some`. async fn chunk_get_try_closest_peers( &self, address: &XorName, peer_count: usize, + diag: Option<&ChunkFetchDiagnostics<'_>>, + sweep: &'static str, ) -> Result { - let peers = self.closest_peers(address, peer_count).await?; + let lookup_start = Instant::now(); + let (peers, peer_contexts) = if diag.is_some() { + match self + .network() + .find_closest_peers_with_diagnostics(address, peer_count) + .await + { + Ok(contexts) => { + let peers = contexts + .iter() + .map(|context| (context.peer_id, context.addresses.clone())) + .collect(); + (peers, Some(contexts)) + } + Err(e) => { + if let Some(diag) = diag { + diag.emit_chunk_level( + sweep, + 0, + DownloadDiagnosticsOutcome::LookupError, + Some(bounded_error("lookup", &e.to_string())), + ); + } + return Err(e); + } + } + } else { + // Preserve the pre-instrumentation lookup path exactly when + // diagnostics are disabled. + match self.closest_peers(address, peer_count).await { + Ok(peers) => (peers, None), + Err(e) => return Err(e), + } + }; + let lookup_duration_ms = + u64::try_from(lookup_start.elapsed().as_millis()).unwrap_or(u64::MAX); + let lookup_duration_opt = Some(lookup_duration_ms); let addr_hex = hex::encode(address); + let lookup_correlation_id = diag.map(|diag| { + let sequence = NEXT_DIAGNOSTIC_LOOKUP_ID.fetch_add(1, AtomicOrdering::Relaxed); + format!( + "{}-{}-{}-{}-{sequence}", + diag.file_attempt, diag.chunk_index, sweep, addr_hex + ) + }); let queried = peers.len(); let mut not_found = 0usize; let mut timeout = 0usize; let mut network_err = 0usize; let mut protocol_err = 0usize; - for (peer, addrs) in &peers { - match self.chunk_get_from_peer(address, peer, addrs).await { + for (peer_attempt, (peer, addrs)) in peers.iter().enumerate() { + let peer_attempt_no = peer_attempt + 1; + let result = if let Some(diag) = diag { + let Some(peer_context) = peer_contexts + .as_ref() + .and_then(|contexts| contexts.get(peer_attempt)) + else { + return Err(Error::Network( + "diagnostics peer context missing for selected peer".to_string(), + )); + }; + let Some(lookup_correlation_id) = lookup_correlation_id.as_deref() else { + return Err(Error::Network( + "diagnostics lookup correlation ID missing".to_string(), + )); + }; + let node = self.network().node(); + let peer_connected_before_request = node.is_peer_connected(peer).await; + let (active_guard, active_requests_at_start) = + ActiveDiagnosticRequestGuard::enter(); + let request_started_unix_ms = unix_now_ms(); + let resp_start = Instant::now(); + let correlation = + DownloadRequestCorrelation::new(self.next_request_id(), node.peer_id()); + let observed = self + .chunk_get_from_peer_with_metadata(address, peer, addrs, &correlation) + .await; + let response_elapsed_ms = + u64::try_from(resp_start.elapsed().as_millis()).unwrap_or(u64::MAX); + let request_completed_unix_ms = unix_now_ms(); + // Count only the network request itself; route classification + // and sidecar emission are diagnostic bookkeeping. + drop(active_guard); + + let (result, source_peer, transport_source, route) = match observed { + Ok(response) => { + let route = node + .classify_peer_transport_route( + &response.source_peer, + response.transport_source.as_ref(), + ) + .await; + ( + response.result, + Some(response.source_peer), + response.transport_source, + route, + ) + } + Err(error) => (Err(error), None, None, PeerRouteKind::Unknown), + }; + let (outcome, bytes, _got_response, error) = classify_peer_attempt(&result); + let lookup = if peer_attempt_no == 1 { + lookup_duration_opt + } else { + None + }; + diag.emit_peer_attempt( + sweep, + peer_attempt_no, + lookup, + lookup_correlation_id, + peer_context, + peer, + source_peer.as_ref(), + transport_source.as_ref(), + route, + peer_connected_before_request, + active_requests_at_start, + request_started_unix_ms, + request_completed_unix_ms, + &correlation, + response_elapsed_ms, + bytes, + outcome, + error, + ); + result + } else { + // Preserve the existing request path exactly when diagnostics + // are disabled: no metadata lookup, clock read, or counter. + self.chunk_get_from_peer(address, peer, addrs).await + }; + match result { Ok(Some(chunk)) => { return Ok(CloseGroupOutcome { chunk: Some(chunk), @@ -829,6 +1193,13 @@ impl Client { } } + // The sweep queried every selected peer without success. Emit an + // explicit exhausted record so the peer-set exhaustion is a record + // rather than a silent gap. + if let Some(diag) = diag { + diag.emit_chunk_level(sweep, 0, DownloadDiagnosticsOutcome::Exhausted, None); + } + Ok(CloseGroupOutcome { chunk: None, queried, @@ -1003,6 +1374,72 @@ impl Client { result } + /// Diagnostics-only variant of [`Self::chunk_get_from_peer`] that retains + /// the authenticated response peer and observed transport source. Request + /// construction, validation, timeouts, and error mapping mirror the normal + /// helper exactly. + async fn chunk_get_from_peer_with_metadata( + &self, + address: &XorName, + peer: &PeerId, + peer_addrs: &[MultiAddr], + correlation: &DownloadRequestCorrelation, + ) -> Result, Error>> { + let node = self.network().node(); + let message_bytes = encode_diagnostic_chunk_get_request(address, correlation)?; + + let timeout = Duration::from_secs(self.config().chunk_get_timeout_secs); + let addr_hex = hex::encode(address); + let timeout_secs = self.config().chunk_get_timeout_secs; + + send_and_await_chunk_response_with_metadata( + node, + peer, + message_bytes, + correlation.request_id, + timeout, + peer_addrs, + |body| match body { + ChunkMessageBody::GetResponse(ChunkGetResponse::Success { + address: addr, + content, + }) => { + if addr != *address { + return Some(Err(Error::InvalidData(format!( + "Mismatched chunk address: expected {addr_hex}, got {}", + hex::encode(addr) + )))); + } + let computed = compute_address(&content); + if computed != addr { + return Some(Err(Error::InvalidData(format!( + "Invalid chunk content: expected hash {addr_hex}, got {}", + hex::encode(computed) + )))); + } + debug!( + "Retrieved chunk {} ({} bytes) from peer {peer}", + hex::encode(addr), + content.len() + ); + Some(Ok(Some(DataChunk::new(addr, Bytes::from(content))))) + } + ChunkMessageBody::GetResponse(ChunkGetResponse::NotFound { .. }) => Some(Ok(None)), + ChunkMessageBody::GetResponse(ChunkGetResponse::Error(e)) => Some(Err( + Error::Protocol(format!("Remote GET error for {addr_hex}: {e}")), + )), + _ => None, + }, + |e| Error::Network(format!("Failed to send GET to peer {peer}: {e}")), + || { + Error::Timeout(format!( + "Timeout waiting for chunk {addr_hex} from {peer} after {timeout_secs}s" + )) + }, + ) + .await + } + /// Check if a chunk exists on the network. /// /// # Errors @@ -1064,6 +1501,108 @@ mod tests { /// Last byte position in the test XOR distance arrays. const TEST_DISTANCE_TAIL_INDEX: usize = TEST_XORNAME_BYTE_LEN - 1; + #[test] + fn diagnostic_correlation_is_identical_on_wire_and_in_record() { + let address = [7u8; 32]; + let correlation = DownloadRequestCorrelation::new( + 9_903, + &PeerId::from_bytes([42; TEST_XORNAME_BYTE_LEN]), + ); + let encoded = encode_diagnostic_chunk_get_request(&address, &correlation).unwrap(); + let wire = ChunkMessage::decode(&encoded).unwrap(); + assert_eq!(wire.request_id, correlation.request_id); + assert!(matches!(wire.body, ChunkMessageBody::GetRequest(_))); + + let record = DownloadDiagnosticsRecord::peer_attempt( + 1, + 1, + &address, + "initial", + 1, + None, + "lookup-1", + "expected-peer", + Vec::new(), + Vec::new(), + None, + None, + None, + None, + None, + "unknown", + None, + Some(false), + Some(1), + Some(8), + 100, + 200, + &correlation, + 100, + 0, + DownloadDiagnosticsOutcome::Timeout, + Some("timeout".to_string()), + ); + assert_eq!(record.request_id, Some(wire.request_id)); + assert_eq!(record.local_peer_id, Some(correlation.local_peer_id)); + } + + #[test] + fn classify_peer_attempt_pins_outcomes_and_response_attribution() { + let chunk = DataChunk::new([0u8; 32], Bytes::from_static(b"payload")); + let cases = [ + ( + Ok(Some(chunk)), + DownloadDiagnosticsOutcome::Found, + 7, + true, + None, + ), + ( + Ok(None), + DownloadDiagnosticsOutcome::NotFound, + 0, + true, + None, + ), + ( + Err(Error::Timeout("late".to_string())), + DownloadDiagnosticsOutcome::Timeout, + 0, + false, + Some("timeout: late"), + ), + ( + Err(Error::Network("dial".to_string())), + DownloadDiagnosticsOutcome::NetworkError, + 0, + false, + Some("network: dial"), + ), + ( + Err(Error::InvalidData("hash".to_string())), + DownloadDiagnosticsOutcome::ProtocolError, + 0, + true, + Some("protocol: hash"), + ), + ( + Err(Error::Protocol("remote".to_string())), + DownloadDiagnosticsOutcome::ProtocolError, + 0, + false, + Some("protocol: remote"), + ), + ]; + + for (result, expected_outcome, expected_bytes, expected_response, expected_error) in cases { + let (outcome, bytes, got_response, error) = classify_peer_attempt(&result); + assert_eq!(outcome, expected_outcome); + assert_eq!(bytes, expected_bytes); + assert_eq!(got_response, expected_response); + assert_eq!(error.as_deref(), expected_error); + } + } + #[test] fn classify_put_failure_maps_remote_timeout_and_dial_reasons() { let remote = |source| Error::RemotePut { diff --git a/ant-core/src/data/client/diagnostics.rs b/ant-core/src/data/client/diagnostics.rs new file mode 100644 index 00000000..d2a6b391 --- /dev/null +++ b/ant-core/src/data/client/diagnostics.rs @@ -0,0 +1,940 @@ +//! Normal-path download diagnostics instrumentation. +//! +//! Runtime-gated sidecar JSONL writer for `ant file download +//! --download-diagnostics `. One record is emitted per normal-path +//! chunk fetch attempt (cache hit, per-peer attempt, lookup failure, or +//! exhausted peer set) while the existing early-return / retry / +//! adaptive-concurrency / stdout behaviour is preserved. +//! +//! When the `--download-diagnostics` flag is absent, no channel, file, or +//! writer is created and the download path is unchanged. The optional sender +//! threaded through the file/chunk download path is `None`, so record +//! construction is skipped entirely (no allocation, no I/O). +//! +//! # Schema v4: exact node/client request correlation +//! +//! Peer-attempt records carry the request ID allocated by this client and the +//! client's local peer ID. The same request ID is encoded on the chunk GET, +//! while the peer ID matches the serving node's `source_peer`, permitting an +//! exact join to node-side GET telemetry. Chunk-level records leave both +//! fields `null` because no individual peer request was sent. +//! +//! The `ant-protocol` diagnostic branch +//! `diagnostics/v2-903-response-transport-metadata` exposes +//! `send_and_await_chunk_response_with_metadata`, which returns a +//! `ChunkProtocolResponse { result, source_peer, transport_source }`. The +//! `saorsa-core` branch `diagnostics/v2-903-peer-route-classification` +//! exposes `P2PNode::classify_peer_transport_route(expected_peer, +//! transport_source)` returning a `PeerRouteKind` +//! (`direct`/`relay`/`lan`/`unverified`/`unknown`). Schema v2 records the +//! *actual* `source_peer` and `transport_source` from the observed response, +//! classifies the route from the actual transport source against the peer's +//! typed DHT addresses, and attaches a `route_note` only when the route is +//! `unknown`. A `peer_connected_before_request` sample +//! (`node.is_peer_connected(peer)` called before the send) and an adaptive +//! `fetch_cap` snapshot are included on every record. +//! +//! # TTFB limitation +//! +//! The protocol event is emitted only after complete message reassembly, so +//! this branch measures complete-response latency (`response_elapsed_ms`), +//! not true network time-to-first-byte. `ttfb_ms` is always `null`, +//! `ttfb_available` is `false`, and `ttfb_unavailable_reason` carries the +//! explanation. This prevents complete-response latency being presented as +//! TTFB. + +use std::fmt; +use std::io::{self, Write}; +use std::path::Path; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{mpsc, Arc}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use ant_protocol::transport::PeerId; +use serde::Serialize; +use tracing::{error, warn}; + +/// Current diagnostic record schema discriminator. +pub const DIAGNOSTICS_SCHEMA_VERSION: u8 = 4; + +/// Correlation values captured once for a peer request and shared by both +/// the encoded protocol message and its diagnostic record. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct DownloadRequestCorrelation { + pub(crate) request_id: u64, + pub(crate) local_peer_id: String, +} + +impl DownloadRequestCorrelation { + pub(crate) fn new(request_id: u64, local_peer_id: &PeerId) -> Self { + Self { + request_id, + local_peer_id: local_peer_id.to_string(), + } + } +} + +/// Bounded capacity of the diagnostics channel. A slow writer must not create +/// unbounded memory growth: when full, further records are dropped (counted +/// via `try_send`), which is acceptable for a best-effort diagnostic sidecar. +const DIAGNOSTICS_CHANNEL_CAPACITY: usize = 1024; + +/// Upper bound on the length of the `error` string we serialize, so a verbose +/// remote error message cannot balloon the sidecar file. The category prefix +/// is always preserved; only the trailing detail is truncated. +const DIAGNOSTICS_ERROR_MAX_CHARS: usize = 240; + +/// The outcome of a single normal-path chunk fetch attempt. +/// +/// Each variant maps to a stable lowercase string used as the `outcome` JSON +/// field. Variants are intentionally exhaustive over the record categories +/// listed in the design doc. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DownloadDiagnosticsOutcome { + /// A chunk was successfully fetched from a peer. + Found, + /// A queried peer responded `NotFound`. + NotFound, + /// A peer attempt timed out waiting for a response. + Timeout, + /// A peer attempt failed at the transport / dial / send layer. + NetworkError, + /// A peer responded with a structured protocol-level error (e.g. a + /// corrupted-chunk `ChunkGetResponse::Error` or a content/address + /// mismatch). + ProtocolError, + /// The chunk was served from the in-memory cache; no peer was contacted. + CacheHit, + /// The DHT closest-peer lookup itself failed before any peer was queried. + LookupError, + /// A sweep queried every selected peer without success. + Exhausted, +} + +impl DownloadDiagnosticsOutcome { + /// Canonical lowercase label for the `outcome` JSON field. + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Found => "found", + Self::NotFound => "not_found", + Self::Timeout => "timeout", + Self::NetworkError => "network_error", + Self::ProtocolError => "protocol_error", + Self::CacheHit => "cache_hit", + Self::LookupError => "lookup_error", + Self::Exhausted => "exhausted", + } + } +} + +impl fmt::Display for DownloadDiagnosticsOutcome { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl Serialize for DownloadDiagnosticsOutcome { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + serializer.serialize_str(self.as_str()) + } +} + +/// One JSONL record for a normal-path chunk fetch attempt. +/// +/// Field names are stable and pinned by the serialization tests. `null` JSON +/// values are used for fields that do not apply to a given record kind (e.g. +/// `peer_attempt` / `expected_peer` / `source_peer` / `lookup_duration_ms` are +/// `null` for a cache hit, which has no peer). +#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +pub struct DownloadDiagnosticsRecord { + /// Stable schema discriminator, currently `4`. + pub schema_version: u8, + /// UTC time the attempt completed, RFC 3339 (`YYYY-MM-DDTHH:MM:SSZ`). + pub timestamp: String, + /// Wall-clock Unix time immediately before the peer request, in + /// milliseconds. Together with `request_completed_unix_ms`, this permits + /// request overlap to be reconstructed against fleet telemetry. + pub request_started_unix_ms: Option, + /// Wall-clock Unix time immediately after the peer request completed, in + /// milliseconds. `None` for chunk-level records. + pub request_completed_unix_ms: Option, + /// Protocol request identifier allocated by this client and sent on the + /// wire; the serving node's record for this GET carries the same value. + /// `None` for chunk-level records where no peer request was sent. + pub request_id: Option, + /// This diagnostic client's local peer ID, matching the serving node's + /// `source_peer` field. `None` for chunk-level records. + pub local_peer_id: Option, + /// Outer file / deferred-retry attempt number (1 = first pass). + pub file_attempt: usize, + /// Chunk index within the file (1-based, matching the progress reports). + pub chunk_index: usize, + /// Hex-encoded chunk address. + pub chunk_address: String, + /// `initial` or `retry` (internal close-group retry sweep). + pub sweep: String, + /// Peer attempt number within the sweep; `None` for chunk-level records + /// (`cache_hit`, `lookup_error`, `exhausted`). + pub peer_attempt: Option, + /// Closest-peer DHT lookup duration, emitted on the first peer attempt + /// associated with that lookup; `None` otherwise. + pub lookup_duration_ms: Option, + /// Process-local identifier shared by attempts from one closest-peer lookup. + pub lookup_correlation_id: Option, + /// One-based ordinal in the DHT-selected peer order. + pub selected_peer_ordinal: Option, + /// Peer selected by the DHT lookup for this attempt; `None` for chunk-level + /// records. + pub expected_peer: Option, + /// Dial addresses selected from the DHT record, in priority order. + pub selected_peer_addresses: Option>, + /// Address-type labels parallel to `selected_peer_addresses`. + pub selected_peer_address_types: Option>, + /// This client's monotonic last-successful-DHT-interaction age. This is + /// local knowledge, not proof of remote uptime. + pub local_last_seen_age_ms: Option, + /// Publisher-clock-derived address-set age. The timestamp is untrusted and + /// is not proof of remote uptime. + pub publisher_address_set_age_ms: Option, + /// Raw publisher wall-clock address-set sequence, when present. + pub publisher_address_set_unix_ns: Option, + /// Authenticated peer that supplied the matching response, from the + /// `ChunkProtocolResponse` metadata; `None` when no response was received + /// (timeout / send failure) or for chunk-level records. + pub source_peer: Option, + /// Transport address that delivered the response, from the + /// `ChunkProtocolResponse` metadata; `None` when no response was received + /// or for chunk-level records. + pub transport_source: Option, + /// `direct`, `relay`, `lan`, `unverified`, or `unknown`, classified from + /// the actual transport source via + /// `P2PNode::classify_peer_transport_route`. `unknown` for chunk-level + /// records with no peer. + pub route: String, + /// Why `route` is `unknown` when that is the case; `None` once a real + /// transport source is classified, and for chunk-level records. + pub route_note: Option, + /// Whether `node.is_peer_connected(peer)` returned `true` when sampled + /// before the send; `None` for chunk-level records. + pub peer_connected_before_request: Option, + /// Number of diagnostics-enabled peer requests active in this process + /// immediately after this request entered the active set. + pub active_requests_at_start: Option, + /// Adaptive fetch concurrency cap snapshot at the time of this record; + /// `None` when diagnostics are disabled (never emitted in that case). + pub fetch_cap: Option, + /// Elapsed time until the complete response was reassembled and + /// delivered; `None` for chunk-level records with no peer attempt. + pub response_elapsed_ms: Option, + /// Time to first byte. Always `null` — see `ttfb_unavailable_reason`. + pub ttfb_ms: Option, + /// Explicitly `false` so complete-response latency is never presented as + /// TTFB. + pub ttfb_available: bool, + /// Why TTFB is unavailable. + pub ttfb_unavailable_reason: String, + /// Valid returned chunk bytes; `0` for non-`found` outcomes. + pub bytes: u64, + /// Attempt outcome. See [`DownloadDiagnosticsOutcome`]. + pub outcome: DownloadDiagnosticsOutcome, + /// Bounded diagnostic error category/detail; no secrets. `None` for + /// successful records. + pub error: Option, +} + +impl DownloadDiagnosticsRecord { + /// The shared TTFB-unavailable reason string used by every record. + pub const TTFB_UNAVAILABLE_REASON: &'static str = + "protocol exposes only a complete-response event; first-byte/first-frame \ + timing is not available"; + + /// The shared route-unknown note used when `classify_peer_transport_route` + /// returns `Unknown`: the transport source was absent (no response) or did + /// not match any known typed peer dial address. + pub const ROUTE_UNKNOWN_NOTE: &'static str = + "transport_source absent or did not match any known typed peer dial address; \ + route could not be classified from the observed response"; + + /// Build a peer-attempt record. `lookup_duration_ms` is attached only when + /// this is the first peer attempt of the sweep (`peer_attempt == 1`). + /// `route_note` should be `Some` only when `route` is `"unknown"`. + #[allow(clippy::too_many_arguments)] + pub(crate) fn peer_attempt( + file_attempt: usize, + chunk_index: usize, + chunk_address: &[u8; 32], + sweep: &'static str, + peer_attempt: usize, + lookup_duration_ms: Option, + lookup_correlation_id: &str, + expected_peer: &str, + selected_peer_addresses: Vec, + selected_peer_address_types: Vec, + local_last_seen_age_ms: Option, + publisher_address_set_age_ms: Option, + publisher_address_set_unix_ns: Option, + source_peer: Option<&str>, + transport_source: Option<&str>, + route: &str, + route_note: Option<&str>, + peer_connected_before_request: Option, + active_requests_at_start: Option, + fetch_cap: Option, + request_started_unix_ms: u64, + request_completed_unix_ms: u64, + correlation: &DownloadRequestCorrelation, + response_elapsed_ms: u64, + bytes: u64, + outcome: DownloadDiagnosticsOutcome, + error: Option, + ) -> Self { + let lookup = if peer_attempt == 1 { + lookup_duration_ms + } else { + None + }; + Self { + schema_version: DIAGNOSTICS_SCHEMA_VERSION, + timestamp: utc_now_rfc3339(), + request_started_unix_ms: Some(request_started_unix_ms), + request_completed_unix_ms: Some(request_completed_unix_ms), + request_id: Some(correlation.request_id), + local_peer_id: Some(correlation.local_peer_id.clone()), + file_attempt, + chunk_index, + chunk_address: hex::encode(chunk_address), + sweep: sweep.to_string(), + peer_attempt: Some(peer_attempt), + lookup_duration_ms: lookup, + lookup_correlation_id: Some(lookup_correlation_id.to_string()), + selected_peer_ordinal: Some(peer_attempt), + expected_peer: Some(expected_peer.to_string()), + selected_peer_addresses: Some(selected_peer_addresses), + selected_peer_address_types: Some(selected_peer_address_types), + local_last_seen_age_ms, + publisher_address_set_age_ms, + publisher_address_set_unix_ns, + source_peer: source_peer.map(str::to_string), + transport_source: transport_source.map(str::to_string), + route: route.to_string(), + route_note: route_note.map(str::to_string), + peer_connected_before_request, + active_requests_at_start, + fetch_cap, + response_elapsed_ms: Some(response_elapsed_ms), + ttfb_ms: None, + ttfb_available: false, + ttfb_unavailable_reason: Self::TTFB_UNAVAILABLE_REASON.to_string(), + bytes, + outcome, + error, + } + } + + /// Build a chunk-level record (no peer attempt): cache hit, lookup error, + /// or exhausted peer set. + #[allow(clippy::too_many_arguments)] + pub fn chunk_level( + file_attempt: usize, + chunk_index: usize, + chunk_address: &[u8; 32], + sweep: &'static str, + fetch_cap: Option, + bytes: u64, + outcome: DownloadDiagnosticsOutcome, + error: Option, + ) -> Self { + Self { + schema_version: DIAGNOSTICS_SCHEMA_VERSION, + timestamp: utc_now_rfc3339(), + request_started_unix_ms: None, + request_completed_unix_ms: None, + request_id: None, + local_peer_id: None, + file_attempt, + chunk_index, + chunk_address: hex::encode(chunk_address), + sweep: sweep.to_string(), + peer_attempt: None, + lookup_duration_ms: None, + lookup_correlation_id: None, + selected_peer_ordinal: None, + expected_peer: None, + selected_peer_addresses: None, + selected_peer_address_types: None, + local_last_seen_age_ms: None, + publisher_address_set_age_ms: None, + publisher_address_set_unix_ns: None, + source_peer: None, + transport_source: None, + route: "unknown".to_string(), + route_note: None, + peer_connected_before_request: None, + active_requests_at_start: None, + fetch_cap, + response_elapsed_ms: None, + ttfb_ms: None, + ttfb_available: false, + ttfb_unavailable_reason: Self::TTFB_UNAVAILABLE_REASON.to_string(), + bytes, + outcome, + error, + } + } +} + +/// A cloneable, bounded sender for diagnostic records. +/// +/// Cloning is cheap (a single `mpsc::Sender` handle). `try_emit` never +/// blocks: when the bounded channel is full the record is dropped, so a slow +/// writer cannot stall the download path. Dropped records are counted and the +/// writer reports the total when it exits. +#[derive(Clone)] +pub struct DownloadDiagnosticsSender { + tx: mpsc::SyncSender, + dropped: Arc, +} + +impl DownloadDiagnosticsSender { + /// Enqueue a record without blocking. Drops the record if the bounded + /// channel is full. + pub fn try_emit(&self, record: DownloadDiagnosticsRecord) { + if self.tx.try_send(record).is_err() { + self.dropped.fetch_add(1, Ordering::Relaxed); + } + } +} + +/// Open `` for writing (truncated), spawn a dedicated OS thread that +/// drains records from a bounded channel and writes one JSON line per record +/// to a buffered writer, flushing on close. +/// +/// The writer runs on a plain OS thread (not a tokio task) so synchronous file +/// writes never block the async runtime. The returned sender is cloneable and +/// can be threaded through the download path; dropping the last clone closes +/// the channel. Joining the returned thread handle waits for the final flush +/// and writer exit. +/// +/// # Errors +/// +/// Returns an error if the file cannot be opened or the writer thread cannot +/// be spawned. +pub fn spawn_download_diagnostics_writer( + path: &Path, +) -> io::Result<(DownloadDiagnosticsSender, std::thread::JoinHandle<()>)> { + let file = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(path)?; + let (tx, rx) = mpsc::sync_channel::(DIAGNOSTICS_CHANNEL_CAPACITY); + let dropped = Arc::new(AtomicU64::new(0)); + let writer_dropped = Arc::clone(&dropped); + let builder = std::thread::Builder::new().name("ant-download-diagnostics-writer".to_string()); + let handle = builder + .spawn(move || { + let mut writer = io::BufWriter::new(file); + for record in rx.iter() { + match serde_json::to_string(&record) { + Ok(line) => { + if let Err(err) = writeln!(writer, "{line}") { + error!(%err, "download diagnostics sidecar write failed"); + break; + } + } + Err(err) => { + // A record that cannot be serialized is skipped rather + // than dropping the whole sidecar; the writer keeps + // draining so later valid records survive. + error!(%err, "download diagnostics record serialization failed"); + continue; + } + } + } + if let Err(err) = writer.flush() { + error!(%err, "download diagnostics sidecar flush failed"); + } + let dropped = writer_dropped.load(Ordering::Relaxed); + if dropped > 0 { + warn!(dropped, "download diagnostics records were dropped"); + } + }) + .map_err(|e| io::Error::other(format!("failed to spawn diagnostics writer thread: {e}")))?; + Ok((DownloadDiagnosticsSender { tx, dropped }, handle)) +} + +/// Bound an error message to [`DIAGNOSTICS_ERROR_MAX_CHARS`] chars, preserving +/// a leading category if one is supplied. +/// +/// `category` is a short stable label (e.g. `"timeout"`); `detail` is the +/// free-form error text that may be truncated. No credentials are added — the +/// caller passes only a bounded diagnostic string. +pub fn bounded_error(category: &str, detail: &str) -> String { + let prefix = if category.is_empty() { + String::new() + } else { + format!("{category}: ") + }; + if prefix.len() + detail.len() <= DIAGNOSTICS_ERROR_MAX_CHARS { + return format!("{prefix}{detail}"); + } + let remaining = DIAGNOSTICS_ERROR_MAX_CHARS.saturating_sub(prefix.len()); + let mut truncated: String = detail.chars().take(remaining.saturating_sub(1)).collect(); + truncated.push('…'); + format!("{prefix}{truncated}") +} + +/// Format the current UTC time as an RFC 3339 string (`YYYY-MM-DDTHH:MM:SSZ`) +/// without a `chrono`/`time` dependency. +fn utc_now_rfc3339() -> String { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default(); + rfc3339_from_unix_secs(now.as_secs()) +} + +/// Current wall-clock Unix time in milliseconds for joining request windows +/// to external fleet telemetry. Saturates if the platform clock representation +/// exceeds `u64`. +pub(crate) fn unix_now_ms() -> u64 { + let millis = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis(); + u64::try_from(millis).unwrap_or(u64::MAX) +} + +/// Convert Unix epoch seconds to an RFC 3339 UTC string. +/// +/// Uses the well-known civil-from-days algorithm (Howard Hinnant). No leap +/// seconds; sufficient precision for a diagnostic timestamp. +fn rfc3339_from_unix_secs(secs: u64) -> String { + let days = (secs / 86_400) as i64; + let secs_of_day = secs % 86_400; + let hour = secs_of_day / 3600; + let minute = (secs_of_day % 3600) / 60; + let second = secs_of_day % 60; + + // Civil date from days since 1970-01-01. + let z = days + 719_468; + let era = if z >= 0 { z } else { z - 146_096 } / 146_097; + let doe = z - era * 146_097; + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let y = yoe + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let m = if mp < 10 { mp + 3 } else { mp - 9 }; + let year = if m <= 2 { y + 1 } else { y }; + + format!("{year:04}-{m:02}-{d:02}T{hour:02}:{minute:02}:{second:02}Z") +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_correlation(request_id: u64) -> DownloadRequestCorrelation { + DownloadRequestCorrelation::new(request_id, &PeerId::from_bytes([42; 32])) + } + + #[test] + fn outcome_as_str_is_stable_lowercase() { + assert_eq!(DownloadDiagnosticsOutcome::Found.as_str(), "found"); + assert_eq!(DownloadDiagnosticsOutcome::NotFound.as_str(), "not_found"); + assert_eq!(DownloadDiagnosticsOutcome::Timeout.as_str(), "timeout"); + assert_eq!( + DownloadDiagnosticsOutcome::NetworkError.as_str(), + "network_error" + ); + assert_eq!( + DownloadDiagnosticsOutcome::ProtocolError.as_str(), + "protocol_error" + ); + assert_eq!(DownloadDiagnosticsOutcome::CacheHit.as_str(), "cache_hit"); + assert_eq!( + DownloadDiagnosticsOutcome::LookupError.as_str(), + "lookup_error" + ); + assert_eq!(DownloadDiagnosticsOutcome::Exhausted.as_str(), "exhausted"); + } + + #[test] + fn outcome_serializes_as_lowercase_string() { + let v = serde_json::to_string(&DownloadDiagnosticsOutcome::NetworkError).unwrap(); + assert_eq!(v, "\"network_error\""); + } + + #[test] + fn peer_attempt_record_pins_field_names_and_null_ttfb() { + let addr = [7u8; 32]; + let correlation = test_correlation(9_001); + let record = DownloadDiagnosticsRecord::peer_attempt( + 1, + 3, + &addr, + "initial", + 1, + Some(42), + "lookup-1", + "peer-abc", + vec!["/ip4/1.2.3.4/udp/9000/quic".to_string()], + vec!["direct".to_string()], + Some(1_500), + Some(2_000), + Some(1_234_000_000), + Some("peer-abc"), + Some("/ip4/1.2.3.4/udp/9000/quic"), + "direct", + None, + Some(true), + Some(8), + Some(8), + 120, + 1024, + &correlation, + 904, + 1024, + DownloadDiagnosticsOutcome::Found, + None, + ); + let json = serde_json::to_value(&record).unwrap(); + let obj = json.as_object().unwrap(); + // Pin field names — schema v4. + for field in [ + "schema_version", + "timestamp", + "request_started_unix_ms", + "request_completed_unix_ms", + "request_id", + "local_peer_id", + "file_attempt", + "chunk_index", + "chunk_address", + "sweep", + "peer_attempt", + "lookup_duration_ms", + "lookup_correlation_id", + "selected_peer_ordinal", + "expected_peer", + "selected_peer_addresses", + "selected_peer_address_types", + "local_last_seen_age_ms", + "publisher_address_set_age_ms", + "publisher_address_set_unix_ns", + "source_peer", + "transport_source", + "route", + "route_note", + "peer_connected_before_request", + "active_requests_at_start", + "fetch_cap", + "response_elapsed_ms", + "ttfb_ms", + "ttfb_available", + "ttfb_unavailable_reason", + "bytes", + "outcome", + "error", + ] { + assert!(obj.contains_key(field), "missing field {field}"); + } + // Explicit unavailable-TTFB representation. + assert_eq!(obj["ttfb_ms"], serde_json::Value::Null); + assert_eq!(obj["ttfb_available"], serde_json::Value::Bool(false)); + assert!( + obj["ttfb_unavailable_reason"] + .as_str() + .unwrap() + .contains("first-byte"), + "ttfb reason must mention first-byte" + ); + // Route classified from actual transport source. + assert_eq!(obj["route"], serde_json::Value::String("direct".into())); + assert_eq!(obj["route_note"], serde_json::Value::Null); + // Actual source_peer and transport_source from response metadata. + assert_eq!(obj["source_peer"], serde_json::json!("peer-abc")); + assert_eq!( + obj["transport_source"], + serde_json::json!("/ip4/1.2.3.4/udp/9000/quic") + ); + // Request bounds, active count, peer state, and fetch cap sampled. + assert_eq!(obj["request_started_unix_ms"], serde_json::json!(120u64)); + assert_eq!(obj["request_completed_unix_ms"], serde_json::json!(1024u64)); + assert_eq!(obj["active_requests_at_start"], serde_json::json!(8usize)); + assert_eq!( + obj["peer_connected_before_request"], + serde_json::json!(true) + ); + assert_eq!(obj["fetch_cap"], serde_json::json!(8usize)); + // First peer attempt carries the lookup duration. + assert_eq!(obj["lookup_duration_ms"], serde_json::json!(42u64)); + assert_eq!(obj["bytes"], serde_json::json!(1024u64)); + assert_eq!(obj["outcome"], serde_json::json!("found")); + assert_eq!(obj["request_id"], serde_json::json!(9_001u64)); + assert_eq!( + obj["local_peer_id"], + serde_json::json!(correlation.local_peer_id) + ); + assert_eq!(obj["schema_version"], serde_json::json!(4u8)); + assert_eq!(obj["lookup_correlation_id"], serde_json::json!("lookup-1")); + assert_eq!(obj["selected_peer_ordinal"], serde_json::json!(1usize)); + assert_eq!(obj["local_last_seen_age_ms"], serde_json::json!(1_500u64)); + assert_eq!( + obj["publisher_address_set_age_ms"], + serde_json::json!(2_000u64) + ); + assert_eq!( + obj["chunk_address"], + serde_json::Value::String(hex::encode(addr)) + ); + } + + #[test] + fn later_peer_attempt_omits_lookup_duration_and_carries_route_note_when_unknown() { + let addr = [9u8; 32]; + let record = DownloadDiagnosticsRecord::peer_attempt( + 1, + 1, + &addr, + "retry", + 3, + Some(10), + "lookup-2", + "peer-x", + vec!["/ip6/2001:db8::1/udp/9000/quic".to_string()], + vec!["unverified".to_string()], + None, + None, + None, + // No response → no source_peer, no transport_source. + None, + None, + "unknown", + Some(DownloadDiagnosticsRecord::ROUTE_UNKNOWN_NOTE), + Some(false), + Some(4), + Some(4), + 500, + 1_000, + &test_correlation(9_002), + 500, + 0, + DownloadDiagnosticsOutcome::Timeout, + Some(bounded_error("timeout", "no response")), + ); + let json = serde_json::to_value(&record).unwrap(); + let obj = json.as_object().unwrap(); + assert_eq!(obj["lookup_duration_ms"], serde_json::Value::Null); + assert_eq!(obj["lookup_correlation_id"], serde_json::json!("lookup-2")); + assert_eq!(obj["selected_peer_ordinal"], serde_json::json!(3usize)); + assert_eq!( + obj["route_note"], + serde_json::json!(DownloadDiagnosticsRecord::ROUTE_UNKNOWN_NOTE) + ); + assert_eq!(obj["source_peer"], serde_json::Value::Null); + assert_eq!(obj["transport_source"], serde_json::Value::Null); + assert_eq!(obj["route"], serde_json::json!("unknown")); + assert_eq!( + obj["peer_connected_before_request"], + serde_json::json!(false) + ); + assert_eq!(obj["active_requests_at_start"], serde_json::json!(4usize)); + assert_eq!(obj["fetch_cap"], serde_json::json!(4usize)); + assert_eq!(obj["outcome"], serde_json::json!("timeout")); + assert_eq!(obj["bytes"], serde_json::json!(0u64)); + assert!(obj["error"].as_str().unwrap().starts_with("timeout: ")); + } + + #[test] + fn cache_hit_record_has_no_peer_fields() { + let addr = [1u8; 32]; + let record = DownloadDiagnosticsRecord::chunk_level( + 1, + 2, + &addr, + "initial", + Some(8), + 4096, + DownloadDiagnosticsOutcome::CacheHit, + None, + ); + let json = serde_json::to_value(&record).unwrap(); + let obj = json.as_object().unwrap(); + assert_eq!(obj["peer_attempt"], serde_json::Value::Null); + assert_eq!(obj["expected_peer"], serde_json::Value::Null); + assert_eq!(obj["source_peer"], serde_json::Value::Null); + assert_eq!(obj["transport_source"], serde_json::Value::Null); + assert_eq!(obj["lookup_duration_ms"], serde_json::Value::Null); + assert_eq!(obj["lookup_correlation_id"], serde_json::Value::Null); + assert_eq!(obj["selected_peer_addresses"], serde_json::Value::Null); + assert_eq!(obj["response_elapsed_ms"], serde_json::Value::Null); + assert_eq!( + obj["peer_connected_before_request"], + serde_json::Value::Null + ); + assert_eq!(obj["request_started_unix_ms"], serde_json::Value::Null); + assert_eq!(obj["request_completed_unix_ms"], serde_json::Value::Null); + assert_eq!(obj["request_id"], serde_json::Value::Null); + assert_eq!(obj["local_peer_id"], serde_json::Value::Null); + assert_eq!(obj["active_requests_at_start"], serde_json::Value::Null); + assert_eq!(obj["fetch_cap"], serde_json::json!(8usize)); + assert_eq!(obj["route"], serde_json::json!("unknown")); + assert_eq!(obj["route_note"], serde_json::Value::Null); + assert_eq!(obj["outcome"], serde_json::json!("cache_hit")); + assert_eq!(obj["bytes"], serde_json::json!(4096u64)); + } + + #[test] + fn exhausted_and_lookup_error_records_classify_correctly() { + let addr = [2u8; 32]; + let exhausted = DownloadDiagnosticsRecord::chunk_level( + 2, + 5, + &addr, + "retry", + Some(2), + 0, + DownloadDiagnosticsOutcome::Exhausted, + None, + ); + assert_eq!( + serde_json::to_value(&exhausted).unwrap()["outcome"], + serde_json::json!("exhausted") + ); + + let lookup_err = DownloadDiagnosticsRecord::chunk_level( + 2, + 5, + &addr, + "initial", + Some(2), + 0, + DownloadDiagnosticsOutcome::LookupError, + Some(bounded_error("lookup", "DHT returned no peers")), + ); + let v = serde_json::to_value(&lookup_err).unwrap(); + assert_eq!(v["outcome"], serde_json::json!("lookup_error")); + assert!(v["error"].as_str().unwrap().starts_with("lookup: ")); + } + + #[test] + fn bounded_error_truncates_long_detail() { + let long = "x".repeat(10_000); + let s = bounded_error("network", &long); + assert!(s.starts_with("network: ")); + // +1 for the ellipsis added on truncation. + assert!(s.chars().count() <= DIAGNOSTICS_ERROR_MAX_CHARS); + assert!(s.ends_with('…')); + } + + #[test] + fn bounded_error_preserves_short_detail_intact() { + let s = bounded_error("protocol", "mismatched address"); + assert_eq!(s, "protocol: mismatched address"); + } + + #[test] + fn rfc3339_formatter_is_valid_for_known_epoch() { + // 2021-01-01T00:00:00Z = 1609459200. + let s = rfc3339_from_unix_secs(1_609_459_200); + assert_eq!(s, "2021-01-01T00:00:00Z"); + // 1970-01-01T00:00:00Z = 0. + assert_eq!(rfc3339_from_unix_secs(0), "1970-01-01T00:00:00Z"); + // Leap-year day: 2024-02-29T00:00:00Z = 1709164800. + assert_eq!( + rfc3339_from_unix_secs(1_709_164_800), + "2024-02-29T00:00:00Z" + ); + } + + #[test] + fn disabled_diagnostics_does_not_create_sidecar() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("disabled.jsonl"); + let diagnostics: Option = None; + + assert!(diagnostics.is_none()); + assert!(!path.exists()); + } + + #[test] + fn writer_emits_one_json_line_per_record_and_flushes_on_drop() { + let dir = std::env::temp_dir(); + let path = dir.join(format!( + "ant-dl-diag-{}-{}.jsonl", + std::process::id(), + std::time::SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let (sender, writer) = spawn_download_diagnostics_writer(&path).unwrap(); + let addr = [3u8; 32]; + sender.try_emit(DownloadDiagnosticsRecord::chunk_level( + 1, + 1, + &addr, + "initial", + Some(8), + 128, + DownloadDiagnosticsOutcome::CacheHit, + None, + )); + sender.try_emit(DownloadDiagnosticsRecord::peer_attempt( + 1, + 1, + &addr, + "initial", + 1, + Some(5), + "lookup-writer", + "peer-z", + vec!["/ip4/1.2.3.4/udp/9000/quic".to_string()], + vec!["direct".to_string()], + Some(50), + Some(100), + Some(1_234_000_000), + Some("peer-z"), + Some("/ip4/1.2.3.4/udp/9000/quic"), + "direct", + None, + Some(true), + Some(8), + Some(8), + 30, + 60, + &test_correlation(9_003), + 30, + 128, + DownloadDiagnosticsOutcome::Found, + None, + )); + // Drop the last sender: the channel closes and the writer flushes. + drop(sender); + writer.join().unwrap(); + let contents = std::fs::read_to_string(&path).unwrap(); + let lines: Vec<&str> = contents.lines().filter(|l| !l.is_empty()).collect(); + assert_eq!( + lines.len(), + 2, + "expected 2 JSONL records, got: {contents:?}" + ); + let first = serde_json::from_str::(lines[0]).unwrap(); + assert_eq!(first["outcome"], serde_json::json!("cache_hit")); + assert_eq!(first["schema_version"], serde_json::json!(4u8)); + let second = serde_json::from_str::(lines[1]).unwrap(); + assert_eq!(second["outcome"], serde_json::json!("found")); + assert_eq!(second["route"], serde_json::json!("direct")); + assert_eq!(second["route_note"], serde_json::Value::Null); + assert_eq!( + second["transport_source"], + serde_json::json!("/ip4/1.2.3.4/udp/9000/quic") + ); + let _ = std::fs::remove_file(&path); + } +} diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index 42bc60dd..cad69036 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -14,8 +14,9 @@ use crate::data::client::adaptive::{observe_op, rebucketed_unordered}; use crate::data::client::batch::{ finalize_batch_payment, PaymentIntent, PreparedChunk, WaveAggregateStats, }; -use crate::data::client::chunk::ChunkPeerGetResult; +use crate::data::client::chunk::{ChunkFetchDiagnostics, ChunkPeerGetResult}; use crate::data::client::classify_error; +use crate::data::client::diagnostics::DownloadDiagnosticsSender; use crate::data::client::merkle::{ chunk_contents_for_upload_addresses, finalize_merkle_batch, merkle_batch_sizes, merkle_billable_leaves, merkle_deferred_retry, merkle_store_with_retry, should_use_merkle, @@ -166,6 +167,10 @@ struct FileDownloadFetchContext { fetched_ref: Arc, progress_ref: Option>, peer_reports: Option>>>, + /// Optional runtime-gated download diagnostics sender. `None` when + /// `--download-diagnostics` was not passed, so the chunk-fetch path + /// skips all record construction and allocation. + diagnostics: Option, } /// Number of chunks per upload wave (matches batch.rs PAYMENT_WAVE_SIZE). @@ -208,7 +213,7 @@ const DOWNLOAD_STREAM_BATCH_BYTES_PER_CHUNK_MULTIPLIER: u64 = 3; /// of a file already live on the network. const ESTIMATE_SAMPLE_CAP: usize = 5; -/// First diagnostic all-peer fetch attempt for a file chunk. +/// First normal-path diagnostic fetch attempt. const FIRST_DIAGNOSTIC_FETCH_ATTEMPT: usize = 1; /// Deferred retry attempt number for retry round 0. @@ -2828,7 +2833,7 @@ impl Client { output: &Path, peer_count: NonZeroUsize, ) -> Result { - self.file_download_with_progress_using_peer_count(data_map, output, None, peer_count.get()) + self.file_download_with_progress_from_closest_peers(data_map, output, None, peer_count) .await } @@ -2854,6 +2859,28 @@ impl Client { output, progress, peer_count.get(), + None, + ) + .await + } + + /// Download a file with progress and optional per-attempt JSONL diagnostics. + /// + /// Passing `None` preserves the standard path without diagnostic records. + pub async fn file_download_with_progress_and_diagnostics_from_closest_peers( + &self, + data_map: &DataMap, + output: &Path, + progress: Option>, + peer_count: NonZeroUsize, + diagnostics: Option, + ) -> Result { + self.file_download_with_progress_using_peer_count( + data_map, + output, + progress, + peer_count.get(), + diagnostics, ) .await } @@ -2885,6 +2912,7 @@ impl Client { progress, peer_count.get(), Some(chunk_reports.clone()), + None, ) .await?; @@ -2964,8 +2992,21 @@ impl Client { } } } else { + // Normal path: early-return after the first peer that has the + // chunk. When diagnostics are enabled we thread a per-chunk + // diagnostics context through so each peer attempt in the sweep + // is recorded; when disabled (`None`) this is a zero-cost pass. + let diag = context.diagnostics.as_ref().map(|sender| { + ChunkFetchDiagnostics::new( + sender, + attempt, + idx + 1, + addr, + self.controller().fetch.current(), + ) + }); match self - .chunk_get_observed_from_closest_peers(&addr, context.peer_count) + .chunk_get_observed_from_closest_peers(&addr, context.peer_count, diag.as_ref()) .await { Ok(Some(chunk)) => Some(chunk.content), @@ -3032,6 +3073,7 @@ impl Client { progress: Option>, peer_count: usize, peer_reports: Option>>>, + diagnostics: Option, mut on_chunk: F, ) -> Result where @@ -3085,7 +3127,9 @@ impl Client { // load-shedding signal for // sustained close-group exhaustion). let chunk = self - .chunk_get_observed_from_closest_peers(&addr, peer_count) + .chunk_get_observed_from_closest_peers( + &addr, peer_count, None, + ) .await .map_err(|e| { self_encryption::Error::Generic(format!( @@ -3144,6 +3188,7 @@ impl Client { let fetched_for_closure = fetched_counter.clone(); let progress_for_closure = progress.clone(); let peer_reports_for_closure = peer_reports.clone(); + let diagnostics_for_closure = diagnostics.clone(); let fetch_limiter_outer = self.controller().fetch.clone(); let usable_memory = usable_memory_bytes(); @@ -3174,15 +3219,16 @@ impl Client { fetched_ref: fetched_for_closure.clone(), progress_ref: progress_for_closure.clone(), peer_reports: peer_reports_for_closure.clone(), + diagnostics: diagnostics_for_closure.clone(), }; let fetch_limiter = fetch_limiter_outer.clone(); tokio::task::block_in_place(|| { handle.block_on(async { - // First pass: try every chunk in the batch. Normal mode - // uses chunk_get_observed (early-return after a found - // peer); diagnostic mode asks every selected closest - // peer and records that sweep before returning bytes. + // First pass: try every chunk in the batch. Both normal + // and diagnostic modes preserve the closest-peer + // early-return path; diagnostics only records the peers + // actually attempted before a chunk is found. // Any missing chunk or transient fetch error is encoded // as Err(hash), so one noisy chunk does not abort the // whole batch before the deferred retry rounds run. @@ -3339,6 +3385,7 @@ impl Client { output, progress, self.config().close_group_size, + None, ) .await } @@ -3354,9 +3401,15 @@ impl Client { output: &Path, progress: Option>, peer_count: usize, + diagnostics: Option, ) -> Result { self.file_download_with_progress_using_peer_count_and_reports( - data_map, output, progress, peer_count, None, + data_map, + output, + progress, + peer_count, + None, + diagnostics, ) .await } @@ -3368,6 +3421,7 @@ impl Client { progress: Option>, peer_count: usize, peer_reports: Option>>>, + diagnostics: Option, ) -> Result { debug!("Downloading file to {}", output.display()); @@ -3383,10 +3437,17 @@ impl Client { let mut file = std::fs::File::create(tmp.path())?; let bytes_written = self - .download_decrypted_chunks(data_map, progress, peer_count, peer_reports, |bytes| { - let r = file.write_all(&bytes).map_err(Error::from); - std::future::ready(r) - }) + .download_decrypted_chunks( + data_map, + progress, + peer_count, + peer_reports, + diagnostics, + |bytes| { + let r = file.write_all(&bytes).map_err(Error::from); + std::future::ready(r) + }, + ) .await?; file.flush()?; drop(file); // close the handle before rename (Windows won't rename an open file) @@ -3425,7 +3486,7 @@ impl Client { progress: Option>, ) -> Result { let peer_count = self.config().close_group_size; - self.download_decrypted_chunks(data_map, progress, peer_count, None, |bytes| { + self.download_decrypted_chunks(data_map, progress, peer_count, None, None, |bytes| { let sink = sink.clone(); async move { sink.send(Ok(bytes)) diff --git a/ant-core/src/data/client/mod.rs b/ant-core/src/data/client/mod.rs index 7a56edff..f2d0ecc6 100644 --- a/ant-core/src/data/client/mod.rs +++ b/ant-core/src/data/client/mod.rs @@ -10,6 +10,7 @@ pub(crate) mod cached_merkle; pub(crate) mod cached_single; pub mod chunk; pub mod data; +pub mod diagnostics; pub mod file; pub mod merkle; pub mod payment; diff --git a/ant-core/src/data/mod.rs b/ant-core/src/data/mod.rs index 07dbccde..139445c3 100644 --- a/ant-core/src/data/mod.rs +++ b/ant-core/src/data/mod.rs @@ -26,6 +26,10 @@ pub use client::batch::{ finalize_batch_payment, PaidChunk, PaymentIntent, PreparedChunk, SingleNodeQuotePayment, }; pub use client::data::DataUploadResult; +pub use client::diagnostics::{ + spawn_download_diagnostics_writer, DownloadDiagnosticsOutcome, DownloadDiagnosticsRecord, + DownloadDiagnosticsSender, +}; pub use client::file::{ CostEstimateConfidence, DownloadEvent, ExternalPaymentInfo, FileChunkPeerReport, FileChunkPeerReportPeer, FileChunkPeerStatus, FileChunkPeerSweepReport, diff --git a/ant-core/src/data/network.rs b/ant-core/src/data/network.rs index 42275452..af9134a2 100644 --- a/ant-core/src/data/network.rs +++ b/ant-core/src/data/network.rs @@ -11,6 +11,19 @@ use ant_protocol::MAX_WIRE_MESSAGE_SIZE; use std::net::SocketAddr; use std::sync::Arc; +/// Read-only DHT context captured for one diagnostics-enabled closest-peer +/// selection. None of these fields influence selection or dialing. +pub(crate) struct ClosestPeerDiagnostics { + pub peer_id: PeerId, + pub addresses: Vec, + pub address_types: Vec, + /// This process's monotonic age since its last successful DHT interaction. + pub local_last_seen_age_ms: Option, + /// Publisher-clock-derived age of the latest address-set publication. + pub publisher_address_set_age_ms: Option, + pub publisher_address_set_unix_ns: Option, +} + /// Network abstraction for the Autonomi client. /// /// Wraps a `P2PNode` providing high-level operations for @@ -131,6 +144,60 @@ impl Network { .collect()) } + /// Find the same peers, in the same order, while capturing read-only DHT + /// context for the explicitly enabled download diagnostics sidecar. + pub(crate) async fn find_closest_peers_with_diagnostics( + &self, + target: &[u8; 32], + count: usize, + ) -> Result> { + let local_peer_id = self.node.peer_id(); + let closest_nodes = self + .node + .dht() + .find_closest_nodes(target, count + 1) + .await + .map_err(|e| Error::Network(format!("DHT closest-nodes lookup failed: {e}")))?; + let now_ns = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + let now_ns = u64::try_from(now_ns).unwrap_or(u64::MAX); + + let mut result = Vec::with_capacity(count); + for node in closest_nodes + .into_iter() + .filter(|node| node.peer_id != *local_peer_id) + .take(count) + { + let publisher_address_set_unix_ns = node.publisher_address_set_unix_ns(); + // A publisher clock may be ahead of ours. In that case, retain the + // raw timestamp but do not misreport its age as zero. + let publisher_address_set_age_ms = publisher_address_set_unix_ns + .and_then(|published| now_ns.checked_sub(published)) + .map(|age_ns| age_ns / 1_000_000); + let local_last_seen_age_ms = self + .node + .peer_last_seen_elapsed(&node.peer_id) + .await + .map(|age| u64::try_from(age.as_millis()).unwrap_or(u64::MAX)); + let address_context = node.address_and_type_labels_by_priority(); + let (addresses, address_types) = address_context + .into_iter() + .map(|(address, label)| (address, label.to_string())) + .unzip(); + result.push(ClosestPeerDiagnostics { + peer_id: node.peer_id, + addresses, + address_types, + local_last_seen_age_ms, + publisher_address_set_age_ms, + publisher_address_set_unix_ns, + }); + } + Ok(result) + } + /// Find a witnessed close-group transcript for a target address. /// /// The underlying DHT method returns the initial client K, each responder's diff --git a/docs/download-diagnostics-design.md b/docs/download-diagnostics-design.md new file mode 100644 index 00000000..0c46a670 --- /dev/null +++ b/docs/download-diagnostics-design.md @@ -0,0 +1,86 @@ +# Normal-path download diagnostics design + +## Status + +Implementation design for the V2-903 investigation. This is temporary, runtime-gated diagnostic instrumentation rather than a change to download policy. + +## Requirement + +Capture one JSON Lines record for each normal-path chunk fetch attempt while preserving the existing early-success behaviour, retry policy, adaptive concurrency, stdout, and default resource use. + +The diagnostic runner is independent of version A/B testing. It should run beside the existing PROD-DL-01 runner in the same region and provider so the network vantage is not changed. + +## Runtime interface + +`ant file download ... --download-diagnostics ` + +- Omitted: the existing path and output remain unchanged; no diagnostic channel or file is created. +- Present: the CLI opens a sidecar JSONL file and passes an optional bounded diagnostics sender through the normal file/chunk download path. +- Diagnostic output never replaces or contaminates normal stdout or `--json` output. +- Records are streamed as attempts finish so a later process or file-download failure does not discard earlier evidence. + +## Record schema + +Each record contains: + +| Field | Meaning | +|---|---| +| `schema_version` | Stable schema discriminator; `4` adds exact node/client request-correlation fields | +| `timestamp` | UTC time when the attempt completed | +| `request_started_unix_ms` / `request_completed_unix_ms` | Exact peer-request wall-clock bounds for joining to retained fleet telemetry and reconstructing overlap | +| `request_id` | The exact protocol request ID allocated by this client and placed on the chunk GET; the serving node records the same value; `null` for chunk-level records | +| `local_peer_id` | This client's PeerId, matching the serving node's `source_peer`; `null` for chunk-level records | +| `file_attempt` | Outer file/deferred-retry attempt number | +| `chunk_index` / `chunk_address` | Chunk identity within the file | +| `sweep` | Initial or internal retry sweep | +| `peer_attempt` | Peer attempt number within the sweep | +| `lookup_duration_ms` | Closest-peer DHT lookup duration; emitted on the first attempt associated with that lookup | +| `lookup_correlation_id` | Process-local ID shared by attempts produced by one closest-peer lookup | +| `selected_peer_ordinal` | One-based position in the unchanged DHT-selected peer order | +| `expected_peer` | Peer selected by the DHT lookup | +| `selected_peer_addresses` / `selected_peer_address_types` | Parallel, priority-ordered advertised addresses and their DHT type labels | +| `local_last_seen_age_ms` | This client's monotonic age since its latest successful DHT interaction with the peer; local knowledge, not remote uptime | +| `publisher_address_set_unix_ns` / `publisher_address_set_age_ms` | Untrusted publisher-clock address-set timestamp and derived age; age is `null` for future-skewed clocks | +| `source_peer` | Peer identified by the received protocol response | +| `transport_source` | Actual response event transport MultiAddr, when available | +| `route` | `direct`, `relay`, `lan`, `unverified`, or `unknown`, classified from the actual transport source against typed DHT addresses | +| `route_note` | Explanation only when the route is `unknown` | +| `peer_connected_before_request` | Connection-state sample immediately before the send | +| `active_requests_at_start` | Process-local diagnostics-enabled peer requests active when this request entered the active set | +| `fetch_cap` | Adaptive fetch-concurrency cap snapshot for the chunk fetch | +| `response_elapsed_ms` | Elapsed time until the complete response was reassembled and delivered | +| `ttfb_ms` | `null` until the protocol exposes a first-byte/first-frame event | +| `ttfb_available` / `ttfb_unavailable_reason` | Explicitly prevents complete-response latency being presented as TTFB | +| `bytes` | Valid returned chunk bytes; otherwise `0` | +| `outcome` | `found`, `not_found`, `timeout`, `network_error`, `protocol_error`, `cache_hit`, `lookup_error`, or `exhausted` | +| `error` | Bounded diagnostic error category/detail; no secrets | + +A cache hit is a chunk-level record without a source peer or transport route. Lookup failures and exhausted peer sets are also explicit records rather than silent gaps. + +## Transport classification + +`ant-protocol` supplies the authenticated response peer and actual +`P2PEvent::Message.transport_source`. `saorsa-core` supplies a small public +route-classification API on `P2PNode`. The client classifies the actual source +against that response peer's typed DHT addresses. It must not infer route type +from address-list position or from the expected destination address. + +## TTFB limitation + +The current protocol event is emitted only after complete message reassembly. Therefore this branch can measure complete-response latency, not true network time-to-first-byte. True TTFB requires a new first-frame/streaming event in `ant-protocol` or the transport layer and is deliberately left unavailable here. + +## Safety and bounds + +- The normal early-return path remains normal; unlike `--all-peers`, diagnostics do not query every close-group peer after success. +- The optional channel is bounded. A slow diagnostic writer must not create unbounded memory growth; its failure is surfaced without altering the fetched data result. +- Error strings are bounded and diagnostics contain no credentials. +- Existing public methods delegate to the diagnostic-capable implementation with diagnostics disabled. + +## Verification + +- Route classification unit tests cover direct, relay, LAN, unverified, and unknown addresses. +- Normal-path tests cover cache hit, first-peer success, retry sweep, and exhausted/error outcomes. +- A disabled-diagnostics regression test confirms the existing path does not allocate or emit records. +- JSON schema/serialization tests pin field names and the explicit unavailable-TTFB representation. +- Run `cargo fmt --check`, focused tests, and `cargo check` in both repositories. +- Independent reviewers check specification compliance, concurrency/back-pressure behaviour, route correctness, and compatibility before either branch is pushed.