use the retryAfterSeconds in OperationRepo

Use the retryAfterSeconds to stall processing in the OperationRepo. We
wait in the OperationRepo as this allows more operations to be grouped
together, as we getGroupableOperations will be called after the delay.

There are number of other places that use HTTPClient other than the
OperationRepo, we will address those in a follow up commit.

Author: jkasten2

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/NetworkUtils.kt

@@ -17,6 +17,7 @@ object NetworkUtils {
             401, 403 -> ResponseStatusType.UNAUTHORIZED
             404, 410 -> ResponseStatusType.MISSING
             409 -> ResponseStatusType.CONFLICT
+            429 -> ResponseStatusType.RETRYABLE
             else -> ResponseStatusType.RETRYABLE
         }
     }

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt

@@ -177,6 +177,18 @@ class ConfigModel : Model() {
             setLongProperty(::opRepoPostCreateRetryUpTo.name, value)
         }
 
+    /**
+     * The number of milliseconds times the number of times FAIL_RETRY
+     * is returned from an executor for a specific operation. AKA this
+     * backoff will increase each time we retry a specific operation
+     * by this value.
+     */
+    var opRepoDefaultFailRetryBackoff: Long
+        get() = getLongProperty(::opRepoDefaultFailRetryBackoff.name) { 15_000 }
+        set(value) {
+            setLongProperty(::opRepoDefaultFailRetryBackoff.name, value)
+        }
+
     /**
      * The minimum number of milliseconds required to pass to allow the fetching of IAM to occur.
      */

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt

@@ -18,6 +18,7 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.newSingleThreadContext
 import kotlinx.coroutines.withTimeoutOrNull
 import java.util.UUID
+import kotlin.math.max
 import kotlin.reflect.KClass
 
 internal class OperationRepo(
@@ -221,6 +222,7 @@ internal class OperationRepo(
                 }
             }
 
+            var highestRetries = 0
             when (response.result) {
                 ExecutionResult.SUCCESS -> {
                     // on success we remove the operation from the store and wake any waiters
@@ -248,7 +250,6 @@ internal class OperationRepo(
                 ExecutionResult.FAIL_RETRY -> {
                     Logging.error("Operation execution failed, retrying: $operations")
                     // add back all operations to the front of the queue to be re-executed.
-                    var highestRetries = 0
                     synchronized(queue) {
                         ops.reversed().forEach {
                             if (++it.retries > highestRetries) {
@@ -257,7 +258,6 @@ internal class OperationRepo(
                             queue.add(0, it)
                         }
                     }
-                    delayBeforeRetry(highestRetries)
                 }
                 ExecutionResult.FAIL_PAUSE_OPREPO -> {
                     Logging.error("Operation execution failed with eventual retry, pausing the operation repo: $operations")
@@ -282,6 +282,8 @@ internal class OperationRepo(
                     }
                 }
             }
+
+            delayBeforeNextExecution(highestRetries, response.retryAfterSeconds)
         } catch (e: Throwable) {
             Logging.log(LogLevel.ERROR, "Error attempting to execute operation: $ops", e)
 
@@ -291,8 +293,18 @@ internal class OperationRepo(
         }
     }
 
-    suspend fun delayBeforeRetry(retries: Int) {
-        val delayFor = retries * 15_000L
+    /**
+     * 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.error("Operations being delay for: $delayFor ms")
         delay(delayFor)

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt

@@ -143,7 +143,7 @@ class OperationRepoTests : FunSpec({
         // Given
         val mocks = Mocks()
         val opRepo = mocks.operationRepo
-        coEvery { opRepo.delayBeforeRetry(any()) } just runs
+        coEvery { opRepo.delayBeforeNextExecution(any(), any()) } just runs
         coEvery {
             mocks.executor.execute(any())
         } returns ExecutionResponse(ExecutionResult.FAIL_RETRY) andThen ExecutionResponse(ExecutionResult.SUCCESS)
@@ -166,7 +166,7 @@ class OperationRepoTests : FunSpec({
                     it[0] shouldBe operation
                 },
             )
-            opRepo.delayBeforeRetry(1)
+            opRepo.delayBeforeNextExecution(1, null)
             mocks.executor.execute(
                 withArg {
                     it.count() shouldBe 1
@@ -177,6 +177,31 @@ class OperationRepoTests : FunSpec({
         }
     }
 
+    test("delays processing queue by retryAfterSeconds from the last executor's results") {
+        // Given
+        val mocks = Mocks()
+        val opRepo = mocks.operationRepo
+        coEvery {
+            mocks.executor.execute(any())
+        } returns ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = 1) andThen ExecutionResponse(ExecutionResult.SUCCESS)
+
+        // When
+        opRepo.start()
+        opRepo.enqueue(mockOperation())
+        val response1 =
+            withTimeoutOrNull(999) {
+                opRepo.enqueueAndWait(mockOperation())
+            }
+        val response2 =
+            withTimeoutOrNull(100) {
+                opRepo.enqueueAndWait(mockOperation())
+            }
+
+        // Then
+        response1 shouldBe null
+        response2 shouldBe true
+    }
+
     test("enqueue operation executes and is removed when executed after fail") {
         // Given
         val mocks = Mocks()

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/operations/IdentityOperationExecutorTests.kt

@@ -56,7 +56,14 @@ class IdentityOperationExecutorTests : FunSpec({
     test("execution of set alias operation with network timeout") {
         // Given
         val mockIdentityBackendService = mockk<IIdentityBackendService>()
-        coEvery { mockIdentityBackendService.setAlias(any(), any(), any(), any()) } throws BackendException(408, "TIMEOUT")
+        coEvery {
+            mockIdentityBackendService.setAlias(
+                any(),
+                any(),
+                any(),
+                any(),
+            )
+        } throws BackendException(408, "TIMEOUT", retryAfterSeconds = 10)
 
         val mockIdentityModelStore = MockHelper.identityModelStore()
         val mockBuildUserService = mockk<IRebuildUserService>()
@@ -71,6 +78,7 @@ class IdentityOperationExecutorTests : FunSpec({
 
         // Then
         response.result shouldBe ExecutionResult.FAIL_RETRY
+        response.retryAfterSeconds shouldBe 10
     }
 
     test("execution of set alias operation with non-retryable error") {

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/operations/LoginUserOperationExecutorTests.kt

@@ -87,7 +87,7 @@ class LoginUserOperationExecutorTests : FunSpec({
     test("login anonymous user fails with retry when network condition exists") {
         // Given
         val mockUserBackendService = mockk<IUserBackendService>()
-        coEvery { mockUserBackendService.createUser(any(), any(), any(), any()) } throws BackendException(408, "TIMEOUT")
+        coEvery { mockUserBackendService.createUser(any(), any(), any(), any()) } throws BackendException(408, "TIMEOUT", retryAfterSeconds = 10)
 
         val mockIdentityOperationExecutor = mockk<IdentityOperationExecutor>()
 
@@ -96,14 +96,25 @@ class LoginUserOperationExecutorTests : FunSpec({
         val mockSubscriptionsModelStore = mockk<SubscriptionModelStore>()
 
         val loginUserOperationExecutor =
-            LoginUserOperationExecutor(mockIdentityOperationExecutor, MockHelper.applicationService(), MockHelper.deviceService(), mockUserBackendService, mockIdentityModelStore, mockPropertiesModelStore, mockSubscriptionsModelStore, MockHelper.configModelStore(), MockHelper.languageContext())
+            LoginUserOperationExecutor(
+                mockIdentityOperationExecutor,
+                MockHelper.applicationService(),
+                MockHelper.deviceService(),
+                mockUserBackendService,
+                mockIdentityModelStore,
+                mockPropertiesModelStore,
+                mockSubscriptionsModelStore,
+                MockHelper.configModelStore(),
+                MockHelper.languageContext(),
+            )
         val operations = listOf<Operation>(LoginUserOperation(appId, localOneSignalId, null, null))
 
         // When
         val response = loginUserOperationExecutor.execute(operations)
 
         // Then
         response.result shouldBe ExecutionResult.FAIL_RETRY
+        response.retryAfterSeconds shouldBe 10
         coVerify(exactly = 1) { mockUserBackendService.createUser(appId, mapOf(), any(), any()) }
     }
 

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/operations/RefreshUserOperationExecutorTests.kt

@@ -177,7 +177,9 @@ class RefreshUserOperationExecutorTests : FunSpec({
     test("refresh user fails with retry when there is a network condition") {
         // Given
         val mockUserBackendService = mockk<IUserBackendService>()
-        coEvery { mockUserBackendService.getUser(appId, IdentityConstants.ONESIGNAL_ID, remoteOneSignalId) } throws BackendException(408)
+        coEvery {
+            mockUserBackendService.getUser(appId, IdentityConstants.ONESIGNAL_ID, remoteOneSignalId)
+        } throws BackendException(408, retryAfterSeconds = 10)
 
         // Given
         val mockIdentityModelStore = MockHelper.identityModelStore()
@@ -203,6 +205,7 @@ class RefreshUserOperationExecutorTests : FunSpec({
 
         // Then
         response.result shouldBe ExecutionResult.FAIL_RETRY
+        response.retryAfterSeconds shouldBe 10
         coVerify(exactly = 1) {
             mockUserBackendService.getUser(appId, IdentityConstants.ONESIGNAL_ID, remoteOneSignalId)
         }

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/operations/SubscriptionOperationExecutorTests.kt

@@ -93,7 +93,7 @@ class SubscriptionOperationExecutorTests : FunSpec({
     test("create subscription fails with retry when there is a network condition") {
         // Given
         val mockSubscriptionBackendService = mockk<ISubscriptionBackendService>()
-        coEvery { mockSubscriptionBackendService.createSubscription(any(), any(), any(), any()) } throws BackendException(408)
+        coEvery { mockSubscriptionBackendService.createSubscription(any(), any(), any(), any()) } throws BackendException(408, retryAfterSeconds = 10)
 
         val mockSubscriptionsModelStore = mockk<SubscriptionModelStore>()
         val mockBuildUserService = mockk<IRebuildUserService>()
@@ -127,6 +127,7 @@ class SubscriptionOperationExecutorTests : FunSpec({
 
         // Then
         response.result shouldBe ExecutionResult.FAIL_RETRY
+        response.retryAfterSeconds shouldBe 10
         coVerify(exactly = 1) {
             mockSubscriptionBackendService.createSubscription(
                 appId,

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/operations/UpdateUserOperationExecutorTests.kt

@@ -311,7 +311,7 @@ class UpdateUserOperationExecutorTests : FunSpec({
     test("update user single operation fails with MISSING, but isInMissingRetryWindow") {
         // Given
         val mockUserBackendService = mockk<IUserBackendService>()
-        coEvery { mockUserBackendService.updateUser(any(), any(), any(), any(), any(), any()) } throws BackendException(404)
+        coEvery { mockUserBackendService.updateUser(any(), any(), any(), any(), any(), any()) } throws BackendException(404, retryAfterSeconds = 10)
 
         // Given
         val mockIdentityModelStore = MockHelper.identityModelStore()
@@ -337,5 +337,6 @@ class UpdateUserOperationExecutorTests : FunSpec({
 
         // Then
         response.result shouldBe ExecutionResult.FAIL_RETRY
+        response.retryAfterSeconds shouldBe 10
     }
 })

OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt

@@ -48,6 +48,7 @@ object MockHelper {
         configModel.opRepoPostWakeDelay = 1
         configModel.opRepoPostCreateDelay = 1
         configModel.opRepoPostCreateRetryUpTo = 1
+        configModel.opRepoDefaultFailRetryBackoff = 1
 
         configModel.appId = DEFAULT_APP_ID