From 1013f95d0916e717f37ae74f66c83855034c3c80 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Wed, 14 Oct 2026 11:12:50 +0700 Subject: [PATCH] fix(irtt): retry terminal targets without restarting healthy sessions --- src/measurement/irtt.rs | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/src/measurement/irtt.rs b/src/measurement/irtt.rs index b25bd90..6d1c63a 100644 --- a/src/measurement/irtt.rs +++ b/src/measurement/irtt.rs @@ -7,7 +7,7 @@ use irtt_client::{ ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError, - ManagedTargetConfig, ManagedTargetEndReason, TargetAuth, TargetId, + ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetLifecycle, TargetAuth, TargetId, }, }; use log::{debug, trace, warn}; @@ -81,6 +81,7 @@ pub struct IrttEngine { client: ManagedClientHandle, bindings: HashMap, + targets: Vec, origin: Instant, @@ -104,6 +105,7 @@ impl IrttEngine { Self { client: handle, bindings: HashMap::new(), + targets: Vec::new(), origin, binding_rx, observation_tx, @@ -126,9 +128,10 @@ impl IrttEngine { } target }) - .collect(); + .collect::>(); - self.client.update_targets(targets)?.await?; + self.client.update_targets(targets.clone())?.await?; + self.targets = targets; self.bindings = bindings .into_iter() @@ -166,6 +169,11 @@ impl IrttEngine { async fn run_events(&mut self) -> Result<(), IrttError> { let mut events = self.client.subscribe()?; + // Durable status catches failures even if their presentation events were lost. + // Resubmitting the full desired set restarts terminal targets only. + let mut retry = tokio::time::interval(std::time::Duration::from_secs(5)); + retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + retry.tick().await; loop { tokio::select! { bindings = self.binding_rx.recv_async() => { @@ -175,6 +183,17 @@ impl IrttEngine { self.set_bindings(bindings).await?; } + _ = retry.tick() => { + if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) { + return Ok(()); + } + if self.client.status().targets.iter().any(|target| { + target.desired && target.lifecycle == ManagedTargetLifecycle::Terminal + }) { + self.client.update_targets(self.targets.clone())?.await?; + } + } + event = events.recv() => { let event = match event { Ok(event) => event,