From 1d211c81ad4283b814e52daa262ac9941cb94a57 Mon Sep 17 00:00:00 2001
From: Nils Andreas Svee <me@lochnair.net>
Date: Sun, 4 Oct 2026 11:37:30 +0700
Subject: [PATCH] fix(client): avoid staggered wakes without a send deadline

---
 crates/irtt-client/src/managed/task.rs | 84 ++++++++++++++++++++++++--
 1 file changed, 79 insertions(+), 5 deletions(-)

diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs
index bc45e7e..8621893 100644
--- a/crates/irtt-client/src/managed/task.rs
+++ b/crates/irtt-client/src/managed/task.rs
@@ -1968,11 +1968,9 @@ impl ManagedClientTask {
             && self.config.pacing == ManagedPacing::Staggered
             && self.active_count() > 0
         {
-            send_deadline = match (send_deadline, self.send_gate) {
-                (Some(target), Some(gate)) => Some(target.max(gate)),
-                (None, gate) => gate,
-                (target, None) => target,
-            };
+            // A release gate delays a send opportunity; it cannot create one.
+            send_deadline = send_deadline
+                .map(|deadline| self.send_gate.map_or(deadline, |gate| deadline.max(gate)));
         }
         non_send_deadline.into_iter().chain(send_deadline).min()
     }
@@ -2398,6 +2396,82 @@ fn close_timeout_failure() -> ManagedTargetFailure {
 mod tests {
     use super::*;
 
+    // Public poll counts depend on reactor/timer timing. Inspect the selected
+    // deadline instead, after a real managed send exhausts a finite schedule.
+    #[test]
+    fn exhausted_staggered_schedule_waits_for_probe_timeout_not_release_gate() {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(async {
+                time::timeout(Duration::from_secs(5), async {
+                    let mut server = irtt_server::Server::bind(
+                        "127.0.0.1:0".parse().unwrap(),
+                        irtt_server::ServerConfig::default(),
+                    )
+                    .await
+                    .unwrap();
+                    let addr = server.local_addr().unwrap();
+                    let server_task = tokio::spawn(async move {
+                        server.run(std::future::pending::<()>()).await.unwrap();
+                    });
+                    let (mut task, handle) = ManagedClient::task(
+                        ManagedClientConfig {
+                            client: ClientConfig {
+                                duration: Some(Duration::from_secs(10)),
+                                interval: Duration::from_secs(15),
+                                probe_timeout: Duration::from_secs(60),
+                                ..ClientConfig::default()
+                            },
+                            pacing: ManagedPacing::Staggered,
+                            ..ManagedClientConfig::default()
+                        },
+                        vec![ManagedTargetConfig::new("finite", addr.to_string())],
+                    )
+                    .unwrap();
+                    let mut events = handle.subscribe().unwrap();
+                    std::future::poll_fn(|cx| {
+                        if let Poll::Ready(outcome) = Pin::new(&mut task).poll(cx) {
+                            panic!("managed target ended before its final probe: {outcome:?}");
+                        }
+                        while let Ok(event) = events.try_recv() {
+                            if matches!(
+                                event,
+                                ManagedEvent::Client {
+                                    event: ClientEvent::EchoSent { .. },
+                                    ..
+                                }
+                            ) {
+                                return Poll::Ready(());
+                            }
+                        }
+                        Poll::Pending
+                    })
+                    .await;
+
+                    let TargetState::Active { client } = &task.targets[0].state else {
+                        panic!("the final probe must remain active while pending");
+                    };
+                    assert!(client.has_pending_probes());
+                    assert!(task.targets[0].schedule.as_ref().unwrap().is_finished());
+                    let timeout = client.next_probe_timeout_deadline().unwrap();
+                    assert!(task.send_gate.unwrap() < timeout);
+                    assert_eq!(task.next_deadline(), Some(timeout));
+
+                    // An expired gate must not register an immediately-ready
+                    // timer either; no sleep or fake clock is needed.
+                    task.send_gate = Some(Instant::now() - Duration::from_secs(1));
+                    assert_eq!(task.next_deadline(), Some(timeout));
+                    let mut cx = Context::from_waker(std::task::Waker::noop());
+                    assert!(!task.register_timer(&mut cx));
+                    server_task.abort();
+                })
+                .await
+                .expect("managed target did not send its final probe");
+            });
+    }
+
     // Poll latency through public events cannot distinguish an O(n) discovery
     // scan from a bounded scan reliably. This tiny inspection trace measures
     // work directly, including empty targets, and checks eventual coverage