From 7c6875d425ba14fe00fce459c9280af5ad0c5f88 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Tue, 11 Aug 2026 18:53:28 +0100 Subject: [PATCH 1/4] feat: add download request diagnostics --- Cargo.lock | 3 +- Cargo.toml | 5 + ant-cli/src/commands/data/file.rs | 58 +- ant-core/src/data/client/chunk.rs | 290 +++++++++- ant-core/src/data/client/diagnostics.rs | 697 ++++++++++++++++++++++++ ant-core/src/data/client/file.rs | 86 ++- ant-core/src/data/client/mod.rs | 1 + ant-core/src/data/mod.rs | 4 + docs/download-diagnostics-design.md | 70 +++ 9 files changed, 1185 insertions(+), 29 deletions(-) create mode 100644 ant-core/src/data/client/diagnostics.rs create mode 100644 docs/download-diagnostics-design.md diff --git a/Cargo.lock b/Cargo.lock index b61d5377..bee17df0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5199,8 +5199,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#008e8c020b8de773e165c99a83ec50a34fbcc5ad" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index f3978b22..51c9bf81 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,8 @@ [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. +[patch.crates-io] +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..51cde961 100644 --- a/ant-core/src/data/client/chunk.rs +++ b/ant-core/src/data/client/chunk.rs @@ -5,6 +5,9 @@ use crate::data::client::adaptive::Outcome; use crate::data::client::batch::{finalize_batch_payment, PreparedChunk}; +use crate::data::client::diagnostics::{ + bounded_error, DownloadDiagnosticsOutcome, DownloadDiagnosticsRecord, DownloadDiagnosticsSender, +}; use crate::data::client::peer_xor_distance; use crate::data::client::Client; use crate::data::error::{Error, Result}; @@ -195,6 +198,140 @@ 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], +} + +impl<'a> ChunkFetchDiagnostics<'a> { + pub(crate) fn new( + sender: &'a DownloadDiagnosticsSender, + file_attempt: usize, + chunk_index: usize, + chunk_address: [u8; 32], + ) -> Self { + Self { + sender, + file_attempt, + chunk_index, + chunk_address, + } + } + + /// 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, + expected_peer: &PeerId, + source_peer: Option<&PeerId>, + 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, + &expected_peer.to_string(), + source_peer.map(|p| p.to_string()).as_deref(), + 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, + 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 +406,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 +414,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 +788,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 +833,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 +893,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,12 +938,35 @@ 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 = match self.closest_peers(address, peer_count).await { + Ok(peers) => peers, + 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); + } + }; + 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 queried = peers.len(); let mut not_found = 0usize; @@ -787,8 +974,33 @@ impl Client { 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 resp_start = Instant::now(); + let result = self.chunk_get_from_peer(address, peer, addrs).await; + let response_elapsed_ms = + u64::try_from(resp_start.elapsed().as_millis()).unwrap_or(u64::MAX); + if let Some(diag) = diag { + let (outcome, bytes, got_response, error) = classify_peer_attempt(&result); + let lookup = if peer_attempt_no == 1 { + lookup_duration_opt + } else { + None + }; + let source_peer = if got_response { Some(peer) } else { None }; + diag.emit_peer_attempt( + sweep, + peer_attempt_no, + lookup, + peer, + source_peer, + response_elapsed_ms, + bytes, + outcome, + error, + ); + } + match result { Ok(Some(chunk)) => { return Ok(CloseGroupOutcome { chunk: Some(chunk), @@ -829,6 +1041,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, @@ -1064,6 +1283,63 @@ mod tests { /// Last byte position in the test XOR distance arrays. const TEST_DISTANCE_TAIL_INDEX: usize = TEST_XORNAME_BYTE_LEN - 1; + #[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..4863caf1 --- /dev/null +++ b/ant-core/src/data/client/diagnostics.rs @@ -0,0 +1,697 @@ +//! 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). +//! +//! # Transport route honesty +//! +//! `ant-protocol`'s `send_and_await_chunk_response` matches the response +//! `P2PEvent::Message` on `topic` and `source` peer but discards the event's +//! `transport_source` (the match arm uses `..`). That field is therefore not +//! surfaced to the client's chunk-fetch path, so this branch cannot classify +//! the route of an *observed* response against the peer's typed DHT +//! addresses. Per the design doc, we record `route: "unknown"` with an +//! explicit `route_note` rather than fabricate a route from the *expected* +//! peer addresses as if it were observed. True route classification requires +//! `send_and_await_chunk_response` (or a sibling helper) to surface +//! `transport_source`, which is a larger `ant-protocol` change left for a +//! follow-up. The pure classifier lives in `saorsa-core` +//! (`classify_peer_route_from_typed`) and is already unit-tested there; it is +//! not re-exported by `ant-protocol` 2.3.1, so `ant-core` cannot name it +//! without breaking the ONE-pin policy. +//! +//! # 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 serde::Serialize; +use tracing::{error, warn}; + +/// Current diagnostic record schema discriminator. +pub const DIAGNOSTICS_SCHEMA_VERSION: u8 = 1; + +/// 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 `1`. + pub schema_version: u8, + /// UTC time the attempt completed, RFC 3339 (`YYYY-MM-DDTHH:MM:SSZ`). + pub timestamp: String, + /// 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, + /// Peer selected by the DHT lookup for this attempt; `None` for chunk-level + /// records. + pub expected_peer: Option, + /// Peer identified by the received protocol response; `None` when no + /// response was received (timeout / send failure) or for chunk-level + /// records. + pub source_peer: Option, + /// Actual response event transport `MultiAddr`. Not surfaced by + /// `send_and_await_chunk_response` in `ant-protocol` 2.3.1, so always + /// `null` in this branch. + pub transport_source: Option, + /// `direct`, `relay`, `lan`, `unverified`, or `unknown`. Always `unknown` + /// in this branch because `transport_source` is not surfaced; see + /// [`DownloadDiagnosticsRecord::route_note`]. + pub route: String, + /// Why `route` is `unknown` when that is the case; `None` once a real + /// transport source is classified. + pub route_note: 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 by every peer-attempt record in this + /// branch. + pub const ROUTE_UNKNOWN_NOTE: &'static str = + "transport_source is not surfaced by ant-protocol \ + send_and_await_chunk_response; route cannot be classified from an \ + observed response and is not inferred from expected peer addresses"; + + /// Build a peer-attempt record. `lookup_duration_ms` is attached only when + /// this is the first peer attempt of the sweep (`peer_attempt == 1`). + #[allow(clippy::too_many_arguments)] + pub fn peer_attempt( + file_attempt: usize, + chunk_index: usize, + chunk_address: &[u8; 32], + sweep: &'static str, + peer_attempt: usize, + lookup_duration_ms: Option, + expected_peer: &str, + source_peer: Option<&str>, + 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(), + file_attempt, + chunk_index, + chunk_address: hex::encode(chunk_address), + sweep: sweep.to_string(), + peer_attempt: Some(peer_attempt), + lookup_duration_ms: lookup, + expected_peer: Some(expected_peer.to_string()), + source_peer: source_peer.map(str::to_string), + transport_source: None, + route: "unknown".to_string(), + route_note: Some(Self::ROUTE_UNKNOWN_NOTE.to_string()), + 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. + pub fn chunk_level( + file_attempt: usize, + chunk_index: usize, + chunk_address: &[u8; 32], + sweep: &'static str, + bytes: u64, + outcome: DownloadDiagnosticsOutcome, + error: Option, + ) -> Self { + Self { + schema_version: DIAGNOSTICS_SCHEMA_VERSION, + timestamp: utc_now_rfc3339(), + file_attempt, + chunk_index, + chunk_address: hex::encode(chunk_address), + sweep: sweep.to_string(), + peer_attempt: None, + lookup_duration_ms: None, + expected_peer: None, + source_peer: None, + transport_source: None, + route: "unknown".to_string(), + route_note: None, + 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()) +} + +/// 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::*; + + #[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 record = DownloadDiagnosticsRecord::peer_attempt( + 1, + 3, + &addr, + "initial", + 1, + Some(42), + "peer-abc", + Some("peer-abc"), + 120, + 1024, + DownloadDiagnosticsOutcome::Found, + None, + ); + let json = serde_json::to_value(&record).unwrap(); + let obj = json.as_object().unwrap(); + // Pin field names. + for field in [ + "schema_version", + "timestamp", + "file_attempt", + "chunk_index", + "chunk_address", + "sweep", + "peer_attempt", + "lookup_duration_ms", + "expected_peer", + "source_peer", + "transport_source", + "route", + "route_note", + "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 honesty. + assert_eq!(obj["route"], serde_json::Value::String("unknown".into())); + assert!( + obj["route_note"].as_str().is_some(), + "first peer attempt must carry a route_note" + ); + // 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")); + // transport_source is null (not surfaced). + assert_eq!(obj["transport_source"], serde_json::Value::Null); + assert_eq!(obj["schema_version"], serde_json::json!(1u8)); + assert_eq!( + obj["chunk_address"], + serde_json::Value::String(hex::encode(addr)) + ); + } + + #[test] + fn later_peer_attempt_omits_lookup_duration_but_explains_unknown_route() { + let addr = [9u8; 32]; + let record = DownloadDiagnosticsRecord::peer_attempt( + 1, + 1, + &addr, + "retry", + 3, + Some(10), + "peer-x", + None, + 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["route_note"], + serde_json::json!(DownloadDiagnosticsRecord::ROUTE_UNKNOWN_NOTE) + ); + assert_eq!(obj["source_peer"], serde_json::Value::Null); + 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", + 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["lookup_duration_ms"], serde_json::Value::Null); + assert_eq!(obj["response_elapsed_ms"], 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", + 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", + 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", + 128, + DownloadDiagnosticsOutcome::CacheHit, + None, + )); + sender.try_emit(DownloadDiagnosticsRecord::peer_attempt( + 1, + 1, + &addr, + "initial", + 1, + Some(5), + "peer-z", + Some("peer-z"), + 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")); + let second = serde_json::from_str::(lines[1]).unwrap(); + assert_eq!(second["outcome"], serde_json::json!("found")); + assert_eq!(second["route"], serde_json::json!("unknown")); + 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..dc20507e 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,16 @@ 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)); 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 +3068,7 @@ impl Client { progress: Option>, peer_count: usize, peer_reports: Option>>>, + diagnostics: Option, mut on_chunk: F, ) -> Result where @@ -3085,7 +3122,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 +3183,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 +3214,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 +3380,7 @@ impl Client { output, progress, self.config().close_group_size, + None, ) .await } @@ -3354,9 +3396,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 +3416,7 @@ impl Client { progress: Option>, peer_count: usize, peer_reports: Option>>>, + diagnostics: Option, ) -> Result { debug!("Downloading file to {}", output.display()); @@ -3383,10 +3432,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 +3481,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/docs/download-diagnostics-design.md b/docs/download-diagnostics-design.md new file mode 100644 index 00000000..eecd8787 --- /dev/null +++ b/docs/download-diagnostics-design.md @@ -0,0 +1,70 @@ +# 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, initially `1` | +| `timestamp` | UTC time when the attempt completed | +| `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 | +| `expected_peer` | Peer selected by the DHT lookup | +| `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 | +| `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 + +`saorsa-core` supplies a small public route-classification API on `P2PNode`. It classifies the actual `P2PEvent::Message.transport_source` against the 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. From d778ca948a96a0e68cc2b5d6dfe9fc28db453500 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Wed, 12 Aug 2026 19:47:55 +0100 Subject: [PATCH 2/4] feat: extend V2-903 download diagnostics --- Cargo.lock | 5 +- Cargo.toml | 5 + ant-core/src/data/client/chunk.rs | 181 ++++++++++++++++++-- ant-core/src/data/client/diagnostics.rs | 210 ++++++++++++++++++------ ant-core/src/data/client/file.rs | 13 +- docs/download-diagnostics-design.md | 13 +- 6 files changed, 358 insertions(+), 69 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bee17df0..cae347af 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]] diff --git a/Cargo.toml b/Cargo.toml index 51c9bf81..1af2f346 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,5 +4,10 @@ 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-core/src/data/client/chunk.rs b/ant-core/src/data/client/chunk.rs index 51cde961..5d5be166 100644 --- a/ant-core/src/data/client/chunk.rs +++ b/ant-core/src/data/client/chunk.rs @@ -6,27 +6,49 @@ use crate::data::client::adaptive::Outcome; use crate::data::client::batch::{finalize_batch_payment, PreparedChunk}; use crate::data::client::diagnostics::{ - bounded_error, DownloadDiagnosticsOutcome, DownloadDiagnosticsRecord, DownloadDiagnosticsSender, + bounded_error, unix_now_ms, DownloadDiagnosticsOutcome, DownloadDiagnosticsRecord, + DownloadDiagnosticsSender, }; use crate::data::client::peer_xor_distance; use crate::data::client::Client; use crate::data::error::{Error, Result}; 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); + +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); + } +} + /// 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 @@ -210,6 +232,7 @@ pub(crate) struct ChunkFetchDiagnostics<'a> { file_attempt: usize, chunk_index: usize, chunk_address: [u8; 32], + fetch_cap: usize, } impl<'a> ChunkFetchDiagnostics<'a> { @@ -218,12 +241,14 @@ impl<'a> ChunkFetchDiagnostics<'a> { file_attempt: usize, chunk_index: usize, chunk_address: [u8; 32], + fetch_cap: usize, ) -> Self { Self { sender, file_attempt, chunk_index, chunk_address, + fetch_cap, } } @@ -237,6 +262,12 @@ impl<'a> ChunkFetchDiagnostics<'a> { lookup_duration_ms: Option, 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, response_elapsed_ms: u64, bytes: u64, outcome: DownloadDiagnosticsOutcome, @@ -251,7 +282,16 @@ impl<'a> ChunkFetchDiagnostics<'a> { peer_attempt, lookup_duration_ms, &expected_peer.to_string(), - source_peer.map(|p| p.to_string()).as_deref(), + 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, response_elapsed_ms, bytes, outcome, @@ -272,6 +312,7 @@ impl<'a> ChunkFetchDiagnostics<'a> { self.chunk_index, &self.chunk_address, sweep, + Some(self.fetch_cap), bytes, outcome, error, @@ -976,30 +1017,69 @@ impl Client { for (peer_attempt, (peer, addrs)) in peers.iter().enumerate() { let peer_attempt_no = peer_attempt + 1; - let resp_start = Instant::now(); - let result = self.chunk_get_from_peer(address, peer, addrs).await; - let response_elapsed_ms = - u64::try_from(resp_start.elapsed().as_millis()).unwrap_or(u64::MAX); - if let Some(diag) = diag { - let (outcome, bytes, got_response, error) = classify_peer_attempt(&result); + let result = if let Some(diag) = diag { + 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 observed = self + .chunk_get_from_peer_with_metadata(address, peer, addrs) + .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 }; - let source_peer = if got_response { Some(peer) } else { None }; diag.emit_peer_attempt( sweep, peer_attempt_no, lookup, peer, - source_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, 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 { @@ -1222,6 +1302,79 @@ 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], + ) -> Result, Error>> { + let node = self.network().node(); + let request_id = self.next_request_id(); + let request = ChunkGetRequest::new(*address); + let message = ChunkMessage { + request_id, + body: ChunkMessageBody::GetRequest(request), + }; + let message_bytes = message + .encode() + .map_err(|e| Error::Protocol(format!("Failed to encode GET request: {e}")))?; + + 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, + 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 diff --git a/ant-core/src/data/client/diagnostics.rs b/ant-core/src/data/client/diagnostics.rs index 4863caf1..535a3429 100644 --- a/ant-core/src/data/client/diagnostics.rs +++ b/ant-core/src/data/client/diagnostics.rs @@ -11,22 +11,22 @@ //! threaded through the file/chunk download path is `None`, so record //! construction is skipped entirely (no allocation, no I/O). //! -//! # Transport route honesty +//! # Schema v2: transport route classification //! -//! `ant-protocol`'s `send_and_await_chunk_response` matches the response -//! `P2PEvent::Message` on `topic` and `source` peer but discards the event's -//! `transport_source` (the match arm uses `..`). That field is therefore not -//! surfaced to the client's chunk-fetch path, so this branch cannot classify -//! the route of an *observed* response against the peer's typed DHT -//! addresses. Per the design doc, we record `route: "unknown"` with an -//! explicit `route_note` rather than fabricate a route from the *expected* -//! peer addresses as if it were observed. True route classification requires -//! `send_and_await_chunk_response` (or a sibling helper) to surface -//! `transport_source`, which is a larger `ant-protocol` change left for a -//! follow-up. The pure classifier lives in `saorsa-core` -//! (`classify_peer_route_from_typed`) and is already unit-tested there; it is -//! not re-exported by `ant-protocol` 2.3.1, so `ant-core` cannot name it -//! without breaking the ONE-pin policy. +//! 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 //! @@ -48,7 +48,7 @@ use serde::Serialize; use tracing::{error, warn}; /// Current diagnostic record schema discriminator. -pub const DIAGNOSTICS_SCHEMA_VERSION: u8 = 1; +pub const DIAGNOSTICS_SCHEMA_VERSION: u8 = 2; /// Bounded capacity of the diagnostics channel. A slow writer must not create /// unbounded memory growth: when full, further records are dropped (counted @@ -127,10 +127,17 @@ impl Serialize for DownloadDiagnosticsOutcome { /// `null` for a cache hit, which has no peer). #[derive(Debug, Clone, Serialize, PartialEq, Eq)] pub struct DownloadDiagnosticsRecord { - /// Stable schema discriminator, currently `1`. + /// Stable schema discriminator, currently `2`. 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, /// Outer file / deferred-retry attempt number (1 = first pass). pub file_attempt: usize, /// Chunk index within the file (1-based, matching the progress reports). @@ -148,21 +155,31 @@ pub struct DownloadDiagnosticsRecord { /// Peer selected by the DHT lookup for this attempt; `None` for chunk-level /// records. pub expected_peer: Option, - /// Peer identified by the received protocol response; `None` when no - /// response was received (timeout / send failure) or for chunk-level - /// records. + /// 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, - /// Actual response event transport `MultiAddr`. Not surfaced by - /// `send_and_await_chunk_response` in `ant-protocol` 2.3.1, so always - /// `null` in this branch. + /// 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`. Always `unknown` - /// in this branch because `transport_source` is not surfaced; see - /// [`DownloadDiagnosticsRecord::route_note`]. + /// `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. + /// 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, @@ -188,15 +205,16 @@ impl DownloadDiagnosticsRecord { "protocol exposes only a complete-response event; first-byte/first-frame \ timing is not available"; - /// The shared route-unknown note used by every peer-attempt record in this - /// branch. + /// 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 is not surfaced by ant-protocol \ - send_and_await_chunk_response; route cannot be classified from an \ - observed response and is not inferred from expected peer addresses"; + "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 fn peer_attempt( file_attempt: usize, @@ -207,6 +225,14 @@ impl DownloadDiagnosticsRecord { lookup_duration_ms: Option, expected_peer: &str, 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, response_elapsed_ms: u64, bytes: u64, outcome: DownloadDiagnosticsOutcome, @@ -220,6 +246,8 @@ impl DownloadDiagnosticsRecord { 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), file_attempt, chunk_index, chunk_address: hex::encode(chunk_address), @@ -228,9 +256,12 @@ impl DownloadDiagnosticsRecord { lookup_duration_ms: lookup, expected_peer: Some(expected_peer.to_string()), source_peer: source_peer.map(str::to_string), - transport_source: None, - route: "unknown".to_string(), - route_note: Some(Self::ROUTE_UNKNOWN_NOTE.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, @@ -243,11 +274,13 @@ impl DownloadDiagnosticsRecord { /// 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, @@ -255,6 +288,8 @@ impl DownloadDiagnosticsRecord { Self { schema_version: DIAGNOSTICS_SCHEMA_VERSION, timestamp: utc_now_rfc3339(), + request_started_unix_ms: None, + request_completed_unix_ms: None, file_attempt, chunk_index, chunk_address: hex::encode(chunk_address), @@ -266,6 +301,9 @@ impl DownloadDiagnosticsRecord { 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, @@ -387,6 +425,17 @@ fn utc_now_rfc3339() -> String { 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 @@ -456,17 +505,27 @@ mod tests { Some(42), "peer-abc", Some("peer-abc"), + Some("/ip4/1.2.3.4/udp/9000/quic"), + "direct", + None, + Some(true), + Some(8), + Some(8), 120, 1024, + 904, + 1024, DownloadDiagnosticsOutcome::Found, None, ); let json = serde_json::to_value(&record).unwrap(); let obj = json.as_object().unwrap(); - // Pin field names. + // Pin field names — schema v2. for field in [ "schema_version", "timestamp", + "request_started_unix_ms", + "request_completed_unix_ms", "file_attempt", "chunk_index", "chunk_address", @@ -478,6 +537,9 @@ mod tests { "transport_source", "route", "route_note", + "peer_connected_before_request", + "active_requests_at_start", + "fetch_cap", "response_elapsed_ms", "ttfb_ms", "ttfb_available", @@ -498,19 +560,29 @@ mod tests { .contains("first-byte"), "ttfb reason must mention first-byte" ); - // Route honesty. - assert_eq!(obj["route"], serde_json::Value::String("unknown".into())); - assert!( - obj["route_note"].as_str().is_some(), - "first peer attempt must carry a route_note" + // 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")); - // transport_source is null (not surfaced). - assert_eq!(obj["transport_source"], serde_json::Value::Null); - assert_eq!(obj["schema_version"], serde_json::json!(1u8)); + assert_eq!(obj["schema_version"], serde_json::json!(2u8)); assert_eq!( obj["chunk_address"], serde_json::Value::String(hex::encode(addr)) @@ -518,7 +590,7 @@ mod tests { } #[test] - fn later_peer_attempt_omits_lookup_duration_but_explains_unknown_route() { + fn later_peer_attempt_omits_lookup_duration_and_carries_route_note_when_unknown() { let addr = [9u8; 32]; let record = DownloadDiagnosticsRecord::peer_attempt( 1, @@ -528,7 +600,16 @@ mod tests { 3, Some(10), "peer-x", + // 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, 500, 0, DownloadDiagnosticsOutcome::Timeout, @@ -542,6 +623,14 @@ mod tests { 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: ")); @@ -555,6 +644,7 @@ mod tests { 2, &addr, "initial", + Some(8), 4096, DownloadDiagnosticsOutcome::CacheHit, None, @@ -564,8 +654,19 @@ mod tests { 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["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["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)); } @@ -578,6 +679,7 @@ mod tests { 5, &addr, "retry", + Some(2), 0, DownloadDiagnosticsOutcome::Exhausted, None, @@ -592,6 +694,7 @@ mod tests { 5, &addr, "initial", + Some(2), 0, DownloadDiagnosticsOutcome::LookupError, Some(bounded_error("lookup", "DHT returned no peers")), @@ -659,6 +762,7 @@ mod tests { 1, &addr, "initial", + Some(8), 128, DownloadDiagnosticsOutcome::CacheHit, None, @@ -672,6 +776,14 @@ mod tests { Some(5), "peer-z", Some("peer-z"), + Some("/ip4/1.2.3.4/udp/9000/quic"), + "direct", + None, + Some(true), + Some(8), + Some(8), + 30, + 60, 30, 128, DownloadDiagnosticsOutcome::Found, @@ -689,9 +801,15 @@ mod tests { ); 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!(2u8)); let second = serde_json::from_str::(lines[1]).unwrap(); assert_eq!(second["outcome"], serde_json::json!("found")); - assert_eq!(second["route"], serde_json::json!("unknown")); + 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 dc20507e..cad69036 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -2996,10 +2996,15 @@ impl Client { // 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)); + 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, diag.as_ref()) .await diff --git a/docs/download-diagnostics-design.md b/docs/download-diagnostics-design.md index eecd8787..949c3b8f 100644 --- a/docs/download-diagnostics-design.md +++ b/docs/download-diagnostics-design.md @@ -25,8 +25,9 @@ Each record contains: | Field | Meaning | |---|---| -| `schema_version` | Stable schema discriminator, initially `1` | +| `schema_version` | Stable schema discriminator; `2` for transport metadata and correlation timestamps | | `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 | | `file_attempt` | Outer file/deferred-retry attempt number | | `chunk_index` / `chunk_address` | Chunk identity within the file | | `sweep` | Initial or internal retry sweep | @@ -36,6 +37,10 @@ Each record contains: | `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 | @@ -47,7 +52,11 @@ A cache hit is a chunk-level record without a source peer or transport route. Lo ## Transport classification -`saorsa-core` supplies a small public route-classification API on `P2PNode`. It classifies the actual `P2PEvent::Message.transport_source` against the peer's typed DHT addresses. It must not infer route type from address-list position or from the expected destination address. +`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 From dfd5c674f6ae85ce122e4460c6a523af87a209f0 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 14 Aug 2026 00:15:58 +0100 Subject: [PATCH 3/4] feat: add selected DHT context to download diagnostics --- Cargo.lock | 2 +- ant-core/src/data/client/chunk.rs | 77 +++++++++++++++++++---- ant-core/src/data/client/diagnostics.rs | 84 +++++++++++++++++++++++-- ant-core/src/data/network.rs | 67 ++++++++++++++++++++ docs/download-diagnostics-design.md | 7 ++- 5 files changed, 218 insertions(+), 19 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cae347af..f422f3cd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5198,7 +5198,7 @@ dependencies = [ [[package]] name = "saorsa-core" version = "0.26.4" -source = "git+https://github.com/WithAutonomi/saorsa-core.git?branch=diagnostics%2Fv2-903-peer-route-classification#008e8c020b8de773e165c99a83ec50a34fbcc5ad" +source = "git+https://github.com/WithAutonomi/saorsa-core.git?branch=diagnostics%2Fv2-903-peer-route-classification#6be87a81f91a119d0761f9ce6f8ee69dec19edcc" dependencies = [ "anyhow", "async-trait", diff --git a/ant-core/src/data/client/chunk.rs b/ant-core/src/data/client/chunk.rs index 5d5be166..3e234c2d 100644 --- a/ant-core/src/data/client/chunk.rs +++ b/ant-core/src/data/client/chunk.rs @@ -12,6 +12,7 @@ use crate::data::client::diagnostics::{ 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, PeerRouteKind}; use ant_protocol::{ @@ -33,6 +34,7 @@ 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; @@ -260,6 +262,8 @@ impl<'a> ChunkFetchDiagnostics<'a> { 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>, @@ -281,7 +285,17 @@ impl<'a> ChunkFetchDiagnostics<'a> { 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(), @@ -991,24 +1005,50 @@ impl Client { sweep: &'static str, ) -> Result { let lookup_start = Instant::now(); - let peers = match self.closest_peers(address, peer_count).await { - Ok(peers) => peers, - Err(e) => { - if let Some(diag) = diag { - diag.emit_chunk_level( - sweep, - 0, - DownloadDiagnosticsOutcome::LookupError, - Some(bounded_error("lookup", &e.to_string())), - ); + 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); } - 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; @@ -1018,6 +1058,19 @@ impl Client { 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) = @@ -1061,6 +1114,8 @@ impl Client { sweep, peer_attempt_no, lookup, + lookup_correlation_id, + peer_context, peer, source_peer.as_ref(), transport_source.as_ref(), diff --git a/ant-core/src/data/client/diagnostics.rs b/ant-core/src/data/client/diagnostics.rs index 535a3429..5d9b1427 100644 --- a/ant-core/src/data/client/diagnostics.rs +++ b/ant-core/src/data/client/diagnostics.rs @@ -11,7 +11,7 @@ //! threaded through the file/chunk download path is `None`, so record //! construction is skipped entirely (no allocation, no I/O). //! -//! # Schema v2: transport route classification +//! # Schema v3: transport route and selected DHT-record context //! //! The `ant-protocol` diagnostic branch //! `diagnostics/v2-903-response-transport-metadata` exposes @@ -48,7 +48,7 @@ use serde::Serialize; use tracing::{error, warn}; /// Current diagnostic record schema discriminator. -pub const DIAGNOSTICS_SCHEMA_VERSION: u8 = 2; +pub const DIAGNOSTICS_SCHEMA_VERSION: u8 = 3; /// Bounded capacity of the diagnostics channel. A slow writer must not create /// unbounded memory growth: when full, further records are dropped (counted @@ -127,7 +127,7 @@ impl Serialize for DownloadDiagnosticsOutcome { /// `null` for a cache hit, which has no peer). #[derive(Debug, Clone, Serialize, PartialEq, Eq)] pub struct DownloadDiagnosticsRecord { - /// Stable schema discriminator, currently `2`. + /// Stable schema discriminator, currently `3`. pub schema_version: u8, /// UTC time the attempt completed, RFC 3339 (`YYYY-MM-DDTHH:MM:SSZ`). pub timestamp: String, @@ -152,9 +152,25 @@ pub struct DownloadDiagnosticsRecord { /// 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. @@ -223,7 +239,13 @@ impl DownloadDiagnosticsRecord { 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, @@ -254,7 +276,14 @@ impl DownloadDiagnosticsRecord { 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(), @@ -296,7 +325,14 @@ impl DownloadDiagnosticsRecord { 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(), @@ -503,7 +539,13 @@ mod tests { "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", @@ -520,7 +562,7 @@ mod tests { ); let json = serde_json::to_value(&record).unwrap(); let obj = json.as_object().unwrap(); - // Pin field names — schema v2. + // Pin field names — schema v3. for field in [ "schema_version", "timestamp", @@ -532,7 +574,14 @@ mod tests { "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", @@ -582,7 +631,14 @@ mod tests { 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["schema_version"], serde_json::json!(2u8)); + assert_eq!(obj["schema_version"], serde_json::json!(3u8)); + 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)) @@ -599,7 +655,13 @@ mod tests { "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, @@ -618,6 +680,8 @@ mod tests { 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) @@ -656,6 +720,8 @@ mod tests { 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"], @@ -774,7 +840,13 @@ mod tests { "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", @@ -801,7 +873,7 @@ mod tests { ); 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!(2u8)); + assert_eq!(first["schema_version"], serde_json::json!(3u8)); 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")); 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 index 949c3b8f..ccbf4893 100644 --- a/docs/download-diagnostics-design.md +++ b/docs/download-diagnostics-design.md @@ -25,7 +25,7 @@ Each record contains: | Field | Meaning | |---|---| -| `schema_version` | Stable schema discriminator; `2` for transport metadata and correlation timestamps | +| `schema_version` | Stable schema discriminator; `3` adds selected DHT-record context | | `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 | | `file_attempt` | Outer file/deferred-retry attempt number | @@ -33,7 +33,12 @@ Each record contains: | `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 | From 7776e74f719beeada710404468d4e231e3c81557 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Tue, 1 Sep 2026 00:10:09 +0100 Subject: [PATCH 4/4] feat: correlate download diagnostics with node GETs --- ant-core/src/data/client/chunk.rs | 79 +++++++++++++++++++++---- ant-core/src/data/client/diagnostics.rs | 67 ++++++++++++++++++--- docs/download-diagnostics-design.md | 4 +- 3 files changed, 130 insertions(+), 20 deletions(-) diff --git a/ant-core/src/data/client/chunk.rs b/ant-core/src/data/client/chunk.rs index 3e234c2d..93c7f3c9 100644 --- a/ant-core/src/data/client/chunk.rs +++ b/ant-core/src/data/client/chunk.rs @@ -7,7 +7,7 @@ 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, + DownloadDiagnosticsSender, DownloadRequestCorrelation, }; use crate::data::client::peer_xor_distance; use crate::data::client::Client; @@ -51,6 +51,18 @@ impl Drop for ActiveDiagnosticRequestGuard { } } +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 @@ -272,6 +284,7 @@ impl<'a> ChunkFetchDiagnostics<'a> { 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, @@ -306,6 +319,7 @@ impl<'a> ChunkFetchDiagnostics<'a> { Some(self.fetch_cap), request_started_unix_ms, request_completed_unix_ms, + correlation, response_elapsed_ms, bytes, outcome, @@ -1077,8 +1091,10 @@ impl Client { 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) + .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); @@ -1124,6 +1140,7 @@ impl Client { active_requests_at_start, request_started_unix_ms, request_completed_unix_ms, + &correlation, response_elapsed_ms, bytes, outcome, @@ -1366,17 +1383,10 @@ impl Client { address: &XorName, peer: &PeerId, peer_addrs: &[MultiAddr], + correlation: &DownloadRequestCorrelation, ) -> Result, Error>> { let node = self.network().node(); - let request_id = self.next_request_id(); - let request = ChunkGetRequest::new(*address); - let message = ChunkMessage { - request_id, - body: ChunkMessageBody::GetRequest(request), - }; - let message_bytes = message - .encode() - .map_err(|e| Error::Protocol(format!("Failed to encode GET request: {e}")))?; + 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); @@ -1386,7 +1396,7 @@ impl Client { node, peer, message_bytes, - request_id, + correlation.request_id, timeout, peer_addrs, |body| match body { @@ -1491,6 +1501,51 @@ 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")); diff --git a/ant-core/src/data/client/diagnostics.rs b/ant-core/src/data/client/diagnostics.rs index 5d9b1427..d2a6b391 100644 --- a/ant-core/src/data/client/diagnostics.rs +++ b/ant-core/src/data/client/diagnostics.rs @@ -11,7 +11,13 @@ //! threaded through the file/chunk download path is `None`, so record //! construction is skipped entirely (no allocation, no I/O). //! -//! # Schema v3: transport route and selected DHT-record context +//! # 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 @@ -44,11 +50,29 @@ 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 = 3; +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 @@ -127,7 +151,7 @@ impl Serialize for DownloadDiagnosticsOutcome { /// `null` for a cache hit, which has no peer). #[derive(Debug, Clone, Serialize, PartialEq, Eq)] pub struct DownloadDiagnosticsRecord { - /// Stable schema discriminator, currently `3`. + /// Stable schema discriminator, currently `4`. pub schema_version: u8, /// UTC time the attempt completed, RFC 3339 (`YYYY-MM-DDTHH:MM:SSZ`). pub timestamp: String, @@ -138,6 +162,13 @@ pub struct DownloadDiagnosticsRecord { /// 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). @@ -232,7 +263,7 @@ impl DownloadDiagnosticsRecord { /// 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 fn peer_attempt( + pub(crate) fn peer_attempt( file_attempt: usize, chunk_index: usize, chunk_address: &[u8; 32], @@ -255,6 +286,7 @@ impl DownloadDiagnosticsRecord { fetch_cap: Option, request_started_unix_ms: u64, request_completed_unix_ms: u64, + correlation: &DownloadRequestCorrelation, response_elapsed_ms: u64, bytes: u64, outcome: DownloadDiagnosticsOutcome, @@ -270,6 +302,8 @@ impl DownloadDiagnosticsRecord { 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), @@ -319,6 +353,8 @@ impl DownloadDiagnosticsRecord { 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), @@ -502,6 +538,10 @@ fn rfc3339_from_unix_secs(secs: u64) -> String { 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"); @@ -532,6 +572,7 @@ mod tests { #[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, @@ -555,6 +596,7 @@ mod tests { Some(8), 120, 1024, + &correlation, 904, 1024, DownloadDiagnosticsOutcome::Found, @@ -562,12 +604,14 @@ mod tests { ); let json = serde_json::to_value(&record).unwrap(); let obj = json.as_object().unwrap(); - // Pin field names — schema v3. + // 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", @@ -631,7 +675,12 @@ mod tests { 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["schema_version"], serde_json::json!(3u8)); + 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)); @@ -672,6 +721,7 @@ mod tests { Some(4), 500, 1_000, + &test_correlation(9_002), 500, 0, DownloadDiagnosticsOutcome::Timeout, @@ -729,6 +779,8 @@ mod tests { ); 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")); @@ -856,6 +908,7 @@ mod tests { Some(8), 30, 60, + &test_correlation(9_003), 30, 128, DownloadDiagnosticsOutcome::Found, @@ -873,7 +926,7 @@ mod tests { ); 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!(3u8)); + 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")); diff --git a/docs/download-diagnostics-design.md b/docs/download-diagnostics-design.md index ccbf4893..0c46a670 100644 --- a/docs/download-diagnostics-design.md +++ b/docs/download-diagnostics-design.md @@ -25,9 +25,11 @@ Each record contains: | Field | Meaning | |---|---| -| `schema_version` | Stable schema discriminator; `3` adds selected DHT-record context | +| `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 |