Merge branch 'fix/managed-finite-lifetime-wake'

Expose the negotiated schedule end only while the schedule is unfinished,
and include it among Active targets' non-send deadlines. This keeps finite
expiry independent of socket writability and of Burst/Staggered pacing.
Staggered release gating continues to apply only to send opportunities.

The existing expiry path marks the schedule finished, removing its lifetime
deadline before pending probes continue draining. No new state, readiness
handling, cadence, or drain/close policy is introduced. Continuous schedules
have no lifetime deadline.

The prior public-API Linux namespace/veth/TBF investigation observed genuine
EAGAIN with Burst pacing, 500 ms duration, 1 ms interval, 1400-byte probes,
and 20 ms probe timeout. Polling stopped around 327 ms after pending probes
disappeared, remained stalled beyond 1.5 s, and resumed on queue release.
Staggered applicability follows from the shared deadline selection; it was
not separately reproduced under TBF.

Retain two deterministic owning-layer tests for schedule deadline lifetime
and managed selection under send_waiting and stagger gating, using ordinary
Instant values and a production socket. Keep the privileged networking
harness as investigation evidence rather than adding bespoke test machinery.

Validation: formatting; focused schedule/managed tests; managed lifecycle,
behavior, and blocking-owner integration tests; workspace all-feature check,
all-target tests, doctests, and strict Clippy; client no-default-features
check/tests/strict Clippy and Tokio-free normal dependency tree; Linux
all-feature client and integration suites, including ancillary transport.

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

@@ -48,6 +48,11 @@ impl ProbeSchedule {
         self.next_send_at
     }
 
+    /// Wake for finite expiry independently of send readiness until finished.
+    pub(crate) fn end_deadline(&self) -> Option<Instant> {
+        self.end_at.filter(|_| !self.is_finished())
+    }
+
     pub(crate) fn interval(&self) -> Duration {
         self.interval
     }
@@ -152,3 +157,42 @@ pub(crate) fn instant_abs_diff(left: Instant, right: Instant) -> Duration {
         .or_else(|| right.checked_duration_since(left))
         .unwrap_or(Duration::ZERO)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn lifetime_deadline_exists_only_for_an_unfinished_finite_schedule() {
+        let start = Instant::now();
+        let end = start + Duration::from_millis(500);
+        let mut negotiated = NegotiatedParams {
+            params: irtt_proto::Params {
+                duration_ns: 500_000_000,
+                interval_ns: 1_000_000,
+                ..irtt_proto::Params::default()
+            },
+            restrictions: Vec::new(),
+        };
+        let mut finite = ProbeSchedule::new(start, &negotiated).unwrap();
+        assert_eq!(finite.end_deadline(), Some(end));
+        assert!(finite.permit_probe_at(end - Duration::from_nanos(1)));
+        assert_eq!(finite.end_deadline(), Some(end));
+        assert!(!finite.permit_probe_at(end));
+        assert!(finite.is_finished());
+        assert_eq!(finite.end_deadline(), None);
+
+        let mut finite = ProbeSchedule::new(start, &negotiated).unwrap();
+        let commit = finite
+            .preflight_managed_commit(start, end - Duration::from_nanos(1))
+            .unwrap();
+        finite.commit(commit);
+        assert!(finite.is_finished());
+        assert_eq!(finite.end_deadline(), None);
+
+        negotiated.params.duration_ns = 0;
+        let mut continuous = ProbeSchedule::new(start, &negotiated).unwrap();
+        assert!(continuous.permit_probe_at(end));
+        assert_eq!(continuous.end_deadline(), None);
+    }
+}

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

@@ -1936,6 +1936,12 @@ impl ManagedClientTask {
                     non_send_deadline = non_send_deadline
                         .into_iter()
                         .chain(client.next_probe_timeout_deadline())
+                        .chain(
+                            target
+                                .schedule
+                                .as_ref()
+                                .and_then(ProbeSchedule::end_deadline),
+                        )
                         .min();
                     if target.desired && target.retirement.is_none() && !target.send_waiting {
                         send_deadline = send_deadline
@@ -2399,6 +2405,71 @@ fn close_timeout_failure() -> ManagedTargetFailure {
 mod tests {
     use super::*;
 
+    // Real EAGAIN needs privileged queue shaping. Inspect only the owning
+    // deadline selection here, using a production socket without sending.
+    #[test]
+    fn finite_lifetime_deadline_survives_send_waiting_and_stagger_gating() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        runtime.block_on(async {
+            let start = Instant::now();
+            let end = start + Duration::from_millis(500);
+            let negotiated = crate::NegotiatedParams {
+                params: irtt_proto::Params {
+                    duration_ns: 500_000_000,
+                    interval_ns: 1_000_000,
+                    ..irtt_proto::Params::default()
+                },
+                restrictions: Vec::new(),
+            };
+            for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] {
+                let (mut task, _handle) = ManagedClient::task(
+                    ManagedClientConfig {
+                        pacing,
+                        ..ManagedClientConfig::default()
+                    },
+                    vec![ManagedTargetConfig::new("target", "127.0.0.1:9")],
+                )
+                .unwrap();
+                let client = AsyncClient::connect(ClientConfig {
+                    server_addr: "127.0.0.1:9".into(),
+                    ..ClientConfig::default()
+                })
+                .await
+                .unwrap();
+                task.state = DriverState::Running;
+                task.install_target_state(0, TargetState::Active { client });
+                task.send_gate = Some(end + Duration::from_secs(1));
+
+                for send_waiting in [false, true] {
+                    task.targets[0].send_waiting = send_waiting;
+                    task.targets[0].schedule =
+                        Some(ProbeSchedule::new(start, &negotiated).unwrap());
+                    let expected = if pacing == ManagedPacing::Burst && !send_waiting {
+                        start
+                    } else {
+                        end
+                    };
+                    assert_eq!(task.next_deadline(), Some(expected));
+
+                    task.targets[0]
+                        .schedule
+                        .as_mut()
+                        .unwrap()
+                        .permit_probe_at(end);
+                    assert_eq!(task.next_deadline(), None);
+                }
+
+                let mut continuous = negotiated.clone();
+                continuous.params.duration_ns = 0;
+                task.targets[0].schedule = Some(ProbeSchedule::new(start, &continuous).unwrap());
+                assert_eq!(task.next_deadline(), None);
+            }
+        });
+    }
+
     #[test]
     fn release_gate_only_delays_an_existing_send_deadline() {
         let earlier = Instant::now();