feat: [SDK-5065] add bounded retry/backoff to remote log export

kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogBatchProcessor.kt

@@ -7,6 +7,7 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.coroutines.cancellation.CancellationException
 
 /**
  * Coroutine-based batch processor. Hand-rolled, dependency-light replacement for
@@ -43,10 +44,14 @@ internal class LogBatchProcessor<T>(
                 withTimeoutOrNull(scheduleDelayMillis) { flushSignal.receive() }
                 try {
                     drainAndExport()
+                } catch (e: CancellationException) {
+                    // CancellationException is an Exception in Kotlin, so the catch below
+                    // would swallow it and spin this loop until isActive flips. Shutdown
+                    // and scope cancellation both depend on it unwinding here.
+                    throw e
                 } catch (_: Exception) {
-                    // Keep the consumer alive. The failed batch was already drained —
-                    // best-effort drop (no retry), matching a failed HTTP post that
-                    // returns success=false without re-queueing.
+                    // Keep the consumer alive. [onExport] owns any retry policy; by the
+                    // time it throws, the batch is unrecoverable and already drained.
                 }
             }
         }

kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogExportRetry.kt

@@ -0,0 +1,142 @@
+package com.onesignal.logger.internal
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.math.min
+import kotlin.random.Random
+
+/** Classification of a single export attempt, driving the retry decision. */
+internal enum class ExportAttempt {
+    SUCCESS,
+
+    /** Transient: worth another attempt after a backoff (5xx-ish, throttling, transport). */
+    RETRYABLE,
+
+    /** Permanent: the same request will keep failing, so retrying only wastes battery. */
+    PERMANENT,
+}
+
+/**
+ * Bounds for [ExportRetrier]. Defaults mirror what OpenTelemetry's okhttp sender applied
+ * by default before it was removed: 5 attempts, 1s initial backoff growing by 1.6x up to
+ * 5s, 20% jitter.
+ *
+ * [maxTotalBackoffMillis] bounds *sleeping only* — see [ExportRetrier] for why there is no
+ * wall-clock bound. At the default attempt count the schedule already sums to ~9.3s
+ * (1 + 1.6 + 2.56 + 4.096, ±20% jitter), so this ceiling binds only if [maxAttempts] is
+ * raised; it exists so that raising it cannot silently produce an unbounded sleep.
+ */
+internal data class RetryPolicy(
+    val maxAttempts: Int = 5,
+    val initialBackoffMillis: Long = 1_000L,
+    val maxBackoffMillis: Long = 5_000L,
+    val backoffMultiplier: Double = 1.6,
+    val jitterFactor: Double = 0.2,
+    val maxTotalBackoffMillis: Long = 15_000L,
+)
+
+/** Statuses OpenTelemetry's `RetryUtil` treated as retryable. */
+private val RETRYABLE_STATUS_CODES = setOf(429, 502, 503, 504)
+
+/**
+ * Sentinel both platform senders already report for a transport-level failure
+ * (no HTTP response at all): DNS, connect/read timeout, socket reset.
+ *
+ * This means "the request went out and nothing usable came back", which is worth retrying.
+ * A failure to *build* the request is not — it will fail identically every time — so senders
+ * must report that separately (iOS uses -3) and let it fall through to permanent below.
+ */
+internal const val TRANSPORT_FAILURE_STATUS_CODE = -1
+
+internal fun classifyStatus(
+    success: Boolean,
+    statusCode: Int,
+): ExportAttempt =
+    when {
+        success -> ExportAttempt.SUCCESS
+        statusCode == TRANSPORT_FAILURE_STATUS_CODE -> ExportAttempt.RETRYABLE
+        statusCode in RETRYABLE_STATUS_CODES -> ExportAttempt.RETRYABLE
+        // Everything else is permanent: 4xx, plus the sender sentinels for conditions that
+        // cannot resolve themselves — -2 "remote logging disabled" and -3 "request could not
+        // be built" (a malformed base URL or app id).
+        else -> ExportAttempt.PERMANENT
+    }
+
+/**
+ * Retries a single export with exponential backoff and jitter, bounded by an attempt count
+ * and by the total time spent *sleeping between* attempts.
+ *
+ * ### There is no wall-clock bound, on purpose
+ *
+ * This retrier cannot cancel a send once [execute]'s `attempt` is running — `ILogHttpSender`
+ * takes no deadline and neither platform implementation accepts one. A wall-clock ceiling
+ * would therefore only ever be checked between attempts, which does not bound anything: an
+ * attempt starting just inside the budget still runs to the sender's own timeout (10s on
+ * both platforms). Worse, it under-delivers precisely where retry matters most — against
+ * connect timeouts, a 15s ceiling with 10s timeouts yields ~2 attempts, not the advertised 5.
+ *
+ * So the attempt count is the real bound and it is honored regardless of how slow each
+ * attempt is. Stated plainly, the worst case for one export is
+ * `maxAttempts × senderTimeout + maxTotalBackoffMillis` — with today's defaults and a 10s
+ * sender timeout, ~59s of a *background* export coroutine, with no caller blocked on it.
+ *
+ * Callers that do need a prompt exit — teardown, in particular — pass an `abortSignal`:
+ * completing it wakes an in-flight backoff immediately and stops further attempts, so the
+ * wait collapses to at most the attempt already in flight rather than the full cycle.
+ *
+ * Waiting is suspension, never a blocking sleep, so a cancelled scope unwinds promptly and
+ * `CancellationException` propagates to the caller rather than being swallowed.
+ *
+ * [nextRandom] is injectable purely so tests can pin the jitter; production uses
+ * [Random.Default].
+ */
+internal class ExportRetrier(
+    private val policy: RetryPolicy = RetryPolicy(),
+    private val nextRandom: () -> Double = { Random.nextDouble() },
+) {
+    /**
+     * Runs [attempt] until it succeeds, fails permanently, or the bounds are reached.
+     * Completing [abortSignal] cuts a backoff short and prevents further attempts; the
+     * attempt already in flight is left to finish.
+     *
+     * Returns true only if [attempt] ultimately reported success.
+     */
+    suspend fun execute(
+        abortSignal: Deferred<Unit> = CompletableDeferred(),
+        attempt: suspend () -> ExportAttempt,
+    ): Boolean {
+        var attemptsMade = 0
+        var backoffMillis = policy.initialBackoffMillis
+        var backoffSpentMillis = 0L
+
+        while (true) {
+            attemptsMade++
+            when (attempt()) {
+                ExportAttempt.SUCCESS -> return true
+                ExportAttempt.PERMANENT -> return false
+                ExportAttempt.RETRYABLE -> Unit
+            }
+
+            if (attemptsMade >= policy.maxAttempts) return false
+            if (abortSignal.isCompleted) return false
+
+            val remainingBackoffMillis = policy.maxTotalBackoffMillis - backoffSpentMillis
+            if (remainingBackoffMillis <= 0) return false
+
+            val waitMillis = min(jittered(backoffMillis), remainingBackoffMillis)
+            backoffSpentMillis += waitMillis
+            backoffMillis = min((backoffMillis * policy.backoffMultiplier).toLong(), policy.maxBackoffMillis)
+
+            // Sleep for the backoff, but wake early if the caller aborts.
+            val aborted = withTimeoutOrNull(waitMillis) { abortSignal.await() } != null
+            if (aborted) return false
+        }
+    }
+
+    /** Spreads the backoff over +/- [RetryPolicy.jitterFactor] so clients do not sync up. */
+    private fun jittered(backoffMillis: Long): Long {
+        val spread = 1.0 - policy.jitterFactor + (2.0 * policy.jitterFactor * nextRandom())
+        return (backoffMillis * spread).toLong().coerceAtLeast(0L)
+    }
+}

kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogTelemetryRemoteImpl.kt

@@ -9,6 +9,7 @@ import com.onesignal.logger.attributes.LogFieldsPerEvent
 import com.onesignal.logger.attributes.LogFieldsTopLevel
 import com.onesignal.logger.otlp.EncodableRecord
 import com.onesignal.logger.otlp.OtlpLogEncoder
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
@@ -17,6 +18,7 @@ import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.coroutines.cancellation.CancellationException
 
 /**
  * Remote telemetry sink: batches records and ships them as OTLP/protobuf over the
@@ -29,13 +31,21 @@ internal class LogTelemetryRemoteImpl(
     private val topLevelFields: LogFieldsTopLevel,
     private val perEventFields: LogFieldsPerEvent,
     private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
+    private val retrier: ExportRetrier = ExportRetrier(),
 ) : ILogTelemetryRemote {
     companion object {
         private const val MAX_QUEUE_SIZE = 100
         private const val MAX_BATCH_SIZE = 100
         private const val SCHEDULE_DELAY_MILLIS = 1_000L
 
-        /** Cap so a hung HTTP send cannot block app teardown indefinitely. */
+        /**
+         * Cap so a hung HTTP send cannot block app teardown indefinitely. It only has to
+         * cover a single send, not a retry cycle: [shutdownSignal] stops the retrier first,
+         * so nothing here can be waiting out a backoff. Deliberately shorter than either
+         * platform sender's own 10s timeout — [shutdown] runs under `runBlocking`, and on
+         * Android that is a lifecycle thread (a log-level change routes through it), so
+         * dropping a hung batch beats blocking that thread for the sender's full timeout.
+         */
         private const val SHUTDOWN_FLUSH_TIMEOUT_MILLIS = 5_000L
     }
 
@@ -52,6 +62,9 @@ internal class LogTelemetryRemoteImpl(
     private val resourceMutex = Mutex()
     private var cachedResourceAttributes: Map<String, String>? = null
 
+    /** Completed by [shutdown] to collapse any in-flight retry cycle. */
+    private val shutdownSignal = CompletableDeferred<Unit>()
+
     private val batchProcessor =
         LogBatchProcessor<EncodableRecord>(
             scope = scope,
@@ -79,13 +92,45 @@ internal class LogTelemetryRemoteImpl(
         )
     }
 
+    /**
+     * Batched export retries transient failures in place. The batch being retried is the
+     * only one held — records arriving meanwhile keep filling the processor's bounded
+     * queue and are dropped past `maxQueueSize`, so memory stays capped at two batches.
+     *
+     * The retry runs under the processor's export mutex, which [forceFlush] and [shutdown]
+     * both need, hence [shutdownSignal].
+     */
     private suspend fun exportBatch(records: List<EncodableRecord>) {
         val payload = OtlpLogEncoder.encode(getResourceAttributes(), records)
-        post(payload)
+        retrier.execute(abortSignal = shutdownSignal) { attemptPost(payload) }
     }
 
     override suspend fun exportEncoded(payload: ByteArray): Boolean = post(payload)
 
+    private suspend fun attemptPost(payload: ByteArray): ExportAttempt =
+        try {
+            val response =
+                httpSender.send(
+                    LogHttpRequest(
+                        url = endpoint,
+                        headers = headers,
+                        contentType = OtlpLogEncoder.CONTENT_TYPE,
+                        body = payload,
+                    ),
+                )
+            classifyStatus(response.success, response.statusCode)
+        } catch (e: CancellationException) {
+            // Not redundant with the catch below: CancellationException is an Exception in
+            // Kotlin, so without this a cancelled scope is misread as a transient backend
+            // failure and the retrier keeps going. On the paths that return without
+            // suspending again — attempt cap reached, backoff budget spent, abort signalled —
+            // nothing would rethrow it and the cancellation would be lost entirely.
+            throw e
+        } catch (_: Exception) {
+            // A thrown sender is a transport failure, same as statusCode -1.
+            ExportAttempt.RETRYABLE
+        }
+
     private suspend fun post(payload: ByteArray): Boolean {
         val response =
             httpSender.send(
@@ -102,6 +147,13 @@ internal class LogTelemetryRemoteImpl(
     override suspend fun forceFlush() = batchProcessor.flush()
 
     override fun shutdown() {
+        // Stop the retrier before asking for the flush. Without this, teardown landing on a
+        // backend blip queues behind a retry cycle whose backoffs alone outlast the flush
+        // budget below: the batch is dropped anyway, only after blocking the caller for the
+        // full timeout. The aborted batch is not requeued — it is mid-retry precisely because
+        // the backend is rejecting it, so a final attempt would just cost another round trip.
+        shutdownSignal.complete(Unit)
+
         // Best-effort flush before teardown. Bounded so a hung sender cannot block
         // app disable/teardown; remaining buffered records are dropped on cancel.
         try {