diff --git a/crates/irtt-stats/src/ipdv.rs b/crates/irtt-stats/src/ipdv.rs index 1ebb716..cf36fd9 100644 --- a/crates/irtt-stats/src/ipdv.rs +++ b/crates/irtt-stats/src/ipdv.rs @@ -48,10 +48,8 @@ impl IpdvTracker { if let Some(pair) = self.try_pair(seq) { pairs.push(pair); } - if let Some(next) = seq.checked_add(1) { - if let Some(pair) = self.try_pair(next) { - pairs.push(pair); - } + if let Some(pair) = self.try_pair(seq.wrapping_add(1)) { + pairs.push(pair); } pairs } @@ -66,15 +64,13 @@ impl IpdvTracker { }; if self.samples.remove(&seq).is_some() { self.completed_pairs.remove(&seq); - if let Some(next) = seq.checked_add(1) { - self.completed_pairs.remove(&next); - } + self.completed_pairs.remove(&seq.wrapping_add(1)); } } } fn try_pair(&mut self, current_seq: u32) -> Option { - let previous_seq = current_seq.checked_sub(1)?; + let previous_seq = current_seq.wrapping_sub(1); if !self.completed_pairs.insert(current_seq) { return None; @@ -237,6 +233,52 @@ mod tests { ); } + #[test] + fn ipdv_tracker_completes_wrapped_adjacent_pair() { + let mut tracker = IpdvTracker::new(None); + assert!(tracker.insert(ipdv_sample(u32::MAX, 10)).is_empty()); + + let pairs = tracker.insert(ipdv_sample(0, 14)); + + assert_eq!( + pairs, + vec![CompletedIpdvPair { + previous_seq: u32::MAX, + current_seq: 0, + rtt_ipdv_ns: 4, + send_ipdv_ns: None, + receive_ipdv_ns: None, + }] + ); + } + + #[test] + fn ipdv_tracker_wrap_gap_does_not_complete_pair() { + let mut tracker = IpdvTracker::new(None); + assert!(tracker.insert(ipdv_sample(u32::MAX - 1, 10)).is_empty()); + + assert!(tracker.insert(ipdv_sample(0, 14)).is_empty()); + } + + #[test] + fn ipdv_tracker_late_wrapped_previous_completes_pair() { + let mut tracker = IpdvTracker::new(None); + assert!(tracker.insert(ipdv_sample(0, 14)).is_empty()); + + let pairs = tracker.insert(ipdv_sample(u32::MAX, 10)); + + assert_eq!( + pairs, + vec![CompletedIpdvPair { + previous_seq: u32::MAX, + current_seq: 0, + rtt_ipdv_ns: 4, + send_ipdv_ns: None, + receive_ipdv_ns: None, + }] + ); + } + #[test] fn ipdv_tracker_duplicate_sequence_does_not_emit_pair_again() { let mut tracker = IpdvTracker::new(None); diff --git a/crates/irtt-stats/tests/stats.rs b/crates/irtt-stats/tests/stats.rs index ccccf6d..c0ee352 100644 --- a/crates/irtt-stats/tests/stats.rs +++ b/crates/irtt-stats/tests/stats.rs @@ -1,9 +1,9 @@ mod common; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, UNIX_EPOCH}; use common::{adjusted_reply, sent, ts, unadjusted_late_reply, unadjusted_reply}; -use irtt_client::{ClientEvent, PacketMeta}; +use irtt_client::{ClientEvent, ClientTimestamp, PacketMeta, RttSample, SignedDuration}; use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig}; #[test] @@ -113,6 +113,34 @@ fn assert_one_ipdv_pair( pair } +fn reply_at(seq: u32, raw_ms: u64, sent_ms: u64) -> ClientEvent { + let sent_at = ClientTimestamp { + mono: Instant::now() + Duration::from_millis(sent_ms), + wall: UNIX_EPOCH + Duration::from_millis(sent_ms), + }; + let received_at = ClientTimestamp { + mono: sent_at.mono + Duration::from_millis(raw_ms), + wall: sent_at.wall + Duration::from_millis(raw_ms), + }; + let raw = Duration::from_millis(raw_ms); + ClientEvent::EchoReply { + seq, + remote: "127.0.0.1:2112".parse().unwrap(), + sent_at, + received_at, + rtt: RttSample { + raw, + adjusted: None, + effective: SignedDuration::from_duration(raw), + }, + server_timing: None, + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: PacketMeta::default(), + } +} + #[test] fn ipdv_is_sequence_adjacent_and_gap_preserving() { let mut collector = StatsCollector::new(StatsConfig::finite()); @@ -133,6 +161,39 @@ fn ipdv_is_sequence_adjacent_and_gap_preserving() { assert_eq!(snapshot.ipdv.round_trip.total_ns, 3_000_000); } +#[test] +fn ipdv_wraparound_sequence_is_adjacent() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + let first = collector.process(&reply_at(u32::MAX, 10, 0)); + let wrapped = collector.process(&reply_at(0, 14, 10)); + + assert!(first.contributed_sample); + assert_no_ipdv_pairs(&first); + + assert!(wrapped.contributed_sample); + assert_one_ipdv_pair(&wrapped, u32::MAX, 0, Duration::from_millis(4)); + + let snapshot = collector.snapshot(); + assert_eq!(snapshot.ipdv.round_trip.count, 1); + assert_eq!(snapshot.ipdv.round_trip.total_ns, 4_000_000); +} + +#[test] +fn ipdv_wraparound_gap_is_preserved() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + let before_wrap = collector.process(&reply_at(u32::MAX - 1, 10, 0)); + let wrapped = collector.process(&reply_at(0, 14, 20)); + + assert!(before_wrap.contributed_sample); + assert_no_ipdv_pairs(&before_wrap); + + assert!(wrapped.contributed_sample); + assert_no_ipdv_pairs(&wrapped); + + let snapshot = collector.snapshot(); + assert_eq!(snapshot.ipdv.round_trip.count, 0); +} + #[test] fn late_reply_can_complete_ipdv_pair() { let mut collector = StatsCollector::new(StatsConfig::finite());