From 419fd0ece0fca2d71ad58ee759abf1e720045a4b 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 | 125 +++++++---- crates/irtt-app/src/cmd/tui/ui.rs | 319 ++++++++++++++++++++++++++--- docs/irtt-tui.md | 25 ++- 3 files changed, 400 insertions(+), 69 deletions(-) diff --git a/crates/irtt-app/src/cmd/tui/run.rs b/crates/irtt-app/src/cmd/tui/run.rs index 70fe21e..4209476 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::{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, -) -> 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, } } diff --git a/crates/irtt-app/src/cmd/tui/ui.rs b/crates/irtt-app/src/cmd/tui/ui.rs index c7512dd..49301ad 100644 --- a/crates/irtt-app/src/cmd/tui/ui.rs +++ b/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, + terminal_generation: Option, remote: Option, session: Option, 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::>(); let block = Block::bordered().title(format!( "{} | {} | window {}", @@ -1271,30 +1361,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 +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) -> 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" + ); + } +} diff --git a/docs/irtt-tui.md b/docs/irtt-tui.md index a080280..7586802 100644 --- a/docs/irtt-tui.md +++ b/docs/irtt-tui.md @@ -31,6 +31,18 @@ a finite run: irtt-tui --duration 30s ``` +Continuous runs reconnect after target-local failures, peer closure, or a +session completing because the server restricted its duration. After a +1.5-second delay, one coalesced retry reopens completed targets while healthy +sessions continue undisturbed. Even if every target fails, the TUI stays alive +until you quit. Explicit stops, removed or replaced targets, and no-test +completion do not trigger retries. Quitting cancels pending retry work. + +Each reconnect keeps the same logical target row. Statistics reset when a new +session generation is accepted, including attempts that fail before opening. +Retained graph history and the age of the last actual reply remain; the graph +leaves a gap between generations instead of connecting their samples. + ## DASHBOARD One dashboard is used for every target count and terminal size. It shows: @@ -88,7 +100,7 @@ redraws still work while the display is paused. The six graph metrics are effective RTT, raw RTT, adjusted RTT, client-to-server delay, server-to-client delay, and server processing time. Signed effective/adjusted RTT and one-way values are preserved. Graph windows -range from 5 seconds to one hour and default to one minute. The footer shows +range from 5 seconds to 24 hours and default to one minute. The footer shows controls appropriate to the currently open graph or details sheet. ## RETAINED HISTORY @@ -116,9 +128,14 @@ corrupt the display. Size a long finite TUI run from the figures in ## EXIT BEHAVIOR -Quitting with `q` or `Ctrl-C` is an interrupted, successful exit. Otherwise, -exit status follows the same peer-close and driver-failure rules as -`irtt-client(1)`. +Quitting with `q` or `Ctrl-C` is an interrupted, successful exit. In continuous +mode, peer closure and target failures trigger reconnect rather than ending the +run. A managed driver failure or an unexpected reconnect-update failure still +ends the TUI with an error. + +Finite runs do not reconnect. They complete after their target sessions end; +a peer closure is a normal session ending, while a run whose targets all fail +exits with an error. ## EXAMPLES