From 75436fa9374367653d7adcf43435860b4c6f65e8 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Thu, 6 Aug 2026 15:14:04 -0700 Subject: [PATCH 1/4] 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/4] 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/4] 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) + } } } From 9692d9cb344a9bdba5c9120c4f41d60b581fd5fd Mon Sep 17 00:00:00 2001 From: Fadi George Date: Wed, 12 Aug 2026 12:48:36 -0700 Subject: [PATCH 4/4] fix(notifications): close durable ingress race conditions Ensure queued notification work cannot be stranded or blocked by poison records, and make job and dispatcher ownership resilient across Android callback and bootstrap races. Co-authored-by: Cursor --- .../common/threading/OneSignalDispatchers.kt | 62 ++++-- .../onesignal/core/services/SyncJobService.kt | 3 +- .../threading/OneSignalDispatchersTests.kt | 20 ++ .../core/services/SyncJobServiceTests.kt | 24 ++- .../internal/ingress/NotificationIngress.kt | 194 ++++++++++++++---- .../impl/NotificationRestoreWorkManager.kt | 4 + .../receivers/BroadcastCompletion.kt | 29 ++- .../receivers/FCMBroadcastReceiver.kt | 5 +- .../receivers/NotificationDismissReceiver.kt | 5 +- .../common/WorkManagerEnqueueTests.kt | 77 +++++++ .../ingress/NotificationIngressTests.kt | 153 +++++++++++++- .../receivers/BroadcastCompletionTests.kt | 16 +- .../receivers/FCMBroadcastReceiverTests.kt | 27 +++ .../java/com/onesignal/mocks/IOMockHelper.kt | 2 +- 14 files changed, 530 insertions(+), 91 deletions(-) create mode 100644 OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/common/WorkManagerEnqueueTests.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 18771ca71e..2041826c0e 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 @@ -58,6 +58,9 @@ object OneSignalDispatchers { @Volatile internal var beforeLaneCreateForTest: ((String) -> Unit)? = null + @Volatile + internal var beforeFallbackCreateForTest: ((String) -> Unit)? = null + private class OptimizedThreadFactory( private val namePrefix: String, private val priority: Int = Thread.NORM_PRIORITY, @@ -160,18 +163,29 @@ object OneSignalDispatchers { @Suppress("TooGenericExceptionCaught") fun initialize() { - val newTarget = + createTargetOrFallback()?.let(::drainPending) + } + + @Suppress("TooGenericExceptionCaught") + private fun createTargetOrFallback(): LaneTarget? = + try { + createTarget(lane) + } catch (e: Exception) { + Logging.warn("OneSignalDispatchers: Using fallback for $lane lane: ${e.message}", e) try { - createTarget(lane) - } catch (e: Exception) { - 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 + Logging.warn("OneSignalDispatchers: Fallback failed for $lane lane: ${t.message}", t) + failPending("OneSignal $lane dispatcher fallback failed") + null } + } catch (t: Throwable) { + Logging.warn("OneSignalDispatchers: Failed to initialize $lane lane: ${t.message}", t) + failPending("OneSignal $lane dispatcher failed to initialize") + null + } + private fun drainPending(newTarget: LaneTarget) { while (true) { val next = synchronized(lock) { @@ -278,17 +292,7 @@ object OneSignalDispatchers { 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 - } - }, + ::runLoop, "$BASE_THREAD_NAME-bootstrap", ).apply { isDaemon = true @@ -304,6 +308,24 @@ object OneSignalDispatchers { } } } + + @Suppress("TooGenericExceptionCaught") + private fun runLoop() { + try { + while (true) { + val gate = queue.poll() ?: return + try { + gate.initialize() + } catch (t: Throwable) { + Logging.warn("OneSignalDispatchers: Bootstrap failed for a lane: ${t.message}", t) + gate.bootstrapFailed() + } + } + } finally { + running.set(false) + if (queue.isNotEmpty()) startIfNeeded() + } + } } private class Pools { @@ -378,6 +400,7 @@ object OneSignalDispatchers { } private fun createFallbackTarget(lane: Lane): LaneTarget { + beforeFallbackCreateForTest?.invoke(lane.name) val dispatcher = when (lane) { Lane.IO -> Dispatchers.IO @@ -487,6 +510,7 @@ object OneSignalDispatchers { } resetPrewarmForTest() beforeLaneCreateForTest = null + beforeFallbackCreateForTest = null } @Suppress("ComplexMethod") @@ -506,7 +530,7 @@ object OneSignalDispatchers { - 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() + """.trimIndent() } internal fun getStatus(): String { 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 3b7e0eb241..21c3a86568 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 @@ -45,6 +45,7 @@ class SyncJobService : JobService() { } private class JobRun(val parameters: JobParameters) { + val jobId = parameters.jobId val state = AtomicReference(RunState.RUNNING) val job = AtomicReference() } @@ -105,7 +106,7 @@ class SyncJobService : JobService() { val run = activeRun.get() val stopped = run != null && - run.parameters === jobParameters && + run.jobId == jobParameters.jobId && run.state.compareAndSet(RunState.RUNNING, RunState.STOPPED) if (stopped) { activeRun.compareAndSet(run, null) 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 2e219cecc8..e97d98f1fe 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 @@ -59,6 +59,26 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.beforeLaneCreateForTest = null } + test("fallback initialization failure does not wedge subsequent lanes") { + OneSignalDispatchers.resetForTest() + val failedJobCompleted = CountDownLatch(1) + val defaultWorkRan = CountDownLatch(1) + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") throw IllegalStateException("primary failed") + } + OneSignalDispatchers.beforeFallbackCreateForTest = { lane -> + if (lane == "IO") throw AssertionError("fallback failed") + } + + OneSignalDispatchers.launchOnIO {}.invokeOnCompletion { failedJobCompleted.countDown() } + failedJobCompleted.await(1, TimeUnit.SECONDS) shouldBe true + + OneSignalDispatchers.beforeLaneCreateForTest = null + OneSignalDispatchers.beforeFallbackCreateForTest = null + OneSignalDispatchers.launchOnDefault { defaultWorkRan.countDown() } + defaultWorkRan.await(1, TimeUnit.SECONDS) shouldBe true + } + test("IO dispatcher should execute work on background thread") { val mainThreadId = Thread.currentThread().id var backgroundThreadId: Long? = null 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 26b1b6d6f8..aaca1db2f3 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 @@ -157,12 +157,15 @@ class SyncJobServiceTests : FunSpec({ verify { mockBackgroundManager.needsJobReschedule = false } } - test("onStopJob cancels the owned coroutine without resolving services") { + test("onStopJob matches distinct parameters by job id and cancels the owned coroutine") { val job = mockk(relaxed = true) + val stopParameters = mockk(relaxed = true) + every { mocks.jobParameters.jobId } returns 42 + every { stopParameters.jobId } returns 42 every { OneSignalDispatchers.launchOnIO(any Unit>()) } returns job mocks.syncJobService.onStartJob(mocks.jobParameters) - val result = mocks.syncJobService.onStopJob(mocks.jobParameters) + val result = mocks.syncJobService.onStopJob(stopParameters) result shouldBe true verify { job.cancel() } @@ -177,6 +180,23 @@ class SyncJobServiceTests : FunSpec({ mocks.syncJobService.onStopJob(mocks.jobParameters) shouldBe false } + test("onStopJob does not cancel a different job id") { + val job = mockk(relaxed = true) + val stopParameters = mockk(relaxed = true) + every { mocks.jobParameters.jobId } returns 42 + every { stopParameters.jobId } returns 43 + every { OneSignalDispatchers.launchOnIO(any Unit>()) } returns job + + mocks.syncJobService.onStartJob(mocks.jobParameters) + + mocks.syncJobService.onStopJob(stopParameters) shouldBe false + verify(exactly = 0) { job.cancel() } + every { OneSignalDispatchers.launchOnIO(any Unit>()) } answers { + runBlocking { firstArg Unit>().invoke() } + mockk(relaxed = true) + } + } + test("onStopJob does not reschedule a run that already completed") { coEvery { OneSignal.initWithContext(any()) } returns false mocks.syncJobService.onStartJob(mocks.jobParameters) 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 b66fdc48c2..2dc719e560 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 @@ -9,9 +9,9 @@ import android.os.Bundle import androidx.work.CoroutineWorker import androidx.work.ExistingWorkPolicy import androidx.work.OneTimeWorkRequest +import androidx.work.Operation 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 @@ -44,7 +44,7 @@ internal object NotificationIngress { createdAtMs = System.currentTimeMillis(), ) IngressStore.get(context).put(record) - scheduleDrainBestEffort(context) + scheduleDrainDurably(context) return true } @@ -65,7 +65,7 @@ internal object NotificationIngress { createdAtMs = System.currentTimeMillis(), ) IngressStore.get(context).put(record) - scheduleDrainBestEffort(context) + scheduleDrainDurably(context) } fun enqueueRestore(context: Context) { @@ -73,32 +73,21 @@ internal object NotificationIngress { } 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) + enqueueDrain(context) } - private fun scheduleDrainBestEffort(context: Context) { - if (drainSchedulerForTest != null) { - scheduleDrainSafely(context) - return - } - OneSignalDispatchers.launchOnIO { - scheduleDrainSafely(context) - } + private fun scheduleDrainDurably(context: Context) { + enqueueDrain(context)?.result?.get() } - @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) + private fun enqueueDrain(context: Context): Operation? { + drainSchedulerForTest?.let { + it(context) + return null } + val request = OneTimeWorkRequest.Builder(NotificationIngressDrainWorker::class.java).build() + return OSWorkManagerHelper.getInstance(context.applicationContext) + .enqueueUniqueWork(DRAIN_WORK_NAME, ExistingWorkPolicy.APPEND_OR_REPLACE, request) } internal fun pendingCountForTest(context: Context): Int = IngressStore.get(context).count() @@ -107,6 +96,21 @@ internal object NotificationIngress { drainSchedulerForTest = null IngressStore.get(context).clear() } + + internal fun putRawForTest( + context: Context, + id: String, + kind: String, + payload: String, + createdAtMs: Long = System.currentTimeMillis(), + ) { + IngressStore.get(context).putRaw(id, kind, payload, createdAtMs) + } + + internal fun attemptCountForTest( + context: Context, + id: String, + ): Int? = IngressStore.get(context).attemptCount(id) } internal class NotificationIngressDrainStarter( @@ -129,23 +133,66 @@ internal class NotificationIngressDrainWorker( @Suppress("TooGenericExceptionCaught") override suspend fun doWork(): Result { val store = IngressStore.get(applicationContext) - if (!OneSignal.initWithContext(applicationContext)) return Result.retry() + if (!OneSignal.initWithContext(applicationContext)) { + return if (runAttemptCount + 1 >= MAX_INIT_ATTEMPTS) Result.failure() else Result.retry() + } - return try { - for (record in store.list()) { - when (record.kind) { - IngressKind.FCM -> processFcm(record) - IngressKind.DISMISS -> processDismiss(record) - } + var retryNeeded = false + for (record in store.list()) { + retryNeeded = processRecord(store, record) || retryNeeded + } + return if (retryNeeded) Result.retry() else Result.success() + } + + @Suppress("TooGenericExceptionCaught") + private suspend fun processRecord( + store: IngressStore, + record: IngressRecord, + ): Boolean = + when { + record.kind == null -> { + Logging.warn("Dropping notification ingress ${record.id} with unknown kind") store.delete(record.id) + false } - Result.success() - } catch (e: Exception) { - Logging.error("Notification ingress drain failed", e) - Result.retry() + isExpired(record) -> { + Logging.warn("Dropping expired notification ingress ${record.id}") + store.delete(record.id) + false + } + else -> + try { + when (record.kind) { + IngressKind.FCM -> processFcm(record) + IngressKind.DISMISS -> processDismiss(record) + } + store.delete(record.id) + false + } catch (e: Exception) { + handleRecordFailure(store, record, e) + } + } + + private fun handleRecordFailure( + store: IngressStore, + record: IngressRecord, + error: Exception, + ): Boolean { + val nextAttempt = record.attemptCount + 1 + return if (nextAttempt >= MAX_RECORD_ATTEMPTS) { + Logging.error("Dropping notification ingress ${record.id} after $nextAttempt attempts", error) + store.delete(record.id) + false + } else { + Logging.warn("Notification ingress ${record.id} failed attempt $nextAttempt", error) + store.setAttemptCount(record.id, nextAttempt) + true } } + private fun isExpired(record: IngressRecord): Boolean = + System.currentTimeMillis() - record.createdAtMs >= MAX_RECORD_AGE_MS + private fun processFcm(record: IngressRecord) { val bundle = BundleCodec.decode(record.payload) OneSignal.getService() @@ -157,6 +204,12 @@ internal class NotificationIngressDrainWorker( OneSignal.getService() .processFromContext(applicationContext, intent) } + + companion object { + internal const val MAX_RECORD_ATTEMPTS = 3 + internal const val MAX_INIT_ATTEMPTS = 3 + internal const val MAX_RECORD_AGE_MS = 24 * 60 * 60 * 1_000L + } } private enum class IngressKind { @@ -166,10 +219,11 @@ private enum class IngressKind { private data class IngressRecord( val id: String, - val kind: IngressKind, + val kind: IngressKind?, val action: String?, val payload: String, val createdAtMs: Long, + val attemptCount: Int = 0, ) private class IngressStore private constructor(context: Context) : @@ -182,7 +236,8 @@ private class IngressStore private constructor(context: Context) : kind TEXT NOT NULL, action TEXT, payload TEXT NOT NULL, - created_at INTEGER NOT NULL + created_at INTEGER NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0 ) """.trimIndent(), ) @@ -193,21 +248,28 @@ private class IngressStore private constructor(context: Context) : oldVersion: Int, newVersion: Int, ) { - database.execSQL("DROP TABLE IF EXISTS $TABLE") - onCreate(database) + if (oldVersion < SCHEMA_WITH_ATTEMPTS_VERSION) { + database.execSQL("ALTER TABLE $TABLE ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0") + } } fun put(record: IngressRecord) { val values = ContentValues().apply { put("id", record.id) - put("kind", record.kind.name) + put("kind", requireNotNull(record.kind).name) put("action", record.action) put("payload", record.payload) put("created_at", record.createdAtMs) + put(ATTEMPT_COUNT_COLUMN, record.attemptCount) + } + val inserted = writableDatabase.insertWithOnConflict(TABLE, null, values, SQLiteDatabase.CONFLICT_IGNORE) + if (inserted == -1L) { + values.remove("created_at") + values.remove(ATTEMPT_COUNT_COLUMN) + check(writableDatabase.update(TABLE, values, "id = ?", arrayOf(record.id)) == 1) { + "Unable to update notification ingress" } - check(writableDatabase.insertWithOnConflict(TABLE, null, values, SQLiteDatabase.CONFLICT_REPLACE) != -1L) { - "Unable to persist notification ingress" } } @@ -215,7 +277,7 @@ private class IngressStore private constructor(context: Context) : val records = mutableListOf() readableDatabase.query( TABLE, - arrayOf("id", "kind", "action", "payload", "created_at"), + arrayOf("id", "kind", "action", "payload", "created_at", ATTEMPT_COUNT_COLUMN), null, null, null, @@ -226,10 +288,11 @@ private class IngressStore private constructor(context: Context) : records += IngressRecord( id = cursor.getString(ID_COLUMN_INDEX), - kind = IngressKind.valueOf(cursor.getString(KIND_COLUMN_INDEX)), + kind = enumValues().firstOrNull { it.name == cursor.getString(KIND_COLUMN_INDEX) }, action = cursor.getString(ACTION_COLUMN_INDEX), payload = cursor.getString(PAYLOAD_COLUMN_INDEX), createdAtMs = cursor.getLong(CREATED_AT_COLUMN_INDEX), + attemptCount = cursor.getInt(ATTEMPT_COUNT_COLUMN_INDEX), ) } } @@ -240,6 +303,44 @@ private class IngressStore private constructor(context: Context) : writableDatabase.delete(TABLE, "id = ?", arrayOf(id)) } + fun setAttemptCount( + id: String, + attemptCount: Int, + ) { + val values = ContentValues().apply { put(ATTEMPT_COUNT_COLUMN, attemptCount) } + writableDatabase.update(TABLE, values, "id = ?", arrayOf(id)) + } + + fun attemptCount(id: String): Int? = + readableDatabase.query( + TABLE, + arrayOf(ATTEMPT_COUNT_COLUMN), + "id = ?", + arrayOf(id), + null, + null, + null, + ).use { cursor -> + if (cursor.moveToFirst()) cursor.getInt(0) else null + } + + fun putRaw( + id: String, + kind: String, + payload: String, + createdAtMs: Long, + ) { + val values = + ContentValues().apply { + put("id", id) + put("kind", kind) + put("payload", payload) + put("created_at", createdAtMs) + put(ATTEMPT_COUNT_COLUMN, 0) + } + writableDatabase.insertWithOnConflict(TABLE, null, values, SQLiteDatabase.CONFLICT_REPLACE) + } + fun count(): Int = readableDatabase.rawQuery("SELECT COUNT(*) FROM $TABLE", null).use { cursor -> cursor.moveToFirst() @@ -252,13 +353,16 @@ private class IngressStore private constructor(context: Context) : companion object { private const val DATABASE_NAME = "OneSignalIngress.db" - private const val DATABASE_VERSION = 1 + private const val SCHEMA_WITH_ATTEMPTS_VERSION = 2 + private const val DATABASE_VERSION = SCHEMA_WITH_ATTEMPTS_VERSION private const val TABLE = "notification_ingress" + private const val ATTEMPT_COUNT_COLUMN = "attempt_count" 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 + private const val ATTEMPT_COUNT_COLUMN_INDEX = 5 @Volatile private var instance: IngressStore? = null 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 04f484e98e..487d73833e 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 @@ -66,5 +66,9 @@ internal class NotificationRestoreWorkManager : INotificationRestoreWorkManager private val NOTIFICATION_RESTORE_WORKER_IDENTIFIER = NotificationRestoreWorker::class.java.canonicalName ?: NotificationRestoreWorker::class.java.name private val restored = AtomicBoolean(false) + + internal fun resetForTest() { + restored.set(false) + } } } 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 b1ab51b2ab..5890fa4f85 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 @@ -1,12 +1,12 @@ package com.onesignal.notifications.receivers 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.ScheduledThreadPoolExecutor +import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean internal fun BroadcastReceiver.runIngressHandoff( @@ -26,16 +26,14 @@ internal class BroadcastCompletion( ) { 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) - } + private val timeoutTask = + timeoutMs?.let { + deadlineExecutor.schedule({ finish("deadline") }, it, TimeUnit.MILLISECONDS) + } fun finish(reason: String = "completed") { if (!finished.compareAndSet(false, true)) return - handler.removeCallbacks(timeout) + timeoutTask?.cancel(false) pendingResult?.finish() val durationMs = SystemClock.elapsedRealtime() - startedAtMs if (durationMs >= SOFT_DEADLINE_MS) { @@ -48,5 +46,18 @@ internal class BroadcastCompletion( companion object { const val RECONSTRUCTIBLE_WORK_TIMEOUT_MS = 8_000L private const val SOFT_DEADLINE_MS = 4_000L + private val deadlineExecutor = + ScheduledThreadPoolExecutor( + 1, + ) { runnable -> + Thread(runnable, "OS_BroadcastDeadline").apply { + isDaemon = true + priority = Thread.NORM_PRIORITY - 1 + } + }.apply { + removeOnCancelPolicy = true + setKeepAliveTime(30, TimeUnit.SECONDS) + allowCoreThreadTimeOut(true) + } } } 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 d8156e9445..e798328649 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 @@ -22,7 +22,10 @@ class FCMBroadcastReceiver : BroadcastReceiver() { return } - runIngressHandoff("FCMBroadcastReceiver") { + runIngressHandoff( + "FCMBroadcastReceiver", + BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, + ) { if (!isFCMMessage(intent)) { setSuccessfulResultCode() return@runIngressHandoff 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 b79610f6d5..f5b5c74dbe 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 @@ -34,7 +34,10 @@ class NotificationDismissReceiver : BroadcastReceiver() { context: Context, intent: Intent, ) { - runIngressHandoff("NotificationDismissReceiver") { + runIngressHandoff( + "NotificationDismissReceiver", + BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, + ) { NotificationIngress.persistDismiss(context, intent) } } diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/common/WorkManagerEnqueueTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/common/WorkManagerEnqueueTests.kt new file mode 100644 index 0000000000..766987d369 --- /dev/null +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/common/WorkManagerEnqueueTests.kt @@ -0,0 +1,77 @@ +package com.onesignal.notifications.internal.common + +import android.content.Context +import androidx.test.core.app.ApplicationProvider +import androidx.work.OneTimeWorkRequest +import androidx.work.WorkManager +import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest +import com.onesignal.notifications.internal.generation.impl.NotificationGenerationWorkManager +import com.onesignal.notifications.internal.restoration.impl.NotificationRestoreWorkManager +import io.kotest.assertions.throwables.shouldThrow +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.shouldBe +import io.mockk.every +import io.mockk.mockk +import io.mockk.mockkObject +import io.mockk.unmockkObject +import io.mockk.verify +import org.json.JSONObject + +@RobolectricTest +class WorkManagerEnqueueTests : FunSpec({ + lateinit var context: Context + lateinit var workManager: WorkManager + + beforeAny { + context = ApplicationProvider.getApplicationContext() + workManager = mockk(relaxed = true) + mockkObject(OSWorkManagerHelper) + every { OSWorkManagerHelper.getInstance(any()) } returns workManager + NotificationRestoreWorkManager.resetForTest() + } + + afterAny { + NotificationRestoreWorkManager.resetForTest() + unmockkObject(OSWorkManagerHelper) + } + + test("notification generation can be enqueued again after WorkManager rejects it") { + val manager = NotificationGenerationWorkManager() + val payload = JSONObject().put("custom", """{"i":"notification-id"}""") + every { + workManager.enqueueUniqueWork(any(), any(), any()) + } throws IllegalStateException("enqueue failed") + + shouldThrow { + manager.beginEnqueueingWork(context, "notification-id", 42, payload, 1L, false, false) + } + + every { + workManager.enqueueUniqueWork(any(), any(), any()) + } returns mockk(relaxed = true) + manager.beginEnqueueingWork(context, "notification-id", 42, payload, 1L, false, false) shouldBe true + verify(exactly = 2) { + workManager.enqueueUniqueWork(any(), any(), any()) + } + NotificationGenerationWorkManager.removeNotificationIdProcessed("notification-id") + } + + test("notification restore can be enqueued again after WorkManager rejects it") { + val manager = NotificationRestoreWorkManager() + every { + workManager.enqueueUniqueWork(any(), any(), any()) + } throws IllegalStateException("enqueue failed") + + shouldThrow { + manager.beginEnqueueingWork(context, shouldDelay = true) + } + + every { + workManager.enqueueUniqueWork(any(), any(), any()) + } returns mockk(relaxed = true) + manager.beginEnqueueingWork(context, shouldDelay = false) + verify(exactly = 2) { + workManager.enqueueUniqueWork(any(), any(), any()) + } + } +}) 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 index cf8438cc90..281a96c6ee 100644 --- 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 @@ -4,17 +4,45 @@ import android.content.Context import android.content.Intent import android.os.Bundle import androidx.test.core.app.ApplicationProvider +import androidx.work.ListenableWorker +import androidx.work.OneTimeWorkRequest +import androidx.work.Operation +import androidx.work.WorkManager +import androidx.work.WorkerParameters import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest +import com.google.common.util.concurrent.SettableFuture +import com.onesignal.OneSignal +import com.onesignal.notifications.internal.bundle.INotificationBundleProcessor +import com.onesignal.notifications.internal.common.OSWorkManagerHelper +import io.kotest.assertions.throwables.shouldThrow import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.shouldBe +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import io.mockk.mockkObject +import io.mockk.unmockkObject +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference @RobolectricTest class NotificationIngressTests : FunSpec({ lateinit var context: Context + lateinit var bundleProcessor: INotificationBundleProcessor beforeAny { context = ApplicationProvider.getApplicationContext() NotificationIngress.resetForTest(context) + bundleProcessor = mockk(relaxed = true) + mockkObject(OneSignal) + coEvery { OneSignal.initWithContext(any()) } returns true + every { OneSignal.getService() } returns bundleProcessor + } + + afterAny { + unmockkObject(OneSignal) + unmockkObject(OSWorkManagerHelper) } test("FCM input remains durable when drain scheduling fails") { @@ -25,11 +53,13 @@ class NotificationIngressTests : FunSpec({ putString("alert", "message") } - NotificationIngress.persistFcm( - context, - Intent("com.google.android.c2dm.intent.RECEIVE"), - bundle, - ) shouldBe true + shouldThrow { + NotificationIngress.persistFcm( + context, + Intent("com.google.android.c2dm.intent.RECEIVE"), + bundle, + ) + } NotificationIngress.pendingCountForTest(context) shouldBe 1 } @@ -60,4 +90,117 @@ class NotificationIngressTests : FunSpec({ countAtSchedule shouldBe 1 } + + test("FCM handoff waits for WorkManager to persist the drain") { + val workManager = mockk() + val operation = mockk() + val operationResult = SettableFuture.create() + val enqueueCalled = CountDownLatch(1) + val completed = CountDownLatch(1) + val failure = AtomicReference() + mockkObject(OSWorkManagerHelper) + every { OSWorkManagerHelper.getInstance(any()) } returns workManager + every { + workManager.enqueueUniqueWork(any(), any(), any()) + } answers { + enqueueCalled.countDown() + operation + } + every { operation.result } returns operationResult + val bundle = Bundle().apply { putString("custom", """{"i":"notification-id"}""") } + + Thread { + try { + NotificationIngress.persistFcm(context, Intent(), bundle) + } catch (error: Throwable) { + failure.set(error) + } finally { + completed.countDown() + } + }.start() + + enqueueCalled.await(1, TimeUnit.SECONDS) shouldBe true + completed.await(50, TimeUnit.MILLISECONDS) shouldBe false + operationResult.set(Operation.SUCCESS) + completed.await(1, TimeUnit.SECONDS) shouldBe true + failure.get() shouldBe null + } + + test("unknown record kind is discarded without blocking the drain") { + NotificationIngress.putRawForTest(context, "unknown", "UNKNOWN", "{}") + + val result = NotificationIngressDrainWorker(context, mockk(relaxed = true)).doWork() + + result.javaClass shouldBe ListenableWorker.Result.success().javaClass + NotificationIngress.pendingCountForTest(context) shouldBe 0 + } + + test("failed record is retried while later records continue") { + NotificationIngress.drainSchedulerForTest = {} + val badBundle = Bundle().apply { putString("custom", """{"i":"bad-id"}""") } + val goodBundle = Bundle().apply { putString("custom", """{"i":"good-id"}""") } + NotificationIngress.persistFcm(context, Intent(), badBundle) + NotificationIngress.persistFcm(context, Intent(), goodBundle) + every { bundleProcessor.processBundleFromReceiver(any(), any()) } answers { + if (secondArg().getString("custom")!!.contains("bad-id")) { + throw IllegalStateException("bad payload") + } + null + } + + val result = NotificationIngressDrainWorker(context, mockk(relaxed = true)).doWork() + + result.javaClass shouldBe ListenableWorker.Result.retry().javaClass + NotificationIngress.pendingCountForTest(context) shouldBe 1 + NotificationIngress.attemptCountForTest(context, "fcm:bad-id") shouldBe 1 + } + + test("record is dropped after the bounded retry limit") { + NotificationIngress.drainSchedulerForTest = {} + val bundle = Bundle().apply { putString("custom", """{"i":"bad-id"}""") } + NotificationIngress.persistFcm(context, Intent(), bundle) + every { bundleProcessor.processBundleFromReceiver(any(), any()) } throws IllegalStateException("bad payload") + val workerParameters = mockk(relaxed = true) + val worker = NotificationIngressDrainWorker(context, workerParameters) + + repeat(NotificationIngressDrainWorker.MAX_RECORD_ATTEMPTS) { worker.doWork() } + + NotificationIngress.pendingCountForTest(context) shouldBe 0 + } + + test("duplicate input does not reset the record attempt count") { + NotificationIngress.drainSchedulerForTest = {} + val bundle = Bundle().apply { putString("custom", """{"i":"bad-id"}""") } + NotificationIngress.persistFcm(context, Intent(), bundle) + every { bundleProcessor.processBundleFromReceiver(any(), any()) } throws IllegalStateException("bad payload") + NotificationIngressDrainWorker(context, mockk(relaxed = true)).doWork() + + NotificationIngress.persistFcm(context, Intent(), bundle) + + NotificationIngress.attemptCountForTest(context, "fcm:bad-id") shouldBe 1 + } + + test("expired record is discarded without processing") { + NotificationIngress.putRawForTest( + context, + id = "expired", + kind = "FCM", + payload = "{}", + createdAtMs = System.currentTimeMillis() - NotificationIngressDrainWorker.MAX_RECORD_AGE_MS, + ) + + NotificationIngressDrainWorker(context, mockk(relaxed = true)).doWork() + + NotificationIngress.pendingCountForTest(context) shouldBe 0 + } + + test("initialization failure stops retrying after the bounded limit") { + coEvery { OneSignal.initWithContext(any()) } returns false + val workerParameters = mockk(relaxed = true) + every { workerParameters.runAttemptCount } returns NotificationIngressDrainWorker.MAX_INIT_ATTEMPTS - 1 + + val result = NotificationIngressDrainWorker(context, workerParameters).doWork() + + result.javaClass shouldBe ListenableWorker.Result.failure().javaClass + } }) 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 index efa562e7ee..998c90f9ea 100644 --- 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 @@ -1,13 +1,12 @@ 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.kotest.matchers.shouldBe +import io.mockk.every import io.mockk.mockk import io.mockk.verify -import org.robolectric.Shadows.shadowOf -import java.time.Duration @RobolectricTest class BroadcastCompletionTests : FunSpec({ @@ -23,10 +22,13 @@ class BroadcastCompletionTests : FunSpec({ test("reconstructible work finishes at its deadline") { val pendingResult = mockk(relaxed = true) - BroadcastCompletion("test", pendingResult, 100) + var finishThread = "" + every { pendingResult.finish() } answers { + finishThread = Thread.currentThread().name + } + BroadcastCompletion("test", pendingResult, 50) - shadowOf(Looper.getMainLooper()).idleFor(Duration.ofMillis(100)) - - verify(exactly = 1) { pendingResult.finish() } + verify(exactly = 1, timeout = 1_000) { pendingResult.finish() } + finishThread shouldBe "OS_BroadcastDeadline" } }) 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 ef37b5643b..08ae32ce41 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 @@ -73,4 +73,31 @@ class FCMBroadcastReceiverTests : FunSpec({ verify(exactly = 0) { OneSignalDispatchers.prewarm() } } + + test("FCMBroadcastReceiver does not claim payloads rejected by durable ingress") { + val context = ApplicationProvider.getApplicationContext() + val intent = + Intent("com.google.android.c2dm.intent.RECEIVE").apply { + putExtra("from", "sender") + putExtra("message_type", "gcm") + } + every { NotificationIngress.persistFcm(any(), any(), any()) } returns false + + FCMBroadcastReceiver().onReceive(context, intent) + + verify(exactly = 1) { NotificationIngress.persistFcm(context, intent, any()) } + } + + test("FCMBroadcastReceiver ignores non-GCM message types") { + val context = ApplicationProvider.getApplicationContext() + val intent = + Intent("com.google.android.c2dm.intent.RECEIVE").apply { + putExtra("from", "sender") + putExtra("message_type", "deleted_messages") + } + + FCMBroadcastReceiver().onReceive(context, intent) + + verify(exactly = 0) { NotificationIngress.persistFcm(any(), any(), any()) } + } }) 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 6fc3db07c7..e0111ba93c 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,8 +2,8 @@ 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.suspendifyOnIngress import com.onesignal.common.threading.suspendifyOnMain import com.onesignal.common.threading.suspendifyOnSerialIO import io.kotest.core.listeners.AfterSpecListener