diff --git a/Cargo.lock b/Cargo.lock index 41eb91f..77ea494 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -778,6 +778,7 @@ name = "irtt-stats" version = "0.5.0" dependencies = [ "irtt-client", + "proptest", ] [[package]] diff --git a/crates/irtt-stats/Cargo.toml b/crates/irtt-stats/Cargo.toml index 76d9b49..be164d3 100644 --- a/crates/irtt-stats/Cargo.toml +++ b/crates/irtt-stats/Cargo.toml @@ -13,3 +13,6 @@ categories = ["network-programming"] [dependencies] irtt-client.workspace = true + +[dev-dependencies] +proptest = "1.11.0" diff --git a/crates/irtt-stats/src/ipdv.rs b/crates/irtt-stats/src/ipdv.rs index 3fa9de6..1254ed2 100644 --- a/crates/irtt-stats/src/ipdv.rs +++ b/crates/irtt-stats/src/ipdv.rs @@ -235,4 +235,59 @@ mod tests { assert!(tracker.completed_pairs.len() <= limit, "seq {seq}"); } } + + mod properties { + use super::*; + use proptest::prelude::*; + + proptest! { + /// The hand-written `ipdv_tracker_bounded_mode_limits_sequence_state` + /// test above only exercises a strictly increasing `0..512` + /// sequence. Real traffic can duplicate, skip, or otherwise not + /// strictly increase sequence numbers (see the client's + /// `LateReply`/`DuplicateReply` handling), so this generates + /// arbitrary, possibly-repeated, possibly-non-monotonic sequence + /// streams and checks the same bounds still hold after every + /// insert: `samples` and `sample_order` never exceed the + /// configured limit and stay in lockstep, and `completed_pairs` + /// never exceeds `samples.len()` (a successful entry can only + /// persist once both its endpoint samples are present, see + /// `try_pair`, so it is always a subset of the live sample keys). + #[test] + fn bounded_tracker_state_never_exceeds_the_limit( + limit in 1usize..8, + seqs in prop::collection::vec(0u32..40, 0..300), + ) { + let mut tracker = IpdvTracker::new(Some(limit)); + for seq in seqs { + tracker.insert(ipdv_sample(seq, i128::from(seq))); + prop_assert!(tracker.samples.len() <= limit); + prop_assert_eq!(tracker.sample_order.len(), tracker.samples.len()); + prop_assert!(tracker.completed_pairs.len() <= tracker.samples.len()); + } + } + + /// A pair for a given adjacent `(seq - 1, seq)` boundary is only + /// ever emitted once, however many times `insert` is called + /// (including for already-seen or unrelated sequences in + /// between). This is the unbounded case, so eviction cannot be + /// the reason a pair fails to repeat. + #[test] + fn each_adjacent_pair_is_emitted_at_most_once( + seqs in prop::collection::vec(0u32..12, 0..200), + ) { + let mut tracker = IpdvTracker::new(None); + let mut seen_pairs = std::collections::HashSet::new(); + for seq in seqs { + for pair in tracker.insert(ipdv_sample(seq, i128::from(seq))) { + let key = (pair.previous_seq, pair.current_seq); + prop_assert!( + seen_pairs.insert(key), + "pair {key:?} emitted more than once" + ); + } + } + } + } + } } diff --git a/crates/irtt-stats/tests/rolling_properties.rs b/crates/irtt-stats/tests/rolling_properties.rs new file mode 100644 index 0000000..ff888f5 --- /dev/null +++ b/crates/irtt-stats/tests/rolling_properties.rs @@ -0,0 +1,338 @@ +//! Property tests for the count-based and time-based rolling windows. +//! +//! [`RollingEvents`] (in `src/rolling.rs`) keeps two independent `VecDeque`s: +//! one bounded by [`StatsConfig::rolling_count`] and one bounded by +//! [`StatsConfig::rolling_time`]. They are pushed to together but evicted +//! independently, and each exposes its own snapshot +//! ([`StatsCollector::rolling_count`] / [`StatsCollector::rolling_time`]). +//! Reading `rolling.rs` and the existing `rolling_count_eviction_recomputes_*` +//! / `rolling_time_eviction_uses_event_timestamps` tests in `tests/stats.rs` +//! confirms there is no single window bounded by *both* limits at once: the +//! two bounds do not compose as AND or OR on shared storage, they are simply +//! two separate parallel windows over the same input stream. This file proves +//! that independence generatively: with both bounds configured +//! simultaneously, each window must behave exactly as if the *other* bound +//! did not exist. +//! +//! The reference model is a plain `Vec<(ClientEvent, at_ms)>` history that is +//! re-filtered from scratch after every operation (once by count, once by +//! time), then replayed through a fresh [`StatsCollector`] to obtain a +//! [`Snapshot`]. Because [`StatsCollector::rolling_count`] / +//! [`StatsCollector::rolling_time`] are themselves computed by replaying the +//! retained events through a fresh `CoreStats` on every call (see +//! `rolling.rs`'s `snapshot_window`), and because both the production window +//! and this reference model process the *same* `ClientEvent` values in the +//! *same* order through the *same* pure, deterministic normalization and +//! aggregation code, the two snapshots are expected to be bit-for-bit equal, +//! not merely close. `Snapshot` derives `PartialEq`, so the comparison below +//! is exact `assert_eq!` with no floating-point tolerance: there is no +//! independent numerical computation on the reference side to accumulate +//! error against, both sides run the identical arithmetic in the identical +//! order. +//! +//! The reference model's time-window recomputation (filter the whole history +//! by `at_ms >= latest_at_ms - window_ms`) is only equivalent to the +//! production incremental sliding-eviction (which pops from the front using +//! the cutoff computed at each push) when event timestamps are pushed in +//! non-decreasing order, exactly as `normalization.rs` documents ("Rolling- +//! window eviction assumes events are pushed in non-decreasing `at()` +//! order"). The operation generator below enforces that by construction: a +//! shared clock only ever moves forward, and every generated event borrows +//! its `at()` timestamp from the clock's current value. + +use std::time::{Duration, Instant, UNIX_EPOCH}; + +use irtt_client::{ + ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample, + ServerTiming, SignedDuration, +}; +use irtt_stats::{LateReplyMode, SampleMode, Snapshot, StatsCollector, StatsConfig}; +use proptest::prelude::*; + +/// One generated operation. `Advance` moves the shared monotonic clock +/// forward without producing an event; every other variant produces exactly +/// one normalized event at the clock's current value. +#[derive(Debug, Clone)] +enum Op { + /// Advance the shared clock by this many milliseconds. + Advance(u16), + /// A probe send. + Send, + /// An on-time unique reply with the given client-observed RTT. + Reply { raw_ms: u16 }, + /// A late reply matched to retained send state (measurable). + LateReplyMatched { raw_ms: u16 }, + /// A late reply that could not be matched to retained state. + LateReplyUnmatched, + /// A duplicate reply for an already-completed sequence. + Duplicate, + /// A probe timeout / loss. + Loss, + /// A diagnostic warning event. + Warning, +} + +fn op_strategy() -> impl Strategy { + prop_oneof![ + 3 => (0u16..40).prop_map(Op::Advance), + 4 => Just(Op::Send), + 4 => (0u16..60).prop_map(|raw_ms| Op::Reply { raw_ms }), + 2 => (0u16..60).prop_map(|raw_ms| Op::LateReplyMatched { raw_ms }), + 1 => Just(Op::LateReplyUnmatched), + 1 => Just(Op::Duplicate), + 2 => Just(Op::Loss), + 1 => Just(Op::Warning), + ] +} + +const ADDR: &str = "127.0.0.1:2112"; + +/// Builds a `ClientTimestamp` purely by adding a millisecond offset to a +/// fixed `base` `Instant` (and to `UNIX_EPOCH` for the wall-clock half). No +/// operation below ever reads the real clock more than once (to establish +/// `base`), so the resulting timestamps are fully deterministic functions of +/// the generated `Op`s, with no wall-clock jitter to make a proptest shrink +/// or CI run flaky. +fn ts_at(base: Instant, ms: u64) -> ClientTimestamp { + ClientTimestamp { + mono: base + Duration::from_millis(ms), + wall: UNIX_EPOCH + Duration::from_millis(ms), + } +} + +fn server_timing(seq: u32) -> ServerTiming { + let base_ns = i64::from(seq) * 10_000_000; + ServerTiming { + receive_mono_ns: Some(base_ns + 1_000_000), + send_mono_ns: Some(base_ns + 2_000_000), + receive_wall_ns: Some(base_ns + 1_000_000), + send_wall_ns: Some(base_ns + 2_000_000), + midpoint_mono_ns: None, + midpoint_wall_ns: None, + processing: Some(Duration::from_micros(100)), + } +} + +fn one_way(seq: u32) -> OneWayDelaySample { + OneWayDelaySample { + client_to_server: Some(SignedDuration::from_nanos(1_000_000 + i128::from(seq))), + server_to_client: Some(SignedDuration::from_nanos(2_000_000 + i128::from(seq))), + } +} + +/// Builds the `ClientEvent` for one non-`Advance` op. `clock_ms` is the +/// shared clock's current value, used as the event's windowing timestamp +/// (`at()`, see `normalization.rs`) for every variant. +fn build_event(op: &Op, seq: u32, clock_ms: u64, base: Instant) -> ClientEvent { + match *op { + Op::Advance(_) => unreachable!("Advance does not produce an event"), + Op::Send => ClientEvent::EchoSent { + seq, + remote: ADDR.parse().unwrap(), + scheduled_at: ts_at(base, clock_ms).mono, + sent_at: ts_at(base, clock_ms), + bytes: 32, + send_call: Duration::from_micros(10), + timer_error: Duration::from_micros(2), + }, + Op::Reply { raw_ms } => { + let sent_at = ts_at(base, clock_ms.saturating_sub(u64::from(raw_ms))); + let received_at = ts_at(base, clock_ms); + ClientEvent::EchoReply { + seq, + remote: ADDR.parse().unwrap(), + sent_at, + received_at, + rtt: RttSample { + raw: Duration::from_millis(u64::from(raw_ms)), + adjusted: None, + effective: SignedDuration::from_duration(Duration::from_millis(u64::from( + raw_ms, + ))), + }, + server_timing: Some(server_timing(seq)), + one_way: Some(one_way(seq)), + received_stats: Some(ReceivedStatsSample { + count: Some(seq + 1), + window: Some(0xff), + }), + bytes: 64, + packet_meta: PacketMeta::default(), + } + } + Op::LateReplyMatched { raw_ms } => { + let sent_at = ts_at(base, clock_ms.saturating_sub(u64::from(raw_ms))); + let received_at = ts_at(base, clock_ms); + ClientEvent::LateReply { + seq, + highest_seen: seq + 1, + remote: ADDR.parse().unwrap(), + sent_at: Some(sent_at), + received_at, + rtt: Some(RttSample { + raw: Duration::from_millis(u64::from(raw_ms)), + adjusted: None, + effective: SignedDuration::from_duration(Duration::from_millis(u64::from( + raw_ms, + ))), + }), + server_timing: Some(server_timing(seq)), + one_way: Some(one_way(seq)), + received_stats: Some(ReceivedStatsSample { + count: Some(seq + 1), + window: Some(0xff), + }), + bytes: 64, + packet_meta: PacketMeta::default(), + } + } + Op::LateReplyUnmatched => ClientEvent::LateReply { + seq, + highest_seen: seq + 1, + remote: ADDR.parse().unwrap(), + sent_at: None, + received_at: ts_at(base, clock_ms), + rtt: None, + server_timing: None, + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: PacketMeta::default(), + }, + Op::Duplicate => ClientEvent::DuplicateReply { + seq, + remote: ADDR.parse().unwrap(), + received_at: ts_at(base, clock_ms), + bytes: 64, + }, + Op::Loss => ClientEvent::EchoLoss { + seq, + sent_at: ts_at(base, clock_ms), + timeout_at: ts_at(base, clock_ms).mono, + }, + Op::Warning => ClientEvent::Warning { + kind: irtt_client::WarningKind::UntrackedReply, + message: "generated".to_owned(), + at: ts_at(base, clock_ms), + }, + } +} + +/// Replays `events` in order through a fresh collector using the rolling +/// windows' own semantics (running statistics only, same late-reply policy), +/// mirroring exactly what `rolling.rs`'s private `snapshot_window` does for +/// the production windows. This is the "recompute from scratch" reference +/// primitive both the count and time reference windows below are built on. +fn replay(events: &[ClientEvent], late_replies: LateReplyMode) -> Snapshot { + let mut collector = StatsCollector::new(StatsConfig { + samples: SampleMode::RunningOnly, + rolling_count: None, + rolling_time: None, + late_replies, + }); + for event in events { + collector.process(event); + } + collector.snapshot() +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 128, .. ProptestConfig::default() })] + + /// With `rolling_count` and `rolling_time` configured simultaneously, + /// each window must match a reference model that applies *only its own* + /// bound to the full event history, re-filtered from scratch after every + /// operation. Any accidental AND/OR coupling between the two bounds (for + /// example the count window also dropping events outside the time + /// window, or vice versa) would show up here as a mismatch. + /// + /// The cumulative snapshot is checked too: it must equal a reference + /// that replays the *entire* history, proving rolling eviction never + /// reaches into the cumulative accounting. + #[test] + fn rolling_count_and_rolling_time_windows_are_independent( + ops in prop::collection::vec(op_strategy(), 1..40), + count_limit in 1usize..6, + time_limit_ms in 1u64..80, + count_only in prop::bool::ANY, + ) { + let late_replies = if count_only { + LateReplyMode::CountOnly + } else { + LateReplyMode::Measure + }; + let config = StatsConfig { + samples: SampleMode::RunningOnly, + rolling_count: Some(count_limit), + rolling_time: Some(Duration::from_millis(time_limit_ms)), + late_replies, + }; + let mut collector = StatsCollector::new(config); + let base = Instant::now(); + let mut clock_ms: u64 = 0; + // Full, unfiltered event history alongside each event's windowing + // timestamp, so the reference windows below can be recomputed from + // scratch after every step. + let mut history: Vec<(ClientEvent, u64)> = Vec::new(); + + for (idx, op) in ops.iter().enumerate() { + if let Op::Advance(delta) = op { + clock_ms += u64::from(*delta); + } else { + let seq = u32::try_from(idx).unwrap(); + let event = build_event(op, seq, clock_ms, base); + collector.process(&event); + history.push((event, clock_ms)); + } + + // Cumulative snapshot: unaffected by rolling eviction, so it must + // equal a fresh replay of the entire history so far. + let full: Vec = history.iter().map(|(event, _)| event.clone()).collect(); + prop_assert_eq!( + collector.snapshot(), + replay(&full, late_replies), + "cumulative snapshot diverged from a full replay after op {}", idx + ); + + // Count window reference: the last `count_limit` retained + // events, ignoring the time bound entirely. + let count_start = history.len().saturating_sub(count_limit); + let count_ref: Vec = history[count_start..] + .iter() + .map(|(event, _)| event.clone()) + .collect(); + prop_assert_eq!( + collector.rolling_count(), + Some(replay(&count_ref, late_replies)), + "count window diverged from the count-only reference after op {}", idx + ); + + // Time window reference: every retained event whose timestamp is + // within `time_limit_ms` of the *latest processed event's* + // timestamp, ignoring the count bound entirely. This recompute- + // from-scratch filter is equivalent to the production window's + // incremental sliding-eviction only because `clock_ms` is + // non-decreasing across pushes (enforced by construction above: + // `Advance` never subtracts). + if let Some((_, latest_at_ms)) = history.last() { + let cutoff = latest_at_ms.checked_sub(time_limit_ms); + let time_ref: Vec = history + .iter() + .filter(|(_, at_ms)| cutoff.is_none_or(|cutoff| *at_ms >= cutoff)) + .map(|(event, _)| event.clone()) + .collect(); + prop_assert_eq!( + collector.rolling_time(), + Some(replay(&time_ref, late_replies)), + "time window diverged from the time-only reference after op {}", idx + ); + } else { + prop_assert_eq!( + collector.rolling_time(), + Some(replay(&[], late_replies)), + "time window should be empty before any event after op {}", idx + ); + } + } + } +}