diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index d5357a6..06d089c 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -1,4 +1,1286 @@ -//! Statistics crate skeleton. -//! -//! Statistics aggregation is intentionally out of scope for the current -//! milestone. +//! Statistics aggregation for `irtt-client` events. + +#![forbid(unsafe_code)] + +use std::{ + collections::{HashMap, HashSet, VecDeque}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; + +use irtt_client::{ + ClientEvent, ClientTimestamp, OneWayDelaySample, ReceivedStatsSample, RttSample, ServerTiming, + SignedDuration, +}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StatsConfig { + pub median: MedianMode, + pub rolling_count: Option, + pub rolling_time: Option, +} + +impl StatsConfig { + pub fn finite() -> Self { + Self { + median: MedianMode::ExactFinite, + rolling_count: None, + rolling_time: None, + } + } + + pub fn continuous() -> Self { + Self { + median: MedianMode::Disabled, + rolling_count: None, + rolling_time: None, + } + } +} + +impl Default for StatsConfig { + fn default() -> Self { + Self::finite() + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MedianMode { + Disabled, + ExactFinite, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct StatsCollector { + config: StatsConfig, + cumulative: CoreStats, + rolling_count: Option>, + rolling_time: Option>, +} + +impl StatsCollector { + pub fn new(config: StatsConfig) -> Self { + Self { + config, + cumulative: CoreStats::new(config.median), + rolling_count: config.rolling_count.map(|_| VecDeque::new()), + rolling_time: config.rolling_time.map(|_| VecDeque::new()), + } + } + + pub fn process(&mut self, event: &ClientEvent) { + let Some(window_event) = WindowEvent::from_client_event(event) else { + return; + }; + + self.cumulative.apply(window_event.clone()); + + if let (Some(limit), Some(window)) = + (self.config.rolling_count, self.rolling_count.as_mut()) + { + window.push_back(window_event.clone()); + while window.len() > limit { + window.pop_front(); + } + } + + if let (Some(duration), Some(window)) = + (self.config.rolling_time, self.rolling_time.as_mut()) + { + let cutoff = window_event.at().checked_sub(duration); + window.push_back(window_event); + if let Some(cutoff) = cutoff { + while window.front().is_some_and(|event| event.at() < cutoff) { + window.pop_front(); + } + } + } + } + + pub fn cumulative(&self) -> CumulativeSnapshot { + self.cumulative.snapshot() + } + + pub fn rolling_count(&self) -> Option { + self.rolling_count.as_ref().map(snapshot_window) + } + + pub fn rolling_time(&self) -> Option { + self.rolling_time.as_ref().map(snapshot_window) + } + + pub fn summary(&self) -> FiniteSummary { + self.cumulative() + } + + #[cfg(test)] + fn retained_median_samples(&self) -> usize { + self.cumulative.retained_median_samples() + } +} + +pub type RollingSnapshot = CumulativeSnapshot; +pub type FiniteSummary = CumulativeSnapshot; + +#[derive(Debug, Clone, PartialEq)] +pub struct CumulativeSnapshot { + pub events: EventCounts, + pub packets: PacketCounts, + pub loss: LossStats, + pub send_call: DurationStats, + pub timer_error: DurationStats, + pub rtt: RttStats, + pub ipdv: IpdvStats, + pub one_way_delay: OneWayDelayStats, + pub server_processing: ServerProcessingStats, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct EventCounts { + pub sent_events: u64, + pub echo_replies: u64, + pub late_unique_replies: u64, + pub duplicate_replies: u64, + pub loss_events: u64, + pub warning_events: u64, + pub untracked_late_replies: u64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct PacketCounts { + pub packets_sent: u64, + pub packets_received: u64, + pub unique_replies: u64, + pub duplicates: u64, + pub late_packets: u64, + pub bytes_sent: u64, + pub bytes_received: u64, + pub server_packets_received: Option, + pub server_received_window: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct LossStats { + pub lost_packets: u64, + pub unknown_loss_packets: u64, + pub upstream_loss_packets: Option, + pub downstream_loss_packets: Option, + pub packet_loss_percent: f64, + pub upstream_loss_percent: f64, + pub downstream_loss_percent: f64, + pub duplicate_percent: f64, + pub late_packets_percent: f64, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct RttStats { + pub primary: SignedDurationStatsWithMedian, + pub raw: DurationStatsWithMedian, + pub adjusted: SignedDurationStatsWithMedian, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct IpdvStats { + pub round_trip: DurationStatsWithMedian, + pub send: DurationStatsWithMedian, + pub receive: DurationStatsWithMedian, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct OneWayDelayStats { + pub send_delay: DurationStatsWithMedian, + pub receive_delay: DurationStatsWithMedian, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct ServerProcessingStats { + pub processing: DurationStats, +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct DurationStats { + pub count: u64, + pub total_ns: u128, + pub min_ns: Option, + pub max_ns: Option, + pub mean_ns: f64, + pub variance_ns2: f64, +} + +impl DurationStats { + pub fn stddev_ns(&self) -> f64 { + self.variance_ns2.sqrt() + } +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct DurationStatsWithMedian { + pub stats: DurationStats, + pub median_ns: Option, +} + +impl DurationStatsWithMedian { + pub fn stddev_ns(&self) -> f64 { + self.stats.stddev_ns() + } +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct SignedDurationStats { + pub count: u64, + pub total_ns: i128, + pub min_ns: Option, + pub max_ns: Option, + pub mean_ns: f64, + pub variance_ns2: f64, +} + +impl SignedDurationStats { + pub fn stddev_ns(&self) -> f64 { + self.variance_ns2.sqrt() + } +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct SignedDurationStatsWithMedian { + pub stats: SignedDurationStats, + pub median_ns: Option, +} + +impl SignedDurationStatsWithMedian { + pub fn stddev_ns(&self) -> f64 { + self.stats.stddev_ns() + } +} + +#[derive(Debug, Clone, PartialEq)] +struct CoreStats { + median: MedianMode, + events: EventCounts, + packets: PacketCounts, + send_call: MetricU64, + timer_error: MetricU64, + rtt_primary: MetricI128, + rtt_raw: MetricU64, + rtt_adjusted: MetricI128, + ipdv_round_trip: MetricU64, + ipdv_send: MetricU64, + ipdv_receive: MetricU64, + send_delay: MetricU64, + receive_delay: MetricU64, + server_processing: MetricU64, + samples: HashMap, + ipdv_pairs: HashSet, +} + +impl CoreStats { + fn new(median: MedianMode) -> Self { + Self { + median, + events: EventCounts::default(), + packets: PacketCounts::default(), + send_call: MetricU64::new(false), + timer_error: MetricU64::new(false), + rtt_primary: MetricI128::new(median == MedianMode::ExactFinite), + rtt_raw: MetricU64::new(median == MedianMode::ExactFinite), + rtt_adjusted: MetricI128::new(median == MedianMode::ExactFinite), + ipdv_round_trip: MetricU64::new(median == MedianMode::ExactFinite), + ipdv_send: MetricU64::new(median == MedianMode::ExactFinite), + ipdv_receive: MetricU64::new(median == MedianMode::ExactFinite), + send_delay: MetricU64::new(median == MedianMode::ExactFinite), + receive_delay: MetricU64::new(median == MedianMode::ExactFinite), + server_processing: MetricU64::new(false), + samples: HashMap::new(), + ipdv_pairs: HashSet::new(), + } + } + + fn apply(&mut self, event: WindowEvent) { + match event { + WindowEvent::Sent { + bytes, + send_call_ns, + timer_error_ns, + .. + } => { + self.events.sent_events += 1; + self.packets.packets_sent += 1; + self.packets.bytes_sent = self.packets.bytes_sent.saturating_add(bytes as u64); + self.send_call.push(send_call_ns); + self.timer_error.push(timer_error_ns); + } + WindowEvent::UniqueReply { + is_late, sample, .. + } => { + let sample = *sample; + self.events.echo_replies += u64::from(!is_late); + self.events.late_unique_replies += u64::from(is_late); + self.packets.packets_received += 1; + self.packets.unique_replies += 1; + self.packets.late_packets += u64::from(is_late); + self.packets.bytes_received = self + .packets + .bytes_received + .saturating_add(sample.bytes as u64); + if let Some(count) = sample.received_count { + self.packets.server_packets_received = Some(u64::from(count)); + } + if let Some(window) = sample.received_window { + self.packets.server_received_window = Some(window); + } + + self.rtt_primary.push(sample.rtt_primary_ns); + self.rtt_raw.push(sample.rtt_raw_ns); + if let Some(adjusted) = sample.rtt_adjusted_ns { + self.rtt_adjusted.push(adjusted); + } + if let Some(processing) = sample.server_processing_ns { + self.server_processing.push(processing); + } + if let Some(delay) = sample.send_delay_ns { + self.send_delay.push(delay); + } + if let Some(delay) = sample.receive_delay_ns { + self.receive_delay.push(delay); + } + + let seq = sample.logical_seq; + if self.samples.insert(seq, sample).is_none() { + self.try_ipdv_pair(seq); + if let Some(next) = seq.checked_add(1) { + self.try_ipdv_pair(next); + } + } + } + WindowEvent::DuplicateReply { .. } => { + self.events.duplicate_replies += 1; + self.packets.packets_received += 1; + self.packets.duplicates += 1; + } + WindowEvent::Loss { .. } => { + self.events.loss_events += 1; + } + WindowEvent::Warning { .. } => { + self.events.warning_events += 1; + } + WindowEvent::UntrackedLate { .. } => { + self.events.untracked_late_replies += 1; + } + } + } + + fn try_ipdv_pair(&mut self, current_seq: u64) { + let Some(previous_seq) = current_seq.checked_sub(1) else { + return; + }; + if !self.ipdv_pairs.insert(current_seq) { + return; + } + let Some(previous) = self.samples.get(&previous_seq) else { + self.ipdv_pairs.remove(¤t_seq); + return; + }; + let Some(current) = self.samples.get(¤t_seq) else { + self.ipdv_pairs.remove(¤t_seq); + return; + }; + + self.ipdv_round_trip.push(abs_i128_to_u64( + current.rtt_primary_ns - previous.rtt_primary_ns, + )); + if let Some(send_ipdv) = send_ipdv_ns(previous, current) { + self.ipdv_send.push(abs_i128_to_u64(send_ipdv)); + } + if let Some(receive_ipdv) = receive_ipdv_ns(previous, current) { + self.ipdv_receive.push(abs_i128_to_u64(receive_ipdv)); + } + } + + fn snapshot(&self) -> CumulativeSnapshot { + let packets = self.packets; + CumulativeSnapshot { + events: self.events, + packets, + loss: loss_stats(packets), + send_call: self.send_call.stats(), + timer_error: self.timer_error.stats(), + rtt: RttStats { + primary: self.rtt_primary.stats_with_median(), + raw: self.rtt_raw.stats_with_median(), + adjusted: self.rtt_adjusted.stats_with_median(), + }, + ipdv: IpdvStats { + round_trip: self.ipdv_round_trip.stats_with_median(), + send: self.ipdv_send.stats_with_median(), + receive: self.ipdv_receive.stats_with_median(), + }, + one_way_delay: OneWayDelayStats { + send_delay: self.send_delay.stats_with_median(), + receive_delay: self.receive_delay.stats_with_median(), + }, + server_processing: ServerProcessingStats { + processing: self.server_processing.stats(), + }, + } + } + + #[cfg(test)] + fn retained_median_samples(&self) -> usize { + self.rtt_primary.retained_samples() + + self.rtt_raw.retained_samples() + + self.rtt_adjusted.retained_samples() + + self.ipdv_round_trip.retained_samples() + + self.ipdv_send.retained_samples() + + self.ipdv_receive.retained_samples() + + self.send_delay.retained_samples() + + self.receive_delay.retained_samples() + } +} + +#[derive(Debug, Clone, PartialEq)] +enum WindowEvent { + Sent { + at: Instant, + bytes: usize, + send_call_ns: u64, + timer_error_ns: u64, + }, + UniqueReply { + at: Instant, + is_late: bool, + sample: Box, + }, + DuplicateReply { + at: Instant, + }, + Loss { + at: Instant, + }, + Warning { + at: Instant, + }, + UntrackedLate { + at: Instant, + }, +} + +impl WindowEvent { + fn from_client_event(event: &ClientEvent) -> Option { + match event { + ClientEvent::EchoSent { + sent_at, + bytes, + send_call, + timer_error, + .. + } => Some(Self::Sent { + at: sent_at.mono, + bytes: *bytes, + send_call_ns: duration_ns_u64(*send_call), + timer_error_ns: duration_ns_u64(*timer_error), + }), + ClientEvent::EchoReply { + logical_seq, + sent_at, + received_at, + rtt, + server_timing, + one_way, + received_stats, + bytes, + .. + } => Some(Self::UniqueReply { + at: received_at.mono, + is_late: false, + sample: Box::new(UniqueSample::new( + *logical_seq, + *sent_at, + *received_at, + *rtt, + *server_timing, + *one_way, + *received_stats, + *bytes, + )), + }), + ClientEvent::LateReply { + logical_seq: Some(logical_seq), + sent_at: Some(sent_at), + received_at, + rtt: Some(rtt), + server_timing, + one_way, + received_stats, + bytes, + .. + } => Some(Self::UniqueReply { + at: received_at.mono, + is_late: true, + sample: Box::new(UniqueSample::new( + *logical_seq, + *sent_at, + *received_at, + *rtt, + *server_timing, + *one_way, + *received_stats, + *bytes, + )), + }), + ClientEvent::LateReply { received_at, .. } => Some(Self::UntrackedLate { + at: received_at.mono, + }), + ClientEvent::DuplicateReply { received_at, .. } => Some(Self::DuplicateReply { + at: received_at.mono, + }), + ClientEvent::EchoLoss { timeout_at, .. } => Some(Self::Loss { at: *timeout_at }), + ClientEvent::Warning { .. } => Some(Self::Warning { at: Instant::now() }), + ClientEvent::SessionStarted { .. } + | ClientEvent::NoTestCompleted { .. } + | ClientEvent::SessionClosed { .. } => None, + } + } + + fn at(&self) -> Instant { + match self { + Self::Sent { at, .. } + | Self::UniqueReply { at, .. } + | Self::DuplicateReply { at } + | Self::Loss { at } + | Self::Warning { at } + | Self::UntrackedLate { at } => *at, + } + } +} + +#[derive(Debug, Clone, PartialEq)] +struct UniqueSample { + logical_seq: u64, + bytes: usize, + rtt_primary_ns: i128, + rtt_raw_ns: u64, + rtt_adjusted_ns: Option, + send_delay_ns: Option, + receive_delay_ns: Option, + server_processing_ns: Option, + received_count: Option, + received_window: Option, + client_send_mono: Instant, + client_receive_mono: Instant, + client_send_wall_ns: Option, + client_receive_wall_ns: Option, + server_receive_mono_ns: Option, + server_send_mono_ns: Option, + server_receive_wall_ns: Option, + server_send_wall_ns: Option, +} + +impl UniqueSample { + #[allow(clippy::too_many_arguments)] + fn new( + logical_seq: u64, + sent_at: ClientTimestamp, + received_at: ClientTimestamp, + rtt: RttSample, + server_timing: Option, + one_way: Option, + received_stats: Option, + bytes: usize, + ) -> Self { + Self { + logical_seq, + bytes, + rtt_primary_ns: signed_duration_ns(rtt.effective_signed), + rtt_raw_ns: duration_ns_u64(rtt.raw), + rtt_adjusted_ns: rtt.adjusted_signed.map(signed_duration_ns), + send_delay_ns: one_way + .and_then(|sample| sample.client_to_server) + .map(duration_ns_u64), + receive_delay_ns: one_way + .and_then(|sample| sample.server_to_client) + .map(duration_ns_u64), + server_processing_ns: server_timing + .and_then(|timing| timing.processing) + .map(duration_ns_u64), + received_count: received_stats.and_then(|stats| stats.count), + received_window: received_stats.and_then(|stats| stats.window), + client_send_mono: sent_at.mono, + client_receive_mono: received_at.mono, + client_send_wall_ns: system_time_ns(sent_at.wall), + client_receive_wall_ns: system_time_ns(received_at.wall), + server_receive_mono_ns: server_timing.and_then(|timing| timing.receive_mono_ns), + server_send_mono_ns: server_timing.and_then(|timing| timing.send_mono_ns), + server_receive_wall_ns: server_timing.and_then(|timing| timing.receive_wall_ns), + server_send_wall_ns: server_timing.and_then(|timing| timing.send_wall_ns), + } + } +} + +#[derive(Debug, Clone, PartialEq)] +struct MetricU64 { + running: RunningU64, + samples: Option>, +} + +impl MetricU64 { + fn new(retain_samples: bool) -> Self { + Self { + running: RunningU64::default(), + samples: retain_samples.then(Vec::new), + } + } + + fn push(&mut self, value: u64) { + self.running.push(value); + if let Some(samples) = self.samples.as_mut() { + samples.push(value); + } + } + + fn stats(&self) -> DurationStats { + self.running.stats() + } + + fn stats_with_median(&self) -> DurationStatsWithMedian { + DurationStatsWithMedian { + stats: self.stats(), + median_ns: self + .samples + .as_ref() + .and_then(|samples| median_u64(samples)), + } + } + + #[cfg(test)] + fn retained_samples(&self) -> usize { + self.samples.as_ref().map_or(0, Vec::len) + } +} + +#[derive(Debug, Clone, PartialEq)] +struct MetricI128 { + running: RunningI128, + samples: Option>, +} + +impl MetricI128 { + fn new(retain_samples: bool) -> Self { + Self { + running: RunningI128::default(), + samples: retain_samples.then(Vec::new), + } + } + + fn push(&mut self, value: i128) { + self.running.push(value); + if let Some(samples) = self.samples.as_mut() { + samples.push(value); + } + } + + fn stats(&self) -> SignedDurationStats { + self.running.stats() + } + + fn stats_with_median(&self) -> SignedDurationStatsWithMedian { + SignedDurationStatsWithMedian { + stats: self.stats(), + median_ns: self + .samples + .as_ref() + .and_then(|samples| median_i128(samples)), + } + } + + #[cfg(test)] + fn retained_samples(&self) -> usize { + self.samples.as_ref().map_or(0, Vec::len) + } +} + +#[derive(Debug, Clone, PartialEq, Default)] +struct RunningU64 { + count: u64, + total: u128, + min: Option, + max: Option, + mean: f64, + m2: f64, +} + +impl RunningU64 { + fn push(&mut self, value: u64) { + self.count += 1; + self.total = self.total.saturating_add(u128::from(value)); + self.min = Some(self.min.map_or(value, |min| min.min(value))); + self.max = Some(self.max.map_or(value, |max| max.max(value))); + let x = value as f64; + let delta = x - self.mean; + self.mean += delta / self.count as f64; + let delta2 = x - self.mean; + self.m2 += delta * delta2; + } + + fn stats(&self) -> DurationStats { + DurationStats { + count: self.count, + total_ns: self.total, + min_ns: self.min, + max_ns: self.max, + mean_ns: if self.count == 0 { 0.0 } else { self.mean }, + variance_ns2: sample_variance(self.count, self.m2), + } + } +} + +#[derive(Debug, Clone, PartialEq, Default)] +struct RunningI128 { + count: u64, + total: i128, + min: Option, + max: Option, + mean: f64, + m2: f64, +} + +impl RunningI128 { + fn push(&mut self, value: i128) { + self.count += 1; + self.total = self.total.saturating_add(value); + self.min = Some(self.min.map_or(value, |min| min.min(value))); + self.max = Some(self.max.map_or(value, |max| max.max(value))); + let x = value as f64; + let delta = x - self.mean; + self.mean += delta / self.count as f64; + let delta2 = x - self.mean; + self.m2 += delta * delta2; + } + + fn stats(&self) -> SignedDurationStats { + SignedDurationStats { + count: self.count, + total_ns: self.total, + min_ns: self.min, + max_ns: self.max, + mean_ns: if self.count == 0 { 0.0 } else { self.mean }, + variance_ns2: sample_variance(self.count, self.m2), + } + } +} + +fn snapshot_window(events: &VecDeque) -> CumulativeSnapshot { + let mut core = CoreStats::new(MedianMode::Disabled); + for event in events { + core.apply(event.clone()); + } + core.snapshot() +} + +fn sample_variance(count: u64, m2: f64) -> f64 { + if count < 2 { + 0.0 + } else { + m2 / (count - 1) as f64 + } +} + +fn median_u64(samples: &[u64]) -> Option { + if samples.is_empty() { + return None; + } + let mut sorted = samples.to_vec(); + sorted.sort_unstable(); + Some(median_sorted_u64(&sorted)) +} + +fn median_sorted_u64(sorted: &[u64]) -> f64 { + let mid = sorted.len() / 2; + if sorted.len() % 2 == 1 { + sorted[mid] as f64 + } else { + (sorted[mid - 1] as f64 + sorted[mid] as f64) / 2.0 + } +} + +fn median_i128(samples: &[i128]) -> Option { + if samples.is_empty() { + return None; + } + let mut sorted = samples.to_vec(); + sorted.sort_unstable(); + let mid = sorted.len() / 2; + Some(if sorted.len() % 2 == 1 { + sorted[mid] as f64 + } else { + (sorted[mid - 1] as f64 + sorted[mid] as f64) / 2.0 + }) +} + +fn loss_stats(packets: PacketCounts) -> LossStats { + let lost = packets.packets_sent.saturating_sub(packets.unique_replies); + let packet_loss_percent = if packets.packets_sent == 0 { + 0.0 + } else if packets.unique_replies == 0 { + 100.0 + } else { + percent(lost as f64, packets.packets_sent as f64) + }; + + let ( + upstream_loss_packets, + upstream_loss_percent, + downstream_loss_packets, + downstream_loss_percent, + ) = if let Some(server_received) = packets.server_packets_received { + let upstream = i128::from(packets.packets_sent) - i128::from(server_received); + let downstream = i128::from(server_received) - i128::from(packets.packets_received); + let upstream_percent = if packets.packets_sent == 0 { + 0.0 + } else { + percent(upstream as f64, packets.packets_sent as f64) + }; + let downstream_percent = if server_received == 0 { + 0.0 + } else { + percent(downstream as f64, server_received as f64) + }; + ( + Some(upstream), + upstream_percent, + Some(downstream), + downstream_percent, + ) + } else { + (None, 0.0, None, 0.0) + }; + + LossStats { + lost_packets: lost, + unknown_loss_packets: lost, + upstream_loss_packets, + downstream_loss_packets, + packet_loss_percent, + upstream_loss_percent, + downstream_loss_percent, + duplicate_percent: if packets.packets_received == 0 { + 0.0 + } else { + percent(packets.duplicates as f64, packets.packets_received as f64) + }, + late_packets_percent: if packets.packets_received == 0 { + 0.0 + } else { + percent(packets.late_packets as f64, packets.packets_received as f64) + }, + } +} + +fn percent(numerator: f64, denominator: f64) -> f64 { + 100.0 * numerator / denominator +} + +fn send_ipdv_ns(previous: &UniqueSample, current: &UniqueSample) -> Option { + if let (Some(prev_server), Some(cur_server)) = ( + previous.server_receive_mono_ns, + current.server_receive_mono_ns, + ) { + return Some( + i128::from(cur_server) + - i128::from(prev_server) + - instant_diff_ns(current.client_send_mono, previous.client_send_mono), + ); + } + if let (Some(prev_server), Some(cur_server), Some(prev_client), Some(cur_client)) = ( + previous.server_receive_wall_ns, + current.server_receive_wall_ns, + previous.client_send_wall_ns, + current.client_send_wall_ns, + ) { + return Some(i128::from(cur_server) - i128::from(prev_server) - (cur_client - prev_client)); + } + None +} + +fn receive_ipdv_ns(previous: &UniqueSample, current: &UniqueSample) -> Option { + if let (Some(prev_server), Some(cur_server)) = + (previous.server_send_mono_ns, current.server_send_mono_ns) + { + return Some( + instant_diff_ns(current.client_receive_mono, previous.client_receive_mono) + - (i128::from(cur_server) - i128::from(prev_server)), + ); + } + if let (Some(prev_server), Some(cur_server), Some(prev_client), Some(cur_client)) = ( + previous.server_send_wall_ns, + current.server_send_wall_ns, + previous.client_receive_wall_ns, + current.client_receive_wall_ns, + ) { + return Some( + (cur_client - prev_client) - (i128::from(cur_server) - i128::from(prev_server)), + ); + } + None +} + +fn instant_diff_ns(current: Instant, previous: Instant) -> i128 { + if let Some(diff) = current.checked_duration_since(previous) { + duration_ns_i128(diff) + } else { + -duration_ns_i128(previous.duration_since(current)) + } +} + +fn system_time_ns(time: SystemTime) -> Option { + if let Ok(duration) = time.duration_since(UNIX_EPOCH) { + return Some(duration_ns_i128(duration)); + } + UNIX_EPOCH + .duration_since(time) + .ok() + .map(|duration| -duration_ns_i128(duration)) +} + +fn duration_ns_u64(duration: Duration) -> u64 { + u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX) +} + +fn duration_ns_i128(duration: Duration) -> i128 { + i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX) +} + +fn signed_duration_ns(duration: SignedDuration) -> i128 { + duration.ns +} + +fn abs_i128_to_u64(value: i128) -> u64 { + u64::try_from(value.saturating_abs()).unwrap_or(u64::MAX) +} + +#[cfg(test)] +mod tests { + use super::*; + use irtt_client::{ + ClientTimestamp, PacketMeta, ReceivedStatsSample, RttSample, ServerTiming, SignedDuration, + }; + use std::time::SystemTime; + + fn ts(ms: u64) -> ClientTimestamp { + ClientTimestamp { + mono: Instant::now() + Duration::from_millis(ms), + wall: UNIX_EPOCH + Duration::from_millis(ms), + } + } + + fn rtt(raw_ms: u64, effective_ms: i128) -> RttSample { + RttSample { + raw: Duration::from_millis(raw_ms), + adjusted: u64::try_from(effective_ms).ok().map(Duration::from_millis), + effective: u64::try_from(effective_ms) + .ok() + .map(Duration::from_millis) + .unwrap_or_else(|| Duration::from_millis(raw_ms)), + adjusted_signed: Some(SignedDuration { + ns: effective_ms * 1_000_000, + }), + effective_signed: SignedDuration { + ns: effective_ms * 1_000_000, + }, + } + } + + fn sent(seq: u32, logical_seq: u64, sent_at: ClientTimestamp) -> ClientEvent { + ClientEvent::EchoSent { + seq, + logical_seq, + remote: "127.0.0.1:2112".parse().unwrap(), + scheduled_at: sent_at.mono, + sent_at, + bytes: 32, + send_call: Duration::from_micros(10), + timer_error: Duration::from_micros(2), + } + } + + fn reply(logical_seq: u64, raw_ms: u64, effective_ms: i128) -> ClientEvent { + let sent_at = ts(logical_seq * 10); + let received_at = ClientTimestamp { + mono: sent_at.mono + Duration::from_millis(raw_ms), + wall: sent_at.wall + Duration::from_millis(raw_ms), + }; + ClientEvent::EchoReply { + seq: logical_seq as u32, + logical_seq, + remote: "127.0.0.1:2112".parse().unwrap(), + sent_at, + received_at, + rtt: rtt(raw_ms, effective_ms), + server_timing: Some(ServerTiming { + receive_wall_ns: Some(system_time_ns(sent_at.wall).unwrap() as i64 + 1_000_000), + receive_mono_ns: Some(logical_seq as i64 * 10_000_000 + 1_000_000), + send_wall_ns: Some(system_time_ns(sent_at.wall).unwrap() as i64 + 2_000_000), + send_mono_ns: Some(logical_seq as i64 * 10_000_000 + 2_000_000), + midpoint_wall_ns: None, + midpoint_mono_ns: None, + processing: Some(Duration::from_millis(1)), + }), + one_way: Some(OneWayDelaySample { + client_to_server: Some(Duration::from_millis(1)), + server_to_client: Some(Duration::from_millis(2)), + }), + received_stats: Some(ReceivedStatsSample { + count: Some((logical_seq + 1) as u32), + window: Some(0xff), + }), + bytes: 64, + packet_meta: PacketMeta::default(), + } + } + + fn late_reply(logical_seq: u64, raw_ms: u64, effective_ms: i128) -> ClientEvent { + let ClientEvent::EchoReply { + seq, + remote, + sent_at, + received_at, + rtt, + server_timing, + one_way, + received_stats, + bytes, + packet_meta, + .. + } = reply(logical_seq, raw_ms, effective_ms) + else { + unreachable!(); + }; + ClientEvent::LateReply { + seq, + logical_seq: Some(logical_seq), + highest_seen: seq + 1, + remote, + sent_at: Some(sent_at), + received_at, + rtt: Some(rtt), + server_timing, + one_way, + received_stats, + bytes, + packet_meta, + } + } + + #[test] + fn running_duration_stats_use_sample_variance() { + let mut metric = MetricU64::new(false); + metric.push(1); + metric.push(2); + metric.push(3); + let stats = metric.stats(); + assert_eq!(stats.count, 3); + assert_eq!(stats.total_ns, 6); + assert_eq!(stats.min_ns, Some(1)); + assert_eq!(stats.max_ns, Some(3)); + assert_eq!(stats.mean_ns, 2.0); + assert_eq!(stats.variance_ns2, 1.0); + assert_eq!(stats.stddev_ns(), 1.0); + } + + #[test] + fn exact_median_handles_odd_and_even_samples() { + assert_eq!(median_u64(&[3, 1, 2]), Some(2.0)); + assert_eq!(median_u64(&[4, 1, 2, 3]), Some(2.5)); + assert_eq!(median_i128(&[-5, 1, 3]), Some(1.0)); + assert_eq!(median_i128(&[-5, 1, 3, 7]), Some(2.0)); + } + + #[test] + fn disabled_median_avoids_finite_retention() { + let mut collector = StatsCollector::new(StatsConfig::continuous()); + collector.process(&reply(0, 10, 9)); + collector.process(&reply(1, 20, 19)); + assert_eq!(collector.retained_median_samples(), 0); + assert_eq!(collector.cumulative().rtt.primary.median_ns, None); + } + + #[test] + fn cumulative_rtt_uses_signed_effective_and_tracks_raw() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&reply(0, 1, -2)); + collector.process(&reply(1, 10, 8)); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.rtt.primary.stats.count, 2); + assert_eq!(snapshot.rtt.primary.stats.min_ns, Some(-2_000_000)); + assert_eq!(snapshot.rtt.primary.median_ns, Some(3_000_000.0)); + assert_eq!(snapshot.rtt.raw.stats.total_ns, 11_000_000); + } + + #[test] + fn late_unique_counts_and_duplicates_do_not_update_measurements() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&sent(0, 0, ts(0))); + collector.process(&sent(1, 1, ts(10))); + collector.process(&reply(1, 10, 9)); + collector.process(&late_reply(0, 20, 19)); + collector.process(&ClientEvent::DuplicateReply { + seq: 0, + remote: "127.0.0.1:2112".parse().unwrap(), + received_at: ts(50), + bytes: 64, + }); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.packets.packets_sent, 2); + assert_eq!(snapshot.packets.packets_received, 3); + assert_eq!(snapshot.packets.unique_replies, 2); + assert_eq!(snapshot.packets.duplicates, 1); + assert_eq!(snapshot.packets.late_packets, 1); + assert_eq!(snapshot.rtt.primary.stats.count, 2); + assert_eq!(snapshot.loss.lost_packets, 0); + assert_eq!(snapshot.loss.duplicate_percent, 100.0 / 3.0); + } + + #[test] + fn final_loss_uses_sent_minus_unique_replies_not_echo_loss_events() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&sent(0, 0, ts(0))); + collector.process(&sent(1, 1, ts(10))); + collector.process(&ClientEvent::EchoLoss { + seq: 0, + logical_seq: 0, + sent_at: ts(0), + timeout_at: Instant::now(), + }); + collector.process(&late_reply(0, 10, 9)); + + let snapshot = collector.summary(); + assert_eq!(snapshot.events.loss_events, 1); + assert_eq!(snapshot.packets.unique_replies, 1); + assert_eq!(snapshot.loss.lost_packets, 1); + assert_eq!(snapshot.loss.packet_loss_percent, 50.0); + } + + #[test] + fn ipdv_is_sequence_adjacent_and_gap_preserving() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&reply(0, 10, 10)); + collector.process(&reply(2, 15, 15)); + collector.process(&reply(3, 12, 12)); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.ipdv.round_trip.stats.count, 1); + assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 3_000_000); + } + + #[test] + fn late_reply_can_complete_ipdv_pair() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&reply(1, 20, 20)); + collector.process(&late_reply(0, 10, 10)); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.ipdv.round_trip.stats.count, 1); + assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 10_000_000); + } + + #[test] + fn server_processing_and_one_way_require_available_samples() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&reply(0, 10, 9)); + collector.process(&ClientEvent::LateReply { + seq: 9, + logical_seq: None, + highest_seen: 10, + remote: "127.0.0.1:2112".parse().unwrap(), + sent_at: None, + received_at: ts(100), + rtt: None, + server_timing: None, + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: PacketMeta::default(), + }); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.server_processing.processing.count, 1); + assert_eq!(snapshot.one_way_delay.send_delay.stats.count, 1); + assert_eq!(snapshot.events.untracked_late_replies, 1); + } + + #[test] + fn rolling_count_eviction_recomputes_from_bounded_events() { + let mut collector = StatsCollector::new(StatsConfig { + median: MedianMode::Disabled, + rolling_count: Some(2), + rolling_time: None, + }); + collector.process(&sent(0, 0, ts(0))); + collector.process(&reply(0, 10, 10)); + collector.process(&reply(1, 20, 20)); + + let rolling = collector.rolling_count().unwrap(); + assert_eq!(rolling.packets.packets_sent, 0); + assert_eq!(rolling.packets.unique_replies, 2); + assert_eq!(rolling.rtt.primary.stats.count, 2); + } + + #[test] + fn rolling_time_eviction_uses_event_timestamps() { + let mut collector = StatsCollector::new(StatsConfig { + median: MedianMode::Disabled, + rolling_count: None, + rolling_time: Some(Duration::from_millis(15)), + }); + collector.process(&sent(0, 0, ts(0))); + collector.process(&sent(1, 1, ts(10))); + collector.process(&sent(2, 2, ts(30))); + + let rolling = collector.rolling_time().unwrap(); + assert_eq!(rolling.packets.packets_sent, 1); + } + + #[test] + fn empty_and_all_lost_edges_are_defined() { + let empty = StatsCollector::new(StatsConfig::finite()).summary(); + assert_eq!(empty.loss.packet_loss_percent, 0.0); + + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&sent(0, 0, ts(0))); + let all_lost = collector.summary(); + assert_eq!(all_lost.loss.lost_packets, 1); + assert_eq!(all_lost.loss.packet_loss_percent, 100.0); + } + + #[test] + fn directional_loss_uses_server_received_count_when_available() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&sent(0, 0, ts(0))); + collector.process(&sent(1, 1, ts(10))); + collector.process(&reply(0, 10, 10)); + + let loss = collector.summary().loss; + assert_eq!(loss.upstream_loss_packets, Some(1)); + assert_eq!(loss.downstream_loss_packets, Some(0)); + assert_eq!(loss.upstream_loss_percent, 50.0); + } + + #[test] + fn single_sample_stddev_is_zero() { + let mut metric = MetricU64::new(false); + metric.push(42); + let stats = metric.stats(); + assert_eq!(stats.variance_ns2, 0.0); + assert_eq!(stats.stddev_ns(), 0.0); + } + + #[test] + fn system_time_before_epoch_is_supported() { + let before = UNIX_EPOCH - Duration::from_nanos(7); + assert_eq!(system_time_ns(before), Some(-7)); + let after = UNIX_EPOCH + Duration::from_nanos(7); + assert_eq!(system_time_ns(after), Some(7)); + let now = SystemTime::now(); + assert!(system_time_ns(now).is_some()); + } +}