diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt
index 8118671035..614903b8f2 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt
@@ -16,9 +16,11 @@ import com.onesignal.core.internal.config.impl.FeatureFlagsRefreshService
 import com.onesignal.core.internal.config.impl.IdentityVerificationService
 import com.onesignal.core.internal.database.IDatabaseProvider
 import com.onesignal.core.internal.database.impl.DatabaseProvider
+import com.onesignal.core.internal.device.IConnectivityMonitor
 import com.onesignal.core.internal.device.IDeviceService
 import com.onesignal.core.internal.device.IFidEnv
 import com.onesignal.core.internal.device.IInstallIdService
+import com.onesignal.core.internal.device.impl.ConnectivityMonitor
 import com.onesignal.core.internal.device.impl.DeviceService
 import com.onesignal.core.internal.device.impl.FidEnvService
 import com.onesignal.core.internal.device.impl.InstallIdService
@@ -34,6 +36,7 @@ import com.onesignal.core.internal.language.impl.LanguageContext
 import com.onesignal.core.internal.operations.IOperationRepo
 import com.onesignal.core.internal.operations.impl.OperationModelStore
 import com.onesignal.core.internal.operations.impl.OperationRepo
+import com.onesignal.core.internal.operations.impl.OperationRetryTrigger
 import com.onesignal.core.internal.permissions.IRequestPermissionService
 import com.onesignal.core.internal.permissions.impl.RequestPermissionService
 import com.onesignal.core.internal.preferences.IPreferencesService
@@ -69,6 +72,7 @@ internal class CoreModule : IModule {
             .provides<IApplicationService>()
         builder.register<DeviceService>().provides<IDeviceService>()
         builder.register<FidEnvService>().provides<IFidEnv>()
+        builder.register<ConnectivityMonitor>().provides<IConnectivityMonitor>()
         builder.register<Time>().provides<ITime>()
         builder.register<DatabaseProvider>().provides<IDatabaseProvider>()
         builder.register<InstallIdService>().provides<IInstallIdService>()
@@ -91,6 +95,7 @@ internal class CoreModule : IModule {
         builder.register<OperationRepo>()
             .provides<IOperationRepo>()
             .provides<IStartableService>()
+        builder.register<OperationRetryTrigger>().provides<IStartableService>()
 
         // Permissions
         builder.register<RequestPermissionService>()
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt
index 959b29f5a1..0eb407906d 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/ConfigModel.kt
@@ -180,15 +180,31 @@ class ConfigModel : Model() {
         }
 
     /**
-     * The number of milliseconds times the number of times FAIL_RETRY
-     * is returned from an executor for a specific operation. AKA this
-     * backoff will increase each time we retry a specific operation
-     * by this value.
+     * The first retry backoff after FAIL_RETRY, in milliseconds. It doubles with each further
+     * failure of the same operation, up to [opRepoMaxFailRetryBackoff], and is then jittered.
      */
-    var opRepoDefaultFailRetryBackoff: Long
-        get() = getLongProperty(::opRepoDefaultFailRetryBackoff.name) { 15_000 }
+    var opRepoBaseFailRetryBackoff: Long
+        get() = getLongProperty(::opRepoBaseFailRetryBackoff.name) { 5_000 }
         set(value) {
-            setLongProperty(::opRepoDefaultFailRetryBackoff.name, value)
+            setLongProperty(::opRepoBaseFailRetryBackoff.name, value)
+        }
+
+    /**
+     * The longest retry backoff, in milliseconds, before jitter.
+     */
+    var opRepoMaxFailRetryBackoff: Long
+        get() = getLongProperty(::opRepoMaxFailRetryBackoff.name) { 300_000 }
+        set(value) {
+            setLongProperty(::opRepoMaxFailRetryBackoff.name, value)
+        }
+
+    /**
+     * The longest a server Retry-After is honored, in milliseconds.
+     */
+    var maxRetryAfter: Long
+        get() = getLongProperty(::maxRetryAfter.name) { 3_600_000 }
+        set(value) {
+            setLongProperty(::maxRetryAfter.name, value)
         }
 
     /**
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/device/IConnectivityMonitor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/device/IConnectivityMonitor.kt
new file mode 100644
index 0000000000..86d535b030
--- /dev/null
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/device/IConnectivityMonitor.kt
@@ -0,0 +1,9 @@
+package com.onesignal.core.internal.device
+
+internal interface IConnectivityMonitor {
+    /**
+     * Calls [onOnline] each time the device goes from having no internet-capable network to
+     * having one. Not called for the state at registration.
+     */
+    fun addOnlineHandler(onOnline: () -> Unit)
+}
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/device/impl/ConnectivityMonitor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/device/impl/ConnectivityMonitor.kt
new file mode 100644
index 0000000000..a48bf27605
--- /dev/null
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/device/impl/ConnectivityMonitor.kt
@@ -0,0 +1,61 @@
+package com.onesignal.core.internal.device.impl
+
+import android.content.Context
+import android.net.ConnectivityManager
+import android.net.Network
+import android.net.NetworkCapabilities
+import android.net.NetworkRequest
+import com.onesignal.core.internal.application.IApplicationService
+import com.onesignal.core.internal.device.IConnectivityMonitor
+import com.onesignal.debug.internal.logging.Logging
+
+internal class ConnectivityMonitor(
+    private val applicationService: IApplicationService,
+) : IConnectivityMonitor {
+    @Suppress("TooGenericExceptionCaught")
+    override fun addOnlineHandler(onOnline: () -> Unit) {
+        try {
+            val connectivityManager =
+                applicationService.appContext.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
+                    ?: return
+            val request =
+                NetworkRequest.Builder()
+                    .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
+                    .build()
+            connectivityManager.registerNetworkCallback(request, OnlineCallback(isOnline(connectivityManager), onOnline))
+        } catch (t: Throwable) {
+            // Can throw, e.g. when the app already holds the maximum number of network callbacks.
+            Logging.warn("ConnectivityMonitor: unable to register network callback", t)
+        }
+    }
+
+    @Suppress("DEPRECATION")
+    private fun isOnline(connectivityManager: ConnectivityManager) = connectivityManager.activeNetworkInfo?.isConnected == true
+
+    // onAvailable also fires for networks already up at registration. Seeding with the current
+    // state keeps those from counting as coming online.
+    private class OnlineCallback(
+        private var online: Boolean,
+        private val onOnline: () -> Unit,
+    ) : ConnectivityManager.NetworkCallback() {
+        private val networks = mutableSetOf<Network>()
+
+        override fun onAvailable(network: Network) {
+            val cameOnline =
+                synchronized(this) {
+                    networks.add(network)
+                    val wasOnline = online
+                    online = true
+                    !wasOnline
+                }
+            if (cameOnline) onOnline()
+        }
+
+        override fun onLost(network: Network) {
+            synchronized(this) {
+                networks.remove(network)
+                if (networks.isEmpty()) online = false
+            }
+        }
+    }
+}
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt
index e1dd89e5bc..ffbc73172a 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/http/impl/HttpClient.kt
@@ -26,7 +26,10 @@ import java.net.HttpURLConnection
 import java.net.URL
 import java.net.UnknownHostException
 import java.util.Scanner
+import java.util.concurrent.TimeUnit
 import javax.net.ssl.HttpsURLConnection
+import kotlin.math.ceil
+import kotlin.math.min
 
 internal const val HTTP_SDK_VERSION_HEADER_KEY = "SDK-Version"
 internal val HTTP_SDK_VERSION_HEADER_VALUE = "onesignal/android/${OneSignalUtils.sdkVersion}"
@@ -40,9 +43,10 @@ internal class HttpClient(
     private val _fidEnv: IFidEnv,
 ) : IHttpClient {
     /**
-     * Delay making network requests until we reach this time.
+     * Delay making network requests until we reach this time, on [ITime.elapsedRealtimeMillis].
      * Used when the OneSignal backend returns a Retry-After value.
      */
+    @Volatile
     private var delayNewRequestsUntil = 0L
 
     override suspend fun post(
@@ -88,7 +92,7 @@ internal class HttpClient(
             return HttpResponse(0, null, null)
         }
 
-        val delayUntil = delayNewRequestsUntil - _time.currentTimeMillis
+        val delayUntil = delayNewRequestsUntil - _time.elapsedRealtimeMillis
         if (delayUntil > 0) delay(delayUntil)
 
         try {
@@ -218,7 +222,8 @@ internal class HttpClient(
 
                     val retryAfter = retryAfterFromResponse(con)
                     val retryLimit = retryLimitFromResponse(con)
-                    val newDelayUntil = _time.currentTimeMillis + (retryAfter ?: 0) * 1_000
+                    val retryAfterMs = min(TimeUnit.SECONDS.toMillis((retryAfter ?: 0).toLong()), _configModelStore.model.maxRetryAfter)
+                    val newDelayUntil = _time.elapsedRealtimeMillis + retryAfterMs
                     if (newDelayUntil > delayNewRequestsUntil) delayNewRequestsUntil = newDelayUntil
 
                     when (httpResponse) {
@@ -316,7 +321,9 @@ internal class HttpClient(
         val retryAfterStr = con.getHeaderField("Retry-After")
         return if (retryAfterStr != null) {
             Logging.debug("HttpClient: Response Retry-After: $retryAfterStr")
-            retryAfterStr.toIntOrNull() ?: _configModelStore.model.httpRetryAfterParseFailFallback
+            val maxSeconds = TimeUnit.MILLISECONDS.toSeconds(_configModelStore.model.maxRetryAfter).toDouble()
+            retryAfterStr.trim().toDoubleOrNull()?.takeIf { it.isFinite() && it >= 0 }?.let { ceil(min(it, maxSeconds)).toInt() }
+                ?: _configModelStore.model.httpRetryAfterParseFailFallback
         } else if (con.responseCode == 429) {
             _configModelStore.model.httpRetryAfterParseFailFallback
         } else {
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt
index 37e64cdcc3..46b30a452a 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt
@@ -27,7 +27,10 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.newSingleThreadContext
 import kotlinx.coroutines.withTimeoutOrNull
 import java.util.UUID
+import java.util.concurrent.TimeUnit
 import kotlin.math.max
+import kotlin.math.min
+import kotlin.random.Random
 import kotlin.reflect.KClass
 
 internal class OperationRepo(
@@ -46,6 +49,10 @@ internal class OperationRepo(
         val bucket: Int,
         var retries: Int = 0,
     ) {
+        // On ITime.elapsedRealtimeMillis. A force may end it early: HttpClient still holds the
+        // request until any Retry-After passes.
+        var retryAt: Long = 0
+
         val waiters = mutableListOf<WaiterWithValue<OperationWaitResult>>()
 
         @Volatile
@@ -78,7 +85,7 @@ internal class OperationRepo(
         }
 
         override fun toString(): String {
-            return "bucket:$bucket, retries:$retries, operation:$operation\n"
+            return "bucket:$bucket, retries:$retries, retryAt:$retryAt, operation:$operation\n"
         }
     }
 
@@ -99,8 +106,13 @@ internal class OperationRepo(
     internal val queue = mutableListOf<OperationQueueItem>()
     private val inFlightLogins = mutableMapOf<String, OperationQueueItem>()
     private val waiter = WaiterWithValue<LoopWaiterMessage>()
-    private val retryWaiter = WaiterWithValue<LoopWaiterMessage>()
     private var paused = false
+
+    // Guarded by queue. Set by a force that lands while ops are out of the queue executing.
+    private var forcedDuringExecute = false
+
+    internal var jitter: () -> Double = { Random.nextDouble(MIN_JITTER, 1.0) }
+
     private val initialized = CompletableDeferred<Unit>()
 
     override suspend fun awaitInitialized() {
@@ -328,8 +340,16 @@ internal class OperationRepo(
     }
 
     override fun forceExecuteOperations() {
-        retryWaiter.wake(LoopWaiterMessage(true))
-        waiter.wake(LoopWaiterMessage(false))
+        val now = _time.elapsedRealtimeMillis
+        val endedWait =
+            synchronized(queue) {
+                forcedDuringExecute = true
+                val backedOff = queue.filter { it.retryAt > now }
+                backedOff.forEach { it.retryAt = 0 }
+                backedOff.isNotEmpty()
+            }
+        // Skip the batching wait only when a retry was actually held back.
+        waiter.wake(LoopWaiterMessage(endedWait))
     }
 
     /**
@@ -393,8 +413,14 @@ internal class OperationRepo(
      *          with a budget.
      */
     private suspend fun waitForNewOperationAndExecutionInterval() {
-        // 1. Wait for an operation to be enqueued
-        var wakeMessage = waiter.waitForWake()
+        // 1. Wait for an operation to be enqueued, or for a backed-off one to become ready
+        val untilNextRetry = msUntilNextRetry()
+        var wakeMessage =
+            if (untilNextRetry == null) {
+                waiter.waitForWake()
+            } else {
+                withTimeoutOrNull(untilNextRetry) { waiter.waitForWake() } ?: LoopWaiterMessage(true)
+            }
 
         // 2. Now wait opRepoExecutionInterval, restart the wait
         //    time everytime something new is enqueued, to ensure
@@ -432,7 +458,6 @@ internal class OperationRepo(
                 response.idTranslations.values.forEach { _newRecordState.add(it) }
             }
 
-            var highestRetries = 0
             when (response.result) {
                 ExecutionResult.SUCCESS -> {
                     // on success we remove the operation from the store and wake any waiters
@@ -465,10 +490,10 @@ internal class OperationRepo(
                     val followers = ops.filter { it != startingOp }
                     if (startingOp.operation is LoginUserOperation && followers.isNotEmpty()) {
                         _operationModelStore.remove(startingOp.operation.id)
-                        startingOp.wakeWaiters(response.toWaitResult(false))
                         synchronized(queue) {
                             followers.reversed().forEach { queue.add(0, it) }
                         }
+                        startingOp.wakeWaiters(response.toWaitResult(false))
                     } else {
                         dropAndWake(ops, response)
                     }
@@ -484,12 +509,22 @@ internal class OperationRepo(
                 }
                 ExecutionResult.FAIL_RETRY -> {
                     Logging.info("Operation execution failed, retrying: $operations")
-                    // add back all operations to the front of the queue to be re-executed.
+                    // Re-queue at the front with a wait on these records only. Other records keep
+                    // going, though HttpClient still holds every request during a Retry-After.
+                    val now = _time.elapsedRealtimeMillis
+                    val config = _configModelStore.model
+                    val retryAfterMs = min(TimeUnit.SECONDS.toMillis((response.retryAfterSeconds ?: 0).toLong()), config.maxRetryAfter)
+                    ops.forEach { it.retries++ }
+                    // One wait for the whole batch so it retries as a single request.
+                    val retries = ops.maxOf { it.retries }
+                    val exponential = config.opRepoBaseFailRetryBackoff shl min(retries - 1, MAX_BACKOFF_DOUBLINGS)
+                    val backoffMs = (min(exponential, config.opRepoMaxFailRetryBackoff) * jitter()).toLong()
                     synchronized(queue) {
+                        // A force that arrived mid-request would otherwise be lost to the new backoff.
+                        val delayMs = if (forcedDuringExecute) 0L else max(backoffMs, retryAfterMs)
+                        Logging.debug("OperationRepo: retry #$retries of $operations in $delayMs ms (backoff: $backoffMs ms, Retry-After: $retryAfterMs ms)")
                         ops.reversed().forEach {
-                            if (++it.retries > highestRetries) {
-                                highestRetries = it.retries
-                            }
+                            it.retryAt = if (delayMs > 0) now + delayMs else 0
                             queue.add(0, it)
                         }
                     }
@@ -524,8 +559,6 @@ internal class OperationRepo(
                 }
             }
 
-            // wait for retry and post create waiters to start next operation
-            delayBeforeNextExecution(highestRetries, response.retryAfterSeconds)
             if (response.idTranslations != null) {
                 delayForPostCreate(_configModelStore.model.opRepoPostCreateDelay)
             }
@@ -551,25 +584,6 @@ internal class OperationRepo(
         ops.forEach { it.wakeWaiters(response.toWaitResult(false)) }
     }
 
-    /**
-     * Wait which ever is longer, retryAfterSeconds returned by the server,
-     * or based on the retry count.
-     */
-    suspend fun delayBeforeNextExecution(
-        retries: Int,
-        retryAfterSeconds: Int?,
-    ) {
-        Logging.debug("retryAfterSeconds: $retryAfterSeconds")
-        val retryAfterSecondsNonNull = retryAfterSeconds?.toLong() ?: 0L
-        val delayForOnRetries = retries * _configModelStore.model.opRepoDefaultFailRetryBackoff
-        val delayFor = max(delayForOnRetries, retryAfterSecondsNonNull * 1_000)
-        if (delayFor < 1) return
-        Logging.debug("Operations being delay for: $delayFor ms")
-        withTimeoutOrNull(delayFor) {
-            retryWaiter.waitForWake()
-        }
-    }
-
     /**
      * Stall processing the queue so the backend's DB has to time
      * reflect the change before we do any other operations to it.
@@ -601,10 +615,18 @@ internal class OperationRepo(
         // Snapshot gate state once per pass so all queue items see the same IV view.
         val newCodePathsRun = _identityVerificationService.newCodePathsRun
         val ivBehaviorActive = _identityVerificationService.ivBehaviorActive
+        val now = _time.elapsedRealtimeMillis
         return synchronized(queue) {
+            // An op must not overtake an earlier op on the same record that is waiting to retry.
+            val waitingRecords = mutableSetOf<String>()
             val startingOp =
                 queue.firstOrNull {
-                    it.operation.canStartExecute &&
+                    if (it.retryAt > now) {
+                        waitingRecords.add(it.operation.applyToRecordId)
+                        return@firstOrNull false
+                    }
+                    it.operation.applyToRecordId !in waitingRecords &&
+                        it.operation.canStartExecute &&
                         _newRecordState.canAccess(it.operation.applyToRecordId) &&
                         it.bucket <= bucketFilter &&
                         // Outer gate: skip IV JWT check entirely on old code path.
@@ -613,7 +635,8 @@ internal class OperationRepo(
 
             if (startingOp != null) {
                 queue.remove(startingOp)
-                val grouped = getGroupableOperations(startingOp)
+                forcedDuringExecute = false
+                val grouped = getGroupableOperations(startingOp, now)
                 grouped.forEach { item ->
                     val login = item.operation as? LoginUserOperation ?: return@forEach
                     inFlightLogins[login.onesignalId] = item
@@ -632,7 +655,10 @@ internal class OperationRepo(
      *
      * THIS SHOULD BE CALLED WHILE THE QUEUE IS SYNCHRONIZED!!
      */
-    private fun getGroupableOperations(startingOp: OperationQueueItem): List<OperationQueueItem> {
+    private fun getGroupableOperations(
+        startingOp: OperationQueueItem,
+        now: Long,
+    ): List<OperationQueueItem> {
         val ops = mutableListOf(startingOp)
 
         if (startingOp.operation.groupComparisonType == GroupComparisonType.NONE) {
@@ -646,7 +672,13 @@ internal class OperationRepo(
                 startingOp.operation.modifyComparisonKey
             }
 
+        val waitingRecords = mutableSetOf<String>()
         for (item in queue.toList()) {
+            if (item.retryAt > now) {
+                waitingRecords.add(item.operation.applyToRecordId)
+                continue
+            }
+
             val itemKey =
                 if (startingOp.operation.groupComparisonType == GroupComparisonType.CREATE) {
                     item.operation.createComparisonKey
@@ -658,11 +690,10 @@ internal class OperationRepo(
                 throw Exception("Both comparison keys can not be blank!")
             }
 
-            if (!_newRecordState.canAccess(item.operation.applyToRecordId)) {
-                continue
-            }
-
-            if (itemKey == startingKey) {
+            if (itemKey == startingKey &&
+                _newRecordState.canAccess(item.operation.applyToRecordId) &&
+                item.operation.applyToRecordId !in waitingRecords
+            ) {
                 queue.remove(item)
                 ops.add(item)
             }
@@ -671,6 +702,13 @@ internal class OperationRepo(
         return ops
     }
 
+    private fun msUntilNextRetry(): Long? {
+        val now = _time.elapsedRealtimeMillis
+        return synchronized(queue) {
+            queue.map { it.retryAt }.filter { it > now }.minOrNull()?.let { it - now }
+        }
+    }
+
     /**
      * Load saved operations from preference service and add them into the queue
      * NOTE: Sometimes the loading might take longer than expected due to I/O reads from disk,
@@ -688,4 +726,11 @@ internal class OperationRepo(
         }
         initialized.complete(Unit)
     }
+
+    companion object {
+        private const val MIN_JITTER = 0.5
+
+        // Keeps the shift from overflowing; the backoff is capped long before this.
+        private const val MAX_BACKOFF_DOUBLINGS = 20
+    }
 }
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt
new file mode 100644
index 0000000000..0e03d8b423
--- /dev/null
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt
@@ -0,0 +1,35 @@
+package com.onesignal.core.internal.operations.impl
+
+import com.onesignal.core.internal.application.IApplicationLifecycleHandler
+import com.onesignal.core.internal.application.IApplicationService
+import com.onesignal.core.internal.device.IConnectivityMonitor
+import com.onesignal.core.internal.operations.IOperationRepo
+import com.onesignal.core.internal.startup.IStartableService
+import java.util.concurrent.atomic.AtomicBoolean
+
+/**
+ * Ends retry backoff on foreground, and on coming back online while in the foreground.
+ */
+internal class OperationRetryTrigger(
+    private val applicationService: IApplicationService,
+    private val connectivityMonitor: IConnectivityMonitor,
+    private val operationRepo: IOperationRepo,
+) : IStartableService, IApplicationLifecycleHandler {
+    private val monitoring = AtomicBoolean(false)
+
+    override fun start() {
+        applicationService.addApplicationLifecycleHandler(this)
+    }
+
+    override fun onFocus(firedOnSubscribe: Boolean) {
+        if (monitoring.compareAndSet(false, true)) {
+            connectivityMonitor.addOnlineHandler {
+                if (applicationService.isInForeground) operationRepo.forceExecuteOperations()
+            }
+        }
+        if (firedOnSubscribe) return
+        operationRepo.forceExecuteOperations()
+    }
+
+    override fun onUnfocused() {}
+}
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/ITime.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/ITime.kt
index 8f1824d481..86f2126547 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/ITime.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/ITime.kt
@@ -15,4 +15,10 @@ interface ITime {
      * Returns how long the app has been running.
      */
     val processUptimeMillis: Long
+
+    /**
+     * Milliseconds since boot, including deep sleep. Unaffected by device clock changes, so use
+     * it for deadlines within this process.
+     */
+    val elapsedRealtimeMillis: Long
 }
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/impl/Time.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/impl/Time.kt
index 753ef124d5..b825023168 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/impl/Time.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/time/impl/Time.kt
@@ -8,6 +8,8 @@ import com.onesignal.core.internal.time.ITime
 internal class Time : ITime {
     override val currentTimeMillis: Long
         get() = System.currentTimeMillis()
+    override val elapsedRealtimeMillis: Long
+        get() = SystemClock.elapsedRealtime()
     override val processUptimeMillis: Long
         @RequiresApi(Build.VERSION_CODES.N)
         get() = SystemClock.uptimeMillis() - android.os.Process.getStartUptimeMillis()
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserFromSubscriptionOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserFromSubscriptionOperationExecutor.kt
index 76d5551854..3d16eb5e19 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserFromSubscriptionOperationExecutor.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserFromSubscriptionOperationExecutor.kt
@@ -90,7 +90,7 @@ internal class LoginUserFromSubscriptionOperationExecutor(
 
             return when (responseType) {
                 NetworkUtils.ResponseStatusType.RETRYABLE ->
-                    ExecutionResponse(ExecutionResult.FAIL_RETRY)
+                    ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = ex.retryAfterSeconds)
                 NetworkUtils.ResponseStatusType.UNAUTHORIZED ->
                     ExecutionResponse(ExecutionResult.FAIL_UNAUTHORIZED)
                 else ->
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserOperationExecutor.kt
index 2a359e9106..200baaf3d9 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserOperationExecutor.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/executors/LoginUserOperationExecutor.kt
@@ -160,7 +160,7 @@ internal class LoginUserOperationExecutor(
                     )
                     createUser(loginUserOp, operations)
                 }
-                else -> ExecutionResponse(result.result)
+                else -> ExecutionResponse(result.result, retryAfterSeconds = result.retryAfterSeconds)
             }
         }
     }
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/device/ConnectivityMonitorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/device/ConnectivityMonitorTests.kt
new file mode 100644
index 0000000000..7266f13c09
--- /dev/null
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/device/ConnectivityMonitorTests.kt
@@ -0,0 +1,98 @@
+package com.onesignal.core.internal.device
+
+import android.content.Context
+import android.net.ConnectivityManager
+import android.net.Network
+import android.net.NetworkInfo
+import android.net.NetworkRequest
+import android.os.Build
+import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest
+import com.onesignal.core.internal.application.IApplicationService
+import com.onesignal.core.internal.device.impl.ConnectivityMonitor
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.shouldBe
+import io.mockk.every
+import io.mockk.just
+import io.mockk.mockk
+import io.mockk.runs
+import io.mockk.slot
+import org.robolectric.annotation.Config
+
+private class MonitorMocks(initiallyOnline: Boolean) {
+    val callback = slot<ConnectivityManager.NetworkCallback>()
+
+    @Suppress("DEPRECATION")
+    val connectivityManager =
+        mockk<ConnectivityManager> {
+            every { registerNetworkCallback(any<NetworkRequest>(), capture(callback)) } just runs
+            every { activeNetworkInfo } returns
+                if (initiallyOnline) mockk<NetworkInfo> { every { isConnected } returns true } else null
+        }
+    val applicationService =
+        mockk<IApplicationService> {
+            every { appContext } returns
+                mockk<Context> {
+                    every { getSystemService(Context.CONNECTIVITY_SERVICE) } returns connectivityManager
+                }
+        }
+    var onlineCount = 0
+
+    fun start() = ConnectivityMonitor(applicationService).addOnlineHandler { onlineCount++ }
+}
+
+@RobolectricTest
+@Config(sdk = [Build.VERSION_CODES.O])
+class ConnectivityMonitorTests : FunSpec({
+    test("networks already up at registration do not count as coming online") {
+        val mocks = MonitorMocks(initiallyOnline = true)
+        mocks.start()
+
+        mocks.callback.captured.onAvailable(mockk<Network>())
+
+        mocks.onlineCount shouldBe 0
+    }
+
+    test("a network appearing while offline counts as coming online") {
+        val mocks = MonitorMocks(initiallyOnline = false)
+        mocks.start()
+
+        mocks.callback.captured.onAvailable(mockk<Network>())
+
+        mocks.onlineCount shouldBe 1
+    }
+
+    test("switching networks without going offline does not count as coming online") {
+        val mocks = MonitorMocks(initiallyOnline = false)
+        val wifi = mockk<Network>()
+        val cellular = mockk<Network>()
+        mocks.start()
+
+        mocks.callback.captured.onAvailable(wifi)
+        mocks.callback.captured.onAvailable(cellular)
+        mocks.callback.captured.onLost(wifi)
+
+        mocks.onlineCount shouldBe 1
+    }
+
+    test("losing every network and getting one back counts as coming online again") {
+        val mocks = MonitorMocks(initiallyOnline = false)
+        val wifi = mockk<Network>()
+        mocks.start()
+
+        mocks.callback.captured.onAvailable(wifi)
+        mocks.callback.captured.onLost(wifi)
+        mocks.callback.captured.onAvailable(wifi)
+
+        mocks.onlineCount shouldBe 2
+    }
+
+    test("a failure registering the network callback does not throw") {
+        val mocks = MonitorMocks(initiallyOnline = true)
+        every { mocks.connectivityManager.registerNetworkCallback(any<NetworkRequest>(), any<ConnectivityManager.NetworkCallback>()) } throws
+            SecurityException()
+
+        mocks.start()
+
+        mocks.onlineCount shouldBe 0
+    }
+})
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt
index 569800cffa..b6e26948b6 100644
--- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/http/HttpClientTests.kt
@@ -6,7 +6,7 @@ import com.onesignal.core.internal.device.impl.HTTP_FID_ENV_HEADER_KEY
 import com.onesignal.core.internal.device.impl.InstallIdService
 import com.onesignal.core.internal.http.impl.HttpClient
 import com.onesignal.core.internal.http.impl.OptionalHeaders
-import com.onesignal.core.internal.time.impl.Time
+import com.onesignal.core.internal.time.ITime
 import com.onesignal.debug.LogLevel
 import com.onesignal.debug.internal.logging.Logging
 import com.onesignal.mocks.MockHelper
@@ -16,6 +16,8 @@ import io.kotest.matchers.should
 import io.kotest.matchers.shouldBe
 import io.kotest.matchers.shouldNotBe
 import io.kotest.matchers.types.beInstanceOf
+import io.mockk.every
+import io.mockk.mockk
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.withTimeoutOrNull
 import org.json.JSONObject
@@ -26,8 +28,10 @@ class Mocks {
     internal val factory = MockHttpConnectionFactory(response)
     internal val installIdService = InstallIdService(MockPreferencesService())
     internal val fidEnv = FakeFidEnv()
+    var now = 1_000_000L
+    private val time = mockk<ITime> { every { elapsedRealtimeMillis } answers { now } }
     internal val httpClient by lazy {
-        HttpClient(factory, MockPreferencesService(), mockConfigModel, Time(), installIdService, fidEnv)
+        HttpClient(factory, MockPreferencesService(), mockConfigModel, time, installIdService, fidEnv)
     }
 }
 
@@ -206,6 +210,40 @@ class HttpClientTests : FunSpec({
         response.retryAfterSeconds shouldBe 60
     }
 
+    test("should use set fallback retryAfterSeconds if Retry-After is negative") {
+        // Given
+        val mocks = Mocks()
+        mocks.response.status = 429
+        mocks.response.mockProps["Retry-After"] = "-5"
+        mocks.response.errorResponseBody = "{}"
+
+        // When
+        val response = mocks.httpClient.post("URL", JSONObject())
+
+        // Then
+        response.retryAfterSeconds shouldBe 60
+    }
+
+    test("should round a fractional Retry-After up and cap it by maxRetryAfter") {
+        // Given
+        val mocks = Mocks()
+        mocks.response.status = 429
+        mocks.response.mockProps["Retry-After"] = " 1.5 "
+        mocks.response.errorResponseBody = "{}"
+        val capped = Mocks()
+        capped.response.status = 429
+        capped.response.mockProps["Retry-After"] = "999999999999"
+        capped.response.errorResponseBody = "{}"
+
+        // When
+        val response = mocks.httpClient.post("URL", JSONObject())
+        val cappedResponse = capped.httpClient.post("URL", JSONObject())
+
+        // Then
+        response.retryAfterSeconds shouldBe 2
+        cappedResponse.retryAfterSeconds shouldBe 3600
+    }
+
     // Since 429 means "Too Many Requests", for some reason the server doesn't give us a
     // Retry-After we should assume a our safe fallback.
     test("should use set fallback retryAfterSeconds if 429 and Retry-After is missing") {
@@ -224,9 +262,7 @@ class HttpClientTests : FunSpec({
     // If the OneSignal server ever responses with a Retry-After we want to
     // make sure we delay any future requests until that delay has past. To
     // be safe we assume the server's Retry-After value means all traffic
-    // should be delayed until then. If we wanted finer grain control we
-    // could work with the the backend to decide which endpoints should be
-    // affected.
+    // should be delayed until then.
     test("ensure next request is delayed by the Retry-After value") {
         // Given
         val mocks = Mocks()
@@ -246,10 +282,11 @@ class HttpClientTests : FunSpec({
 
         mocks.factory.mockResponse = mockSuccessfulResponse
         val response2 =
-            withTimeoutOrNull(975) {
+            withTimeoutOrNull(500) {
                 mocks.httpClient.post("URL", JSONObject())
             }
 
+        mocks.now += 1_000
         val response3 =
             withTimeoutOrNull(100) {
                 mocks.httpClient.post("URL", JSONObject())
@@ -257,7 +294,35 @@ class HttpClientTests : FunSpec({
 
         // Then
         response2 shouldBe null
-        response3 shouldNotBe null
+        response3?.statusCode shouldBe 200
+    }
+
+    test("the Retry-After delay is capped by maxRetryAfter") {
+        // Given
+        val mocks = Mocks()
+        mocks.mockConfigModel.model.maxRetryAfter = 100
+
+        val mockRetryAfterResponse = MockHttpConnectionFactory.MockResponse()
+        mockRetryAfterResponse.status = 429
+        mockRetryAfterResponse.mockProps["Retry-After"] = "3600"
+        mockRetryAfterResponse.errorResponseBody = "{}"
+
+        val mockSuccessfulResponse = MockHttpConnectionFactory.MockResponse()
+        mockSuccessfulResponse.status = 200
+        mockSuccessfulResponse.responseBody = "{}"
+
+        // When
+        mocks.factory.mockResponse = mockRetryAfterResponse
+        mocks.httpClient.post("URL", JSONObject())
+
+        mocks.factory.mockResponse = mockSuccessfulResponse
+        val response =
+            withTimeoutOrNull(1_000) {
+                mocks.httpClient.post("URL", JSONObject())
+            }
+
+        // Then
+        response?.statusCode shouldBe 200
     }
 
     test("Authorization header is set when OptionalHeaders.jwt is provided") {
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt
index 7b67d3ccdc..c3ca1dffb8 100644
--- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt
@@ -8,7 +8,6 @@ import com.onesignal.core.internal.operations.impl.OperationRepo
 import com.onesignal.core.internal.operations.impl.OperationRepo.OperationQueueItem
 import com.onesignal.core.internal.preferences.PreferenceOneSignalKeys
 import com.onesignal.core.internal.preferences.PreferenceStores
-import com.onesignal.core.internal.time.impl.Time
 import com.onesignal.debug.LogLevel
 import com.onesignal.debug.internal.logging.Logging
 import com.onesignal.mocks.CoreInternalMocks
@@ -38,6 +37,7 @@ import io.mockk.runs
 import io.mockk.slot
 import io.mockk.spyk
 import io.mockk.verify
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
@@ -90,7 +90,7 @@ private class Mocks {
                 listOf(executor),
                 operationModelStore,
                 configModelStore,
-                Time(),
+                MockHelper.systemTime(),
                 getNewRecordState(configModelStore),
                 jwtTokenStore,
                 identityVerificationService,
@@ -117,7 +117,7 @@ class OperationRepoTests : FunSpec({
                     listOf(mocks.executor),
                     operationModelStore,
                     mocks.configModelStore,
-                    Time(),
+                    MockHelper.systemTime(),
                     getNewRecordState(mocks.configModelStore),
                     JwtTokenStore(MockPreferencesService()),
                     CoreInternalMocks.identityVerificationService(),
@@ -609,7 +609,6 @@ class OperationRepoTests : FunSpec({
         // Given
         val mocks = Mocks()
         val opRepo = mocks.operationRepo
-        coEvery { opRepo.delayBeforeNextExecution(any(), any()) } just runs
         coEvery {
             mocks.executor.execute(any())
         } returns ExecutionResponse(ExecutionResult.FAIL_RETRY) andThen ExecutionResponse(ExecutionResult.SUCCESS)
@@ -633,7 +632,6 @@ class OperationRepoTests : FunSpec({
                     it[0] shouldBe operation
                 },
             )
-            opRepo.delayBeforeNextExecution(1, null)
             mocks.executor.execute(
                 withArg {
                     it.count() shouldBe 1
@@ -1074,7 +1072,7 @@ class OperationRepoTests : FunSpec({
             // ensure the order: IDs are translated, operation removed from the store, then delay for postCreateDelay
             operation.translateIds(idTranslation)
             mocks.operationModelStore.remove(opId)
-            mocks.operationRepo.delayBeforeNextExecution(any(), any())
+            mocks.operationRepo.delayForPostCreate(any())
         }
     }
 
@@ -1152,13 +1150,188 @@ class OperationRepoTests : FunSpec({
         mocks.operationRepo.queue[1].operation shouldBe op2
     }
 
+    test("a retrying operation does not block operations on other records") {
+        // Given
+        val mocks = Mocks()
+        mocks.configModelStore.model.opRepoBaseFailRetryBackoff = 60_000
+        val opRepo = mocks.operationRepo
+        val failing = mockOperation(applyToRecordId = "record-a")
+        val unrelated = mockOperation(applyToRecordId = "record-b")
+        coEvery { mocks.executor.execute(listOf(failing)) } returns ExecutionResponse(ExecutionResult.FAIL_RETRY)
+
+        // When
+        opRepo.start()
+        opRepo.enqueue(failing)
+        mocks.waitForInternalEnqueue()
+        val response = withTimeoutOrNull(1_000) { opRepo.enqueueAndWait(unrelated) }
+
+        // Then
+        response shouldBe true
+        opRepo.queue.map { it.operation } shouldBe listOf(failing)
+    }
+
+    test("an operation waits behind a retrying operation on the same record and keeps its order") {
+        // Given
+        val mocks = Mocks()
+        mocks.configModelStore.model.opRepoBaseFailRetryBackoff = 200
+        val opRepo = mocks.operationRepo
+        val first = mockOperation(applyToRecordId = "record-a")
+        val second = mockOperation(applyToRecordId = "record-a")
+        val executed = mutableListOf<Operation>()
+        var firstAttempts = 0
+        coEvery { mocks.executor.execute(any()) } answers {
+            val ops = firstArg<List<Operation>>()
+            executed.addAll(ops)
+            if (ops.first() === first && firstAttempts++ == 0) {
+                ExecutionResponse(ExecutionResult.FAIL_RETRY)
+            } else {
+                ExecutionResponse(ExecutionResult.SUCCESS)
+            }
+        }
+
+        // When
+        opRepo.start()
+        opRepo.enqueue(first)
+        mocks.waitForInternalEnqueue()
+        val response = withTimeoutOrNull(2_000) { opRepo.enqueueAndWait(second) }
+
+        // Then
+        response shouldBe true
+        executed shouldBe listOf(first, first, second)
+    }
+
+    test("a waiting operation is not grouped into another operation's request") {
+        // Given
+        val mocks = Mocks()
+        mocks.configModelStore.model.opRepoBaseFailRetryBackoff = 60_000
+        val opRepo = mocks.operationRepo
+        val waiting = mockOperation(groupComparisonType = GroupComparisonType.ALTER, applyToRecordId = "record-a")
+        val starter = mockOperation(groupComparisonType = GroupComparisonType.ALTER, applyToRecordId = "record-b")
+        opRepo.queue.add(OperationQueueItem(waiting, bucket = 0).apply { retryAt = Long.MAX_VALUE })
+        opRepo.queue.add(OperationQueueItem(starter, bucket = 0))
+
+        // When
+        val ops = opRepo.getNextOps(0)
+
+        // Then
+        ops?.map { it.operation } shouldBe listOf(starter)
+        opRepo.queue.map { it.operation } shouldBe listOf(waiting)
+    }
+
+    test("backoff doubles from opRepoBaseFailRetryBackoff up to opRepoMaxFailRetryBackoff, scaled by jitter") {
+        // Given
+        val mocks = Mocks()
+        mocks.configModelStore.model.opRepoBaseFailRetryBackoff = 5_000
+        mocks.configModelStore.model.opRepoMaxFailRetryBackoff = 30_000
+        val opRepo = mocks.operationRepo
+        opRepo.jitter = { 0.5 }
+        coEvery { mocks.executor.execute(any()) } returns ExecutionResponse(ExecutionResult.FAIL_RETRY)
+        val item = OperationQueueItem(mockOperation(), bucket = 0)
+
+        // When
+        val backoffs =
+            (1..5).map {
+                val before = System.nanoTime() / 1_000_000
+                opRepo.executeOperations(listOf(item))
+                opRepo.queue.clear()
+                item.retryAt - before
+            }
+
+        // Then
+        backoffs.zip(listOf(2_500L, 5_000L, 10_000L, 15_000L, 15_000L)).forEach { (actual, expected) ->
+            (actual >= expected) shouldBe true
+            (actual <= expected + 1_000) shouldBe true
+        }
+    }
+
+    test("a failed batch shares one retry time so it retries together") {
+        // Given
+        val mocks = Mocks()
+        mocks.configModelStore.model.opRepoBaseFailRetryBackoff = 5_000
+        val opRepo = mocks.operationRepo
+        var nextJitter = 0.5
+        opRepo.jitter = { nextJitter.also { nextJitter += 0.1 } }
+        coEvery { mocks.executor.execute(any()) } returns ExecutionResponse(ExecutionResult.FAIL_RETRY)
+        val retried = OperationQueueItem(mockOperation(), bucket = 0, retries = 2)
+        val fresh = OperationQueueItem(mockOperation(), bucket = 0)
+
+        // When
+        opRepo.executeOperations(listOf(retried, fresh))
+
+        // Then
+        retried.retries shouldBe 3
+        fresh.retries shouldBe 1
+        (retried.retryAt > 0) shouldBe true
+        fresh.retryAt shouldBe retried.retryAt
+    }
+
+    test("forceExecuteOperations ends the backoff of retrying operations") {
+        // Given
+        val mocks = Mocks()
+        val opRepo = mocks.operationRepo
+        val backedOff = OperationQueueItem(mockOperation(), bucket = 0).apply { retryAt = Long.MAX_VALUE }
+        opRepo.queue.add(backedOff)
+
+        // When
+        opRepo.forceExecuteOperations()
+
+        // Then
+        backedOff.retryAt shouldBe 0
+    }
+
+    test("forceExecuteOperations skips the batching wait when it ends a retry wait") {
+        // Given
+        val mocks = Mocks()
+        mocks.configModelStore.model.opRepoBaseFailRetryBackoff = 60_000
+        val opRepo = mocks.operationRepo
+        var attempts = 0
+        val retried = CompletableDeferred<Unit>()
+        coEvery { mocks.executor.execute(any()) } answers {
+            if (attempts++ == 0) {
+                ExecutionResponse(ExecutionResult.FAIL_RETRY)
+            } else {
+                retried.complete(Unit)
+                ExecutionResponse(ExecutionResult.SUCCESS)
+            }
+        }
+        opRepo.start()
+        opRepo.enqueue(mockOperation())
+        mocks.waitForInternalEnqueue()
+        withTimeoutOrNull(1_000) { while (opRepo.queue.firstOrNull()?.retries != 1) delay(10) }
+        mocks.configModelStore.model.opRepoExecutionInterval = 60_000
+
+        // When
+        opRepo.forceExecuteOperations()
+
+        // Then
+        withTimeoutOrNull(1_000) { retried.await() } shouldBe Unit
+    }
+
+    test("a force during execute skips the next backoff") {
+        // Given
+        val mocks = Mocks()
+        mocks.configModelStore.model.opRepoBaseFailRetryBackoff = 60_000
+        val opRepo = mocks.operationRepo
+        coEvery { mocks.executor.execute(any()) } answers {
+            opRepo.forceExecuteOperations()
+            ExecutionResponse(ExecutionResult.FAIL_RETRY)
+        }
+
+        // When
+        opRepo.executeOperations(listOf(OperationQueueItem(mockOperation(), bucket = 0)))
+
+        // Then
+        opRepo.queue.single().retryAt shouldBe 0
+    }
+
     test("ensure forceExecuteOperations immediately wakes retry waiters") {
         // Given
         val mocks = Mocks()
+        mocks.configModelStore.model.opRepoBaseFailRetryBackoff = 5_000
         val opRepo = mocks.operationRepo
         coEvery {
             mocks.executor.execute(any())
-        } returns ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = 5) andThen ExecutionResponse(ExecutionResult.SUCCESS)
+        } returns ExecutionResponse(ExecutionResult.FAIL_RETRY) andThen ExecutionResponse(ExecutionResult.SUCCESS)
 
         // When
         opRepo.start()
@@ -1168,7 +1341,7 @@ class OperationRepoTests : FunSpec({
             withTimeoutOrNull(999) {
                 opRepo.enqueueAndWait(mockOperation())
             }
-        // Forcing execution to bypass 5 second retry period
+        // Forcing execution to bypass 5 second backoff
         opRepo.forceExecuteOperations()
         val response2 =
             withTimeoutOrNull(100) {
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt
new file mode 100644
index 0000000000..428dd6b62e
--- /dev/null
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt
@@ -0,0 +1,83 @@
+package com.onesignal.core.internal.operations
+
+import com.onesignal.core.internal.application.IApplicationLifecycleHandler
+import com.onesignal.core.internal.application.IApplicationService
+import com.onesignal.core.internal.device.IConnectivityMonitor
+import com.onesignal.core.internal.operations.impl.OperationRetryTrigger
+import io.kotest.core.spec.style.FunSpec
+import io.mockk.every
+import io.mockk.just
+import io.mockk.mockk
+import io.mockk.runs
+import io.mockk.slot
+import io.mockk.verify
+
+private class TriggerMocks {
+    val lifecycleHandler = slot<IApplicationLifecycleHandler>()
+    val onOnline = slot<() -> Unit>()
+    var isInForeground = true
+    val applicationService =
+        mockk<IApplicationService> {
+            every { addApplicationLifecycleHandler(capture(lifecycleHandler)) } just runs
+            every { isInForeground } answers { this@TriggerMocks.isInForeground }
+        }
+    val connectivityMonitor =
+        mockk<IConnectivityMonitor> {
+            every { addOnlineHandler(capture(onOnline)) } just runs
+        }
+    val operationRepo = mockk<IOperationRepo>(relaxed = true)
+
+    fun start() = OperationRetryTrigger(applicationService, connectivityMonitor, operationRepo).start()
+}
+
+class OperationRetryTriggerTests : FunSpec({
+    test("coming to the foreground forces queued operations to retry") {
+        val mocks = TriggerMocks()
+        mocks.start()
+
+        mocks.lifecycleHandler.captured.onFocus(firedOnSubscribe = false)
+
+        verify(exactly = 1) { mocks.operationRepo.forceExecuteOperations() }
+    }
+
+    test("focus reported on subscribe does not force a retry") {
+        val mocks = TriggerMocks()
+        mocks.start()
+
+        mocks.lifecycleHandler.captured.onFocus(firedOnSubscribe = true)
+
+        verify(exactly = 0) { mocks.operationRepo.forceExecuteOperations() }
+    }
+
+    test("connectivity is not watched until the app first comes to the foreground, then only once") {
+        val mocks = TriggerMocks()
+        mocks.start()
+        verify(exactly = 0) { mocks.connectivityMonitor.addOnlineHandler(any()) }
+
+        mocks.lifecycleHandler.captured.onFocus(firedOnSubscribe = true)
+        mocks.lifecycleHandler.captured.onFocus(firedOnSubscribe = false)
+
+        verify(exactly = 1) { mocks.connectivityMonitor.addOnlineHandler(any()) }
+    }
+
+    test("coming online in the foreground forces queued operations to retry") {
+        val mocks = TriggerMocks()
+        mocks.start()
+        mocks.lifecycleHandler.captured.onFocus(firedOnSubscribe = true)
+
+        mocks.onOnline.captured()
+
+        verify(exactly = 1) { mocks.operationRepo.forceExecuteOperations() }
+    }
+
+    test("coming online in the background does not force a retry") {
+        val mocks = TriggerMocks()
+        mocks.start()
+        mocks.lifecycleHandler.captured.onFocus(firedOnSubscribe = true)
+        mocks.isInForeground = false
+
+        mocks.onOnline.captured()
+
+        verify(exactly = 0) { mocks.operationRepo.forceExecuteOperations() }
+    }
+})
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt
index cb16466bc2..52ed67075e 100644
--- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt
@@ -2,7 +2,6 @@ package com.onesignal.user.internal.migrations
 
 import com.onesignal.core.internal.operations.impl.OperationModelStore
 import com.onesignal.core.internal.operations.impl.OperationRepo
-import com.onesignal.core.internal.time.impl.Time
 import com.onesignal.debug.LogLevel
 import com.onesignal.debug.internal.logging.Logging
 import com.onesignal.mocks.CoreInternalMocks
@@ -44,7 +43,7 @@ private class Mocks {
                 listOf(),
                 operationModelStore,
                 configModelStore,
-                Time(),
+                MockHelper.systemTime(),
                 ExecutorMocks.getNewRecordState(configModelStore),
                 JwtTokenStore(MockPreferencesService()),
                 CoreInternalMocks.identityVerificationService(),
diff --git a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt
index 500b736e79..fc4b5da2b6 100644
--- a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt
+++ b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/MockHelper.kt
@@ -27,10 +27,19 @@ object MockHelper {
     fun time(time: Long): ITime {
         val mockTime = mockk<ITime>()
         every { mockTime.currentTimeMillis } returns time
+        every { mockTime.elapsedRealtimeMillis } returns time
 
         return mockTime
     }
 
+    /** Real clocks for JVM tests, where android.os.SystemClock is not available. */
+    fun systemTime(): ITime =
+        object : ITime {
+            override val currentTimeMillis: Long get() = System.currentTimeMillis()
+            override val processUptimeMillis: Long get() = System.nanoTime() / 1_000_000
+            override val elapsedRealtimeMillis: Long get() = System.nanoTime() / 1_000_000
+        }
+
     fun applicationService(): IApplicationService {
         val mockAppService = mockk<IApplicationService>()
 
@@ -49,7 +58,7 @@ object MockHelper {
         configModel.opRepoPostWakeDelay = 1
         configModel.opRepoPostCreateDelay = 1
         configModel.opRepoPostCreateRetryUpTo = 1
-        configModel.opRepoDefaultFailRetryBackoff = 1
+        configModel.opRepoBaseFailRetryBackoff = 1
         configModel.foregroundFetchNotificationPermissionInterval = 1
 
         configModel.appId = DEFAULT_APP_ID