fix(monitor): Prevent NPE in cache merge from freezing device flow

Battery slot percent comparisons in mergeBatterySlot/hasStateChanged compiled to Intrinsics.areEqual on boxed Float; R8 optimization on Android 10/11 dropped a null check during inlining and the resulting NPE escaped onEach { persistLiveDevices }, cancelling the upstream combine and freezing every observer of DeviceMonitor.devices. Comparisons now operate on primitive float (cmpg-float in dex) so no Intrinsics.areEqual call remains in the merge path. The persist loop also catches and reports per-profile, and AAP-only profiles with active DeviceInfo are now persisted even when no BLE pod is in range.
2026-09-14 18:26:11 -04:00 · 2026-05-03 14:20:49 +02:00
parent 82dc88faac
commit 7d93e772b4
6 changed files with 299 additions and 28 deletions
@@ -4,7 +4,10 @@ import eu.darken.capod.common.bluetooth.BluetoothAddress
 import eu.darken.capod.common.bluetooth.BluetoothManager2
 import eu.darken.capod.common.TimeSource
 import eu.darken.capod.common.coroutine.AppScope
+import eu.darken.capod.common.debug.Bugs
+import eu.darken.capod.common.debug.logging.Logging.Priority.ERROR
 import eu.darken.capod.common.debug.logging.Logging.Priority.VERBOSE
+import eu.darken.capod.common.debug.logging.asLog
 import eu.darken.capod.common.debug.logging.log
 import eu.darken.capod.common.debug.logging.logTag
 import eu.darken.capod.common.flow.replayingShare
@@ -20,6 +23,7 @@ import eu.darken.capod.profiles.core.AppleDeviceProfile
 import eu.darken.capod.profiles.core.toReactionConfig
 import eu.darken.capod.profiles.core.DeviceProfile
 import eu.darken.capod.profiles.core.DeviceProfilesRepo
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.combine
@@ -86,7 +90,34 @@ class DeviceMonitor @Inject constructor(
            connectedAddresses = connectedAddresses,
        )
    }.onEach { liveState ->
-        persistLiveDevices(liveState.liveDevices)
+        persistLiveDevices(liveState.liveDevices + aapOnlyForPersistence(liveState))
+    }
+
+    /**
+     * AAP-only profiles whose state didn't make it into [LiveMergeState.liveDevices] (no BLE pod
+     * in the current scan). Without this, [persistLiveDevices] would only ever see BLE-backed
+     * devices, and AAP-delivered DeviceInfo (earbud serials, marketing version) for an out-of-BLE
+     * pod would never reach the cache even though the AAP socket is alive.
+     */
+    private fun aapOnlyForPersistence(state: LiveMergeState): List<PodDevice> {
+        val coveredProfileIds = state.liveDevices.mapNotNull { it.profileId }.toSet()
+        return state.profiles.mapNotNull { profile ->
+            if (profile.id in coveredProfileIds) return@mapNotNull null
+            val aap = state.aapStates.forProfile(profile) ?: return@mapNotNull null
+            PodDevice(
+                profileId = profile.id,
+                label = profile.label,
+                ble = null,
+                aap = aap,
+                profileAddress = profile.address,
+                profileModel = profile.model,
+                profileKeyState = profile.toBleKeyState(),
+                profileLearnedAllowOffEnabled = (profile as? AppleDeviceProfile)?.learnedAllowOffEnabled,
+                profileLastRequestedListeningModeCycleMask = (profile as? AppleDeviceProfile)?.lastRequestedListeningModeCycleMask,
+                reactions = profile.toReactionConfig(),
+                isSystemConnected = profile.address in state.connectedAddresses,
+            )
+        }
    }

    val devices: Flow<List<PodDevice>> = combine(
@@ -181,14 +212,25 @@ class DeviceMonitor @Inject constructor(
        dedupedLiveDevices + nonLiveDevices
    }.replayingShare(appScope)

+    private val reportedPersistFailures = mutableSetOf<String>()
+
    private suspend fun persistLiveDevices(devices: List<PodDevice>) {
        for (device in devices) {
            val profileId = device.profileId ?: continue
-            val existing = deviceStateCache.cachedStates.value[profileId]
-            val newState = device.copy(cached = existing).toCachedState(existing, timeSource.now()) ?: continue
+            try {
+                val existing = deviceStateCache.cachedStates.value[profileId]
+                val newState = device.copy(cached = existing).toCachedState(existing, timeSource.now()) ?: continue

-            log(TAG, VERBOSE) { "Persisting state for $profileId" }
-            deviceStateCache.save(profileId, newState)
+                log(TAG, VERBOSE) { "Persisting state for $profileId" }
+                deviceStateCache.save(profileId, newState)
+            } catch (e: CancellationException) {
+                throw e
+            } catch (e: Exception) {
+                log(TAG, ERROR) { "Failed to persist state for $profileId: ${e.asLog()}" }
+                if (reportedPersistFailures.add(profileId)) {
+                    runCatching { Bugs.report(tag = TAG, message = "persistLiveDevices failed for $profileId", exception = e) }
+                }
+            }
        }
    }

@@ -36,7 +36,9 @@ data class CachedDeviceState(
 ) {
    val deviceInfo: AapDeviceInfo?
        get() {
-            if (deviceName == null && serialNumber == null && firmwareVersion == null) return null
+            if (deviceName == null && serialNumber == null && firmwareVersion == null
+                && leftEarbudSerial == null && rightEarbudSerial == null && marketingVersion == null
+            ) return null
            return AapDeviceInfo(
                name = deviceName ?: "",
                modelNumber = "",
@@ -14,7 +14,7 @@ import java.time.Instant
 * Returns null if:
 * - The device is not live (cached-only)
 * - The device has no profile
- * - All live battery values are null
+ * - All live battery values AND live DeviceInfo are null (nothing fresh to persist)
 * - The state hasn't changed from [existing] (dedup)
 */
 fun PodDevice.toCachedState(
@@ -28,11 +28,12 @@ fun PodDevice.toCachedState(
    val liveRight = aap?.batteryRight ?: (ble as? DualBlePodSnapshot)?.batteryRightPodPercent
    val liveCase = aap?.batteryCase ?: (ble as? HasCase)?.batteryCasePercent
    val liveHeadset = aap?.batteryHeadset ?: (ble as? SingleBlePodSnapshot)?.batteryHeadsetPercent
-
-    if (liveLeft == null && liveRight == null && liveCase == null && liveHeadset == null) return null
-
    val liveDeviceInfo = aap?.deviceInfo

+    if (liveLeft == null && liveRight == null && liveCase == null && liveHeadset == null && liveDeviceInfo == null) {
+        return null
+    }
+
    val newState = CachedDeviceState(
        profileId = pid,
        model = model,
@@ -64,36 +65,34 @@ private fun mergeBatterySlot(
    existing: CachedBatterySlot?,
    now: Instant,
 ): CachedBatterySlot? {
-    if (livePercent == null) return existing
-    if (existing == null) return CachedBatterySlot(livePercent, now)
+    val live: Float = livePercent ?: return existing
+    val current: CachedBatterySlot = existing ?: return CachedBatterySlot(live, now)

-    val isStale = Duration.between(existing.updatedAt, now).abs() > Duration.ofMinutes(1)
-    return if (existing.percent == livePercent && !isStale) existing else CachedBatterySlot(livePercent, now)
+    val isStale = Duration.between(current.updatedAt, now).abs() > Duration.ofMinutes(1)
+    return if (current.percent == live && !isStale) current else CachedBatterySlot(live, now)
 }

 private fun hasStateChanged(old: CachedDeviceState, new: CachedDeviceState): Boolean {
    if (Duration.between(old.lastSeenAt, new.lastSeenAt).abs() > Duration.ofMinutes(1)) return true
-    if (hasSlotTimestampChanged(old.left, new.left)) return true
-    if (hasSlotTimestampChanged(old.right, new.right)) return true
-    if (hasSlotTimestampChanged(old.case, new.case)) return true
-    if (hasSlotTimestampChanged(old.headset, new.headset)) return true
-    return old.left?.percent != new.left?.percent
-        || old.right?.percent != new.right?.percent
-        || old.case?.percent != new.case?.percent
-        || old.headset?.percent != new.headset?.percent
-        || old.isLeftCharging != new.isLeftCharging
+    if (hasSlotChanged(old.left, new.left)) return true
+    if (hasSlotChanged(old.right, new.right)) return true
+    if (hasSlotChanged(old.case, new.case)) return true
+    if (hasSlotChanged(old.headset, new.headset)) return true
+    return old.isLeftCharging != new.isLeftCharging
        || old.isRightCharging != new.isRightCharging
        || old.isCaseCharging != new.isCaseCharging
        || old.isHeadsetCharging != new.isHeadsetCharging
        || old.deviceName != new.deviceName
        || old.serialNumber != new.serialNumber
        || old.firmwareVersion != new.firmwareVersion
+        || old.leftEarbudSerial != new.leftEarbudSerial
+        || old.rightEarbudSerial != new.rightEarbudSerial
+        || old.marketingVersion != new.marketingVersion
 }

-private fun hasSlotTimestampChanged(
-    old: CachedBatterySlot?,
-    new: CachedBatterySlot?,
-): Boolean {
-    if (old == null || new == null) return false
+private fun hasSlotChanged(old: CachedBatterySlot?, new: CachedBatterySlot?): Boolean {
+    if (old == null && new == null) return false
+    if (old == null || new == null) return true
+    if (old.percent != new.percent) return true
    return Duration.between(old.updatedAt, new.updatedAt).abs() > Duration.ofMinutes(1)
 }
@@ -3,6 +3,8 @@ package eu.darken.capod.monitor.core
 import eu.darken.capod.common.TimeSource
 import eu.darken.capod.common.bluetooth.BluetoothAddress
 import eu.darken.capod.common.bluetooth.BluetoothManager2
+import eu.darken.capod.common.debug.Bugs
+import eu.darken.capod.common.debug.autoreport.AutomaticBugReporter
 import eu.darken.capod.monitor.core.aap.AapLifecycleManager
 import eu.darken.capod.monitor.core.ble.BlePodMonitor
 import eu.darken.capod.monitor.core.cache.CachedDeviceState
@@ -22,6 +24,7 @@ import io.mockk.coEvery
 import io.mockk.coVerify
 import io.mockk.every
 import io.mockk.mockk
+import io.mockk.verify
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.collect
@@ -31,6 +34,7 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Test
 import testhelpers.BaseTest
 import testhelpers.TestTimeSource
@@ -536,6 +540,96 @@ class DeviceMonitorTest : BaseTest() {
        devices.size shouldBe 2
    }

+    @AfterEach
+    fun resetBugsReporter() {
+        Bugs.reporter = null
+    }
+
+    /**
+     * Regression: a NPE in the cache merge path used to throw out of `onEach { persistLiveDevices }`,
+     * cancelling the upstream `combine` and freezing every downstream observer (overview, widgets,
+     * etc.) for the rest of the process lifetime. The persist loop now catches and reports.
+     *
+     * Throws from inside `toCachedState` (via a BLE getter) rather than from `save()` so the test
+     * locks in that the catch covers the actual NPE boundary, not just the cache I/O boundary.
+     */
+    @Test
+    fun `flow keeps emitting after persist failure and only reports the bug once per profile`() =
+        runTest(testDispatcher) {
+            val reporter = mockk<AutomaticBugReporter>(relaxed = true)
+            Bugs.reporter = reporter
+
+            val bleFlow = MutableStateFlow(listOf(mockThrowingDualBlePodWithProfile(testProfile)))
+            val aapFlow = MutableStateFlow(emptyMap<BluetoothAddress, AapPodState>())
+            val cacheFlow = MutableStateFlow<Map<String, CachedDeviceState>>(emptyMap())
+            val profilesFlow = MutableStateFlow<List<DeviceProfile>>(listOf(testProfile))
+
+            val blePodMonitor: BlePodMonitor = mockk { every { devices } returns bleFlow }
+            val aapManager: AapConnectionManager = mockk { every { allStates } returns aapFlow }
+            val deviceStateCache: DeviceStateCache = mockk(relaxed = true) {
+                every { cachedStates } returns cacheFlow
+                coEvery { load(any()) } answers { cacheFlow.value[firstArg<String>()] }
+            }
+            val profilesRepo: DeviceProfilesRepo = mockk { every { profiles } returns profilesFlow }
+            val aapLifecycleManager: AapLifecycleManager = mockk(relaxed = true)
+            val bluetoothManager: BluetoothManager2 = mockk {
+                every { connectedDevices } returns MutableStateFlow(emptyList())
+            }
+
+            val monitor = DeviceMonitor(
+                appScope = backgroundScope,
+                blePodMonitor = blePodMonitor,
+                aapManager = aapManager,
+                bluetoothManager = bluetoothManager,
+                deviceStateCache = deviceStateCache,
+                profilesRepo = profilesRepo,
+                aapLifecycleManager = aapLifecycleManager,
+                timeSource = timeSource,
+            )
+
+            val received = mutableListOf<List<PodDevice>>()
+            val collector = backgroundScope.launch {
+                monitor.devices.collect { received += it }
+            }
+            advanceUntilIdle()
+            val initialEmissionCount = received.size
+            initialEmissionCount shouldNotBe 0
+
+            // Trigger more emissions; toCachedState keeps throwing inside the persist loop, but
+            // the flow must survive instead of cancelling its upstream combine.
+            bleFlow.value = listOf(mockThrowingDualBlePodWithProfile(testProfile))
+            advanceUntilIdle()
+            bleFlow.value = listOf(mockThrowingDualBlePodWithProfile(testProfile))
+            advanceUntilIdle()
+
+            // The flow survived: at least two more emissions arrived after the failing merge.
+            (received.size - initialEmissionCount) shouldBe 2
+            // The cache write must NOT have been attempted — the failure was upstream of save().
+            coVerify(exactly = 0) { deviceStateCache.save(any(), any()) }
+            // Dedup: even though merge failed on every emission for the same profile, only one report.
+            verify(exactly = 1) { reporter.notify(any()) }
+
+            collector.cancel()
+        }
+
+    /**
+     * A live BLE pod whose battery getter throws on read. Used to simulate the NPE that R8/JIT
+     * was producing inside the cache merge path — the throw originates inside `toCachedState`,
+     * before `save()` is called.
+     */
+    private fun mockThrowingDualBlePodWithProfile(profile: DeviceProfile): BlePodSnapshot {
+        val bleMeta = object : BlePodSnapshot.Meta {
+            override val profile: DeviceProfile? = profile
+        }
+        return mockk<DualBlePodSnapshot>(relaxed = true) {
+            every { meta } returns bleMeta
+            every { this@mockk.model } returns profile.model
+            every { seenFirstAt } returns Instant.parse("2026-04-05T17:50:00Z")
+            every { seenLastAt } returns Instant.parse("2026-04-05T18:00:00Z")
+            every { batteryLeftPodPercent } throws NullPointerException("synthetic merge failure")
+        }
+    }
+
    @Test
    fun `cache-only refresh does not trigger another persist cycle`() = runTest(testDispatcher) {
        val bleFlow = MutableStateFlow(listOf(mockLiveDualBlePodWithProfile(testProfile)))
@@ -71,4 +71,22 @@ class CachedDeviceStateMigrationTest : BaseTest() {
        val state = json.decodeFromString(CachedDeviceState.serializer(), legacy)
        state.deviceInfo!!.marketingVersion shouldBe "8454480"
    }
+
+    @Test
+    fun `deviceInfo is non-null when only new earbud-serial fields are present`() {
+        val state = CachedDeviceState(
+            profileId = "earbud-only",
+            model = PodModel.AIRPODS_PRO3,
+            leftEarbudSerial = "L-9",
+            rightEarbudSerial = "R-9",
+            marketingVersion = "8888",
+            lastSeenAt = java.time.Instant.ofEpochMilli(1767364074000L),
+        )
+
+        val info = state.deviceInfo!!
+        info.leftEarbudSerial shouldBe "L-9"
+        info.rightEarbudSerial shouldBe "R-9"
+        info.marketingVersion shouldBe "8888"
+        info.name shouldBe ""
+    }
 }
@@ -2,6 +2,8 @@ package eu.darken.capod.monitor.core.cache

 import eu.darken.capod.monitor.core.PodDevice
 import eu.darken.capod.pods.core.apple.PodModel
+import eu.darken.capod.pods.core.apple.aap.AapPodState
+import eu.darken.capod.pods.core.apple.aap.protocol.AapDeviceInfo
 import eu.darken.capod.pods.core.apple.ble.devices.DualApplePods
 import io.kotest.matchers.nulls.shouldBeNull
 import io.kotest.matchers.nulls.shouldNotBeNull
@@ -154,5 +156,119 @@ class ToCachedStateTest : BaseTest() {
                .toCachedState(existing, now)
                .shouldNotBeNull()
        }
+
+        @Test
+        fun `returns new state when only earbud serial changes`() {
+            val existing = CachedDeviceState(
+                profileId = "test-profile",
+                model = PodModel.AIRPODS_PRO3,
+                left = CachedDeviceState.CachedBatterySlot(0.8f, now),
+                right = CachedDeviceState.CachedBatterySlot(0.7f, now),
+                case = CachedDeviceState.CachedBatterySlot(0.5f, now),
+                isLeftCharging = false,
+                isRightCharging = false,
+                isCaseCharging = false,
+                isHeadsetCharging = false,
+                deviceName = "AirPods",
+                serialNumber = "",
+                firmwareVersion = "",
+                leftEarbudSerial = "OLD-LEFT",
+                rightEarbudSerial = "R-1",
+                marketingVersion = "1234",
+                lastSeenAt = now,
+            )
+            val device = PodDevice(
+                profileId = "test-profile",
+                ble = mockDualPod(leftBattery = 0.8f, rightBattery = 0.7f, caseBattery = 0.5f),
+                aap = AapPodState(
+                    deviceInfo = deviceInfo(
+                        leftEarbudSerial = "NEW-LEFT",
+                        rightEarbudSerial = "R-1",
+                        marketingVersion = "1234",
+                    ),
+                ),
+            )
+
+            val result = device.toCachedState(existing, now).shouldNotBeNull()
+            result.leftEarbudSerial shouldBe "NEW-LEFT"
+            result.rightEarbudSerial shouldBe "R-1"
+            result.marketingVersion shouldBe "1234"
+        }
+
+        @Test
+        fun `returns new state when only marketing version changes`() {
+            val existing = CachedDeviceState(
+                profileId = "test-profile",
+                model = PodModel.AIRPODS_PRO3,
+                left = CachedDeviceState.CachedBatterySlot(0.8f, now),
+                isLeftCharging = false,
+                isRightCharging = false,
+                isCaseCharging = false,
+                isHeadsetCharging = false,
+                deviceName = "AirPods",
+                serialNumber = "",
+                firmwareVersion = "",
+                marketingVersion = "1234",
+                lastSeenAt = now,
+            )
+            val device = PodDevice(
+                profileId = "test-profile",
+                ble = mockDualPod(leftBattery = 0.8f),
+                aap = AapPodState(deviceInfo = deviceInfo(marketingVersion = "5678")),
+            )
+
+            val result = device.toCachedState(existing, now).shouldNotBeNull()
+            result.marketingVersion shouldBe "5678"
+        }
    }
+
+    @Nested
+    inner class DeviceInfoOnly {
+
+        @Test
+        fun `persists DeviceInfo when no live battery is available`() {
+            // Mirror how DeviceMonitor.aapOnlyForPersistence builds an AAP-only PodDevice:
+            // profileModel and profileAddress carry the model/address since `ble` is null.
+            val device = PodDevice(
+                profileId = "test-profile",
+                ble = null,
+                aap = AapPodState(
+                    deviceInfo = deviceInfo(
+                        name = "Pro 3",
+                        leftEarbudSerial = "L-1",
+                        rightEarbudSerial = "R-1",
+                        marketingVersion = "9999",
+                    ),
+                ),
+                profileModel = PodModel.AIRPODS_PRO3,
+                profileAddress = "AA:BB:CC:DD:EE:FF",
+            )
+
+            val result = device.toCachedState(existing = null, now = now).shouldNotBeNull()
+            result.model shouldBe PodModel.AIRPODS_PRO3
+            result.address shouldBe "AA:BB:CC:DD:EE:FF"
+            result.deviceName shouldBe "Pro 3"
+            result.leftEarbudSerial shouldBe "L-1"
+            result.rightEarbudSerial shouldBe "R-1"
+            result.marketingVersion shouldBe "9999"
+        }
+    }
+
+    private fun deviceInfo(
+        name: String = "AirPods",
+        serialNumber: String = "",
+        firmwareVersion: String = "",
+        leftEarbudSerial: String? = null,
+        rightEarbudSerial: String? = null,
+        marketingVersion: String? = null,
+    ) = AapDeviceInfo(
+        name = name,
+        modelNumber = "",
+        manufacturer = "",
+        serialNumber = serialNumber,
+        firmwareVersion = firmwareVersion,
+        leftEarbudSerial = leftEarbudSerial,
+        rightEarbudSerial = rightEarbudSerial,
+        marketingVersion = marketingVersion,
+    )
 }