diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index 3af8947..b931a7a 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -399,6 +399,9 @@ fn run_group_stream( } } } + if let Some(warning) = dropped_event_warning(events.dropped_events()) { + eprintln!("{warning}"); + } if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted { return Err("managed client group was cancelled".into()); @@ -446,6 +449,14 @@ fn report_target_failure(target: &ManagedTargetOutcome) { } } +fn dropped_event_warning(dropped_events: u64) -> Option { + (dropped_events > 0).then(|| { + format!( + "irtt-rs: warning: dropped {dropped_events} managed group events; output and statistics may be incomplete" + ) + }) +} + fn estimated_group_completion_grace(args: &ClientArgs) -> Duration { let open_timeout: Duration = args.to_client_config().open_timeouts.iter().sum(); open_timeout @@ -598,6 +609,14 @@ mod tests { OutputConfig::new(format, columns, header, verbose, false).unwrap() } + #[test] + fn dropped_event_warning_marks_statistics_incomplete() { + assert_eq!(dropped_event_warning(0), None); + let warning = dropped_event_warning(7).unwrap(); + assert!(warning.contains("dropped 7 managed group events")); + assert!(warning.contains("statistics may be incomplete")); + } + #[test] fn output_helper_streams_and_collects_events() { let sent_at = test_timestamp(Duration::from_secs(1)); diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index bc53bea..a5cb14e 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -242,6 +242,7 @@ fn run_group_tui( } } } + state.mark_dropped_events(events.dropped_events()); for target in &outcome.targets { if terminal_targets.insert(target.id.as_str().to_owned()) { state.process_target_outcome(target); diff --git a/crates/irtt-cli/src/cmd/tui/ui.rs b/crates/irtt-cli/src/cmd/tui/ui.rs index f30bcfa..0104d17 100644 --- a/crates/irtt-cli/src/cmd/tui/ui.rs +++ b/crates/irtt-cli/src/cmd/tui/ui.rs @@ -96,6 +96,7 @@ pub(super) struct TuiState { targets: Vec, recent_events: VecDeque, last_warning: Option, + dropped_events: u64, graph_metric: GraphMetric, graph_viewport: GraphViewport, view: TuiView, @@ -130,6 +131,7 @@ impl TuiState { targets, recent_events: VecDeque::with_capacity(RECENT_EVENT_LIMIT), last_warning: None, + dropped_events: 0, graph_metric: GraphMetric::EffectiveRtt, graph_viewport: GraphViewport::default(), view: TuiView::Graph, @@ -386,6 +388,17 @@ impl TuiState { self.status = status; } + pub(super) fn mark_dropped_events(&mut self, dropped_events: u64) { + if dropped_events == 0 { + return; + } + self.dropped_events = dropped_events; + let warning = + format!("dropped {dropped_events} managed group events; statistics may be incomplete"); + self.last_warning = Some(warning.clone()); + self.push_event(format!("warning {warning}")); + } + pub(super) fn set_error(&mut self, message: String) { self.status = TuiStatus::Error; self.last_warning = Some(message.clone()); @@ -1811,6 +1824,11 @@ fn status_line(state: &TuiState) -> Paragraph<'_> { TuiView::Graph => "g dashboard", TuiView::Dashboard => "g graph", }; + let incomplete = if state.dropped_events > 0 { + format!(" incomplete:dropped={}", state.dropped_events) + } else { + String::new() + }; let controls = if state.graph_viewport.mode == GraphViewportMode::Follow { format!( "q quit | r reset | p pause | {view_hint} | m metric | arrows pan | +/- zoom | 0 reset window" @@ -1821,10 +1839,11 @@ fn status_line(state: &TuiState) -> Paragraph<'_> { ) }; let lines = vec![Line::from(format!( - "{}{}{} | {}", + "{}{}{}{} | {}", state.status.label(), paused, quitting, + incomplete, controls ))]; Paragraph::new(lines).block(Block::default().borders(Borders::ALL)) @@ -2451,6 +2470,23 @@ mod tests { .contains("warning WrongToken: wrong token")); } + #[test] + fn dropped_events_mark_tui_statistics_incomplete() { + let mut state = TuiState::default(); + + state.mark_dropped_events(9); + + assert_eq!(state.dropped_events, 9); + assert!(state + .last_warning + .as_deref() + .is_some_and(|warning| warning.contains("statistics may be incomplete"))); + assert!(state + .recent_events + .back() + .is_some_and(|event| event.contains("dropped 9 managed group events"))); + } + #[test] fn recent_event_buffer_stays_bounded() { let mut state = TuiState::default(); diff --git a/crates/irtt-client/src/managed/hub.rs b/crates/irtt-client/src/managed/hub.rs index 5be63c1..8f52c05 100644 --- a/crates/irtt-client/src/managed/hub.rs +++ b/crates/irtt-client/src/managed/hub.rs @@ -73,7 +73,9 @@ struct HubState { /// /// Each subscription has its own bounded queue configured by /// [`SubscriberConfig`]. The configured [`SubscriberOverflow`] policy applies -/// when that queue is full. Dropping the handle unregisters the subscriber. +/// when that queue is full. Use [`dropped_events`](Self::dropped_events) to +/// detect incomplete consumption. Dropping the handle unregisters the +/// subscriber. #[must_use = "dropping the subscription unregisters it"] #[derive(Debug)] pub struct EventSubscription { @@ -93,6 +95,7 @@ struct SubscriberInner { struct SubscriberState { queue: VecDeque, connected: bool, + dropped_events: u64, } impl Clone for EventHub { @@ -140,6 +143,7 @@ impl EventHub { state: Mutex::new(SubscriberState { queue: VecDeque::with_capacity(config.capacity), connected: true, + dropped_events: 0, }), available: Condvar::new(), config, @@ -266,6 +270,19 @@ impl EventSubscription { } Ok(None) } + + /// Return the number of events discarded by this subscription's overflow + /// policy. + /// + /// The count is monotonic for the lifetime of the subscription and remains + /// available after the hub is sealed or the subscription disconnects. + pub fn dropped_events(&self) -> u64 { + self.inner + .state + .lock() + .expect("subscriber mutex poisoned") + .dropped_events + } } impl Drop for EventSubscription { @@ -295,14 +312,22 @@ impl SubscriberInner { } match self.config.overflow { - SubscriberOverflow::DropNewest => true, + SubscriberOverflow::DropNewest => { + state.dropped_events = state.dropped_events.saturating_add(1); + true + } SubscriberOverflow::DropOldest => { state.queue.pop_front(); state.queue.push_back(event); + state.dropped_events = state.dropped_events.saturating_add(1); self.available.notify_one(); true } SubscriberOverflow::Disconnect => { + let discarded = u64::try_from(state.queue.len()) + .unwrap_or(u64::MAX) + .saturating_add(1); + state.dropped_events = state.dropped_events.saturating_add(discarded); state.queue.clear(); state.connected = false; self.available.notify_all(); @@ -369,20 +394,23 @@ mod tests { SubscriberOverflow::DropNewest, Ok(Some("event-1")), Ok(None), + 1, ), ( SubscriberOverflow::DropOldest, Ok(Some("event-2")), Ok(None), + 1, ), ( SubscriberOverflow::Disconnect, Err(EventSubscriptionError::Disconnected), Err(EventSubscriptionError::Disconnected), + 2, ), ]; - for (overflow, first_expected, second_expected) in cases { + for (overflow, first_expected, second_expected, expected_drops) in cases { let hub = EventHub::new(); let sub = hub .subscribe(SubscriberConfig { @@ -396,9 +424,32 @@ mod tests { assert_next(&sub, first_expected); assert_next(&sub, second_expected); + assert_eq!(sub.dropped_events(), expected_drops); } } + #[test] + fn drop_oldest_counter_accumulates_and_survives_hub_seal() { + let hub = EventHub::new(); + let sub = hub + .subscribe(SubscriberConfig { + capacity: 2, + overflow: SubscriberOverflow::DropOldest, + }) + .unwrap(); + + for n in 1..=5 { + hub.publish(event(n)); + } + hub.disconnect_all(); + + assert_eq!(sub.dropped_events(), 3); + assert_next(&sub, Ok(Some("event-4"))); + assert_next(&sub, Ok(Some("event-5"))); + assert_next(&sub, Err(EventSubscriptionError::Disconnected)); + assert_eq!(sub.dropped_events(), 3); + } + #[test] fn full_subscriber_does_not_prevent_other_delivery() { let hub = EventHub::new();