Telemetry

.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.

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<String> {
+        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<String> =
+        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<String> {
+        val evicted = mutableListOf<String>()
+        val cutoff = System.currentTimeMillis() - maxAgeMillis
+        val kept = mutableListOf<Pair<String, Long>>()
+        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
+    }
+}

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<String, Any>) -> 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<String, Any> = buildMap {
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) {
+            put("thermal", thermalName(powerManager.currentThermalStatus))
+        }
+        put("lowPower", powerManager.isPowerSaveMode)
+    }
+}

android/src/main/java/com/livekit/reactnative/LivekitReactNativeModule.kt

@@ -1,13 +1,15 @@
 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
 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"
     }

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/'],
   },
 ]);

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
+    }
+}