diff --git a/kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogBatchProcessor.kt b/kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogBatchProcessor.kt index d22fe38..5de5d3a 100644 --- a/kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogBatchProcessor.kt +++ b/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( 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. } } } diff --git a/kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogExportRetry.kt b/kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogExportRetry.kt new file mode 100644 index 0000000..a9710f7 --- /dev/null +++ b/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 = 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) + } +} diff --git a/kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogTelemetryRemoteImpl.kt b/kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogTelemetryRemoteImpl.kt index 1f8cb71..d949814 100644 --- a/kmp/src/commonMain/kotlin/com/onesignal/logger/internal/LogTelemetryRemoteImpl.kt +++ b/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? = null + /** Completed by [shutdown] to collapse any in-flight retry cycle. */ + private val shutdownSignal = CompletableDeferred() + private val batchProcessor = LogBatchProcessor( 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) { 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 { diff --git a/kmp/src/commonTest/kotlin/com/onesignal/logger/LogExportRetryTest.kt b/kmp/src/commonTest/kotlin/com/onesignal/logger/LogExportRetryTest.kt new file mode 100644 index 0000000..0694366 --- /dev/null +++ b/kmp/src/commonTest/kotlin/com/onesignal/logger/LogExportRetryTest.kt @@ -0,0 +1,357 @@ +package com.onesignal.logger + +import com.onesignal.logger.attributes.LogFieldsPerEvent +import com.onesignal.logger.attributes.LogFieldsTopLevel +import com.onesignal.logger.internal.ExportAttempt +import com.onesignal.logger.internal.ExportRetrier +import com.onesignal.logger.internal.LogTelemetryRemoteImpl +import com.onesignal.logger.internal.RetryPolicy +import com.onesignal.logger.internal.classifyStatus +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import kotlin.coroutines.cancellation.CancellationException +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +@OptIn(ExperimentalCoroutinesApi::class) +class LogExportRetryTest { + private fun remote( + scope: CoroutineScope, + http: FakeHttpSender, + ): LogTelemetryRemoteImpl { + val provider = FakePlatformProvider() + return LogTelemetryRemoteImpl( + platformProvider = provider, + httpSender = http, + topLevelFields = LogFieldsTopLevel(provider), + perEventFields = LogFieldsPerEvent(provider), + scope = scope, + ) + } + + private fun failure(statusCode: Int) = LogHttpResponse(success = false, statusCode = statusCode) + + private val ok = LogHttpResponse(success = true, statusCode = 200) + + @Test + fun happyPathPostsExactlyOnce() = runTest { + val http = FakeHttpSender() + val telemetry = remote(backgroundScope, http) + + telemetry.emit(LogRecord(LogSeverity.ERROR, "hello", emptyMap())) + telemetry.forceFlush() + + assertEquals(1, http.sentRequests.size) + } + + @Test + fun retryableStatusIsRetriedUntilSuccess() = runTest { + val http = FakeHttpSender(responses = ArrayDeque(listOf(failure(503), failure(429), ok))) + val telemetry = remote(backgroundScope, http) + + telemetry.emit(LogRecord(LogSeverity.ERROR, "hello", emptyMap())) + telemetry.forceFlush() + + assertEquals(3, http.sentRequests.size) + // Every attempt re-posts the same batch, not a truncated or re-encoded one. Comparing + // lengths would pass for any two distinct payloads that happen to be the same size, + // including a re-encode of the same records. + assertTrue(http.sentRequests[0].body.contentEquals(http.sentRequests[2].body)) + } + + // The crash path must stay single-shot. LogCrashUploader already retries across launches + // and stops on first failure, so routing exportEncoded through the retrier would give + // crash records 5 in-process attempts on top of that. These pin the split, which is + // otherwise invisible: every pre-existing test uses status 500, which classifies as + // PERMANENT and yields one attempt either way. + + @Test + fun exportEncodedIsNotRetriedOnARetryableStatus() = runTest { + val http = FakeHttpSender(responses = ArrayDeque(listOf(failure(503), ok))) + val telemetry = remote(backgroundScope, http) + + val succeeded = telemetry.exportEncoded(byteArrayOf(1, 2, 3)) + + assertFalse(succeeded) + assertEquals(1, http.sentRequests.size) + } + + @Test + fun exportEncodedIsNotRetriedWhenTheSenderThrows() = runTest { + val http = + FakeHttpSender( + responses = ArrayDeque(listOf(ok)), + exceptions = ArrayDeque(listOf(RuntimeException("socket reset"))), + ) + val telemetry = remote(backgroundScope, http) + + // A thrown sender maps to RETRYABLE on the batched path. The crash path must not + // reinterpret it that way — it surfaces to LogCrashUploader, which stops on first + // failure and keeps the record for the next launch. + assertFailsWith { + telemetry.exportEncoded(byteArrayOf(1, 2, 3)) + } + + assertEquals(1, http.sentRequests.size) + } + + @Test + fun transportFailureIsRetried() = runTest { + val http = + FakeHttpSender( + responses = ArrayDeque(listOf(failure(-1), ok)), + exceptions = ArrayDeque(listOf(RuntimeException("connection reset"))), + ) + val telemetry = remote(backgroundScope, http) + + telemetry.emit(LogRecord(LogSeverity.ERROR, "hello", emptyMap())) + telemetry.forceFlush() + + // Thrown send, then statusCode -1, then success. + assertEquals(3, http.sentRequests.size) + } + + @Test + fun permanentStatusIsNotRetried() = runTest { + val http = FakeHttpSender(defaultResponse = failure(400)) + val telemetry = remote(backgroundScope, http) + + telemetry.emit(LogRecord(LogSeverity.ERROR, "hello", emptyMap())) + telemetry.forceFlush() + + assertEquals(1, http.sentRequests.size) + } + + @Test + fun attemptCapIsHonoredWhenBackendKeepsFailing() = runTest { + val http = FakeHttpSender(defaultResponse = failure(503)) + val telemetry = remote(backgroundScope, http) + + telemetry.emit(LogRecord(LogSeverity.ERROR, "hello", emptyMap())) + telemetry.forceFlush() + + assertEquals(RetryPolicy().maxAttempts, http.sentRequests.size) + } + + /** + * Records the virtual time at which each attempt starts, so the gaps between them are + * the backoffs the retrier actually slept. `nextRandom = 0.5` is the midpoint of the + * jitter spread, i.e. a factor of exactly 1.0, which makes the schedule deterministic. + */ + private suspend fun TestScope.backoffsFor( + policy: RetryPolicy, + nextRandom: () -> Double = { 0.5 }, + ): List { + val attemptTimes = mutableListOf() + ExportRetrier(policy = policy, nextRandom = nextRandom).execute { + attemptTimes += testScheduler.currentTime + ExportAttempt.RETRYABLE + } + return attemptTimes.zipWithNext { previous, next -> next - previous } + } + + @Test + fun backoffFollowsTheAdvertisedSchedule() = runTest { + // The numbers in RetryPolicy's KDoc and in the PR table are a promise; pin them. + // 1s initial, 1.6x each time, clamped at the 5s per-delay ceiling. + val backoffs = backoffsFor(RetryPolicy(maxAttempts = 7, maxTotalBackoffMillis = 60_000L)) + + assertEquals(listOf(1_000L, 1_600L, 2_560L, 4_096L, 5_000L, 5_000L), backoffs) + } + + @Test + fun jitterScalesEachDelayByTheConfiguredFactor() = runTest { + // +/-20% around the nominal delay, so a fleet backing off together re-spreads. + val policy = RetryPolicy(maxAttempts = 3, maxTotalBackoffMillis = 60_000L) + + assertEquals(listOf(800L, 1_280L), backoffsFor(policy, nextRandom = { 0.0 })) + assertEquals(listOf(1_200L, 1_920L), backoffsFor(policy, nextRandom = { 1.0 })) + } + + @Test + fun theBackoffBudgetClampsTheLastDelayAndThenStopsRetrying() = runTest { + // The budget is charged against sleeping only, so it is exact rather than dependent + // on how long each attempt took: 1000 + 1600 leaves 400 of a 3s budget, and the + // fourth attempt never starts. Asserting the clamped 400 rather than just the count + // is what stops a "return early instead of clamping" regression from passing. + var attempts = 0 + val attemptTimes = mutableListOf() + val succeeded = + ExportRetrier( + policy = RetryPolicy(maxAttempts = 10, maxTotalBackoffMillis = 3_000L), + nextRandom = { 0.5 }, + ).execute { + attempts++ + attemptTimes += testScheduler.currentTime + ExportAttempt.RETRYABLE + } + + assertFalse(succeeded) + assertEquals(4, attempts) + assertEquals(listOf(1_000L, 1_600L, 400L), attemptTimes.zipWithNext { a, b -> b - a }) + } + + @Test + fun theAttemptCapHoldsNoMatterHowSlowEachAttemptIs() = runTest { + // The bound this policy actually enforces. A wall-clock ceiling would cut this to + // two attempts against a sender sitting on its 10s connect timeout — the slow-failure + // case retry exists for — so the attempt count has to survive slow attempts intact. + var attempts = 0 + ExportRetrier(policy = RetryPolicy(), nextRandom = { 0.5 }).execute { + attempts++ + delay(10_000L) // a connect timeout, not a fast 503 + ExportAttempt.RETRYABLE + } + + assertEquals(RetryPolicy().maxAttempts, attempts) + } + + @Test + fun anAbortSignalWakesAnInFlightBackoffImmediately() = runTest { + // Teardown's lever: without it, shutdown queues behind the whole retry cycle holding + // the export mutex, which outlasts its own flush timeout. + val abort = CompletableDeferred() + var attempts = 0 + val job = + backgroundScope.launch { + ExportRetrier( + policy = RetryPolicy(maxAttempts = 10, initialBackoffMillis = 30_000L), + nextRandom = { 0.5 }, + ).execute(abortSignal = abort) { + attempts++ + ExportAttempt.RETRYABLE + } + } + + runCurrent() + assertEquals(1, attempts) // parked in a 30s backoff + + abort.complete(Unit) + runCurrent() + + assertTrue(job.isCompleted) + assertEquals(0L, testScheduler.currentTime) // returned without waiting out the backoff + assertEquals(1, attempts) + } + + @Test + fun anAlreadyAbortedSignalStillLetsTheAttemptInFlightFinish() = runTest { + // Abort stops retrying, it does not cancel work. Shutdown's own flush goes through + // this path with the signal already completed and must still post once. + val abort = CompletableDeferred(Unit) + var attempts = 0 + + val succeeded = + ExportRetrier(policy = RetryPolicy(), nextRandom = { 0.5 }) + .execute(abortSignal = abort) { + attempts++ + ExportAttempt.SUCCESS + } + + assertTrue(succeeded) + assertEquals(1, attempts) + } + + @Test + fun shutdownFlushesOnceWithoutEnteringARetryCycle() = runTest { + // shutdown() blocks its caller — on Android a lifecycle thread — so its flush must + // not be able to start a retry cycle whose backoffs outlast the flush budget. + // + // This is the integration-level guard on the abort wiring: it goes through the real + // telemetry, and dropping `abortSignal = shutdownSignal` from exportBatch makes it + // post three times instead of once. A variant that parks in a backoff *before* + // teardown would read better, but shutdown() is deliberately non-suspend and uses + // runBlocking, so it does not interleave with runTest's virtual clock and such a test + // passes whether or not the abort is wired. anAbortSignalWakesAnInFlightBackoff- + // Immediately covers the wake-from-backoff behaviour directly on the retrier instead. + val http = FakeHttpSender(defaultResponse = failure(503)) + val telemetry = remote(backgroundScope, http) + + telemetry.emit(LogRecord(LogSeverity.ERROR, "hello", emptyMap())) + telemetry.shutdown() + + assertEquals(1, http.sentRequests.size) + } + + @Test + fun aCancelledSenderIsNotReclassifiedAsATransientFailure() = runTest { + // Pins the `catch (e: CancellationException) { throw e }` in attemptPost, which reads + // as redundant next to the catch-all below it. It is not: CancellationException is an + // Exception in Kotlin, so without it a cancelled send is classified RETRYABLE and the + // retrier keeps posting — burning the full attempt budget on a scope that is already + // going away, and losing the cancellation entirely on the paths that return before + // the next delay(). + // + // This has to go through the real telemetry rather than a bare retrier: passing a + // throwing lambda straight to ExportRetrier.execute bypasses attemptPost, so such a + // test passes whether or not the guard exists. + val http = + FakeHttpSender( + exceptions = ArrayDeque(listOf(CancellationException("scope cancelled mid-send"))), + ) + val telemetry = remote(backgroundScope, http) + + telemetry.emit(LogRecord(LogSeverity.ERROR, "hello", emptyMap())) + assertFailsWith { telemetry.forceFlush() } + + // One send, not maxAttempts: cancellation stopped the loop instead of feeding it. + assertEquals(1, http.sentRequests.size) + } + + @Test + fun cancellationDuringBackoffDelayPropagates() = runTest { + var attempts = 0 + val retrier = + ExportRetrier( + policy = RetryPolicy(maxAttempts = 10), + nextRandom = { 0.5 }, + ) + val job = + backgroundScope.launch { + retrier.execute { + attempts++ + ExportAttempt.RETRYABLE + } + } + + runCurrent() + assertEquals(1, attempts) // parked in the first backoff delay + + job.cancel() + runCurrent() + assertTrue(job.isCancelled) + + advanceTimeBy(60_000) + runCurrent() + assertEquals(1, attempts) // never woke up for another attempt + } + + @Test + fun aRequestThatCouldNotBeBuiltIsPermanent() { + // iOS reports -3 when URL construction fails. Retrying cannot help, and reusing the + // -1 transport sentinel for it would burn the full budget on every batch against a + // misconfiguration that never resolves. + assertEquals(ExportAttempt.PERMANENT, classifyStatus(success = false, statusCode = -3)) + } + + @Test + fun classifiesStatusCodes() { + assertEquals(ExportAttempt.SUCCESS, classifyStatus(success = true, statusCode = 200)) + listOf(429, 502, 503, 504, -1).forEach { + assertEquals(ExportAttempt.RETRYABLE, classifyStatus(success = false, statusCode = it), "status $it") + } + listOf(400, 401, 403, 404, 500, -2).forEach { + assertEquals(ExportAttempt.PERMANENT, classifyStatus(success = false, statusCode = it), "status $it") + } + } +} diff --git a/kmp/src/commonTest/kotlin/com/onesignal/logger/TestFakes.kt b/kmp/src/commonTest/kotlin/com/onesignal/logger/TestFakes.kt index 20f2660..6d1ceea 100644 --- a/kmp/src/commonTest/kotlin/com/onesignal/logger/TestFakes.kt +++ b/kmp/src/commonTest/kotlin/com/onesignal/logger/TestFakes.kt @@ -45,13 +45,29 @@ internal class FakePlatformProvider( internal class FakeHttpSender( var responses: ArrayDeque = ArrayDeque(), var defaultResponse: LogHttpResponse = LogHttpResponse(success = true, statusCode = 200), + /** Drained before [responses]; each entry makes one send throw, as a transport failure would. */ + var exceptions: ArrayDeque = ArrayDeque(), ) : ILogHttpSender { private val mutex = Mutex() val sentRequests = mutableListOf() override suspend fun send(request: LogHttpRequest): LogHttpResponse { - mutex.withLock { sentRequests.add(request) } - return if (responses.isNotEmpty()) responses.removeFirst() else defaultResponse + // Both queues are drained under the same lock as the recording. The retry tests are + // the first to drive this repeatedly, and a fake that advertises thread-safety it + // only half-provides is worse than one that makes no claim. + val queued = + mutex.withLock { + sentRequests.add(request) + // A queued exception short-circuits without consuming a response, so the two + // queues stay independent the way callers expect. + if (exceptions.isNotEmpty()) { + exceptions.removeFirst() to null + } else { + null to (if (responses.isNotEmpty()) responses.removeFirst() else defaultResponse) + } + } + queued.first?.let { throw it } + return queued.second!! } fun lastBodyAsString(): String = sentRequests.last().body.decodeToString()