fix(client): improve probe tracking and reply handling
This commit is contained in:
1 parent
7cbc30a3a2
commit
548363ad05
2 files changed
+19
-6
No files matched your search
@@ -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)"
|
||||||
|
),
|
||||||
|
}])
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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(())
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user