From dfe0534905ad0bc85cc27a21498f855e438f0f29 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Jul 2026 15:55:18 +0700 Subject: [PATCH] test(managed): cover group late replies during final drain Exercise a real one-target ManagedClientGroup through timeout, Draining, a gated retained reply, LateReply classification, local close, and authoritative TargetFinished publication. Synchronize release after EchoLoss and a scheduler-handled no-op update so the ordering is deterministic rather than delay-dependent. Add a silent-server control that verifies completion only after the final-drain deadline, and assert bounded join, queued-event draining through disconnection, target counters, terminal outcome equality, and group aggregates in both cases. Harden the analogous single-client late-drain test with a viable one-probe run window and packet-shape checks. No production final-drain behavior change was required. --- crates/irtt-client/src/managed/group.rs | 283 +++++++++++++++++++++++ crates/irtt-client/src/managed/runner.rs | 12 +- 2 files changed, 293 insertions(+), 2 deletions(-) diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 21315c6..862f609 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -2084,6 +2084,65 @@ mod tests { } } + fn start_gated_reply_server(params: Params) -> (FakeServer, mpsc::SyncSender<()>) { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let addr = socket.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::sync_channel(0); + 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(); + socket + .set_read_timeout(Some(Duration::from_millis(800))) + .unwrap(); + + let (echo, peer) = recv_request_timeout(&socket).expect("missing echo request"); + assert_eq!(echo[3] & flags::FLAG_CLOSE, 0); + assert!(echo.len() >= 16, "echo request too short: {}", echo.len()); + let seq = u32::from_le_bytes(echo[12..16].try_into().unwrap()); + tx.send(ServerObservation::Echo { + seq, + at: Instant::now(), + }) + .unwrap(); + release_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting to release delayed reply"); + + let timestamps = TimestampFields { + recv_wall: Some(1_000_000_000), + recv_mono: Some(100_000), + send_wall: Some(1_001_000_000), + send_mono: Some(1_100_000), + ..Default::default() + }; + socket + .send_to(&echo_reply_packet(TOKEN, seq, ¶ms, ×tamps), peer) + .unwrap(); + + while let Some((packet, _)) = recv_request_timeout(&socket) { + if packet[3] & flags::FLAG_CLOSE != 0 { + tx.send(ServerObservation::Close { at: Instant::now() }) + .unwrap(); + break; + } + } + }); + ( + FakeServer { + addr, + _observations: rx, + done, + }, + release_tx, + ) + } + fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec, SocketAddr)> { let mut buf = [0_u8; 2048]; socket @@ -2195,6 +2254,36 @@ mod tests { } } + fn final_drain_event_sequence(events: &[ManagedGroupEvent]) -> Vec<&'static str> { + events + .iter() + .filter_map(|event| match event { + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::SessionStarted { .. }, + .. + }) => Some("started"), + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::EchoSent { seq: 0, .. }, + .. + }) => Some("sent"), + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::EchoLoss { seq: 0, .. }, + .. + }) => Some("loss"), + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::LateReply { seq: 0, .. }, + .. + }) => Some("late"), + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::SessionClosed { .. }, + .. + }) => Some("closed"), + ManagedGroupEvent::TargetFinished(_) => Some("finished"), + _ => None, + }) + .collect() + } + fn join_group_with_timeout( session: ManagedClientGroupSession, timeout: Duration, @@ -3504,6 +3593,200 @@ mod tests { .all(|target| target.end_reason == ManagedTargetEndReason::Removed)); } + #[test] + fn finite_group_drains_late_reply_during_final_drain() { + let interval = Duration::from_millis(100); + let duration = interval; + let params = test_params(Some(duration), interval); + let (server, release_reply) = start_gated_reply_server(params); + let target = target("late", server.addr); + let mut config = group_config(Some(duration), interval, ManagedGroupPacing::Staggered); + config.client.probe_timeout = Duration::from_millis(20); + let (session, sub) = ManagedClientGroup::start_with_subscription( + config, + vec![target.clone()], + SubscriberConfig { + capacity: 16, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + let mut events = Vec::new(); + loop { + let event = sub + .recv_timeout(Duration::from_secs(2)) + .expect("subscription disconnected before EchoLoss") + .expect("timed out waiting for EchoLoss"); + let saw_loss = matches!( + &event, + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::EchoLoss { seq: 0, .. }, + .. + }) + ); + events.push(event); + if saw_loss { + break; + } + } + + // EchoLoss is published before this scheduler iteration transitions the + // target to Draining. A synchronous no-op update is handled at the next + // loop head, so its return proves that transition completed before the + // server releases the retained reply. + session.update_targets(vec![target]).unwrap(); + release_reply.send(()).unwrap(); + let outcome = join_group_with_timeout(session, Duration::from_secs(1)); + events.extend(collect_group_events_until_disconnected(&sub)); + server.join(); + + assert_eq!( + final_drain_event_sequence(&events), + ["started", "sent", "loss", "late", "closed", "finished"] + ); + assert_eq!( + events + .iter() + .filter(|event| matches!( + event, + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::EchoLoss { seq: 0, .. }, + .. + }) + )) + .count(), + 1 + ); + assert_eq!( + events + .iter() + .filter(|event| matches!( + event, + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::LateReply { + seq: 0, + sent_at: Some(_), + rtt: Some(_), + .. + }, + .. + }) + )) + .count(), + 1 + ); + + assert_eq!( + outcome.end_reason, + ManagedGroupEndReason::AllTargetsComplete + ); + assert_eq!(outcome.targets.len(), 1); + let target = &outcome.targets[0]; + assert_eq!(target.end_reason, ManagedTargetEndReason::TestComplete); + assert_eq!(target.packets_sent, 1); + assert_eq!(target.replies_received, 0); + assert_eq!(target.duplicates, 0); + assert_eq!(target.late, 1); + assert_eq!(target.warning_events, 0); + assert_eq!(outcome.total_target_outcomes, 1); + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(outcome.peer_closed_target_outcomes, 0); + assert_eq!(outcome.failed_target_outcomes, 0); + assert_eq!(outcome.discarded_target_outcomes, 0); + assert_eq!( + events.iter().find_map(|event| match event { + ManagedGroupEvent::TargetFinished(target) => Some(target), + _ => None, + }), + Some(target) + ); + assert_eq!(sub.dropped_events(), 0); + assert_eq!(sub.try_recv(), Err(EventSubscriptionError::Disconnected)); + } + + #[test] + fn finite_group_final_drain_expires_without_late_reply() { + let interval = Duration::from_millis(100); + let duration = interval; + let server = start_silent_runtime_server(test_params(Some(duration), interval)); + let mut config = group_config(Some(duration), interval, ManagedGroupPacing::Staggered); + config.client.probe_timeout = Duration::from_millis(20); + let (session, sub) = ManagedClientGroup::start_with_subscription( + config, + vec![target("silent", server.addr)], + SubscriberConfig { + capacity: 16, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + let outcome = join_group_with_timeout(session, Duration::from_secs(1)); + let events = collect_group_events_until_disconnected(&sub); + server.join(); + + assert_eq!( + final_drain_event_sequence(&events), + ["started", "sent", "loss", "closed", "finished"] + ); + assert!(!events.iter().any(|event| matches!( + event, + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::LateReply { .. }, + .. + }) + ))); + let timeout_at = events + .iter() + .find_map(|event| match event { + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::EchoLoss { timeout_at, .. }, + .. + }) => Some(*timeout_at), + _ => None, + }) + .expect("missing EchoLoss"); + let closed_at = events + .iter() + .find_map(|event| match event { + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::SessionClosed { at, .. }, + .. + }) => Some(at.mono), + _ => None, + }) + .expect("missing SessionClosed"); + assert!(closed_at >= timeout_at + GROUP_FINAL_DRAIN); + + assert_eq!( + outcome.end_reason, + ManagedGroupEndReason::AllTargetsComplete + ); + assert_eq!(outcome.targets.len(), 1); + let target = &outcome.targets[0]; + assert_eq!(target.end_reason, ManagedTargetEndReason::TestComplete); + assert_eq!(target.packets_sent, 1); + assert_eq!(target.replies_received, 0); + assert_eq!(target.duplicates, 0); + assert_eq!(target.late, 0); + assert_eq!(target.warning_events, 0); + assert_eq!(outcome.total_target_outcomes, 1); + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(outcome.peer_closed_target_outcomes, 0); + assert_eq!(outcome.failed_target_outcomes, 0); + assert_eq!(outcome.discarded_target_outcomes, 0); + assert_eq!( + events.iter().find_map(|event| match event { + ManagedGroupEvent::TargetFinished(target) => Some(target), + _ => None, + }), + Some(target) + ); + assert_eq!(sub.dropped_events(), 0); + assert_eq!(sub.try_recv(), Err(EventSubscriptionError::Disconnected)); + } + #[test] fn finite_static_group_still_completes_naturally() { let interval = Duration::from_millis(10); diff --git a/crates/irtt-client/src/managed/runner.rs b/crates/irtt-client/src/managed/runner.rs index f45848d..ed05c20 100644 --- a/crates/irtt-client/src/managed/runner.rs +++ b/crates/irtt-client/src/managed/runner.rs @@ -465,6 +465,12 @@ mod tests { let Some((packet, peer)) = recv_request_timeout(&socket) else { return; }; + assert_eq!(packet[3] & flags::FLAG_CLOSE, 0); + assert!( + packet.len() >= 16, + "echo request too short: {}", + packet.len() + ); let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); thread::sleep(delay); let ts = TimestampFields { @@ -725,10 +731,12 @@ mod tests { #[test] fn finite_managed_run_drains_late_reply_after_timeout_before_close() { - let duration = Duration::from_millis(1); - let params = test_params(Some(duration), Duration::from_millis(10)); + let interval = Duration::from_millis(100); + let duration = interval; + let params = test_params(Some(duration), interval); let server = start_delayed_reply_server(params, Duration::from_millis(60)); let mut cfg = config(server.addr, Some(duration)); + cfg.interval = interval; cfg.probe_timeout = Duration::from_millis(20); let (session, sub) = ManagedClient::start_with_subscription(