diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt index 8205f23b3..199f23a1a 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt +++ b/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) } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/SessionModel.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/SessionModel.kt index f76f40408..8768ed3ab 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/SessionModel.kt +++ b/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]. diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt index 246b48148..ea002d6d4 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt +++ b/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, ) } diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt index 9b662ad3a..7ed654cae 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt +++ b/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() diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt index b19465312..79d803d7c 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt @@ -41,7 +41,11 @@ import io.mockk.verify import io.mockk.verifyOrder import kotlinx.coroutines.Job import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.launch import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent import kotlinx.coroutines.test.runTest // Mocks used by every test in this file @@ -53,10 +57,13 @@ private class Mocks( private val mockSessionModelStore = MockHelper.sessionModelStore() + // Runs the heartbeat in virtual time; elapsedRealtime advances with it. + val heartbeatScope = TestScope() + val time: ITime = mockk().also { every { it.currentTimeMillis } answers { currentTime } - every { it.elapsedRealtimeMillis } answers { elapsedRealtime } + every { it.elapsedRealtimeMillis } answers { elapsedRealtime + heartbeatScope.testScheduler.currentTime } } val featureManager: IFeatureManager = @@ -117,7 +124,18 @@ private class Mocks( featureManager, identityModelStore, operationRepo, - ) + ).also { it.launchHeartbeat = ::launchInHeartbeatScope } + + var heartbeatJob: Job? = null + + fun launchInHeartbeatScope(block: suspend () -> Unit): Job = heartbeatScope.launch { block() }.also { heartbeatJob = it } + + val heartbeats get() = enqueued.filterIsInstance().filter { !it.isEnd } + + fun advanceTime(ms: Long) { + heartbeatScope.advanceTimeBy(ms) + heartbeatScope.runCurrent() + } val spyCallback = spyk() @@ -127,6 +145,9 @@ private class Mocks( } } +private const val MINUTE_MS = 60_000L +private const val HEARTBEAT_MS = 30 * MINUTE_MS + class SessionServiceTests : FunSpec({ // SessionService.onFocus/onUnfocused dispatch their state mutation through the now-always-async // runOnSerialIO. IOMockHelper.beforeSpec stubs runOnSerialIO to run inline so the tests below @@ -828,6 +849,269 @@ class SessionServiceTests : FunSpec({ sessionModelStore.model.activeDuration shouldBe 200L } + test("heartbeat sends the cumulative duration after 30 minutes in the foreground") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + val sessionModelStore = mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + + // When + mocks.advanceTime(HEARTBEAT_MS - 1) + + // Then + mocks.heartbeats shouldBe emptyList() + + // When + mocks.advanceTime(1) + + // Then + val heartbeat = mocks.heartbeats.single() + heartbeat.sessionId shouldBe sessionModelStore.model.localSessionId + heartbeat.onesignalId shouldBe Mocks.ONESIGNAL_ID + heartbeat.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID + heartbeat.activeDuration shouldBe HEARTBEAT_MS + heartbeat.endTime shouldBe null + } + + test("heartbeat sends again every 30 minutes in the foreground") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + + // When + mocks.advanceTime(2 * HEARTBEAT_MS) + + // Then + mocks.heartbeats.map { it.activeDuration } shouldBe listOf(HEARTBEAT_MS, 2 * HEARTBEAT_MS) + } + + test("heartbeat pauses in the background and counts only foreground time") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + mocks.advanceTime(20 * MINUTE_MS) + sessionService.onUnfocused() + + // When + mocks.advanceTime(2 * HEARTBEAT_MS) + + // Then + mocks.heartbeats shouldBe emptyList() + + // When + sessionService.onFocus(false) + mocks.advanceTime(10 * MINUTE_MS) + + // Then + mocks.heartbeats.single().activeDuration shouldBe HEARTBEAT_MS + } + + test("new session restarts the heartbeat interval") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + mocks.advanceTime(HEARTBEAT_MS) + sessionService.onUnfocused() + sessionService.backgroundRun() + mocks.enqueued.clear() + + // When + sessionService.onFocus(false) + mocks.advanceTime(HEARTBEAT_MS) + + // Then + mocks.heartbeats.single().activeDuration shouldBe HEARTBEAT_MS + } + + test("cold start after a kill following a heartbeat ends with at least the heartbeat's duration and time") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + mocks.currentTime += HEARTBEAT_MS + mocks.advanceTime(HEARTBEAT_MS) + val heartbeatTime = mocks.currentTime + + // When + val restarted = + SessionService( + MockHelper.applicationService(), + mocks.configModelStore, + mocks.sessionModelStore(), + mocks.time, + mocks.featureManager, + mocks.identityModelStore, + mocks.operationRepo, + ).also { it.launchHeartbeat = { _ -> mockk(relaxed = true) } } + mocks.enqueued.clear() + restarted.bootstrap() + restarted.start() + restarted.onFocus(false) + + // Then + val end = mocks.enqueued.filterIsInstance().single() + end.isEnd shouldBe true + end.activeDuration shouldBe HEARTBEAT_MS + end.endTime shouldBe heartbeatTime + } + + test("cold start after a kill following a clock-jumped heartbeat ends at the heartbeat's foreground time") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + val focusTime = mocks.currentTime + mocks.currentTime += 2 * HEARTBEAT_MS + mocks.advanceTime(HEARTBEAT_MS) + + // When + val restarted = + SessionService( + MockHelper.applicationService(), + mocks.configModelStore, + mocks.sessionModelStore(), + mocks.time, + mocks.featureManager, + mocks.identityModelStore, + mocks.operationRepo, + ).also { it.launchHeartbeat = { _ -> mockk(relaxed = true) } } + mocks.enqueued.clear() + restarted.bootstrap() + restarted.start() + restarted.onFocus(false) + + // Then + val end = mocks.enqueued.filterIsInstance().single() + end.activeDuration shouldBe HEARTBEAT_MS + end.endTime shouldBe focusTime + HEARTBEAT_MS + } + + test("session that unfocused after a heartbeat ends at its unfocus even if the heartbeat's clock ran ahead") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + mocks.currentTime += 2 * HEARTBEAT_MS + mocks.advanceTime(HEARTBEAT_MS) + mocks.currentTime -= HEARTBEAT_MS + val unfocusTime = mocks.currentTime + sessionService.onUnfocused() + + // When + sessionService.backgroundRun() + + // Then + val end = mocks.enqueued.filterIsInstance().single { it.isEnd } + end.endTime shouldBe unfocusTime + end.activeDuration shouldBe HEARTBEAT_MS + } + + test("session end clears the last heartbeat") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionModelStore = + mocks.sessionModelStore { + it.isValid = true + it.lastHeartbeatDuration = HEARTBEAT_MS + it.lastHeartbeatTime = 999L + } + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + + // When + sessionService.backgroundRun() + + // Then + sessionModelStore.model.lastHeartbeatDuration shouldBe 0L + sessionModelStore.model.lastHeartbeatTime shouldBe null + } + + test("heartbeat that fails stops without an uncaught exception and restarts on the next focus") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + every { mocks.operationRepo.enqueue(ofType(), any()) } throws RuntimeException("enqueue failed") + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + + // When + mocks.advanceTime(2 * HEARTBEAT_MS) + + // Then + verify(exactly = 1) { mocks.operationRepo.enqueue(ofType(), any()) } + val failed = mocks.heartbeatJob!! + failed.isCompleted shouldBe true + failed.isCancelled shouldBe false + + // When + sessionService.onUnfocused() + sessionService.onFocus(false) + mocks.advanceTime(HEARTBEAT_MS) + + // Then + verify(exactly = 2) { mocks.operationRepo.enqueue(ofType(), any()) } + } + + test("heartbeat does not run when the session uses the legacy path") { + // Given + val mocks = Mocks(sessionsApiEnabled = false) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + + // When + sessionService.onFocus(false) + mocks.advanceTime(2 * HEARTBEAT_MS) + + // Then + mocks.enqueued shouldBe emptyList() + } + + test("heartbeat without a push subscription ID sends nothing and waits for the next interval") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + mocks.configModelStore.model.pushSubscriptionId = null + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + + // When + mocks.advanceTime(2 * HEARTBEAT_MS) + + // Then + mocks.enqueued shouldBe emptyList() + } + test("session model fields survive a reload from preferences") { // Given val prefs = MockPreferencesService() @@ -840,6 +1124,8 @@ class SessionServiceTests : FunSpec({ store.model.usesSessionsApi = true store.model.onesignalId = Mocks.ONESIGNAL_ID store.model.subscriptionId = Mocks.SUBSCRIPTION_ID + store.model.lastHeartbeatDuration = 654L + store.model.lastHeartbeatTime = 987L // When val reloaded = SessionModelStore(prefs).model @@ -853,6 +1139,8 @@ class SessionServiceTests : FunSpec({ reloaded.usesSessionsApi shouldBe true reloaded.onesignalId shouldBe Mocks.ONESIGNAL_ID reloaded.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID + reloaded.lastHeartbeatDuration shouldBe 654L + reloaded.lastHeartbeatTime shouldBe 987L } test("onFocus dispatches the session-mutation body through runOnSerialIO (SDK-4508)") {