diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 5dfb7fd..cd3a160 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -975,6 +975,10 @@ fn poll_active_timeouts(shared: &GroupShared, now: Instant) { ) { continue; } + if target.runtime.is_peer_closed() { + target.mark_finished(ManagedTargetEndReason::PeerClosed); + continue; + } match target.runtime.poll_timeouts_at(now) { Ok(events) => publish_events(&shared.hub, &mut target, events), 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, ¶ms), 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, + ¶ms, + &TimestampFields::default(), + FLAG_REPLY | flags::FLAG_CLOSE, + ), + peer, + ) + .unwrap(); + }); + FakeServer { + addr, + _observations: rx, + done, + } + } + fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec, SocketAddr)> { let mut buf = [0_u8; 2048]; socket @@ -1523,13 +1567,23 @@ mod tests { seq: u32, params: &Params, timestamps: &TimestampFields, + ) -> Vec { + 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 { let layout = PacketLayout::echo(false, params); let packet_len = echo_packet_len(false, params).unwrap(); let mut packet = Vec::with_capacity(packet_len); packet.extend_from_slice(&MAGIC); - packet.push(FLAG_REPLY); + packet.push(flags); packet.extend_from_slice(&token.to_le_bytes()); packet.extend_from_slice(&seq.to_le_bytes()); @@ -1704,6 +1758,10 @@ mod tests { ManagedGroupEndReason::AllTargetsComplete ); 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")] { assert!(events.iter().any(|event| { 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] fn staggered_pacing_does_not_burst_active_targets() { let duration = Duration::from_millis(170); diff --git a/crates/irtt-client/src/managed/runner.rs b/crates/irtt-client/src/managed/runner.rs index 7f1de6e..7c5c47d 100644 --- a/crates/irtt-client/src/managed/runner.rs +++ b/crates/irtt-client/src/managed/runner.rs @@ -72,6 +72,8 @@ pub enum SessionEndReason { /// The negotiated finite test duration completed and the client closed the /// session normally. TestComplete, + /// The peer ended the session with an authenticated close-flagged reply. + PeerClosed, /// Cancellation was requested through [`ManagedClientSession::stop`] or by /// dropping the session handle. Cancelled, @@ -267,12 +269,16 @@ fn run_client( publish_events(&hub, &mut counters, close_events); } + let end_reason = if cancelled { + SessionEndReason::Cancelled + } else if client.is_peer_closed() { + SessionEndReason::PeerClosed + } else { + SessionEndReason::TestComplete + }; + Ok(SessionOutcome { - end_reason: if cancelled { - SessionEndReason::Cancelled - } else { - SessionEndReason::TestComplete - }, + end_reason, packets_sent, replies_received: counters.replies_received, duplicates: counters.duplicates, @@ -683,6 +689,8 @@ mod tests { server.join(); assert_eq!(outcome.end_reason, SessionEndReason::TestComplete); + assert!(outcome.packets_sent > 0); + assert!(outcome.replies_received > 0); assert!(events .iter() .any(|event| matches!(event, ClientEvent::SessionStarted { .. }))); @@ -716,6 +724,9 @@ mod tests { server.join(); 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 .iter() .any(|event| matches!(event, ClientEvent::EchoLoss { seq: 0, .. }))); @@ -755,7 +766,9 @@ mod tests { let outcome = session.join().unwrap(); 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 .iter() .any(|event| matches!(event, ClientEvent::EchoReply { .. }))); @@ -786,6 +799,8 @@ mod tests { server.join(); assert_eq!(outcome.end_reason, SessionEndReason::Cancelled); + assert!(outcome.packets_sent > 0); + assert!(outcome.replies_received > 0); assert!(events .iter() .any(|event| matches!(event, ClientEvent::SessionClosed { .. }))); @@ -834,5 +849,6 @@ mod tests { done.join().unwrap(); assert_eq!(outcome.end_reason, SessionEndReason::NoTestComplete); + assert_eq!(outcome.packets_sent, 0); } } diff --git a/crates/irtt-client/src/runtime.rs b/crates/irtt-client/src/runtime.rs index 041a5f4..9e679c8 100644 --- a/crates/irtt-client/src/runtime.rs +++ b/crates/irtt-client/src/runtime.rs @@ -32,6 +32,7 @@ pub(crate) struct SessionRuntime { negotiated: Option, phase: ClientPhase, session: Option, + final_packets_sent: u64, } #[derive(Debug, Clone, Copy)] @@ -65,6 +66,7 @@ impl SessionRuntime { negotiated: None, phase: ClientPhase::Connected, session: None, + final_packets_sent: 0, }) } @@ -346,13 +348,7 @@ impl SessionRuntime { let packet = encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?; send(&packet)?; - self.phase = ClientPhase::Closed { - source: CloseSource::Local, - }; - if let Some(session) = self.session.as_mut() { - session.timed_out.clear(); - } - self.session = None; + self.transition_to_closed(CloseSource::Local); Ok(vec![ClientEvent::SessionClosed { remote: self.remote, @@ -389,7 +385,7 @@ impl SessionRuntime { pub(crate) fn packets_sent(&self) -> u64 { self.session .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 { @@ -620,19 +616,21 @@ impl SessionRuntime { } fn close_from_peer(&mut self, token: u64, now: ClientTimestamp, events: &mut Vec) { - self.phase = ClientPhase::Closed { - source: CloseSource::Peer, - }; - if let Some(session) = self.session.as_mut() { - session.timed_out.clear(); - } - self.session = None; + self.transition_to_closed(CloseSource::Peer); events.push(ClientEvent::SessionClosed { remote: self.remote, token, 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(