Merge pull request #2233 from OneSignal/fix-ANR-operationRepo-load-with-coroutine-enqueue

Fix ANR caused by operationRepo.enqueue while loading is in progress

Author: jinliu9508

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OSPrimaryCoroutineScope.kt

@@ -0,0 +1,19 @@
+package com.onesignal.common.threading
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.newSingleThreadContext
+
+object OSPrimaryCoroutineScope {
+    // CoroutineScope tied to the main thread
+    private val mainScope = CoroutineScope(newSingleThreadContext(name = "OSPrimaryCoroutineScope"))
+
+    /**
+     * Executes the given [block] on the OS primary coroutine scope.
+     */
+    fun execute(block: suspend () -> Unit) {
+        mainScope.launch {
+            block()
+        }
+    }
+}

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionListener.kt

@@ -1,5 +1,6 @@
 package com.onesignal.session.internal.session.impl
 
+import com.onesignal.common.threading.OSPrimaryCoroutineScope
 import com.onesignal.common.threading.suspendifyOnThread
 import com.onesignal.core.internal.config.ConfigModelStore
 import com.onesignal.core.internal.operations.IOperationRepo
@@ -40,7 +41,10 @@ internal class SessionListener(
     }
 
     override fun onSessionStarted() {
-        _operationRepo.enqueue(TrackSessionStartOperation(_configModelStore.model.appId, _identityModelStore.model.onesignalId), true)
+        // enqueue the operation in background
+        OSPrimaryCoroutineScope.execute {
+            _operationRepo.enqueue(TrackSessionStartOperation(_configModelStore.model.appId, _identityModelStore.model.onesignalId), true)
+        }
     }
 
     override fun onSessionActive() {
@@ -54,9 +58,12 @@ internal class SessionListener(
             Logging.error("SessionListener.onSessionEnded sending duration of $durationInSeconds seconds")
         }
 
-        _operationRepo.enqueue(
-            TrackSessionEndOperation(_configModelStore.model.appId, _identityModelStore.model.onesignalId, durationInSeconds),
-        )
+        // enqueue the operation in background
+        OSPrimaryCoroutineScope.execute {
+            _operationRepo.enqueue(
+                TrackSessionEndOperation(_configModelStore.model.appId, _identityModelStore.model.onesignalId, durationInSeconds),
+            )
+        }
 
         suspendifyOnThread {
             _outcomeEventsController.sendSessionEndOutcomeEvent(durationInSeconds)

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/service/UserRefreshService.kt

@@ -1,6 +1,7 @@
 package com.onesignal.user.internal.service
 
 import com.onesignal.common.IDManager
+import com.onesignal.common.threading.OSPrimaryCoroutineScope
 import com.onesignal.core.internal.application.IApplicationService
 import com.onesignal.core.internal.config.ConfigModelStore
 import com.onesignal.core.internal.operations.IOperationRepo
@@ -28,12 +29,14 @@ class UserRefreshService(
             return
         }
 
-        _operationRepo.enqueue(
-            RefreshUserOperation(
-                _configModelStore.model.appId,
-                _identityModelStore.model.onesignalId,
-            ),
-        )
+        OSPrimaryCoroutineScope.execute {
+            _operationRepo.enqueue(
+                RefreshUserOperation(
+                    _configModelStore.model.appId,
+                    _identityModelStore.model.onesignalId,
+                ),
+            )
+        }
     }
 
     override fun start() = _sessionService.subscribe(this)

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

@@ -5,11 +5,15 @@ import com.onesignal.common.threading.WaiterWithValue
 import com.onesignal.core.internal.operations.impl.OperationModelStore
 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.MockHelper
+import com.onesignal.mocks.MockPreferencesService
 import com.onesignal.user.internal.operations.ExecutorMocks.Companion.getNewRecordState
+import com.onesignal.user.internal.operations.LoginUserOperation
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.shouldBe
 import io.mockk.CapturingSlot
@@ -28,6 +32,7 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.withTimeoutOrNull
 import kotlinx.coroutines.yield
+import org.json.JSONArray
 import java.util.UUID
 
 // Mocks used by every test in this file
@@ -76,6 +81,65 @@ class OperationRepoTests : FunSpec({
         Logging.logLevel = LogLevel.NONE
     }
 
+    test("ensure loading in the background thread does not block enqueue") {
+        // Given
+        val prefs = MockPreferencesService()
+        val mocks = Mocks()
+        val operationModelStore: OperationModelStore = spyk(OperationModelStore(prefs))
+        val operationRepo =
+            spyk(
+                OperationRepo(
+                    listOf(mocks.executor),
+                    operationModelStore,
+                    mocks.configModelStore,
+                    Time(),
+                    getNewRecordState(mocks.configModelStore),
+                ),
+            )
+
+        val cachedOperation = LoginUserOperation()
+        val newOperation = LoginUserOperation()
+        val jsonArray = JSONArray()
+
+        // cache the operation
+        jsonArray.put(cachedOperation.toJSON())
+        prefs.saveString(PreferenceStores.ONESIGNAL, PreferenceOneSignalKeys.MODEL_STORE_PREFIX + "operations", jsonArray.toString())
+
+        cachedOperation.id = UUID.randomUUID().toString()
+        newOperation.id = UUID.randomUUID().toString()
+        every { operationModelStore.create(any()) } answers {
+            // simulate a prolonged loading from cache
+            Thread.sleep(1000)
+            cachedOperation
+        }
+
+        // simulate a background thread to load operations
+        val backgroundThread =
+            Thread {
+                operationRepo.loadSavedOperations()
+            }
+
+        val mainThread =
+            Thread {
+                operationRepo.enqueue(newOperation)
+            }
+
+        // When
+        backgroundThread.start()
+        mainThread.start()
+
+        // Then
+        // insertion from the main thread is done without blocking
+        mainThread.join(500)
+        operationRepo.queue.size shouldBe 1
+        mainThread.state shouldBe Thread.State.TERMINATED
+
+        // after loading is completed, the cached operation should be at the beginning of the queue
+        backgroundThread.join()
+        operationRepo.queue.size shouldBe 2
+        operationRepo.queue.first().operation shouldBe cachedOperation
+    }
+
     test("containsInstanceOf") {
         // Given
         val operationRepo = Mocks().operationRepo