fix: [SDK-4995] harden background notification ingress against ANRs

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.
@@ -110,7 +134,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 +146,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)
+            }
         }
     }
 }

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt

@@ -30,11 +30,28 @@ 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 jobId = parameters.jobId
+        val state = AtomicReference(RunState.RUNNING)
+        val job = AtomicReference<Job?>()
+    }
+
+    private val activeRun = AtomicReference<JobRun?>()
+
     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 +62,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<IBackgroundManager>()
-                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<IBackgroundManager>()
-            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.jobId == jobParameters.jobId &&
+                run.state.compareAndSet(RunState.RUNNING, RunState.STOPPED)
+        if (stopped) {
+            activeRun.compareAndSet(run, null)
+            run?.job?.get()?.cancel()
         }
+        return stopped
     }
 }

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,59 @@ 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("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
@@ -77,15 +132,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,59 +342,25 @@ 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"
         status shouldContain "SerialIO Executor: Active"
         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
     }
 })

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({
@@ -139,6 +140,45 @@ 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("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