1use std::{2 collections::VecDeque,3 sync::{Mutex as StdMutex, MutexGuard as StdMutexGuard},4 time::{Duration, Instant},5};67use iroh::endpoint::{Connection, PathId};8use serde::Serialize;910const LATENCY_BUCKETS_MS: usize = 5_001;11pub(crate) const INPUT_ACK_PENDING_CAPACITY: usize = 1_024;1213pub(crate) fn lock_network_diagnostics(14 diagnostics: &StdMutex<NetworkSessionDiagnostics>,15) -> StdMutexGuard<'_, NetworkSessionDiagnostics> {16 diagnostics17 .lock()18 .unwrap_or_else(std::sync::PoisonError::into_inner)19}20212223pub(crate) enum PathMode {24 25 Unknown,26 Direct,27 Relay,28 Custom,29}3031impl PathMode {32 pub(crate) const fn as_str(self) -> &'static str {33 match self {34 Self::Unknown => "unknown",35 Self::Direct => "direct",36 Self::Relay => "relay",37 Self::Custom => "custom",38 }39 }40}414243pub(crate) enum NetworkLeg {44 Media,45 Input,46 Audio,47}484950struct BoundedLatencyHistogram {51 buckets: Box<[u64]>,52 sample_count: u64,53 overflow_total: u64,54 max_ms: Option<u64>,55}5657impl BoundedLatencyHistogram {58 fn new() -> Self {59 Self {60 buckets: vec![0; LATENCY_BUCKETS_MS].into_boxed_slice(),61 ..Self::default()62 }63 }6465 fn record(&mut self, duration: Duration) {66 let millis = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX);67 self.sample_count = self.sample_count.saturating_add(1);68 self.max_ms = Some(self.max_ms.unwrap_or_default().max(millis));69 if let Ok(index) = usize::try_from(millis)70 && let Some(bucket) = self.buckets.get_mut(index)71 {72 *bucket = bucket.saturating_add(1);73 } else {74 self.overflow_total = self.overflow_total.saturating_add(1);75 }76 }7778 fn percentile(&self, percentile: u64) -> Option<u64> {79 if self.sample_count == 0 {80 return None;81 }82 let rank = self.sample_count.saturating_mul(percentile).div_ceil(100);83 let mut seen = 0_u64;84 for (millis, count) in self.buckets.iter().copied().enumerate() {85 seen = seen.saturating_add(count);86 if seen >= rank {87 return u64::try_from(millis).ok();88 }89 }90 None91 }9293 fn summary(&self) -> LatencySummary {94 LatencySummary {95 sample_count: self.sample_count,96 p50_ms: self.percentile(50),97 p95_ms: self.percentile(95),98 max_ms: self.max_ms,99 overflow_total: self.overflow_total,100 }101 }102}103104105struct LatencySummary {106 sample_count: u64,107 p50_ms: Option<u64>,108 p95_ms: Option<u64>,109 max_ms: Option<u64>,110 overflow_total: u64,111}112113114struct LegObservation {115 selected_path_id: Option<PathId>,116 mode: PathMode,117 rtt: Option<Duration>,118 cwnd_bytes: Option<u64>,119 mtu: Option<u16>,120 congestion_events: Option<u64>,121 black_holes_detected: Option<u64>,122 tx_datagrams: u64,123 tx_bytes: u64,124 rx_datagrams: u64,125 rx_bytes: u64,126 lost_packets: u64,127 lost_bytes: u64,128}129130impl LegObservation {131 fn from_connection(connection: &Connection) -> Self {132 let paths = connection.paths();133 let selected = paths.iter().find(|path| path.is_selected());134 let (selected_path_id, mode, rtt, cwnd_bytes, mtu, congestion_events, black_holes_detected) =135 if let Some(path) = selected {136 let stats = path.stats();137 let mode = classify_path_mode(path.is_ip(), path.is_relay());138 (139 Some(path.id()),140 mode,141 Some(stats.rtt),142 Some(stats.cwnd),143 Some(stats.current_mtu),144 Some(stats.congestion_events),145 Some(stats.black_holes_detected),146 )147 } else {148 (None, PathMode::Unknown, None, None, None, None, None)149 };150 let stats = connection.stats();151 Self {152 selected_path_id,153 mode,154 rtt,155 cwnd_bytes,156 mtu,157 congestion_events,158 black_holes_detected,159 tx_datagrams: stats.udp_tx.datagrams,160 tx_bytes: stats.udp_tx.bytes,161 rx_datagrams: stats.udp_rx.datagrams,162 rx_bytes: stats.udp_rx.bytes,163 lost_packets: stats.lost_packets,164 lost_bytes: stats.lost_bytes,165 }166 }167}168169const fn classify_path_mode(is_ip: bool, is_relay: bool) -> PathMode {170 match (is_ip, is_relay) {171 (true, false) => PathMode::Direct,172 (false, true) => PathMode::Relay,173 _ => PathMode::Custom,174 }175}176177178struct ConnectionCounters {179 tx_datagrams: u64,180 tx_bytes: u64,181 rx_datagrams: u64,182 rx_bytes: u64,183 lost_packets: u64,184 lost_bytes: u64,185}186187impl ConnectionCounters {188 fn from_observation(observation: &LegObservation) -> Self {189 Self {190 tx_datagrams: observation.tx_datagrams,191 tx_bytes: observation.tx_bytes,192 rx_datagrams: observation.rx_datagrams,193 rx_bytes: observation.rx_bytes,194 lost_packets: observation.lost_packets,195 lost_bytes: observation.lost_bytes,196 }197 }198199 fn regressed_from(self, previous: Self) -> bool {200 self.tx_datagrams < previous.tx_datagrams201 || self.tx_bytes < previous.tx_bytes202 || self.rx_datagrams < previous.rx_datagrams203 || self.rx_bytes < previous.rx_bytes204 || self.lost_packets < previous.lost_packets205 || self.lost_bytes < previous.lost_bytes206 }207}208209210struct LegDiagnostics {211 mode: PathMode,212 selected_path_id: Option<PathId>,213 sample_count: u64,214 direct_samples: u64,215 relay_samples: u64,216 custom_samples: u64,217 unknown_samples: u64,218 path_epoch: u64,219 path_switch_total: u64,220 mode_transition_total: u64,221 rtt: BoundedLatencyHistogram,222 rtt_current_ms: Option<f64>,223 counters: ConnectionCounters,224 has_counters: bool,225 cwnd_bytes: Option<u64>,226 mtu: Option<u16>,227 congestion_events: Option<u64>,228 black_holes_detected: Option<u64>,229 counter_regression_total: u64,230}231232impl Default for LegDiagnostics {233 fn default() -> Self {234 Self {235 mode: PathMode::Unknown,236 selected_path_id: None,237 sample_count: 0,238 direct_samples: 0,239 relay_samples: 0,240 custom_samples: 0,241 unknown_samples: 0,242 path_epoch: 0,243 path_switch_total: 0,244 mode_transition_total: 0,245 rtt: BoundedLatencyHistogram::new(),246 rtt_current_ms: None,247 counters: ConnectionCounters::default(),248 has_counters: false,249 cwnd_bytes: None,250 mtu: None,251 congestion_events: None,252 black_holes_detected: None,253 counter_regression_total: 0,254 }255 }256}257258impl LegDiagnostics {259 fn observe(&mut self, observation: LegObservation) {260 let previous_mode = self.mode;261 let previous_path_id = self.selected_path_id;262 self.sample_count = self.sample_count.saturating_add(1);263 match observation.mode {264 PathMode::Direct => self.direct_samples = self.direct_samples.saturating_add(1),265 PathMode::Relay => self.relay_samples = self.relay_samples.saturating_add(1),266 PathMode::Custom => self.custom_samples = self.custom_samples.saturating_add(1),267 PathMode::Unknown => self.unknown_samples = self.unknown_samples.saturating_add(1),268 }269 if observation.selected_path_id != previous_path_id {270 if previous_path_id.is_some() {271 self.path_switch_total = self.path_switch_total.saturating_add(1);272 }273 if observation.selected_path_id.is_some() {274 self.path_epoch = self.path_epoch.saturating_add(1);275 }276 }277 if self.sample_count > 1 && previous_mode != observation.mode {278 self.mode_transition_total = self.mode_transition_total.saturating_add(1);279 }280 if let Some(rtt) = observation.rtt {281 self.rtt.record(rtt);282 }283 self.rtt_current_ms = observation.rtt.map(|rtt| rtt.as_secs_f64() * 1_000.0);284 let counters = ConnectionCounters::from_observation(&observation);285 if self.has_counters && counters.regressed_from(self.counters) {286 self.counter_regression_total = self.counter_regression_total.saturating_add(1);287 }288 self.mode = observation.mode;289 self.selected_path_id = observation.selected_path_id;290 self.counters = counters;291 self.has_counters = true;292 self.cwnd_bytes = observation.cwnd_bytes;293 self.mtu = observation.mtu;294 self.congestion_events = observation.congestion_events;295 self.black_holes_detected = observation.black_holes_detected;296 }297298 fn snapshot(&self) -> NetworkLegSnapshot {299 let rtt = self.rtt.summary();300 NetworkLegSnapshot {301 mode: self.mode,302 sample_count: self.sample_count,303 direct_samples: self.direct_samples,304 relay_samples: self.relay_samples,305 custom_samples: self.custom_samples,306 unknown_samples: self.unknown_samples,307 path_epoch: self.path_epoch,308 path_switch_total: self.path_switch_total,309 mode_transition_total: self.mode_transition_total,310 rtt_sample_count: rtt.sample_count,311 rtt_current_ms: self.rtt_current_ms,312 rtt_p50_ms: rtt.p50_ms,313 rtt_p95_ms: rtt.p95_ms,314 rtt_max_ms: rtt.max_ms,315 rtt_overflow_total: rtt.overflow_total,316 tx_datagrams: self.counters.tx_datagrams,317 tx_bytes: self.counters.tx_bytes,318 rx_datagrams: self.counters.rx_datagrams,319 rx_bytes: self.counters.rx_bytes,320 lost_packets: self.counters.lost_packets,321 lost_bytes: self.counters.lost_bytes,322 cwnd_bytes: self.cwnd_bytes,323 mtu: self.mtu,324 congestion_events: self.congestion_events,325 black_holes_detected: self.black_holes_detected,326 counter_regression_total: self.counter_regression_total,327 complete: self.counter_regression_total == 0 && rtt.overflow_total == 0,328 }329 }330}331332333pub(crate) struct NetworkLegSnapshot {334 pub(crate) mode: PathMode,335 sample_count: u64,336 direct_samples: u64,337 relay_samples: u64,338 custom_samples: u64,339 unknown_samples: u64,340 path_epoch: u64,341 path_switch_total: u64,342 mode_transition_total: u64,343 rtt_sample_count: u64,344 pub(crate) rtt_current_ms: Option<f64>,345 pub(crate) rtt_p50_ms: Option<u64>,346 rtt_p95_ms: Option<u64>,347 rtt_max_ms: Option<u64>,348 rtt_overflow_total: u64,349 tx_datagrams: u64,350 tx_bytes: u64,351 rx_datagrams: u64,352 rx_bytes: u64,353 lost_packets: u64,354 lost_bytes: u64,355 cwnd_bytes: Option<u64>,356 mtu: Option<u16>,357 congestion_events: Option<u64>,358 black_holes_detected: Option<u64>,359 counter_regression_total: u64,360 complete: bool,361}362363364struct InputAckDiagnostics {365 negotiated: bool,366 sent_total: u64,367 acknowledged_total: u64,368 last_ack: Option<u64>,369 pending: VecDeque<(u64, Instant)>,370 duplicate_total: u64,371 untracked_total: u64,372 malformed: bool,373 closed: bool,374 latency: BoundedLatencyHistogram,375}376377impl InputAckDiagnostics {378 fn new(negotiated: bool) -> Self {379 Self {380 negotiated,381 sent_total: 0,382 acknowledged_total: 0,383 last_ack: None,384 pending: VecDeque::with_capacity(INPUT_ACK_PENDING_CAPACITY),385 duplicate_total: 0,386 untracked_total: 0,387 malformed: false,388 closed: false,389 latency: BoundedLatencyHistogram::new(),390 }391 }392393 fn begin_send(&mut self, now: Instant) -> Option<u64> {394 if !self.negotiated {395 return None;396 }397 self.sent_total = self.sent_total.saturating_add(1);398 if self.pending.len() == INPUT_ACK_PENDING_CAPACITY {399 self.pending.pop_front();400 self.untracked_total = self.untracked_total.saturating_add(1);401 }402 self.pending.push_back((self.sent_total, now));403 Some(self.sent_total)404 }405406 fn observe_ack(&mut self, sequence: u64, now: Instant) -> Result<(), String> {407 if !self.negotiated {408 self.malformed = true;409 return Err("Host sent an input acknowledgement that was not negotiated".to_string());410 }411 if self.last_ack == Some(sequence) {412 self.duplicate_total = self.duplicate_total.saturating_add(1);413 return Ok(());414 }415 if self.last_ack.is_some_and(|last| sequence < last) || sequence > self.sent_total {416 self.malformed = true;417 return Err(format!(418 "Host sent invalid input acknowledgement {sequence} after {} local events",419 self.sent_total420 ));421 }422 if sequence == 0 && self.last_ack.is_none() {423 self.last_ack = Some(0);424 return Ok(());425 }426 while self427 .pending428 .front()429 .is_some_and(|(pending, _)| *pending <= sequence)430 {431 let (_, sent_at) = self.pending.pop_front().expect("front checked");432 self.latency.record(now.saturating_duration_since(sent_at));433 }434 self.acknowledged_total = sequence;435 self.last_ack = Some(sequence);436 Ok(())437 }438439 fn snapshot(&self) -> InputAckSnapshot {440 let latency = self.latency.summary();441 InputAckSnapshot {442 negotiated: self.negotiated,443 sent_total: self.sent_total,444 acknowledged_total: self.acknowledged_total,445 pending_count: self.pending.len(),446 pending_capacity: INPUT_ACK_PENDING_CAPACITY,447 duplicate_total: self.duplicate_total,448 untracked_total: self.untracked_total,449 malformed: self.malformed,450 closed: self.closed,451 latency_sample_count: latency.sample_count,452 latency_p50_ms: latency.p50_ms,453 latency_p95_ms: latency.p95_ms,454 latency_max_ms: latency.max_ms,455 latency_overflow_total: latency.overflow_total,456 complete: !self.malformed457 && !self.closed458 && self.untracked_total == 0459 && latency.overflow_total == 0,460 }461 }462}463464465pub(crate) struct InputAckSnapshot {466 negotiated: bool,467 sent_total: u64,468 acknowledged_total: u64,469 pending_count: usize,470 pending_capacity: usize,471 duplicate_total: u64,472 untracked_total: u64,473 malformed: bool,474 closed: bool,475 latency_sample_count: u64,476 latency_p50_ms: Option<u64>,477 latency_p95_ms: Option<u64>,478 latency_max_ms: Option<u64>,479 latency_overflow_total: u64,480 complete: bool,481}482483484pub(crate) struct NetworkSessionDiagnostics {485 started_at: Instant,486 media: LegDiagnostics,487 input: LegDiagnostics,488 audio: Option<LegDiagnostics>,489 input_ack: InputAckDiagnostics,490}491492impl NetworkSessionDiagnostics {493 pub(crate) fn new(started_at: Instant, input_ack_negotiated: bool) -> Self {494 Self {495 started_at,496 media: LegDiagnostics::default(),497 input: LegDiagnostics::default(),498 audio: None,499 input_ack: InputAckDiagnostics::new(input_ack_negotiated),500 }501 }502503 pub(crate) fn observe_connection(&mut self, leg: NetworkLeg, connection: &Connection) {504 let diagnostics = match leg {505 NetworkLeg::Media => &mut self.media,506 NetworkLeg::Input => &mut self.input,507 NetworkLeg::Audio => self.audio.get_or_insert_with(LegDiagnostics::default),508 };509 diagnostics.observe(LegObservation::from_connection(connection));510 }511512 pub(crate) fn begin_input_send(&mut self, now: Instant) {513 let _ = self.input_ack.begin_send(now);514 }515516 pub(crate) fn observe_input_ack(&mut self, sequence: u64, now: Instant) -> Result<(), String> {517 self.input_ack.observe_ack(sequence, now)518 }519520 pub(crate) fn mark_input_feedback_closed(&mut self) {521 if self.input_ack.negotiated {522 self.input_ack.closed = true;523 }524 }525526 pub(crate) fn mark_input_feedback_malformed(&mut self) {527 if self.input_ack.negotiated {528 self.input_ack.malformed = true;529 }530 }531532 pub(crate) fn snapshot(&self, now: Instant) -> NetworkDiagnosticsSnapshot {533 NetworkDiagnosticsSnapshot {534 version: 1,535 session_elapsed_ms: u64::try_from(536 now.saturating_duration_since(self.started_at).as_millis(),537 )538 .unwrap_or(u64::MAX),539 media: self.media.snapshot(),540 input: self.input.snapshot(),541 audio: self.audio.as_ref().map(LegDiagnostics::snapshot),542 input_ack: self.input_ack.snapshot(),543 }544 }545}546547548pub(crate) struct NetworkDiagnosticsSnapshot {549 version: u8,550 session_elapsed_ms: u64,551 pub(crate) media: NetworkLegSnapshot,552 input: NetworkLegSnapshot,553 audio: Option<NetworkLegSnapshot>,554 input_ack: InputAckSnapshot,555}556557558impl NetworkDiagnosticsSnapshot {559 pub(crate) fn test_fixture(mode: PathMode, rtt: Option<Duration>) -> Self {560 let mut diagnostics = NetworkSessionDiagnostics::new(Instant::now(), false);561 diagnostics.media.mode = mode;562 diagnostics.media.sample_count = 1;563 match mode {564 PathMode::Direct => diagnostics.media.direct_samples = 1,565 PathMode::Relay => diagnostics.media.relay_samples = 1,566 PathMode::Custom => diagnostics.media.custom_samples = 1,567 PathMode::Unknown => diagnostics.media.unknown_samples = 1,568 }569 if let Some(rtt) = rtt {570 diagnostics.media.rtt.record(rtt);571 diagnostics.media.rtt_current_ms = Some(rtt.as_secs_f64() * 1_000.0);572 }573 diagnostics.snapshot(Instant::now())574 }575}576577578mod tests {579 use super::*;580581 fn observation(mode: PathMode, counters: u64) -> LegObservation {582 LegObservation {583 selected_path_id: Some(PathId::ZERO),584 mode,585 rtt: Some(Duration::from_millis(10 + counters)),586 cwnd_bytes: Some(64_000),587 mtu: Some(1_200),588 congestion_events: Some(0),589 black_holes_detected: Some(0),590 tx_datagrams: counters,591 tx_bytes: counters * 100,592 rx_datagrams: counters,593 rx_bytes: counters * 100,594 lost_packets: 0,595 lost_bytes: 0,596 }597 }598599 600 fn path_mode_adapter_is_fail_closed_for_ambiguous_paths() {601 assert_eq!(classify_path_mode(true, false), PathMode::Direct);602 assert_eq!(classify_path_mode(false, true), PathMode::Relay);603 assert_eq!(classify_path_mode(false, false), PathMode::Custom);604 assert_eq!(classify_path_mode(true, true), PathMode::Custom);605 }606607 608 fn path_residency_and_transitions_remain_bounded() {609 let mut diagnostics = LegDiagnostics::default();610 diagnostics.observe(observation(PathMode::Direct, 1));611 diagnostics.observe(observation(PathMode::Direct, 2));612 diagnostics.observe(observation(PathMode::Relay, 3));613 let snapshot = diagnostics.snapshot();614 assert_eq!(snapshot.sample_count, 3);615 assert_eq!(snapshot.direct_samples, 2);616 assert_eq!(snapshot.relay_samples, 1);617 assert_eq!(snapshot.mode_transition_total, 1);618 assert!(snapshot.complete);619 }620621 622 fn same_mode_path_switch_advances_epoch_without_mode_transition() {623 let mut diagnostics = LegDiagnostics::default();624 diagnostics.observe(observation(PathMode::Direct, 1));625 let mut replacement = observation(PathMode::Direct, 2);626 replacement.selected_path_id = Some(PathId::MAX);627 diagnostics.observe(replacement);628 let snapshot = diagnostics.snapshot();629 assert_eq!(snapshot.path_epoch, 2);630 assert_eq!(snapshot.path_switch_total, 1);631 assert_eq!(snapshot.mode_transition_total, 0);632 }633634 635 fn current_rtt_preserves_sub_millisecond_precision() {636 let mut diagnostics = LegDiagnostics::default();637 let mut sample = observation(PathMode::Direct, 1);638 sample.rtt = Some(Duration::from_micros(800));639 diagnostics.observe(sample);640 let current = diagnostics.snapshot().rtt_current_ms.unwrap();641 assert!((current - 0.8).abs() < f64::EPSILON);642 }643644 645 fn counter_regression_marks_the_leg_incomplete() {646 let mut diagnostics = LegDiagnostics::default();647 diagnostics.observe(observation(PathMode::Direct, 5));648 diagnostics.observe(observation(PathMode::Direct, 4));649 let snapshot = diagnostics.snapshot();650 assert_eq!(snapshot.counter_regression_total, 1);651 assert!(!snapshot.complete);652 }653654 655 fn latency_overflow_is_visible_instead_of_clamped() {656 let mut histogram = BoundedLatencyHistogram::new();657 histogram.record(Duration::from_millis(5_001));658 let summary = histogram.summary();659 assert_eq!(summary.sample_count, 1);660 assert_eq!(summary.p95_ms, None);661 assert_eq!(summary.max_ms, Some(5_001));662 assert_eq!(summary.overflow_total, 1);663 }664665 666 fn cumulative_ack_accounts_for_each_pending_event() {667 let started = Instant::now();668 let mut diagnostics = InputAckDiagnostics::new(true);669 diagnostics.begin_send(started);670 diagnostics.begin_send(started + Duration::from_millis(1));671 diagnostics.begin_send(started + Duration::from_millis(2));672 diagnostics673 .observe_ack(3, started + Duration::from_millis(12))674 .unwrap();675 let snapshot = diagnostics.snapshot();676 assert_eq!(snapshot.acknowledged_total, 3);677 assert_eq!(snapshot.pending_count, 0);678 assert_eq!(snapshot.latency_sample_count, 3);679 assert_eq!(snapshot.latency_p50_ms, Some(11));680 }681682 683 fn initial_and_duplicate_ack_do_not_invent_latency() {684 let now = Instant::now();685 let mut diagnostics = InputAckDiagnostics::new(true);686 diagnostics.observe_ack(0, now).unwrap();687 diagnostics.observe_ack(0, now).unwrap();688 let snapshot = diagnostics.snapshot();689 assert_eq!(snapshot.duplicate_total, 1);690 assert_eq!(snapshot.latency_sample_count, 0);691 }692693 694 fn lower_and_future_ack_fail_closed() {695 let now = Instant::now();696 let mut diagnostics = InputAckDiagnostics::new(true);697 diagnostics.begin_send(now);698 diagnostics.observe_ack(1, now).unwrap();699 assert!(diagnostics.observe_ack(0, now).is_err());700 assert!(diagnostics.observe_ack(2, now).is_err());701 assert!(diagnostics.snapshot().malformed);702 }703704 705 fn pending_overflow_drops_telemetry_not_input() {706 let now = Instant::now();707 let mut diagnostics = InputAckDiagnostics::new(true);708 for _ in 0..=INPUT_ACK_PENDING_CAPACITY {709 diagnostics.begin_send(now);710 }711 let snapshot = diagnostics.snapshot();712 assert_eq!(snapshot.sent_total, (INPUT_ACK_PENDING_CAPACITY + 1) as u64);713 assert_eq!(snapshot.pending_count, INPUT_ACK_PENDING_CAPACITY);714 assert_eq!(snapshot.untracked_total, 1);715 assert!(!snapshot.complete);716 }717718 719 fn serialized_snapshot_omits_path_and_peer_identity_material() {720 let now = Instant::now();721 let snapshot = NetworkSessionDiagnostics::new(now, true).snapshot(now);722 let json = serde_json::to_string(&snapshot).unwrap();723 for forbidden in [724 "path_id",725 "remote_addr",726 "local_addr",727 "node_id",728 "relay_url",729 ] {730 assert!(731 !json.contains(forbidden),732 "serialized forbidden field {forbidden}"733 );734 }735 }736}