fix(client): handle sequence wrap in reply ordering

This commit is contained in:
Nils Andreas Svee committed 2026-05-07 21:25:51 +07:00
1 parent 1f167071fb
commit 4cd0201c50
4 files changed
+128 -9

No files matched your search

+22 -9
View File
@@ -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 {
@@ -17,4 +17,5 @@ mod negotiation;
mod no_test;
mod open;
mod probes_replies;
mod sequence;
mod ttl;
@@ -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, &params, 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, &params, &ts, None);
socket.send_to(&reply_max, peer).unwrap();
thread::sleep(Duration::from_millis(10));
let reply_zero = echo_reply_packet(TOKEN, 0, &params, &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();
@@ -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));
}