From c5a234c3779894588fe8f2521471c9ff0b224962 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 6 Oct 2026 16:26:46 +0700 Subject: [PATCH] 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 | 122 ++++++++++++++++------ crates/irtt-app/src/cmd/tui/ui.rs | 162 +++++++++++++++++++++++++---- 2 files changed, 230 insertions(+), 54 deletions(-) diff --git a/crates/irtt-app/src/cmd/tui/run.rs b/crates/irtt-app/src/cmd/tui/run.rs index 70fe21e..6bba6d6 100644 --- a/crates/irtt-app/src/cmd/tui/run.rs +++ b/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, + terminal_targets: &mut HashMap, ) -> 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, } } diff --git a/crates/irtt-app/src/cmd/tui/ui.rs b/crates/irtt-app/src/cmd/tui/ui.rs index c7512dd..ff0a4a7 100644 --- a/crates/irtt-app/src/cmd/tui/ui.rs +++ b/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, + stats_generation: Option, remote: Option, session: Option, 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::>(); let block = Block::bordered().title(format!( "{} | {} | window {}", @@ -1271,30 +1313,51 @@ fn target_metric_series( target_idx: usize, viewport: GraphViewportRange, metric: GraphMetric, -) -> Option { +) -> Vec { // 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) -> 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" + ); + } +}