diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 63b4845..a49468e 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -138,6 +138,73 @@ pub struct ManagedClientHandle { max_live_target_generations: usize, status: watch::Receiver>, events: broadcast::WeakSender, + #[cfg(test)] + update_submission_hook: Option>, +} + +#[cfg(test)] +#[derive(Default)] +struct UpdateSubmissionTestHook { + state: std::sync::Mutex, + arrived: std::sync::Condvar, + released: std::sync::Condvar, +} + +#[cfg(test)] +#[derive(Default)] +struct UpdateSubmissionTestState { + armed: bool, + arrived: bool, + released: bool, +} + +#[cfg(test)] +impl UpdateSubmissionTestHook { + fn arm(&self) { + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .armed = true; + } + + fn pause_after_admission(&self) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !state.armed { + return; + } + state.arrived = true; + self.arrived.notify_all(); + while !state.released { + state = self + .released + .wait(state) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + } + + fn wait_until_arrived(&self) { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let (state, timeout) = self + .arrived + .wait_timeout_while(state, Duration::from_secs(2), |state| !state.arrived) + .unwrap_or_else(std::sync::PoisonError::into_inner); + assert!(state.arrived && !timeout.timed_out()); + } + + fn release(&self) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.released = true; + self.released.notify_all(); + } } impl fmt::Debug for ManagedClientHandle { @@ -185,6 +252,11 @@ impl ManagedClientHandle { }); } + #[cfg(test)] + if let Some(hook) = &self.update_submission_hook { + hook.pause_after_admission(); + } + // Submission linearizes at the successful `try_send` below. The // driver checks stop immediately before non-suspending application; // terminal sealing closes the receiver before draining accepted work. @@ -2076,6 +2148,8 @@ fn build_task( max_live_target_generations, status: status_receiver, events: weak_events, + #[cfg(test)] + update_submission_hook: None, }; Ok((task, handle)) } diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index f8c97b1..d0dbf15 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -27,6 +27,7 @@ enum ServerBehavior { PeerClose, DelayedEcho(Duration), DeferredBurst(usize), + NoisyOpen(usize), } #[derive(Clone, Debug)] @@ -47,6 +48,7 @@ struct TestServer { records: Arc>>, probe_seen: Arc<(Mutex, Condvar)>, reply_sent: Arc<(Mutex, Condvar)>, + open_reply_sent: Arc<(Mutex, Condvar)>, thread: JoinHandle<()>, } @@ -110,6 +112,8 @@ fn start_server_with_gates( let thread_probe_seen = Arc::clone(&probe_seen); let reply_sent = Arc::new((Mutex::new(false), Condvar::new())); let thread_reply_sent = Arc::clone(&reply_sent); + let open_reply_sent = Arc::new((Mutex::new(false), Condvar::new())); + let thread_open_reply_sent = Arc::clone(&open_reply_sent); let thread = thread::spawn(move || { let mut buffer = [0_u8; 8192]; let (open_len, peer) = socket.recv_from(&mut buffer).unwrap(); @@ -137,8 +141,17 @@ fn start_server_with_gates( key.as_deref(), ) .unwrap(); + let noisy_open = matches!(behavior, ServerBehavior::NoisyOpen(_)); + if let ServerBehavior::NoisyOpen(count) = behavior { + for _ in 0..count { + socket.send_to(&[0_u8], peer).unwrap(); + } + } socket.send_to(&open_reply, peer).unwrap(); - if no_test { + let (sent, ready) = &*thread_open_reply_sent; + *sent.lock().unwrap() = true; + ready.notify_all(); + if no_test || noisy_open { return; } @@ -227,6 +240,7 @@ fn start_server_with_gates( records, probe_seen, reply_sent, + open_reply_sent, thread, } } @@ -805,6 +819,289 @@ fn dynamic_empty_update_orders_stopping_before_acknowledgement() { assert!(Arc::ptr_eq(&acknowledgement.status, &events[stopping].1)); } +#[test] +fn noisy_opening_yields_before_stop_observation() { + let open_gate = Arc::new(PacketGate::default()); + let server = start_server_with_gates( + ServerBehavior::NoisyOpen(256), + None, + None, + Some(Arc::clone(&open_gate)), + None, + ); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (task, handle) = ManagedClient::task(managed, vec![target("noisy", server.addr)]).unwrap(); + let runtime = runtime(); + let mut task = Box::pin(task); + + drive_task_until(&runtime, &mut task, |task| { + matches!(task.targets[0].state, TargetState::Opening { .. }) + }); + drive_task_until(&runtime, &mut task, |task| { + matches!( + task.targets[0].state, + TargetState::Opening { ref open, .. } if open.has_in_flight_work() + ) + }); + let seen = open_gate.seen.lock().unwrap(); + let (seen, timeout) = open_gate + .seen_ready + .wait_timeout_while(seen, Duration::from_secs(2), |seen| !*seen) + .unwrap(); + assert!(*seen && !timeout.timed_out()); + drop(seen); + open_gate.release(); + wait_flag(&server.open_reply_sent); + + // One managed task poll must leave a valid open reply behind the noisy + // queue, rather than letting this target consume all immediate work. + let poll = runtime.block_on(poll_fn(|cx| Poll::Ready(task.as_mut().poll(cx)))); + assert!(poll.is_pending()); + assert!(matches!(task.targets[0].state, TargetState::Opening { .. })); + + let receipt = handle.stop(); + let poll = runtime.block_on(poll_fn(|cx| Poll::Ready(task.as_mut().poll(cx)))); + assert!(poll.is_pending()); + assert!(handle.status().stop_requested); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Stopping); + let outcome = runtime.block_on(task); + runtime.block_on(receipt); + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + server.finish(); +} + +#[test] +fn stop_during_quiescent_stopping_is_durable_without_duplicate_event() { + let server = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Staggered); + managed.client.duration = None; + managed.final_drain = Duration::from_millis(80); + let (mut task, handle) = + ManagedClient::task(managed, vec![target("old", server.addr)]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + let mut task = Box::pin(task); + drive_task_until(&runtime, &mut task, |task| task.active_count() == 1); + + let removal = handle.update_targets(vec![]).unwrap(); + drive_task_until(&runtime, &mut task, |task| { + task.state == DriverState::Stopping + && !task.stop_observed + && task.targets.iter().any(|target| !target.desired) + }); + let acknowledgement = runtime.block_on(removal).unwrap(); + assert_eq!(acknowledgement.status.lifecycle, ManagedLifecycle::Stopping); + + let stop = handle.stop(); + let outcome = runtime.block_on(task); + runtime.block_on(stop); + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + assert!(handle.status().stop_requested); + assert!(outcome + .recent_target_outcomes + .iter() + .any(|outcome| { matches!(outcome.end_reason, ManagedTargetEndReason::Removed) })); + assert_eq!( + observations + .lock() + .unwrap() + .iter() + .filter(|(event, _)| matches!(event, ManagedEvent::Stopping)) + .count(), + 1 + ); + server.finish(); +} + +#[test] +fn stop_before_enqueue_rejects_without_allocating_a_generation() { + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + let (task, handle) = ManagedClient::task(managed, vec![]).unwrap(); + let initial = handle.status(); + let stop = handle.stop(); + assert!(matches!( + handle.update_targets(vec![ManagedTargetConfig::new("later", "127.0.0.1:9")]), + Err(ManagedCommandError::Stopping) + )); + assert_eq!(task.next_generation, 1); + assert_eq!(task.applied_command_sequence, 0); + assert_eq!(handle.status(), initial); + let outcome = runtime().block_on(task); + runtime().block_on(stop); + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); +} + +#[test] +fn paused_submission_cannot_outlive_terminal_seal() { + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + let (task, mut handle) = ManagedClient::task(managed, vec![]).unwrap(); + let hook = Arc::new(UpdateSubmissionTestHook::default()); + hook.arm(); + handle.update_submission_hook = Some(Arc::clone(&hook)); + let submitter = handle.clone(); + let sender = thread::spawn(move || { + submitter.update_targets(vec![ManagedTargetConfig::new("late", "127.0.0.1:9")]) + }); + hook.wait_until_arrived(); + + let stop = handle.stop(); + let runtime = runtime(); + let mut task = Box::pin(task); + let Poll::Ready(outcome) = runtime.block_on(poll_fn(|cx| Poll::Ready(task.as_mut().poll(cx)))) + else { + panic!("empty stopped task did not seal") + }; + runtime.block_on(stop); + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + + hook.release(); + assert!(matches!( + sender.join().unwrap(), + Err(ManagedCommandError::DriverClosed) + )); + drop(task); +} + +#[test] +fn accepted_update_before_stop_resolves_stopping() { + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + let (task, handle) = ManagedClient::task(managed, vec![]).unwrap(); + let receipt = handle + .update_targets(vec![ManagedTargetConfig::new("queued", "127.0.0.1:9")]) + .unwrap(); + let stop = handle.stop(); + let outcome = runtime().block_on(task); + runtime().block_on(stop); + assert_eq!(outcome.applied_command_sequence, 0); + assert!(matches!( + runtime().block_on(receipt), + Err(ManagedCommandApplyError::Stopping) + )); +} + +#[test] +fn accepted_update_before_driver_failure_resolves_driver_failed() { + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + let (task, handle) = ManagedClient::task(managed, vec![]).unwrap(); + let receipt = handle + .update_targets(vec![ManagedTargetConfig::new("queued", "127.0.0.1:9")]) + .unwrap(); + let mut task = Box::pin(task); + let Poll::Ready(outcome) = poll_once(task.as_mut()) else { + panic!("first poll outside Tokio did not fail") + }; + assert_eq!( + outcome.end_reason, + ManagedEndReason::DriverFailed(ManagedDriverFailure::NoTokioRuntime) + ); + let mut receipt = Box::pin(receipt); + assert!(matches!( + poll_once(Pin::as_mut(&mut receipt)), + Poll::Ready(Err(ManagedCommandApplyError::DriverFailed( + ManagedDriverFailure::NoTokioRuntime + ))) + )); + assert!(matches!( + handle.update_targets(vec![]), + Err(ManagedCommandError::DriverClosed) + )); +} + +#[test] +fn extreme_event_capacity_is_rejected_without_panic() { + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.event_capacity = usize::MAX; + assert!(matches!( + ManagedClient::task(managed, vec![]), + Err(ManagedConfigError::EventCapacityTooLarge { + configured: usize::MAX, + .. + }) + )); +} + +#[test] +fn extreme_command_capacity_is_rejected_without_panic() { + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.command_capacity = usize::MAX; + assert!(matches!( + ManagedClient::task(managed, vec![]), + Err(ManagedConfigError::CommandCapacityTooLarge { + configured: usize::MAX, + .. + }) + )); +} + +#[test] +fn oversized_desired_set_is_rejected_before_enqueue() { + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.max_live_target_generations = 1; + let (task, handle) = ManagedClient::task(managed, vec![]).unwrap(); + let initial = handle.status(); + let capacity = handle.commands.capacity(); + assert!(matches!( + handle.update_targets(vec![ + ManagedTargetConfig::new("one", "127.0.0.1:9"), + ManagedTargetConfig::new("two", "127.0.0.1:10"), + ]), + Err(ManagedCommandError::TooManyTargets { + configured: 2, + limit: 1 + }) + )); + assert_eq!(handle.commands.capacity(), capacity); + assert_eq!(task.next_generation, 1); + assert_eq!(task.applied_command_sequence, 0); + assert_eq!(handle.status(), initial); +} + +#[test] +fn legal_desired_set_can_still_fail_at_overlap_application() { + let server = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = 1; + let (task, handle) = ManagedClient::task(managed, vec![target("same", server.addr)]).unwrap(); + let runtime = runtime(); + let mut task = Box::pin(task); + drive_task_until(&runtime, &mut task, |task| task.active_count() == 1); + + let receipt = handle + .update_targets(vec![ManagedTargetConfig::new("same", "127.0.0.1:10")]) + .unwrap(); + let mut receipt = Box::pin(receipt); + let result = runtime.block_on(poll_fn(|cx| { + assert!(task.as_mut().poll(cx).is_pending()); + match receipt.as_mut().poll(cx) { + Poll::Ready(result) => Poll::Ready(result), + Poll::Pending => Poll::Pending, + } + })); + assert!(matches!( + result, + Err(ManagedCommandApplyError::LiveGenerationLimitExceeded { + required: 2, + limit: 1 + }) + )); + let stop = handle.stop(); + runtime.block_on(task); + runtime.block_on(stop); + server.finish(); +} + #[test] fn dynamic_explicit_remove_and_readd_uses_new_generation() { let first = start_server(ServerBehavior::Echo, None);