Break cache feedback loops in monitor and reactions

This commit is contained in:
darken
2026-04-14 11:28:41 +02:00
committed by Matthias Urhahn
parent 6825abaa41
commit 71ba531025
7 changed files with 307 additions and 16 deletions
@@ -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<BleScanResult>.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<ScanResult>.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]"
}
@@ -49,15 +49,14 @@ class DeviceMonitor @Inject constructor(
aapLifecycleManager.start() aapLifecycleManager.start()
} }
val devices: Flow<List<PodDevice>> = combine( private val liveState: Flow<LiveMergeState> = combine(
blePodMonitor.devices, blePodMonitor.devices,
aapManager.allStates, aapManager.allStates,
deviceStateCache.cachedStates,
profilesRepo.profiles, profilesRepo.profiles,
) { pods, aapStates, cachedStates, profiles -> ) { pods, aapStates, profiles ->
val profilesById = profiles.associateBy { it.id } 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 liveDevices = pods.map { pod ->
val profile = pod.meta.profile?.id?.let(profilesById::get) val profile = pod.meta.profile?.id?.let(profilesById::get)
PodDevice( PodDevice(
@@ -65,7 +64,6 @@ class DeviceMonitor @Inject constructor(
label = profile?.label, label = profile?.label,
ble = pod, ble = pod,
aap = aapStates.forProfile(profile), aap = aapStates.forProfile(profile),
cached = profile?.id?.let { cachedStates[it] },
profileAddress = profile?.address, profileAddress = profile?.address,
profileModel = profile?.model, profileModel = profile?.model,
profileKeyState = profile.toBleKeyState(), 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<List<PodDevice>> = 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 // Collapse live duplicates sharing an identity-backed profile — e.g. when the legacy
// signal-quality fallback misattributed ambient strangers to our profile. // signal-quality fallback misattributed ambient strangers to our profile.
// Only applied when the group has at least one IRK-verified candidate; for no-IRK // 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 dedupedLiveDevices + nonLiveDevices
} }.replayingShare(appScope)
.onEach { devices -> persistLiveDevices(devices) }
.replayingShare(appScope)
private suspend fun persistLiveDevices(devices: List<PodDevice>) { private suspend fun persistLiveDevices(devices: List<PodDevice>) {
for (device in devices) { for (device in devices) {
val profileId = device.profileId ?: continue val profileId = device.profileId ?: continue
val existing = deviceStateCache.cachedStates.value[profileId] 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" } log(TAG, VERBOSE) { "Persisting state for $profileId" }
deviceStateCache.save(profileId, newState) deviceStateCache.save(profileId, newState)
} }
} }
private data class LiveMergeState(
val liveDevices: List<PodDevice>,
val profiles: List<DeviceProfile>,
val aapStates: Map<BluetoothAddress, AapPodState>,
)
suspend fun getDeviceForProfile(profileId: String): PodDevice? { suspend fun getDeviceForProfile(profileId: String): PodDevice? {
log(TAG) { "getDeviceForProfile(profileId=$profileId)" } log(TAG) { "getDeviceForProfile(profileId=$profileId)" }
@@ -37,10 +37,10 @@ fun PodDevice.toCachedState(
profileId = pid, profileId = pid,
model = model, model = model,
address = address, address = address,
left = liveLeft?.let { CachedBatterySlot(it, now) } ?: existing?.left, left = mergeBatterySlot(liveLeft, existing?.left, now),
right = liveRight?.let { CachedBatterySlot(it, now) } ?: existing?.right, right = mergeBatterySlot(liveRight, existing?.right, now),
case = liveCase?.let { CachedBatterySlot(it, now) } ?: existing?.case, case = mergeBatterySlot(liveCase, existing?.case, now),
headset = liveHeadset?.let { CachedBatterySlot(it, now) } ?: existing?.headset, headset = mergeBatterySlot(liveHeadset, existing?.headset, now),
isLeftCharging = isLeftPodCharging, isLeftCharging = isLeftPodCharging,
isRightCharging = isRightPodCharging, isRightCharging = isRightPodCharging,
isCaseCharging = isCaseCharging, isCaseCharging = isCaseCharging,
@@ -56,6 +56,18 @@ fun PodDevice.toCachedState(
return newState 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 { private fun hasStateChanged(old: CachedDeviceState, new: CachedDeviceState): Boolean {
if (Duration.between(old.lastSeenAt, new.lastSeenAt).abs() > Duration.ofMinutes(1)) return true if (Duration.between(old.lastSeenAt, new.lastSeenAt).abs() > Duration.ofMinutes(1)) return true
if (hasSlotTimestampChanged(old.left, new.left)) return true if (hasSlotTimestampChanged(old.left, new.left)) return true
@@ -29,9 +29,11 @@ import eu.darken.capod.main.core.MonitorMode
import eu.darken.capod.main.core.PermissionTool import eu.darken.capod.main.core.PermissionTool
import eu.darken.capod.monitor.core.DeviceMonitor import eu.darken.capod.monitor.core.DeviceMonitor
import eu.darken.capod.monitor.core.MonitorCoroutineScope 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.ble.BlePodMonitor
import eu.darken.capod.monitor.core.primaryDevice import eu.darken.capod.monitor.core.primaryDevice
import eu.darken.capod.monitor.ui.MonitorNotifications 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.pods.core.apple.aap.AapConnectionManager
import eu.darken.capod.profiles.core.DeviceProfile import eu.darken.capod.profiles.core.DeviceProfile
import eu.darken.capod.profiles.core.DeviceProfilesRepo import eu.darken.capod.profiles.core.DeviceProfilesRepo
@@ -46,7 +48,7 @@ import kotlinx.coroutines.cancelChildren
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.catch import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.distinctUntilChangedBy
import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.flow.emptyFlow
import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flatMapLatest import kotlinx.coroutines.flow.flatMapLatest
@@ -189,7 +191,7 @@ class MonitorService : Service() {
val monitorJob = deviceMonitor.primaryDevice() val monitorJob = deviceMonitor.primaryDevice()
.setupCommonEventHandlers(TAG) { "BlePodMonitor" } .setupCommonEventHandlers(TAG) { "BlePodMonitor" }
.distinctUntilChanged() .distinctUntilChangedBy { it?.toNotificationKey() }
.throttleLatest(1000) .throttleLatest(1000)
.onEach { currentDevice -> .onEach { currentDevice ->
val useExtraNotification = generalSettings.useExtraMonitorNotification.valueBlocking 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,
)
@@ -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(')')
}
@@ -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.setupCommonEventHandlers
import eu.darken.capod.common.flow.withPrevious import eu.darken.capod.common.flow.withPrevious
import eu.darken.capod.monitor.core.DeviceMonitor import eu.darken.capod.monitor.core.DeviceMonitor
import eu.darken.capod.monitor.core.PodDevice
import eu.darken.capod.monitor.core.primaryDevice import eu.darken.capod.monitor.core.primaryDevice
import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.distinctUntilChangedBy
import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.flow.emptyFlow
import kotlinx.coroutines.flow.filter import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flatMapLatest import kotlinx.coroutines.flow.flatMapLatest
@@ -50,7 +52,8 @@ class PlayPause @Inject constructor(
deviceMonitor.primaryDevice() deviceMonitor.primaryDevice()
} }
} }
.distinctUntilChanged() // Cache persistence can update battery timestamps without changing any reaction-relevant state.
.distinctUntilChangedBy { it?.toPlayPauseMonitorKey() }
.withPrevious() .withPrevious()
.filter { (previous, current) -> .filter { (previous, current) ->
if (previous == null || current == null) return@filter false if (previous == null || current == null) return@filter false
@@ -363,6 +366,35 @@ class PlayPause @Inject constructor(
val stagedConfirmation: Boolean, 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 { companion object {
private val TAG = logTag("Reaction", "PlayPause") private val TAG = logTag("Reaction", "PlayPause")
} }
@@ -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.AapConnectionManager
import eu.darken.capod.pods.core.apple.aap.AapPodState 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.BlePodSnapshot
import eu.darken.capod.pods.core.apple.ble.DualBlePodSnapshot
import eu.darken.capod.pods.core.apple.ble.devices.ApplePods import eu.darken.capod.pods.core.apple.ble.devices.ApplePods
import eu.darken.capod.profiles.core.AppleDeviceProfile import eu.darken.capod.profiles.core.AppleDeviceProfile
import eu.darken.capod.profiles.core.DeviceProfile 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.shouldBe
import io.kotest.matchers.shouldNotBe import io.kotest.matchers.shouldNotBe
import io.mockk.coEvery import io.mockk.coEvery
import io.mockk.coVerify
import io.mockk.every import io.mockk.every
import io.mockk.mockk import io.mockk.mockk
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.first
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.advanceUntilIdle
import kotlinx.coroutines.test.UnconfinedTestDispatcher import kotlinx.coroutines.test.UnconfinedTestDispatcher
import kotlinx.coroutines.test.runTest import kotlinx.coroutines.test.runTest
import org.junit.jupiter.api.Test 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<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 } returns batteryLeft
every { batteryRightPodPercent } returns batteryRight
}
}
private fun createMonitor( private fun createMonitor(
ble: List<BlePodSnapshot> = emptyList(), ble: List<BlePodSnapshot> = emptyList(),
aap: Map<BluetoothAddress, AapPodState> = emptyMap(), aap: Map<BluetoothAddress, AapPodState> = emptyMap(),
@@ -508,4 +532,62 @@ class DeviceMonitorTest : BaseTest() {
devices.size shouldBe 2 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<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>()] }
coEvery { save(any(), any()) } answers {
val profileId = firstArg<String>()
val state = secondArg<CachedDeviceState>()
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()
}
} }