fix(client): improve probe tracking and reply handling

This commit is contained in:
Nils Andreas Svee committed 2026-04-28 21:55:49 +07:00
1 parent 2380d76da3
commit 3534a59fce
2 files changed
+19 -6

No files matched your search

+13 -5
View File
@@ -173,6 +173,8 @@ impl Client {
} }
} }
session.pending.check_capacity()?;
let negotiated = self let negotiated = self
.negotiated .negotiated
.as_ref() .as_ref()
@@ -230,7 +232,7 @@ impl Client {
Ok(vec![]) Ok(vec![])
} }
pub fn recv_once(&mut self, now: ClientTimestamp) -> Result<Vec<ClientEvent>, ClientError> { pub fn recv_once(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
match self.phase { match self.phase {
ClientPhase::Open { .. } => {} ClientPhase::Open { .. } => {}
ClientPhase::Closed => return Err(ClientError::AlreadyClosed), ClientPhase::Closed => return Err(ClientError::AlreadyClosed),
@@ -394,13 +396,13 @@ impl Client {
remote: self.remote, remote: self.remote,
received_at: now, received_at: now,
}]) }])
} else { } else if session.highest_received_seq.is_some_and(|h| wire_seq < h) {
let highest_seen = session.highest_received_seq.unwrap_or(wire_seq); // TODO: u32 ordering is acceptable for finite tests but continuous
update_highest_received(session, wire_seq); // probing will eventually require wrap-aware comparison.
Ok(vec![ClientEvent::LateReply { Ok(vec![ClientEvent::LateReply {
seq: wire_seq, seq: wire_seq,
logical_seq: None, logical_seq: None,
highest_seen, highest_seen: session.highest_received_seq.unwrap(),
remote: self.remote, remote: self.remote,
sent_at: None, sent_at: None,
received_at: now, received_at: now,
@@ -410,6 +412,12 @@ impl Client {
received_stats: build_received_stats(&reply), received_stats: build_received_stats(&reply),
packet_meta: PacketMeta::default(), packet_meta: PacketMeta::default(),
}]) }])
} else {
Ok(vec![ClientEvent::Warning {
message: format!(
"dropped reply with untracked seq {wire_seq} (no pending or completed entry)"
),
}])
} }
} }
+6 -1
View File
@@ -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 { if self.map.len() >= self.max_capacity {
return Err(ClientError::PendingLimitExceeded { return Err(ClientError::PendingLimitExceeded {
limit: self.max_capacity, limit: self.max_capacity,
}); });
} }
Ok(())
}
pub fn insert(&mut self, probe: PendingProbe) -> Result<(), ClientError> {
self.check_capacity()?;
self.map.insert(probe.wire_seq, probe); self.map.insert(probe.wire_seq, probe);
Ok(()) Ok(())
} }