diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index 0e5a814..c8425f7 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -2400,7 +2400,13 @@ async fn run_peer_network(node: Arc) -> Result<(), String> { // and find_providers finds nobody. This forwards every PoolEvent (add → insert, remove → // evict) into routing so cross-node discovery works as the pool fills. if let Some(dht) = dht.clone() { - spawn_dht_routing_feed(dht, handle.clone()); + spawn_dht_routing_feed(dht.clone(), handle.clone()); + // 4a-ii. Publish this node's local inventory into the DHT — in the BACKGROUND, and only now + // that routing is being fed from the live pool (#1974). Two reasons for this position: + // bring-up must not wait on it (the listener bind is still ahead of us), and every + // announce run before the feed exists queries the empty table bootstrap left behind, + // so it times out instead of reaching the peers that would store the record. + crate::dht::spawn_initial_inventory_announce(dht); } // 4b. Bring up the P2P CONTENT engine (#164/#165) over the live DHT + this node's mTLS identity: the @@ -2666,12 +2672,20 @@ async fn bring_up_dht( tracing::debug!(error = %e, "DHT bootstrap found no peers yet; records republish once the pool fills"); } - // Announce the node's CURRENT inventory so peers can immediately find the content it holds. + // Derive the node's CURRENT inventory and the content ids it provides, and record them on the + // handle so every later reconcile diffs against the truth from the first moment. + // + // The ANNOUNCE itself — one Kademlia lookup + PUT per id, each RPC bounded by the DHT timeout — + // is deliberately NOT run here. Awaiting it on the bring-up path is what left the mTLS peer-RPC + // listener (bound near the end of `run_peer_network`) unbound for 12m40s on a node holding 44 + // capsules: undialable and undiscoverable, yet holding a relay reservation that advertised it as + // up, and logging nothing for the whole window (dig_ecosystem#1974). The caller starts it in the + // background via `spawn_initial_inventory_announce` once the pool→routing feed is live. let cached = node.cache_list_cached().await; - let announced = crate::dht::announce_inventory(&service, &cached).await; let initial_ids = crate::dht::inventory_content_ids(&cached); println!( - "dig-node peer network: DHT up — announced {announced} content id(s) for local inventory" + "dig-node peer network: DHT up — {} content id(s) to announce for local inventory", + initial_ids.len() ); let dht = crate::dht::DhtHandle::new(service, initial_ids); diff --git a/crates/dig-node-core/src/seams/capsule/capsule_store.rs b/crates/dig-node-core/src/seams/capsule/capsule_store.rs index 7305ba0..d872e0b 100644 --- a/crates/dig-node-core/src/seams/capsule/capsule_store.rs +++ b/crates/dig-node-core/src/seams/capsule/capsule_store.rs @@ -19,6 +19,59 @@ use digstore_core::Bytes32; use crate::{module_exists, CachedCapsule, Node, PeerNetwork}; +/// Walk `//.dig` and describe every capsule this node holds. +/// +/// BLOCKING (`read_dir` plus one `stat` per entry) — drive it from a blocking thread, never from an +/// async worker: on a network-mounted cache each of those is a round trip. It reads only directory +/// metadata (name, size, mtime) and never opens a capsule's bytes, so its cost is exactly one `stat` +/// per held generation. +fn list_cached_capsules(modules_root: &std::path::Path) -> Vec { + let mut out = Vec::new(); + // Outer level: one directory per store id (hex). Inner: `.dig` (or a legacy `.module`). + let Ok(stores) = std::fs::read_dir(modules_root) else { + return out; // no modules cached yet + }; + for store_entry in stores.flatten() { + if !store_entry.path().is_dir() { + continue; + } + let Some(store_hex) = store_entry.file_name().to_str().map(str::to_string) else { + continue; + }; + let Ok(modules) = std::fs::read_dir(store_entry.path()) else { + continue; + }; + for m in modules.flatten() { + let path = m.path(); + // A capsule module is `.dig` (or a legacy `.module` a prior binary + // wrote — #1896); either names a held capsule, so stripping BOTH suffixes from one + // authority is what keeps a legacy holder discoverable through the upgrade. + let Some(root_hex) = path + .file_name() + .and_then(|f| f.to_str()) + .and_then(crate::capsule_key::cached_root_stem) + .map(str::to_string) + else { + continue; + }; + let Ok(md) = m.metadata() else { continue }; + let last_used_unix_ms = md + .modified() + .ok() + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + out.push(CachedCapsule { + store_id: store_hex.clone(), + root: root_hex, + size_bytes: md.len(), + last_used_unix_ms, + }); + } + } + out +} + /// Seam 6 (capsule management) — the node's on-disk `.dig` capsule cache: list/remove/fetch a held /// capsule, gap-fill a missing chain-confirmed generation, and the self-reference plumbing that lets /// `&self` read handlers spawn an owned background backfill. @@ -134,51 +187,14 @@ pub trait CapsuleStore: Send + Sync { #[async_trait::async_trait] impl CapsuleStore for Node { async fn cache_list_cached(&self) -> Vec { + // Handed to a blocking thread rather than run inline: this is `std::fs` `read_dir` + `stat` + // per held capsule, and on a node whose cache is a NETWORK mount (the S3-backed store, #1943) + // every one of those is a round trip. Running them on an async worker parks a runtime thread + // for the duration, stalling whatever else was scheduled onto it (dig_ecosystem#1974). let modules_root = self.cache_dir.join("modules"); - let mut out = Vec::new(); - // Outer level: one directory per store id (hex). Inner: `.dig` (or a legacy `.module`). - let Ok(stores) = std::fs::read_dir(&modules_root) else { - return out; // no modules cached yet - }; - for store_entry in stores.flatten() { - if !store_entry.path().is_dir() { - continue; - } - let Some(store_hex) = store_entry.file_name().to_str().map(str::to_string) else { - continue; - }; - let Ok(modules) = std::fs::read_dir(store_entry.path()) else { - continue; - }; - for m in modules.flatten() { - let path = m.path(); - // A capsule module is `.dig` (or a legacy `.module` a prior binary - // wrote — #1896); either names a held capsule, so stripping BOTH suffixes from one - // authority is what keeps a legacy holder discoverable through the upgrade. - let Some(root_hex) = path - .file_name() - .and_then(|f| f.to_str()) - .and_then(crate::capsule_key::cached_root_stem) - .map(str::to_string) - else { - continue; - }; - let Ok(md) = m.metadata() else { continue }; - let last_used_unix_ms = md - .modified() - .ok() - .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) - .map(|d| d.as_millis() as u64) - .unwrap_or(0); - out.push(CachedCapsule { - store_id: store_hex.clone(), - root: root_hex, - size_bytes: md.len(), - last_used_unix_ms, - }); - } - } - out + tokio::task::spawn_blocking(move || list_cached_capsules(&modules_root)) + .await + .unwrap_or_default() } async fn cache_remove_cached( diff --git a/crates/dig-node-core/src/seams/dig_peer/dht.rs b/crates/dig-node-core/src/seams/dig_peer/dht.rs index 810304c..a8951d3 100644 --- a/crates/dig-node-core/src/seams/dig_peer/dht.rs +++ b/crates/dig-node-core/src/seams/dig_peer/dht.rs @@ -377,16 +377,100 @@ pub fn inventory_content_ids(cached: &[CachedCapsule]) -> Vec usize { + total.div_ceil(10).max(1) +} + /// Announce EVERY content id for the node's current inventory into the DHT (`announce_provider` per -/// id). Called on startup once the DHT is bootstrapped, so peers can immediately find the content this -/// node holds. Returns the number of content ids announced. Best-effort: a PUT that reaches no peers -/// (empty routing table) still stores the record locally + is retried by `republish`. +/// id), up to `concurrency` at a time, logging progress as it goes. Returns the number of content ids +/// announced. Best-effort: a PUT that reaches no peers (empty routing table) still stores the record +/// locally + is retried by `republish`. +/// +/// The announces run CONCURRENTLY because each one is dominated by network wait, not by local work — +/// see [`INITIAL_ANNOUNCE_CONCURRENCY`] for what the sequential version cost. The progress logging is +/// not cosmetic: this is a multi-minute operation on a content-heavy node, and a silent one is +/// indistinguishable from a hung bring-up — it was read as a crash twice (dig_ecosystem#1974). +pub async fn announce_inventory_ids( + dht: &DhtService, + ids: &[dig_dht::ContentId], + concurrency: usize, +) -> usize { + use futures::stream::StreamExt; + use std::sync::atomic::{AtomicUsize, Ordering}; + + let total = ids.len(); + if total == 0 { + return 0; + } + let stride = announce_progress_stride(total); + // `tokio::time::Instant` so the reported elapsed tracks the same clock the RPC timeouts do. + let started = tokio::time::Instant::now(); + let done = AtomicUsize::new(0); + + println!( + "dig-node peer network: DHT announcing {total} content id(s) for local inventory \ + ({concurrency} at a time)" + ); + + futures::stream::iter(ids.iter()) + .for_each_concurrent(concurrency.max(1), |id| { + let done = &done; + async move { + let _ = dht.announce_provider(id).await; + let n = done.fetch_add(1, Ordering::Relaxed) + 1; + if n % stride == 0 && n != total { + println!( + "dig-node peer network: DHT announced {n} of {total} content id(s) ({:.0}s elapsed)", + started.elapsed().as_secs_f64() + ); + } + } + }) + .await; + + println!( + "dig-node peer network: DHT announced {total} content id(s) for local inventory in {:.1}s", + started.elapsed().as_secs_f64() + ); + total +} + +/// [`announce_inventory_ids`] over the content ids derived from `cached`, at the default concurrency. pub async fn announce_inventory(dht: &DhtService, cached: &[CachedCapsule]) -> usize { let ids = inventory_content_ids(cached); - for id in &ids { - let _ = dht.announce_provider(id).await; - } - ids.len() + announce_inventory_ids(dht, &ids, INITIAL_ANNOUNCE_CONCURRENCY).await +} + +/// Announce the node's initial inventory into the DHT in the BACKGROUND. +/// +/// The ids are already recorded on `handle` by bring-up, so every later reconcile diffs against the +/// truth whether or not this task has finished; what runs here is only the network publishing. It is +/// deliberately NOT awaited on the bring-up path: doing so left the mTLS peer-RPC listener unbound — +/// the node undialable and undiscoverable, while holding a relay reservation that advertised it as up +/// — for 12m40s on a node holding 44 capsules (dig_ecosystem#1974). +/// +/// Call it AFTER the live pool→routing feed is installed, so each announce runs against a routing +/// table that is filling rather than the empty one bootstrap leaves behind. +pub fn spawn_initial_inventory_announce(handle: Arc) { + tokio::spawn(async move { + let ids = handle.announced_ids().await; + announce_inventory_ids(handle.service(), &ids, INITIAL_ANNOUNCE_CONCURRENCY).await; + }); } /// The diff between a previous and a current inventory content-id set: `(to_announce, to_withdraw)`. @@ -522,6 +606,16 @@ impl DhtHandle { &self.service } + /// A snapshot of the content ids this node currently intends to provide. + /// + /// Recorded at bring-up from the on-disk inventory and kept current by + /// [`reconcile_inventory`](Self::reconcile_inventory). Read by + /// [`spawn_initial_inventory_announce`] so the background announce publishes exactly the set the + /// handle already considers announced — the two can never disagree about what this node holds. + pub async fn announced_ids(&self) -> Vec { + self.announced.lock().await.clone() + } + /// Re-derive the inventory content-id set from `cached` and reconcile it with the DHT: announce /// new ids, actively retract gone ids (see [`sync_inventory`]). Updates the remembered set. Call /// whenever the node's inventory changes (a capsule cached, a root advanced, a store removed). @@ -679,6 +773,139 @@ mod tests { } use dig_dht::ContentId; + /// A transport whose every RPC sleeps for `delay` and then reports the peer unreachable. + /// + /// This is the shape of the real bring-up announce: against an almost-empty routing table every + /// PUT grinds to the per-RPC timeout rather than resolving, so an announce's cost is entirely + /// network wait. It makes the wall-clock of [`announce_inventory_ids`] a pure function of how + /// many announces are allowed in flight, which is exactly what the concurrency test measures. + struct SlowTransport { + delay: Duration, + } + + #[async_trait] + impl DhtTransport for SlowTransport { + async fn rpc( + &self, + _from: &Contact, + _peer: &Contact, + _request: &DhtRequest, + ) -> Result { + tokio::time::sleep(self.delay).await; + Err(DhtError::Timeout) + } + } + + /// A `DhtService` over [`SlowTransport`], with `seeds` contacts already in its routing table so + /// `announce_provider` actually reaches the transport (with an empty table it short-circuits). + async fn slow_service(delay: Duration, seeds: usize) -> Arc { + let service = Arc::new(DhtService::new( + PeerId::from_bytes([0x77; 32]), + vec![CandidateAddr::direct("::1", 9444)], + dig_dht::DhtConfig::default(), + Arc::new(SlowTransport { delay }), + )); + for i in 0..seeds { + service + .add_peer( + &PeerId::from_bytes([i as u8 + 1; 32]), + vec![CandidateAddr::direct("::1", 9500 + i as u16)], + ) + .await; + } + service + } + + /// `total` capsules spread one-per-store, as the inventory content-id list the bring-up announces. + fn inventory_ids(capsules: usize) -> Vec { + let cached: Vec = (0..capsules) + .map(|i| { + cap( + &format!("{:064x}", i + 1), + &format!("{:064x}", 0xF000 + i + 1), + ) + }) + .collect(); + inventory_content_ids(&cached) + } + + /// **Proves:** the initial inventory announce runs its per-id Kademlia round trips CONCURRENTLY, + /// so bring-up cost is sub-linear in how much content the node holds. + /// + /// **Catches:** the dig_ecosystem#1974 defect — `announce_inventory` was a bare + /// `for id in &ids { dht.announce_provider(id).await }`, so a node holding 44 capsules (68 content + /// ids) paid 68 sequential lookup-plus-PUT round trips at the 5s per-RPC timeout and took 12m40s. + /// A revert to the sequential loop makes the measured elapsed equal the full sequential cost and + /// fails the bound below. + /// + /// The bound is SELF-CALIBRATING: it times one real `announce_provider` against this same + /// transport rather than hardcoding a duration, so it cannot rot when dig-dht changes how many + /// RPCs an announce performs. Time is PAUSED, so the measurement is the deterministic virtual + /// clock, not a wall-clock race that could flake on a loaded CI box. + #[tokio::test(start_paused = true)] + async fn the_initial_inventory_announce_runs_concurrently_not_one_id_at_a_time() { + let service = slow_service(Duration::from_secs(5), 3).await; + let ids = inventory_ids(34); // 34 stores x 1 capsule each = 68 content ids, as on rpc.dig.net + assert_eq!(ids.len(), 68, "34 single-capsule stores announce 68 content ids"); + + // Calibrate: what does ONE announce cost against this transport? + let t0 = tokio::time::Instant::now(); + let _ = service.announce_provider(&ids[0]).await; + let one = t0.elapsed(); + assert!( + one >= Duration::from_secs(5), + "the calibration announce must actually reach the slow transport, took {one:?}" + ); + + let concurrency = 8; + let t1 = tokio::time::Instant::now(); + let announced = announce_inventory_ids(&service, &ids, concurrency).await; + let all = t1.elapsed(); + + assert_eq!(announced, ids.len(), "every content id is announced"); + + // The sequential implementation costs `ids.len() * one`. With 8 in flight the whole set must + // land in well under a quarter of that; a generous margin keeps the assertion about + // concurrency rather than about an exact scheduling shape. + let sequential = one * ids.len() as u32; + assert!( + all < sequential / 4, + "announcing {} ids with {concurrency} in flight took {all:?}, which is not meaningfully \ + faster than the {sequential:?} a one-at-a-time loop would cost (#1974)", + ids.len(), + ); + } + + /// **Proves:** an empty inventory announces nothing and does no network work — the state every + /// fresh node boots in. + /// + /// **Catches:** a progress-logging or concurrency rewrite that divides by a zero total, or that + /// emits bring-up noise on a node with no content to announce. + #[tokio::test] + async fn an_empty_inventory_announces_nothing() { + let service = slow_service(Duration::from_secs(5), 3).await; + assert_eq!(announce_inventory_ids(&service, &[], 8).await, 0); + } + + /// **Proves:** the progress stride emits at most ~10 lines no matter how much content is held, and + /// never zero (which would panic the `n % stride` modulo). + /// + /// **Catches:** a stride of 0 on a small inventory, and per-id log spam on a large one — the + /// tier-0 eager cache (#1934) deliberately grows holdings, so this must stay bounded. + #[test] + fn announce_progress_is_bounded_regardless_of_inventory_size() { + assert_eq!(announce_progress_stride(0), 1, "stride is never zero"); + assert_eq!(announce_progress_stride(1), 1); + assert_eq!(announce_progress_stride(68), 7); + for total in [1usize, 9, 68, 1_000, 250_000] { + let lines = total / announce_progress_stride(total); + assert!( + lines <= 10, + "{total} ids would emit {lines} progress lines; the cap is ~10" + ); + } + } + fn cap(store: &str, root: &str) -> CachedCapsule { CachedCapsule { store_id: store.to_string(), diff --git a/crates/dig-node-core/src/seams/dig_peer/mod.rs b/crates/dig-node-core/src/seams/dig_peer/mod.rs index 86ff555..7cd156d 100644 --- a/crates/dig-node-core/src/seams/dig_peer/mod.rs +++ b/crates/dig-node-core/src/seams/dig_peer/mod.rs @@ -21,6 +21,7 @@ pub mod module_transport; pub mod net; pub mod peer_network; pub mod pex; +pub mod ping; pub mod pool_locator; /// Wire-level range-stream conformance tests (#1668): the real serve path over a real loopback mTLS /// connection, decoded through the real dig-nat codec. Test-only — there is no production surface here. diff --git a/crates/dig-node-core/src/seams/dig_peer/ping.rs b/crates/dig-node-core/src/seams/dig_peer/ping.rs new file mode 100644 index 0000000..3312441 --- /dev/null +++ b/crates/dig-node-core/src/seams/dig_peer/ping.rs @@ -0,0 +1,978 @@ +//! Peer PING — run the connection ladder against one peer and report WHICH tier reached it. +//! +//! This answers the question "is this node actually reachable, and how?", which until now was +//! answered by hand with a TCP port probe across a list of addresses. An open port is not a peer +//! connection: it says nothing about whether the mTLS handshake succeeds, whether the certificate +//! binds the `peer_id` you asked for, or whether the path that worked was the direct one or the +//! relay of last resort (dig_ecosystem#1985). +//! +//! **It reports the LADDER, not just the winner.** SPEC §19.1 ranks the tiers +//! `Direct → UPnP → NAT-PMP → PCP → hole-punch → Relayed` and makes the relay the LAST resort, so +//! "connected" is not one fact but two: whether the peer was reached, and what it cost to reach it. +//! A peer reachable only through the relay is a different operational state from one reachable +//! directly, and collapsing them to a boolean is what hid dig_ecosystem#1929. +//! +//! **A relay-only success is expected, not broken.** Most peers on the network are behind NAT and +//! are relay-reachable only; that is the normal shape of the network, and [`PingVerdict::severity`] +//! deliberately grades it `warn`, never `error`. The finding worth surfacing is narrower: a peer +//! that ADVERTISES a routable address and still cannot be reached directly. +//! +//! **It is read-only.** Each tier is a bare `dig-nat` dial that is dropped as soon as it is graded: +//! it joins no pool, announces nothing, writes nothing, and leaves no relay reservation behind — a +//! diagnostic that mutates network state is not a diagnostic. It reuses the SAME +//! [`crate::net::full_nat_config`] every other node dial is built from, narrowed to one tier at a +//! time, so what it reports is what the real dialer does rather than a parallel prober that could +//! drift away from it. + +use std::net::SocketAddr; +use std::time::Duration; + +use async_trait::async_trait; +use dig_nat::{PeerTarget, TraversalKind}; +use serde_json::{json, Value}; + +/// The ladder as SPEC §19.1 defines it, in canonical rank order (relay last). +/// +/// Derived from [`TraversalKind::rank`] rather than written out, so a tier added to dig-nat is +/// probed here automatically instead of being silently missed by a hardcoded list. +pub fn ladder_tiers() -> Vec { + let mut tiers = vec![ + TraversalKind::Direct, + TraversalKind::Upnp, + TraversalKind::NatPmp, + TraversalKind::Pcp, + TraversalKind::HolePunch, + TraversalKind::Relayed, + ]; + tiers.sort_by_key(|t| t.rank()); + tiers +} + +/// The wire token for a tier — stable, lower-case, and part of the control-method contract. +pub fn tier_name(tier: TraversalKind) -> &'static str { + match tier { + TraversalKind::Direct => "direct", + TraversalKind::Upnp => "upnp", + TraversalKind::NatPmp => "nat-pmp", + TraversalKind::Pcp => "pcp", + TraversalKind::HolePunch => "hole-punch", + TraversalKind::Relayed => "relayed", + } +} + +/// What one tier of the ladder did. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TierOutcome { + /// The tier established an authenticated mTLS peer connection. + Connected { + /// The address the connection actually landed on. + remote_addr: SocketAddr, + /// The `peer_id` the presented certificate derives (`SHA-256(TLS SPKI DER)`) — the identity + /// that answered, which is not necessarily the identity that was asked for. + observed_peer_id: String, + elapsed_ms: u64, + }, + /// The tier was attempted and did not connect. `reason` is dig-nat's own failure text. + Failed { reason: String, elapsed_ms: u64 }, + /// The tier was not attempted at all (the overall deadline elapsed first). + Skipped { reason: String }, +} + +impl TierOutcome { + /// The connected peer's observed identity, if this tier connected. + pub fn observed_peer_id(&self) -> Option<&str> { + match self { + TierOutcome::Connected { + observed_peer_id, .. + } => Some(observed_peer_id), + _ => None, + } + } +} + +/// One rung of the ladder and what it did. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TierReport { + pub tier: TraversalKind, + pub outcome: TierOutcome, +} + +impl TierReport { + fn skipped(tier: TraversalKind, reason: impl Into) -> Self { + TierReport { + tier, + outcome: TierOutcome::Skipped { + reason: reason.into(), + }, + } + } +} + +/// The overall reading of a ladder run. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PingVerdict { + /// Reached WITHOUT the relay — the healthy shape. + Direct { tier: TraversalKind }, + /// Reached, but only through the relay. SPEC §19.1 makes the relay the last resort, so this is a + /// yellow reading rather than a green one — and the ordinary state for a peer behind NAT. + RelayedOnly, + /// No tier reached the peer. + Unreachable, + /// A tier connected, but the certificate derives a DIFFERENT `peer_id` than the one asked for. + /// + /// This outranks every other reading, including a successful direct connection: reaching the + /// right address and the wrong identity is a failure that must be loud, never a pass. + IdentityMismatch { expected: String, observed: String }, +} + +/// Grade a completed ladder run. +/// +/// `expected_peer_id` is the identity the caller asked for, when it asked by `peer_id`; a ping by +/// bare address has none to check against and reports whichever identity answered. +pub fn verdict(expected_peer_id: Option<&str>, tiers: &[TierReport]) -> PingVerdict { + // Identity first, and before any success is reported: a connection to the wrong peer is the one + // outcome that must never be graded on how nicely it connected. + if let Some(expected) = expected_peer_id { + for report in tiers { + if let Some(observed) = report.outcome.observed_peer_id() { + if !observed.eq_ignore_ascii_case(expected) { + return PingVerdict::IdentityMismatch { + expected: expected.to_ascii_lowercase(), + observed: observed.to_ascii_lowercase(), + }; + } + } + } + } + // The best (lowest-rank) tier that connected wins the reading. + let best = tiers + .iter() + .filter(|r| matches!(r.outcome, TierOutcome::Connected { .. })) + .min_by_key(|r| r.tier.rank()); + match best { + Some(r) if r.tier == TraversalKind::Relayed => PingVerdict::RelayedOnly, + Some(r) => PingVerdict::Direct { tier: r.tier }, + None => PingVerdict::Unreachable, + } +} + +impl PingVerdict { + /// The wire token for this reading. + pub fn code(&self) -> &'static str { + match self { + PingVerdict::Direct { .. } => "direct", + PingVerdict::RelayedOnly => "relayed-only", + PingVerdict::Unreachable => "unreachable", + PingVerdict::IdentityMismatch { .. } => "identity-mismatch", + } + } + + /// How loudly to render this reading: `ok` / `warn` / `error`. + /// + /// A relay-only peer is `warn`, NOT `error`: most peers on the network are behind NAT and are + /// relay-reachable only, so grading that as a failure would report a healthy network as broken + /// to every user who ran the diagnostic. + pub fn severity(&self) -> &'static str { + match self { + PingVerdict::Direct { .. } => "ok", + PingVerdict::RelayedOnly => "warn", + PingVerdict::Unreachable | PingVerdict::IdentityMismatch { .. } => "error", + } + } + + /// A one-line reading in plain language, saying what happened AND whether it is a problem. + pub fn summary(&self) -> String { + match self { + PingVerdict::Direct { tier } => format!( + "reachable over the {} tier, without the relay", + tier_name(*tier) + ), + PingVerdict::RelayedOnly => "reachable, but only through the relay — normal for a peer \ + behind NAT; a finding only if this peer advertises a routable address" + .to_string(), + PingVerdict::Unreachable => { + "not reachable on any tier of the connection ladder".to_string() + } + PingVerdict::IdentityMismatch { expected, observed } => format!( + "WRONG PEER: the address answered with peer_id {observed}, not the {expected} asked for" + ), + } + } +} + +/// A peer reached by one tier: who answered and where. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DialedPeer { + pub observed_peer_id: String, + pub remote_addr: SocketAddr, +} + +/// Attempt exactly ONE tier of the ladder. +/// +/// The seam exists so [`run_ladder`]'s grading is testable without a network; the production +/// implementation ([`NatTierDialer`]) is a thin wrapper over the same `dig_nat::connect_with_runtime` +/// every other node dial goes through. +#[async_trait] +pub trait TierDialer: Send + Sync { + /// Dial `target` with ONLY `tier` enabled. `Err` carries dig-nat's failure text verbatim. + async fn dial_tier(&self, tier: TraversalKind, target: &PeerTarget) -> Result; +} + +/// Run every tier of the ladder against `target` and report each one. +/// +/// Every tier is attempted even after one succeeds — the point of the diagnostic is the whole +/// ladder, since "connected" hides whether the direct path was available. Attempts stop only at +/// `deadline`, after which the remaining tiers are reported as skipped rather than silently dropped. +pub async fn run_ladder( + dialer: &dyn TierDialer, + target: &PeerTarget, + deadline: Duration, +) -> Vec { + // `tokio::time::Instant`, not `std::time::Instant`: it is the clock `tokio::time` bounds are + // measured against, so the deadline holds under a paused test clock as well as in production. + let started = tokio::time::Instant::now(); + let mut reports = Vec::new(); + for tier in ladder_tiers() { + if started.elapsed() >= deadline { + reports.push(TierReport::skipped( + tier, + format!("overall deadline of {}s reached first", deadline.as_secs()), + )); + continue; + } + let tier_started = tokio::time::Instant::now(); + let outcome = match dialer.dial_tier(tier, target).await { + Ok(peer) => TierOutcome::Connected { + remote_addr: peer.remote_addr, + observed_peer_id: peer.observed_peer_id, + elapsed_ms: tier_started.elapsed().as_millis() as u64, + }, + Err(reason) => TierOutcome::Failed { + reason, + elapsed_ms: tier_started.elapsed().as_millis() as u64, + }, + }; + reports.push(TierReport { tier, outcome }); + } + reports +} + +/// Everything a ping needs from the running peer network: this node's mTLS identity, the shared NAT +/// runtime (which carries the live relay reservation the relayed tier rides), the network id, and the +/// STUN server that feeds the hole-punch tier. +/// +/// Assembled once by bring-up and kept on the [`Node`](crate::Node) so the control surface can run a +/// ladder with exactly the inputs the node's own dials use — the ping cannot drift from the real +/// dialer because it is given the real dialer's configuration. +pub struct PeerPingContext { + identity: std::sync::Arc, + runtime: std::sync::Arc, + network_id: String, + stun_server: Option, + per_tier_timeout: Duration, +} + +impl PeerPingContext { + pub fn new( + identity: std::sync::Arc, + runtime: std::sync::Arc, + network_id: impl Into, + stun_server: Option, + per_tier_timeout: Duration, + ) -> Self { + PeerPingContext { + identity, + runtime, + network_id: network_id.into(), + stun_server, + per_tier_timeout, + } + } + + /// The network id peers are registered under, for relay lookups + hole-punch coordination. + pub fn network_id(&self) -> &str { + &self.network_id + } + + /// The per-tier timeout a ladder run bounds each attempt by. + pub fn per_tier_timeout(&self) -> Duration { + self.per_tier_timeout + } +} + +/// The production [`TierDialer`]: one tier of the REAL `dig-nat` ladder per attempt. +/// +/// It builds [`crate::net::full_nat_config`] — the one shared config constructor SPEC §19.1 requires +/// every dial site to use — and narrows `enabled_methods` to the single tier under test. So each +/// attempt is the genuine dialer restricted to one rung, not a reimplementation that could disagree +/// with what the node actually does when it connects to a peer. +pub struct NatTierDialer<'a> { + ctx: &'a PeerPingContext, +} + +impl<'a> NatTierDialer<'a> { + pub fn new(ctx: &'a PeerPingContext) -> Self { + NatTierDialer { ctx } + } +} + +#[async_trait] +impl TierDialer for NatTierDialer<'_> { + async fn dial_tier( + &self, + tier: TraversalKind, + target: &PeerTarget, + ) -> Result { + let config = dig_nat::NatConfig::builder() + .per_method_timeout(self.ctx.per_tier_timeout) + .enabled_methods(vec![tier]); + let config = match self.ctx.stun_server { + Some(stun) => config.stun_server(stun).build(), + None => config.build(), + }; + + let conn = dig_nat::connect_with_runtime( + target, + &self.ctx.identity, + &config, + &self.ctx.runtime, + ) + .await + .map_err(|e| e.to_string())?; + + let dialed = DialedPeer { + observed_peer_id: conn.peer_id.to_hex(), + remote_addr: conn.remote_addr, + }; + // Dropped immediately, before the result is even graded: a diagnostic must leave nothing + // behind — no pooled session, no held relay circuit (#1985). + drop(conn); + Ok(dialed) + } +} + +/// How a ping resolved the `peer` argument into something dialable. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TargetResolution { + /// A `peer_id` and at least one candidate address — a full identity-verified ladder can run. + Resolved { + peer_id: String, + addrs: Vec, + }, + /// The argument named a peer whose address this node does not know. + NoKnownAddress { peer_id: String }, + /// An address was given with no identity to verify it against. + /// + /// dig-nat pins the expected `peer_id` in the TLS verifier, so there is no anonymous dial to + /// fall back to — and degrading to a bare TCP probe would report exactly the "open port means + /// connected" answer this diagnostic exists to replace. The caller is told what to supply. + IdentityRequired { addr: SocketAddr }, + /// The argument was neither a 64-hex `peer_id` nor a dialable `host:port`. + Unparseable, +} + +/// Turn the `peer` argument into a dialable target. +/// +/// `known` is the node's view of who is where — `(peer_id_hex, addr)` for every peer it can name, +/// sourced from the connected pool. `explicit_peer_id` is an identity the caller pinned outright, +/// which always wins: pinning a `peer_id` that does NOT match the address is exactly how the +/// identity-mismatch case is exercised, so it must not be second-guessed here. +pub fn resolve_target( + peer: &str, + explicit_peer_id: Option<&str>, + known: &[(String, SocketAddr)], +) -> TargetResolution { + let peer = peer.trim(); + + // A dialable address: the identity comes from the caller's pin, else from what this node already + // knows is listening there. + if let Ok(addr) = peer.parse::() { + if let Some(pinned) = explicit_peer_id { + return TargetResolution::Resolved { + peer_id: pinned.to_ascii_lowercase(), + addrs: vec![addr], + }; + } + return match known.iter().find(|(_, a)| *a == addr) { + Some((peer_id, _)) => TargetResolution::Resolved { + peer_id: peer_id.to_ascii_lowercase(), + addrs: vec![addr], + }, + None => TargetResolution::IdentityRequired { addr }, + }; + } + + // A bare peer_id: look up every address this node knows for it. + if is_peer_id(peer) { + let peer_id = peer.to_ascii_lowercase(); + let addrs: Vec = known + .iter() + .filter(|(id, _)| id.eq_ignore_ascii_case(&peer_id)) + .map(|(_, a)| *a) + .collect(); + return if addrs.is_empty() { + TargetResolution::NoKnownAddress { peer_id } + } else { + TargetResolution::Resolved { peer_id, addrs } + }; + } + + TargetResolution::Unparseable +} + +/// Whether `s` is a canonical 64-hex `peer_id`. +fn is_peer_id(s: &str) -> bool { + s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit()) +} + +/// Build the dig-nat [`PeerTarget`] for a resolved ping, ordering candidates IPv6-first (§5.2). +pub fn peer_target( + peer_id_hex: &str, + addrs: &[SocketAddr], + network_id: &str, +) -> Option { + let bytes = hex::decode(peer_id_hex).ok().filter(|b| b.len() == 32)?; + let mut raw = [0u8; 32]; + raw.copy_from_slice(&bytes); + // IPv6 candidates lead: the ecosystem is IPv6-first with IPv4 as the fallback, and the ping must + // exercise the ladder in the same family order a real dial would. + let mut ordered: Vec = addrs.to_vec(); + ordered.sort_by_key(|a| !a.is_ipv6()); + Some(PeerTarget::with_addrs( + dig_nat::PeerId::from_bytes(raw), + ordered, + network_id, + )) +} + +/// The ping result as the control method returns it. +pub fn report_json( + peer: &str, + expected_peer_id: Option<&str>, + tiers: &[TierReport], + verdict: &PingVerdict, +) -> Value { + json!({ + "peer": peer, + "expected_peer_id": expected_peer_id, + "verdict": verdict.code(), + "severity": verdict.severity(), + "summary": verdict.summary(), + "ladder": tiers.iter().map(tier_json).collect::>(), + }) +} + +/// One ladder rung as JSON. The TCP/mTLS distinction lives in `reason`: dig-nat reports a refused +/// port and a rejected handshake as different failures, and #1985 needs them told apart. +fn tier_json(report: &TierReport) -> Value { + match &report.outcome { + TierOutcome::Connected { + remote_addr, + observed_peer_id, + elapsed_ms, + } => json!({ + "tier": tier_name(report.tier), + "result": "connected", + "remote_addr": remote_addr.to_string(), + // §5.2 is IPv6-first, so an IPv4-only success is itself a finding worth reading off. + "family": if remote_addr.is_ipv6() { "ipv6" } else { "ipv4" }, + "observed_peer_id": observed_peer_id, + "elapsed_ms": elapsed_ms, + }), + TierOutcome::Failed { reason, elapsed_ms } => json!({ + "tier": tier_name(report.tier), + "result": "failed", + "reason": reason, + "elapsed_ms": elapsed_ms, + }), + TierOutcome::Skipped { reason } => json!({ + "tier": tier_name(report.tier), + "result": "skipped", + "reason": reason, + }), + } +} + +/// Run a full ping: resolve `peer`, walk the ladder, grade it, and return the report JSON. +/// +/// `known` is `(peer_id_hex, addr)` for every peer this node can currently name. A resolution +/// failure is reported as a result with an `error` severity, never as a transport error: "I could +/// not work out what to dial" is a diagnostic answer, and the caller asked a diagnostic question. +pub async fn ping_peer( + ctx: &PeerPingContext, + peer: &str, + explicit_peer_id: Option<&str>, + known: &[(String, SocketAddr)], + deadline: Duration, +) -> Value { + let (peer_id, addrs) = match resolve_target(peer, explicit_peer_id, known) { + TargetResolution::Resolved { peer_id, addrs } => (peer_id, addrs), + TargetResolution::NoKnownAddress { peer_id } => { + return unresolved_json( + peer, + Some(&peer_id), + "this node knows no address for that peer_id — dial it by address, or wait for \ + discovery to fold it into the connected pool", + ) + } + TargetResolution::IdentityRequired { addr } => { + return unresolved_json( + peer, + None, + &format!( + "no peer_id is known for {addr}; supply peer_id so the mTLS certificate can be \ + verified — an identity-less dial could only report whether a port is open, \ + which is not a peer connection" + ), + ) + } + TargetResolution::Unparseable => { + return unresolved_json( + peer, + None, + "not a dialable address (host:port, IPv6 in brackets) nor a 64-hex peer_id", + ) + } + }; + + let Some(target) = peer_target(&peer_id, &addrs, ctx.network_id()) else { + return unresolved_json(peer, Some(&peer_id), "peer_id is not valid 64-hex"); + }; + + let dialer = NatTierDialer::new(ctx); + let tiers = run_ladder(&dialer, &target, deadline).await; + let verdict = verdict(Some(&peer_id), &tiers); + report_json(peer, Some(&peer_id), &tiers, &verdict) +} + +/// The result shape for a ping that never got as far as dialing. +fn unresolved_json(peer: &str, peer_id: Option<&str>, reason: &str) -> Value { + json!({ + "peer": peer, + "expected_peer_id": peer_id, + "verdict": "unresolved", + "severity": "error", + "summary": reason, + "ladder": Vec::::new(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn addr(s: &str) -> SocketAddr { + s.parse().expect("test address") + } + + fn connected(tier: TraversalKind, peer_id: &str, at: &str) -> TierReport { + TierReport { + tier, + outcome: TierOutcome::Connected { + remote_addr: addr(at), + observed_peer_id: peer_id.to_string(), + elapsed_ms: 12, + }, + } + } + + fn failed(tier: TraversalKind, reason: &str) -> TierReport { + TierReport { + tier, + outcome: TierOutcome::Failed { + reason: reason.to_string(), + elapsed_ms: 5000, + }, + } + } + + const PEER_A: &str = "aa"; + const PEER_B: &str = "bb"; + + /// **Proves:** the probed ladder is SPEC §19.1's canonical order, relay last. + /// + /// **Catches:** a reordering that probes the relay before the direct tier, which would report + /// "relayed" for a peer that was directly reachable all along. + #[test] + fn the_ladder_is_probed_in_spec_rank_order_with_the_relay_last() { + let tiers = ladder_tiers(); + assert_eq!( + tiers, + vec![ + TraversalKind::Direct, + TraversalKind::Upnp, + TraversalKind::NatPmp, + TraversalKind::Pcp, + TraversalKind::HolePunch, + TraversalKind::Relayed, + ] + ); + assert_eq!( + *tiers.last().expect("non-empty ladder"), + TraversalKind::Relayed, + "SPEC §19.1 makes the relay the LAST resort" + ); + } + + /// **Proves:** a direct success is graded green. + #[test] + fn a_direct_connection_is_a_green_reading() { + let tiers = vec![ + connected(TraversalKind::Direct, PEER_A, "[2001:db8::1]:9444"), + failed(TraversalKind::Relayed, "no reservation"), + ]; + let v = verdict(Some(PEER_A), &tiers); + assert_eq!( + v, + PingVerdict::Direct { + tier: TraversalKind::Direct + } + ); + assert_eq!(v.severity(), "ok"); + } + + /// **Proves:** a peer reachable ONLY through the relay reads as a yellow result — reached, but + /// not on the tier SPEC prefers. + /// + /// **Catches:** grading a relay fallback as a plain success, which is what let dig_ecosystem#1929 + /// (peers relayed when direct was possible) hide behind a green "connected". + #[test] + fn a_relay_only_success_is_yellow_not_green() { + let tiers = vec![ + failed(TraversalKind::Direct, "connection refused"), + failed(TraversalKind::HolePunch, "no path formed"), + connected(TraversalKind::Relayed, PEER_A, "44.217.228.224:9444"), + ]; + let v = verdict(Some(PEER_A), &tiers); + assert_eq!(v, PingVerdict::RelayedOnly); + assert_eq!(v.severity(), "warn", "relay-only is a warning, not a pass"); + } + + /// **Proves:** a NAT'd peer — direct refused, relay worked — is NOT reported as an error, and its + /// summary says so in words. + /// + /// **Catches:** the failure mode #1985 calls out explicitly: six of ten peers on the network are + /// relay-only, so grading that shape `error` would report a healthy network as broken to every + /// user who ran the diagnostic. + #[test] + fn a_natted_peer_reads_as_expected_rather_than_as_breakage() { + let tiers = vec![ + failed(TraversalKind::Direct, "connection refused"), + connected(TraversalKind::Relayed, PEER_A, "10.0.0.5:9444"), + ]; + let v = verdict(Some(PEER_A), &tiers); + assert_ne!(v.severity(), "error", "a NAT'd peer is normal, not an error"); + let summary = v.summary().to_lowercase(); + assert!( + summary.contains("normal") && summary.contains("nat"), + "the summary must say a relay-only peer behind NAT is expected, got: {summary}" + ); + } + + /// **Proves:** connecting to a reachable address that presents the WRONG identity is a loud + /// failure — and outranks the fact that the connection itself succeeded directly. + /// + /// **Catches:** grading on reachability before identity, which would report "direct, ok" for a + /// dial that reached an entirely different node than the one asked for. + #[test] + fn a_wrong_peer_id_on_a_reachable_address_is_an_error_not_a_success() { + let tiers = vec![connected( + TraversalKind::Direct, + PEER_B, + "[2001:db8::1]:9444", + )]; + let v = verdict(Some(PEER_A), &tiers); + assert_eq!( + v, + PingVerdict::IdentityMismatch { + expected: PEER_A.to_string(), + observed: PEER_B.to_string(), + }, + "identity is checked before reachability is graded" + ); + assert_eq!(v.severity(), "error"); + } + + /// **Proves:** a ping by bare ADDRESS (no expected identity) reports whoever answered instead of + /// inventing a mismatch. + #[test] + fn a_ping_by_address_alone_reports_the_identity_that_answered() { + let tiers = vec![connected( + TraversalKind::Direct, + PEER_B, + "[2001:db8::1]:9444", + )]; + assert_eq!( + verdict(None, &tiers), + PingVerdict::Direct { + tier: TraversalKind::Direct + } + ); + } + + /// **Proves:** a peer no tier reached is unreachable, and that is an error. + #[test] + fn a_peer_no_tier_reached_is_unreachable() { + let tiers = vec![ + failed(TraversalKind::Direct, "connection refused"), + failed(TraversalKind::Relayed, "no reservation"), + ]; + let v = verdict(Some(PEER_A), &tiers); + assert_eq!(v, PingVerdict::Unreachable); + assert_eq!(v.severity(), "error"); + } + + /// **Proves:** when several tiers connect, the reading names the BEST (lowest-rank) one. + /// + /// **Catches:** reporting the last tier that happened to succeed, which would call a directly + /// reachable peer "relayed". + #[test] + fn the_best_tier_wins_the_reading_not_the_last_one_tried() { + let tiers = vec![ + connected(TraversalKind::Direct, PEER_A, "[2001:db8::1]:9444"), + connected(TraversalKind::Relayed, PEER_A, "44.217.228.224:9444"), + ]; + assert_eq!( + verdict(Some(PEER_A), &tiers), + PingVerdict::Direct { + tier: TraversalKind::Direct + } + ); + } + + /// **Proves:** the JSON reports EVERY rung — including the ones that failed — plus the address + /// family that won, which §5.2 makes a finding in its own right. + /// + /// **Catches:** emitting only the winning tier, which is exactly the "connected: true" answer + /// #1985 exists to replace. + #[test] + fn the_json_reports_every_rung_and_the_winning_address_family() { + let tiers = vec![ + failed(TraversalKind::Direct, "connection refused"), + connected(TraversalKind::Relayed, PEER_A, "44.217.228.224:9444"), + TierReport::skipped(TraversalKind::HolePunch, "overall deadline"), + ]; + let v = verdict(Some(PEER_A), &tiers); + let out = report_json("44.217.228.224:9444", Some(PEER_A), &tiers, &v); + + let ladder = out["ladder"].as_array().expect("ladder array"); + assert_eq!(ladder.len(), 3, "every attempted rung is reported"); + assert_eq!(ladder[0]["result"], "failed"); + assert_eq!(ladder[0]["reason"], "connection refused"); + assert_eq!(ladder[1]["result"], "connected"); + assert_eq!(ladder[1]["family"], "ipv4"); + assert_eq!(ladder[2]["result"], "skipped"); + assert_eq!(out["verdict"], "relayed-only"); + assert_eq!(out["severity"], "warn"); + } + + /// **Proves:** an IPv6 win is reported as such, so an IPv4-only success is visible as the §5.2 + /// finding it is. + #[test] + fn an_ipv6_connection_is_reported_as_the_ipv6_family() { + let tiers = vec![connected( + TraversalKind::Direct, + PEER_A, + "[2001:db8::1]:9444", + )]; + let out = report_json("x", Some(PEER_A), &tiers, &verdict(Some(PEER_A), &tiers)); + assert_eq!(out["ladder"][0]["family"], "ipv6"); + } + + // -- resolve_target -------------------------------------------------------------------------- + + fn known_peer(id: &str, at: &str) -> (String, SocketAddr) { + (id.repeat(32), addr(at)) + } + + /// **Proves:** a bare `peer_id` is resolved to every address this node knows for it. + #[test] + fn a_peer_id_resolves_to_the_addresses_this_node_knows_for_it() { + let known = vec![ + known_peer("aa", "[2001:db8::1]:9444"), + known_peer("bb", "10.0.0.9:9444"), + ]; + assert_eq!( + resolve_target(&"aa".repeat(32), None, &known), + TargetResolution::Resolved { + peer_id: "aa".repeat(32), + addrs: vec![addr("[2001:db8::1]:9444")], + } + ); + } + + /// **Proves:** an address this node already knows an identity for resolves without the caller + /// having to supply the `peer_id` — the "ping by ip address" form the request asked for. + #[test] + fn a_known_address_resolves_to_the_identity_listening_there() { + let known = vec![known_peer("aa", "[2001:db8::1]:9444")]; + assert_eq!( + resolve_target("[2001:db8::1]:9444", None, &known), + TargetResolution::Resolved { + peer_id: "aa".repeat(32), + addrs: vec![addr("[2001:db8::1]:9444")], + } + ); + } + + /// **Proves:** an address with NO known identity is refused with an explanation, rather than + /// silently downgraded to a bare TCP probe. + /// + /// **Catches:** the failure #1985 names directly — "an open port is not a peer connection". + /// dig-nat pins the expected `peer_id` in its TLS verifier, so there is no anonymous dial; the + /// honest answer is to say what is missing, not to answer a different, weaker question. + #[test] + fn an_unknown_address_asks_for_the_peer_id_instead_of_probing_the_port() { + assert_eq!( + resolve_target("203.0.113.7:9444", None, &[]), + TargetResolution::IdentityRequired { + addr: addr("203.0.113.7:9444") + } + ); + } + + /// **Proves:** an explicitly pinned `peer_id` beats what this node believes is at that address. + /// + /// **Catches:** "helpfully" correcting the caller's pin to the known identity, which would make + /// the wrong-identity acceptance case in #1985 impossible to test — the mismatch would be + /// silently repaired into a pass. + #[test] + fn an_explicitly_pinned_peer_id_is_never_second_guessed() { + let known = vec![known_peer("aa", "[2001:db8::1]:9444")]; + assert_eq!( + resolve_target("[2001:db8::1]:9444", Some(&"bb".repeat(32)), &known), + TargetResolution::Resolved { + peer_id: "bb".repeat(32), + addrs: vec![addr("[2001:db8::1]:9444")], + } + ); + } + + /// **Proves:** a `peer_id` with no known address says so, instead of dialing nothing and calling + /// the peer unreachable. + #[test] + fn a_peer_id_with_no_known_address_says_so() { + assert_eq!( + resolve_target(&"cc".repeat(32), None, &[]), + TargetResolution::NoKnownAddress { + peer_id: "cc".repeat(32) + } + ); + } + + /// **Proves:** junk input is rejected deterministically. + #[test] + fn an_unparseable_argument_is_rejected() { + for junk in ["", "not-a-peer", "zz".repeat(32).as_str(), "1.2.3.4"] { + assert_eq!( + resolve_target(junk, None, &[]), + TargetResolution::Unparseable, + "{junk:?} is neither an address nor a peer_id" + ); + } + } + + /// **Proves:** the dial target orders its candidates IPv6-first (§5.2), so the ping exercises the + /// ladder in the family order a real dial would. + #[test] + fn the_dial_target_puts_ipv6_candidates_first() { + let target = peer_target( + &"aa".repeat(32), + &[addr("10.0.0.9:9444"), addr("[2001:db8::1]:9444")], + "DIG_MAINNET", + ) + .expect("valid peer_id"); + assert!( + target.direct_addrs()[0].is_ipv6(), + "IPv6 leads the candidate list, got {:?}", + target.direct_addrs() + ); + } + + // -- run_ladder ------------------------------------------------------------------------------ + + /// A dialer that answers from a scripted per-tier table. + struct ScriptedDialer { + connect_on: Vec, + peer_id: String, + /// How long each attempt "takes", so the deadline path is exercisable. + per_tier: Duration, + } + + #[async_trait] + impl TierDialer for ScriptedDialer { + async fn dial_tier( + &self, + tier: TraversalKind, + _target: &PeerTarget, + ) -> Result { + tokio::time::sleep(self.per_tier).await; + if self.connect_on.contains(&tier) { + Ok(DialedPeer { + observed_peer_id: self.peer_id.clone(), + remote_addr: addr("[2001:db8::1]:9444"), + }) + } else { + Err(format!("{} unavailable", tier_name(tier))) + } + } + } + + fn test_target() -> PeerTarget { + PeerTarget::with_addr( + dig_nat::PeerId::from_bytes([0x22; 32]), + addr("[2001:db8::1]:9444"), + "DIG_MAINNET", + ) + } + + /// **Proves:** the ladder keeps probing AFTER a tier succeeds, so the report says whether the + /// direct path was available rather than stopping at the first thing that worked. + /// + /// **Catches:** a short-circuit on first success — which would make the diagnostic unable to + /// answer "was it relayed when direct would have worked?", the question #1929 needed. + #[tokio::test(start_paused = true)] + async fn every_tier_is_probed_even_after_one_succeeds() { + let dialer = ScriptedDialer { + connect_on: vec![TraversalKind::Direct, TraversalKind::Relayed], + peer_id: PEER_A.to_string(), + per_tier: Duration::from_millis(10), + }; + let reports = run_ladder(&dialer, &test_target(), Duration::from_secs(60)).await; + assert_eq!( + reports.len(), + ladder_tiers().len(), + "every rung of the ladder is reported" + ); + assert!( + reports + .iter() + .all(|r| !matches!(r.outcome, TierOutcome::Skipped { .. })), + "nothing is skipped when the deadline is generous" + ); + } + + /// **Proves:** the run is bounded — once the overall deadline passes, the remaining tiers are + /// reported as skipped rather than attempted. + /// + /// **Catches:** an unbounded probe, which #1985 forbids: a black-holed address must not be able + /// to hang the caller. + #[tokio::test(start_paused = true)] + async fn the_run_is_bounded_and_says_which_tiers_it_skipped() { + let dialer = ScriptedDialer { + connect_on: vec![], + peer_id: PEER_A.to_string(), + per_tier: Duration::from_secs(5), + }; + let reports = run_ladder(&dialer, &test_target(), Duration::from_secs(11)).await; + assert_eq!(reports.len(), ladder_tiers().len(), "every rung is accounted for"); + let skipped = reports + .iter() + .filter(|r| matches!(r.outcome, TierOutcome::Skipped { .. })) + .count(); + assert!( + skipped > 0, + "an 11s deadline over 5s-per-tier attempts must skip the tail, got {reports:#?}" + ); + } +}