diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 0398d48..187f13f 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -417,7 +417,9 @@ impl Client { let server_timing = build_server_timing(&reply.timestamps); let one_way = compute_one_way(&pending.sent_at, &now, &reply.timestamps); let received_stats = build_received_stats(&reply); - let is_late = session.highest_received_seq.is_some_and(|h| wire_seq < h); + let is_late = session + .highest_received_seq + .is_some_and(|h| sequence_is_before(wire_seq, h)); let highest_seen = session.highest_received_seq.unwrap_or(wire_seq); update_highest_received(session, wire_seq); @@ -486,9 +488,10 @@ impl Client { bytes: packet.len(), packet_meta: meta.into(), }]) - } else if session.highest_received_seq.is_some_and(|h| wire_seq < h) { - // TODO: u32 ordering is acceptable for finite tests but continuous - // probing will eventually require wrap-aware comparison. + } else if session + .highest_received_seq + .is_some_and(|h| sequence_is_before(wire_seq, h)) + { Ok(vec![ClientEvent::LateReply { seq: wire_seq, logical_seq: None, @@ -635,11 +638,21 @@ impl Client { } fn update_highest_received(session: &mut ActiveSession, wire_seq: u32) { - session.highest_received_seq = Some( - session - .highest_received_seq - .map_or(wire_seq, |h| h.max(wire_seq)), - ); + session.highest_received_seq = Some(session.highest_received_seq.map_or(wire_seq, |h| { + if sequence_is_after(wire_seq, h) { + wire_seq + } else { + h + } + })); +} + +fn sequence_is_after(candidate: u32, current: u32) -> bool { + candidate != current && candidate.wrapping_sub(current) < (1 << 31) +} + +fn sequence_is_before(candidate: u32, current: u32) -> bool { + current != candidate && current.wrapping_sub(candidate) < (1 << 31) } fn instant_abs_diff(left: Instant, right: Instant) -> Duration { diff --git a/crates/irtt-client/src/client/tests/mod.rs b/crates/irtt-client/src/client/tests/mod.rs index 65975bb..8212f33 100644 --- a/crates/irtt-client/src/client/tests/mod.rs +++ b/crates/irtt-client/src/client/tests/mod.rs @@ -17,4 +17,5 @@ mod negotiation; mod no_test; mod open; mod probes_replies; +mod sequence; mod ttl; diff --git a/crates/irtt-client/src/client/tests/probes_replies.rs b/crates/irtt-client/src/client/tests/probes_replies.rs index ace51df..c478ff9 100644 --- a/crates/irtt-client/src/client/tests/probes_replies.rs +++ b/crates/irtt-client/src/client/tests/probes_replies.rs @@ -1584,6 +1584,66 @@ fn send_probe_wraps_wire_sequence_at_u32_max() { server.join(); } +#[test] +fn wrapped_reply_after_u32_max_is_not_late() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut seqs = Vec::new(); + for _ in 0..2 { + let mut buf = [0_u8; 2048]; + let (size, _) = socket.recv_from(&mut buf).unwrap(); + tx.send(buf[..size].to_vec()).unwrap(); + seqs.push(u32::from_le_bytes(buf[12..16].try_into().unwrap())); + } + assert_eq!(seqs, vec![u32::MAX, 0]); + + let ts = TimestampFields::default(); + let reply_max = echo_reply_packet(TOKEN, u32::MAX, ¶ms, &ts, None); + socket.send_to(&reply_max, peer).unwrap(); + thread::sleep(Duration::from_millis(10)); + let reply_zero = echo_reply_packet(TOKEN, 0, ¶ms, &ts, None); + socket.send_to(&reply_zero, peer).unwrap(); + }); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + let session = client.session.as_mut().unwrap(); + session.next_wire_seq = u32::MAX; + session.next_logical_seq = 41; + + client.send_probe().unwrap(); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let first = client.recv_once().unwrap(); + assert!(matches!( + &first[0], + ClientEvent::EchoReply { seq: u32::MAX, .. } + )); + + thread::sleep(Duration::from_millis(30)); + let second = client.recv_once().unwrap(); + assert!( + matches!(&second[0], ClientEvent::EchoReply { seq: 0, .. }), + "freshly wrapped reply should not be late, got {:?}", + second[0] + ); + server.join(); +} + #[test] fn send_probe_after_sending_done_is_noop() { let params = default_params(); diff --git a/crates/irtt-client/src/client/tests/sequence.rs b/crates/irtt-client/src/client/tests/sequence.rs new file mode 100644 index 0000000..d546cb5 --- /dev/null +++ b/crates/irtt-client/src/client/tests/sequence.rs @@ -0,0 +1,45 @@ +use super::*; +use std::time::Instant; + +use crate::{ + probe::{CompletedSet, PendingMap, TimedOutMap}, + session::ActiveSession, +}; + +#[test] +fn sequence_ordering_handles_normal_and_wrapped_values() { + assert!(sequence_is_after(11, 10)); + assert!(sequence_is_before(9, 10)); + assert!(!sequence_is_after(10, 10)); + assert!(!sequence_is_before(10, 10)); + + assert!(sequence_is_after(0, u32::MAX - 1)); + assert!(!sequence_is_before(0, u32::MAX - 1)); + assert!(sequence_is_after(1, u32::MAX)); + assert!(!sequence_is_before(1, u32::MAX)); + + assert!(sequence_is_before(u32::MAX, 1)); + assert!(!sequence_is_after(u32::MAX, 1)); +} + +#[test] +fn highest_received_seq_updates_across_wrap() { + let mut session = ActiveSession { + next_wire_seq: 0, + next_logical_seq: 0, + highest_received_seq: Some(u32::MAX), + packets_sent: 0, + start_mono: Instant::now(), + end_mono: None, + next_send_at: Instant::now(), + pending: PendingMap::new(8), + timed_out: TimedOutMap::new(8), + completed: CompletedSet::new(8), + sending_done: false, + }; + + update_highest_received(&mut session, 0); + assert_eq!(session.highest_received_seq, Some(0)); + update_highest_received(&mut session, u32::MAX); + assert_eq!(session.highest_received_seq, Some(0)); +}