From 548363ad055e3f2ba92d8d16e2198428a5532854 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Apr 2026 21:52:04 +0700 Subject: [PATCH] fix(client): improve probe tracking and reply handling --- crates/irtt-client/src/client.rs | 18 +++++++++++++----- crates/irtt-client/src/probe.rs | 7 ++++++- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index db49dc7..f0b030e 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -173,6 +173,8 @@ impl Client { } } + session.pending.check_capacity()?; + let negotiated = self .negotiated .as_ref() @@ -230,7 +232,7 @@ impl Client { Ok(vec![]) } - pub fn recv_once(&mut self, now: ClientTimestamp) -> Result, ClientError> { + pub fn recv_once(&mut self) -> Result, ClientError> { match self.phase { ClientPhase::Open { .. } => {} ClientPhase::Closed => return Err(ClientError::AlreadyClosed), @@ -394,13 +396,13 @@ impl Client { remote: self.remote, received_at: now, }]) - } else { - let highest_seen = session.highest_received_seq.unwrap_or(wire_seq); - update_highest_received(session, wire_seq); + } 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. Ok(vec![ClientEvent::LateReply { seq: wire_seq, logical_seq: None, - highest_seen, + highest_seen: session.highest_received_seq.unwrap(), remote: self.remote, sent_at: None, received_at: now, @@ -410,6 +412,12 @@ impl Client { received_stats: build_received_stats(&reply), packet_meta: PacketMeta::default(), }]) + } else { + Ok(vec![ClientEvent::Warning { + message: format!( + "dropped reply with untracked seq {wire_seq} (no pending or completed entry)" + ), + }]) } } diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 5a520ce..c0fd246 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -24,12 +24,17 @@ impl PendingMap { } } - pub fn insert(&mut self, probe: PendingProbe) -> Result<(), ClientError> { + pub fn check_capacity(&self) -> Result<(), ClientError> { if self.map.len() >= self.max_capacity { return Err(ClientError::PendingLimitExceeded { limit: self.max_capacity, }); } + Ok(()) + } + + pub fn insert(&mut self, probe: PendingProbe) -> Result<(), ClientError> { + self.check_capacity()?; self.map.insert(probe.wire_seq, probe); Ok(()) }