From 6adec064425376abf94be5106e6f16ff0a1cb851 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Jul 2026 00:31:36 +0700 Subject: [PATCH] fix(managed): expose dropped subscription events Track a monotonic discarded-event count per bounded subscription for all overflow policies, including after hub sealing or disconnection. Preserve nonblocking publication and existing queue policies. Warn once in streaming CLI output and mark TUI summaries/status as incomplete when DropOldest discarded events. Add exact overflow-counter and incomplete-marker regressions. --- crates/irtt-cli/src/cmd/client/run.rs | 19 +++++++++ crates/irtt-cli/src/cmd/tui/run.rs | 1 + crates/irtt-cli/src/cmd/tui/ui.rs | 38 +++++++++++++++++- crates/irtt-client/src/managed/hub.rs | 57 +++++++++++++++++++++++++-- 4 files changed, 111 insertions(+), 4 deletions(-) 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();