feat: [SDK-5396] send a heartbeat every 30 minutes of foreground time

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt

@@ -92,7 +92,8 @@ internal class SessionQueuePolicy(
                 val replaced = it.operation as UpdateSessionOperation
                 op.activeDuration = maxOf(op.activeDuration, replaced.activeDuration)
                 incoming.retries = maxOf(incoming.retries, it.retries)
-                incoming.retryAt = maxOf(incoming.retryAt, it.retryAt)
+                // An end is sent while the background job waits for it, so it doesn't wait out a heartbeat's backoff.
+                if (!op.isEnd) incoming.retryAt = maxOf(incoming.retryAt, it.retryAt)
                 val waiters = synchronized(it) { it.waiters.toList().also { _ -> it.waiters.clear() } }
                 incoming.attachOrWake(waiters)
             }

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

@@ -64,6 +64,28 @@ class SessionModel : Model() {
             setLongProperty(::activeDuration.name, value)
         }
 
+    /**
+     * The duration at this session's last heartbeat, in milliseconds, set even when the heartbeat
+     * couldn't be built. Persisted so an end after a kill while focused doesn't report less than
+     * the server already has.
+     */
+    var lastHeartbeatDuration: Long
+        get() = getLongProperty(::lastHeartbeatDuration.name) { 0L }
+        set(value) {
+            setLongProperty(::lastHeartbeatDuration.name, value)
+        }
+
+    /**
+     * When this session's last heartbeat was sent, in Unix time milliseconds, measured from [focusTime]
+     * on the monotonic clock so a wall-clock jump mid-focus can't move a killed session's end. Null
+     * until the first heartbeat, and cleared on unfocus, which then marks the end.
+     */
+    var lastHeartbeatTime: Long?
+        get() = getOptLongProperty(::lastHeartbeatTime.name)
+        set(value) {
+            setOptLongProperty(::lastHeartbeatTime.name, value)
+        }
+
     /**
      * [focusTime] from a monotonic clock ([com.onesignal.core.internal.time.ITime.elapsedRealtimeMillis]).
      * Used instead of [focusTime] to measure [activeDuration] when [usesSessionsApi].

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

@@ -5,6 +5,7 @@ import com.onesignal.common.events.EventProducer
 import com.onesignal.common.modeling.ISingletonModelStoreChangeHandler
 import com.onesignal.common.modeling.Model
 import com.onesignal.common.modeling.ModelChangedArgs
+import com.onesignal.common.threading.OneSignalDispatchers
 import com.onesignal.common.threading.runOnSerialIO
 import com.onesignal.core.internal.application.IApplicationLifecycleHandler
 import com.onesignal.core.internal.application.IApplicationService
@@ -27,8 +28,11 @@ import com.onesignal.session.internal.session.operations.CreateSessionOperation
 import com.onesignal.session.internal.session.operations.UpdateSessionOperation
 import com.onesignal.user.internal.backend.IdentityConstants
 import com.onesignal.user.internal.identity.IdentityModelStore
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeoutOrNull
 import java.util.UUID
@@ -44,6 +48,9 @@ import kotlin.reflect.KMutableProperty1
  * some amount of time after losing focus.
  *
  * The time threshold for a session to expire is a configuration option: [ConfigModel.sessionFocusTimeout].
+ *
+ * While focused, a session that [SessionModel.usesSessionsApi] reports its duration every
+ * [HEARTBEAT_INTERVAL_MS] of active time, so the server keeps it open and a crash loses little duration.
  */
 internal class SessionService(
     private val applicationService: IApplicationService,
@@ -58,6 +65,11 @@ internal class SessionService(
     // Tests replace it with a test dispatcher so the end send stays on the virtual clock.
     internal var ioDispatcher: CoroutineDispatcher = Dispatchers.IO
 
+    // Runs on the serial IO lane so it doesn't race the focus handlers. Not a constructor parameter
+    // for the same reason as ioDispatcher; tests replace it to run the heartbeat on a virtual clock.
+    internal var launchHeartbeat: (suspend () -> Unit) -> Job = OneSignalDispatchers::launchOnSerialIO
+    private val heartbeat = Heartbeat()
+
     override val startTime: Long
         // Pre-bootstrap default returns "now" so call sites computing `_time.currentTimeMillis - startTime`
         // (e.g. IAM session-duration / SESSION_TIME triggers) see ~0ms elapsed instead of ~58 years
@@ -169,6 +181,8 @@ internal class SessionService(
         val end = if (session.usesSessionsApi) endSessionOperation(session, config?.appId) else null
         sessionLifeCycleNotifier.fire { it.onSessionEnded(activeDuration) }
         session.activeDuration = 0L
+        session.lastHeartbeatDuration = 0L
+        session.lastHeartbeatTime = null
         return end
     }
 
@@ -238,6 +252,52 @@ internal class SessionService(
             session.focusElapsedRealtime = focusElapsedMs
             sessionLifeCycleNotifier.fire { it.onSessionActive() }
         }
+        if (session.usesSessionsApi) heartbeat.start(session) else heartbeat.stop()
+    }
+
+    /** Sends the session's cumulative duration every [HEARTBEAT_INTERVAL_MS] of active time while focused. */
+    private inner class Heartbeat {
+        private var job: Job? = null
+
+        @Suppress("TooGenericExceptionCaught")
+        fun start(session: SessionModel) {
+            stop()
+            job =
+                launchHeartbeat {
+                    // The serial IO scope has no exception handler, so an escaping failure would crash
+                    // the app. Stop until the next focus instead, which also can't spin on a failed tick.
+                    try {
+                        tickForever(session)
+                    } catch (e: CancellationException) {
+                        throw e
+                    } catch (e: Exception) {
+                        Logging.error("SessionService: heartbeat stopped", e)
+                    }
+                }
+        }
+
+        private suspend fun tickForever(session: SessionModel) {
+            while (true) {
+                val focusDuration = session.focusDurationAt(time.elapsedRealtimeMillis)
+                val duration = session.activeDuration + focusDuration
+                val remaining = session.lastHeartbeatDuration + HEARTBEAT_INTERVAL_MS - duration
+                if (remaining > 0) {
+                    // Rechecked after waking since delay and elapsedRealtime can disagree.
+                    delay(remaining)
+                    continue
+                }
+                Logging.debug("SessionService: heartbeat with activeDuration: $duration")
+                session.lastHeartbeatDuration = duration
+                // Measured from the focus, so a wall-clock jump mid-focus can't move a killed session's end.
+                session.lastHeartbeatTime = session.focusTime + focusDuration
+                updateSessionOperation(session, config?.appId, duration)?.let { operationRepo.enqueue(it) }
+            }
+        }
+
+        fun stop() {
+            job?.cancel()
+            job = null
+        }
     }
 
     override fun onUnfocused() {
@@ -258,16 +318,17 @@ internal class SessionService(
             Logging.warn("SessionService.onUnfocused called before bootstrap; ignoring.")
             return
         }
+        heartbeat.stop()
         // capture the amount of time the app was focused
         val dt =
             if (session.usesSessionsApi) {
-                // elapsedRealtime resets on reboot; drop the interval rather than count a negative one.
-                (unfocusElapsedMs - session.focusElapsedRealtime).coerceAtLeast(0L)
+                session.focusDurationAt(unfocusElapsedMs)
             } else {
                 unfocusTimeMs - session.focusTime
             }
         session.activeDuration += dt
         session.unfocusTime = unfocusTimeMs
+        session.lastHeartbeatTime = null
         Logging.log(LogLevel.DEBUG, "SessionService.onUnfocused adding time $dt for total: ${session.activeDuration}")
     }
 
@@ -285,6 +346,9 @@ internal class SessionService(
     companion object {
         // Covers the post-create delay when the create is still queued, plus the request itself.
         private const val END_SESSION_SEND_TIMEOUT_MS = 30_000L
+
+        // Hardcoded until the interval is remotely configured.
+        internal const val HEARTBEAT_INTERVAL_MS = 30 * 60 * 1000L
     }
 }
 
@@ -304,19 +368,40 @@ private fun createSessionOperation(
         )
     }
 
+/** Time in the current focus, which [SessionModel.activeDuration] doesn't include until unfocus. */
+private fun SessionModel.focusDurationAt(elapsedRealtimeMs: Long): Long =
+    // elapsedRealtime resets on reboot; drop the interval rather than count a negative one.
+    (elapsedRealtimeMs - focusElapsedRealtime).coerceAtLeast(0L)
+
+/**
+ * An app killed while focused never adds its last focus to [SessionModel.activeDuration] or sets
+ * [SessionModel.unfocusTime], so fall back to the last focus or heartbeat, whichever is later.
+ */
 private fun endSessionOperation(
     session: SessionModel,
     appId: String?,
 ): UpdateSessionOperation? =
-    withSessionIds(session, appId, "ending") { resolvedAppId, onesignalId, subscriptionId ->
+    updateSessionOperation(
+        session,
+        appId,
+        activeDuration = maxOf(session.activeDuration, session.lastHeartbeatDuration),
+        endTime = maxOf(session.unfocusTime ?: 0L, session.focusTime, session.lastHeartbeatTime ?: 0L),
+    )
+
+private fun updateSessionOperation(
+    session: SessionModel,
+    appId: String?,
+    activeDuration: Long,
+    endTime: Long? = null,
+): UpdateSessionOperation? =
+    withSessionIds(session, appId, if (endTime != null) "ending" else "updating") { resolvedAppId, onesignalId, subscriptionId ->
         UpdateSessionOperation(
             appId = resolvedAppId,
             sessionId = session.localSessionId,
             onesignalId = onesignalId,
             subscriptionId = subscriptionId,
-            activeDuration = session.activeDuration,
-            // An app killed while focused has no unfocus after its last focus.
-            endTime = maxOf(session.unfocusTime ?: 0L, session.focusTime),
+            activeDuration = activeDuration,
+            endTime = endTime,
         )
     }
 

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

@@ -181,6 +181,20 @@ class SessionOperationQueueTests : FunSpec({
         incoming.retryAt shouldBe 60_000L
     }
 
+    test("an end that replaces a backed-off update keeps its retries but not its backoff") {
+        val policy = SessionQueuePolicy()
+        val backedOff = item(update()).apply {
+            retries = 3
+            retryAt = 60_000L
+        }
+        val end = item(update(activeDuration = 2_000L, endTime = 9L))
+
+        policy.onEnqueue(mutableListOf(backedOff), end)
+
+        end.retries shouldBe 3
+        end.retryAt shouldBe 0L
+    }
+
     test("an update after an end that is executing is dropped") {
         val policy = SessionQueuePolicy()
         val queue = mutableListOf<OperationRepo.OperationQueueItem>()