From 1cb6e07706ea2cc6814bdd0e0427401bee732e0a Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 20:59:42 +0700 Subject: [PATCH] test(managed): verify unified task lifecycle --- crates/irtt-client/src/managed/task.rs | 17 + crates/irtt-client/src/managed/tests.rs | 626 ++++++++++++++++++++++++ 2 files changed, 643 insertions(+) create mode 100644 crates/irtt-client/src/managed/tests.rs diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 1e7e680..411ef70 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -142,6 +142,8 @@ impl Future for ManagedStopReceipt { type ConnectFuture = Pin> + Send + 'static>>; type WakeFuture = Pin>> + Send + 'static>>; +#[cfg(test)] +type EventObservations = Arc)>>>; fn arm_wake(mut receiver: watch::Receiver<()>) -> WakeFuture { Box::pin(async move { @@ -313,6 +315,8 @@ pub struct ManagedClientTask { stop_observed: bool, final_outcome: Option>, _next_generation: u64, + #[cfg(test)] + event_observations: Option, } impl ManagedClientTask { @@ -323,6 +327,13 @@ impl ManagedClientTask { } fn publish_event(&self, event: ManagedEvent) { + #[cfg(test)] + if let Some(observations) = &self.event_observations { + observations + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push((event.clone(), Arc::clone(&self.resources().status.borrow()))); + } if let Some(events) = &self.resources().events { let _ = events.send(event); } @@ -1180,6 +1191,8 @@ fn build_task( stop_observed: false, final_outcome: None, _next_generation: next_generation, + #[cfg(test)] + event_observations: None, }; let handle = ManagedClientHandle { stop, @@ -1218,3 +1231,7 @@ fn close_timeout_failure() -> ManagedTargetFailure { message: Arc::from("best-effort close did not become writable before its deadline"), } } + +#[cfg(test)] +#[path = "tests.rs"] +mod tests; diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs new file mode 100644 index 0000000..2ce1d95 --- /dev/null +++ b/crates/irtt-client/src/managed/tests.rs @@ -0,0 +1,626 @@ +use std::{ + future::Future, + net::{SocketAddr, UdpSocket}, + pin::Pin, + sync::{Arc, Condvar, Mutex}, + task::{Context, Poll, Waker}, + thread::{self, JoinHandle}, + time::{Duration, Instant}, +}; + +use irtt_proto::{ + decode_close_request, decode_echo_request, decode_open_request, encode_echo_reply, + encode_open_reply, flags, EchoReply, OpenReply, ReceivedStats, StampAt, TimestampFields, +}; +use tokio::runtime::{Builder, Runtime}; + +use super::*; +use crate::{socket::resolution_call_counts, ClientAuthConfig, RunMode}; + +const TOKEN: u64 = 0x1234_5678_90ab_cdef; + +#[derive(Clone, Copy)] +enum ServerBehavior { + Echo, + NoTest, + PeerClose, + DelayedEcho(Duration), +} + +#[derive(Clone, Debug)] +enum PacketKind { + Open, + Probe, + Close, +} + +#[derive(Clone, Debug)] +struct PacketRecord { + kind: PacketKind, + at: Instant, +} + +struct TestServer { + addr: SocketAddr, + records: Arc>>, + probe_seen: Arc<(Mutex, Condvar)>, + thread: JoinHandle<()>, +} + +impl TestServer { + fn finish(self) -> Vec { + self.thread.join().unwrap(); + self.records.lock().unwrap().clone() + } +} + +fn start_server(behavior: ServerBehavior, key: Option>) -> TestServer { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let records = Arc::new(Mutex::new(Vec::new())); + let thread_records = Arc::clone(&records); + let probe_seen = Arc::new((Mutex::new(false), Condvar::new())); + let thread_probe_seen = Arc::clone(&probe_seen); + let thread = thread::spawn(move || { + let mut buffer = [0_u8; 8192]; + let (open_len, peer) = socket.recv_from(&mut buffer).unwrap(); + thread_records.lock().unwrap().push(PacketRecord { + kind: PacketKind::Open, + at: Instant::now(), + }); + let request = decode_open_request(&buffer[..open_len], key.as_deref()).unwrap(); + let no_test = matches!(behavior, ServerBehavior::NoTest); + let open_reply = encode_open_reply( + &OpenReply { + flags: flags::FLAG_OPEN + | flags::FLAG_REPLY + | if no_test { flags::FLAG_CLOSE } else { 0 }, + token: if no_test { 0 } else { TOKEN }, + params: request.params.clone(), + }, + key.as_deref(), + ) + .unwrap(); + socket.send_to(&open_reply, peer).unwrap(); + if no_test { + return; + } + + loop { + let Ok((len, packet_peer)) = socket.recv_from(&mut buffer) else { + return; + }; + let packet = &buffer[..len]; + if let Ok(probe) = decode_echo_request(packet, &request.params, key.as_deref()) { + thread_records.lock().unwrap().push(PacketRecord { + kind: PacketKind::Probe, + at: Instant::now(), + }); + let (seen, ready) = &*thread_probe_seen; + *seen.lock().unwrap() = true; + ready.notify_all(); + if let ServerBehavior::DelayedEcho(delay) = behavior { + thread::sleep(delay); + } + let peer_close = matches!(behavior, ServerBehavior::PeerClose); + let reply = encode_echo_reply( + &EchoReply { + flags: flags::FLAG_REPLY | if peer_close { flags::FLAG_CLOSE } else { 0 }, + token: TOKEN, + sequence: probe.sequence, + recv_count: None, + recv_window: None, + timestamps: TimestampFields::default(), + payload: Vec::new(), + }, + &request.params, + key.as_deref(), + ) + .unwrap(); + socket.send_to(&reply, packet_peer).unwrap(); + if peer_close { + return; + } + } else if decode_close_request(packet, key.as_deref()).is_ok() { + thread_records.lock().unwrap().push(PacketRecord { + kind: PacketKind::Close, + at: Instant::now(), + }); + return; + } + } + }); + TestServer { + addr, + records, + probe_seen, + thread, + } +} + +fn runtime() -> Runtime { + Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .unwrap() +} + +fn config(pacing: ManagedPacing) -> ManagedClientConfig { + ManagedClientConfig { + client: ClientConfig { + duration: Some(Duration::from_millis(130)), + interval: Duration::from_millis(60), + probe_timeout: Duration::from_millis(35), + open_timeouts: vec![Duration::from_millis(200)], + received_stats: ReceivedStats::None, + stamp_at: StampAt::None, + ..ClientConfig::default() + }, + pacing, + final_drain: Duration::from_millis(5), + ..ManagedClientConfig::default() + } +} + +fn target(id: &str, addr: SocketAddr) -> ManagedTargetConfig { + ManagedTargetConfig::new(id, addr.to_string()) +} + +fn poll_once(future: Pin<&mut F>) -> Poll { + Future::poll(future, &mut Context::from_waker(Waker::noop())) +} + +fn probes(records: &[PacketRecord]) -> Vec { + records + .iter() + .filter_map(|record| matches!(record.kind, PacketKind::Probe).then_some(record.at)) + .collect() +} + +fn has_close(records: &[PacketRecord]) -> bool { + records + .iter() + .any(|record| matches!(record.kind, PacketKind::Close)) +} + +fn echo_sends( + observations: &Mutex)>>, +) -> Vec<(TargetInstance, u32, Instant, Instant)> { + observations + .lock() + .unwrap() + .iter() + .filter_map(|(event, _)| match event { + ManagedEvent::Client { + target, + event: + ClientEvent::EchoSent { + seq, + scheduled_at, + sent_at, + .. + }, + } => Some((target.clone(), *seq, *scheduled_at, sent_at.mono)), + _ => None, + }) + .collect() +} + +fn assert_rotated_without_catchup(sends: &[(TargetInstance, u32, Instant, Instant)]) { + assert!(sends.len() >= 4); + assert_ne!(sends[0].0, sends[1].0); + for target in [&sends[0].0, &sends[1].0] { + let target_sends = sends + .iter() + .filter(|(instance, ..)| instance == target) + .collect::>(); + assert!(target_sends.len() >= 2); + assert!(target_sends[1].2 > target_sends[0].3); + } +} + +#[test] +fn construction_is_runtime_and_io_free() { + let before = resolution_call_counts(); + let (task, handle) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![ManagedTargetConfig::new("dns", "no-such-host.invalid")], + ) + .unwrap(); + assert_eq!(resolution_call_counts(), before); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::NotStarted); + drop(task); +} + +#[test] +fn quiescent_rejects_empty_initial_targets() { + assert!(matches!( + ManagedClient::task(config(ManagedPacing::Staggered), vec![]), + Err(ManagedConfigError::EmptyInitialTargets) + )); +} + +#[test] +fn explicit_empty_waits_for_stop() { + let mut config = config(ManagedPacing::Staggered); + config.completion = ManagedCompletionPolicy::ExplicitStop; + let (task, handle) = ManagedClient::task(config, vec![]).unwrap(); + let outcome = runtime().block_on(async move { + let mut task = Box::pin(task); + assert!( + tokio::time::timeout(Duration::from_millis(20), task.as_mut()) + .await + .is_err() + ); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running); + let receipt = handle.stop(); + let outcome = task.await; + receipt.await; + outcome + }); + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); +} + +#[test] +fn duplicate_and_limit_rejections_are_transactional() { + let mut limited = config(ManagedPacing::Staggered); + limited.max_live_target_generations = 1; + assert!(matches!( + ManagedClient::task( + limited, + vec![ + target("a", "127.0.0.1:1".parse().unwrap()), + target("b", "127.0.0.1:2".parse().unwrap()) + ] + ), + Err(ManagedConfigError::TooManyTargets { .. }) + )); + let duplicate = vec![ + target("same", "127.0.0.1:1".parse().unwrap()), + target("same", "127.0.0.1:2".parse().unwrap()), + ]; + assert!(matches!( + ManagedClient::task(config(ManagedPacing::Staggered), duplicate), + Err(ManagedConfigError::DuplicateTargetId { .. }) + )); +} + +#[test] +fn first_poll_without_runtime_fails_durably() { + let (task, handle) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("one", "127.0.0.1:9".parse().unwrap())], + ) + .unwrap(); + let mut task = Box::pin(task); + let Poll::Ready(outcome) = poll_once(task.as_mut()) else { + panic!("task did not fail on its first poll") + }; + assert_eq!( + outcome.end_reason, + ManagedEndReason::DriverFailed(ManagedDriverFailure::NoTokioRuntime) + ); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Failed); +} + +#[test] +fn pre_poll_stop() { + let before = resolution_call_counts(); + let (task, handle) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![ManagedTargetConfig::new("one", "no-such-host.invalid")], + ) + .unwrap(); + let mut receipt = Box::pin(handle.stop()); + let mut task = Box::pin(task); + let Poll::Ready(outcome) = poll_once(task.as_mut()) else { + panic!("pre-stopped task did not complete") + }; + assert!(poll_once(receipt.as_mut()).is_ready()); + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + assert_eq!(resolution_call_counts(), before); +} + +#[test] +fn finite_one_target_lifecycle() { + let server = start_server(ServerBehavior::Echo, None); + let (task, _) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("one", server.addr)], + ) + .unwrap(); + let outcome = runtime().block_on(task); + let records = server.finish(); + assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete); + assert!(matches!( + outcome.recent_target_outcomes[0].end_reason, + ManagedTargetEndReason::TestComplete + )); + assert!(!probes(&records).is_empty() && has_close(&records)); +} + +#[test] +fn no_test_target_completion() { + let server = start_server(ServerBehavior::NoTest, None); + let mut config = config(ManagedPacing::Staggered); + config.client.run_mode = RunMode::NoTest; + let (task, _) = ManagedClient::task(config, vec![target("no-test", server.addr)]).unwrap(); + let outcome = runtime().block_on(task); + let records = server.finish(); + assert!(matches!( + outcome.recent_target_outcomes[0].end_reason, + ManagedTargetEndReason::NoTestComplete + )); + assert!(probes(&records).is_empty() && !has_close(&records)); +} + +#[test] +fn authenticated_peer_close_outcome() { + let key = b"managed-peer-close".to_vec(); + let server = start_server(ServerBehavior::PeerClose, Some(key.clone())); + let mut configured = target("peer", server.addr); + configured.auth = Some(ClientAuthConfig { + hmac_key: Some(key), + }); + let (task, _) = + ManagedClient::task(config(ManagedPacing::Staggered), vec![configured]).unwrap(); + let outcome = runtime().block_on(task); + let records = server.finish(); + assert_eq!(outcome.peer_closed_target_outcomes, 1); + assert!(matches!( + outcome.recent_target_outcomes[0].end_reason, + ManagedTargetEndReason::PeerClosed + )); + assert!(!has_close(&records)); +} + +#[test] +fn static_multi_target_completion() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let (task, _) = ManagedClient::task( + config(ManagedPacing::Burst), + vec![target("first", first.addr), target("second", second.addr)], + ) + .unwrap(); + let outcome = runtime().block_on(task); + assert_eq!(outcome.total_target_outcomes, 2); + assert_eq!(outcome.successful_target_outcomes, 2); + assert!(!probes(&first.finish()).is_empty()); + assert!(!probes(&second.finish()).is_empty()); +} + +#[test] +fn sibling_failure_isolation() { + let good = start_server(ServerBehavior::Echo, None); + let unused = UdpSocket::bind("127.0.0.1:0").unwrap(); + let bad_addr = unused.local_addr().unwrap(); + drop(unused); + let (task, _) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("good", good.addr), target("bad", bad_addr)], + ) + .unwrap(); + let outcome = runtime().block_on(task); + assert_eq!(outcome.total_target_outcomes, 2); + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(outcome.failed_target_outcomes, 1); + assert!(!probes(&good.finish()).is_empty()); +} + +#[test] +fn burst_pacing_fairness() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let (mut task, _) = ManagedClient::task( + config(ManagedPacing::Burst), + vec![target("first", first.addr), target("second", second.addr)], + ) + .unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + runtime().block_on(task); + assert_rotated_without_catchup(&echo_sends(&observations)); + first.finish(); + second.finish(); +} + +#[test] +fn staggered_pacing_fairness() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let (mut task, _) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("first", first.addr), target("second", second.addr)], + ) + .unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + runtime().block_on(task); + assert_rotated_without_catchup(&echo_sends(&observations)); + first.finish(); + second.finish(); +} + +#[test] +fn committed_probe_graceful_drain() { + let server = start_server(ServerBehavior::DelayedEcho(Duration::from_millis(25)), None); + let mut config = config(ManagedPacing::Staggered); + config.completion = ManagedCompletionPolicy::ExplicitStop; + config.client.duration = None; + config.client.interval = Duration::from_millis(100); + config.client.probe_timeout = Duration::from_millis(80); + config.final_drain = Duration::from_millis(5); + let (task, handle) = ManagedClient::task(config, vec![target("one", server.addr)]).unwrap(); + let stop_handle = handle.clone(); + let seen = Arc::clone(&server.probe_seen); + let stopper = thread::spawn(move || { + let (flag, ready) = &*seen; + let guard = flag.lock().unwrap(); + let (guard, timeout) = ready + .wait_timeout_while(guard, Duration::from_secs(2), |seen| !*seen) + .unwrap(); + let probe_seen = *guard; + drop(guard); + drop(stop_handle.stop()); + assert!(probe_seen && !timeout.timed_out()); + }); + let outcome = runtime().block_on(task); + stopper.join().unwrap(); + let records = server.finish(); + let target = &outcome.recent_target_outcomes[0]; + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + assert!(matches!(target.end_reason, ManagedTargetEndReason::Stopped)); + assert_eq!(target.packets_sent, 1); + assert_eq!(target.replies_received, 1); + assert_eq!(probes(&records).len(), 1); +} + +#[test] +fn event_loss_independence() { + let server = start_server(ServerBehavior::Echo, None); + let mut config = config(ManagedPacing::Staggered); + config.event_capacity = 1; + let (task, handle) = ManagedClient::task(config, vec![target("one", server.addr)]).unwrap(); + let mut events = handle.subscribe().unwrap(); + let outcome = runtime().block_on(task); + assert!(matches!( + events.try_recv(), + Err(broadcast::error::TryRecvError::Lagged(_)) + )); + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Completed); + server.finish(); +} + +#[test] +fn outcome_history_eviction_preserves_aggregates() { + let servers = [ + start_server(ServerBehavior::NoTest, None), + start_server(ServerBehavior::NoTest, None), + start_server(ServerBehavior::NoTest, None), + ]; + let mut config = config(ManagedPacing::Staggered); + config.client.run_mode = RunMode::NoTest; + config.outcome_history_limit = 1; + let targets = servers + .iter() + .enumerate() + .map(|(index, server)| target(&format!("target-{index}"), server.addr)) + .collect(); + let (mut task, _) = ManagedClient::task(config, targets).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let outcome = runtime().block_on(task); + assert_eq!(outcome.total_target_outcomes, 3); + assert_eq!(outcome.successful_target_outcomes, 3); + assert_eq!(outcome.discarded_target_outcomes, 2); + assert_eq!(outcome.recent_target_outcomes.len(), 1); + let last_finished = observations + .lock() + .unwrap() + .iter() + .filter_map(|(event, _)| match event { + ManagedEvent::TargetFinished { outcome } => Some(outcome.target.clone()), + _ => None, + }) + .next_back() + .unwrap(); + assert_eq!(outcome.recent_target_outcomes[0].target, last_finished); + for server in servers { + server.finish(); + } +} + +#[test] +fn status_precedes_corresponding_events() { + let server = start_server(ServerBehavior::NoTest, None); + let mut config = config(ManagedPacing::Staggered); + config.client.run_mode = RunMode::NoTest; + let (mut task, _) = ManagedClient::task(config, vec![target("one", server.addr)]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + runtime().block_on(task); + let mut saw_started = false; + let mut saw_state = false; + let mut saw_finished = false; + let mut saw_stopping = false; + let mut saw_completed = false; + for (event, status) in observations.lock().unwrap().iter() { + match event { + ManagedEvent::Started => { + saw_started = true; + assert_eq!(status.lifecycle, ManagedLifecycle::Running); + } + ManagedEvent::TargetStateChanged { target, lifecycle } => { + saw_state = true; + assert!(status + .targets + .iter() + .any(|entry| entry.target == *target && entry.lifecycle == *lifecycle)); + } + ManagedEvent::TargetFinished { .. } => { + saw_finished = true; + assert!(status.total_target_outcomes >= 1); + } + ManagedEvent::Stopping => { + saw_stopping = true; + assert_eq!(status.lifecycle, ManagedLifecycle::Stopping); + } + ManagedEvent::Completed { .. } => { + saw_completed = true; + assert_eq!(status.lifecycle, ManagedLifecycle::Completed); + } + _ => {} + } + } + assert!(saw_started && saw_state && saw_finished && saw_stopping && saw_completed); + server.finish(); +} + +#[test] +fn dropping_all_handles_does_not_stop() { + let server = start_server(ServerBehavior::Echo, None); + let (task, handle) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("one", server.addr)], + ) + .unwrap(); + drop(handle); + let outcome = runtime().block_on(task); + assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete); + server.finish(); +} + +#[test] +fn task_drop_abandonment() { + let (task, handle) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("one", "127.0.0.1:9".parse().unwrap())], + ) + .unwrap(); + let mut events = handle.subscribe().unwrap(); + drop(task); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Abandoned); + assert_eq!(handle.status().total_target_outcomes, 0); + assert!(matches!(events.try_recv(), Ok(ManagedEvent::Abandoned))); +} + +#[test] +fn terminal_subscription_is_closed() { + let (task, handle) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("one", "127.0.0.1:9".parse().unwrap())], + ) + .unwrap(); + let mut task = Box::pin(task); + assert!(poll_once(task.as_mut()).is_ready()); + assert!(matches!( + handle.subscribe(), + Err(ManagedSubscribeError::Closed) + )); +}