feat(tui): enable continuous target re-provisioning with generation tracking

Allow targets to automatically restart after completing or failing, and ensure distinct state and statistics across successive target generations.

crates/irtt-app/src/cmd/tui/run.rs

@@ -1,13 +1,10 @@
-use std::{collections::HashSet, future::poll_fn, io, time::Duration};
+use std::{collections::HashMap, future::poll_fn, io, time::Duration};
 
 use super::ui::{TuiConfig, TuiState, TuiStatus, TuiTerminal};
 use crate::{
     cmd::tui::args::TuiArgs,
     shared::client::{
-        session::{
-            drain_final_events, peer_close_run_error, request_managed_stop_for_peer_close,
-            request_managed_stop_once,
-        },
+        session::{drain_final_events, request_managed_stop_once},
         worker::ManagedWorker,
     },
 };
@@ -17,7 +14,10 @@ use crossterm::{
 };
 use futures_core::Stream;
 use irtt_client::{
-    managed::{ManagedClient, ManagedEndReason, ManagedEvent, TargetInstance},
+    managed::{
+        ManagedClient, ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy,
+        ManagedEndReason, ManagedEvent, ManagedTargetEndReason, ManagedTargetLifecycle, TargetId,
+    },
     ClientEvent,
 };
 use ratatui::layout::Rect;
@@ -26,6 +26,8 @@ use tokio::{
     time::{sleep, Instant},
 };
 
+const RETRY_DELAY: Duration = Duration::from_millis(1500);
+
 const RENDER_INTERVAL: Duration = Duration::from_millis(100);
 
 pub async fn run_tui(
@@ -46,8 +48,12 @@ pub async fn run_tui(
     );
     state.set_status(TuiStatus::Opening);
     terminal.draw(&state)?;
-    let (task, handle) = match ManagedClient::task(setup.managed_config(), setup.managed_targets())
-    {
+    let desired_targets = setup.managed_targets();
+    let mut config = setup.managed_config();
+    if continuous {
+        config.completion = ManagedCompletionPolicy::ExplicitStop;
+    }
+    let (task, handle) = match ManagedClient::task(config, desired_targets.clone()) {
         Ok(value) => value,
         Err(error) => {
             state.set_error(error.to_string());
@@ -75,7 +81,11 @@ pub async fn run_tui(
     tokio::pin!(render);
     let mut interrupted = false;
     let mut stop_requested = false;
-    let mut terminal_targets = HashSet::new();
+    let mut terminal_targets = HashMap::new();
+    let retry = sleep(RETRY_DELAY);
+    tokio::pin!(retry);
+    let mut retry_armed = false;
+    let mut update_receipt = None;
     let mut dropped_events = 0_u64;
     let mut events_closed = false;
     let mut status_closed = false;
@@ -92,13 +102,7 @@ pub async fn run_tui(
             }
             changed = status.changed(), if !status_closed => {
                 status_closed = changed.is_err();
-                let snapshot = status.borrow_and_update().clone();
-                if request_managed_stop_for_peer_close(
-                    continuous, interrupted, snapshot.peer_closed_target_outcomes,
-                    &mut stop_requested,
-                ) {
-                    drop(handle.stop());
-                }
+                status.borrow_and_update();
                 force_render = true;
             }
             event = events.recv(), if !events_closed => match event {
@@ -138,11 +142,59 @@ pub async fn run_tui(
                     _ => {}
                 }
             }
+            _ = &mut retry, if retry_armed && !interrupted && !*shutdown.borrow() => {
+                retry_armed = false;
+                update_receipt = match handle.update_targets(desired_targets.clone()) {
+                    Ok(receipt) => Some(receipt),
+                    Err(ManagedCommandError::Stopping | ManagedCommandError::DriverClosed)
+                        if stop_requested || *shutdown.borrow() => None,
+                    Err(error) => return Err(format!("reconnect update failed: {error}").into()),
+                };
+            }
+            result = async { update_receipt.as_mut().unwrap().await },
+                if update_receipt.is_some() => {
+                update_receipt = None;
+                match result {
+                    Ok(ack) => {
+                        for target in ack.status.targets.iter() {
+                            if matches!(target.lifecycle, ManagedTargetLifecycle::Pending
+                                | ManagedTargetLifecycle::Connecting | ManagedTargetLifecycle::Opening) {
+                                state.process_target_opening(&target.target);
+                            }
+                        }
+                        force_render = true;
+                    }
+                    Err(ManagedCommandApplyError::Stopping
+                        | ManagedCommandApplyError::AcknowledgementDisconnected)
+                        if stop_requested || *shutdown.borrow() => {}
+                    Err(error) => return Err(format!("reconnect update failed: {error}").into()),
+                }
+            }
             _ = &mut render, if !state.paused => force_render = true,
         }
         if interrupted && request_managed_stop_once(&mut stop_requested) {
             drop(handle.stop());
         }
+        if continuous
+            && !interrupted
+            && !stop_requested
+            && !retry_armed
+            && update_receipt.is_none()
+            && status.borrow().targets.iter().any(|target| {
+                target.desired
+                    && target.outcome.as_ref().is_some_and(|outcome| {
+                        matches!(
+                            outcome.end_reason,
+                            ManagedTargetEndReason::Failed(_)
+                                | ManagedTargetEndReason::PeerClosed
+                                | ManagedTargetEndReason::TestComplete
+                        )
+                    })
+            })
+        {
+            retry.as_mut().reset(Instant::now() + RETRY_DELAY);
+            retry_armed = true;
+        }
         if force_render {
             terminal.draw(&state)?;
             // Schedule from the completed draw, never catch up missed frames.
@@ -164,14 +216,15 @@ pub async fn run_tui(
         Ok::<(), std::convert::Infallible>(())
     })?;
     state.mark_dropped_managed_events(dropped_events);
-    if outcome.discarded_target_outcomes != 0 {
+    if !continuous && outcome.discarded_target_outcomes != 0 {
         state.set_run_error(format!(
             "{} final target outcomes were discarded",
             outcome.discarded_target_outcomes
         ));
     }
     for target in outcome.recent_target_outcomes.iter() {
-        if terminal_targets.insert(target.target.clone()) {
+        if terminal_targets.get(&target.target.id) != Some(&target.target.generation) {
+            terminal_targets.insert(target.target.id.clone(), target.target.generation);
             state.process_target_outcome(target);
         }
     }
@@ -180,17 +233,14 @@ pub async fn run_tui(
         ManagedEndReason::DriverFailed(failure) => {
             Some(format!("managed driver failed: {failure}"))
         }
-        _ => peer_close_run_error(continuous, interrupted, outcome.peer_closed_target_outcomes)
-            .or_else(|| {
-                (!interrupted
-                    && outcome.successful_target_outcomes == 0
-                    && outcome.failed_target_outcomes > 0)
-                    .then(|| {
-                        format!(
-                            "no managed target completed successfully ({} failed)",
-                            outcome.failed_target_outcomes
-                        )
-                    })
+        _ => (!interrupted
+            && outcome.successful_target_outcomes == 0
+            && outcome.failed_target_outcomes > 0)
+            .then(|| {
+                format!(
+                    "no managed target completed successfully ({} failed)",
+                    outcome.failed_target_outcomes
+                )
             }),
     };
     if let Some(error) = error {
@@ -206,7 +256,7 @@ pub async fn run_tui(
 fn process_tui_event(
     event: ManagedEvent,
     state: &mut TuiState,
-    terminal_targets: &mut HashSet<TargetInstance>,
+    terminal_targets: &mut HashMap<TargetId, u64>,
 ) -> bool {
     match event {
         ManagedEvent::Client { target, event } => {
@@ -221,10 +271,20 @@ fn process_tui_event(
             force_render
         }
         ManagedEvent::TargetFinished { outcome } => {
-            terminal_targets.insert(outcome.target.clone());
+            terminal_targets.insert(outcome.target.id.clone(), outcome.target.generation);
             state.process_target_outcome(&outcome);
             true
         }
+        ManagedEvent::TargetStateChanged {
+            target,
+            lifecycle:
+                ManagedTargetLifecycle::Pending
+                | ManagedTargetLifecycle::Connecting
+                | ManagedTargetLifecycle::Opening,
+        } => {
+            state.process_target_opening(&target);
+            true
+        }
         _ => false,
     }
 }

crates/irtt-app/src/cmd/tui/ui.rs

@@ -183,13 +183,47 @@ impl TuiState {
 
     pub(super) fn process_target_event(&mut self, target: &TargetInstance, event: &ClientEvent) {
         let idx = self.ensure_target(target.id.as_str());
+        if !self.accept_generation(idx, target.generation) {
+            return;
+        }
+        if self.targets[idx].stats_generation != Some(target.generation)
+            || matches!(event, ClientEvent::SessionStarted(_))
+        {
+            self.targets[idx].stats_generation = Some(target.generation);
+            // A dropped SessionStarted must never merge sequence namespaces.
+            self.targets[idx].stats =
+                StatsCollector::new(stats_config(self.config.duration.is_none()));
+        }
         self.process_event_for_target(idx, event);
     }
 
+    fn accept_generation(&mut self, idx: usize, generation: u64) -> bool {
+        let target = &mut self.targets[idx];
+        if target
+            .generation
+            .is_some_and(|current| current > generation)
+        {
+            return false;
+        }
+        target.generation = Some(generation);
+        true
+    }
+
+    pub(super) fn process_target_opening(&mut self, instance: &TargetInstance) {
+        let idx = self.ensure_target(instance.id.as_str());
+        let new_generation = self.targets[idx].generation != Some(instance.generation);
+        if self.accept_generation(idx, instance.generation) && new_generation {
+            self.targets[idx].status = TargetStatus::Opening;
+        }
+    }
+
     pub(super) fn process_target_outcome(&mut self, outcome: &ManagedTargetOutcome) {
         let label = outcome.target.id.as_str();
         let idx = self.ensure_target(label);
 
+        if !self.accept_generation(idx, outcome.target.generation) {
+            return;
+        }
         let (status, recent, primary_warning) = match &outcome.end_reason {
             ManagedTargetEndReason::TestComplete => {
                 (TargetStatus::Closed, "test completed".to_owned(), None)
@@ -250,10 +284,11 @@ impl TuiState {
             self.push_event(format!("{label}: {cleanup_warning}"));
         }
 
-        if self
-            .targets
-            .iter()
-            .all(|target| target.status.is_terminal())
+        if self.config.duration.is_some()
+            && self
+                .targets
+                .iter()
+                .all(|target| target.status.is_terminal())
         {
             self.status = if self.targets.iter().any(|target| target.status.is_success()) {
                 TuiStatus::Complete
@@ -319,6 +354,7 @@ impl TuiState {
                     ..
                 } => {
                     target.push_graph_sample(GraphSample {
+                        generation: target.generation.expect("client event has a generation"),
                         timestamp: *received_at,
                         seq: *seq,
                         rtt: *rtt,
@@ -364,10 +400,11 @@ impl TuiState {
         };
         if let Some(status) = global_status {
             self.status = if status == TuiStatus::Complete
-                && !self
-                    .targets
-                    .iter()
-                    .all(|target| target.status.is_terminal())
+                && (self.config.duration.is_none()
+                    || !self
+                        .targets
+                        .iter()
+                        .all(|target| target.status.is_terminal()))
             {
                 TuiStatus::Running
             } else {
@@ -519,6 +556,8 @@ impl TuiState {
 #[derive(Debug)]
 struct TuiTargetState {
     label: String,
+    generation: Option<u64>,
+    stats_generation: Option<u64>,
     remote: Option<String>,
     session: Option<String>,
     status: TargetStatus,
@@ -533,6 +572,8 @@ impl TuiTargetState {
     fn new(label: String, stats_config: irtt_stats::StatsConfig) -> Self {
         Self {
             label,
+            generation: None,
+            stats_generation: None,
             remote: None,
             session: None,
             status: TargetStatus::Opening,
@@ -886,6 +927,7 @@ impl GraphMetric {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 struct GraphSample {
+    generation: u64,
     timestamp: ClientTimestamp,
     seq: u32,
     rtt: RttSample,
@@ -1222,7 +1264,7 @@ fn render_graph_area(frame: &mut Frame<'_>, area: Rect, state: &TuiState) {
         .targets
         .iter()
         .enumerate()
-        .filter_map(|(idx, target)| target_metric_series(target, idx, viewport, metric))
+        .flat_map(|(idx, target)| target_metric_series(target, idx, viewport, metric))
         .collect::<Vec<_>>();
     let block = Block::bordered().title(format!(
         "{} | {} | window {}",
@@ -1271,30 +1313,51 @@ fn target_metric_series(
     target_idx: usize,
     viewport: GraphViewportRange,
     metric: GraphMetric,
-) -> Option<ChartSeries> {
+) -> Vec<ChartSeries> {
     // Histories are monotonic. Skip old samples without converting every
     // retained measurement on each repaint, then keep the nearest valid
     // predecessor for interpolation across the left boundary.
     let first = target
         .graph_history
         .partition_point(|sample| sample.timestamp.mono < viewport.start);
+    let predecessor_generation = target
+        .graph_history
+        .get(first.saturating_sub(1))
+        .map(|sample| sample.generation);
     let mut previous = target
         .graph_history
         .range(..first)
         .rev()
+        .take_while(|sample| Some(sample.generation) == predecessor_generation)
         .find_map(|sample| {
-            metric
-                .value_ns(sample)
-                .map(|ns| (sample.timestamp.mono, ns as f64 / 1_000_000.0))
+            metric.value_ns(sample).map(|ns| {
+                (
+                    sample.generation,
+                    (sample.timestamp.mono, ns as f64 / 1_000_000.0),
+                )
+            })
         });
     let points = target.graph_history.range(first..).filter_map(|sample| {
-        metric
-            .value_ns(sample)
-            .map(|ns| (sample.timestamp.mono, ns as f64 / 1_000_000.0))
+        metric.value_ns(sample).map(|ns| {
+            (
+                sample.generation,
+                (sample.timestamp.mono, ns as f64 / 1_000_000.0),
+            )
+        })
     });
+    let mut series = Vec::new();
     let mut data = Vec::new();
-    for point in points {
-        if let Some(before) = previous {
+    for (generation, point) in points {
+        if previous.is_some_and(|(before_generation, _)| before_generation != generation) {
+            if !data.is_empty() {
+                series.push(ChartSeries {
+                    style: target_style(target_idx),
+                    data: std::mem::take(&mut data),
+                });
+            }
+            previous = None;
+        }
+        if let Some((_, before)) = previous {
             // Only genuine crossing segments contribute boundary points. In
             // particular, never extend the latest sample to the live edge.
             if before.0 < viewport.start && point.0 > viewport.start {
@@ -1316,12 +1379,15 @@ fn target_metric_series(
                 point.1,
             ));
         }
-        previous = Some(point);
+        previous = Some((generation, point));
     }
-    (!data.is_empty()).then_some(ChartSeries {
-        style: target_style(target_idx),
-        data,
-    })
+    if !data.is_empty() {
+        series.push(ChartSeries {
+            style: target_style(target_idx),
+            data,
+        });
+    }
+    series
 }
 
 fn interpolate_graph_value(a: (Instant, f64), b: (Instant, f64), at: Instant) -> f64 {
@@ -1692,3 +1758,53 @@ fn format_optional_hex(value: Option<u64>) -> String {
         .map(|value| format!("0x{value:x}"))
         .unwrap_or_else(|| ABSENT.to_owned())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn reconnect_history_clips_within_sessions_and_leaves_the_outage_empty() {
+        let start = Instant::now();
+        let mut target = TuiTargetState::new("monitor".into(), stats_config(true));
+        for (generation, seconds) in [(0, 0), (0, 2), (1, 6), (1, 8)] {
+            let raw = Duration::from_millis(seconds + 1);
+            target.push_graph_sample(GraphSample {
+                generation,
+                timestamp: ClientTimestamp {
+                    mono: start + Duration::from_secs(seconds),
+                    wall: SystemTime::UNIX_EPOCH + Duration::from_secs(seconds),
+                },
+                seq: 0,
+                rtt: RttSample {
+                    raw,
+                    adjusted: None,
+                    effective: SignedDuration::from_nanos(raw.as_nanos() as i128),
+                },
+                one_way: None,
+                server_processing: None,
+            });
+        }
+        let series = |left, right| {
+            target_metric_series(
+                &target,
+                0,
+                GraphViewportRange {
+                    start: start + Duration::from_secs(left),
+                    end: start + Duration::from_secs(right),
+                    window: Duration::from_secs(right - left),
+                },
+                GraphMetric::RawRtt,
+            )
+        };
+        let clipped = series(1, 7);
+        assert_eq!(clipped.len(), 2);
+        assert_eq!(clipped[0].data, [(0.0, 2.0), (1.0, 3.0)]);
+        assert_eq!(clipped[1].data, [(5.0, 7.0), (6.0, 8.0)]);
+        assert!(series(3, 5).is_empty(), "no interpolation across sessions");
+        assert!(
+            series(9, 10).is_empty(),
+            "no extrapolation to the live edge"
+        );
+    }
+}