fix(client): handle sequence wrap in reply ordering
This commit is contained in:
1 parent
294d97a13d
commit
b53e650c53
4 files changed
+128
-9
No files matched your search
@@ -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, ¶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();
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
Reference in new issue
Block a user