diff --git a/app/src/foss/java/eu/darken/capod/common/upgrade/core/UpgradeRepoFoss.kt b/app/src/foss/java/eu/darken/capod/common/upgrade/core/UpgradeRepoFoss.kt index f5f64cca..46fe0f25 100644 --- a/app/src/foss/java/eu/darken/capod/common/upgrade/core/UpgradeRepoFoss.kt +++ b/app/src/foss/java/eu/darken/capod/common/upgrade/core/UpgradeRepoFoss.kt @@ -3,15 +3,20 @@ package eu.darken.capod.common.upgrade.core import eu.darken.capod.common.WebpageTool import eu.darken.capod.common.coroutine.AppScope import eu.darken.capod.common.debug.logging.Logging.Priority.WARN +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.setupCommonEventHandlers import eu.darken.capod.common.upgrade.UpgradeRepo +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.SharingStarted -import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.catch +import kotlinx.coroutines.flow.flatMapLatest +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.shareIn import java.time.Instant import java.util.UUID @@ -31,20 +36,39 @@ class UpgradeRepoFoss @Inject constructor( private val refreshTrigger = MutableStateFlow(UUID.randomUUID()) - override val upgradeInfo: Flow = combine( - fossCache.upgrade.flow, - refreshTrigger - ) { data, _ -> - if (data == null) { - Info() - } else { - Info( - isPro = true, - upgradedAt = data.upgradedAt, - upgradeReason = data.reason, - ) + // Written only from the sharing coroutine (single collector) — no synchronization needed. + private var lastKnownInfo: Info? = null + + override val upgradeInfo: Flow = refreshTrigger + .flatMapLatest { + fossCache.upgrade.flow + .map { data -> + if (data == null) { + Info() + } else { + Info( + isPro = true, + upgradedAt = data.upgradedAt, + upgradeReason = data.reason, + ) + } + } + .catch { e -> + // A SharedFlow cannot fail: without this, a thrown cache read dies inside + // shareIn's sharing coroutine and every collector hangs forever (VM state stuck + // on Loading, checkSponsorReturn suspended mid-unlock). The catch sits INSIDE + // flatMapLatest so the error completes only this inner subscription — refresh() + // resubscribes the cache and recovery stays possible. Last-known preservation: + // a late read failure must not revoke an entitlement we already saw; the error + // rides on the previous Info instead. Contrast: gplay keeps a retryWhen loop + // because billing re-settles in-place — the FOSS read is a local one-shot, and + // refresh-driven resubscription IS the retry. + if (e is CancellationException) throw e + log(TAG, WARN) { "upgradeInfo read failed: ${e.asLog()}" } + emit((lastKnownInfo ?: Info()).copy(error = e)) + } } - } + .onEach { if (it.error == null) lastKnownInfo = it } .setupCommonEventHandlers(TAG) { "upgradeInfo" } .shareIn(appScope, SharingStarted.WhileSubscribed(3000L, 0L), replay = 1) @@ -68,6 +92,8 @@ class UpgradeRepoFoss @Inject constructor( * decode reads as null and therefore counts as ABSENT to this transaction, i.e. it gets * replaced. That matches the pre-existing read behaviour — such a record already presents the * user as free — and re-creating it on the next successful sponsor visit is the recovery path. + * Decode failures therefore fall back to absent by design (the flag), so the error path around + * [upgradeInfo] covers IO/corruption throws, not schema mismatches. * * @return true if a new record was created, false if an existing record was kept. */ @@ -79,6 +105,9 @@ class UpgradeRepoFoss @Inject constructor( reason = FossUpgrade.Reason.DONATED, ) } + // A returned transaction proves the store is readable again: revive a possibly error-stuck + // inner flow so the record propagates to collectors still holding the error replay. + refresh() return if (updated.old == null) { true } else { diff --git a/app/src/testFoss/java/eu/darken/capod/common/upgrade/core/UpgradeRepoFossTest.kt b/app/src/testFoss/java/eu/darken/capod/common/upgrade/core/UpgradeRepoFossTest.kt index 35459148..d0436367 100644 --- a/app/src/testFoss/java/eu/darken/capod/common/upgrade/core/UpgradeRepoFossTest.kt +++ b/app/src/testFoss/java/eu/darken/capod/common/upgrade/core/UpgradeRepoFossTest.kt @@ -1,12 +1,32 @@ package eu.darken.capod.common.upgrade.core +import eu.darken.capod.common.datastore.DataStoreValue import eu.darken.capod.common.upgrade.UpgradeRepo import io.kotest.matchers.shouldBe +import io.kotest.matchers.types.shouldBeInstanceOf +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout import org.junit.jupiter.api.AfterEach import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test import testhelpers.BaseTest +import java.io.IOException import java.time.Instant +import java.util.concurrent.atomic.AtomicInteger class UpgradeRepoFossTest : BaseTest() { @@ -20,6 +40,24 @@ class UpgradeRepoFossTest : BaseTest() { } + private val record = FossUpgrade( + upgradedAt = Instant.EPOCH, + reason = FossUpgrade.Reason.DONATED, + ) + + private fun createUpgradeValue(cacheFlow: Flow) = mockk>().apply { + every { flow } returns cacheFlow + coEvery { update(any()) } returns DataStoreValue.Updated(old = null, new = record) + } + + // Real dispatchers, not a test scheduler: the shareIn sharing coroutine and the collectors have + // to actually interleave here, and the whole point is that a failure settles instead of hanging. + private fun createRepo(appScope: CoroutineScope, upgradeValue: DataStoreValue) = UpgradeRepoFoss( + appScope = appScope, + fossCache = mockk().apply { every { upgrade } returns upgradeValue }, + webpageTool = mockk(), + ) + @Test fun `test upgrade info pro status mapping`() { UpgradeRepoFoss.Info( isPro = false, @@ -34,4 +72,82 @@ class UpgradeRepoFossTest : BaseTest() { upgradedAt = Instant.EPOCH, ).isPro shouldBe true } -} \ No newline at end of file + + @Test fun `a failing cache read surfaces as a settled error Info instead of hanging`(): Unit = runBlocking { + val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + try { + val repo = createRepo(scope, createUpgradeValue(flow { throw IOException("cache broken") })) + + withTimeout(10_000) { + repo.upgradeInfo.first().apply { + // Type and message: a bare non-null check would also pass on a swallow-and-wrap. + error.shouldBeInstanceOf().message shouldBe "cache broken" + isPro shouldBe false + // The UI must be able to render this: an unsettled error is an endless spinner. + isSettled shouldBe true + } + } + } finally { + scope.cancel() + } + } + + @Test fun `a late cache failure keeps the last known entitlement`(): Unit = runBlocking { + val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + try { + val repo = createRepo(scope, createUpgradeValue(flow { + emit(record) + throw IOException("cache broken later") + })) + + withTimeout(10_000) { + val infos = repo.upgradeInfo.take(2).toList() + + infos[0].apply { + isPro shouldBe true + error shouldBe null + } + // The entitlement we already saw must survive the read failure - a revoked Pro + // status would kick a paying supporter back to the pitch. + infos[1].apply { + isPro shouldBe true + error.shouldBeInstanceOf() + } + } + } finally { + scope.cancel() + } + } + + @Test fun `a successful persist revives an error-stuck upgradeInfo`(): Unit = runBlocking { + val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + try { + // First subscription fails, later ones read fine: the store recovered, but the shared + // flow is still replaying the error Info to everyone. + val subscriptions = AtomicInteger(0) + val upgradeValue = createUpgradeValue(flow { + if (subscriptions.getAndIncrement() == 0) throw IOException("cache broken") + emit(record) + }) + val repo = createRepo(scope, upgradeValue) + + val received = Channel(Channel.UNLIMITED) + scope.launch { repo.upgradeInfo.collect { received.send(it) } } + + withTimeout(10_000) { + received.receive().error.shouldBeInstanceOf() + + // No explicit refresh() from the test: persist has to do the reviving itself, + // otherwise the user's unlock never reaches the screen they are looking at. + repo.persistUpgrade() shouldBe true + + received.receive().apply { + isPro shouldBe true + error shouldBe null + } + } + } finally { + scope.cancel() + } + } +}