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::{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,
+    },
     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,10 @@ pub async fn run_tui(
     tokio::pin!(render);
     let mut interrupted = false;
     let mut stop_requested = false;
-    let mut terminal_targets = HashSet::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,18 +101,12 @@ 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());
-                }
+                state.process_managed_status(&status.borrow_and_update());
                 force_render = true;
             }
             event = events.recv(), if !events_closed => match event {
                 Ok(event) => force_render = process_tui_event(
-                    event, &mut state, &mut terminal_targets,
+                    event, &mut state,
                 ),
                 Err(RecvError::Lagged(count)) => {
                     dropped_events = dropped_events.saturating_add(count);
@@ -138,11 +141,55 @@ pub async fn run_tui(
                     _ => {}
                 }
             }
+            _ = &mut retry, if retry_armed && !interrupted && !*shutdown.borrow() => {
+                state.process_managed_status(&status.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) => {
+                        state.process_managed_status(&ack.status);
+                        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.
@@ -160,37 +207,32 @@ pub async fn run_tui(
         }
     };
     drain_final_events(&mut events, &mut dropped_events, |event| {
-        process_tui_event(event, &mut state, &mut terminal_targets);
+        process_tui_event(event, &mut state);
         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()) {
-            state.process_target_outcome(target);
-        }
+        state.process_target_outcome(target);
     }
     interrupted |= *shutdown.borrow();
     let error = match &outcome.end_reason {
         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 {
@@ -203,11 +245,7 @@ pub async fn run_tui(
     Ok(())
 }
 
-fn process_tui_event(
-    event: ManagedEvent,
-    state: &mut TuiState,
-    terminal_targets: &mut HashSet<TargetInstance>,
-) -> bool {
+fn process_tui_event(event: ManagedEvent, state: &mut TuiState) -> bool {
     match event {
         ManagedEvent::Client { target, event } => {
             let force_render = matches!(
@@ -221,10 +259,19 @@ fn process_tui_event(
             force_render
         }
         ManagedEvent::TargetFinished { outcome } => {
-            terminal_targets.insert(outcome.target.clone());
             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

@@ -11,7 +11,10 @@ use crossterm::{
     execute,
     terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
 };
-use irtt_client::managed::{ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance};
+use irtt_client::managed::{
+    ManagedStatus, ManagedTargetEndReason, ManagedTargetLifecycle, ManagedTargetOutcome,
+    TargetInstance,
+};
 use irtt_client::{
     ClientEvent, ClientTimestamp, NegotiationResult, OneWayDelaySample, RttSample, SignedDuration,
 };
@@ -183,13 +186,90 @@ 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].terminal_generation == Some(target.generation)
+            && matches!(
+                event,
+                ClientEvent::SessionStarted(_)
+                    | ClientEvent::NoTestCompleted(_)
+                    | ClientEvent::SessionClosed { .. }
+            )
+        {
+            return;
+        }
         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;
+        }
+        if target.generation != Some(generation) {
+            target.generation = Some(generation);
+            target.stats = StatsCollector::new(stats_config(self.config.duration.is_none()));
+            target.session = None;
+            target.negotiated = None;
+        }
+        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_managed_status(&mut self, status: &ManagedStatus) {
+        // Consume durable endings before accepting replacement generations.
+        // Receipts retain this history even if a watch update was coalesced.
+        for outcome in status.recent_target_outcomes.iter().chain(
+            status
+                .targets
+                .iter()
+                .filter_map(|target| target.outcome.as_deref()),
+        ) {
+            self.process_target_outcome(outcome);
+        }
+        for target in status.targets.iter() {
+            if matches!(
+                target.lifecycle,
+                ManagedTargetLifecycle::Pending
+                    | ManagedTargetLifecycle::Connecting
+                    | ManagedTargetLifecycle::Opening
+            ) {
+                self.process_target_opening(&target.target);
+            } else {
+                let idx = self.ensure_target(target.target.id.as_str());
+                if self.accept_generation(idx, target.target.generation)
+                    && self.targets[idx].terminal_generation != Some(target.target.generation)
+                    && target.lifecycle == ManagedTargetLifecycle::Active
+                {
+                    self.targets[idx].status = TargetStatus::Active;
+                }
+            }
+        }
+    }
+
     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.targets[idx]
+            .terminal_generation
+            .is_some_and(|generation| generation >= outcome.target.generation)
+        {
+            return;
+        }
+        let current = self.accept_generation(idx, outcome.target.generation);
+        self.targets[idx].terminal_generation = Some(outcome.target.generation);
         let (status, recent, primary_warning) = match &outcome.end_reason {
             ManagedTargetEndReason::TestComplete => {
                 (TargetStatus::Closed, "test completed".to_owned(), None)
@@ -237,8 +317,10 @@ impl TuiState {
         };
 
         let target = &mut self.targets[idx];
-        target.remote = outcome.remote.map(|remote| remote.to_string());
-        target.status = status;
+        if current {
+            target.remote = outcome.remote.map(|remote| remote.to_string());
+            target.status = status;
+        }
         if let Some(warning) = &warning {
             target.last_warning = Some(warning.clone());
         }
@@ -250,10 +332,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 +402,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 +448,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 +604,8 @@ impl TuiState {
 #[derive(Debug)]
 struct TuiTargetState {
     label: String,
+    generation: Option<u64>,
+    terminal_generation: Option<u64>,
     remote: Option<String>,
     session: Option<String>,
     status: TargetStatus,
@@ -533,6 +620,8 @@ impl TuiTargetState {
     fn new(label: String, stats_config: irtt_stats::StatsConfig) -> Self {
         Self {
             label,
+            generation: None,
+            terminal_generation: None,
             remote: None,
             session: None,
             status: TargetStatus::Opening,
@@ -886,6 +975,7 @@ impl GraphMetric {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 struct GraphSample {
+    generation: u64,
     timestamp: ClientTimestamp,
     seq: u32,
     rtt: RttSample,
@@ -1222,7 +1312,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 +1361,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 +1427,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 +1806,156 @@ 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 durable_failure_and_replacement_reset_stats_without_client_events() {
+        use irtt_client::managed::{
+            ManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedTargetConfig,
+        };
+
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        runtime.block_on(async {
+            let desired = vec![ManagedTargetConfig::new("monitor", "127.0.0.1:99999")];
+            let (task, handle) = ManagedClient::task(
+                ManagedClientConfig {
+                    completion: ManagedCompletionPolicy::ExplicitStop,
+                    ..Default::default()
+                },
+                desired.clone(),
+            )
+            .unwrap();
+            let mut status = handle.subscribe_status();
+            let instance = status.borrow().targets[0].target.clone();
+            let mut state = TuiState::with_target_labels(
+                TuiConfig {
+                    interval: Duration::from_secs(1),
+                    duration: None,
+                    timeout: Duration::from_secs(4),
+                    target_probes: None,
+                    pacing: GroupPacingArg::Staggered,
+                },
+                ["monitor".to_owned()],
+            );
+            state.process_target_event(
+                &instance,
+                &ClientEvent::EchoSent {
+                    seq: 42,
+                    remote: "127.0.0.1:2112".parse().unwrap(),
+                    scheduled_at: None,
+                    sent_at: ClientTimestamp::now(),
+                    bytes: 60,
+                    send_call: Duration::ZERO,
+                    timer_error: None,
+                },
+            );
+            assert_eq!(state.targets[0].stats.snapshot().packets.packets_sent, 1);
+            let run = tokio::spawn(task);
+            tokio::time::timeout(Duration::from_secs(5), async {
+                loop {
+                    status.changed().await.unwrap();
+                    if status.borrow().targets[0].outcome.is_some() {
+                        break;
+                    }
+                }
+                // Never consume broadcast events: status owns the failure evidence.
+                let snapshot = status.borrow_and_update().clone();
+                let outcome = snapshot.targets[0].outcome.as_ref().unwrap();
+                state.process_managed_status(&snapshot);
+                assert_eq!(state.targets[0].status, TargetStatus::Failed);
+                assert!(state.targets[0].last_warning.is_some());
+                let event_count = state.recent_events.len();
+                state.process_target_outcome(outcome);
+                assert_eq!(
+                    state.recent_events.len(),
+                    event_count,
+                    "no duplicate ending"
+                );
+
+                let ack = handle.update_targets(desired).unwrap().await.unwrap();
+                // Opening events may arrive before their update receipt.
+                let mut late_state =
+                    TuiState::with_target_labels(state.config.clone(), ["monitor".to_owned()]);
+                late_state.process_target_opening(&ack.status.targets[0].target);
+                late_state.process_managed_status(&ack.status);
+                assert_eq!(late_state.targets[0].status, TargetStatus::Opening);
+                assert!(late_state.targets[0].last_warning.is_some());
+
+                state.process_managed_status(&ack.status);
+                assert_eq!(state.targets[0].status, TargetStatus::Opening);
+                assert_eq!(state.targets[0].stats.snapshot().packets.packets_sent, 0);
+                assert!(
+                    state.targets[0].last_warning.is_some(),
+                    "failure evidence survives reprovisioning"
+                );
+                loop {
+                    status.changed().await.unwrap();
+                    if status.borrow().targets[0].outcome.is_some() {
+                        break;
+                    }
+                }
+                state.process_managed_status(&status.borrow_and_update());
+                assert_eq!(state.targets[0].status, TargetStatus::Failed);
+                assert_eq!(state.targets[0].stats.snapshot().packets.packets_sent, 0);
+                // An older receipt must not revive a durably failed generation.
+                state.process_managed_status(&ack.status);
+                assert_eq!(state.targets[0].status, TargetStatus::Failed);
+                handle.stop().await;
+                run.await.unwrap();
+            })
+            .await
+            .unwrap();
+        });
+    }
+
+    #[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"
+        );
+    }
+}