diff --git a/.changeset/client-telemetry-native-store.md b/.changeset/client-telemetry-native-store.md new file mode 100644 index 00000000..a36f274a --- /dev/null +++ b/.changeset/client-telemetry-native-store.md @@ -0,0 +1,5 @@ +--- +'@livekit/react-native': minor +--- + +Client telemetry: `registerGlobals` now names the platform to `livekit-client`'s telemetry pipeline, flushes it when the app leaves the foreground, and gives it a write-ahead cache backed by a directory of files (`LKBatchStore.swift`, `BatchStore.kt`) mirroring the Rust core's `FileCache` — so a session that ends with the app being killed, or an hour spent offline, replays at the next launch instead of being lost. Thermal state, low power mode and memory pressure are read natively and reported as the events SPEC names, stretching the upload cadence under pressure. diff --git a/android/src/main/java/com/livekit/reactnative/BatchStore.kt b/android/src/main/java/com/livekit/reactnative/BatchStore.kt new file mode 100644 index 00000000..b833a15c --- /dev/null +++ b/android/src/main/java/com/livekit/reactnative/BatchStore.kt @@ -0,0 +1,86 @@ +package com.livekit.reactnative + +import android.content.Context +import java.io.File + +/** + * A directory of batches, mirroring the Rust core's `FileCache` so that a React Native app keeps + * the caching semantics an iOS or Android app gets: a batch is written before the network is + * tried, survives the process, and is removed only once the collector has taken it. + * + * Batch ids sort oldest-first as plain strings, so pruning never has to stat or parse anything. + * Writes go to a temporary name and are renamed into place, so a crash never leaves half a batch + * readable. Eviction is oldest-first above the byte, count and age budgets — and one batch always + * survives, because a cache that prunes itself empty is worse than one that is slightly too big. + */ +class BatchStore( + context: Context, + private val maxBytes: Long, + private val maxBatches: Int, + private val maxAgeMillis: Long, +) { + private val directory = File(context.cacheDir, "livekit-telemetry").apply { mkdirs() } + + init { + prune() + } + + /** Stores a batch and returns the ids evicted to stay inside the budgets. */ + fun put(id: String, body: ByteArray): List { + val temporary = File(directory, "$id.tmp") + val destination = File(directory, id) + return try { + temporary.writeBytes(body) + if (!temporary.renameTo(destination)) { + temporary.delete() + return emptyList() + } + prune() + } catch (error: Exception) { + temporary.delete() + emptyList() + } + } + + fun pending(): List = + directory.list()?.filterNot { it.endsWith(".tmp") }?.sorted() ?: emptyList() + + fun read(id: String): ByteArray? = File(directory, id).takeIf { it.isFile }?.readBytes() + + fun remove(id: String) { + File(directory, id).delete() + } + + fun clear() { + pending().forEach { remove(it) } + } + + /** Drops what is too old, then the oldest until the rest fits. Returns what it dropped. */ + private fun prune(): List { + val evicted = mutableListOf() + val cutoff = System.currentTimeMillis() - maxAgeMillis + val kept = mutableListOf>() + var total = 0L + + for (id in pending()) { + val file = File(directory, id) + if (file.lastModified() < cutoff) { + file.delete() + evicted.add(id) + continue + } + kept.add(id to file.length()) + total += file.length() + } + + var index = 0 + while ((total > maxBytes || kept.size - index > maxBatches) && kept.size - index > 1) { + val (id, bytes) = kept[index] + remove(id) + evicted.add(id) + total -= bytes + index += 1 + } + return evicted + } +} diff --git a/android/src/main/java/com/livekit/reactnative/DeviceStateMonitor.kt b/android/src/main/java/com/livekit/reactnative/DeviceStateMonitor.kt new file mode 100644 index 00000000..9133d124 --- /dev/null +++ b/android/src/main/java/com/livekit/reactnative/DeviceStateMonitor.kt @@ -0,0 +1,99 @@ +package com.livekit.reactnative + +import android.content.BroadcastReceiver +import android.content.ComponentCallbacks2 +import android.content.Context +import android.content.Intent +import android.content.IntentFilter +import android.content.res.Configuration +import android.os.Build +import android.os.PowerManager +import androidx.annotation.RequiresApi + +/** + * Thermal state, power save mode and memory pressure: the three signals SPEC's cadence policy needs + * and no browser can answer. The pipeline never measures any of them itself — measuring CPU costs + * CPU — so each value here is the OS's own judgement, reported when it changes. + * + * Values are SPEC's names, not Android's, so that a record from Android and a record from iOS say + * the same thing. + */ +class DeviceStateMonitor( + private val context: Context, + private val onChange: (Map) -> Unit, +) { + companion object { + const val EVENT_NAME = "LK_DEVICE_STATE" + + /** `PowerManager.THERMAL_STATUS_*` collapsed onto SPEC's four levels. */ + fun thermalName(status: Int): String = when (status) { + PowerManager.THERMAL_STATUS_NONE -> "nominal" + PowerManager.THERMAL_STATUS_LIGHT -> "fair" + PowerManager.THERMAL_STATUS_MODERATE, PowerManager.THERMAL_STATUS_SEVERE -> "serious" + else -> "critical" + } + + /** `onTrimMemory` levels, as SPEC maps them. */ + fun memoryName(level: Int): String = when { + level >= ComponentCallbacks2.TRIM_MEMORY_COMPLETE -> "critical" + level >= ComponentCallbacks2.TRIM_MEMORY_RUNNING_CRITICAL -> "critical" + level >= ComponentCallbacks2.TRIM_MEMORY_BACKGROUND -> "warning" + level >= ComponentCallbacks2.TRIM_MEMORY_RUNNING_LOW -> "warning" + else -> "normal" + } + } + + private val powerManager = context.getSystemService(Context.POWER_SERVICE) as PowerManager + + @RequiresApi(Build.VERSION_CODES.Q) + private val thermalListener = PowerManager.OnThermalStatusChangedListener { status -> + onChange(mapOf("thermal" to thermalName(status))) + } + + private val powerSaveReceiver = object : BroadcastReceiver() { + override fun onReceive(context: Context?, intent: Intent?) { + onChange(mapOf("lowPower" to powerManager.isPowerSaveMode)) + } + } + + private val memoryCallbacks = object : ComponentCallbacks2 { + override fun onTrimMemory(level: Int) = onChange(mapOf("memory" to memoryName(level))) + override fun onConfigurationChanged(newConfig: Configuration) = Unit + + @Deprecated("Required by ComponentCallbacks2 below API 34") + override fun onLowMemory() = onChange(mapOf("memory" to "critical")) + } + + private var started = false + + fun start() { + if (started) return + started = true + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) { + powerManager.addThermalStatusListener(thermalListener) + } + context.registerReceiver( + powerSaveReceiver, + IntentFilter(PowerManager.ACTION_POWER_SAVE_MODE_CHANGED), + ) + context.applicationContext.registerComponentCallbacks(memoryCallbacks) + } + + fun stop() { + if (!started) return + started = false + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) { + powerManager.removeThermalStatusListener(thermalListener) + } + runCatching { context.unregisterReceiver(powerSaveReceiver) } + context.applicationContext.unregisterComponentCallbacks(memoryCallbacks) + } + + /** Everything that is a state rather than an edge, for the first report. */ + fun snapshot(): Map = buildMap { + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) { + put("thermal", thermalName(powerManager.currentThermalStatus)) + } + put("lowPower", powerManager.isPowerSaveMode) + } +} diff --git a/android/src/main/java/com/livekit/reactnative/LivekitReactNativeModule.kt b/android/src/main/java/com/livekit/reactnative/LivekitReactNativeModule.kt index ece4fcb5..063e7716 100644 --- a/android/src/main/java/com/livekit/reactnative/LivekitReactNativeModule.kt +++ b/android/src/main/java/com/livekit/reactnative/LivekitReactNativeModule.kt @@ -1,6 +1,7 @@ package com.livekit.reactnative import android.media.AudioAttributes +import android.util.Base64 import android.util.Log import com.facebook.react.bridge.Arguments import com.facebook.react.bridge.Promise @@ -8,6 +9,7 @@ import com.facebook.react.bridge.ReactApplicationContext import com.facebook.react.bridge.ReactContextBaseJavaModule import com.facebook.react.bridge.ReactMethod import com.facebook.react.bridge.ReadableMap +import com.facebook.react.bridge.WritableArray import com.facebook.react.modules.core.DeviceEventManagerModule import com.livekit.reactnative.audio.AudioDeviceKind import com.livekit.reactnative.audio.AudioManagerUtils @@ -31,6 +33,86 @@ class LivekitReactNativeModule(reactContext: ReactApplicationContext) : ReactCon val audioSinkManager = AudioSinkManager(reactContext) val audioManager = AudioSwitchManager(reactContext.applicationContext) + + /** + * Thermal state, power save mode and memory pressure, for SPEC's cadence policy. These reach + * the Rust core, never JavaScript: `livekit-client` has no business knowing a phone gets hot. + * Called once by whoever hosts the core; nothing is observed until it is. + */ + private val deviceStateMonitor = DeviceStateMonitor(reactContext.applicationContext) { change -> + val payload = Arguments.createMap() + change.forEach { (key, value) -> + when (value) { + is Boolean -> payload.putBoolean(key, value) + else -> payload.putString(key, value.toString()) + } + } + reactContext.getJSModule(DeviceEventManagerModule.RCTDeviceEventEmitter::class.java) + .emit(DeviceStateMonitor.EVENT_NAME, payload) + } + + /** + * The first state comes back on the promise rather than as an event — an event sent from inside + * this call would race the listener JS is registering around it, and be dropped. + */ + /** + * The write-ahead cache `livekit-client`'s pipeline stores batches in. Synchronous, because its + * queue path has no await in it; base64 because the bridge does not carry bytes. + * + * SPEC's budget: 4 MiB across at most 512 batches, nothing older than a day. + */ + private val batchStore by lazy { + BatchStore(reactContext.applicationContext, 4L * 1024 * 1024, 512, 24L * 60 * 60 * 1000) + } + + @ReactMethod(isBlockingSynchronousMethod = true) + fun batchStorePut(id: String, body: String): WritableArray { + val evicted = batchStore.put(id, Base64.decode(body, Base64.NO_WRAP)) + val out = Arguments.createArray() + evicted.forEach { out.pushString(it) } + return out + } + + @ReactMethod(isBlockingSynchronousMethod = true) + fun batchStorePending(): WritableArray { + val out = Arguments.createArray() + batchStore.pending().forEach { out.pushString(it) } + return out + } + + @ReactMethod(isBlockingSynchronousMethod = true) + fun batchStoreRead(id: String): String? = + batchStore.read(id)?.let { Base64.encodeToString(it, Base64.NO_WRAP) } + + @ReactMethod(isBlockingSynchronousMethod = true) + fun batchStoreRemove(id: String): Boolean { + batchStore.remove(id) + return true + } + + @ReactMethod(isBlockingSynchronousMethod = true) + fun batchStoreClear(): Boolean { + batchStore.clear() + return true + } + + @ReactMethod + fun startDeviceStateUpdates(promise: Promise) { + deviceStateMonitor.start() + val snapshot = Arguments.createMap() + deviceStateMonitor.snapshot().forEach { (key, value) -> + when (value) { + is Boolean -> snapshot.putBoolean(key, value) + else -> snapshot.putString(key, value.toString()) + } + } + promise.resolve(snapshot) + } + + override fun invalidate() { + deviceStateMonitor.stop() + super.invalidate() + } override fun getName(): String { return "LivekitReactNativeModule" } diff --git a/eslint.config.mjs b/eslint.config.mjs index 9bd580de..b0f8b700 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -33,6 +33,7 @@ export default defineConfig([ }, }, { - ignores: ['node_modules/', 'lib/', 'docs/', 'src/polyfills/', '**/.yalc/'], + // telemetry-poc is a stand-alone React Native app with its own config; see its README. + ignores: ['node_modules/', 'lib/', 'docs/', 'src/polyfills/', '**/.yalc/', 'telemetry-poc/'], }, ]); diff --git a/ios/LKBatchStore.swift b/ios/LKBatchStore.swift new file mode 100644 index 00000000..3d349945 --- /dev/null +++ b/ios/LKBatchStore.swift @@ -0,0 +1,93 @@ +import Foundation + +/** + * A directory of batches, mirroring the Rust core's `FileCache` so that a React Native app keeps + * the caching semantics an iOS or Android app gets: a batch is written before the network is + * tried, survives the process, and is removed only once the collector has taken it. + * + * Batch ids sort oldest-first as plain strings, so pruning never has to stat or parse anything. + * Writes go to a temporary name and are renamed into place, so a crash never leaves half a batch + * readable. Eviction is oldest-first above the byte, count and age budgets — and one batch always + * survives, because a cache that prunes itself empty is worse than one that is slightly too big. + */ +@objc(LKBatchStore) +public class LKBatchStore: NSObject { + private let directory: URL + private let maxBytes: Int + private let maxBatches: Int + private let maxAge: TimeInterval + + @objc public init(maxBytes: Int, maxBatches: Int, maxAgeSeconds: Double) { + let caches = FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask)[0] + directory = caches.appendingPathComponent("livekit-telemetry", isDirectory: true) + self.maxBytes = maxBytes + self.maxBatches = maxBatches + maxAge = maxAgeSeconds + super.init() + try? FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + _ = prune() + } + + /// Stores a batch and returns the ids evicted to stay inside the budgets. + @objc public func put(id: String, body: Data) -> [String] { + let destination = directory.appendingPathComponent(id) + let temporary = directory.appendingPathComponent("\(id).tmp") + guard (try? body.write(to: temporary)) != nil else { return [] } + do { + _ = try FileManager.default.replaceItemAt(destination, withItemAt: temporary) + } catch { + try? FileManager.default.removeItem(at: temporary) + return [] + } + return prune() + } + + @objc public func pending() -> [String] { + let names = (try? FileManager.default.contentsOfDirectory(atPath: directory.path)) ?? [] + return names.filter { !$0.hasSuffix(".tmp") }.sorted() + } + + @objc public func read(id: String) -> Data? { + try? Data(contentsOf: directory.appendingPathComponent(id)) + } + + @objc public func remove(id: String) { + try? FileManager.default.removeItem(at: directory.appendingPathComponent(id)) + } + + @objc public func clear() { + for id in pending() { remove(id: id) } + } + + /// Drops what is too old, then the oldest until the rest fits. Returns what it dropped. + private func prune() -> [String] { + var evicted: [String] = [] + var sizes: [(id: String, bytes: Int)] = [] + var total = 0 + let cutoff = Date().addingTimeInterval(-maxAge) + + for id in pending() { + let path = directory.appendingPathComponent(id) + let attributes = try? FileManager.default.attributesOfItem(atPath: path.path) + let modified = attributes?[.modificationDate] as? Date ?? Date() + let bytes = (attributes?[.size] as? NSNumber)?.intValue ?? 0 + if modified < cutoff { + remove(id: id) + evicted.append(id) + continue + } + sizes.append((id, bytes)) + total += bytes + } + + var index = 0 + while (total > maxBytes || sizes.count - index > maxBatches) && sizes.count - index > 1 { + let oldest = sizes[index] + remove(id: oldest.id) + evicted.append(oldest.id) + total -= oldest.bytes + index += 1 + } + return evicted + } +} diff --git a/ios/LKDeviceState.swift b/ios/LKDeviceState.swift new file mode 100644 index 00000000..762c3206 --- /dev/null +++ b/ios/LKDeviceState.swift @@ -0,0 +1,87 @@ +import Foundation + +/** + * Thermal state, low power mode and memory pressure: the three signals SPEC's cadence policy needs + * and no browser can answer. The pipeline never measures any of them itself — measuring CPU costs + * CPU — so each value here is the OS's own judgement, reported when it changes. + * + * Values are SPEC's names, not Apple's, so that a record from iOS and a record from Android say the + * same thing. + */ +@objc(LKDeviceState) +public class LKDeviceState: NSObject { + /// One event carrying whatever changed; JS merges it into the state it already has. + @objc public static let eventName = "LK_DEVICE_STATE" + + private let onChange: ([String: Any]) -> Void + private var memorySource: DispatchSourceMemoryPressure? + + @objc public init(onChange: @escaping ([String: Any]) -> Void) { + self.onChange = onChange + super.init() + + NotificationCenter.default.addObserver( + self, + selector: #selector(thermalChanged), + name: ProcessInfo.thermalStateDidChangeNotification, + object: nil + ) + NotificationCenter.default.addObserver( + self, + selector: #selector(powerChanged), + name: Notification.Name.NSProcessInfoPowerStateDidChange, + object: nil + ) + + // DispatchSource reports all three levels, including the return to normal that + // `didReceiveMemoryWarning` never sends. + let source = DispatchSource.makeMemoryPressureSource( + eventMask: [.normal, .warning, .critical], + queue: .main + ) + source.setEventHandler { [weak self, weak source] in + guard let self, let event = source?.data else { return } + self.onChange(["memory": LKDeviceState.memoryName(event)]) + } + source.resume() + memorySource = source + } + + deinit { + NotificationCenter.default.removeObserver(self) + memorySource?.cancel() + } + + /// Everything that is a state rather than an edge, for the first report. + @objc public func snapshot() -> [String: Any] { + let info = ProcessInfo.processInfo + return [ + "thermal": LKDeviceState.thermalName(info.thermalState), + "lowPower": info.isLowPowerModeEnabled, + ] + } + + @objc private func thermalChanged() { + onChange(["thermal": LKDeviceState.thermalName(ProcessInfo.processInfo.thermalState)]) + } + + @objc private func powerChanged() { + onChange(["lowPower": ProcessInfo.processInfo.isLowPowerModeEnabled]) + } + + private static func thermalName(_ state: ProcessInfo.ThermalState) -> String { + switch state { + case .nominal: return "nominal" + case .fair: return "fair" + case .serious: return "serious" + case .critical: return "critical" + @unknown default: return "nominal" + } + } + + private static func memoryName(_ event: DispatchSource.MemoryPressureEvent) -> String { + if event.contains(.critical) { return "critical" } + if event.contains(.warning) { return "warning" } + return "normal" + } +} diff --git a/ios/LiveKitReactNativeModule.swift b/ios/LiveKitReactNativeModule.swift index 435b339b..13e583c3 100644 --- a/ios/LiveKitReactNativeModule.swift +++ b/ios/LiveKitReactNativeModule.swift @@ -7,11 +7,19 @@ struct LKEvents { static let kEventVolumeProcessed = "LK_VOLUME_PROCESSED"; static let kEventMultibandProcessed = "LK_MULTIBAND_PROCESSED"; static let kEventAudioData = "LK_AUDIO_DATA"; + static let kEventDeviceState = LKDeviceState.eventName; } @objc(LivekitReactNativeModule) public class LivekitReactNativeModule: RCTEventEmitter { + private var deviceState: LKDeviceState? = nil + + /// SPEC's cache budget: 4 MiB across at most 512 batches, nothing older than a day. + private lazy var batchStore = LKBatchStore(maxBytes: 4 * 1024 * 1024, + maxBatches: 512, + maxAgeSeconds: 24 * 60 * 60) + // This cannot be initialized in init as self.bridge is given afterwards. private var _audioRendererManager: AudioRendererManager? = nil public var audioRendererManager: AudioRendererManager { @@ -256,11 +264,63 @@ public class LivekitReactNativeModule: RCTEventEmitter { return nil } + /// Thermal state, low power mode and memory pressure, for SPEC's cadence policy. These reach + /// the Rust core, never JavaScript: `livekit-client` has no business knowing a phone gets hot. + /// Called once by whoever hosts the core; nothing is observed until it is. + /// + /// The first state comes back on the promise rather than as an event — an event sent from + /// inside this call would race the listener a caller registers around it, and be dropped. + @objc(startDeviceStateUpdates:withRejecter:) + public func startDeviceStateUpdates( + _ resolve: @escaping RCTPromiseResolveBlock, + withRejecter reject: @escaping RCTPromiseRejectBlock + ) { + if deviceState == nil { + deviceState = LKDeviceState { [weak self] change in + self?.sendEvent(withName: LKEvents.kEventDeviceState, body: change) + } + } + resolve(deviceState?.snapshot() ?? [:]) + } + + /// The write-ahead cache `livekit-client`'s pipeline stores batches in. Synchronous, because + /// its queue path has no await in it; base64 because the bridge does not carry bytes. + @objc(batchStorePut:body:) + public func batchStorePut(_ id: String, body: String) -> [String] { + guard let data = Data(base64Encoded: body) else { return [] } + return batchStore.put(id: id, body: data) + } + + @objc(batchStorePending) + public func batchStorePending() -> [String] { + batchStore.pending() + } + + @objc(batchStoreRead:) + public func batchStoreRead(_ id: String) -> String? { + batchStore.read(id: id)?.base64EncodedString() + } + + // A blocking synchronous method must return an *object*: the TurboModule interop retains + // whatever it gets back, so a `Bool` return is read as a pointer and segfaults the app. + @objc(batchStoreRemove:) + public func batchStoreRemove(_ id: String) -> Any? { + batchStore.remove(id: id) + return nil + } + + @objc(batchStoreClear) + public func batchStoreClear() -> Any? { + batchStore.clear() + return nil + } + override public func supportedEvents() -> [String]! { return [ LKEvents.kEventVolumeProcessed, LKEvents.kEventMultibandProcessed, LKEvents.kEventAudioData, + LKEvents.kEventDeviceState, ] } } diff --git a/ios/LivekitReactNativeModule.m b/ios/LivekitReactNativeModule.m index dfe83d6c..057e4f8f 100644 --- a/ios/LivekitReactNativeModule.m +++ b/ios/LivekitReactNativeModule.m @@ -12,6 +12,16 @@ @interface RCT_EXTERN_MODULE(LivekitReactNativeModule, RCTEventEmitter) RCT_EXTERN_METHOD(setDefaultAudioTrackVolume:(nonnull NSNumber *) volume) +RCT_EXTERN_METHOD(startDeviceStateUpdates:(RCTPromiseResolveBlock)resolve + withRejecter:(RCTPromiseRejectBlock)reject) + +RCT_EXTERN__BLOCKING_SYNCHRONOUS_METHOD(batchStorePut:(nonnull NSString *)id + body:(nonnull NSString *)body) +RCT_EXTERN__BLOCKING_SYNCHRONOUS_METHOD(batchStorePending) +RCT_EXTERN__BLOCKING_SYNCHRONOUS_METHOD(batchStoreRead:(nonnull NSString *)id) +RCT_EXTERN__BLOCKING_SYNCHRONOUS_METHOD(batchStoreRemove:(nonnull NSString *)id) +RCT_EXTERN__BLOCKING_SYNCHRONOUS_METHOD(batchStoreClear) + RCT_EXTERN_METHOD(showAudioRoutePicker) RCT_EXTERN_METHOD(getAudioOutputsWithResolver:(RCTPromiseResolveBlock)resolve withRejecter:(RCTPromiseRejectBlock)reject) diff --git a/src/index.tsx b/src/index.tsx index 3d5ed530..7c02641a 100644 --- a/src/index.tsx +++ b/src/index.tsx @@ -31,6 +31,7 @@ import type { LogLevel, SetLogLevelOptions } from './logger'; import RNE2EEManager from './e2ee/RNE2EEManager'; import RNKeyProvider, { type RNKeyProviderOptions } from './e2ee/RNKeyProvider'; import { setupNativeEvents } from './events/EventEmitter'; +import { registerTelemetry } from './telemetry'; import { ReadableStream, WritableStream, @@ -74,6 +75,7 @@ export function registerGlobals(options?: RegisterGlobalsOptions) { shimCryptoUuid(); shimWebstreams(); setupNativeEvents(); + registerTelemetry(); } function livekitRegisterGlobals() { @@ -168,6 +170,9 @@ export * from './logger'; export * from './audio/AudioManager'; export * from './audio/AudioManagerLegacy'; export * from './audio/MediaRecorder'; +export { registerTelemetry } from './telemetry'; +export { nativeBatchStore } from './telemetryStorage'; +export { Telemetry } from 'livekit-client'; export { AudioSession, diff --git a/src/telemetry.ts b/src/telemetry.ts new file mode 100644 index 00000000..eed5413b --- /dev/null +++ b/src/telemetry.ts @@ -0,0 +1,100 @@ +import { Telemetry } from 'livekit-client'; +import { + AppState, + NativeEventEmitter, + NativeModules, + Platform, +} from 'react-native'; +import { version } from '../package.json'; +import { nativeBatchStore } from './telemetryStorage'; + +/** + * The React Native half of client telemetry. + * + * `livekit-client` owns the instrumentation — when a connect span starts, what a stats window + * holds, how a subscribe ends at first media — and this package reuses it unchanged. What it adds + * is what a phone knows and a browser does not: a filesystem to cache batches in, and a device + * that gets hot, runs low on memory and asks for less work. + * + * None of that vocabulary crosses into `livekit-client`. It receives a `TelemetryStorage` it never + * inspects, an event name it never interprets, and a cadence factor without a reason attached. + */ + +/** What `LK_DEVICE_STATE` carries, in SPEC's names — the native side does the mapping. */ +interface NativeDeviceState { + thermal?: 'nominal' | 'fair' | 'serious' | 'critical'; + lowPower?: boolean; + memory?: 'normal' | 'warning' | 'critical'; +} + +/** SPEC's cadence table, for the rows only a device can fill in. Factors multiply, capped at 4×. */ +const THERMAL_FACTOR = { nominal: 1, fair: 1, serious: 2, critical: 4 }; +const MEMORY_FACTOR = { normal: 1, warning: 2, critical: 4 }; + +let device: NativeDeviceState = {}; + +function reportDevice(next: NativeDeviceState) { + if (next.thermal !== undefined && next.thermal !== device.thermal) { + Telemetry.emit('lk.device.thermal.changed', { + 'lk.device.thermal.state': next.thermal, + }); + } + if (next.lowPower !== undefined && next.lowPower !== device.lowPower) { + Telemetry.emit('lk.device.low_power.changed', { + 'lk.device.low_power.enabled': next.lowPower, + }); + } + if (next.memory !== undefined && next.memory !== device.memory) { + Telemetry.emit('lk.device.memory.changed', { + 'lk.device.memory.pressure': next.memory, + }); + } + device = { ...device, ...next }; + Telemetry.setCadenceFactor( + Math.min( + 4, + THERMAL_FACTOR[device.thermal ?? 'nominal'] * + MEMORY_FACTOR[device.memory ?? 'normal'] * + (device.lowPower ? 2 : 1) + ) + ); +} + +export function registerTelemetry() { + Telemetry.configure({ + resource: { + 'service.name': 'livekit-client-react-native', + 'service.version': version, + 'os.name': Platform.OS, + 'os.version': String(Platform.Version), + }, + storage: nativeBatchStore(), + }); + + // A browser flushes on `visibilitychange`; there is no such event here, so an app leaving the + // foreground is both the record and the last chance to upload. + const report = (state: string) => + Telemetry.deviceState({ + appState: state === 'active' ? 'foreground' : 'background', + }); + AppState.addEventListener('change', (state) => { + report(state); + if (state !== 'active') { + Telemetry.flush().catch(() => {}); + } + }); + report(AppState.currentState ?? 'active'); + + const native = NativeModules.LivekitReactNativeModule; + // An app on an older native build simply reports less; it must not crash. + if (!native?.startDeviceStateUpdates) { + return; + } + new NativeEventEmitter(native).addListener('LK_DEVICE_STATE', reportDevice); + // The first state is the promise's answer, not an event: an event sent while this call is still + // in flight would arrive before the listener above is registered natively, and be dropped. + native + .startDeviceStateUpdates() + .then(reportDevice) + .catch(() => {}); +} diff --git a/src/telemetryStorage.ts b/src/telemetryStorage.ts new file mode 100644 index 00000000..935f23a2 --- /dev/null +++ b/src/telemetryStorage.ts @@ -0,0 +1,36 @@ +import { fromByteArray, toByteArray } from 'base64-js'; +import type { TelemetryStorage } from 'livekit-client'; +import { NativeModules } from 'react-native'; + +/** + * The write-ahead cache `livekit-client`'s pipeline stores batches in, backed by a directory of + * files — the same shape the Rust core's `FileCache` has, so a React Native app keeps the caching + * semantics an iOS or Android app gets: a batch survives the process and is removed only once the + * collector has taken it. + * + * The calls are synchronous blocking bridge calls because the pipeline's queue path has no await + * in it, and base64 because the bridge does not carry bytes. That is also why this is native code + * in this package rather than one of the filesystem packages on npm — all of those are async. + */ +export function nativeBatchStore(): TelemetryStorage | undefined { + const native = NativeModules.LivekitReactNativeModule; + // An app on an older native build simply keeps its batches in memory. + if (!native?.batchStorePut) { + return undefined; + } + return { + put: (id: string, body: Uint8Array) => + native.batchStorePut(id, fromByteArray(body)) ?? [], + pending: () => native.batchStorePending() ?? [], + read: (id: string) => { + const body = native.batchStoreRead(id); + return body ? toByteArray(body) : undefined; + }, + remove: (id: string) => { + native.batchStoreRemove(id); + }, + clear: () => { + native.batchStoreClear(); + }, + }; +} diff --git a/telemetry-poc/.bundle/config b/telemetry-poc/.bundle/config new file mode 100644 index 00000000..848943bb --- /dev/null +++ b/telemetry-poc/.bundle/config @@ -0,0 +1,2 @@ +BUNDLE_PATH: "vendor/bundle" +BUNDLE_FORCE_RUBY_PLATFORM: 1 diff --git a/telemetry-poc/.eslintrc.js b/telemetry-poc/.eslintrc.js new file mode 100644 index 00000000..187894b6 --- /dev/null +++ b/telemetry-poc/.eslintrc.js @@ -0,0 +1,4 @@ +module.exports = { + root: true, + extends: '@react-native', +}; diff --git a/telemetry-poc/.gitignore b/telemetry-poc/.gitignore new file mode 100644 index 00000000..de999559 --- /dev/null +++ b/telemetry-poc/.gitignore @@ -0,0 +1,75 @@ +# OSX +# +.DS_Store + +# Xcode +# +build/ +*.pbxuser +!default.pbxuser +*.mode1v3 +!default.mode1v3 +*.mode2v3 +!default.mode2v3 +*.perspectivev3 +!default.perspectivev3 +xcuserdata +*.xccheckout +*.moved-aside +DerivedData +*.hmap +*.ipa +*.xcuserstate +**/.xcode.env.local + +# Android/IntelliJ +# +build/ +.idea +.gradle +local.properties +*.iml +*.hprof +.cxx/ +*.keystore +!debug.keystore +.kotlin/ + +# node.js +# +node_modules/ +npm-debug.log +yarn-error.log + +# fastlane +# +# It is recommended to not store the screenshots in the git repo. Instead, use fastlane to re-generate the +# screenshots whenever they are needed. +# For more information about the recommended setup visit: +# https://docs.fastlane.tools/best-practices/source-control/ + +**/fastlane/report.xml +**/fastlane/Preview.html +**/fastlane/screenshots +**/fastlane/test_output + +# Bundle artifact +*.jsbundle + +# Ruby / CocoaPods +**/Pods/ +/vendor/bundle/ + +# Temporary files created by Metro to check the health of the file watcher +.metro-health-check* + +# testing +/coverage + +# Yarn +.yarn/* +!.yarn/patches +!.yarn/plugins +!.yarn/releases +!.yarn/sdks +!.yarn/versions diff --git a/telemetry-poc/.prettierrc.js b/telemetry-poc/.prettierrc.js new file mode 100644 index 00000000..06860c8d --- /dev/null +++ b/telemetry-poc/.prettierrc.js @@ -0,0 +1,5 @@ +module.exports = { + arrowParens: 'avoid', + singleQuote: true, + trailingComma: 'all', +}; diff --git a/telemetry-poc/.watchmanconfig b/telemetry-poc/.watchmanconfig new file mode 100644 index 00000000..0967ef42 --- /dev/null +++ b/telemetry-poc/.watchmanconfig @@ -0,0 +1 @@ +{} diff --git a/telemetry-poc/App.tsx b/telemetry-poc/App.tsx new file mode 100644 index 00000000..85a59c6c --- /dev/null +++ b/telemetry-poc/App.tsx @@ -0,0 +1,74 @@ +/** + * Telemetry PoC — the integrated pipeline on React Native: `livekit-client`'s telemetry module + * (the same one a browser runs) plus this SDK's own `src/telemetry.ts` seam, which names the + * platform and flushes when the app leaves the foreground. + * + * The iOS simulator shares the host's network stack, so 127.0.0.1 is the Mac running + * `otelcol-contrib --config ../../client-sdk-js-telemetry/src/telemetry/otelcol-web.yaml`. + */ +import React, {useCallback, useEffect, useState} from 'react'; +import {Button, NativeModules, SafeAreaView, ScrollView, Text} from 'react-native'; +import {Telemetry} from 'livekit-client'; +import {registerTelemetry} from '../src/telemetry'; + +const endpoint = 'http://127.0.0.1:4320/v1/logs'; + +export default function App() { + const [log, setLog] = useState([]); + const say = useCallback((line: string) => setLog(prev => [...prev, line]), []); + + // The PoC has no debugger attached, so warnings have to be visible on screen. + useEffect(() => { + const warn = console.warn; + console.warn = (...args: unknown[]) => { + say(`warn: ${args.map(String).join(' ')}`.slice(0, 200)); + warn(...args); + }; + return () => { + console.warn = warn; + }; + }, [say]); + + const send = useCallback( + async (encoding: 'protobuf' | 'json') => { + try { + // A real app gets the destination from its first Cloud connect; here it is a local + // collector, named before registerGlobals' seam reports the device it is running on. + Telemetry.configure({endpoint, encoding, flushInterval: 1}); + registerTelemetry(); + const native = NativeModules.LivekitReactNativeModule; + say( + `native module: ${ + native + ? Object.keys(native) + .filter(k => k.toLowerCase().includes('device')) + .join(', ') || 'linked, no device methods' + : 'missing' + }`, + ); + await Telemetry.ping(encoding === 'json' ? 2 : 1); + say(`${encoding}: ${Telemetry.diagnostics()}`); + } catch (error) { + say(`${encoding}: ${String(error)}`); + } + }, + [say], + ); + + // Ping on mount as well as on tap, so a headless `simctl launch` proves the path on its own. + useEffect(() => { + send('protobuf').then(() => send('json')); + }, [send]); + + return ( + +