From 28837d59828c10a6574c24201619d7c92676e408 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Tue, 13 Oct 2026 16:40:21 -0700 Subject: [PATCH 1/7] fix: [SDK-5535] retry failed operations without blocking the queue A FAIL_RETRY op now waits on its own instead of stalling the whole OperationRepo loop. Ops on the same record still wait behind it, so a create runs before its updates. Retry-After is honored and capped at 1h. Foreground and network return end the backoff early. Co-Authored-By: Claude Opus 5.5 --- .../java/com/onesignal/core/CoreModule.kt | 2 + .../internal/operations/impl/OperationRepo.kt | 101 ++++++++++------ .../operations/impl/OperationRetryTrigger.kt | 57 +++++++++ .../internal/operations/OperationRepoTests.kt | 111 +++++++++++++++++- .../operations/OperationRetryTriggerTests.kt | 80 +++++++++++++ 5 files changed, 307 insertions(+), 44 deletions(-) create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt create mode 100644 OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt index 811867103..a1fa95a7c 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt @@ -34,6 +34,7 @@ import com.onesignal.core.internal.language.impl.LanguageContext import com.onesignal.core.internal.operations.IOperationRepo import com.onesignal.core.internal.operations.impl.OperationModelStore import com.onesignal.core.internal.operations.impl.OperationRepo +import com.onesignal.core.internal.operations.impl.OperationRetryTrigger import com.onesignal.core.internal.permissions.IRequestPermissionService import com.onesignal.core.internal.permissions.impl.RequestPermissionService import com.onesignal.core.internal.preferences.IPreferencesService @@ -91,6 +92,7 @@ internal class CoreModule : IModule { builder.register() .provides() .provides() + builder.register().provides() // Permissions builder.register() diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt index 37e64cdcc..ac91d818c 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt @@ -27,7 +27,9 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.newSingleThreadContext import kotlinx.coroutines.withTimeoutOrNull import java.util.UUID +import java.util.concurrent.TimeUnit import kotlin.math.max +import kotlin.math.min import kotlin.reflect.KClass internal class OperationRepo( @@ -46,6 +48,12 @@ internal class OperationRepo( val bucket: Int, var retries: Int = 0, ) { + // Backoff can be cut short by forceExecuteOperations; Retry-After cannot. + var backoffUntil: Long = 0 + var retryAfterUntil: Long = 0 + + fun retryAt(): Long = max(backoffUntil, retryAfterUntil) + val waiters = mutableListOf>() @Volatile @@ -78,7 +86,7 @@ internal class OperationRepo( } override fun toString(): String { - return "bucket:$bucket, retries:$retries, operation:$operation\n" + return "bucket:$bucket, retries:$retries, retryAt:${retryAt()}, operation:$operation\n" } } @@ -99,7 +107,6 @@ internal class OperationRepo( internal val queue = mutableListOf() private val inFlightLogins = mutableMapOf() private val waiter = WaiterWithValue() - private val retryWaiter = WaiterWithValue() private var paused = false private val initialized = CompletableDeferred() @@ -328,7 +335,9 @@ internal class OperationRepo( } override fun forceExecuteOperations() { - retryWaiter.wake(LoopWaiterMessage(true)) + synchronized(queue) { + queue.forEach { it.backoffUntil = 0 } + } waiter.wake(LoopWaiterMessage(false)) } @@ -393,8 +402,14 @@ internal class OperationRepo( * with a budget. */ private suspend fun waitForNewOperationAndExecutionInterval() { - // 1. Wait for an operation to be enqueued - var wakeMessage = waiter.waitForWake() + // 1. Wait for an operation to be enqueued, or for a backed-off one to become ready + val untilNextRetry = msUntilNextRetry() + var wakeMessage = + if (untilNextRetry == null) { + waiter.waitForWake() + } else { + withTimeoutOrNull(untilNextRetry) { waiter.waitForWake() } ?: LoopWaiterMessage(true) + } // 2. Now wait opRepoExecutionInterval, restart the wait // time everytime something new is enqueued, to ensure @@ -432,7 +447,6 @@ internal class OperationRepo( response.idTranslations.values.forEach { _newRecordState.add(it) } } - var highestRetries = 0 when (response.result) { ExecutionResult.SUCCESS -> { // on success we remove the operation from the store and wake any waiters @@ -484,12 +498,15 @@ internal class OperationRepo( } ExecutionResult.FAIL_RETRY -> { Logging.info("Operation execution failed, retrying: $operations") - // add back all operations to the front of the queue to be re-executed. + // Re-queue at the front with a per-op wait so unrelated ops keep running. + val now = _time.currentTimeMillis + val retryAfterMs = min(TimeUnit.SECONDS.toMillis((response.retryAfterSeconds ?: 0).toLong()), MAX_RETRY_DELAY_MS) synchronized(queue) { ops.reversed().forEach { - if (++it.retries > highestRetries) { - highestRetries = it.retries - } + it.retries++ + val backoffMs = min(it.retries * _configModelStore.model.opRepoDefaultFailRetryBackoff, MAX_RETRY_DELAY_MS) + it.backoffUntil = now + backoffMs + it.retryAfterUntil = now + retryAfterMs queue.add(0, it) } } @@ -524,8 +541,6 @@ internal class OperationRepo( } } - // wait for retry and post create waiters to start next operation - delayBeforeNextExecution(highestRetries, response.retryAfterSeconds) if (response.idTranslations != null) { delayForPostCreate(_configModelStore.model.opRepoPostCreateDelay) } @@ -551,25 +566,6 @@ internal class OperationRepo( ops.forEach { it.wakeWaiters(response.toWaitResult(false)) } } - /** - * Wait which ever is longer, retryAfterSeconds returned by the server, - * or based on the retry count. - */ - suspend fun delayBeforeNextExecution( - retries: Int, - retryAfterSeconds: Int?, - ) { - Logging.debug("retryAfterSeconds: $retryAfterSeconds") - val retryAfterSecondsNonNull = retryAfterSeconds?.toLong() ?: 0L - val delayForOnRetries = retries * _configModelStore.model.opRepoDefaultFailRetryBackoff - val delayFor = max(delayForOnRetries, retryAfterSecondsNonNull * 1_000) - if (delayFor < 1) return - Logging.debug("Operations being delay for: $delayFor ms") - withTimeoutOrNull(delayFor) { - retryWaiter.waitForWake() - } - } - /** * Stall processing the queue so the backend's DB has to time * reflect the change before we do any other operations to it. @@ -601,10 +597,18 @@ internal class OperationRepo( // Snapshot gate state once per pass so all queue items see the same IV view. val newCodePathsRun = _identityVerificationService.newCodePathsRun val ivBehaviorActive = _identityVerificationService.ivBehaviorActive + val now = _time.currentTimeMillis return synchronized(queue) { + // An op must not overtake an earlier op on the same record that is waiting to retry. + val waitingRecords = mutableSetOf() val startingOp = queue.firstOrNull { - it.operation.canStartExecute && + if (it.retryAt() > now) { + waitingRecords.add(it.operation.applyToRecordId) + return@firstOrNull false + } + it.operation.applyToRecordId !in waitingRecords && + it.operation.canStartExecute && _newRecordState.canAccess(it.operation.applyToRecordId) && it.bucket <= bucketFilter && // Outer gate: skip IV JWT check entirely on old code path. @@ -613,7 +617,7 @@ internal class OperationRepo( if (startingOp != null) { queue.remove(startingOp) - val grouped = getGroupableOperations(startingOp) + val grouped = getGroupableOperations(startingOp, now) grouped.forEach { item -> val login = item.operation as? LoginUserOperation ?: return@forEach inFlightLogins[login.onesignalId] = item @@ -632,7 +636,10 @@ internal class OperationRepo( * * THIS SHOULD BE CALLED WHILE THE QUEUE IS SYNCHRONIZED!! */ - private fun getGroupableOperations(startingOp: OperationQueueItem): List { + private fun getGroupableOperations( + startingOp: OperationQueueItem, + now: Long, + ): List { val ops = mutableListOf(startingOp) if (startingOp.operation.groupComparisonType == GroupComparisonType.NONE) { @@ -646,7 +653,13 @@ internal class OperationRepo( startingOp.operation.modifyComparisonKey } + val waitingRecords = mutableSetOf() for (item in queue.toList()) { + if (item.retryAt() > now) { + waitingRecords.add(item.operation.applyToRecordId) + continue + } + val itemKey = if (startingOp.operation.groupComparisonType == GroupComparisonType.CREATE) { item.operation.createComparisonKey @@ -658,11 +671,10 @@ internal class OperationRepo( throw Exception("Both comparison keys can not be blank!") } - if (!_newRecordState.canAccess(item.operation.applyToRecordId)) { - continue - } - - if (itemKey == startingKey) { + if (itemKey == startingKey && + _newRecordState.canAccess(item.operation.applyToRecordId) && + item.operation.applyToRecordId !in waitingRecords + ) { queue.remove(item) ops.add(item) } @@ -671,6 +683,13 @@ internal class OperationRepo( return ops } + private fun msUntilNextRetry(): Long? { + val now = _time.currentTimeMillis + return synchronized(queue) { + queue.map { it.retryAt() }.filter { it > now }.minOrNull()?.let { it - now } + } + } + /** * Load saved operations from preference service and add them into the queue * NOTE: Sometimes the loading might take longer than expected due to I/O reads from disk, @@ -688,4 +707,8 @@ internal class OperationRepo( } initialized.complete(Unit) } + + companion object { + private val MAX_RETRY_DELAY_MS = TimeUnit.HOURS.toMillis(1) + } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt new file mode 100644 index 000000000..19932ed5b --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt @@ -0,0 +1,57 @@ +package com.onesignal.core.internal.operations.impl + +import android.content.Context +import android.net.ConnectivityManager +import android.net.Network +import android.net.NetworkCapabilities +import android.net.NetworkRequest +import com.onesignal.core.internal.application.IApplicationLifecycleHandler +import com.onesignal.core.internal.application.IApplicationService +import com.onesignal.core.internal.operations.IOperationRepo +import com.onesignal.core.internal.startup.IStartableService +import com.onesignal.debug.internal.logging.Logging + +/** + * Ends the retry backoff of queued operations when the app comes to the foreground + * or a network becomes available. + */ +internal class OperationRetryTrigger( + private val applicationService: IApplicationService, + private val operationRepo: IOperationRepo, +) : IStartableService, IApplicationLifecycleHandler { + override fun start() { + applicationService.addApplicationLifecycleHandler(this) + registerNetworkCallback() + } + + override fun onFocus(firedOnSubscribe: Boolean) { + if (firedOnSubscribe) return + operationRepo.forceExecuteOperations() + } + + override fun onUnfocused() {} + + @Suppress("TooGenericExceptionCaught") + private fun registerNetworkCallback() { + try { + val connectivityManager = + applicationService.appContext.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager + ?: return + val request = + NetworkRequest.Builder() + .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) + .build() + connectivityManager.registerNetworkCallback( + request, + object : ConnectivityManager.NetworkCallback() { + override fun onAvailable(network: Network) { + operationRepo.forceExecuteOperations() + } + }, + ) + } catch (t: Throwable) { + // Can throw, e.g. when the app already holds the maximum number of network callbacks. + Logging.warn("OperationRetryTrigger: unable to register network callback", t) + } + } +} diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt index 7b67d3ccd..dec98ece1 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt @@ -609,7 +609,6 @@ class OperationRepoTests : FunSpec({ // Given val mocks = Mocks() val opRepo = mocks.operationRepo - coEvery { opRepo.delayBeforeNextExecution(any(), any()) } just runs coEvery { mocks.executor.execute(any()) } returns ExecutionResponse(ExecutionResult.FAIL_RETRY) andThen ExecutionResponse(ExecutionResult.SUCCESS) @@ -633,7 +632,6 @@ class OperationRepoTests : FunSpec({ it[0] shouldBe operation }, ) - opRepo.delayBeforeNextExecution(1, null) mocks.executor.execute( withArg { it.count() shouldBe 1 @@ -1074,7 +1072,7 @@ class OperationRepoTests : FunSpec({ // ensure the order: IDs are translated, operation removed from the store, then delay for postCreateDelay operation.translateIds(idTranslation) mocks.operationModelStore.remove(opId) - mocks.operationRepo.delayBeforeNextExecution(any(), any()) + mocks.operationRepo.delayForPostCreate(any()) } } @@ -1152,13 +1150,116 @@ class OperationRepoTests : FunSpec({ mocks.operationRepo.queue[1].operation shouldBe op2 } + test("a retrying operation does not block operations on other records") { + // Given + val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 60_000 + val opRepo = mocks.operationRepo + val failing = mockOperation(applyToRecordId = "record-a") + val unrelated = mockOperation(applyToRecordId = "record-b") + coEvery { mocks.executor.execute(listOf(failing)) } returns ExecutionResponse(ExecutionResult.FAIL_RETRY) + + // When + opRepo.start() + opRepo.enqueue(failing) + mocks.waitForInternalEnqueue() + val response = withTimeoutOrNull(1_000) { opRepo.enqueueAndWait(unrelated) } + + // Then + response shouldBe true + opRepo.queue.map { it.operation } shouldBe listOf(failing) + } + + test("an operation waits behind a retrying operation on the same record and keeps its order") { + // Given + val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 200 + val opRepo = mocks.operationRepo + val first = mockOperation(applyToRecordId = "record-a") + val second = mockOperation(applyToRecordId = "record-a") + val executed = mutableListOf() + var firstAttempts = 0 + coEvery { mocks.executor.execute(any()) } answers { + val ops = firstArg>() + executed.addAll(ops) + if (ops.first() === first && firstAttempts++ == 0) { + ExecutionResponse(ExecutionResult.FAIL_RETRY) + } else { + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + + // When + opRepo.start() + opRepo.enqueue(first) + mocks.waitForInternalEnqueue() + val response = withTimeoutOrNull(2_000) { opRepo.enqueueAndWait(second) } + + // Then + response shouldBe true + executed shouldBe listOf(first, first, second) + } + + test("a waiting operation is not grouped into another operation's request") { + // Given + val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 60_000 + val opRepo = mocks.operationRepo + val waiting = mockOperation(groupComparisonType = GroupComparisonType.ALTER, applyToRecordId = "record-a") + val starter = mockOperation(groupComparisonType = GroupComparisonType.ALTER, applyToRecordId = "record-b") + opRepo.queue.add(OperationQueueItem(waiting, bucket = 0).apply { backoffUntil = Long.MAX_VALUE }) + opRepo.queue.add(OperationQueueItem(starter, bucket = 0)) + + // When + val ops = opRepo.getNextOps(0) + + // Then + ops?.map { it.operation } shouldBe listOf(starter) + opRepo.queue.map { it.operation } shouldBe listOf(waiting) + } + + test("Retry-After is capped at one hour") { + // Given + val mocks = Mocks() + val opRepo = mocks.operationRepo + coEvery { mocks.executor.execute(any()) } returns ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = 10 * 60 * 60) + + // When + val before = System.currentTimeMillis() + opRepo.executeOperations(listOf(OperationQueueItem(mockOperation(), bucket = 0))) + val after = System.currentTimeMillis() + + // Then + val retryAfterUntil = opRepo.queue.single().retryAfterUntil + (retryAfterUntil >= before + 60 * 60 * 1_000L) shouldBe true + (retryAfterUntil <= after + 60 * 60 * 1_000L) shouldBe true + } + + test("forceExecuteOperations ends backoff but still honors Retry-After") { + // Given + val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 60_000 + val opRepo = mocks.operationRepo + val backedOff = OperationQueueItem(mockOperation(), bucket = 0).apply { backoffUntil = Long.MAX_VALUE } + val rateLimited = OperationQueueItem(mockOperation(), bucket = 0).apply { retryAfterUntil = Long.MAX_VALUE } + opRepo.queue.addAll(listOf(backedOff, rateLimited)) + + // When + opRepo.forceExecuteOperations() + + // Then + backedOff.retryAt() shouldBe 0 + rateLimited.retryAt() shouldBe Long.MAX_VALUE + } + test("ensure forceExecuteOperations immediately wakes retry waiters") { // Given val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 5_000 val opRepo = mocks.operationRepo coEvery { mocks.executor.execute(any()) - } returns ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = 5) andThen ExecutionResponse(ExecutionResult.SUCCESS) + } returns ExecutionResponse(ExecutionResult.FAIL_RETRY) andThen ExecutionResponse(ExecutionResult.SUCCESS) // When opRepo.start() @@ -1168,7 +1269,7 @@ class OperationRepoTests : FunSpec({ withTimeoutOrNull(999) { opRepo.enqueueAndWait(mockOperation()) } - // Forcing execution to bypass 5 second retry period + // Forcing execution to bypass 5 second backoff opRepo.forceExecuteOperations() val response2 = withTimeoutOrNull(100) { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt new file mode 100644 index 000000000..b627c1178 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt @@ -0,0 +1,80 @@ +package com.onesignal.core.internal.operations + +import android.content.Context +import android.net.ConnectivityManager +import android.net.Network +import android.net.NetworkRequest +import android.os.Build +import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest +import com.onesignal.core.internal.application.IApplicationLifecycleHandler +import com.onesignal.core.internal.application.IApplicationService +import com.onesignal.core.internal.operations.impl.OperationRetryTrigger +import io.kotest.core.spec.style.FunSpec +import io.mockk.every +import io.mockk.just +import io.mockk.mockk +import io.mockk.runs +import io.mockk.slot +import io.mockk.verify +import org.robolectric.annotation.Config + +private class TriggerMocks { + val lifecycleHandler = slot() + val networkCallback = slot() + val connectivityManager = + mockk { + every { registerNetworkCallback(any(), capture(networkCallback)) } just runs + } + val applicationService = + mockk { + every { addApplicationLifecycleHandler(capture(lifecycleHandler)) } just runs + every { appContext } returns + mockk { + every { getSystemService(Context.CONNECTIVITY_SERVICE) } returns connectivityManager + } + } + val operationRepo = mockk(relaxed = true) + + fun start() = OperationRetryTrigger(applicationService, operationRepo).start() +} + +@RobolectricTest +@Config(sdk = [Build.VERSION_CODES.O]) +class OperationRetryTriggerTests : FunSpec({ + test("coming to the foreground forces queued operations to retry") { + val mocks = TriggerMocks() + mocks.start() + + mocks.lifecycleHandler.captured.onFocus(firedOnSubscribe = false) + + verify(exactly = 1) { mocks.operationRepo.forceExecuteOperations() } + } + + test("focus reported on subscribe does not force a retry") { + val mocks = TriggerMocks() + mocks.start() + + mocks.lifecycleHandler.captured.onFocus(firedOnSubscribe = true) + + verify(exactly = 0) { mocks.operationRepo.forceExecuteOperations() } + } + + test("network becoming available forces queued operations to retry") { + val mocks = TriggerMocks() + mocks.start() + + mocks.networkCallback.captured.onAvailable(mockk()) + + verify(exactly = 1) { mocks.operationRepo.forceExecuteOperations() } + } + + test("a failure registering the network callback does not crash start") { + val mocks = TriggerMocks() + every { mocks.connectivityManager.registerNetworkCallback(any(), any()) } throws + SecurityException() + + mocks.start() + + verify(exactly = 1) { mocks.applicationService.addApplicationLifecycleHandler(any()) } + } +}) From 7b8d7c35f5de4938b0ae1d11bd59e02a43b4d8ce Mon Sep 17 00:00:00 2001 From: Fadi George Date: Tue, 13 Oct 2026 18:34:16 -0700 Subject: [PATCH 2/7] fix: [SDK-5535] stop HttpClient from holding every request during Retry-After Operations and the IAM fetch wait out Retry-After themselves, matching iOS. Co-Authored-By: Claude Opus 5.5 --- .../core/internal/http/impl/HttpClient.kt | 14 ----------- .../core/internal/http/HttpClientTests.kt | 25 ++++++------------- 2 files changed, 7 insertions(+), 32 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt index e1dd89e5b..f457d8793 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt @@ -15,10 +15,8 @@ import com.onesignal.core.internal.http.IHttpClient import com.onesignal.core.internal.preferences.IPreferencesService import com.onesignal.core.internal.preferences.PreferenceOneSignalKeys import com.onesignal.core.internal.preferences.PreferenceStores -import com.onesignal.core.internal.time.ITime import com.onesignal.debug.internal.logging.Logging import kotlinx.coroutines.TimeoutCancellationException -import kotlinx.coroutines.delay import kotlinx.coroutines.withTimeout import org.json.JSONObject import java.net.ConnectException @@ -35,16 +33,9 @@ internal class HttpClient( private val _connectionFactory: IHttpConnectionFactory, private val _prefs: IPreferencesService, private val _configModelStore: ConfigModelStore, - private val _time: ITime, private val _installIdService: IInstallIdService, private val _fidEnv: IFidEnv, ) : IHttpClient { - /** - * Delay making network requests until we reach this time. - * Used when the OneSignal backend returns a Retry-After value. - */ - private var delayNewRequestsUntil = 0L - override suspend fun post( url: String, body: JSONObject, @@ -88,9 +79,6 @@ internal class HttpClient( return HttpResponse(0, null, null) } - val delayUntil = delayNewRequestsUntil - _time.currentTimeMillis - if (delayUntil > 0) delay(delayUntil) - try { return withTimeout(getThreadTimeout(timeout).toLong()) { return@withTimeout makeRequestIODispatcher(url, method, jsonBody, timeout, headers) @@ -218,8 +206,6 @@ internal class HttpClient( val retryAfter = retryAfterFromResponse(con) val retryLimit = retryLimitFromResponse(con) - val newDelayUntil = _time.currentTimeMillis + (retryAfter ?: 0) * 1_000 - if (newDelayUntil > delayNewRequestsUntil) delayNewRequestsUntil = newDelayUntil when (httpResponse) { HttpURLConnection.HTTP_NOT_MODIFIED -> { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt index 569800cff..76c7c812f 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt @@ -6,7 +6,6 @@ import com.onesignal.core.internal.device.impl.HTTP_FID_ENV_HEADER_KEY import com.onesignal.core.internal.device.impl.InstallIdService import com.onesignal.core.internal.http.impl.HttpClient import com.onesignal.core.internal.http.impl.OptionalHeaders -import com.onesignal.core.internal.time.impl.Time import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging import com.onesignal.mocks.MockHelper @@ -27,7 +26,7 @@ class Mocks { internal val installIdService = InstallIdService(MockPreferencesService()) internal val fidEnv = FakeFidEnv() internal val httpClient by lazy { - HttpClient(factory, MockPreferencesService(), mockConfigModel, Time(), installIdService, fidEnv) + HttpClient(factory, MockPreferencesService(), mockConfigModel, installIdService, fidEnv) } } @@ -221,19 +220,14 @@ class HttpClientTests : FunSpec({ response.retryAfterSeconds shouldBe 60 } - // If the OneSignal server ever responses with a Retry-After we want to - // make sure we delay any future requests until that delay has past. To - // be safe we assume the server's Retry-After value means all traffic - // should be delayed until then. If we wanted finer grain control we - // could work with the the backend to decide which endpoints should be - // affected. - test("ensure next request is delayed by the Retry-After value") { + // Callers wait out Retry-After themselves, so other requests are not held. + test("a Retry-After response does not delay the next request") { // Given val mocks = Mocks() val mockRetryAfterResponse = MockHttpConnectionFactory.MockResponse() mockRetryAfterResponse.status = 429 - mockRetryAfterResponse.mockProps["Retry-After"] = "1" + mockRetryAfterResponse.mockProps["Retry-After"] = "10" mockRetryAfterResponse.errorResponseBody = "{}" val mockSuccessfulResponse = MockHttpConnectionFactory.MockResponse() @@ -246,18 +240,13 @@ class HttpClientTests : FunSpec({ mocks.factory.mockResponse = mockSuccessfulResponse val response2 = - withTimeoutOrNull(975) { - mocks.httpClient.post("URL", JSONObject()) - } - - val response3 = - withTimeoutOrNull(100) { + withTimeoutOrNull(1_000) { mocks.httpClient.post("URL", JSONObject()) } // Then - response2 shouldBe null - response3 shouldNotBe null + response2 shouldNotBe null + response2!!.statusCode shouldBe 200 } test("Authorization header is set when OptionalHeaders.jwt is provided") { From b94a1cdb1087fdd027312a0d2b6d6373e35a4a63 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Wed, 14 Oct 2026 12:02:26 -0700 Subject: [PATCH 3/7] fix: [SDK-5535] address review on retry backoff - Keep HttpClient's shared Retry-After gate, capped by maxRetryDelay, but fail fast with a 429 instead of suspending, so no caller stalls on it. - Login executors pass Retry-After through. - One retry deadline per operation, on elapsedRealtime. A force ends it; an active Retry-After still holds through the gate. - A force during execute is kept, and skips the batching wait when it ends a retry wait. - Drop the network callback; foreground still ends the wait. - maxRetryDelay moves to ConfigModel; one debug line per retry. Co-Authored-By: Claude Opus 5.5 --- .../core/internal/config/ConfigModel.kt | 9 +++ .../core/internal/http/impl/HttpClient.kt | 22 ++++++ .../internal/operations/impl/OperationRepo.kt | 57 ++++++++------ .../operations/impl/OperationRetryTrigger.kt | 34 +------- .../com/onesignal/core/internal/time/ITime.kt | 6 ++ .../onesignal/core/internal/time/impl/Time.kt | 2 + ...inUserFromSubscriptionOperationExecutor.kt | 2 +- .../executors/LoginUserOperationExecutor.kt | 2 +- .../core/internal/http/HttpClientTests.kt | 54 ++++++++++--- .../internal/operations/OperationRepoTests.kt | 78 +++++++++++++++---- .../operations/OperationRetryTriggerTests.kt | 37 --------- .../RecoverFromDroppedLoginBugTests.kt | 3 +- .../java/com/onesignal/mocks/MockHelper.kt | 9 +++ 13 files changed, 191 insertions(+), 124 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt index 959b29f5a..408dcfbba 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt @@ -191,6 +191,15 @@ class ConfigModel : Model() { setLongProperty(::opRepoDefaultFailRetryBackoff.name, value) } + /** + * The longest any retry waits, whether from backoff or a server Retry-After, in milliseconds. + */ + var maxRetryDelay: Long + get() = getLongProperty(::maxRetryDelay.name) { 3_600_000 } + set(value) { + setLongProperty(::maxRetryDelay.name, value) + } + /** * How long [com.onesignal.core.internal.operations.IOperationRepo.enqueueAndAwaitResult] waits. * The operation stays queued after the waiter is released. diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt index f457d8793..96d296251 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt @@ -15,6 +15,7 @@ import com.onesignal.core.internal.http.IHttpClient import com.onesignal.core.internal.preferences.IPreferencesService import com.onesignal.core.internal.preferences.PreferenceOneSignalKeys import com.onesignal.core.internal.preferences.PreferenceStores +import com.onesignal.core.internal.time.ITime import com.onesignal.debug.internal.logging.Logging import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.withTimeout @@ -24,7 +25,9 @@ import java.net.HttpURLConnection import java.net.URL import java.net.UnknownHostException import java.util.Scanner +import java.util.concurrent.TimeUnit import javax.net.ssl.HttpsURLConnection +import kotlin.math.min internal const val HTTP_SDK_VERSION_HEADER_KEY = "SDK-Version" internal val HTTP_SDK_VERSION_HEADER_VALUE = "onesignal/android/${OneSignalUtils.sdkVersion}" @@ -33,9 +36,17 @@ internal class HttpClient( private val _connectionFactory: IHttpConnectionFactory, private val _prefs: IPreferencesService, private val _configModelStore: ConfigModelStore, + private val _time: ITime, private val _installIdService: IInstallIdService, private val _fidEnv: IFidEnv, ) : IHttpClient { + /** + * Set from a backend Retry-After, on [ITime.elapsedRealtimeMillis]. Until then requests fail + * fast with a 429 instead of being sent, so callers back off without stalling here. + */ + @Volatile + private var delayNewRequestsUntil = 0L + override suspend fun post( url: String, body: JSONObject, @@ -79,6 +90,13 @@ internal class HttpClient( return HttpResponse(0, null, null) } + val delayMs = delayNewRequestsUntil - _time.elapsedRealtimeMillis + if (delayMs > 0) { + val retryAfterSeconds = TimeUnit.MILLISECONDS.toSeconds(delayMs - 1).toInt() + 1 + Logging.debug("HttpClient: Not sending ${method ?: "GET"} $url, Retry-After has ${retryAfterSeconds}s left") + return HttpResponse(HTTP_TOO_MANY_REQUESTS, null, retryAfterSeconds = retryAfterSeconds) + } + try { return withTimeout(getThreadTimeout(timeout).toLong()) { return@withTimeout makeRequestIODispatcher(url, method, jsonBody, timeout, headers) @@ -206,6 +224,9 @@ internal class HttpClient( val retryAfter = retryAfterFromResponse(con) val retryLimit = retryLimitFromResponse(con) + val retryAfterMs = min(TimeUnit.SECONDS.toMillis((retryAfter ?: 0).toLong()), _configModelStore.model.maxRetryDelay) + val newDelayUntil = _time.elapsedRealtimeMillis + retryAfterMs + if (newDelayUntil > delayNewRequestsUntil) delayNewRequestsUntil = newDelayUntil when (httpResponse) { HttpURLConnection.HTTP_NOT_MODIFIED -> { @@ -339,5 +360,6 @@ internal class HttpClient( private const val OS_API_VERSION = "1" private const val OS_ACCEPT_HEADER = "application/vnd.onesignal.v$OS_API_VERSION+json" private const val THREAD_ID = 10000 + private const val HTTP_TOO_MANY_REQUESTS = 429 } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt index ac91d818c..b6e5f8344 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt @@ -48,11 +48,8 @@ internal class OperationRepo( val bucket: Int, var retries: Int = 0, ) { - // Backoff can be cut short by forceExecuteOperations; Retry-After cannot. - var backoffUntil: Long = 0 - var retryAfterUntil: Long = 0 - - fun retryAt(): Long = max(backoffUntil, retryAfterUntil) + // On ITime.elapsedRealtimeMillis. + var retryAt: Long = 0 val waiters = mutableListOf>() @@ -86,7 +83,7 @@ internal class OperationRepo( } override fun toString(): String { - return "bucket:$bucket, retries:$retries, retryAt:${retryAt()}, operation:$operation\n" + return "bucket:$bucket, retries:$retries, retryAt:$retryAt, operation:$operation\n" } } @@ -108,6 +105,10 @@ internal class OperationRepo( private val inFlightLogins = mutableMapOf() private val waiter = WaiterWithValue() private var paused = false + + // Guarded by queue. Set by a force that lands while ops are out of the queue executing. + private var forcedDuringExecute = false + private val initialized = CompletableDeferred() override suspend fun awaitInitialized() { @@ -335,10 +336,17 @@ internal class OperationRepo( } override fun forceExecuteOperations() { - synchronized(queue) { - queue.forEach { it.backoffUntil = 0 } - } - waiter.wake(LoopWaiterMessage(false)) + val now = _time.elapsedRealtimeMillis + val endedWait = + synchronized(queue) { + forcedDuringExecute = true + val waiting = queue.filter { it.retryAt > now } + waiting.forEach { it.retryAt = 0 } + waiting.isNotEmpty() + } + // Skip the batching wait only when a retry was actually held back. An active Retry-After + // still holds: HttpClient fails the request fast and the op is rescheduled. + waiter.wake(LoopWaiterMessage(endedWait)) } /** @@ -499,14 +507,18 @@ internal class OperationRepo( ExecutionResult.FAIL_RETRY -> { Logging.info("Operation execution failed, retrying: $operations") // Re-queue at the front with a per-op wait so unrelated ops keep running. - val now = _time.currentTimeMillis - val retryAfterMs = min(TimeUnit.SECONDS.toMillis((response.retryAfterSeconds ?: 0).toLong()), MAX_RETRY_DELAY_MS) + val now = _time.elapsedRealtimeMillis + val maxDelay = _configModelStore.model.maxRetryDelay + val retryAfterMs = min(TimeUnit.SECONDS.toMillis((response.retryAfterSeconds ?: 0).toLong()), maxDelay) synchronized(queue) { + // A force that arrived mid-request would otherwise be lost to the new backoff. + val forced = forcedDuringExecute ops.reversed().forEach { it.retries++ - val backoffMs = min(it.retries * _configModelStore.model.opRepoDefaultFailRetryBackoff, MAX_RETRY_DELAY_MS) - it.backoffUntil = now + backoffMs - it.retryAfterUntil = now + retryAfterMs + val backoffMs = if (forced) 0L else min(it.retries * _configModelStore.model.opRepoDefaultFailRetryBackoff, maxDelay) + val delayMs = max(backoffMs, retryAfterMs) + it.retryAt = if (delayMs > 0) now + delayMs else 0 + Logging.debug("OperationRepo: retry #${it.retries} of ${it.operation.name} in $delayMs ms (backoff: $backoffMs ms, Retry-After: $retryAfterMs ms)") queue.add(0, it) } } @@ -597,13 +609,13 @@ internal class OperationRepo( // Snapshot gate state once per pass so all queue items see the same IV view. val newCodePathsRun = _identityVerificationService.newCodePathsRun val ivBehaviorActive = _identityVerificationService.ivBehaviorActive - val now = _time.currentTimeMillis + val now = _time.elapsedRealtimeMillis return synchronized(queue) { // An op must not overtake an earlier op on the same record that is waiting to retry. val waitingRecords = mutableSetOf() val startingOp = queue.firstOrNull { - if (it.retryAt() > now) { + if (it.retryAt > now) { waitingRecords.add(it.operation.applyToRecordId) return@firstOrNull false } @@ -617,6 +629,7 @@ internal class OperationRepo( if (startingOp != null) { queue.remove(startingOp) + forcedDuringExecute = false val grouped = getGroupableOperations(startingOp, now) grouped.forEach { item -> val login = item.operation as? LoginUserOperation ?: return@forEach @@ -655,7 +668,7 @@ internal class OperationRepo( val waitingRecords = mutableSetOf() for (item in queue.toList()) { - if (item.retryAt() > now) { + if (item.retryAt > now) { waitingRecords.add(item.operation.applyToRecordId) continue } @@ -684,9 +697,9 @@ internal class OperationRepo( } private fun msUntilNextRetry(): Long? { - val now = _time.currentTimeMillis + val now = _time.elapsedRealtimeMillis return synchronized(queue) { - queue.map { it.retryAt() }.filter { it > now }.minOrNull()?.let { it - now } + queue.map { it.retryAt }.filter { it > now }.minOrNull()?.let { it - now } } } @@ -707,8 +720,4 @@ internal class OperationRepo( } initialized.complete(Unit) } - - companion object { - private val MAX_RETRY_DELAY_MS = TimeUnit.HOURS.toMillis(1) - } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt index 19932ed5b..34d7d907a 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt @@ -1,19 +1,12 @@ package com.onesignal.core.internal.operations.impl -import android.content.Context -import android.net.ConnectivityManager -import android.net.Network -import android.net.NetworkCapabilities -import android.net.NetworkRequest import com.onesignal.core.internal.application.IApplicationLifecycleHandler import com.onesignal.core.internal.application.IApplicationService import com.onesignal.core.internal.operations.IOperationRepo import com.onesignal.core.internal.startup.IStartableService -import com.onesignal.debug.internal.logging.Logging /** - * Ends the retry backoff of queued operations when the app comes to the foreground - * or a network becomes available. + * Ends the retry backoff of queued operations when the app comes to the foreground. */ internal class OperationRetryTrigger( private val applicationService: IApplicationService, @@ -21,7 +14,6 @@ internal class OperationRetryTrigger( ) : IStartableService, IApplicationLifecycleHandler { override fun start() { applicationService.addApplicationLifecycleHandler(this) - registerNetworkCallback() } override fun onFocus(firedOnSubscribe: Boolean) { @@ -30,28 +22,4 @@ internal class OperationRetryTrigger( } override fun onUnfocused() {} - - @Suppress("TooGenericExceptionCaught") - private fun registerNetworkCallback() { - try { - val connectivityManager = - applicationService.appContext.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager - ?: return - val request = - NetworkRequest.Builder() - .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) - .build() - connectivityManager.registerNetworkCallback( - request, - object : ConnectivityManager.NetworkCallback() { - override fun onAvailable(network: Network) { - operationRepo.forceExecuteOperations() - } - }, - ) - } catch (t: Throwable) { - // Can throw, e.g. when the app already holds the maximum number of network callbacks. - Logging.warn("OperationRetryTrigger: unable to register network callback", t) - } - } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/ITime.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/ITime.kt index 8f1824d48..86f212654 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/ITime.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/ITime.kt @@ -15,4 +15,10 @@ interface ITime { * Returns how long the app has been running. */ val processUptimeMillis: Long + + /** + * Milliseconds since boot, including deep sleep. Unaffected by device clock changes, so use + * it for deadlines within this process. + */ + val elapsedRealtimeMillis: Long } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/impl/Time.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/impl/Time.kt index 753ef124d..b82502316 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/impl/Time.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/impl/Time.kt @@ -8,6 +8,8 @@ import com.onesignal.core.internal.time.ITime internal class Time : ITime { override val currentTimeMillis: Long get() = System.currentTimeMillis() + override val elapsedRealtimeMillis: Long + get() = SystemClock.elapsedRealtime() override val processUptimeMillis: Long @RequiresApi(Build.VERSION_CODES.N) get() = SystemClock.uptimeMillis() - android.os.Process.getStartUptimeMillis() diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserFromSubscriptionOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserFromSubscriptionOperationExecutor.kt index 76d555185..3d16eb5e1 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserFromSubscriptionOperationExecutor.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserFromSubscriptionOperationExecutor.kt @@ -90,7 +90,7 @@ internal class LoginUserFromSubscriptionOperationExecutor( return when (responseType) { NetworkUtils.ResponseStatusType.RETRYABLE -> - ExecutionResponse(ExecutionResult.FAIL_RETRY) + ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = ex.retryAfterSeconds) NetworkUtils.ResponseStatusType.UNAUTHORIZED -> ExecutionResponse(ExecutionResult.FAIL_UNAUTHORIZED) else -> diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserOperationExecutor.kt index 2a359e910..200baaf3d 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserOperationExecutor.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserOperationExecutor.kt @@ -160,7 +160,7 @@ internal class LoginUserOperationExecutor( ) createUser(loginUserOp, operations) } - else -> ExecutionResponse(result.result) + else -> ExecutionResponse(result.result, retryAfterSeconds = result.retryAfterSeconds) } } } diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt index 76c7c812f..e8767f672 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt @@ -6,6 +6,7 @@ import com.onesignal.core.internal.device.impl.HTTP_FID_ENV_HEADER_KEY import com.onesignal.core.internal.device.impl.InstallIdService import com.onesignal.core.internal.http.impl.HttpClient import com.onesignal.core.internal.http.impl.OptionalHeaders +import com.onesignal.core.internal.time.ITime import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging import com.onesignal.mocks.MockHelper @@ -15,8 +16,9 @@ import io.kotest.matchers.should import io.kotest.matchers.shouldBe import io.kotest.matchers.shouldNotBe import io.kotest.matchers.types.beInstanceOf +import io.mockk.every +import io.mockk.mockk import kotlinx.coroutines.TimeoutCancellationException -import kotlinx.coroutines.withTimeoutOrNull import org.json.JSONObject class Mocks { @@ -25,8 +27,10 @@ class Mocks { internal val factory = MockHttpConnectionFactory(response) internal val installIdService = InstallIdService(MockPreferencesService()) internal val fidEnv = FakeFidEnv() + var now = 1_000_000L + private val time = mockk { every { elapsedRealtimeMillis } answers { now } } internal val httpClient by lazy { - HttpClient(factory, MockPreferencesService(), mockConfigModel, installIdService, fidEnv) + HttpClient(factory, MockPreferencesService(), mockConfigModel, time, installIdService, fidEnv) } } @@ -220,8 +224,9 @@ class HttpClientTests : FunSpec({ response.retryAfterSeconds shouldBe 60 } - // Callers wait out Retry-After themselves, so other requests are not held. - test("a Retry-After response does not delay the next request") { + // Retry-After applies to all traffic, but requests fail fast instead of waiting so a single + // caller, like the operation queue, never stalls on it. + test("after a Retry-After, requests fail fast without being sent until it passes") { // Given val mocks = Mocks() @@ -239,14 +244,43 @@ class HttpClientTests : FunSpec({ mocks.httpClient.post("URL", JSONObject()) mocks.factory.mockResponse = mockSuccessfulResponse - val response2 = - withTimeoutOrNull(1_000) { - mocks.httpClient.post("URL", JSONObject()) - } + mocks.now += 4_500 + val response2 = mocks.httpClient.post("URL", JSONObject()) + + mocks.now += 5_500 + val response3 = mocks.httpClient.post("URL", JSONObject()) + + // Then + response2.statusCode shouldBe 429 + response2.retryAfterSeconds shouldBe 6 + response3.statusCode shouldBe 200 + mocks.factory.connections.size shouldBe 2 + } + + test("the Retry-After window is capped by maxRetryDelay") { + // Given + val mocks = Mocks() + mocks.mockConfigModel.model.maxRetryDelay = 2_000 + + val mockRetryAfterResponse = MockHttpConnectionFactory.MockResponse() + mockRetryAfterResponse.status = 429 + mockRetryAfterResponse.mockProps["Retry-After"] = "3600" + mockRetryAfterResponse.errorResponseBody = "{}" + + val mockSuccessfulResponse = MockHttpConnectionFactory.MockResponse() + mockSuccessfulResponse.status = 200 + mockSuccessfulResponse.responseBody = "{}" + + // When + mocks.factory.mockResponse = mockRetryAfterResponse + mocks.httpClient.post("URL", JSONObject()) + + mocks.factory.mockResponse = mockSuccessfulResponse + mocks.now += 2_000 + val response = mocks.httpClient.post("URL", JSONObject()) // Then - response2 shouldNotBe null - response2!!.statusCode shouldBe 200 + response.statusCode shouldBe 200 } test("Authorization header is set when OptionalHeaders.jwt is provided") { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt index dec98ece1..463b0c6b9 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt @@ -8,7 +8,6 @@ import com.onesignal.core.internal.operations.impl.OperationRepo import com.onesignal.core.internal.operations.impl.OperationRepo.OperationQueueItem import com.onesignal.core.internal.preferences.PreferenceOneSignalKeys import com.onesignal.core.internal.preferences.PreferenceStores -import com.onesignal.core.internal.time.impl.Time import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging import com.onesignal.mocks.CoreInternalMocks @@ -38,6 +37,7 @@ import io.mockk.runs import io.mockk.slot import io.mockk.spyk import io.mockk.verify +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.delay import kotlinx.coroutines.launch @@ -90,7 +90,7 @@ private class Mocks { listOf(executor), operationModelStore, configModelStore, - Time(), + MockHelper.systemTime(), getNewRecordState(configModelStore), jwtTokenStore, identityVerificationService, @@ -117,7 +117,7 @@ class OperationRepoTests : FunSpec({ listOf(mocks.executor), operationModelStore, mocks.configModelStore, - Time(), + MockHelper.systemTime(), getNewRecordState(mocks.configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(), @@ -1207,7 +1207,7 @@ class OperationRepoTests : FunSpec({ val opRepo = mocks.operationRepo val waiting = mockOperation(groupComparisonType = GroupComparisonType.ALTER, applyToRecordId = "record-a") val starter = mockOperation(groupComparisonType = GroupComparisonType.ALTER, applyToRecordId = "record-b") - opRepo.queue.add(OperationQueueItem(waiting, bucket = 0).apply { backoffUntil = Long.MAX_VALUE }) + opRepo.queue.add(OperationQueueItem(waiting, bucket = 0).apply { retryAt = Long.MAX_VALUE }) opRepo.queue.add(OperationQueueItem(starter, bucket = 0)) // When @@ -1218,38 +1218,84 @@ class OperationRepoTests : FunSpec({ opRepo.queue.map { it.operation } shouldBe listOf(waiting) } - test("Retry-After is capped at one hour") { + test("Retry-After is capped by maxRetryDelay") { // Given val mocks = Mocks() val opRepo = mocks.operationRepo coEvery { mocks.executor.execute(any()) } returns ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = 10 * 60 * 60) // When - val before = System.currentTimeMillis() + val before = System.nanoTime() / 1_000_000 opRepo.executeOperations(listOf(OperationQueueItem(mockOperation(), bucket = 0))) - val after = System.currentTimeMillis() + val after = System.nanoTime() / 1_000_000 // Then - val retryAfterUntil = opRepo.queue.single().retryAfterUntil - (retryAfterUntil >= before + 60 * 60 * 1_000L) shouldBe true - (retryAfterUntil <= after + 60 * 60 * 1_000L) shouldBe true + val retryAt = opRepo.queue.single().retryAt + (retryAt >= before + 60 * 60 * 1_000L) shouldBe true + (retryAt <= after + 60 * 60 * 1_000L) shouldBe true } - test("forceExecuteOperations ends backoff but still honors Retry-After") { + test("forceExecuteOperations ends the wait of retrying operations") { + // Given + val mocks = Mocks() + val opRepo = mocks.operationRepo + val waiting = OperationQueueItem(mockOperation(), bucket = 0).apply { retryAt = Long.MAX_VALUE } + opRepo.queue.add(waiting) + + // When + opRepo.forceExecuteOperations() + + // Then + waiting.retryAt shouldBe 0 + } + + test("forceExecuteOperations skips the batching wait when it ends a retry wait") { // Given val mocks = Mocks() mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 60_000 val opRepo = mocks.operationRepo - val backedOff = OperationQueueItem(mockOperation(), bucket = 0).apply { backoffUntil = Long.MAX_VALUE } - val rateLimited = OperationQueueItem(mockOperation(), bucket = 0).apply { retryAfterUntil = Long.MAX_VALUE } - opRepo.queue.addAll(listOf(backedOff, rateLimited)) + var attempts = 0 + val retried = CompletableDeferred() + coEvery { mocks.executor.execute(any()) } answers { + if (attempts++ == 0) { + ExecutionResponse(ExecutionResult.FAIL_RETRY) + } else { + retried.complete(Unit) + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + opRepo.start() + opRepo.enqueue(mockOperation()) + mocks.waitForInternalEnqueue() + withTimeoutOrNull(1_000) { while (opRepo.queue.firstOrNull()?.retries != 1) delay(10) } + mocks.configModelStore.model.opRepoExecutionInterval = 60_000 // When opRepo.forceExecuteOperations() // Then - backedOff.retryAt() shouldBe 0 - rateLimited.retryAt() shouldBe Long.MAX_VALUE + withTimeoutOrNull(1_000) { retried.await() } shouldBe Unit + } + + test("a force during execute skips the next backoff but keeps Retry-After") { + // Given + val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 60_000 + val opRepo = mocks.operationRepo + coEvery { mocks.executor.execute(any()) } answers { + opRepo.forceExecuteOperations() + ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = 30) + } + + // When + val before = System.nanoTime() / 1_000_000 + opRepo.executeOperations(listOf(OperationQueueItem(mockOperation(), bucket = 0))) + val after = System.nanoTime() / 1_000_000 + + // Then + val retryAt = opRepo.queue.single().retryAt + (retryAt >= before + 30_000) shouldBe true + (retryAt <= after + 30_000) shouldBe true } test("ensure forceExecuteOperations immediately wakes retry waiters") { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt index b627c1178..6ca94f9bc 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt @@ -1,11 +1,5 @@ package com.onesignal.core.internal.operations -import android.content.Context -import android.net.ConnectivityManager -import android.net.Network -import android.net.NetworkRequest -import android.os.Build -import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest import com.onesignal.core.internal.application.IApplicationLifecycleHandler import com.onesignal.core.internal.application.IApplicationService import com.onesignal.core.internal.operations.impl.OperationRetryTrigger @@ -16,30 +10,18 @@ import io.mockk.mockk import io.mockk.runs import io.mockk.slot import io.mockk.verify -import org.robolectric.annotation.Config private class TriggerMocks { val lifecycleHandler = slot() - val networkCallback = slot() - val connectivityManager = - mockk { - every { registerNetworkCallback(any(), capture(networkCallback)) } just runs - } val applicationService = mockk { every { addApplicationLifecycleHandler(capture(lifecycleHandler)) } just runs - every { appContext } returns - mockk { - every { getSystemService(Context.CONNECTIVITY_SERVICE) } returns connectivityManager - } } val operationRepo = mockk(relaxed = true) fun start() = OperationRetryTrigger(applicationService, operationRepo).start() } -@RobolectricTest -@Config(sdk = [Build.VERSION_CODES.O]) class OperationRetryTriggerTests : FunSpec({ test("coming to the foreground forces queued operations to retry") { val mocks = TriggerMocks() @@ -58,23 +40,4 @@ class OperationRetryTriggerTests : FunSpec({ verify(exactly = 0) { mocks.operationRepo.forceExecuteOperations() } } - - test("network becoming available forces queued operations to retry") { - val mocks = TriggerMocks() - mocks.start() - - mocks.networkCallback.captured.onAvailable(mockk()) - - verify(exactly = 1) { mocks.operationRepo.forceExecuteOperations() } - } - - test("a failure registering the network callback does not crash start") { - val mocks = TriggerMocks() - every { mocks.connectivityManager.registerNetworkCallback(any(), any()) } throws - SecurityException() - - mocks.start() - - verify(exactly = 1) { mocks.applicationService.addApplicationLifecycleHandler(any()) } - } }) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt index cb16466bc..52ed67075 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt @@ -2,7 +2,6 @@ package com.onesignal.user.internal.migrations import com.onesignal.core.internal.operations.impl.OperationModelStore import com.onesignal.core.internal.operations.impl.OperationRepo -import com.onesignal.core.internal.time.impl.Time import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging import com.onesignal.mocks.CoreInternalMocks @@ -44,7 +43,7 @@ private class Mocks { listOf(), operationModelStore, configModelStore, - Time(), + MockHelper.systemTime(), ExecutorMocks.getNewRecordState(configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(), diff --git a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt index 500b736e7..3b189da8f 100644 --- a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt +++ b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt @@ -27,10 +27,19 @@ object MockHelper { fun time(time: Long): ITime { val mockTime = mockk() every { mockTime.currentTimeMillis } returns time + every { mockTime.elapsedRealtimeMillis } returns time return mockTime } + /** Real clocks for JVM tests, where android.os.SystemClock is not available. */ + fun systemTime(): ITime = + object : ITime { + override val currentTimeMillis: Long get() = System.currentTimeMillis() + override val processUptimeMillis: Long get() = System.nanoTime() / 1_000_000 + override val elapsedRealtimeMillis: Long get() = System.nanoTime() / 1_000_000 + } + fun applicationService(): IApplicationService { val mockAppService = mockk() From 317e00a6eb6e7e6132165e4388525bb3aff15d47 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Wed, 14 Oct 2026 12:14:23 -0700 Subject: [PATCH 4/7] fix: [SDK-5535] match iOS retry backoff and wake triggers - Backoff is exponential from 5s to 300s with 0.5-1x jitter, like iOS. - Retry-After is kept per operation as a minimum a force cannot end. - Foreground forces a retry; coming back online forces one only while in the foreground. Connectivity is watched from the first focus. - Retry-After parsing accepts fractions and falls back on negatives. - maxRetryDelay is now maxRetryAfter; opRepoMaxFailRetryBackoff caps backoff. Co-Authored-By: Claude Opus 5.5 --- .../java/com/onesignal/core/CoreModule.kt | 3 + .../core/internal/config/ConfigModel.kt | 25 +++-- .../internal/device/IConnectivityMonitor.kt | 9 ++ .../device/impl/ConnectivityMonitor.kt | 61 ++++++++++++ .../core/internal/http/impl/HttpClient.kt | 7 +- .../internal/operations/impl/OperationRepo.kt | 41 +++++--- .../operations/impl/OperationRetryTrigger.kt | 14 ++- .../device/ConnectivityMonitorTests.kt | 98 +++++++++++++++++++ .../core/internal/http/HttpClientTests.kt | 38 ++++++- .../internal/operations/OperationRepoTests.kt | 49 ++++++++-- .../operations/OperationRetryTriggerTests.kt | 42 +++++++- 11 files changed, 352 insertions(+), 35 deletions(-) create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/device/IConnectivityMonitor.kt create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/device/impl/ConnectivityMonitor.kt create mode 100644 OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/device/ConnectivityMonitorTests.kt diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt index a1fa95a7c..614903b8f 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt @@ -16,9 +16,11 @@ import com.onesignal.core.internal.config.impl.FeatureFlagsRefreshService import com.onesignal.core.internal.config.impl.IdentityVerificationService import com.onesignal.core.internal.database.IDatabaseProvider import com.onesignal.core.internal.database.impl.DatabaseProvider +import com.onesignal.core.internal.device.IConnectivityMonitor import com.onesignal.core.internal.device.IDeviceService import com.onesignal.core.internal.device.IFidEnv import com.onesignal.core.internal.device.IInstallIdService +import com.onesignal.core.internal.device.impl.ConnectivityMonitor import com.onesignal.core.internal.device.impl.DeviceService import com.onesignal.core.internal.device.impl.FidEnvService import com.onesignal.core.internal.device.impl.InstallIdService @@ -70,6 +72,7 @@ internal class CoreModule : IModule { .provides() builder.register().provides() builder.register().provides() + builder.register().provides() builder.register