test(managed): verify bounded control progress

crates/irtt-client/src/managed/task.rs

@@ -138,6 +138,73 @@ pub struct ManagedClientHandle {
     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 {
@@ -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))
 }

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,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);