diff --git a/.changeset/livekit-net-redirect-scheme.md b/.changeset/livekit-net-redirect-scheme.md new file mode 100644 index 000000000..6ccefaa99 --- /dev/null +++ b/.changeset/livekit-net-redirect-scheme.md @@ -0,0 +1,5 @@ +--- +livekit-net: patch +--- + +The native HTTP client no longer follows a redirect that changes the scheme, so an `Authorization` header never reaches a plaintext URL on the same host and port. diff --git a/.changeset/telemetry-transport.md b/.changeset/telemetry-transport.md new file mode 100644 index 000000000..b2b6ee119 --- /dev/null +++ b/.changeset/telemetry-transport.md @@ -0,0 +1,5 @@ +--- +livekit-telemetry: minor +--- + +Add the `TelemetryTransport` seam, a `net` HTTP transport over `livekit-net`, and LiveKit Cloud destination routing from a Room's server URL and token. diff --git a/Cargo.lock b/Cargo.lock index e43a2f1cd..f54f435ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4016,11 +4016,19 @@ dependencies = [ name = "livekit-telemetry" version = "0.1.0" dependencies = [ + "async-trait", + "base64 0.22.1", + "livekit-net", "log", "opentelemetry-proto", "prost 0.14.4", + "prost-types 0.14.4", "rand 0.9.5", + "serde_json", + "thiserror 2.0.19", + "tokio", "uniffi", + "url", ] [[package]] diff --git a/livekit-net/src/native/mod.rs b/livekit-net/src/native/mod.rs index 6ca7ef78d..e0cb48bd6 100644 --- a/livekit-net/src/native/mod.rs +++ b/livekit-net/src/native/mod.rs @@ -42,10 +42,29 @@ pub struct NativeTransport { impl NativeTransport { pub(crate) fn new() -> Self { - Self { http: reqwest::Client::new() } + let http = reqwest::Client::builder() + .redirect(redirect_policy()) + .build() + // `reqwest::Client::new` panics the same way. + .expect("the TLS backend initializes"); + Self { http } } } +/// reqwest's default policy (at most 10 redirects), except that a redirect to another scheme is +/// returned instead of followed: reqwest drops `Authorization` only when the host or port +/// changes, so an `https` to `http` redirect on port 443 would send a credential in plaintext. +fn redirect_policy() -> reqwest::redirect::Policy { + reqwest::redirect::Policy::custom(|attempt| { + let from = attempt.previous().last().map(url::Url::scheme); + if from.is_some_and(|scheme| scheme != attempt.url().scheme()) { + attempt.stop() + } else { + reqwest::redirect::Policy::default().redirect(attempt) + } + }) +} + /// The shared default transport behind the registry fallback. Memoized because /// consumers resolve a client per request: a fresh transport per resolution /// would defeat its connection pooling. @@ -139,3 +158,43 @@ impl HttpClient for NativeTransport { } } } + +#[cfg(test)] +mod tests { + use std::{ + io::{Read, Write}, + net::TcpListener, + time::Duration, + }; + + use super::*; + + /// A redirect that changes only the scheme, on the same port, comes back as the 3xx: the + /// request with its `Authorization` header is not sent again under the new scheme. + #[tokio::test] + async fn a_redirect_to_another_scheme_is_not_followed() { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let port = listener.local_addr().expect("address").port(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("the first request"); + let _ = stream.read(&mut [0; 4096]); + let answer = format!( + "HTTP/1.1 307 Temporary Redirect\r\nLocation: https://127.0.0.1:{port}/v1/logs\r\n\ + Content-Length: 0\r\nConnection: close\r\n\r\n" + ); + stream.write_all(answer.as_bytes()).expect("answer"); + drop(stream); + listener.set_nonblocking(true).expect("non-blocking"); + std::thread::sleep(Duration::from_millis(300)); + listener.accept().is_ok() + }); + let headers = vec![Header { name: "Authorization".into(), value: "Bearer token".into() }]; + let url = format!("http://127.0.0.1:{port}/v1/logs"); + let response = NativeTransport::new() + .request(HttpMethod::Post, url, headers, Some(b"batch".to_vec())) + .await + .expect("the redirect itself comes back"); + assert_eq!(response.status, 307); + assert!(!server.join().expect("server"), "nothing was sent to the new scheme"); + } +} diff --git a/livekit-telemetry/Cargo.toml b/livekit-telemetry/Cargo.toml index 7364e25b3..29eb5d1d5 100644 --- a/livekit-telemetry/Cargo.toml +++ b/livekit-telemetry/Cargo.toml @@ -8,17 +8,34 @@ edition.workspace = true repository.workspace = true [dependencies] +tokio = { workspace = true, default-features = false, features = ["macros", "sync", "time"] } log = { workspace = true } +thiserror = { workspace = true } +async-trait = "0.1" prost = { workspace = true } +# `google.protobuf.Any`/`Duration` for the `google.rpc.Status` error body (RetryInfo). +prost-types = { workspace = true } rand = { workspace = true } +# Unverified JWT claims (observability grant, expiry); both already linked by livekit-api. +base64 = "0.22" +serde_json = { workspace = true } +# Server URLs are parsed, never string-matched, before a token may follow them. +url = "2.3" # OTLP message types only (`gen-tonic-messages` = prost structs, no tonic). Same prost as # livekit-protocol so a single prost is linked. opentelemetry-proto = { version = "0.32", default-features = false, features = ["logs", "trace", "gen-tonic-messages"] } +livekit-net = { workspace = true, optional = true } uniffi = { workspace = true, features = ["scaffolding-ffi-buffer-fns"], optional = true } [features] +# HTTP transport over a `livekit-net` client. The host registers one with +# `livekit_net::set_http_client`, or the build enables `livekit-net/native` and a TLS feature. +net = ["dep:livekit-net"] uniffi = ["dep:uniffi"] +[dev-dependencies] +tokio = { workspace = true, default-features = false, features = ["rt", "rt-multi-thread", "macros", "sync", "time", "test-util", "net", "io-util"] } + # How CI checks this crate's features, read by # `.github/workflows/feature-combinations-curated.yml` via `cargo metadata`. [package.metadata.feature-combinations] diff --git a/livekit-telemetry/README.md b/livekit-telemetry/README.md index 83ca1fa3a..1128d148e 100644 --- a/livekit-telemetry/README.md +++ b/livekit-telemetry/README.md @@ -12,6 +12,9 @@ It currently holds the shared data model and its encoding: (trace) it belongs to; every drop is counted. - **Encoding.** [OTLP/HTTP](https://opentelemetry.io/docs/specs/otlp/) protobuf logs and traces, using the `opentelemetry-proto` message types only (no gRPC stack). +- **Transport and destinations.** `TelemetryTransport`, the seam that moves an encoded batch off + the device (`NetTransport` over `livekit-net` with the `net` feature), and the routing that + turns a Room's server URL and token into its LiveKit Cloud ingest URL and credential. The pipeline that batches, caches and uploads records is not part of the crate yet. diff --git a/livekit-telemetry/SPEC.md b/livekit-telemetry/SPEC.md index 1c58613d3..ffe4c89e7 100644 --- a/livekit-telemetry/SPEC.md +++ b/livekit-telemetry/SPEC.md @@ -8,9 +8,11 @@ Every record carries a wall-clock timestamp: one without `timestamp_ns` is stamp captured (queued), never at export; an explicit `timestamp_ns` is kept. Its attributes, like its owner and timestamp, are taken at capture, its session's (`lk.room.*`, `lk.participant.*`, `session.id`) included, so a record captured before a disconnect keeps its room; only -pipeline-wide attributes are added at export. 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)). +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)). ## Resource attributes @@ -24,6 +26,61 @@ Set once per pipeline (`TelemetryConfig.resource`): | `device.model.identifier` | platform SDK | `iPhone16,1` | | `telemetry.sdk.name/language/version` | core | `livekit-telemetry`, `rust`, `0.1.0` | +## Pipeline, scopes and destination + +### Destination and credentials + +The platform passes exactly two things per room, through `Scope::set_server(url, token)`: the +LiveKit server URL the room connects to and the participant token it connects with — at connect, +and again with **every** refreshed token (the SFU sends one right after join, then every few +minutes). The call is cheap and idempotent. There is no client-side endpoint, header or sink. + +- **Ingest URL:** derived from the server URL's host, `https:///observability/client/{logs,traces}/otlp/v0`, + for LiveKit Cloud hosts only (`*.livekit.cloud`, including `*.staging.livekit.cloud`). Any other + host (self-hosted, OSS) has no ingest: one local warning, and that room's records are dropped + at the door instead of cached. +- **Token:** `Authorization: Bearer `. The core reads the token's *unverified* claims: the + observability grant (`observability.write` or `observability.clientWrite`) and `exp`. It never + sends a token known to be expired or refused; batches wait for the next token instead + (a hard hold). A project whose first token has no grant never opted in: nothing is collected + for it. A refreshed token without the grant (today's SFU drops it) does not replace a granted + one that is still valid. Tokens live in memory only — never in a batch, never on disk. +- **Server URL validation:** parsed with WHATWG URL rules, never string-matched. A token is only + sent to `https://.livekit.cloud/…` built from the parsed host alone: TLS scheme + (`wss`/`https`), a domain under the Cloud suffix with a label of its own, the default port, no + userinfo. +- **Ownership:** every record captures its owner when it is captured — its session and the + project that session is routed to at that moment — and keeps it: a Room that reconnects to + another project takes nothing queued or cached along. Credentials are keyed by (project, + session): a live Room uploads with its **own** latest token for that project, never another + Room's. A Room's records captured before it had a server go to its own first project, never to + another Room's. Process-level records (device state, pre-room errors, self-telemetry) go to + the project most recently handed a token, with that project's latest token; so do batches from + a previous launch, which wait — up to the 24 h age limit — for a token of the same project. + The answer to a request is attributed to the project it was sent to — its 404, disable or + pause never lands on another project. A session's credentials stay while the session is alive + (its Room, or records, windows or spans still referencing it) — for every project it was routed + to, so records captured for an earlier project can still be sent; that is the residual: a live + Room keeps one credential per project it has used — and while cached batches need them; + project-level copies only while a live session is routed there or the backlog has batches for + it. A token the collector refused is recorded by identity (a hash) and never sent again from + any slot, until it expires (a token without `exp` stays refused for the process). A past, + negative or non-numeric `exp` counts as expired. A project's entry goes once no live session + and no cached batch needs it. An entry known to receive nothing (not Cloud, no grant, 404, + disabled) stays for the process. +- **Routing key:** a Cloud project is keyed by its host, which must be a DNS hostname (letters, + digits, `-` and `.`). Any other URL is keyed by its scheme, host and port, so a plaintext or + userinfo URL on a Cloud host can neither feed that project a token nor silence it. Its host and + port are escaped (`%XX` for anything but a lowercase letter, a digit, `.`, `-` or `_`): the key + ends each cached batch's file name, which must be valid on every platform. +- **Waiting** for a first destination or a usable token is uncapped, bounded only by the cache. + +Local end-to-end tests point everything at an OpenTelemetry collector of their own with the +`LK_TELEMETRY_ENDPOINT` environment variable, read by the core at start (a base URL gets +`/v1/logs` and `/v1/traces`; a URL with a `logs` path segment is used as is, and its traces URL +replaces that segment). It is not part of any platform API. With it, every batch goes there +without Cloud rules or tokens. + ## Events An event with no `body` is exported with its name as the body as well as in `event_name`: log @@ -46,14 +103,14 @@ severity: info attributes: # counts since the previous report lk.telemetry.uploads.sent: int # batches accepted lk.telemetry.uploads.bytes: int # compressed bytes accepted — what telemetry cost the uplink - lk.telemetry.uploads.failed: int # attempts that failed transiently (no answer, 429, 5xx) + lk.telemetry.uploads.failed: int # attempts that failed transiently (a network error, 429 or 5xx, before the timeout) lk.telemetry.cache.batches: int # batches waiting in the cache right now (a gauge) # the rest only when non-zero: lk.telemetry.uploads.timeouts: int # attempts that hit export_timeout_ms lk.telemetry.uploads.unauthorized: int # 401/403 answers: tokens the collector refused lk.telemetry.holds.capped: int # soft holds that reached the 60 s cap lk.telemetry.cache.write_errors: int # batches the disk refused (full, gone), kept in memory - lk.telemetry.dropped.queue_full: int # records evicted from the in-memory queue + lk.telemetry.dropped.queue_full: int # records evicted from the in-memory queue, and spans dropped from a full span buffer lk.telemetry.dropped.cache_error: int # records no cache, not even memory, could take lk.telemetry.dropped.cache_full: int # records evicted by the cache's size / file-count bound lk.telemetry.dropped.expired: int # records in batches past the 24 h age limit @@ -170,8 +227,8 @@ A `TelemetryEvent` with an empty `name` is a plain log record (OTLP log without (semconv), `lk.log.source` (`sdk` | `ffi` | `webrtc`) and `lk.log.logger` (type, module or file). The platform hands the core a typed `LogRecord` via `log(record)`; the core applies the floor: WebRTC only at `error`, the SDK and the core at the configured `log_severity`, the core's own telemetry module -never. Only `warn` and `error` records -leave the device; `trace`/`debug`/`info` are dropped in `emit`. +never. Only plain records at or above `log_severity` (default `warn`) leave the device. Lower ones +are dropped in `emit`. The core's own warnings and errors (Rust `log` records from `livekit*` targets, never `livekit_telemetry*`) reach the pipeline by themselves through `livekit-uniffi`'s log forwarder: @@ -188,8 +245,9 @@ lines. ## Spans A span is **one attempt** at an operation. The scope (one Room connection lifetime, across -reconnects) is the trace; its id is generated by the core when the pipeline starts and rides on -every span and log record. Spans are exported when they end — never a long-lived scope span. +reconnects) is the trace. The core generates its id in `begin_scope`, and it rides on every span +and log record. Records outside any Room carry the id of the process session, which the pipeline +makes at start. Spans are exported when they end — never a long-lived scope span. | Rule | Value | |---|---| diff --git a/livekit-telemetry/src/destination.rs b/livekit-telemetry/src/destination.rs new file mode 100644 index 000000000..33316be43 --- /dev/null +++ b/livekit-telemetry/src/destination.rs @@ -0,0 +1,948 @@ +// 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. + +//! Where batches go and with which credential — decided here, once, for every platform. +//! +//! A platform hands over exactly two things: the LiveKit server URL a room connects to and the +//! participant token it connects with (again on every token refresh). The core derives the +//! ingest URL, checks that the host is LiveKit Cloud, reads the token's unverified claims (the +//! observability grant, the expiry), and routes every batch to the project its session belongs +//! to, so two rooms talking to two projects never share a token or a destination. + +use std::{ + collections::HashMap, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use base64::{ + alphabet, + engine::{DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig}, + Engine, +}; +use tokio::time::Instant; + +/// The environment variable that points every upload at a collector of your own — a local +/// OpenTelemetry collector for end-to-end tests. Not part of any platform API: set it in the +/// test process's environment before the pipeline starts. Either a URL with a `logs` path +/// segment (its traces URL replaces that segment with `traces`) or a base URL +/// (`http://localhost:4318`, OTLP paths `/v1/logs` and `/v1/traces` are appended). +pub const ENDPOINT_OVERRIDE_ENV: &str = "LK_TELEMETRY_ENDPOINT"; + +/// The longest a token is believed to stay valid, whatever its `exp` says. +const MAX_TOKEN_LIFETIME: Duration = Duration::from_secs(366 * 24 * 60 * 60); + +/// LiveKit Cloud project hosts: `.livekit.cloud` in production, +/// `.staging.livekit.cloud` in staging. Self-hosted servers have no ingest. +const CLOUD_SUFFIX: &str = ".livekit.cloud"; + +/// Which OTLP signal a batch carries; picks the ingest path. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Signal { + Logs, + Traces, +} + +/// A request target: URL per signal and the bearer token, when one is due. +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct Target { + /// The project the batch is sent to (`None` for the test override): what its answer is + /// attributed to — a 404, a disable, a pause. + pub project: Option, + /// The OTLP logs endpoint. + pub logs: String, + /// The OTLP traces endpoint. + pub traces: String, + /// The bearer token to send, `None` for the test override. + pub token: Option, +} + +impl Target { + /// The endpoint for `signal`. + pub fn url(&self, signal: Signal) -> &str { + match signal { + Signal::Logs => &self.logs, + Signal::Traces => &self.traces, + } + } +} + +/// What the unverified claims of a participant token say — enough to not send with a token the +/// collector is known to refuse. The signature is the collector's business. +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct Token { + raw: String, + /// `observability.write` or `observability.clientWrite`. + grant: bool, + /// From `exp`, on the monotonic clock (so tests can move it). + expires: Option, + /// The credential's identity (a hash of the raw token): what a refusal is recorded against. + id: u64, +} + +impl From<&str> for Token { + /// Read a raw participant token's claims; a malformed one carries no grant. + fn from(raw: &str) -> Self { + let claims = raw + .split('.') + .nth(1) + .and_then(|payload| { + const URL_SAFE: GeneralPurpose = GeneralPurpose::new( + &alphabet::URL_SAFE, + GeneralPurposeConfig::new() + .with_decode_padding_mode(DecodePaddingMode::Indifferent), + ); + URL_SAFE.decode(payload).ok() + }) + .and_then(|json| serde_json::from_slice::(&json).ok()); + let grant = claims.as_ref().is_some_and(|c| { + let observability = &c["observability"]; + observability["write"].as_bool() == Some(true) + || observability["clientWrite"].as_bool() == Some(true) + }); + // Untrusted input, read conservatively: a past or negative `exp` is expired, and so is + // one that is not a number at all; a fractional one counts in whole seconds; one beyond + // any real token lifetime is clamped (an `Instant` that far out would overflow). Only a + // token without `exp` has no known expiry. + let expires = claims.as_ref().and_then(|c| c.get("exp")).map(|exp| { + let now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs_f64(); + let left = match exp.as_f64() { + Some(exp) if exp.is_finite() && exp > now => { + Duration::from_secs_f64((exp - now).min(MAX_TOKEN_LIFETIME.as_secs_f64())) + } + _ => Duration::ZERO, + }; + Instant::now().checked_add(left).unwrap_or_else(Instant::now) + }); + let id = { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + raw.hash(&mut hasher); + hasher.finish() + }; + Self { raw: raw.to_owned(), grant, expires, id } + } +} + +impl Token { + fn expired(&self) -> bool { + self.expires.is_some_and(|at| Instant::now() >= at) + } + + /// May be sent: carries the grant, not expired, not refused. + fn usable(&self, refused: &Refusals) -> bool { + self.grant && !self.expired() && !refused.contains(self) + } +} + +/// Why a project receives nothing (anymore). Each is logged once. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Dead { + /// Not a LiveKit Cloud host (self-hosted, a typo): there is no ingest to talk to. + NotCloud, + /// The project's own token carries no observability grant: the customer did not opt in. + NoGrant, + /// The collector answered 404: no client ingest on this host. Revived by the next token + /// (a new connect or refresh), so a misrouted answer cannot silence a project for good. + NotFound, + /// The collector said the project's data recording is disabled by its owner. + Disabled, +} + +/// One LiveKit Cloud project, keyed by its host. +#[derive(Debug, Default)] +struct Project { + /// The latest token any room handed over for the project: what process-level batches and + /// batches from a previous launch are sent with. + latest: Option, + /// A token with the grant was seen: the customer opted in. Refreshed tokens may lack the + /// grant (today's server drops it on refresh) without withdrawing that. + consented: bool, + dead: Option, +} + +/// Offer `new` for a token slot: a refused token is never taken, into any slot; a token with the +/// grant wins; one without it does not replace a usable granted token (today's refreshes drop +/// the grant). +fn offer(slot: &mut Option, new: &Token, refused: &Refusals) { + if refused.contains(new) { + return; + } + let keep = slot.as_ref().is_some_and(|t| t.raw == new.raw || (!new.grant && t.usable(refused))); + if !keep { + *slot = Some(new.clone()); + } +} + +/// Credentials the collector refused, by identity, shared by every slot: a refused token is never +/// sent again, whichever Room or project hands it over. An entry lives until its token expires; +/// tokens without `exp` stay refused for the life of the process. +// ponytail: grows with refusals of tokens without `exp` (LiveKit tokens always carry one); a cap +// if a server ever mints tokens without expiry. +#[derive(Debug, Default)] +struct Refusals(HashMap>); + +impl Refusals { + fn contains(&self, token: &Token) -> bool { + self.0.contains_key(&token.id) + } + + /// Returns whether the refusal is new. + fn add(&mut self, token: &Token) -> bool { + let now = Instant::now(); + self.0.retain(|_, expires| expires.is_none_or(|at| at > now)); + self.0.insert(token.id, token.expires).is_none() + } +} + +/// The owner of process-level records (device state, pre-room errors, self-telemetry): no Room, +/// so they go to the latest project with its latest token. +pub(crate) const PROCESS_OWNER: &str = "process"; + +/// Where a batch is headed. +#[derive(Debug, Clone, PartialEq)] +pub(crate) enum Route { + /// Ready: send now. + Send(Target), + /// Keep it cached: no destination yet, or no usable token (missing, expired, grant-less, + /// refused) until the platform hands over a fresh one. A hard hold: no escape hatch sends + /// it anyway. + Wait, + /// Drop it: its project receives nothing. + Drop, +} + +/// Every project and session this process talks to, plus the test override. Tokens live here, +/// in memory only: never persisted, never in a batch. +#[derive(Debug, Default)] +pub(crate) struct Destinations { + projects: HashMap, + /// Each Room session's own latest token per project, by (host, session id): its batches for + /// that project are sent with it and never with another Room's, nor with the token it later + /// got for another project. + sessions: HashMap<(String, String), Token>, + /// Each Room session's first project: where the records it captured before it had a server + /// go. A Room's records never fall back to another Room's project. + rooms: HashMap, + refused: Refusals, + /// The host process-level batches (device state, pre-room logs, self-telemetry) go to: the + /// project most recently handed a token, while it is alive. + latest: Option, + /// [`ENDPOINT_OVERRIDE_ENV`]: every batch goes there, no Cloud rules, no token. + endpoint_override: Option, +} + +impl Destinations { + /// No projects yet. With `endpoint_override` (see [`ENDPOINT_OVERRIDE_ENV`]) every batch goes + /// there instead. + pub fn new(endpoint_override: Option<&str>) -> Self { + Self { + endpoint_override: endpoint_override.filter(|e| !e.is_empty()).map(override_target), + ..Self::default() + } + } + + /// The first project a Room session was routed to: where what it captured before it had a + /// server belongs. + pub fn first_project(&self, owner: &str) -> Option { + self.rooms.get(owner).cloned() + } + + /// Whether every batch goes to the test override. + pub fn has_override(&self) -> bool { + self.endpoint_override.is_some() + } + + /// Session `owner`'s server URL and token (at connect, and again on every refresh). Returns + /// the project key its batches carry, or `None` when the URL has no host. + pub fn set(&mut self, url: &str, token: &str, owner: &str) -> Option { + let Server { key: host, cloud } = Server::parse(url)?; + if self.endpoint_override.is_some() { + return Some(host); + } + // The first project is the Room's even when it receives nothing: what the Room captured + // before it had a server belongs there, and is dropped with it. + self.rooms.entry(owner.to_owned()).or_insert_with(|| host.clone()); + let project = self.projects.entry(host.clone()).or_default(); + if project.dead.is_none() && !cloud { + project.dead = Some(Dead::NotCloud); + log::warn!("{host} is not LiveKit Cloud: client telemetry stays on this device"); + } + if matches!(project.dead, Some(Dead::NotCloud | Dead::Disabled)) { + return Some(host); + } + let token = Token::from(token); + if project.dead == Some(Dead::NotFound) { + log::debug!("new token for {host}: trying its ingest again"); + project.dead = None; + } + if token.grant { + project.consented = true; + if project.dead == Some(Dead::NoGrant) { + project.dead = None; + } + } else if !project.consented && project.dead.is_none() { + project.dead = Some(Dead::NoGrant); + log::info!("the token for {host} has no observability grant: telemetry stays local"); + } + offer(&mut project.latest, &token, &self.refused); + let key = (host.clone(), owner.to_owned()); + let mut own = self.sessions.remove(&key); + offer(&mut own, &token, &self.refused); + self.sessions.extend(own.map(|own| (key, own))); + self.latest = Some(host.clone()); + Some(host) + } + + /// Where a batch goes: `host` is its project as captured with its records, `owner` the + /// session that produced them. A Room known to this process sends with its own latest token + /// for that project and waits when it has none; process-level batches and a previous + /// launch's use the project's latest. A Room's records captured before it had a server go + /// to its first project, never to another Room's. + pub fn route(&self, host: Option<&str>, owner: Option<&str>) -> Route { + if let Some(target) = &self.endpoint_override { + return Route::Send(target.clone()); + } + let room = owner.filter(|o| *o != PROCESS_OWNER); + let host = match (host, room) { + (Some(host), _) => host, + (None, Some(room)) => match self.rooms.get(room) { + Some(first) => first.as_str(), + None => return Route::Wait, + }, + (None, None) => match self.fallback() { + Some(host) => host, + // Nothing to talk to yet: wait. Only dead projects so far: nobody will ever + // take process-level data, so it is not kept either. + None if self.all_dead() => return Route::Drop, + None => return Route::Wait, + }, + }; + let Some(project) = self.projects.get(host) else { return Route::Wait }; + if project.dead.is_some() { + return Route::Drop; + } + let token = match room { + Some(room) if self.rooms.contains_key(room) => { + self.sessions.get(&(host.to_owned(), room.to_owned())) + } + _ => project.latest.as_ref(), + }; + match token { + Some(token) if token.usable(&self.refused) => Route::Send(Target { + project: Some(host.to_owned()), + logs: ingest_url(host, Signal::Logs), + traces: ingest_url(host, Signal::Traces), + token: Some(token.raw.clone()), + }), + _ => Route::Wait, + } + } + + /// The host of the project process-level batches go to: the latest, else any alive one. + fn fallback(&self) -> Option<&str> { + let alive = |host: &&String| self.projects.get(*host).is_some_and(|p| p.dead.is_none()); + self.latest + .as_ref() + .filter(alive) + .or_else(|| self.projects.keys().find(alive)) + .map(String::as_str) + } + + fn all_dead(&self) -> bool { + !self.projects.is_empty() && self.projects.values().all(|p| p.dead.is_some()) + } + + /// Whether session `owner`, routed to `host`, should still collect: `false` once its + /// project is known to receive nothing, so its records are dropped at the door instead of + /// cached. A Room without a server yet waits for its first project, like [`route`](Self::route). + pub fn alive(&self, host: Option<&str>, owner: Option<&str>) -> bool { + !matches!(self.route(host, owner), Route::Drop) + } + + /// The collector refused `token` (401/403): it is not sent again, from any slot; batches that + /// would use it wait for the next one. + pub fn refuse(&mut self, token: &str) { + if self.refused.add(&Token::from(token)) { + log::warn!("the collector refused a token; waiting for a new one"); + } + } + + /// Keep credentials only where something still needs them. `live` maps every session still + /// alive — held by its Room, or by records, windows or spans not yet cached — to its current + /// project; its credentials stay (for each project it was routed to: records it captured for + /// an earlier project may still be queued). `backlog` is (project or none, owner) of every + /// cached batch: their credentials stay too. Project-level copies stay while a live session + /// is routed there or the backlog has batches for the project (process-level batches: the + /// latest project). + pub fn retain_owners( + &mut self, + live: &HashMap>, + backlog: &std::collections::HashSet<(Option, String)>, + ) { + // A Room's batch captured before it had a server is cached without a project: it goes + // to that Room's first project, so it backs that project's credential. + let backlog: std::collections::HashSet<(Option, String)> = backlog + .iter() + .map(|(host, owner)| match (host, self.rooms.get(owner)) { + (None, Some(first)) if owner != PROCESS_OWNER => { + (Some(first.clone()), owner.clone()) + } + _ => (host.clone(), owner.clone()), + }) + .collect(); + let backed = + |host: &str, owner: &str| backlog.contains(&(Some(host.to_owned()), owner.to_owned())); + self.sessions.retain(|(host, owner), _| backed(host, owner) || live.contains_key(owner)); + self.rooms + .retain(|owner, _| live.contains_key(owner) || backlog.iter().any(|(_, o)| o == owner)); + let unrouted = backlog.iter().any(|(host, _)| host.is_none()); + let latest = self.latest.clone(); + // A kept Room credential or first project keeps its project's entry too. + let held: std::collections::HashSet = self + .sessions + .keys() + .map(|(host, _)| host) + .chain(self.rooms.values()) + .cloned() + .collect(); + // An entry nothing needs goes: the table follows the live sessions and the backlog. A + // dead entry stays for the process, so a later Room or process-level record is not + // collected for a project that takes nothing. + self.projects.retain(|host, project| { + let needed = live.values().any(|route| route.as_deref() == Some(host)) + || backlog.iter().any(|(h, _)| h.as_deref() == Some(host)) + || (unrouted && latest.as_deref() == Some(host)) + || held.contains(host); + if !needed { + project.latest = None; + } + needed || project.dead.is_some() + }); + } + + /// How many credentials are held, Room and project copies together. + #[cfg(test)] + pub fn credentials(&self) -> usize { + self.sessions.len() + self.projects.values().filter(|p| p.latest.is_some()).count() + } + + /// Try `host`'s ingest again if it answered 404 (a reconnect). Returns whether it did. + pub fn revive(&mut self, host: &str) -> bool { + match self.projects.get_mut(host) { + Some(project) if project.dead == Some(Dead::NotFound) => { + project.dead = None; + true + } + _ => false, + } + } + + /// The project on `host` receives nothing (see [`Dead`] for how long). + pub fn kill(&mut self, host: &str, why: Dead) { + let project = self.projects.entry(host.to_owned()).or_default(); + if project.dead.is_none() { + match why { + Dead::NotFound => log::warn!("{host} has no client telemetry ingest; going silent"), + Dead::Disabled => log::warn!("{host} disabled data recording; going silent"), + Dead::NotCloud | Dead::NoGrant => {} + } + project.dead = Some(why); + } + } +} + +/// A server URL, parsed (WHATWG URL rules, the same a browser or `URLSession` applies): its +/// routing key (a Cloud project's host, or `+` and the escaped host and port for any other +/// URL, for example `ws+localhost%3A7880`) and whether it is a LiveKit Cloud project a token may be +/// sent to. The key ends every cached batch's id, which is a file name: it holds no character a +/// file name cannot hold on any platform. +struct Server { + key: String, + cloud: bool, +} + +impl Server { + fn parse(url: &str) -> Option { + let url = url::Url::parse(url).ok()?; + let host = url.host_str().filter(|h| !h.is_empty())?.to_ascii_lowercase(); + let key = match url.port() { + Some(port) => format!("{host}:{port}"), + None => host.clone(), + }; + // Credentials only ever go to `https://.livekit.cloud/…`, built from the + // validated host alone: a TLS scheme, a domain of letters, digits, `-` and `.` (a DNS + // hostname) under the Cloud suffix with a label of its own, the default port, no + // userinfo. Anything else — a look-alike that only a string check would accept + // (`evil.example\.livekit.cloud` is host `evil.example`), a plaintext scheme, an odd port + // — is not Cloud. + let cloud = matches!(url.scheme(), "wss" | "https") + && matches!(url.host(), Some(url::Host::Domain(_))) + && host.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'.') + && host.strip_suffix(CLOUD_SUFFIX).is_some_and(|label| !label.is_empty()) + && url.port().is_none() + && url.username().is_empty() + && url.password().is_none(); + // A non-Cloud URL never shares a key with a Cloud host: it can neither feed nor kill it. + let key = if cloud { key } else { format!("{}+{}", url.scheme(), file_safe(&key)) }; + Some(Self { key, cloud }) + } +} + +/// `text` with every byte other than a lowercase letter, a digit, `.`, `-` or `_` written as +/// `%XX`: a port's `:`, the brackets and colons of an IPv6 address and the `*` or `"` a URL host +/// may hold are not allowed in a Windows file name. +fn file_safe(text: &str) -> String { + let mut safe = String::with_capacity(text.len()); + for byte in text.bytes() { + match byte { + b'a'..=b'z' | b'0'..=b'9' | b'.' | b'-' | b'_' => safe.push(byte as char), + _ => safe.push_str(&format!("%{byte:02X}")), + } + } + safe +} + +/// `https:///observability/client/{logs,traces}/otlp/v0`: the client table set. +fn ingest_url(host: &str, signal: Signal) -> String { + let signal = match signal { + Signal::Logs => "logs", + Signal::Traces => "traces", + }; + format!("https://{host}/observability/client/{signal}/otlp/v0") +} + +/// The targets of [`ENDPOINT_OVERRIDE_ENV`]. +fn override_target(endpoint: &str) -> Target { + let endpoint = endpoint.trim_end_matches('/'); + // A logs URL has a whole `logs` path segment; its traces URL swaps that segment. + let traces = url::Url::parse(endpoint).ok().and_then(|mut url| { + let mut segments: Vec = url.path_segments()?.map(str::to_owned).collect(); + let logs = segments.iter().rposition(|s| s == "logs")?; + segments[logs] = "traces".to_owned(); + url.set_path(&segments.join("/")); + Some(url.to_string()) + }); + let (logs, traces) = match traces { + Some(traces) => (endpoint.to_owned(), traces), + None => (format!("{endpoint}/v1/logs"), format!("{endpoint}/v1/traces")), + }; + Target { project: None, logs, traces, token: None } +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + + /// An unsigned JWT with these claims — enough for the core, which never verifies. + pub(crate) fn jwt(claims: serde_json::Value) -> String { + let engine = base64::engine::general_purpose::URL_SAFE_NO_PAD; + format!( + "{}.{}.sig", + engine.encode(br#"{"alg":"HS256"}"#), + engine.encode(claims.to_string()) + ) + } + + pub(crate) fn granted(ttl_secs: u64) -> String { + let exp = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs() + ttl_secs; + jwt(serde_json::json!({ "exp": exp, "observability": { "write": true } })) + } + + pub(crate) fn grantless(ttl_secs: u64) -> String { + let exp = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs() + ttl_secs; + jwt(serde_json::json!({ "exp": exp, "video": { "roomJoin": true } })) + } + + fn send(route: Route) -> Target { + match route { + Route::Send(target) => target, + other => panic!("expected a target, got {other:?}"), + } + } + + #[test] + fn only_livekit_cloud_hosts_get_an_ingest_url() { + let mut d = Destinations::default(); + let token = granted(600); + let host = d.set("wss://Proj.LiveKit.Cloud/rtc?access_token=x#f", &token, "s"); + assert_eq!(host.as_deref(), Some("proj.livekit.cloud")); + let target = send(d.route(host.as_deref(), Some("s"))); + assert_eq!(target.logs, "https://proj.livekit.cloud/observability/client/logs/otlp/v0"); + assert_eq!(target.traces, "https://proj.livekit.cloud/observability/client/traces/otlp/v0"); + assert_eq!(target.token.as_deref(), Some(token.as_str())); + + let staging = d.set("https://p.staging.livekit.cloud", &token, "s"); + assert!(matches!(d.route(staging.as_deref(), None), Route::Send(_)), "staging is Cloud"); + + for url in + ["ws://localhost:7880", "wss://livekit.example.com", "wss://livekit.cloud.evil.io"] + { + let host = d.set(url, &token, "s"); + assert_eq!(d.route(host.as_deref(), None), Route::Drop, "{url}: no ingest"); + } + assert_eq!(d.set("nonsense", &token, "s"), None); + let empty = d.set("wss:///rtc", &token, "s"); // the parser reads host `rtc` + assert_eq!(d.route(empty.as_deref(), None), Route::Drop); + } + + /// Whatever a string check would accept, the token only goes to a host the URL parser agrees is + /// `.livekit.cloud`, over TLS, on the default port, without userinfo — and the + /// endpoint is built from that host alone. + #[test] + fn look_alike_server_urls_never_get_the_token() { + let token = granted(600); + for url in [ + "wss://evil.example\\.livekit.cloud", + "wss://evil.example\\x.livekit.cloud/rtc", + "wss://user:pass@p.livekit.cloud", + "wss://evil.example@p.livekit.cloud", + "wss://p.livekit.cloud:8443", + "ws://p.livekit.cloud", + "http://p.livekit.cloud", + "wss://livekit.cloud", + "wss://.livekit.cloud", + "wss://p.livekit.cloud.evil.io", + "wss://p.livekit.cloud%2eevil.io", + "wss://evil.io#.livekit.cloud", + "wss://evil.io?.livekit.cloud", + "wss://[::1]", + "wss://1.2.3.4", + "wss://p.livekit.cloud\u{0}.evil", + ] { + let mut d = Destinations::default(); + let host = d.set(url, &token, "s"); + let route = d.route(host.as_deref(), Some("s")); + assert!(!matches!(route, Route::Send(_)), "{url} → {host:?} must not get the token"); + } + let mut d = Destinations::default(); + let host = d.set("WSS://P.LiveKit.Cloud:443/rtc?x=1#y", &token, "s"); + let target = send(d.route(host.as_deref(), Some("s"))); + assert_eq!(target.logs, "https://p.livekit.cloud/observability/client/logs/otlp/v0"); + } + + #[test] + fn a_token_without_the_grant_is_not_consent() { + let mut d = Destinations::default(); + let host = d.set("wss://p.livekit.cloud", &grantless(600), "s"); + assert_eq!(d.route(host.as_deref(), Some("s")), Route::Drop, "never opted in"); + assert_eq!(d.route(None, None), Route::Drop, "and no project takes process data"); + let host = d.set("wss://p.livekit.cloud", &granted(600), "s"); + assert!(matches!(d.route(host.as_deref(), Some("s")), Route::Send(_)), "a grant opts in"); + } + + #[tokio::test(start_paused = true)] + async fn a_refresh_that_drops_the_grant_keeps_the_granted_token_until_it_expires() { + let mut d = Destinations::default(); + let join = granted(60); + let host = d.set("wss://p.livekit.cloud", &join, "s"); + d.set("wss://p.livekit.cloud", &grantless(600), "s"); + assert_eq!(send(d.route(host.as_deref(), Some("s"))).token, Some(join), "still granted"); + tokio::time::advance(Duration::from_secs(61)).await; + assert_eq!(d.route(host.as_deref(), Some("s")), Route::Wait, "expired: hold, never drop"); + let refreshed = granted(600); + d.set("wss://p.livekit.cloud", &refreshed, "s"); + assert_eq!(send(d.route(host.as_deref(), Some("s"))).token, Some(refreshed)); + } + + #[test] + fn a_refused_token_is_never_sent_again() { + let mut d = Destinations::default(); + let first = granted(600); + let host = d.set("wss://p.livekit.cloud", &first, "s").expect("host"); + d.refuse(&first); + assert_eq!(d.route(Some(&host), Some("s")), Route::Wait); + d.set("wss://p.livekit.cloud", &first, "s"); + assert_eq!(d.route(Some(&host), Some("s")), Route::Wait, "handing it over again: still no"); + d.set("wss://p.livekit.cloud", &granted(900), "s"); + assert!(matches!(d.route(Some(&host), Some("s")), Route::Send(_))); + } + + #[test] + fn rooms_never_borrow_each_others_tokens() { + let mut d = Destinations::default(); + let (a, b, c) = (granted(600), granted(700), granted(800)); + let host_a = d.set("wss://a.livekit.cloud", &a, "room-a"); + let host_b = d.set("wss://b.livekit.cloud", &b, "room-b"); + d.set("wss://a.livekit.cloud", &c, "room-c"); // a second room on project a + let target = send(d.route(host_a.as_deref(), Some("room-a"))); + assert!(target.logs.starts_with("https://a.") && target.token == Some(a.clone())); + let target = send(d.route(host_a.as_deref(), Some("room-c"))); + assert_eq!(target.token, Some(c.clone()), "same project, its own token"); + let target = send(d.route(host_b.as_deref(), Some("room-b"))); + assert!(target.logs.starts_with("https://b.") && target.token == Some(b)); + d.refuse(&a); + assert_eq!(d.route(host_a.as_deref(), Some("room-a")), Route::Wait, "no borrowing c"); + let previous_launch = send(d.route(host_a.as_deref(), Some("gone"))); + assert_eq!(previous_launch.token, Some(c), "a previous launch's batch: project's latest"); + d.kill("b.livekit.cloud", Dead::Disabled); + assert!(send(d.route(None, None)).logs.starts_with("https://a."), "process: a live one"); + } + + /// A Room's batches keep the project they were captured for — after it reconnects to another + /// project they still go to the first with the first project's token — and records it captured + /// before it had a server never go to another Room's project. + #[test] + fn ownership_survives_a_room_changing_projects() { + let mut d = Destinations::default(); + let (a, b) = (granted(600), granted(700)); + let host_a = d.set("wss://a.livekit.cloud", &a, "room").expect("a"); + d.set("wss://b.livekit.cloud", &b, "room"); + let target = send(d.route(Some(&host_a), Some("room"))); + assert!(target.logs.starts_with("https://a.") && target.token == Some(a)); + + d.set("wss://c.livekit.cloud", &granted(800), "other"); + assert_eq!(d.route(None, Some("unconnected")), Route::Wait, "no borrowing c"); + let early = send(d.route(None, Some("room"))); + assert!(early.logs.starts_with("https://a."), "a Room's first project"); + assert!(matches!(d.route(None, Some(PROCESS_OWNER)), Route::Send(_))); + } + + /// A refusal belongs to the credential — no slot, Room, project or process route can make it + /// usable again — and credentials follow the live Rooms and their backlog, project-level copies + /// included. + #[test] + fn refusals_stick_and_credentials_follow_live_rooms() { + let mut d = Destinations::default(); + let tokens: Vec = (0..40).map(|n| granted(600 + n)).collect(); + for (n, token) in tokens.iter().enumerate() { + d.set("wss://p.livekit.cloud", token, &format!("room-{n}")); + d.refuse(token); + } + for (n, token) in tokens.iter().enumerate() { + let room = format!("room-{n}"); + d.set("wss://p.livekit.cloud", token, &room); // handed over again + assert_eq!(d.route(Some("p.livekit.cloud"), Some(&room)), Route::Wait, "{n}"); + d.set("wss://p.livekit.cloud", token, &format!("new-{n}")); // by a new Room + assert_eq!(d.route(Some("p.livekit.cloud"), Some(&format!("new-{n}"))), Route::Wait); + assert_eq!(d.route(None, Some(PROCESS_OWNER)), Route::Wait, "{n}: process route"); + assert_eq!(d.route(Some("p.livekit.cloud"), Some("gone")), Route::Wait, "{n}: restart"); + } + // Refuse A, let B become the latest, hand A over again: A stays refused everywhere. + let (a, b) = (granted(900), granted(901)); + d.set("wss://q.livekit.cloud", &a, "ra"); + d.refuse(&a); + d.set("wss://q.livekit.cloud", &b, "rb"); + d.set("wss://q.livekit.cloud", &a, "ra"); + let process = send(d.route(Some("q.livekit.cloud"), Some("gone"))); + assert_eq!(process.token, Some(b), "the project copy is b, never the refused a"); + + // Churn: 50 projects, their Rooms gone, nothing cached: nothing is retained. + for n in 0..50 { + d.set(&format!("wss://c{n}.livekit.cloud"), &granted(1000 + n), &format!("churn-{n}")); + } + let live: HashMap> = + [("rb".to_owned(), Some("q.livekit.cloud".to_owned()))].into(); + d.retain_owners(&live, &Default::default()); + assert_eq!(d.credentials(), 2, "rb's token and q's project copy, nothing else"); + let backlog = [(Some("q.livekit.cloud".to_owned()), "ra".to_owned())].into(); + d.set("wss://q.livekit.cloud", &granted(902), "ra"); + d.retain_owners(&Default::default(), &backlog); + assert_eq!(d.credentials(), 2, "a gone Room's backlog keeps its credential and q's copy"); + } + + #[test] + fn not_found_recovers_with_the_next_token_disabled_does_not() { + let mut d = Destinations::default(); + let host = d.set("wss://p.livekit.cloud", &granted(600), "s").expect("host"); + d.kill(&host, Dead::NotFound); + assert_eq!(d.route(Some(&host), Some("s")), Route::Drop); + d.set("wss://p.livekit.cloud", &granted(700), "s"); + assert!(matches!(d.route(Some(&host), Some("s")), Route::Send(_)), "a new connect retries"); + d.kill(&host, Dead::Disabled); + d.set("wss://p.livekit.cloud", &granted(800), "s"); + assert_eq!(d.route(Some(&host), Some("s")), Route::Drop, "the owner's switch stands"); + } + + #[test] + fn the_override_takes_everything_without_a_token() { + let d = Destinations::new(Some("http://localhost:4318/")); + let target = send(d.route(None, None)); + assert_eq!(target.logs, "http://localhost:4318/v1/logs"); + assert_eq!(target.traces, "http://localhost:4318/v1/traces"); + assert_eq!(target.token, None); + let full = Destinations::new(Some("https://h/observability/client/logs/otlp/v0")); + assert_eq!( + send(full.route(Some("ignored"), None)).traces, + "https://h/observability/client/traces/otlp/v0" + ); + let mut local = Destinations::new(Some("http://localhost:4318")); + let host = local.set("ws://localhost:7880", "not-a-jwt", "s"); + assert!(matches!(local.route(host.as_deref(), Some("s")), Route::Send(_))); + } + + #[test] + fn the_override_swaps_only_a_whole_logs_segment() { + let targets = |endpoint| { + let target = send(Destinations::new(Some(endpoint)).route(None, None)); + (target.logs, target.traces) + }; + assert_eq!( + targets("http://logs-collector:4318"), + ( + "http://logs-collector:4318/v1/logs".to_owned(), + "http://logs-collector:4318/v1/traces".to_owned() + ) + ); + assert_eq!( + targets("http://collector:4318/catalogs"), + ( + "http://collector:4318/catalogs/v1/logs".to_owned(), + "http://collector:4318/catalogs/v1/traces".to_owned() + ) + ); + } + + /// A plaintext or userinfo URL on a Cloud host gets a routing key of its own: it never + /// revives the Cloud project with a token from an insecure connect, and never silences it. + #[test] + fn non_cloud_urls_never_share_a_routing_key_with_cloud() { + let cloud = "p.livekit.cloud"; + let mut d = Destinations::default(); + d.set("wss://p.livekit.cloud", &grantless(600), "a"); + let plain = d.set("ws://p.livekit.cloud", &granted(600), "b"); + assert_ne!(plain.as_deref(), Some(cloud)); + assert_eq!(d.route(Some(cloud), Some("b")), Route::Drop, "still never opted in"); + assert_eq!(d.route(plain.as_deref(), Some("b")), Route::Drop, "not Cloud"); + + let mut d = Destinations::default(); + d.set("wss://p.livekit.cloud", &granted(600), "a"); + d.kill(cloud, Dead::NotFound); + d.set("ws://p.livekit.cloud", &granted(700), "b"); + assert_eq!(d.route(Some(cloud), Some("a")), Route::Drop, "only a Cloud connect revives"); + + for first in ["ws://p.livekit.cloud", "wss://user@p.livekit.cloud"] { + let mut d = Destinations::default(); + d.set(first, &granted(600), "a"); + let host = d.set("wss://p.livekit.cloud", &granted(600), "b"); + assert!(matches!(d.route(host.as_deref(), Some("b")), Route::Send(_)), "{first}"); + } + } + + #[test] + fn a_room_without_a_project_waits_for_its_first_one() { + let mut d = Destinations::default(); + d.set("ws://localhost:7880", &granted(600), "a"); + assert!(!d.alive(None, None), "process data: only dead projects so far"); + assert!(d.alive(None, Some("b")), "a Room without a server waits"); + } + + #[test] + fn the_project_table_follows_the_live_sessions() { + let mut d = Destinations::default(); + for n in 0..1000 { + d.set(&format!("wss://p{n}.livekit.cloud"), &granted(600), &format!("room-{n}")); + } + d.kill("p0.livekit.cloud", Dead::Disabled); + d.retain_owners(&Default::default(), &Default::default()); + assert_eq!(d.projects.len(), 1, "only the owner's switch stays"); + let route = d.route(Some("p0.livekit.cloud"), Some("room-0")); + assert_eq!(route, Route::Drop, "and keeps the project silent"); + } + + #[test] + fn routing_keys_are_file_names_on_every_platform() { + // What Windows refuses in a file name: a routing key ends every cached batch's id. + let refused = |c: char| c.is_control() || r#"<>:"/\|?*"#.contains(c); + let urls = [ + "ws://localhost:7880", + "wss://[::1]:7880", + "wss://a*b\"c.example", + "wss://x*y.livekit.cloud", + "https://p.livekit.cloud:8443", + "wss://p.livekit.cloud", + ]; + let mut d = Destinations::default(); + let keys: Vec = + urls.iter().map(|url| d.set(url, &granted(600), "s").expect(url)).collect(); + assert!(keys.iter().all(|key| !key.contains(refused)), "{keys:?}"); + let distinct: std::collections::HashSet<&String> = keys.iter().collect(); + assert_eq!(distinct.len(), keys.len(), "distinct servers keep distinct keys: {keys:?}"); + assert_eq!(keys[5], "p.livekit.cloud", "a Cloud project is keyed by its host"); + } + + #[test] + fn early_records_belong_to_a_first_project_that_is_not_cloud() { + let mut d = Destinations::default(); + let token = granted(600); + let first = d.set("ws://localhost:7880", &token, "room").expect("a host"); + d.set("wss://later.livekit.cloud", &token, "room"); + assert_eq!(d.first_project("room"), Some(first)); + assert_eq!(d.route(None, Some("room")), Route::Drop, "dropped with that project"); + } + + #[test] + fn early_records_belong_to_a_first_project_that_is_disabled() { + let mut d = Destinations::default(); + let token = granted(600); + d.set("wss://disabled.livekit.cloud", &token, "other"); + d.kill("disabled.livekit.cloud", Dead::Disabled); + d.set("wss://disabled.livekit.cloud", &token, "room"); + d.set("wss://later.livekit.cloud", &token, "room"); + assert_eq!(d.first_project("room").as_deref(), Some("disabled.livekit.cloud")); + assert_eq!(d.route(None, Some("room")), Route::Drop, "dropped with that project"); + } + + #[test] + fn a_dead_project_outlives_its_room() { + for (url, token) in + [("ws://localhost:7880", granted(600)), ("wss://p.livekit.cloud", grantless(600))] + { + let mut d = Destinations::default(); + d.set(url, &token, "room"); + d.retain_owners(&Default::default(), &Default::default()); + assert_eq!(d.route(None, Some(PROCESS_OWNER)), Route::Drop, "{url}"); + assert!(!d.alive(None, Some(PROCESS_OWNER)), "{url}: process data is not collected"); + } + } + + /// A past, negative, fractional-past or malformed `exp` is expired (a hard hold); only a + /// missing one means no known expiry. + #[tokio::test(start_paused = true)] + async fn untrusted_expiry_claims_fail_closed() { + let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs_f64(); + let token = |exp: serde_json::Value| { + jwt(serde_json::json!({ "exp": exp, "observability": { "write": true } })) + }; + for exp in [ + serde_json::json!(-1), + serde_json::json!(0), + serde_json::json!(now - 0.5), + serde_json::json!(-1e308), + serde_json::json!("tomorrow"), + serde_json::json!(true), + serde_json::json!(null), + serde_json::json!({}), + ] { + let mut d = Destinations::default(); + let host = d.set("wss://p.livekit.cloud", &token(exp.clone()), "s"); + assert_eq!(d.route(host.as_deref(), Some("s")), Route::Wait, "exp {exp}: expired"); + } + let mut d = Destinations::default(); + let host = d.set("wss://p.livekit.cloud", &token(serde_json::json!(now + 60.9)), "s"); + assert!(matches!(d.route(host.as_deref(), Some("s")), Route::Send(_)), "fractional future"); + let host = d.set("wss://p.livekit.cloud", &token(serde_json::json!(1e300)), "s"); + assert!(matches!(d.route(host.as_deref(), Some("s")), Route::Send(_)), "extreme: clamped"); + } + + #[test] + fn tokens_are_read_not_verified() { + assert!(Token::from(granted(60).as_str()).grant); + let client_write = jwt(serde_json::json!({ "observability": { "clientWrite": true } })); + let parsed = Token::from(client_write.as_str()); + assert!(parsed.grant && parsed.expires.is_none(), "no exp: no known expiry"); + assert!(!Token::from("garbage").grant); + assert!(!Token::from("a.!!!.c").grant); + } +} diff --git a/livekit-telemetry/src/event.rs b/livekit-telemetry/src/event.rs index 79cfb1bcc..a0dd3cfb1 100644 --- a/livekit-telemetry/src/event.rs +++ b/livekit-telemetry/src/event.rs @@ -25,10 +25,12 @@ use std::time::{SystemTime, UNIX_EPOCH}; pub struct TelemetryEvent { /// Event name. LiveKit-defined events use the `lk.` prefix (e.g. `lk.ping`); see `SPEC.md`. pub name: String, + /// How serious it is (OTLP severity number and text). pub severity: Severity, /// Optional human-readable message (the OTLP log record body). #[cfg_attr(feature = "uniffi", uniffi(default))] pub body: Option, + /// The record's own attributes; the session's are added when it is captured. pub attributes: Vec, /// Wall-clock time in nanoseconds since the Unix epoch. `None` is stamped when the record is /// queued, never at export. @@ -43,14 +45,7 @@ pub struct TelemetryEvent { impl TelemetryEvent { /// An `Info` event without attributes, stamped when emitted. pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - severity: Severity::Info, - body: None, - attributes: Vec::new(), - timestamp_ns: None, - span_id: None, - } + Self { name: name.into(), ..Self::default() } } /// Link this record to an in-flight span. @@ -109,21 +104,42 @@ impl TelemetryEvent { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default)] pub enum Severity { + /// OTel `TRACE` (1). Trace, + /// OTel `DEBUG` (5). Debug, + /// OTel `INFO` (9). #[default] Info, + /// OTel `WARN` (13). Warn, + /// OTel `ERROR` (17). Error, } +impl Severity { + /// The OTel severity text (`WARN`). + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Trace => "TRACE", + Self::Debug => "DEBUG", + Self::Info => "INFO", + Self::Warn => "WARN", + Self::Error => "ERROR", + } + } +} + /// Where a log line came from. WebRTC is chatty at warn, so only its errors become records; /// the SDK and the core use the configured floor. #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum LogSource { + /// The platform SDK's own code (Swift, Kotlin, Dart, JS). Sdk, + /// The Rust core, through the FFI. Ffi, + /// libwebrtc. WebRtc, } @@ -143,20 +159,28 @@ impl LogSource { #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq)] pub struct LogRecord { + /// How serious the line is; the core applies the source's floor to it. pub severity: Severity, + /// Who wrote the line: it picks the floor, and is exported as `lk.log.source`. pub source: LogSource, /// The line itself (the OTLP log body). Not `message`: AGENTS.md keeps that name off every /// exported record. pub body: String, - /// The logger: a type, module or file name (`Room`, `livekit::rtc_engine`, `sctp.cc`). + /// The logger: a type, module or file name (`Room`, `livekit::rtc_engine`, `sctp.cc`), + /// exported as `lk.log.logger` to group lines in the backend. #[cfg_attr(feature = "uniffi", uniffi(default))] pub logger: Option, + /// The function that logged, exported as `code.function.name`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub function: Option, + /// The source file, exported as `code.file.path`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub file: Option, + /// The line in `file`, exported as `code.line.number` (0 is no line). #[cfg_attr(feature = "uniffi", uniffi(default))] pub line: Option, + /// When the line was logged, in nanoseconds since the Unix epoch; `None` is stamped when the + /// record is queued. #[cfg_attr(feature = "uniffi", uniffi(default))] pub timestamp_ns: Option, /// The in-flight span this line was logged under, if any. @@ -180,7 +204,7 @@ impl From for TelemetryEvent { event = event.with_attribute("code.file.path", file); } if let Some(line) = record.line.filter(|l| *l > 0) { - event = event.with_attribute("code.line.number", line as i64); + event = event.with_attribute("code.line.number", i64::from(line)); } event.timestamp_ns = record.timestamp_ns; event.span_id = record.span_id; @@ -192,7 +216,9 @@ impl From for TelemetryEvent { #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq)] pub struct Attribute { + /// The attribute's name (`lk.room.sid`, `acme.checkout.step`). pub key: String, + /// Its value. pub value: AttributeValue, } @@ -207,9 +233,13 @@ impl Attribute { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, PartialEq)] pub enum AttributeValue { + /// A string (OTLP `string_value`). Str(String), + /// A 64-bit integer (OTLP `int_value`). Int(i64), + /// A 64-bit float (OTLP `double_value`). Double(f64), + /// A boolean (OTLP `bool_value`). Bool(bool), } @@ -245,8 +275,10 @@ impl From for AttributeValue { /// Current wall-clock time in nanoseconds since the Unix epoch (0 if the clock is before 1970). pub(crate) fn now_unix_nanos() -> u64 { - let now = - SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_nanos() as u64).unwrap_or(0); + let now = SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| { + // Past the year 2554 a u64 of nanoseconds runs out: saturate rather than wrap. + u64::try_from(d.as_nanos()).unwrap_or(u64::MAX) + }); #[cfg(test)] let now = now.saturating_add_signed(CLOCK_JUMP_NS.with(|jump| jump.get())); now @@ -254,7 +286,9 @@ pub(crate) fn now_unix_nanos() -> u64 { #[cfg(test)] thread_local! { - /// How far a test moved the wall clock, on this (current-thread runtime) thread. + /// How far a test moved the wall clock, on this (current-thread runtime) thread: the + /// capture-time test (`queued_records_carry_their_capture_time_not_their_export_time`) moves + /// it between capture and export, which no real clock does on demand. pub(crate) static CLOCK_JUMP_NS: std::cell::Cell = const { std::cell::Cell::new(0) }; } @@ -263,7 +297,9 @@ thread_local! { /// truncated id silently collides with another. #[expect(dead_code, reason = "enforced by the pipeline, not in place yet")] pub(crate) const MAX_NAME_BYTES: usize = 128; +/// The longest attribute key an app may hand over, in bytes. pub(crate) const MAX_KEY_BYTES: usize = 128; +/// The longest string attribute value an app may hand over, in bytes. pub(crate) const MAX_VALUE_BYTES: usize = 1024; /// Custom attributes per room, and per custom event. pub(crate) const MAX_CUSTOM_ATTRIBUTES: usize = 64; @@ -271,7 +307,7 @@ 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`. -pub(crate) fn reserved(key: &str) -> bool { +pub(crate) fn is_reserved(key: &str) -> bool { key.starts_with("lk.") || key.starts_with("otel.") || key.starts_with("code.") @@ -285,7 +321,7 @@ pub(crate) fn valid_custom(key: &str, value: Option<&AttributeValue>) -> bool { Some(AttributeValue::Str(s)) => s.len() <= MAX_VALUE_BYTES, _ => true, }; - !key.is_empty() && key.len() <= MAX_KEY_BYTES && !reserved(key) && value_ok + !key.is_empty() && key.len() <= MAX_KEY_BYTES && !is_reserved(key) && value_ok } /// Encoded bytes of a log record beyond its strings: timestamps, severity, trace id and the @@ -294,6 +330,9 @@ pub(crate) fn valid_custom(key: &str, value: Option<&AttributeValue>) -> bool { const RECORD_OVERHEAD_BYTES: usize = 78; /// Encoded bytes of an attribute beyond its key and string value: tags, lengths and wrappers. const ATTRIBUTE_OVERHEAD_BYTES: usize = 8; +/// Encoded bytes of a number or boolean value: a varint takes 1 to 10 and a double 8, so 8 is +/// the round figure the encoder test calibrates. +const NUMBER_VALUE_BYTES: usize = 8; /// Rough encoded size of `attributes`. pub(crate) fn attributes_size_hint(attributes: &[Attribute]) -> usize { @@ -305,7 +344,7 @@ impl AttributeValue { fn size_hint(&self) -> usize { match self { AttributeValue::Str(s) => s.len(), - _ => 8, + _ => NUMBER_VALUE_BYTES, } } } diff --git a/livekit-telemetry/src/lib.rs b/livekit-telemetry/src/lib.rs index 4620e0f31..56c68873b 100644 --- a/livekit-telemetry/src/lib.rs +++ b/livekit-telemetry/src/lib.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +#![doc = include_str!("../README.md")] + /// Event data model: what SDKs push in. mod event; @@ -29,12 +31,13 @@ mod store; )] mod stats; -/// Spans: one attempt at an operation, with explicit handles across the FFI. +/// Session identity: the trace id, attributes and route of one Room. #[expect( dead_code, reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" )] mod scope; +/// Spans: one attempt at an operation, with explicit handles across the FFI. #[expect( dead_code, reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" @@ -51,9 +54,30 @@ mod otlp; /// OTLP protobuf types (re-exported from `opentelemetry-proto`). mod proto; +/// Transport seam: how encoded batches leave the device. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" + ) +)] +mod transport; + +/// Where batches go: server URL + token → ingest URL, grant, expiry, per-project routing. +#[expect( + dead_code, + reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" +)] +mod destination; + +pub use destination::ENDPOINT_OVERRIDE_ENV; pub use event::*; pub use span::SpanOutcome; pub use stats::{TelemetryStats, TelemetryStatus}; +#[cfg(feature = "net")] +pub use transport::NetTransport; +pub use transport::{ExportError, ExportRequest, ExportResponse, TelemetryTransport}; #[cfg(feature = "uniffi")] uniffi::setup_scaffolding!(); @@ -61,11 +85,17 @@ uniffi::setup_scaffolding!(); /// Warnings logged by the tests, so a test can assert the ones it caused. #[cfg(test)] pub(crate) mod test_log { + use std::cell::RefCell; use std::sync::{Mutex, Once}; use std::thread::{self, ThreadId}; static LINES: Mutex> = Mutex::new(Vec::new()); + thread_local! { + /// Asked at every line this thread logs; a `true` marks the line `held: `. + static HELD: RefCell bool>>> = const { RefCell::new(None) }; + } + struct Capture; impl log::Log for Capture { @@ -75,7 +105,9 @@ pub(crate) mod test_log { fn log(&self, record: &log::Record) { if self.enabled(record.metadata()) { - let line = (thread::current().id(), record.args().to_string()); + let held = HELD.with(|held| held.borrow().as_ref().is_some_and(|held| held())); + let line = format!("{}{}", if held { "held: " } else { "" }, record.args()); + let line = (thread::current().id(), line); LINES.lock().unwrap_or_else(|e| e.into_inner()).push(line); } } @@ -92,6 +124,12 @@ pub(crate) mod test_log { }); } + /// Mark every line this thread logs while `held` returns `true` (a lock taken, say) with a + /// `held: ` prefix: the host logger would have run at that moment. + pub fn mark_while(held: impl Fn() -> bool + 'static) { + HELD.with(|slot| *slot.borrow_mut() = Some(Box::new(held))); + } + /// The warnings logged on this thread since the last call: this test's own, since every /// test runs on a thread of its own. pub fn take() -> Vec { diff --git a/livekit-telemetry/src/otlp.rs b/livekit-telemetry/src/otlp.rs index e34e486f3..02539d1ec 100644 --- a/livekit-telemetry/src/otlp.rs +++ b/livekit-telemetry/src/otlp.rs @@ -29,6 +29,7 @@ use crate::{ Attribute, AttributeValue, Severity, SpanOutcome, }; +/// The `Content-Type` of an OTLP/HTTP protobuf request. pub(crate) const CONTENT_TYPE: &str = "application/x-protobuf"; fn resource(attributes: &[Attribute]) -> Option { @@ -133,14 +134,20 @@ fn log_record( // duplicates `EventName` for backends that do not read the field yet (Loki); an app cannot // set it (`otel.*` is reserved). let body = event.body.or_else(|| (!event.name.is_empty()).then(|| event.name.clone())); + // The core's copy is the only one: a pipeline attribute of that name never ships. + event.attributes.retain(|a| a.key != "otel.event.name"); if !event.name.is_empty() { event.attributes.push(Attribute::new("otel.event.name", event.name.clone())); } + last_value_wins(&mut event.attributes); LogRecord { time_unix_nano, + // When the collecting side saw the record: for one recorded in process, OTel SDKs set + // it equal to the timestamp (9 bytes). observed_time_unix_nano: time_unix_nano, severity_number: SeverityNumber::from(event.severity) as i32, - severity_text: severity_text(event.severity).to_owned(), + // The text as well as the number, as OTel SDKs send them: a log view shows the text. + severity_text: event.severity.as_str().to_owned(), body: body.map(|text| AnyValue { value: Some(any_value::Value::StringValue(text)) }), attributes: event.attributes.iter().map(KeyValue::from).collect(), event_name: event.name, @@ -151,14 +158,13 @@ fn log_record( } fn otlp_span(mut record: SpanRecord, global: &[Attribute]) -> Span { - let session = record.session.clone(); fill_in_global(&mut record.attributes, global); // The outcome is the core's, as `session.id` is: a span's own copies never ship. record.attributes.retain(|a| a.key != "lk.outcome" && a.key != "error.type"); let mut attributes: Vec = record.attributes.iter().map(KeyValue::from).collect(); attributes.extend(record.outcome_attributes().iter().map(KeyValue::from)); Span { - trace_id: session.trace_id.to_vec(), + trace_id: record.session.trace_id.to_vec(), span_id: record.span_id.to_be_bytes().to_vec(), parent_span_id: record.parent_span_id.map(|p| p.to_be_bytes().to_vec()).unwrap_or_default(), name: record.name, @@ -196,6 +202,21 @@ fn fill_in_global(own: &mut Vec, global: &[Attribute]) { } } +/// One attribute per key, as OTLP requires (a receiver's choice among repeats is unpredictable): +/// the last value supplied wins, in the place of that last copy. Only a caller's own list can +/// repeat a key, so the precedence above holds: the session's attributes, `session.id` and +/// `otel.event.name` replaced every copy of their keys, and the rest only fill in missing keys. +fn last_value_wins(attributes: &mut Vec) { + let mut i = 0; + while i < attributes.len() { + if attributes[i + 1..].iter().any(|a| a.key == attributes[i].key) { + attributes.remove(i); + } else { + i += 1; + } + } +} + impl From for span::SpanKind { /// The OTLP value of a span kind. fn from(kind: SpanKind) -> Self { @@ -218,16 +239,6 @@ impl From for SeverityNumber { } } -fn severity_text(severity: Severity) -> &'static str { - match severity { - Severity::Trace => "TRACE", - Severity::Debug => "DEBUG", - Severity::Info => "INFO", - Severity::Warn => "WARN", - Severity::Error => "ERROR", - } -} - impl From<&Attribute> for KeyValue { fn from(attribute: &Attribute) -> Self { let value = match &attribute.value { @@ -249,6 +260,59 @@ mod tests { use super::*; use crate::TelemetryEvent; + /// A pipeline attribute named `otel.event.name` never ships a second copy of the key: the + /// core's own, from the event's name, is the only one. + #[test] + fn otel_event_name_is_sent_once() { + let session = crate::scope::ScopeState::new(); + let global = [Attribute::new("otel.event.name", "sdk-meta")]; + let queued = vec![QueuedEvent::new(TelemetryEvent::new("lk.ping"), session)]; + let decoded = ExportLogsServiceRequest::decode(&encode_logs(&[], &global, queued)[..]) + .expect("valid OTLP"); + let record = &decoded.resource_logs[0].scope_logs[0].log_records[0]; + let names: Vec<_> = + record.attributes.iter().filter(|kv| kv.key == "otel.event.name").collect(); + assert_eq!(names.len(), 1, "{names:?}"); + let value = names[0].value.as_ref().and_then(|v| v.value.as_ref()); + assert_eq!(value, Some(&any_value::Value::StringValue("lk.ping".into()))); + } + + /// OTLP keys are unique: of a caller's repeated keys the last value ships, and a session key + /// the caller repeats still ships the session's value. + #[test] + fn repeated_caller_keys_ship_once_with_the_last_value() { + let session = crate::scope::ScopeState::with_trace_id([7u8; 16]); + let custom = TelemetryEvent::custom( + "checkout", + vec![Attribute::new("app.step", "first"), Attribute::new("app.step", "second")], + ); + let sdk = TelemetryEvent::new("lk.ping") + .with_attribute("session.id", "spoofed") + .with_attribute("session.id", "spoofed again"); + let queued = + vec![QueuedEvent::new(custom, session.clone()), QueuedEvent::new(sdk, session)]; + let decoded = ExportLogsServiceRequest::decode(&encode_logs(&[], &[], queued)[..]) + .expect("valid OTLP"); + let records = &decoded.resource_logs[0].scope_logs[0].log_records; + let values = |record: &LogRecord, key: &str| -> Vec<_> { + let values = record.attributes.iter().filter(|kv| kv.key == key); + values.filter_map(|kv| kv.value.clone()?.value).collect() + }; + assert_eq!( + values(&records[0], "app.step"), + [any_value::Value::StringValue("second".into())] + ); + let session_id = any_value::Value::StringValue("07".repeat(16)); + assert_eq!(values(&records[1], "session.id"), [session_id]); + for record in records { + let mut keys: Vec<_> = record.attributes.iter().map(|kv| &kv.key).collect(); + let count = keys.len(); + keys.sort_unstable(); + keys.dedup(); + assert_eq!(keys.len(), count, "unique keys: {:?}", record.attributes); + } + } + #[test] fn encodes_events_as_otlp_log_records() { let resource = [Attribute::new("service.name", "test")]; diff --git a/livekit-telemetry/src/span.rs b/livekit-telemetry/src/span.rs index 5987d7d60..53dee7655 100644 --- a/livekit-telemetry/src/span.rs +++ b/livekit-telemetry/src/span.rs @@ -23,7 +23,7 @@ use crate::{ /// OTel span kind, restricted to what client operations need; implied by [`crate::SpanName`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SpanKind { +pub(crate) enum SpanKind { /// An operation inside the SDK (publish, subscribe). Internal, /// A call to the SFU that waits for its answer (connect, reconnect). @@ -36,8 +36,11 @@ pub enum SpanKind { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SpanOutcome { + /// It did what it set out to do (OTel status `Ok`). Ok, + /// It failed; `error.type` says how (OTel status `Error`). Error, + /// It was called off before it finished, which is not a failure (OTel status `Unset`). Cancelled, } @@ -89,8 +92,9 @@ const SPAN_OVERHEAD_BYTES: usize = 94; /// Encoded bytes of a span event beyond its name: its timestamp and framing. const SPAN_EVENT_OVERHEAD_BYTES: usize = 13; -/// OTel default span limits. +/// Checkpoints one span keeps: OTel's default span limit. const MAX_EVENTS_PER_SPAN: usize = 128; +/// Attributes one span keeps: OTel's default span limit. const MAX_ATTRIBUTES_PER_SPAN: usize = 128; /// Spans a host can leave open before the oldest is abandoned (counted as dropped). const MAX_OPEN_SPANS: usize = 256; @@ -104,7 +108,7 @@ pub(crate) struct Spans { open: HashMap, /// Insertion order of `open`, to abandon the oldest when the cap is hit. open_order: VecDeque, - finished: Vec, + finished: VecDeque, finished_capacity: usize, /// Which session every recent span belongs to — open, finished or already exported — so a /// log record that arrives after its span ended (a warning logged right before a failing @@ -114,7 +118,7 @@ pub(crate) struct Spans { /// open span keeps its session however many spans start after it. session_order: VecDeque, next_id: u64, - pub dropped: u64, + dropped: u64, warned: Warned, /// The opt-out, checked under the lock that guards the registry (see `Store`). revoked: Arc, @@ -150,13 +154,13 @@ impl Spans { Self { open: HashMap::new(), open_order: VecDeque::new(), - finished: Vec::new(), + finished: VecDeque::new(), finished_capacity: finished_capacity.max(1), sessions: HashMap::new(), session_order: VecDeque::new(), - // Span ids must be non-zero (OTLP treats all-zero as absent); start at 1 and mix in - // randomness so ids from two pipelines in one process never collide. 63 bits: a - // platform whose integers are signed (Dart) must be able to hand an id back. + // A random odd 63-bit start: never zero (OTLP treats all-zero as absent), and two + // pipelines in one process are very unlikely to overlap. 63 bits: a platform whose + // integers are signed (Dart) must be able to hand an id back. next_id: (rand::random::() >> 1) | 1, dropped: 0, warned: Warned::default(), @@ -174,7 +178,8 @@ impl Spans { self.revoked.load(std::sync::atomic::Ordering::SeqCst) } - /// Open a span in `session`'s trace. + /// Open a span in `session`'s trace. After the opt-out nothing opens: returns 0, which is + /// never a span id. pub fn begin_in( &mut self, name: &str, @@ -216,8 +221,6 @@ impl Spans { id } - /// Whether a span with one of these names is still open (the exporter holds uploads while - /// `lk.connect` / `lk.reconnect` are). /// The session a recent span belongs to — open, ended or exported (log records emitted inside /// a span are filed there, and they may arrive after the span ended). pub fn scope_of(&self, id: u64) -> Option> { @@ -234,6 +237,8 @@ impl Spans { self.open.len() } + /// Whether a span with one of these names is still open (the exporter holds uploads while + /// `lk.connect` / `lk.reconnect` are). pub fn any_open(&self, names: &[&str]) -> bool { self.open.values().any(|span| names.contains(&span.name.as_str())) } @@ -302,10 +307,10 @@ impl Spans { span.attributes = attributes; self.retire(id); if self.finished.len() >= self.finished_capacity { - self.finished.remove(0); + self.finished.pop_front(); self.count_drop("finished", self.finished_capacity); } - self.finished.push(span); + self.finished.push_back(span); } /// A span ended or was abandoned: its session stays resolvable until [`REMEMBERED_SPANS`] @@ -329,8 +334,8 @@ impl Spans { } /// Take the finished spans, oldest first. - /// At most `max` spans and about `max_bytes` (always at least one, so an oversized span - /// still ships). + /// At most `max` spans and about `max_bytes` (at least one when `max > 0`, so an oversized + /// span still ships). pub fn drain(&mut self, max: usize, max_bytes: usize) -> Vec { self.warned = Warned::default(); let (mut n, mut bytes) = (0, 0); diff --git a/livekit-telemetry/src/stats.rs b/livekit-telemetry/src/stats.rs index 26a5d759e..f2a10ea4e 100644 --- a/livekit-telemetry/src/stats.rs +++ b/livekit-telemetry/src/stats.rs @@ -50,7 +50,8 @@ macro_rules! counters { } counters! { - /// Records evicted from the in-memory queue (`max_queue_size`). + /// Records evicted from the in-memory queue (`max_queue_size`), and spans dropped from a full + /// span buffer (open spans abandoned, finished ones past its capacity). queue_full, /// Records lost because no cache, not even memory, could take their batch. cache_error, @@ -101,18 +102,13 @@ impl Counters { impl Snapshot { /// Everything that was lost, by any reason. pub fn dropped(&self) -> u64 { - self.queue_full - + self.cache_error - + self.cache_full - + self.expired - + self.corrupt - + self.invalid - + self.rejected - + self.oversized - + self.throttled - + self.disabled - + self.rate_limited - + self.purged + self.problem_drops() + self.disabled + self.purged + } + + /// Upload attempts that did not get through: failures and timeouts together, the "failed" + /// of the report's body (the `lk.telemetry.uploads.failed` attribute counts failures only). + pub fn failed_attempts(&self) -> u64 { + self.upload_failures + self.upload_timeouts } /// Records lost other than by policy (a project that receives nothing, an opt-out). @@ -150,14 +146,14 @@ impl Snapshot { "telemetry: {} batches sent ({} B), {} failed, {} dropped, {} cached", self.uploads_sent, self.upload_bytes, - self.upload_failures + self.upload_timeouts, + self.failed_attempts(), self.problem_drops(), cached_batches )) - .with_attribute("lk.telemetry.uploads.sent", self.uploads_sent as i64) - .with_attribute("lk.telemetry.uploads.bytes", self.upload_bytes as i64) - .with_attribute("lk.telemetry.uploads.failed", self.upload_failures as i64) - .with_attribute("lk.telemetry.cache.batches", cached_batches as i64); + .with_attribute("lk.telemetry.uploads.sent", int(self.uploads_sent)) + .with_attribute("lk.telemetry.uploads.bytes", int(self.upload_bytes)) + .with_attribute("lk.telemetry.uploads.failed", int(self.upload_failures)) + .with_attribute("lk.telemetry.cache.batches", int(cached_batches)); for (key, value) in [ ("lk.telemetry.uploads.timeouts", self.upload_timeouts), ("lk.telemetry.uploads.unauthorized", self.auth_denied), @@ -175,13 +171,18 @@ impl Snapshot { ("lk.telemetry.dropped.rate_limited", self.rate_limited), ] { if value > 0 { - event = event.with_attribute(key, value as i64); + event = event.with_attribute(key, int(value)); } } event } } +/// A count as an OTLP integer: past `i64::MAX` it saturates rather than wraps negative. +fn int(count: u64) -> i64 { + i64::try_from(count).unwrap_or(i64::MAX) +} + /// What the upload policy is doing right now. #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -224,17 +225,29 @@ pub struct TelemetryStats { pub status: TelemetryStatus, /// Records lost for any reason (sum of the `dropped_*` fields). pub dropped: u64, + /// Records evicted from the in-memory queue, and spans dropped from a full span buffer. pub dropped_queue_full: u64, + /// Records lost because no cache, not even memory, could take their batch. pub dropped_cache_error: u64, + /// Records evicted from the cache by its size or file-count bound. pub dropped_cache_full: u64, + /// Records in cached batches past the 24 h age limit. pub dropped_expired: u64, + /// Records in cached batches that failed their integrity check. pub dropped_corrupt: u64, + /// Custom events and attributes refused at the door, over the limits. pub dropped_invalid: u64, + /// Records the collector rejected (a final 4xx/5xx, or a partial success). pub dropped_rejected: u64, + /// Single records larger than the collector accepts. pub dropped_oversized: u64, + /// Records evicted from the cache while the collector held uploads off. pub dropped_throttled: u64, + /// Records for a project that receives nothing (self-hosted, no grant, disabled, 404). pub dropped_disabled: u64, + /// Discrete events dropped by the flood guard. pub dropped_rate_limited: u64, + /// Records deleted by the opt-out. pub dropped_purged: u64, /// Batches the collector accepted. pub uploads_sent: u64, @@ -254,8 +267,9 @@ pub struct TelemetryStats { pub cached_batches: u64, } -/// One line for a debug console: status, throughput, one backlog number, one loss number, then -/// the loss breakdown. +/// One line for a debug console: status, throughput, one backlog number, "failed" (failures and +/// timeouts together, [`TelemetryStats::failed_attempts`]), one loss number, then the loss +/// breakdown. impl std::fmt::Display for TelemetryStats { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( @@ -267,7 +281,7 @@ impl std::fmt::Display for TelemetryStats { self.uploads_sent, self.upload_bytes, self.cached_batches, - self.upload_failures + self.upload_timeouts, + self.failed_attempts(), self.dropped, self.dropped_queue_full, self.dropped_cache_full + self.dropped_cache_error, @@ -288,6 +302,12 @@ impl std::fmt::Display for TelemetryStats { } impl TelemetryStats { + /// Upload attempts that did not get through: `upload_failures` and `upload_timeouts` + /// together, the "failed" of the `Display` line and of the report's body. + pub fn failed_attempts(&self) -> u64 { + self.upload_failures + self.upload_timeouts + } + pub(crate) fn new(snapshot: Snapshot, cached_batches: u64, status: TelemetryStatus) -> Self { Self { status, @@ -315,3 +335,20 @@ impl TelemetryStats { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::AttributeValue; + + /// A count past `i64::MAX` reaches the report as `i64::MAX`, never as a negative number. + #[test] + fn report_counts_saturate_instead_of_wrapping() { + let snapshot = Snapshot { upload_failures: u64::MAX, ..Snapshot::default() }; + let report = snapshot.report(u64::MAX); + let value = + |key: &str| report.attributes.iter().find(|a| a.key == key).map(|a| a.value.clone()); + assert_eq!(value("lk.telemetry.uploads.failed"), Some(AttributeValue::Int(i64::MAX))); + assert_eq!(value("lk.telemetry.cache.batches"), Some(AttributeValue::Int(i64::MAX))); + } +} diff --git a/livekit-telemetry/src/store.rs b/livekit-telemetry/src/store.rs index 075c9d4a8..44120ea39 100644 --- a/livekit-telemetry/src/store.rs +++ b/livekit-telemetry/src/store.rs @@ -59,7 +59,8 @@ pub(crate) struct Store { queue: Mutex, /// The opt-out, checked under the queue lock `clear` also takes: nothing lands after a purge. revoked: Arc, - /// Test-only: runs at the start of `push`, before the lock (to hold a producer there). + /// Test-only: runs at the start of `push`, before the lock, so the opt-out race test + /// (`a_capture_racing_the_opt_out_leaves_nothing_behind`) can hold a producer exactly there. #[cfg(test)] pub(crate) pause: Mutex>>, capacity: usize, @@ -110,23 +111,27 @@ impl Store { if self.revoked.load(Ordering::SeqCst) { return false; } + let mut warn = false; if queue.events.len() >= self.capacity { if let Some(oldest) = queue.events.pop_front() { queue.bytes = queue.bytes.saturating_sub(oldest.event.size_hint()); } Counters::add(&self.counters.queue_full, 1); - if !queue.full_warned { - queue.full_warned = true; - log::warn!( - "queue full ({} records): dropping oldest until the exporter drains", - self.capacity - ); - } + warn = !std::mem::replace(&mut queue.full_warned, true); } let before = queue.bytes; queue.bytes += queued.event.size_hint(); queue.events.push_back(queued); - before < self.flush_threshold && queue.bytes >= self.flush_threshold + let crossed = before < self.flush_threshold && queue.bytes >= self.flush_threshold; + // Logged after the lock: a log line may call into the host. + drop(queue); + if warn { + log::warn!( + "queue full ({} records): dropping oldest until the exporter drains", + self.capacity + ); + } + crossed } /// Remove and return the oldest events: at most `min(max, len)` of them and about `max_bytes` @@ -172,6 +177,23 @@ mod tests { QueuedEvent::new(event, ScopeState::new()) } + /// The queue-full warning reaches the logger once the queue's lock is released: a host logger + /// that calls back into the pipeline cannot block on it. + #[test] + fn the_queue_full_warning_is_logged_outside_the_queue_lock() { + crate::test_log::capture(); + let store = Arc::new(Store::new(1, usize::MAX, Arc::new(Counters::default()))); + let held = Arc::downgrade(&store); + crate::test_log::mark_while(move || { + held.upgrade().is_some_and(|store| store.queue.try_lock().is_err()) + }); + store.push(queued(TelemetryEvent::new("a"))); + store.push(queued(TelemetryEvent::new("b"))); + let lines = crate::test_log::take(); + assert!(lines.iter().any(|line| line.contains("queue full")), "{lines:?}"); + assert!(lines.iter().all(|line| !line.starts_with("held: ")), "{lines:?}"); + } + #[test] fn drops_oldest_when_full() { crate::test_log::capture(); diff --git a/livekit-telemetry/src/transport.rs b/livekit-telemetry/src/transport.rs new file mode 100644 index 000000000..fdc22d273 --- /dev/null +++ b/livekit-telemetry/src/transport.rs @@ -0,0 +1,601 @@ +// 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; + +use prost::Message; + +/// One fully composed OTLP/HTTP export request. +/// +/// The core fills in the URL, the headers (content type, auth, …) and the protobuf body; +/// a transport only moves the bytes. The body is an `ExportLogsServiceRequest` or an +/// `ExportTraceServiceRequest`, as `url` says. +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +#[derive(Debug, Clone, PartialEq)] +pub struct ExportRequest { + /// Where to POST: the OTLP logs or traces endpoint of the batch's project. + pub url: String, + /// Headers to send exactly as given: content type, content encoding, `Authorization`. + pub headers: HashMap, + /// The encoded request body, sent as is. + pub body: Vec, +} + +/// What the collector answered. A transport moves bytes both ways and never interprets them: +/// status classification, `Retry-After`, the `google.rpc.Status` body and the Cloud "disabled" +/// contract are read once, here, for every platform (`Verdict::of`). +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +#[derive(Debug, Clone, PartialEq, Default)] +pub struct ExportResponse { + /// The HTTP status code. + pub status: u16, + /// The response headers; the core reads `Retry-After` from them. + pub headers: HashMap, + /// The response body: an OTLP partial success, a `google.rpc.Status` or plain text. + pub body: Vec, +} + +impl ExportResponse { + /// `2xx`, nothing else: what an accepting collector answers. + pub fn accepted() -> Self { + Self { status: 200, ..Self::default() } + } +} + +/// LiveKit Cloud's answer when the project owner switched data recording off (401 or 403, plain +/// text, no machine-readable code yet). Matched as a phrase, not a word. +const DISABLED_ANSWER: &str = "data recording is disabled by owner"; + +/// The most characters of an error body that a verdict's reason keeps: enough to read the cause in +/// a log line, short enough that a large HTML error page does not fill it. +const MAX_REASON_CHARS: usize = 200; + +/// How long a 429 that names no delay holds uploads: LiveKit Cloud's quota answer names none. +const THROTTLE_DEFAULT_MS: u64 = 60_000; + +/// Why a transport could not get an answer. A 4xx/5xx is an answer, not an error: return it as an +/// [`ExportResponse`] and the core classifies it. +#[cfg_attr(feature = "uniffi", derive(uniffi::Error))] +#[derive(Debug, Clone, PartialEq, thiserror::Error)] +pub enum ExportError { + /// No response: connection refused or reset, DNS or TLS failure, offline, timed out. The + /// batch stays cached and is retried with backoff. + #[error("retryable export error: {reason}")] + Retryable { + /// What went wrong, for the log. + reason: String, + /// How long to wait before the next try, when known. `None` uses the backoff. + retry_after_ms: Option, + }, + /// The request can never succeed (an invalid URL): the batch is dropped. + #[error("export rejected: {reason}")] + Rejected { + /// What went wrong, for the log. + reason: String, + }, +} + +/// A foreign transport threw something its binding did not declare (a Swift `Error`, a Kotlin +/// `RuntimeException`): no answer, retried with backoff — never a panic in the exporter. +#[cfg(feature = "uniffi")] +impl From for ExportError { + fn from(error: uniffi::UnexpectedUniFFICallbackError) -> Self { + Self::Retryable { + reason: format!("transport threw: {}", error.reason), + retry_after_ms: None, + } + } +} + +/// What a collector's answer means for the batch: OTLP/HTTP failure semantics, plus the answers +/// LiveKit Cloud gives on top of them. +#[derive(Debug, Clone, PartialEq)] +pub(crate) enum Verdict { + /// 2xx. `rejected` counts the records an OTLP partial success refused (never retried). + Accepted { rejected: u64, reason: String }, + /// 400, 3xx (a redirect the transport did not follow), and every other 4xx/5xx the OTLP spec + /// does not call retryable: drop the batch. + Rejected(String), + /// 413: split the batch and retry the halves, down to single records. + TooLarge, + /// 401/403 that is not the "disabled" answer: the credential is the problem, not the data — + /// hold the project's batches until the platform hands over a new token. + Unauthorized(String), + /// 404: the host has no client ingest; the project goes silent. + NotFound, + /// 401/403 "data recording is disabled by owner": the project goes silent and its cache is + /// purged. + Disabled, + /// 429, or 503 naming a delay: pause every upload for `delay_ms`, keep collecting. + Throttled { delay_ms: u64, reason: String }, + /// 502/503/504, or 500 carrying `RetryInfo`: the server failed transiently. Retry the batch + /// after `delay_ms` or a backoff; a batch that keeps failing is dropped eventually. + Retry { delay_ms: Option, reason: String }, +} + +impl Verdict { + /// Classify a collector response. + pub(crate) fn of(response: &ExportResponse) -> Self { + let status = response.status; + if (200..300).contains(&status) { + return match PartialSuccessResponse::decode(response.body.as_slice()) { + Ok(PartialSuccessResponse { partial_success: Some(p) }) if p.rejected > 0 => { + Self::Accepted { rejected: p.rejected as u64, reason: p.error_message } + } + _ => Self::Accepted { rejected: 0, reason: String::new() }, + }; + } + // OTLP/HTTP errors are a protobuf `google.rpc.Status`; Cloud's auth answers are text. + let rpc = RpcStatus::decode(response.body.as_slice()).ok(); + let text = match rpc.as_ref().map(|s| s.message.trim()).filter(|m| !m.is_empty()) { + Some(message) => message.to_owned(), + None => String::from_utf8_lossy(&response.body).trim().to_owned(), + }; + // The whole answer is searched for the phrase; only the logged reason is cut. + if matches!(status, 401 | 403) && text.to_ascii_lowercase().contains(DISABLED_ANSWER) { + return Self::Disabled; + } + let text: String = text.chars().take(MAX_REASON_CHARS).collect(); + // `reason`, not `message`: a UniFFI field named `message` collides with + // `Throwable.message` in Kotlin. + let reason = if text.is_empty() { + format!("HTTP {status}") + } else { + format!("HTTP {status}: {text}") + }; + if matches!(status, 401 | 403) { + return Self::Unauthorized(reason); + } + let header = response + .headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case("retry-after")) + .and_then(|(_, value)| retry_after_ms(value)); + let retry_info = rpc.as_ref().and_then(retry_info_ms); + let delay = header.or(retry_info); + // The status decides whether a batch is retried; a delay hint only says when. A hint on a + // final status (400, 422, 501, a redirect) changes nothing. + match status { + 404 => Self::NotFound, + 413 => Self::TooLarge, + // A rate limit is an instruction to stop, so 429 always yields a wait: LiveKit Cloud + // answers a quota check with a bare `ResourceExhausted` — no `Retry-After`, no + // `RetryInfo` — and without one the exporter would spend its retries hammering the + // endpoint that just asked for quiet. + 429 => Self::Throttled { delay_ms: delay.unwrap_or(THROTTLE_DEFAULT_MS), reason }, + 503 if delay.is_some() => { + Self::Throttled { delay_ms: delay.unwrap_or_default(), reason } + } + 502..=504 => Self::Retry { delay_ms: delay, reason }, + // LiveKit Cloud's one retryable 500 carries `RetryInfo` (custom: OTLP retries only + // 429/502/503/504). + 500 if retry_info.is_some() => Self::Retry { delay_ms: retry_info, reason }, + _ => Self::Rejected(reason), + } + } +} + +/// `ExportLogsServiceResponse` and `ExportTraceServiceResponse` share one shape: field 1 is the +/// partial success, whose field 1 counts the rejected records and field 2 says why. +#[derive(Clone, PartialEq, prost::Message)] +struct PartialSuccessResponse { + #[prost(message, optional, tag = "1")] + partial_success: Option, +} + +#[derive(Clone, PartialEq, prost::Message)] +struct PartialSuccess { + #[prost(int64, tag = "1")] + rejected: i64, + #[prost(string, tag = "2")] + error_message: String, +} + +/// `google.rpc.Status`, the OTLP/HTTP error body. Two messages, three fields: not worth a +/// generated crate. +#[derive(Clone, PartialEq, prost::Message)] +struct RpcStatus { + #[prost(int32, tag = "1")] + code: i32, + #[prost(string, tag = "2")] + message: String, + #[prost(message, repeated, tag = "3")] + details: Vec, +} + +/// `google.rpc.RetryInfo`. +#[derive(Clone, PartialEq, prost::Message)] +struct RetryInfo { + #[prost(message, optional, tag = "1")] + retry_delay: Option, +} + +/// RFC 9110 §10.2.3 `Retry-After`: delay-seconds or an IMF-fixdate. Untrusted input: garbage is +/// ignored, a date in the past means now; the caller clamps the far end. +fn retry_after_ms(value: &str) -> Option { + let value = value.trim(); + // delay-seconds: digits only; more than a u64 holds saturates (the exporter clamps it). + if !value.is_empty() && value.bytes().all(|b| b.is_ascii_digit()) { + return Some(value.parse::().unwrap_or(u64::MAX).saturating_mul(1000)); + } + let at = imf_fixdate_secs(value)?; + let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).ok()?.as_secs(); + Some(at.saturating_sub(now).saturating_mul(1000)) +} + +/// `Sun, 06 Nov 1994 08:49:37 GMT` → unix seconds. The obsolete RFC 850 / asctime forms are not +/// accepted (RFC 9110 lets a recipient ignore them). +fn imf_fixdate_secs(value: &str) -> Option { + let mut parts = value.split_ascii_whitespace(); + let (_weekday, day, month, year, time, zone) = + (parts.next()?, parts.next()?, parts.next()?, parts.next()?, parts.next()?, parts.next()?); + if zone != "GMT" || parts.next().is_some() { + return None; + } + const MONTHS: [&str; 12] = + ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]; + let month = MONTHS.iter().position(|m| *m == month)? as i64 + 1; + let (day, year): (i64, i64) = (day.parse().ok()?, year.parse().ok()?); + let leap = year % 4 == 0 && (year % 100 != 0 || year % 400 == 0); + let month_days = match month { + 2 if leap => 29, + 2 => 28, + 4 | 6 | 9 | 11 => 30, + _ => 31, + }; + let mut hms = time.split(':').map(|n| n.parse::().ok()); + let (h, m, s) = (hms.next()??, hms.next()??, hms.next()??); + // IMF-fixdate has a four-digit year; bounding it keeps the arithmetic below in range. + if !(1970..=9999).contains(&year) + || hms.next().is_some() + || !(1..=month_days).contains(&day) + || !(0..24).contains(&h) + || !(0..60).contains(&m) + || !(0..61).contains(&s) + { + return None; + } + // Days from the civil date (proleptic Gregorian), after Howard Hinnant's algorithm. + let y = if month <= 2 { year - 1 } else { year }; + let era = y.div_euclid(400); + let yoe = y - era * 400; + let doy = (153 * (month + if month > 2 { -3 } else { 9 }) + 2) / 5 + day - 1; + let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; + let days = era * 146_097 + doe - 719_468; + u64::try_from(days * 86_400 + h * 3600 + m * 60 + s).ok() +} + +fn retry_info_ms(status: &RpcStatus) -> Option { + status + .details + .iter() + .filter(|detail| detail.type_url.ends_with("google.rpc.RetryInfo")) + .find_map(|detail| RetryInfo::decode(detail.value.as_slice()).ok()?.retry_delay) + .map(|delay| { + (delay.seconds.max(0) as u64) + .saturating_mul(1000) + .saturating_add(delay.nanos.clamp(0, 999_999_999) as u64 / 1_000_000) + }) +} + +/// Moves an encoded batch off the device and hands back whatever came back. +/// +/// Implemented in Rust (`NetTransport`, feature `net`) or by the host platform through UniFFI. +/// The contract: +/// +/// - POST `url` with `headers` and `body` as given, and return the response whatever its status; +/// fail only when there was no response (connection, DNS, TLS, timeout) — +/// [`ExportError::Retryable`] — or the request cannot be made at all +/// ([`ExportError::Rejected`], e.g. an invalid URL). +/// - Never retry: the core owns the retry policy. +/// - Never forward the `Authorization` header across origins: do not follow a redirect to +/// another scheme, host or port with it (returning the 3xx is fine — the core drops the batch). +/// `NetTransport` over the native `livekit-net` client follows at most 10 redirects, never one +/// that changes the scheme, and drops `Authorization` when a redirect changes the host or port. +/// - Keep the request off the call's critical path: lowest priority where the platform allows it. +#[cfg_attr(feature = "uniffi", uniffi::export(with_foreign))] +#[async_trait::async_trait] +pub trait TelemetryTransport: Send + Sync { + /// Send `request` once and return the collector's answer, whatever its status. + async fn send(&self, request: ExportRequest) -> Result; +} + +#[cfg(feature = "net")] +mod net { + use std::sync::Arc; + + use livekit_net::{Header, HttpClient, HttpClientExt, HttpResponse, TransportError}; + + use super::{ExportError, ExportRequest, ExportResponse, TelemetryTransport}; + + /// Default transport: HTTP POST through a `livekit-net` client — the native backend, or + /// whatever the host registered with `livekit_net::set_http_client`. + pub struct NetTransport(Arc); + + impl NetTransport { + /// Post through `client`. + pub(crate) fn new(client: Arc) -> Self { + Self(client) + } + + /// Resolve the process-wide `livekit-net` client; `None` when none is available. + pub fn from_registry() -> Option { + livekit_net::http_client().map(Self::new) + } + } + + impl From for ExportResponse { + fn from(response: HttpResponse) -> Self { + Self { + status: response.status, + headers: response.headers.into_iter().map(|h| (h.name, h.value)).collect(), + body: response.body, + } + } + } + + #[async_trait::async_trait] + impl TelemetryTransport for NetTransport { + async fn send(&self, request: ExportRequest) -> Result { + let headers = + request.headers.into_iter().map(|(name, value)| Header { name, value }).collect(); + match self.0.post(request.url, headers, request.body).await { + Ok(response) => Ok(response.into()), + // The client swallowed the body with the status; the core still classifies it. + Err(TransportError::Http { status }) => { + Ok(ExportResponse { status, ..Default::default() }) + } + Err(other) => { + Err(ExportError::Retryable { reason: other.to_string(), retry_after_ms: None }) + } + } + } + } +} + +#[cfg(feature = "net")] +pub use net::NetTransport; + +#[cfg(test)] +mod tests { + use prost::Message; + + use super::*; + + fn status_body(message: &str, retry: Option<(i64, i32)>) -> Vec { + let details = retry + .map(|(seconds, nanos)| prost_types::Any { + type_url: "type.googleapis.com/google.rpc.RetryInfo".into(), + value: RetryInfo { retry_delay: Some(prost_types::Duration { seconds, nanos }) } + .encode_to_vec(), + }) + .into_iter() + .collect(); + RpcStatus { code: 8, message: message.into(), details }.encode_to_vec() + } + + fn response(status: u16, headers: &[(&str, &str)], body: Vec) -> ExportResponse { + ExportResponse { + status, + headers: headers.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(), + body, + } + } + + fn verdict(status: u16, headers: &[(&str, &str)], body: Vec) -> Verdict { + Verdict::of(&response(status, headers, body)) + } + + #[test] + fn success_and_partial_success() { + assert_eq!( + Verdict::of(&ExportResponse::accepted()), + Verdict::Accepted { rejected: 0, reason: String::new() } + ); + assert!(matches!(verdict(204, &[], vec![]), Verdict::Accepted { rejected: 0, .. })); + let partial = PartialSuccessResponse { + partial_success: Some(PartialSuccess { rejected: 3, error_message: "too old".into() }), + } + .encode_to_vec(); + assert_eq!( + verdict(200, &[], partial), + Verdict::Accepted { rejected: 3, reason: "too old".into() } + ); + assert!(matches!(verdict(200, &[], b"not protobuf".to_vec()), Verdict::Accepted { .. })); + } + + #[test] + fn client_errors() { + assert_eq!( + verdict(400, &[], status_body("bad batch", None)), + Verdict::Rejected("HTTP 400: bad batch".into()) + ); + assert_eq!(verdict(413, &[], vec![]), Verdict::TooLarge); + assert!(matches!(verdict(307, &[], vec![]), Verdict::Rejected(_)), "never follow"); + assert!(matches!(verdict(422, &[], vec![]), Verdict::Rejected(_))); + assert_eq!(verdict(404, &[], vec![]), Verdict::NotFound); + for body in ["invalid token", "operation requires observability write grant"] { + assert_eq!( + verdict(401, &[], body.as_bytes().to_vec()), + Verdict::Unauthorized(format!("HTTP 401: {body}")) + ); + } + assert!(matches!(verdict(403, &[], vec![]), Verdict::Unauthorized(_))); + } + + #[test] + fn disabled_by_owner() { + let text = b"project data recording is disabled by owner".to_vec(); + assert_eq!(verdict(401, &[], text), Verdict::Disabled); + let rpc = status_body("Project data recording is disabled by owner", None); + assert_eq!(verdict(403, &[], rpc), Verdict::Disabled); + // The phrase counts wherever it sits; only the logged reason is cut short. + let long = format!("{} data recording is disabled by owner", "x".repeat(250)); + assert_eq!(verdict(401, &[], long.into_bytes()), Verdict::Disabled); + let Verdict::Unauthorized(reason) = verdict(401, &[], "y".repeat(500).into_bytes()) else { + panic!("a credential answer") + }; + assert_eq!(reason.len(), "HTTP 401: ".len() + MAX_REASON_CHARS); + // Only the documented phrase: another answer that merely mentions "disabled" is a + // credential problem, not a project switch. + for body in ["account disabled", "token disabled for this room"] { + assert!(matches!( + verdict(401, &[], body.as_bytes().to_vec()), + Verdict::Unauthorized(_) + )); + } + } + + /// The status decides; a delay hint on a final status is ignored. Only 429, 502–504 and Cloud's + /// 500-with-RetryInfo are retried. + #[test] + fn delay_hints_never_make_a_final_status_retryable() { + type Hint<'a> = (&'a [(&'a str, &'a str)], Vec); + let hints: [Hint; 3] = [ + (&[("Retry-After", "5")], vec![]), + (&[], status_body("later", Some((5, 0)))), + (&[("Retry-After", "5")], status_body("later", Some((5, 0)))), + ]; + for (headers, body) in &hints { + for status in [300, 302, 307, 400, 405, 409, 410, 422, 501, 505, 599] { + assert!( + matches!(verdict(status, headers, body.clone()), Verdict::Rejected(_)), + "{status} with {headers:?}" + ); + } + assert_eq!(verdict(413, headers, body.clone()), Verdict::TooLarge); + assert_eq!(verdict(404, headers, body.clone()), Verdict::NotFound); + assert!(matches!(verdict(429, headers, body.clone()), Verdict::Throttled { .. })); + assert!(matches!(verdict(503, headers, body.clone()), Verdict::Throttled { .. })); + for status in [502, 504] { + assert!(matches!( + verdict(status, headers, body.clone()), + Verdict::Retry { delay_ms: Some(5_000), .. } + )); + } + } + // 500: only with RetryInfo in the body, never on a Retry-After header alone. + assert!(matches!(verdict(500, &[("Retry-After", "5")], vec![]), Verdict::Rejected(_))); + assert!(matches!( + verdict(500, &[], status_body("later", Some((5, 0)))), + Verdict::Retry { delay_ms: Some(5_000), .. } + )); + } + + /// Untrusted time values never panic or wrap; oversized ones saturate. + #[test] + fn untrusted_time_values_are_bounded() { + assert_eq!(retry_after_ms("99999999999999999999999"), Some(u64::MAX)); + assert_eq!(retry_after_ms("18446744073709551615"), Some(u64::MAX)); + assert_eq!(retry_after_ms("-5"), None); + assert_eq!(retry_after_ms(""), None); + for date in [ + "Sun, 06 Nov 99999999999999 08:49:37 GMT", + "Sun, 06 Nov -9223372036854775808 08:49:37 GMT", + "Sun, 06 Nov 1969 08:49:37 GMT", + "Sun, 06 Nov 10000 08:49:37 GMT", + "Sun, 32 Nov 2026 08:49:37 GMT", + "Sun, 06 Nov 2026 08:49:37:99 GMT", + "Sun, 06 Nov 2026 99:49:37 GMT", + ] { + assert_eq!(imf_fixdate_secs(date), None, "{date}"); + } + assert_eq!(imf_fixdate_secs("Fri, 31 Dec 9999 23:59:59 GMT"), Some(253_402_300_799)); + let huge = verdict(429, &[("Retry-After", "99999999999999999999")], vec![]); + assert!(matches!(huge, Verdict::Throttled { delay_ms: u64::MAX, .. }), "clamped later"); + let far = verdict(429, &[], status_body("q", Some((i64::MAX, 999_999_999)))); + assert!(matches!(far, Verdict::Throttled { delay_ms: u64::MAX, .. })); + } + + #[test] + fn throttling_takes_the_header_then_the_body_then_the_default() { + let both = verdict(429, &[("Retry-After", "7")], status_body("quota", Some((30, 0)))); + assert!(matches!(both, Verdict::Throttled { delay_ms: 7_000, .. })); + let body = verdict(429, &[], status_body("quota", Some((30, 500_000_000)))); + assert!(matches!(body, Verdict::Throttled { delay_ms: 30_500, .. })); + // What LiveKit Cloud answers over quota: ResourceExhausted, no header, no RetryInfo. + assert_eq!( + verdict(429, &[], status_body("QuotaStatusExceeded", None)), + Verdict::Throttled { delay_ms: 60_000, reason: "HTTP 429: QuotaStatusExceeded".into() } + ); + let unavailable = verdict(503, &[("retry-after", " 12 ")], vec![]); + assert!(matches!(unavailable, Verdict::Throttled { delay_ms: 12_000, .. })); + let past = verdict(429, &[("Retry-After", "Wed, 21 Oct 2015 07:28:00 GMT")], vec![]); + assert!(matches!(past, Verdict::Throttled { delay_ms: 0, .. }), "a past date: now"); + let garbage = verdict(429, &[("Retry-After", "soon-ish")], vec![]); + assert!(matches!(garbage, Verdict::Throttled { delay_ms: 60_000, .. }), "ignored"); + let negative = verdict(429, &[], status_body("q", Some((-5, -1)))); + assert!(matches!(negative, Verdict::Throttled { delay_ms: 0, .. }), "clamped"); + } + + #[test] + fn impossible_calendar_dates_are_ignored() { + assert_eq!(imf_fixdate_secs("Tue, 30 Feb 2027 12:00:00 GMT"), None); + assert_eq!(imf_fixdate_secs("Mon, 29 Feb 2027 12:00:00 GMT"), None, "not a leap year"); + assert_eq!(imf_fixdate_secs("Thu, 31 Apr 2027 12:00:00 GMT"), None); + assert_eq!(imf_fixdate_secs("Tue, 29 Feb 2100 12:00:00 GMT"), None, "a century"); + assert!(imf_fixdate_secs("Tue, 29 Feb 2000 12:00:00 GMT").is_some(), "every 400 years"); + let ignored = verdict(429, &[("Retry-After", "Tue, 30 Feb 2027 12:00:00 GMT")], vec![]); + assert!(matches!(ignored, Verdict::Throttled { delay_ms: 60_000, .. }), "the default"); + } + + #[test] + fn retry_after_http_dates() { + assert_eq!(imf_fixdate_secs("Sun, 06 Nov 1994 08:49:37 GMT"), Some(784_111_777)); + assert_eq!(imf_fixdate_secs("Thu, 01 Jan 1970 00:00:00 GMT"), Some(0)); + assert_eq!(imf_fixdate_secs("Tue, 29 Feb 2028 12:00:00 GMT"), Some(1_835_438_400)); + assert_eq!(imf_fixdate_secs("Sunday, 06-Nov-94 08:49:37 GMT"), None, "obsolete form"); + assert_eq!(imf_fixdate_secs("Sun, 06 Nov 1994 08:49:37 PST"), None); + // A far-future date: the delay is the distance from now, whatever now is. + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock") + .as_secs(); + let expected = (253_402_300_799 - now) * 1000; + let ms = retry_after_ms("Fri, 31 Dec 9999 23:59:59 GMT").expect("date"); + assert!(expected.abs_diff(ms) <= 2_000, "{ms} vs {expected}"); + } + + #[test] + fn server_errors() { + for status in [502, 503, 504] { + assert!(matches!(verdict(status, &[], vec![]), Verdict::Retry { delay_ms: None, .. })); + } + assert!(matches!( + verdict(502, &[("Retry-After", "3")], vec![]), + Verdict::Retry { delay_ms: Some(3_000), .. } + )); + // Cloud's retryable 500 carries RetryInfo; a bare 500 is final, per OTLP/HTTP. + let internal = verdict(500, &[], status_body("try later", Some((5, 0)))); + assert!(matches!(internal, Verdict::Retry { delay_ms: Some(5_000), .. })); + assert!(matches!(verdict(500, &[], vec![]), Verdict::Rejected(_))); + assert!(matches!(verdict(501, &[], vec![]), Verdict::Rejected(_))); + } + + /// An exception the foreign transport did not declare becomes a retryable transport error. + /// Without the conversion UniFFI panics on the exporter's thread. + #[cfg(feature = "uniffi")] + #[test] + fn an_undeclared_foreign_exception_is_a_retryable_failure() { + let lifted = as uniffi::LiftReturn< + crate::UniFfiTag, + >>::handle_callback_unexpected_error( + uniffi::UnexpectedUniFFICallbackError::new("java.lang.IllegalStateException: boom"), + ); + assert!(matches!( + lifted, + Err(ExportError::Retryable { retry_after_ms: None, ref reason }) if reason.contains("boom") + )); + } +}