fix(upgrade): Settle FOSS entitlement errors instead of hanging

A thrown cache read used to die inside shareIn's sharing coroutine, leaving every collector waiting forever. Catch inside flatMapLatest, keep the last known entitlement on late failures, and let a successful persist revive an error-stuck inner flow.
2026-09-14 18:26:11 -04:00 · 2026-08-03 20:28:01 +02:00
parent b77f4a9581
commit 9b60ed945f
2 changed files with 160 additions and 15 deletions
@@ -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<UpgradeRepo.Info> = 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<UpgradeRepo.Info> = 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 {
@@ -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<FossUpgrade?>) = mockk<DataStoreValue<FossUpgrade?>>().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<FossUpgrade?>) = UpgradeRepoFoss(
+        appScope = appScope,
+        fossCache = mockk<FossCache>().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
    }
-}
+
+    @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<IOException>().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<IOException>()
+                }
+            }
+        } 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<UpgradeRepo.Info>(Channel.UNLIMITED)
+            scope.launch { repo.upgradeInfo.collect { received.send(it) } }
+
+            withTimeout(10_000) {
+                received.receive().error.shouldBeInstanceOf<IOException>()
+
+                // 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()
+        }
+    }
+}