Merge pull request #53 from Lochnair/fix/managed-liveness-control

fix(managed): harden liveness and control bounds

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<Result<OpenOutcome, ClientError>> {
-        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()))
             }
         }
     }

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| {

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")]

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,8 +135,76 @@ impl StopSignal {
 pub struct ManagedClientHandle {
     stop: Arc<StopSignal>,
     commands: mpsc::Sender<ManagedCommand>,
+    max_live_target_generations: usize,
     status: watch::Receiver<Arc<ManagedStatus>>,
     events: broadcast::WeakSender<ManagedEvent>,
+    #[cfg(test)]
+    update_submission_hook: Option<Arc<UpdateSubmissionTestHook>>,
+}
+
+#[cfg(test)]
+#[derive(Default)]
+struct UpdateSubmissionTestHook {
+    state: std::sync::Mutex<UpdateSubmissionTestState>,
+    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 {
@@ -130,16 +240,41 @@ impl ManagedClientHandle {
         &self,
         targets: Vec<ManagedTargetConfig>,
     ) -> Result<ManagedCommandReceipt, ManagedCommandError> {
-        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,
+            });
+        }
+
+        #[cfg(test)]
+        if let Some(hook) = &self.update_submission_hook {
+            hook.pause_after_admission();
+        }
+
+        // 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,
             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 +792,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 +822,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 +906,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 +1816,22 @@ impl ManagedClientTask {
         }
     }
 
-    fn seal(&mut self, end_reason: ManagedEndReason, failed: bool) -> Poll<ManagedOutcome> {
+    fn seal(&mut self, mut end_reason: ManagedEndReason, failed: bool) -> Poll<ManagedOutcome> {
+        // 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 +1913,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 +1931,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 +1952,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 +1977,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 +1996,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 +2074,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 +2112,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,8 +2149,11 @@ fn build_task(
     let handle = ManagedClientHandle {
         stop,
         commands: command_sender,
+        max_live_target_generations,
         status: status_receiver,
         events: weak_events,
+        #[cfg(test)]
+        update_submission_hook: None,
     };
     Ok((task, handle))
 }

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<Mutex<Vec<PacketRecord>>>,
     probe_seen: Arc<(Mutex<bool>, Condvar)>,
     reply_sent: Arc<(Mutex<bool>, Condvar)>,
+    open_reply_sent: Arc<(Mutex<bool>, 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,374 @@ 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 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::<Vec<_>>();
+    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::<Vec<_>>(),
+        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);
+    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);