1use std::{2 collections::VecDeque,3 sync::Arc,4 sync::atomic::{AtomicU64, Ordering},5 sync::{Mutex as StdMutex, MutexGuard as StdMutexGuard},6 time::{Duration, Instant},7};89use iroh::endpoint::{Connection, PathId};10use serde::Serialize;11use sigil_protocol::{12 AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1, FocusStateV2,13 MediaFeedbackFlags, MediaFeedbackReportV1, SessionSnapshotV2,14};1516const LATENCY_BUCKETS_MS: usize = 5_001;17pub(crate) const INPUT_ACK_PENDING_CAPACITY: usize = 1_024;18const MAX_DIAGNOSTIC_VIEWERS: usize = 8;1920pub(crate) fn lock_network_diagnostics(21 diagnostics: &StdMutex<NetworkSessionDiagnostics>,22) -> StdMutexGuard<'_, NetworkSessionDiagnostics> {23 diagnostics24 .lock()25 .unwrap_or_else(std::sync::PoisonError::into_inner)26}27282930pub(crate) enum PathMode {31 32 Unknown,33 Direct,34 Relay,35 Custom,36}3738impl PathMode {39 pub(crate) const fn as_str(self) -> &'static str {40 match self {41 Self::Unknown => "unknown",42 Self::Direct => "direct",43 Self::Relay => "relay",44 Self::Custom => "custom",45 }46 }47}484950pub(crate) enum NetworkLeg {51 Media,52 Input,53 Audio,54}55565758pub(crate) enum MediaDeliveryRole {59 60 DirectHost,61 62 ViewerRelay,63}646566struct MediaAuthenticationDiagnostics {67 mode: &'static str,68 generation_certified: bool,69 delivery_role: MediaDeliveryRole,70 verification_failures: Arc<AtomicU64>,71}727374struct SessionProtocolDiagnostics {75 mode: &'static str,76 local_handle: Option<String>,77 focus_generation: Option<u64>,78 roster_revision: Option<u64>,79 roster_updated_at: Option<Instant>,80 subscription_expires_at_unix: Option<u64>,81 stale_snapshot_total: u64,82 invalid_snapshot_total: u64,83}8485impl Default for SessionProtocolDiagnostics {86 fn default() -> Self {87 Self {88 mode: "legacy_exclusive_v1",89 local_handle: None,90 focus_generation: None,91 roster_revision: None,92 roster_updated_at: None,93 subscription_expires_at_unix: None,94 stale_snapshot_total: 0,95 invalid_snapshot_total: 0,96 }97 }98}99100impl SessionProtocolDiagnostics {101 fn configure_v2(102 &mut self,103 snapshot: &SessionSnapshotV2,104 subscription_expires_at_unix: u64,105 now: Instant,106 ) -> Result<(), String> {107 self.mode = "moq_multi_viewer_v2";108 self.subscription_expires_at_unix = Some(subscription_expires_at_unix);109 self.observe_snapshot(snapshot, now, true)110 }111112 fn observe_snapshot(113 &mut self,114 snapshot: &SessionSnapshotV2,115 now: Instant,116 initial: bool,117 ) -> Result<(), String> {118 if let Err(error) = snapshot.validate() {119 self.invalid_snapshot_total = self.invalid_snapshot_total.saturating_add(1);120 return Err(format!("Invalid session diagnostics snapshot: {error}"));121 }122 if snapshot.viewers.len() > MAX_DIAGNOSTIC_VIEWERS {123 self.invalid_snapshot_total = self.invalid_snapshot_total.saturating_add(1);124 return Err("Session diagnostics roster exceeds the hard viewer bound".to_string());125 }126 if !initial127 && self128 .roster_revision129 .is_some_and(|revision| snapshot.revision <= revision)130 {131 self.stale_snapshot_total = self.stale_snapshot_total.saturating_add(1);132 return Ok(());133 }134 let local = snapshot.self_viewer();135 self.local_handle = Some(local.presence_id.as_str().to_owned());136 self.focus_generation = match &snapshot.focus {137 FocusStateV2::Held {138 holder,139 session_id,140 focus_generation,141 ..142 } if holder == &local.presence_id && *session_id == local.session_id => {143 Some(*focus_generation)144 }145 _ => None,146 };147 self.roster_revision = Some(snapshot.revision);148 self.roster_updated_at = Some(now);149 Ok(())150 }151152 fn snapshot(&self, now: Instant, now_unix: u64) -> SessionProtocolSnapshot {153 SessionProtocolSnapshot {154 mode: self.mode,155 local_handle: self.local_handle.clone(),156 focus_generation: self.focus_generation,157 roster_revision: self.roster_revision,158 roster_age_ms: self.roster_updated_at.map(|updated| {159 u64::try_from(now.saturating_duration_since(updated).as_millis())160 .unwrap_or(u64::MAX)161 }),162 subscription_expires_at_unix: self.subscription_expires_at_unix,163 subscription_seconds_remaining: self164 .subscription_expires_at_unix165 .map(|expiry| expiry.saturating_sub(now_unix)),166 stale_snapshot_total: self.stale_snapshot_total,167 invalid_snapshot_total: self.invalid_snapshot_total,168 }169 }170}171172173pub(crate) struct SessionProtocolSnapshot {174 mode: &'static str,175 local_handle: Option<String>,176 focus_generation: Option<u64>,177 roster_revision: Option<u64>,178 roster_age_ms: Option<u64>,179 subscription_expires_at_unix: Option<u64>,180 subscription_seconds_remaining: Option<u64>,181 stale_snapshot_total: u64,182 invalid_snapshot_total: u64,183}184185186pub(crate) struct AdaptiveViewerSnapshot {187 scope: &'static str,188 local_pressure: &'static str,189 aggregate_target_kbps: Option<u32>,190 aggregate_state: &'static str,191 recovery_state: &'static str,192 applied: bool,193}194195impl Default for AdaptiveViewerSnapshot {196 fn default() -> Self {197 Self {198 scope: "local_viewer",199 local_pressure: "awaiting_report",200 aggregate_target_kbps: None,201 aggregate_state: "unavailable",202 recovery_state: "active",203 applied: false,204 }205 }206}207208impl AdaptiveViewerSnapshot {209 fn observe_report(&mut self, report: &MediaFeedbackReportV1) {210 let drops = report211 .transport_dropped_delta212 .saturating_add(report.frontend_dropped_delta)213 .saturating_add(report.decoder_dropped_delta)214 .saturating_add(report.presenter_dropped_delta);215 let queued = report.frontend_queue_depth > 0216 || report.decode_queue_depth > 0217 || report.presenter_queue_depth > 0;218 self.local_pressure = if report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE) {219 "recovering"220 } else if drops > 0 || queued {221 "pressured"222 } else {223 "clean"224 };225 }226227 fn observe_decision(&mut self, decision: &AdaptiveBitrateDecisionV1) {228 self.aggregate_target_kbps = Some(decision.target_kbps);229 self.aggregate_state = match decision.state {230 AdaptiveBitrateStateV1::Hold => "hold",231 AdaptiveBitrateStateV1::Decrease => "decrease",232 AdaptiveBitrateStateV1::Increase => "increase",233 };234 let local_pressure = decision235 .reasons236 .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE)237 || decision238 .reasons239 .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)240 || decision241 .reasons242 .contains(AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY);243 self.recovery_state = if local_pressure && decision.target_kbps == decision.floor_kbps {244 "floor_limited"245 } else {246 "active"247 };248 self.applied = decision.applied;249 }250251 fn detach(&mut self) {252 self.recovery_state = "detached";253 }254}255256impl Default for MediaAuthenticationDiagnostics {257 fn default() -> Self {258 Self {259 mode: "transport-authenticated-v1",260 generation_certified: false,261 delivery_role: MediaDeliveryRole::DirectHost,262 verification_failures: Arc::new(AtomicU64::new(0)),263 }264 }265}266267impl MediaAuthenticationDiagnostics {268 fn snapshot(&self) -> MediaAuthenticationSnapshot {269 MediaAuthenticationSnapshot {270 mode: self.mode,271 generation_certified: self.generation_certified,272 delivery_role: self.delivery_role,273 verification_failures: self.verification_failures.load(Ordering::Relaxed),274 }275 }276}277278279pub(crate) struct MediaAuthenticationSnapshot {280 mode: &'static str,281 generation_certified: bool,282 delivery_role: MediaDeliveryRole,283 verification_failures: u64,284}285286287struct BoundedLatencyHistogram {288 buckets: Box<[u64]>,289 sample_count: u64,290 overflow_total: u64,291 max_ms: Option<u64>,292}293294impl BoundedLatencyHistogram {295 fn new() -> Self {296 Self {297 buckets: vec![0; LATENCY_BUCKETS_MS].into_boxed_slice(),298 ..Self::default()299 }300 }301302 fn record(&mut self, duration: Duration) {303 let millis = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX);304 self.sample_count = self.sample_count.saturating_add(1);305 self.max_ms = Some(self.max_ms.unwrap_or_default().max(millis));306 if let Ok(index) = usize::try_from(millis)307 && let Some(bucket) = self.buckets.get_mut(index)308 {309 *bucket = bucket.saturating_add(1);310 } else {311 self.overflow_total = self.overflow_total.saturating_add(1);312 }313 }314315 fn percentile(&self, percentile: u64) -> Option<u64> {316 if self.sample_count == 0 {317 return None;318 }319 let rank = self.sample_count.saturating_mul(percentile).div_ceil(100);320 let mut seen = 0_u64;321 for (millis, count) in self.buckets.iter().copied().enumerate() {322 seen = seen.saturating_add(count);323 if seen >= rank {324 return u64::try_from(millis).ok();325 }326 }327 None328 }329330 fn summary(&self) -> LatencySummary {331 LatencySummary {332 sample_count: self.sample_count,333 p50_ms: self.percentile(50),334 p95_ms: self.percentile(95),335 max_ms: self.max_ms,336 overflow_total: self.overflow_total,337 }338 }339}340341342struct LatencySummary {343 sample_count: u64,344 p50_ms: Option<u64>,345 p95_ms: Option<u64>,346 max_ms: Option<u64>,347 overflow_total: u64,348}349350351struct LegObservation {352 selected_path_id: Option<PathId>,353 mode: PathMode,354 rtt: Option<Duration>,355 cwnd_bytes: Option<u64>,356 mtu: Option<u16>,357 congestion_events: Option<u64>,358 black_holes_detected: Option<u64>,359 tx_datagrams: u64,360 tx_bytes: u64,361 rx_datagrams: u64,362 rx_bytes: u64,363 lost_packets: u64,364 lost_bytes: u64,365}366367impl LegObservation {368 fn from_connection(connection: &Connection) -> Self {369 let paths = connection.paths();370 let selected = paths.iter().find(|path| path.is_selected());371 let (selected_path_id, mode, rtt, cwnd_bytes, mtu, congestion_events, black_holes_detected) =372 if let Some(path) = selected {373 let stats = path.stats();374 let mode = classify_path_mode(path.is_ip(), path.is_relay());375 (376 Some(path.id()),377 mode,378 Some(stats.rtt),379 Some(stats.cwnd),380 Some(stats.current_mtu),381 Some(stats.congestion_events),382 Some(stats.black_holes_detected),383 )384 } else {385 (None, PathMode::Unknown, None, None, None, None, None)386 };387 let stats = connection.stats();388 Self {389 selected_path_id,390 mode,391 rtt,392 cwnd_bytes,393 mtu,394 congestion_events,395 black_holes_detected,396 tx_datagrams: stats.udp_tx.datagrams,397 tx_bytes: stats.udp_tx.bytes,398 rx_datagrams: stats.udp_rx.datagrams,399 rx_bytes: stats.udp_rx.bytes,400 lost_packets: stats.lost_packets,401 lost_bytes: stats.lost_bytes,402 }403 }404}405406const fn classify_path_mode(is_ip: bool, is_relay: bool) -> PathMode {407 match (is_ip, is_relay) {408 (true, false) => PathMode::Direct,409 (false, true) => PathMode::Relay,410 _ => PathMode::Custom,411 }412}413414415struct ConnectionCounters {416 tx_datagrams: u64,417 tx_bytes: u64,418 rx_datagrams: u64,419 rx_bytes: u64,420 lost_packets: u64,421 lost_bytes: u64,422}423424impl ConnectionCounters {425 fn from_observation(observation: &LegObservation) -> Self {426 Self {427 tx_datagrams: observation.tx_datagrams,428 tx_bytes: observation.tx_bytes,429 rx_datagrams: observation.rx_datagrams,430 rx_bytes: observation.rx_bytes,431 lost_packets: observation.lost_packets,432 lost_bytes: observation.lost_bytes,433 }434 }435436 fn regressed_from(self, previous: Self) -> bool {437 self.tx_datagrams < previous.tx_datagrams438 || self.tx_bytes < previous.tx_bytes439 || self.rx_datagrams < previous.rx_datagrams440 || self.rx_bytes < previous.rx_bytes441 || self.lost_packets < previous.lost_packets442 || self.lost_bytes < previous.lost_bytes443 }444}445446447struct LegDiagnostics {448 mode: PathMode,449 selected_path_id: Option<PathId>,450 sample_count: u64,451 direct_samples: u64,452 relay_samples: u64,453 custom_samples: u64,454 unknown_samples: u64,455 path_epoch: u64,456 path_switch_total: u64,457 mode_transition_total: u64,458 rtt: BoundedLatencyHistogram,459 rtt_current_ms: Option<f64>,460 counters: ConnectionCounters,461 has_counters: bool,462 cwnd_bytes: Option<u64>,463 mtu: Option<u16>,464 congestion_events: Option<u64>,465 black_holes_detected: Option<u64>,466 counter_regression_total: u64,467}468469impl Default for LegDiagnostics {470 fn default() -> Self {471 Self {472 mode: PathMode::Unknown,473 selected_path_id: None,474 sample_count: 0,475 direct_samples: 0,476 relay_samples: 0,477 custom_samples: 0,478 unknown_samples: 0,479 path_epoch: 0,480 path_switch_total: 0,481 mode_transition_total: 0,482 rtt: BoundedLatencyHistogram::new(),483 rtt_current_ms: None,484 counters: ConnectionCounters::default(),485 has_counters: false,486 cwnd_bytes: None,487 mtu: None,488 congestion_events: None,489 black_holes_detected: None,490 counter_regression_total: 0,491 }492 }493}494495impl LegDiagnostics {496 fn observe(&mut self, observation: LegObservation) {497 let previous_mode = self.mode;498 let previous_path_id = self.selected_path_id;499 self.sample_count = self.sample_count.saturating_add(1);500 match observation.mode {501 PathMode::Direct => self.direct_samples = self.direct_samples.saturating_add(1),502 PathMode::Relay => self.relay_samples = self.relay_samples.saturating_add(1),503 PathMode::Custom => self.custom_samples = self.custom_samples.saturating_add(1),504 PathMode::Unknown => self.unknown_samples = self.unknown_samples.saturating_add(1),505 }506 if observation.selected_path_id != previous_path_id {507 if previous_path_id.is_some() {508 self.path_switch_total = self.path_switch_total.saturating_add(1);509 }510 if observation.selected_path_id.is_some() {511 self.path_epoch = self.path_epoch.saturating_add(1);512 }513 }514 if self.sample_count > 1 && previous_mode != observation.mode {515 self.mode_transition_total = self.mode_transition_total.saturating_add(1);516 }517 if let Some(rtt) = observation.rtt {518 self.rtt.record(rtt);519 }520 self.rtt_current_ms = observation.rtt.map(|rtt| rtt.as_secs_f64() * 1_000.0);521 let counters = ConnectionCounters::from_observation(&observation);522 if self.has_counters && counters.regressed_from(self.counters) {523 self.counter_regression_total = self.counter_regression_total.saturating_add(1);524 }525 self.mode = observation.mode;526 self.selected_path_id = observation.selected_path_id;527 self.counters = counters;528 self.has_counters = true;529 self.cwnd_bytes = observation.cwnd_bytes;530 self.mtu = observation.mtu;531 self.congestion_events = observation.congestion_events;532 self.black_holes_detected = observation.black_holes_detected;533 }534535 fn snapshot(&self) -> NetworkLegSnapshot {536 let rtt = self.rtt.summary();537 NetworkLegSnapshot {538 mode: self.mode,539 sample_count: self.sample_count,540 direct_samples: self.direct_samples,541 relay_samples: self.relay_samples,542 custom_samples: self.custom_samples,543 unknown_samples: self.unknown_samples,544 path_epoch: self.path_epoch,545 path_switch_total: self.path_switch_total,546 mode_transition_total: self.mode_transition_total,547 rtt_sample_count: rtt.sample_count,548 rtt_current_ms: self.rtt_current_ms,549 rtt_p50_ms: rtt.p50_ms,550 rtt_p95_ms: rtt.p95_ms,551 rtt_max_ms: rtt.max_ms,552 rtt_overflow_total: rtt.overflow_total,553 tx_datagrams: self.counters.tx_datagrams,554 tx_bytes: self.counters.tx_bytes,555 rx_datagrams: self.counters.rx_datagrams,556 rx_bytes: self.counters.rx_bytes,557 lost_packets: self.counters.lost_packets,558 lost_bytes: self.counters.lost_bytes,559 cwnd_bytes: self.cwnd_bytes,560 mtu: self.mtu,561 congestion_events: self.congestion_events,562 black_holes_detected: self.black_holes_detected,563 counter_regression_total: self.counter_regression_total,564 complete: self.counter_regression_total == 0 && rtt.overflow_total == 0,565 }566 }567}568569570pub(crate) struct NetworkLegSnapshot {571 pub(crate) mode: PathMode,572 sample_count: u64,573 direct_samples: u64,574 relay_samples: u64,575 custom_samples: u64,576 unknown_samples: u64,577 path_epoch: u64,578 path_switch_total: u64,579 mode_transition_total: u64,580 rtt_sample_count: u64,581 pub(crate) rtt_current_ms: Option<f64>,582 pub(crate) rtt_p50_ms: Option<u64>,583 rtt_p95_ms: Option<u64>,584 rtt_max_ms: Option<u64>,585 rtt_overflow_total: u64,586 tx_datagrams: u64,587 tx_bytes: u64,588 rx_datagrams: u64,589 rx_bytes: u64,590 lost_packets: u64,591 lost_bytes: u64,592 cwnd_bytes: Option<u64>,593 mtu: Option<u16>,594 congestion_events: Option<u64>,595 black_holes_detected: Option<u64>,596 counter_regression_total: u64,597 complete: bool,598}599600601struct InputAckDiagnostics {602 negotiated: bool,603 sent_total: u64,604 acknowledged_total: u64,605 last_ack: Option<u64>,606 pending: VecDeque<(u64, Instant)>,607 duplicate_total: u64,608 untracked_total: u64,609 malformed: bool,610 closed: bool,611 latency: BoundedLatencyHistogram,612}613614impl InputAckDiagnostics {615 fn new(negotiated: bool) -> Self {616 Self {617 negotiated,618 sent_total: 0,619 acknowledged_total: 0,620 last_ack: None,621 pending: VecDeque::with_capacity(INPUT_ACK_PENDING_CAPACITY),622 duplicate_total: 0,623 untracked_total: 0,624 malformed: false,625 closed: false,626 latency: BoundedLatencyHistogram::new(),627 }628 }629630 fn begin_send(&mut self, now: Instant) -> Option<u64> {631 if !self.negotiated {632 return None;633 }634 self.sent_total = self.sent_total.saturating_add(1);635 if self.pending.len() == INPUT_ACK_PENDING_CAPACITY {636 self.pending.pop_front();637 self.untracked_total = self.untracked_total.saturating_add(1);638 }639 self.pending.push_back((self.sent_total, now));640 Some(self.sent_total)641 }642643 fn observe_ack(&mut self, sequence: u64, now: Instant) -> Result<(), String> {644 if !self.negotiated {645 self.malformed = true;646 return Err("Host sent an input acknowledgement that was not negotiated".to_string());647 }648 if self.last_ack == Some(sequence) {649 self.duplicate_total = self.duplicate_total.saturating_add(1);650 return Ok(());651 }652 if self.last_ack.is_some_and(|last| sequence < last) || sequence > self.sent_total {653 self.malformed = true;654 return Err(format!(655 "Host sent invalid input acknowledgement {sequence} after {} local events",656 self.sent_total657 ));658 }659 if sequence == 0 && self.last_ack.is_none() {660 self.last_ack = Some(0);661 return Ok(());662 }663 while self664 .pending665 .front()666 .is_some_and(|(pending, _)| *pending <= sequence)667 {668 let (_, sent_at) = self.pending.pop_front().expect("front checked");669 self.latency.record(now.saturating_duration_since(sent_at));670 }671 self.acknowledged_total = sequence;672 self.last_ack = Some(sequence);673 Ok(())674 }675676 fn snapshot(&self) -> InputAckSnapshot {677 let latency = self.latency.summary();678 InputAckSnapshot {679 negotiated: self.negotiated,680 sent_total: self.sent_total,681 acknowledged_total: self.acknowledged_total,682 pending_count: self.pending.len(),683 pending_capacity: INPUT_ACK_PENDING_CAPACITY,684 duplicate_total: self.duplicate_total,685 untracked_total: self.untracked_total,686 malformed: self.malformed,687 closed: self.closed,688 latency_sample_count: latency.sample_count,689 latency_p50_ms: latency.p50_ms,690 latency_p95_ms: latency.p95_ms,691 latency_max_ms: latency.max_ms,692 latency_overflow_total: latency.overflow_total,693 complete: !self.malformed694 && !self.closed695 && self.untracked_total == 0696 && latency.overflow_total == 0,697 }698 }699}700701702pub(crate) struct InputAckSnapshot {703 negotiated: bool,704 sent_total: u64,705 acknowledged_total: u64,706 pending_count: usize,707 pending_capacity: usize,708 duplicate_total: u64,709 untracked_total: u64,710 malformed: bool,711 closed: bool,712 latency_sample_count: u64,713 latency_p50_ms: Option<u64>,714 latency_p95_ms: Option<u64>,715 latency_max_ms: Option<u64>,716 latency_overflow_total: u64,717 complete: bool,718}719720721pub(crate) struct NetworkSessionDiagnostics {722 started_at: Instant,723 media: LegDiagnostics,724 input: LegDiagnostics,725 audio: Option<LegDiagnostics>,726 input_ack: InputAckDiagnostics,727 media_authentication: MediaAuthenticationDiagnostics,728 adaptive: AdaptiveViewerSnapshot,729 session: SessionProtocolDiagnostics,730}731732impl NetworkSessionDiagnostics {733 pub(crate) fn new(started_at: Instant, input_ack_negotiated: bool) -> Self {734 Self {735 started_at,736 media: LegDiagnostics::default(),737 input: LegDiagnostics::default(),738 audio: None,739 input_ack: InputAckDiagnostics::new(input_ack_negotiated),740 media_authentication: MediaAuthenticationDiagnostics::default(),741 adaptive: AdaptiveViewerSnapshot::default(),742 session: SessionProtocolDiagnostics::default(),743 }744 }745746 pub(crate) fn configure_v2_session(747 &mut self,748 snapshot: &SessionSnapshotV2,749 subscription_expires_at_unix: u64,750 now: Instant,751 ) -> Result<(), String> {752 self.session753 .configure_v2(snapshot, subscription_expires_at_unix, now)754 }755756 pub(crate) fn observe_session_snapshot(757 &mut self,758 snapshot: &SessionSnapshotV2,759 now: Instant,760 ) -> Result<(), String> {761 self.session.observe_snapshot(snapshot, now, false)762 }763764 pub(crate) fn mark_session_snapshot_invalid(&mut self) {765 self.session.invalid_snapshot_total = self.session.invalid_snapshot_total.saturating_add(1);766 }767768 pub(crate) fn configure_authenticated_media(&mut self, delivery_role: MediaDeliveryRole) {769 self.media_authentication.mode = sigil_protocol::MediaAuthMode::Ed25519.label();770 self.media_authentication.generation_certified = true;771 self.media_authentication.delivery_role = delivery_role;772 }773774 pub(crate) fn media_verification_failure_counter(&self) -> Arc<AtomicU64> {775 Arc::clone(&self.media_authentication.verification_failures)776 }777778 pub(crate) fn set_input_ack_negotiated(&mut self, negotiated: bool) {779 debug_assert_eq!(self.input_ack.sent_total, 0);780 self.input_ack = InputAckDiagnostics::new(negotiated);781 }782783 pub(crate) fn observe_connection(&mut self, leg: NetworkLeg, connection: &Connection) {784 let diagnostics = match leg {785 NetworkLeg::Media => &mut self.media,786 NetworkLeg::Input => &mut self.input,787 NetworkLeg::Audio => self.audio.get_or_insert_with(LegDiagnostics::default),788 };789 diagnostics.observe(LegObservation::from_connection(connection));790 }791792 pub(crate) fn begin_input_send(&mut self, now: Instant) {793 let _ = self.input_ack.begin_send(now);794 }795796 pub(crate) fn observe_input_ack(&mut self, sequence: u64, now: Instant) -> Result<(), String> {797 self.input_ack.observe_ack(sequence, now)798 }799800 pub(crate) fn mark_input_feedback_closed(&mut self) {801 if self.input_ack.negotiated {802 self.input_ack.closed = true;803 }804 }805806 pub(crate) fn mark_input_feedback_malformed(&mut self) {807 if self.input_ack.negotiated {808 self.input_ack.malformed = true;809 }810 }811812 pub(crate) fn observe_adaptive_report(&mut self, report: &MediaFeedbackReportV1) {813 self.adaptive.observe_report(report);814 }815816 pub(crate) fn observe_adaptive_decision(&mut self, decision: &AdaptiveBitrateDecisionV1) {817 self.adaptive.observe_decision(decision);818 }819820 pub(crate) fn mark_adaptive_detached(&mut self) {821 self.adaptive.detach();822 }823824 pub(crate) fn snapshot(&self, now: Instant) -> NetworkDiagnosticsSnapshot {825 let now_unix = std::time::SystemTime::now()826 .duration_since(std::time::UNIX_EPOCH)827 .map_or(0, |duration| duration.as_secs());828 NetworkDiagnosticsSnapshot {829 version: 2,830 session_elapsed_ms: u64::try_from(831 now.saturating_duration_since(self.started_at).as_millis(),832 )833 .unwrap_or(u64::MAX),834 media: self.media.snapshot(),835 input: self.input.snapshot(),836 audio: self.audio.as_ref().map(LegDiagnostics::snapshot),837 input_ack: self.input_ack.snapshot(),838 media_authentication: self.media_authentication.snapshot(),839 adaptive: self.adaptive.clone(),840 session: self.session.snapshot(now, now_unix),841 }842 }843}844845846pub(crate) struct NetworkDiagnosticsSnapshot {847 version: u8,848 session_elapsed_ms: u64,849 pub(crate) media: NetworkLegSnapshot,850 input: NetworkLegSnapshot,851 audio: Option<NetworkLegSnapshot>,852 input_ack: InputAckSnapshot,853 media_authentication: MediaAuthenticationSnapshot,854 adaptive: AdaptiveViewerSnapshot,855 session: SessionProtocolSnapshot,856}857858859impl NetworkDiagnosticsSnapshot {860 pub(crate) fn test_fixture(mode: PathMode, rtt: Option<Duration>) -> Self {861 let mut diagnostics = NetworkSessionDiagnostics::new(Instant::now(), false);862 diagnostics.media.mode = mode;863 diagnostics.media.sample_count = 1;864 match mode {865 PathMode::Direct => diagnostics.media.direct_samples = 1,866 PathMode::Relay => diagnostics.media.relay_samples = 1,867 PathMode::Custom => diagnostics.media.custom_samples = 1,868 PathMode::Unknown => diagnostics.media.unknown_samples = 1,869 }870 if let Some(rtt) = rtt {871 diagnostics.media.rtt.record(rtt);872 diagnostics.media.rtt_current_ms = Some(rtt.as_secs_f64() * 1_000.0);873 }874 diagnostics.snapshot(Instant::now())875 }876}877878879mod tests {880 use super::*;881882 fn observation(mode: PathMode, counters: u64) -> LegObservation {883 LegObservation {884 selected_path_id: Some(PathId::ZERO),885 mode,886 rtt: Some(Duration::from_millis(10 + counters)),887 cwnd_bytes: Some(64_000),888 mtu: Some(1_200),889 congestion_events: Some(0),890 black_holes_detected: Some(0),891 tx_datagrams: counters,892 tx_bytes: counters * 100,893 rx_datagrams: counters,894 rx_bytes: counters * 100,895 lost_packets: 0,896 lost_bytes: 0,897 }898 }899900 901 fn path_mode_adapter_is_fail_closed_for_ambiguous_paths() {902 assert_eq!(classify_path_mode(true, false), PathMode::Direct);903 assert_eq!(classify_path_mode(false, true), PathMode::Relay);904 assert_eq!(classify_path_mode(false, false), PathMode::Custom);905 assert_eq!(classify_path_mode(true, true), PathMode::Custom);906 }907908 909 fn path_residency_and_transitions_remain_bounded() {910 let mut diagnostics = LegDiagnostics::default();911 diagnostics.observe(observation(PathMode::Direct, 1));912 diagnostics.observe(observation(PathMode::Direct, 2));913 diagnostics.observe(observation(PathMode::Relay, 3));914 let snapshot = diagnostics.snapshot();915 assert_eq!(snapshot.sample_count, 3);916 assert_eq!(snapshot.direct_samples, 2);917 assert_eq!(snapshot.relay_samples, 1);918 assert_eq!(snapshot.mode_transition_total, 1);919 assert!(snapshot.complete);920 }921922 923 fn same_mode_path_switch_advances_epoch_without_mode_transition() {924 let mut diagnostics = LegDiagnostics::default();925 diagnostics.observe(observation(PathMode::Direct, 1));926 let mut replacement = observation(PathMode::Direct, 2);927 replacement.selected_path_id = Some(PathId::MAX);928 diagnostics.observe(replacement);929 let snapshot = diagnostics.snapshot();930 assert_eq!(snapshot.path_epoch, 2);931 assert_eq!(snapshot.path_switch_total, 1);932 assert_eq!(snapshot.mode_transition_total, 0);933 }934935 936 fn current_rtt_preserves_sub_millisecond_precision() {937 let mut diagnostics = LegDiagnostics::default();938 let mut sample = observation(PathMode::Direct, 1);939 sample.rtt = Some(Duration::from_micros(800));940 diagnostics.observe(sample);941 let current = diagnostics.snapshot().rtt_current_ms.unwrap();942 assert!((current - 0.8).abs() < f64::EPSILON);943 }944945 946 fn counter_regression_marks_the_leg_incomplete() {947 let mut diagnostics = LegDiagnostics::default();948 diagnostics.observe(observation(PathMode::Direct, 5));949 diagnostics.observe(observation(PathMode::Direct, 4));950 let snapshot = diagnostics.snapshot();951 assert_eq!(snapshot.counter_regression_total, 1);952 assert!(!snapshot.complete);953 }954955 956 fn latency_overflow_is_visible_instead_of_clamped() {957 let mut histogram = BoundedLatencyHistogram::new();958 histogram.record(Duration::from_millis(5_001));959 let summary = histogram.summary();960 assert_eq!(summary.sample_count, 1);961 assert_eq!(summary.p95_ms, None);962 assert_eq!(summary.max_ms, Some(5_001));963 assert_eq!(summary.overflow_total, 1);964 }965966 967 fn cumulative_ack_accounts_for_each_pending_event() {968 let started = Instant::now();969 let mut diagnostics = InputAckDiagnostics::new(true);970 diagnostics.begin_send(started);971 diagnostics.begin_send(started + Duration::from_millis(1));972 diagnostics.begin_send(started + Duration::from_millis(2));973 diagnostics974 .observe_ack(3, started + Duration::from_millis(12))975 .unwrap();976 let snapshot = diagnostics.snapshot();977 assert_eq!(snapshot.acknowledged_total, 3);978 assert_eq!(snapshot.pending_count, 0);979 assert_eq!(snapshot.latency_sample_count, 3);980 assert_eq!(snapshot.latency_p50_ms, Some(11));981 }982983 984 fn initial_and_duplicate_ack_do_not_invent_latency() {985 let now = Instant::now();986 let mut diagnostics = InputAckDiagnostics::new(true);987 diagnostics.observe_ack(0, now).unwrap();988 diagnostics.observe_ack(0, now).unwrap();989 let snapshot = diagnostics.snapshot();990 assert_eq!(snapshot.duplicate_total, 1);991 assert_eq!(snapshot.latency_sample_count, 0);992 }993994 995 fn lower_and_future_ack_fail_closed() {996 let now = Instant::now();997 let mut diagnostics = InputAckDiagnostics::new(true);998 diagnostics.begin_send(now);999 diagnostics.observe_ack(1, now).unwrap();1000 assert!(diagnostics.observe_ack(0, now).is_err());1001 assert!(diagnostics.observe_ack(2, now).is_err());1002 assert!(diagnostics.snapshot().malformed);1003 }10041005 1006 fn pending_overflow_drops_telemetry_not_input() {1007 let now = Instant::now();1008 let mut diagnostics = InputAckDiagnostics::new(true);1009 for _ in 0..=INPUT_ACK_PENDING_CAPACITY {1010 diagnostics.begin_send(now);1011 }1012 let snapshot = diagnostics.snapshot();1013 assert_eq!(snapshot.sent_total, (INPUT_ACK_PENDING_CAPACITY + 1) as u64);1014 assert_eq!(snapshot.pending_count, INPUT_ACK_PENDING_CAPACITY);1015 assert_eq!(snapshot.untracked_total, 1);1016 assert!(!snapshot.complete);1017 }10181019 1020 fn serialized_snapshot_omits_path_and_peer_identity_material() {1021 let now = Instant::now();1022 let snapshot = NetworkSessionDiagnostics::new(now, true).snapshot(now);1023 let json = serde_json::to_string(&snapshot).unwrap();1024 for forbidden in [1025 "path_id",1026 "remote_addr",1027 "local_addr",1028 "node_id",1029 "relay_url",1030 ] {1031 assert!(1032 !json.contains(forbidden),1033 "serialized forbidden field {forbidden}"1034 );1035 }1036 }10371038 1039 fn authenticated_media_diagnostics_are_truthful_and_expose_no_key_material() {1040 let now = Instant::now();1041 let mut diagnostics = NetworkSessionDiagnostics::new(now, false);1042 diagnostics.configure_authenticated_media(MediaDeliveryRole::ViewerRelay);1043 diagnostics1044 .media_verification_failure_counter()1045 .fetch_add(2, Ordering::Relaxed);1046 let json = serde_json::to_value(diagnostics.snapshot(now)).unwrap();1047 assert_eq!(json["media_authentication"]["mode"], "ed25519-v1");1048 assert_eq!(json["media_authentication"]["generation_certified"], true);1049 assert_eq!(1050 json["media_authentication"]["delivery_role"],1051 "viewer_relay"1052 );1053 assert_eq!(json["media_authentication"]["verification_failures"], 2);1054 let serialized = json.to_string();1055 assert!(!serialized.contains("certificate"));1056 assert!(!serialized.contains("capability"));1057 assert!(!serialized.contains("public_key"));1058 }10591060 1061 fn adaptive_diagnostics_separate_local_pressure_from_the_aggregate_decision() {1062 let now = Instant::now();1063 let mut diagnostics = NetworkSessionDiagnostics::new(now, false);1064 let mut report = MediaFeedbackReportV1 {1065 report_id: 1,1066 interval_ms: 1_000,1067 flags: MediaFeedbackFlags::RESYNC_ACTIVE,1068 last_sequence: Some(10),1069 transport_dropped_delta: 1,1070 frontend_dropped_delta: 0,1071 decoder_dropped_delta: 0,1072 presenter_dropped_delta: 0,1073 frontend_queue_depth: 1,1074 frontend_queue_capacity: 4,1075 decode_queue_depth: 0,1076 decode_queue_capacity: 4,1077 presenter_queue_depth: 0,1078 presenter_queue_capacity: 2,1079 transport_delivery_p95_ms: Some(10),1080 decode_p95_ms: Some(3),1081 presentation_p95_ms: Some(5),1082 };1083 diagnostics.observe_adaptive_report(&report);1084 let decision = AdaptiveBitrateDecisionV1 {1085 decision_id: 1,1086 report_id: 1,1087 target_kbps: 1_000,1088 floor_kbps: 1_000,1089 ceiling_kbps: 12_000,1090 state: AdaptiveBitrateStateV1::Decrease,1091 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,1092 applied: false,1093 };1094 diagnostics.observe_adaptive_decision(&decision);1095 let value = serde_json::to_value(diagnostics.snapshot(now)).unwrap();1096 assert_eq!(value["adaptive"]["scope"], "local_viewer");1097 assert_eq!(value["adaptive"]["local_pressure"], "recovering");1098 assert_eq!(value["adaptive"]["aggregate_target_kbps"], 1_000);1099 assert_eq!(value["adaptive"]["recovery_state"], "floor_limited");1100 report.flags = MediaFeedbackFlags::NONE;1101 diagnostics.observe_adaptive_report(&report);1102 assert_eq!(1103 serde_json::to_value(diagnostics.snapshot(now)).unwrap()["adaptive"]["local_pressure"],1104 "pressured"1105 );1106 }1107}