From 8556e1a9b14df9183f6eb7d447b6ba0e3c2e68b2 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Wed, 14 Oct 2026 16:23:35 -0700 Subject: [PATCH 1/8] feat: [SDK-5378] persist and retry sessions API requests through OperationRepo Co-Authored-By: Claude Opus 5.5 --- .../operations/impl/OperationModelStore.kt | 5 + .../internal/operations/impl/OperationRepo.kt | 34 ++- .../operations/impl/SessionQueuePolicy.kt | 103 ++++++++ .../com/onesignal/session/SessionModule.kt | 3 + .../session/internal/session/SessionModel.kt | 10 + .../internal/session/impl/SessionService.kt | 1 + .../operations/CreateSessionOperation.kt | 82 ++++++ .../operations/UpdateSessionOperation.kt | 95 +++++++ .../impl/SessionOperationExecutor.kt | 110 ++++++++ .../operations/SessionOperationQueueTests.kt | 241 ++++++++++++++++++ .../internal/session/SessionServiceTests.kt | 5 + .../SessionOperationExecutorTests.kt | 152 +++++++++++ 12 files changed, 835 insertions(+), 6 deletions(-) create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt create mode 100644 OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt create mode 100644 OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationModelStore.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationModelStore.kt index c3d6e45910..6235e6e324 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationModelStore.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationModelStore.kt @@ -4,6 +4,9 @@ import com.onesignal.common.modeling.ModelStore import com.onesignal.core.internal.operations.Operation import com.onesignal.core.internal.preferences.IPreferencesService import com.onesignal.debug.internal.logging.Logging +import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.UpdateSessionOperation +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor import com.onesignal.user.internal.operations.CreateSubscriptionOperation import com.onesignal.user.internal.operations.DeleteAliasOperation import com.onesignal.user.internal.operations.DeleteSubscriptionOperation @@ -63,6 +66,8 @@ internal class OperationModelStore(prefs: IPreferencesService) : ModelStore TrackSessionEndOperation() UpdateUserOperationExecutor.TRACK_PURCHASE -> TrackPurchaseOperation() CustomEventOperationExecutor.CUSTOM_EVENT -> TrackCustomEventOperation() + SessionOperationExecutor.CREATE_SESSION -> CreateSessionOperation() + SessionOperationExecutor.UPDATE_SESSION -> UpdateSessionOperation() else -> throw Exception("Unrecognized operation: $operationName") } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt index 46b30a452a..46efcd548d 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt @@ -107,6 +107,7 @@ internal class OperationRepo( private val inFlightLogins = mutableMapOf() private val waiter = WaiterWithValue() private var paused = false + private val sessionQueuePolicy = SessionQueuePolicy() // Guarded by queue. Set by a force that lands while ops are out of the queue executing. private var forcedDuringExecute = false @@ -187,7 +188,7 @@ internal class OperationRepo( operation.id = UUID.randomUUID().toString() scope.launch { - internalEnqueue(OperationQueueItem(operation, bucket = enqueueIntoBucket), flush, true) + enqueueNew(OperationQueueItem(operation, bucket = enqueueIntoBucket), flush) } } @@ -202,7 +203,7 @@ internal class OperationRepo( operation.id = UUID.randomUUID().toString() val waiter = WaiterWithValue() scope.launch { - internalEnqueue(OperationQueueItem(operation, waiter, bucket = enqueueIntoBucket), flush, true) + enqueueNew(OperationQueueItem(operation, waiter, bucket = enqueueIntoBucket), flush) } val timeoutMs = _configModelStore.model.opRepoAwaitTimeout return withTimeoutOrNull(timeoutMs) { @@ -218,12 +219,13 @@ internal class OperationRepo( * they cannot be authenticated and would otherwise sit in the queue forever, blocked by * `hasValidJwtIfRequired`. LoginUserOperation is exempt because it's enqueued * intentionally during logout and purged later by [removeOperationsWithoutExternalId] - * if needed. Outer-gated on `newCodePathsRun` so Phase 1 customers stay byte-for-byte + * if needed. Ops with `requiresJwt = false` are exempt since they can still execute. + * Outer-gated on `newCodePathsRun` so Phase 1 customers stay byte-for-byte * on the legacy enqueue path. */ private fun shouldSuppressAnonymousOp(op: Operation): Boolean { if (!_identityVerificationService.newCodePathsRun) return false - if (op is LoginUserOperation) return false + if (op is LoginUserOperation || !op.requiresJwt) return false val suppress = _configModelStore.model.useIdentityVerification == JwtRequirement.REQUIRED && op.externalId == null @@ -233,6 +235,23 @@ internal class OperationRepo( return suppress } + private fun enqueueNew( + queueItem: OperationQueueItem, + flush: Boolean, + ) { + val accepted = + synchronized(queue) { + val result = sessionQueuePolicy.onEnqueue(queue, queueItem) + dropAndWake(result.removed) + result.accept + } + if (accepted) { + internalEnqueue(queueItem, flush, true) + } else { + queueItem.wakeWaiters(OperationWaitResult(false)) + } + } + /** * Only used inside this class, adds OperationQueueItem to queue * WARNING: Never set flush=true until budget rules are added, even for internal use! @@ -353,14 +372,14 @@ internal class OperationRepo( } /** - * Drops queued operations whose externalId is null. Called by the IV-aware HYDRATE + * Drops queued operations whose externalId is null and that require a JWT. Called by the IV-aware HYDRATE * choreography in [OperationRepoIvExtensions] when `jwt_required` becomes REQUIRED * to evict anon ops that can no longer execute. */ internal fun removeOperationsWithoutExternalId() { val removedIds: List = synchronized(queue) { - val anonymous = queue.filter { it.operation.externalId == null } + val anonymous = queue.filter { it.operation.externalId == null && it.operation.requiresJwt } anonymous.forEach { it.wakeWaiters(OperationWaitResult(false)) } queue.removeAll(anonymous) // IV=ON never transfers anonymous state; clear existingOnesignalId so the @@ -454,6 +473,7 @@ internal class OperationRepo( ops.forEach { it.operation.translateIds(response.idTranslations) } synchronized(queue) { queue.forEach { it.operation.translateIds(response.idTranslations) } + sessionQueuePolicy.recordTranslations(response.idTranslations) } response.idTranslations.values.forEach { _newRecordState.add(it) } } @@ -582,6 +602,8 @@ internal class OperationRepo( ) { ops.forEach { _operationModelStore.remove(it.operation.id) } ops.forEach { it.wakeWaiters(response.toWaitResult(false)) } + val orphans = synchronized(queue) { ops.flatMap { sessionQueuePolicy.onDropped(queue, it.operation) } } + if (orphans.isNotEmpty()) dropAndWake(orphans) } /** 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 new file mode 100644 index 0000000000..73f846f42d --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt @@ -0,0 +1,103 @@ +package com.onesignal.core.internal.operations.impl + +import com.onesignal.core.internal.operations.Operation +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.UpdateSessionOperation + +/** + * Queue rules for sessions API operations in [OperationRepo]. Every function must be called while + * holding the queue lock. Functions return the queue items they removed; the caller drops them + * from the store and wakes their waiters. + */ +internal class SessionQueuePolicy( + private val maxQueued: Int = MAX_QUEUED_SESSION_OPERATIONS, +) { + // Recent local -> backend ID translations, for updates built from a local ID just before the + // create succeeded but enqueued after the queue was translated. + private val recentTranslations = + object : LinkedHashMap() { + override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_RECENT_TRANSLATIONS + } + + fun recordTranslations(idTranslations: Map) { + recentTranslations.putAll(idTranslations) + } + + class EnqueueResult( + val accept: Boolean, + val removed: List = emptyList(), + ) + + fun onEnqueue( + queue: MutableList, + incoming: OperationRepo.OperationQueueItem, + ): EnqueueResult { + val op = incoming.operation + if (op.isSessionOperation) op.translateIds(recentTranslations) + + if (op is UpdateSessionOperation && hasQueuedEnd(queue, op.sessionId)) { + Logging.debug("SessionQueuePolicy: session ${op.sessionId} already has a queued end, dropping $op") + return EnqueueResult(accept = false) + } + + val removed = mutableListOf() + if (op is UpdateSessionOperation) { + val unsent = queue.filter { (it.operation as? UpdateSessionOperation)?.sessionId == op.sessionId } + // Durations are cumulative, so the newest update carries everything the unsent ones would. + unsent.forEach { + val replaced = it.operation as UpdateSessionOperation + op.activeDuration = maxOf(op.activeDuration, replaced.activeDuration) + incoming.waiters.addAll(it.waiters) + it.waiters.clear() + } + queue.removeAll(unsent) + removed.addAll(unsent) + } + + var accept = true + while (queue.count { it.operation.isSessionOperation } + (if (accept) 1 else 0) > maxQueued) { + val oldestSession = queue.first { it.operation.isSessionOperation }.operation.sessionKey + val dropped = queue.filter { it.operation.sessionKey == oldestSession } + Logging.warn("SessionQueuePolicy: more than $maxQueued sessions API operations queued, dropping session $oldestSession") + queue.removeAll(dropped) + removed.addAll(dropped) + if (op.sessionKey == oldestSession) accept = false + } + return EnqueueResult(accept, removed) + } + + /** Removes queued updates that can never send because [dropped] was their session's create. */ + fun onDropped( + queue: MutableList, + dropped: Operation, + ): List { + if (dropped !is CreateSessionOperation) return emptyList() + val orphans = queue.filter { (it.operation as? UpdateSessionOperation)?.sessionId == dropped.localSessionId } + if (orphans.isNotEmpty()) { + Logging.debug("SessionQueuePolicy: create for session ${dropped.localSessionId} was dropped, dropping ${orphans.size} updates") + queue.removeAll(orphans) + } + return orphans + } + + private fun hasQueuedEnd( + queue: List, + sessionId: String, + ) = queue.any { (it.operation as? UpdateSessionOperation)?.let { update -> update.sessionId == sessionId && update.isEnd } == true } + + private val Operation.isSessionOperation get() = this is CreateSessionOperation || this is UpdateSessionOperation + + private val Operation.sessionKey: String? + get() = + when (this) { + is CreateSessionOperation -> localSessionId + is UpdateSessionOperation -> sessionId + else -> null + } + + companion object { + const val MAX_QUEUED_SESSION_OPERATIONS = 100 + private const val MAX_RECENT_TRANSLATIONS = 50 + } +} 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..dcee3dc238 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 @@ -3,6 +3,7 @@ package com.onesignal.session import com.onesignal.common.modules.IModule import com.onesignal.common.services.ServiceBuilder import com.onesignal.core.internal.background.IBackgroundService +import com.onesignal.core.internal.operations.IOperationExecutor import com.onesignal.core.internal.startup.IBootstrapService import com.onesignal.core.internal.startup.IStartableService import com.onesignal.session.internal.SessionManager @@ -22,6 +23,7 @@ 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.operations.impl.SessionOperationExecutor internal class SessionModule : IModule { override fun register(builder: ServiceBuilder) { @@ -46,5 +48,6 @@ internal class SessionModule : IModule { builder.register().provides() builder.register().provides() builder.register().provides() + builder.register().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..3c44417549 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 @@ -96,6 +96,16 @@ class SessionModel : Model() { setOptStringProperty(::subscriptionId.name, value) } + /** + * Identifies this session's queued sessions API requests until [serverSessionId] is known. + * A new local ID for each session. + */ + var localSessionId: String + get() = getStringProperty(::localSessionId.name) { "" } + set(value) { + setStringProperty(::localSessionId.name, value) + } + /** * Null until the sessions API creates this session. Cleared when a new session starts. */ 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..f5b8ba0a99 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 @@ -197,6 +197,7 @@ internal class SessionService( .takeIf { it.hasProperty(IdentityConstants.ONESIGNAL_ID) } ?.onesignalId session.subscriptionId = config?.pushSubscriptionId + session.localSessionId = IDManager.createLocalId() session.serverSessionId = null session.isValid = true Logging.debug("SessionService: New session started at ${session.startTime}") diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt new file mode 100644 index 0000000000..b8cd44dca7 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt @@ -0,0 +1,82 @@ +package com.onesignal.session.internal.session.operations + +import com.onesignal.common.IDManager +import com.onesignal.core.internal.operations.GroupComparisonType +import com.onesignal.core.internal.operations.Operation +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor +import java.util.UUID + +/** + * An [Operation] to create a session through the sessions API. + */ +internal class CreateSessionOperation() : Operation(SessionOperationExecutor.CREATE_SESSION) { + var appId: String + get() = getStringProperty(::appId.name) + private set(value) { + setStringProperty(::appId.name, value) + } + + /** + * Stands in for the backend session ID until the create succeeds. Updates queued before then + * reference it and are translated to the backend ID. + */ + var localSessionId: String + get() = getStringProperty(::localSessionId.name) + private set(value) { + setStringProperty(::localSessionId.name, value) + } + + var onesignalId: String + get() = getStringProperty(::onesignalId.name) + private set(value) { + setStringProperty(::onesignalId.name, value) + } + + var subscriptionId: String + get() = getStringProperty(::subscriptionId.name) + private set(value) { + setStringProperty(::subscriptionId.name, value) + } + + /** Unix time in milliseconds. */ + var startTime: Long + get() = getLongProperty(::startTime.name) + private set(value) { + setLongProperty(::startTime.name, value) + } + + /** Persisted so a retry after a restart is recognized by the backend as the same request. */ + var idempotencyKey: String + get() = getStringProperty(::idempotencyKey.name) + private set(value) { + setStringProperty(::idempotencyKey.name, value) + } + + override val requiresJwt: Boolean get() = false + override val createComparisonKey: String get() = "" + override val modifyComparisonKey: String get() = "" + override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE + override val canStartExecute: Boolean + get() = !IDManager.isLocalId(onesignalId) && !IDManager.isLocalId(subscriptionId) + override val applyToRecordId: String get() = localSessionId + + constructor( + appId: String, + localSessionId: String, + onesignalId: String, + subscriptionId: String, + startTime: Long, + ) : this() { + this.appId = appId + this.localSessionId = localSessionId + this.onesignalId = onesignalId + this.subscriptionId = subscriptionId + this.startTime = startTime + this.idempotencyKey = UUID.randomUUID().toString() + } + + override fun translateIds(map: Map) { + map[onesignalId]?.let { onesignalId = it } + map[subscriptionId]?.let { subscriptionId = it } + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt new file mode 100644 index 0000000000..767b43644f --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt @@ -0,0 +1,95 @@ +package com.onesignal.session.internal.session.operations + +import com.onesignal.common.IDManager +import com.onesignal.core.internal.operations.GroupComparisonType +import com.onesignal.core.internal.operations.Operation +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor +import java.util.UUID + +/** + * An [Operation] to report a session's cumulative duration through the sessions API, and to end + * the session when [endTime] is set. + */ +internal class UpdateSessionOperation() : Operation(SessionOperationExecutor.UPDATE_SESSION) { + var appId: String + get() = getStringProperty(::appId.name) + private set(value) { + setStringProperty(::appId.name, value) + } + + /** + * The backend session ID, or the [CreateSessionOperation.localSessionId] while that create is + * still queued. + */ + var sessionId: String + get() = getStringProperty(::sessionId.name) + private set(value) { + setStringProperty(::sessionId.name, value) + } + + var onesignalId: String + get() = getStringProperty(::onesignalId.name) + private set(value) { + setStringProperty(::onesignalId.name, value) + } + + var subscriptionId: String + get() = getStringProperty(::subscriptionId.name) + private set(value) { + setStringProperty(::subscriptionId.name, value) + } + + /** Cumulative active time of the session, in milliseconds. */ + var activeDuration: Long + get() = getLongProperty(::activeDuration.name) + internal set(value) { + setLongProperty(::activeDuration.name, value) + } + + /** Unix time in milliseconds. Null for a duration-only update. */ + var endTime: Long? + get() = getOptLongProperty(::endTime.name) + private set(value) { + setOptLongProperty(::endTime.name, value) + } + + /** Persisted so a retry after a restart is recognized by the backend as the same request. */ + var idempotencyKey: String + get() = getStringProperty(::idempotencyKey.name) + private set(value) { + setStringProperty(::idempotencyKey.name, value) + } + + val isEnd: Boolean get() = endTime != null + + override val requiresJwt: Boolean get() = false + override val createComparisonKey: String get() = "" + override val modifyComparisonKey: String get() = "" + override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE + override val canStartExecute: Boolean + get() = !IDManager.isLocalId(sessionId) && !IDManager.isLocalId(onesignalId) && !IDManager.isLocalId(subscriptionId) + override val applyToRecordId: String get() = sessionId + + constructor( + appId: String, + sessionId: String, + onesignalId: String, + subscriptionId: String, + activeDuration: Long, + endTime: Long? = null, + ) : this() { + this.appId = appId + this.sessionId = sessionId + this.onesignalId = onesignalId + this.subscriptionId = subscriptionId + this.activeDuration = activeDuration + this.endTime = endTime + this.idempotencyKey = UUID.randomUUID().toString() + } + + override fun translateIds(map: Map) { + map[sessionId]?.let { sessionId = it } + map[onesignalId]?.let { onesignalId = it } + map[subscriptionId]?.let { subscriptionId = it } + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt new file mode 100644 index 0000000000..ff0f43d161 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt @@ -0,0 +1,110 @@ +package com.onesignal.session.internal.session.operations.impl + +import com.onesignal.common.exceptions.BackendException +import com.onesignal.core.internal.device.IDeviceService +import com.onesignal.core.internal.operations.ExecutionResponse +import com.onesignal.core.internal.operations.ExecutionResult +import com.onesignal.core.internal.operations.IOperationExecutor +import com.onesignal.core.internal.operations.Operation +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.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.UpdateSessionOperation +import com.onesignal.user.internal.operations.impl.states.NewRecordsState +import java.util.concurrent.TimeUnit + +internal class SessionOperationExecutor( + private val sessionsBackend: ISessionsBackendService, + private val deviceService: IDeviceService, + private val sessionModelStore: SessionModelStore, + private val newRecordState: NewRecordsState, +) : IOperationExecutor { + override val operations: List + get() = listOf(CREATE_SESSION, UPDATE_SESSION) + + override suspend fun execute(operations: List): ExecutionResponse = + when (val operation = operations.single()) { + is CreateSessionOperation -> createSession(operation) + is UpdateSessionOperation -> updateSession(operation) + else -> error("Unrecognized operation: $operation") + } + + private suspend fun createSession(operation: CreateSessionOperation): ExecutionResponse { + val sessionId = + try { + sessionsBackend.createSession( + operation.appId, + CreateSessionRequest( + onesignalId = operation.onesignalId, + subscriptionId = operation.subscriptionId, + deviceType = deviceService.deviceType.value, + startTime = TimeUnit.MILLISECONDS.toSeconds(operation.startTime), + idempotencyKey = operation.idempotencyKey, + ), + ) + } catch (ex: BackendException) { + return failure(ex, retryMissing = false) + } + + return if (sessionId == null) { + Logging.warn("SessionOperationExecutor: create session response has no session_id, dropping $operation") + ExecutionResponse(ExecutionResult.FAIL_NORETRY) + } else { + // Updates enqueued from here on read the backend ID from the model; queued ones are translated. + val session = sessionModelStore.model + if (session.localSessionId == operation.localSessionId) session.serverSessionId = sessionId + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(operation.localSessionId to sessionId)) + } + } + + private suspend fun updateSession(operation: UpdateSessionOperation): ExecutionResponse { + try { + sessionsBackend.updateSession( + operation.appId, + operation.sessionId, + UpdateSessionRequest( + onesignalId = operation.onesignalId, + subscriptionId = operation.subscriptionId, + durationSeconds = TimeUnit.MILLISECONDS.toSeconds(operation.activeDuration), + idempotencyKey = operation.idempotencyKey, + endTime = operation.endTime?.let { TimeUnit.MILLISECONDS.toSeconds(it) }, + ), + ) + } catch (ex: BackendException) { + // The create's 202 is processed asynchronously, so a fresh session can briefly 404. + return failure(ex, retryMissing = newRecordState.isInMissingRetryWindow(operation.sessionId)) + } + return ExecutionResponse(ExecutionResult.SUCCESS) + } + + private fun failure( + ex: BackendException, + retryMissing: Boolean, + ): ExecutionResponse { + val code = ex.statusCode + val retryable = + code == 0 || + code == HTTP_REQUEST_TIMEOUT || + code == HTTP_TOO_MANY_REQUESTS || + code >= HTTP_SERVER_ERROR || + (code == HTTP_NOT_FOUND && retryMissing) + return if (retryable) { + ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = ex.retryAfterSeconds) + } else { + ExecutionResponse(ExecutionResult.FAIL_NORETRY, httpStatusCode = code, httpResponse = ex.response) + } + } + + companion object { + const val CREATE_SESSION = "create-session" + const val UPDATE_SESSION = "update-session" + + private const val HTTP_NOT_FOUND = 404 + private const val HTTP_REQUEST_TIMEOUT = 408 + private const val HTTP_TOO_MANY_REQUESTS = 429 + private const val HTTP_SERVER_ERROR = 500 + } +} 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 new file mode 100644 index 0000000000..802c296cf4 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt @@ -0,0 +1,241 @@ +package com.onesignal.core.internal.operations + +import com.onesignal.core.internal.operations.impl.OperationModelStore +import com.onesignal.core.internal.operations.impl.OperationRepo +import com.onesignal.core.internal.operations.impl.SessionQueuePolicy +import com.onesignal.debug.LogLevel +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.mocks.CoreInternalMocks +import com.onesignal.mocks.MockHelper +import com.onesignal.mocks.MockPreferencesService +import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.UpdateSessionOperation +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor +import com.onesignal.user.internal.jwt.JwtRequirement +import com.onesignal.user.internal.jwt.JwtTokenStore +import com.onesignal.user.internal.operations.ExecutorMocks.Companion.getNewRecordState +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.collections.shouldContainExactly +import io.kotest.matchers.shouldBe +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import io.mockk.spyk +import io.mockk.verify +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.delay +import kotlinx.coroutines.withTimeout + +private class SessionQueueMocks( + val prefs: MockPreferencesService = MockPreferencesService(), +) { + val configModelStore = + MockHelper.configModelStore { + it.isInitializedWithRemote = true + it.useIdentityVerification = JwtRequirement.NOT_REQUIRED + } + + val operationModelStore = OperationModelStore(prefs) + + val executed = mutableListOf() + var respond: (Operation) -> ExecutionResponse = { ExecutionResponse(ExecutionResult.SUCCESS) } + + val executor: IOperationExecutor = + mockk().also { + every { it.operations } returns listOf(SessionOperationExecutor.CREATE_SESSION, SessionOperationExecutor.UPDATE_SESSION) + coEvery { it.execute(any()) } answers { + val op = firstArg>().single() + executed.add(op) + respond(op) + } + } + + var identityVerificationService = CoreInternalMocks.identityVerificationService() + + val operationRepo: OperationRepo by lazy { + spyk( + OperationRepo( + listOf(executor), + operationModelStore, + configModelStore, + MockHelper.systemTime(), + getNewRecordState(configModelStore), + JwtTokenStore(MockPreferencesService()), + identityVerificationService, + ), + recordPrivateCalls = true, + ) + } + + fun queued() = synchronized(operationRepo.queue) { operationRepo.queue.map { it.operation } } + + fun enqueueAll(vararg ops: Operation) { + ops.forEach { operationRepo.enqueue(it) } + verify(timeout = 1_000, exactly = ops.size) { + operationRepo["enqueueNew"](any(), any()) + } + } +} + +private fun create( + localSessionId: String = "local-s1", + onesignalId: String = "onesignalId", +) = CreateSessionOperation("appId", localSessionId, onesignalId, "subscriptionId", startTime = 1_000L) + +private fun update( + sessionId: String = "local-s1", + activeDuration: Long = 1_000L, + endTime: Long? = null, +) = UpdateSessionOperation("appId", sessionId, "onesignalId", "subscriptionId", activeDuration, endTime) + +private suspend fun awaitExecuted( + mocks: SessionQueueMocks, + count: Int, +) = withTimeout(5_000) { + while (mocks.executed.size < count) delay(10) +} + +class SessionOperationQueueTests : FunSpec({ + beforeAny { + Logging.logLevel = LogLevel.NONE + } + + test("duration-only updates collapse to the highest duration") { + val mocks = SessionQueueMocks() + + mocks.enqueueAll(update(activeDuration = 2_000L), update(activeDuration = 5_000L), update(activeDuration = 4_000L)) + + val queued = mocks.queued().single() as UpdateSessionOperation + queued.activeDuration shouldBe 5_000L + mocks.operationModelStore.list().map { it.id } shouldContainExactly listOf(queued.id) + } + + test("an end replaces an unsent update and keeps its duration") { + val mocks = SessionQueueMocks() + + mocks.enqueueAll(update(activeDuration = 5_000L), update(activeDuration = 3_000L, endTime = 9L)) + + val queued = mocks.queued().single() as UpdateSessionOperation + queued.endTime shouldBe 9L + queued.activeDuration shouldBe 5_000L + } + + test("an update after a queued end is dropped") { + val mocks = SessionQueueMocks() + + mocks.enqueueAll(update(endTime = 9L), update(activeDuration = 8_000L)) + + val queued = mocks.queued().single() as UpdateSessionOperation + queued.endTime shouldBe 9L + } + + test("updates for different sessions are not coalesced") { + val mocks = SessionQueueMocks() + + mocks.enqueueAll(update(sessionId = "s1"), update(sessionId = "s2")) + + mocks.queued().size shouldBe 2 + } + + test("cap drops the oldest session with all its operations") { + val mocks = SessionQueueMocks() + val max = SessionQueuePolicy.MAX_QUEUED_SESSION_OPERATIONS + val others = (1 until max).map { create(localSessionId = "local-other-$it") } + + mocks.enqueueAll(create(), update(), *others.toTypedArray()) + + mocks.queued() shouldContainExactly others + mocks.operationModelStore.list().size shouldBe others.size + } + + test("an update waits for its create and is sent with the backend session ID") { + val mocks = SessionQueueMocks() + mocks.respond = { op -> + if (op is CreateSessionOperation) { + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to "server-s1")) + } else { + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + + mocks.enqueueAll(create(), update()) + mocks.operationRepo.start() + awaitExecuted(mocks, 2) + + mocks.executed[0].shouldBeCreate() + (mocks.executed[1] as UpdateSessionOperation).sessionId shouldBe "server-s1" + } + + test("an update built from the local ID after its create succeeded is still sent") { + val mocks = SessionQueueMocks() + val created = CompletableDeferred() + mocks.respond = { op -> + if (op is CreateSessionOperation) { + created.complete(Unit) + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to "server-s1")) + } else { + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + mocks.operationRepo.start() + mocks.operationRepo.enqueue(create()) + created.await() + awaitExecuted(mocks, 1) + + mocks.operationRepo.enqueue(update()) + awaitExecuted(mocks, 2) + + (mocks.executed[1] as UpdateSessionOperation).sessionId shouldBe "server-s1" + } + + test("a dropped create drops its queued updates") { + val mocks = SessionQueueMocks() + mocks.respond = { ExecutionResponse(ExecutionResult.FAIL_NORETRY) } + + mocks.enqueueAll(create(), update(), update(sessionId = "server-other")) + mocks.operationRepo.start() + awaitExecuted(mocks, 2) + + mocks.executed.map { it.applyToRecordId } shouldContainExactly listOf("local-s1", "server-other") + mocks.queued() shouldBe emptyList() + mocks.operationModelStore.list() shouldBe emptyList() + } + + test("a create waits for a backend onesignalId") { + val op = create(onesignalId = "local-user") + op.canStartExecute shouldBe false + + op.translateIds(mapOf("local-user" to "user")) + + op.canStartExecute shouldBe true + op.onesignalId shouldBe "user" + } + + test("a persisted create replays on the next launch with the same idempotency key") { + // Given + val prefs = MockPreferencesService() + val op = create().apply { id = "op-id" } + OperationModelStore(prefs).apply { loadOperations() }.add(op) + + // When + val mocks = SessionQueueMocks(prefs) + mocks.operationRepo.start() + awaitExecuted(mocks, 1) + + // Then + (mocks.executed.single() as CreateSessionOperation).idempotencyKey shouldBe op.idempotencyKey + } + + test("session operations are kept for anonymous users when identity verification is required") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.useIdentityVerification = JwtRequirement.REQUIRED + mocks.identityVerificationService = CoreInternalMocks.identityVerificationService(newCodePathsRun = true, ivBehaviorActive = true) + + mocks.enqueueAll(create()) + mocks.operationRepo.removeOperationsWithoutExternalId() + + mocks.queued().size shouldBe 1 + } +}) + +private fun Operation.shouldBeCreate() = (this is CreateSessionOperation) shouldBe true 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..b061b94d83 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 @@ -1,5 +1,6 @@ package com.onesignal.session.internal.session +import com.onesignal.common.IDManager import com.onesignal.common.modeling.ISingletonModelStoreChangeHandler import com.onesignal.common.modeling.ModelChangeTags import com.onesignal.common.modeling.ModelChangedArgs @@ -17,6 +18,7 @@ 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.every import io.mockk.just import io.mockk.mockk @@ -310,6 +312,7 @@ class SessionServiceTests : FunSpec({ mocks.sessionModelStore { it.isValid = false it.serverSessionId = "previous-server-session" + it.localSessionId = "local-previous" } // When @@ -319,6 +322,8 @@ class SessionServiceTests : FunSpec({ sessionModelStore.model.onesignalId shouldBe Mocks.ONESIGNAL_ID sessionModelStore.model.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID sessionModelStore.model.serverSessionId shouldBe null + sessionModelStore.model.localSessionId shouldNotBe "local-previous" + IDManager.isLocalId(sessionModelStore.model.localSessionId) shouldBe true } test("pinned IDs stay the same after login within a session") { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt new file mode 100644 index 0000000000..9238919929 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt @@ -0,0 +1,152 @@ +package com.onesignal.session.internal.session.operations + +import com.onesignal.common.exceptions.BackendException +import com.onesignal.core.internal.operations.ExecutionResult +import com.onesignal.core.internal.operations.impl.OperationModelStore +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.SessionModel +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.operations.impl.SessionOperationExecutor +import com.onesignal.user.internal.operations.impl.states.NewRecordsState +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.shouldBe +import io.kotest.matchers.types.shouldBeInstanceOf +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk + +private class Mocks { + val backend = mockk() + val session = SessionModel().apply { localSessionId = "local-session" } + val newRecordState = mockk().also { every { it.isInMissingRetryWindow(any()) } returns false } + val executor = + SessionOperationExecutor( + backend, + MockHelper.deviceService(), + MockHelper.sessionModelStore().also { every { it.model } returns session }, + newRecordState, + ) +} + +private fun create() = CreateSessionOperation("appId", "local-session", "onesignalId", "subscriptionId", startTime = 1_700_000_000_500L) + +private fun update(endTime: Long? = null) = + UpdateSessionOperation("appId", "server-session", "onesignalId", "subscriptionId", activeDuration = 61_900L, endTime = endTime) + +class SessionOperationExecutorTests : FunSpec({ + beforeAny { + Logging.logLevel = LogLevel.NONE + } + + test("create sends the request and translates the local session ID") { + // Given + val mocks = Mocks() + val op = create() + coEvery { mocks.backend.createSession(any(), any()) } returns "server-session" + + // When + val response = mocks.executor.execute(listOf(op)) + + // Then + response.result shouldBe ExecutionResult.SUCCESS + response.idTranslations shouldBe mapOf("local-session" to "server-session") + mocks.session.serverSessionId shouldBe "server-session" + coVerify { + mocks.backend.createSession( + "appId", + CreateSessionRequest("onesignalId", "subscriptionId", 1, 1_700_000_000L, op.idempotencyKey), + ) + } + } + + test("create for an earlier session does not overwrite the current session's server ID") { + val mocks = Mocks() + mocks.session.localSessionId = "local-newer" + coEvery { mocks.backend.createSession(any(), any()) } returns "server-session" + + mocks.executor.execute(listOf(create())) + + mocks.session.serverSessionId shouldBe null + } + + test("create without a session_id in the response is dropped") { + val mocks = Mocks() + coEvery { mocks.backend.createSession(any(), any()) } returns null + + mocks.executor.execute(listOf(create())).result shouldBe ExecutionResult.FAIL_NORETRY + } + + test("update sends the duration and end time in seconds") { + val mocks = Mocks() + val op = update(endTime = 1_700_000_100_999L) + coEvery { mocks.backend.updateSession(any(), any(), any()) } returns Unit + + mocks.executor.execute(listOf(op)).result shouldBe ExecutionResult.SUCCESS + + coVerify { + mocks.backend.updateSession( + "appId", + "server-session", + UpdateSessionRequest("onesignalId", "subscriptionId", 61L, op.idempotencyKey, 1_700_000_100L), + ) + } + } + + for (status in listOf(0, 408, 429, 500, 503)) { + test("status $status is retried with its Retry-After") { + val mocks = Mocks() + coEvery { mocks.backend.createSession(any(), any()) } throws BackendException(status, retryAfterSeconds = 7) + + val response = mocks.executor.execute(listOf(create())) + + response.result shouldBe ExecutionResult.FAIL_RETRY + response.retryAfterSeconds shouldBe 7 + } + } + + for (status in listOf(400, 401, 403, 404, 409, 422)) { + test("status $status is dropped") { + val mocks = Mocks() + coEvery { mocks.backend.updateSession(any(), any(), any()) } throws BackendException(status) + + mocks.executor.execute(listOf(update())).result shouldBe ExecutionResult.FAIL_NORETRY + } + } + + test("update 404 right after the create is retried") { + val mocks = Mocks() + every { mocks.newRecordState.isInMissingRetryWindow("server-session") } returns true + coEvery { mocks.backend.updateSession(any(), any(), any()) } throws BackendException(404) + + mocks.executor.execute(listOf(update())).result shouldBe ExecutionResult.FAIL_RETRY + } + + test("idempotency keys survive persistence") { + // Given + val store = OperationModelStore(MockPreferencesService()) + val create = create() + val update = update(endTime = 1L) + + // When + val restoredCreate = store.create(create.toJSON()) + val restoredUpdate = store.create(update.toJSON()) + + // Then + restoredCreate.shouldBeInstanceOf() + restoredCreate.idempotencyKey shouldBe create.idempotencyKey + restoredCreate.startTime shouldBe create.startTime + restoredUpdate.shouldBeInstanceOf() + restoredUpdate.idempotencyKey shouldBe update.idempotencyKey + restoredUpdate.endTime shouldBe 1L + } + + test("each operation gets its own idempotency key") { + (create().idempotencyKey == create().idempotencyKey) shouldBe false + } +}) From 60eccf842c778234aae553a6ed4b460bc50fe44f Mon Sep 17 00:00:00 2001 From: Fadi George Date: Wed, 14 Oct 2026 16:47:47 -0700 Subject: [PATCH 2/8] fix: [SDK-5378] retry offline session requests and tighten session queue rules Co-Authored-By: Claude Opus 5.5 --- .../operations/impl/SessionQueuePolicy.kt | 34 +++++++++++--- .../impl/SessionOperationExecutor.kt | 8 +++- .../operations/SessionOperationQueueTests.kt | 46 +++++++++++++++++++ .../SessionOperationExecutorTests.kt | 10 +++- 4 files changed, 89 insertions(+), 9 deletions(-) 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 73f846f42d..af25096f23 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 @@ -20,8 +20,17 @@ internal class SessionQueuePolicy( override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_RECENT_TRANSLATIONS } + // Sessions with an accepted end. The queue alone misses an end that is executing. + private val endedSessions = + object : LinkedHashMap() { + override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_ENDED_SESSIONS + } + fun recordTranslations(idTranslations: Map) { recentTranslations.putAll(idTranslations) + idTranslations.forEach { (localId, backendId) -> + if (endedSessions.remove(localId) != null) endedSessions[backendId] = Unit + } } class EnqueueResult( @@ -34,13 +43,22 @@ internal class SessionQueuePolicy( incoming: OperationRepo.OperationQueueItem, ): EnqueueResult { val op = incoming.operation - if (op.isSessionOperation) op.translateIds(recentTranslations) + if (!op.isSessionOperation) return EnqueueResult(accept = true) + op.translateIds(recentTranslations) - if (op is UpdateSessionOperation && hasQueuedEnd(queue, op.sessionId)) { - Logging.debug("SessionQueuePolicy: session ${op.sessionId} already has a queued end, dropping $op") - return EnqueueResult(accept = false) + return if (op is UpdateSessionOperation && (op.sessionId in endedSessions || hasQueuedEnd(queue, op.sessionId))) { + Logging.debug("SessionQueuePolicy: session ${op.sessionId} already ended, dropping $op") + EnqueueResult(accept = false) + } else { + admit(queue, incoming) } + } + private fun admit( + queue: MutableList, + incoming: OperationRepo.OperationQueueItem, + ): EnqueueResult { + val op = incoming.operation val removed = mutableListOf() if (op is UpdateSessionOperation) { val unsent = queue.filter { (it.operation as? UpdateSessionOperation)?.sessionId == op.sessionId } @@ -48,8 +66,10 @@ internal class SessionQueuePolicy( unsent.forEach { val replaced = it.operation as UpdateSessionOperation op.activeDuration = maxOf(op.activeDuration, replaced.activeDuration) - incoming.waiters.addAll(it.waiters) - it.waiters.clear() + incoming.retries = maxOf(incoming.retries, it.retries) + incoming.retryAt = maxOf(incoming.retryAt, it.retryAt) + val waiters = synchronized(it) { it.waiters.toList().also { _ -> it.waiters.clear() } } + incoming.attachOrWake(waiters) } queue.removeAll(unsent) removed.addAll(unsent) @@ -64,6 +84,7 @@ internal class SessionQueuePolicy( removed.addAll(dropped) if (op.sessionKey == oldestSession) accept = false } + if (accept && op is UpdateSessionOperation && op.isEnd) endedSessions[op.sessionId] = Unit return EnqueueResult(accept, removed) } @@ -99,5 +120,6 @@ internal class SessionQueuePolicy( companion object { const val MAX_QUEUED_SESSION_OPERATIONS = 100 private const val MAX_RECENT_TRANSLATIONS = 50 + private const val MAX_ENDED_SESSIONS = 50 } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt index ff0f43d161..c7b4def78f 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt @@ -46,7 +46,10 @@ internal class SessionOperationExecutor( ), ) } catch (ex: BackendException) { - return failure(ex, retryMissing = false) + val retryMissing = + newRecordState.isInMissingRetryWindow(operation.onesignalId) || + newRecordState.isInMissingRetryWindow(operation.subscriptionId) + return failure(ex, retryMissing) } return if (sessionId == null) { @@ -86,7 +89,8 @@ internal class SessionOperationExecutor( ): ExecutionResponse { val code = ex.statusCode val retryable = - code == 0 || + // HttpClient reports offline and other network errors as 0 or -1. + code <= 0 || code == HTTP_REQUEST_TIMEOUT || code == HTTP_TOO_MANY_REQUESTS || code >= HTTP_SERVER_ERROR || 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 802c296cf4..49f98bb7b6 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 @@ -14,6 +14,7 @@ import com.onesignal.session.internal.session.operations.impl.SessionOperationEx import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.ExecutorMocks.Companion.getNewRecordState +import com.onesignal.user.internal.operations.SetTagOperation import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.collections.shouldContainExactly import io.kotest.matchers.shouldBe @@ -88,6 +89,8 @@ private fun update( endTime: Long? = null, ) = UpdateSessionOperation("appId", sessionId, "onesignalId", "subscriptionId", activeDuration, endTime) +private fun item(op: Operation) = OperationRepo.OperationQueueItem(op, bucket = 0) + private suspend fun awaitExecuted( mocks: SessionQueueMocks, count: Int, @@ -148,6 +151,49 @@ class SessionOperationQueueTests : FunSpec({ mocks.operationModelStore.list().size shouldBe others.size } + test("non-session operations do not count toward the cap") { + val policy = SessionQueuePolicy(maxQueued = 2) + val queue = mutableListOf(item(create(localSessionId = "local-a")), item(create(localSessionId = "local-b"))) + + val result = policy.onEnqueue(queue, item(SetTagOperation("appId", "onesignalId", null, "key", "value"))) + + result.accept shouldBe true + result.removed shouldBe emptyList() + queue.size shouldBe 2 + } + + test("a coalesced update keeps the backoff of the update it replaces") { + val policy = SessionQueuePolicy() + val backedOff = item(update()).apply { + retries = 3 + retryAt = 60_000L + } + val incoming = item(update(activeDuration = 2_000L)) + + policy.onEnqueue(mutableListOf(backedOff), incoming) + + incoming.retries shouldBe 3 + incoming.retryAt shouldBe 60_000L + } + + test("an update after an end that is executing is dropped") { + val policy = SessionQueuePolicy() + val queue = mutableListOf() + policy.onEnqueue(queue, item(update(endTime = 9L))).accept shouldBe true + + // The end leaves the queue while it executes. + policy.onEnqueue(queue, item(update(activeDuration = 8_000L))).accept shouldBe false + } + + test("an end recorded under the local ID still drops updates after the create succeeds") { + val policy = SessionQueuePolicy() + policy.onEnqueue(mutableListOf(), item(update(endTime = 9L))) + + policy.recordTranslations(mapOf("local-s1" to "server-s1")) + + policy.onEnqueue(mutableListOf(), item(update(sessionId = "server-s1"))).accept shouldBe false + } + test("an update waits for its create and is sent with the backend session ID") { val mocks = SessionQueueMocks() mocks.respond = { op -> diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt index 9238919929..b7be1a3acb 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt @@ -98,7 +98,7 @@ class SessionOperationExecutorTests : FunSpec({ } } - for (status in listOf(0, 408, 429, 500, 503)) { + for (status in listOf(-1, 0, 408, 429, 500, 503)) { test("status $status is retried with its Retry-After") { val mocks = Mocks() coEvery { mocks.backend.createSession(any(), any()) } throws BackendException(status, retryAfterSeconds = 7) @@ -127,6 +127,14 @@ class SessionOperationExecutorTests : FunSpec({ mocks.executor.execute(listOf(update())).result shouldBe ExecutionResult.FAIL_RETRY } + test("create 404 while the user is new is retried") { + val mocks = Mocks() + every { mocks.newRecordState.isInMissingRetryWindow("onesignalId") } returns true + coEvery { mocks.backend.createSession(any(), any()) } throws BackendException(404) + + mocks.executor.execute(listOf(create())).result shouldBe ExecutionResult.FAIL_RETRY + } + test("idempotency keys survive persistence") { // Given val store = OperationModelStore(MockPreferencesService()) From b02b5746ce2af9d7cbc25175d3f8da111e639f32 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Wed, 14 Oct 2026 19:38:41 -0700 Subject: [PATCH 3/8] fix(session): avoid stalling queue on create delay --- .../internal/operations/impl/OperationRepo.kt | 10 ++++++- .../operations/impl/SessionQueuePolicy.kt | 20 +++++++------- .../impl/SessionOperationExecutor.kt | 10 ++++--- .../operations/SessionOperationQueueTests.kt | 26 +++++++++++++++++++ .../SessionOperationExecutorTests.kt | 3 +++ 5 files changed, 56 insertions(+), 13 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt index 46efcd548d..4b68987c16 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt @@ -16,6 +16,7 @@ import com.onesignal.core.internal.startup.IStartableService import com.onesignal.core.internal.time.ITime import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging +import com.onesignal.session.internal.session.operations.CreateSessionOperation import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.LoginUserOperation @@ -580,7 +581,14 @@ internal class OperationRepo( } if (response.idTranslations != null) { - delayForPostCreate(_configModelStore.model.opRepoPostCreateDelay) + val postCreateDelay = _configModelStore.model.opRepoPostCreateDelay + // A new session's only dependents are its updates, which NewRecordsState holds back, + // so only the wake is delayed. + if (ops.all { it.operation is CreateSessionOperation }) { + scope.launch { delayForPostCreate(postCreateDelay) } + } else { + delayForPostCreate(postCreateDelay) + } } } catch (e: Throwable) { Logging.log(LogLevel.ERROR, "Error attempting to execute operation: $ops", e) 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 af25096f23..3c466a4698 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 @@ -20,16 +20,17 @@ internal class SessionQueuePolicy( override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_RECENT_TRANSLATIONS } - // Sessions with an accepted end. The queue alone misses an end that is executing. - private val endedSessions = + // Sessions that take no more updates: ended, or their create was dropped. The queue alone + // misses an end that is executing. + private val closedSessions = object : LinkedHashMap() { - override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_ENDED_SESSIONS + override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_CLOSED_SESSIONS } fun recordTranslations(idTranslations: Map) { recentTranslations.putAll(idTranslations) idTranslations.forEach { (localId, backendId) -> - if (endedSessions.remove(localId) != null) endedSessions[backendId] = Unit + if (closedSessions.remove(localId) != null) closedSessions[backendId] = Unit } } @@ -46,8 +47,8 @@ internal class SessionQueuePolicy( if (!op.isSessionOperation) return EnqueueResult(accept = true) op.translateIds(recentTranslations) - return if (op is UpdateSessionOperation && (op.sessionId in endedSessions || hasQueuedEnd(queue, op.sessionId))) { - Logging.debug("SessionQueuePolicy: session ${op.sessionId} already ended, dropping $op") + return if (op is UpdateSessionOperation && (op.sessionId in closedSessions || hasQueuedEnd(queue, op.sessionId))) { + Logging.debug("SessionQueuePolicy: session ${op.sessionId} is closed, dropping $op") EnqueueResult(accept = false) } else { admit(queue, incoming) @@ -84,16 +85,17 @@ internal class SessionQueuePolicy( removed.addAll(dropped) if (op.sessionKey == oldestSession) accept = false } - if (accept && op is UpdateSessionOperation && op.isEnd) endedSessions[op.sessionId] = Unit + if (accept && op is UpdateSessionOperation && op.isEnd) closedSessions[op.sessionId] = Unit return EnqueueResult(accept, removed) } - /** Removes queued updates that can never send because [dropped] was their session's create. */ + /** Removes and rejects updates that can never send because [dropped] was their session's create. */ fun onDropped( queue: MutableList, dropped: Operation, ): List { if (dropped !is CreateSessionOperation) return emptyList() + closedSessions[dropped.localSessionId] = Unit val orphans = queue.filter { (it.operation as? UpdateSessionOperation)?.sessionId == dropped.localSessionId } if (orphans.isNotEmpty()) { Logging.debug("SessionQueuePolicy: create for session ${dropped.localSessionId} was dropped, dropping ${orphans.size} updates") @@ -120,6 +122,6 @@ internal class SessionQueuePolicy( companion object { const val MAX_QUEUED_SESSION_OPERATIONS = 100 private const val MAX_RECENT_TRANSLATIONS = 50 - private const val MAX_ENDED_SESSIONS = 50 + private const val MAX_CLOSED_SESSIONS = 50 } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt index c7b4def78f..196b148227 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt @@ -1,6 +1,7 @@ package com.onesignal.session.internal.session.operations.impl import com.onesignal.common.exceptions.BackendException +import com.onesignal.common.threading.runOnSerialIO import com.onesignal.core.internal.device.IDeviceService import com.onesignal.core.internal.operations.ExecutionResponse import com.onesignal.core.internal.operations.ExecutionResult @@ -56,9 +57,12 @@ internal class SessionOperationExecutor( Logging.warn("SessionOperationExecutor: create session response has no session_id, dropping $operation") ExecutionResponse(ExecutionResult.FAIL_NORETRY) } else { - // Updates enqueued from here on read the backend ID from the model; queued ones are translated. - val session = sessionModelStore.model - if (session.localSessionId == operation.localSessionId) session.serverSessionId = sessionId + // Updates enqueued later read the backend ID from the model; queued ones are translated. + // Serial IO is where SessionService starts new sessions, so a newer one can't get this ID. + runOnSerialIO { + val session = sessionModelStore.model + if (session.localSessionId == operation.localSessionId) session.serverSessionId = sessionId + } ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(operation.localSessionId to sessionId)) } } 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 49f98bb7b6..43b1b241b1 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 @@ -194,6 +194,32 @@ class SessionOperationQueueTests : FunSpec({ policy.onEnqueue(mutableListOf(), item(update(sessionId = "server-s1"))).accept shouldBe false } + test("an update for a session whose create was dropped is rejected") { + val policy = SessionQueuePolicy() + + policy.onDropped(mutableListOf(), create()) + + policy.onEnqueue(mutableListOf(), item(update())).accept shouldBe false + } + + test("a session create does not stall other operations for the post-create delay") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.opRepoPostCreateDelay = 60_000L + mocks.respond = { op -> + if (op is CreateSessionOperation) { + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to "server-s1")) + } else { + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + + mocks.enqueueAll(create(), update(sessionId = "server-other")) + mocks.operationRepo.start() + awaitExecuted(mocks, 2) + + (mocks.executed[1] as UpdateSessionOperation).sessionId shouldBe "server-other" + } + test("an update waits for its create and is sent with the backend session ID") { val mocks = SessionQueueMocks() mocks.respond = { op -> diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt index b7be1a3acb..dd256ba216 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt @@ -5,6 +5,7 @@ import com.onesignal.core.internal.operations.ExecutionResult import com.onesignal.core.internal.operations.impl.OperationModelStore import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging +import com.onesignal.mocks.IOMockHelper import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService import com.onesignal.session.internal.session.SessionModel @@ -40,6 +41,8 @@ private fun update(endTime: Long? = null) = UpdateSessionOperation("appId", "server-session", "onesignalId", "subscriptionId", activeDuration = 61_900L, endTime = endTime) class SessionOperationExecutorTests : FunSpec({ + listener(IOMockHelper) + beforeAny { Logging.logLevel = LogLevel.NONE } From 34f233c9eacb1e6f8864e9897ef4bad02f6ae20d Mon Sep 17 00:00:00 2001 From: Fadi George Date: Wed, 14 Oct 2026 22:38:39 -0700 Subject: [PATCH 4/8] fix(session): tighten queue rules for updates/consent --- .../internal/operations/impl/OperationRepo.kt | 18 +++- .../operations/impl/SessionQueuePolicy.kt | 45 +++++++-- .../operations/CreateSessionOperation.kt | 17 ++-- .../operations/UpdateSessionOperation.kt | 6 ++ .../impl/SessionOperationExecutor.kt | 39 ++++++-- .../operations/SessionOperationQueueTests.kt | 91 +++++++++++++++++-- .../SessionOperationExecutorTests.kt | 42 +++++++-- 7 files changed, 213 insertions(+), 45 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt index 4b68987c16..cbe55fa380 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt @@ -108,7 +108,10 @@ internal class OperationRepo( private val inFlightLogins = mutableMapOf() private val waiter = WaiterWithValue() private var paused = false - private val sessionQueuePolicy = SessionQueuePolicy() + private val sessionQueuePolicy = + SessionQueuePolicy( + awaitingConsent = { _configModelStore.model.let { it.consentRequired == true && it.consentGiven != true } }, + ) // Guarded by queue. Set by a force that lands while ops are out of the queue executing. private var forcedDuringExecute = false @@ -373,16 +376,18 @@ internal class OperationRepo( } /** - * Drops queued operations whose externalId is null and that require a JWT. Called by the IV-aware HYDRATE + * Drops queued operations whose externalId is null and that require a JWT, or that wait for an + * anonymous user, which is never created under IV. Called by the IV-aware HYDRATE * choreography in [OperationRepoIvExtensions] when `jwt_required` becomes REQUIRED * to evict anon ops that can no longer execute. */ internal fun removeOperationsWithoutExternalId() { val removedIds: List = synchronized(queue) { - val anonymous = queue.filter { it.operation.externalId == null && it.operation.requiresJwt } + val unsendable = queue.filter { it.operation.externalId == null && (it.operation.requiresJwt || it.operation.waitsForAnonymousUser) } + queue.removeAll(unsendable) + val anonymous = unsendable + unsendable.flatMap { sessionQueuePolicy.onDropped(queue, it.operation) } anonymous.forEach { it.wakeWaiters(OperationWaitResult(false)) } - queue.removeAll(anonymous) // IV=ON never transfers anonymous state; clear existingOnesignalId so the // executor takes the createUser (upsert) path. The merge-anon-into-identified // path can't dispatch under IV — anon user creation requires a JWT-less call @@ -402,6 +407,10 @@ internal class OperationRepo( removedIds.forEach { _operationModelStore.remove(it) } } + // Session ops take no JWT, but a create can't be sent until its user exists. + private val Operation.waitsForAnonymousUser: Boolean + get() = this is CreateSessionOperation && IDManager.isLocalId(onesignalId) + /** * Post-HYDRATE maintenance: scheduled on IO so it runs *after* `loadSavedOperations` * populates the queue (fix for an earlier race where the purge ran against an empty @@ -747,6 +756,7 @@ internal class OperationRepo( internal fun loadSavedOperations() { _operationModelStore.loadOperations() for (operation in _operationModelStore.list().reversed()) { + synchronized(queue) { sessionQueuePolicy.onLoaded(operation) } internalEnqueue( OperationQueueItem(operation, bucket = enqueueIntoBucket), flush = false, 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 3c466a4698..35f05442cf 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 @@ -1,5 +1,6 @@ package com.onesignal.core.internal.operations.impl +import com.onesignal.common.IDManager import com.onesignal.core.internal.operations.Operation import com.onesignal.debug.internal.logging.Logging import com.onesignal.session.internal.session.operations.CreateSessionOperation @@ -12,6 +13,8 @@ import com.onesignal.session.internal.session.operations.UpdateSessionOperation */ internal class SessionQueuePolicy( private val maxQueued: Int = MAX_QUEUED_SESSION_OPERATIONS, + // Without consent HttpClient fails every request, so these must not be stored or sent. + private val awaitingConsent: () -> Boolean = { false }, ) { // Recent local -> backend ID translations, for updates built from a local ID just before the // create succeeded but enqueued after the queue was translated. @@ -27,13 +30,26 @@ internal class SessionQueuePolicy( override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_CLOSED_SESSIONS } + // Creates by local session ID, and by backend session ID once sent. Kept after they leave the + // queue so updates enqueued while one executes, or after it succeeded, still find it. + private val creates = + object : LinkedHashMap() { + override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > maxQueued * 2 + } + fun recordTranslations(idTranslations: Map) { recentTranslations.putAll(idTranslations) idTranslations.forEach { (localId, backendId) -> if (closedSessions.remove(localId) != null) closedSessions[backendId] = Unit + creates[localId]?.let { creates[backendId] = it } } } + /** For operations loaded from the store, which skip [onEnqueue]. */ + fun onLoaded(op: Operation) { + if (op is CreateSessionOperation) creates[op.localSessionId] = op + } + class EnqueueResult( val accept: Boolean, val removed: List = emptyList(), @@ -47,12 +63,28 @@ internal class SessionQueuePolicy( if (!op.isSessionOperation) return EnqueueResult(accept = true) op.translateIds(recentTranslations) - return if (op is UpdateSessionOperation && (op.sessionId in closedSessions || hasQueuedEnd(queue, op.sessionId))) { - Logging.debug("SessionQueuePolicy: session ${op.sessionId} is closed, dropping $op") - EnqueueResult(accept = false) - } else { - admit(queue, incoming) - } + val result = + when { + awaitingConsent() -> { + Logging.debug("SessionQueuePolicy: privacy consent not given, dropping $op") + EnqueueResult(accept = false) + } + op is UpdateSessionOperation && (op.sessionId in closedSessions || hasQueuedEnd(queue, op.sessionId)) -> { + Logging.debug("SessionQueuePolicy: session ${op.sessionId} is closed, dropping $op") + EnqueueResult(accept = false) + } + op is UpdateSessionOperation && IDManager.isLocalId(op.sessionId) && op.sessionId !in creates -> { + Logging.debug("SessionQueuePolicy: session ${op.sessionId} has no create, dropping $op") + EnqueueResult(accept = false) + } + else -> { + if (op is UpdateSessionOperation) creates[op.sessionId]?.let { op.useIdsOf(it) } + admit(queue, incoming) + } + } + // Its updates could never be sent. + if (!result.accept && op is CreateSessionOperation) closedSessions[op.localSessionId] = Unit + return result } private fun admit( @@ -85,6 +117,7 @@ internal class SessionQueuePolicy( removed.addAll(dropped) if (op.sessionKey == oldestSession) accept = false } + if (accept && op is CreateSessionOperation) creates[op.localSessionId] = op if (accept && op is UpdateSessionOperation && op.isEnd) closedSessions[op.sessionId] = Unit return EnqueueResult(accept, removed) } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt index b8cd44dca7..c4daa9c3dd 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt @@ -4,7 +4,6 @@ import com.onesignal.common.IDManager import com.onesignal.core.internal.operations.GroupComparisonType import com.onesignal.core.internal.operations.Operation import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor -import java.util.UUID /** * An [Operation] to create a session through the sessions API. @@ -45,11 +44,14 @@ internal class CreateSessionOperation() : Operation(SessionOperationExecutor.CRE setLongProperty(::startTime.name, value) } - /** Persisted so a retry after a restart is recognized by the backend as the same request. */ - var idempotencyKey: String - get() = getStringProperty(::idempotencyKey.name) - private set(value) { - setStringProperty(::idempotencyKey.name, value) + /** Stable across retries and restarts, so the backend recognizes a resent create. */ + val idempotencyKey: String get() = localSessionId + + /** Responses so far that had no session_id. */ + var missingSessionIdAttempts: Int + get() = getOptIntProperty(::missingSessionIdAttempts.name) ?: 0 + internal set(value) { + setIntProperty(::missingSessionIdAttempts.name, value) } override val requiresJwt: Boolean get() = false @@ -66,13 +68,14 @@ internal class CreateSessionOperation() : Operation(SessionOperationExecutor.CRE onesignalId: String, subscriptionId: String, startTime: Long, + externalId: String?, ) : this() { this.appId = appId this.localSessionId = localSessionId this.onesignalId = onesignalId this.subscriptionId = subscriptionId this.startTime = startTime - this.idempotencyKey = UUID.randomUUID().toString() + this.externalId = externalId } override fun translateIds(map: Map) { diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt index 767b43644f..aea92b0305 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt @@ -87,6 +87,12 @@ internal class UpdateSessionOperation() : Operation(SessionOperationExecutor.UPD this.idempotencyKey = UUID.randomUUID().toString() } + /** The backend requires an update to repeat the user and subscription its create sent. */ + fun useIdsOf(create: CreateSessionOperation) { + onesignalId = create.onesignalId + subscriptionId = create.subscriptionId + } + override fun translateIds(map: Map) { map[sessionId]?.let { sessionId = it } map[onesignalId]?.let { onesignalId = it } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt index 196b148227..bebcea10bd 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt @@ -53,17 +53,35 @@ internal class SessionOperationExecutor( return failure(ex, retryMissing) } - return if (sessionId == null) { - Logging.warn("SessionOperationExecutor: create session response has no session_id, dropping $operation") - ExecutionResponse(ExecutionResult.FAIL_NORETRY) - } else { - // Updates enqueued later read the backend ID from the model; queued ones are translated. - // Serial IO is where SessionService starts new sessions, so a newer one can't get this ID. - runOnSerialIO { - val session = sessionModelStore.model - if (session.localSessionId == operation.localSessionId) session.serverSessionId = sessionId + return if (sessionId == null) missingSessionId(operation) else created(operation, sessionId) + } + + private suspend fun created( + operation: CreateSessionOperation, + sessionId: String, + ): ExecutionResponse { + // Updates enqueued later read these from the model; queued ones are translated. + // Serial IO is where SessionService starts new sessions, so a newer one can't get them. + runOnSerialIO { + val session = sessionModelStore.model + if (session.localSessionId == operation.localSessionId) { + session.serverSessionId = sessionId + session.onesignalId = operation.onesignalId + session.subscriptionId = operation.subscriptionId } - ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(operation.localSessionId to sessionId)) + } + return ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(operation.localSessionId to sessionId)) + } + + /** A retry with the same idempotency key should return the session the backend already created. */ + private fun missingSessionId(operation: CreateSessionOperation): ExecutionResponse { + operation.missingSessionIdAttempts++ + return if (operation.missingSessionIdAttempts < MAX_MISSING_SESSION_ID_ATTEMPTS) { + Logging.warn("SessionOperationExecutor: create session response has no session_id, retrying $operation") + ExecutionResponse(ExecutionResult.FAIL_RETRY) + } else { + Logging.warn("SessionOperationExecutor: create session responses never had a session_id, dropping $operation") + ExecutionResponse(ExecutionResult.FAIL_NORETRY) } } @@ -109,6 +127,7 @@ internal class SessionOperationExecutor( companion object { const val CREATE_SESSION = "create-session" const val UPDATE_SESSION = "update-session" + const val MAX_MISSING_SESSION_ID_ATTEMPTS = 3 private const val HTTP_NOT_FOUND = 404 private const val HTTP_REQUEST_TIMEOUT = 408 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 43b1b241b1..702553825d 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 @@ -81,13 +81,15 @@ private class SessionQueueMocks( private fun create( localSessionId: String = "local-s1", onesignalId: String = "onesignalId", -) = CreateSessionOperation("appId", localSessionId, onesignalId, "subscriptionId", startTime = 1_000L) + externalId: String? = null, +) = CreateSessionOperation("appId", localSessionId, onesignalId, "subscriptionId", startTime = 1_000L, externalId) private fun update( - sessionId: String = "local-s1", + sessionId: String = "server-s1", activeDuration: Long = 1_000L, endTime: Long? = null, -) = UpdateSessionOperation("appId", sessionId, "onesignalId", "subscriptionId", activeDuration, endTime) + onesignalId: String = "onesignalId", +) = UpdateSessionOperation("appId", sessionId, onesignalId, "subscriptionId", activeDuration, endTime) private fun item(op: Operation) = OperationRepo.OperationQueueItem(op, bucket = 0) @@ -145,7 +147,7 @@ class SessionOperationQueueTests : FunSpec({ val max = SessionQueuePolicy.MAX_QUEUED_SESSION_OPERATIONS val others = (1 until max).map { create(localSessionId = "local-other-$it") } - mocks.enqueueAll(create(), update(), *others.toTypedArray()) + mocks.enqueueAll(create(), update(sessionId = "local-s1"), *others.toTypedArray()) mocks.queued() shouldContainExactly others mocks.operationModelStore.list().size shouldBe others.size @@ -187,7 +189,9 @@ class SessionOperationQueueTests : FunSpec({ test("an end recorded under the local ID still drops updates after the create succeeds") { val policy = SessionQueuePolicy() - policy.onEnqueue(mutableListOf(), item(update(endTime = 9L))) + val queue = mutableListOf() + policy.onEnqueue(queue, item(create())) + policy.onEnqueue(queue, item(update(sessionId = "local-s1", endTime = 9L))).accept shouldBe true policy.recordTranslations(mapOf("local-s1" to "server-s1")) @@ -199,7 +203,63 @@ class SessionOperationQueueTests : FunSpec({ policy.onDropped(mutableListOf(), create()) - policy.onEnqueue(mutableListOf(), item(update())).accept shouldBe false + policy.onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe false + } + + test("an update with a local session ID and no create is rejected") { + SessionQueuePolicy().onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe false + } + + test("an update carries the IDs its create was sent with") { + val policy = SessionQueuePolicy() + val queue = mutableListOf() + val create = create(onesignalId = "local-user") + policy.onEnqueue(queue, item(create)) + val queued = update(sessionId = "local-s1", onesignalId = "user-b") + + policy.onEnqueue(queue, item(queued)) + + queued.onesignalId shouldBe "local-user" + } + + test("an update after its create succeeded carries the create's IDs") { + val policy = SessionQueuePolicy() + val create = create(onesignalId = "user-a") + policy.onEnqueue(mutableListOf(), item(create)) + policy.recordTranslations(mapOf("local-s1" to "server-s1")) + val later = update(sessionId = "server-s1", onesignalId = "user-b") + + policy.onEnqueue(mutableListOf(), item(later)) + + later.onesignalId shouldBe "user-a" + } + + test("an update for a create loaded from the store is accepted") { + val policy = SessionQueuePolicy() + policy.onLoaded(create()) + + policy.onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe true + } + + test("session operations are not stored before privacy consent") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.consentRequired = true + mocks.configModelStore.model.consentGiven = false + + mocks.enqueueAll(create(), update()) + + mocks.queued() shouldBe emptyList() + mocks.operationModelStore.list() shouldBe emptyList() + } + + test("an update for a session whose create was rejected is rejected") { + var awaitingConsent = true + val policy = SessionQueuePolicy(awaitingConsent = { awaitingConsent }) + policy.onEnqueue(mutableListOf(), item(create())).accept shouldBe false + + awaitingConsent = false + + policy.onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe false } test("a session create does not stall other operations for the post-create delay") { @@ -230,7 +290,7 @@ class SessionOperationQueueTests : FunSpec({ } } - mocks.enqueueAll(create(), update()) + mocks.enqueueAll(create(), update(sessionId = "local-s1")) mocks.operationRepo.start() awaitExecuted(mocks, 2) @@ -254,7 +314,7 @@ class SessionOperationQueueTests : FunSpec({ created.await() awaitExecuted(mocks, 1) - mocks.operationRepo.enqueue(update()) + mocks.operationRepo.enqueue(update(sessionId = "local-s1")) awaitExecuted(mocks, 2) (mocks.executed[1] as UpdateSessionOperation).sessionId shouldBe "server-s1" @@ -264,7 +324,7 @@ class SessionOperationQueueTests : FunSpec({ val mocks = SessionQueueMocks() mocks.respond = { ExecutionResponse(ExecutionResult.FAIL_NORETRY) } - mocks.enqueueAll(create(), update(), update(sessionId = "server-other")) + mocks.enqueueAll(create(), update(sessionId = "local-s1"), update(sessionId = "server-other")) mocks.operationRepo.start() awaitExecuted(mocks, 2) @@ -308,6 +368,19 @@ class SessionOperationQueueTests : FunSpec({ mocks.queued().size shouldBe 1 } + + test("a create for an anonymous user that was never created is purged with its updates under identity verification") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.useIdentityVerification = JwtRequirement.REQUIRED + mocks.identityVerificationService = CoreInternalMocks.identityVerificationService(newCodePathsRun = true, ivBehaviorActive = true) + val kept = create(localSessionId = "local-s2", onesignalId = "local-user", externalId = "alice") + + mocks.enqueueAll(create(onesignalId = "local-anon"), update(sessionId = "local-s1"), kept) + mocks.operationRepo.removeOperationsWithoutExternalId() + + mocks.queued() shouldContainExactly listOf(kept) + mocks.operationModelStore.list() shouldContainExactly listOf(kept) + } }) private fun Operation.shouldBeCreate() = (this is CreateSessionOperation) shouldBe true diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt index dd256ba216..b838e33655 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt @@ -1,11 +1,11 @@ package com.onesignal.session.internal.session.operations import com.onesignal.common.exceptions.BackendException +import com.onesignal.common.threading.runOnSerialIO import com.onesignal.core.internal.operations.ExecutionResult import com.onesignal.core.internal.operations.impl.OperationModelStore import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging -import com.onesignal.mocks.IOMockHelper import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService import com.onesignal.session.internal.session.SessionModel @@ -21,10 +21,16 @@ import io.mockk.coEvery import io.mockk.coVerify import io.mockk.every import io.mockk.mockk +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.withTimeout private class Mocks { val backend = mockk() - val session = SessionModel().apply { localSessionId = "local-session" } + val session = + SessionModel().apply { + localSessionId = "local-session" + onesignalId = "local-user" + } val newRecordState = mockk().also { every { it.isInMissingRetryWindow(any()) } returns false } val executor = SessionOperationExecutor( @@ -35,14 +41,19 @@ private class Mocks { ) } -private fun create() = CreateSessionOperation("appId", "local-session", "onesignalId", "subscriptionId", startTime = 1_700_000_000_500L) +private fun create() = CreateSessionOperation("appId", "local-session", "onesignalId", "subscriptionId", startTime = 1_700_000_000_500L, externalId = null) private fun update(endTime: Long? = null) = UpdateSessionOperation("appId", "server-session", "onesignalId", "subscriptionId", activeDuration = 61_900L, endTime = endTime) -class SessionOperationExecutorTests : FunSpec({ - listener(IOMockHelper) +/** The executor writes the session on serial IO; this waits for work queued there before it. */ +private suspend fun awaitSerialIO() { + val done = CompletableDeferred() + runOnSerialIO { done.complete(Unit) } + withTimeout(5_000) { done.await() } +} +class SessionOperationExecutorTests : FunSpec({ beforeAny { Logging.logLevel = LogLevel.NONE } @@ -55,11 +66,14 @@ class SessionOperationExecutorTests : FunSpec({ // When val response = mocks.executor.execute(listOf(op)) + awaitSerialIO() // Then response.result shouldBe ExecutionResult.SUCCESS response.idTranslations shouldBe mapOf("local-session" to "server-session") mocks.session.serverSessionId shouldBe "server-session" + mocks.session.onesignalId shouldBe "onesignalId" + mocks.session.subscriptionId shouldBe "subscriptionId" coVerify { mocks.backend.createSession( "appId", @@ -74,15 +88,20 @@ class SessionOperationExecutorTests : FunSpec({ coEvery { mocks.backend.createSession(any(), any()) } returns "server-session" mocks.executor.execute(listOf(create())) + awaitSerialIO() mocks.session.serverSessionId shouldBe null } - test("create without a session_id in the response is dropped") { + test("create without a session_id in the response is retried, then dropped") { val mocks = Mocks() + val op = create() coEvery { mocks.backend.createSession(any(), any()) } returns null - mocks.executor.execute(listOf(create())).result shouldBe ExecutionResult.FAIL_NORETRY + val results = (1..SessionOperationExecutor.MAX_MISSING_SESSION_ID_ATTEMPTS).map { mocks.executor.execute(listOf(op)).result } + + results shouldBe listOf(ExecutionResult.FAIL_RETRY, ExecutionResult.FAIL_RETRY, ExecutionResult.FAIL_NORETRY) + coVerify(exactly = 3) { mocks.backend.createSession(any(), match { it.idempotencyKey == op.idempotencyKey }) } } test("update sends the duration and end time in seconds") { @@ -152,12 +171,17 @@ class SessionOperationExecutorTests : FunSpec({ restoredCreate.shouldBeInstanceOf() restoredCreate.idempotencyKey shouldBe create.idempotencyKey restoredCreate.startTime shouldBe create.startTime + restoredCreate.missingSessionIdAttempts shouldBe 0 restoredUpdate.shouldBeInstanceOf() restoredUpdate.idempotencyKey shouldBe update.idempotencyKey restoredUpdate.endTime shouldBe 1L } - test("each operation gets its own idempotency key") { - (create().idempotencyKey == create().idempotencyKey) shouldBe false + test("a create is keyed by its local session ID") { + create().idempotencyKey shouldBe "local-session" + } + + test("each update gets its own idempotency key") { + (update().idempotencyKey == update().idempotencyKey) shouldBe false } }) From ee433af99c8dd904dbc60792474f8491e678d693 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Thu, 15 Oct 2026 11:47:25 -0700 Subject: [PATCH 5/8] fix(session): drop anonymous creates at enqueue and record restored ends Co-Authored-By: Claude Opus 5.5 --- .../internal/operations/impl/OperationRepo.kt | 9 +++++---- .../operations/impl/SessionQueuePolicy.kt | 1 + .../operations/SessionOperationQueueTests.kt | 20 ++++++++++++++++++- 3 files changed, 25 insertions(+), 5 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt index cbe55fa380..cc6d64528c 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt @@ -223,13 +223,14 @@ internal class OperationRepo( * they cannot be authenticated and would otherwise sit in the queue forever, blocked by * `hasValidJwtIfRequired`. LoginUserOperation is exempt because it's enqueued * intentionally during logout and purged later by [removeOperationsWithoutExternalId] - * if needed. Ops with `requiresJwt = false` are exempt since they can still execute. + * if needed. Ops with `requiresJwt = false` are exempt since they can still execute, except a + * session create that waits on an anonymous user or subscription that will never be created. * Outer-gated on `newCodePathsRun` so Phase 1 customers stay byte-for-byte * on the legacy enqueue path. */ private fun shouldSuppressAnonymousOp(op: Operation): Boolean { if (!_identityVerificationService.newCodePathsRun) return false - if (op is LoginUserOperation || !op.requiresJwt) return false + if (op is LoginUserOperation || !(op.requiresJwt || op.waitsForAnonymousUser)) return false val suppress = _configModelStore.model.useIdentityVerification == JwtRequirement.REQUIRED && op.externalId == null @@ -407,9 +408,9 @@ internal class OperationRepo( removedIds.forEach { _operationModelStore.remove(it) } } - // Session ops take no JWT, but a create can't be sent until its user exists. + // Session ops take no JWT, but a create can't be sent until its user and subscription exist. private val Operation.waitsForAnonymousUser: Boolean - get() = this is CreateSessionOperation && IDManager.isLocalId(onesignalId) + get() = this is CreateSessionOperation && (IDManager.isLocalId(onesignalId) || IDManager.isLocalId(subscriptionId)) /** * Post-HYDRATE maintenance: scheduled on IO so it runs *after* `loadSavedOperations` 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 35f05442cf..ec34ab21d9 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 @@ -48,6 +48,7 @@ internal class SessionQueuePolicy( /** For operations loaded from the store, which skip [onEnqueue]. */ fun onLoaded(op: Operation) { if (op is CreateSessionOperation) creates[op.localSessionId] = op + if (op is UpdateSessionOperation && op.isEnd) closedSessions[op.sessionId] = Unit } class EnqueueResult( 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 702553825d..d44d793310 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 @@ -241,6 +241,13 @@ class SessionOperationQueueTests : FunSpec({ policy.onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe true } + test("an update for a session whose end was loaded from the store is rejected") { + val policy = SessionQueuePolicy() + policy.onLoaded(update(endTime = 2_000L)) + + policy.onEnqueue(mutableListOf(), item(update())).accept shouldBe false + } + test("session operations are not stored before privacy consent") { val mocks = SessionQueueMocks() mocks.configModelStore.model.consentRequired = true @@ -371,16 +378,27 @@ class SessionOperationQueueTests : FunSpec({ test("a create for an anonymous user that was never created is purged with its updates under identity verification") { val mocks = SessionQueueMocks() - mocks.configModelStore.model.useIdentityVerification = JwtRequirement.REQUIRED mocks.identityVerificationService = CoreInternalMocks.identityVerificationService(newCodePathsRun = true, ivBehaviorActive = true) val kept = create(localSessionId = "local-s2", onesignalId = "local-user", externalId = "alice") mocks.enqueueAll(create(onesignalId = "local-anon"), update(sessionId = "local-s1"), kept) + mocks.configModelStore.model.useIdentityVerification = JwtRequirement.REQUIRED mocks.operationRepo.removeOperationsWithoutExternalId() mocks.queued() shouldContainExactly listOf(kept) mocks.operationModelStore.list() shouldContainExactly listOf(kept) } + + test("a create for an anonymous user that was never created is not enqueued under identity verification") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.useIdentityVerification = JwtRequirement.REQUIRED + mocks.identityVerificationService = CoreInternalMocks.identityVerificationService(newCodePathsRun = true, ivBehaviorActive = true) + + mocks.operationRepo.enqueueAndAwaitResult(create(onesignalId = "local-anon")).success shouldBe false + + mocks.queued() shouldBe emptyList() + mocks.operationModelStore.list() shouldBe emptyList() + } }) private fun Operation.shouldBeCreate() = (this is CreateSessionOperation) shouldBe true From e03fff635b7fce2cdde241e630ff6596bee5c8b4 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Thu, 15 Oct 2026 11:58:51 -0700 Subject: [PATCH 6/8] test: deflake post-create delay and enqueue debounce tests Co-Authored-By: Claude Opus 5.5 --- .../onesignal/core/internal/operations/OperationRepoTests.kt | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt index c3ca1dffb8..168e008efc 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt @@ -1068,6 +1068,8 @@ class OperationRepoTests : FunSpec({ // Then response shouldBe true + // waiters wake before the delay starts + coVerify(timeout = 1_000) { mocks.operationRepo.delayForPostCreate(any()) } coVerifyOrder { // ensure the order: IDs are translated, operation removed from the store, then delay for postCreateDelay operation.translateIds(idTranslation) @@ -1083,7 +1085,7 @@ class OperationRepoTests : FunSpec({ test("ensure each time enqueue is called it restarts the delay time") { // Given val mocks = Mocks() - mocks.configModelStore.model.opRepoExecutionInterval = 100 + mocks.configModelStore.model.opRepoExecutionInterval = 200 // When mocks.operationRepo.start() From 95f3c221d8c2e141491bac4e53684a05965a306b Mon Sep 17 00:00:00 2001 From: Fadi George Date: Thu, 15 Oct 2026 12:07:58 -0700 Subject: [PATCH 7/8] test: clear prefs before each SDKInitTests case Co-Authored-By: Claude Opus 5.5 --- .../core/internal/application/SDKInitTests.kt | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/SDKInitTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/SDKInitTests.kt index 7f5c9e8550..daa182c08e 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/SDKInitTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/SDKInitTests.kt @@ -83,11 +83,7 @@ class SDKInitTests : FunSpec({ error.get() shouldBe null } - beforeAny { - Logging.logLevel = LogLevel.NONE - } - - afterAny { + fun clearPrefs() { val context = getApplicationContext() val prefs = context.getSharedPreferences("OneSignal", Context.MODE_PRIVATE) prefs.edit() @@ -101,6 +97,16 @@ class SDKInitTests : FunSpec({ .commit() } + // Earlier specs in the same JVM can leave an appId behind. + beforeAny { + Logging.logLevel = LogLevel.NONE + clearPrefs() + } + + afterAny { + clearPrefs() + } + test("accessor instances after initWithContext without appID shows the failure reason") { // Given val context = getApplicationContext() From 355c8a0a864364ad1267c68f98aea8ce32766c83 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Fri, 16 Oct 2026 15:03:39 -0700 Subject: [PATCH 8/8] refactor: [SDK-5378] keep a session's created and closed state in one place SessionQueuePolicy's creates and closed sessions merge into one persisted SessionStateModelStore, so they survive a restart instead of being rebuilt from the queue. The executor no longer writes the server session ID to SessionModel, which drops that field. Co-Authored-By: Claude Opus 5.5 --- .../java/com/onesignal/core/CoreModule.kt | 2 + .../internal/operations/impl/OperationRepo.kt | 5 +- .../operations/impl/SessionQueuePolicy.kt | 112 +++++++++++------- .../session/internal/session/SessionModel.kt | 12 +- .../internal/session/impl/SessionService.kt | 1 - .../session/operations/SessionStateModel.kt | 49 ++++++++ .../operations/UpdateSessionOperation.kt | 9 +- .../impl/SessionOperationExecutor.kt | 24 +--- .../internal/operations/OperationRepoTests.kt | 3 + .../operations/SessionOperationQueueTests.kt | 50 ++++++-- .../internal/session/SessionServiceTests.kt | 6 +- .../SessionOperationExecutorTests.kt | 32 ----- .../RecoverFromDroppedLoginBugTests.kt | 2 + 13 files changed, 183 insertions(+), 124 deletions(-) create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionStateModel.kt diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt index 614903b8f2..be054b384d 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt @@ -55,6 +55,7 @@ import com.onesignal.logger.IObservabilityEventRecorder import com.onesignal.logger.LoggerFactory import com.onesignal.notifications.INotificationsManager import com.onesignal.notifications.internal.MisconfiguredNotificationsManager +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.user.internal.jwt.JwtTokenStore internal class CoreModule : IModule { @@ -92,6 +93,7 @@ internal class CoreModule : IModule { // Operations builder.register().provides() + builder.register().provides() builder.register() .provides() .provides() diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt index cc6d64528c..4502ff5c83 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt @@ -17,6 +17,7 @@ import com.onesignal.core.internal.time.ITime import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.LoginUserOperation @@ -34,6 +35,7 @@ import kotlin.math.min import kotlin.random.Random import kotlin.reflect.KClass +@Suppress("LongParameterList") internal class OperationRepo( executors: List, private val _operationModelStore: OperationModelStore, @@ -42,6 +44,7 @@ internal class OperationRepo( private val _newRecordState: NewRecordsState, private val _jwtTokenStore: JwtTokenStore, private val _identityVerificationService: IdentityVerificationService, + sessionStateStore: SessionStateModelStore, ) : IOperationRepo, IStartableService { internal class OperationQueueItem( @@ -111,6 +114,7 @@ internal class OperationRepo( private val sessionQueuePolicy = SessionQueuePolicy( awaitingConsent = { _configModelStore.model.let { it.consentRequired == true && it.consentGiven != true } }, + sessions = sessionStateStore, ) // Guarded by queue. Set by a force that lands while ops are out of the queue executing. @@ -757,7 +761,6 @@ internal class OperationRepo( internal fun loadSavedOperations() { _operationModelStore.loadOperations() for (operation in _operationModelStore.list().reversed()) { - synchronized(queue) { sessionQueuePolicy.onLoaded(operation) } internalEnqueue( OperationQueueItem(operation, bucket = enqueueIntoBucket), flush = false, 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 ec34ab21d9..8205f23b31 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 @@ -1,9 +1,11 @@ package com.onesignal.core.internal.operations.impl import com.onesignal.common.IDManager +import com.onesignal.common.modeling.SimpleModelStore import com.onesignal.core.internal.operations.Operation import com.onesignal.debug.internal.logging.Logging import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.SessionStateModel import com.onesignal.session.internal.session.operations.UpdateSessionOperation /** @@ -15,40 +17,18 @@ internal class SessionQueuePolicy( private val maxQueued: Int = MAX_QUEUED_SESSION_OPERATIONS, // Without consent HttpClient fails every request, so these must not be stored or sent. private val awaitingConsent: () -> Boolean = { false }, + private val sessions: SimpleModelStore = SimpleModelStore({ SessionStateModel() }), ) { - // Recent local -> backend ID translations, for updates built from a local ID just before the - // create succeeded but enqueued after the queue was translated. + // Recent local -> backend ID translations, for an operation built from a local ID just before + // its record was created but enqueued after the queue was translated. private val recentTranslations = object : LinkedHashMap() { override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_RECENT_TRANSLATIONS } - // Sessions that take no more updates: ended, or their create was dropped. The queue alone - // misses an end that is executing. - private val closedSessions = - object : LinkedHashMap() { - override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_CLOSED_SESSIONS - } - - // Creates by local session ID, and by backend session ID once sent. Kept after they leave the - // queue so updates enqueued while one executes, or after it succeeded, still find it. - private val creates = - object : LinkedHashMap() { - override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > maxQueued * 2 - } - fun recordTranslations(idTranslations: Map) { recentTranslations.putAll(idTranslations) - idTranslations.forEach { (localId, backendId) -> - if (closedSessions.remove(localId) != null) closedSessions[backendId] = Unit - creates[localId]?.let { creates[backendId] = it } - } - } - - /** For operations loaded from the store, which skip [onEnqueue]. */ - fun onLoaded(op: Operation) { - if (op is CreateSessionOperation) creates[op.localSessionId] = op - if (op is UpdateSessionOperation && op.isEnd) closedSessions[op.sessionId] = Unit + sessions.list().forEach { it.translateIds(idTranslations) } } class EnqueueResult( @@ -70,24 +50,35 @@ internal class SessionQueuePolicy( Logging.debug("SessionQueuePolicy: privacy consent not given, dropping $op") EnqueueResult(accept = false) } - op is UpdateSessionOperation && (op.sessionId in closedSessions || hasQueuedEnd(queue, op.sessionId)) -> { - Logging.debug("SessionQueuePolicy: session ${op.sessionId} is closed, dropping $op") - EnqueueResult(accept = false) - } - op is UpdateSessionOperation && IDManager.isLocalId(op.sessionId) && op.sessionId !in creates -> { - Logging.debug("SessionQueuePolicy: session ${op.sessionId} has no create, dropping $op") - EnqueueResult(accept = false) - } - else -> { - if (op is UpdateSessionOperation) creates[op.sessionId]?.let { op.useIdsOf(it) } - admit(queue, incoming) - } + op is UpdateSessionOperation -> admitUpdate(queue, incoming, op) + else -> admit(queue, incoming) } // Its updates could never be sent. - if (!result.accept && op is CreateSessionOperation) closedSessions[op.localSessionId] = Unit + if (!result.accept && op is CreateSessionOperation) updateState(op.localSessionId, queue) { it.isClosed = true } return result } + private fun admitUpdate( + queue: MutableList, + incoming: OperationRepo.OperationQueueItem, + op: UpdateSessionOperation, + ): EnqueueResult { + val state = stateFor(op.sessionId) + state?.serverSessionId?.let { op.translateIds(mapOf(op.sessionId to it)) } + val rejection = + when { + state?.isClosed == true || hasQueuedEnd(queue, op.sessionId) -> "is closed" + IDManager.isLocalId(op.sessionId) && state?.onesignalId == null -> "has no create" + else -> null + } + if (rejection != null) { + Logging.debug("SessionQueuePolicy: session ${op.sessionId} $rejection, dropping $op") + return EnqueueResult(accept = false) + } + state?.let { op.useCreateIds(it.onesignalId, it.subscriptionId) } + return admit(queue, incoming) + } + private fun admit( queue: MutableList, incoming: OperationRepo.OperationQueueItem, @@ -118,8 +109,13 @@ internal class SessionQueuePolicy( removed.addAll(dropped) if (op.sessionKey == oldestSession) accept = false } - if (accept && op is CreateSessionOperation) creates[op.localSessionId] = op - if (accept && op is UpdateSessionOperation && op.isEnd) closedSessions[op.sessionId] = Unit + if (accept && op is CreateSessionOperation) { + updateState(op.localSessionId, queue) { + it.onesignalId = op.onesignalId + it.subscriptionId = op.subscriptionId + } + } + if (accept && op is UpdateSessionOperation && op.isEnd) updateState(op.sessionId, queue) { it.isClosed = true } return EnqueueResult(accept, removed) } @@ -129,7 +125,7 @@ internal class SessionQueuePolicy( dropped: Operation, ): List { if (dropped !is CreateSessionOperation) return emptyList() - closedSessions[dropped.localSessionId] = Unit + updateState(dropped.localSessionId, queue) { it.isClosed = true } val orphans = queue.filter { (it.operation as? UpdateSessionOperation)?.sessionId == dropped.localSessionId } if (orphans.isNotEmpty()) { Logging.debug("SessionQueuePolicy: create for session ${dropped.localSessionId} was dropped, dropping ${orphans.size} updates") @@ -138,6 +134,37 @@ internal class SessionQueuePolicy( return orphans } + private fun stateFor(sessionId: String) = sessions.list().firstOrNull { it.id == sessionId || it.serverSessionId == sessionId } + + private fun updateState( + sessionKey: String, + queue: List, + change: (SessionStateModel) -> Unit, + ) { + val existing = stateFor(sessionKey) + if (existing != null) { + change(existing) + } else { + sessions.add(SessionStateModel().also { it.id = sessionKey }.also(change)) + } + trimSessions(queue, sessionKey) + } + + /** Drops the oldest sessions past the cap, but never one with operations still queued, which need its state. */ + private fun trimSessions( + queue: List, + keep: String, + ) { + val all = sessions.list() + val excess = all.size - maxQueued + if (excess <= 0) return + val queuedSessions = queue.mapNotNull { it.operation.sessionKey }.toSet() + keep + all + .filter { it.id !in queuedSessions && it.serverSessionId !in queuedSessions } + .take(excess) + .forEach { sessions.remove(it.id) } + } + private fun hasQueuedEnd( queue: List, sessionId: String, @@ -156,6 +183,5 @@ internal class SessionQueuePolicy( companion object { const val MAX_QUEUED_SESSION_OPERATIONS = 100 private const val MAX_RECENT_TRANSLATIONS = 50 - private const val MAX_CLOSED_SESSIONS = 50 } } 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 3c44417549..dc55546b3a 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 @@ -97,21 +97,11 @@ class SessionModel : Model() { } /** - * Identifies this session's queued sessions API requests until [serverSessionId] is known. - * A new local ID for each session. + * Identifies this session's sessions API requests. A new local ID for each session. */ var localSessionId: String get() = getStringProperty(::localSessionId.name) { "" } set(value) { setStringProperty(::localSessionId.name, value) } - - /** - * Null until the sessions API creates this session. Cleared when a new session starts. - */ - var serverSessionId: String? - get() = getOptStringProperty(::serverSessionId.name) - set(value) { - setOptStringProperty(::serverSessionId.name, value) - } } 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 f5b8ba0a99..a2a45bdfb8 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 @@ -198,7 +198,6 @@ internal class SessionService( ?.onesignalId session.subscriptionId = config?.pushSubscriptionId session.localSessionId = IDManager.createLocalId() - session.serverSessionId = null session.isValid = true Logging.debug("SessionService: New session started at ${session.startTime}") sessionLifeCycleNotifier.fire { it.onSessionStarted() } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionStateModel.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionStateModel.kt new file mode 100644 index 0000000000..93fbd1b8f4 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionStateModel.kt @@ -0,0 +1,49 @@ +package com.onesignal.session.internal.session.operations + +import com.onesignal.common.modeling.Model +import com.onesignal.common.modeling.SimpleModelStore +import com.onesignal.core.internal.preferences.IPreferencesService + +/** + * The one place that knows whether a session was created and whether it ended. Persisted, since a + * session's operations can leave the queue, or the process can restart, while more of its updates + * are still coming. [id] is the session's local ID, or its backend ID when only that is known. + */ +internal class SessionStateModel : Model() { + /** Set once the create succeeds. */ + var serverSessionId: String? + get() = getOptStringProperty(::serverSessionId.name) + set(value) { + setOptStringProperty(::serverSessionId.name, value) + } + + /** The user the create is sent with, which every update repeats. Null when no create is known. */ + var onesignalId: String? + get() = getOptStringProperty(::onesignalId.name) + set(value) { + setOptStringProperty(::onesignalId.name, value) + } + + /** The subscription the create is sent with, which every update repeats. */ + var subscriptionId: String? + get() = getOptStringProperty(::subscriptionId.name) + set(value) { + setOptStringProperty(::subscriptionId.name, value) + } + + /** Takes no more updates: it ended, or its create was dropped. */ + var isClosed: Boolean + get() = getBooleanProperty(::isClosed.name) { false } + set(value) { + setBooleanProperty(::isClosed.name, value) + } + + fun translateIds(map: Map) { + if (serverSessionId == null) map[id]?.let { serverSessionId = it } + onesignalId?.let { map[it] }?.let { onesignalId = it } + subscriptionId?.let { map[it] }?.let { subscriptionId = it } + } +} + +internal class SessionStateModelStore(prefs: IPreferencesService) : + SimpleModelStore({ SessionStateModel() }, "sessionsApiState", prefs) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt index aea92b0305..5340057089 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt @@ -88,9 +88,12 @@ internal class UpdateSessionOperation() : Operation(SessionOperationExecutor.UPD } /** The backend requires an update to repeat the user and subscription its create sent. */ - fun useIdsOf(create: CreateSessionOperation) { - onesignalId = create.onesignalId - subscriptionId = create.subscriptionId + fun useCreateIds( + onesignalId: String?, + subscriptionId: String?, + ) { + onesignalId?.let { this.onesignalId = it } + subscriptionId?.let { this.subscriptionId = it } } override fun translateIds(map: Map) { diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt index bebcea10bd..bf716f63a0 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt @@ -1,14 +1,12 @@ package com.onesignal.session.internal.session.operations.impl import com.onesignal.common.exceptions.BackendException -import com.onesignal.common.threading.runOnSerialIO import com.onesignal.core.internal.device.IDeviceService import com.onesignal.core.internal.operations.ExecutionResponse import com.onesignal.core.internal.operations.ExecutionResult import com.onesignal.core.internal.operations.IOperationExecutor import com.onesignal.core.internal.operations.Operation 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 @@ -20,7 +18,6 @@ import java.util.concurrent.TimeUnit internal class SessionOperationExecutor( private val sessionsBackend: ISessionsBackendService, private val deviceService: IDeviceService, - private val sessionModelStore: SessionModelStore, private val newRecordState: NewRecordsState, ) : IOperationExecutor { override val operations: List @@ -53,24 +50,11 @@ internal class SessionOperationExecutor( return failure(ex, retryMissing) } - return if (sessionId == null) missingSessionId(operation) else created(operation, sessionId) - } - - private suspend fun created( - operation: CreateSessionOperation, - sessionId: String, - ): ExecutionResponse { - // Updates enqueued later read these from the model; queued ones are translated. - // Serial IO is where SessionService starts new sessions, so a newer one can't get them. - runOnSerialIO { - val session = sessionModelStore.model - if (session.localSessionId == operation.localSessionId) { - session.serverSessionId = sessionId - session.onesignalId = operation.onesignalId - session.subscriptionId = operation.subscriptionId - } + return if (sessionId == null) { + missingSessionId(operation) + } else { + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(operation.localSessionId to sessionId)) } - return ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(operation.localSessionId to sessionId)) } /** A retry with the same idempotency key should return the session the backend already created. */ diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt index 168e008efc..866a246564 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt @@ -13,6 +13,7 @@ import com.onesignal.debug.internal.logging.Logging import com.onesignal.mocks.CoreInternalMocks import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.CreateSubscriptionOperation @@ -94,6 +95,7 @@ private class Mocks { getNewRecordState(configModelStore), jwtTokenStore, identityVerificationService, + SessionStateModelStore(MockPreferencesService()), ), recordPrivateCalls = true, ) @@ -121,6 +123,7 @@ class OperationRepoTests : FunSpec({ getNewRecordState(mocks.configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(), + SessionStateModelStore(MockPreferencesService()), ), ) 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 d44d793310..9b662ad3a6 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 @@ -9,6 +9,7 @@ import com.onesignal.mocks.CoreInternalMocks import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.session.internal.session.operations.UpdateSessionOperation import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor import com.onesignal.user.internal.jwt.JwtRequirement @@ -37,6 +38,7 @@ private class SessionQueueMocks( } val operationModelStore = OperationModelStore(prefs) + val sessionStateStore = SessionStateModelStore(prefs) val executed = mutableListOf() var respond: (Operation) -> ExecutionResponse = { ExecutionResponse(ExecutionResult.SUCCESS) } @@ -63,6 +65,7 @@ private class SessionQueueMocks( getNewRecordState(configModelStore), JwtTokenStore(MockPreferencesService()), identityVerificationService, + sessionStateStore, ), recordPrivateCalls = true, ) @@ -234,18 +237,49 @@ class SessionOperationQueueTests : FunSpec({ later.onesignalId shouldBe "user-a" } - test("an update for a create loaded from the store is accepted") { - val policy = SessionQueuePolicy() - policy.onLoaded(create()) + test("an update after a restart carries the IDs of a create from before it") { + val prefs = MockPreferencesService() + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(create(onesignalId = "user-a"))) + val later = update(sessionId = "local-s1", onesignalId = "user-b") + + val accepted = SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(later)).accept - policy.onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe true + accepted shouldBe true + later.onesignalId shouldBe "user-a" } - test("an update for a session whose end was loaded from the store is rejected") { - val policy = SessionQueuePolicy() - policy.onLoaded(update(endTime = 2_000L)) + test("an update after an end that already left the queue is rejected after a restart") { + val prefs = MockPreferencesService() + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(update(endTime = 2_000L))) + + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(update())).accept shouldBe false + } + + test("an update after a restart is sent with the backend session ID of a create from before it") { + val prefs = MockPreferencesService() + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).apply { + onEnqueue(mutableListOf(), item(create())) + recordTranslations(mapOf("local-s1" to "server-s1")) + } + val later = update(sessionId = "local-s1") + + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(later)) + + later.sessionId shouldBe "server-s1" + } + + test("the session state cap never drops a session with operations still queued") { + val policy = SessionQueuePolicy(maxQueued = 2) + val queue = mutableListOf() + val queued = item(create(localSessionId = "local-a")) + policy.onEnqueue(queue, queued) + queue.add(queued) + + policy.onEnqueue(queue, item(create(localSessionId = "local-b"))) + policy.onEnqueue(queue, item(create(localSessionId = "local-c"))) - policy.onEnqueue(mutableListOf(), item(update())).accept shouldBe false + policy.onEnqueue(queue, item(update(sessionId = "local-a"))).accept shouldBe true + policy.onEnqueue(queue, item(update(sessionId = "local-b"))).accept shouldBe false } test("session operations are not stored before privacy consent") { 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 b061b94d83..49e564867e 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 @@ -302,7 +302,7 @@ class SessionServiceTests : FunSpec({ sessionModelStore.model.usesSessionsApi shouldBe true } - test("new session pins onesignal and subscription IDs and clears the server session ID") { + test("new session pins onesignal and subscription IDs and a new local session ID") { // Given val mocks = Mocks() val sessionService = mocks.sessionService @@ -311,7 +311,6 @@ class SessionServiceTests : FunSpec({ val sessionModelStore = mocks.sessionModelStore { it.isValid = false - it.serverSessionId = "previous-server-session" it.localSessionId = "local-previous" } @@ -321,7 +320,6 @@ class SessionServiceTests : FunSpec({ // Then sessionModelStore.model.onesignalId shouldBe Mocks.ONESIGNAL_ID sessionModelStore.model.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID - sessionModelStore.model.serverSessionId shouldBe null sessionModelStore.model.localSessionId shouldNotBe "local-previous" IDManager.isLocalId(sessionModelStore.model.localSessionId) shouldBe true } @@ -513,7 +511,6 @@ class SessionServiceTests : FunSpec({ store.model.usesSessionsApi = true store.model.onesignalId = Mocks.ONESIGNAL_ID store.model.subscriptionId = Mocks.SUBSCRIPTION_ID - store.model.serverSessionId = "server-session-id" // When val reloaded = SessionModelStore(prefs).model @@ -526,7 +523,6 @@ class SessionServiceTests : FunSpec({ reloaded.usesSessionsApi shouldBe true reloaded.onesignalId shouldBe Mocks.ONESIGNAL_ID reloaded.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID - reloaded.serverSessionId shouldBe "server-session-id" } test("onFocus dispatches the session-mutation body through runOnSerialIO (SDK-4508)") { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt index b838e33655..8119777021 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt @@ -1,14 +1,12 @@ package com.onesignal.session.internal.session.operations import com.onesignal.common.exceptions.BackendException -import com.onesignal.common.threading.runOnSerialIO import com.onesignal.core.internal.operations.ExecutionResult import com.onesignal.core.internal.operations.impl.OperationModelStore 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.SessionModel import com.onesignal.session.internal.session.backend.CreateSessionRequest import com.onesignal.session.internal.session.backend.ISessionsBackendService import com.onesignal.session.internal.session.backend.UpdateSessionRequest @@ -21,22 +19,14 @@ import io.mockk.coEvery import io.mockk.coVerify import io.mockk.every import io.mockk.mockk -import kotlinx.coroutines.CompletableDeferred -import kotlinx.coroutines.withTimeout private class Mocks { val backend = mockk() - val session = - SessionModel().apply { - localSessionId = "local-session" - onesignalId = "local-user" - } val newRecordState = mockk().also { every { it.isInMissingRetryWindow(any()) } returns false } val executor = SessionOperationExecutor( backend, MockHelper.deviceService(), - MockHelper.sessionModelStore().also { every { it.model } returns session }, newRecordState, ) } @@ -46,13 +36,6 @@ private fun create() = CreateSessionOperation("appId", "local-session", "onesign private fun update(endTime: Long? = null) = UpdateSessionOperation("appId", "server-session", "onesignalId", "subscriptionId", activeDuration = 61_900L, endTime = endTime) -/** The executor writes the session on serial IO; this waits for work queued there before it. */ -private suspend fun awaitSerialIO() { - val done = CompletableDeferred() - runOnSerialIO { done.complete(Unit) } - withTimeout(5_000) { done.await() } -} - class SessionOperationExecutorTests : FunSpec({ beforeAny { Logging.logLevel = LogLevel.NONE @@ -66,14 +49,10 @@ class SessionOperationExecutorTests : FunSpec({ // When val response = mocks.executor.execute(listOf(op)) - awaitSerialIO() // Then response.result shouldBe ExecutionResult.SUCCESS response.idTranslations shouldBe mapOf("local-session" to "server-session") - mocks.session.serverSessionId shouldBe "server-session" - mocks.session.onesignalId shouldBe "onesignalId" - mocks.session.subscriptionId shouldBe "subscriptionId" coVerify { mocks.backend.createSession( "appId", @@ -82,17 +61,6 @@ class SessionOperationExecutorTests : FunSpec({ } } - test("create for an earlier session does not overwrite the current session's server ID") { - val mocks = Mocks() - mocks.session.localSessionId = "local-newer" - coEvery { mocks.backend.createSession(any(), any()) } returns "server-session" - - mocks.executor.execute(listOf(create())) - awaitSerialIO() - - mocks.session.serverSessionId shouldBe null - } - test("create without a session_id in the response is retried, then dropped") { val mocks = Mocks() val op = create() diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt index 52ed67075e..5f32d79117 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt @@ -7,6 +7,7 @@ import com.onesignal.debug.internal.logging.Logging import com.onesignal.mocks.CoreInternalMocks import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.ExecutorMocks @@ -47,6 +48,7 @@ private class Mocks { ExecutorMocks.getNewRecordState(configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(), + SessionStateModelStore(MockPreferencesService()), ), )