fix(upgrade): Add a persistent acknowledgement safety net for Play purchases

Play auto-refunds (and revokes) purchases not acknowledged within 3 days. The in-process ack machinery covers every case where the process lives long enough; what it cannot cover is a process death around the Play sheet (aggressive OEM task killers) followed by the user not reopening the app before the deadline. Add a gplay-only WorkManager safety net: - PurchaseAckWorker: self-completing sweep via a new bounded BillingManager.ensureAllAcknowledged() that refreshes and acknowledges in the same coroutine (the reactive ack collector is async, so a worker cannot prove its acks happened through it). Retries with exponential backoff until the purchase's refund deadline, then gives up visibly. - PurchaseAckScheduler: two unique work identities. A launch watch (REPLACE, armed and awaited before launchBillingFlow with a 30min delay so it cannot complete while the user is still in the sheet) and a discovered-purchase rescue (KEEP, 1min delay, armed directly from an ack pass that finds unacknowledged purchases, pre-attempt). Separate identities so a new purchase flow can never displace a pending rescue. Both triggers are fail-open: a broken WorkManager never blocks a purchase or an ack. WorkManager resolves via Provider at first arm so eager Application-time construction of the billing stack cannot trigger WorkManager's on-demand initialization prematurely. - Nothing cancels the work from the foreground path: an ack pass can see zero unacked purchases while the sheet is still open, so the worker completes itself after its own reconciliation instead. The ack pass now runs under a mutex (the worker sweep and the reactive collector would otherwise race the token bookkeeping) and reports per-outcome counts for the sweep result mapping. This is a port of d4rken-org/sdmaid-se#2685; the ported sources are byte-identical to the donor apart from the package rename. CAPod had no explicit WorkManager wiring at all (work-runtime only arrived transitively through Glance), so this also adds it: - addWorkerManager() pinning androidx.work 2.7.1, the version already resolved via Glance, plus androidx.hilt:hilt-work and its KSP compiler. work-runtime-ktx is required at 2.7.1: CoroutineWorker, Operation.await, OneTimeWorkRequestBuilder and workDataOf all still live in the ktx artifact at that version. androidx.hilt moves 1.0.0 -> 1.2.0 (by conflict resolution) because 1.0.0's hilt-compiler ships no KSP SymbolProcessorProvider, so @HiltWorker would generate nothing. - WorkManagerModule providing the singleton WorkManager. - App implements Configuration.Provider with the injected HiltWorkerFactory. WorkManager 2.7.1 still declares that interface as getWorkManagerConfiguration(), not the later property form. - The manifest removes androidx.work's startup initializer so the on-demand configuration is the one that takes effect. FOSS stays untouched behaviour-wise: all new billing types live in src/gplay, workers need no manifest entry, and the worker factory resolves the worker only in gplay variants.
2026-09-16 11:16:12 -04:00 · 2026-08-18 17:55:19 +02:00
parent cebed0a60d
commit 971fcbd34c
13 changed files with 613 additions and 9 deletions
@@ -23,6 +23,7 @@ import eu.darken.capod.common.upgrade.core.billing.Sku
 import eu.darken.capod.common.upgrade.core.billing.SkuDetails
 import eu.darken.capod.common.upgrade.core.billing.UserCanceledBillingException
 import eu.darken.capod.common.upgrade.core.billing.client.redacted
+import eu.darken.capod.common.upgrade.core.billing.work.PurchaseAckScheduler
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
@@ -58,6 +59,7 @@ class UpgradeRepoGplay @Inject constructor(
    private val billingManager: BillingManager,
    private val billingCache: BillingCache,
    private val curriculumVitae: CurriculumVitae,
+    private val ackScheduler: PurchaseAckScheduler,
 ) : UpgradeRepo {

    override val storeSite: String = STORE_SITE
@@ -137,6 +139,7 @@ class UpgradeRepoGplay @Inject constructor(
            .onEach { failedAt -> recordProUnconfirmed(failedAt) }
            .setupCommonEventHandlers(TAG) { "connectionFailureRecorder" }
            .launchIn(scope)
+
    }

    // Settledness travels WITH the ownership data (Info.isSettled), never on a parallel flow —
@@ -255,6 +258,18 @@ class UpgradeRepoGplay @Inject constructor(
            return
        }
        try {
+            // Persistent ack safety net, launch trigger: armed and AWAITED before the Play sheet
+            // can open, so the WorkManager DB transaction lands even if the process dies around
+            // the sheet — the exact window behind Play's unacknowledged-purchase auto-refunds.
+            // Failure to arm never blocks the purchase; the foreground ack path still exists.
+            try {
+                ackScheduler.armForBillingFlowLaunch()
+            } catch (e: CancellationException) {
+                throw e
+            } catch (e: Exception) {
+                log(TAG, WARN) { "Failed to arm ack safety net for launch: ${e.asLog()}" }
+            }
+
            // Bounded, like every other Play path (refresh, restore, SKU query, ack). useConnection
            // waits for a healthy connection indefinitely, so a Play outage between rendering the
            // offers and this tap would park the launch forever — with launchBusySku still held,
@@ -16,6 +16,7 @@ import eu.darken.capod.common.upgrade.core.billing.client.BillingConnection
 import eu.darken.capod.common.upgrade.core.billing.client.BillingConnectionProvider
 import eu.darken.capod.common.upgrade.core.billing.client.isPurchased
 import eu.darken.capod.common.upgrade.core.billing.client.redacted
+import eu.darken.capod.common.upgrade.core.billing.work.PurchaseAckScheduler
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.Channel
@@ -25,6 +26,8 @@ import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.flow.SharingStarted.Companion.WhileSubscribed
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withTimeoutOrNull
 import javax.inject.Inject
 import javax.inject.Singleton
@@ -33,6 +36,7 @@ import javax.inject.Singleton
 class BillingManager @Inject constructor(
    @AppScope private val scope: CoroutineScope,
    connectionProvider: BillingConnectionProvider,
+    private val ackScheduler: PurchaseAckScheduler,
 ) {

    // Fresh Play data plus its provenance: a query result covers owned products of the queried
@@ -174,6 +178,11 @@ class BillingManager @Inject constructor(
        }
    }

+    // Serializes acknowledgement work between the reactive ack collector and explicit
+    // ensureAllAcknowledged() sweeps (PurchaseAckWorker): both paths mutate the token bookkeeping
+    // sets and both must never double-drive the same purchase's inline retry sequence.
+    private val ackMutex = Mutex()
+
    // Re-drives the ack pass WITHOUT a new purchases emission: `purchases` is distinctUntilChanged,
    // so a refresh returning a byte-identical (still unacknowledged) list is deduped and could never
    // retry a failed ack -- the pipeline starved until Play sent something different. Declared ahead
@@ -256,13 +265,13 @@ class BillingManager @Inject constructor(
    // below. The immutable Purchase snapshot keeps reporting isAcknowledged=false until a fresh Play
    // query supersedes it, so the ack re-fires every emission until then; re-acking is a documented
    // no-op on Play's side, whereas skipping a needed ack gets the purchase auto-refunded after 3
-    // days -- so the ack stays unconditional and this set only quiets the log spam. Single
-    // sequential collector (the ack pass below), no locking needed.
+    // days -- so the ack stays unconditional and this set only quiets the log spam. Confined by
+    // ackMutex (the collector's pass and explicit sweeps both run under it).
    private val loggedAckTokens = mutableSetOf<String>()

    // Tokens whose PERMANENT ack failure was already reported. Play will keep rejecting these
    // (developer error, item not owned, unsupported feature), so the bug report fires once per token
-    // instead of once per pass. Same single-collector confinement as loggedAckTokens.
+    // instead of once per pass. Same ackMutex confinement as loggedAckTokens.
    private val reportedAckFailures = mutableSetOf<String>()

    // At most one reschedule timer in flight: repeated failures must not stack timers.
@@ -296,10 +305,13 @@ class BillingManager @Inject constructor(
    // .isAcknowledged, whose immutable snapshot stays false until a fresh Play query.
    private enum class AckOutcome { SUCCESS, TRANSIENT, PERMANENT }

+    // Aggregate outcome of one ack pass; ensureAllAcknowledged() maps it to a sweep result.
+    data class AckPassOutcome(val transient: Int, val permanent: Int)
+
    // One acknowledgement pass over the canonical purchase list. Never throws except cancellation:
    // transient failures schedule a re-drive, permanent ones are reported and left to organic fresh
    // -data signals.
-    private suspend fun runAckPass(purchases: Collection<Purchase>) {
+    private suspend fun runAckPass(purchases: Collection<Purchase>): AckPassOutcome = ackMutex.withLock {
        val needAck = purchases.filter {
            // The canonical list carries pending payments too. Play rejects acknowledging one
            // PERMANENTLY, so an unfiltered pass would fire a bug report for every pending purchase,
@@ -317,7 +329,23 @@ class BillingManager @Inject constructor(
            needsAck
        }

+        if (needAck.isNotEmpty()) {
+            // Arm the persistent safety net BEFORE attempting anything, and AWAIT the enqueue (the
+            // scheduler bounds it): the inline retries below can span minutes, and a process death
+            // inside them must not strand the purchase until Play's 3-day auto-refund. A deferred
+            // signal (channel + collector) would reintroduce exactly that window. Fail-open: the
+            // net is an extra layer, never a reason to skip the acks themselves.
+            try {
+                ackScheduler.armForUnackedPurchases(needAck.maxOf { it.purchaseTime } + ACK_SAFETY_NET_DEADLINE_MS)
+            } catch (e: CancellationException) {
+                throw e
+            } catch (e: Exception) {
+                log(TAG, WARN) { "Failed to arm ack safety net: ${e.asLog()}" }
+            }
+        }
+
        var transientFailures = 0
+        var permanentFailures = 0

        for (purchase in needAck) {
            // First ack of a token is INFO; idempotent repeats drop to DEBUG. This never gates the
@@ -381,6 +409,7 @@ class BillingManager @Inject constructor(
            }

            if (outcome == AckOutcome.TRANSIENT) transientFailures++
+            if (outcome == AckOutcome.PERMANENT) permanentFailures++
            if (abortPass) break
        }

@@ -391,6 +420,41 @@ class BillingManager @Inject constructor(
            }
            scheduleAckRetry()
        }
+
+        AckPassOutcome(transient = transientFailures, permanent = permanentFailures)
+    }
+
+    // Outcome of an explicit safety-net sweep, see ensureAllAcknowledged().
+    enum class AckSweepResult { COMPLETE, RETRY, PERMANENT_FAILURE }
+
+    /**
+     * One self-contained acknowledgement sweep for the persistent safety net (PurchaseAckWorker):
+     * refresh from Play, then acknowledge everything unacknowledged IN THIS COROUTINE. The reactive
+     * ack collector consumes purchase state asynchronously, so a caller that needs proof the acks
+     * actually happened before it reports success (a worker deciding success vs retry) cannot rely
+     * on it. Never throws except cancellation.
+     */
+    suspend fun ensureAllAcknowledged(): AckSweepResult {
+        log(TAG) { "ensureAllAcknowledged()" }
+        val fresh = try {
+            useConnection { refreshPurchases() }
+        } catch (e: CancellationException) {
+            throw e
+        } catch (e: Exception) {
+            log(TAG, WARN) { "ensureAllAcknowledged(): refresh failed: ${e.asLog()}" }
+            return AckSweepResult.RETRY
+        }
+        // Same bookkeeping every other refresh exit owes: grace episode clock + dead-binder teardown.
+        processReconciliation(fresh)
+        val outcome = runAckPass(fresh.purchases)
+        return when {
+            // An incomplete refresh may be hiding an unacknowledged purchase of the failed type,
+            // and a transient ack failure is retriable by definition.
+            outcome.transient > 0 || !fresh.isComplete -> AckSweepResult.RETRY
+            // Play will keep rejecting these no matter how often the worker comes back.
+            outcome.permanent > 0 -> AckSweepResult.PERMANENT_FAILURE
+            else -> AckSweepResult.COMPLETE
+        }
    }

    // A purchase Play will keep rejecting: report it once per token, then stay quiet. The pass still
@@ -586,6 +650,10 @@ class BillingManager @Inject constructor(
            BillingResponseCode.ITEM_NOT_OWNED,
        )

+        // Play auto-refunds purchases not acknowledged within 3 days; every safety-net deadline
+        // derives from this.
+        const val ACK_SAFETY_NET_DEADLINE_MS = 3 * 24 * 60 * 60 * 1000L
+
        private const val INITIAL_REFRESH_TIMEOUT_MS = 30_000L
        private const val MAX_BACKOFF_MS = 300_000L

@@ -0,0 +1,114 @@
+package eu.darken.capod.common.upgrade.core.billing.work
+
+import androidx.work.BackoffPolicy
+import androidx.work.Constraints
+import androidx.work.ExistingWorkPolicy
+import androidx.work.NetworkType
+import androidx.work.OneTimeWorkRequestBuilder
+import androidx.work.WorkManager
+import androidx.work.await
+import androidx.work.workDataOf
+import eu.darken.capod.common.BuildConfigWrap
+import eu.darken.capod.common.debug.logging.Logging.Priority.WARN
+import eu.darken.capod.common.debug.logging.log
+import eu.darken.capod.common.debug.logging.logTag
+import eu.darken.capod.common.upgrade.core.billing.BillingManager
+import kotlinx.coroutines.withTimeoutOrNull
+import java.util.concurrent.TimeUnit
+import javax.inject.Inject
+import javax.inject.Provider
+import javax.inject.Singleton
+
+/**
+ * Arms the [PurchaseAckWorker] safety net. Two triggers:
+ * - a billing flow is about to launch (armed and awaited BEFORE the Play sheet, so the WorkManager
+ *   DB transaction lands even if the process dies around the sheet),
+ * - an ack pass discovered unacknowledged purchases (called directly, pre-attempt, from
+ *   BillingManager's runAckPass).
+ */
+@Singleton
+class PurchaseAckScheduler @Inject constructor(
+    // Resolved on the first arm, not at construction: fleet App classes eagerly inject the billing
+    // stack during Application field injection, and resolving WorkManager there can trigger its
+    // on-demand initialization before the Application's worker factory field is set.
+    private val workManager: Provider<WorkManager>,
+) {
+
+    // A genuinely new flow refreshes the watch window: REPLACE the previous LAUNCH watch. The
+    // worker sweeps ALL unacknowledged purchases, so replacing an older watch loses nothing — and a
+    // pending rescue for an already-discovered purchase has its own identity, so starting another
+    // purchase can never displace it. The long delay keeps the worker out of the window where the
+    // user may still be in the Play sheet.
+    suspend fun armForBillingFlowLaunch() = arm(
+        name = WORK_NAME_LAUNCH,
+        policy = ExistingWorkPolicy.REPLACE,
+        expiresAt = System.currentTimeMillis() + BillingManager.ACK_SAFETY_NET_DEADLINE_MS,
+        initialDelayMs = LAUNCH_DELAY_MS,
+    )
+
+    // Any pending rescue already covers every unacknowledged purchase: KEEP it. Once completed
+    // work exists, KEEP inserts a fresh request. Short delay — the purchase already EXISTS (unlike
+    // the launch trigger), possibly for days, so waiting 30min could waste real deadline time.
+    // Accepted edge of KEEP, within the rescue lane only: a pending request keeps its original
+    // (possibly earlier) expiry; a newer purchase with a later deadline is only re-covered once a
+    // later pass re-arms after the old work completed. Bounded residual, only reachable via
+    // out-of-band purchases.
+    suspend fun armForUnackedPurchases(expiresAt: Long) = arm(
+        name = WORK_NAME_RESCUE,
+        policy = ExistingWorkPolicy.KEEP,
+        expiresAt = expiresAt,
+        initialDelayMs = DISCOVERY_DELAY_MS,
+    )
+
+    private suspend fun arm(
+        name: String,
+        policy: ExistingWorkPolicy,
+        expiresAt: Long,
+        initialDelayMs: Long,
+    ) {
+        if (expiresAt <= System.currentTimeMillis()) {
+            // Play has already voided (or is about to void) such a purchase; a sweep can't help.
+            log(TAG, WARN) { "arm($policy): deadline $expiresAt already passed, not scheduling" }
+            return
+        }
+        val request = OneTimeWorkRequestBuilder<PurchaseAckWorker>().apply {
+            setConstraints(
+                Constraints.Builder().apply {
+                    setRequiredNetworkType(NetworkType.CONNECTED)
+                }.build()
+            )
+            // Launch trigger: the worker must not run while the user may still be in the Play
+            // sheet — an immediate sweep would find nothing unacknowledged, report success, and
+            // complete the net before the purchase it exists for even happened.
+            setInitialDelay(initialDelayMs, TimeUnit.MILLISECONDS)
+            setBackoffCriteria(BackoffPolicy.EXPONENTIAL, BACKOFF_DELAY_MS, TimeUnit.MILLISECONDS)
+            setInputData(workDataOf(PurchaseAckWorker.KEY_EXPIRES_AT to expiresAt))
+        }.build()
+
+        // Await the enqueue: the caller arms this because the process may die at any moment — a
+        // fire-and-forget enqueue could be lost with it. Cancellable and BOUNDED: every caller
+        // needs a durable enqueue without an unbounded stall — a WorkManager that never settles
+        // must become an exception (handled fail-open by every caller) instead of a hang (which on
+        // the launch lane would park the purchase and its busy guard forever).
+        val operation = workManager.get().enqueueUniqueWork(name, policy, request)
+        withTimeoutOrNull(ENQUEUE_TIMEOUT_MS) { operation.await() }
+            ?: throw IllegalStateException("WorkManager enqueue did not settle within ${ENQUEUE_TIMEOUT_MS}ms")
+        log(TAG) { "arm($policy): safety net armed, expiresAt=$expiresAt" }
+    }
+
+    companion object {
+        // WorkManager persists these names AND the worker's class name in its DB across app
+        // updates: keep all of them stable while old work may exist (hence the version suffix for
+        // future changes). Separate identities per trigger: the launch watch's REPLACE must not be
+        // able to displace a pending rescue for a purchase that already exists.
+        private val WORK_NAME_LAUNCH = "${BuildConfigWrap.APPLICATION_ID}.gplay.purchase-ack.launch.v1"
+        private val WORK_NAME_RESCUE = "${BuildConfigWrap.APPLICATION_ID}.gplay.purchase-ack.rescue.v1"
+
+        private const val LAUNCH_DELAY_MS = 30 * 60 * 1000L
+        private const val DISCOVERY_DELAY_MS = 60 * 1000L
+        private const val BACKOFF_DELAY_MS = 30 * 60 * 1000L
+        private const val ENQUEUE_TIMEOUT_MS = 10 * 1000L
+
+        val TAG: String = logTag("Upgrade", "Gplay", "Billing", "AckScheduler")
+    }
+}
@@ -0,0 +1,86 @@
+package eu.darken.capod.common.upgrade.core.billing.work
+
+import android.content.Context
+import androidx.hilt.work.HiltWorker
+import androidx.work.CoroutineWorker
+import androidx.work.ListenableWorker.Result
+import androidx.work.WorkerParameters
+import dagger.assisted.Assisted
+import dagger.assisted.AssistedInject
+import eu.darken.capod.common.debug.logging.Logging.Priority.INFO
+import eu.darken.capod.common.debug.logging.Logging.Priority.WARN
+import eu.darken.capod.common.debug.logging.log
+import eu.darken.capod.common.debug.logging.logTag
+import eu.darken.capod.common.upgrade.core.billing.BillingManager
+import kotlinx.coroutines.withTimeoutOrNull
+
+/**
+ * Persistent acknowledgement safety net, armed by [PurchaseAckScheduler].
+ *
+ * Play auto-refunds (and revokes) any purchase not acknowledged within 3 days. The in-process ack
+ * machinery in [BillingManager] handles every case where the process lives long enough — this
+ * worker covers the case it can't: the process dies around the Play purchase sheet (OEM task
+ * killers) and the user doesn't reopen the app before the deadline. Play voids such purchases and
+ * revokes the entitlement, so the user loses what they signed up for.
+ *
+ * Self-completing by design: nothing cancels this work from the foreground ack path (an ack pass
+ * can legitimately see zero unacknowledged purchases while the Play sheet is still open, which
+ * must not tear down the net). The redundant sweep after a successful foreground ack is one
+ * purchase query.
+ */
+@HiltWorker
+class PurchaseAckWorker @AssistedInject constructor(
+    @Assisted private val context: Context,
+    @Assisted private val params: WorkerParameters,
+    private val billingManager: BillingManager,
+) : CoroutineWorker(context, params) {
+
+    override suspend fun doWork(): Result {
+        val expiresAt = inputData.getLong(KEY_EXPIRES_AT, 0L)
+        log(TAG) { "doWork(): attempt=$runAttemptCount, expiresAt=$expiresAt" }
+
+        if (!isWorthSweeping(System.currentTimeMillis(), expiresAt)) {
+            // Past Play's refund deadline (or malformed input): retrying can't achieve anything.
+            // failure() is deliberate over success() — it is visible in WorkManager diagnostics,
+            // and a completed state lets a later KEEP enqueue insert fresh work.
+            log(TAG, WARN) { "doWork(): deadline passed, giving up" }
+            return Result.failure()
+        }
+
+        // Bounded well below WorkManager's 10-minute execution limit, but generous enough for the
+        // connection wait plus the per-purchase inline retries. A sweep that ran out of time is a
+        // transient outcome, not a verdict. External cancellation propagates out of doWork — it
+        // must never be converted into success.
+        val sweep = withTimeoutOrNull(SWEEP_TIMEOUT_MS) {
+            billingManager.ensureAllAcknowledged()
+        }
+        log(TAG, INFO) { "doWork(): sweep=$sweep" }
+
+        return mapSweep(sweep, System.currentTimeMillis(), expiresAt)
+    }
+
+    companion object {
+        // Persisted in WorkManager's request data — keep the key stable while old work may exist.
+        const val KEY_EXPIRES_AT = "purchase.ack.expiresAt"
+
+        private const val SWEEP_TIMEOUT_MS = 4 * 60 * 1000L
+
+        // Pure so the retry/expiry decision is unit-testable without a WorkManager test harness.
+        internal fun isWorthSweeping(now: Long, expiresAt: Long): Boolean =
+            expiresAt > 0L && now < expiresAt
+
+        internal fun mapSweep(
+            sweep: BillingManager.AckSweepResult?,
+            now: Long,
+            expiresAt: Long,
+        ): Result = when (sweep) {
+            BillingManager.AckSweepResult.COMPLETE -> Result.success()
+            BillingManager.AckSweepResult.PERMANENT_FAILURE -> Result.failure()
+            // RETRY or timeout (null): keep trying until the deadline. WorkManager's exponential
+            // backoff caps at 5h, so the 3-day window still yields many attempts.
+            else -> if (now < expiresAt) Result.retry() else Result.failure()
+        }
+
+        val TAG: String = logTag("Upgrade", "Gplay", "Billing", "AckWorker")
+    }
+}