feat(telemetry): add the pipeline, sessions, spans and exporter by pblazej · Pull Request #1483 · livekit/rust-sdks

pblazej · 2026-10-01T10:48:03Z

Summary

The pipeline: the synchronous handle, one scope per Room, typed spans, and the exporter actor. The README becomes the crate docs (its usage example is a doc test) and SPEC.md is complete.

Changes

Telemetry: the synchronous handle — flood guard, flush / shutdown / purge, stats
Scope: one Room — server URL and token, attributes, RTC stats, the lk.subscribe lifecycle
Typed Spans
Exporter: the actor that encodes, caches and uploads under the upload policy
README as crate docs; SPEC.md complete
Ending the last open lk.connect or lk.reconnect wakes the exporter, so a batch held while connecting goes out at once
The span registry holds sessions weakly, so a Room dropped without disconnected is freed and no longer counts as in a call
Plain log records have a flood guard of their own, and disconnected reports once per connection
Checkpoints and attributes a span drops past its caps are counted in the exported span
A session waits on at most 256 tracks of each kind, and the exporter's clock retires a published track that never sends
The config is normalized once, with a 1 s floor for the periods and the timeout; Telemetry::shutdown_bound names the longest wait of shutdown
A second shutdown returns only once the exporter stopped, and a request starts only under the opt-out's admission barrier
The first flush no longer subtracts from the monotonic clock, which panicked on Windows right after boot
A purge reports the result of its last attempt, so a retry that deletes everything reports success
A soft hold that ends while the cache is empty no longer silences the escape warning of the next hold

Why this stage is larger

The largest stage, at about 4.7k lines: Telemetry ⇄ Shared ⇄ Exporter and Scope ⇄ Span ⇄ Telemetry are one dependency cycle, so they land together. Their tests follow in #1484, together with the backend contract.

Temporary

#[expect(dead_code)] on Telemetry::weak_commands, which serves global (#1485).

Verification

At 58f3a6e5, from a clean checkout (CI's test workflow runs only for PRs into main, so these were run locally; there is no clippy job in CI):

cargo fmt --check
cargo clippy -p livekit-telemetry --all-targets -- -D warnings, also with --all-features and as a cargo hack --feature-powerset --depth 2 run
cargo test -p livekit-telemetry: 146 unit, 2 doc
RUSTDOCFLAGS="-D warnings" cargo doc -p livekit-telemetry --no-deps

devin-ai-integration

Note

Newer findings are available below. Devin Review posted a newer report on this PR, in addition to the findings presented here.

Devin Review found 6 potential issues.

2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)

devin-ai-integration · 2026-10-01T14:37:19Z

+        if self.offline() {
+            self.log_hold(Some("offline"), backlog);
+            return (0, false);
+        }


🟡 Offline time exhausts the upload hold

When the device goes offline during a soft hold, budget retains held_since throughout the hard hold. Reconnection immediately releases a batch even if the constrained network or low battery still requires quiet.

Learn more

A soft hold uses held_since as its 60-second timer. Going offline is a hard hold and stops requests indefinitely, but the early offline return leaves that timer running. On reconnection, the next pass sees a cap that has already elapsed and sends a batch despite the continuing soft-hold condition.

Example: Low Data Mode starts holding at 12:00:00, the phone goes offline at 12:00:05 and reconnects at 12:02:00. The first pass sends a batch immediately rather than waiting under the still-active Low Data Mode hold.

Recommended fix: Suspend or reset the soft-hold timer when the hard offline hold begins, and resume a fresh or remaining timer when it lifts. Check the same behavior for destination/token hard holds so waiting for credentials does not consume the soft-hold allowance.

Was this helpful? React with 👍 or 👎 to provide feedback.

Offline time counts toward the 60 s soft-hold cap on purpose, because the cap bounds the policy when its signals are wrong. SPEC.md states this, and nothing is uploaded while the device is offline.

devin-ai-integration · 2026-10-01T14:37:24Z

TelemetryEvent::custom

+        let valid = !name.is_empty()
+            && name.len() <= crate::event::MAX_NAME_BYTES
+            && attributes.len() <= crate::event::MAX_CUSTOM_ATTRIBUTES
+            && attributes.iter().all(|a| crate::event::valid_custom(&a.key, Some(&a.value)));


🟡 Custom event prefix exceeds name limit

A 128-byte name passes emit_custom, but TelemetryEvent::custom prepends custom. afterward. The exported name exceeds the 128-byte limit instead of being rejected and counted.

Was this helpful? React with 👍 or 👎 to provide feedback.

The 128-byte limit applies to the name the app passes, and the custom. prefix that the core adds is not part of that budget.

…#1479) ## Summary Adds `livekit-telemetry`, the shared client-telemetry core, starting with its data model and the OTLP encoding. ## Changes - Event and log records with attributes, and the bounded in-memory queue - Health counters and the `lk.telemetry.report` event - The span store - Session identity: `ScopeState` (trace id, SDK and app attributes, route) - OTLP/HTTP protobuf encoding on `opentelemetry-proto` types - `README.md`, `SPEC.md` (resource attributes, events, log records, spans; the contract for the whole crate) and the crate changeset — the only PR the changeset check sees ## Temporary Until #1483: a crate-level `#![allow(dead_code)]` for internals the pipeline consumes, and `ScopeState` without subscribe tracking (#1483 completes `scope.rs`). <details><summary><b>Architecture</b></summary> ```text platform ──emit / span / getStats report / device state──▶ Telemetry ─▶ Store ─▶ Exporter ─▶ BatchCache ─▶ TelemetryTransport (instruments, one Scope per Room) (sync, never blocks) (actor: passes, (write-ahead: (host HTTP, livekit-net, one request out) memory | disk) or Dart's pull queue) ``` - **One pipeline per process, one scope per Room.** A scope is a trace id plus the Room's attributes; a span is one attempt; checkpoints are span events; `lk.outcome` carries `ok | error | cancelled`. - **The platform keeps only what it alone can do:** OS signals and moving bytes. The core owns destination and credentials, batching, OTLP encoding, retry, holds, persistence, loss accounting, the subscribe state machine, stats mapping (one raw `getStats()` report per peer connection) and polling cadence. - **Conservative by default:** one export and one RTC window per minute, stretched up to 4× under device pressure; while a Room is in a call, at most 4 requests per interval (a backlog waits its turn; with no call nothing is metered) — wake-ups (subscribes, device changes, tokens) never add exports; entering the background uploads everything at once. - **The core reports on itself:** its own Rust warnings and errors reach the pipeline through the existing log forwarder once a platform installs it, any compact JWT (also inside punctuation) and bearer/token credential values (quoted or spaced) masked; the console output is unchanged. - **Transport stays flexible:** the `TelemetryTransport` foreign trait (Swift/Kotlin), a bounded pull queue (Dart, whose callbacks are isolate-bound: polled with `try_next()` from a timer, no instruments passed, so Rust never calls into Dart), or `NetTransport` over the unmodified `livekit-net` registry. </details> <details><summary><b>Verification</b></summary> At `ec9508a4`, from a clean checkout (CI's test workflow runs only for PRs into `main`, so these were run locally; there is no clippy job in CI): - `cargo fmt -- --check` - `cargo clippy -p livekit-telemetry --all-targets --all-features -- -D warnings` - `cargo check -p livekit-telemetry --all-targets --no-default-features` with features `[]`, `[uniffi]` - `cargo test -p livekit-telemetry`: 23 unit; `--all-features`: 23 unit - `cargo publish --dry-run -p livekit-telemetry` `cargo doc -D warnings` reports one unresolved link, to `Telemetry::stats`, which #1483 adds. </details>

`Telemetry` is the synchronous handle (emit, log, device state, flush, shutdown, purge, stats) behind a flood guard; `Scope` is one Room on it, with its own trace id, server URL and token, attributes, RTC stats and the `lk.subscribe` lifecycle; `Span` is one typed attempt (`SpanName`, `SpanStep`). The `Exporter` actor drains the queue and finished spans, encodes and gzips one batch per project into the cache before any network is involved, then uploads oldest first: it acts on every collector answer, pauses a destination with jittered backoff or for the delay the server asked for, splits a 413, holds uploads while a Room connects or the device asks for quiet, and meters requests while a Room is in a call.

…n drops Ending the last open `lk.connect` or `lk.reconnect` now wakes the exporter, so a batch held while connecting goes out at once instead of at the next tick. The span registry keeps weak session references, so a Room dropped without `disconnected` is freed once its spans are exported and no longer counts as in a call. Plain log records get a flood guard of their own, so a burst of events never silences warnings. `disconnected` reports once per connection. Checkpoints and attributes a span drops past its caps are now counted in the exported span, as the registry does for the rest. The first flush no longer subtracts the period from the monotonic clock, which panics on Windows right after boot. Server changes to one session run one at a time, and a session remembers a server only once its destination and route are in place. A session waits on at most 256 tracks of each kind (a track past that is counted as `queue_full`, and each overflow is warned about once; it ends when a new track is admitted), keeps no track sid longer than an attribute value, and the exporter's clock retires a published track that never sends. The opt-out forgets published tracks and checks consent under the lock it clears them under. A second shutdown returns only once the exporter stopped, and a shutdown drain obeys a device hold that begins meanwhile. The config is normalized once, with a 1 s floor for the periods and the timeout, instead of scattered clamps. App-defined spans refuse the SDK's keys, and the pipeline-wide attributes have the custom limits. A soft hold warns on its first escape only, and a hold that lifts while nothing is cached ends its episode too, so the next hold warns again. The flood guard logs after its lock. Protocol reasons convert with `From<i32>`, a transport's `retry_after_ms` pauses the destination as a server's `Retry-After` does, and the routing calls pass the session, whose trace id is formatted once. Test-only hooks and `Telemetry` methods with no production caller (`start`, `emit_custom`, `trace_id`, `record_stats`) are gone, the public exports are listed, internal types have clearer names, and the docs, defaults and SPEC state the real behavior. A changeset records the stage. A test checks that a batch for a self-hosted server with an IPv6 address and a port is cached under a file name every platform accepts. `Telemetry::shutdown_bound` names the longest wait of `shutdown`, the export timeout after its floor plus a second, so a caller that waits for the exporter uses the same bound. Device change events are stamped under the device lock and emitted after it, and the span registry's warnings, attributes a span's handle dropped included, reach the logger after its lock, and after the subscribe lock when a subscribe opens the span. A request starts only under the opt-out's admission barrier; the exporter stops on a cancel signal whether it runs or drains, `Shared::purge` waits until it is gone and reports what its last attempt left, and purges racing each other count once. The gzip ratio is documented as a measured typical result. The exporter counts the pipeline-wide attributes in every batch budget, so batches go out full instead of cut too large and halved, and a record that names a span the pipeline cannot place goes to the process session without that span id.

pblazej · 2026-11-04T12:36:15Z

+                Some((batch.host.map(str::to_owned), batch.owner?.to_owned()))
+            })
+            .collect();
+        self.destinations().retain_owners(&live, &backlog);


Reminder for #1483: include the credential-retirement fix already published in #1396 (426902ea) together with its deterministic regression (4fe88a53). List the cache backlog before taking the scopes registry lock, then keep that guard through retain_owners (scopes before destinations), so a concurrently registering Room cannot lose its token.

Still present at 58f3a6e: retire_credentials() releases the scopes guard after building live, then lists the backlog and calls retain_owners(). A Room registered between those steps can have its new credential removed. Please include the existing fix and deterministic regression linked above; no separate thread needed.

xianshijing-lk

Reviewed at 58f3a6e. The inline comments cover blocking storage, report atomicity, logging re-entry, split bookkeeping and smaller accounting/lifecycle issues.

Validation: 147 unit tests and two doc tests pass with all features; generated telemetry Kotlin bindings compile; formatting passes. Additional deterministic probes reproduced the reported correctness issues, and 60 repeated lifecycle cycles released shared state across shutdown, purge and partial initialization. This was a telemetry binding compilation check, not the full Android packaging job.

Related documentation points:

Telemetry::new() performs synchronous cache recovery/pruning, and stats() synchronously enumerates the cache. Document their disk I/O and caller-thread requirements.
State that transport futures must return promptly from each poll. The admission barrier spans the first poll, so synchronous transport work can delay revocation and cannot be preempted by the upload timeout.
purge() can wait export_timeout + 1 s if the returned exporter is still held but never spawned. If it is dropped, the watch channel closes promptly. Please document this distinction as shutdown() does.

The existing credential-retirement thread also remains applicable; I will follow up there rather than duplicate it.

xianshijing-lk · 2026-11-06T00:52:00Z

+    pub async fn purge(&self) -> bool {
+        self.revoke();
+        // A failure here is logged; the last attempt below tries every file again.
+        self.clear();


[P2] Move cache I/O off the async executor

clear() synchronously lists, deletes and syncs cache files before purge() first yields. The exporter likewise calls synchronous push, read, remove and pending from its async loop. This blocks a worker, or the whole executor on a single-thread runtime; an async timeout cannot preempt it. With an injected cache taking 100 ms per operation, timeout(20 ms, purge()) returned success after about 620 ms.

Please run storage operations on a blocking worker and signal cancellation before starting deletion. Signaling alone does not wait for the exporter to stop; retain the synchronization/final sweep needed to handle writes already in progress. Repeated directory listings within a pass can also be consolidated, updating the listing after mutations.

xianshijing-lk · 2026-11-06T00:52:00Z

+        for sample in
+            crate::rtc::samples_from_report(track_sid, kind, direction, &report, timestamp_ns)
+        {
+            self.record_stats(sample);


[P2] Admit all layers of one report atomically

Each record_stats() call acquires and releases the windows lock separately. An exporter close or correlation-attribute split between layers ends a partial poll; the next partial poll replaces the previous stream map and loses counter growth.

A deterministic probe with two layers growing 100 → 200 → 300 bytes exported a cumulative count of 200 instead of 400 when a close landed between layers. Please ingest the complete report under one windows lock, keeping callbacks and diagnostics outside it. record_peer_stats() needs the same treatment.

xianshijing-lk · 2026-11-06T00:52:00Z

+                return;
+            }
+            self.shared.before_commit();
+            let host = self.shared.destinations.lock().unwrap_or_else(|e| e.into_inner()).set(


[P2] Defer destination diagnostics until the locks are released

Destinations::set() emits log messages while this destinations mutex and the session transition mutex are held. refuse() and kill() also log under the destinations mutex. A host logger that synchronously re-enters emit() or another destination operation can deadlock. I confirmed that the destination warning runs while the mutex is held.

Please use diagnostic()/logged_after() as for the other registries, releasing both locks before logging here. Correctly tagged telemetry FFI logs are filtered before re-entry, so ordinary forwarding of those logs is safe; other host callbacks still expose the lock.

xianshijing-lk · 2026-11-06T00:52:00Z

+        let evicted = self.shared.cache.replace(id, &new)?;
+        self.count_evicted(&evicted);
+        self.born.remove(id);
+        let now = Instant::now();


[P2] Preserve the parent's age when splitting

Both halves receive Instant::now() rather than the parent's birth time. Since expiry requires both wall-clock and monotonic age to exceed 24 hours, splitting a 23-hour-old batch gives its records another 24 hours. A paused-clock probe confirmed that the halves remain unexpired at 25 hours of record age.

Please inherit the parent's optional birth timestamp, preserving its absence for batches from an earlier launch. Splitting should change batch size, not retention age.

xianshijing-lk · 2026-11-06T00:52:00Z

+            self.undeletable.insert(id.to_owned());
+        }
+        self.born.remove(id);
+        if let Some(counter) = lost {


[P2] Count a failed-to-delete batch as lost only once

When deletion fails, forget() adds the ID to undeletable and increments the loss counter. Although upload advancement skips these IDs, the expiry scan and purge_dead() still call forget() again on later passes.

With deletion forced to return PermissionDenied, one disabled record was counted four times across the initial purge and three additional passes. Expired batches have the same problem. Please count the initial discard once and make subsequent deletion attempts accounting-neutral.

xianshijing-lk · 2026-11-06T00:52:00Z

+        self.born.remove(id);
+        let now = Instant::now();
+        for (half, _) in &new {
+            self.born.insert(half.clone(), now);


[P2] Do not recreate birth metadata for evicted halves

replace() can immediately evict a half to satisfy the cache bound. count_evicted() removes its birth entry, but this loop then inserts it again even though the batch no longer exists. If the pass ends before trying that missing half, future cache listings never revisit it.

Under cache pressure, a probe performing 100 splits left 101 birth entries for one cached batch. Please install the halves' metadata before processing evictions, or keep metadata only for surviving halves.

xianshijing-lk · 2026-11-06T00:52:00Z

        while out.len() < max {
            let Some(next) = queue.events.front() else { break };
-            let size = next.event.size_hint();
+            let size = next.event.size_hint() + extra;


[P3] Subtract only the queued record's size

push() adds event.size_hint() to queue.bytes, but this subtracts size_hint() + extra. Pipeline attributes belong in the batch budget, not the queue's own accounting. A partial-drain probe left 5,600 bytes queued but accounted for only 800.

enqueue() normally drains to empty, which resets the accounting through saturation, so this is not persistent drift across normal passes. Please keep extra in the local batch budget and subtract only size_hint() from queue.bytes.

xianshijing-lk · 2026-11-06T00:52:00Z

+/// A cached batch is intact when it is one complete gzip member (RFC 1952) whose CRC checks
+/// out: a truncated or bit-flipped file is caught here, never sent.
+fn intact(body: &[u8]) -> bool {
+    body.starts_with(&[0x1f, 0x8b]) && gunzip(body).is_some()


[P3] Validate cached gzip bodies without retaining the decoded payload

Every replay calls intact(), which expands the entire body into a Vec just to check the CRC. There is no decoded-size cap: a probe expanded 76,531 compressed bytes into 16 MiB. The compressed cache bound does not bound this allocation, including for batches written under an earlier, larger max_batch_bytes setting.

Please stream CRC validation through a fixed-size buffer and enforce a decoded-size bound when materializing a body for splitting. This is defensive handling of persisted cache data, rather than a failure of normal default-sized batches.

xianshijing-lk · 2026-11-06T00:52:00Z

+        for pause in self.pauses.values_mut() {
+            pause.until = pause.until.filter(|t| now < *t);
+            pause.server_until = pause.server_until.filter(|t| now < *t);
+        }


[P3] Retire pause state when its destination disappears

Expired deadlines are cleared, but the map entry and failure history remain until recovered() sees a successful upload. A destination that fails and then loses all owners and cached batches never recovers, so its entry stays for the exporter's lifetime.

Please retire entries once their destination has no live owners or cached batches. Keep failure history for destinations still in use; removing every expired pause would reset exponential backoff.

xianshijing-lk · 2026-11-06T00:52:00Z

+        }
+        self.telemetry.retire_stats(&self.state, None);
+        // Out of the call: uploads no longer yield to it.
+        self.state.server.lock().unwrap_or_else(|e| e.into_inner()).take();


[P3] Serialize disconnect with server updates

set_server() holds transition, but disconnected() clears server without it. If disconnect lands between the server check and remember_server(), the in-flight update restores server, leaving in_call() true after disconnect and keeping uploads metered. I reproduced this interleaving with the existing test hook.

Please coordinate disconnect with the session transition and make in_call() account for disconnected state. Keep host logging outside the transition lock.

pblazej added the internal to tag changes that don't require changelog documentation label Oct 1, 2026

pblazej force-pushed the blaze/telemetry-stack/5-pipeline branch from 463a051 to c16912c Compare October 1, 2026 13:53

pblazej mentioned this pull request Oct 1, 2026

feat(uniffi): expose the telemetry core #1396

Open

pblazej added this pull request to stack #1486 October 1, 2026 14:23

pblazej marked this pull request as ready for review October 1, 2026 14:33

pblazej requested a review from ladvoc as a code owner October 1, 2026 14:33

pblazej requested review from 1egoman, davidliu, hiroshihorie, lukasIO and xianshijing-lk October 1, 2026 14:33

devin-ai-integration Bot reviewed Oct 1, 2026

View reviewed changes

pblazej force-pushed the blaze/telemetry-stack/5-pipeline branch from c16912c to c79bf94 Compare October 23, 2026 06:09

This comment was marked as resolved.

Sign in to view

pblazej added 3 commits October 23, 2026 11:02

docs(telemetry): specify the upload policy, app API and typed surface

b05b14d

pblazej force-pushed the blaze/telemetry-stack/5-pipeline branch from c79bf94 to 58f3a6e Compare October 23, 2026 09:56

pblazej commented Nov 4, 2026

View reviewed changes

xianshijing-lk reviewed Nov 6, 2026

View reviewed changes

Conversation

pblazej commented Oct 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

Why this stage is larger

Temporary

Uh oh!

devin-ai-integration Bot left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

devin-ai-integration Bot Oct 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

devin-ai-integration Bot Oct 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

This comment was marked as resolved.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xianshijing-lk left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

pblazej commented Oct 1, 2026 •

edited

Loading

devin-ai-integration Bot left a comment •

edited

Loading

devin-ai-integration Bot Oct 1, 2026 •

edited

Loading

devin-ai-integration Bot Oct 1, 2026 •

edited

Loading