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 ffe4c89e7..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,78 @@ 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 +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`): diff --git a/livekit-telemetry/src/device.rs b/livekit-telemetry/src/device.rs new file mode 100644 index 000000000..ce7714181 --- /dev/null +++ b/livekit-telemetry/src/device.rs @@ -0,0 +1,655 @@ +// 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, + /// 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, +} + +/// 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 { + /// The app is in front of the user; sent as `foreground`. + #[default] + Foreground, + /// The app runs in the background; sent as `background`. + 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 { + /// 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, +} + +/// 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 { + /// 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, + /// 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, + /// No network path: the device is offline; sent as `unavailable`. + Unavailable, +} + +/// 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 +/// 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, 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))] + 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, + /// 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 + /// [`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, + 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() != BatteryBucket::Fine, + self.network_constrained, + ] + .iter() + .filter(|&&on| on) + .count() as u32; + ((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(crate) fn holds_uploads(&self) -> bool { + self.network_constrained || self.battery_bucket() == BatteryBucket::NearlyEmpty + } + + /// 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 => { + 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` (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. + 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 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: {level}%{}", + if self.battery_charging { ", charging" } else { "" } + )) + .with_attribute("lk.device.battery.level", i64::from(level)) + .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", + } + } +} + +/// 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::*; + + #[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"); + } + + #[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))); + } + + #[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.*`"); + } + + /// 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: TelemetryEvent = DeviceEvent::AudioRouteChanged { + outputs: vec![AudioOutput::Speaker], + reason: AudioRouteReason::OldDeviceUnavailable, + } + .into(); + assert_eq!(event.body.as_deref(), Some("audio route: speaker (old_device_unavailable)")); + let denied: TelemetryEvent = DeviceEvent::CaptureFailed { + device: CaptureDevice::ScreenShare, + reason: CaptureFailure::PermissionDenied, + } + .into(); + 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/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 7b2b3a526..75788c4f9 100644 --- a/livekit-telemetry/src/lib.rs +++ b/livekit-telemetry/src/lib.rs @@ -31,6 +31,26 @@ 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. +#[cfg_attr( + not(test), + 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 +96,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..69a988dc7 --- /dev/null +++ b/livekit-telemetry/src/rtc.rs @@ -0,0 +1,1600 @@ +// 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, HashSet}, + sync::Arc, + time::Duration, +}; + +use tokio::time::Instant; + +use crate::{ + 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, +} + +/// Received (inbound) or sent (outbound) media. +#[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, +} + +/// 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. 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. + #[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 { + /// 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), + AttributeValue::Double(d) => Some(*d), + 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) +} + +/// 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) +} + +/// 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()), + _ => 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 selected 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 { + // 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() + .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 timestamp_ns = Some(timestamp_ns.unwrap_or_else(now_unix_nanos)); + let index = ReportIndex::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 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 ReportIndex<'a> { + by_id: HashMap<&'a str, &'a RtcStat>, + remote_by_local: HashMap, + pair_rtt_ms: Option, + rtp: Vec<(&'a RtcStat, StreamDirection)>, +} + +impl<'a> ReportIndex<'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(), + }; + // 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() { + "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); + } + } + "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 + } + + /// 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| ms_f64(remote, "roundTripTime")); + } + 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 = ms_f64(stat, "jitter"); + sample.frames_per_second = num(stat, "framesPerSecond"); + sample.audio_level = num(stat, "audioLevel"); + sample.rtt_ms = self.pair_rtt_ms; + } + } + sample + } +} + +/// 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 { + 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; + } + + /// 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 { + 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"), 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, + 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, + fps: Gauge, + audio_level: Gauge, +} + +impl Window { + fn open(start_ns: u64, first: RtcStatsSample, session: Arc) -> Self { + let mut window = Self { + session, + start_ns, + samples: 0, + first: first.clone(), + last: first, + jitter: Gauge::default(), + rtt: Gauge::default(), + fps: Gauge::default(), + audio_level: Gauge::default(), + }; + window.commit(); + window + } + + /// `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?)); + 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), + 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 let Some(rtt) = self.rtt.avg() { + parts.push(format!("rtt {rtt:.0} ms")); + } + 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) + { + 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 record captured now (see `QueuedEvent::new`). + fn into_queued(self, end_ns: u64) -> QueuedEvent { + let session = Arc::clone(&self.session); + QueuedEvent::new(self.into_event(end_ns), session) + } + + /// 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(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 + /// 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(); + 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, + /// 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, +} + +/// 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) +} + +/// 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 StreamCounts { + bytes: Option, + packets: Option, +} + +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) + } +} + +/// 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. +/// `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() } + } + + /// 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.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.clone(), Window::open(timestamp_ns, reading, session)); + } + } + } + + /// 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 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 { + 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() + } + + /// 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.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() + .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 { + self.end_polls(|(trace_id, ..)| *trace_id == session.trace_id); + 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, tracks, CPU counters), with the + /// polls recorded so far read into their windows. + #[cfg(test)] + 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 [`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 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 (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); + } + } + + /// 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 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 + } +} + +/// 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::*; + + /// 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, 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, 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, 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"); + 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("16 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"); + } + + #[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()) + } + + #[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)), + ("state", AttributeValue::Str("succeeded".into())), + ("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))); + } + + /// 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"); + } +}