From 1c6b4be9159134b13b839b150666799db568c97d Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 9 Aug 2026 23:45:19 +0700 Subject: [PATCH 1/4] fix(client): bound asynchronous opening work --- crates/irtt-client/src/async_client.rs | 93 +++++++++++--------- crates/irtt-client/src/async_client/tests.rs | 50 +++++++++++ 2 files changed, 102 insertions(+), 41 deletions(-) diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index 8523746..ecee6c3 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -34,6 +34,13 @@ use crate::{ timing::ClientTimestamp, }; +/// Maximum amount of immediately available opening work performed by one poll. +/// +/// Opening may consume ignored UDP datagrams, roll over an expired attempt, or +/// encounter readiness false positives without ever awaiting. Bound that work +/// so one managed target cannot retain the current-thread runtime. +const OPEN_POLL_WORK_BUDGET: usize = 64; + #[cfg(test)] use crate::client::ProbeSendTimestamps; @@ -315,7 +322,7 @@ impl AsyncClient { unreachable!("connected async clients retain an open request"); } - loop { + for _ in 0..OPEN_POLL_WORK_BUDGET { if state.cleanup.is_some() { return self.poll_open_cleanup(state, cx); } @@ -430,6 +437,11 @@ impl AsyncClient { }; return Poll::Ready(Ok(self.commit_async_open(prepared))); } + + // We deliberately truncated known immediate opening work. Arrange for + // another poll even if no new socket-readiness edge occurs. + cx.waker().wake_by_ref(); + Poll::Pending } pub(crate) fn poll_send_probe( @@ -855,49 +867,48 @@ impl AsyncClient { state: &mut AsyncOpenState, cx: &mut Context<'_>, ) -> Poll> { - loop { - let Some(packet) = state - .cleanup - .as_ref() - .expect("cleanup polling requires a retained primary error") - .packet() - else { - return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())); - }; + let Some(packet) = state + .cleanup + .as_ref() + .expect("cleanup polling requires a retained primary error") + .packet() + else { + return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())); + }; - #[cfg(test)] - if self - .test_hooks - .pause_open_cleanup_before_send - .replace(false) - { - return Poll::Pending; - } + #[cfg(test)] + if self + .test_hooks + .pause_open_cleanup_before_send + .replace(false) + { + return Poll::Pending; + } - #[cfg(test)] - if self.test_hooks.fail_cleanup_send.replace(false) { - return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())); - } - #[cfg(test)] - let send_result = self.test_hooks.try_send(&self.socket, packet); - #[cfg(not(test))] - let send_result = self.socket.try_send(packet); - match send_result { - Ok(_) => { - return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())); - } - Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} - Err(_) => { - return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())); - } - } + #[cfg(test)] + if self.test_hooks.fail_cleanup_send.replace(false) { + return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())); + } + #[cfg(test)] + let send_result = self.test_hooks.try_send(&self.socket, packet); + #[cfg(not(test))] + let send_result = self.socket.try_send(packet); + match send_result { + Ok(_) => return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())), + Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} + Err(_) => return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())), + } - match poll_writable_until(&self.socket, state.deadline_timer(), cx) { - Poll::Pending => return Poll::Pending, - Poll::Ready(Ok(true)) => {} - Poll::Ready(Ok(false) | Err(_)) => { - return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())); - } + match poll_writable_until(&self.socket, state.deadline_timer(), cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Ok(true)) => { + // Retain cleanup state and resume the send in a fresh poll. A + // writable readiness false positive must not spin this future. + cx.waker().wake_by_ref(); + Poll::Pending + } + Poll::Ready(Ok(false) | Err(_)) => { + Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())) } } } diff --git a/crates/irtt-client/src/async_client/tests.rs b/crates/irtt-client/src/async_client/tests.rs index ec4df19..6447c5a 100644 --- a/crates/irtt-client/src/async_client/tests.rs +++ b/crates/irtt-client/src/async_client/tests.rs @@ -334,6 +334,56 @@ fn ignored_open_traffic_uses_one_request_and_deadline_per_attempt() { assert_eq!(server.finish().len(), 2); } +#[test] +fn opening_poll_budget_yields_then_public_open_completes() { + let (queued, queue_ready) = mpsc::channel(); + let server = start_server(move |socket, tx| { + let (open_packet, peer) = recv_packet(&socket, &tx); + let request = decode_open_request(&open_packet, None).unwrap(); + for _ in 0..=OPEN_POLL_WORK_BUDGET { + socket.send_to(&[0_u8], peer).unwrap(); + } + send_open_reply( + &socket, + peer, + request.params, + None, + flags::FLAG_OPEN | flags::FLAG_REPLY, + TOKEN, + ); + queued.send(()).unwrap(); + }); + + runtime().block_on(async { + let mut client = AsyncClient::connect(config(server.addr, None, 0)) + .await + .unwrap(); + client.socket.writable().await.unwrap(); + client.test_hooks.pause_open_before_readable.set(true); + let mut opening = Box::pin(client.open()); + + // Submit the one opening request, then let the server queue all noise + // before performing the explicit bounded poll below. + assert!(matches!( + poll_once(Pin::as_mut(&mut opening)), + Poll::Pending + )); + queue_ready.recv_timeout(Duration::from_secs(2)).unwrap(); + + // The valid reply is behind more ignored datagrams than one opening + // poll may consume, so this cannot complete without the budget yield. + assert!(matches!( + poll_once(Pin::as_mut(&mut opening)), + Poll::Pending + )); + assert!(matches!( + opening.await.unwrap(), + OpenOutcome::Started { .. } + )); + }); + assert_eq!(server.finish().len(), 1); +} + #[test] fn ignored_open_traffic_cannot_extend_the_absolute_attempt_deadline() { let server = start_server(move |socket, tx| { -- 2.54.0 From dd1182d24d4b6a1793cfff5caff132da1fdc836b Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 9 Aug 2026 23:50:09 +0700 Subject: [PATCH 2/4] fix(managed): harden control admission --- crates/irtt-client/src/managed/model.rs | 6 ++ crates/irtt-client/src/managed/task.rs | 135 +++++++++++++++++++++--- 2 files changed, 124 insertions(+), 17 deletions(-) diff --git a/crates/irtt-client/src/managed/model.rs b/crates/irtt-client/src/managed/model.rs index f972c2d..fac057f 100644 --- a/crates/irtt-client/src/managed/model.rs +++ b/crates/irtt-client/src/managed/model.rs @@ -268,8 +268,12 @@ pub enum ManagedConfigError { TooManyTargets { configured: usize, limit: usize }, #[error("managed event capacity must be greater than zero")] ZeroEventCapacity, + #[error("managed event capacity {configured} exceeds Tokio's maximum {maximum}")] + EventCapacityTooLarge { configured: usize, maximum: usize }, #[error("managed command capacity must be greater than zero")] ZeroCommandCapacity, + #[error("managed command capacity {configured} exceeds Tokio's maximum {maximum}")] + CommandCapacityTooLarge { configured: usize, maximum: usize }, #[error("managed live-generation limit must be greater than zero")] ZeroLiveGenerationLimit, #[error("managed final drain {duration:?} cannot be scheduled from the current instant")] @@ -289,6 +293,8 @@ pub enum ManagedConfigError { pub enum ManagedCommandError { #[error("managed target updates are no longer accepted")] Stopping, + #[error("managed target update contains {configured} targets but the limit is {limit}")] + TooManyTargets { configured: usize, limit: usize }, #[error("managed command queue is full")] QueueFull, #[error("managed command receiver is closed")] diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index c68cb40..63b4845 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -5,7 +5,7 @@ use std::{ mem, pin::Pin, sync::{ - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicU8, Ordering}, Arc, }, task::{Context, Poll}, @@ -38,6 +38,9 @@ use super::{ const TARGET_WORK_BUDGET: usize = 128; const COMMAND_WORK_BUDGET: usize = 32; const POST_DEADLINE_RECEIVE_BUDGET: usize = TARGET_WORK_BUDGET; +// Tokio 1.53 broadcast asserts this exact bound before rounding capacity to a +// power of two; Tokio does not expose it as a public constant. +const MAX_BROADCAST_CHANNEL_CAPACITY: usize = usize::MAX >> 1; /// Entry point for constructing a unified Tokio managed task. #[derive(Debug, Default)] @@ -56,25 +59,64 @@ impl ManagedClient { #[derive(Debug)] struct StopSignal { requested: AtomicBool, - updates_accepted: AtomicBool, + update_admission: AtomicU8, wake: watch::Sender<()>, acknowledged: AtomicBool, acknowledgement: watch::Sender<()>, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[repr(u8)] +enum UpdateAdmission { + Open, + Stopping, + Closed, +} + +impl UpdateAdmission { + fn from_raw(value: u8) -> Self { + match value { + value if value == Self::Open as u8 => Self::Open, + value if value == Self::Stopping as u8 => Self::Stopping, + value if value == Self::Closed as u8 => Self::Closed, + _ => unreachable!("managed update admission stores a known state"), + } + } +} + impl StopSignal { fn request(&self) { - self.updates_accepted.store(false, Ordering::Release); + self.begin_stopping(); if !self.requested.swap(true, Ordering::AcqRel) { self.wake.send_replace(()); } } - fn updates_accepted(&self) -> bool { - self.updates_accepted.load(Ordering::Acquire) + fn update_admission(&self) -> UpdateAdmission { + UpdateAdmission::from_raw(self.update_admission.load(Ordering::Acquire)) } + + fn begin_stopping(&self) { + let mut current = self.update_admission.load(Ordering::Acquire); + loop { + match UpdateAdmission::from_raw(current) { + UpdateAdmission::Open => match self.update_admission.compare_exchange_weak( + UpdateAdmission::Open as u8, + UpdateAdmission::Stopping as u8, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => return, + Err(observed) => current = observed, + }, + UpdateAdmission::Stopping | UpdateAdmission::Closed => return, + } + } + } + fn close_updates(&self) { - self.updates_accepted.store(false, Ordering::Release); + self.update_admission + .store(UpdateAdmission::Closed as u8, Ordering::Release); } fn is_requested(&self) -> bool { @@ -93,6 +135,7 @@ impl StopSignal { pub struct ManagedClientHandle { stop: Arc, commands: mpsc::Sender, + max_live_target_generations: usize, status: watch::Receiver>, events: broadcast::WeakSender, } @@ -130,16 +173,32 @@ impl ManagedClientHandle { &self, targets: Vec, ) -> Result { - if !self.stop.updates_accepted() { - return Err(ManagedCommandError::Stopping); + match self.stop.update_admission() { + UpdateAdmission::Open => {} + UpdateAdmission::Stopping => return Err(ManagedCommandError::Stopping), + UpdateAdmission::Closed => return Err(ManagedCommandError::DriverClosed), } + if targets.len() > self.max_live_target_generations { + return Err(ManagedCommandError::TooManyTargets { + configured: targets.len(), + limit: self.max_live_target_generations, + }); + } + + // 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. let (acknowledgement, receiver) = oneshot::channel(); match self.commands.try_send(ManagedCommand::UpdateTargets { targets, acknowledgement, }) { Ok(()) => Ok(ManagedCommandReceipt { receiver }), - Err(mpsc::error::TrySendError::Full(_)) => Err(ManagedCommandError::QueueFull), + Err(mpsc::error::TrySendError::Full(_)) => match self.stop.update_admission() { + UpdateAdmission::Open => Err(ManagedCommandError::QueueFull), + UpdateAdmission::Stopping => Err(ManagedCommandError::Stopping), + UpdateAdmission::Closed => Err(ManagedCommandError::DriverClosed), + }, Err(mpsc::error::TrySendError::Closed(_)) => Err(ManagedCommandError::DriverClosed), } } @@ -657,11 +716,15 @@ impl ManagedClientTask { return; } self.stop_observed = true; - self.resources().stop.close_updates(); + self.resources().stop.begin_stopping(); + let lifecycle_transition = + self.state != DriverState::Stopping || self.lifecycle != ManagedLifecycle::Stopping; self.state = DriverState::Stopping; self.lifecycle = ManagedLifecycle::Stopping; self.replace_status(); - self.publish_event(ManagedEvent::Stopping); + if lifecycle_transition { + self.publish_event(ManagedEvent::Stopping); + } self.resources().stop.acknowledge(); self.scan_remaining = self.targets.len(); self.burst_remaining = 0; @@ -683,7 +746,7 @@ impl ManagedClientTask { // `apply_targets` cannot suspend, so a stop observed here wins over // this command; once it passes, this command may complete atomically. let result = if self.state != DriverState::Running - || self.resources().stop.is_requested() + || self.resources().stop.update_admission() != UpdateAdmission::Open { Err(ManagedCommandApplyError::Stopping) } else { @@ -767,7 +830,7 @@ impl ManagedClientTask { let stopping = self.targets.iter().all(|runtime| !runtime.desired) && self.config.completion == ManagedCompletionPolicy::FinishWhenQuiescent; if stopping { - self.resources().stop.close_updates(); + self.resources().stop.begin_stopping(); self.state = DriverState::Stopping; self.lifecycle = ManagedLifecycle::Stopping; } @@ -1677,7 +1740,22 @@ impl ManagedClientTask { } } - fn seal(&mut self, end_reason: ManagedEndReason, failed: bool) -> Poll { + fn seal(&mut self, mut end_reason: ManagedEndReason, failed: bool) -> Poll { + // Sealing is the final stop-observation point. A stop that reaches the + // latch before this terminal linearization becomes durable even when + // quiescence had already entered `Stopping`. + if self.resources().stop.is_requested() { + self.observe_stop(); + } + if self.stop_observed && matches!(end_reason, ManagedEndReason::TargetsComplete) { + end_reason = ManagedEndReason::StopRequested; + } + + // Closing admission before closing and draining the receiver makes every + // accepted command task-owned. Tokio still permits buffered commands + // to be drained after `Receiver::close`, but rejects later sends. + self.resources().stop.close_updates(); + self.commands.close(); while let Ok(ManagedCommand::UpdateTargets { acknowledgement, .. }) = self.commands.try_recv() @@ -1759,11 +1837,14 @@ impl Future for ManagedClientTask { return this.fail_driver(failure); } } - if this.state == DriverState::Running && this.resources().stop.is_requested() { + if this.resources().stop.is_requested() { this.observe_stop(); } let mut immediate = this.process_commands(cx); + if this.resources().stop.is_requested() { + this.observe_stop(); + } let now = Instant::now(); immediate |= this.poll_target_pass(cx, now); this.prune_undesired_terminal(); @@ -1774,6 +1855,10 @@ impl Future for ManagedClientTask { }; } + if this.resources().stop.is_requested() { + this.observe_stop(); + } + if this.all_targets_terminal() { match this.state { DriverState::Stopping => { @@ -1791,7 +1876,7 @@ impl Future for ManagedClientTask { { this.state = DriverState::Stopping; this.lifecycle = ManagedLifecycle::Stopping; - this.resources().stop.close_updates(); + this.resources().stop.begin_stopping(); this.replace_status(); this.publish_event(ManagedEvent::Stopping); return this.seal(ManagedEndReason::TargetsComplete, false); @@ -1816,6 +1901,8 @@ impl Drop for ManagedClientTask { } self.lifecycle = ManagedLifecycle::Abandoned; self.final_outcome = None; + self.resources().stop.close_updates(); + self.commands.close(); self.replace_status(); self.publish_event(ManagedEvent::Abandoned); self.resources().stop.acknowledge(); @@ -1833,9 +1920,21 @@ fn build_task( if config.event_capacity == 0 { return Err(ManagedConfigError::ZeroEventCapacity); } + if config.event_capacity > MAX_BROADCAST_CHANNEL_CAPACITY { + return Err(ManagedConfigError::EventCapacityTooLarge { + configured: config.event_capacity, + maximum: MAX_BROADCAST_CHANNEL_CAPACITY, + }); + } if config.command_capacity == 0 { return Err(ManagedConfigError::ZeroCommandCapacity); } + if config.command_capacity > tokio::sync::Semaphore::MAX_PERMITS { + return Err(ManagedConfigError::CommandCapacityTooLarge { + configured: config.command_capacity, + maximum: tokio::sync::Semaphore::MAX_PERMITS, + }); + } if config.max_live_target_generations == 0 { return Err(ManagedConfigError::ZeroLiveGenerationLimit); } @@ -1899,7 +1998,7 @@ fn build_task( let (acknowledgement_sender, _) = watch::channel(()); let stop = Arc::new(StopSignal { requested: AtomicBool::new(false), - updates_accepted: AtomicBool::new(true), + update_admission: AtomicU8::new(UpdateAdmission::Open as u8), wake: wake_sender, acknowledged: AtomicBool::new(false), acknowledgement: acknowledgement_sender, @@ -1937,6 +2036,7 @@ fn build_task( }); let (status_sender, status_receiver) = watch::channel(initial); let (command_sender, command_receiver) = mpsc::channel(config.command_capacity); + let max_live_target_generations = config.max_live_target_generations; let resources = TaskResources { status: status_sender, events: Some(event_sender), @@ -1973,6 +2073,7 @@ fn build_task( let handle = ManagedClientHandle { stop, commands: command_sender, + max_live_target_generations, status: status_receiver, events: weak_events, }; -- 2.54.0 From dd396a8b5a7b646643660bceb01ad5ee9d0614a7 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 9 Aug 2026 23:57:24 +0700 Subject: [PATCH 3/4] test(managed): verify bounded control progress --- crates/irtt-client/src/managed/task.rs | 74 ++++++ crates/irtt-client/src/managed/tests.rs | 299 +++++++++++++++++++++++- 2 files changed, 372 insertions(+), 1 deletion(-) 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); -- 2.54.0 From 8d47dc9c6e6f0c0e5056a0d777edb89cebf6f5de Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 00:27:18 +0700 Subject: [PATCH 4/4] test(managed): cover stopping admission overlap --- crates/irtt-client/src/managed/task.rs | 10 ++- crates/irtt-client/src/managed/tests.rs | 85 +++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 3 deletions(-) diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index a49468e..88a2b95 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -257,9 +257,13 @@ impl ManagedClientHandle { 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. + // Observing `Open` only permits attempting submission. A successful + // `try_send` transfers command ownership to the driver. A concurrent + // stop may move admission to `Stopping` before this send, allowing the + // command to enqueue; the driver-side check immediately before + // `apply_targets` determines whether it applies or resolves as + // `Stopping`. Terminal sealing closes the receiver before draining + // accepted commands. let (acknowledgement, receiver) = oneshot::channel(); match self.commands.try_send(ManagedCommand::UpdateTargets { targets, diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index d0dbf15..f767145 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -967,6 +967,91 @@ fn paused_submission_cannot_outlive_terminal_seal() { drop(task); } +#[test] +fn paused_submission_can_enqueue_during_stopping() { + let reply_gate = Arc::new(PacketGate::default()); + let server = start_server_with_gates( + ServerBehavior::Echo, + None, + None, + None, + Some(Arc::clone(&reply_gate)), + ); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.client.interval = Duration::from_secs(1); + managed.client.probe_timeout = Duration::from_secs(1); + let (task, mut handle) = + ManagedClient::task(managed, vec![target("old", server.addr)]).unwrap(); + let hook = Arc::new(UpdateSubmissionTestHook::default()); + hook.arm(); + handle.update_submission_hook = Some(Arc::clone(&hook)); + let runtime = runtime(); + let mut task = Box::pin(task); + drive_task_until(&runtime, &mut task, |task| { + task.targets[0].counters.packets_sent == 1 + }); + wait_flag(&server.probe_seen); + + let next_generation = task.next_generation; + let applied_command_sequence = task.applied_command_sequence; + let desired_membership = task + .targets + .iter() + .map(|target| (target.instance.clone(), target.desired)) + .collect::>(); + let submitter = handle.clone(); + let sender = thread::spawn(move || { + submitter.update_targets(vec![ManagedTargetConfig::new("new", "127.0.0.1:9")]) + }); + hook.wait_until_arrived(); + + let stop = handle.stop(); + drive_task_until(&runtime, &mut task, |task| { + task.state == DriverState::Stopping + && task.stop_observed + && task.resources().stop.update_admission() == UpdateAdmission::Stopping + && matches!(task.targets[0].state, TargetState::Draining { .. }) + }); + let stopping_status = handle.status(); + assert_eq!(stopping_status.lifecycle, ManagedLifecycle::Stopping); + + hook.release(); + let receipt = sender + .join() + .unwrap() + .expect("send must succeed while the receiver remains open during stopping"); + + runtime.block_on(poll_fn(|cx| { + assert!(task.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + })); + let result = runtime.block_on(receipt); + assert!(!matches!( + &result, + Err(ManagedCommandApplyError::AcknowledgementDisconnected) + )); + assert!(matches!(result, Err(ManagedCommandApplyError::Stopping))); + assert_eq!(task.applied_command_sequence, applied_command_sequence); + assert_eq!(task.next_generation, next_generation); + assert_eq!( + task.targets + .iter() + .map(|target| (target.instance.clone(), target.desired)) + .collect::>(), + desired_membership + ); + assert_eq!(handle.status(), stopping_status); + + reply_gate.release(); + wait_flag(&server.reply_sent); + let outcome = runtime.block_on(task); + runtime.block_on(stop); + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + server.finish(); +} + #[test] fn accepted_update_before_stop_resolves_stopping() { let mut managed = config(ManagedPacing::Staggered); -- 2.54.0