From 75436fa9374367653d7adcf43435860b4c6f65e8 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Thu, 6 Aug 2026 15:14:04 -0700 Subject: [PATCH 1/3] fix(threading): always call onComplete after suspendifyWithCompletion --- .../onesignal/common/threading/ThreadUtils.kt | 9 +++- .../common/threading/ThreadUtilsTests.kt | 19 ++++++++ .../notifications/receivers/BootUpReceiver.kt | 21 ++++----- .../receivers/FCMBroadcastReceiver.kt | 43 +++++++++---------- .../receivers/NotificationDismissReceiver.kt | 27 ++++++------ .../receivers/UpgradeReceiver.kt | 21 ++++----- .../receivers/FCMBroadcastReceiverTests.kt | 2 +- .../receivers/PrewarmEntryPointTests.kt | 6 +-- .../java/com/onesignal/mocks/IOMockHelper.kt | 13 ++++++ 9 files changed, 100 insertions(+), 61 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt index 24ca1241ea..971d303376 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt @@ -110,7 +110,7 @@ fun runOnSerialIO(block: () -> Unit) { * * @param useIO Whether to use IO scope (true) or Default scope (false) * @param block The suspending code to execute - * @param onComplete Optional callback to execute after completion + * @param onComplete Optional callback that always executes after [block], including on failure. */ fun suspendifyWithCompletion( useIO: Boolean = true, @@ -122,9 +122,14 @@ fun suspendifyWithCompletion( launch { try { block() - onComplete?.invoke() } catch (e: Exception) { Logging.error("Exception in suspendifyWithCompletion", e) + } finally { + try { + onComplete?.invoke() + } catch (e: Exception) { + Logging.error("Exception in suspendifyWithCompletion onComplete", e) + } } } } diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsTests.kt index 0c372c427b..d47fa33096 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsTests.kt @@ -139,6 +139,25 @@ class ThreadUtilsTests : FunSpec({ onCompleteCalled shouldBe true } + test("suspendifyWithCompletion should execute onComplete when block throws") { + val latch = CountDownLatch(1) + var onCompleteCalled = false + + suspendifyWithCompletion( + useIO = true, + block = { + throw RuntimeException("Test error") + }, + onComplete = { + onCompleteCalled = true + latch.countDown() + }, + ) + + latch.await() + onCompleteCalled shouldBe true + } + test("suspendifyWithErrorHandling should handle errors properly") { var errorHandled = false var onCompleteCalled = false diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt index 1bede49770..5da3677b37 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt @@ -46,16 +46,17 @@ class BootUpReceiver : BroadcastReceiver() { val pendingResult: BroadcastReceiver.PendingResult? = goAsync() // in background, init onesignal and begin enqueueing restore work - suspendifyOnIO { - if (!OneSignal.initWithContext(context.applicationContext)) { - Logging.warn("NotificationRestoreReceiver skipped due to failed OneSignal init") - pendingResult?.finish() - return@suspendifyOnIO - } + suspendifyOnIO( + block = { + if (!OneSignal.initWithContext(context.applicationContext)) { + Logging.warn("NotificationRestoreReceiver skipped due to failed OneSignal init") + return@suspendifyOnIO + } - val restoreWorkManager = OneSignal.getService() - restoreWorkManager.beginEnqueueingWork(context, true) - pendingResult?.finish() - } + val restoreWorkManager = OneSignal.getService() + restoreWorkManager.beginEnqueueingWork(context, true) + }, + onComplete = { pendingResult?.finish() }, + ) } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt index 09fa617d9a..f0552a733e 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt @@ -33,33 +33,32 @@ class FCMBroadcastReceiver : BroadcastReceiver() { val pendingResult: BroadcastReceiver.PendingResult? = goAsync() // process in background - suspendifyOnIO { - if (!OneSignal.initWithContext(context.applicationContext)) { - Logging.warn("FCMBroadcastReceiver skipped due to failed OneSignal init") - pendingResult?.finish() - return@suspendifyOnIO - } + suspendifyOnIO( + block = { + if (!OneSignal.initWithContext(context.applicationContext)) { + Logging.warn("FCMBroadcastReceiver skipped due to failed OneSignal init") + return@suspendifyOnIO + } - val bundleProcessor = OneSignal.getService() + val bundleProcessor = OneSignal.getService() - if (!isFCMMessage(intent)) { - setSuccessfulResultCode() - pendingResult?.finish() - return@suspendifyOnIO - } + if (!isFCMMessage(intent)) { + setSuccessfulResultCode() + return@suspendifyOnIO + } - val processedResult = bundleProcessor.processBundleFromReceiver(context, bundle) + val processedResult = bundleProcessor.processBundleFromReceiver(context, bundle) - // Prevent other FCM receivers from firing if work manager is processing the notification - if (processedResult?.isWorkManagerProcessing == true) { - setAbort() - pendingResult?.finish() - return@suspendifyOnIO - } + // Prevent other FCM receivers from firing if work manager is processing the notification + if (processedResult?.isWorkManagerProcessing == true) { + setAbort() + return@suspendifyOnIO + } - setSuccessfulResultCode() - pendingResult?.finish() - } + setSuccessfulResultCode() + }, + onComplete = { pendingResult?.finish() }, + ) } private fun setSuccessfulResultCode() { diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt index fa90a9cd18..a553f36ae9 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt @@ -46,20 +46,21 @@ class NotificationDismissReceiver : BroadcastReceiver() { val pendingResult: BroadcastReceiver.PendingResult? = goAsync() - suspendifyOnIO { - if (!OneSignal.initWithContext(context.applicationContext)) { - Logging.warn("NotificationOpenedReceiver skipped due to failed OneSignal init") - pendingResult?.finish() - return@suspendifyOnIO - } + suspendifyOnIO( + block = { + if (!OneSignal.initWithContext(context.applicationContext)) { + Logging.warn("NotificationOpenedReceiver skipped due to failed OneSignal init") + return@suspendifyOnIO + } - val notificationOpenedProcessor = OneSignal.getService() + val notificationOpenedProcessor = OneSignal.getService() - // init OneSignal in background but process in main - withContext(Dispatchers.Main) { - notificationOpenedProcessor.processFromContext(context, intent) - } - pendingResult?.finish() - } + // init OneSignal in background but process in main + withContext(Dispatchers.Main) { + notificationOpenedProcessor.processFromContext(context, intent) + } + }, + onComplete = { pendingResult?.finish() }, + ) } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt index f48c050197..5ab5a8fe00 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt @@ -56,16 +56,17 @@ class UpgradeReceiver : BroadcastReceiver() { val pendingResult: BroadcastReceiver.PendingResult? = goAsync() // init OneSignal and enqueue restore work in background - suspendifyOnIO { - if (!OneSignal.initWithContext(context.applicationContext)) { - Logging.warn("UpgradeReceiver skipped due to failed OneSignal init") - pendingResult?.finish() - return@suspendifyOnIO - } + suspendifyOnIO( + block = { + if (!OneSignal.initWithContext(context.applicationContext)) { + Logging.warn("UpgradeReceiver skipped due to failed OneSignal init") + return@suspendifyOnIO + } - val restoreWorkManager = OneSignal.getService() - restoreWorkManager.beginEnqueueingWork(context, true) - pendingResult?.finish() - } + val restoreWorkManager = OneSignal.getService() + restoreWorkManager.beginEnqueueingWork(context, true) + }, + onComplete = { pendingResult?.finish() }, + ) } } diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt index ff0416ce76..fa3ca42ee3 100644 --- a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt @@ -53,7 +53,7 @@ class FCMBroadcastReceiverTests : FunSpec({ verify(exactly = 1) { OneSignalDispatchers.prewarm() } verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>()) + suspendifyOnIO(any Unit>(), any<() -> Unit>()) } } diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt index 99cdcecef2..cb0da39d72 100644 --- a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt @@ -52,7 +52,7 @@ class PrewarmEntryPointTests : FunSpec({ verify(exactly = 1) { OneSignalDispatchers.prewarm() } verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>()) + suspendifyOnIO(any Unit>(), any<() -> Unit>()) } } @@ -62,7 +62,7 @@ class PrewarmEntryPointTests : FunSpec({ verify(exactly = 1) { OneSignalDispatchers.prewarm() } verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>()) + suspendifyOnIO(any Unit>(), any<() -> Unit>()) } } @@ -72,7 +72,7 @@ class PrewarmEntryPointTests : FunSpec({ verify(exactly = 1) { OneSignalDispatchers.prewarm() } verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>()) + suspendifyOnIO(any Unit>(), any<() -> Unit>()) } } }) diff --git a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt index a2b2751530..c41b5e6544 100644 --- a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt +++ b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt @@ -79,6 +79,7 @@ object IOMockHelper : BeforeSpecListener, AfterSpecListener, BeforeTestListener, } } + @Suppress("LongMethod") override suspend fun beforeSpec(spec: Spec) { // ThreadUtilsKt = file that contains suspendifyOnIO mockkStatic(THREADUTILS_PATH) @@ -136,6 +137,18 @@ object IOMockHelper : BeforeSpecListener, AfterSpecListener, BeforeTestListener, trackAsyncWork(block) } + every { suspendifyOnIO(any Unit>(), any<() -> Unit>()) } answers { + val block = firstArg Unit>() + val onComplete = secondArg<() -> Unit>() + trackAsyncWork { + try { + block() + } finally { + onComplete() + } + } + } + every { suspendifyOnSerialIO(any Unit>()) } answers { val block = firstArg Unit>() trackAsyncWork(block) From 4aaf1b9e3cd60cdfc99016b78f7ca52177ca584d Mon Sep 17 00:00:00 2001 From: Fadi George Date: Fri, 7 Aug 2026 10:18:32 -0700 Subject: [PATCH 2/3] feat(notifications): add durable ingress lane for FCM/dismiss events --- .../common/threading/OneSignalDispatchers.kt | 532 +++++++++++------- .../onesignal/common/threading/ThreadUtils.kt | 24 + .../onesignal/core/services/SyncJobService.kt | 98 ++-- .../threading/OneSignalDispatchersTests.kt | 85 +-- .../common/threading/ThreadUtilsTests.kt | 21 + .../core/services/SyncJobServiceTests.kt | 56 +- .../notifications/NotificationsModule.kt | 2 + .../impl/NotificationGenerationWorkManager.kt | 50 +- .../internal/ingress/NotificationIngress.kt | 320 +++++++++++ .../impl/NotificationRestoreWorkManager.kt | 46 +- .../notifications/receivers/BootUpReceiver.kt | 27 +- .../receivers/BroadcastCompletion.kt | 40 ++ .../receivers/FCMBroadcastReceiver.kt | 31 +- .../receivers/NotificationDismissReceiver.kt | 26 +- .../receivers/UpgradeReceiver.kt | 27 +- .../ingress/NotificationIngressTests.kt | 63 +++ .../receivers/BroadcastCompletionTests.kt | 32 ++ .../receivers/FCMBroadcastReceiverTests.kt | 13 +- .../receivers/PrewarmEntryPointTests.kt | 16 +- .../java/com/onesignal/mocks/IOMockHelper.kt | 19 + 20 files changed, 1098 insertions(+), 430 deletions(-) create mode 100644 OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/ingress/NotificationIngress.kt create mode 100644 OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt create mode 100644 OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/ingress/NotificationIngressTests.kt create mode 100644 OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/BroadcastCompletionTests.kt diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt index e1b651f8b0..9836d7bf20 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt @@ -10,13 +10,15 @@ import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.cancel import kotlinx.coroutines.isActive import kotlinx.coroutines.launch -import java.util.concurrent.ExecutorService -import java.util.concurrent.Executors +import java.util.concurrent.CancellationException +import java.util.concurrent.ConcurrentLinkedQueue import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.ThreadFactory import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger +import kotlin.coroutines.CoroutineContext /** * Optimized threading manager for the OneSignal SDK. @@ -31,6 +33,7 @@ import java.util.concurrent.atomic.AtomicInteger * * Made public to allow mocking in tests via IOMockHelper. */ +@Suppress("TooManyFunctions", "StringLiteralDuplication") object OneSignalDispatchers { // Optimized pool sizes based on CPU cores and workload analysis private const val IO_CORE_POOL_SIZE = 2 // Increased for better concurrency @@ -41,6 +44,8 @@ object OneSignalDispatchers { 30L // Keep threads alive longer to reduce recreation private const val QUEUE_CAPACITY = 200 // Increased to handle more queued operations during init, while still preventing memory bloat + private const val TEST_READY_TIMEOUT_MS = 2_000L + private const val TEST_READY_POLL_MS = 5L internal const val BASE_THREAD_NAME = "OneSignal" // Base thread name prefix private const val IO_THREAD_NAME_PREFIX = "$BASE_THREAD_NAME-IO" // Thread name prefix for I/O operations @@ -48,6 +53,10 @@ object OneSignalDispatchers { "$BASE_THREAD_NAME-Default" // Thread name prefix for CPU operations private const val SERIAL_IO_THREAD_NAME = "$BASE_THREAD_NAME-SerialIO" // Single, named thread for order-sensitive work + private const val INGRESS_THREAD_NAME = "$BASE_THREAD_NAME-Ingress" + + @Volatile + internal var beforeLaneCreateForTest: ((String) -> Unit)? = null private class OptimizedThreadFactory( private val namePrefix: String, @@ -63,131 +72,328 @@ object OneSignalDispatchers { } } + private enum class Lane { + IO, + DEFAULT, + SERIAL_IO, + INGRESS, + } + + private enum class LaneState { + COLD, + STARTING, + DRAINING, + READY, + CLOSED, + } + + private class LaneTarget( + val dispatcher: CoroutineDispatcher, + val executor: ThreadPoolExecutor?, + ) { + fun close() { + executor?.shutdownNow() + } + } + /** - * Holds one generation of executors, dispatchers, and scopes. Everything stays lazily - * initialized (so production cold-start only pays for what the first caller actually uses), - * but bundling them lets the test-only [resetForTest] hook atomically swap in a clean - * generation and tear down the old one — including [shutdownNow] to interrupt worker threads - * that are parked in non-cancellable JVM waits (e.g. `CountDownLatch.await()` from a prior - * spec), which a plain coroutine cancellation cannot free. + * A stable dispatcher that queues work until its backing executor has been built and + * prestarted on the bootstrap thread. CoroutineScope.launch can therefore return its real Job + * without forcing executor construction or waiting on a lazy lock on the caller thread. */ - private class Pools { - val ioExecutorLazy = - lazy { - try { - ThreadPoolExecutor( - IO_CORE_POOL_SIZE, - IO_MAX_POOL_SIZE, - KEEP_ALIVE_TIME_SECONDS, - TimeUnit.SECONDS, - LinkedBlockingQueue(QUEUE_CAPACITY), - OptimizedThreadFactory( - namePrefix = IO_THREAD_NAME_PREFIX, - priority = Thread.NORM_PRIORITY - 1, - // Slightly lower priority for I/O tasks - ), - ).apply { - allowCoreThreadTimeOut(false) // Keep core threads alive + private class GateDispatcher( + private val lane: Lane, + private val requestBootstrap: (GateDispatcher) -> Unit, + ) : CoroutineDispatcher() { + private val lock = Any() + private val pending = ArrayDeque>() + private var state = LaneState.COLD + private var target: LaneTarget? = null + + override fun dispatch( + context: CoroutineContext, + block: Runnable, + ) { + var readyDispatcher: CoroutineDispatcher? = null + var shouldBootstrap = false + var rejected = false + + synchronized(lock) { + when (state) { + LaneState.COLD -> { + pending.addLast(context to block) + state = LaneState.STARTING + shouldBootstrap = true } - } catch (e: Exception) { - Logging.error("OneSignalDispatchers: Failed to create IO executor: ${e.message}") - throw e // Let the dispatcher fallback handle this + LaneState.STARTING, LaneState.DRAINING -> pending.addLast(context to block) + LaneState.READY -> readyDispatcher = target!!.dispatcher + LaneState.CLOSED -> rejected = true } } - val ioExecutor: ThreadPoolExecutor get() = ioExecutorLazy.value - /** Single-thread executor for order-sensitive lifecycle work (focus / unfocus handlers). */ - val serialIOExecutorLazy = - lazy { + when { + shouldBootstrap -> requestBootstrap(this) + readyDispatcher != null -> dispatchSafely(readyDispatcher!!, context, block) + rejected -> cancelAndComplete(context, block, "OneSignal dispatcher generation is closed") + } + } + + fun requestWarmup() { + val shouldBootstrap = + synchronized(lock) { + if (state == LaneState.COLD) { + state = LaneState.STARTING + true + } else { + false + } + } + if (shouldBootstrap) requestBootstrap(this) + } + + @Suppress("TooGenericExceptionCaught") + fun initialize() { + val newTarget = try { - Executors.newSingleThreadExecutor( - OptimizedThreadFactory( - namePrefix = SERIAL_IO_THREAD_NAME, - priority = Thread.NORM_PRIORITY - 1, - ), - ) + createTarget(lane) } catch (e: Exception) { - Logging.error("OneSignalDispatchers: Failed to create SerialIO executor: ${e.message}") - throw e + Logging.warn("OneSignalDispatchers: Using fallback for $lane lane: ${e.message}", e) + createFallbackTarget(lane) + } catch (t: Throwable) { + Logging.warn("OneSignalDispatchers: Failed to initialize $lane lane: ${t.message}", t) + failPending("OneSignal $lane dispatcher failed to initialize") + return } - } - val serialIOExecutor: ExecutorService get() = serialIOExecutorLazy.value - val defaultExecutorLazy = - lazy { - try { - ThreadPoolExecutor( - DEFAULT_CORE_POOL_SIZE, - DEFAULT_MAX_POOL_SIZE, - KEEP_ALIVE_TIME_SECONDS, - TimeUnit.SECONDS, - LinkedBlockingQueue(QUEUE_CAPACITY), - OptimizedThreadFactory(DEFAULT_THREAD_NAME_PREFIX), - ).apply { - allowCoreThreadTimeOut(false) // Keep core threads alive + while (true) { + val next = + synchronized(lock) { + if (state == LaneState.CLOSED) { + null + } else { + state = LaneState.DRAINING + pending.removeFirstOrNull().also { + if (it == null) { + target = newTarget + state = LaneState.READY + } + } + } } - } catch (e: Exception) { - Logging.error("OneSignalDispatchers: Failed to create Default executor: ${e.message}") - throw e // Let the dispatcher fallback handle this + + if (next == null) { + val closed = synchronized(lock) { state == LaneState.CLOSED } + if (closed) newTarget.close() + return } + dispatchSafely(newTarget.dispatcher, next.first, next.second) } - val defaultExecutor: ThreadPoolExecutor get() = defaultExecutorLazy.value + } - // Dispatchers - also lazy initialized - val IO: CoroutineDispatcher by lazy { - try { - ioExecutor.asCoroutineDispatcher() - } catch (e: Exception) { - Logging.error("OneSignalDispatchers: Using fallback Dispatchers.IO dispatcher: ${e.message}") - Dispatchers.IO + fun close() { + val queued: List> + val oldTarget: LaneTarget? + synchronized(lock) { + if (state == LaneState.CLOSED) return + state = LaneState.CLOSED + queued = pending.toList() + pending.clear() + oldTarget = target + target = null } + queued.forEach { + cancelAndComplete(it.first, it.second, "OneSignal dispatcher generation was reset") + } + oldTarget?.close() } - val Default: CoroutineDispatcher by lazy { - try { - defaultExecutor.asCoroutineDispatcher() - } catch (e: Exception) { - Logging.error("OneSignalDispatchers: Using fallback Dispatchers.Default dispatcher: ${e.message}") - Dispatchers.Default + fun bootstrapFailed() { + failPending("OneSignal dispatcher bootstrap thread failed to start") + } + + fun status(): String = + synchronized(lock) { + when (state) { + LaneState.READY -> "Active" + LaneState.CLOSED -> "Shutdown" + else -> state.name.lowercase().replaceFirstChar { it.uppercase() } + } } + + fun executor(): ThreadPoolExecutor? = synchronized(lock) { target?.executor } + + private fun failPending(reason: String) { + val queued = + synchronized(lock) { + val copy = pending.toList() + pending.clear() + state = LaneState.COLD + copy + } + queued.forEach { cancelAndComplete(it.first, it.second, reason) } } - val SerialIO: CoroutineDispatcher by lazy { + private fun dispatchSafely( + dispatcher: CoroutineDispatcher, + context: CoroutineContext, + block: Runnable, + ) { try { - serialIOExecutor.asCoroutineDispatcher() - } catch (e: Exception) { - // Fall back to a limitedParallelism(1) view of Dispatchers.IO so submissions stay serialized. - Logging.error("OneSignalDispatchers: Using fallback serialized Dispatchers.IO: ${e.message}") - @Suppress("OPT_IN_USAGE") - Dispatchers.IO.limitedParallelism(1) + dispatcher.dispatch(context, block) + } catch (e: RuntimeException) { + Logging.error("OneSignalDispatchers: $lane dispatch rejected: ${e.message}", e) + cancelAndComplete(context, block, "OneSignal $lane dispatch rejected") } } - val ioScopeLazy = lazy { CoroutineScope(SupervisorJob() + IO) } - val IOScope: CoroutineScope get() = ioScopeLazy.value + private fun cancelAndComplete( + context: CoroutineContext, + block: Runnable, + reason: String, + ) { + val job = context[Job] ?: return + job.cancel(CancellationException(reason)) + block.run() + } + } - val defaultScopeLazy = lazy { CoroutineScope(SupervisorJob() + Default) } - val DefaultScope: CoroutineScope get() = defaultScopeLazy.value + private class BootstrapCoordinator { + private val queue = ConcurrentLinkedQueue() + private val running = AtomicBoolean(false) - val serialIOScopeLazy = lazy { CoroutineScope(SupervisorJob() + SerialIO) } - val SerialIOScope: CoroutineScope get() = serialIOScopeLazy.value + fun request(gate: GateDispatcher) { + queue.add(gate) + startIfNeeded() + } - /** - * Cancel scopes and forcibly stop executors for this generation. Only touches what was - * actually initialized so we never spin up a pool just to tear it down. [shutdownNow] - * interrupts parked worker threads so they can't outlive the spec that created them. - */ @Suppress("TooGenericExceptionCaught") + private fun startIfNeeded() { + if (!running.compareAndSet(false, true)) return + try { + Thread( + { + while (true) { + val gate = queue.poll() + if (gate != null) { + gate.initialize() + continue + } + running.set(false) + if (queue.isEmpty() || !running.compareAndSet(false, true)) return@Thread + } + }, + "$BASE_THREAD_NAME-bootstrap", + ).apply { + isDaemon = true + priority = Thread.NORM_PRIORITY - 2 + start() + } + } catch (t: Throwable) { + running.set(false) + Logging.warn("OneSignalDispatchers: Failed to start bootstrap thread: ${t.message}", t) + while (true) { + val gate = queue.poll() ?: break + gate.bootstrapFailed() + } + } + } + } + + private class Pools { + private val coordinator = BootstrapCoordinator() + val IO = GateDispatcher(Lane.IO, coordinator::request) + val Default = GateDispatcher(Lane.DEFAULT, coordinator::request) + val SerialIO = GateDispatcher(Lane.SERIAL_IO, coordinator::request) + val Ingress = GateDispatcher(Lane.INGRESS, coordinator::request) + val IOScope = CoroutineScope(SupervisorJob() + IO) + val DefaultScope = CoroutineScope(SupervisorJob() + Default) + val SerialIOScope = CoroutineScope(SupervisorJob() + SerialIO) + val IngressScope = CoroutineScope(SupervisorJob() + Ingress) + + fun prewarm() { + IO.requestWarmup() + Default.requestWarmup() + SerialIO.requestWarmup() + Ingress.requestWarmup() + } + fun shutdown() { - if (ioScopeLazy.isInitialized()) runCatching { ioScopeLazy.value.cancel() } - if (defaultScopeLazy.isInitialized()) runCatching { defaultScopeLazy.value.cancel() } - if (serialIOScopeLazy.isInitialized()) runCatching { serialIOScopeLazy.value.cancel() } - if (ioExecutorLazy.isInitialized()) runCatching { ioExecutorLazy.value.shutdownNow() } - if (defaultExecutorLazy.isInitialized()) runCatching { defaultExecutorLazy.value.shutdownNow() } - if (serialIOExecutorLazy.isInitialized()) runCatching { serialIOExecutorLazy.value.shutdownNow() } + IOScope.cancel() + DefaultScope.cancel() + SerialIOScope.cancel() + IngressScope.cancel() + IO.close() + Default.close() + SerialIO.close() + Ingress.close() } } + @Suppress("LongMethod") + private fun createTarget(lane: Lane): LaneTarget { + beforeLaneCreateForTest?.invoke(lane.name) + val executor = + when (lane) { + Lane.IO -> + ThreadPoolExecutor( + IO_CORE_POOL_SIZE, + IO_MAX_POOL_SIZE, + KEEP_ALIVE_TIME_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(QUEUE_CAPACITY), + OptimizedThreadFactory(IO_THREAD_NAME_PREFIX, Thread.NORM_PRIORITY - 1), + ) + Lane.DEFAULT -> + ThreadPoolExecutor( + DEFAULT_CORE_POOL_SIZE, + DEFAULT_MAX_POOL_SIZE, + KEEP_ALIVE_TIME_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(QUEUE_CAPACITY), + OptimizedThreadFactory(DEFAULT_THREAD_NAME_PREFIX), + ) + Lane.SERIAL_IO -> + ThreadPoolExecutor( + 1, + 1, + KEEP_ALIVE_TIME_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(QUEUE_CAPACITY), + OptimizedThreadFactory(SERIAL_IO_THREAD_NAME, Thread.NORM_PRIORITY - 1), + ) + Lane.INGRESS -> + ThreadPoolExecutor( + 1, + 1, + KEEP_ALIVE_TIME_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(QUEUE_CAPACITY), + OptimizedThreadFactory(INGRESS_THREAD_NAME, Thread.NORM_PRIORITY - 1), + ) + } + executor.allowCoreThreadTimeOut(false) + executor.prestartAllCoreThreads() + return LaneTarget(executor.asCoroutineDispatcher(), executor) + } + + private fun createFallbackTarget(lane: Lane): LaneTarget { + val dispatcher = + when (lane) { + Lane.IO -> Dispatchers.IO + Lane.DEFAULT -> Dispatchers.Default + Lane.SERIAL_IO -> { + @Suppress("OPT_IN_USAGE") + Dispatchers.IO.limitedParallelism(1) + } + Lane.INGRESS -> { + @Suppress("OPT_IN_USAGE") + Dispatchers.IO.limitedParallelism(1) + } + } + dispatcher.dispatch(kotlin.coroutines.EmptyCoroutineContext, Runnable {}) + return LaneTarget(dispatcher, null) + } + @Volatile private var pools = Pools() @@ -211,57 +417,19 @@ object OneSignalDispatchers { return pools.SerialIOScope.launch { block() } } + /** Launches short durable-ingress work on a pool isolated from general SDK I/O. */ + fun launchOnIngress(block: suspend () -> Unit): Job { + return pools.IngressScope.launch { block() } + } + @Volatile private var prewarmStarted = false private val prewarmLock = Any() /** - * Triggers the lazy initialization of [IO], [Default], and [SerialIO] (and their backing - * executors, dispatchers, and scopes) on a short-lived background thread. - * - * Background: - * The lazy `by lazy` properties below construct `ThreadPoolExecutor` instances and wrap them - * in `asCoroutineDispatcher() + SupervisorJob() + CoroutineScope(...)`. Production OTel - * shows that when the **first** caller of [launchOnIO] / [launchOnSerialIO] is on the main - * thread (Activity-lifecycle handler, `JobService.onStartJob`, etc.), the construction cost - * — which includes a `kotlinx.coroutines.BuildersKt.launch` that hits - * `ThreadPoolExecutor.execute` and `LinkedBlockingQueue.offer` synchronously — is paid on - * the calling thread, blocking the main thread for many seconds on cold start. - * - * Calling [prewarm] from a non-time-sensitive spot in [com.onesignal.OneSignal.initWithContext] - * shifts that cost to a dedicated `OneSignal-prewarm` daemon thread, so the first - * production caller — including main-thread lifecycle handlers — only pays the much cheaper - * "submit work to an already-constructed executor" cost. - * - * Idempotent and fire-and-forget: a no-op on second and subsequent calls. Safe to invoke - * from any thread; the heavy lifting always happens on the daemon thread we spawn here, not - * on the caller. Failures are logged and swallowed because the executors retain their - * existing fallback paths (e.g. `Dispatchers.IO.limitedParallelism(1)` for [SerialIO]) and a - * failed prewarm will simply mean the first production caller pays the original cost. - * - * **Best-effort, not a hard guarantee.** Because [prewarm] is fire-and-forget, a caller that - * dispatches immediately afterward can still win the lazy-init race and pay construction on - * its own thread. The fix relies on placing [prewarm] at cold-start entry points where there - * is meaningful lead time (e.g. a `goAsync()` handoff or `initWithContext` work) before the - * first `suspendify*` / `launchOn*` dispatch. - * - * [suspendifyOnIO], [suspendifyOnDefault], [launchOnIO], [launchOnDefault], and - * [suspendifyOnSerialIO] always route through [IO] / [Default] / [SerialIO], so [prewarm] - * benefits the very first real dispatch from any of them. - * - * The known main-thread cold-start entry points: - * | Entry point | Class | - * |---|---| - * | process-start activity lifecycle registration | `core.internal.application.impl.ActivityLifecycleInitializer` | - * | `initWithContext` / `initWithContextSuspend` | `OneSignalImp` | - * | `onStartJob` | `core.services.SyncJobService` | - * | `onReceive` | `FCMBroadcastReceiver`, `NotificationDismissReceiver`, `BootUpReceiver`, `UpgradeReceiver` | - * | `onMessage` / registration callbacks | `ADMMessageHandler`, `ADMMessageHandlerJob` | - * | `onNewToken` / `onMessageReceived` | `OneSignalHmsEventBridge` | - * | `processIntent` / `processOpen` | `NotificationOpenedActivityBase`, `NotificationOpenedActivityHMS` | - * - * When adding a new cold-start entry point (receiver, job, activity trampoline, push bridge), - * call [prewarm] at the top of it before the first dispatch. + * Requests asynchronous initialization and worker prestart for every lane. Dispatch correctness + * does not depend on this call: a cold first dispatch queues behind the same bootstrap gate and + * returns without constructing or waiting for a pool on its caller thread. */ @Suppress("TooGenericExceptionCaught") fun prewarm() { @@ -271,35 +439,10 @@ object OneSignalDispatchers { prewarmStarted = true } try { - val prewarmThread = Thread( - { - try { - // Each launch* call below triggers the corresponding lazy chain - // (executor -> dispatcher -> scope) and submits an empty coroutine, - // which forces the worker thread(s) to start as well. - launchOnIO { /* warm IOScope + ioExecutor */ } - launchOnDefault { /* warm DefaultScope + defaultExecutor */ } - launchOnSerialIO { /* warm SerialIOScope + serialIOExecutor */ } - } catch (e: Throwable) { - synchronized(prewarmLock) { prewarmStarted = false } - Logging.warn("OneSignalDispatchers.prewarm failed: ${e.message}", e) - } - }, - "$BASE_THREAD_NAME-prewarm", - ) - prewarmThread.isDaemon = true - prewarmThread.priority = Thread.NORM_PRIORITY - 2 - prewarmThread.start() + pools.prewarm() } catch (t: Throwable) { - // Constructing, configuring, or starting the daemon can itself fail before the body - // ever runs (e.g. OutOfMemoryError "unable to create new native thread", a - // SecurityManager denial, or InternalError). Swallow it so a prewarm failure never - // propagates onto the cold-start entry point's caller thread, and reset the guard so a - // later entry point can retry. The dispatchers keep their lazy fallbacks, so the only - // cost of a failed prewarm is that the first real dispatch pays the original - // construction cost. synchronized(prewarmLock) { prewarmStarted = false } - Logging.warn("OneSignalDispatchers.prewarm failed to start daemon: ${t.message}", t) + Logging.warn("OneSignalDispatchers.prewarm failed: ${t.message}", t) } } @@ -342,43 +485,56 @@ object OneSignalDispatchers { Logging.error("OneSignalDispatchers.resetForTest failed: ${e.message}", e) } resetPrewarmForTest() + beforeLaneCreateForTest = null } + @Suppress("ComplexMethod") internal fun getPerformanceMetrics(): String { - return try { - val current = pools - val serialQueueSize = - (current.serialIOExecutor as? ThreadPoolExecutor)?.queue?.size?.toString() ?: "n/a" - val serialCompleted = - (current.serialIOExecutor as? ThreadPoolExecutor)?.completedTaskCount ?: 0L - """ + val current = pools + val io = current.IO.executor() + val default = current.Default.executor() + val serial = current.SerialIO.executor() + val ingress = current.Ingress.executor() + return """ OneSignalDispatchers Performance Metrics: - - IO Pool: ${current.ioExecutor.activeCount}/${current.ioExecutor.corePoolSize} active/core threads - - IO Queue: ${current.ioExecutor.queue.size} pending tasks - - Default Pool: ${current.defaultExecutor.activeCount}/${current.defaultExecutor.corePoolSize} active/core threads - - Default Queue: ${current.defaultExecutor.queue.size} pending tasks - - SerialIO Queue: $serialQueueSize pending tasks - - Total completed tasks: ${current.ioExecutor.completedTaskCount + current.defaultExecutor.completedTaskCount + serialCompleted} - - Memory usage: ~${(current.ioExecutor.activeCount + current.defaultExecutor.activeCount + 1) * 1024}KB (thread stacks, ~1MB each) + - IO Pool: ${io?.let { "${it.activeCount}/${it.corePoolSize}" } ?: "n/a"} active/core threads + - IO Queue: ${io?.queue?.size ?: "n/a"} pending tasks + - Default Pool: ${default?.let { "${it.activeCount}/${it.corePoolSize}" } ?: "n/a"} active/core threads + - Default Queue: ${default?.queue?.size ?: "n/a"} pending tasks + - SerialIO Queue: ${serial?.queue?.size ?: "n/a"} pending tasks + - Ingress Queue: ${ingress?.queue?.size ?: "n/a"} pending tasks + - Total completed tasks: ${(io?.completedTaskCount ?: 0L) + (default?.completedTaskCount ?: 0L) + (serial?.completedTaskCount ?: 0L) + (ingress?.completedTaskCount ?: 0L)} + - Memory usage: ~${((io?.activeCount ?: 0) + (default?.activeCount ?: 0) + (serial?.activeCount ?: 0) + (ingress?.activeCount ?: 0)) * 1024}KB (thread stacks, ~1MB each) """.trimIndent() - } catch (e: Exception) { - "OneSignalDispatchers not initialized or using fallback dispatchers ${e.message}" - } } internal fun getStatus(): String { val current = pools return """ OneSignalDispatchers Status: - - IO Executor: ${executorStatus("ioExecutor") { current.ioExecutor.isShutdown }} - - Default Executor: ${executorStatus("defaultExecutor") { current.defaultExecutor.isShutdown }} - - SerialIO Executor: ${executorStatus("serialIOExecutor") { current.serialIOExecutor.isShutdown }} + - IO Executor: ${current.IO.status()} + - Default Executor: ${current.Default.status()} + - SerialIO Executor: ${current.SerialIO.status()} + - Ingress Executor: ${current.Ingress.status()} - IO Scope: ${scopeStatus("IOScope") { current.IOScope.isActive }} - Default Scope: ${scopeStatus("DefaultScope") { current.DefaultScope.isActive }} - SerialIO Scope: ${scopeStatus("SerialIOScope") { current.SerialIOScope.isActive }} + - Ingress Scope: ${scopeStatus("IngressScope") { current.IngressScope.isActive }} """.trimIndent() } + internal fun awaitReadyForTest(timeoutMs: Long = TEST_READY_TIMEOUT_MS): Boolean { + val deadline = System.currentTimeMillis() + timeoutMs + while (System.currentTimeMillis() < deadline) { + val current = pools + if (listOf(current.IO, current.Default, current.SerialIO, current.Ingress).all { it.status() == "Active" }) { + return true + } + Thread.sleep(TEST_READY_POLL_MS) + } + return false + } + // internal so tests can exercise the failure branch (when `isShutdown()` itself throws, // which happens when the lazy initializer threw and re-throws on every access). internal fun executorStatus( diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt index 971d303376..0097cc863e 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt @@ -1,3 +1,5 @@ +@file:Suppress("TooManyFunctions") + package com.onesignal.common.threading import com.onesignal.debug.internal.logging.Logging @@ -67,6 +69,28 @@ fun suspendifyOnIO(block: suspend () -> Unit) { suspendifyWithCompletion(useIO = true, block = block, onComplete = null) } +/** Runs short, deadline-sensitive ingress work on its isolated serial dispatcher. */ +fun suspendifyOnIngress( + block: suspend () -> Unit, + onComplete: (() -> Unit)? = null, +) { + val job = + OneSignalDispatchers.launchOnIngress { + try { + block() + } catch (e: Exception) { + Logging.error("Exception in suspendifyOnIngress", e) + } + } + job.invokeOnCompletion { + try { + onComplete?.invoke() + } catch (e: Exception) { + Logging.error("Exception in suspendifyOnIngress onComplete", e) + } + } +} + /** * Modern utility for executing suspending code on the default dispatcher. * Uses OneSignal's centralized thread management for CPU-intensive operations. diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt index ff36ae649a..3b7e0eb241 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt @@ -30,11 +30,27 @@ import android.app.job.JobParameters import android.app.job.JobService import com.onesignal.OneSignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIO +import com.onesignal.common.threading.launchOnIO import com.onesignal.core.internal.background.IBackgroundManager import com.onesignal.debug.internal.logging.Logging +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.Job +import java.util.concurrent.atomic.AtomicReference class SyncJobService : JobService() { + private enum class RunState { + RUNNING, + STOPPED, + FINISHED, + } + + private class JobRun(val parameters: JobParameters) { + val state = AtomicReference(RunState.RUNNING) + val job = AtomicReference() + } + + private val activeRun = AtomicReference() + override fun onStartJob(jobParameters: JobParameters): Boolean { // Android delivers JobService.onStartJob on the main thread. The suspendifyOnIO call // below is the SDK's first IO-pool consumer on cold start in this process, and the @@ -45,52 +61,56 @@ class SyncJobService : JobService() { // be too late because the cold-init cost has already been paid on entry to the helper. OneSignalDispatchers.prewarm() - suspendifyOnIO { - var reschedule = false - - try { - // Init OneSignal in background - if (!OneSignal.initWithContext(this)) { - return@suspendifyOnIO - } - - val backgroundService = OneSignal.getService() - backgroundService.runBackgroundServices() - - Logging.debug("LollipopSyncRunnable:JobFinished needsJobReschedule: " + backgroundService.needsJobReschedule) - - // Reschedule if needed - reschedule = backgroundService.needsJobReschedule - backgroundService.needsJobReschedule = false - } finally { - // Always call jobFinished to finish the job; onStopJob will handle the case when init failed - jobFinished(jobParameters, reschedule) - } + val run = JobRun(jobParameters) + activeRun.getAndSet(run)?.let { previous -> + previous.state.compareAndSet(RunState.RUNNING, RunState.STOPPED) + previous.job.get()?.cancel() + } + val job = launchOnIO { executeRun(run) } + run.job.set(job) + if (run.state.get() == RunState.STOPPED) { + job.cancel() } - // Returning true means the job will always continue running and do everything else in IO thread - // When initWithContext failed, the background task will simply end return true } - override fun onStopJob(jobParameters: JobParameters): Boolean { - /* - * After 5.4, onStartJob calls initWithContext in background. That introduced a small possibility - * when onStopJob is called before the initialization completes in the background. When that happens, - * OneSignal.getService will run into a NPE. In that case, we just need to omit the job and do not - * reschedule. - */ - - // Additional hardening in the event of getService failure + private suspend fun executeRun(run: JobRun) { + var reschedule = false try { - // We assume init has been called via onStartJob\ + if (!OneSignal.initWithContext(this)) { + return + } + val backgroundService = OneSignal.getService() - val reschedule = backgroundService.cancelRunBackgroundServices() - Logging.debug("SyncJobService onStopJob called, system conditions not available reschedule: $reschedule") - return reschedule + backgroundService.runBackgroundServices() + reschedule = backgroundService.needsJobReschedule + backgroundService.needsJobReschedule = false + Logging.debug("LollipopSyncRunnable:JobFinished needsJobReschedule: $reschedule") + } catch (e: CancellationException) { + reschedule = true + throw e } catch (e: Exception) { - Logging.error("SyncJobService onStopJob failed, omit and do not reschedule") - return false + reschedule = true + Logging.error("SyncJobService background execution failed", e) + } finally { + if (run.state.compareAndSet(RunState.RUNNING, RunState.FINISHED)) { + activeRun.compareAndSet(run, null) + jobFinished(run.parameters, reschedule) + } + } + } + + override fun onStopJob(jobParameters: JobParameters): Boolean { + val run = activeRun.get() + val stopped = + run != null && + run.parameters === jobParameters && + run.state.compareAndSet(RunState.RUNNING, RunState.STOPPED) + if (stopped) { + activeRun.compareAndSet(run, null) + run?.job?.get()?.cancel() } + return stopped } } diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt index 121a195dc6..2e219cecc8 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt @@ -10,7 +10,9 @@ import kotlinx.coroutines.delay import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withContext import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong class OneSignalDispatchersTests : FunSpec({ @@ -24,6 +26,39 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.Default shouldNotBe null } + test("first launch returns while its lane is still being created off caller") { + OneSignalDispatchers.resetForTest() + val createStarted = CountDownLatch(1) + val allowCreate = CountDownLatch(1) + val launchReturned = CountDownLatch(1) + val workRan = CountDownLatch(1) + val createThreadId = AtomicLong() + val callerThreadId = AtomicLong() + + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") { + createThreadId.set(Thread.currentThread().id) + createStarted.countDown() + allowCreate.await() + } + } + + val caller = + Thread { + callerThreadId.set(Thread.currentThread().id) + OneSignalDispatchers.launchOnIO { workRan.countDown() } + launchReturned.countDown() + } + caller.start() + + createStarted.await(1, TimeUnit.SECONDS) shouldBe true + launchReturned.await(1, TimeUnit.SECONDS) shouldBe true + createThreadId.get() shouldNotBe callerThreadId.get() + allowCreate.countDown() + workRan.await(1, TimeUnit.SECONDS) shouldBe true + OneSignalDispatchers.beforeLaneCreateForTest = null + } + test("IO dispatcher should execute work on background thread") { val mainThreadId = Thread.currentThread().id var backgroundThreadId: Long? = null @@ -77,15 +112,19 @@ class OneSignalDispatchersTests : FunSpec({ } test("getStatus should return meaningful status information") { + OneSignalDispatchers.prewarm() + OneSignalDispatchers.awaitReadyForTest() shouldBe true val status = OneSignalDispatchers.getStatus() status shouldContain "OneSignalDispatchers Status:" status shouldContain "IO Executor: Active" status shouldContain "Default Executor: Active" status shouldContain "SerialIO Executor: Active" + status shouldContain "Ingress Executor: Active" status shouldContain "IO Scope: Active" status shouldContain "Default Scope: Active" status shouldContain "SerialIO Scope: Active" + status shouldContain "Ingress Scope: Active" } test("getPerformanceMetrics should include SerialIO queue and total completed task counters") { @@ -283,28 +322,10 @@ class OneSignalDispatchersTests : FunSpec({ } test("prewarm returns immediately and warms IO / Default / SerialIO dispatchers on a background thread") { - // SDK-4507: regression coverage for the cold-init main-thread block. prewarm() must - // (a) return on the caller's thread without ever doing the executor / dispatcher / - // scope construction work inline, and (b) leave all three dispatchers + scopes in the - // "Active" state once the dedicated daemon thread finishes its empty launches. - OneSignalDispatchers.resetPrewarmForTest() - val callerThreadId = Thread.currentThread().id - - // Call from the test thread (which stands in for the main thread under production - // usage). The call must return microseconds-fast; we don't assert wall-clock latency, - // just that the heavy work didn't happen on this thread. + OneSignalDispatchers.resetForTest() OneSignalDispatchers.prewarm() - // Resolve the prewarm thread by name from the JVM's thread set; its name is set by - // the prewarm() impl. We `join()` on it so the subsequent status assertions don't - // race a still-running prewarm thread. - val prewarmThread = - Thread.getAllStackTraces().keys.firstOrNull { it.name == "OneSignal-prewarm" } - prewarmThread?.join(2_000) - // After prewarm has finished, getStatus must report all three executors and scopes - // as Active. If the prewarm thread itself failed it would be a no-op for getStatus - // because the lazy chain wouldn't have run; this assertion proves both ends of the - // contract (heavy work was done, and it ran on the prewarm thread, not the caller). + OneSignalDispatchers.awaitReadyForTest() shouldBe true val status = OneSignalDispatchers.getStatus() status shouldContain "IO Executor: Active" status shouldContain "Default Executor: Active" @@ -312,30 +333,14 @@ class OneSignalDispatchersTests : FunSpec({ status shouldContain "IO Scope: Active" status shouldContain "Default Scope: Active" status shouldContain "SerialIO Scope: Active" - - // Sanity: the prewarm thread was a separate thread, not the test thread. - prewarmThread?.id shouldNotBe callerThreadId } test("prewarm is idempotent: a second call is a no-op and does not spawn a second prewarm thread") { - // The first prewarm() may have already run in earlier tests (or in the previous test - // above). Reset the latch so we get a deterministic "first call" here, then verify - // that the second call does NOT spawn another OneSignal-prewarm thread. - OneSignalDispatchers.resetPrewarmForTest() - + OneSignalDispatchers.resetForTest() OneSignalDispatchers.prewarm() - val firstPrewarmThread = - Thread.getAllStackTraces().keys.firstOrNull { it.name == "OneSignal-prewarm" } - firstPrewarmThread?.join(2_000) - - // Snapshot any straggling "OneSignal-prewarm" threads -- there should be at most one - // (the one above, possibly still in TERMINATED state in the JVM's thread set briefly). - val countBefore = Thread.getAllStackTraces().keys.count { it.name == "OneSignal-prewarm" } - - // Second call must be a no-op. No new prewarm thread, no exception. + OneSignalDispatchers.awaitReadyForTest() shouldBe true + val statusBefore = OneSignalDispatchers.getStatus() OneSignalDispatchers.prewarm() - val countAfter = Thread.getAllStackTraces().keys.count { it.name == "OneSignal-prewarm" } - - countAfter shouldBe countBefore + OneSignalDispatchers.getStatus() shouldBe statusBefore } }) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsTests.kt index d47fa33096..3f149ed940 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsTests.kt @@ -8,6 +8,7 @@ import io.kotest.matchers.shouldBe import io.kotest.matchers.shouldNotBe import kotlinx.coroutines.delay import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger class ThreadUtilsTests : FunSpec({ @@ -158,6 +159,26 @@ class ThreadUtilsTests : FunSpec({ onCompleteCalled shouldBe true } + test("suspendifyOnIngress completes when a cold queued dispatch is cancelled") { + OneSignalDispatchers.resetForTest() + val createStarted = CountDownLatch(1) + val allowCreate = CountDownLatch(1) + val completed = CountDownLatch(1) + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "INGRESS") { + createStarted.countDown() + allowCreate.await() + } + } + + suspendifyOnIngress(block = {}, onComplete = { completed.countDown() }) + createStarted.await() + OneSignalDispatchers.resetForTest() + + completed.await(1, TimeUnit.SECONDS) shouldBe true + allowCreate.countDown() + } + test("suspendifyWithErrorHandling should handle errors properly") { var errorHandled = false var onCompleteCalled = false diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/services/SyncJobServiceTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/services/SyncJobServiceTests.kt index 7b603a199f..26b1b6d6f8 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/services/SyncJobServiceTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/services/SyncJobServiceTests.kt @@ -3,7 +3,6 @@ package com.onesignal.core.services import android.app.job.JobParameters import com.onesignal.OneSignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIO import com.onesignal.core.internal.background.IBackgroundManager import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging @@ -21,6 +20,7 @@ import io.mockk.spyk import io.mockk.unmockkObject import io.mockk.verify import io.mockk.verifyOrder +import kotlinx.coroutines.runBlocking private class Mocks { val syncJobService = spyk(SyncJobService(), recordPrivateCalls = true) @@ -51,7 +51,7 @@ class SyncJobServiceTests : FunSpec({ unmockkObject(OneSignal) } - test("onStartJob calls prewarm before suspendifyOnIO") { + test("onStartJob calls prewarm before launchOnIO") { coEvery { OneSignal.initWithContext(any()) } returns false mocks.syncJobService.onStartJob(mocks.jobParameters) @@ -61,7 +61,7 @@ class SyncJobServiceTests : FunSpec({ // would already be paid on the caller (main) thread by the time the helper is entered. verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>()) + OneSignalDispatchers.launchOnIO(any Unit>()) } } @@ -157,44 +157,30 @@ class SyncJobServiceTests : FunSpec({ verify { mockBackgroundManager.needsJobReschedule = false } } - test("onStopJob returns false when OneSignal.getService throws") { - // Given - val syncJobService = mocks.syncJobService - val jobParameters = mocks.jobParameters - coEvery { OneSignal.getService() } throws NullPointerException() - - // When - val result = syncJobService.onStopJob(jobParameters) - - // Then - result shouldBe false - } - - test("onStopJob calls cancelRunBackgroundServices and returns its result") { - // Given - val mockBackgroundManager = mocks.mockBackgroundManager - val syncJobService = mocks.syncJobService - val jobParameters = mocks.jobParameters - every { mockBackgroundManager.cancelRunBackgroundServices() } returns true + test("onStopJob cancels the owned coroutine without resolving services") { + val job = mockk(relaxed = true) + every { OneSignalDispatchers.launchOnIO(any Unit>()) } returns job - // When - val result = syncJobService.onStopJob(jobParameters) + mocks.syncJobService.onStartJob(mocks.jobParameters) + val result = mocks.syncJobService.onStopJob(mocks.jobParameters) - // Then result shouldBe true - verify { mockBackgroundManager.cancelRunBackgroundServices() } + verify { job.cancel() } + verify(exactly = 0) { OneSignal.getService() } + every { OneSignalDispatchers.launchOnIO(any Unit>()) } answers { + runBlocking { firstArg Unit>().invoke() } + mockk(relaxed = true) + } } - test("onStopJob returns false when cancelRunBackgroundServices returns false") { - // Given - val mockBackgroundManager = mocks.mockBackgroundManager - every { mockBackgroundManager.cancelRunBackgroundServices() } returns false + test("onStopJob returns false when no run is active") { + mocks.syncJobService.onStopJob(mocks.jobParameters) shouldBe false + } - // When - val result = mocks.syncJobService.onStopJob(mocks.jobParameters) + test("onStopJob does not reschedule a run that already completed") { + coEvery { OneSignal.initWithContext(any()) } returns false + mocks.syncJobService.onStartJob(mocks.jobParameters) - // Then - result shouldBe false - verify { mockBackgroundManager.cancelRunBackgroundServices() } + mocks.syncJobService.onStopJob(mocks.jobParameters) shouldBe false } }) diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/NotificationsModule.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/NotificationsModule.kt index c26dea5a45..e2ec80fd0a 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/NotificationsModule.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/NotificationsModule.kt @@ -34,6 +34,7 @@ import com.onesignal.notifications.internal.generation.INotificationGenerationPr import com.onesignal.notifications.internal.generation.INotificationGenerationWorkManager import com.onesignal.notifications.internal.generation.impl.NotificationGenerationProcessor import com.onesignal.notifications.internal.generation.impl.NotificationGenerationWorkManager +import com.onesignal.notifications.internal.ingress.NotificationIngressDrainStarter import com.onesignal.notifications.internal.lifecycle.INotificationLifecycleService import com.onesignal.notifications.internal.lifecycle.impl.NotificationLifecycleService import com.onesignal.notifications.internal.limiting.INotificationLimitManager @@ -141,6 +142,7 @@ internal class NotificationsModule : IModule { // Startable services builder.register().provides() + builder.register().provides() builder.register() .provides() diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/generation/impl/NotificationGenerationWorkManager.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/generation/impl/NotificationGenerationWorkManager.kt index 0dc570df7e..17637acf3b 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/generation/impl/NotificationGenerationWorkManager.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/generation/impl/NotificationGenerationWorkManager.kt @@ -18,6 +18,7 @@ import org.json.JSONObject import java.util.concurrent.ConcurrentHashMap internal class NotificationGenerationWorkManager : INotificationGenerationWorkManager { + @Suppress("ReturnCount", "TooGenericExceptionCaught") override fun beginEnqueueingWork( context: Context, osNotificationId: String, @@ -39,26 +40,30 @@ internal class NotificationGenerationWorkManager : INotificationGenerationWorkMa return true } - // TODO: Need to figure out how to implement the isHighPriority param - val inputData = - Data.Builder() - .putString(OS_ID_DATA_PARAM, id) - .putInt(ANDROID_NOTIF_ID_WORKER_DATA_PARAM, androidNotificationId) - .putString(JSON_PAYLOAD_WORKER_DATA_PARAM, jsonPayload.toString()) - .putLong(TIMESTAMP_WORKER_DATA_PARAM, timestamp) - .putBoolean(IS_RESTORING_WORKER_DATA_PARAM, isRestoring) - .build() - val workRequest = - OneTimeWorkRequest.Builder(NotificationGenerationWorker::class.java) - .setInputData(inputData) - .build() - Logging.debug( - "NotificationWorkManager enqueueing notification work with notificationId: $osNotificationId and jsonPayload: $jsonPayload", - ) - OSWorkManagerHelper.getInstance(context) - .enqueueUniqueWork(osNotificationId, ExistingWorkPolicy.KEEP, workRequest) - - return true + return try { + // TODO: Need to figure out how to implement the isHighPriority param + val inputData = + Data.Builder() + .putString(OS_ID_DATA_PARAM, id) + .putInt(ANDROID_NOTIF_ID_WORKER_DATA_PARAM, androidNotificationId) + .putString(JSON_PAYLOAD_WORKER_DATA_PARAM, jsonPayload.toString()) + .putLong(TIMESTAMP_WORKER_DATA_PARAM, timestamp) + .putBoolean(IS_RESTORING_WORKER_DATA_PARAM, isRestoring) + .build() + val workRequest = + OneTimeWorkRequest.Builder(NotificationGenerationWorker::class.java) + .setInputData(inputData) + .build() + Logging.debug( + "NotificationWorkManager enqueueing notification work with notificationId: $osNotificationId and jsonPayload: $jsonPayload", + ) + OSWorkManagerHelper.getInstance(context) + .enqueueUniqueWork(osNotificationId, ExistingWorkPolicy.KEEP, workRequest) + true + } catch (e: Exception) { + removeNotificationIdProcessed(id) + throw e + } } class NotificationGenerationWorker(context: Context, workerParams: WorkerParameters) : CoroutineWorker(context, workerParams) { @@ -71,12 +76,13 @@ internal class NotificationGenerationWorkManager : INotificationGenerationWorkMa val notificationProcessor: INotificationGenerationProcessor = OneSignal.getService() val inputData = inputData val id = inputData.getString(OS_ID_DATA_PARAM) ?: return Result.failure() + val payload = inputData.getString(JSON_PAYLOAD_WORKER_DATA_PARAM) ?: return Result.failure() return try { Logging.debug("NotificationWorker running doWork with data: $inputData") val androidNotificationId = inputData.getInt(ANDROID_NOTIF_ID_WORKER_DATA_PARAM, 0) - val jsonPayload = JSONObject(inputData.getString(JSON_PAYLOAD_WORKER_DATA_PARAM)) + val jsonPayload = JSONObject(payload) val timestamp = inputData.getLong( TIMESTAMP_WORKER_DATA_PARAM, @@ -96,7 +102,7 @@ internal class NotificationGenerationWorkManager : INotificationGenerationWorkMa Logging.error("Error occurred doing work for job with id: $id", e) Result.failure() } finally { - removeNotificationIdProcessed(id!!) + removeNotificationIdProcessed(id) } } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/ingress/NotificationIngress.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/ingress/NotificationIngress.kt new file mode 100644 index 0000000000..f18336fcb2 --- /dev/null +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/ingress/NotificationIngress.kt @@ -0,0 +1,320 @@ +package com.onesignal.notifications.internal.ingress + +import android.content.ContentValues +import android.content.Context +import android.content.Intent +import android.database.sqlite.SQLiteDatabase +import android.database.sqlite.SQLiteOpenHelper +import android.os.Bundle +import androidx.work.CoroutineWorker +import androidx.work.ExistingWorkPolicy +import androidx.work.OneTimeWorkRequest +import androidx.work.WorkerParameters +import com.onesignal.OneSignal +import com.onesignal.common.threading.OneSignalDispatchers +import com.onesignal.core.internal.application.IApplicationService +import com.onesignal.core.internal.startup.IStartableService +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.notifications.internal.bundle.INotificationBundleProcessor +import com.onesignal.notifications.internal.common.NotificationConstants +import com.onesignal.notifications.internal.common.NotificationFormatHelper +import com.onesignal.notifications.internal.common.OSWorkManagerHelper +import com.onesignal.notifications.internal.open.INotificationOpenedProcessor +import com.onesignal.notifications.internal.restoration.impl.NotificationRestoreWorkManager +import org.json.JSONObject + +internal object NotificationIngress { + private const val DRAIN_WORK_NAME = "OneSignalNotificationIngressDrain" + + @Volatile + internal var drainSchedulerForTest: ((Context) -> Unit)? = null + + fun persistFcm( + context: Context, + intent: Intent, + bundle: Bundle, + ): Boolean { + val notificationId = NotificationFormatHelper.getOSNotificationIdFromJson(BundleCodec.toJson(bundle)) ?: return false + val record = + IngressRecord( + id = "fcm:$notificationId", + kind = IngressKind.FCM, + action = intent.action, + payload = BundleCodec.encode(bundle), + createdAtMs = System.currentTimeMillis(), + ) + IngressStore.get(context).put(record) + scheduleDrainBestEffort(context) + return true + } + + fun persistDismiss( + context: Context, + intent: Intent, + ) { + val bundle = intent.extras ?: Bundle() + val notificationId = bundle.getInt(NotificationConstants.BUNDLE_KEY_ANDROID_NOTIFICATION_ID, 0) + val summary = bundle.getString("summary").orEmpty() + val payload = BundleCodec.encode(bundle) + val record = + IngressRecord( + id = "dismiss:$notificationId:$summary:${payload.hashCode()}", + kind = IngressKind.DISMISS, + action = intent.action, + payload = payload, + createdAtMs = System.currentTimeMillis(), + ) + IngressStore.get(context).put(record) + scheduleDrainBestEffort(context) + } + + fun enqueueRestore(context: Context) { + NotificationRestoreWorkManager().beginEnqueueingWork(context, true) + } + + fun scheduleDrain(context: Context) { + drainSchedulerForTest?.let { + it(context) + return + } + val request = OneTimeWorkRequest.Builder(NotificationIngressDrainWorker::class.java).build() + OSWorkManagerHelper.getInstance(context.applicationContext) + .enqueueUniqueWork(DRAIN_WORK_NAME, ExistingWorkPolicy.KEEP, request) + } + + @Suppress("TooGenericExceptionCaught") + private fun scheduleDrainBestEffort(context: Context) { + if (drainSchedulerForTest != null) { + try { + scheduleDrain(context) + } catch (e: Exception) { + Logging.warn("Notification ingress persisted; drain scheduling will retry on next startup", e) + } + return + } + OneSignalDispatchers.launchOnIO { + try { + scheduleDrain(context) + } catch (e: Exception) { + Logging.warn("Notification ingress persisted; drain scheduling will retry on next startup", e) + } + } + } + + internal fun pendingCountForTest(context: Context): Int = IngressStore.get(context).count() + + internal fun resetForTest(context: Context) { + drainSchedulerForTest = null + IngressStore.get(context).clear() + } +} + +internal class NotificationIngressDrainStarter( + private val applicationService: IApplicationService, +) : IStartableService { + @Suppress("TooGenericExceptionCaught") + override fun start() { + try { + NotificationIngress.scheduleDrain(applicationService.appContext) + } catch (e: Exception) { + Logging.warn("Notification ingress startup drain scheduling failed", e) + } + } +} + +internal class NotificationIngressDrainWorker( + context: Context, + workerParameters: WorkerParameters, +) : CoroutineWorker(context, workerParameters) { + @Suppress("TooGenericExceptionCaught") + override suspend fun doWork(): Result { + val store = IngressStore.get(applicationContext) + if (!OneSignal.initWithContext(applicationContext)) return Result.retry() + + return try { + for (record in store.list()) { + when (record.kind) { + IngressKind.FCM -> processFcm(record) + IngressKind.DISMISS -> processDismiss(record) + } + store.delete(record.id) + } + Result.success() + } catch (e: Exception) { + Logging.error("Notification ingress drain failed", e) + Result.retry() + } + } + + private fun processFcm(record: IngressRecord) { + val bundle = BundleCodec.decode(record.payload) + OneSignal.getService() + .processBundleFromReceiver(applicationContext, bundle) + } + + private suspend fun processDismiss(record: IngressRecord) { + val intent = Intent(record.action).putExtras(BundleCodec.decode(record.payload)) + OneSignal.getService() + .processFromContext(applicationContext, intent) + } +} + +private enum class IngressKind { + FCM, + DISMISS, +} + +private data class IngressRecord( + val id: String, + val kind: IngressKind, + val action: String?, + val payload: String, + val createdAtMs: Long, +) + +private class IngressStore private constructor(context: Context) : + SQLiteOpenHelper(context.applicationContext, DATABASE_NAME, null, DATABASE_VERSION) { + override fun onCreate(database: SQLiteDatabase) { + database.execSQL( + """ + CREATE TABLE $TABLE ( + id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + action TEXT, + payload TEXT NOT NULL, + created_at INTEGER NOT NULL + ) + """.trimIndent(), + ) + } + + override fun onUpgrade( + database: SQLiteDatabase, + oldVersion: Int, + newVersion: Int, + ) { + database.execSQL("DROP TABLE IF EXISTS $TABLE") + onCreate(database) + } + + fun put(record: IngressRecord) { + val values = + ContentValues().apply { + put("id", record.id) + put("kind", record.kind.name) + put("action", record.action) + put("payload", record.payload) + put("created_at", record.createdAtMs) + } + check(writableDatabase.insertWithOnConflict(TABLE, null, values, SQLiteDatabase.CONFLICT_REPLACE) != -1L) { + "Unable to persist notification ingress" + } + } + + fun list(): List { + val records = mutableListOf() + readableDatabase.query( + TABLE, + arrayOf("id", "kind", "action", "payload", "created_at"), + null, + null, + null, + null, + "created_at ASC", + ).use { cursor -> + while (cursor.moveToNext()) { + records += + IngressRecord( + id = cursor.getString(ID_COLUMN_INDEX), + kind = IngressKind.valueOf(cursor.getString(KIND_COLUMN_INDEX)), + action = cursor.getString(ACTION_COLUMN_INDEX), + payload = cursor.getString(PAYLOAD_COLUMN_INDEX), + createdAtMs = cursor.getLong(CREATED_AT_COLUMN_INDEX), + ) + } + } + return records + } + + fun delete(id: String) { + writableDatabase.delete(TABLE, "id = ?", arrayOf(id)) + } + + fun count(): Int = + readableDatabase.rawQuery("SELECT COUNT(*) FROM $TABLE", null).use { cursor -> + cursor.moveToFirst() + cursor.getInt(0) + } + + fun clear() { + writableDatabase.delete(TABLE, null, null) + } + + companion object { + private const val DATABASE_NAME = "OneSignalIngress.db" + private const val DATABASE_VERSION = 1 + private const val TABLE = "notification_ingress" + private const val ID_COLUMN_INDEX = 0 + private const val KIND_COLUMN_INDEX = 1 + private const val ACTION_COLUMN_INDEX = 2 + private const val PAYLOAD_COLUMN_INDEX = 3 + private const val CREATED_AT_COLUMN_INDEX = 4 + + @Volatile + private var instance: IngressStore? = null + + fun get(context: Context): IngressStore = + instance ?: synchronized(this) { + instance ?: IngressStore(context).also { instance = it } + } + } +} + +@Suppress("DEPRECATION") +private object BundleCodec { + private const val TYPE = "type" + private const val VALUE = "value" + + fun encode(bundle: Bundle): String { + val root = JSONObject() + for (key in bundle.keySet()) { + val value = bundle[key] + val encoded = JSONObject() + when (value) { + is Boolean -> encoded.put(TYPE, "boolean").put(VALUE, value) + is Int -> encoded.put(TYPE, "int").put(VALUE, value) + is Long -> encoded.put(TYPE, "long").put(VALUE, value) + is Double -> encoded.put(TYPE, "double").put(VALUE, value) + else -> encoded.put(TYPE, "string").put(VALUE, value?.toString()) + } + root.put(key, encoded) + } + return root.toString() + } + + fun decode(encoded: String): Bundle { + val root = JSONObject(encoded) + val bundle = Bundle() + val keys = root.keys() + while (keys.hasNext()) { + val key = keys.next() + val value = root.getJSONObject(key) + when (value.getString(TYPE)) { + "boolean" -> bundle.putBoolean(key, value.getBoolean(VALUE)) + "int" -> bundle.putInt(key, value.getInt(VALUE)) + "long" -> bundle.putLong(key, value.getLong(VALUE)) + "double" -> bundle.putDouble(key, value.getDouble(VALUE)) + else -> bundle.putString(key, if (value.isNull(VALUE)) null else value.getString(VALUE)) + } + } + return bundle + } + + fun toJson(bundle: Bundle): JSONObject { + val json = JSONObject() + for (key in bundle.keySet()) { + json.put(key, bundle[key]) + } + return json + } +} diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/restoration/impl/NotificationRestoreWorkManager.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/restoration/impl/NotificationRestoreWorkManager.kt index 78d9c4718d..04f484e98e 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/restoration/impl/NotificationRestoreWorkManager.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/restoration/impl/NotificationRestoreWorkManager.kt @@ -12,38 +12,32 @@ import com.onesignal.notifications.internal.common.OSWorkManagerHelper import com.onesignal.notifications.internal.restoration.INotificationRestoreProcessor import com.onesignal.notifications.internal.restoration.INotificationRestoreWorkManager import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean internal class NotificationRestoreWorkManager : INotificationRestoreWorkManager { - // Notifications will never be force removed when the app's process is running, - // so we only need to restore at most once per cold start of the app. - private var restored = false - private val lock = Any() - + @Suppress("TooGenericExceptionCaught") override fun beginEnqueueingWork( context: Context, shouldDelay: Boolean, ) { - // Only allow one piece of work to be enqueued. - synchronized(lock) { - if (restored) { - return - } + if (!restored.compareAndSet(false, true)) return - restored = true + try { + val restoreDelayInSeconds = if (shouldDelay) 15 else 0 + val workRequest = + OneTimeWorkRequest.Builder(NotificationRestoreWorker::class.java) + .setInitialDelay(restoreDelayInSeconds.toLong(), TimeUnit.SECONDS) + .build() + OSWorkManagerHelper.getInstance(context) + .enqueueUniqueWork( + NOTIFICATION_RESTORE_WORKER_IDENTIFIER, + ExistingWorkPolicy.KEEP, + workRequest, + ) + } catch (e: Exception) { + restored.set(false) + throw e } - - // When boot or upgrade, add a 15 second delay to alleviate app doing to much work all at once - val restoreDelayInSeconds = if (shouldDelay) 15 else 0 - val workRequest = - OneTimeWorkRequest.Builder(NotificationRestoreWorker::class.java) - .setInitialDelay(restoreDelayInSeconds.toLong(), TimeUnit.SECONDS) - .build() - OSWorkManagerHelper.getInstance(context!!) - .enqueueUniqueWork( - NOTIFICATION_RESTORE_WORKER_IDENTIFIER, - ExistingWorkPolicy.KEEP, - workRequest, - ) } class NotificationRestoreWorker(context: Context, workerParams: WorkerParameters) : CoroutineWorker(context, workerParams) { @@ -69,6 +63,8 @@ internal class NotificationRestoreWorkManager : INotificationRestoreWorkManager } companion object { - private val NOTIFICATION_RESTORE_WORKER_IDENTIFIER = NotificationRestoreWorker::class.java.canonicalName + private val NOTIFICATION_RESTORE_WORKER_IDENTIFIER = + NotificationRestoreWorker::class.java.canonicalName ?: NotificationRestoreWorker::class.java.name + private val restored = AtomicBoolean(false) } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt index 5da3677b37..2382bc820d 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt @@ -29,11 +29,9 @@ package com.onesignal.notifications.receivers import android.content.BroadcastReceiver import android.content.Context import android.content.Intent -import com.onesignal.OneSignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIO -import com.onesignal.debug.internal.logging.Logging -import com.onesignal.notifications.internal.restoration.INotificationRestoreWorkManager +import com.onesignal.common.threading.suspendifyOnIngress +import com.onesignal.notifications.internal.ingress.NotificationIngress class BootUpReceiver : BroadcastReceiver() { override fun onReceive( @@ -44,19 +42,18 @@ class BootUpReceiver : BroadcastReceiver() { // goAsync() so the daemon has lead time before the first suspendifyOnIO dispatch. OneSignalDispatchers.prewarm() - val pendingResult: BroadcastReceiver.PendingResult? = goAsync() - // in background, init onesignal and begin enqueueing restore work - suspendifyOnIO( + val completion = + BroadcastCompletion( + "BootUpReceiver", + goAsync(), + BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, + ) + // Persist the reconstructible restore request without waiting for full SDK initialization. + suspendifyOnIngress( block = { - if (!OneSignal.initWithContext(context.applicationContext)) { - Logging.warn("NotificationRestoreReceiver skipped due to failed OneSignal init") - return@suspendifyOnIO - } - - val restoreWorkManager = OneSignal.getService() - restoreWorkManager.beginEnqueueingWork(context, true) + NotificationIngress.enqueueRestore(context) }, - onComplete = { pendingResult?.finish() }, + onComplete = { completion.finish() }, ) } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt new file mode 100644 index 0000000000..60287ac890 --- /dev/null +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt @@ -0,0 +1,40 @@ +package com.onesignal.notifications.receivers + +import android.content.BroadcastReceiver +import android.os.Handler +import android.os.Looper +import android.os.SystemClock +import com.onesignal.debug.internal.logging.Logging +import java.util.concurrent.atomic.AtomicBoolean + +internal class BroadcastCompletion( + private val receiverName: String, + private val pendingResult: BroadcastReceiver.PendingResult?, + timeoutMs: Long? = null, +) { + private val startedAtMs = SystemClock.elapsedRealtime() + private val finished = AtomicBoolean(false) + private val handler = Handler(Looper.getMainLooper()) + private val timeout = Runnable { finish("deadline") } + + init { + if (timeoutMs != null) handler.postDelayed(timeout, timeoutMs) + } + + fun finish(reason: String = "completed") { + if (!finished.compareAndSet(false, true)) return + handler.removeCallbacks(timeout) + pendingResult?.finish() + val durationMs = SystemClock.elapsedRealtime() - startedAtMs + if (durationMs >= SOFT_DEADLINE_MS) { + Logging.warn("$receiverName durable handoff finished after ${durationMs}ms ($reason)") + } else { + Logging.debug("$receiverName durable handoff finished after ${durationMs}ms ($reason)") + } + } + + companion object { + const val RECONSTRUCTIBLE_WORK_TIMEOUT_MS = 8_000L + private const val SOFT_DEADLINE_MS = 4_000L + } +} diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt index f0552a733e..ab0968d40b 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt @@ -4,11 +4,9 @@ import android.app.Activity import android.content.BroadcastReceiver import android.content.Context import android.content.Intent -import com.onesignal.OneSignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIO -import com.onesignal.debug.internal.logging.Logging -import com.onesignal.notifications.internal.bundle.INotificationBundleProcessor +import com.onesignal.common.threading.suspendifyOnIngress +import com.onesignal.notifications.internal.ingress.NotificationIngress // This is the entry point when a FCM payload is received from the Google Play services app // OneSignal does not use FirebaseMessagingService.onMessageReceived as it does not allow multiple @@ -31,33 +29,22 @@ class FCMBroadcastReceiver : BroadcastReceiver() { // likely to be warm by the time the suspendifyOnIO below submits its work. OneSignalDispatchers.prewarm() - val pendingResult: BroadcastReceiver.PendingResult? = goAsync() + val completion = BroadcastCompletion("FCMBroadcastReceiver", goAsync()) // process in background - suspendifyOnIO( + suspendifyOnIngress( block = { - if (!OneSignal.initWithContext(context.applicationContext)) { - Logging.warn("FCMBroadcastReceiver skipped due to failed OneSignal init") - return@suspendifyOnIO - } - - val bundleProcessor = OneSignal.getService() - if (!isFCMMessage(intent)) { setSuccessfulResultCode() - return@suspendifyOnIO + return@suspendifyOnIngress } - val processedResult = bundleProcessor.processBundleFromReceiver(context, bundle) - - // Prevent other FCM receivers from firing if work manager is processing the notification - if (processedResult?.isWorkManagerProcessing == true) { + if (NotificationIngress.persistFcm(context, intent, bundle)) { setAbort() - return@suspendifyOnIO + } else { + setSuccessfulResultCode() } - - setSuccessfulResultCode() }, - onComplete = { pendingResult?.finish() }, + onComplete = { completion.finish() }, ) } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt index a553f36ae9..86640e2b95 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt @@ -27,13 +27,9 @@ package com.onesignal.notifications.receivers import android.content.BroadcastReceiver import android.content.Context import android.content.Intent -import com.onesignal.OneSignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIO -import com.onesignal.debug.internal.logging.Logging -import com.onesignal.notifications.internal.open.INotificationOpenedProcessor -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.withContext +import com.onesignal.common.threading.suspendifyOnIngress +import com.onesignal.notifications.internal.ingress.NotificationIngress class NotificationDismissReceiver : BroadcastReceiver() { override fun onReceive( @@ -44,23 +40,13 @@ class NotificationDismissReceiver : BroadcastReceiver() { // goAsync() so the daemon has lead time before the first suspendifyOnIO dispatch. OneSignalDispatchers.prewarm() - val pendingResult: BroadcastReceiver.PendingResult? = goAsync() + val completion = BroadcastCompletion("NotificationDismissReceiver", goAsync()) - suspendifyOnIO( + suspendifyOnIngress( block = { - if (!OneSignal.initWithContext(context.applicationContext)) { - Logging.warn("NotificationOpenedReceiver skipped due to failed OneSignal init") - return@suspendifyOnIO - } - - val notificationOpenedProcessor = OneSignal.getService() - - // init OneSignal in background but process in main - withContext(Dispatchers.Main) { - notificationOpenedProcessor.processFromContext(context, intent) - } + NotificationIngress.persistDismiss(context, intent) }, - onComplete = { pendingResult?.finish() }, + onComplete = { completion.finish() }, ) } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt index 5ab5a8fe00..f150b128d1 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt @@ -30,11 +30,9 @@ import android.content.BroadcastReceiver import android.content.Context import android.content.Intent import android.os.Build -import com.onesignal.OneSignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIO -import com.onesignal.debug.internal.logging.Logging -import com.onesignal.notifications.internal.restoration.INotificationRestoreWorkManager +import com.onesignal.common.threading.suspendifyOnIngress +import com.onesignal.notifications.internal.ingress.NotificationIngress class UpgradeReceiver : BroadcastReceiver() { override fun onReceive( @@ -53,20 +51,19 @@ class UpgradeReceiver : BroadcastReceiver() { // goAsync() so the daemon has lead time before the first suspendifyOnIO dispatch. OneSignalDispatchers.prewarm() - val pendingResult: BroadcastReceiver.PendingResult? = goAsync() + val completion = + BroadcastCompletion( + "UpgradeReceiver", + goAsync(), + BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, + ) - // init OneSignal and enqueue restore work in background - suspendifyOnIO( + // Persist the reconstructible restore request without waiting for full SDK initialization. + suspendifyOnIngress( block = { - if (!OneSignal.initWithContext(context.applicationContext)) { - Logging.warn("UpgradeReceiver skipped due to failed OneSignal init") - return@suspendifyOnIO - } - - val restoreWorkManager = OneSignal.getService() - restoreWorkManager.beginEnqueueingWork(context, true) + NotificationIngress.enqueueRestore(context) }, - onComplete = { pendingResult?.finish() }, + onComplete = { completion.finish() }, ) } } diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/ingress/NotificationIngressTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/ingress/NotificationIngressTests.kt new file mode 100644 index 0000000000..cf8438cc90 --- /dev/null +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/ingress/NotificationIngressTests.kt @@ -0,0 +1,63 @@ +package com.onesignal.notifications.internal.ingress + +import android.content.Context +import android.content.Intent +import android.os.Bundle +import androidx.test.core.app.ApplicationProvider +import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.shouldBe + +@RobolectricTest +class NotificationIngressTests : FunSpec({ + lateinit var context: Context + + beforeAny { + context = ApplicationProvider.getApplicationContext() + NotificationIngress.resetForTest(context) + } + + test("FCM input remains durable when drain scheduling fails") { + NotificationIngress.drainSchedulerForTest = { throw IllegalStateException("scheduler unavailable") } + val bundle = + Bundle().apply { + putString("custom", """{"i":"notification-id"}""") + putString("alert", "message") + } + + NotificationIngress.persistFcm( + context, + Intent("com.google.android.c2dm.intent.RECEIVE"), + bundle, + ) shouldBe true + + NotificationIngress.pendingCountForTest(context) shouldBe 1 + } + + test("duplicate FCM input replaces the same durable record") { + NotificationIngress.drainSchedulerForTest = {} + val bundle = Bundle().apply { putString("custom", """{"i":"notification-id"}""") } + val intent = Intent("com.google.android.c2dm.intent.RECEIVE") + + NotificationIngress.persistFcm(context, intent, bundle) + NotificationIngress.persistFcm(context, intent, bundle) + + NotificationIngress.pendingCountForTest(context) shouldBe 1 + } + + test("dismiss input is persisted before scheduling") { + var countAtSchedule = 0 + NotificationIngress.drainSchedulerForTest = { + countAtSchedule = NotificationIngress.pendingCountForTest(context) + } + val intent = + Intent().apply { + putExtra("androidNotificationId", 42) + putExtra("dismissed", true) + } + + NotificationIngress.persistDismiss(context, intent) + + countAtSchedule shouldBe 1 + } +}) diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/BroadcastCompletionTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/BroadcastCompletionTests.kt new file mode 100644 index 0000000000..efa562e7ee --- /dev/null +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/BroadcastCompletionTests.kt @@ -0,0 +1,32 @@ +package com.onesignal.notifications.receivers + +import android.content.BroadcastReceiver +import android.os.Looper +import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest +import io.kotest.core.spec.style.FunSpec +import io.mockk.mockk +import io.mockk.verify +import org.robolectric.Shadows.shadowOf +import java.time.Duration + +@RobolectricTest +class BroadcastCompletionTests : FunSpec({ + test("finish is exact once") { + val pendingResult = mockk(relaxed = true) + val completion = BroadcastCompletion("test", pendingResult) + + completion.finish() + completion.finish("deadline") + + verify(exactly = 1) { pendingResult.finish() } + } + + test("reconstructible work finishes at its deadline") { + val pendingResult = mockk(relaxed = true) + BroadcastCompletion("test", pendingResult, 100) + + shadowOf(Looper.getMainLooper()).idleFor(Duration.ofMillis(100)) + + verify(exactly = 1) { pendingResult.finish() } + } +}) diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt index fa3ca42ee3..ef37b5643b 100644 --- a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt @@ -6,11 +6,13 @@ import androidx.test.core.app.ApplicationProvider import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest import com.onesignal.OneSignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIO +import com.onesignal.common.threading.suspendifyOnIngress import com.onesignal.mocks.IOMockHelper +import com.onesignal.notifications.internal.ingress.NotificationIngress import io.kotest.core.spec.style.FunSpec import io.mockk.clearMocks import io.mockk.coEvery +import io.mockk.every import io.mockk.mockkObject import io.mockk.unmockkObject import io.mockk.verify @@ -27,6 +29,8 @@ class FCMBroadcastReceiverTests : FunSpec({ clearMocks(OneSignalDispatchers, answers = false) mockkObject(OneSignal) coEvery { OneSignal.initWithContext(any()) } returns false + mockkObject(NotificationIngress) + every { NotificationIngress.persistFcm(any(), any(), any()) } returns true } afterAny { @@ -34,12 +38,13 @@ class FCMBroadcastReceiverTests : FunSpec({ // are owned by IOMockHelper and torn down in its afterSpec — unmockkAll() here would strip // them mid-spec and break the remaining tests. unmockkObject(OneSignal) + unmockkObject(NotificationIngress) } test("FCMBroadcastReceiver.onReceive makes the explicit prewarm() head-start call before dispatch for a normal push") { // Scope of this test: it asserts the explicit `OneSignalDispatchers.prewarm()` call in - // onReceive (the goAsync() head start) happens before the suspendifyOnIO dispatch. - // IOMockHelper stubs `suspendifyOnIO` (run inline) and prewarm(), so this verifies + // onReceive (the goAsync() head start) happens before the ingress dispatch. + // IOMockHelper stubs `suspendifyOnIngress` (run inline) and prewarm(), so this verifies // placement/ordering, not end-to-end cold-init behavior. val context = ApplicationProvider.getApplicationContext() val intent = @@ -53,7 +58,7 @@ class FCMBroadcastReceiverTests : FunSpec({ verify(exactly = 1) { OneSignalDispatchers.prewarm() } verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>(), any<() -> Unit>()) + suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } } diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt index cb0da39d72..0ec4a01a0e 100644 --- a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt @@ -6,11 +6,13 @@ import androidx.test.core.app.ApplicationProvider import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest import com.onesignal.OneSignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIO +import com.onesignal.common.threading.suspendifyOnIngress import com.onesignal.mocks.IOMockHelper +import com.onesignal.notifications.internal.ingress.NotificationIngress import io.kotest.core.spec.style.FunSpec import io.mockk.clearMocks import io.mockk.coEvery +import io.mockk.every import io.mockk.mockkObject import io.mockk.unmockkObject import io.mockk.verify @@ -19,7 +21,7 @@ import io.mockk.verifyOrder /** * Verifies each cold-start broadcast-receiver entry point makes the explicit * [OneSignalDispatchers.prewarm] head-start call before its first dispatch. [IOMockHelper] stubs - * `suspendifyOnIO` (run inline) and `prewarm()`, so these assert placement/ordering of the explicit + * `suspendifyOnIngress` (run inline) and `prewarm()`, so these assert placement/ordering of the explicit * call, not end-to-end cold-init behavior. * * Not covered here (no unit-test path): `ADMMessageHandler` / `ADMMessageHandlerJob` (the @@ -37,6 +39,9 @@ class PrewarmEntryPointTests : FunSpec({ clearMocks(OneSignalDispatchers, answers = false) mockkObject(OneSignal) coEvery { OneSignal.initWithContext(any()) } returns false + mockkObject(NotificationIngress) + every { NotificationIngress.enqueueRestore(any()) } returns Unit + every { NotificationIngress.persistDismiss(any(), any()) } returns Unit } afterAny { @@ -44,6 +49,7 @@ class PrewarmEntryPointTests : FunSpec({ // are owned by IOMockHelper and torn down in its afterSpec — unmockkAll() here would strip // them mid-spec and break the remaining tests. unmockkObject(OneSignal) + unmockkObject(NotificationIngress) } test("BootUpReceiver.onReceive prewarms before dispatch") { @@ -52,7 +58,7 @@ class PrewarmEntryPointTests : FunSpec({ verify(exactly = 1) { OneSignalDispatchers.prewarm() } verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>(), any<() -> Unit>()) + suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } } @@ -62,7 +68,7 @@ class PrewarmEntryPointTests : FunSpec({ verify(exactly = 1) { OneSignalDispatchers.prewarm() } verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>(), any<() -> Unit>()) + suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } } @@ -72,7 +78,7 @@ class PrewarmEntryPointTests : FunSpec({ verify(exactly = 1) { OneSignalDispatchers.prewarm() } verifyOrder { OneSignalDispatchers.prewarm() - suspendifyOnIO(any Unit>(), any<() -> Unit>()) + suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } } }) diff --git a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt index c41b5e6544..6fc3db07c7 100644 --- a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt +++ b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt @@ -2,6 +2,7 @@ package com.onesignal.mocks import com.onesignal.common.threading.OneSignalDispatchers import com.onesignal.common.threading.runOnSerialIO +import com.onesignal.common.threading.suspendifyOnIngress import com.onesignal.common.threading.suspendifyOnIO import com.onesignal.common.threading.suspendifyOnMain import com.onesignal.common.threading.suspendifyOnSerialIO @@ -149,6 +150,18 @@ object IOMockHelper : BeforeSpecListener, AfterSpecListener, BeforeTestListener, } } + every { suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } answers { + val block = firstArg Unit>() + val onComplete = secondArg<() -> Unit>() + trackAsyncWork { + try { + block() + } finally { + onComplete() + } + } + } + every { suspendifyOnSerialIO(any Unit>()) } answers { val block = firstArg Unit>() trackAsyncWork(block) @@ -187,6 +200,12 @@ object IOMockHelper : BeforeSpecListener, AfterSpecListener, BeforeTestListener, // Return a mock Job (launchOnDefault returns a Job) mockk(relaxed = true) } + + every { OneSignalDispatchers.launchOnIngress(any Unit>()) } answers { + val block = firstArg Unit>() + trackAsyncWork(block) + mockk(relaxed = true) + } } override suspend fun beforeTest(testCase: TestCase) { From 804cf2592ceb3a648e9e0b4c60c7c60aebace46a Mon Sep 17 00:00:00 2001 From: Fadi George Date: Fri, 7 Aug 2026 11:50:29 -0700 Subject: [PATCH 3/3] refactor(notifications): extract runIngressHandoff helper --- .../common/threading/OneSignalDispatchers.kt | 103 +++++++++--------- .../internal/ingress/NotificationIngress.kt | 22 ++-- .../notifications/receivers/BootUpReceiver.kt | 24 +--- .../receivers/BroadcastCompletion.kt | 12 ++ .../receivers/FCMBroadcastReceiver.kt | 34 ++---- .../receivers/NotificationDismissReceiver.kt | 17 +-- .../receivers/UpgradeReceiver.kt | 25 +---- 7 files changed, 101 insertions(+), 136 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt index 9836d7bf20..18771ca71e 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt @@ -96,6 +96,13 @@ object OneSignalDispatchers { } } + private data class LaneConfig( + val corePoolSize: Int, + val maxPoolSize: Int, + val threadName: String, + val priority: Int = Thread.NORM_PRIORITY, + ) + /** * A stable dispatcher that queues work until its backing executor has been built and * prestarted on the bootstrap thread. CoroutineScope.launch can therefore return its real Job @@ -309,73 +316,67 @@ object OneSignalDispatchers { val DefaultScope = CoroutineScope(SupervisorJob() + Default) val SerialIOScope = CoroutineScope(SupervisorJob() + SerialIO) val IngressScope = CoroutineScope(SupervisorJob() + Ingress) + private val gates = listOf(IO, Default, SerialIO, Ingress) + private val scopes = listOf(IOScope, DefaultScope, SerialIOScope, IngressScope) fun prewarm() { - IO.requestWarmup() - Default.requestWarmup() - SerialIO.requestWarmup() - Ingress.requestWarmup() + gates.forEach { it.requestWarmup() } } fun shutdown() { - IOScope.cancel() - DefaultScope.cancel() - SerialIOScope.cancel() - IngressScope.cancel() - IO.close() - Default.close() - SerialIO.close() - Ingress.close() + scopes.forEach { it.cancel() } + gates.forEach { it.close() } } } - @Suppress("LongMethod") private fun createTarget(lane: Lane): LaneTarget { beforeLaneCreateForTest?.invoke(lane.name) + val config = lane.config() val executor = - when (lane) { - Lane.IO -> - ThreadPoolExecutor( - IO_CORE_POOL_SIZE, - IO_MAX_POOL_SIZE, - KEEP_ALIVE_TIME_SECONDS, - TimeUnit.SECONDS, - LinkedBlockingQueue(QUEUE_CAPACITY), - OptimizedThreadFactory(IO_THREAD_NAME_PREFIX, Thread.NORM_PRIORITY - 1), - ) - Lane.DEFAULT -> - ThreadPoolExecutor( - DEFAULT_CORE_POOL_SIZE, - DEFAULT_MAX_POOL_SIZE, - KEEP_ALIVE_TIME_SECONDS, - TimeUnit.SECONDS, - LinkedBlockingQueue(QUEUE_CAPACITY), - OptimizedThreadFactory(DEFAULT_THREAD_NAME_PREFIX), - ) - Lane.SERIAL_IO -> - ThreadPoolExecutor( - 1, - 1, - KEEP_ALIVE_TIME_SECONDS, - TimeUnit.SECONDS, - LinkedBlockingQueue(QUEUE_CAPACITY), - OptimizedThreadFactory(SERIAL_IO_THREAD_NAME, Thread.NORM_PRIORITY - 1), - ) - Lane.INGRESS -> - ThreadPoolExecutor( - 1, - 1, - KEEP_ALIVE_TIME_SECONDS, - TimeUnit.SECONDS, - LinkedBlockingQueue(QUEUE_CAPACITY), - OptimizedThreadFactory(INGRESS_THREAD_NAME, Thread.NORM_PRIORITY - 1), - ) - } + ThreadPoolExecutor( + config.corePoolSize, + config.maxPoolSize, + KEEP_ALIVE_TIME_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(QUEUE_CAPACITY), + OptimizedThreadFactory(config.threadName, config.priority), + ) executor.allowCoreThreadTimeOut(false) executor.prestartAllCoreThreads() return LaneTarget(executor.asCoroutineDispatcher(), executor) } + private fun Lane.config(): LaneConfig = + when (this) { + Lane.IO -> + LaneConfig( + IO_CORE_POOL_SIZE, + IO_MAX_POOL_SIZE, + IO_THREAD_NAME_PREFIX, + Thread.NORM_PRIORITY - 1, + ) + Lane.DEFAULT -> + LaneConfig( + DEFAULT_CORE_POOL_SIZE, + DEFAULT_MAX_POOL_SIZE, + DEFAULT_THREAD_NAME_PREFIX, + ) + Lane.SERIAL_IO -> + LaneConfig( + 1, + 1, + SERIAL_IO_THREAD_NAME, + Thread.NORM_PRIORITY - 1, + ) + Lane.INGRESS -> + LaneConfig( + 1, + 1, + INGRESS_THREAD_NAME, + Thread.NORM_PRIORITY - 1, + ) + } + private fun createFallbackTarget(lane: Lane): LaneTarget { val dispatcher = when (lane) { diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/ingress/NotificationIngress.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/ingress/NotificationIngress.kt index f18336fcb2..b66fdc48c2 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/ingress/NotificationIngress.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/ingress/NotificationIngress.kt @@ -82,22 +82,22 @@ internal object NotificationIngress { .enqueueUniqueWork(DRAIN_WORK_NAME, ExistingWorkPolicy.KEEP, request) } - @Suppress("TooGenericExceptionCaught") private fun scheduleDrainBestEffort(context: Context) { if (drainSchedulerForTest != null) { - try { - scheduleDrain(context) - } catch (e: Exception) { - Logging.warn("Notification ingress persisted; drain scheduling will retry on next startup", e) - } + scheduleDrainSafely(context) return } OneSignalDispatchers.launchOnIO { - try { - scheduleDrain(context) - } catch (e: Exception) { - Logging.warn("Notification ingress persisted; drain scheduling will retry on next startup", e) - } + scheduleDrainSafely(context) + } + } + + @Suppress("TooGenericExceptionCaught") + private fun scheduleDrainSafely(context: Context) { + try { + scheduleDrain(context) + } catch (e: Exception) { + Logging.warn("Notification ingress persisted; drain scheduling will retry on next startup", e) } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt index 2382bc820d..4aa8b9a454 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt @@ -29,8 +29,6 @@ package com.onesignal.notifications.receivers import android.content.BroadcastReceiver import android.content.Context import android.content.Intent -import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIngress import com.onesignal.notifications.internal.ingress.NotificationIngress class BootUpReceiver : BroadcastReceiver() { @@ -38,22 +36,12 @@ class BootUpReceiver : BroadcastReceiver() { context: Context, intent: Intent, ) { - // Boot can cold-start the process before initWithContext. Warm dispatchers before - // goAsync() so the daemon has lead time before the first suspendifyOnIO dispatch. - OneSignalDispatchers.prewarm() - - val completion = - BroadcastCompletion( - "BootUpReceiver", - goAsync(), - BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, - ) // Persist the reconstructible restore request without waiting for full SDK initialization. - suspendifyOnIngress( - block = { - NotificationIngress.enqueueRestore(context) - }, - onComplete = { completion.finish() }, - ) + runIngressHandoff( + "BootUpReceiver", + BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, + ) { + NotificationIngress.enqueueRestore(context) + } } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt index 60287ac890..b1ab51b2ab 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt @@ -4,9 +4,21 @@ import android.content.BroadcastReceiver import android.os.Handler import android.os.Looper import android.os.SystemClock +import com.onesignal.common.threading.OneSignalDispatchers +import com.onesignal.common.threading.suspendifyOnIngress import com.onesignal.debug.internal.logging.Logging import java.util.concurrent.atomic.AtomicBoolean +internal fun BroadcastReceiver.runIngressHandoff( + receiverName: String, + timeoutMs: Long? = null, + block: suspend () -> Unit, +) { + OneSignalDispatchers.prewarm() + val completion = BroadcastCompletion(receiverName, goAsync(), timeoutMs) + suspendifyOnIngress(block = block, onComplete = { completion.finish() }) +} + internal class BroadcastCompletion( private val receiverName: String, private val pendingResult: BroadcastReceiver.PendingResult?, diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt index ab0968d40b..d8156e9445 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt @@ -4,8 +4,6 @@ import android.app.Activity import android.content.BroadcastReceiver import android.content.Context import android.content.Intent -import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIngress import com.onesignal.notifications.internal.ingress.NotificationIngress // This is the entry point when a FCM payload is received from the Google Play services app @@ -24,28 +22,18 @@ class FCMBroadcastReceiver : BroadcastReceiver() { return } - // FCM can cold-start the process before initWithContext. Warm dispatchers before goAsync() - // so the prewarm daemon gets a head start during the handoff, making the dispatchers more - // likely to be warm by the time the suspendifyOnIO below submits its work. - OneSignalDispatchers.prewarm() - - val completion = BroadcastCompletion("FCMBroadcastReceiver", goAsync()) - // process in background - suspendifyOnIngress( - block = { - if (!isFCMMessage(intent)) { - setSuccessfulResultCode() - return@suspendifyOnIngress - } + runIngressHandoff("FCMBroadcastReceiver") { + if (!isFCMMessage(intent)) { + setSuccessfulResultCode() + return@runIngressHandoff + } - if (NotificationIngress.persistFcm(context, intent, bundle)) { - setAbort() - } else { - setSuccessfulResultCode() - } - }, - onComplete = { completion.finish() }, - ) + if (NotificationIngress.persistFcm(context, intent, bundle)) { + setAbort() + } else { + setSuccessfulResultCode() + } + } } private fun setSuccessfulResultCode() { diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt index 86640e2b95..b79610f6d5 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt @@ -27,8 +27,6 @@ package com.onesignal.notifications.receivers import android.content.BroadcastReceiver import android.content.Context import android.content.Intent -import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIngress import com.onesignal.notifications.internal.ingress.NotificationIngress class NotificationDismissReceiver : BroadcastReceiver() { @@ -36,17 +34,8 @@ class NotificationDismissReceiver : BroadcastReceiver() { context: Context, intent: Intent, ) { - // A dismiss can cold-start the process before initWithContext. Warm dispatchers before - // goAsync() so the daemon has lead time before the first suspendifyOnIO dispatch. - OneSignalDispatchers.prewarm() - - val completion = BroadcastCompletion("NotificationDismissReceiver", goAsync()) - - suspendifyOnIngress( - block = { - NotificationIngress.persistDismiss(context, intent) - }, - onComplete = { completion.finish() }, - ) + runIngressHandoff("NotificationDismissReceiver") { + NotificationIngress.persistDismiss(context, intent) + } } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt index f150b128d1..9788beb5de 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt @@ -30,8 +30,6 @@ import android.content.BroadcastReceiver import android.content.Context import android.content.Intent import android.os.Build -import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.common.threading.suspendifyOnIngress import com.onesignal.notifications.internal.ingress.NotificationIngress class UpgradeReceiver : BroadcastReceiver() { @@ -47,23 +45,12 @@ class UpgradeReceiver : BroadcastReceiver() { return } - // App upgrade can cold-start the process before initWithContext. Warm dispatchers before - // goAsync() so the daemon has lead time before the first suspendifyOnIO dispatch. - OneSignalDispatchers.prewarm() - - val completion = - BroadcastCompletion( - "UpgradeReceiver", - goAsync(), - BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, - ) - // Persist the reconstructible restore request without waiting for full SDK initialization. - suspendifyOnIngress( - block = { - NotificationIngress.enqueueRestore(context) - }, - onComplete = { completion.finish() }, - ) + runIngressHandoff( + "UpgradeReceiver", + BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, + ) { + NotificationIngress.enqueueRestore(context) + } } }