test(managed): verify unified task lifecycle

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<Mutex<Vec<PacketRecord>>>,
+    probe_seen: Arc<(Mutex<bool>, Condvar)>,
+    thread: JoinHandle<()>,
+}
+
+impl TestServer {
+    fn finish(self) -> Vec<PacketRecord> {
+        self.thread.join().unwrap();
+        self.records.lock().unwrap().clone()
+    }
+}
+
+fn start_server(behavior: ServerBehavior, key: Option<Vec<u8>>) -> 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<F: Future>(future: Pin<&mut F>) -> Poll<F::Output> {
+    Future::poll(future, &mut Context::from_waker(Waker::noop()))
+}
+
+fn probes(records: &[PacketRecord]) -> Vec<Instant> {
+    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<(ManagedEvent, Arc<ManagedStatus>)>>,
+) -> 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::<Vec<_>>();
+        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)
+    ));
+}