diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/SessionModule.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/SessionModule.kt index 1dc8cc9888..1f74337f90 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/SessionModule.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/SessionModule.kt @@ -22,6 +22,9 @@ import com.onesignal.session.internal.session.backend.ISessionsBackendService import com.onesignal.session.internal.session.backend.impl.SessionsBackendService import com.onesignal.session.internal.session.impl.SessionListener import com.onesignal.session.internal.session.impl.SessionService +import com.onesignal.session.internal.session.requests.ISessionRequestQueue +import com.onesignal.session.internal.session.requests.SessionRequestStore +import com.onesignal.session.internal.session.requests.impl.SessionRequestQueue internal class SessionModule : IModule { override fun register(builder: ServiceBuilder) { @@ -46,5 +49,9 @@ internal class SessionModule : IModule { builder.register().provides() builder.register().provides() builder.register().provides() + builder.register().provides() + builder.register() + .provides() + .provides() } } 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 9f520e0593..9bac041d67 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 @@ -46,6 +46,15 @@ class SessionModel : Model() { setLongProperty(::focusTime.name, value) } + /** + * When this app was last unfocused, in Unix time milliseconds. Null until the first unfocus. + */ + var unfocusTime: Long? + get() = getOptLongProperty(::unfocusTime.name) + set(value) { + setOptLongProperty(::unfocusTime.name, value) + } + /** * How long this session has spent as active, in milliseconds. */ 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 7ac57d91be..ad90e666ec 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 @@ -22,8 +22,11 @@ import com.onesignal.session.internal.session.ISessionLifecycleHandler import com.onesignal.session.internal.session.ISessionService import com.onesignal.session.internal.session.SessionModel import com.onesignal.session.internal.session.SessionModelStore +import com.onesignal.session.internal.session.requests.ISessionRequestQueue +import com.onesignal.session.internal.session.requests.SessionRequest import com.onesignal.user.internal.backend.IdentityConstants import com.onesignal.user.internal.identity.IdentityModelStore +import kotlinx.coroutines.withTimeoutOrNull import java.util.UUID import kotlin.reflect.KMutableProperty1 @@ -45,6 +48,7 @@ internal class SessionService( private val time: ITime, private val featureManager: IFeatureManager, private val identityModelStore: IdentityModelStore, + private val sessionRequestQueue: ISessionRequestQueue, ) : ISessionService, IBootstrapService, IStartableService, IBackgroundService, IApplicationLifecycleHandler { override val startTime: Long // Pre-bootstrap default returns "now" so call sites computing `_time.currentTimeMillis - startTime` @@ -81,18 +85,33 @@ internal class SessionService( override fun start() { applicationService.addApplicationLifecycleHandler(this) - identityModelStore.subscribe(PinnedLocalIdTranslator(IdentityConstants.ONESIGNAL_ID, SessionModel::onesignalId)) + identityModelStore.subscribe( + PinnedLocalIdTranslator( + IdentityConstants.ONESIGNAL_ID, + SessionModel::onesignalId, + translateQueued = sessionRequestQueue::translateOnesignalId, + ), + ) // Login before the anonymous user is created carries the local push subscription over to // the new user, so its backend ID belongs to that user, not the one pinned on this session. configModelStore.subscribe( - PinnedLocalIdTranslator(ConfigModel::pushSubscriptionId.name, SessionModel::subscriptionId) { session -> - val identityModel = identityModelStore.model - val currentOnesignalId = if (identityModel.hasProperty(IdentityConstants.ONESIGNAL_ID)) identityModel.onesignalId else null - session.onesignalId == currentOnesignalId - }, + PinnedLocalIdTranslator( + ConfigModel::pushSubscriptionId.name, + SessionModel::subscriptionId, + appliesTo = { session -> session.onesignalId == currentOnesignalId }, + translateQueued = { localId, backendId -> + sessionRequestQueue.translateSubscriptionId(localId, backendId, currentOnesignalId) + }, + ), ) } + private val currentOnesignalId: String? + get() { + val identityModel = identityModelStore.model + return if (identityModel.hasProperty(IdentityConstants.ONESIGNAL_ID)) identityModel.onesignalId else null + } + /** * A session that starts before the user is created pins local IDs. When the backend replaces * a local ID in place, carry it over to the pinned copy. Changes from a non-local ID, such as a @@ -102,6 +121,7 @@ internal class SessionService( private val sourceProperty: String, private val pinnedId: KMutableProperty1, private val appliesTo: (SessionModel) -> Boolean = { true }, + private val translateQueued: (localId: String, backendId: String) -> Unit, ) : ISingletonModelStoreChangeHandler { override fun onModelReplaced( model: TModel, @@ -118,6 +138,7 @@ internal class SessionService( if (localId == null || backendId == null) return runOnSerialIO { + translateQueued(localId, backendId) val session = this@SessionService.session ?: return@runOnSerialIO if (pinnedId.get(session) == localId && appliesTo(session)) pinnedId.set(session, backendId) } @@ -129,21 +150,29 @@ internal class SessionService( * the SDK. */ override suspend fun backgroundRun() { - endSession() + val end = endSession() ?: return + // Android 15+ blocks network requests started after this job finishes, so wait for the end + // here. On timeout it stays queued and is sent on the next app open. + val sent = withTimeoutOrNull(END_SESSION_SEND_TIMEOUT_MS) { sessionRequestQueue.enqueueAndWait(end) } + Logging.debug("SessionService.backgroundRun: session end sent: $sent") } - private fun endSession() { + /** + * @return the request that ends the session through the sessions API, for the caller to enqueue. + */ + private fun endSession(): SessionRequest? { // Defensive: if bootstrap() has not run yet, there is no session state to end. // This can happen when SyncJobService races with // an in-flight initWithContext that has not yet reached bootstrapServices(). - val session = this.session ?: return - if (!session.isValid) return + val session = this.session?.takeIf { it.isValid } ?: return null val activeDuration = session.activeDuration Logging.debug("SessionService.backgroundRun: Session ended. activeDuration: $activeDuration") session.isValid = false + val end = if (session.usesSessionsApi) session.toEndRequest(config?.appId) else null sessionLifeCycleNotifier.fire { it.onSessionEnded(activeDuration) } session.activeDuration = 0L + return end } /** @@ -181,7 +210,7 @@ internal class SessionService( // Treat app cold starts as a new session, we attempt to end any previous session to do this. if (!hasFocused) { hasFocused = true - endSession() + endSession()?.let { sessionRequestQueue.enqueue(it) } } if (!session.isValid) { @@ -200,6 +229,7 @@ internal class SessionService( session.serverSessionId = null session.isValid = true Logging.debug("SessionService: New session started at ${session.startTime}") + if (session.usesSessionsApi) session.toCreateRequest(config?.appId)?.let { sessionRequestQueue.enqueue(it) } sessionLifeCycleNotifier.fire { it.onSessionStarted() } } else { // existing session: just remember the focus time so we can calculate the active time @@ -237,6 +267,7 @@ internal class SessionService( unfocusTimeMs - session.focusTime } session.activeDuration += dt + session.unfocusTime = unfocusTimeMs Logging.log(LogLevel.DEBUG, "SessionService.onUnfocused adding time $dt for total: ${session.activeDuration}") } @@ -250,4 +281,51 @@ internal class SessionService( override val hasSubscribers: Boolean get() = sessionLifeCycleNotifier.hasSubscribers + + companion object { + // Covers the new record delay when the create was just sent, plus the request itself. + private const val END_SESSION_SEND_TIMEOUT_MS = 30_000L + } +} + +private fun SessionModel.toCreateRequest(appId: String?): SessionRequest? = + withSessionIds(appId, "creating") { resolvedAppId, onesignalId, subscriptionId -> + SessionRequest.create( + appId = resolvedAppId, + localSessionId = sessionId, + onesignalId = onesignalId, + subscriptionId = subscriptionId, + startTime = startTime, + ) + } + +private fun SessionModel.toEndRequest(appId: String?): SessionRequest? = + withSessionIds(appId, "ending") { resolvedAppId, onesignalId, subscriptionId -> + SessionRequest.update( + appId = resolvedAppId, + localSessionId = sessionId, + serverSessionId = serverSessionId, + onesignalId = onesignalId, + subscriptionId = subscriptionId, + activeDuration = activeDuration, + // An app killed while focused has no unfocus after its last focus. + endTime = maxOf(unfocusTime ?: 0L, focusTime), + ) + } + +private fun SessionModel.withSessionIds( + appId: String?, + action: String, + block: (appId: String, onesignalId: String, subscriptionId: String) -> T, +): T? { + val onesignalId = onesignalId + val subscriptionId = subscriptionId + if (appId == null || onesignalId == null || subscriptionId == null) { + Logging.warn( + "SessionService: not $action session $sessionId, missing appId: ${appId == null}, " + + "onesignalId: ${onesignalId == null}, subscriptionId: ${subscriptionId == null}", + ) + return null + } + return block(appId, onesignalId, subscriptionId) } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/ISessionRequestQueue.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/ISessionRequestQueue.kt new file mode 100644 index 0000000000..9a12759401 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/ISessionRequestQueue.kt @@ -0,0 +1,34 @@ +package com.onesignal.session.internal.session.requests + +/** + * Sends sessions API requests in their own persisted queue, so a sessions outage never delays + * the operation repo's user and subscription operations. + */ +internal interface ISessionRequestQueue { + fun enqueue(request: SessionRequest) + + /** + * Enqueues [request] and suspends until it is sent, or dropped without being sent. + * + * @return true if the backend accepted [request]. + */ + suspend fun enqueueAndWait(request: SessionRequest): Boolean + + /** + * Replaces a local OneSignal ID once the backend creates the user. + */ + fun translateOnesignalId( + localId: String, + backendId: String, + ) + + /** + * Replaces a local subscription ID once the backend creates the subscription, only in requests + * pinned to [onesignalId], the user it was created for. + */ + fun translateSubscriptionId( + localId: String, + backendId: String, + onesignalId: String?, + ) +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/SessionRequest.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/SessionRequest.kt new file mode 100644 index 0000000000..1edbcc9cf1 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/SessionRequest.kt @@ -0,0 +1,154 @@ +package com.onesignal.session.internal.session.requests + +import com.onesignal.common.IDManager +import com.onesignal.common.modeling.Model +import java.util.UUID + +/** + * A persisted sessions API request, waiting in [ISessionRequestQueue] to be sent. + */ +internal class SessionRequest : Model() { + enum class Type { CREATE, UPDATE } + + var type: Type + get() = getEnumProperty(::type.name) + private set(value) { + setEnumProperty(::type.name, value) + } + + var appId: String + get() = getStringProperty(::appId.name) + private set(value) { + setStringProperty(::appId.name, value) + } + + /** + * The SDK's [com.onesignal.session.internal.session.SessionModel.sessionId], which ties a + * session's updates to its create. + */ + var localSessionId: String + get() = getStringProperty(::localSessionId.name) + private set(value) { + setStringProperty(::localSessionId.name, value) + } + + /** + * Null until the session's create succeeds. Always null for a create. + */ + var serverSessionId: String? + get() = getOptStringProperty(::serverSessionId.name) + set(value) { + setOptStringProperty(::serverSessionId.name, value) + } + + /** + * May be a local ID until the user is created; see [IDManager.isLocalId]. + */ + var onesignalId: String + get() = getStringProperty(::onesignalId.name) + set(value) { + setStringProperty(::onesignalId.name, value) + } + + /** + * May be a local ID until the subscription is created; see [IDManager.isLocalId]. + */ + var subscriptionId: String + get() = getStringProperty(::subscriptionId.name) + set(value) { + setStringProperty(::subscriptionId.name, value) + } + + /** Unix time in milliseconds. Set on a create. */ + var startTime: Long + get() = getLongProperty(::startTime.name) { 0L } + private set(value) { + setLongProperty(::startTime.name, value) + } + + /** Cumulative active time in milliseconds. Set on an update. */ + var activeDuration: Long + get() = getLongProperty(::activeDuration.name) { 0L } + set(value) { + setLongProperty(::activeDuration.name, value) + } + + /** Unix time in milliseconds; set when this update ends the session. */ + var endTime: Long? + get() = getOptLongProperty(::endTime.name) + private set(value) { + setOptLongProperty(::endTime.name, value) + } + + /** + * Generated once and persisted, so every retry, including after a restart, is deduplicated by the backend. + */ + var idempotencyKey: String + get() = getStringProperty(::idempotencyKey.name) + private set(value) { + setStringProperty(::idempotencyKey.name, value) + } + + /** Failed attempts the backend answered, persisted so the limit holds across restarts. */ + var failedAttempts: Int + get() = getOptIntProperty(::failedAttempts.name) ?: 0 + set(value) { + setIntProperty(::failedAttempts.name, value) + } + + val isCreate: Boolean get() = type == Type.CREATE + + val isEnd: Boolean get() = type == Type.UPDATE && endTime != null + + val hasBackendIds: Boolean + get() = !IDManager.isLocalId(onesignalId) && !IDManager.isLocalId(subscriptionId) + + fun isSameSession(other: SessionRequest) = appId == other.appId && localSessionId == other.localSessionId + + override fun toString() = "SessionRequest(type: $type, session: $localSessionId, end: $isEnd, failedAttempts: $failedAttempts)" + + companion object { + fun create( + appId: String, + localSessionId: String, + onesignalId: String, + subscriptionId: String, + startTime: Long, + ) = SessionRequest().apply { + init(Type.CREATE, appId, localSessionId, onesignalId, subscriptionId) + this.startTime = startTime + } + + @Suppress("LongParameterList") + fun update( + appId: String, + localSessionId: String, + serverSessionId: String?, + onesignalId: String, + subscriptionId: String, + activeDuration: Long, + endTime: Long? = null, + ) = SessionRequest().apply { + init(Type.UPDATE, appId, localSessionId, onesignalId, subscriptionId) + this.serverSessionId = serverSessionId + this.activeDuration = activeDuration + this.endTime = endTime + } + + private fun SessionRequest.init( + type: Type, + appId: String, + localSessionId: String, + onesignalId: String, + subscriptionId: String, + ) { + id = UUID.randomUUID().toString() + this.type = type + this.appId = appId + this.localSessionId = localSessionId + this.onesignalId = onesignalId + this.subscriptionId = subscriptionId + idempotencyKey = UUID.randomUUID().toString() + } + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/SessionRequestStore.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/SessionRequestStore.kt new file mode 100644 index 0000000000..ccba5d841f --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/SessionRequestStore.kt @@ -0,0 +1,7 @@ +package com.onesignal.session.internal.session.requests + +import com.onesignal.common.modeling.SimpleModelStore +import com.onesignal.core.internal.preferences.IPreferencesService + +internal open class SessionRequestStore(prefs: IPreferencesService) : + SimpleModelStore({ SessionRequest() }, "session_requests", prefs) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/impl/SessionRequestQueue.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/impl/SessionRequestQueue.kt new file mode 100644 index 0000000000..1496fc1678 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/requests/impl/SessionRequestQueue.kt @@ -0,0 +1,408 @@ +package com.onesignal.session.internal.session.requests.impl + +import android.content.Context +import android.net.ConnectivityManager +import android.net.Network +import android.net.NetworkCapabilities +import android.net.NetworkRequest +import com.onesignal.common.NetworkUtils +import com.onesignal.common.exceptions.BackendException +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 +import com.onesignal.core.internal.device.IDeviceService +import com.onesignal.core.internal.startup.IStartableService +import com.onesignal.core.internal.time.ITime +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.session.internal.session.SessionModelStore +import com.onesignal.session.internal.session.backend.CreateSessionRequest +import com.onesignal.session.internal.session.backend.ISessionsBackendService +import com.onesignal.session.internal.session.backend.UpdateSessionRequest +import com.onesignal.session.internal.session.requests.ISessionRequestQueue +import com.onesignal.session.internal.session.requests.SessionRequest +import com.onesignal.session.internal.session.requests.SessionRequestStore +import com.onesignal.user.internal.operations.impl.states.NewRecordsState +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.withTimeoutOrNull +import java.util.concurrent.TimeUnit.MILLISECONDS +import kotlin.math.pow +import kotlin.random.Random + +/** + * Sends one request at a time, oldest first among those ready: a create waits for backend user and + * subscription IDs, and an update waits for its session's create. A request that fails is retried + * with backoff without holding up the others. + */ +@Suppress("TooManyFunctions") +internal class SessionRequestQueue( + private val store: SessionRequestStore, + private val sessionsBackend: ISessionsBackendService, + private val deviceService: IDeviceService, + private val newRecordsState: NewRecordsState, + private val applicationService: IApplicationService, + private val sessionModelStore: SessionModelStore, + private val time: ITime, +) : ISessionRequestQueue, IStartableService, IApplicationLifecycleHandler { + private val lock = Any() + private val wakeSignal = Channel(Channel.CONFLATED) + private var inFlightId: String? = null + + // Backoff is per run; a restart retries saved requests right away. + private val retryAt = mutableMapOf() + private val attempts = mutableMapOf() + private val waiters = mutableMapOf>>() + + // A create can succeed after the session's next update was built without its server ID. + private val createdSessionIds = mutableMapOf() + + override fun start() { + applicationService.addApplicationLifecycleHandler(this) + registerNetworkCallback() + OneSignalDispatchers.launchOnIO { processForever() } + } + + override fun enqueue(request: SessionRequest) { + admit(request, waiter = null) + } + + override suspend fun enqueueAndWait(request: SessionRequest): Boolean { + val waiter = CompletableDeferred() + admit(request, waiter) + return waiter.await() + } + + override fun translateOnesignalId( + localId: String, + backendId: String, + ) = translate { if (it.onesignalId == localId) it.onesignalId = backendId } + + override fun translateSubscriptionId( + localId: String, + backendId: String, + onesignalId: String?, + ) = translate { if (it.subscriptionId == localId && it.onesignalId == onesignalId) it.subscriptionId = backendId } + + private fun translate(apply: (SessionRequest) -> Unit) { + synchronized(lock) { store.list().forEach(apply) } + wake() + } + + override fun onFocus(firedOnSubscribe: Boolean) = retryNow() + + override fun onUnfocused() = Unit + + private fun retryNow() { + synchronized(lock) { retryAt.clear() } + wake() + } + + private fun wake() { + wakeSignal.trySend(Unit) + } + + private fun admit( + request: SessionRequest, + waiter: CompletableDeferred?, + ) { + synchronized(lock) { + waiter?.let { waiters.getOrPut(request.id) { mutableListOf() }.add(it) } + if (!request.isCreate && !admitUpdate(request)) { + complete(request.id, sent = false) + return + } + store.add(request) + enforceCap() + } + wake() + } + + /** + * Folds unsent updates for the same session into [incoming], which then carries the highest + * duration. An end replaces them the same way. + * + * @return false if [incoming] should be discarded: its session's end is already queued or + * sending, or the session has no server ID and no queued create, so it could never be sent. + */ + private fun admitUpdate(incoming: SessionRequest): Boolean { + val requests = store.list() + val updates = requests.filter { !it.isCreate && it.isSameSession(incoming) } + if (incoming.serverSessionId == null) incoming.serverSessionId = createdSessionIds[incoming.localSessionId] + val hasEnd = updates.any { it.isEnd } + val isOrphaned = incoming.serverSessionId == null && requests.none { it.isCreate && it.isSameSession(incoming) } + if (hasEnd || isOrphaned) { + Logging.debug("SessionRequestQueue: discarding $incoming, session ended: $hasEnd, never created: $isOrphaned") + return false + } + coalesce(incoming, updates.filter { it.id != inFlightId }) + return true + } + + private fun coalesce( + incoming: SessionRequest, + unsent: List, + ) { + if (unsent.isEmpty()) return + incoming.activeDuration = maxOf(incoming.activeDuration, unsent.maxOf { it.activeDuration }) + // Otherwise a session that keeps sending updates would never reach the attempt limit. + incoming.failedAttempts = maxOf(incoming.failedAttempts, unsent.maxOf { it.failedAttempts }) + unsent.forEach { replaced -> + remove(replaced) + waiters.remove(replaced.id)?.let { waiters.getOrPut(incoming.id) { mutableListOf() }.addAll(it) } + } + Logging.debug("SessionRequestQueue: coalesced ${unsent.size} unsent update(s) for session ${incoming.localSessionId}") + } + + private fun enforceCap() { + while (true) { + val unsent = store.list().filter { it.id != inFlightId } + if (unsent.size <= MAX_QUEUED_REQUESTS) return + + val oldest = unsent.first() + Logging.warn("SessionRequestQueue: more than $MAX_QUEUED_REQUESTS requests queued, dropping oldest: $oldest") + drop(oldest) + } + } + + internal suspend fun processForever() { + while (true) { + val next = synchronized(lock) { nextReady()?.also { inFlightId = it.id } } + if (next == null) { + val waitMs = synchronized(lock) { nextWakeDelay() } + if (waitMs == null) wakeSignal.receive() else withTimeoutOrNull(waitMs) { wakeSignal.receive() } + continue + } + try { + send(next) + } catch (e: CancellationException) { + throw e + } catch (@Suppress("TooGenericExceptionCaught") e: Throwable) { + Logging.error("SessionRequestQueue: error sending $next, dropping", e) + synchronized(lock) { drop(next) } + } finally { + synchronized(lock) { inFlightId = null } + } + } + } + + private fun nextReady(): SessionRequest? { + val now = time.currentTimeMillis + return store.list().firstOrNull { request -> + (retryAt[request.id] ?: 0L) <= now && hasAccessibleIds(request) + } + } + + private fun hasAccessibleIds(request: SessionRequest): Boolean { + val serverSessionId = request.serverSessionId + if (!request.isCreate && (serverSessionId == null || !newRecordsState.canAccess(serverSessionId))) return false + return request.hasBackendIds && + newRecordsState.canAccess(request.onesignalId) && + newRecordsState.canAccess(request.subscriptionId) + } + + /** + * @return how long until a waiting request may become ready, or null to wait for a wake. + */ + private fun nextWakeDelay(): Long? { + val now = time.currentTimeMillis + val pending = store.list() + val untilRetry = pending.mapNotNull { retryAt[it.id] }.minOrNull()?.let { (it - now).coerceAtLeast(0L) } + // NewRecordsState doesn't expose when a record becomes accessible, so poll for it. + val awaitingNewRecord = pending.any { (retryAt[it.id] ?: 0L) <= now && it.hasBackendIds && !hasAccessibleIds(it) } + val untilNewRecord = if (awaitingNewRecord) NEW_RECORD_POLL_MS else null + return listOfNotNull(untilRetry, untilNewRecord).minOrNull() + } + + private suspend fun send(request: SessionRequest) { + try { + if (request.isCreate) sendCreate(request) else sendUpdate(request) + } catch (ex: BackendException) { + synchronized(lock) { onFailure(request, ex) } + } + } + + private suspend fun sendCreate(request: SessionRequest) { + val serverSessionId = + sessionsBackend.createSession( + request.appId, + CreateSessionRequest( + onesignalId = request.onesignalId, + subscriptionId = request.subscriptionId, + deviceType = deviceService.deviceType.value, + startTime = MILLISECONDS.toSeconds(request.startTime), + idempotencyKey = request.idempotencyKey, + ), + ) + synchronized(lock) { + if (serverSessionId == null) { + // Retrying with the same idempotency key lets the backend return the session it already created. + Logging.warn("SessionRequestQueue: create session response is missing data.session_id") + retryOrDrop(request, backendResponded = true, retryAfterSeconds = null) + } else { + onCreated(request, serverSessionId) + } + } + } + + private suspend fun sendUpdate(request: SessionRequest) { + sessionsBackend.updateSession( + request.appId, + checkNotNull(request.serverSessionId), + UpdateSessionRequest( + onesignalId = request.onesignalId, + subscriptionId = request.subscriptionId, + durationSeconds = MILLISECONDS.toSeconds(request.activeDuration), + idempotencyKey = request.idempotencyKey, + endTime = request.endTime?.let { MILLISECONDS.toSeconds(it) }, + ), + ) + synchronized(lock) { + remove(request) + complete(request.id, sent = true) + } + } + + private fun onCreated( + create: SessionRequest, + serverSessionId: String, + ) { + remove(create) + complete(create.id, sent = true) + newRecordsState.add(serverSessionId) + createdSessionIds[create.localSessionId] = serverSessionId + store.list().filter { !it.isCreate && it.isSameSession(create) }.forEach { it.serverSessionId = serverSessionId } + + // Serialized with SessionService, which clears the server ID when the next session starts. + runOnSerialIO { + val session = sessionModelStore.model + if (session.sessionId == create.localSessionId) session.serverSessionId = serverSessionId + } + } + + private fun onFailure( + request: SessionRequest, + ex: BackendException, + ) { + val code = ex.statusCode + val missingId = if (request.isCreate) request.onesignalId else request.serverSessionId + val retryable = + when (NetworkUtils.getResponseStatusType(code)) { + // A record the backend just created may not be readable yet. + NetworkUtils.ResponseStatusType.MISSING -> missingId != null && newRecordsState.isInMissingRetryWindow(missingId) + // RETRYABLE also covers 4xx codes NetworkUtils doesn't list, like 413 and 422. Those won't succeed on retry. + NetworkUtils.ResponseStatusType.RETRYABLE -> code !in CLIENT_ERROR_CODES || code in RETRYABLE_CLIENT_ERROR_CODES + else -> false + } + if (!retryable) { + Logging.warn("SessionRequestQueue: $request failed with $code, dropping") + drop(request) + return + } + // A status of 0 or less means no response, e.g. offline, which keeps retrying until reconnect. + retryOrDrop(request, backendResponded = code > 0, retryAfterSeconds = ex.retryAfterSeconds) + } + + /** + * Attempts the backend answered are capped; attempts that never reached it are not. + */ + private fun retryOrDrop( + request: SessionRequest, + backendResponded: Boolean, + retryAfterSeconds: Int?, + ) { + if (backendResponded && ++request.failedAttempts >= MAX_FAILED_ATTEMPTS) { + Logging.warn("SessionRequestQueue: $request failed ${request.failedAttempts} times, dropping") + drop(request) + return + } + val attempt = (attempts[request.id] ?: 0) + 1 + attempts[request.id] = attempt + val delayMs = retryDelayMs(attempt, retryAfterSeconds, Random) + retryAt[request.id] = time.currentTimeMillis + delayMs + Logging.info("SessionRequestQueue: $request failed, retrying in $delayMs ms") + } + + /** + * Drops [request] unsent, along with any updates that could then never get a server session ID. + */ + private fun drop(request: SessionRequest) { + remove(request) + complete(request.id, sent = false) + if (!request.isCreate) return + + store.list() + .filter { !it.isCreate && it.isSameSession(request) && it.serverSessionId == null } + .forEach { orphan -> + Logging.debug("SessionRequestQueue: dropping $orphan, its session was never created") + remove(orphan) + complete(orphan.id, sent = false) + } + } + + private fun remove(request: SessionRequest) { + store.remove(request.id) + retryAt.remove(request.id) + attempts.remove(request.id) + } + + private fun complete( + requestId: String, + sent: Boolean, + ) { + waiters.remove(requestId)?.forEach { it.complete(sent) } + } + + private fun registerNetworkCallback() { + val connectivityManager = + applicationService.appContext.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager ?: return + val request = + NetworkRequest.Builder() + .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) + .build() + try { + connectivityManager.registerNetworkCallback( + request, + object : ConnectivityManager.NetworkCallback() { + override fun onAvailable(network: Network) { + // Android 15+ blocks requests from cached background apps; a blocked attempt only lengthens the backoff. + if (applicationService.isInForeground) retryNow() + } + }, + ) + } catch (e: RuntimeException) { + // Throws SecurityException on some OEM builds and when the per-app callback limit is reached. + Logging.warn("SessionRequestQueue: unable to register a network callback", e) + } + } + + companion object { + internal const val MAX_QUEUED_REQUESTS = 100 + internal const val MAX_FAILED_ATTEMPTS = 5 + + private const val BASE_RETRY_DELAY_MS = 5_000L + private const val MAX_RETRY_DELAY_MS = 300_000L + private const val NEW_RECORD_POLL_MS = 1_000L + private const val MILLIS_PER_SECOND = 1_000L + + private val CLIENT_ERROR_CODES = 400..499 + + // Request Timeout and Too Many Requests + private val RETRYABLE_CLIENT_ERROR_CODES = setOf(408, 429) + + /** + * Exponential backoff, jittered over its upper half, or the server's Retry-After when longer. + */ + internal fun retryDelayMs( + attempt: Int, + retryAfterSeconds: Int?, + random: Random, + ): Long { + val ceiling = (BASE_RETRY_DELAY_MS * 2.0.pow(attempt - 1)).toLong().coerceAtMost(MAX_RETRY_DELAY_MS) + val backoff = ceiling / 2 + random.nextLong(ceiling / 2 + 1) + val retryAfterMs = (retryAfterSeconds ?: 0) * MILLIS_PER_SECOND + return maxOf(backoff, retryAfterMs) + } + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/states/NewRecordsState.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/states/NewRecordsState.kt index cdc44a4c38..f4e1723c7f 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/states/NewRecordsState.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/operations/impl/states/NewRecordsState.kt @@ -13,19 +13,23 @@ class NewRecordsState( private val _time: ITime, private val _configModelStore: ConfigModelStore, ) { + // Shared by the operation repo and the session request queue, which run on different threads. // Key = a string id // Value = A Timestamp in ms of when the id was created private val records: MutableMap = mutableMapOf() + @Synchronized fun add(key: String) { records[key] = _time.currentTimeMillis } + @Synchronized fun canAccess(key: String): Boolean { val timeLastMovedOrCreated = records[key] ?: return true return _time.currentTimeMillis - timeLastMovedOrCreated >= _configModelStore.model.opRepoPostCreateDelay } + @Synchronized fun isInMissingRetryWindow(key: String): Boolean { val timeLastMovedOrCreated = records[key] ?: return false return _time.currentTimeMillis - timeLastMovedOrCreated <= _configModelStore.model.opRepoPostCreateRetryUpTo 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 746653d72b..7a23f96e1d 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 @@ -13,10 +13,15 @@ import com.onesignal.mocks.IOMockHelper import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService import com.onesignal.session.internal.session.impl.SessionService +import com.onesignal.session.internal.session.requests.ISessionRequestQueue +import com.onesignal.session.internal.session.requests.SessionRequest import com.onesignal.user.internal.backend.IdentityConstants import com.onesignal.user.internal.identity.IdentityModel import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.shouldBe +import io.kotest.matchers.shouldNotBe +import io.mockk.coEvery +import io.mockk.coVerify import io.mockk.every import io.mockk.just import io.mockk.mockk @@ -30,19 +35,21 @@ import io.mockk.unmockkStatic import io.mockk.verify import io.mockk.verifyOrder import kotlinx.coroutines.Job +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.test.runTest // Mocks used by every test in this file private class Mocks( var sessionsApiEnabled: Boolean = false, ) { - val currentTime = 1111L + var currentTime = 1111L var elapsedRealtime = 5000L private val mockSessionModelStore = MockHelper.sessionModelStore() val time: ITime = mockk().also { - every { it.currentTimeMillis } returns currentTime + every { it.currentTimeMillis } answers { currentTime } every { it.elapsedRealtimeMillis } answers { elapsedRealtime } } @@ -54,6 +61,13 @@ private class Mocks( val identityModelStore = MockHelper.identityModelStore { it.onesignalId = ONESIGNAL_ID } val configModelStore = MockHelper.configModelStore { it.pushSubscriptionId = SUBSCRIPTION_ID } + val enqueued = mutableListOf() + val sessionRequestQueue: ISessionRequestQueue = + mockk(relaxUnitFun = true).also { + every { it.enqueue(capture(enqueued)) } just runs + coEvery { it.enqueueAndWait(capture(enqueued)) } returns true + } + val identityChangeHandler = slot>() val configChangeHandler = slot>() @@ -96,6 +110,7 @@ private class Mocks( time, featureManager, identityModelStore, + sessionRequestQueue, ) val spyCallback = spyk() @@ -437,6 +452,303 @@ class SessionServiceTests : FunSpec({ sessionModelStore.model.onesignalId shouldBe "local-user" } + test("local ID translations are forwarded to queued session requests") { + // Given + val mocks = Mocks() + mocks.identityModelStore.model.onesignalId = "local-user" + mocks.configModelStore.model.pushSubscriptionId = "local-subscription" + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + + // When + mocks.hydrateOnesignalId("backend-user") + mocks.hydratePushSubscriptionId("backend-subscription") + + // Then + verifyOrder { + mocks.sessionRequestQueue.translateOnesignalId("local-user", "backend-user") + mocks.sessionRequestQueue.translateSubscriptionId("local-subscription", "backend-subscription", "backend-user") + } + } + + test("new session enqueues a create with the start time and pinned IDs when the flag is on") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + val sessionModelStore = mocks.sessionModelStore { it.isValid = false } + + // When + sessionService.onFocus(false) + + // Then + val create = mocks.enqueued.single() + create.appId shouldBe MockHelper.DEFAULT_APP_ID + create.localSessionId shouldBe sessionModelStore.model.sessionId + create.onesignalId shouldBe Mocks.ONESIGNAL_ID + create.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID + create.startTime shouldBe mocks.currentTime + } + + test("each new session enqueues its own create with a new idempotency key") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + sessionService.onUnfocused() + + // When + sessionService.backgroundRun() + sessionService.onFocus(false) + + // Then + val creates = mocks.enqueued.filter { it.isCreate } + creates.size shouldBe 2 + creates.map { it.localSessionId }.distinct().size shouldBe 2 + creates.map { it.idempotencyKey }.distinct().size shouldBe 2 + } + + test("resuming within the session timeout does not enqueue another create") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + + // When + sessionService.onUnfocused() + sessionService.onFocus(false) + + // Then + mocks.enqueued.size shouldBe 1 + } + + test("cold start with a persisted valid session enqueues a create for the new session") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionModelStore = + mocks.sessionModelStore { + it.isValid = true + it.sessionId = "previous-session" + } + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + + // When + sessionService.onFocus(true) + + // Then + val create = mocks.enqueued.single() + create.localSessionId shouldBe sessionModelStore.model.sessionId + sessionModelStore.model.sessionId shouldNotBe "previous-session" + } + + test("new session does not enqueue a create when the flag is off") { + // Given + val mocks = Mocks(sessionsApiEnabled = false) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + mocks.sessionModelStore { it.isValid = false } + sessionService.subscribe(mocks.spyCallback) + + // When + sessionService.onFocus(false) + + // Then + mocks.enqueued shouldBe emptyList() + verify(exactly = 1) { mocks.spyCallback.onSessionStarted() } + } + + test("new session does not enqueue a create without a push subscription ID") { + // 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.subscribe(mocks.spyCallback) + + // When + sessionService.onFocus(false) + + // Then + mocks.enqueued shouldBe emptyList() + verify(exactly = 1) { mocks.spyCallback.onSessionStarted() } + } + + test("background run sends the end with the unfocus time and final duration when the flag is on") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + val sessionModelStore = mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + mocks.elapsedRealtime += 750L + mocks.currentTime = 2222L + sessionService.onUnfocused() + mocks.currentTime = 32222L + + // When + sessionService.backgroundRun() + + // Then + val end = mocks.enqueued.filter { !it.isCreate }.single() + end.appId shouldBe MockHelper.DEFAULT_APP_ID + end.localSessionId shouldBe sessionModelStore.model.sessionId + end.serverSessionId shouldBe null + end.onesignalId shouldBe Mocks.ONESIGNAL_ID + end.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID + end.activeDuration shouldBe 750L + end.endTime shouldBe 2222L + coVerify(exactly = 1) { mocks.sessionRequestQueue.enqueueAndWait(end) } + sessionModelStore.model.isValid shouldBe false + } + + test("background run ends the session by its server session ID once created") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + val sessionModelStore = mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + sessionModelStore.model.serverSessionId = "server-session-id" + sessionService.onUnfocused() + + // When + sessionService.backgroundRun() + + // Then + val end = mocks.enqueued.filter { !it.isCreate }.single() + end.serverSessionId shouldBe "server-session-id" + } + + test("background run returns when the end is not sent in time and leaves it queued") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + coEvery { mocks.sessionRequestQueue.enqueueAndWait(any()) } coAnswers { awaitCancellation() } + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + val sessionModelStore = mocks.sessionModelStore { it.isValid = false } + sessionService.onFocus(false) + sessionService.onUnfocused() + + // When + runTest { sessionService.backgroundRun() } + + // Then + coVerify(exactly = 1) { mocks.sessionRequestQueue.enqueueAndWait(match { it.isEnd }) } + sessionModelStore.model.isValid shouldBe false + } + + test("cold start ends the previous sessions API session before creating a new one") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + mocks.sessionModelStore { + it.isValid = true + it.sessionId = "previous-session" + it.usesSessionsApi = true + it.onesignalId = Mocks.ONESIGNAL_ID + it.subscriptionId = Mocks.SUBSCRIPTION_ID + it.serverSessionId = "previous-server-session-id" + it.focusTime = 500L + it.unfocusTime = 900L + it.activeDuration = 400L + } + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + + // When + sessionService.onFocus(false) + + // Then + val end = mocks.enqueued[0] + end.localSessionId shouldBe "previous-session" + end.serverSessionId shouldBe "previous-server-session-id" + end.activeDuration shouldBe 400L + end.endTime shouldBe 900L + mocks.enqueued[1].isCreate shouldBe true + coVerify(exactly = 0) { mocks.sessionRequestQueue.enqueueAndWait(any()) } + } + + test("cold start after a kill while focused ends the previous session at its last focus") { + // Given + val mocks = Mocks(sessionsApiEnabled = true) + mocks.sessionModelStore { + it.isValid = true + it.sessionId = "previous-session" + it.usesSessionsApi = true + it.onesignalId = Mocks.ONESIGNAL_ID + it.subscriptionId = Mocks.SUBSCRIPTION_ID + it.unfocusTime = 300L + it.focusTime = 500L + } + val sessionService = mocks.sessionService + sessionService.bootstrap() + sessionService.start() + + // When + sessionService.onFocus(false) + + // Then + val end = mocks.enqueued.filter { !it.isCreate }.single() + end.endTime shouldBe 500L + } + + test("session end does not send an end 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 } + sessionService.subscribe(mocks.spyCallback) + sessionService.onFocus(false) + sessionService.onUnfocused() + mocks.sessionsApiEnabled = true + + // When + sessionService.backgroundRun() + + // Then + mocks.enqueued shouldBe emptyList() + coVerify(exactly = 0) { mocks.sessionRequestQueue.enqueueAndWait(any()) } + verify(exactly = 1) { mocks.spyCallback.onSessionEnded(any()) } + } + + test("session end does not send an end without a push subscription ID") { + // 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.subscribe(mocks.spyCallback) + sessionService.onFocus(false) + sessionService.onUnfocused() + + // When + sessionService.backgroundRun() + + // Then + mocks.enqueued shouldBe emptyList() + verify(exactly = 1) { mocks.spyCallback.onSessionEnded(any()) } + } + test("sessions API active duration uses the monotonic clock") { // Given val mocks = Mocks(sessionsApiEnabled = true) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/requests/SessionRequestQueueTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/requests/SessionRequestQueueTests.kt new file mode 100644 index 0000000000..db1d963fe1 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/requests/SessionRequestQueueTests.kt @@ -0,0 +1,407 @@ +package com.onesignal.session.internal.session.requests + +import com.onesignal.common.exceptions.BackendException +import com.onesignal.core.internal.time.ITime +import com.onesignal.debug.LogLevel +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.mocks.MockHelper +import com.onesignal.mocks.MockPreferencesService +import com.onesignal.session.internal.session.backend.CreateSessionRequest +import com.onesignal.session.internal.session.backend.ISessionsBackendService +import com.onesignal.session.internal.session.backend.UpdateSessionRequest +import com.onesignal.session.internal.session.requests.impl.SessionRequestQueue +import com.onesignal.user.internal.operations.impl.states.NewRecordsState +import io.kotest.assertions.nondeterministic.eventually +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.collections.shouldHaveSize +import io.kotest.matchers.longs.shouldBeGreaterThanOrEqual +import io.kotest.matchers.longs.shouldBeLessThanOrEqual +import io.kotest.matchers.shouldBe +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlin.random.Random +import kotlin.time.Duration.Companion.seconds + +private const val APP_ID = MockHelper.DEFAULT_APP_ID +private const val ONESIGNAL_ID = "onesignal-id" +private const val SUBSCRIPTION_ID = "subscription-id" + +private class Mocks( + val prefs: MockPreferencesService = MockPreferencesService(), +) { + val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + + var timeOffset = 0L + val time: ITime = + mockk().also { + every { it.currentTimeMillis } answers { System.currentTimeMillis() + timeOffset } + } + + val newRecordsState = NewRecordsState(time, MockHelper.configModelStore()) + val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "session-1" } + val store = SessionRequestStore(prefs) + + val creates = mutableListOf() + val updates = mutableListOf>() + val backend: ISessionsBackendService = + mockk().also { + coEvery { it.createSession(APP_ID, any()) } answers { + val request = secondArg() + creates.add(request) + "server-${request.idempotencyKey}" + } + coEvery { it.updateSession(APP_ID, any(), any()) } answers { + updates.add(secondArg() to thirdArg()) + } + } + + val queue = + SessionRequestQueue( + store, + backend, + MockHelper.deviceService(), + newRecordsState, + MockHelper.applicationService(), + sessionModelStore, + time, + ) + + fun start() = queue.also { scope.launch { it.processForever() } } +} + +private fun create( + session: String = "session-1", + onesignalId: String = ONESIGNAL_ID, + subscriptionId: String = SUBSCRIPTION_ID, +) = SessionRequest.create(APP_ID, session, onesignalId, subscriptionId, startTime = 1_000L) + +private fun update( + session: String = "session-1", + serverSessionId: String? = null, + activeDuration: Long = 1_000L, + endTime: Long? = null, + onesignalId: String = ONESIGNAL_ID, +) = SessionRequest.update(APP_ID, session, serverSessionId, onesignalId, SUBSCRIPTION_ID, activeDuration, endTime) + +private fun backendError( + statusCode: Int, + retryAfterSeconds: Int? = null, +) = BackendException(statusCode, retryAfterSeconds = retryAfterSeconds) + +class SessionRequestQueueTests : FunSpec({ + timeout = 20_000 + + // android.util.Log isn't available in unit tests. + val originalLogLevel = Logging.logLevel + beforeSpec { Logging.logLevel = LogLevel.NONE } + afterSpec { Logging.logLevel = originalLogLevel } + val scopes = mutableListOf() + + fun mocks(prefs: MockPreferencesService = MockPreferencesService()) = Mocks(prefs).also { scopes.add(it.scope) } + + // Each queue's loop runs in its mocks' scope so it can be cancelled after the test. + afterTest { + scopes.forEach { it.cancel() } + scopes.clear() + } + + test("sends a create and records the server session ID on the current session") { + val mocks = mocks() + val queue = mocks.start() + val request = create() + + queue.enqueueAndWait(request) shouldBe true + + mocks.creates.single().let { + it.onesignalId shouldBe ONESIGNAL_ID + it.subscriptionId shouldBe SUBSCRIPTION_ID + it.startTime shouldBe 1L + it.idempotencyKey shouldBe request.idempotencyKey + } + mocks.store.list() shouldBe emptyList() + eventually(5.seconds) { mocks.sessionModelStore.model.serverSessionId shouldBe "server-${request.idempotencyKey}" } + } + + test("an update waits for its session's create, then uses the server session ID") { + val mocks = mocks() + val create = create() + val end = update(endTime = 5_000L) + mocks.queue.enqueue(create) + mocks.queue.enqueue(end) + + mocks.start() + + eventually(5.seconds) { mocks.updates shouldHaveSize 1 } + mocks.updates.single().let { (sessionId, request) -> + sessionId shouldBe "server-${create.idempotencyKey}" + request.endTime shouldBe 5L + request.idempotencyKey shouldBe end.idempotencyKey + } + } + + test("an update built after its create succeeded still gets the server session ID") { + val mocks = mocks() + val queue = mocks.start() + val create = create() + queue.enqueueAndWait(create) shouldBe true + + queue.enqueueAndWait(update(serverSessionId = null)) shouldBe true + + mocks.updates.single().first shouldBe "server-${create.idempotencyKey}" + } + + test("a create waits for backend user and subscription IDs") { + val mocks = mocks() + val queue = mocks.start() + val sent = mocks.scope.async { queue.enqueueAndWait(create(onesignalId = "local-user", subscriptionId = "local-sub")) } + + delay(200) + mocks.creates shouldBe emptyList() + + queue.translateOnesignalId("local-user", "backend-user") + queue.translateSubscriptionId("local-sub", "backend-sub", "backend-user") + + sent.await() shouldBe true + mocks.creates.single().let { + it.onesignalId shouldBe "backend-user" + it.subscriptionId shouldBe "backend-sub" + } + } + + test("a subscription ID is only translated for requests pinned to the user it was created for") { + val mocks = mocks() + val request = create(onesignalId = "other-user", subscriptionId = "local-sub") + mocks.queue.enqueue(request) + + mocks.queue.translateSubscriptionId("local-sub", "backend-sub", ONESIGNAL_ID) + + request.subscriptionId shouldBe "local-sub" + } + + test("unsent updates for a session coalesce into the latest with the highest duration") { + val mocks = mocks() + val queue = mocks.queue + queue.enqueue(create(onesignalId = "local-user")) + queue.enqueue(update(activeDuration = 3_000L)) + queue.enqueue(update(activeDuration = 2_000L)) + + val updates = mocks.store.list().filter { !it.isCreate } + updates shouldHaveSize 1 + updates.single().activeDuration shouldBe 3_000L + } + + test("an end replaces unsent updates and later updates for the session are discarded") { + val mocks = mocks() + val queue = mocks.queue + queue.enqueue(create(onesignalId = "local-user")) + queue.enqueue(update(activeDuration = 1_000L)) + val end = update(activeDuration = 2_000L, endTime = 5_000L) + queue.enqueue(end) + + queue.enqueueAndWait(update(activeDuration = 9_000L)) shouldBe false + + mocks.store.list().filter { !it.isCreate } shouldBe listOf(end) + } + + test("updates for other sessions are not coalesced") { + val mocks = mocks() + val queue = mocks.queue + queue.enqueue(update(session = "session-1", serverSessionId = "server-1", onesignalId = "local-user")) + queue.enqueue(update(session = "session-2", serverSessionId = "server-2", onesignalId = "local-user")) + + mocks.store.list() shouldHaveSize 2 + } + + test("an update for a session that was never created is discarded") { + val mocks = mocks() + + mocks.start().enqueueAndWait(update(serverSessionId = null)) shouldBe false + + mocks.store.list() shouldBe emptyList() + } + + test("the oldest request is dropped past the cap, along with its session's updates") { + val mocks = mocks() + val queue = mocks.queue + queue.enqueue(create(session = "oldest", onesignalId = "local-user")) + queue.enqueue(update(session = "oldest", onesignalId = "local-user")) + + repeat(SessionRequestQueue.MAX_QUEUED_REQUESTS - 1) { queue.enqueue(create(session = "session-$it", onesignalId = "local-user")) } + + val requests = mocks.store.list() + requests shouldHaveSize SessionRequestQueue.MAX_QUEUED_REQUESTS - 1 + requests.none { it.localSessionId == "oldest" } shouldBe true + } + + test("a failed request retries with the same idempotency key") { + val mocks = mocks() + val keys = mutableListOf() + var calls = 0 + coEvery { mocks.backend.createSession(APP_ID, any()) } answers { + keys.add(secondArg().idempotencyKey) + if (++calls == 1) throw backendError(500) else "server-id" + } + val queue = mocks.start() + val request = create() + val sent = mocks.scope.async { queue.enqueueAndWait(request) } + + eventually(5.seconds) { request.failedAttempts shouldBe 1 } + queue.onFocus(false) + + sent.await() shouldBe true + keys shouldBe listOf(request.idempotencyKey, request.idempotencyKey) + } + + test("a request backing off does not block other sessions") { + val mocks = mocks() + coEvery { mocks.backend.createSession(APP_ID, match { it.onesignalId == "failing-user" }) } throws backendError(503) + val queue = mocks.start() + val failing = create(session = "failing", onesignalId = "failing-user") + queue.enqueue(failing) + eventually(5.seconds) { failing.failedAttempts shouldBe 1 } + + queue.enqueueAndWait(create(session = "healthy")) shouldBe true + + mocks.store.list() shouldBe listOf(failing) + } + + test("a request is dropped after the maximum failed attempts, along with its session's updates") { + val mocks = mocks() + coEvery { mocks.backend.createSession(APP_ID, any()) } throws backendError(500) + val queue = mocks.start() + val create = create() + val createSent = mocks.scope.async { queue.enqueueAndWait(create) } + eventually(5.seconds) { mocks.store.list() shouldHaveSize 1 } + val updateSent = mocks.scope.async { queue.enqueueAndWait(update()) } + + repeat(SessionRequestQueue.MAX_FAILED_ATTEMPTS - 1) { attempt -> + eventually(5.seconds) { create.failedAttempts shouldBe attempt + 1 } + queue.onFocus(false) + } + + createSent.await() shouldBe false + updateSent.await() shouldBe false + mocks.store.list() shouldBe emptyList() + coVerify(exactly = SessionRequestQueue.MAX_FAILED_ATTEMPTS) { mocks.backend.createSession(any(), any()) } + } + + test("a non-retryable client error drops the request") { + val mocks = mocks() + coEvery { mocks.backend.createSession(APP_ID, any()) } throws backendError(400) + + mocks.start().enqueueAndWait(create()) shouldBe false + + mocks.store.list() shouldBe emptyList() + } + + test("a request that never reached the backend keeps retrying without counting an attempt") { + val mocks = mocks() + var calls = 0 + coEvery { mocks.backend.createSession(APP_ID, any()) } answers { + if (++calls <= SessionRequestQueue.MAX_FAILED_ATTEMPTS) throw backendError(0) else "server-id" + } + val queue = mocks.start() + val request = create() + val sent = mocks.scope.async { queue.enqueueAndWait(request) } + + repeat(SessionRequestQueue.MAX_FAILED_ATTEMPTS) { attempt -> + eventually(5.seconds) { calls shouldBe attempt + 1 } + delay(50) + queue.onFocus(false) + } + + sent.await() shouldBe true + request.failedAttempts shouldBe 0 + } + + test("a 404 for a just-created user is retried") { + val mocks = mocks() + var calls = 0 + coEvery { mocks.backend.createSession(APP_ID, any()) } answers { + if (++calls == 1) throw backendError(404) else "server-id" + } + val configModelStore = MockHelper.configModelStore { it.opRepoPostCreateRetryUpTo = 60_000L } + val newRecordsState = NewRecordsState(mocks.time, configModelStore).also { it.add(ONESIGNAL_ID) } + val queue = + SessionRequestQueue( + mocks.store, + mocks.backend, + MockHelper.deviceService(), + newRecordsState, + MockHelper.applicationService(), + mocks.sessionModelStore, + mocks.time, + ).also { queue -> mocks.scope.launch { queue.processForever() } } + val request = create() + val sent = mocks.scope.async { queue.enqueueAndWait(request) } + + eventually(5.seconds) { request.failedAttempts shouldBe 1 } + queue.onFocus(false) + + sent.await() shouldBe true + } + + test("a request killed mid-send is replayed on the next launch with the same idempotency key") { + val prefs = MockPreferencesService() + val first = mocks(prefs) + val inFlight = CompletableDeferred() + coEvery { first.backend.createSession(APP_ID, any()) } coAnswers { + inFlight.complete(Unit) + awaitCancellation() + } + val request = create() + first.start().enqueue(request) + inFlight.await() + first.scope.cancel() + + val relaunched = mocks(prefs) + relaunched.store.list().single().idempotencyKey shouldBe request.idempotencyKey + relaunched.start() + + eventually(5.seconds) { relaunched.creates shouldHaveSize 1 } + relaunched.creates.single().idempotencyKey shouldBe request.idempotencyKey + } + + test("enqueueing does not wait on a request in flight") { + val mocks = mocks() + val inFlight = CompletableDeferred() + coEvery { mocks.backend.createSession(APP_ID, any()) } coAnswers { + inFlight.complete(Unit) + awaitCancellation() + } + val queue = mocks.start() + queue.enqueue(create(session = "stuck")) + inFlight.await() + + queue.enqueue(create(session = "next")) + + mocks.store.list().map { it.localSessionId } shouldBe listOf("stuck", "next") + } + + test("retry delay backs off exponentially within bounds and honors Retry-After") { + val random = Random(0) + repeat(20) { + SessionRequestQueue.retryDelayMs(1, null, random).let { + it shouldBeGreaterThanOrEqual 2_500L + it shouldBeLessThanOrEqual 5_000L + } + SessionRequestQueue.retryDelayMs(3, null, random).let { + it shouldBeGreaterThanOrEqual 10_000L + it shouldBeLessThanOrEqual 20_000L + } + SessionRequestQueue.retryDelayMs(30, null, random) shouldBeLessThanOrEqual 5 * 60_000L + } + SessionRequestQueue.retryDelayMs(1, 120, random) shouldBe 120_000L + } +})