diff --git a/Cargo.lock b/Cargo.lock index ff1eba29fe..6a9ecdf8f9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2700,6 +2700,7 @@ dependencies = [ "postage", "project", "serde", + "serde_json", "settings", "telemetry", "util", @@ -3526,6 +3527,7 @@ dependencies = [ "serde_json", "settings", "smallvec", + "smol", "telemetry", "theme", "theme_settings", diff --git a/crates/call/Cargo.toml b/crates/call/Cargo.toml index eb9e3c8d86..eafb713d7e 100644 --- a/crates/call/Cargo.toml +++ b/crates/call/Cargo.toml @@ -51,6 +51,7 @@ fs = { workspace = true, features = ["test-support"] } gpui = { workspace = true, features = ["test-support"] } language = { workspace = true, features = ["test-support"] } project = { workspace = true, features = ["test-support"] } +serde_json.workspace = true util = { workspace = true, features = ["test-support"] } workspace = { workspace = true, features = ["test-support"] } diff --git a/crates/call/src/call_impl/diagnostics.rs b/crates/call/src/call_impl/diagnostics.rs index 1aa1774dfb..9410151049 100644 --- a/crates/call/src/call_impl/diagnostics.rs +++ b/crates/call/src/call_impl/diagnostics.rs @@ -1,33 +1,91 @@ -use gpui::{Context, Task, WeakEntity}; -use livekit_client::ConnectionQuality; -use std::time::Duration; +use gpui::{Context, Entity, Task, WeakEntity}; +use livekit_client::{ConnectionQuality, RemoteAudioPlaybackStats}; +use serde::{Serialize, Serializer}; +use std::{ + collections::{HashMap, VecDeque}, + sync::Arc, + time::{Duration, Instant}, +}; use super::room::Room; -#[derive(Clone, Default)] +const MAX_HISTORY_SAMPLES: usize = 60; + +#[derive(Copy, Clone, Default)] +pub struct DurationDTO(pub Duration); + +#[derive(Copy, Clone)] +pub struct ConnectionQualityDTO(pub ConnectionQuality); + +#[derive(Clone, Default, Serialize)] pub struct CallStats { - pub connection_quality: Option, - pub effective_quality: Option, + pub connection_quality: Option, + pub effective_quality: Option, pub latency_ms: Option, pub jitter_ms: Option, pub packet_loss_pct: Option, - pub input_lag: Option, + #[serde(rename = "input_lag_ms")] + pub input_lag: Option, +} + +#[derive(Clone, Default, Serialize)] +pub struct RemoteAudioDiagnostics { + pub participant_id: String, + #[serde(skip)] + pub participant_name: String, + pub track_id: String, + pub packets_received: u64, + pub packets_lost: i64, + pub packet_loss_pct: Option, + pub jitter_ms: f64, + pub jitter_buffer_delay_ms: Option, + pub concealed_samples: u64, + pub concealment_events: u64, + pub inserted_samples_for_deceleration: u64, + pub removed_samples_for_acceleration: u64, + pub frames_received: u64, + pub frames_dropped: u64, + pub queue_underflows: u64, + pub current_queue_depth: u64, + pub maximum_queue_depth: u64, +} + +#[derive(Clone, Serialize)] +pub struct CallDiagnosticsSnapshot { + #[serde(rename = "elapsed_ms")] + pub elapsed: DurationDTO, + pub stats: CallStats, + pub remote_audio: Vec, +} + +#[derive(Serialize)] +pub struct CallDiagnosticsReport { + schema_version: u32, + samples: Vec>, } pub struct CallDiagnostics { stats: CallStats, + history: VecDeque>, + previous_inbound: HashMap, + participant_ids: HashMap, + track_ids: HashMap, + started_at: Instant, room: WeakEntity, poll_task: Option>, - stats_update_task: Option>, } impl CallDiagnostics { pub fn new(room: WeakEntity, cx: &mut Context) -> Self { let mut this = Self { stats: CallStats::default(), + history: VecDeque::with_capacity(MAX_HISTORY_SAMPLES), + previous_inbound: HashMap::default(), + participant_ids: HashMap::default(), + track_ids: HashMap::default(), + started_at: Instant::now(), room, poll_task: None, - stats_update_task: None, }; this.start_polling(cx); this @@ -37,10 +95,35 @@ impl CallDiagnostics { &self.stats } + pub fn latest(&self) -> Option<&CallDiagnosticsSnapshot> { + self.history.back().map(Arc::as_ref) + } + + pub fn history(&self) -> &VecDeque> { + &self.history + } + + pub fn report(&self) -> CallDiagnosticsReport { + CallDiagnosticsReport { + schema_version: 1, + samples: self.history.iter().cloned().collect(), + } + } + fn start_polling(&mut self, cx: &mut Context) { self.poll_task = Some(cx.spawn(async move |this, cx| { loop { - if this.update(cx, |this, cx| this.poll_stats(cx)).is_err() { + let Ok(Some(room)) = this.update(cx, |this, _| this.room.upgrade()) else { + break; + }; + let Ok(poll_task) = this.update(cx, |this, cx| this.poll_stats(room, cx)) else { + break; + }; + let result = poll_task.await; + if this + .update(cx, |this, cx| this.apply_poll_result(result, cx)) + .is_err() + { break; } cx.background_executor().timer(Duration::from_secs(1)).await; @@ -48,41 +131,136 @@ impl CallDiagnostics { })); } - fn poll_stats(&mut self, cx: &mut Context) { - let Some(room) = self.room.upgrade() else { - return; - }; - + fn poll_stats(&mut self, room: Entity, cx: &mut Context) -> Task { let connection_quality = room.read(cx).connection_quality(); - self.stats.connection_quality = Some(connection_quality); - self.stats.input_lag = room.read(cx).input_lag(); - + let input_lag = room.read(cx).input_lag().map(DurationDTO); let stats_future = room.read(cx).get_stats(cx); - let background_task = cx.background_executor().spawn(async move { - let session_stats = stats_future.await; - session_stats.map(|stats| compute_network_stats(&stats)) - }); + let mut remote_tracks = Vec::new(); + for (user_id, participant) in room.read(cx).remote_participants() { + let next_participant_number = self.participant_ids.len() + 1; + let participant_id = self + .participant_ids + .entry(*user_id) + .or_insert_with(|| format!("participant-{next_participant_number}")) + .clone(); + for (track_id, (track, stream)) in &participant.audio_tracks { + let track_id = track_id.to_string(); + let next_track_number = self.track_ids.len() + 1; + let track_id = self + .track_ids + .entry(track_id) + .or_insert_with(|| format!("audio-track-{next_track_number}")) + .clone(); + remote_tracks.push(TrackContext { + participant_id: participant_id.clone(), + participant_name: participant.user.username.to_string(), + track_id, + rtc_track_id: track.rtc_track_id(), + playback: stream.remote_playback_stats().unwrap_or_default(), + }); + } + } - self.stats_update_task = Some(cx.spawn(async move |this, cx| { - let result = background_task.await; - this.update(cx, |this, cx| { - if let Some(computed) = result { - this.stats.latency_ms = computed.latency_ms; - this.stats.jitter_ms = computed.jitter_ms; - this.stats.packet_loss_pct = computed.packet_loss_pct; - } - let quality = this - .stats - .connection_quality - .unwrap_or(ConnectionQuality::Lost); - this.stats.effective_quality = - Some(effective_connection_quality(quality, &this.stats)); - cx.notify(); - }) - .ok(); - })); + let previous_inbound = std::mem::take(&mut self.previous_inbound); + let started_at = self.started_at; + cx.background_executor().spawn(async move { + let Some(session_stats) = stats_future.await else { + return PollResult { + snapshot: None, + previous_inbound, + }; + }; + let computed = compute_network_stats(&session_stats); + let (remote_audio, previous_inbound) = + compute_remote_audio_stats(&session_stats, &remote_tracks, previous_inbound); + let mut stats = CallStats { + connection_quality: Some(ConnectionQualityDTO(connection_quality)), + effective_quality: None, + latency_ms: computed.latency_ms, + jitter_ms: computed.jitter_ms, + packet_loss_pct: computed.packet_loss_pct, + input_lag, + }; + stats.effective_quality = Some(ConnectionQualityDTO(effective_connection_quality( + connection_quality, + &stats, + ))); + PollResult { + snapshot: Some(CallDiagnosticsSnapshot { + elapsed: DurationDTO(started_at.elapsed()), + stats, + remote_audio, + }), + previous_inbound, + } + }) } + + fn apply_poll_result(&mut self, result: PollResult, cx: &mut Context) { + self.previous_inbound = result.previous_inbound; + let Some(snapshot) = result.snapshot else { + return; + }; + self.stats = snapshot.stats.clone(); + if self.history.len() + 1 > MAX_HISTORY_SAMPLES { + self.history.pop_front(); + } + self.history.push_back(Arc::new(snapshot)); + cx.notify(); + } +} + +impl Serialize for DurationDTO { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.0.as_millis().serialize(serializer) + } +} + +impl Serialize for ConnectionQualityDTO { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + match self.0 { + ConnectionQuality::Excellent => "excellent", + ConnectionQuality::Good => "good", + ConnectionQuality::Poor => "poor", + ConnectionQuality::Lost => "lost", + } + .serialize(serializer) + } +} + +#[cfg_attr(any(test, feature = "test-support"), allow(dead_code))] +struct TrackContext { + participant_id: String, + participant_name: String, + track_id: String, + rtc_track_id: String, + playback: RemoteAudioPlaybackStats, +} + +struct PollResult { + snapshot: Option, + previous_inbound: HashMap, +} + +#[derive(Clone, Copy)] +#[cfg_attr(any(test, feature = "test-support"), allow(dead_code))] +struct InboundCounters { + packets_received: u64, + packets_lost: i64, + jitter_buffer_delay: f64, + jitter_buffer_emitted_count: u64, + concealed_samples: u64, + concealment_events: u64, + inserted_samples_for_deceleration: u64, + removed_samples_for_acceleration: u64, + playback: RemoteAudioPlaybackStats, } struct ComputedNetworkStats { @@ -126,6 +304,159 @@ fn compute_network_stats(stats: &livekit_client::SessionStats) -> ComputedNetwor } } +#[cfg(all( + not(rust_analyzer), + any( + test, + feature = "test-support", + all(target_os = "windows", target_env = "gnu"), + target_os = "freebsd" + ) +))] +fn compute_remote_audio_stats( + _stats: &livekit_client::SessionStats, + _remote_tracks: &[TrackContext], + previous_inbound: HashMap, +) -> ( + Vec, + HashMap, +) { + (Vec::new(), previous_inbound) +} + +#[cfg(any( + rust_analyzer, + not(any( + test, + feature = "test-support", + all(target_os = "windows", target_env = "gnu"), + target_os = "freebsd" + )) +))] +fn compute_remote_audio_stats( + stats: &livekit_client::SessionStats, + remote_tracks: &[TrackContext], + previous_inbound: HashMap, +) -> ( + Vec, + HashMap, +) { + use livekit_client::RtcStats; + + let mut diagnostics = Vec::new(); + let mut next_inbound = HashMap::default(); + for stat in &stats.subscriber_stats { + let RtcStats::InboundRtp(inbound) = stat else { + continue; + }; + if inbound.stream.kind != "audio" { + continue; + } + + let Some(track) = remote_tracks + .iter() + .find(|track| track.rtc_track_id == inbound.inbound.track_identifier) + else { + continue; + }; + + let current = InboundCounters { + packets_received: inbound.received.packets_received, + packets_lost: inbound.received.packets_lost, + jitter_buffer_delay: inbound.inbound.jitter_buffer_delay, + jitter_buffer_emitted_count: inbound.inbound.jitter_buffer_emitted_count, + concealed_samples: inbound.inbound.concealed_samples, + concealment_events: inbound.inbound.concealment_events, + inserted_samples_for_deceleration: inbound.inbound.inserted_samples_for_deceleration, + removed_samples_for_acceleration: inbound.inbound.removed_samples_for_acceleration, + playback: track.playback, + }; + let previous = previous_inbound + .get(&inbound.rtc.id) + .copied() + .unwrap_or(current); + next_inbound.insert(inbound.rtc.id.clone(), current); + + let packets_received = current + .packets_received + .checked_sub(previous.packets_received) + .unwrap_or_default(); + let packets_lost = current + .packets_lost + .checked_sub(previous.packets_lost) + .unwrap_or_default(); + + let packet_loss_pct = { + let packets_lost = packets_lost.max(0); + let expected_packets = packets_received as i128 + i128::from(packets_lost); + if expected_packets > 0 { + Some((packets_lost as f64 / expected_packets as f64) * 100.0) + } else { + None + } + }; + + let emitted_count = current + .jitter_buffer_emitted_count + .checked_sub(previous.jitter_buffer_emitted_count) + .unwrap_or_default(); + let jitter_buffer_delay_ms = { + let delay = current.jitter_buffer_delay - previous.jitter_buffer_delay; + if emitted_count > 0 && delay >= 0.0 { + Some((delay / emitted_count as f64) * 1000.0) + } else { + None + } + }; + + diagnostics.push(RemoteAudioDiagnostics { + participant_id: track.participant_id.clone(), + participant_name: track.participant_name.clone(), + track_id: track.track_id.clone(), + packets_received, + packets_lost, + packet_loss_pct, + jitter_ms: inbound.received.jitter * 1000.0, + jitter_buffer_delay_ms, + concealed_samples: current + .concealed_samples + .checked_sub(previous.concealed_samples) + .unwrap_or_default(), + concealment_events: current + .concealment_events + .checked_sub(previous.concealment_events) + .unwrap_or_default(), + inserted_samples_for_deceleration: current + .inserted_samples_for_deceleration + .checked_sub(previous.inserted_samples_for_deceleration) + .unwrap_or_default(), + removed_samples_for_acceleration: current + .removed_samples_for_acceleration + .checked_sub(previous.removed_samples_for_acceleration) + .unwrap_or_default(), + frames_received: current + .playback + .frames_received + .checked_sub(previous.playback.frames_received) + .unwrap_or_default(), + frames_dropped: current + .playback + .frames_dropped + .checked_sub(previous.playback.frames_dropped) + .unwrap_or_default(), + queue_underflows: current + .playback + .queue_underflows + .checked_sub(previous.playback.queue_underflows) + .unwrap_or_default(), + current_queue_depth: current.playback.current_queue_depth, + maximum_queue_depth: current.playback.maximum_queue_depth, + }); + } + + (diagnostics, next_inbound) +} + #[cfg(all( not(rust_analyzer), any( @@ -224,9 +555,37 @@ fn effective_connection_quality( worst = worst.max(metric_quality(loss, 1.0, 5.0)); } if let Some(lag) = stats.input_lag { - let lag_ms = lag.as_secs_f64() * 1000.0; + let lag_ms = lag.0.as_secs_f64() * 1000.0; worst = worst.max(metric_quality(lag_ms, 20.0, 50.0)); } worst } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn report_redacts_participant_names() -> anyhow::Result<()> { + let report = CallDiagnosticsReport { + schema_version: 1, + samples: vec![Arc::new(CallDiagnosticsSnapshot { + elapsed: DurationDTO(Duration::from_secs(1)), + stats: CallStats::default(), + remote_audio: vec![RemoteAudioDiagnostics { + participant_id: "participant-1".to_string(), + participant_name: "private-username".to_string(), + track_id: "audio-track-1".to_string(), + ..Default::default() + }], + })], + }; + + let serialized = serde_json::to_string(&report)?; + assert!(!serialized.contains("private-username")); + assert!(serialized.contains("participant-1")); + assert!(serialized.contains("audio-track-1")); + Ok(()) + } +} diff --git a/crates/call/src/call_impl/mod.rs b/crates/call/src/call_impl/mod.rs index 702242b133..f7fa9d3b7d 100644 --- a/crates/call/src/call_impl/mod.rs +++ b/crates/call/src/call_impl/mod.rs @@ -391,6 +391,7 @@ pub struct IncomingCall { /// Singleton global maintaining the user's participation in a room across workspaces. pub struct ActiveCall { room: Option<(Entity, Vec)>, + last_call_diagnostics: Option>, pending_room_creation: Option, Arc>>>>, location: Option>, _join_debouncer: OneAtATime, @@ -410,6 +411,7 @@ impl ActiveCall { fn new(client: Arc, user_store: Entity, cx: &mut Context) -> Self { Self { room: None, + last_call_diagnostics: None, pending_room_creation: None, location: None, pending_invites: Default::default(), @@ -690,6 +692,7 @@ impl ActiveCall { Audio::end_call(cx); let channel_id = self.channel_id(cx); + self.retain_room_diagnostics(cx); if let Some((room, _)) = self.room.take() { cx.emit(Event::RoomLeft { channel_id }); room.update(cx, |room, cx| room.leave(cx)) @@ -744,6 +747,7 @@ impl ActiveCall { Task::ready(Ok(())) } else { cx.notify(); + self.retain_room_diagnostics(cx); if let Some(room) = room { if room.read(cx).status().is_offline() { self.room = None; @@ -779,6 +783,23 @@ impl ActiveCall { self.room.as_ref().map(|(room, _)| room) } + pub fn call_diagnostics(&self, cx: &App) -> Option> { + self.room() + .and_then(|room| room.read(cx).diagnostics()) + .cloned() + .or_else(|| self.last_call_diagnostics.clone()) + } + + fn retain_room_diagnostics(&mut self, cx: &App) { + if let Some(diagnostics) = self + .room() + .and_then(|room| room.read(cx).diagnostics()) + .cloned() + { + self.last_call_diagnostics = Some(diagnostics); + } + } + pub fn client(&self) -> Arc { self.client.clone() } diff --git a/crates/collab_ui/Cargo.toml b/crates/collab_ui/Cargo.toml index 920f620e0e..80ade0d74a 100644 --- a/crates/collab_ui/Cargo.toml +++ b/crates/collab_ui/Cargo.toml @@ -50,6 +50,7 @@ serde.workspace = true serde_json.workspace = true settings.workspace = true smallvec.workspace = true +smol.workspace = true telemetry.workspace = true theme.workspace = true theme_settings.workspace = true diff --git a/crates/collab_ui/src/call_stats_modal.rs b/crates/collab_ui/src/call_stats_modal.rs index cfbdf82c3b..99293bcd7f 100644 --- a/crates/collab_ui/src/call_stats_modal.rs +++ b/crates/collab_ui/src/call_stats_modal.rs @@ -1,13 +1,24 @@ -use call::{ActiveCall, Room, room}; +use anyhow::Result; +use call::{ + ActiveCall, + diagnostics::{CallDiagnostics, RemoteAudioDiagnostics}, + room, +}; use gpui::{ - DismissEvent, Entity, EventEmitter, FocusHandle, Focusable, FontWeight, Render, Subscription, - Window, + ClipboardItem, DismissEvent, Entity, EventEmitter, FocusHandle, Focusable, FontWeight, Render, + Subscription, Task, TaskExt as _, Window, }; use livekit_client::ConnectionQuality; +use release_channel::{AppVersion, ReleaseChannel}; +use serde::Serialize; +use std::{cmp::Reverse, path::PathBuf}; use ui::prelude::*; use workspace::{ModalView, Workspace}; use zed_actions::ShowCallStats; +const WEBRTC_AUDIO_SAMPLES_PER_MILLISECOND: f64 = 48.0; +const PLAYBACK_FRAME_DURATION_MILLISECONDS: u64 = 10; + pub fn init(cx: &mut App) { cx.observe_new(|workspace: &mut Workspace, _, _cx| { workspace.register_action(|workspace, _: &ShowCallStats, window, cx| { @@ -23,6 +34,15 @@ pub struct CallStatsModal { _diagnostics_subscription: Option, } +#[derive(Serialize)] +struct ExportedCallDiagnostics { + application_version: String, + release_channel: String, + operating_system: &'static str, + architecture: &'static str, + diagnostics: call::diagnostics::CallDiagnosticsReport, +} + impl CallStatsModal { fn new(cx: &mut Context) -> Self { let mut this = Self { @@ -41,7 +61,7 @@ impl CallStatsModal { } fn observe_diagnostics(&mut self, cx: &mut Context) { - let diagnostics = active_room(cx).and_then(|room| room.read(cx).diagnostics().cloned()); + let diagnostics = call_diagnostics(cx); if let Some(diagnostics) = diagnostics { self._diagnostics_subscription = Some(cx.observe(&diagnostics, |_, _, cx| cx.notify())); @@ -61,7 +81,7 @@ impl CallStatsModal { self.observe_diagnostics(cx); } room::Event::RoomLeft { .. } => { - self._diagnostics_subscription = None; + self.observe_diagnostics(cx); cx.notify(); } _ => {} @@ -71,10 +91,66 @@ impl CallStatsModal { fn dismiss(&mut self, _: &menu::Cancel, _: &mut Window, cx: &mut Context) { cx.emit(DismissEvent); } + + fn report(cx: &App) -> Option { + let diagnostics = call_diagnostics(cx)?.read(cx).report(); + let release_channel = ReleaseChannel::try_global(cx) + .map(|channel| channel.dev_name().to_string()) + .unwrap_or_else(|| "unknown".to_string()); + Some(ExportedCallDiagnostics { + application_version: AppVersion::global(cx).to_string(), + release_channel, + operating_system: std::env::consts::OS, + architecture: std::env::consts::ARCH, + diagnostics, + }) + } + + fn serialize_report(cx: &App) -> Option>> { + let report = Self::report(cx)?; + Some( + cx.background_executor() + .spawn(async move { Ok(serde_json::to_string_pretty(&report)?) }), + ) + } + + fn copy_report(&mut self, cx: &mut Context) { + let Some(report) = Self::serialize_report(cx) else { + return; + }; + cx.spawn(async move |_, cx| { + let report = report.await?; + cx.update(|cx| { + cx.write_to_clipboard(ClipboardItem::new_string(report)); + }); + anyhow::Ok(()) + }) + .detach_and_log_err(cx); + } + + fn save_report(&mut self, cx: &mut Context) { + let Some(report) = Self::report(cx) else { + return; + }; + let save_dialog = + cx.prompt_for_new_path(&PathBuf::default(), Some("zed-call-diagnostics.json")); + cx.spawn(async move |_, cx| { + let Some(path) = save_dialog.await?? else { + return anyhow::Ok(()); + }; + cx.background_spawn(async move { + let report = serde_json::to_string_pretty(&report)?; + smol::fs::write(path, report).await?; + anyhow::Ok(()) + }) + .await + }) + .detach_and_log_err(cx); + } } -fn active_room(cx: &App) -> Option> { - ActiveCall::try_global(cx)?.read(cx).room().cloned() +fn call_diagnostics(cx: &App) -> Option> { + ActiveCall::try_global(cx)?.read(cx).call_diagnostics(cx) } fn quality_label(quality: Option) -> (&'static str, Color) { @@ -111,10 +187,10 @@ fn metric_rating(label: &str, value_ms: f64) -> (&'static str, Color) { } } -fn input_lag_rating(value_ms: f64) -> (&'static str, Color) { - if value_ms < 20.0 { +fn input_lag_rating(value_ms: u128) -> (&'static str, Color) { + if value_ms < 20 { ("Normal", Color::Success) - } else if value_ms < 50.0 { + } else if value_ms < 50 { ("High", Color::Warning) } else { ("Poor", Color::Error) @@ -131,6 +207,26 @@ fn packet_loss_rating(loss_pct: f64) -> (&'static str, Color) { } } +fn audio_issue_score(audio: &RemoteAudioDiagnostics) -> u64 { + audio + .concealment_events + .saturating_add(audio.frames_dropped) + .saturating_add(audio.queue_underflows) + .saturating_add(u64::from( + audio.packet_loss_pct.is_some_and(|loss| loss >= 1.0), + )) +} + +fn format_audio_duration(milliseconds: f64) -> String { + if milliseconds > 0.0 && milliseconds < 1.0 { + "<1ms".to_string() + } else if milliseconds < 10.0 && milliseconds.fract() != 0.0 { + format!("{milliseconds:.1}ms") + } else { + format!("{milliseconds:.0}ms") + } +} + impl EventEmitter for CallStatsModal {} impl ModalView for CallStatsModal {} @@ -142,23 +238,59 @@ impl Focusable for CallStatsModal { impl Render for CallStatsModal { fn render(&mut self, _window: &mut Window, cx: &mut Context) -> impl IntoElement { - let room = active_room(cx); - let is_connected = room.is_some(); - let stats = room - .and_then(|room| { - let diagnostics = room.read(cx).diagnostics()?; - Some(diagnostics.read(cx).stats().clone()) + let is_connected = ActiveCall::try_global(cx) + .is_some_and(|active_call| active_call.read(cx).room().is_some()); + let diagnostics = call_diagnostics(cx); + let (latest, sample_count, retained_duration, recent_issue_count) = diagnostics + .as_ref() + .map(|diagnostics| { + let diagnostics = diagnostics.read(cx); + let retained_duration = diagnostics + .history() + .front() + .zip(diagnostics.history().back()) + .and_then(|(first, last)| last.elapsed.0.checked_sub(first.elapsed.0)) + .unwrap_or_default(); + let recent_issue_count = diagnostics + .history() + .iter() + .rev() + .take(60) + .filter(|snapshot| { + snapshot + .remote_audio + .iter() + .any(|audio| audio_issue_score(audio) > 0) + }) + .count(); + ( + diagnostics.latest().cloned(), + diagnostics.history().len(), + retained_duration, + recent_issue_count, + ) }) .unwrap_or_default(); + let stats = latest + .as_ref() + .map(|snapshot| snapshot.stats.clone()) + .unwrap_or_default(); + let mut remote_audio = latest + .map(|snapshot| snapshot.remote_audio) + .unwrap_or_default(); + remote_audio.sort_by_key(|audio| Reverse(audio_issue_score(audio))); - let (quality_text, quality_color) = quality_label(stats.connection_quality); + let (quality_text, quality_color) = + quality_label(stats.connection_quality.map(|inner| inner.0)); + let has_diagnostics = sample_count > 0; v_flex() .key_context("CallStatsModal") .on_action(cx.listener(Self::dismiss)) .track_focus(&self.focus_handle) .elevation_3(cx) - .w(rems(24.)) + .w(rems(36.)) + .max_h(rems(42.)) .p_4() .gap_3() .child( @@ -171,64 +303,205 @@ impl Render for CallStatsModal { .color(quality_color), ), ) - .when(!is_connected, |this| { + .when(!is_connected && has_diagnostics, |this| { + this.child( + h_flex() + .justify_center() + .child(Label::new("Showing diagnostics from the most recent call").color(Color::Muted)), + ) + }) + .when(!has_diagnostics, |this| { this.child( h_flex() .justify_center() .py_4() - .child(Label::new("Not in a call").color(Color::Muted)), + .child(Label::new("No call diagnostics available").color(Color::Muted)), ) }) - .when(is_connected, |this| { + .when(has_diagnostics, |this| { this.child( v_flex() - .gap_1() + .id("call-diagnostics-scroll") + .gap_3() + .max_h(rems(32.)) + .overflow_y_scroll() .child( - h_flex() - .gap_2() - .child(Label::new("Network").weight(FontWeight::SEMIBOLD)), + Label::new(format!( + "{sample_count} samples · {:.0}s retained · {recent_issue_count} affected intervals in the last 60s", + retained_duration.as_secs_f64() + )) + .size(LabelSize::Small) + .color(Color::Muted), ) - .child(self.render_metric_row( - "Latency", - "Time for data to travel to the server", - stats.latency_ms, - |v| format!("{:.0}ms", v), - |v| metric_rating("Latency", v), - )) - .child(self.render_metric_row( - "Jitter", - "Variance or fluctuation in latency", - stats.jitter_ms, - |v| format!("{:.0}ms", v), - |v| metric_rating("Jitter", v), - )) - .child(self.render_metric_row( - "Packet loss", - "Amount of data lost during transfer", - stats.packet_loss_pct, - |v| format!("{:.1}%", v), - |v| packet_loss_rating(v), - )) - .child(self.render_metric_row( - "Input lag", - "Delay from audio capture to WebRTC", - stats.input_lag.map(|d| d.as_secs_f64() * 1000.0), - |v| format!("{:.1}ms", v), - |v| input_lag_rating(v), - )), + .child( + v_flex() + .gap_1() + .child(Label::new("Network").weight(FontWeight::SEMIBOLD)) + .child(self.render_metric_row( + "Latency", + "Time for data to travel to the server", + stats.latency_ms, + |v| format!("{:.0}ms", v), + |v| metric_rating("Latency", v), + )) + .child(self.render_metric_row( + "Jitter", + "Variance or fluctuation in latency", + stats.jitter_ms, + |v| format!("{:.0}ms", v), + |v| metric_rating("Jitter", v), + )) + .child(self.render_metric_row( + "Packet loss", + "Amount of data lost during transfer", + stats.packet_loss_pct, + |v| format!("{:.1}%", v), + packet_loss_rating, + )) + .child(self.render_metric_row( + "Input lag", + "Delay from audio capture to WebRTC", + stats.input_lag.map(|d| d.0.as_millis()), + |v| format!("{}ms", v), + input_lag_rating, + )), + ) + .child( + v_flex() + .gap_1() + .child(Label::new("Inbound audio").weight(FontWeight::SEMIBOLD)) + .when(remote_audio.is_empty(), |this| { + this.child( + Label::new("Waiting for inbound audio statistics") + .color(Color::Muted), + ) + }) + .children( + remote_audio + .into_iter() + .map(|audio| self.render_remote_audio(audio)), + ), + ), + ) + }) + .when(has_diagnostics, |this| { + this.child( + h_flex() + .justify_end() + .gap_2() + .child( + Button::new("copy-call-diagnostics", "Copy Report") + .on_click(cx.listener(|this, _, _, cx| this.copy_report(cx))), + ) + .child( + Button::new("save-call-diagnostics", "Save Report…") + .on_click(cx.listener(|this, _, _, cx| this.save_report(cx))), + ), ) }) } } impl CallStatsModal { - fn render_metric_row( + fn render_remote_audio(&self, audio: RemoteAudioDiagnostics) -> impl IntoElement { + let issue_score = audio_issue_score(&audio); + let (status, color) = if issue_score > 0 { + ("Affected", Color::Warning) + } else { + ("Healthy", Color::Success) + }; + let packet_loss = audio + .packet_loss_pct + .map(|loss| format!("{loss:.1}%")) + .unwrap_or_else(|| "—".to_string()); + let jitter_buffer_delay = audio + .jitter_buffer_delay_ms + .map(|delay| format!("{delay:.1}ms")) + .unwrap_or_else(|| "—".to_string()); + let repaired_audio_duration = format_audio_duration( + audio.concealed_samples as f64 / WEBRTC_AUDIO_SAMPLES_PER_MILLISECOND, + ); + let starved_audio_duration = format_audio_duration( + audio + .queue_underflows + .saturating_mul(PLAYBACK_FRAME_DURATION_MILLISECONDS) as f64, + ); + let dropped_audio_duration = format_audio_duration( + audio + .frames_dropped + .saturating_mul(PLAYBACK_FRAME_DURATION_MILLISECONDS) as f64, + ); + let buffered_audio_duration = format_audio_duration( + audio + .current_queue_depth + .saturating_mul(PLAYBACK_FRAME_DURATION_MILLISECONDS) as f64, + ); + let peak_buffered_audio_duration = format_audio_duration( + audio + .maximum_queue_depth + .saturating_mul(PLAYBACK_FRAME_DURATION_MILLISECONDS) as f64, + ); + let repair_event_label = if audio.concealment_events == 1 { + "event" + } else { + "events" + }; + + v_flex() + .px_2() + .py_1() + .gap_1() + .rounded_md() + .child( + h_flex() + .justify_between() + .child( + h_flex() + .gap_1() + .child(Label::new(audio.participant_name)) + .child( + Label::new(format!( + "{} · {}", + audio.participant_id, audio.track_id + )) + .size(LabelSize::Small) + .color(Color::Muted), + ), + ) + .child(Label::new(status).color(color)), + ) + .child( + Label::new(format!( + "Loss {packet_loss} · jitter {:.1}ms · jitter buffer {jitter_buffer_delay}", + audio.jitter_ms + )) + .size(LabelSize::Small) + .color(Color::Muted), + ) + .child( + Label::new(format!( + "WebRTC repaired {repaired_audio_duration} in {} {repair_event_label}", + audio.concealment_events, + )) + .size(LabelSize::Small) + .color(Color::Muted), + ) + .child( + Label::new(format!( + "Local playback starved for {starved_audio_duration} · dropped {dropped_audio_duration} · buffered {buffered_audio_duration} (peak {peak_buffered_audio_duration})", + )) + .size(LabelSize::Small) + .color(Color::Muted), + ) + } + + fn render_metric_row( &self, title: &str, description: &str, - value: Option, - format_value: impl Fn(f64) -> String, - rate: impl Fn(f64) -> (&'static str, Color), + value: Option, + format_value: impl Fn(T) -> String, + rate: impl Fn(T) -> (&'static str, Color), ) -> impl IntoElement { let (rating_text, rating_color, value_text) = match value { Some(v) => { diff --git a/crates/livekit_client/src/lib.rs b/crates/livekit_client/src/lib.rs index aa4831562c..709db8112f 100644 --- a/crates/livekit_client/src/lib.rs +++ b/crates/livekit_client/src/lib.rs @@ -5,6 +5,7 @@ use cpal::DeviceId; mod remote_video_track_view; pub use remote_video_track_view::{RemoteVideoTrackView, RemoteVideoTrackViewEvent}; use rodio::DeviceTrait as _; +use serde::{Deserialize, Serialize}; mod record; pub use record::CaptureInput; @@ -75,6 +76,15 @@ pub enum ConnectionQuality { Lost, } +#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)] +pub struct RemoteAudioPlaybackStats { + pub frames_received: u64, + pub frames_dropped: u64, + pub queue_underflows: u64, + pub current_queue_depth: u64, + pub maximum_queue_depth: u64, +} + #[derive(Debug, Clone)] pub enum TrackPublication { Local(LocalTrackPublication), diff --git a/crates/livekit_client/src/livekit_client.rs b/crates/livekit_client/src/livekit_client.rs index 1c1cc5c3b7..fbad39880b 100644 --- a/crates/livekit_client/src/livekit_client.rs +++ b/crates/livekit_client/src/livekit_client.rs @@ -297,6 +297,10 @@ impl RemoteAudioTrack { pub fn sid(&self) -> TrackSid { self.0.sid() } + + pub fn rtc_track_id(&self) -> String { + self.0.rtc_track().id() + } } impl RemoteVideoTrack { diff --git a/crates/livekit_client/src/livekit_client/playback.rs b/crates/livekit_client/src/livekit_client/playback.rs index bed43d22d6..f00b64768e 100644 --- a/crates/livekit_client/src/livekit_client/playback.rs +++ b/crates/livekit_client/src/livekit_client/playback.rs @@ -31,6 +31,8 @@ use std::time::{Duration, Instant}; use std::{borrow::Cow, collections::VecDeque, sync::Arc}; use util::{ResultExt as _, maybe}; +use crate::RemoteAudioPlaybackStats; + struct TimestampedFrame { frame: AudioFrame<'static>, captured_at: Instant, @@ -91,6 +93,7 @@ impl AudioStack { sample_rate: SAMPLE_RATE.get(), num_channels: CHANNEL_COUNT.get() as u32, buffer: Arc::default(), + diagnostics: Arc::default(), }; self.mixer.lock().add_source(source.clone()); @@ -109,6 +112,7 @@ impl AudioStack { } }); + let diagnostics = source.diagnostics.clone(); let mixer = self.mixer.clone(); let on_drop = util::defer(move || { mixer.lock().remove_source(source.ssrc); @@ -116,8 +120,9 @@ impl AudioStack { drop(output_task); }); - AudioStream::Output { + AudioStream { _drop: Box::new(on_drop), + remote_playback_diagnostics: Some(diagnostics), } } @@ -211,8 +216,9 @@ impl AudioStack { }); Ok(( super::LocalAudioTrack(track), - AudioStream::Output { + AudioStream { _drop: Box::new(on_drop), + remote_playback_diagnostics: None, }, input_lag_us, )) @@ -417,9 +423,17 @@ pub struct Speaker { use super::LocalVideoTrack; -pub enum AudioStream { - Input { _task: Task<()> }, - Output { _drop: Box }, +pub struct AudioStream { + _drop: Box, + remote_playback_diagnostics: Option>, +} + +impl AudioStream { + pub fn remote_playback_stats(&self) -> Option { + self.remote_playback_diagnostics + .as_ref() + .map(|diagnostics| diagnostics.snapshot()) + } } pub(crate) async fn capture_local_video_track( @@ -468,6 +482,28 @@ struct AudioMixerSource { sample_rate: u32, num_channels: u32, buffer: Arc>>>, + diagnostics: Arc, +} + +#[derive(Default)] +struct RemoteAudioPlaybackCounters { + frames_received: AtomicU64, + frames_dropped: AtomicU64, + queue_underflows: AtomicU64, + current_queue_depth: AtomicU64, + maximum_queue_depth: AtomicU64, +} + +impl RemoteAudioPlaybackCounters { + fn snapshot(&self) -> RemoteAudioPlaybackStats { + RemoteAudioPlaybackStats { + frames_received: self.frames_received.load(Ordering::Relaxed), + frames_dropped: self.frames_dropped.load(Ordering::Relaxed), + queue_underflows: self.queue_underflows.load(Ordering::Relaxed), + current_queue_depth: self.current_queue_depth.load(Ordering::Relaxed), + maximum_queue_depth: self.maximum_queue_depth.load(Ordering::Relaxed), + } + } } impl AudioMixerSource { @@ -479,8 +515,23 @@ impl AudioMixerSource { let mut buffer = self.buffer.lock(); buffer.push_back(frame.data.to_vec()); + self.diagnostics + .frames_received + .fetch_add(1, Ordering::Relaxed); while buffer.len() > 10 { buffer.pop_front(); + self.diagnostics + .frames_dropped + .fetch_add(1, Ordering::Relaxed); + } + let queue_depth = buffer.len() as u64; + self.diagnostics + .current_queue_depth + .store(queue_depth, Ordering::Relaxed); + if queue_depth > self.diagnostics.maximum_queue_depth.load(Ordering::Relaxed) { + self.diagnostics + .maximum_queue_depth + .fetch_max(queue_depth, Ordering::Relaxed); } } } @@ -496,7 +547,16 @@ impl libwebrtc::native::audio_mixer::AudioMixerSource for AudioMixerSource { fn get_audio_frame_with_info<'a>(&self, target_sample_rate: u32) -> Option> { assert_eq!(self.sample_rate, target_sample_rate); - let buf = self.buffer.lock().pop_front()?; + let mut buffer = self.buffer.lock(); + let Some(buf) = buffer.pop_front() else { + self.diagnostics + .queue_underflows + .fetch_add(1, Ordering::Relaxed); + return None; + }; + self.diagnostics + .current_queue_depth + .store(buffer.len() as u64, Ordering::Relaxed); Some(AudioFrame { data: Cow::Owned(buf), sample_rate: self.sample_rate, diff --git a/crates/livekit_client/src/mock_client.rs b/crates/livekit_client/src/mock_client.rs index d1cd399d25..fff81d341a 100644 --- a/crates/livekit_client/src/mock_client.rs +++ b/crates/livekit_client/src/mock_client.rs @@ -19,6 +19,12 @@ pub use test::{ConnectionState, ParticipantIdentity, RtcStats, SessionStats, Tra pub struct AudioStream {} +impl AudioStream { + pub fn remote_playback_stats(&self) -> Option { + None + } +} + #[cfg(not(target_os = "macos"))] pub type RemoteVideoFrame = std::sync::Arc; diff --git a/crates/livekit_client/src/mock_client/track.rs b/crates/livekit_client/src/mock_client/track.rs index e1565e8d71..1557967925 100644 --- a/crates/livekit_client/src/mock_client/track.rs +++ b/crates/livekit_client/src/mock_client/track.rs @@ -28,6 +28,10 @@ impl RemoteAudioTrack { self.server_track.sid.clone() } + pub fn rtc_track_id(&self) -> String { + self.server_track.sid.to_string() + } + pub fn publisher_id(&self) -> ParticipantIdentity { self.server_track.publisher_id.clone() } diff --git a/crates/title_bar/src/collab.rs b/crates/title_bar/src/collab.rs index 3d498c6fd8..fe418bef36 100644 --- a/crates/title_bar/src/collab.rs +++ b/crates/title_bar/src/collab.rs @@ -27,7 +27,7 @@ use zed_actions::ShowCallStats; use crate::TitleBar; -fn format_stat(value: Option, format: impl Fn(f64) -> String) -> String { +fn format_stat(value: Option, format: impl Fn(T) -> String) -> String { match value { Some(v) => format(v), None => "—".to_string(), @@ -373,7 +373,10 @@ impl TitleBar { .channel_id() .and_then(|channel_id| channel_store.read(cx).channel_for_id(channel_id).cloned()); - let effective_quality = stats.effective_quality.unwrap_or(ConnectionQuality::Lost); + let effective_quality = stats + .effective_quality + .map(|inner| inner.0) + .unwrap_or(ConnectionQuality::Lost); let (signal_icon, signal_color, quality_label) = match effective_quality { ConnectionQuality::Excellent => { (IconName::SignalHigh, Some(Color::Success), "Excellent") @@ -414,8 +417,8 @@ impl TitleBar { let packet_loss = format_stat(stats.packet_loss_pct, |v| format!("{:.1}%", v)); let input_lag = - format_stat(stats.input_lag.map(|d| d.as_secs_f64() * 1000.0), |v| { - format!("{:.1}ms", v) + format_stat(stats.input_lag.map(|d| d.0.as_millis()), |v| { + format!("{}ms", v) }); let key_binding = KeyBinding::for_action(&ShowCallStats, cx);