From 71ba531025c61eaa30eb58f6c9d59e77cf820c03 Mon Sep 17 00:00:00 2001 From: darken Date: Mon, 13 Apr 2026 17:32:38 +0200 Subject: [PATCH] Break cache feedback loops in monitor and reactions --- .../capod/common/bluetooth/BleLogSummaries.kt | 46 +++++++++++ .../capod/monitor/core/DeviceMonitor.kt | 39 +++++++-- .../core/cache/DeviceStateCacheExtensions.kt | 20 ++++- .../monitor/core/worker/MonitorService.kt | 54 +++++++++++- .../pods/core/apple/ble/PodLogSummaries.kt | 48 +++++++++++ .../reaction/core/playpause/PlayPause.kt | 34 +++++++- .../capod/monitor/core/DeviceMonitorTest.kt | 82 +++++++++++++++++++ 7 files changed, 307 insertions(+), 16 deletions(-) create mode 100644 app/src/main/java/eu/darken/capod/common/bluetooth/BleLogSummaries.kt create mode 100644 app/src/main/java/eu/darken/capod/pods/core/apple/ble/PodLogSummaries.kt diff --git a/app/src/main/java/eu/darken/capod/common/bluetooth/BleLogSummaries.kt b/app/src/main/java/eu/darken/capod/common/bluetooth/BleLogSummaries.kt new file mode 100644 index 00000000..1ac670ca --- /dev/null +++ b/app/src/main/java/eu/darken/capod/common/bluetooth/BleLogSummaries.kt @@ -0,0 +1,46 @@ +package eu.darken.capod.common.bluetooth + +import android.bluetooth.le.ScanResult +import androidx.core.util.forEach + +fun BluetoothAddress.redactedForLogs(): String { + val parts = split(':') + if (parts.size < 2) return takeLast(5) + return "XX:XX:XX:XX:${parts.takeLast(2).joinToString(":")}" +} + +fun BleScanResult.logSummary(): String { + val payloadSummary = manufacturerSpecificData.entries + .sortedBy { it.key } + .joinToString(separator = ",") { (manufacturerId, data) -> "$manufacturerId:${data.size}B" } + .ifEmpty { "-" } + return "addr=${address.redactedForLogs()}, rssi=$rssi, payloads=[$payloadSummary]" +} + +@JvmName("logBleScanResultCollectionSummary") +fun Collection.logSummary(limit: Int = 3): String { + if (isEmpty()) return "count=0" + val sample = take(limit).joinToString(separator = ", ") { it.logSummary() } + val suffix = if (size > limit) ", ..." else "" + return "count=$size, sample=[$sample$suffix]" +} + +fun ScanResult.logSummary(): String { + val payloadSummary = buildList { + scanRecord?.manufacturerSpecificData?.forEach { manufacturerId, data -> + add("$manufacturerId:${data.size}B") + } + } + .sorted() + .joinToString(separator = ",") + .ifEmpty { "-" } + return "addr=${device.address.redactedForLogs()}, rssi=$rssi, payloads=[$payloadSummary]" +} + +@JvmName("logFrameworkScanResultCollectionSummary") +fun Collection.logSummary(limit: Int = 3): String { + if (isEmpty()) return "count=0" + val sample = take(limit).joinToString(separator = ", ") { it.logSummary() } + val suffix = if (size > limit) ", ..." else "" + return "count=$size, sample=[$sample$suffix]" +} diff --git a/app/src/main/java/eu/darken/capod/monitor/core/DeviceMonitor.kt b/app/src/main/java/eu/darken/capod/monitor/core/DeviceMonitor.kt index 0abfe9af..d8c55fb1 100644 --- a/app/src/main/java/eu/darken/capod/monitor/core/DeviceMonitor.kt +++ b/app/src/main/java/eu/darken/capod/monitor/core/DeviceMonitor.kt @@ -49,15 +49,14 @@ class DeviceMonitor @Inject constructor( aapLifecycleManager.start() } - val devices: Flow> = combine( + private val liveState: Flow = combine( blePodMonitor.devices, aapManager.allStates, - deviceStateCache.cachedStates, profilesRepo.profiles, - ) { pods, aapStates, cachedStates, profiles -> + ) { pods, aapStates, profiles -> val profilesById = profiles.associateBy { it.id } - // Live devices — BLE + AAP + cached fallback for missing fields + // Live devices — BLE + AAP. Cache is merged later so cache writes don't feed back here. val liveDevices = pods.map { pod -> val profile = pod.meta.profile?.id?.let(profilesById::get) PodDevice( @@ -65,7 +64,6 @@ class DeviceMonitor @Inject constructor( label = profile?.label, ble = pod, aap = aapStates.forProfile(profile), - cached = profile?.id?.let { cachedStates[it] }, profileAddress = profile?.address, profileModel = profile?.model, profileKeyState = profile.toBleKeyState(), @@ -73,6 +71,25 @@ class DeviceMonitor @Inject constructor( ) } + LiveMergeState( + liveDevices = liveDevices, + profiles = profiles, + aapStates = aapStates, + ) + }.onEach { liveState -> + persistLiveDevices(liveState.liveDevices) + } + + val devices: Flow> = combine( + liveState, + deviceStateCache.cachedStates, + ) { liveState, cachedStates -> + val liveDevices = liveState.liveDevices.map { device -> + device.copy(cached = device.profileId?.let { cachedStates[it] }) + } + val profiles = liveState.profiles + val aapStates = liveState.aapStates + // Collapse live duplicates sharing an identity-backed profile — e.g. when the legacy // signal-quality fallback misattributed ambient strangers to our profile. // Only applied when the group has at least one IRK-verified candidate; for no-IRK @@ -149,21 +166,25 @@ class DeviceMonitor @Inject constructor( } dedupedLiveDevices + nonLiveDevices - } - .onEach { devices -> persistLiveDevices(devices) } - .replayingShare(appScope) + }.replayingShare(appScope) private suspend fun persistLiveDevices(devices: List) { for (device in devices) { val profileId = device.profileId ?: continue val existing = deviceStateCache.cachedStates.value[profileId] - val newState = device.toCachedState(existing, timeSource.now()) ?: continue + val newState = device.copy(cached = existing).toCachedState(existing, timeSource.now()) ?: continue log(TAG, VERBOSE) { "Persisting state for $profileId" } deviceStateCache.save(profileId, newState) } } + private data class LiveMergeState( + val liveDevices: List, + val profiles: List, + val aapStates: Map, + ) + suspend fun getDeviceForProfile(profileId: String): PodDevice? { log(TAG) { "getDeviceForProfile(profileId=$profileId)" } diff --git a/app/src/main/java/eu/darken/capod/monitor/core/cache/DeviceStateCacheExtensions.kt b/app/src/main/java/eu/darken/capod/monitor/core/cache/DeviceStateCacheExtensions.kt index cefbf599..73040e9a 100644 --- a/app/src/main/java/eu/darken/capod/monitor/core/cache/DeviceStateCacheExtensions.kt +++ b/app/src/main/java/eu/darken/capod/monitor/core/cache/DeviceStateCacheExtensions.kt @@ -37,10 +37,10 @@ fun PodDevice.toCachedState( profileId = pid, model = model, address = address, - left = liveLeft?.let { CachedBatterySlot(it, now) } ?: existing?.left, - right = liveRight?.let { CachedBatterySlot(it, now) } ?: existing?.right, - case = liveCase?.let { CachedBatterySlot(it, now) } ?: existing?.case, - headset = liveHeadset?.let { CachedBatterySlot(it, now) } ?: existing?.headset, + left = mergeBatterySlot(liveLeft, existing?.left, now), + right = mergeBatterySlot(liveRight, existing?.right, now), + case = mergeBatterySlot(liveCase, existing?.case, now), + headset = mergeBatterySlot(liveHeadset, existing?.headset, now), isLeftCharging = isLeftPodCharging, isRightCharging = isRightPodCharging, isCaseCharging = isCaseCharging, @@ -56,6 +56,18 @@ fun PodDevice.toCachedState( return newState } +private fun mergeBatterySlot( + livePercent: Float?, + existing: CachedBatterySlot?, + now: Instant, +): CachedBatterySlot? { + if (livePercent == null) return existing + if (existing == null) return CachedBatterySlot(livePercent, now) + + val isStale = Duration.between(existing.updatedAt, now).abs() > Duration.ofMinutes(1) + return if (existing.percent == livePercent && !isStale) existing else CachedBatterySlot(livePercent, 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 diff --git a/app/src/main/java/eu/darken/capod/monitor/core/worker/MonitorService.kt b/app/src/main/java/eu/darken/capod/monitor/core/worker/MonitorService.kt index cff0cf0c..8a1eb4b9 100644 --- a/app/src/main/java/eu/darken/capod/monitor/core/worker/MonitorService.kt +++ b/app/src/main/java/eu/darken/capod/monitor/core/worker/MonitorService.kt @@ -29,9 +29,11 @@ import eu.darken.capod.main.core.MonitorMode import eu.darken.capod.main.core.PermissionTool import eu.darken.capod.monitor.core.DeviceMonitor import eu.darken.capod.monitor.core.MonitorCoroutineScope +import eu.darken.capod.monitor.core.PodDevice import eu.darken.capod.monitor.core.ble.BlePodMonitor import eu.darken.capod.monitor.core.primaryDevice import eu.darken.capod.monitor.ui.MonitorNotifications +import eu.darken.capod.pods.core.apple.PodModel import eu.darken.capod.pods.core.apple.aap.AapConnectionManager import eu.darken.capod.profiles.core.DeviceProfile import eu.darken.capod.profiles.core.DeviceProfilesRepo @@ -46,7 +48,7 @@ import kotlinx.coroutines.cancelChildren import kotlinx.coroutines.delay import kotlinx.coroutines.flow.catch import kotlinx.coroutines.flow.combine -import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.distinctUntilChangedBy import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.flatMapLatest @@ -189,7 +191,7 @@ class MonitorService : Service() { val monitorJob = deviceMonitor.primaryDevice() .setupCommonEventHandlers(TAG) { "BlePodMonitor" } - .distinctUntilChanged() + .distinctUntilChangedBy { it?.toNotificationKey() } .throttleLatest(1000) .onEach { currentDevice -> val useExtraNotification = generalSettings.useExtraMonitorNotification.valueBlocking @@ -337,3 +339,51 @@ class MonitorService : Service() { } } } + +private data class NotificationDeviceKey( + val profileId: String?, + val label: String?, + val model: PodModel, + val hasDualPods: Boolean, + val hasCase: Boolean, + val hasEarDetection: Boolean, + val batteryLeft: Float?, + val batteryRight: Float?, + val batteryCase: Float?, + val batteryHeadset: Float?, + val isLeftPodCharging: Boolean?, + val isRightPodCharging: Boolean?, + val isCaseCharging: Boolean?, + val isHeadsetBeingCharged: Boolean?, + val isLeftInEar: Boolean?, + val isRightInEar: Boolean?, + val isBeingWorn: Boolean?, + val iconRes: Int, + val leftPodIcon: Int, + val rightPodIcon: Int, + val caseIcon: Int, +) + +private fun PodDevice.toNotificationKey(): NotificationDeviceKey = NotificationDeviceKey( + profileId = profileId, + label = label, + model = model, + hasDualPods = hasDualPods, + hasCase = hasCase, + hasEarDetection = hasEarDetection, + batteryLeft = batteryLeft, + batteryRight = batteryRight, + batteryCase = batteryCase, + batteryHeadset = batteryHeadset, + isLeftPodCharging = isLeftPodCharging, + isRightPodCharging = isRightPodCharging, + isCaseCharging = isCaseCharging, + isHeadsetBeingCharged = isHeadsetBeingCharged, + isLeftInEar = isLeftInEar, + isRightInEar = isRightInEar, + isBeingWorn = isBeingWorn, + iconRes = iconRes, + leftPodIcon = leftPodIcon, + rightPodIcon = rightPodIcon, + caseIcon = caseIcon, +) diff --git a/app/src/main/java/eu/darken/capod/pods/core/apple/ble/PodLogSummaries.kt b/app/src/main/java/eu/darken/capod/pods/core/apple/ble/PodLogSummaries.kt new file mode 100644 index 00000000..7c67f2da --- /dev/null +++ b/app/src/main/java/eu/darken/capod/pods/core/apple/ble/PodLogSummaries.kt @@ -0,0 +1,48 @@ +package eu.darken.capod.pods.core.apple.ble + +import eu.darken.capod.common.bluetooth.redactedForLogs +import eu.darken.capod.pods.core.apple.ble.history.KnownDevice +import eu.darken.capod.profiles.core.DeviceProfile +import java.util.Locale + +private fun String.shortIdForLogs(): String = take(8) + +fun DeviceProfile.logSummary(): String = buildString { + append("profile(id=") + append(id.shortIdForLogs()) + append(", model=") + append(model) + append(", addr=") + append(address?.redactedForLogs() ?: "-") + append(')') +} + +fun BlePodSnapshot.logSummary(): String = buildString { + append("pod(model=") + append(model) + append(", id=") + append(identifier.toString().shortIdForLogs()) + append(", profile=") + append(meta.profile?.id?.shortIdForLogs() ?: "-") + append(", addr=") + append(address.redactedForLogs()) + append(", rssi=") + append(rssi) + append(", reliability=") + append(String.format(Locale.US, "%.2f", reliability)) + append(')') +} + +fun KnownDevice.logSummary(): String = buildString { + append("known(id=") + append(id.toString().shortIdForLogs()) + append(", model=") + append(history.last().model) + append(", history=") + append(history.size) + append(", seen=") + append(seenCounter) + append(", addr=") + append(lastAddress.redactedForLogs()) + append(')') +} diff --git a/app/src/main/java/eu/darken/capod/reaction/core/playpause/PlayPause.kt b/app/src/main/java/eu/darken/capod/reaction/core/playpause/PlayPause.kt index e42a4958..330ee7c5 100644 --- a/app/src/main/java/eu/darken/capod/reaction/core/playpause/PlayPause.kt +++ b/app/src/main/java/eu/darken/capod/reaction/core/playpause/PlayPause.kt @@ -9,8 +9,10 @@ import eu.darken.capod.common.debug.logging.logTag import eu.darken.capod.common.flow.setupCommonEventHandlers import eu.darken.capod.common.flow.withPrevious import eu.darken.capod.monitor.core.DeviceMonitor +import eu.darken.capod.monitor.core.PodDevice import eu.darken.capod.monitor.core.primaryDevice import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.distinctUntilChangedBy import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.flow.filter import kotlinx.coroutines.flow.flatMapLatest @@ -50,7 +52,8 @@ class PlayPause @Inject constructor( deviceMonitor.primaryDevice() } } - .distinctUntilChanged() + // Cache persistence can update battery timestamps without changing any reaction-relevant state. + .distinctUntilChangedBy { it?.toPlayPauseMonitorKey() } .withPrevious() .filter { (previous, current) -> if (previous == null || current == null) return@filter false @@ -363,6 +366,35 @@ class PlayPause @Inject constructor( val stagedConfirmation: Boolean, ) + internal data class PlayPauseMonitorKey( + val profileId: String?, + val autoPlay: Boolean, + val autoPause: Boolean, + val onePodMode: Boolean, + val hasEarDetection: Boolean, + val hasDualPods: Boolean, + val leftInEar: Boolean?, + val rightInEar: Boolean?, + val isBeingWorn: Boolean?, + val hasAapEarDetection: Boolean, + val hasBleSnapshot: Boolean, + ) + + internal fun PodDevice.toPlayPauseMonitorKey(): PlayPauseMonitorKey = + PlayPauseMonitorKey( + profileId = profileId, + autoPlay = reactions.autoPlay, + autoPause = reactions.autoPause, + onePodMode = reactions.onePodMode, + hasEarDetection = hasEarDetection, + hasDualPods = hasDualPods, + leftInEar = isLeftInEar, + rightInEar = isRightInEar, + isBeingWorn = isBeingWorn, + hasAapEarDetection = aap?.aapEarDetection != null, + hasBleSnapshot = ble != null, + ) + companion object { private val TAG = logTag("Reaction", "PlayPause") } diff --git a/app/src/test/java/eu/darken/capod/monitor/core/DeviceMonitorTest.kt b/app/src/test/java/eu/darken/capod/monitor/core/DeviceMonitorTest.kt index e088d9a6..411d6251 100644 --- a/app/src/test/java/eu/darken/capod/monitor/core/DeviceMonitorTest.kt +++ b/app/src/test/java/eu/darken/capod/monitor/core/DeviceMonitorTest.kt @@ -10,6 +10,7 @@ import eu.darken.capod.pods.core.apple.PodModel import eu.darken.capod.pods.core.apple.aap.AapConnectionManager import eu.darken.capod.pods.core.apple.aap.AapPodState import eu.darken.capod.pods.core.apple.ble.BlePodSnapshot +import eu.darken.capod.pods.core.apple.ble.DualBlePodSnapshot import eu.darken.capod.pods.core.apple.ble.devices.ApplePods import eu.darken.capod.profiles.core.AppleDeviceProfile import eu.darken.capod.profiles.core.DeviceProfile @@ -17,12 +18,17 @@ import eu.darken.capod.profiles.core.DeviceProfilesRepo import io.kotest.matchers.shouldBe import io.kotest.matchers.shouldNotBe import io.mockk.coEvery +import io.mockk.coVerify import io.mockk.every import io.mockk.mockk import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.Job +import kotlinx.coroutines.flow.collect import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.UnconfinedTestDispatcher import kotlinx.coroutines.test.runTest import org.junit.jupiter.api.Test @@ -107,6 +113,24 @@ class DeviceMonitorTest : BaseTest() { } } + private fun mockLiveDualBlePodWithProfile( + profile: DeviceProfile, + batteryLeft: Float = 0.9f, + batteryRight: Float = 1.0f, + ): BlePodSnapshot { + val bleMeta = object : BlePodSnapshot.Meta { + override val profile: DeviceProfile? = profile + } + return mockk(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 } returns batteryLeft + every { batteryRightPodPercent } returns batteryRight + } + } + private fun createMonitor( ble: List = emptyList(), aap: Map = emptyMap(), @@ -508,4 +532,62 @@ class DeviceMonitorTest : BaseTest() { devices.size shouldBe 2 } + + @Test + fun `cache-only refresh does not trigger another persist cycle`() = runTest(testDispatcher) { + val bleFlow = MutableStateFlow(listOf(mockLiveDualBlePodWithProfile(testProfile))) + val aapFlow = MutableStateFlow(emptyMap()) + val cacheFlow = MutableStateFlow>(emptyMap()) + val profilesFlow = MutableStateFlow>(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()] } + coEvery { save(any(), any()) } answers { + val profileId = firstArg() + val state = secondArg() + cacheFlow.value += (profileId to state) + Unit + } + } + val profilesRepo: DeviceProfilesRepo = mockk { + every { profiles } returns profilesFlow + } + val aapLifecycleManager: AapLifecycleManager = mockk(relaxed = true) + + val monitor = DeviceMonitor( + appScope = backgroundScope, + blePodMonitor = blePodMonitor, + aapManager = aapManager, + deviceStateCache = deviceStateCache, + profilesRepo = profilesRepo, + aapLifecycleManager = aapLifecycleManager, + timeSource = timeSource, + ) + + val collector: Job = backgroundScope.launch { + monitor.devices.collect { } + } + advanceUntilIdle() + + coVerify(exactly = 1) { deviceStateCache.save(testProfile.id, any()) } + + val persisted = cacheFlow.value.getValue(testProfile.id) + cacheFlow.value += testProfile.id to persisted.copy( + left = CachedDeviceState.CachedBatterySlot( + percent = 0.1f, + updatedAt = persisted.left?.updatedAt ?: Instant.parse("2026-04-05T18:00:00Z"), + ), + ) + advanceUntilIdle() + + coVerify(exactly = 1) { deviceStateCache.save(testProfile.id, any()) } + collector.cancel() + } }