fix(client): preserve peer-close counters and provenance

Snapshot sent counters when a runtime leaves its live session state, and expose PeerClosed from the single managed runner instead of folding it into TestComplete. Group timeout polling now recognizes a peer-closed runtime before attempting another operation, preserving the same provenance and counters.

Regression coverage verifies single and group peer-close outcomes, finite completion, cancellation, no-test completion, and final counters.
This commit is contained in:
Nils Andreas Svee committed 2026-07-27 23:55:04 +07:00
1 parent 40167ab51a
commit 91db85cc63
3 files changed
+117 -20

No files matched your search

+84 -1
View File
@@ -975,6 +975,10 @@ fn poll_active_timeouts(shared: &GroupShared, now: Instant) {
) { ) {
continue; continue;
} }
if target.runtime.is_peer_closed() {
target.mark_finished(ManagedTargetEndReason::PeerClosed);
continue;
}
match target.runtime.poll_timeouts_at(now) { match target.runtime.poll_timeouts_at(now) {
Ok(events) => publish_events(&shared.hub, &mut target, events), Ok(events) => publish_events(&shared.hub, &mut target, events),
Err(err) => target.mark_finished(ManagedTargetEndReason::Failed { Err(err) => target.mark_finished(ManagedTargetEndReason::Failed {
@@ -1501,6 +1505,46 @@ mod tests {
} }
} }
fn start_peer_close_server(params: Params) -> FakeServer {
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
let addr = socket.local_addr().unwrap();
let (tx, rx) = mpsc::channel();
let done = thread::spawn(move || {
let (open, peer) = recv_request_timeout(&socket).expect("missing open request");
assert_ne!(open[3] & FLAG_OPEN, 0);
tx.send(ServerObservation::Open { at: Instant::now() })
.unwrap();
socket
.send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
.unwrap();
let (echo, peer) = recv_request_timeout(&socket).expect("missing echo request");
let seq = u32::from_le_bytes(echo[12..16].try_into().unwrap());
tx.send(ServerObservation::Echo {
seq,
at: Instant::now(),
})
.unwrap();
socket
.send_to(
&echo_reply_packet_with_flags(
TOKEN,
seq,
&params,
&TimestampFields::default(),
FLAG_REPLY | flags::FLAG_CLOSE,
),
peer,
)
.unwrap();
});
FakeServer {
addr,
_observations: rx,
done,
}
}
fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> { fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> {
let mut buf = [0_u8; 2048]; let mut buf = [0_u8; 2048];
socket socket
@@ -1523,13 +1567,23 @@ mod tests {
seq: u32, seq: u32,
params: &Params, params: &Params,
timestamps: &TimestampFields, timestamps: &TimestampFields,
) -> Vec<u8> {
echo_reply_packet_with_flags(token, seq, params, timestamps, FLAG_REPLY)
}
fn echo_reply_packet_with_flags(
token: u64,
seq: u32,
params: &Params,
timestamps: &TimestampFields,
flags: u8,
) -> Vec<u8> { ) -> Vec<u8> {
let layout = PacketLayout::echo(false, params); let layout = PacketLayout::echo(false, params);
let packet_len = echo_packet_len(false, params).unwrap(); let packet_len = echo_packet_len(false, params).unwrap();
let mut packet = Vec::with_capacity(packet_len); let mut packet = Vec::with_capacity(packet_len);
packet.extend_from_slice(&MAGIC); packet.extend_from_slice(&MAGIC);
packet.push(FLAG_REPLY); packet.push(flags);
packet.extend_from_slice(&token.to_le_bytes()); packet.extend_from_slice(&token.to_le_bytes());
packet.extend_from_slice(&seq.to_le_bytes()); packet.extend_from_slice(&seq.to_le_bytes());
@@ -1704,6 +1758,10 @@ mod tests {
ManagedGroupEndReason::AllTargetsComplete ManagedGroupEndReason::AllTargetsComplete
); );
assert_eq!(outcome.targets.len(), 2); assert_eq!(outcome.targets.len(), 2);
assert!(outcome
.targets
.iter()
.all(|target| target.packets_sent > 0 && target.replies_received > 0));
for id in [TargetId::from("a"), TargetId::from("b")] { for id in [TargetId::from("a"), TargetId::from("b")] {
assert!(events.iter().any(|event| { assert!(events.iter().any(|event| {
event.target == id && matches!(event.event, ClientEvent::SessionStarted { .. }) event.target == id && matches!(event.event, ClientEvent::SessionStarted { .. })
@@ -1717,6 +1775,31 @@ mod tests {
} }
} }
#[test]
fn peer_close_preserves_group_outcome_counters_and_provenance() {
let interval = Duration::from_millis(10);
let server = start_peer_close_server(test_params(None, interval));
let session = ManagedClientGroup::start(
group_config(None, interval, ManagedGroupPacing::Staggered),
vec![target("peer", server.addr)],
)
.unwrap();
let outcome = session.join().unwrap();
server.join();
assert_eq!(
outcome.end_reason,
ManagedGroupEndReason::AllTargetsComplete
);
assert_eq!(outcome.targets.len(), 1);
let target = &outcome.targets[0];
assert_eq!(target.end_reason, ManagedTargetEndReason::PeerClosed);
assert_eq!(target.packets_sent, 1);
assert_eq!(target.replies_received, 1);
}
#[test] #[test]
fn staggered_pacing_does_not_burst_active_targets() { fn staggered_pacing_does_not_burst_active_targets() {
let duration = Duration::from_millis(170); let duration = Duration::from_millis(170);
+20 -4
View File
@@ -72,6 +72,8 @@ pub enum SessionEndReason {
/// The negotiated finite test duration completed and the client closed the /// The negotiated finite test duration completed and the client closed the
/// session normally. /// session normally.
TestComplete, TestComplete,
/// The peer ended the session with an authenticated close-flagged reply.
PeerClosed,
/// Cancellation was requested through [`ManagedClientSession::stop`] or by /// Cancellation was requested through [`ManagedClientSession::stop`] or by
/// dropping the session handle. /// dropping the session handle.
Cancelled, Cancelled,
@@ -267,12 +269,16 @@ fn run_client(
publish_events(&hub, &mut counters, close_events); publish_events(&hub, &mut counters, close_events);
} }
Ok(SessionOutcome { let end_reason = if cancelled {
end_reason: if cancelled {
SessionEndReason::Cancelled SessionEndReason::Cancelled
} else if client.is_peer_closed() {
SessionEndReason::PeerClosed
} else { } else {
SessionEndReason::TestComplete SessionEndReason::TestComplete
}, };
Ok(SessionOutcome {
end_reason,
packets_sent, packets_sent,
replies_received: counters.replies_received, replies_received: counters.replies_received,
duplicates: counters.duplicates, duplicates: counters.duplicates,
@@ -683,6 +689,8 @@ mod tests {
server.join(); server.join();
assert_eq!(outcome.end_reason, SessionEndReason::TestComplete); assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
assert!(outcome.packets_sent > 0);
assert!(outcome.replies_received > 0);
assert!(events assert!(events
.iter() .iter()
.any(|event| matches!(event, ClientEvent::SessionStarted { .. }))); .any(|event| matches!(event, ClientEvent::SessionStarted { .. })));
@@ -716,6 +724,9 @@ mod tests {
server.join(); server.join();
assert_eq!(outcome.end_reason, SessionEndReason::TestComplete); assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
assert_eq!(outcome.packets_sent, 1);
assert_eq!(outcome.replies_received, 0);
assert_eq!(outcome.late, 1);
assert!(events assert!(events
.iter() .iter()
.any(|event| matches!(event, ClientEvent::EchoLoss { seq: 0, .. }))); .any(|event| matches!(event, ClientEvent::EchoLoss { seq: 0, .. })));
@@ -755,7 +766,9 @@ mod tests {
let outcome = session.join().unwrap(); let outcome = session.join().unwrap();
server.join(); server.join();
assert_eq!(outcome.end_reason, SessionEndReason::TestComplete); assert_eq!(outcome.end_reason, SessionEndReason::PeerClosed);
assert_eq!(outcome.packets_sent, 1);
assert_eq!(outcome.replies_received, 1);
assert!(events assert!(events
.iter() .iter()
.any(|event| matches!(event, ClientEvent::EchoReply { .. }))); .any(|event| matches!(event, ClientEvent::EchoReply { .. })));
@@ -786,6 +799,8 @@ mod tests {
server.join(); server.join();
assert_eq!(outcome.end_reason, SessionEndReason::Cancelled); assert_eq!(outcome.end_reason, SessionEndReason::Cancelled);
assert!(outcome.packets_sent > 0);
assert!(outcome.replies_received > 0);
assert!(events assert!(events
.iter() .iter()
.any(|event| matches!(event, ClientEvent::SessionClosed { .. }))); .any(|event| matches!(event, ClientEvent::SessionClosed { .. })));
@@ -834,5 +849,6 @@ mod tests {
done.join().unwrap(); done.join().unwrap();
assert_eq!(outcome.end_reason, SessionEndReason::NoTestComplete); assert_eq!(outcome.end_reason, SessionEndReason::NoTestComplete);
assert_eq!(outcome.packets_sent, 0);
} }
} }
+13 -15
View File
@@ -32,6 +32,7 @@ pub(crate) struct SessionRuntime {
negotiated: Option<NegotiatedParams>, negotiated: Option<NegotiatedParams>,
phase: ClientPhase, phase: ClientPhase,
session: Option<ActiveSession>, session: Option<ActiveSession>,
final_packets_sent: u64,
} }
#[derive(Debug, Clone, Copy)] #[derive(Debug, Clone, Copy)]
@@ -65,6 +66,7 @@ impl SessionRuntime {
negotiated: None, negotiated: None,
phase: ClientPhase::Connected, phase: ClientPhase::Connected,
session: None, session: None,
final_packets_sent: 0,
}) })
} }
@@ -346,13 +348,7 @@ impl SessionRuntime {
let packet = let packet =
encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?; encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?;
send(&packet)?; send(&packet)?;
self.phase = ClientPhase::Closed { self.transition_to_closed(CloseSource::Local);
source: CloseSource::Local,
};
if let Some(session) = self.session.as_mut() {
session.timed_out.clear();
}
self.session = None;
Ok(vec![ClientEvent::SessionClosed { Ok(vec![ClientEvent::SessionClosed {
remote: self.remote, remote: self.remote,
@@ -389,7 +385,7 @@ impl SessionRuntime {
pub(crate) fn packets_sent(&self) -> u64 { pub(crate) fn packets_sent(&self) -> u64 {
self.session self.session
.as_ref() .as_ref()
.map_or(0, |session| session.packets_sent) .map_or(self.final_packets_sent, |session| session.packets_sent)
} }
pub(crate) fn is_open(&self) -> bool { pub(crate) fn is_open(&self) -> bool {
@@ -620,19 +616,21 @@ impl SessionRuntime {
} }
fn close_from_peer(&mut self, token: u64, now: ClientTimestamp, events: &mut Vec<ClientEvent>) { fn close_from_peer(&mut self, token: u64, now: ClientTimestamp, events: &mut Vec<ClientEvent>) {
self.phase = ClientPhase::Closed { self.transition_to_closed(CloseSource::Peer);
source: CloseSource::Peer,
};
if let Some(session) = self.session.as_mut() {
session.timed_out.clear();
}
self.session = None;
events.push(ClientEvent::SessionClosed { events.push(ClientEvent::SessionClosed {
remote: self.remote, remote: self.remote,
token, token,
at: now, at: now,
}); });
} }
fn transition_to_closed(&mut self, source: CloseSource) {
if let Some(mut session) = self.session.take() {
session.timed_out.clear();
self.final_packets_sent = session.packets_sent;
}
self.phase = ClientPhase::Closed { source };
}
} }
pub(crate) fn recv_buffer_size( pub(crate) fn recv_buffer_size(