From 33a7af07d29137baf0e0e112c9fbe28ffed397ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:32:15 +0200 Subject: [PATCH 1/3] feat(telemetry): add device state and RTC stats windowing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `DeviceState` is what the host observes (thermal, power, memory, battery, network, lifecycle); the core derives the `lk.device.*.changed` events, a cadence factor of up to 4× under pressure, and soft and hard upload holds. `DeviceEvent` files audio route changes, interruptions and capture failures with shared enums. RTC stats map one raw `getStats()` report (per track or per peer connection) to samples, fold simulcast layers into one series per track, and window them on device into `lk.rtc.stats.sample`, per track, direction and session. A window closes into a captured record that carries its session's attributes and `session.id` as of that moment. --- livekit-telemetry/src/device.rs | 520 ++++++++++++++++ livekit-telemetry/src/lib.rs | 19 + livekit-telemetry/src/rtc.rs | 1006 +++++++++++++++++++++++++++++++ 3 files changed, 1545 insertions(+) create mode 100644 livekit-telemetry/src/device.rs create mode 100644 livekit-telemetry/src/rtc.rs diff --git a/livekit-telemetry/src/device.rs b/livekit-telemetry/src/device.rs new file mode 100644 index 000000000..d99e03e50 --- /dev/null +++ b/livekit-telemetry/src/device.rs @@ -0,0 +1,520 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::TelemetryEvent; + +/// Thermal pressure as the OS reports it (`ProcessInfo.thermalState`, `PowerManager` thermal +/// status, the web Compute Pressure API — all four share this vocabulary). +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum ThermalState { + /// The platform has no thermal source: no pressure assumed, no change event. + #[default] + Unknown, + Nominal, + Fair, + Serious, + Critical, +} + +/// Whether the app is in front of the user. +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum AppState { + #[default] + Foreground, + Background, +} + +/// Memory pressure as the OS reports it (`DISPATCH_MEMORYPRESSURE_*`, `onTrimMemory` levels: +/// `RUNNING_LOW`/`BACKGROUND` → warning, `RUNNING_CRITICAL`/`COMPLETE` → critical). +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum MemoryPressure { + #[default] + Normal, + Warning, + Critical, +} + +/// The active network path, in OTel `network.connection.type` terms. +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum NetworkType { + #[default] + Unknown, + Wifi, + Cell, + Wired, + /// A VPN tunnel is the active interface (Android `TRANSPORT_VPN`, WebRTC candidate `vpn`). + Vpn, + /// Bluetooth tethering (Android `TRANSPORT_BLUETOOTH`, WebRTC candidate `bluetooth`). + Bluetooth, + /// Connected over something the platform does not name (`NWInterface.InterfaceType.other`, + /// Android `TRANSPORT_USB` / `LOWPAN`, …). + Other, + Unavailable, +} + +/// What the host observes about the device; pushed with +/// [`Telemetry::set_device_state`](crate::Telemetry::set_device_state) whenever anything changes. +/// +/// The host owns the OS APIs (thermal, power, memory, battery, network path, lifecycle) — they +/// are not reachable from Rust without a JVM/ObjC bridge — and the core owns what to do with +/// them: emit the change events from `SPEC.md`, stretch its own cadence and hold uploads so +/// telemetry never competes with the call. CPU is deliberately not measured here (measuring CPU +/// costs CPU): thermal state is the OS's judgement, and WebRTC's own `qualityLimitationReason` +/// (see [`RtcStatsSample`](crate::RtcStatsSample)) says when the encoder is CPU-starved. +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +#[derive(Debug, Clone, Copy, PartialEq, Default)] +pub struct DeviceState { + pub thermal: ThermalState, + /// Low Power Mode / Battery Saver; `None` when the platform has no source (no change event, + /// no stretch). + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub low_power_mode: Option, + pub app_state: AppState, + // No `uniffi(default)` on enum fields: uniffi 0.31's Swift generator rejects `Default` there. + pub memory: MemoryPressure, + pub network: NetworkType, + /// Cellular / personal hotspot (`NWPath.isExpensive`, metered on Android). + #[cfg_attr(feature = "uniffi", uniffi(default = false))] + pub network_expensive: bool, + /// The user asked for less traffic: Low Data Mode (`NWPath.isConstrained`), Data Saver, + /// `navigator.connection.saveData`. + #[cfg_attr(feature = "uniffi", uniffi(default = false))] + pub network_constrained: bool, + /// Battery charge in percent; `None` when unknown (desktops, tvOS, monitoring disabled). + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub battery_level: Option, + #[cfg_attr(feature = "uniffi", uniffi(default = false))] + pub battery_charging: bool, +} + +impl DeviceState { + /// Below this, unplugged, the pipeline still records but stops uploading (a soft hold). + const BATTERY_HOLD_PERCENT: u32 = 10; + /// Below this, unplugged, the cadence doubles. + const BATTERY_LOW_PERCENT: u32 = 20; + + /// Multiplier on the flush interval and stats window: 1× at rest, up to 4× under pressure + /// (60 s → 4 min at the default cadence). Serious thermal, memory + /// warning, low-power mode, background, low battery and a constrained network each double + /// it; critical thermal or memory quadruple it. + pub fn cadence_factor(&self) -> u32 { + let thermal = match self.thermal { + ThermalState::Unknown | ThermalState::Nominal | ThermalState::Fair => 1, + ThermalState::Serious => 2, + ThermalState::Critical => 4, + }; + let memory = match self.memory { + MemoryPressure::Normal => 1, + MemoryPressure::Warning => 2, + MemoryPressure::Critical => 4, + }; + let doubled = [ + self.low_power_mode == Some(true), + self.app_state == AppState::Background, + self.battery_bucket() >= 1, + self.network_constrained, + ] + .iter() + .filter(|&&on| on) + .count() as u32; + ((thermal * memory) << doubled).min(4) + } + + /// A soft hold: the user asked for less traffic (Low Data Mode / Data Saver) or the battery + /// is nearly empty and unplugged. Uploads wait (at most a minute at a time); data keeps + /// flowing into the cache meanwhile. Being offline (`network: Unavailable`) is a hard hold: + /// nothing is attempted until the network is back. + pub fn holds_uploads(&self) -> bool { + self.network_constrained || self.battery_bucket() == 2 + } + + /// 0 = fine or unknown, 1 = low (≤ 20 %), 2 = nearly empty (≤ 10 %); always 0 on charge. + /// Per-percent battery updates only matter when they cross one of these. + fn battery_bucket(&self) -> u8 { + match self.battery_level { + Some(level) if !self.battery_charging && level <= Self::BATTERY_HOLD_PERCENT => 2, + Some(level) if !self.battery_charging && level <= Self::BATTERY_LOW_PERCENT => 1, + _ => 0, + } + } + + /// The `SPEC.md` events describing what changed between `previous` and `self` + /// (everything, when there is no previous state: "initial value at session start"). + pub(crate) fn change_events(&self, previous: Option<&DeviceState>) -> Vec { + let mut events = Vec::new(); + // An unknown reading is no reading: it never makes a change event. + if self.thermal != ThermalState::Unknown + && previous.is_none_or(|p| p.thermal != self.thermal) + { + events.push( + TelemetryEvent::new("lk.device.thermal.changed") + .with_body(format!("thermal: {}", self.thermal.as_str())) + .with_attribute("lk.device.thermal.state", self.thermal.as_str()), + ); + } + if let Some(low_power) = self + .low_power_mode + .filter(|_| previous.is_none_or(|p| p.low_power_mode != self.low_power_mode)) + { + events.push( + TelemetryEvent::new("lk.device.low_power.changed") + .with_body(format!("low power mode: {}", if low_power { "on" } else { "off" })) + .with_attribute("lk.device.low_power.enabled", low_power), + ); + } + if previous.is_none_or(|p| p.app_state != self.app_state) { + events.push( + TelemetryEvent::new("lk.device.app_state.changed") + .with_body(format!("app: {}", self.app_state.as_str())) + .with_attribute("lk.device.app_state", self.app_state.as_str()), + ); + } + if previous.is_none_or(|p| p.memory != self.memory) { + events.push( + TelemetryEvent::new("lk.device.memory.changed") + .with_body(format!("memory pressure: {}", self.memory.as_str())) + .with_attribute("lk.device.memory.pressure", self.memory.as_str()), + ); + } + if previous.is_none_or(|p| { + (p.network, p.network_expensive, p.network_constrained) + != (self.network, self.network_expensive, self.network_constrained) + }) { + events.push( + TelemetryEvent::new("lk.device.network.changed") + .with_body(format!( + "network: {}{}{}", + self.network.as_str(), + if self.network_expensive { ", expensive" } else { "" }, + if self.network_constrained { ", constrained" } else { "" } + )) + .with_attribute("network.connection.type", self.network.as_str()) + .with_attribute("lk.device.network.expensive", self.network_expensive) + .with_attribute("lk.device.network.constrained", self.network_constrained), + ); + } + // Per-percent updates are noise; report charging flips and bucket crossings. + if self.battery_level.is_some() + && previous.is_none_or(|p| { + p.battery_charging != self.battery_charging + || p.battery_bucket() != self.battery_bucket() + || p.battery_level.is_none() + }) + { + events.push( + TelemetryEvent::new("lk.device.battery.changed") + .with_body(format!( + "battery: {}%{}", + self.battery_level.unwrap_or(0), + if self.battery_charging { ", charging" } else { "" } + )) + .with_attribute( + "lk.device.battery.level", + self.battery_level.unwrap_or(0) as i64, + ) + .with_attribute("lk.device.battery.charging", self.battery_charging), + ); + } + events + } +} + +impl ThermalState { + fn as_str(self) -> &'static str { + match self { + ThermalState::Unknown => "unknown", + ThermalState::Nominal => "nominal", + ThermalState::Fair => "fair", + ThermalState::Serious => "serious", + ThermalState::Critical => "critical", + } + } +} + +impl AppState { + fn as_str(self) -> &'static str { + match self { + AppState::Foreground => "foreground", + AppState::Background => "background", + } + } +} + +impl MemoryPressure { + fn as_str(self) -> &'static str { + match self { + MemoryPressure::Normal => "normal", + MemoryPressure::Warning => "warning", + MemoryPressure::Critical => "critical", + } + } +} + +impl NetworkType { + fn as_str(self) -> &'static str { + match self { + NetworkType::Unknown => "unknown", + NetworkType::Wifi => "wifi", + NetworkType::Cell => "cell", + NetworkType::Wired => "wired", + NetworkType::Vpn => "vpn", + NetworkType::Bluetooth => "bluetooth", + NetworkType::Other => "other", + NetworkType::Unavailable => "unavailable", + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cadence_stretches_under_pressure_and_caps_at_four() { + assert_eq!(DeviceState::default().cadence_factor(), 1); + let serious = DeviceState { thermal: ThermalState::Serious, ..Default::default() }; + assert_eq!(serious.cadence_factor(), 2); + let low_power = DeviceState { low_power_mode: Some(true), ..Default::default() }; + assert_eq!(low_power.cadence_factor(), 2); + let tight = DeviceState { memory: MemoryPressure::Warning, ..Default::default() }; + assert_eq!(tight.cadence_factor(), 2); + let low_battery = DeviceState { battery_level: Some(15), ..Default::default() }; + assert_eq!(low_battery.cadence_factor(), 2); + let charging = DeviceState { battery_charging: true, ..low_battery }; + assert_eq!(charging.cadence_factor(), 1, "a charging battery is not low"); + let everything = DeviceState { + thermal: ThermalState::Critical, + low_power_mode: Some(true), + app_state: AppState::Background, + memory: MemoryPressure::Critical, + network_constrained: true, + ..Default::default() + }; + assert_eq!(everything.cadence_factor(), 4); + } + + #[test] + fn uploads_hold_on_constrained_network_or_empty_battery() { + assert!(!DeviceState::default().holds_uploads()); + let saver = DeviceState { network_constrained: true, ..Default::default() }; + assert!(saver.holds_uploads()); + let empty = DeviceState { battery_level: Some(8), ..Default::default() }; + assert!(empty.holds_uploads()); + let plugged = DeviceState { battery_charging: true, ..empty }; + assert!(!plugged.holds_uploads()); + let low = DeviceState { battery_level: Some(15), ..Default::default() }; + assert!(!low.holds_uploads(), "low only stretches the cadence"); + } + + #[test] + fn only_changed_fields_produce_events() { + let initial = DeviceState { + thermal: ThermalState::Nominal, + low_power_mode: Some(false), + ..Default::default() + }; + let names: Vec<_> = initial.change_events(None).into_iter().map(|e| e.name).collect(); + assert_eq!( + names, + [ + "lk.device.thermal.changed", + "lk.device.low_power.changed", + "lk.device.app_state.changed", + "lk.device.memory.changed", + "lk.device.network.changed", + ], + "initial values at session start; battery is unknown so it stays silent" + ); + let hot = DeviceState { thermal: ThermalState::Critical, ..initial }; + let events = hot.change_events(Some(&initial)); + assert_eq!(events.len(), 1); + assert_eq!(events[0].name, "lk.device.thermal.changed"); + assert!(hot.change_events(Some(&hot)).is_empty()); + } + + /// A platform without a thermal or low-power source reports them unknown — never a made-up + /// nominal/off: no change event, no stretch, no hold. + #[test] + fn unknown_thermal_and_low_power_are_silent_and_neutral() { + let unknown = DeviceState::default(); + assert_eq!((unknown.thermal, unknown.low_power_mode), (ThermalState::Unknown, None)); + let names: Vec<_> = unknown.change_events(None).into_iter().map(|e| e.name).collect(); + assert_eq!( + names, + [ + "lk.device.app_state.changed", + "lk.device.memory.changed", + "lk.device.network.changed" + ] + ); + assert_eq!(unknown.cadence_factor(), 1); + assert!(!unknown.holds_uploads()); + let known = DeviceState { + thermal: ThermalState::Serious, + low_power_mode: Some(true), + ..Default::default() + }; + assert!(unknown.change_events(Some(&known)).is_empty(), "losing the source is silent"); + assert_eq!(known.change_events(Some(&unknown)).len(), 2, "a first real reading reports"); + } + + #[test] + fn battery_reports_bucket_crossings_not_every_percent() { + let at = |level| DeviceState { battery_level: Some(level), ..Default::default() }; + assert_eq!(at(80).change_events(Some(&DeviceState::default())).len(), 1, "first reading"); + assert!(at(79).change_events(Some(&at(80))).is_empty()); + assert_eq!(at(20).change_events(Some(&at(21))).len(), 1, "crossed into low"); + assert_eq!(at(10).change_events(Some(&at(11))).len(), 1, "crossed into hold"); + let charging = DeviceState { battery_charging: true, ..at(10) }; + assert_eq!(charging.change_events(Some(&at(10))).len(), 1, "plugged in"); + } +} + +/// Why the audio route changed, in platform-neutral terms (AVAudioSession reasons; Android's +/// device added/removed callbacks map to `NewDevice` / `OldDeviceUnavailable`). +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AudioRouteReason { + NewDevice, + OldDeviceUnavailable, + CategoryChange, + Override, + WakeFromSleep, + NoSuitableRoute, + RouteConfigurationChange, + Unknown, +} + +/// Where the audio goes, in platform-neutral terms (AVAudioSession port types, Android's +/// `AudioDevice`; the web's `MediaDeviceInfo.kind` only ever gives `other`). +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AudioOutput { + Speaker, + /// The phone earpiece (`builtInReceiver`, Android `Earpiece`). + Receiver, + WiredHeadset, + Bluetooth, + CarAudio, + AirPlay, + Hdmi, + Usb, + Other, +} + +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CaptureDevice { + Camera, + Microphone, + ScreenShare, +} + +/// Why a capture device could not be used: the `getUserMedia` failure taxonomy, which every +/// platform maps onto (authorization status, `onCameraDisconnected`, `NotReadableError`, …). +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CaptureFailure { + /// The user (or MDM) said no: the most common pre-connect failure on mobile. + PermissionDenied, + NotFound, + /// Another app holds the device (`NotReadableError`, `TrackStartError`). + InUse, + /// The device went away mid-capture (unplugged, `onCameraDisconnected`). + Disconnected, + Other, +} + +/// Things that happen to the device mid-call and explain what the media did next. Not state +/// (see [`DeviceState`]): each one is a record with a display body. +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DeviceEvent { + /// The outputs after the change, in the order the platform lists them. + AudioRouteChanged { outputs: Vec, reason: AudioRouteReason }, + /// A phone call, Siri, another app taking audio focus: `began`, then the matching `ended`. + AudioInterruption { began: bool }, + /// A capture device could not be, or stay, used. + CaptureFailed { device: CaptureDevice, reason: CaptureFailure }, +} + +pub(crate) fn snake(debug: impl std::fmt::Debug) -> String { + let mut out = String::new(); + for (i, c) in format!("{debug:?}").chars().enumerate() { + if c.is_uppercase() && i > 0 { + out.push('_'); + } + out.push(c.to_ascii_lowercase()); + } + out +} + +impl DeviceEvent { + pub(crate) fn into_event(self) -> crate::TelemetryEvent { + use crate::{Severity, TelemetryEvent}; + match self { + Self::AudioRouteChanged { outputs, reason } => { + let outputs = outputs.into_iter().map(snake).collect::>().join(","); + let reason = snake(reason); + TelemetryEvent::new("lk.device.audio_route.changed") + .with_body(format!("audio route: {outputs} ({reason})")) + .with_attribute("lk.device.audio_route.outputs", outputs) + .with_attribute("lk.device.audio_route.reason", reason) + } + Self::AudioInterruption { began } => { + let phase = if began { "began" } else { "ended" }; + TelemetryEvent::new("lk.device.audio.interruption") + .with_body(format!("audio interruption {phase}")) + .with_attribute("lk.device.audio.interruption", phase) + } + Self::CaptureFailed { device, reason } => { + let (device, reason) = (snake(device), snake(reason)); + TelemetryEvent::new("lk.device.capture.failed") + .with_severity(Severity::Warn) + .with_body(format!("capture failed: {device} ({reason})")) + .with_attribute("lk.device.capture.device", device) + .with_attribute("lk.device.capture.reason", reason) + } + } + } +} + +#[cfg(test)] +mod event_tests { + use super::*; + + #[test] + fn device_events_have_bodies_and_snake_case_values() { + let event = DeviceEvent::AudioRouteChanged { + outputs: vec![AudioOutput::Speaker], + reason: AudioRouteReason::OldDeviceUnavailable, + } + .into_event(); + assert_eq!(event.body.as_deref(), Some("audio route: speaker (old_device_unavailable)")); + let denied = DeviceEvent::CaptureFailed { + device: CaptureDevice::ScreenShare, + reason: CaptureFailure::PermissionDenied, + } + .into_event(); + assert_eq!( + denied.body.as_deref(), + Some("capture failed: screen_share (permission_denied)") + ); + assert_eq!(denied.severity, crate::Severity::Warn); + } +} diff --git a/livekit-telemetry/src/lib.rs b/livekit-telemetry/src/lib.rs index 7b2b3a526..b3920cf6b 100644 --- a/livekit-telemetry/src/lib.rs +++ b/livekit-telemetry/src/lib.rs @@ -31,6 +31,23 @@ mod store; )] mod stats; +/// Host-reported device state and the cadence policy derived from it. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" + ) +)] +mod device; + +/// RTC stats samples and their on-device windowing. +#[expect( + dead_code, + reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" +)] +mod rtc; + /// Session identity: the trace id, attributes and route of one Room. #[expect( dead_code, @@ -76,7 +93,9 @@ mod destination; pub use cache::{BatchCache, FileCache, MemoryCache}; pub use destination::ENDPOINT_OVERRIDE_ENV; +pub use device::*; pub use event::*; +pub use rtc::{RtcStat, RtcStatsSample, StreamDirection, TrackKind}; pub use span::SpanOutcome; pub use stats::{TelemetryStats, TelemetryStatus}; #[cfg(feature = "net")] diff --git a/livekit-telemetry/src/rtc.rs b/livekit-telemetry/src/rtc.rs new file mode 100644 index 000000000..7076cc8ee --- /dev/null +++ b/livekit-telemetry/src/rtc.rs @@ -0,0 +1,1006 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::{collections::HashMap, sync::Arc, time::Duration}; + +use tokio::time::Instant; + +use crate::{ + event::now_unix_nanos, scope::ScopeState, store::QueuedEvent, Attribute, AttributeValue, + TelemetryEvent, +}; + +/// Audio or video. +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum TrackKind { + Audio, + Video, +} + +/// Received (inbound) or sent (outbound) media. +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum StreamDirection { + Inbound, + Outbound, +} + +/// One reading of a track's RTP statistics, as `getStats()` reports them. +/// +/// Cumulative counters (`bytes`, `packets`, `freeze_count`, …) are passed through as reported — +/// monotonic, paired with their denominators (the W3C webrtc-stats model) so any layer can +/// recompute rates and a dropped window never corrupts the next. Gauges (`jitter_ms`, `rtt_ms`, +/// `frames_per_second`, `audio_level`) are summarised per window as min/max/avg. Fields a +/// platform or direction does not have stay `None` and are omitted from the wire. +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +#[derive(Debug, Clone, PartialEq)] +pub struct RtcStatsSample { + pub track_sid: String, + pub kind: TrackKind, + pub direction: StreamDirection, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub codec: Option, + // Cumulative counters. + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub bytes: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub packets: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub packets_lost: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub freeze_count: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub freezes_duration_ms: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub concealed_samples: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub concealment_events: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub jitter_buffer_delay_ms: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub jitter_buffer_emitted_count: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub quality_limitation_bandwidth_ms: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub quality_limitation_cpu_ms: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub quality_limitation_other_ms: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub pause_count: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub pauses_duration_ms: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub silent_concealed_samples: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub interruption_count: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub interruptions_duration_ms: Option, + // Gauges. + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub jitter_ms: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub rtt_ms: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub frames_per_second: Option, + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub audio_level: Option, + /// When the reading was taken; `None` = now. + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub timestamp_ns: Option, + /// The RTP stream this sample describes when a track has several (simulcast layers): any + /// stable id (`rid`, ssrc, the stats id). The core folds layers into the track; a platform + /// never sums them itself. + #[cfg_attr(feature = "uniffi", uniffi(default))] + pub layer: Option, +} + +/// One entry of a WebRTC `RTCStatsReport`, as the platform got it: the entry's `type`, `id` and +/// its standard members (W3C webrtc-stats names; nested maps flattened with a dot, e.g. +/// `qualityLimitationDurations.cpu`). Numbers may arrive as `Int`, `Double` or numeric `Str`. +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +#[derive(Debug, Clone, PartialEq)] +pub struct RtcStat { + pub kind: String, + pub id: String, + pub members: HashMap, +} + +fn num(stat: &RtcStat, key: &str) -> Option { + match stat.members.get(key)? { + AttributeValue::Int(i) => Some(*i as f64), + AttributeValue::Double(d) => Some(*d), + AttributeValue::Str(s) => s.parse().ok(), + AttributeValue::Bool(_) => None, + } +} + +fn count(stat: &RtcStat, key: &str) -> Option { + num(stat, key).map(|v| v.max(0.0) as u64) +} + +/// A duration member (seconds, per webrtc-stats) as whole milliseconds. +fn ms(stat: &RtcStat, key: &str) -> Option { + num(stat, key).map(|s| (s.max(0.0) * 1000.0) as u64) +} + +fn text(stat: &RtcStat, key: &str) -> Option { + match stat.members.get(key)? { + AttributeValue::Str(s) => Some(s.clone()), + _ => None, + } +} + +/// The samples in one track's `getStats()` report: one per `outbound-rtp` (tagged with its +/// layer) or `inbound-rtp` entry of `direction`, codec resolved through `codecId`, RTT from the +/// `remote-inbound-rtp` that reports on the stream (outbound) or the nominated candidate pair +/// (inbound), durations converted from seconds to milliseconds. One mapping for every SDK. +pub(crate) fn samples_from_report( + track_sid: &str, + kind: TrackKind, + direction: StreamDirection, + report: &[RtcStat], + timestamp_ns: Option, +) -> Vec { + let index = Index::new(report); + index + .rtp + .iter() + .filter(|(_, d)| *d == direction) + .map(|(stat, direction)| index.sample(stat, track_sid, kind, *direction, timestamp_ns)) + .collect() +} + +/// The samples in a whole peer connection's `getStats()` report: every RTP stream whose track +/// is in `tracks` (MediaStreamTrack id → track sid; `trackIdentifier` of an `inbound-rtp`, or of +/// the `media-source` an `outbound-rtp` names), kind from the stream's `kind`. Streams of +/// unknown tracks are skipped. One call per peer connection per poll: no platform walks its +/// participants and tracks. +pub(crate) fn samples_from_peer_report( + report: &[RtcStat], + tracks: &HashMap, + timestamp_ns: Option, +) -> Vec { + let index = Index::new(report); + index + .rtp + .iter() + .filter_map(|(stat, direction)| { + let track_id = match direction { + StreamDirection::Inbound => text(stat, "trackIdentifier"), + StreamDirection::Outbound => text(stat, "mediaSourceId") + .and_then(|id| index.by_id.get(id.as_str()).copied()) + .filter(|source| source.kind == "media-source") + .and_then(|source| text(source, "trackIdentifier")), + }?; + let sid = tracks.get(&track_id)?; + let kind = match text(stat, "kind").or_else(|| text(stat, "mediaType"))?.as_str() { + "audio" => TrackKind::Audio, + "video" => TrackKind::Video, + _ => return None, + }; + Some(index.sample(stat, sid, kind, *direction, timestamp_ns)) + }) + .collect() +} + +/// A report indexed once: entries by id, the `remote-inbound-rtp` reporting on each outbound +/// stream, the nominated candidate pair's RTT, the RTP streams with their direction. Resolving a +/// stream's codec, RTT or media source is then a lookup, not a scan of the report. +struct Index<'a> { + by_id: HashMap<&'a str, &'a RtcStat>, + remote_by_local: HashMap, + pair_rtt_ms: Option, + rtp: Vec<(&'a RtcStat, StreamDirection)>, +} + +impl<'a> Index<'a> { + fn new(report: &'a [RtcStat]) -> Self { + let mut index = Self { + by_id: HashMap::with_capacity(report.len()), + remote_by_local: HashMap::new(), + pair_rtt_ms: None, + rtp: Vec::new(), + }; + for stat in report { + index.by_id.insert(stat.id.as_str(), stat); + match stat.kind.as_str() { + "outbound-rtp" => index.rtp.push((stat, StreamDirection::Outbound)), + "inbound-rtp" => index.rtp.push((stat, StreamDirection::Inbound)), + "remote-inbound-rtp" => { + if let Some(local) = text(stat, "localId") { + index.remote_by_local.insert(local, stat); + } + } + "candidate-pair" if index.pair_rtt_ms.is_none() => { + let nominated = + matches!(stat.members.get("nominated"), Some(AttributeValue::Bool(true))) + || text(stat, "state").as_deref() == Some("succeeded"); + if nominated { + index.pair_rtt_ms = num(stat, "currentRoundTripTime").map(|s| s * 1000.0); + } + } + _ => {} + } + } + index + } + + /// One RTP stream entry as a sample. + fn sample( + &self, + stat: &RtcStat, + track_sid: &str, + kind: TrackKind, + direction: StreamDirection, + timestamp_ns: Option, + ) -> RtcStatsSample { + let mut sample = RtcStatsSample::new(track_sid, kind, direction); + sample.codec = text(stat, "codecId") + .and_then(|id| self.by_id.get(id.as_str()).copied()) + .filter(|codec| codec.kind == "codec") + .and_then(|codec| text(codec, "mimeType")); + sample.timestamp_ns = timestamp_ns; + match direction { + StreamDirection::Outbound => { + sample.layer = Some(text(stat, "rid").unwrap_or_else(|| stat.id.clone())); + sample.bytes = count(stat, "bytesSent"); + sample.packets = count(stat, "packetsSent"); + sample.frames_per_second = num(stat, "framesPerSecond"); + sample.quality_limitation_bandwidth_ms = + ms(stat, "qualityLimitationDurations.bandwidth"); + sample.quality_limitation_cpu_ms = ms(stat, "qualityLimitationDurations.cpu"); + sample.quality_limitation_other_ms = ms(stat, "qualityLimitationDurations.other"); + sample.rtt_ms = self + .remote_by_local + .get(&stat.id) + .and_then(|remote| num(remote, "roundTripTime")) + .map(|s| s * 1000.0); + } + StreamDirection::Inbound => { + sample.bytes = count(stat, "bytesReceived"); + sample.packets = count(stat, "packetsReceived"); + sample.packets_lost = count(stat, "packetsLost"); + sample.freeze_count = count(stat, "freezeCount"); + sample.freezes_duration_ms = ms(stat, "totalFreezesDuration"); + sample.pause_count = count(stat, "pauseCount"); + sample.pauses_duration_ms = ms(stat, "totalPausesDuration"); + sample.concealed_samples = count(stat, "concealedSamples"); + sample.silent_concealed_samples = count(stat, "silentConcealedSamples"); + sample.concealment_events = count(stat, "concealmentEvents"); + sample.interruption_count = count(stat, "interruptionCount"); + sample.interruptions_duration_ms = ms(stat, "totalInterruptionDuration"); + sample.jitter_buffer_delay_ms = ms(stat, "jitterBufferDelay"); + sample.jitter_buffer_emitted_count = count(stat, "jitterBufferEmittedCount"); + sample.jitter_ms = num(stat, "jitter").map(|s| s * 1000.0); + sample.frames_per_second = num(stat, "framesPerSecond"); + sample.audio_level = num(stat, "audioLevel"); + sample.rtt_ms = self.pair_rtt_ms; + } + } + sample + } +} + +impl RtcStatsSample { + /// A sample with every optional field unset. + pub fn new(track_sid: impl Into, kind: TrackKind, direction: StreamDirection) -> Self { + Self { + track_sid: track_sid.into(), + kind, + direction, + codec: None, + bytes: None, + packets: None, + packets_lost: None, + freeze_count: None, + freezes_duration_ms: None, + concealed_samples: None, + concealment_events: None, + jitter_buffer_delay_ms: None, + jitter_buffer_emitted_count: None, + quality_limitation_bandwidth_ms: None, + quality_limitation_cpu_ms: None, + quality_limitation_other_ms: None, + pause_count: None, + pauses_duration_ms: None, + silent_concealed_samples: None, + interruption_count: None, + interruptions_duration_ms: None, + jitter_ms: None, + rtt_ms: None, + frames_per_second: None, + audio_level: None, + timestamp_ns: None, + layer: None, + } + } +} + +/// min / max / avg of a gauge over a window. +#[derive(Debug, Default, Clone, Copy)] +struct Gauge { + min: f64, + max: f64, + sum: f64, + n: u32, +} + +impl Gauge { + fn add(&mut self, value: Option) { + let Some(value) = value else { return }; + if self.n == 0 { + self.min = value; + self.max = value; + } else { + self.min = self.min.min(value); + self.max = self.max.max(value); + } + self.sum += value; + self.n += 1; + } + + fn attach(&self, event: TelemetryEvent, key: &str) -> TelemetryEvent { + if self.n == 0 { + return event; + } + event + .with_attribute(format!("{key}.min"), self.min) + .with_attribute(format!("{key}.max"), self.max) + .with_attribute(format!("{key}.avg"), self.sum / self.n as f64) + } +} + +/// Samples of one track in one direction accumulated since the window opened. +struct Window { + session: Arc, + /// The window's owner, captured when it opened: the session's project and its correlation + /// attributes then. A change to either closes the window first (see `StatsWindows::split`). + route: Option, + custom: Vec, + start_ns: u64, + samples: u32, + /// Cumulative counters at the window's first reading, for the display body's deltas. + first: RtcStatsSample, + last: RtcStatsSample, + jitter: Gauge, + rtt: Gauge, + fps: Gauge, + audio_level: Gauge, +} + +impl Window { + fn open(start_ns: u64, first: RtcStatsSample, session: Arc) -> Self { + let mut window = Self { + route: session.route(), + custom: session.custom_snapshot(), + session, + start_ns, + samples: 0, + first: first.clone(), + last: first.clone(), + jitter: Gauge::default(), + rtt: Gauge::default(), + fps: Gauge::default(), + audio_level: Gauge::default(), + }; + window.add(first); + window + } + + /// `video outbound: 1204 kbps, loss 0.4%, rtt 48 ms, 30 fps` — deltas over the window. + fn summary(&self, window_ms: u64) -> String { + let mut parts = Vec::new(); + let delta = |a: Option, b: Option| Some(a?.saturating_sub(b?)); + if let (Some(bytes), true) = (delta(self.last.bytes, self.first.bytes), window_ms > 0) { + parts.push(format!("{} kbps", bytes * 8 / window_ms)); + } + if let (Some(lost), Some(packets)) = ( + delta(self.last.packets_lost, self.first.packets_lost), + delta(self.last.packets, self.first.packets), + ) { + if lost + packets > 0 { + parts.push(format!("loss {:.1}%", lost as f64 * 100.0 / (lost + packets) as f64)); + } + } + if self.rtt.n > 0 { + parts.push(format!("rtt {:.0} ms", self.rtt.sum / self.rtt.n as f64)); + } + if self.fps.n > 0 { + parts.push(format!("{:.0} fps", self.fps.sum / self.fps.n as f64)); + } + if let Some(freezes) = + delta(self.last.freeze_count, self.first.freeze_count).filter(|f| *f > 0) + { + parts.push(format!("{freezes} freezes")); + } + format!( + "{} {}: {}", + self.last.kind.as_str(), + self.last.direction.as_str(), + if parts.is_empty() { "no data".to_owned() } else { parts.join(", ") } + ) + } + + /// The closed window as a queued record, with the owner it captured when it opened and its + /// session's attributes as of now, the window's capture time (see `QueuedEvent::new`). + fn into_queued(self, end_ns: u64) -> QueuedEvent { + let (session, route, custom) = + (self.session.clone(), self.route.clone(), self.custom.clone()); + let mut event = self.into_event(end_ns); + for attribute in custom { + if !event.attributes.iter().any(|a| a.key == attribute.key) { + event.attributes.push(attribute); + } + } + session.decorate(&mut event.attributes); + QueuedEvent { event, session, route } + } + + fn add(&mut self, sample: RtcStatsSample) { + self.samples += 1; + self.jitter.add(sample.jitter_ms); + self.rtt.add(sample.rtt_ms); + self.fps.add(sample.frames_per_second); + self.audio_level.add(sample.audio_level); + self.last = sample; + } + + /// The `lk.rtc.stats.sample` event for this window, stamped at `end_ns`. Its body is the + /// one-line human summary a log view shows (OTel: an event's body is its display message); + /// the attributes carry the numbers. + fn into_event(self, end_ns: u64) -> TelemetryEvent { + let window_ms = end_ns.saturating_sub(self.start_ns) / 1_000_000; + let body = self.summary(window_ms); + let last = self.last; + let mut event = TelemetryEvent::new("lk.rtc.stats.sample") + .with_body(body) + .with_attribute("lk.track.sid", last.track_sid) + .with_attribute("lk.track.kind", last.kind.as_str()) + .with_attribute("lk.track.direction", last.direction.as_str()) + .with_attribute("lk.rtc.window_ms", window_ms as i64) + .with_attribute("lk.rtc.samples", self.samples as i64); + if let Some(codec) = last.codec { + event = event.with_attribute("lk.rtc.codec", codec); + } + let counters = [ + ("lk.rtc.bytes", last.bytes), + ("lk.rtc.packets", last.packets), + ("lk.rtc.packets_lost", last.packets_lost), + ("lk.rtc.freeze_count", last.freeze_count), + ("lk.rtc.freezes_duration_ms", last.freezes_duration_ms), + ("lk.rtc.concealed_samples", last.concealed_samples), + ("lk.rtc.concealment_events", last.concealment_events), + ("lk.rtc.jitter_buffer_delay_ms", last.jitter_buffer_delay_ms), + ("lk.rtc.jitter_buffer_emitted_count", last.jitter_buffer_emitted_count), + ("lk.rtc.quality_limitation.bandwidth_ms", last.quality_limitation_bandwidth_ms), + ("lk.rtc.quality_limitation.cpu_ms", last.quality_limitation_cpu_ms), + ("lk.rtc.quality_limitation.other_ms", last.quality_limitation_other_ms), + ("lk.rtc.pause_count", last.pause_count), + ("lk.rtc.pauses_duration_ms", last.pauses_duration_ms), + ("lk.rtc.silent_concealed_samples", last.silent_concealed_samples), + ("lk.rtc.interruption_count", last.interruption_count), + ("lk.rtc.interruptions_duration_ms", last.interruptions_duration_ms), + ]; + for (key, value) in counters { + if let Some(value) = value { + event = event.with_attribute(key, value as i64); + } + } + event = self.jitter.attach(event, "lk.rtc.jitter_ms"); + event = self.rtt.attach(event, "lk.rtc.rtt_ms"); + event = self.fps.attach(event, "lk.rtc.fps"); + event = self.audio_level.attach(event, "lk.rtc.audio_level"); + event.timestamp_ns = Some(end_ns); + event + } +} + +/// Open RTC stats windows, one per track and direction. +/// +/// The platform pushes raw `getStats()` readings when the core asks; the core ships one +/// `lk.rtc.stats.sample` per window (60 s by default, stretched under device pressure) — on +/// device: 1 Hz raw sampling across ~100k concurrent participants would be the same ~100k +/// records/s fleet-wide. +#[derive(Default)] +pub(crate) struct StatsWindows { + windows: HashMap, + /// Last cumulative `qualityLimitationDurations.cpu` per outbound track. + limitation: HashMap, + cpu_limited_until: Option, + /// Last cumulative counters per simulcast layer, per outbound track: the track's counters + /// are their sum, so a suspended top layer cannot freeze them. + layers: HashMap>, + /// The opt-out, checked under the lock that guards the windows (see `Store`). + revoked: Arc, +} + +/// One track in one direction *in one session*: two rooms in a process may subscribe to the same +/// published track (same sid, same direction), and their readings must never merge. +type TrackKey = ([u8; 16], String, StreamDirection); + +fn track_key(session: &ScopeState, sample: &RtcStatsSample) -> TrackKey { + (session.trace_id, sample.track_sid.clone(), sample.direction) +} + +#[derive(Debug, Clone, Copy, Default)] +struct LayerCounters { + bytes: Option, + packets: Option, + fps: Option, + limitation_bandwidth_ms: Option, + limitation_cpu_ms: Option, + limitation_other_ms: Option, +} + +fn sum(values: impl Iterator>) -> Option { + values.flatten().reduce(|a, b| a + b) +} + +fn max_u64(values: impl Iterator>) -> Option { + values.flatten().max() +} + +/// CPU starvation is sticky: stretch the cadence for a while after the last sign of it. +/// `qualityLimitationDurations.bandwidth` is deliberately not a signal: WebRTC reports it for +/// minutes during a normal ramp-up and for as long as an encoder stalls (an iPhone camera at +/// 0 kbps held uploads for 8 minutes); the congestion controller does not need our help. +const CPU_LIMITED_HOLD: Duration = Duration::from_secs(60); + +impl StatsWindows { + /// Tie the windows to a pipeline's opt-out. + pub fn with_consent(revoked: Arc) -> Self { + Self { revoked, ..Self::default() } + } + + pub fn record_in(&mut self, mut sample: RtcStatsSample, session: &Arc) { + if self.revoked.load(std::sync::atomic::Ordering::SeqCst) { + return; + } + let key = track_key(session, &sample); + self.fold_layers(&key, &mut sample); + self.track_limitation(&key, &sample); + let timestamp = *sample.timestamp_ns.get_or_insert_with(now_unix_nanos); + match self.windows.get_mut(&key) { + Some(window) => window.add(sample), + None => { + self.windows.insert(key, Window::open(timestamp, sample, session.clone())); + } + } + } + + /// Simulcast publishes one RTP stream per layer under one track sid. Remember each layer's + /// latest cumulative counters and rewrite the sample as the track's total, so the window + /// sees one monotonic series per track. + fn fold_layers(&mut self, key: &TrackKey, sample: &mut RtcStatsSample) { + let Some(layer) = sample.layer.clone() else { return }; + let layers = self.layers.entry(key.clone()).or_default(); + layers.insert( + layer, + LayerCounters { + bytes: sample.bytes, + packets: sample.packets, + fps: sample.frames_per_second, + limitation_bandwidth_ms: sample.quality_limitation_bandwidth_ms, + limitation_cpu_ms: sample.quality_limitation_cpu_ms, + limitation_other_ms: sample.quality_limitation_other_ms, + }, + ); + sample.bytes = sum(layers.values().map(|l| l.bytes)); + sample.packets = sum(layers.values().map(|l| l.packets)); + sample.frames_per_second = layers + .values() + .filter_map(|l| l.fps) + .fold(None, |m, v| Some(m.map_or(v, |m: f64| m.max(v)))); + sample.quality_limitation_bandwidth_ms = + max_u64(layers.values().map(|l| l.limitation_bandwidth_ms)); + sample.quality_limitation_cpu_ms = max_u64(layers.values().map(|l| l.limitation_cpu_ms)); + sample.quality_limitation_other_ms = + max_u64(layers.values().map(|l| l.limitation_other_ms)); + } + + /// Close every open window into its event, filed under the window's session, and start fresh. + /// A track that produced no reading for a whole window has left (unpublished, unsubscribed, + /// its room gone): its per-layer and CPU counters are retired with it, so memory follows the + /// live tracks rather than every track the process ever saw. + pub fn close(&mut self) -> Vec { + let windows = &self.windows; + self.layers.retain(|key, _| windows.contains_key(key)); + self.limitation.retain(|key, _| windows.contains_key(key)); + let end = now_unix_nanos(); + self.windows.drain().map(|(_, window)| window.into_queued(end)).collect() + } + + /// A track left `session` (`track_sid`), or the session ended (`None`): close its windows now + /// — the last partial window still ships — and forget everything kept for it. + pub fn retire(&mut self, session: &ScopeState, track_sid: Option<&str>) -> Vec { + let gone = |(trace_id, sid, _): &TrackKey| { + *trace_id == session.trace_id && track_sid.is_none_or(|t| t == sid) + }; + self.layers.retain(|key, _| !gone(key)); + self.limitation.retain(|key, _| !gone(key)); + let keys: Vec = self.windows.keys().filter(|k| gone(k)).cloned().collect(); + let end = now_unix_nanos(); + keys.into_iter() + .filter_map(|key| self.windows.remove(&key)) + .map(|window| window.into_queued(end)) + .collect() + } + + /// Close `session`'s open windows now — their records then carry the attributes they were + /// captured under — keeping each track's cumulative state for the windows that follow. + pub fn split(&mut self, session: &ScopeState) -> Vec { + let keys: Vec = self + .windows + .keys() + .filter(|(trace_id, ..)| *trace_id == session.trace_id) + .cloned() + .collect(); + let end = now_unix_nanos(); + keys.into_iter() + .filter_map(|key| self.windows.remove(&key)) + .map(|window| window.into_queued(end)) + .collect() + } + + /// How many tracks the accumulator holds state for (windows, layers, CPU counters). + #[cfg(test)] + pub fn tracked(&self) -> (usize, usize, usize) { + (self.windows.len(), self.layers.len(), self.limitation.len()) + } + + #[cfg(test)] + pub fn record(&mut self, sample: RtcStatsSample) { + self.record_in(sample, &test_session()); + } + + #[cfg(test)] + pub fn close_events(&mut self) -> Vec { + self.close().into_iter().map(|q| q.event).collect() + } + + /// The encoder was CPU-starved within the last minute: stretch the cadence, like thermal + /// pressure. + pub fn cpu_limited(&self) -> bool { + self.cpu_limited_until.is_some_and(|t| Instant::now() < t) + } + + /// A cpu limitation counter that grew since the previous reading means the encoder was + /// starved in between. + fn track_limitation(&mut self, key: &TrackKey, sample: &RtcStatsSample) { + if sample.direction != StreamDirection::Outbound { + return; + } + let current = sample.quality_limitation_cpu_ms.unwrap_or(0); + if let Some(previous) = self.limitation.insert(key.clone(), current) { + if current > previous { + self.cpu_limited_until = Some(Instant::now() + CPU_LIMITED_HOLD); + } + } + } + + /// Drop everything (opt-out); returns how many open windows went. + pub fn clear(&mut self) -> u64 { + let open = self.windows.len() as u64; + *self = Self::with_consent(self.revoked.clone()); + open + } +} + +/// The one session every `StatsWindows::record` in a unit test files under. +#[cfg(test)] +fn test_session() -> Arc { + ScopeState::with_trace_id([7; 16]) +} + +impl TrackKind { + fn as_str(self) -> &'static str { + match self { + TrackKind::Audio => "audio", + TrackKind::Video => "video", + } + } +} + +impl StreamDirection { + fn as_str(self) -> &'static str { + match self { + StreamDirection::Inbound => "inbound", + StreamDirection::Outbound => "outbound", + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn simulcast_layers_fold_into_one_track_series() { + let mut windows = StatsWindows::default(); + let layer = |id: &str, bytes: u64, fps: f64| RtcStatsSample { + bytes: Some(bytes), + frames_per_second: Some(fps), + layer: Some(id.into()), + ..RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound) + }; + windows.record(layer("h", 1_000, 30.0)); + windows.record(layer("f", 4_000, 30.0)); + // The top layer stalls; the half layer keeps sending. + windows.record(layer("h", 3_000, 30.0)); + windows.record(layer("f", 4_000, 0.0)); + let events = windows.close_events(); + assert_eq!(events.len(), 1, "one window per track, not per layer"); + let bytes = + events[0].attributes.iter().find(|a| a.key == "lk.rtc.bytes").map(|a| a.value.clone()); + assert_eq!(bytes, Some(AttributeValue::Int(7_000)), "last folded total"); + let body = events[0].body.clone().unwrap_or_default(); + assert!(!body.contains(" 0 kbps"), "a stalled layer is not a stalled track: {body}"); + } + + #[tokio::test(start_paused = true)] + async fn cpu_limitation_counter_drives_cadence_pressure() { + let mut windows = StatsWindows::default(); + let reading = |bandwidth_ms, cpu_ms| RtcStatsSample { + quality_limitation_bandwidth_ms: Some(bandwidth_ms), + quality_limitation_cpu_ms: Some(cpu_ms), + ..RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound) + }; + windows.record(reading(0, 0)); + assert!(!windows.cpu_limited(), "first reading: no delta"); + windows.record(reading(5_000, 0)); + assert!(!windows.cpu_limited(), "bandwidth limitation is not pressure"); + windows.record(reading(5_000, 250)); + assert!(windows.cpu_limited()); + tokio::time::advance(Duration::from_secs(59)).await; + assert!(windows.cpu_limited(), "cpu hold is sticky"); + tokio::time::advance(Duration::from_secs(2)).await; + assert!(!windows.cpu_limited(), "and expires"); + let mut inbound = RtcStatsSample::new("TR_2", TrackKind::Audio, StreamDirection::Inbound); + inbound.quality_limitation_cpu_ms = Some(9_999); + windows.record(inbound.clone()); + windows.record(inbound); + assert!(!windows.cpu_limited(), "inbound counters are ignored"); + } + use crate::AttributeValue; + + fn attr(event: &TelemetryEvent, key: &str) -> Option { + event.attributes.iter().find(|a| a.key == key).map(|a| a.value.clone()) + } + + #[test] + fn window_keeps_last_counter_and_summarises_gauges() { + let mut windows = StatsWindows::default(); + for (bytes, jitter) in [(100, 1.0), (200, 3.0), (300, 2.0)] { + let mut sample = + RtcStatsSample::new("TR_1", TrackKind::Audio, StreamDirection::Inbound); + sample.bytes = Some(bytes); + sample.jitter_ms = Some(jitter); + windows.record(sample); + } + let mut other = RtcStatsSample::new("TR_1", TrackKind::Audio, StreamDirection::Outbound); + other.packets = Some(7); + windows.record(other); + + let mut events = windows.close_events(); + assert!(windows.close_events().is_empty(), "closing again yields nothing"); + events.sort_by_key(|e| attr(e, "lk.track.direction").map(|v| format!("{v:?}"))); + assert_eq!(events.len(), 2, "one event per track and direction"); + let inbound = &events[0]; + assert_eq!(inbound.name, "lk.rtc.stats.sample"); + assert_eq!( + attr(inbound, "lk.track.direction"), + Some(AttributeValue::Str("inbound".into())) + ); + assert_eq!( + attr(inbound, "lk.rtc.bytes"), + Some(AttributeValue::Int(300)), + "cumulative: last value" + ); + assert_eq!(attr(inbound, "lk.rtc.samples"), Some(AttributeValue::Int(3))); + assert_eq!(attr(inbound, "lk.rtc.jitter_ms.min"), Some(AttributeValue::Double(1.0))); + assert_eq!(attr(inbound, "lk.rtc.jitter_ms.max"), Some(AttributeValue::Double(3.0))); + assert_eq!(attr(inbound, "lk.rtc.jitter_ms.avg"), Some(AttributeValue::Double(2.0))); + assert_eq!(attr(inbound, "lk.rtc.rtt_ms.avg"), None, "absent gauges are omitted"); + assert_eq!(attr(&events[1], "lk.rtc.packets"), Some(AttributeValue::Int(7))); + } + + #[test] + fn a_report_maps_to_samples_with_codec_rtt_layers_and_milliseconds() { + let stat = |kind: &str, id: &str, members: &[(&str, AttributeValue)]| RtcStat { + kind: kind.into(), + id: id.into(), + members: members.iter().map(|(k, v)| (k.to_string(), v.clone())).collect(), + }; + let s = |v: &str| AttributeValue::Str(v.into()); + let report = vec![ + stat("codec", "C1", &[("mimeType", s("video/VP8"))]), + stat( + "outbound-rtp", + "OUT_f", + &[ + ("rid", s("f")), + ("codecId", s("C1")), + ("bytesSent", AttributeValue::Int(7_000)), + ("packetsSent", s("70")), + ("framesPerSecond", AttributeValue::Double(29.5)), + ("qualityLimitationDurations.cpu", AttributeValue::Double(1.25)), + ], + ), + stat( + "outbound-rtp", + "OUT_h", + &[("rid", s("h")), ("bytesSent", AttributeValue::Int(500))], + ), + stat( + "remote-inbound-rtp", + "RI_f", + &[("localId", s("OUT_f")), ("roundTripTime", AttributeValue::Double(0.042))], + ), + ]; + let out = samples_from_report( + "TR_1", + TrackKind::Video, + StreamDirection::Outbound, + &report, + Some(5), + ); + assert_eq!(out.len(), 2); + let f = out.iter().find(|x| x.layer.as_deref() == Some("f")).expect("layer f"); + assert_eq!(f.codec.as_deref(), Some("video/VP8")); + assert_eq!((f.bytes, f.packets, f.frames_per_second), (Some(7_000), Some(70), Some(29.5))); + assert_eq!(f.quality_limitation_cpu_ms, Some(1_250)); + assert_eq!(f.rtt_ms, Some(42.0)); + assert_eq!(f.timestamp_ns, Some(5)); + assert_eq!( + out.iter().find(|x| x.layer.as_deref() == Some("h")).and_then(|x| x.rtt_ms), + None + ); + + let report = vec![ + stat( + "candidate-pair", + "CP", + &[ + ("nominated", AttributeValue::Bool(true)), + ("currentRoundTripTime", AttributeValue::Double(0.1)), + ], + ), + stat( + "inbound-rtp", + "IN", + &[ + ("bytesReceived", AttributeValue::Int(12)), + ("packetsLost", AttributeValue::Int(-3)), + ("jitter", AttributeValue::Double(0.02)), + ("totalFreezesDuration", AttributeValue::Double(2.5)), + ("jitterBufferDelay", AttributeValue::Double(3.0)), + ("audioLevel", AttributeValue::Double(0.5)), + ], + ), + ]; + let out = + samples_from_report("TR_2", TrackKind::Audio, StreamDirection::Inbound, &report, None); + assert_eq!(out.len(), 1); + let x = &out[0]; + assert_eq!((x.bytes, x.packets_lost, x.rtt_ms), (Some(12), Some(0), Some(100.0))); + assert_eq!( + (x.jitter_ms, x.freezes_duration_ms, x.jitter_buffer_delay_ms), + (Some(20.0), Some(2_500), Some(3_000)) + ); + assert_eq!(x.layer, None); + } + + /// Two rooms in one process subscribed to the same published track keep two windows, each filed + /// under its own session. + #[test] + fn the_same_track_in_two_sessions_never_merges() { + let mut windows = StatsWindows::default(); + let (a, b) = (ScopeState::new(), ScopeState::new()); + let reading = |bytes| RtcStatsSample { + bytes: Some(bytes), + ..RtcStatsSample::new("TR_shared", TrackKind::Audio, StreamDirection::Inbound) + }; + windows.record_in(reading(100), &a); + windows.record_in(reading(900), &b); + windows.record_in(reading(200), &a); + let closed = windows.close(); + assert_eq!(closed.len(), 2, "one window per session"); + for (session, bytes) in [(&a, 200), (&b, 900)] { + let window = closed.iter().find(|q| q.session == *session).expect("own window"); + assert_eq!(attr(&window.event, "lk.rtc.bytes"), Some(AttributeValue::Int(bytes))); + let id = Some(AttributeValue::Str(session.hex())); + assert_eq!(attr(&window.event, "session.id"), id, "decorated at capture"); + } + } + + /// Per-track state follows the live tracks — retired when a track leaves, when its session + /// ends, and when it goes a whole window without a reading. + #[test] + fn per_track_state_is_retired_with_the_track() { + let mut windows = StatsWindows::default(); + let (a, b) = (ScopeState::new(), ScopeState::new()); + let outbound = |sid: &str| RtcStatsSample { + bytes: Some(1), + quality_limitation_cpu_ms: Some(0), + layer: Some("f".into()), + ..RtcStatsSample::new(sid, TrackKind::Video, StreamDirection::Outbound) + }; + for sid in ["TR_1", "TR_2"] { + windows.record_in(outbound(sid), &a); + } + windows.record_in(outbound("TR_3"), &b); + assert_eq!(windows.tracked(), (3, 3, 3)); + + let last = windows.retire(&a, Some("TR_1")); + assert_eq!(last.len(), 1, "the unpublished track's partial window ships now"); + assert_eq!(windows.tracked(), (2, 2, 2)); + + let last = windows.retire(&a, None); + assert_eq!(last.len(), 1, "the session's last window ships at disconnect"); + assert_eq!(windows.tracked(), (1, 1, 1), "the other session keeps its track"); + + windows.close(); + assert_eq!(windows.tracked(), (0, 1, 1), "counters survive an ordinary window…"); + windows.close(); + assert_eq!(windows.tracked(), (0, 0, 0), "…not a whole window without a reading"); + } + + #[test] + fn a_peer_connection_report_maps_to_its_tracks() { + let stat = |kind: &str, id: &str, members: &[(&str, &str)]| RtcStat { + kind: kind.into(), + id: id.into(), + members: members + .iter() + .map(|(k, v)| (k.to_string(), AttributeValue::Str(v.to_string()))) + .collect(), + }; + let report = vec![ + stat("media-source", "MS1", &[("trackIdentifier", "local-cam")]), + stat( + "outbound-rtp", + "O1", + &[("kind", "video"), ("mediaSourceId", "MS1"), ("rid", "f"), ("bytesSent", "10")], + ), + stat( + "outbound-rtp", + "O2", + &[("kind", "video"), ("mediaSourceId", "MS1"), ("rid", "h"), ("bytesSent", "5")], + ), + stat( + "inbound-rtp", + "I1", + &[("kind", "audio"), ("trackIdentifier", "TR_remote"), ("bytesReceived", "7")], + ), + stat( + "inbound-rtp", + "I2", + &[("kind", "audio"), ("trackIdentifier", "unknown"), ("bytesReceived", "1")], + ), + ]; + let tracks = HashMap::from([ + ("local-cam".to_owned(), "TR_cam".to_owned()), + ("TR_remote".to_owned(), "TR_remote".to_owned()), + ]); + let out = samples_from_peer_report(&report, &tracks, None); + assert_eq!(out.len(), 3, "two camera layers and the known remote track"); + assert!(out + .iter() + .filter(|s| s.track_sid == "TR_cam") + .all(|s| { s.kind == TrackKind::Video && s.direction == StreamDirection::Outbound })); + let remote = out.iter().find(|s| s.track_sid == "TR_remote").expect("remote"); + assert_eq!((remote.kind, remote.bytes), (TrackKind::Audio, Some(7))); + } +} From 87ebe37cc71e72171e8ff81335d5a897491f9f37 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:32:15 +0200 Subject: [PATCH 2/3] docs(telemetry): specify the cadence policy --- livekit-telemetry/SPEC.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/livekit-telemetry/SPEC.md b/livekit-telemetry/SPEC.md index ffe4c89e7..83d949d0a 100644 --- a/livekit-telemetry/SPEC.md +++ b/livekit-telemetry/SPEC.md @@ -220,6 +220,30 @@ cadence: on failure platforms: all — ios: authorization status, capture interruptions; android: permission checks, camera callbacks; web: DOMException names ``` +## Cadence policy + +`flush_interval × factor` and `stats_window × factor`, capped at 4× (60 s → 4 min at the default +cadence: one export and one RTC window per minute, conservative for the collector's shared +per-project quota). Factors multiply; a change applies at the next tick, and a *shorter* period +applies at once (pressure relieved → no waiting out a stretched period). + +| Condition | factor | source | +|---|---|---| +| thermal `serious` | 2 | host, `DeviceState.thermal` | +| thermal `critical` | 4 | host | +| memory pressure `warning` | 2 | host, `DeviceState.memory` | +| memory pressure `critical` | 4 | host | +| low-power mode | 2 | host | +| background | 2 | host | +| battery ≤ 20 % and unplugged | 2 | host, `DeviceState.battery_*` | +| constrained network (Low Data Mode / Data Saver) | 2 | host, `DeviceState.network_constrained` | +| encoder CPU-limited: an outbound track's `qualityLimitationDurations.cpu` grew within the last 60 s | 2 | core, from `record_stats` | + +CPU is never measured by the pipeline itself (measuring CPU costs CPU): thermal state is the OS's +judgement and `qualityLimitationReason` is WebRTC's. The core also paces the platform's +`getStats()` polling (`Scope::stats_poll_interval_ms`): every second while a subscribe waits for +its first media, else twice per (stretched) window — 30 s by default. + ## Log records A `TelemetryEvent` with an empty `name` is a plain log record (OTLP log without `event_name`): From 4c6c812df34317f2051cde1cf50daaaff9a3d9be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Thu, 22 Oct 2026 13:47:19 +0200 Subject: [PATCH 3/3] fix(telemetry): take inbound RTT from the selected pair and keep live tracks Inbound RTT now comes from the candidate pair the transport names as selected, else the first pair that is selected, or nominated and succeeded, as the JS port does; a stale succeeded pair listed first no longer wins. A reading without the CPU limitation counter keeps the previous baseline instead of resetting it to 0, which made the next reading look like CPU pressure. A track keeps its layer and CPU state through a close that follows a split, because state now follows the tracks that read since the last close, not the open windows. A NaN or infinite stats number counts as no reading. A report gets one timestamp, and each poll is read as a whole: each field of its one window reading is the largest value its streams report, so a stream without a value never erases another's and the order of a report changes nothing. The display bitrate spans the window's first and last reading and is left out without two. A layered outbound track's bytes and packets (its readings name their RTP streams) are a running total of every stream's growth since the previous poll; any other track, inbound included, passes its counters through as reported. A stream's first poll only sets its starting point, so a stream that comes and goes is never counted twice, and a track keeps only the previous poll's streams. Clearing counts a first poll that has not opened its window yet as one purged record. `network.connection.type` is reserved for the core. SPEC names the battery attributes the code emits (`lk.device.battery.level`, `lk.device.battery.charging`), now pinned by a test, and the `lk.rtc.stats.sample` contract moves into this stage. A closed window is captured through `QueuedEvent::new`, so its owner and attributes are taken at one moment. Device events convert with `From`, variant names are documented as the wire contract and pinned to SPEC, the cadence cap and battery buckets are named, the cadence helpers are crate-private, and every public field and variant has a doc line: the ones this commit adds name the value or attribute they are sent as. A changeset records the stage. --- .changeset/telemetry-device-rtc.md | 5 + livekit-telemetry/SPEC.md | 58 +- livekit-telemetry/src/device.rs | 419 +++++++++----- livekit-telemetry/src/event.rs | 4 +- livekit-telemetry/src/lib.rs | 9 +- livekit-telemetry/src/rtc.rs | 862 ++++++++++++++++++++++++----- 6 files changed, 1072 insertions(+), 285 deletions(-) create mode 100644 .changeset/telemetry-device-rtc.md diff --git a/.changeset/telemetry-device-rtc.md b/.changeset/telemetry-device-rtc.md new file mode 100644 index 000000000..caa117afa --- /dev/null +++ b/.changeset/telemetry-device-rtc.md @@ -0,0 +1,5 @@ +--- +livekit-telemetry: minor +--- + +Add device state and device events (`DeviceState`, `DeviceEvent`) and RTC stats samples (`RtcStatsSample`, `RtcStat`) with their on-device windowing. diff --git a/livekit-telemetry/SPEC.md b/livekit-telemetry/SPEC.md index 83d949d0a..f5424301b 100644 --- a/livekit-telemetry/SPEC.md +++ b/livekit-telemetry/SPEC.md @@ -10,9 +10,9 @@ owner and timestamp, are taken at capture, its session's (`lk.room.*`, `lk.parti `session.id`) included, so a record captured before a disconnect keeps its room; only pipeline-wide attributes are added at export. A record ships each key once: of a key it repeats, the last value supplied wins, and its session's keys replace its own. Attribute keys the SDK -owns are `lk.*`, `otel.*`, `code.*`, `session.id` and `error.type`: an app's attributes and -custom events cannot set them, and on a span only the core writes `lk.outcome` and `error.type` -(see [Spans](#spans)). +owns are `lk.*`, `otel.*`, `code.*`, `session.id`, `error.type` and `network.connection.type`: +an app's attributes and custom events cannot set them, and on a span only the core writes +`lk.outcome` and `error.type` (see [Spans](#spans)). ## Resource attributes @@ -183,8 +183,8 @@ platforms: ios, macos, android — web: Chromium only event: lk.device.battery.changed area: device attributes: - hw.battery.charge: double # 0.0–1.0 (OTel hardware semconv) - hw.battery.state: enum(charging | discharging) # OTel hardware semconv + lk.device.battery.level: int # percent, 0–100 + lk.device.battery.charging: bool cadence: on charging change and when the level crosses 20 % or 10 % unplugged — never per percent; silent where the level is unknown (desktops, tvOS) platforms: ios, android — optional elsewhere @@ -220,6 +220,54 @@ cadence: on failure platforms: all — ios: authorization status, capture interruptions; android: permission checks, camera callbacks; web: DOMException names ``` +```yaml +event: lk.rtc.stats.sample +area: rtc +severity: info +cadence: one per track and direction per stats window (default 60 s, stretched by the cadence + factor); closed early on background, when the track leaves (`track_ended`), at + disconnect and at shutdown. Produced by the core from raw readings — one getStats() + report per peer connection (`record_peer_stats`), polled when the core says + (`stats_poll_interval_ms`). Keyed by session and track: two rooms receiving the same + track keep two windows. +attributes: + lk.track.sid: string + lk.track.kind: enum(audio | video) + lk.track.direction: enum(inbound | outbound) + lk.rtc.codec: string # mimeType, when known + lk.rtc.window_ms: int # actual window length + lk.rtc.samples: int # readings (polls) in the window; one poll's layers are one reading + # cumulative counters — the last reading's value, monotonic (W3C webrtc-stats model). A + # layered outbound track's (one whose readings name their RTP streams) bytes and packets are its + # running total: each poll adds every stream's growth since the previous poll, so a stream's + # first poll only sets its starting point (a new or returning stream's first poll is not + # counted, never a double count), and the order of a report's streams changes nothing + lk.rtc.bytes: int + lk.rtc.packets: int + lk.rtc.packets_lost: int # inbound + lk.rtc.freeze_count: int # inbound video + lk.rtc.freezes_duration_ms: int # inbound video + lk.rtc.concealed_samples: int # inbound audio + lk.rtc.concealment_events: int # inbound audio + lk.rtc.jitter_buffer_delay_ms: int # inbound + lk.rtc.jitter_buffer_emitted_count: int # inbound + lk.rtc.quality_limitation.bandwidth_ms: int # outbound video + lk.rtc.quality_limitation.cpu_ms: int # outbound video + lk.rtc.quality_limitation.other_ms: int # outbound video + lk.rtc.pause_count: int # inbound video + lk.rtc.pauses_duration_ms: int # inbound video + lk.rtc.silent_concealed_samples: int # inbound audio + lk.rtc.interruption_count: int # inbound audio + lk.rtc.interruptions_duration_ms: int # inbound audio + # gauges — min / max / avg over the window's readings, one reading per poll; a poll's value is + # the largest any of its streams reports, so a value one stream lacks comes from the others + lk.rtc.jitter_ms.{min,max,avg}: double + lk.rtc.rtt_ms.{min,max,avg}: double # remote-inbound RTT for outbound, the selected candidate pair for inbound + lk.rtc.fps.{min,max,avg}: double # video + lk.rtc.audio_level.{min,max,avg}: double # audio, 0.0–1.0 linear +platforms: all +``` + ## Cadence policy `flush_interval × factor` and `stats_window × factor`, capped at 4× (60 s → 4 min at the default diff --git a/livekit-telemetry/src/device.rs b/livekit-telemetry/src/device.rs index d99e03e50..ce7714181 100644 --- a/livekit-telemetry/src/device.rs +++ b/livekit-telemetry/src/device.rs @@ -22,9 +22,13 @@ pub enum ThermalState { /// The platform has no thermal source: no pressure assumed, no change event. #[default] Unknown, + /// No thermal pressure; sent as `nominal`. Nominal, + /// Slightly raised thermal pressure; sent as `fair`. Fair, + /// High thermal pressure; sent as `serious`. Serious, + /// Pressure high enough to cut performance until the device cools; sent as `critical`. Critical, } @@ -32,8 +36,10 @@ pub enum ThermalState { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum AppState { + /// The app is in front of the user; sent as `foreground`. #[default] Foreground, + /// The app runs in the background; sent as `background`. Background, } @@ -42,9 +48,12 @@ pub enum AppState { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum MemoryPressure { + /// No memory pressure; sent as `normal`. #[default] Normal, + /// The OS warns that memory runs low; sent as `warning`. Warning, + /// Memory is critically low; sent as `critical`. Critical, } @@ -52,10 +61,14 @@ pub enum MemoryPressure { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum NetworkType { + /// The platform cannot tell the network type; sent as `unknown`. #[default] Unknown, + /// Wi-Fi; sent as `wifi`. Wifi, + /// A cellular network; sent as `cell`. Cell, + /// Ethernet or another wired link; sent as `wired`. Wired, /// A VPN tunnel is the active interface (Android `TRANSPORT_VPN`, WebRTC candidate `vpn`). Vpn, @@ -64,11 +77,12 @@ pub enum NetworkType { /// Connected over something the platform does not name (`NWInterface.InterfaceType.other`, /// Android `TRANSPORT_USB` / `LOWPAN`, …). Other, + /// No network path: the device is offline; sent as `unavailable`. Unavailable, } -/// What the host observes about the device; pushed with -/// [`Telemetry::set_device_state`](crate::Telemetry::set_device_state) whenever anything changes. +/// What the host observes about the device; pushed with `Telemetry::set_device_state` whenever +/// anything changes. /// /// The host owns the OS APIs (thermal, power, memory, battery, network path, lifecycle) — they /// are not reachable from Rust without a JVM/ObjC bridge — and the core owns what to do with @@ -77,16 +91,20 @@ pub enum NetworkType { /// costs CPU): thermal state is the OS's judgement, and WebRTC's own `qualityLimitationReason` /// (see [`RtcStatsSample`](crate::RtcStatsSample)) says when the encoder is CPU-starved. #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] -#[derive(Debug, Clone, Copy, PartialEq, Default)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub struct DeviceState { + /// The OS's thermal state; `Unknown` when the platform has no source. pub thermal: ThermalState, /// Low Power Mode / Battery Saver; `None` when the platform has no source (no change event, /// no stretch). #[cfg_attr(feature = "uniffi", uniffi(default))] pub low_power_mode: Option, + /// Whether the app is in front of the user. pub app_state: AppState, // No `uniffi(default)` on enum fields: uniffi 0.31's Swift generator rejects `Default` there. + /// The OS's memory pressure. pub memory: MemoryPressure, + /// The active network path. pub network: NetworkType, /// Cellular / personal hotspot (`NWPath.isExpensive`, metered on Android). #[cfg_attr(feature = "uniffi", uniffi(default = false))] @@ -98,21 +116,36 @@ pub struct DeviceState { /// Battery charge in percent; `None` when unknown (desktops, tvOS, monitoring disabled). #[cfg_attr(feature = "uniffi", uniffi(default))] pub battery_level: Option, + /// Plugged in and charging: a charging battery is never low. #[cfg_attr(feature = "uniffi", uniffi(default = false))] pub battery_charging: bool, } +/// The largest cadence factor: the flush interval and stats window stretch at most this much. +pub(crate) const MAX_CADENCE_FACTOR: u32 = 4; + +/// How much the battery level matters; per-percent updates only count when they cross a bucket. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum BatteryBucket { + /// Charging, unknown, or above 20 %. + Fine, + /// At most 20 % and unplugged: the cadence doubles. + Low, + /// At most 10 % and unplugged: uploads hold. + NearlyEmpty, +} + impl DeviceState { /// Below this, unplugged, the pipeline still records but stops uploading (a soft hold). const BATTERY_HOLD_PERCENT: u32 = 10; /// Below this, unplugged, the cadence doubles. const BATTERY_LOW_PERCENT: u32 = 20; - /// Multiplier on the flush interval and stats window: 1× at rest, up to 4× under pressure - /// (60 s → 4 min at the default cadence). Serious thermal, memory - /// warning, low-power mode, background, low battery and a constrained network each double - /// it; critical thermal or memory quadruple it. - pub fn cadence_factor(&self) -> u32 { + /// Multiplier on the flush interval and stats window: 1× at rest, up to + /// [`MAX_CADENCE_FACTOR`] under pressure (60 s → 4 min at the default cadence). Serious + /// thermal, memory warning, low-power mode, background, low battery and a constrained network + /// each double it; critical thermal or memory quadruple it. + pub(crate) fn cadence_factor(&self) -> u32 { let thermal = match self.thermal { ThermalState::Unknown | ThermalState::Nominal | ThermalState::Fair => 1, ThermalState::Serious => 2, @@ -126,35 +159,41 @@ impl DeviceState { let doubled = [ self.low_power_mode == Some(true), self.app_state == AppState::Background, - self.battery_bucket() >= 1, + self.battery_bucket() != BatteryBucket::Fine, self.network_constrained, ] .iter() .filter(|&&on| on) .count() as u32; - ((thermal * memory) << doubled).min(4) + ((thermal * memory) << doubled).min(MAX_CADENCE_FACTOR) } /// A soft hold: the user asked for less traffic (Low Data Mode / Data Saver) or the battery /// is nearly empty and unplugged. Uploads wait (at most a minute at a time); data keeps /// flowing into the cache meanwhile. Being offline (`network: Unavailable`) is a hard hold: /// nothing is attempted until the network is back. - pub fn holds_uploads(&self) -> bool { - self.network_constrained || self.battery_bucket() == 2 + pub(crate) fn holds_uploads(&self) -> bool { + self.network_constrained || self.battery_bucket() == BatteryBucket::NearlyEmpty } - /// 0 = fine or unknown, 1 = low (≤ 20 %), 2 = nearly empty (≤ 10 %); always 0 on charge. - /// Per-percent battery updates only matter when they cross one of these. - fn battery_bucket(&self) -> u8 { + /// Where the battery level stands; always [`BatteryBucket::Fine`] on charge. + fn battery_bucket(&self) -> BatteryBucket { match self.battery_level { - Some(level) if !self.battery_charging && level <= Self::BATTERY_HOLD_PERCENT => 2, - Some(level) if !self.battery_charging && level <= Self::BATTERY_LOW_PERCENT => 1, - _ => 0, + Some(level) if !self.battery_charging && level <= Self::BATTERY_HOLD_PERCENT => { + BatteryBucket::NearlyEmpty + } + Some(level) if !self.battery_charging && level <= Self::BATTERY_LOW_PERCENT => { + BatteryBucket::Low + } + _ => BatteryBucket::Fine, } } - /// The `SPEC.md` events describing what changed between `previous` and `self` - /// (everything, when there is no previous state: "initial value at session start"). + /// The `SPEC.md` events describing what changed between `previous` and `self` (every known + /// field, when there is no previous state: "initial value at session start"). Each event + /// carries a display body that repeats its values: the body takes the place of the name a + /// body-less event would show (see `SPEC.md`, Events), so it costs about what that name copy + /// would. pub(crate) fn change_events(&self, previous: Option<&DeviceState>) -> Vec { let mut events = Vec::new(); // An unknown reading is no reading: it never makes a change event. @@ -209,24 +248,20 @@ impl DeviceState { ); } // Per-percent updates are noise; report charging flips and bucket crossings. - if self.battery_level.is_some() - && previous.is_none_or(|p| { + if let Some(level) = self.battery_level.filter(|_| { + previous.is_none_or(|p| { p.battery_charging != self.battery_charging || p.battery_bucket() != self.battery_bucket() || p.battery_level.is_none() }) - { + }) { events.push( TelemetryEvent::new("lk.device.battery.changed") .with_body(format!( - "battery: {}%{}", - self.battery_level.unwrap_or(0), + "battery: {level}%{}", if self.battery_charging { ", charging" } else { "" } )) - .with_attribute( - "lk.device.battery.level", - self.battery_level.unwrap_or(0) as i64, - ) + .with_attribute("lk.device.battery.level", i64::from(level)) .with_attribute("lk.device.battery.charging", self.battery_charging), ); } @@ -280,6 +315,153 @@ impl NetworkType { } } +/// Why the audio route changed, in platform-neutral terms (AVAudioSession reasons; Android's +/// device added/removed callbacks map to `NewDevice` / `OldDeviceUnavailable`). +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AudioRouteReason { + /// An audio device became available, such as plugged-in headphones; sent as `new_device`. + NewDevice, + /// The audio device in use went away; sent as `old_device_unavailable`. + OldDeviceUnavailable, + /// The audio session category changed; sent as `category_change`. + CategoryChange, + /// The app overrode the route, such as forcing the speaker; sent as `override`. + Override, + /// The device woke from sleep; sent as `wake_from_sleep`. + WakeFromSleep, + /// No route suits the audio session category; sent as `no_suitable_route`. + NoSuitableRoute, + /// The configuration of the route's ports changed; sent as `route_configuration_change`. + RouteConfigurationChange, + /// The platform gives no reason; sent as `unknown`. + Unknown, +} + +/// Where the audio goes, in platform-neutral terms (AVAudioSession port types, Android's +/// `AudioDevice`; the web's `MediaDeviceInfo.kind` only ever gives `other`). +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AudioOutput { + /// The built-in loudspeaker; sent as `speaker`. + Speaker, + /// The phone earpiece (`builtInReceiver`, Android `Earpiece`). + Receiver, + /// Wired headphones or a wired headset; sent as `wired_headset`. + WiredHeadset, + /// A Bluetooth headset, headphones or speaker; sent as `bluetooth`. + Bluetooth, + /// A car's audio system; sent as `car_audio`. + CarAudio, + /// An AirPlay receiver; sent as `air_play`. + AirPlay, + /// An HDMI display or receiver; sent as `hdmi`. + Hdmi, + /// A USB audio device; sent as `usb`. + Usb, + /// An output the platform does not name; sent as `other`. + Other, +} + +/// Which capture device a failure is about. +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CaptureDevice { + /// A camera; sent as `camera`. + Camera, + /// A microphone; sent as `microphone`. + Microphone, + /// Screen capture; sent as `screen_share`. + ScreenShare, +} + +/// Why a capture device could not be used: the `getUserMedia` failure taxonomy, which every +/// platform maps onto (authorization status, `onCameraDisconnected`, `NotReadableError`, …). +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CaptureFailure { + /// The user (or MDM) said no: the most common pre-connect failure on mobile. + PermissionDenied, + /// No such device exists (`NotFoundError`); sent as `not_found`. + NotFound, + /// Another app holds the device (`NotReadableError`, `TrackStartError`). + InUse, + /// The device went away mid-capture (unplugged, `onCameraDisconnected`). + Disconnected, + /// Any other failure; sent as `other`. + Other, +} + +/// Things that happen to the device mid-call and explain what the media did next. Not state +/// (see [`DeviceState`]): each one is a record with a display body. +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DeviceEvent { + /// The audio output route changed. + AudioRouteChanged { + /// The outputs now in use, in platform order; sent in `lk.device.audio_route.outputs`. + outputs: Vec, + /// Why the route changed; sent in `lk.device.audio_route.reason`. + reason: AudioRouteReason, + }, + /// A phone call, Siri, another app taking audio focus: `began`, then the matching `ended`. + AudioInterruption { + /// Began (`true`) or ended (`false`); sent in `lk.device.audio.interruption`. + began: bool, + }, + /// A capture device could not be, or stay, used. + CaptureFailed { + /// The device that failed; sent in `lk.device.capture.device`. + device: CaptureDevice, + /// Why it failed; sent in `lk.device.capture.reason`. + reason: CaptureFailure, + }, +} + +/// A variant's wire value: its `Debug` name in snake case (`OldDeviceUnavailable` → +/// `old_device_unavailable`). Variant names are therefore the wire contract: renaming a variant +/// changes what ships, and the tests pin every value to `SPEC.md`. +pub(crate) fn snake_case(debug: impl std::fmt::Debug) -> String { + let mut out = String::new(); + for (i, c) in format!("{debug:?}").chars().enumerate() { + if c.is_uppercase() && i > 0 { + out.push('_'); + } + out.push(c.to_ascii_lowercase()); + } + out +} + +impl From for TelemetryEvent { + /// The `SPEC.md` event for a device event, with its display body. + fn from(event: DeviceEvent) -> Self { + match event { + DeviceEvent::AudioRouteChanged { outputs, reason } => { + let outputs = outputs.into_iter().map(snake_case).collect::>().join(","); + let reason = snake_case(reason); + TelemetryEvent::new("lk.device.audio_route.changed") + .with_body(format!("audio route: {outputs} ({reason})")) + .with_attribute("lk.device.audio_route.outputs", outputs) + .with_attribute("lk.device.audio_route.reason", reason) + } + DeviceEvent::AudioInterruption { began } => { + let phase = if began { "began" } else { "ended" }; + TelemetryEvent::new("lk.device.audio.interruption") + .with_body(format!("audio interruption {phase}")) + .with_attribute("lk.device.audio.interruption", phase) + } + DeviceEvent::CaptureFailed { device, reason } => { + let (device, reason) = (snake_case(device), snake_case(reason)); + TelemetryEvent::new("lk.device.capture.failed") + .with_severity(crate::Severity::Warn) + .with_body(format!("capture failed: {device} ({reason})")) + .with_attribute("lk.device.capture.device", device) + .with_attribute("lk.device.capture.reason", reason) + } + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -383,134 +565,87 @@ mod tests { let charging = DeviceState { battery_charging: true, ..at(10) }; assert_eq!(charging.change_events(Some(&at(10))).len(), 1, "plugged in"); } -} - -/// Why the audio route changed, in platform-neutral terms (AVAudioSession reasons; Android's -/// device added/removed callbacks map to `NewDevice` / `OldDeviceUnavailable`). -#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AudioRouteReason { - NewDevice, - OldDeviceUnavailable, - CategoryChange, - Override, - WakeFromSleep, - NoSuitableRoute, - RouteConfigurationChange, - Unknown, -} - -/// Where the audio goes, in platform-neutral terms (AVAudioSession port types, Android's -/// `AudioDevice`; the web's `MediaDeviceInfo.kind` only ever gives `other`). -#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AudioOutput { - Speaker, - /// The phone earpiece (`builtInReceiver`, Android `Earpiece`). - Receiver, - WiredHeadset, - Bluetooth, - CarAudio, - AirPlay, - Hdmi, - Usb, - Other, -} - -#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum CaptureDevice { - Camera, - Microphone, - ScreenShare, -} - -/// Why a capture device could not be used: the `getUserMedia` failure taxonomy, which every -/// platform maps onto (authorization status, `onCameraDisconnected`, `NotReadableError`, …). -#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum CaptureFailure { - /// The user (or MDM) said no: the most common pre-connect failure on mobile. - PermissionDenied, - NotFound, - /// Another app holds the device (`NotReadableError`, `TrackStartError`). - InUse, - /// The device went away mid-capture (unplugged, `onCameraDisconnected`). - Disconnected, - Other, -} -/// Things that happen to the device mid-call and explain what the media did next. Not state -/// (see [`DeviceState`]): each one is a record with a display body. -#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum DeviceEvent { - /// The outputs after the change, in the order the platform lists them. - AudioRouteChanged { outputs: Vec, reason: AudioRouteReason }, - /// A phone call, Siri, another app taking audio focus: `began`, then the matching `ended`. - AudioInterruption { began: bool }, - /// A capture device could not be, or stay, used. - CaptureFailed { device: CaptureDevice, reason: CaptureFailure }, -} - -pub(crate) fn snake(debug: impl std::fmt::Debug) -> String { - let mut out = String::new(); - for (i, c) in format!("{debug:?}").chars().enumerate() { - if c.is_uppercase() && i > 0 { - out.push('_'); - } - out.push(c.to_ascii_lowercase()); + #[test] + fn battery_events_carry_the_level_and_the_charging_flag() { + let low = DeviceState { battery_level: Some(15), ..Default::default() }; + let events = low.change_events(Some(&DeviceState::default())); + let [event] = &events[..] else { panic!("one battery event: {events:?}") }; + let value = |key: &str| event.attributes.iter().find(|a| a.key == key).map(|a| &a.value); + assert_eq!(value("lk.device.battery.level"), Some(&crate::AttributeValue::Int(15))); + assert_eq!(value("lk.device.battery.charging"), Some(&crate::AttributeValue::Bool(false))); } - out -} -impl DeviceEvent { - pub(crate) fn into_event(self) -> crate::TelemetryEvent { - use crate::{Severity, TelemetryEvent}; - match self { - Self::AudioRouteChanged { outputs, reason } => { - let outputs = outputs.into_iter().map(snake).collect::>().join(","); - let reason = snake(reason); - TelemetryEvent::new("lk.device.audio_route.changed") - .with_body(format!("audio route: {outputs} ({reason})")) - .with_attribute("lk.device.audio_route.outputs", outputs) - .with_attribute("lk.device.audio_route.reason", reason) - } - Self::AudioInterruption { began } => { - let phase = if began { "began" } else { "ended" }; - TelemetryEvent::new("lk.device.audio.interruption") - .with_body(format!("audio interruption {phase}")) - .with_attribute("lk.device.audio.interruption", phase) - } - Self::CaptureFailed { device, reason } => { - let (device, reason) = (snake(device), snake(reason)); - TelemetryEvent::new("lk.device.capture.failed") - .with_severity(Severity::Warn) - .with_body(format!("capture failed: {device} ({reason})")) - .with_attribute("lk.device.capture.device", device) - .with_attribute("lk.device.capture.reason", reason) - } - } + #[test] + fn the_core_owns_the_network_connection_type() { + assert!(!crate::event::valid_custom("network.connection.type", None)); + assert!(crate::event::valid_custom("network.peer.address", None), "other `network.*`"); } -} -#[cfg(test)] -mod event_tests { - use super::*; + /// Every enum value that ships, pinned to the lists in `SPEC.md`. + #[test] + fn wire_values_match_the_spec() { + let names = |values: Vec| values.join(" | "); + use AudioRouteReason as R; + let reasons = [ + R::NewDevice, + R::OldDeviceUnavailable, + R::CategoryChange, + R::Override, + R::WakeFromSleep, + R::NoSuitableRoute, + R::RouteConfigurationChange, + R::Unknown, + ]; + assert_eq!( + names(reasons.into_iter().map(snake_case).collect()), + "new_device | old_device_unavailable | category_change | override | wake_from_sleep \ + | no_suitable_route | route_configuration_change | unknown" + ); + use AudioOutput as O; + let outputs = [ + O::Speaker, + O::Receiver, + O::WiredHeadset, + O::Bluetooth, + O::CarAudio, + O::AirPlay, + O::Hdmi, + O::Usb, + O::Other, + ]; + assert_eq!( + names(outputs.into_iter().map(snake_case).collect()), + "speaker | receiver | wired_headset | bluetooth | car_audio | air_play | hdmi | usb \ + | other" + ); + let devices = + [CaptureDevice::Camera, CaptureDevice::Microphone, CaptureDevice::ScreenShare]; + assert_eq!( + names(devices.into_iter().map(snake_case).collect()), + "camera | microphone | screen_share" + ); + use CaptureFailure as F; + let failures = [F::PermissionDenied, F::NotFound, F::InUse, F::Disconnected, F::Other]; + assert_eq!( + names(failures.into_iter().map(snake_case).collect()), + "permission_denied | not_found | in_use | disconnected | other" + ); + } #[test] fn device_events_have_bodies_and_snake_case_values() { - let event = DeviceEvent::AudioRouteChanged { + let event: TelemetryEvent = DeviceEvent::AudioRouteChanged { outputs: vec![AudioOutput::Speaker], reason: AudioRouteReason::OldDeviceUnavailable, } - .into_event(); + .into(); assert_eq!(event.body.as_deref(), Some("audio route: speaker (old_device_unavailable)")); - let denied = DeviceEvent::CaptureFailed { + let denied: TelemetryEvent = DeviceEvent::CaptureFailed { device: CaptureDevice::ScreenShare, reason: CaptureFailure::PermissionDenied, } - .into_event(); + .into(); assert_eq!( denied.body.as_deref(), Some("capture failed: screen_share (permission_denied)") diff --git a/livekit-telemetry/src/event.rs b/livekit-telemetry/src/event.rs index a0dd3cfb1..535e011f6 100644 --- a/livekit-telemetry/src/event.rs +++ b/livekit-telemetry/src/event.rs @@ -306,13 +306,15 @@ pub(crate) const MAX_CUSTOM_ATTRIBUTES: usize = 64; /// Keys the SDK owns: an app can neither set nor override them — `lk.*` (room, participant, /// track, outcome), `otel.*` (OTel's own, `otel.event.name` among them), `code.*` (where a log -/// line was written), `session.id` and `error.type`. +/// line was written), `session.id`, `error.type` and `network.connection.type` (the device's +/// network path; other `network.*` keys stay the app's). pub(crate) fn is_reserved(key: &str) -> bool { key.starts_with("lk.") || key.starts_with("otel.") || key.starts_with("code.") || key == "session.id" || key == "error.type" + || key == "network.connection.type" } /// Whether an app-provided attribute is within the limits and outside the SDK's namespace. diff --git a/livekit-telemetry/src/lib.rs b/livekit-telemetry/src/lib.rs index b3920cf6b..75788c4f9 100644 --- a/livekit-telemetry/src/lib.rs +++ b/livekit-telemetry/src/lib.rs @@ -42,9 +42,12 @@ mod stats; mod device; /// RTC stats samples and their on-device windowing. -#[expect( - dead_code, - reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" + ) )] mod rtc; diff --git a/livekit-telemetry/src/rtc.rs b/livekit-telemetry/src/rtc.rs index 7076cc8ee..69a988dc7 100644 --- a/livekit-telemetry/src/rtc.rs +++ b/livekit-telemetry/src/rtc.rs @@ -12,20 +12,25 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{collections::HashMap, sync::Arc, time::Duration}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, + time::Duration, +}; use tokio::time::Instant; use crate::{ - event::now_unix_nanos, scope::ScopeState, store::QueuedEvent, Attribute, AttributeValue, - TelemetryEvent, + event::now_unix_nanos, scope::ScopeState, store::QueuedEvent, AttributeValue, TelemetryEvent, }; /// Audio or video. #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum TrackKind { + /// An audio track; sent as `audio`. Audio, + /// A video track; sent as `video`. Video, } @@ -33,7 +38,9 @@ pub enum TrackKind { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum StreamDirection { + /// Media this device receives; sent as `inbound`. Inbound, + /// Media this device sends; sent as `outbound`. Outbound, } @@ -41,59 +48,94 @@ pub enum StreamDirection { /// /// Cumulative counters (`bytes`, `packets`, `freeze_count`, …) are passed through as reported — /// monotonic, paired with their denominators (the W3C webrtc-stats model) so any layer can -/// recompute rates and a dropped window never corrupts the next. Gauges (`jitter_ms`, `rtt_ms`, +/// recompute rates and a dropped window never corrupts the next. An outbound track whose readings +/// name their RTP streams (`layer`) ships its bytes and packets as a running total of each +/// stream's growth instead, so a stream that comes and goes is never counted twice. Gauges (`jitter_ms`, `rtt_ms`, /// `frames_per_second`, `audio_level`) are summarised per window as min/max/avg. Fields a /// platform or direction does not have stay `None` and are omitted from the wire. #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq)] pub struct RtcStatsSample { + /// The track's sid (`TR_…`). pub track_sid: String, + /// Audio or video. pub kind: TrackKind, + /// Sent (outbound) or received (inbound). pub direction: StreamDirection, + /// The codec's `mimeType` (`video/VP8`), resolved through the stream's `codecId`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub codec: Option, // Cumulative counters. + /// `bytesSent` (outbound) or `bytesReceived` (inbound). #[cfg_attr(feature = "uniffi", uniffi(default))] pub bytes: Option, + /// `packetsSent` (outbound) or `packetsReceived` (inbound). #[cfg_attr(feature = "uniffi", uniffi(default))] pub packets: Option, + /// `packetsLost`, inbound. #[cfg_attr(feature = "uniffi", uniffi(default))] pub packets_lost: Option, + /// `freezeCount`, inbound video: short stalls, a frame shown at least 3× the average frame + /// time, and at least 150 ms more than it, after the one before. A longer gap is a pause + /// instead. #[cfg_attr(feature = "uniffi", uniffi(default))] pub freeze_count: Option, + /// `totalFreezesDuration` in milliseconds, inbound video. #[cfg_attr(feature = "uniffi", uniffi(default))] pub freezes_duration_ms: Option, + /// `concealedSamples`, inbound audio: samples the receiver made up for lost or late packets. #[cfg_attr(feature = "uniffi", uniffi(default))] pub concealed_samples: Option, + /// `concealmentEvents`, inbound audio: how many separate runs of concealment started (one loss + /// burst is one event, however many samples it conceals). #[cfg_attr(feature = "uniffi", uniffi(default))] pub concealment_events: Option, + /// `jitterBufferDelay` in milliseconds, inbound: the total time emitted samples or frames spent + /// in the jitter buffer. Divide by `jitter_buffer_emitted_count` for the average. #[cfg_attr(feature = "uniffi", uniffi(default))] pub jitter_buffer_delay_ms: Option, + /// `jitterBufferEmittedCount`, inbound: samples or frames that left the jitter buffer. #[cfg_attr(feature = "uniffi", uniffi(default))] pub jitter_buffer_emitted_count: Option, + /// `qualityLimitationDurations.bandwidth` in milliseconds, outbound video. #[cfg_attr(feature = "uniffi", uniffi(default))] pub quality_limitation_bandwidth_ms: Option, + /// `qualityLimitationDurations.cpu` in milliseconds, outbound video: time the encoder was + /// limited by the CPU. #[cfg_attr(feature = "uniffi", uniffi(default))] pub quality_limitation_cpu_ms: Option, + /// `qualityLimitationDurations.other` in milliseconds, outbound video. #[cfg_attr(feature = "uniffi", uniffi(default))] pub quality_limitation_other_ms: Option, + /// `pauseCount`, inbound video: gaps of more than 5 s without a frame (usually the sender + /// paused). A shorter stall is a freeze instead. #[cfg_attr(feature = "uniffi", uniffi(default))] pub pause_count: Option, + /// `totalPausesDuration` in milliseconds, inbound video. #[cfg_attr(feature = "uniffi", uniffi(default))] pub pauses_duration_ms: Option, + /// `silentConcealedSamples`, inbound audio: the concealed samples that were silence, which a + /// listener cannot hear. Part of `concealed_samples`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub silent_concealed_samples: Option, + /// `interruptionCount`, inbound audio: concealment runs of 150 ms or more, long enough to hear. #[cfg_attr(feature = "uniffi", uniffi(default))] pub interruption_count: Option, + /// `totalInterruptionDuration` in milliseconds, inbound audio. #[cfg_attr(feature = "uniffi", uniffi(default))] pub interruptions_duration_ms: Option, // Gauges. + /// `jitter` in milliseconds, inbound. #[cfg_attr(feature = "uniffi", uniffi(default))] pub jitter_ms: Option, + /// Round-trip time in milliseconds: the `remote-inbound-rtp` `roundTripTime` (outbound), the + /// selected candidate pair's `currentRoundTripTime` (inbound). #[cfg_attr(feature = "uniffi", uniffi(default))] pub rtt_ms: Option, + /// `framesPerSecond`, video. #[cfg_attr(feature = "uniffi", uniffi(default))] pub frames_per_second: Option, + /// `audioLevel`, inbound audio: 0.0 to 1.0, linear (not dB). #[cfg_attr(feature = "uniffi", uniffi(default))] pub audio_level: Option, /// When the reading was taken; `None` = now. @@ -112,11 +154,15 @@ pub struct RtcStatsSample { #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq)] pub struct RtcStat { + /// The stat's `type`: `inbound-rtp`, `candidate-pair`, `codec`, … pub kind: String, + /// The stat's `id`, which other stats name (`codecId`, `localId`, `mediaSourceId`). pub id: String, + /// Its members by W3C name. pub members: HashMap, } +/// A numeric member as a finite `f64`: a NaN or an infinity is no reading. fn num(stat: &RtcStat, key: &str) -> Option { match stat.members.get(key)? { AttributeValue::Int(i) => Some(*i as f64), @@ -124,8 +170,10 @@ fn num(stat: &RtcStat, key: &str) -> Option { AttributeValue::Str(s) => s.parse().ok(), AttributeValue::Bool(_) => None, } + .filter(|v: &f64| v.is_finite()) } +/// A counter member as a whole number (negative readings count as 0). fn count(stat: &RtcStat, key: &str) -> Option { num(stat, key).map(|v| v.max(0.0) as u64) } @@ -135,6 +183,12 @@ fn ms(stat: &RtcStat, key: &str) -> Option { num(stat, key).map(|s| (s.max(0.0) * 1000.0) as u64) } +/// A duration member (seconds) as fractional milliseconds, for gauges. +fn ms_f64(stat: &RtcStat, key: &str) -> Option { + num(stat, key).map(|s| s * 1000.0) +} + +/// A string member. fn text(stat: &RtcStat, key: &str) -> Option { match stat.members.get(key)? { AttributeValue::Str(s) => Some(s.clone()), @@ -144,7 +198,7 @@ fn text(stat: &RtcStat, key: &str) -> Option { /// The samples in one track's `getStats()` report: one per `outbound-rtp` (tagged with its /// layer) or `inbound-rtp` entry of `direction`, codec resolved through `codecId`, RTT from the -/// `remote-inbound-rtp` that reports on the stream (outbound) or the nominated candidate pair +/// `remote-inbound-rtp` that reports on the stream (outbound) or the selected candidate pair /// (inbound), durations converted from seconds to milliseconds. One mapping for every SDK. pub(crate) fn samples_from_report( track_sid: &str, @@ -153,7 +207,9 @@ pub(crate) fn samples_from_report( report: &[RtcStat], timestamp_ns: Option, ) -> Vec { - let index = Index::new(report); + // One poll, one time: its layers then fold into one reading (see `Window::add`). + let timestamp_ns = Some(timestamp_ns.unwrap_or_else(now_unix_nanos)); + let index = ReportIndex::new(report); index .rtp .iter() @@ -172,7 +228,8 @@ pub(crate) fn samples_from_peer_report( tracks: &HashMap, timestamp_ns: Option, ) -> Vec { - let index = Index::new(report); + let timestamp_ns = Some(timestamp_ns.unwrap_or_else(now_unix_nanos)); + let index = ReportIndex::new(report); index .rtp .iter() @@ -196,16 +253,16 @@ pub(crate) fn samples_from_peer_report( } /// A report indexed once: entries by id, the `remote-inbound-rtp` reporting on each outbound -/// stream, the nominated candidate pair's RTT, the RTP streams with their direction. Resolving a +/// stream, the selected candidate pair's RTT, the RTP streams with their direction. Resolving a /// stream's codec, RTT or media source is then a lookup, not a scan of the report. -struct Index<'a> { +struct ReportIndex<'a> { by_id: HashMap<&'a str, &'a RtcStat>, remote_by_local: HashMap, pair_rtt_ms: Option, rtp: Vec<(&'a RtcStat, StreamDirection)>, } -impl<'a> Index<'a> { +impl<'a> ReportIndex<'a> { fn new(report: &'a [RtcStat]) -> Self { let mut index = Self { by_id: HashMap::with_capacity(report.len()), @@ -213,6 +270,11 @@ impl<'a> Index<'a> { pair_rtt_ms: None, rtp: Vec::new(), }; + // The transport's selected pair, else the first pair that is selected, or nominated and + // succeeded (not every platform names the selected pair). + let (mut selected, mut fallback) = (None, None); + let flag = + |stat: &RtcStat, key| matches!(stat.members.get(key), Some(AttributeValue::Bool(true))); for stat in report { index.by_id.insert(stat.id.as_str(), stat); match stat.kind.as_str() { @@ -223,17 +285,23 @@ impl<'a> Index<'a> { index.remote_by_local.insert(local, stat); } } - "candidate-pair" if index.pair_rtt_ms.is_none() => { - let nominated = - matches!(stat.members.get("nominated"), Some(AttributeValue::Bool(true))) - || text(stat, "state").as_deref() == Some("succeeded"); - if nominated { - index.pair_rtt_ms = num(stat, "currentRoundTripTime").map(|s| s * 1000.0); + "transport" => { + selected = selected.or_else(|| text(stat, "selectedCandidatePairId")) + } + "candidate-pair" if fallback.is_none() => { + let succeeded = text(stat, "state").as_deref() == Some("succeeded"); + if flag(stat, "selected") || (flag(stat, "nominated") && succeeded) { + fallback = Some(stat); } } _ => {} } } + let pair = selected + .and_then(|id| index.by_id.get(id.as_str()).copied()) + .filter(|pair| pair.kind == "candidate-pair") + .or(fallback); + index.pair_rtt_ms = pair.and_then(|pair| ms_f64(pair, "currentRoundTripTime")); index } @@ -265,8 +333,7 @@ impl<'a> Index<'a> { sample.rtt_ms = self .remote_by_local .get(&stat.id) - .and_then(|remote| num(remote, "roundTripTime")) - .map(|s| s * 1000.0); + .and_then(|remote| ms_f64(remote, "roundTripTime")); } StreamDirection::Inbound => { sample.bytes = count(stat, "bytesReceived"); @@ -283,7 +350,7 @@ impl<'a> Index<'a> { sample.interruptions_duration_ms = ms(stat, "totalInterruptionDuration"); sample.jitter_buffer_delay_ms = ms(stat, "jitterBufferDelay"); sample.jitter_buffer_emitted_count = count(stat, "jitterBufferEmittedCount"); - sample.jitter_ms = num(stat, "jitter").map(|s| s * 1000.0); + sample.jitter_ms = ms_f64(stat, "jitter"); sample.frames_per_second = num(stat, "framesPerSecond"); sample.audio_level = num(stat, "audioLevel"); sample.rtt_ms = self.pair_rtt_ms; @@ -293,7 +360,44 @@ impl<'a> Index<'a> { } } +/// Keep the larger of `mine` and `theirs`, whichever is known. +fn larger(mine: &mut Option, theirs: Option) { + if let Some(theirs) = theirs { + if mine.as_ref().is_none_or(|mine| theirs > *mine) { + *mine = Some(theirs); + } + } +} + impl RtcStatsSample { + /// Merge another stream of the same poll into this reading: each field keeps the larger of + /// the values known, so a stream without a field never erases it and the report's order + /// changes nothing. + fn merge(&mut self, other: RtcStatsSample) { + larger(&mut self.codec, other.codec); + larger(&mut self.bytes, other.bytes); + larger(&mut self.packets, other.packets); + larger(&mut self.packets_lost, other.packets_lost); + larger(&mut self.freeze_count, other.freeze_count); + larger(&mut self.freezes_duration_ms, other.freezes_duration_ms); + larger(&mut self.concealed_samples, other.concealed_samples); + larger(&mut self.concealment_events, other.concealment_events); + larger(&mut self.jitter_buffer_delay_ms, other.jitter_buffer_delay_ms); + larger(&mut self.jitter_buffer_emitted_count, other.jitter_buffer_emitted_count); + larger(&mut self.quality_limitation_bandwidth_ms, other.quality_limitation_bandwidth_ms); + larger(&mut self.quality_limitation_cpu_ms, other.quality_limitation_cpu_ms); + larger(&mut self.quality_limitation_other_ms, other.quality_limitation_other_ms); + larger(&mut self.pause_count, other.pause_count); + larger(&mut self.pauses_duration_ms, other.pauses_duration_ms); + larger(&mut self.silent_concealed_samples, other.silent_concealed_samples); + larger(&mut self.interruption_count, other.interruption_count); + larger(&mut self.interruptions_duration_ms, other.interruptions_duration_ms); + larger(&mut self.jitter_ms, other.jitter_ms); + larger(&mut self.rtt_ms, other.rtt_ms); + larger(&mut self.frames_per_second, other.frames_per_second); + larger(&mut self.audio_level, other.audio_level); + } + /// A sample with every optional field unset. pub fn new(track_sid: impl Into, kind: TrackKind, direction: StreamDirection) -> Self { Self { @@ -351,28 +455,32 @@ impl Gauge { self.n += 1; } + /// The average over the window; `None` without a reading. + fn avg(&self) -> Option { + (self.n > 0).then(|| self.sum / self.n as f64) + } + fn attach(&self, event: TelemetryEvent, key: &str) -> TelemetryEvent { - if self.n == 0 { - return event; - } + let Some(avg) = self.avg() else { return event }; event .with_attribute(format!("{key}.min"), self.min) .with_attribute(format!("{key}.max"), self.max) - .with_attribute(format!("{key}.avg"), self.sum / self.n as f64) + .with_attribute(format!("{key}.avg"), avg) } } /// Samples of one track in one direction accumulated since the window opened. struct Window { + /// The session the window ships under. A change to its project or correlation attributes + /// closes the window first (see `StatsWindows::split`), so the owner at close is the owner the + /// readings were taken under. session: Arc, - /// The window's owner, captured when it opened: the session's project and its correlation - /// attributes then. A change to either closes the window first (see `StatsWindows::split`). - route: Option, - custom: Vec, start_ns: u64, + /// Readings in the window: one per poll. samples: u32, /// Cumulative counters at the window's first reading, for the display body's deltas. first: RtcStatsSample, + /// The newest reading. last: RtcStatsSample, jitter: Gauge, rtt: Gauge, @@ -383,28 +491,32 @@ struct Window { impl Window { fn open(start_ns: u64, first: RtcStatsSample, session: Arc) -> Self { let mut window = Self { - route: session.route(), - custom: session.custom_snapshot(), session, start_ns, samples: 0, first: first.clone(), - last: first.clone(), + last: first, jitter: Gauge::default(), rtt: Gauge::default(), fps: Gauge::default(), audio_level: Gauge::default(), }; - window.add(first); + window.commit(); window } - /// `video outbound: 1204 kbps, loss 0.4%, rtt 48 ms, 30 fps` — deltas over the window. - fn summary(&self, window_ms: u64) -> String { + /// `video inbound: 1204 kbps, loss 0.4%, rtt 48 ms, 30 fps` — deltas between the window's + /// first and last reading. The bitrate is over the time between those two readings, and is + /// left out without two distinct reading times. + fn summary(&self) -> String { let mut parts = Vec::new(); let delta = |a: Option, b: Option| Some(a?.saturating_sub(b?)); - if let (Some(bytes), true) = (delta(self.last.bytes, self.first.bytes), window_ms > 0) { - parts.push(format!("{} kbps", bytes * 8 / window_ms)); + let interval_ms = + delta(self.last.timestamp_ns, self.first.timestamp_ns).map(|ns| ns / 1_000_000); + if let (Some(bytes), Some(ms)) = + (delta(self.last.bytes, self.first.bytes), interval_ms.filter(|ms| *ms > 0)) + { + parts.push(format!("{} kbps", bytes * 8 / ms)); } if let (Some(lost), Some(packets)) = ( delta(self.last.packets_lost, self.first.packets_lost), @@ -414,11 +526,11 @@ impl Window { parts.push(format!("loss {:.1}%", lost as f64 * 100.0 / (lost + packets) as f64)); } } - if self.rtt.n > 0 { - parts.push(format!("rtt {:.0} ms", self.rtt.sum / self.rtt.n as f64)); + if let Some(rtt) = self.rtt.avg() { + parts.push(format!("rtt {rtt:.0} ms")); } - if self.fps.n > 0 { - parts.push(format!("{:.0} fps", self.fps.sum / self.fps.n as f64)); + if let Some(fps) = self.fps.avg() { + parts.push(format!("{fps:.0} fps")); } if let Some(freezes) = delta(self.last.freeze_count, self.first.freeze_count).filter(|f| *f > 0) @@ -433,28 +545,25 @@ impl Window { ) } - /// The closed window as a queued record, with the owner it captured when it opened and its - /// session's attributes as of now, the window's capture time (see `QueuedEvent::new`). + /// The closed window as a record captured now (see `QueuedEvent::new`). fn into_queued(self, end_ns: u64) -> QueuedEvent { - let (session, route, custom) = - (self.session.clone(), self.route.clone(), self.custom.clone()); - let mut event = self.into_event(end_ns); - for attribute in custom { - if !event.attributes.iter().any(|a| a.key == attribute.key) { - event.attributes.push(attribute); - } - } - session.decorate(&mut event.attributes); - QueuedEvent { event, session, route } + let session = Arc::clone(&self.session); + QueuedEvent::new(self.into_event(end_ns), session) } - fn add(&mut self, sample: RtcStatsSample) { + /// One more poll's reading. + fn add(&mut self, reading: RtcStatsSample) { + self.last = reading; + self.commit(); + } + + /// Put the newest reading into the gauges. + fn commit(&mut self) { self.samples += 1; - self.jitter.add(sample.jitter_ms); - self.rtt.add(sample.rtt_ms); - self.fps.add(sample.frames_per_second); - self.audio_level.add(sample.audio_level); - self.last = sample; + self.jitter.add(self.last.jitter_ms); + self.rtt.add(self.last.rtt_ms); + self.fps.add(self.last.frames_per_second); + self.audio_level.add(self.last.audio_level); } /// The `lk.rtc.stats.sample` event for this window, stamped at `end_ns`. Its body is the @@ -462,7 +571,7 @@ impl Window { /// the attributes carry the numbers. fn into_event(self, end_ns: u64) -> TelemetryEvent { let window_ms = end_ns.saturating_sub(self.start_ns) / 1_000_000; - let body = self.summary(window_ms); + let body = self.summary(); let last = self.last; let mut event = TelemetryEvent::new("lk.rtc.stats.sample") .with_body(body) @@ -519,9 +628,11 @@ pub(crate) struct StatsWindows { /// Last cumulative `qualityLimitationDurations.cpu` per outbound track. limitation: HashMap, cpu_limited_until: Option, - /// Last cumulative counters per simulcast layer, per outbound track: the track's counters - /// are their sum, so a suspended top layer cannot freeze them. - layers: HashMap>, + /// Per track: the poll in progress and the counting state across polls. + tracks: HashMap, + /// Tracks with a reading since the last [`close`](Self::close): only these keep their stream + /// and CPU state. A window closed early (a split) does not retire its track. + seen: HashSet, /// The opt-out, checked under the lock that guards the windows (see `Store`). revoked: Arc, } @@ -534,22 +645,92 @@ fn track_key(session: &ScopeState, sample: &RtcStatsSample) -> TrackKey { (session.trace_id, sample.track_sid.clone(), sample.direction) } +/// One track's state across polls. +#[derive(Default)] +struct Track { + /// The poll being gathered: a report's streams arrive one by one, and the poll is read as a + /// whole once the next poll comes or its window closes. + poll: Option, + /// The previous poll's bytes and packets per RTP stream (`layer`): one report's streams at + /// most, no older ones. + previous: HashMap, + /// A layered outbound track's bytes and packets so far: every poll's growth, summed. + total: StreamCounts, +} + +/// One poll's readings, as the streams of one report came in. +struct Poll { + session: Arc, + timestamp_ns: u64, + readings: Vec, +} + +/// A stream's, or a track's, bytes and packets. #[derive(Debug, Clone, Copy, Default)] -struct LayerCounters { +struct StreamCounts { bytes: Option, packets: Option, - fps: Option, - limitation_bandwidth_ms: Option, - limitation_cpu_ms: Option, - limitation_other_ms: Option, } -fn sum(values: impl Iterator>) -> Option { - values.flatten().reduce(|a, b| a + b) +impl Track { + /// A poll's readings as one: each field the largest value its streams report, so their + /// order changes nothing. On a layered outbound track (streams named by `layer`), the bytes + /// and packets are the track's running total instead: each named stream adds its growth since + /// the previous poll. Any other track passes its counters through as reported. A stream that was not in the previous poll only sets its starting point, + /// so a new or returning stream's first poll is not counted (an undercount, never a double + /// count), and an absent counter keeps the starting point it had. + fn reading(&mut self, readings: Vec) -> Option { + // A track's readings all share its direction (it is part of the track's key). + let named = + readings.iter().any(|r| r.direction == StreamDirection::Outbound && r.layer.is_some()); + let mut streams: HashMap = HashMap::new(); + let mut reading: Option = None; + for sample in readings { + if named { + let stream = streams.entry(sample.layer.clone().unwrap_or_default()).or_default(); + larger(&mut stream.bytes, sample.bytes); + larger(&mut stream.packets, sample.packets); + } + match &mut reading { + Some(reading) => reading.merge(sample), + None => reading = Some(sample), + } + } + let mut reading = reading?; + reading.layer = None; + if named { + let mut grown = StreamCounts::default(); + for (id, now) in &mut streams { + let before = self.previous.get(id).copied().unwrap_or_default(); + grow(&mut grown.bytes, now.bytes, before.bytes); + grow(&mut grown.packets, now.packets, before.packets); + now.bytes = now.bytes.or(before.bytes); + now.packets = now.packets.or(before.packets); + } + self.previous = streams; + add(&mut self.total.bytes, grown.bytes); + add(&mut self.total.packets, grown.packets); + reading.bytes = self.total.bytes; + reading.packets = self.total.packets; + } + Some(reading) + } } -fn max_u64(values: impl Iterator>) -> Option { - values.flatten().max() +/// Count a stream's growth into `grown`: known `now` against a known `before`. A stream with +/// `now` known still makes the poll's count known (`Some(0)` at least), so a track's first poll +/// reports a total. +fn grow(grown: &mut Option, now: Option, before: Option) { + let Some(now) = now else { return }; + let growth = before.map_or(0, |before| now.saturating_sub(before)); + *grown = Some(grown.unwrap_or(0) + growth); +} + +/// Add a poll's growth to a running total. +fn add(total: &mut Option, grown: Option) { + if let Some(grown) = grown { + *total = Some(total.unwrap_or(0) + grown); + } } /// CPU starvation is sticky: stretch the cadence for a while after the last sign of it. @@ -564,60 +745,66 @@ impl StatsWindows { Self { revoked, ..Self::default() } } + /// Add one reading to its track's poll in progress. A reading with another timestamp starts + /// the next poll, and the finished one goes into the track's window as one reading. pub fn record_in(&mut self, mut sample: RtcStatsSample, session: &Arc) { if self.revoked.load(std::sync::atomic::Ordering::SeqCst) { return; } let key = track_key(session, &sample); - self.fold_layers(&key, &mut sample); - self.track_limitation(&key, &sample); - let timestamp = *sample.timestamp_ns.get_or_insert_with(now_unix_nanos); - match self.windows.get_mut(&key) { - Some(window) => window.add(sample), + self.seen.insert(key.clone()); + let timestamp_ns = *sample.timestamp_ns.get_or_insert_with(now_unix_nanos); + let track = self.tracks.entry(key.clone()).or_default(); + if track.poll.as_ref().is_some_and(|poll| poll.timestamp_ns != timestamp_ns) { + self.end_poll(&key); + } + self.check_limitation(&key, &sample); + let track = self.tracks.entry(key).or_default(); + let poll = track.poll.get_or_insert_with(|| Poll { + session: session.clone(), + timestamp_ns, + readings: Vec::new(), + }); + poll.readings.push(sample); + } + + /// Read the track's poll in progress as one reading into its window, opening the window if + /// needed. + fn end_poll(&mut self, key: &TrackKey) { + let Some(track) = self.tracks.get_mut(key) else { return }; + let Some(Poll { session, timestamp_ns, readings }) = track.poll.take() else { return }; + let Some(reading) = track.reading(readings) else { return }; + self.set_limitation_baseline(key, &reading); + match self.windows.get_mut(key) { + Some(window) => window.add(reading), None => { - self.windows.insert(key, Window::open(timestamp, sample, session.clone())); + self.windows.insert(key.clone(), Window::open(timestamp_ns, reading, session)); } } } - /// Simulcast publishes one RTP stream per layer under one track sid. Remember each layer's - /// latest cumulative counters and rewrite the sample as the track's total, so the window - /// sees one monotonic series per track. - fn fold_layers(&mut self, key: &TrackKey, sample: &mut RtcStatsSample) { - let Some(layer) = sample.layer.clone() else { return }; - let layers = self.layers.entry(key.clone()).or_default(); - layers.insert( - layer, - LayerCounters { - bytes: sample.bytes, - packets: sample.packets, - fps: sample.frames_per_second, - limitation_bandwidth_ms: sample.quality_limitation_bandwidth_ms, - limitation_cpu_ms: sample.quality_limitation_cpu_ms, - limitation_other_ms: sample.quality_limitation_other_ms, - }, - ); - sample.bytes = sum(layers.values().map(|l| l.bytes)); - sample.packets = sum(layers.values().map(|l| l.packets)); - sample.frames_per_second = layers - .values() - .filter_map(|l| l.fps) - .fold(None, |m, v| Some(m.map_or(v, |m: f64| m.max(v)))); - sample.quality_limitation_bandwidth_ms = - max_u64(layers.values().map(|l| l.limitation_bandwidth_ms)); - sample.quality_limitation_cpu_ms = max_u64(layers.values().map(|l| l.limitation_cpu_ms)); - sample.quality_limitation_other_ms = - max_u64(layers.values().map(|l| l.limitation_other_ms)); + /// End every poll in progress that `ended` selects. + fn end_polls(&mut self, ended: impl Fn(&TrackKey) -> bool) { + let keys: Vec = self + .tracks + .iter() + .filter(|(key, track)| track.poll.is_some() && ended(key)) + .map(|(key, _)| key.clone()) + .collect(); + for key in keys { + self.end_poll(&key); + } } /// Close every open window into its event, filed under the window's session, and start fresh. /// A track that produced no reading for a whole window has left (unpublished, unsubscribed, - /// its room gone): its per-layer and CPU counters are retired with it, so memory follows the + /// its room gone): its stream and CPU counters are retired with it, so memory follows the /// live tracks rather than every track the process ever saw. pub fn close(&mut self) -> Vec { - let windows = &self.windows; - self.layers.retain(|key, _| windows.contains_key(key)); - self.limitation.retain(|key, _| windows.contains_key(key)); + self.end_polls(|_| true); + let seen = std::mem::take(&mut self.seen); + self.tracks.retain(|key, _| seen.contains(key)); + self.limitation.retain(|key, _| seen.contains(key)); let end = now_unix_nanos(); self.windows.drain().map(|(_, window)| window.into_queued(end)).collect() } @@ -628,8 +815,10 @@ impl StatsWindows { let gone = |(trace_id, sid, _): &TrackKey| { *trace_id == session.trace_id && track_sid.is_none_or(|t| t == sid) }; - self.layers.retain(|key, _| !gone(key)); + self.end_polls(gone); + self.tracks.retain(|key, _| !gone(key)); self.limitation.retain(|key, _| !gone(key)); + self.seen.retain(|key| !gone(key)); let keys: Vec = self.windows.keys().filter(|k| gone(k)).cloned().collect(); let end = now_unix_nanos(); keys.into_iter() @@ -641,6 +830,7 @@ impl StatsWindows { /// Close `session`'s open windows now — their records then carry the attributes they were /// captured under — keeping each track's cumulative state for the windows that follow. pub fn split(&mut self, session: &ScopeState) -> Vec { + self.end_polls(|(trace_id, ..)| *trace_id == session.trace_id); let keys: Vec = self .windows .keys() @@ -654,45 +844,69 @@ impl StatsWindows { .collect() } - /// How many tracks the accumulator holds state for (windows, layers, CPU counters). + /// How many tracks the accumulator holds state for (windows, tracks, CPU counters), with the + /// polls recorded so far read into their windows. #[cfg(test)] - pub fn tracked(&self) -> (usize, usize, usize) { - (self.windows.len(), self.layers.len(), self.limitation.len()) + pub fn tracked(&mut self) -> (usize, usize, usize) { + self.end_polls(|_| true); + (self.windows.len(), self.tracks.len(), self.limitation.len()) } + /// Record a reading under the unit tests' one session. #[cfg(test)] pub fn record(&mut self, sample: RtcStatsSample) { self.record_in(sample, &test_session()); } + /// Close every window and keep only the events. #[cfg(test)] pub fn close_events(&mut self) -> Vec { self.close().into_iter().map(|q| q.event).collect() } - /// The encoder was CPU-starved within the last minute: stretch the cadence, like thermal - /// pressure. + /// The encoder was CPU-starved within the last [`CPU_LIMITED_HOLD`]: stretch the cadence, + /// like thermal pressure. pub fn cpu_limited(&self) -> bool { self.cpu_limited_until.is_some_and(|t| Instant::now() < t) } - /// A cpu limitation counter that grew since the previous reading means the encoder was - /// starved in between. - fn track_limitation(&mut self, key: &TrackKey, sample: &RtcStatsSample) { + /// A cpu limitation counter that grew since the previous poll means the encoder was starved + /// in between. Checked as each reading arrives, against the previous poll only, so the order + /// of a poll's streams changes nothing. + fn check_limitation(&mut self, key: &TrackKey, sample: &RtcStatsSample) { if sample.direction != StreamDirection::Outbound { return; } - let current = sample.quality_limitation_cpu_ms.unwrap_or(0); - if let Some(previous) = self.limitation.insert(key.clone(), current) { - if current > previous { - self.cpu_limited_until = Some(Instant::now() + CPU_LIMITED_HOLD); - } + let (Some(current), Some(&previous)) = + (sample.quality_limitation_cpu_ms, self.limitation.get(key)) + else { + return; + }; + if current > previous { + self.cpu_limited_until = Some(Instant::now() + CPU_LIMITED_HOLD); } } - /// Drop everything (opt-out); returns how many open windows went. + /// A finished poll's cpu limitation counter is the baseline for the next poll. A poll without + /// the counter says nothing: it must not reset the baseline. + fn set_limitation_baseline(&mut self, key: &TrackKey, reading: &RtcStatsSample) { + if reading.direction != StreamDirection::Outbound { + return; + } + if let Some(current) = reading.quality_limitation_cpu_ms { + self.limitation.insert(key.clone(), current); + } + } + + /// Drop everything (opt-out); returns how many RTC records went, each once: the open windows, + /// and the first polls that had not opened their window yet. pub fn clear(&mut self) -> u64 { - let open = self.windows.len() as u64; + let first_polls = self + .tracks + .iter() + .filter(|(key, track)| track.poll.is_some() && !self.windows.contains_key(*key)) + .count(); + let open = (self.windows.len() + first_polls) as u64; *self = Self::with_consent(self.revoked.clone()); open } @@ -726,27 +940,28 @@ impl StreamDirection { mod tests { use super::*; + /// Simulcast layers make one series per track: its bytes grow by every layer's growth, so a + /// stalled layer does not stall the track. #[test] fn simulcast_layers_fold_into_one_track_series() { let mut windows = StatsWindows::default(); - let layer = |id: &str, bytes: u64, fps: f64| RtcStatsSample { + let layer = |id: &str, bytes: u64, fps: f64, second: u64| RtcStatsSample { bytes: Some(bytes), frames_per_second: Some(fps), layer: Some(id.into()), + timestamp_ns: Some(second * 1_000_000_000), ..RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound) }; - windows.record(layer("h", 1_000, 30.0)); - windows.record(layer("f", 4_000, 30.0)); + windows.record(layer("h", 1_000, 30.0, 1)); + windows.record(layer("f", 4_000, 30.0, 1)); // The top layer stalls; the half layer keeps sending. - windows.record(layer("h", 3_000, 30.0)); - windows.record(layer("f", 4_000, 0.0)); + windows.record(layer("h", 3_000, 30.0, 2)); + windows.record(layer("f", 4_000, 0.0, 2)); let events = windows.close_events(); assert_eq!(events.len(), 1, "one window per track, not per layer"); - let bytes = - events[0].attributes.iter().find(|a| a.key == "lk.rtc.bytes").map(|a| a.value.clone()); - assert_eq!(bytes, Some(AttributeValue::Int(7_000)), "last folded total"); + assert_eq!(attr(&events[0], "lk.rtc.bytes"), Some(AttributeValue::Int(2_000)), "h grew"); let body = events[0].body.clone().unwrap_or_default(); - assert!(!body.contains(" 0 kbps"), "a stalled layer is not a stalled track: {body}"); + assert!(body.contains("16 kbps"), "a stalled layer is not a stalled track: {body}"); } #[tokio::test(start_paused = true)] @@ -773,7 +988,92 @@ mod tests { windows.record(inbound); assert!(!windows.cpu_limited(), "inbound counters are ignored"); } - use crate::AttributeValue; + + #[tokio::test(start_paused = true)] + async fn a_reading_without_the_cpu_counter_keeps_the_baseline() { + let mut windows = StatsWindows::default(); + let reading = |cpu_ms| RtcStatsSample { + quality_limitation_cpu_ms: cpu_ms, + ..RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound) + }; + for cpu_ms in [Some(5_000), None, Some(5_000)] { + windows.record(reading(cpu_ms)); + } + assert!(!windows.cpu_limited(), "the counter did not grow"); + } + + #[test] + fn a_split_window_keeps_its_tracks_state_through_the_next_close() { + let mut windows = StatsWindows::default(); + let session = test_session(); + windows.record_in( + RtcStatsSample { + bytes: Some(1_000), + quality_limitation_cpu_ms: Some(0), + layer: Some("f".into()), + ..RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound) + }, + &session, + ); + windows.split(&session); + windows.close(); + assert_eq!(windows.tracked(), (0, 1, 1), "the track is live: it read since the last close"); + windows.close(); + assert_eq!(windows.tracked(), (0, 0, 0), "a whole window without a reading retires it"); + } + + fn stat(kind: &str, id: &str, members: &[(&str, AttributeValue)]) -> RtcStat { + RtcStat { + kind: kind.into(), + id: id.into(), + members: members.iter().map(|(k, v)| (k.to_string(), v.clone())).collect(), + } + } + + #[test] + fn inbound_rtt_comes_from_the_selected_candidate_pair() { + let inbound = stat("inbound-rtp", "IN", &[("bytesReceived", AttributeValue::Int(1))]); + let pair = |id: &str, rtt: f64, nominated: bool| { + stat( + "candidate-pair", + id, + &[ + ("nominated", AttributeValue::Bool(nominated)), + ("state", AttributeValue::Str("succeeded".into())), + ("currentRoundTripTime", AttributeValue::Double(rtt)), + ], + ) + }; + let rtt = |report: Vec| { + samples_from_report("TR_1", TrackKind::Audio, StreamDirection::Inbound, &report, None) + [0] + .rtt_ms + }; + // Without a name from the transport: the first pair that is nominated and succeeded. + let fallback = vec![inbound.clone(), pair("A", 0.3, false), pair("B", 0.04, true)]; + assert_eq!(rtt(fallback), Some(40.0), "a stale succeeded pair listed first is skipped"); + // The pair the transport names wins over any other. + let transport = + stat("transport", "T", &[("selectedCandidatePairId", AttributeValue::Str("B".into()))]); + let named = vec![inbound, pair("A", 0.3, true), pair("B", 0.04, false), transport]; + assert_eq!(rtt(named), Some(40.0)); + } + + #[test] + fn non_finite_numbers_are_no_reading() { + let report = [stat( + "inbound-rtp", + "IN", + &[ + ("jitter", AttributeValue::Str("NaN".into())), + ("bytesReceived", AttributeValue::Double(f64::INFINITY)), + ], + )]; + let sample = + &samples_from_report("TR_1", TrackKind::Audio, StreamDirection::Inbound, &report, None) + [0]; + assert_eq!((sample.jitter_ms, sample.bytes), (None, None)); + } fn attr(event: &TelemetryEvent, key: &str) -> Option { event.attributes.iter().find(|a| a.key == key).map(|a| a.value.clone()) @@ -874,6 +1174,7 @@ mod tests { "CP", &[ ("nominated", AttributeValue::Bool(true)), + ("state", AttributeValue::Str("succeeded".into())), ("currentRoundTripTime", AttributeValue::Double(0.1)), ], ), @@ -1003,4 +1304,297 @@ mod tests { let remote = out.iter().find(|s| s.track_sid == "TR_remote").expect("remote"); assert_eq!((remote.kind, remote.bytes), (TrackKind::Audio, Some(7))); } + + /// The bitrate is the bytes between the window's first and last reading over the time between + /// them, not over the window: readings twice per window end before the window does. + #[test] + fn the_bitrate_spans_the_readings_not_the_window() { + let mut first = RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound); + first.bytes = Some(0); + first.timestamp_ns = Some(0); + let mut last = first.clone(); + last.bytes = Some(300_000); + last.timestamp_ns = Some(30_000_000_000); + let mut window = Window::open(0, first.clone(), test_session()); + window.add(last); + let body = window.into_event(60_000_000_000).body.unwrap_or_default(); + assert!(body.contains("80 kbps"), "{body}"); + let alone = Window::open(0, first, test_session()).into_event(60_000_000_000); + let body = alone.body.unwrap_or_default(); + assert!(!body.contains("kbps"), "one reading has no bitrate: {body}"); + } + + /// One poll's simulcast layers are one reading: the order the report lists them in does not + /// change the frame-rate gauges. + #[test] + fn the_order_of_a_reports_layers_does_not_change_the_gauges() { + let gauges = |order: [(&str, f64); 2]| { + let mut windows = StatsWindows::default(); + for (layer, fps) in order { + let mut sample = + RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound); + sample.layer = Some(layer.into()); + sample.frames_per_second = Some(fps); + sample.timestamp_ns = Some(1); + windows.record(sample); + } + let event = windows.close_events().pop().expect("a window"); + let fps: Vec<_> = + event.attributes.into_iter().filter(|a| a.key.starts_with("lk.rtc.fps")).collect(); + (fps, event.body) + }; + let (h_first, body) = gauges([("h", 10.0), ("f", 30.0)]); + assert_eq!(h_first, gauges([("f", 30.0), ("h", 10.0)]).0); + assert!(body.unwrap_or_default().contains("30 fps")); + } + + /// A poll's layers make one reading in any report order: a layer without an RTT does not + /// erase the RTT another layer has, and of two RTTs the larger counts. + #[test] + fn a_polls_reading_keeps_what_any_layer_reports_in_either_order() { + let layer = |rid: &str| { + stat("outbound-rtp", &format!("OUT_{rid}"), &[("rid", AttributeValue::Str(rid.into()))]) + }; + let rtt = |rid: &str, secs: f64| { + stat( + "remote-inbound-rtp", + &format!("RI_{rid}"), + &[ + ("localId", AttributeValue::Str(format!("OUT_{rid}"))), + ("roundTripTime", AttributeValue::Double(secs)), + ], + ) + }; + let rtt_avg = |report: Vec| { + let mut windows = StatsWindows::default(); + let outbound = StreamDirection::Outbound; + for sample in samples_from_report("TR_1", TrackKind::Video, outbound, &report, Some(1)) + { + windows.record(sample); + } + attr(&windows.close_events().pop().expect("a window"), "lk.rtc.rtt_ms.avg") + }; + let (f, h, q) = (layer("f"), layer("h"), layer("q")); + for report in [vec![f.clone(), h.clone()], vec![h, f.clone()]] { + let report = [report, vec![rtt("f", 0.042)]].concat(); + assert_eq!(rtt_avg(report), Some(AttributeValue::Double(42.0)), "only `f` has one"); + } + for report in [vec![f.clone(), q.clone()], vec![q, f]] { + let report = [report, vec![rtt("f", 0.042), rtt("q", 0.05)]].concat(); + assert_eq!(rtt_avg(report), Some(AttributeValue::Double(50.0)), "the larger one"); + } + } + + /// One outbound reading: stream `id` with `bytes` bytes and as many packets, at `second`. + fn stream(id: &str, bytes: Option, second: u64) -> RtcStatsSample { + RtcStatsSample { + layer: Some(id.into()), + bytes, + packets: bytes, + timestamp_ns: Some(second * 1_000_000_000), + ..RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound) + } + } + + /// The track's bytes and packets in the window `windows` closes now. + fn totals(windows: &mut StatsWindows) -> (Option, Option) { + let event = windows.close_events().pop().expect("a window"); + (attr(&event, "lk.rtc.bytes"), attr(&event, "lk.rtc.packets")) + } + + fn both(n: i64) -> (Option, Option) { + (Some(AttributeValue::Int(n)), Some(AttributeValue::Int(n))) + } + + /// A stream that leaves and reports again only sets its starting point on its return, so + /// nothing it sent before is counted twice; what it sends after counts. + #[test] + fn a_stream_that_reports_again_adds_only_its_growth() { + let mut windows = StatsWindows::default(); + for id in 0..9 { + windows.record(stream(&format!("layer-{id}"), Some(1), id + 1)); + } + assert_eq!(totals(&mut windows), both(0), "first polls only set starting points"); + windows.record(stream("layer-0", Some(1), 10)); + assert_eq!(totals(&mut windows), both(0), "no stream sent another byte"); + windows.record(stream("layer-0", Some(3), 11)); + assert_eq!(totals(&mut windows), both(2), "what it sent since counts"); + } + + /// A counter a stream leaves out never becomes a starting point of zero: the next value + /// known only sets the starting point, and growth after it counts. + #[test] + fn an_absent_counter_never_becomes_a_zero_starting_point() { + let mut windows = StatsWindows::default(); + for id in 0..9 { + windows.record(stream(&format!("layer-{id}"), Some(1), id + 1)); + } + windows.close_events(); + windows.record(stream("layer-0", None, 10)); + windows.close_events(); + windows.record(stream("layer-0", Some(1), 11)); + assert_eq!(totals(&mut windows), both(0), "no stream sent anything more"); + windows.record(stream("layer-0", None, 12)); + windows.record(stream("layer-0", Some(4), 13)); + assert_eq!(totals(&mut windows), both(3), "growth across a poll without the counter"); + } + + /// The same poll in any order gives the same totals, here ten streams at once. + #[test] + fn a_polls_totals_do_not_depend_on_its_order() { + let totals_for = |reverse: bool| { + let mut windows = StatsWindows::default(); + let mut ids: Vec = (1..=10).collect(); + if reverse { + ids.reverse(); + } + for second in [1, 2] { + for &id in &ids { + windows.record(stream(&format!("layer-{id}"), Some(id * second), second)); + } + } + totals(&mut windows) + }; + assert_eq!(totals_for(false), totals_for(true), "same polls in reverse order"); + assert_eq!(totals_for(false), both(55), "each stream grew by its own count"); + } + + /// Every order of a poll's streams gives the same window: totals, gauges and body. + #[test] + fn every_order_of_a_poll_gives_the_same_window() { + let poll = |second: u64| { + [("f", 3_000, 30.0, 0.05), ("h", 2_000, 15.0, 0.04), ("q", 1_000, 7.0, 0.06)].map( + |(id, bytes, fps, rtt)| RtcStatsSample { + frames_per_second: Some(fps / second as f64), + rtt_ms: Some(rtt * 1_000.0), + ..stream(id, Some(bytes * second), second) + }, + ) + }; + let orders = [[0, 1, 2], [0, 2, 1], [1, 0, 2], [1, 2, 0], [2, 0, 1], [2, 1, 0]]; + let windows: Vec<_> = orders + .iter() + .map(|order| { + let mut windows = StatsWindows::default(); + for second in [1, 2] { + let readings = poll(second); + for &i in order { + windows.record(readings[i].clone()); + } + } + let event = windows.close_events().pop().expect("a window"); + let mut attributes: Vec<_> = event + .attributes + .into_iter() + // All but the window's length, which depends on when it closes. + .filter(|a| a.key.starts_with("lk.rtc.") && a.key != "lk.rtc.window_ms") + .map(|a| format!("{}={:?}", a.key, a.value)) + .collect(); + attributes.sort(); + (attributes, event.body) + }) + .collect(); + assert!(windows.iter().all(|w| *w == windows[0]), "{windows:#?}"); + } + + /// A falling frame rate comes from the current poll only, in either order of its layers. + #[test] + fn a_falling_frame_rate_comes_from_the_current_poll_only() { + let reading = |layer: &str, fps, timestamp_ns| RtcStatsSample { + layer: Some(layer.into()), + frames_per_second: Some(fps), + timestamp_ns: Some(timestamp_ns), + ..RtcStatsSample::new("TR_1", TrackKind::Video, StreamDirection::Outbound) + }; + for order in [[("h", 5.0), ("f", 10.0)], [("f", 10.0), ("h", 5.0)]] { + let mut windows = StatsWindows::default(); + windows.record(reading("f", 30.0, 1)); + windows.record(reading("h", 10.0, 1)); + windows.close_events(); + for (layer, fps) in order { + windows.record(reading(layer, fps, 2)); + } + let event = windows.close_events().pop().expect("a window"); + assert_eq!(attr(&event, "lk.rtc.fps.avg"), Some(AttributeValue::Double(10.0))); + } + } + + /// A track keeps one report's streams at most, however many stream ids it went through. + #[test] + fn a_track_keeps_one_reports_streams() { + let mut windows = StatsWindows::default(); + for id in 0..1000 { + windows.record(stream(&format!("outbound-rtp-{id}"), Some(1), id + 1)); + windows.close_events(); + } + let kept = windows.tracks.values().next().expect("the track").previous.len(); + assert_eq!(kept, 1, "the previous poll's one stream"); + } + + /// Clearing forgets every stream, and the track starts over afterwards. + #[test] + fn clear_forgets_every_stream() { + let mut windows = StatsWindows::default(); + for _ in 0..2 { + for id in 0..9 { + windows.record(stream(&format!("layer-{id}"), Some(1), id + 1)); + } + windows.clear(); + assert_eq!(windows.tracked(), (0, 0, 0)); + } + } + + /// A retired track's last window holds its session until it is sent, and clearing lets every + /// session go. + #[test] + fn retire_and_clear_let_the_session_go() { + let mut windows = StatsWindows::default(); + let inbound = + |sid: &str| RtcStatsSample::new(sid, TrackKind::Audio, StreamDirection::Inbound); + let scope = ScopeState::new(); + let session = Arc::downgrade(&scope); + windows.record_in(inbound("TR_1"), &scope); + let queued = windows.retire(&scope, None); + drop(scope); + assert!(session.upgrade().is_some(), "the last window holds it"); + drop(queued); + assert!(session.upgrade().is_none(), "gone with the window"); + let scope = ScopeState::new(); + let session = Arc::downgrade(&scope); + windows.record_in(inbound("TR_2"), &scope); + drop(scope); + windows.clear(); + assert!(session.upgrade().is_none(), "nothing kept after clear"); + } + + /// The running total is a layered outbound track's: an inbound track passes its counters + /// through as reported, whatever stream ids its readings carry, while an outbound track with + /// stream ids counts each stream's growth. + #[test] + fn only_a_layered_outbound_track_keeps_a_running_total() { + let totals_for = |direction| { + let mut windows = StatsWindows::default(); + [123, 200].map(|bytes| { + windows + .record(RtcStatsSample { direction, ..stream("ssrc-1", Some(bytes), bytes) }); + totals(&mut windows) + }) + }; + assert_eq!(totals_for(StreamDirection::Inbound), [both(123), both(200)], "as reported"); + assert_eq!(totals_for(StreamDirection::Outbound), [both(0), both(77)], "growth"); + } + + /// Clearing counts every RTC record it drops, each once: an open window, and a first poll + /// that has not opened its window yet. + #[test] + fn clear_counts_every_window_and_first_poll_once() { + let mut windows = StatsWindows::default(); + windows.record(stream("f", Some(100), 1)); + assert_eq!(windows.clear(), 1, "a pending first poll is one record"); + // A window with a pending poll of its own, and another track with only a pending poll. + windows.record(stream("f", Some(100), 1)); + windows.record(stream("f", Some(200), 2)); + windows.record(RtcStatsSample { track_sid: "TR_2".into(), ..stream("f", Some(100), 1) }); + assert_eq!(windows.clear(), 2, "one per track, its pending poll included"); + } }