feat(telemetry): add client telemetry through the shared Rust core

.changes/client-telemetry

@@ -0,0 +1 @@
+minor type="added" "Client telemetry through the shared Rust core: each Room reports its connect, reconnect, publish and subscribe spans, RTC statistics, SDK warnings and errors and device state to its LiveKit Cloud project when the token carries the observability grant; apps can add Room.emitTelemetryEvent(_:attributes:) and Room.setTelemetryAttribute(_:value:), and opt out with LiveKitSDK.disableTelemetry()"

.github/workflows/ci.yaml

@@ -111,6 +111,16 @@ jobs:
             # Data track schema definitions are stored as participant data blobs.
             enable_participant_data_blob: true
 
+      # The telemetry end-to-end test posts OTLP to 127.0.0.1:4319 and reads back what the
+      # collector wrote (Tests/LiveKitCoreTests/Telemetry/otelcol.yaml). A plain binary, like the
+      # dev server above: macOS runners have no container runtime.
+      - name: Run OpenTelemetry Collector
+        run: |
+          arch="$(uname -m | sed 's/x86_64/amd64/')"
+          curl -sSfL -o otelcol.tar.gz "https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.160.0/otelcol-contrib_0.160.0_darwin_${arch}.tar.gz"
+          tar -xzf otelcol.tar.gz otelcol-contrib
+          nohup ./otelcol-contrib --config Tests/LiveKitCoreTests/Telemetry/otelcol.yaml > "$RUNNER_TEMP/otelcol.log" 2>&1 &
+
       - uses: maxim-lobanov/setup-xcode@v1
         with:
           xcode-version: ${{ matrix.xcode }}

Benchmarks/LiveKitBenchmark/BenchmarkSupport/TokenGenerator.swift

@@ -56,6 +56,7 @@ struct TokenGenerator {
             ingressAdmin: false,
             hidden: false,
             recorder: false,
+            agent: false,
         )
 
         let options = TokenOptions(

Benchmarks/Package.swift

@@ -2,14 +2,21 @@
 
 import PackageDescription
 
+// Same switch as the SDK's manifests: `LIVEKIT_UNIFFI_PATH` points at a local livekit-uniffi build.
+let uniFFI: Package.Dependency = if let path = Context.environment["LIVEKIT_UNIFFI_PATH"] {
+    .package(name: "livekit-uniffi-xcframework", path: path)
+} else {
+    .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", from: "0.0.1")
+}
+
 let package = Package(
     name: "LiveKitBenchmark",
     platforms: [
         .macOS(.v26),
     ],
     dependencies: [
         .package(name: "client-sdk-swift", path: "../"),
-        .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", from: "0.0.1"),
+        uniFFI,
         // No Jemalloc trait: jemalloc's malloc-zone hooks crash on macOS 26 (malloc metrics read 0 without it)
         .package(url: "https://github.com/ordo-one/benchmark.git", from: "1.29.0", traits: []),
     ],

Package.swift

@@ -3,6 +3,15 @@
 
 import PackageDescription
 
+// Development wiring for unreleased Rust bindings: set `LIVEKIT_UNIFFI_PATH` to a local
+// livekit-uniffi Swift package (`cargo make swift-package-debug` in rust-sdks/livekit-uniffi builds
+// it into `packages/swift/LiveKitUniFFI`). Unset, the released package is used.
+let uniFFI: Package.Dependency = if let path = Context.environment["LIVEKIT_UNIFFI_PATH"] {
+    .package(name: "livekit-uniffi-xcframework", path: path)
+} else {
+    .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", exact: "0.1.9")
+}
+
 let package = Package(
     name: "LiveKit",
     platforms: [
@@ -20,7 +29,7 @@ let package = Package(
     dependencies: [
         // LK-Prefixed Dynamic WebRTC XCFramework
         .package(url: "https://github.com/livekit/webrtc-xcframework.git", exact: "150.7871.02"),
-        .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", exact: "0.1.9"),
+        uniFFI,
         // Test-only: conformance oracle for the nanopb facades.
         .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.31.0"),
         // Only used for DocC generation

Package@swift-6.2.swift

@@ -3,6 +3,15 @@
 
 import PackageDescription
 
+// Development wiring for unreleased Rust bindings: set `LIVEKIT_UNIFFI_PATH` to a local
+// livekit-uniffi Swift package (`cargo make swift-package-debug` in rust-sdks/livekit-uniffi builds
+// it into `packages/swift/LiveKitUniFFI`). Unset, the released package is used.
+let uniFFI: Package.Dependency = if let path = Context.environment["LIVEKIT_UNIFFI_PATH"] {
+    .package(name: "livekit-uniffi-xcframework", path: path)
+} else {
+    .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", exact: "0.1.9")
+}
+
 let package = Package(
     name: "LiveKit",
     platforms: [
@@ -21,7 +30,7 @@ let package = Package(
     dependencies: [
         // LK-Prefixed Dynamic WebRTC XCFramework
         .package(url: "https://github.com/livekit/webrtc-xcframework.git", exact: "150.7871.02"),
-        .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", exact: "0.1.9"),
+        uniFFI,
         // Test-only: conformance oracle for the nanopb facades.
         .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.31.0"),
         // Only used for DocC generation

Sources/LiveKit/Core/Room+Engine.swift

@@ -16,6 +16,7 @@
 
 // swiftlint:disable file_length
 
+internal import LiveKitUniFFI
 import Foundation
 
 #if canImport(Network)
@@ -355,7 +356,9 @@
     }
 
     // swiftlint:disable:next cyclomatic_complexity function_body_length
-    func startReconnect(reason: StartReconnectReason, nextReconnectMode: ReconnectMode? = nil) async throws {
+    func startReconnect(reason: StartReconnectReason, nextReconnectMode: ReconnectMode? = nil,
+                        telemetryReason: ReconnectReason? = nil) async throws
+    {
         log("[Connect] Starting, reason: \(reason)")
 
         guard case .connected = _state.connectionState else {
@@ -385,6 +388,9 @@
             $0.nextReconnectMode = nextReconnectMode
         }
 
+        // One reconnect cycle = one span; attempts are its checkpoints.
+        let reconnectSpan = telemetryScope?.start(name: .reconnect(reason: telemetryReason ?? reason.telemetry), parent: nil)
+
         // quick connect sequence, does not update connection state
         @Sendable func quickReconnectSequence() async throws {
             log("[Connect] Starting .quick reconnect sequence...")
@@ -475,83 +481,90 @@
             _state.mutate { $0.connectedUrl = finalUrl }
 
             dataTracks?.handleReconnect(fullReconnect: true)
+            rtcTelemetry?.joined(self) // tracks re-announced while reconnecting raised no didPublish
         }
 
-        do {
-            let reconnectTask = Task.retrying(totalAttempts: _state.connectOptions.reconnectAttempts,
-                                              retryDelay: { @Sendable attempt in
-                                                  let delay = TimeInterval.computeReconnectDelay(forAttempt: attempt,
-                                                                                                 baseDelay: self._state.connectOptions.reconnectAttemptDelay,
-                                                                                                 maxDelay: self._state.connectOptions.reconnectMaxDelay,
-                                                                                                 totalAttempts: self._state.connectOptions.reconnectAttempts,
-                                                                                                 addJitter: true)
-                                                  self.log("[Connect] Retry cycle waiting for \(String(format: "%.2f", delay)) seconds before attempt \(attempt + 1)")
-                                                  return delay
-                                              }) { currentAttempt, totalAttempts in
-                // Not reconnecting state anymore
-                guard let currentMode = self._state.isReconnectingWithMode else {
-                    self.log("[Connect] Not in reconnect state anymore, exiting retry cycle.")
-                    return
-                }
+        await TelemetrySpan.$current.withValue(reconnectSpan) {
+            do {
+                let reconnectTask = Task.retrying(totalAttempts: _state.connectOptions.reconnectAttempts,
+                                                  retryDelay: { @Sendable attempt in
+                                                      let delay = TimeInterval.computeReconnectDelay(forAttempt: attempt,
+                                                                                                     baseDelay: self._state.connectOptions.reconnectAttemptDelay,
+                                                                                                     maxDelay: self._state.connectOptions.reconnectMaxDelay,
+                                                                                                     totalAttempts: self._state.connectOptions.reconnectAttempts,
+                                                                                                     addJitter: true)
+                                                      self.log("[Connect] Retry cycle waiting for \(String(format: "%.2f", delay)) seconds before attempt \(attempt + 1)")
+                                                      return delay
+                                                  }) { currentAttempt, totalAttempts in
+                    // Not reconnecting state anymore
+                    guard let currentMode = self._state.isReconnectingWithMode else {
+                        self.log("[Connect] Not in reconnect state anymore, exiting retry cycle.")
+                        return
+                    }
 
-                // Full reconnect failed, give up
-                guard currentMode != .full else { return }
+                    // Full reconnect failed, give up
+                    guard currentMode != .full else { return }
 
-                self.log("[Connect] Starting retry attempt \(currentAttempt)/\(totalAttempts) with mode: \(currentMode)")
+                    self.log("[Connect] Starting retry attempt \(currentAttempt)/\(totalAttempts) with mode: \(currentMode)")
 
-                // Try full reconnect for the final attempt
-                if totalAttempts == currentAttempt, self._state.nextReconnectMode == nil {
-                    self._state.mutate { $0.nextReconnectMode = .full }
-                }
+                    // Try full reconnect for the final attempt
+                    if totalAttempts == currentAttempt, self._state.nextReconnectMode == nil {
+                        self._state.mutate { $0.nextReconnectMode = .full }
+                    }
 
-                let mode: ReconnectMode = self._state.mutate {
-                    let mode: ReconnectMode = ($0.nextReconnectMode == .full || $0.isReconnectingWithMode == .full) ? .full : .quick
-                    $0.isReconnectingWithMode = mode
-                    $0.nextReconnectMode = nil
-                    return mode
-                }
+                    let mode: ReconnectMode = self._state.mutate {
+                        let mode: ReconnectMode = ($0.nextReconnectMode == .full || $0.isReconnectingWithMode == .full) ? .full : .quick
+                        $0.isReconnectingWithMode = mode
+                        $0.nextReconnectMode = nil
+                        return mode
+                    }
 
-                do {
-                    if case .quick = mode {
-                        try await quickReconnectSequence()
-                        self.log("[Connect] Quick reconnect succeeded for attempt \(currentAttempt)")
-                    } else if case .full = mode {
-                        try await fullReconnectSequence()
-                        self.log("[Connect] Full reconnect succeeded for attempt \(currentAttempt)")
+                    reconnectSpan?.step(step: .attempt(number: UInt32(currentAttempt), full: mode == .full))
+
+                    do {
+                        if case .quick = mode {
+                            try await quickReconnectSequence()
+                            self.log("[Connect] Quick reconnect succeeded for attempt \(currentAttempt)")
+                        } else if case .full = mode {
+                            try await fullReconnectSequence()
+                            self.log("[Connect] Full reconnect succeeded for attempt \(currentAttempt)")
+                        }
+                    } catch {
+                        self.log("[Connect] Reconnect mode: \(mode) failed with error: \(error)", .error)
+                        // Re-throw
+                        throw error
                     }
-                } catch {
-                    self.log("[Connect] Reconnect mode: \(mode) failed with error: \(error)", .error)
-                    // Re-throw
-                    throw error
                 }
-            }
 
-            _state.mutate {
-                $0.reconnectTask = reconnectTask.cancellable()
-            }
+                _state.mutate {
+                    $0.reconnectTask = reconnectTask.cancellable()
+                }
 
-            try await reconnectTask.value
+                try await reconnectTask.value
 
-            // Re-connect sequence successful
-            log("[Connect] Sequence completed")
-            _state.mutate {
-                $0.connectionState = .connected
-                $0.reconnectTask = nil
-                $0.isReconnectingWithMode = nil
-                $0.nextReconnectMode = nil
-            }
+                // Re-connect sequence successful
+                log("[Connect] Sequence completed")
+                reconnectSpan?.end(outcome: .ok, error: nil)
+                _state.mutate {
+                    $0.connectionState = .connected
+                    $0.reconnectTask = nil
+                    $0.isReconnectingWithMode = nil
+                    $0.nextReconnectMode = nil
+                }
 
-            if let providedUrl = _state.providedUrl, providedUrl.isCloud, let regionManager = await regionManager(for: providedUrl) {
-                // Clear failed region attempts after a successful reconnect.
-                await regionManager.resetAttempts()
-            }
-        } catch {
-            log("[Connect] Sequence failed with error: \(error)")
+                if let providedUrl = _state.providedUrl, providedUrl.isCloud, let regionManager = await regionManager(for: providedUrl) {
+                    // Clear failed region attempts after a successful reconnect.
+                    await regionManager.resetAttempts()
+                }
+            } catch {
+                log("[Connect] Sequence failed with error: \(error)")
+                if Task.isCancelled { reconnectSpan?.cancel() } else { reconnectSpan?.end(with: error) }
 
-            // Only clean up if the reconnect task wasn't cancelled — when cancelled,
-            // the caller (disconnect() or a new reconnect) handles cleanup separately.
-            if !Task.isCancelled {
-                await cleanUp(withError: error)
+                // Only clean up if the reconnect task wasn't cancelled — when cancelled,
+                // the caller (disconnect() or a new reconnect) handles cleanup separately.
+                if !Task.isCancelled {
+                    await cleanUp(withError: error, telemetryReason: .reconnectFailed) // the cause is on the reconnect span
+                }
             }
         }
     }

Sources/LiveKit/Core/Room+SignalClientDelegate.swift

@@ -19,6 +19,7 @@
 #if !COCOAPODS && !LK_XCFRAMEWORK
 import LiveKitNanopb
 #endif
+internal import LiveKitUniFFI
 import Foundation
 
 internal import LiveKitWebRTC
@@ -64,7 +65,7 @@ extension Room: SignalClientDelegate {
             // Abort current attempt
             await signalClient.cleanUp(withError: error)
         case .disconnect:
-            await cleanUp(withError: error)
+            await cleanUp(withError: error, telemetryReason: telemetryDisconnectReason(proto: Int32(reason.rawValue)))
         default:
             log("Unknown leave action: \(action), ignoring", .warning)
         }
@@ -127,6 +128,8 @@ extension Room: SignalClientDelegate {
                     }
                 }
             }
+            updateTelemetryRoom() // a full reconnect keeps the Room's sid but not the participant's
+            rtcTelemetry?.joined(self)
         }
     }
 
@@ -146,6 +149,7 @@ extension Room: SignalClientDelegate {
 
         if response.hasParticipant {
             localParticipant.set(info: response.participant, connectionState: _state.connectionState)
+            updateTelemetryRoom()
         }
 
         _republishLocalTracks()

Sources/LiveKit/Core/Room+TransportDelegate.swift

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+internal import LiveKitUniFFI
 import Foundation
 
 internal import LiveKitWebRTC
@@ -59,7 +60,8 @@ extension Room: TransportDelegate {
             if transport.isPrimary || (_state.hasPublished && transport.target == .publisher), pcState.isDisconnected {
                 Task {
                     do {
-                        try await startReconnect(reason: .transport)
+                        try await startReconnect(reason: .transport,
+                                                 telemetryReason: transport.target == .publisher ? .publisherFailed : .subscriberFailed)
                     } catch {
                         log("Failed calling startReconnect, error: \(error)", .error)
                     }