fix(upgrade): Harden billing storage, restore races, and offer retry

2026-09-14 18:26:11 -04:00 · 2026-07-23 13:01:14 +02:00
parent f629f05f23
commit 36c54d5d19
10 changed files with 603 additions and 52 deletions
@@ -24,8 +24,10 @@ import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
-import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.filter
@@ -35,6 +37,7 @@ import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onStart
+import kotlinx.coroutines.flow.retryWhen
 import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
@@ -44,7 +47,6 @@ import java.time.Instant
 import javax.inject.Inject
 import javax.inject.Singleton
 import eu.darken.capod.common.datastore.value
-import eu.darken.capod.common.datastore.valueBlocking

@Singleton
 class UpgradeRepoGplay @Inject constructor(
@@ -54,14 +56,18 @@ class UpgradeRepoGplay @Inject constructor(
    private val timeSource: TimeSource,
 ) : UpgradeRepo {

-    private val lastProStateAt: Long
-        get() = billingCache.lastProStateAt.valueBlocking
-
    // Serializes the sticky check-then-write anchor logic: concurrent fresh observations (init
    // collector, direct restores, failure events) must not interleave between reading the current
    // anchor and stamping the new one.
    private val proStateLock = Mutex()

+    // True while the invisible already-owned recovery (the async ITEM_ALREADY_OWNED collector below)
+    // is restoring. The ViewModel gates buy actions on it so a buy tap can't race the silent restore
+    // and buy the OTHER product on top of what the user already owns (a different-SKU double charge —
+    // the ITEM_ALREADY_OWNED reconciliation only covers the same SKU).
+    private val autoRestoreBusyState = MutableStateFlow(false)
+    val autoRestoreBusy: StateFlow<Boolean> = autoRestoreBusyState.asStateFlow()
+
    init {
        // Fresh-provenance grace stamping: freshBillingData carries every successful query result
        // and push payload as an event — unlike the equality-deduped billingData state, an
@@ -107,12 +113,15 @@ class UpgradeRepoGplay @Inject constructor(
            .filter { it.responseCode == BillingClient.BillingResponseCode.ITEM_ALREADY_OWNED }
            .onEach {
                log(TAG, INFO) { "Async already-owned event -> restoring purchase" }
+                autoRestoreBusyState.value = true
                try {
                    withTimeoutOrNull(RESTORE_ON_OWNED_TIMEOUT_MS) { restorePurchaseNow() }
                } catch (e: CancellationException) {
                    throw e
                } catch (e: Exception) {
                    log(TAG, WARN) { "Async already-owned restore failed: ${e.asLog()}" }
+                } finally {
+                    autoRestoreBusyState.value = false
                }
            }
            .setupCommonEventHandlers(TAG) { "asyncAlreadyOwned" }
@@ -121,9 +130,36 @@ class UpgradeRepoGplay @Inject constructor(

    // Grace window depends on what was last owned: a permanent one-time purchase should almost
    // never be dropped on a Play hiccup, so it gets a long window; a subscription legitimately
-    // lapses, so it keeps the short one (also used for unknown/legacy last SKUs).
-    private fun graceWindowMs(): Long =
-        if (billingCache.lastProStateSku.valueBlocking.isIapSku()) GRACE_PERIOD_IAP_MS else GRACE_PERIOD_MS
+    // lapses, so it keeps the short one (also used for unknown/legacy last SKUs). Suspend + the
+    // cancellable value() read (not valueBlocking's runBlocking) so a hung DataStore read can be
+    // cancelled/retried instead of pinning a dispatcher thread.
+    private suspend fun graceWindowMs(): Long =
+        if (billingCache.lastProStateSku.value().isIapSku()) GRACE_PERIOD_IAP_MS else GRACE_PERIOD_MS
+
+    // Was the last confirmed Pro state within the grace window? Guarded AND bounded: a read failure
+    // — the same DataStore a caller may have just failed on — is treated as "not recently Pro"
+    // rather than propagating; a hung read is bounded by a timeout so it can't wedge the sequential
+    // upgradeInfo mapping and block a later confirmed purchase behind it. Shared by the reactive
+    // mapping, the reactive retry and the direct restore so their grace decision can't diverge.
+    private suspend fun isRecentlyPro(): Boolean = try {
+        val recent = withTimeoutOrNull(GRACE_PROBE_TIMEOUT_MS) {
+            (timeSource.currentTimeMillis() - billingCache.lastProStateAt.value()) < graceWindowMs()
+        }
+        if (recent == null) log(TAG, WARN) { "Grace probe timed out, treating as not-recently-pro" }
+        recent ?: false
+    } catch (e: CancellationException) {
+        throw e
+    } catch (e: Exception) {
+        log(TAG, WARN) { "Grace probe read failed, treating as not-recently-pro: ${e.asLog()}" }
+        false
+    }
+
+    // Reactive fallback when the upgradeInfo mapping throws (only local DataStore reads can fail here
+    // now — the connection loop retries billing errors itself): keep a recently-Pro user in grace,
+    // otherwise surface the error. Never throws — a second cache failure resolves to the error Info.
+    private suspend fun graceOrError(error: Throwable): Info =
+        if (isRecentlyPro()) Info(gracePeriod = true, billingData = null)
+        else Info(billingData = null, error = error)

    private fun String.isIapSku(): Boolean =
        CapodSku.PRO_SKUS.singleOrNull { it.id == this }?.type == Sku.Type.IAP
@@ -131,6 +167,13 @@ class UpgradeRepoGplay @Inject constructor(
    // Grace is time-based, but billingData is equality-deduped state kept hot by a
    // process-lifetime subscriber — without this deadline tick, a lapsed grace window would keep
    // isPro=true until the next distinct billing emission or a process restart.
+    //
+    // Storage-failure resilient: the leading onStart emits immediately so a confirmed purchase in
+    // billingData never waits on the first DataStore read (combine can fire, and toUpgradeInfo()'s
+    // mapped-first branch surfaces the purchase without touching the cache). The trailing retryWhen
+    // catches a failure from EITHER the lastProStateAt.flow source OR graceWindowMs() and keeps the
+    // tick alive (emit + capped backoff), so a broken cache can't starve combine or terminate the
+    // stream.
    private val graceDeadlineTick: Flow<Unit> = billingCache.lastProStateAt.flow
        .flatMapLatest { lastProAt ->
            flow {
@@ -144,6 +187,14 @@ class UpgradeRepoGplay @Inject constructor(
                }
            }
        }
+        .onStart { emit(Unit) }
+        .retryWhen { error, attempt ->
+            if (error is CancellationException) return@retryWhen false
+            log(TAG, WARN) { "graceDeadlineTick failed (attempt=$attempt): ${error.asLog()}" }
+            emit(Unit)
+            delay(retryDelayMs(attempt))
+            true
+        }

    override val upgradeInfo: Flow<UpgradeRepo.Info> = combine(
        billingDataRepo.billingData
@@ -152,26 +203,48 @@ class UpgradeRepoGplay @Inject constructor(
        graceDeadlineTick,
    ) { data, _ -> data }
        .map { data -> data.toUpgradeInfo() }
-        .catch { error ->
-            log(TAG, WARN) { "upgradeInfo error: ${error.asLog()}" }
-            val now = timeSource.currentTimeMillis()
-            if ((now - lastProStateAt) < graceWindowMs()) {
-                emit(Info(gracePeriod = true, billingData = null))
-            } else {
-                emit(Info(billingData = null, error = error))
-            }
+        .retryWhen { error, attempt ->
+            // Defensive backstop: toUpgradeInfo() now routes its cache access through the
+            // guarded+bounded isRecentlyPro() and so never throws for a failing/hung DataStore, and
+            // graceDeadlineTick keeps itself alive. This only fires on a genuinely unexpected
+            // upstream error — keep the flow alive (a terminal .catch would complete the shared flow
+            // and the process-lifetime subscriber would never let it recover) and emit a guarded
+            // fallback rather than terminating.
+            if (error is CancellationException) return@retryWhen false
+            log(TAG, WARN) { "upgradeInfo mapping failed unexpectedly (attempt=$attempt): ${error.asLog()}" }
+            emit(graceOrError(error))
+            delay(retryDelayMs(attempt))
+            true
        }
        .shareIn(scope, SharingStarted.WhileSubscribed(3000L, 0L), replay = 1)

    // True once we've ever confirmed a known Pro purchase on this install; drives the proactive
    // restore banner. Local signal only — a fresh install or switched Google account starts false.
+    // Fail-soft: this is combined in the ViewModel OUTSIDE the hardened upgradeInfo, so a DataStore
+    // read failure here would otherwise terminate that combine and strand the screen even when
+    // upgradeInfo correctly reports Pro. On failure fall back to false and retry.
    val wasEverPro: Flow<Boolean> = billingCache.lastProStateAt.flow
        .map { it > 0 }
+        .retryWhen { error, attempt ->
+            if (error is CancellationException) return@retryWhen false
+            log(TAG, WARN) { "wasEverPro read failed (attempt=$attempt): ${error.asLog()}" }
+            emit(false)
+            delay(retryDelayMs(attempt))
+            true
+        }
        .distinctUntilChanged()

    // Start of the current "fresh data can't confirm Pro" episode (0 = none open). Drives the
    // two-stage grace UI: calm confirmation phase first, diagnostics once the episode has aged.
+    // Fail-soft for the same reason as wasEverPro: fall back to 0 (no open episode) and retry.
    val proUnconfirmedSince: Flow<Long> = billingCache.proUnconfirmedAt.flow
+        .retryWhen { error, attempt ->
+            if (error is CancellationException) return@retryWhen false
+            log(TAG, WARN) { "proUnconfirmedSince read failed (attempt=$attempt): ${error.asLog()}" }
+            emit(0L)
+            delay(retryDelayMs(attempt))
+            true
+        }

    // True once any fresh billing observation arrived this process. The pre-reconciliation empty
    // purchase state must not enable purchase actions — an owner on a fresh install would briefly
@@ -193,17 +266,27 @@ class UpgradeRepoGplay @Inject constructor(
        return try {
            val fresh = billingDataRepo.refresh()
            // Returned data is fresh by definition — stamp it even if the flows dedupe the
-            // unchanged result and the init collector never sees a new emission.
-            recordProState(fresh)
+            // unchanged result and the init collector never sees a new emission. Best-effort: a
+            // failed cache write must not turn a successful Play restore into the grace/error path
+            // (the mapped info below is returned regardless).
+            try {
+                recordProState(fresh)
+            } catch (e: CancellationException) {
+                throw e
+            } catch (e: Exception) {
+                log(TAG, WARN) { "restore: failed to record pro state: ${e.asLog()}" }
+            }
            fresh.data.toUpgradeInfo()
        } catch (e: CancellationException) {
            throw e
        } catch (e: Exception) {
-            // Mirror the reactive flow's catch: a transient Play error while we were Pro recently
-            // keeps us Pro via the grace period; otherwise surface the error so the caller can show
-            // the proper "Play unavailable" message instead of a generic restore failure.
-            if ((timeSource.currentTimeMillis() - lastProStateAt) < graceWindowMs()) {
-                log(TAG, VERBOSE) { "Restore hit a Play error but we were Pro recently -> grace" }
+            // A transient Play error (or a cache read failure in the mapping) while we were Pro
+            // recently keeps us Pro via grace; otherwise surface the error so the caller can show the
+            // proper "Play unavailable" message instead of a generic restore failure. isRecentlyPro
+            // guards its own probe, so a second failure of the same broken cache resolves to "throw
+            // the original error" rather than escaping with the probe's exception.
+            if (isRecentlyPro()) {
+                log(TAG, VERBOSE) { "Restore hit an error but we were Pro recently -> grace" }
                Info(gracePeriod = true, billingData = null)
            } else {
                throw e
@@ -215,19 +298,26 @@ class UpgradeRepoGplay @Inject constructor(
    // Only relinquishes Pro if we haven't had it for a while (grace period). READ-ONLY: this also
    // runs on replayed shared-flow data, so it must never stamp the grace cache — a refunded
    // purchase could otherwise keep re-stamping its own grace window. See recordProState().
-    private fun BillingData?.toUpgradeInfo(): Info {
-        val now = timeSource.currentTimeMillis()
-        val proSku = this?.getProSku()
-        log(TAG) { "toUpgradeInfo(): now=$now, lastProStateAt=$lastProStateAt, data=$this" }
-        return when {
-            proSku != null -> Info(billingData = this, upgrades = this!!.getProSkus())
+    //
+    // Branch on MAPPED upgrades before any cache read: a confirmed known purchase is Pro even when
+    // local storage is unreadable (mapped-first return, no DataStore access), and a purchase list
+    // containing only products this app doesn't know maps to zero upgrades and correctly falls
+    // through to the grace check instead of masquerading as a confirmed purchase.
+    private suspend fun BillingData?.toUpgradeInfo(): Info {
+        val mapped = Info(billingData = this, upgrades = this?.getProSkus() ?: emptyList())
+        if (mapped.upgrades.isNotEmpty()) return mapped

-            (now - lastProStateAt) < graceWindowMs() -> {
-                log(TAG, VERBOSE) { "We are not pro, but were recently, did GPlay try annoy us again?" }
-                Info(gracePeriod = true, billingData = null)
-            }
-
-            else -> Info(billingData = this, upgrades = this?.getProSkus() ?: emptyList())
+        // No confirmed purchase (incl. the null pre-data placeholder the combine seeds): fall back to
+        // the grace window via the guarded+bounded probe. Routing through isRecentlyPro() — instead
+        // of reading the cache inline — means a failing/hung DataStore can NOT throw out of this
+        // mapping. If it could, the map's exception would tear down the flow and the retry would
+        // re-inject the null placeholder, looping forever and never processing a later confirmed
+        // purchase that arrives behind it in the sequential map.
+        return if (isRecentlyPro()) {
+            log(TAG, VERBOSE) { "We are not pro, but were recently, did GPlay try annoy us again?" }
+            Info(gracePeriod = true, billingData = null)
+        } else {
+            mapped
        }
    }

@@ -338,6 +428,16 @@ class UpgradeRepoGplay @Inject constructor(
        internal fun preferredProSku(upgrades: Collection<PurchasedSku>): Sku? =
            upgrades.firstOrNull { it.sku.type == Sku.Type.IAP }?.sku ?: upgrades.firstOrNull()?.sku

+        // Backoff for the local-DataStore-failure retries in upgradeInfo and graceDeadlineTick:
+        // 30s/60s/120s/240s, capped at 5min. Integer math on purpose — a Double-pow formula could
+        // overflow into a hot loop at extreme attempt counts. Pure and unit-tested.
+        internal fun retryDelayMs(attempt: Long): Long =
+            if (attempt >= 4) 300_000L else 30_000L shl attempt.toInt()
+
+        // Upper bound on a single grace-cache probe: a hung DataStore read resolves to "not
+        // recently pro" instead of wedging the sequential upgradeInfo mapping behind it.
+        private const val GRACE_PROBE_TIMEOUT_MS = 2_000L
+
        private const val RESTORE_ON_OWNED_TIMEOUT_MS = 15_000L

        val TAG: String = logTag("Upgrade", "Gplay", "Control")
@@ -74,16 +74,34 @@ data class BillingClientConnection(
        purchasesGlobal,
        queryCache,
    ) { global, cached ->
-        val combined = mutableSetOf<Purchase>()
+        // Dedup by purchaseToken, not Purchase identity: the query-cache snapshot and the listener
+        // overlay can hold the SAME purchase with a different ack-state — Play's immutable Purchase
+        // keeps reporting isAcknowledged=false until a fresh query supersedes it — so a Set/equals
+        // dedup (originalJson+signature differ) would keep BOTH. Insert the query-cache entries
+        // first, then let the listener overlay overwrite by token: reconcileListenerRecords() has
+        // already dropped same-token listener records after a non-raced query and deliberately KEEPS
+        // them when a purchase raced the query, so a surviving overlay entry is by construction the
+        // newer one — the overlay is not unconditionally fresher, it only wins when reconciliation
+        // left it in place.
+        val byToken = LinkedHashMap<String, Purchase>()

-        cached.iaps?.let { combined.addAll(it) }
-        cached.subs?.let { combined.addAll(it) }
+        fun keep(purchase: Purchase) {
+            if (purchase.purchaseToken.isBlank()) {
+                // Play supplies a non-empty token for PURCHASED purchases; a blank one is malformed
+                // and must not collapse every such record under the "" key.
+                log(TAG, WARN) { "Ignoring PURCHASED record with blank purchaseToken: $purchase" }
+                return
+            }
+            byToken[purchase.purchaseToken] = purchase
+        }

+        cached.iaps?.forEach { keep(it) }
+        cached.subs?.forEach { keep(it) }
        global
            .filter { it.purchaseState == Purchase.PurchaseState.PURCHASED }
-            .let { combined.addAll(it) }
+            .forEach { keep(it) }

-        combined.sortedByDescending { it.purchaseTime }
+        byToken.values.sortedByDescending { it.purchaseTime }
    }
        .setupCommonEventHandlers(TAG) { "purchases" }

@@ -77,6 +77,7 @@ object UpgradeScreenTags {
    const val SUB_BUTTON = "upgrade.sub.button"
    const val IAP_BUTTON = "upgrade.iap.button"
    const val RESTORE_BUTTON = "upgrade.restore.button"
+    const val RETRY_BUTTON = "upgrade.retry.button"
    const val RESTORE_BANNER = "upgrade.restore.banner"
    const val OWNER_HERO = "upgrade.owner.hero"
    const val OWNER_SUB_CARD = "upgrade.owner.subCard"
@@ -145,6 +146,7 @@ fun UpgradeScreenHost(
        onSubscriptionTrial = { activity?.let { vm.onGoSubscriptionTrial(it) } },
        onIap = { activity?.let { vm.onGoIap(it) } },
        onRestore = { vm.restorePurchase() },
+        onRetry = { vm.retrySkuQuery() },
        onManageSubscription = { vm.onManageSubscription() },
    )

@@ -248,6 +250,7 @@ fun UpgradeScreen(
    onIap: () -> Unit,
    onRestore: () -> Unit,
    onManageSubscription: () -> Unit,
+    onRetry: () -> Unit = {},
 ) {
    Scaffold(
        containerColor = MaterialTheme.colorScheme.surface,
@@ -307,6 +310,7 @@ fun UpgradeScreen(
                        onSubscriptionTrial = onSubscriptionTrial,
                        onIap = onIap,
                        onRestore = onRestore,
+                        onRetry = onRetry,
                    )

                    state is UpgradeUiState.Loaded -> AcquisitionContent(
@@ -315,6 +319,7 @@ fun UpgradeScreen(
                        onSubscriptionTrial = onSubscriptionTrial,
                        onIap = onIap,
                        onRestore = onRestore,
+                        onRetry = onRetry,
                    )
                }

@@ -544,6 +549,7 @@ private fun GraceContent(
    onSubscriptionTrial: () -> Unit,
    onIap: () -> Unit,
    onRestore: () -> Unit,
+    onRetry: () -> Unit = {},
 ) {
    val grace = state.grace ?: return

@@ -609,6 +615,7 @@ private fun GraceContent(
            onSubscriptionTrial = onSubscriptionTrial,
            onIap = onIap,
            onRestore = onRestore,
+            onRetry = onRetry,
            showRestore = false,
        )
    }
@@ -623,6 +630,7 @@ private fun AcquisitionContent(
    onSubscriptionTrial: () -> Unit,
    onIap: () -> Unit,
    onRestore: () -> Unit,
+    onRetry: () -> Unit = {},
 ) {
    val benefits = listOf(
        Benefit(Icons.TwoTone.Palette, R.string.upgrade_benefit_themes),
@@ -713,6 +721,7 @@ private fun AcquisitionContent(
        onSubscriptionTrial = onSubscriptionTrial,
        onIap = onIap,
        onRestore = onRestore,
+        onRetry = onRetry,
    )
 }

@@ -795,6 +804,7 @@ private fun PricingContent(
    onSubscriptionTrial: () -> Unit,
    onIap: () -> Unit,
    onRestore: () -> Unit,
+    onRetry: () -> Unit = {},
    showRestore: Boolean = true,
 ) {
    // Subscription button (primary)
@@ -868,7 +878,9 @@ private fun PricingContent(
        }
    }

-    // If no details loaded at all, show a simple fallback upgrade button
+    // If no details loaded at all, show a simple fallback upgrade button (which re-queries the SKU
+    // at purchase time) plus a Retry that reloads the offers — a cold/slow Play store can otherwise
+    // leave price and subscription/trial selection unavailable for the whole screen visit.
    if (!state.subAvailable && !state.iapAvailable) {
        Button(
            onClick = onIap,
@@ -892,6 +904,25 @@ private fun PricingContent(
                color = MaterialTheme.colorScheme.onPrimary,
            )
        }
+
+        Spacer(modifier = Modifier.height(8.dp))
+
+        OutlinedButton(
+            onClick = onRetry,
+            // Disabled while a query is running so repeated taps can't thrash the query flow —
+            // owners/grace users keep this fallback visible price-independently.
+            enabled = !state.skuQueryInProgress,
+            modifier = Modifier
+                .fillMaxWidth()
+                .height(52.dp)
+                .testTag(UpgradeScreenTags.RETRY_BUTTON),
+            shape = RoundedCornerShape(12.dp),
+        ) {
+            Text(
+                text = stringResource(R.string.general_retry_action),
+                style = MaterialTheme.typography.titleMedium,
+            )
+        }
    }

    Spacer(modifier = Modifier.height(8.dp))
@@ -20,6 +20,9 @@ sealed interface UpgradeUiState {
        val settled: Boolean = true,
        val restoreInProgress: Boolean = false,
        val verificationInProgress: Boolean = false,
+        // A SKU-detail query is still running. Owners/grace users render the fallback + Retry
+        // price-independently, so the Retry affordance disables itself while this is true.
+        val skuQueryInProgress: Boolean = false,
    ) : UpgradeUiState {
        val subAvailable: Boolean get() = subscriptionAction != SubscriptionAction.UNAVAILABLE
        val iapAvailable: Boolean get() = iapPrice != null
@@ -69,6 +72,7 @@ fun toLoadedState(
    settled: Boolean,
    restoreInProgress: Boolean,
    verificationInProgress: Boolean,
+    skuQueryInProgress: Boolean = false,
 ): UpgradeUiState.Loaded {
    val iapOffer = skus.iap?.details?.oneTimePurchaseOfferDetails
    val subOffers = skus.sub?.details?.subscriptionOfferDetails
@@ -94,5 +98,6 @@ fun toLoadedState(
        settled = settled,
        restoreInProgress = restoreInProgress,
        verificationInProgress = verificationInProgress,
+        skuQueryInProgress = skuQueryInProgress,
    )
 }
@@ -22,6 +22,7 @@ import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
@@ -37,6 +38,7 @@ import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.withTimeoutOrNull
 import java.time.Duration
 import javax.inject.Inject
@@ -86,17 +88,24 @@ class UpgradeViewModel @Inject constructor(
        },
    ).stateIn(vmScope, SharingStarted.Eagerly, false)

-    // One aggregate SKU-detail query per ViewModel lifetime, both types concurrently. Failures
+    // Bumped by retrySkuQuery() to re-run the aggregate query after a cold/slow-Play failure left
+    // the offers unavailable — without it the Lazily-cached failure would brick offer selection for
+    // the whole ViewModel lifetime (only leaving and reopening the screen recovered).
+    private val retryTrigger = MutableStateFlow(0)
+
+    // One aggregate SKU-detail query per retry generation, both types concurrently. Failures
    // resolve to null details — owners/grace render price-independently, acquisition users get
    // the fallback purchase UI.
-    private val skuQueries = flow {
-        emit(SkuQueryState())
-        val result = coroutineScope {
-            val iap = async { querySkuDetailsSafe(CapodSku.Iap.PRO_UPGRADE) }
-            val sub = async { querySkuDetailsSafe(CapodSku.Sub.PRO_UPGRADE) }
-            SkuQueryState(done = true, iap = iap.await(), sub = sub.await())
+    private val skuQueries = retryTrigger.flatMapLatest {
+        flow {
+            emit(SkuQueryState())
+            val result = coroutineScope {
+                val iap = async { querySkuDetailsSafe(CapodSku.Iap.PRO_UPGRADE) }
+                val sub = async { querySkuDetailsSafe(CapodSku.Sub.PRO_UPGRADE) }
+                SkuQueryState(done = true, iap = iap.await(), sub = sub.await())
+            }
+            emit(result)
        }
-        emit(result)
    }.shareIn(vmScope, SharingStarted.Lazily, replay = 1)

    private suspend fun querySkuDetailsSafe(sku: Sku): SkuDetails? = try {
@@ -110,6 +119,23 @@ class UpgradeViewModel @Inject constructor(
        null
    }

+    // Re-runs the SKU queries from the fallback "Retry" affordance. The button that calls this is
+    // disabled while a query is in flight (skuQueryInProgress), which is the actual thrash guard for
+    // owners/grace users who keep the fallback visible price-independently; a re-trigger that still
+    // slips through only cancels-and-restarts the flatMapLatest query (latest wins, each attempt is
+    // bounded by SKU_QUERY_TIMEOUT_MS), so it can't leak or wedge.
+    fun retrySkuQuery() {
+        log(TAG) { "retrySkuQuery()" }
+        retryTrigger.update { it + 1 }
+    }
+
+    // Manual restore OR the repo's invisible already-owned recovery — either one pauses the buy
+    // actions, so the two can't be raced against each other from the UI.
+    private val effectiveRestore: Flow<Boolean> = combine(
+        restoring,
+        upgradeRepo.autoRestoreBusy,
+    ) { manual, auto -> manual || auto }
+
    // Re-evaluates the grace presentation when the open episode crosses the diagnostics
    // threshold — every other combined flow is distinct-until-changed and would never re-fire.
    private val graceTick = upgradeRepo.proUnconfirmedSince
@@ -149,7 +175,7 @@ class UpgradeViewModel @Inject constructor(
        billingState,
        skuQueries,
        settled,
-        restoring,
+        effectiveRestore,
        purchaseBusy,
    ) { billing, skus, isSettled, isRestoring, isBusy ->
        val info = billing.info
@@ -178,6 +204,9 @@ class UpgradeViewModel @Inject constructor(
                settled = isSettled,
                restoreInProgress = isRestoring,
                verificationInProgress = isBusy,
+                // Owners/grace users keep the fallback + Retry visible while a query is still
+                // running; disable Retry then so repeated taps can't thrash the query flow.
+                skuQueryInProgress = !skus.done,
            )
        }
    }.stateIn(vmScope, SharingStarted.WhileSubscribed(5_000), UpgradeUiState.Loading)
@@ -249,6 +278,14 @@ class UpgradeViewModel @Inject constructor(
    // Single-flight for purchase actions: the guard is held from the tap until the Play sheet
    // launch has resolved, so repeated taps can't stack verification queries or billing flows.
    private suspend fun runExclusive(block: suspend () -> Unit) {
+        // Authoritative gate for the invisible already-owned recovery: button disabling is
+        // best-effort (recomposition lags a tap), so a subscribe/buy tap dispatched while the silent
+        // restore runs must be refused here, or it could buy the OTHER product on top of what the
+        // user already owns (a different-SKU double charge the ITEM_ALREADY_OWNED path won't catch).
+        if (upgradeRepo.autoRestoreBusy.value) {
+            log(TAG) { "Purchase action ignored, auto-restore in progress" }
+            return
+        }
        if (restoring.value) {
            log(TAG) { "Purchase action ignored, restore in progress" }
            return
@@ -279,6 +316,11 @@ class UpgradeViewModel @Inject constructor(
    }

    fun restorePurchase() = launch {
+        // Don't overlap the invisible already-owned recovery — it is itself a restore.
+        if (upgradeRepo.autoRestoreBusy.value) {
+            log(TAG) { "restorePurchase() ignored, auto-restore in progress" }
+            return@launch
+        }
        // Symmetric to runExclusive: a restore must not overlap an in-flight verification or
        // billing launch either, or the user could end up with two result dialogs stacked.
        if (purchaseBusy.value) {
@@ -5,6 +5,7 @@
    <string name="general_copy_action">Copy</string>
    <string name="general_thank_you_label">Thank you</string>
    <string name="general_upgrade_action">Upgrade</string>
+    <string name="general_retry_action">Retry</string>
    <string name="common_feature_requires_pro_msg">Upgrade to unlock.</string>
    <string name="common_upgrade_required_label">Requires upgrade</string>
    <string name="general_donate_action">Donate</string>
@@ -23,9 +23,11 @@ import io.mockk.mockk
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.launch
@@ -38,6 +40,7 @@ import testhelpers.BaseTest
 import testhelpers.TestTimeSource
 import testhelpers.coroutine.runTest2
 import java.io.File
+import java.io.IOException
 import java.time.Duration

 class UpgradeRepoGplayTest : BaseTest() {
@@ -106,8 +109,9 @@ class UpgradeRepoGplayTest : BaseTest() {

        billingDataFlow.emit(BillingData(purchases = emptyList()))

-        // First emission is onStart, second is billing data
-        emissions.size shouldBe 2
+        // The startup ticks (grace-deadline onStart + the initial cache read) plus the empty
+        // billing emission all map to the same not-pro state — assert the contract, not the
+        // incidental emission count.
        val info = emissions.last()
        info.isPro shouldBe false
        info.error.shouldBeNull()
@@ -779,4 +783,164 @@ class UpgradeRepoGplayTest : BaseTest() {

        testScope.cancel()
    }
+
+    // --- Storage-failure resilience (P1) ---
+
+    // A cache whose reads/writes fail like a full or corrupt DataStore.
+    private fun failingCache(): BillingCache = mockk(relaxed = true) {
+        every { lastProStateAt } returns mockk { every { flow } returns flow { throw IOException("disk full") } }
+        every { lastProStateSku } returns mockk { every { flow } returns flow { throw IOException("disk full") } }
+        every { proUnconfirmedAt } returns mockk { every { flow } returns flow { throw IOException("disk full") } }
+        coEvery { stampLastProState(any(), any()) } throws IOException("disk full")
+        coEvery { recordProUnconfirmed(any()) } throws IOException("disk full")
+    }
+
+    private fun repoWith(cache: BillingCache, scope: TestScope): UpgradeRepoGplay = UpgradeRepoGplay(
+        scope = scope,
+        billingDataRepo = billingDataRepo,
+        billingCache = cache,
+        timeSource = timeSource,
+    )
+
+    @Test
+    fun `a known purchase is pro even when the grace cache is unreadable and unwritable`() = runTest2 {
+        val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
+        // restorePurchaseNow maps the fresh data directly: mapped-first must return Pro without
+        // touching the cache, and the best-effort stamp must swallow the write failure.
+        coEvery { billingDataRepo.refresh() } returns freshData(
+            purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id))
+        )
+        val repo = repoWith(failingCache(), testScope)
+
+        repo.restorePurchaseNow().isPro shouldBe true
+
+        testScope.cancel()
+    }
+
+    @Test
+    fun `unknown-only purchases still fall through to the grace check`() = runTest2 {
+        val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
+        // A purchase list of only products this app doesn't recognize maps to zero upgrades — it
+        // must NOT masquerade as a confirmed purchase, but fall through to grace for a recent owner.
+        coEvery { billingDataRepo.refresh() } returns freshData(
+            purchases = listOf(mockPurchase("some.unknown.product"))
+        )
+        billingCache.lastProStateAt.valueBlocking = now() - 1_000L
+        val repo = createRepo(testScope)
+
+        repo.restorePurchaseNow().isPro shouldBe true
+
+        testScope.cancel()
+    }
+
+    @Test
+    fun `unknown-only purchases without recent grace are not pro`() = runTest2 {
+        val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
+        coEvery { billingDataRepo.refresh() } returns freshData(
+            purchases = listOf(mockPurchase("some.unknown.product"))
+        )
+        val repo = createRepo(testScope)
+
+        repo.restorePurchaseNow().isPro shouldBe false
+
+        testScope.cancel()
+    }
+
+    @Test
+    fun `a confirmed purchase surfaces on upgradeInfo even when the grace cache is unreadable`() = runTest2 {
+        val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
+        // The pre-data null placeholder and the empty snapshot both hit the (unreadable) grace probe
+        // first; the mapping must NOT throw and loop there, or a real Pro purchase arriving behind
+        // them would never be processed. mapped-first must surface the purchase regardless.
+        val billing = MutableSharedFlow<BillingData>(replay = 1)
+        every { billingDataRepo.billingData } returns billing
+        val repo = repoWith(failingCache(), testScope)
+
+        val emissions = mutableListOf<eu.darken.capod.common.upgrade.UpgradeRepo.Info>()
+        val job = testScope.launch { repo.upgradeInfo.toList(emissions) }
+
+        billing.emit(BillingData(purchases = emptyList()))
+        emissions.last().isPro shouldBe false
+
+        billing.emit(BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id))))
+        emissions.last().isPro shouldBe true
+
+        job.cancel()
+        testScope.cancel()
+    }
+
+    @Test
+    fun `a persistently failing grace cache degrades to not-pro without erroring or terminating`() = runTest2 {
+        val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
+        every { billingDataRepo.billingData } returns flowOf(BillingData(purchases = emptyList()))
+        val repo = repoWith(failingCache(), testScope)
+
+        // The grace probe reads the broken cache but is guarded+bounded, so the mapping never throws:
+        // the flow emits a plain not-pro Info (no error) instead of terminating. The companion test
+        // above proves it keeps processing a later confirmed purchase, so the flow stays alive.
+        val info = repo.upgradeInfo.first()
+        info.isPro shouldBe false
+        info.error.shouldBeNull()
+
+        testScope.cancel()
+    }
+
+    @Test
+    fun `wasEverPro falls back to false when its cache read fails`() = runTest2 {
+        val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
+        val repo = repoWith(failingCache(), testScope)
+
+        repo.wasEverPro.first() shouldBe false
+
+        testScope.cancel()
+    }
+
+    @Test
+    fun `proUnconfirmedSince falls back to 0 when its cache read fails`() = runTest2 {
+        val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
+        val repo = repoWith(failingCache(), testScope)
+
+        repo.proUnconfirmedSince.first() shouldBe 0L
+
+        testScope.cancel()
+    }
+
+    @Test
+    fun `retryDelayMs grows and caps at five minutes`() {
+        UpgradeRepoGplay.retryDelayMs(0) shouldBe 30_000L
+        UpgradeRepoGplay.retryDelayMs(1) shouldBe 60_000L
+        UpgradeRepoGplay.retryDelayMs(2) shouldBe 120_000L
+        UpgradeRepoGplay.retryDelayMs(3) shouldBe 240_000L
+        UpgradeRepoGplay.retryDelayMs(4) shouldBe 300_000L
+        UpgradeRepoGplay.retryDelayMs(100) shouldBe 300_000L
+        UpgradeRepoGplay.retryDelayMs(Long.MAX_VALUE) shouldBe 300_000L
+    }
+
+    // --- autoRestoreBusy gate (P2) ---
+
+    @Test
+    fun `autoRestoreBusy rises during the invisible already-owned restore and falls after`() = runTest2 {
+        val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
+        coEvery { billingDataRepo.refresh() } coAnswers {
+            delay(1_000)
+            freshData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id)))
+        }
+        every { billingDataRepo.purchaseFailures } returns
+            flowOf(mockBillingResult(BillingResponseCode.ITEM_ALREADY_OWNED))
+
+        // The event fires on construction; the restore then suspends in refresh().
+        val repo = createRepo(testScope)
+
+        val states = mutableListOf<Boolean>()
+        val job = testScope.launch { repo.autoRestoreBusy.toList(states) }
+        testScope.testScheduler.runCurrent()
+        states.last() shouldBe true
+
+        testScope.testScheduler.advanceTimeBy(1_100)
+        testScope.testScheduler.runCurrent()
+        states.last() shouldBe false
+
+        job.cancel()
+        testScope.cancel()
+    }
 }
@@ -328,4 +328,46 @@ class BillingClientConnectionTest : BaseTest() {
            )
        }
    }
+
+    // --- Token-based dedup of the merged purchases view (P4) ---
+
+    @Test
+    fun `a same-token listener overlay overwrites the query-cache record`() = runTest2 {
+        // The query cache holds a snapshot of a purchase; the listener then pushes a fresher copy
+        // (differing ack-state) under the same token. Dedup by token must keep exactly one entry,
+        // and the listener overlay — left in place by reconciliation — wins.
+        val cached = mockPurchase(CapodSku.Iap.PRO_UPGRADE.id, token = "shared")
+        val overlay = mockPurchase(CapodSku.Iap.PRO_UPGRADE.id, token = "shared")
+        val harness = Harness()
+        coEvery { harness.client.queryPurchasesAsync(any<QueryPurchasesParams>()) } returnsMany listOf(
+            harness.okResult(listOf(cached)),
+            harness.okResult(emptyList()),
+        )
+        harness.connection.refreshPurchases()
+
+        // A listener push arriving after the refresh — same token, fresher instance.
+        harness.purchasesGlobal.value = listOf(overlay)
+
+        harness.connection.purchases.first() shouldContainExactly listOf(overlay)
+    }
+
+    @Test
+    fun `distinct-token purchases all survive and stay ordered by purchase time`() = runTest2 {
+        val older = mockPurchase(CapodSku.Iap.PRO_UPGRADE.id, token = "a", purchaseTime = 1_000)
+        val newer = mockPurchase(CapodSku.Sub.PRO_UPGRADE.id, token = "b", purchaseTime = 2_000)
+        val harness = Harness(purchasesGlobal = MutableStateFlow(listOf(older, newer)))
+
+        harness.connection.purchases.first() shouldContainExactly listOf(newer, older)
+    }
+
+    @Test
+    fun `a purchased record with a blank token is discarded`() = runTest2 {
+        // Play supplies a non-empty token for PURCHASED purchases; a blank one is malformed and
+        // must not collapse every such record under the "" key.
+        val valid = mockPurchase(CapodSku.Iap.PRO_UPGRADE.id, token = "valid", purchaseTime = 2_000)
+        val blank = mockPurchase(CapodSku.Sub.PRO_UPGRADE.id, token = "", purchaseTime = 1_000)
+        val harness = Harness(purchasesGlobal = MutableStateFlow(listOf(valid, blank)))
+
+        harness.connection.purchases.first() shouldContainExactly listOf(valid)
+    }
 }
@@ -53,6 +53,7 @@ class UpgradeScreenComposeTest {
        onIap: () -> Unit = {},
        onRestore: () -> Unit = {},
        onManageSubscription: () -> Unit = {},
+        onRetry: () -> Unit = {},
    ) {
        composeRule.setContent {
            UpgradeScreen(
@@ -63,10 +64,43 @@ class UpgradeScreenComposeTest {
                onIap = onIap,
                onRestore = onRestore,
                onManageSubscription = onManageSubscription,
+                onRetry = onRetry,
            )
        }
    }

+    // No-offer fallback state (cold/slow Play store returned no product details).
+    private fun noOffers(skuQueryInProgress: Boolean = false) = UpgradeUiState.Loaded(
+        subscriptionAction = SubscriptionAction.UNAVAILABLE,
+        subscriptionEnabled = false,
+        subscriptionPrice = null,
+        iapEnabled = true,
+        iapPrice = null,
+        skuQueryInProgress = skuQueryInProgress,
+    )
+
+    @Test
+    fun `the no-offers fallback shows a Retry that fires the callback`() {
+        var retries = 0
+        setScreen(state = noOffers(), onRetry = { retries++ })
+
+        composeRule.onNodeWithTag(UpgradeScreenTags.RETRY_BUTTON)
+            .performScrollTo()
+            .assertIsEnabled()
+            .performClick()
+
+        retries shouldBe 1
+    }
+
+    @Test
+    fun `the Retry button is disabled while a SKU query is running`() {
+        setScreen(state = noOffers(skuQueryInProgress = true))
+
+        composeRule.onNodeWithTag(UpgradeScreenTags.RETRY_BUTTON)
+            .performScrollTo()
+            .assertIsNotEnabled()
+    }
+
    // --- Owner states ---

    @Test
@@ -61,6 +61,9 @@ class UpgradeViewModelTest : BaseTest() {
        every { wasEverPro } returns MutableStateFlow(false)
        every { proUnconfirmedSince } returns MutableStateFlow(0L)
        every { isSettled } returns MutableStateFlow(true)
+        // A relaxed mock returns a Flow that never emits — the effectiveRestore combine would
+        // starve and the state flow would never leave Loading.
+        every { autoRestoreBusy } returns MutableStateFlow(false)
        coEvery { queryCurrentSubscriptions() } returns emptyList()
        coEvery { querySkus(any()) } returns emptyList()
    }
@@ -628,4 +631,115 @@ class UpgradeViewModelTest : BaseTest() {
        state.iapEnabled shouldBe false
        state.subscriptionEnabled shouldBe false
    }
+
+    @Test
+    fun `toLoadedState disables the buy actions during a restore`() {
+        val state = toLoadedState(
+            skus = SkuQueryState(done = true),
+            ownership = Ownership(),
+            grace = null,
+            showRestoreBanner = false,
+            settled = true,
+            restoreInProgress = true,
+            verificationInProgress = false,
+        )
+
+        state.iapEnabled shouldBe false
+        state.subscriptionEnabled shouldBe false
+    }
+
+    // --- SKU-query retry (P3) ---
+
+    @Test
+    fun `a slow but healthy Play store waits for the query instead of tripping the timeout`() = runTest2 {
+        // 9s is under the 15s SKU-query timeout — the store is slow, not broken. Prove the screen
+        // WAITED for the slow query (still Loading at 5s) and only loaded once it answered at 9s; a
+        // shorter timeout would have flipped to Loaded early with null details.
+        val repo = mockRepo()
+        coEvery { repo.querySkus(any()) } coAnswers {
+            delay(9_000)
+            emptyList()
+        }
+        val vm = createVm(repo)
+
+        val states = mutableListOf<UpgradeUiState>()
+        val job = launch(UnconfinedTestDispatcher(testScheduler)) { vm.state.collect { states.add(it) } }
+
+        testScheduler.advanceTimeBy(5_000)
+        testScheduler.runCurrent()
+        states.last().shouldBeInstanceOf<UpgradeUiState.Loading>()
+
+        testScheduler.advanceTimeBy(4_500)
+        testScheduler.runCurrent()
+        states.last().shouldBeInstanceOf<UpgradeUiState.Loaded>()
+
+        job.cancel()
+    }
+
+    @Test
+    fun `retry re-runs the SKU queries`() = runTest2 {
+        val repo = mockRepo()
+        val vm = createVm(repo)
+        vm.state.first { it is UpgradeUiState.Loaded }
+
+        vm.retrySkuQuery()
+        advanceUntilIdle()
+
+        // Initial aggregate query + the retried generation.
+        coVerify(exactly = 2) { repo.querySkus(CapodSku.Iap.PRO_UPGRADE) }
+    }
+
+    @Test
+    fun `retry is disabled while a SKU query is still running`() = runTest2 {
+        // Grace users render the fallback + Retry price-independently while the query runs; the
+        // Retry must be disabled then so repeated taps can't thrash the query flow.
+        val repo = mockRepo()
+        every { repo.upgradeInfo } returns MutableStateFlow(
+            UpgradeRepoGplay.Info(gracePeriod = true, billingData = null)
+        )
+        every { repo.proUnconfirmedSince } returns MutableStateFlow(now() - Duration.ofHours(25).toMillis())
+        coEvery { repo.querySkus(any()) } coAnswers { awaitCancellation() }
+        val vm = createVm(repo)
+
+        val state = vm.state.first { it is UpgradeUiState.Loaded } as UpgradeUiState.Loaded
+
+        state.skuQueryInProgress shouldBe true
+    }
+
+    // --- autoRestoreBusy gate (P2) ---
+
+    @Test
+    fun `the invisible auto-restore disables the buy buttons`() = runTest2 {
+        val repo = mockRepo()
+        val autoBusy = MutableStateFlow(false)
+        every { repo.autoRestoreBusy } returns autoBusy
+        val vm = createVm(repo)
+
+        val states = mutableListOf<UpgradeUiState>()
+        val job = launch(UnconfinedTestDispatcher(testScheduler)) { vm.state.collect { states.add(it) } }
+        testScheduler.runCurrent()
+        (states.last() as UpgradeUiState.Loaded).iapEnabled shouldBe true
+
+        autoBusy.value = true
+        testScheduler.runCurrent()
+
+        (states.last() as UpgradeUiState.Loaded).iapEnabled shouldBe false
+        (states.last() as UpgradeUiState.Loaded).restoreInProgress shouldBe true
+
+        job.cancel()
+    }
+
+    @Test
+    fun `a purchase tap during the invisible auto-restore is refused authoritatively`() = runTest2 {
+        // Button disabling lags a tap; the authoritative gate in runExclusive must refuse a tap
+        // dispatched while the silent restore runs, or a subscribe would buy on top of the owned IAP.
+        val repo = mockRepo()
+        every { repo.autoRestoreBusy } returns MutableStateFlow(true)
+        val vm = createVm(repo)
+
+        vm.onGoSubscription(mockk<Activity>())
+        advanceUntilIdle()
+
+        coVerify(exactly = 0) { repo.launchBillingFlow(any(), any(), any()) }
+    }
 }