From 788faa12da7dd28c1e0e4106223683aae7286456 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Tue, 6 Oct 2026 17:03:33 -0700 Subject: [PATCH 1/9] feat: [SDK-5378] persist and retry sessions API requests in the operation queue Co-Authored-By: Claude Opus 5.5 (1M context) --- .../operations/impl/OperationModelStore.kt | 5 + .../internal/operations/impl/OperationRepo.kt | 38 ++- .../impl/OperationRepoSessionExtensions.kt | 99 ++++++ .../com/onesignal/session/SessionModule.kt | 3 + .../session/internal/session/SessionModel.kt | 7 + .../operations/CreateSessionOperation.kt | 98 ++++++ .../operations/UpdateSessionOperation.kt | 97 ++++++ .../impl/SessionOperationExecutor.kt | 88 ++++++ .../operations/SessionOperationQueueTests.kt | 294 ++++++++++++++++++ .../SessionOperationExecutorTests.kt | 199 ++++++++++++ 10 files changed, 921 insertions(+), 7 deletions(-) create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.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 c1a8db7ba2..23549db190 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 @@ -14,6 +14,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.UpdateSessionOperation import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.LoginUserOperation @@ -67,6 +68,9 @@ internal class OperationRepo( private val waiter = WaiterWithValue() private val retryWaiter = WaiterWithValue() private var paused = false + + // Ops enqueued after their local IDs were translated would otherwise keep the local ID forever. + private val appliedIdTranslations = mutableMapOf() private val initialized = CompletableDeferred() override suspend fun awaitInitialized() { @@ -171,7 +175,7 @@ internal class OperationRepo( */ 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 @@ -194,11 +198,7 @@ internal class OperationRepo( index: Int? = null, ) { synchronized(queue) { - val hasExisting = queue.any { it.operation.id == queueItem.operation.id } - if (hasExisting) { - Logging.debug("OperationRepo: internalEnqueue - operation.id: ${queueItem.operation.id} already exists in the queue.") - return - } + if (!prepareForQueue(queueItem, addToStore)) return // Dedupe LoginUserOperation by onesignalId. val op = queueItem.operation @@ -244,11 +244,31 @@ internal class OperationRepo( if (addToStore) { _operationModelStore.add(queueItem.operation) } + enforceSessionOperationCap(op, _operationModelStore) } waiter.wake(LoopWaiterMessage(flush, 0)) } + /** + * THIS SHOULD BE CALLED WHILE THE QUEUE IS SYNCHRONIZED!! + * + * @return false if [queueItem] should not be added. + */ + private fun prepareForQueue( + queueItem: OperationQueueItem, + isNew: Boolean, + ): Boolean { + val op = queueItem.operation + val isDuplicate = queue.any { it.operation.id == op.id } + if (isDuplicate) { + Logging.debug("OperationRepo: internalEnqueue - operation.id: ${op.id} already exists in the queue.") + } else if (isNew && appliedIdTranslations.isNotEmpty()) { + op.translateIds(appliedIdTranslations) + } + return !isDuplicate && (!isNew || op !is UpdateSessionOperation || coalesceSessionUpdate(queueItem, _operationModelStore)) + } + /** * The background processing that will never return. This should be called on it's own * dedicated thread. @@ -293,7 +313,7 @@ internal class OperationRepo( 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.waiter?.wake(false) } queue.removeAll(anonymous) // IV=ON never transfers anonymous state; clear existingOnesignalId so the @@ -381,6 +401,7 @@ internal class OperationRepo( ops.forEach { it.operation.translateIds(response.idTranslations) } synchronized(queue) { queue.forEach { it.operation.translateIds(response.idTranslations) } + appliedIdTranslations.putAll(response.idTranslations) } response.idTranslations.values.forEach { _newRecordState.add(it) } } @@ -480,6 +501,9 @@ internal class OperationRepo( private fun dropAndWake(ops: List) { ops.forEach { _operationModelStore.remove(it.operation.id) } ops.forEach { it.waiter?.wake(false) } + synchronized(queue) { + dropOrphanedSessionUpdates(ops, _operationModelStore) + } } /** diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt new file mode 100644 index 0000000000..37b0269dd8 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt @@ -0,0 +1,99 @@ +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 + +/** + * Session operations accumulate while offline, one or more per app open. Coalescing keeps + * each session to at most a create and one update, and the cap bounds how many sessions are kept. + * All functions here must be called while the queue is synchronized. + */ + +internal const val MAX_QUEUED_SESSION_OPERATIONS = 100 + +/** + * Folds queued, unsent updates for the same session into [incoming], which then carries the + * highest duration. An update that ends the session replaces them the same way. + * + * @return false if [incoming] should be discarded because the session's end is already queued. + */ +internal fun OperationRepo.coalesceSessionUpdate( + incoming: OperationRepo.OperationQueueItem, + operationModelStore: OperationModelStore, +): Boolean { + val op = incoming.operation as UpdateSessionOperation + val queued = + queue.filter { + val other = it.operation + other is UpdateSessionOperation && other.appId == op.appId && other.sessionId == op.sessionId + } + if (queued.any { (it.operation as UpdateSessionOperation).isEnd }) { + Logging.debug("OperationRepo: session ${op.sessionId} already has an end queued, discarding $op") + incoming.waiter?.wake(false) + return false + } + queued.maxOfOrNull { (it.operation as UpdateSessionOperation).activeDuration }?.let { + op.activeDuration = maxOf(op.activeDuration, it) + } + queued.forEach { item -> + if (incoming.waiter == null) { + incoming.waiter = item.waiter + } else { + item.waiter?.wake(true) + } + } + if (queued.isNotEmpty()) { + removeQueuedOperations(queued, operationModelStore) + Logging.debug("OperationRepo: coalesced ${queued.size} queued update(s) for session ${op.sessionId}") + } + return true +} + +/** + * Drops the oldest session operations once more than [MAX_QUEUED_SESSION_OPERATIONS] are queued. + */ +internal fun OperationRepo.enforceSessionOperationCap( + added: Operation, + operationModelStore: OperationModelStore, +) { + if (added !is CreateSessionOperation && added !is UpdateSessionOperation) return + + while (true) { + val sessionItems = queue.filter { it.operation is CreateSessionOperation || it.operation is UpdateSessionOperation } + if (sessionItems.size <= MAX_QUEUED_SESSION_OPERATIONS) return + + val oldest = sessionItems.first() + Logging.warn("OperationRepo: more than $MAX_QUEUED_SESSION_OPERATIONS session operations queued, dropping oldest: ${oldest.operation}") + oldest.waiter?.wake(false) + removeQueuedOperations(listOf(oldest), operationModelStore) + dropOrphanedSessionUpdates(listOf(oldest), operationModelStore) + } +} + +/** + * Updates for a session whose create was dropped can never get a backend session ID, so drop them too. + */ +internal fun OperationRepo.dropOrphanedSessionUpdates( + dropped: List, + operationModelStore: OperationModelStore, +) { + val localSessionIds = dropped.mapNotNull { (it.operation as? CreateSessionOperation)?.localSessionId }.toSet() + if (localSessionIds.isEmpty()) return + + val orphans = queue.filter { (it.operation as? UpdateSessionOperation)?.sessionId in localSessionIds } + if (orphans.isEmpty()) return + + Logging.warn("OperationRepo: dropping ${orphans.size} update(s) for sessions that were never created") + orphans.forEach { it.waiter?.wake(false) } + removeQueuedOperations(orphans, operationModelStore) +} + +private fun OperationRepo.removeQueuedOperations( + items: List, + operationModelStore: OperationModelStore, +) { + queue.removeAll(items) + items.forEach { operationModelStore.remove(it.operation.id) } +} 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..576c1368a7 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 @@ -1,5 +1,6 @@ package com.onesignal.session.internal.session +import com.onesignal.common.IDManager import com.onesignal.common.modeling.Model /** @@ -104,4 +105,10 @@ class SessionModel : Model() { set(value) { setOptStringProperty(::serverSessionId.name, value) } + + /** + * Stands in for [serverSessionId] in queued session operations until the backend assigns one. + */ + internal val localSessionId: String + get() = "${IDManager.LOCAL_PREFIX}$sessionId" } 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..30395e01cc --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt @@ -0,0 +1,98 @@ +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. + */ +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 this operation succeeds, so queued + * [UpdateSessionOperation]s can reference the session and be translated. + */ + var localSessionId: String + get() = getStringProperty(::localSessionId.name) + private set(value) { + setStringProperty(::localSessionId.name, value) + } + + /** + * May be a local ID; see [IDManager.isLocalId]. + */ + var onesignalId: String + get() = getStringProperty(::onesignalId.name) + private set(value) { + setStringProperty(::onesignalId.name, value) + } + + /** + * May be a local ID; see [IDManager.isLocalId]. + */ + 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) + } + + var directAttributionId: String? + get() = getOptStringProperty(::directAttributionId.name) + private set(value) { + setOptStringProperty(::directAttributionId.name, value) + } + + /** + * Generated once and persisted, so every retry, including after a restart, is deduplicated by the backend. + */ + var idempotencyKey: String + get() = getStringProperty(::idempotencyKey.name) + private set(value) { + setStringProperty(::idempotencyKey.name, value) + } + + override val createComparisonKey: String get() = "$appId.Session.$localSessionId" + override val modifyComparisonKey: String get() = "$appId.Session.$localSessionId" + override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE + override val canStartExecute: Boolean get() = !IDManager.isLocalId(onesignalId) && !IDManager.isLocalId(subscriptionId) + override val applyToRecordId: String get() = onesignalId + override val requiresJwt: Boolean get() = false + + constructor( + appId: String, + localSessionId: String, + onesignalId: String, + subscriptionId: String, + startTime: Long, + directAttributionId: String? = null, + idempotencyKey: String = UUID.randomUUID().toString(), + ) : this() { + this.appId = appId + this.localSessionId = localSessionId + this.onesignalId = onesignalId + this.subscriptionId = subscriptionId + this.startTime = startTime + this.directAttributionId = directAttributionId + this.idempotencyKey = idempotencyKey + } + + 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..2487efc237 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt @@ -0,0 +1,97 @@ +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. + */ +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] until the create succeeds. + */ + 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 in milliseconds. */ + var activeDuration: Long + get() = getLongProperty(::activeDuration.name) + internal set(value) { + setLongProperty(::activeDuration.name, value) + } + + /** Unix time in milliseconds; set when this update ends the session. */ + var endTime: Long? + get() = getOptLongProperty(::endTime.name) + private set(value) { + setOptLongProperty(::endTime.name, value) + } + + /** + * Generated once and persisted, so every retry, including after a restart, is deduplicated by the backend. + */ + var idempotencyKey: String + get() = getStringProperty(::idempotencyKey.name) + private set(value) { + setStringProperty(::idempotencyKey.name, value) + } + + val isEnd: Boolean get() = endTime != null + + override val createComparisonKey: String get() = "$appId.Session.$sessionId" + override val modifyComparisonKey: String get() = "$appId.Session.$sessionId" + 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 + override val requiresJwt: Boolean get() = false + + constructor( + appId: String, + sessionId: String, + onesignalId: String, + subscriptionId: String, + activeDuration: Long, + endTime: Long? = null, + idempotencyKey: String = UUID.randomUUID().toString(), + ) : this() { + this.appId = appId + this.sessionId = sessionId + this.onesignalId = onesignalId + this.subscriptionId = subscriptionId + this.activeDuration = activeDuration + this.endTime = endTime + this.idempotencyKey = idempotencyKey + } + + 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..51fb1ed4cf --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt @@ -0,0 +1,88 @@ +package com.onesignal.session.internal.session.operations.impl + +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.SessionsApiResult +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 java.util.concurrent.TimeUnit.MILLISECONDS + +internal class SessionOperationExecutor( + private val sessionsBackend: ISessionsBackendService, + private val deviceService: IDeviceService, + private val sessionModelStore: SessionModelStore, +) : IOperationExecutor { + override val operations: List + get() = listOf(CREATE_SESSION, UPDATE_SESSION) + + override suspend fun execute(operations: List): ExecutionResponse = + when (val operation = operations.first()) { + is CreateSessionOperation -> createSession(operation) + is UpdateSessionOperation -> updateSession(operation) + else -> throw IllegalArgumentException("Unrecognized operation: $operation") + } + + private suspend fun createSession(op: CreateSessionOperation): ExecutionResponse { + val request = + CreateSessionRequest( + onesignalId = op.onesignalId, + subscriptionId = op.subscriptionId, + deviceType = deviceService.deviceType.value, + startTime = MILLISECONDS.toSeconds(op.startTime), + idempotencyKey = op.idempotencyKey, + directAttributionId = op.directAttributionId, + ) + return when (val result = sessionsBackend.createSession(op.appId, request)) { + is SessionsApiResult.Success -> { + val serverSessionId = result.value + // Persisted so updates enqueued after this point, including after a restart, use the backend ID directly. + val session = sessionModelStore.model + if (session.localSessionId == op.localSessionId) { + session.serverSessionId = serverSessionId + } + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to serverSessionId)) + } + is SessionsApiResult.Retry -> retry(result) + is SessionsApiResult.Drop -> drop("create", result) + } + } + + private suspend fun updateSession(op: UpdateSessionOperation): ExecutionResponse { + val request = + UpdateSessionRequest( + onesignalId = op.onesignalId, + subscriptionId = op.subscriptionId, + durationSeconds = MILLISECONDS.toSeconds(op.activeDuration), + idempotencyKey = op.idempotencyKey, + endTime = op.endTime?.let { MILLISECONDS.toSeconds(it) }, + ) + return when (val result = sessionsBackend.updateSession(op.appId, op.sessionId, request)) { + is SessionsApiResult.Success -> ExecutionResponse(ExecutionResult.SUCCESS) + is SessionsApiResult.Retry -> retry(result) + is SessionsApiResult.Drop -> drop("update", result) + } + } + + private fun retry(result: SessionsApiResult.Retry) = ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = result.retryAfterSeconds) + + private fun drop( + action: String, + result: SessionsApiResult.Drop, + ): ExecutionResponse { + Logging.warn("SessionOperationExecutor: $action session failed with ${result.statusCode}, dropping") + return ExecutionResponse(ExecutionResult.FAIL_NORETRY) + } + + companion object { + const val CREATE_SESSION = "create-session" + const val UPDATE_SESSION = "update-session" + } +} 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..c90e616b82 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt @@ -0,0 +1,294 @@ +package com.onesignal.core.internal.operations + +import com.onesignal.core.internal.operations.impl.MAX_QUEUED_SESSION_OPERATIONS +import com.onesignal.core.internal.operations.impl.OperationModelStore +import com.onesignal.core.internal.operations.impl.OperationRepo +import com.onesignal.core.internal.time.impl.Time +import com.onesignal.debug.LogLevel +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.mocks.CoreInternalMocks +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.just +import io.mockk.mockk +import io.mockk.runs +import io.mockk.verify +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.delay +import kotlinx.coroutines.withTimeout +import java.util.UUID +import java.util.concurrent.ConcurrentHashMap + +private const val APP_ID = "appId" +private const val LOCAL_SESSION_ID = "local-session" + +private class SessionQueueMocks( + ivRequired: Boolean = false, +) { + val configModelStore = + MockHelper.configModelStore { + it.isInitializedWithRemote = true + it.useIdentityVerification = if (ivRequired) JwtRequirement.REQUIRED else JwtRequirement.NOT_REQUIRED + } + + val storedOperations = mutableListOf() + private val barriers = ConcurrentHashMap>() + val operationModelStore: OperationModelStore = + mockk().also { store -> + every { store.loadOperations() } just runs + every { store.list() } answers { storedOperations.toList() } + every { store.add(any()) } answers { + val op = firstArg() + val barrier = barriers.remove(op) + if (barrier != null) barrier.complete(Unit) else storedOperations.add(op) + } + every { store.add(any(), any()) } answers { storedOperations.add(firstArg(), secondArg()) } + every { store.remove(any()) } answers { storedOperations.removeAll { it.id == firstArg() } } + } + + val executed = mutableListOf() + val executor: IOperationExecutor = + mockk().also { + every { it.operations } returns listOf(SessionOperationExecutor.CREATE_SESSION, SessionOperationExecutor.UPDATE_SESSION) + } + + val operationRepo = + OperationRepo( + listOf(executor), + operationModelStore, + configModelStore, + Time(), + getNewRecordState(configModelStore), + JwtTokenStore(MockPreferencesService()), + CoreInternalMocks.identityVerificationService(newCodePathsRun = ivRequired, ivBehaviorActive = ivRequired), + ) + + fun respondWith(vararg responses: (Operation) -> ExecutionResponse) { + var call = 0 + coEvery { executor.execute(any()) } answers { + val op = firstArg>().single() + executed.add(op) + responses[minOf(call++, responses.size - 1)](op) + } + } + + /** + * Enqueues [ops], then waits until the repo's thread has processed them by enqueueing a + * barrier operation behind them. + */ + suspend fun enqueueAll(vararg ops: Operation) { + ops.forEach { operationRepo.enqueue(it) } + val barrier = mockk(relaxed = true) + every { barrier.id } returns UUID.randomUUID().toString() + val processed = CompletableDeferred() + barriers[barrier] = processed + operationRepo.enqueue(barrier) + withTimeout(2_000) { processed.await() } + } + + val queuedOperations: List + get() = + synchronized(operationRepo.queue) { + operationRepo.queue.map { it.operation }.filter { it is CreateSessionOperation || it is UpdateSessionOperation } + } +} + +private fun create( + localSessionId: String = LOCAL_SESSION_ID, + onesignalId: String = "onesignal-id", +) = CreateSessionOperation(APP_ID, localSessionId, onesignalId, "subscription-id", startTime = 1_000L) + +private fun update( + activeDuration: Long, + sessionId: String = LOCAL_SESSION_ID, + endTime: Long? = null, +) = UpdateSessionOperation(APP_ID, sessionId, "onesignal-id", "subscription-id", activeDuration, endTime) + +class SessionOperationQueueTests : FunSpec({ + beforeAny { + Logging.logLevel = LogLevel.NONE + } + + test("queued duration-only updates collapse into the newest update with the highest duration") { + // Given + val mocks = SessionQueueMocks() + val first = update(activeDuration = 20_000) + val second = update(activeDuration = 10_000) + + // When + mocks.enqueueAll(first, second) + + // Then + mocks.queuedOperations shouldContainExactly listOf(second) + second.activeDuration shouldBe 20_000 + mocks.storedOperations shouldContainExactly listOf(second) + } + + test("an end replaces an unsent update for the same session") { + // Given + val mocks = SessionQueueMocks() + val duration = update(activeDuration = 10_000) + val end = update(activeDuration = 15_000, endTime = 5_000L) + + // When + mocks.enqueueAll(duration, end) + + // Then + mocks.queuedOperations shouldContainExactly listOf(end) + end.activeDuration shouldBe 15_000 + end.isEnd shouldBe true + } + + test("an update for a session whose end is already queued is discarded") { + // Given + val mocks = SessionQueueMocks() + val end = update(activeDuration = 15_000, endTime = 5_000L) + val late = update(activeDuration = 20_000) + + // When + mocks.enqueueAll(end, late) + + // Then + mocks.queuedOperations shouldContainExactly listOf(end) + mocks.storedOperations shouldContainExactly listOf(end) + } + + test("updates for different sessions are not coalesced") { + // Given + val mocks = SessionQueueMocks() + val a = update(activeDuration = 10_000, sessionId = "local-a") + val b = update(activeDuration = 10_000, sessionId = "local-b") + + // When + mocks.enqueueAll(a, b) + + // Then + mocks.queuedOperations shouldContainExactly listOf(a, b) + } + + test("the oldest session operation is dropped past the cap, along with its orphaned updates") { + // Given + val mocks = SessionQueueMocks() + val oldestCreate = create(localSessionId = "local-oldest", onesignalId = "local-user") + val oldestUpdate = update(activeDuration = 1_000, sessionId = "local-oldest") + val others = (1..MAX_QUEUED_SESSION_OPERATIONS - 2).map { create(localSessionId = "local-$it", onesignalId = "local-user") } + mocks.enqueueAll(oldestCreate, oldestUpdate, *others.toTypedArray()) + mocks.queuedOperations.size shouldBe MAX_QUEUED_SESSION_OPERATIONS + + // When + val newest = create(localSessionId = "local-newest", onesignalId = "local-user") + mocks.enqueueAll(newest) + + // Then + mocks.queuedOperations shouldContainExactly others + newest + mocks.storedOperations shouldContainExactly others + newest + } + + test("updates wait for their create, are translated to the backend session ID, and run in order") { + // Given + val mocks = SessionQueueMocks() + mocks.respondWith( + { ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(LOCAL_SESSION_ID to "server-session")) }, + { ExecutionResponse(ExecutionResult.SUCCESS) }, + ) + val createOp = create() + val updateOp = update(activeDuration = 10_000) + mocks.enqueueAll(updateOp, createOp) + + // When + mocks.operationRepo.start() + withTimeout(2_000) { while (mocks.executed.size < 2) delay(10) } + + // Then + mocks.executed shouldContainExactly listOf(createOp, updateOp) + updateOp.sessionId shouldBe "server-session" + } + + test("an update enqueued after its create was translated still gets the backend session ID") { + // Given + val mocks = SessionQueueMocks() + mocks.respondWith( + { ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(LOCAL_SESSION_ID to "server-session")) }, + { ExecutionResponse(ExecutionResult.SUCCESS) }, + ) + mocks.enqueueAll(create()) + mocks.operationRepo.start() + withTimeout(2_000) { while (mocks.executed.size < 1) delay(10) } + + // When + val lateUpdate = update(activeDuration = 10_000) + mocks.operationRepo.enqueue(lateUpdate) + withTimeout(2_000) { while (mocks.executed.size < 2) delay(10) } + + // Then + mocks.executed[1] shouldBe lateUpdate + lateUpdate.sessionId shouldBe "server-session" + } + + test("a create that fails without retry drops its queued updates") { + // Given + val mocks = SessionQueueMocks() + mocks.respondWith({ ExecutionResponse(ExecutionResult.FAIL_NORETRY) }) + val createOp = create() + val updateOp = update(activeDuration = 10_000) + mocks.enqueueAll(createOp, updateOp) + + // When + mocks.operationRepo.start() + withTimeout(2_000) { while (mocks.queuedOperations.isNotEmpty()) delay(10) } + + // Then + mocks.executed shouldContainExactly listOf(createOp) + mocks.storedOperations shouldBe emptyList() + } + + test("a retried create keeps its idempotency key") { + // Given + val mocks = SessionQueueMocks() + val keys = mutableListOf() + mocks.respondWith( + { + keys.add((it as CreateSessionOperation).idempotencyKey) + ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = null) + }, + { + keys.add((it as CreateSessionOperation).idempotencyKey) + ExecutionResponse(ExecutionResult.SUCCESS) + }, + ) + val createOp = create() + mocks.enqueueAll(createOp) + + // When + mocks.operationRepo.start() + withTimeout(2_000) { while (keys.size < 2) delay(10) } + + // Then + keys shouldContainExactly listOf(createOp.idempotencyKey, createOp.idempotencyKey) + } + + test("session operations are not suppressed or purged as anonymous under identity verification") { + // Given + val mocks = SessionQueueMocks(ivRequired = true) + val createOp = create() + + // When + mocks.enqueueAll(createOp) + mocks.operationRepo.removeOperationsWithoutExternalId() + + // Then + mocks.queuedOperations shouldContainExactly listOf(createOp) + verify(exactly = 0) { mocks.operationModelStore.remove(any()) } + } +}) 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..cb49c935bd --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt @@ -0,0 +1,199 @@ +package com.onesignal.session.internal.session.operations + +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.backend.CreateSessionRequest +import com.onesignal.session.internal.session.backend.ISessionsBackendService +import com.onesignal.session.internal.session.backend.SessionsApiResult +import com.onesignal.session.internal.session.backend.UpdateSessionRequest +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.shouldBe +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.mockk + +private const val APP_ID = "appId" + +private fun createOp(localSessionId: String = "local-session-uuid") = + CreateSessionOperation( + appId = APP_ID, + localSessionId = localSessionId, + onesignalId = "onesignal-id", + subscriptionId = "subscription-id", + startTime = 1_700_000_000_123L, + directAttributionId = "notification-id", + idempotencyKey = "create-key", + ) + +private fun updateOp(endTime: Long? = null) = + UpdateSessionOperation( + appId = APP_ID, + sessionId = "server-session", + onesignalId = "onesignal-id", + subscriptionId = "subscription-id", + activeDuration = 42_900L, + endTime = endTime, + idempotencyKey = "update-key", + ) + +class SessionOperationExecutorTests : FunSpec({ + beforeAny { + Logging.logLevel = LogLevel.NONE + } + + test("create sends the request in seconds and translates the local session ID") { + // Given + val backend = mockk() + coEvery { backend.createSession(any(), any()) } returns SessionsApiResult.Success("server-session") + val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "session-uuid" } + val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), sessionModelStore) + + // When + val response = executor.execute(listOf(createOp(localSessionId = sessionModelStore.model.localSessionId))) + + // Then + response.result shouldBe ExecutionResult.SUCCESS + response.idTranslations shouldBe mapOf("local-session-uuid" to "server-session") + sessionModelStore.model.serverSessionId shouldBe "server-session" + coVerify { + backend.createSession( + APP_ID, + CreateSessionRequest( + onesignalId = "onesignal-id", + subscriptionId = "subscription-id", + deviceType = 1, + startTime = 1_700_000_000L, + idempotencyKey = "create-key", + directAttributionId = "notification-id", + ), + ) + } + } + + test("create for a session that is no longer current leaves the current session alone") { + // Given + val backend = mockk() + coEvery { backend.createSession(any(), any()) } returns SessionsApiResult.Success("server-session") + val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "newer-session" } + val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), sessionModelStore) + + // When + val response = executor.execute(listOf(createOp(localSessionId = "local-older-session"))) + + // Then + response.idTranslations shouldBe mapOf("local-older-session" to "server-session") + sessionModelStore.model.serverSessionId shouldBe null + } + + test("a retryable failure retries and passes Retry-After through") { + // Given + val backend = mockk() + coEvery { backend.createSession(any(), any()) } returns SessionsApiResult.Retry(429, retryAfterSeconds = 30) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + + // When + val response = executor.execute(listOf(createOp())) + + // Then + response.result shouldBe ExecutionResult.FAIL_RETRY + response.retryAfterSeconds shouldBe 30 + } + + test("a permanent failure is not retried") { + // Given + val backend = mockk() + coEvery { backend.updateSession(any(), any(), any()) } returns SessionsApiResult.Drop(400) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + + // When + val response = executor.execute(listOf(updateOp())) + + // Then + response.result shouldBe ExecutionResult.FAIL_NORETRY + } + + test("update sends the duration and end time in seconds") { + // Given + val backend = mockk() + coEvery { backend.updateSession(any(), any(), any()) } returns SessionsApiResult.Success(Unit) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + + // When + val response = executor.execute(listOf(updateOp(endTime = 1_700_000_100_999L))) + + // Then + response.result shouldBe ExecutionResult.SUCCESS + coVerify { + backend.updateSession( + APP_ID, + "server-session", + UpdateSessionRequest( + onesignalId = "onesignal-id", + subscriptionId = "subscription-id", + durationSeconds = 42L, + idempotencyKey = "update-key", + endTime = 1_700_000_100L, + ), + ) + } + } + + test("operations wait for backend IDs and translate local IDs") { + // Given + val create = CreateSessionOperation(APP_ID, "local-session", "local-user", "local-sub", 0L) + val update = UpdateSessionOperation(APP_ID, "local-session", "local-user", "local-sub", 0L) + + // Then + create.canStartExecute shouldBe false + update.canStartExecute shouldBe false + + // When + val translations = mapOf("local-user" to "user", "local-sub" to "sub") + create.translateIds(translations) + update.translateIds(translations) + + // Then + create.canStartExecute shouldBe true + update.canStartExecute shouldBe false + + // When + update.translateIds(mapOf("local-session" to "server-session")) + + // Then + update.canStartExecute shouldBe true + update.sessionId shouldBe "server-session" + } + + test("persisted operations reload with the same idempotency keys") { + // Given + val prefs = MockPreferencesService() + val create = createOp().apply { id = "create-op" } + val duration = updateOp().apply { id = "duration-op" } + val end = updateOp(endTime = 5_000L).apply { id = "end-op" } + OperationModelStore(prefs).apply { + loadOperations() + add(create) + add(duration) + add(end) + } + + // When + val reloaded = OperationModelStore(prefs).apply { loadOperations() }.list().toList() + + // Then + val reloadedCreate = reloaded[0] as CreateSessionOperation + reloadedCreate.idempotencyKey shouldBe "create-key" + reloadedCreate.localSessionId shouldBe "local-session-uuid" + reloadedCreate.startTime shouldBe 1_700_000_000_123L + val reloadedDuration = reloaded[1] as UpdateSessionOperation + reloadedDuration.idempotencyKey shouldBe "update-key" + reloadedDuration.endTime shouldBe null + reloadedDuration.activeDuration shouldBe 42_900L + (reloaded[2] as UpdateSessionOperation).endTime shouldBe 5_000L + } +}) From c08b5d1c89a75b937da5df36dd82bb0ba59bbda7 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Tue, 6 Oct 2026 17:13:51 -0700 Subject: [PATCH 2/9] feat: [SDK-5378] retry backed-off operations on app open and network return Co-Authored-By: Claude Opus 5.5 (1M context) --- .../java/com/onesignal/core/CoreModule.kt | 2 + .../internal/operations/IOperationRepo.kt | 6 ++ .../internal/operations/impl/OperationRepo.kt | 32 ++++++-- .../operations/impl/OperationRetryTrigger.kt | 50 ++++++++++++ .../internal/operations/OperationRepoTests.kt | 69 ++++++++++++++++ .../operations/OperationRetryTriggerTests.kt | 78 +++++++++++++++++++ 6 files changed, 232 insertions(+), 5 deletions(-) create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt create mode 100644 OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.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 8118671035..a1fa95a7cd 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 @@ -34,6 +34,7 @@ import com.onesignal.core.internal.language.impl.LanguageContext import com.onesignal.core.internal.operations.IOperationRepo import com.onesignal.core.internal.operations.impl.OperationModelStore import com.onesignal.core.internal.operations.impl.OperationRepo +import com.onesignal.core.internal.operations.impl.OperationRetryTrigger import com.onesignal.core.internal.permissions.IRequestPermissionService import com.onesignal.core.internal.permissions.impl.RequestPermissionService import com.onesignal.core.internal.preferences.IPreferencesService @@ -91,6 +92,7 @@ internal class CoreModule : IModule { builder.register() .provides() .provides() + builder.register().provides() // Permissions builder.register() diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/IOperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/IOperationRepo.kt index d2dceea5c3..a0fbcf924f 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/IOperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/IOperationRepo.kt @@ -42,6 +42,12 @@ interface IOperationRepo { suspend fun awaitInitialized() fun forceExecuteOperations() + + /** + * Retry operations waiting on failure backoff now, such as when the network returns. + * Unlike [forceExecuteOperations], a backend-requested Retry-After is still honored. + */ + fun retryNow() } // Extension function so the syntax containsInstanceOf() can be used over 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 23549db190..c24784b3b7 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 @@ -28,6 +28,7 @@ import kotlinx.coroutines.withTimeoutOrNull import java.util.UUID import kotlin.math.max import kotlin.reflect.KClass +import kotlin.time.TimeSource internal class OperationRepo( executors: List, @@ -69,6 +70,9 @@ internal class OperationRepo( private val retryWaiter = WaiterWithValue() private var paused = false + @Volatile + private var isWaitingToRetry = false + // Ops enqueued after their local IDs were translated would otherwise keep the local ID forever. private val appliedIdTranslations = mutableMapOf() private val initialized = CompletableDeferred() @@ -305,6 +309,13 @@ internal class OperationRepo( waiter.wake(LoopWaiterMessage(false)) } + override fun retryNow() { + // Only while waiting: the waiter is conflated, so a stale wake would cut short the next backoff. + if (isWaitingToRetry) { + retryWaiter.wake(LoopWaiterMessage(false)) + } + } + /** * Drops queued operations whose externalId is null. Called by the IV-aware HYDRATE * choreography in [OperationRepoIvExtensions] when `jwt_required` becomes REQUIRED @@ -508,20 +519,31 @@ internal class OperationRepo( /** * Wait which ever is longer, retryAfterSeconds returned by the server, - * or based on the retry count. + * or based on the retry count. [forceExecuteOperations] ends the wait; [retryNow] ends + * only the backoff part and still waits out retryAfterSeconds. */ suspend fun delayBeforeNextExecution( retries: Int, retryAfterSeconds: Int?, ) { Logging.debug("retryAfterSeconds: $retryAfterSeconds") - val retryAfterSecondsNonNull = retryAfterSeconds?.toLong() ?: 0L + val retryAfterMs = (retryAfterSeconds?.toLong() ?: 0L) * 1_000 val delayForOnRetries = retries * _configModelStore.model.opRepoDefaultFailRetryBackoff - val delayFor = max(delayForOnRetries, retryAfterSecondsNonNull * 1_000) + val delayFor = max(delayForOnRetries, retryAfterMs) if (delayFor < 1) return Logging.debug("Operations being delay for: $delayFor ms") - withTimeoutOrNull(delayFor) { - retryWaiter.waitForWake() + + val start = TimeSource.Monotonic.markNow() + var remaining = delayFor + isWaitingToRetry = true + try { + while (remaining > 0) { + val message = withTimeoutOrNull(remaining) { retryWaiter.waitForWake() } + val waitUntil = if (message == null || message.force) 0L else retryAfterMs + remaining = waitUntil - start.elapsedNow().inWholeMilliseconds + } + } finally { + isWaitingToRetry = false } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt new file mode 100644 index 0000000000..4f36f3c9f6 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt @@ -0,0 +1,50 @@ +package com.onesignal.core.internal.operations.impl + +import android.content.Context +import android.net.ConnectivityManager +import android.net.Network +import android.net.NetworkCapabilities +import android.net.NetworkRequest +import com.onesignal.core.internal.application.IApplicationLifecycleHandler +import com.onesignal.core.internal.application.IApplicationService +import com.onesignal.core.internal.operations.IOperationRepo +import com.onesignal.core.internal.startup.IStartableService +import com.onesignal.debug.internal.logging.Logging + +/** + * Retries operations waiting on failure backoff when the app opens or the network returns, + * rather than leaving them until the backoff runs out. + */ +internal class OperationRetryTrigger( + private val applicationService: IApplicationService, + private val operationRepo: IOperationRepo, +) : IStartableService, IApplicationLifecycleHandler { + override fun start() { + applicationService.addApplicationLifecycleHandler(this) + registerNetworkCallback() + } + + override fun onFocus(firedOnSubscribe: Boolean) = operationRepo.retryNow() + + override fun onUnfocused() = Unit + + private fun registerNetworkCallback() { + val connectivityManager = + applicationService.appContext.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager ?: return + val request = + NetworkRequest.Builder() + .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) + .build() + try { + connectivityManager.registerNetworkCallback( + request, + object : ConnectivityManager.NetworkCallback() { + override fun onAvailable(network: Network) = operationRepo.retryNow() + }, + ) + } catch (e: RuntimeException) { + // Throws SecurityException on some OEM builds and when the per-app callback limit is reached. + Logging.warn("OperationRetryTrigger: unable to register a network callback", e) + } + } +} 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 f6ec6a2740..8986e11f09 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 @@ -20,6 +20,7 @@ import com.onesignal.user.internal.operations.LoginUserOperation import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.ints.shouldBeGreaterThan import io.kotest.matchers.ints.shouldBeLessThan +import io.kotest.matchers.longs.shouldBeGreaterThanOrEqual import io.kotest.matchers.shouldBe import io.mockk.CapturingSlot import io.mockk.coEvery @@ -875,6 +876,64 @@ class OperationRepoTests : FunSpec({ mocks.operationRepo.queue[1].operation shouldBe op2 } + test("retryNow ends the failure backoff early") { + // Given + val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 60_000 + val executions = mocks.recordExecutions(ExecutionResponse(ExecutionResult.FAIL_RETRY), ExecutionResponse(ExecutionResult.SUCCESS)) + mocks.operationRepo.enqueue(mockOperation()) + mocks.operationRepo.start() + withTimeout(2_000) { while (executions.size < 1) delay(10) } + delay(100) + + // When + mocks.operationRepo.retryNow() + + // Then + withTimeout(2_000) { while (executions.size < 2) delay(10) } + } + + test("retryNow still waits out Retry-After") { + // Given + val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 60_000 + val executions = + mocks.recordExecutions( + ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = 1), + ExecutionResponse(ExecutionResult.SUCCESS), + ) + mocks.operationRepo.enqueue(mockOperation()) + mocks.operationRepo.start() + withTimeout(2_000) { while (executions.size < 1) delay(10) } + delay(100) + + // When + mocks.operationRepo.retryNow() + + // Then - retried once Retry-After passes, not immediately and not after the full backoff + delay(500) + executions.size shouldBe 1 + withTimeout(2_000) { while (executions.size < 2) delay(10) } + (executions[1] - executions[0]) shouldBeGreaterThanOrEqual 1_000L + } + + test("retryNow while not waiting to retry does not shorten a later backoff") { + // Given + val mocks = Mocks() + mocks.configModelStore.model.opRepoDefaultFailRetryBackoff = 60_000 + val executions = mocks.recordExecutions(ExecutionResponse(ExecutionResult.FAIL_RETRY), ExecutionResponse(ExecutionResult.SUCCESS)) + mocks.operationRepo.retryNow() + + // When + mocks.operationRepo.enqueue(mockOperation()) + mocks.operationRepo.start() + withTimeout(2_000) { while (executions.size < 1) delay(10) } + + // Then + delay(500) + executions.size shouldBe 1 + } + test("ensure forceExecuteOperations immediately wakes retry waiters") { // Given val mocks = Mocks() @@ -1204,6 +1263,16 @@ class OperationRepoTests : FunSpec({ } } +/** Responds with [responses] in order, repeating the last, and returns the time of each execution. */ +private fun Mocks.recordExecutions(vararg responses: ExecutionResponse): List { + val executions = java.util.Collections.synchronizedList(mutableListOf()) + coEvery { executor.execute(any()) } answers { + executions.add(System.currentTimeMillis()) + responses[minOf(executions.size - 1, responses.size - 1)] + } + return executions +} + private fun Mocks.waitForInternalEnqueue() { verify(timeout = 100) { operationRepo["internalEnqueue"]( diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt new file mode 100644 index 0000000000..31d1656c3d --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt @@ -0,0 +1,78 @@ +package com.onesignal.core.internal.operations + +import android.content.Context +import android.content.ContextWrapper +import android.net.ConnectivityManager +import android.net.Network +import android.net.NetworkRequest +import androidx.test.core.app.ApplicationProvider +import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest +import com.onesignal.core.internal.application.IApplicationService +import com.onesignal.core.internal.operations.impl.OperationRetryTrigger +import io.kotest.core.spec.style.FunSpec +import io.mockk.every +import io.mockk.just +import io.mockk.mockk +import io.mockk.runs +import io.mockk.slot +import io.mockk.verify + +private class ConnectivityContext( + base: Context, + private val connectivityManager: ConnectivityManager, +) : ContextWrapper(base) { + override fun getSystemService(name: String): Any? = + if (name == Context.CONNECTIVITY_SERVICE) connectivityManager else super.getSystemService(name) +} + +@RobolectricTest +class OperationRetryTriggerTests : FunSpec({ + fun setUp(connectivityManager: ConnectivityManager): Pair { + val operationRepo = mockk(relaxed = true) + val applicationService = mockk(relaxed = true) + every { applicationService.appContext } returns ConnectivityContext(ApplicationProvider.getApplicationContext(), connectivityManager) + return OperationRetryTrigger(applicationService, operationRepo) to operationRepo + } + + test("retries when the network becomes available") { + // Given + val callback = slot() + val connectivityManager = mockk() + every { connectivityManager.registerNetworkCallback(any(), capture(callback)) } just runs + val (trigger, operationRepo) = setUp(connectivityManager) + trigger.start() + + // When + callback.captured.onAvailable(mockk()) + + // Then + verify(exactly = 1) { operationRepo.retryNow() } + } + + test("retries when the app comes to the foreground") { + // Given + val connectivityManager = mockk(relaxed = true) + val (trigger, operationRepo) = setUp(connectivityManager) + + // When + trigger.onFocus(firedOnSubscribe = false) + + // Then + verify(exactly = 1) { operationRepo.retryNow() } + } + + test("starts even when the network callback cannot be registered") { + // Given + val connectivityManager = mockk() + every { connectivityManager.registerNetworkCallback(any(), any()) } throws + SecurityException("denied") + val (trigger, operationRepo) = setUp(connectivityManager) + + // When + trigger.start() + trigger.onFocus(firedOnSubscribe = false) + + // Then + verify(exactly = 1) { operationRepo.retryNow() } + } +}) From 16190c0af521133a415354e43451944c5273f523 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Tue, 6 Oct 2026 18:56:22 -0700 Subject: [PATCH 3/9] fix(retry): skip network retry when app backgrounded --- .../operations/impl/OperationRetryTrigger.kt | 9 +++++--- .../operations/OperationRetryTriggerTests.kt | 23 +++++++++++++++++-- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt index 4f36f3c9f6..98ab2d4efe 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRetryTrigger.kt @@ -12,8 +12,8 @@ import com.onesignal.core.internal.startup.IStartableService import com.onesignal.debug.internal.logging.Logging /** - * Retries operations waiting on failure backoff when the app opens or the network returns, - * rather than leaving them until the backoff runs out. + * Retries operations waiting on failure backoff when the app opens or the network returns while + * the app is in the foreground, rather than leaving them until the backoff runs out. */ internal class OperationRetryTrigger( private val applicationService: IApplicationService, @@ -39,7 +39,10 @@ internal class OperationRetryTrigger( connectivityManager.registerNetworkCallback( request, object : ConnectivityManager.NetworkCallback() { - override fun onAvailable(network: Network) = operationRepo.retryNow() + override fun onAvailable(network: Network) { + // Android 15+ blocks requests from cached background apps; a blocked attempt only lengthens the backoff. + if (applicationService.isInForeground) operationRepo.retryNow() + } }, ) } catch (e: RuntimeException) { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt index 31d1656c3d..ab7d5f6382 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRetryTriggerTests.kt @@ -27,14 +27,18 @@ private class ConnectivityContext( @RobolectricTest class OperationRetryTriggerTests : FunSpec({ - fun setUp(connectivityManager: ConnectivityManager): Pair { + fun setUp( + connectivityManager: ConnectivityManager, + isInForeground: Boolean = true, + ): Pair { val operationRepo = mockk(relaxed = true) val applicationService = mockk(relaxed = true) + every { applicationService.isInForeground } returns isInForeground every { applicationService.appContext } returns ConnectivityContext(ApplicationProvider.getApplicationContext(), connectivityManager) return OperationRetryTrigger(applicationService, operationRepo) to operationRepo } - test("retries when the network becomes available") { + test("retries when the network becomes available while the app is in the foreground") { // Given val callback = slot() val connectivityManager = mockk() @@ -49,6 +53,21 @@ class OperationRetryTriggerTests : FunSpec({ verify(exactly = 1) { operationRepo.retryNow() } } + test("does not retry when the network becomes available while the app is in the background") { + // Given + val callback = slot() + val connectivityManager = mockk() + every { connectivityManager.registerNetworkCallback(any(), capture(callback)) } just runs + val (trigger, operationRepo) = setUp(connectivityManager, isInForeground = false) + trigger.start() + + // When + callback.captured.onAvailable(mockk()) + + // Then + verify(exactly = 0) { operationRepo.retryNow() } + } + test("retries when the app comes to the foreground") { // Given val connectivityManager = mockk(relaxed = true) From 5d5509058ec8c3922dccd359e0d2ae4e9f3d0fe9 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Tue, 6 Oct 2026 19:03:02 -0700 Subject: [PATCH 4/9] fix: [SDK-5378] address review: serialize serverSessionId write, discard updates after in-flight end or for orphaned sessions Co-Authored-By: Claude Opus 5.5 (1M context) --- .../internal/operations/impl/OperationRepo.kt | 11 ++- .../impl/OperationRepoSessionExtensions.kt | 26 ++++--- .../impl/SessionOperationExecutor.kt | 11 ++- .../operations/SessionOperationQueueTests.kt | 77 +++++++++++++++++++ 4 files changed, 112 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 c24784b3b7..e2353e5182 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 @@ -73,6 +73,12 @@ internal class OperationRepo( @Volatile private var isWaitingToRetry = false + /** Ops removed from [queue] by [getNextOps] and not yet finished; guarded by [queue]. */ + internal var inFlight: List = emptyList() + private set + + internal val isInitialized: Boolean get() = initialized.isCompleted + // Ops enqueued after their local IDs were translated would otherwise keep the local ID forever. private val appliedIdTranslations = mutableMapOf() private val initialized = CompletableDeferred() @@ -270,7 +276,7 @@ internal class OperationRepo( } else if (isNew && appliedIdTranslations.isNotEmpty()) { op.translateIds(appliedIdTranslations) } - return !isDuplicate && (!isNew || op !is UpdateSessionOperation || coalesceSessionUpdate(queueItem, _operationModelStore)) + return !isDuplicate && (!isNew || op !is UpdateSessionOperation || admitSessionUpdate(queueItem, _operationModelStore)) } /** @@ -395,6 +401,7 @@ internal class OperationRepo( } internal suspend fun executeOperations(ops: List) { + synchronized(queue) { inFlight = ops } try { val startingOp = ops.first() val executor = @@ -505,6 +512,8 @@ internal class OperationRepo( } catch (e: Throwable) { Logging.log(LogLevel.ERROR, "Error attempting to execute operation: $ops", e) dropAndWake(ops) + } finally { + synchronized(queue) { inFlight = emptyList() } } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt index 37b0269dd8..e7ef2fdf42 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.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 @@ -17,23 +18,28 @@ internal const val MAX_QUEUED_SESSION_OPERATIONS = 100 * Folds queued, unsent updates for the same session into [incoming], which then carries the * highest duration. An update that ends the session replaces them the same way. * - * @return false if [incoming] should be discarded because the session's end is already queued. + * @return false if [incoming] should be discarded: the session's end is already queued or + * sending, or its session ID is local and no create for it remains, so it could never be sent. */ -internal fun OperationRepo.coalesceSessionUpdate( +internal fun OperationRepo.admitSessionUpdate( incoming: OperationRepo.OperationQueueItem, operationModelStore: OperationModelStore, ): Boolean { val op = incoming.operation as UpdateSessionOperation - val queued = - queue.filter { - val other = it.operation - other is UpdateSessionOperation && other.appId == op.appId && other.sessionId == op.sessionId - } - if (queued.any { (it.operation as UpdateSessionOperation).isEnd }) { - Logging.debug("OperationRepo: session ${op.sessionId} already has an end queued, discarding $op") + val pending = (queue + inFlight).map { it.operation } + val hasEnd = pending.any { it is UpdateSessionOperation && it.isSameSession(op) && it.isEnd } + // Saved ops may not be loaded yet, so a missing create only proves an orphan after initialization. + val isOrphaned = + isInitialized && + IDManager.isLocalId(op.sessionId) && + pending.none { it is CreateSessionOperation && it.localSessionId == op.sessionId } + if (hasEnd || isOrphaned) { + Logging.debug("OperationRepo: discarding $op, session ended: $hasEnd, never created: $isOrphaned") incoming.waiter?.wake(false) return false } + + val queued = queue.filter { (it.operation as? UpdateSessionOperation)?.isSameSession(op) == true } queued.maxOfOrNull { (it.operation as UpdateSessionOperation).activeDuration }?.let { op.activeDuration = maxOf(op.activeDuration, it) } @@ -90,6 +96,8 @@ internal fun OperationRepo.dropOrphanedSessionUpdates( removeQueuedOperations(orphans, operationModelStore) } +private fun UpdateSessionOperation.isSameSession(other: UpdateSessionOperation) = appId == other.appId && sessionId == other.sessionId + private fun OperationRepo.removeQueuedOperations( items: List, operationModelStore: OperationModelStore, 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 51fb1ed4cf..431ed50b68 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,5 +1,6 @@ package com.onesignal.session.internal.session.operations.impl +import com.onesignal.common.threading.OneSignalDispatchers import com.onesignal.core.internal.device.IDeviceService import com.onesignal.core.internal.operations.ExecutionResponse import com.onesignal.core.internal.operations.ExecutionResult @@ -13,6 +14,7 @@ import com.onesignal.session.internal.session.backend.SessionsApiResult 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 kotlinx.coroutines.withContext import java.util.concurrent.TimeUnit.MILLISECONDS internal class SessionOperationExecutor( @@ -44,9 +46,12 @@ internal class SessionOperationExecutor( is SessionsApiResult.Success -> { val serverSessionId = result.value // Persisted so updates enqueued after this point, including after a restart, use the backend ID directly. - val session = sessionModelStore.model - if (session.localSessionId == op.localSessionId) { - session.serverSessionId = serverSessionId + // On SerialIO so the check and write can't interleave with SessionService starting a new session. + withContext(OneSignalDispatchers.SerialIO) { + val session = sessionModelStore.model + if (session.localSessionId == op.localSessionId) { + session.serverSessionId = serverSessionId + } } ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to serverSessionId)) } 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 c90e616b82..180f3cd573 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 @@ -74,6 +74,20 @@ private class SessionQueueMocks( CoreInternalMocks.identityVerificationService(newCodePathsRun = ivRequired, ivBehaviorActive = ivRequired), ) + /** The first execution suspends until [release] completes; [started] completes once it begins. */ + fun holdFirstExecution( + started: CompletableDeferred, + release: CompletableDeferred, + response: (Operation) -> ExecutionResponse, + ) { + coEvery { executor.execute(any()) } coAnswers { + val op = firstArg>().single() + executed.add(op) + if (started.complete(Unit)) release.await() + response(op) + } + } + fun respondWith(vararg responses: (Operation) -> ExecutionResponse) { var call = 0 coEvery { executor.execute(any()) } answers { @@ -278,6 +292,69 @@ class SessionOperationQueueTests : FunSpec({ keys shouldContainExactly listOf(createOp.idempotencyKey, createOp.idempotencyKey) } + test("an update for a session whose end is being sent is discarded") { + // Given + val mocks = SessionQueueMocks() + val started = CompletableDeferred() + val release = CompletableDeferred() + mocks.holdFirstExecution(started, release) { ExecutionResponse(ExecutionResult.SUCCESS) } + val end = update(activeDuration = 15_000, sessionId = "server-session", endTime = 5_000L) + mocks.enqueueAll(end) + mocks.operationRepo.start() + withTimeout(2_000) { started.await() } + + // When + mocks.enqueueAll(update(activeDuration = 20_000, sessionId = "server-session")) + release.complete(Unit) + withTimeout(2_000) { while (mocks.storedOperations.isNotEmpty()) delay(10) } + delay(100) + + // Then + mocks.executed shouldContainExactly listOf(end) + } + + test("an update for a local session with no pending create is discarded") { + // Given + val mocks = SessionQueueMocks() + mocks.operationRepo.start() + withTimeout(2_000) { mocks.operationRepo.awaitInitialized() } + + // When + mocks.enqueueAll(update(activeDuration = 10_000)) + + // Then + mocks.queuedOperations shouldBe emptyList() + mocks.storedOperations shouldBe emptyList() + } + + test("an update for a session whose create is being sent is kept and translated") { + // Given + val mocks = SessionQueueMocks() + val started = CompletableDeferred() + val release = CompletableDeferred() + mocks.holdFirstExecution(started, release) { + if (it is CreateSessionOperation) { + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(LOCAL_SESSION_ID to "server-session")) + } else { + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + val createOp = create() + mocks.enqueueAll(createOp) + mocks.operationRepo.start() + withTimeout(2_000) { started.await() } + + // When + val updateOp = update(activeDuration = 10_000) + mocks.enqueueAll(updateOp) + release.complete(Unit) + withTimeout(2_000) { while (mocks.executed.size < 2) delay(10) } + + // Then + mocks.executed shouldContainExactly listOf(createOp, updateOp) + updateOp.sessionId shouldBe "server-session" + } + test("session operations are not suppressed or purged as anonymous under identity verification") { // Given val mocks = SessionQueueMocks(ivRequired = true) From 6a20fa67819d1eaf2465ad6a8050f9d03d072476 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Tue, 6 Oct 2026 19:13:15 -0700 Subject: [PATCH 5/9] test: [SDK-5378] wait for the post-create delay before verifying call order Co-Authored-By: Claude Opus 5.5 (1M context) --- .../onesignal/core/internal/operations/OperationRepoTests.kt | 2 ++ 1 file changed, 2 insertions(+) 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 8986e11f09..61c5aef175 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 @@ -794,6 +794,8 @@ class OperationRepoTests : FunSpec({ // Then response shouldBe true + // The waiter wakes before the delay starts, so wait for the delay call before checking order. + coVerify(timeout = 1_000) { mocks.operationRepo.delayBeforeNextExecution(any(), any()) } coVerifyOrder { // ensure the order: IDs are translated, operation removed from the store, then delay for postCreateDelay operation.translateIds(idTranslation) From a186ec22a75ed093cf5786881a2e1e85ee37fa9b Mon Sep 17 00:00:00 2001 From: Fadi George Date: Wed, 7 Oct 2026 17:01:56 -0700 Subject: [PATCH 6/9] refactor: [SDK-5378] classify BackendException with NetworkUtils in the session executor Co-Authored-By: Claude Opus 5.5 (1M context) --- .../impl/SessionOperationExecutor.kt | 74 +++++++++++++------ .../SessionOperationExecutorTests.kt | 62 +++++++++++----- 2 files changed, 93 insertions(+), 43 deletions(-) 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 431ed50b68..b04f4ad69e 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,5 +1,7 @@ package com.onesignal.session.internal.session.operations.impl +import com.onesignal.common.NetworkUtils +import com.onesignal.common.exceptions.BackendException import com.onesignal.common.threading.OneSignalDispatchers import com.onesignal.core.internal.device.IDeviceService import com.onesignal.core.internal.operations.ExecutionResponse @@ -10,7 +12,6 @@ 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.SessionsApiResult import com.onesignal.session.internal.session.backend.UpdateSessionRequest import com.onesignal.session.internal.session.operations.CreateSessionOperation import com.onesignal.session.internal.session.operations.UpdateSessionOperation @@ -42,24 +43,36 @@ internal class SessionOperationExecutor( idempotencyKey = op.idempotencyKey, directAttributionId = op.directAttributionId, ) - return when (val result = sessionsBackend.createSession(op.appId, request)) { - is SessionsApiResult.Success -> { - val serverSessionId = result.value - // Persisted so updates enqueued after this point, including after a restart, use the backend ID directly. - // On SerialIO so the check and write can't interleave with SessionService starting a new session. - withContext(OneSignalDispatchers.SerialIO) { - val session = sessionModelStore.model - if (session.localSessionId == op.localSessionId) { - session.serverSessionId = serverSessionId - } - } - ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to serverSessionId)) + val serverSessionId = + try { + sessionsBackend.createSession(op.appId, request) + } catch (ex: BackendException) { + return failure("create", ex) } - is SessionsApiResult.Retry -> retry(result) - is SessionsApiResult.Drop -> drop("create", result) + return if (serverSessionId == null) { + // Retrying with the same idempotency key lets the backend return the session it already created. + Logging.warn("SessionOperationExecutor: create session response is missing data.session_id, retrying") + ExecutionResponse(ExecutionResult.FAIL_RETRY) + } else { + sessionCreated(op, serverSessionId) } } + private suspend fun sessionCreated( + op: CreateSessionOperation, + serverSessionId: String, + ): ExecutionResponse { + // Persisted so updates enqueued after this point, including after a restart, use the backend ID directly. + // On SerialIO so the check and write can't interleave with SessionService starting a new session. + withContext(OneSignalDispatchers.SerialIO) { + val session = sessionModelStore.model + if (session.localSessionId == op.localSessionId) { + session.serverSessionId = serverSessionId + } + } + return ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to serverSessionId)) + } + private suspend fun updateSession(op: UpdateSessionOperation): ExecutionResponse { val request = UpdateSessionRequest( @@ -69,25 +82,38 @@ internal class SessionOperationExecutor( idempotencyKey = op.idempotencyKey, endTime = op.endTime?.let { MILLISECONDS.toSeconds(it) }, ) - return when (val result = sessionsBackend.updateSession(op.appId, op.sessionId, request)) { - is SessionsApiResult.Success -> ExecutionResponse(ExecutionResult.SUCCESS) - is SessionsApiResult.Retry -> retry(result) - is SessionsApiResult.Drop -> drop("update", result) + try { + sessionsBackend.updateSession(op.appId, op.sessionId, request) + } catch (ex: BackendException) { + return failure("update", ex) } + return ExecutionResponse(ExecutionResult.SUCCESS) } - private fun retry(result: SessionsApiResult.Retry) = ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = result.retryAfterSeconds) - - private fun drop( + private fun failure( action: String, - result: SessionsApiResult.Drop, + ex: BackendException, ): ExecutionResponse { - Logging.warn("SessionOperationExecutor: $action session failed with ${result.statusCode}, dropping") + val code = ex.statusCode + // RETRYABLE also covers 4xx codes NetworkUtils doesn't list, like 413 and 422. Those won't succeed on retry, + // and OperationRepo retries without a cap at the front of the queue, so they would block every other operation. + val retryable = + NetworkUtils.getResponseStatusType(code) == NetworkUtils.ResponseStatusType.RETRYABLE && + (code !in CLIENT_ERROR_CODES || code in RETRYABLE_CLIENT_ERROR_CODES) + if (retryable) { + return ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = ex.retryAfterSeconds) + } + Logging.warn("SessionOperationExecutor: $action session failed with $code, dropping") return ExecutionResponse(ExecutionResult.FAIL_NORETRY) } companion object { const val CREATE_SESSION = "create-session" const val UPDATE_SESSION = "update-session" + + private val CLIENT_ERROR_CODES = 400..499 + + // Request Timeout and Too Many Requests + private val RETRYABLE_CLIENT_ERROR_CODES = setOf(408, 429) } } 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 cb49c935bd..bcff2b93e2 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,5 +1,6 @@ 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 @@ -8,7 +9,6 @@ import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService import com.onesignal.session.internal.session.backend.CreateSessionRequest import com.onesignal.session.internal.session.backend.ISessionsBackendService -import com.onesignal.session.internal.session.backend.SessionsApiResult import com.onesignal.session.internal.session.backend.UpdateSessionRequest import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor import io.kotest.core.spec.style.FunSpec @@ -49,7 +49,7 @@ class SessionOperationExecutorTests : FunSpec({ test("create sends the request in seconds and translates the local session ID") { // Given val backend = mockk() - coEvery { backend.createSession(any(), any()) } returns SessionsApiResult.Success("server-session") + coEvery { backend.createSession(any(), any()) } returns "server-session" val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "session-uuid" } val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), sessionModelStore) @@ -78,7 +78,7 @@ class SessionOperationExecutorTests : FunSpec({ test("create for a session that is no longer current leaves the current session alone") { // Given val backend = mockk() - coEvery { backend.createSession(any(), any()) } returns SessionsApiResult.Success("server-session") + coEvery { backend.createSession(any(), any()) } returns "server-session" val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "newer-session" } val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), sessionModelStore) @@ -90,37 +90,61 @@ class SessionOperationExecutorTests : FunSpec({ sessionModelStore.model.serverSessionId shouldBe null } - test("a retryable failure retries and passes Retry-After through") { - // Given - val backend = mockk() - coEvery { backend.createSession(any(), any()) } returns SessionsApiResult.Retry(429, retryAfterSeconds = 30) - val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + test("network errors, 408, 429, and 5xx are retried with Retry-After") { + listOf(-1, 0, 408, 429, 500, 503).forEach { statusCode -> + // Given + val backend = mockk() + coEvery { backend.createSession(any(), any()) } throws BackendException(statusCode, retryAfterSeconds = 30) + coEvery { backend.updateSession(any(), any(), any()) } throws BackendException(statusCode, retryAfterSeconds = 30) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + + // When + val responses = listOf(executor.execute(listOf(createOp())), executor.execute(listOf(updateOp()))) + + // Then + responses.forEach { + it.result shouldBe ExecutionResult.FAIL_RETRY + it.retryAfterSeconds shouldBe 30 + } + } + } - // When - val response = executor.execute(listOf(createOp())) + test("other 4xx failures are not retried") { + listOf(400, 401, 403, 404, 409, 410, 413, 422).forEach { statusCode -> + // Given + val backend = mockk() + coEvery { backend.createSession(any(), any()) } throws BackendException(statusCode) + coEvery { backend.updateSession(any(), any(), any()) } throws BackendException(statusCode) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) - // Then - response.result shouldBe ExecutionResult.FAIL_RETRY - response.retryAfterSeconds shouldBe 30 + // When + val responses = listOf(executor.execute(listOf(createOp())), executor.execute(listOf(updateOp()))) + + // Then + responses.forEach { it.result shouldBe ExecutionResult.FAIL_NORETRY } + } } - test("a permanent failure is not retried") { + test("create retries a success response without a session ID") { // Given val backend = mockk() - coEvery { backend.updateSession(any(), any(), any()) } returns SessionsApiResult.Drop(400) - val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + coEvery { backend.createSession(any(), any()) } returns null + val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "session-uuid" } + val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), sessionModelStore) // When - val response = executor.execute(listOf(updateOp())) + val response = executor.execute(listOf(createOp(localSessionId = sessionModelStore.model.localSessionId))) // Then - response.result shouldBe ExecutionResult.FAIL_NORETRY + response.result shouldBe ExecutionResult.FAIL_RETRY + response.idTranslations shouldBe null + sessionModelStore.model.serverSessionId shouldBe null } test("update sends the duration and end time in seconds") { // Given val backend = mockk() - coEvery { backend.updateSession(any(), any(), any()) } returns SessionsApiResult.Success(Unit) + coEvery { backend.updateSession(any(), any(), any()) } returns Unit val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) // When From 890c79dc1780afc56651ede76bf8743d1f0a78bf Mon Sep 17 00:00:00 2001 From: Fadi George Date: Thu, 8 Oct 2026 12:34:48 -0700 Subject: [PATCH 7/9] fix: [SDK-5378] address review: cap session op attempts, translate session pins in OperationRepo, drop unsendable ops under IV - Session ops drop after 5 failed attempts the backend answered. The count is persisted, and network errors don't count, so offline still retries until reconnect. - OperationRepo writes each ID translation it applies into the current session's pinned IDs and server session ID, replacing SessionService's PinnedLocalIdTranslator. - Under required identity verification, session ops for a local OneSignal ID that no identified login will create are discarded at enqueue and dropped with the anonymous purge. - CreateSessionOperation and UpdateSessionOperation share a SessionOperation base. Co-Authored-By: Claude Opus 5.5 --- .../internal/operations/impl/OperationRepo.kt | 29 ++-- .../impl/OperationRepoSessionExtensions.kt | 50 ++++++- .../session/internal/session/SessionModel.kt | 10 ++ .../internal/session/impl/SessionService.kt | 46 ------ .../operations/CreateSessionOperation.kt | 43 +----- .../session/operations/SessionOperation.kt | 58 ++++++++ .../operations/UpdateSessionOperation.kt | 35 +---- .../impl/SessionOperationExecutor.kt | 63 +++++---- .../internal/operations/OperationRepoTests.kt | 5 + .../operations/SessionOperationQueueTests.kt | 78 +++++++++++ .../internal/session/SessionServiceTests.kt | 132 ------------------ .../SessionOperationExecutorTests.kt | 70 ++++++---- .../RecoverFromDroppedLoginBugTests.kt | 1 + 13 files changed, 299 insertions(+), 321 deletions(-) create mode 100644 OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionOperation.kt 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 e2353e5182..d3ac89301a 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 @@ -2,6 +2,7 @@ package com.onesignal.core.internal.operations.impl import com.onesignal.common.IDManager import com.onesignal.common.threading.WaiterWithValue +import com.onesignal.common.threading.runOnSerialIO import com.onesignal.common.threading.suspendifyOnIO import com.onesignal.core.internal.config.ConfigModelStore import com.onesignal.core.internal.config.impl.IdentityVerificationService @@ -14,7 +15,8 @@ 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.UpdateSessionOperation +import com.onesignal.session.internal.session.SessionModelStore +import com.onesignal.session.internal.session.operations.SessionOperation import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.LoginUserOperation @@ -30,6 +32,7 @@ import kotlin.math.max import kotlin.reflect.KClass import kotlin.time.TimeSource +@Suppress("LongParameterList") internal class OperationRepo( executors: List, private val _operationModelStore: OperationModelStore, @@ -38,6 +41,7 @@ internal class OperationRepo( private val _newRecordState: NewRecordsState, private val _jwtTokenStore: JwtTokenStore, private val _identityVerificationService: IdentityVerificationService, + private val sessionModelStore: SessionModelStore, ) : IOperationRepo, IStartableService { internal class OperationQueueItem( @@ -184,17 +188,19 @@ internal class OperationRepo( * on the legacy enqueue path. */ private fun shouldSuppressAnonymousOp(op: Operation): Boolean { - if (!_identityVerificationService.newCodePathsRun) return false if (op is LoginUserOperation || !op.requiresJwt) return false - val suppress = - _configModelStore.model.useIdentityVerification == JwtRequirement.REQUIRED && - op.externalId == null + val suppress = isIdentityVerificationRequired && op.externalId == null if (suppress) { Logging.debug("OperationRepo: suppressing anonymous op under IV-required: $op") } return suppress } + internal val isIdentityVerificationRequired: Boolean + get() = + _identityVerificationService.newCodePathsRun && + _configModelStore.model.useIdentityVerification == JwtRequirement.REQUIRED + /** * Only used inside this class, adds OperationQueueItem to queue * WARNING: Never set flush=true until budget rules are added, even for internal use! @@ -270,13 +276,14 @@ internal class OperationRepo( isNew: Boolean, ): Boolean { val op = queueItem.operation - val isDuplicate = queue.any { it.operation.id == op.id } - if (isDuplicate) { + if (queue.any { it.operation.id == op.id }) { Logging.debug("OperationRepo: internalEnqueue - operation.id: ${op.id} already exists in the queue.") - } else if (isNew && appliedIdTranslations.isNotEmpty()) { + return false + } + if (isNew && appliedIdTranslations.isNotEmpty()) { op.translateIds(appliedIdTranslations) } - return !isDuplicate && (!isNew || op !is UpdateSessionOperation || admitSessionUpdate(queueItem, _operationModelStore)) + return !isNew || op !is SessionOperation || admitSessionOperation(queueItem, _operationModelStore) } /** @@ -346,6 +353,7 @@ internal class OperationRepo( } } Logging.debug("OperationRepo: removeOperationsWithoutExternalId removed ${anonymous.size} of ${anonymous.size + queue.size} operations") + dropUnsendableSessionOperations(_operationModelStore) anonymous.map { it.operation.id } } // Persistent store removal outside the queue lock; ModelStore has its own locking. @@ -422,6 +430,9 @@ internal class OperationRepo( appliedIdTranslations.putAll(response.idTranslations) } response.idTranslations.values.forEach { _newRecordState.add(it) } + // On SerialIO, where SessionService starts sessions, so the session can't change mid-write. + // Updates built from the old IDs before this runs are still translated at enqueue. + runOnSerialIO { sessionModelStore.model.translateIds(response.idTranslations) } } var highestRetries = 0 diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt index e7ef2fdf42..593bc8d913 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt @@ -4,7 +4,9 @@ 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 +import com.onesignal.session.internal.session.operations.SessionOperation import com.onesignal.session.internal.session.operations.UpdateSessionOperation +import com.onesignal.user.internal.operations.LoginUserOperation /** * Session operations accumulate while offline, one or more per app open. Coalescing keeps @@ -14,6 +16,48 @@ import com.onesignal.session.internal.session.operations.UpdateSessionOperation internal const val MAX_QUEUED_SESSION_OPERATIONS = 100 +/** + * @return false if [incoming] was discarded because it could never be sent; see [admitSessionUpdate]. + */ +internal fun OperationRepo.admitSessionOperation( + incoming: OperationRepo.OperationQueueItem, + operationModelStore: OperationModelStore, +): Boolean { + val op = incoming.operation as SessionOperation + if (isUnsendable(op)) { + Logging.debug("OperationRepo: discarding $op, its anonymous user is never created under identity verification") + incoming.waiter?.wake(false) + return false + } + return op !is UpdateSessionOperation || admitSessionUpdate(incoming, operationModelStore) +} + +/** + * Drops queued session operations that can never be sent, after anonymous operations are purged. + */ +internal fun OperationRepo.dropUnsendableSessionOperations(operationModelStore: OperationModelStore) { + val unsendable = queue.filter { (it.operation as? SessionOperation)?.let(::isUnsendable) == true } + if (unsendable.isEmpty()) return + + Logging.warn("OperationRepo: dropping ${unsendable.size} session operation(s) for an anonymous user under identity verification") + unsendable.forEach { it.waiter?.wake(false) } + removeQueuedOperations(unsendable, operationModelStore) + dropOrphanedSessionUpdates(unsendable, operationModelStore) +} + +/** + * Under required identity verification the anonymous user is never created, so a local OneSignal ID + * that no identified login will create never gets a backend ID. + */ +private fun OperationRepo.isUnsendable(op: SessionOperation): Boolean = + isInitialized && + isIdentityVerificationRequired && + IDManager.isLocalId(op.onesignalId) && + (queue + inFlight).none { + val login = it.operation as? LoginUserOperation + login != null && login.onesignalId == op.onesignalId && login.externalId != null + } + /** * Folds queued, unsent updates for the same session into [incoming], which then carries the * highest duration. An update that ends the session replaces them the same way. @@ -21,7 +65,7 @@ internal const val MAX_QUEUED_SESSION_OPERATIONS = 100 * @return false if [incoming] should be discarded: the session's end is already queued or * sending, or its session ID is local and no create for it remains, so it could never be sent. */ -internal fun OperationRepo.admitSessionUpdate( +private fun OperationRepo.admitSessionUpdate( incoming: OperationRepo.OperationQueueItem, operationModelStore: OperationModelStore, ): Boolean { @@ -64,10 +108,10 @@ internal fun OperationRepo.enforceSessionOperationCap( added: Operation, operationModelStore: OperationModelStore, ) { - if (added !is CreateSessionOperation && added !is UpdateSessionOperation) return + if (added !is SessionOperation) return while (true) { - val sessionItems = queue.filter { it.operation is CreateSessionOperation || it.operation is UpdateSessionOperation } + val sessionItems = queue.filter { it.operation is SessionOperation } if (sessionItems.size <= MAX_QUEUED_SESSION_OPERATIONS) return val oldest = sessionItems.first() 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 576c1368a7..b70fe89928 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 @@ -111,4 +111,14 @@ class SessionModel : Model() { */ internal val localSessionId: String get() = "${IDManager.LOCAL_PREFIX}$sessionId" + + /** + * Applies the same backend IDs the operation queue gives this session's operations, so + * updates built from these fields, including after a restart, match them. + */ + internal fun translateIds(map: Map) { + onesignalId?.let { map[it] }?.let { onesignalId = it } + subscriptionId?.let { map[it] }?.let { subscriptionId = it } + map[localSessionId]?.let { serverSessionId = it } + } } 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..689f774e18 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 @@ -1,10 +1,6 @@ package com.onesignal.session.internal.session.impl -import com.onesignal.common.IDManager import com.onesignal.common.events.EventProducer -import com.onesignal.common.modeling.ISingletonModelStoreChangeHandler -import com.onesignal.common.modeling.Model -import com.onesignal.common.modeling.ModelChangedArgs import com.onesignal.common.threading.runOnSerialIO import com.onesignal.core.internal.application.IApplicationLifecycleHandler import com.onesignal.core.internal.application.IApplicationService @@ -25,7 +21,6 @@ import com.onesignal.session.internal.session.SessionModelStore import com.onesignal.user.internal.backend.IdentityConstants import com.onesignal.user.internal.identity.IdentityModelStore import java.util.UUID -import kotlin.reflect.KMutableProperty1 /** * The implementation for [ISessionService] will continue a session as long as the app remains @@ -81,47 +76,6 @@ internal class SessionService( override fun start() { applicationService.addApplicationLifecycleHandler(this) - identityModelStore.subscribe(PinnedLocalIdTranslator(IdentityConstants.ONESIGNAL_ID, SessionModel::onesignalId)) - // Login before the anonymous user is created carries the local push subscription over to - // the new user, so its backend ID belongs to that user, not the one pinned on this session. - configModelStore.subscribe( - PinnedLocalIdTranslator(ConfigModel::pushSubscriptionId.name, SessionModel::subscriptionId) { session -> - val identityModel = identityModelStore.model - val currentOnesignalId = if (identityModel.hasProperty(IdentityConstants.ONESIGNAL_ID)) identityModel.onesignalId else null - session.onesignalId == currentOnesignalId - }, - ) - } - - /** - * A session that starts before the user is created pins local IDs. When the backend replaces - * a local ID in place, carry it over to the pinned copy. Changes from a non-local ID, such as a - * login or user switch, leave the pinned IDs alone. - */ - private inner class PinnedLocalIdTranslator( - private val sourceProperty: String, - private val pinnedId: KMutableProperty1, - private val appliesTo: (SessionModel) -> Boolean = { true }, - ) : ISingletonModelStoreChangeHandler { - override fun onModelReplaced( - model: TModel, - tag: String, - ) = Unit - - override fun onModelUpdated( - args: ModelChangedArgs, - tag: String, - ) { - val localId = - (args.oldValue as? String)?.takeIf { args.property == sourceProperty && IDManager.isLocalId(it) } - val backendId = (args.newValue as? String)?.takeUnless { IDManager.isLocalId(it) } - if (localId == null || backendId == null) return - - runOnSerialIO { - val session = this@SessionService.session ?: return@runOnSerialIO - if (pinnedId.get(session) == localId && appliesTo(session)) pinnedId.set(session, backendId) - } - } } /** NOTE: This triggers more often than scheduleBackgroundRunIn defined above, 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 30395e01cc..e03b268821 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 @@ -1,7 +1,6 @@ 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 @@ -9,13 +8,7 @@ import java.util.UUID /** * An [Operation] to create a session through the sessions API. */ -class CreateSessionOperation() : Operation(SessionOperationExecutor.CREATE_SESSION) { - var appId: String - get() = getStringProperty(::appId.name) - private set(value) { - setStringProperty(::appId.name, value) - } - +class CreateSessionOperation() : SessionOperation(SessionOperationExecutor.CREATE_SESSION) { /** * Stands in for the backend session ID until this operation succeeds, so queued * [UpdateSessionOperation]s can reference the session and be translated. @@ -26,24 +19,6 @@ class CreateSessionOperation() : Operation(SessionOperationExecutor.CREATE_SESSI setStringProperty(::localSessionId.name, value) } - /** - * May be a local ID; see [IDManager.isLocalId]. - */ - var onesignalId: String - get() = getStringProperty(::onesignalId.name) - private set(value) { - setStringProperty(::onesignalId.name, value) - } - - /** - * May be a local ID; see [IDManager.isLocalId]. - */ - 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) @@ -57,21 +32,10 @@ class CreateSessionOperation() : Operation(SessionOperationExecutor.CREATE_SESSI setOptStringProperty(::directAttributionId.name, value) } - /** - * Generated once and persisted, so every retry, including after a restart, is deduplicated by the backend. - */ - var idempotencyKey: String - get() = getStringProperty(::idempotencyKey.name) - private set(value) { - setStringProperty(::idempotencyKey.name, value) - } - override val createComparisonKey: String get() = "$appId.Session.$localSessionId" override val modifyComparisonKey: String get() = "$appId.Session.$localSessionId" - override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE override val canStartExecute: Boolean get() = !IDManager.isLocalId(onesignalId) && !IDManager.isLocalId(subscriptionId) override val applyToRecordId: String get() = onesignalId - override val requiresJwt: Boolean get() = false constructor( appId: String, @@ -90,9 +54,4 @@ class CreateSessionOperation() : Operation(SessionOperationExecutor.CREATE_SESSI this.directAttributionId = directAttributionId this.idempotencyKey = idempotencyKey } - - 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/SessionOperation.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionOperation.kt new file mode 100644 index 0000000000..04c369b2ec --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionOperation.kt @@ -0,0 +1,58 @@ +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 + +/** + * Fields shared by [CreateSessionOperation] and [UpdateSessionOperation]. + */ +abstract class SessionOperation(name: String) : Operation(name) { + var appId: String + get() = getStringProperty(::appId.name) + protected set(value) { + setStringProperty(::appId.name, value) + } + + /** + * May be a local ID; see [IDManager.isLocalId]. + */ + var onesignalId: String + get() = getStringProperty(::onesignalId.name) + protected set(value) { + setStringProperty(::onesignalId.name, value) + } + + /** + * May be a local ID; see [IDManager.isLocalId]. + */ + var subscriptionId: String + get() = getStringProperty(::subscriptionId.name) + protected set(value) { + setStringProperty(::subscriptionId.name, value) + } + + /** + * Generated once and persisted, so every retry, including after a restart, is deduplicated by the backend. + */ + var idempotencyKey: String + get() = getStringProperty(::idempotencyKey.name) + protected set(value) { + setStringProperty(::idempotencyKey.name, value) + } + + /** Failed attempts the backend answered, persisted so the limit holds across restarts. */ + var failedAttempts: Int + get() = getOptIntProperty(::failedAttempts.name) ?: 0 + internal set(value) { + setIntProperty(::failedAttempts.name, value) + } + + override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE + override val requiresJwt: Boolean get() = false + + 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 index 2487efc237..d58bfbcf5c 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 @@ -1,7 +1,6 @@ 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 @@ -10,13 +9,7 @@ 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. */ -class UpdateSessionOperation() : Operation(SessionOperationExecutor.UPDATE_SESSION) { - var appId: String - get() = getStringProperty(::appId.name) - private set(value) { - setStringProperty(::appId.name, value) - } - +class UpdateSessionOperation() : SessionOperation(SessionOperationExecutor.UPDATE_SESSION) { /** * The backend session ID, or the [CreateSessionOperation.localSessionId] until the create succeeds. */ @@ -26,18 +19,6 @@ class UpdateSessionOperation() : Operation(SessionOperationExecutor.UPDATE_SESSI 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 in milliseconds. */ var activeDuration: Long get() = getLongProperty(::activeDuration.name) @@ -52,24 +33,13 @@ class UpdateSessionOperation() : Operation(SessionOperationExecutor.UPDATE_SESSI setOptLongProperty(::endTime.name, value) } - /** - * Generated once and persisted, so every retry, including after a restart, is deduplicated by the backend. - */ - var idempotencyKey: String - get() = getStringProperty(::idempotencyKey.name) - private set(value) { - setStringProperty(::idempotencyKey.name, value) - } - val isEnd: Boolean get() = endTime != null override val createComparisonKey: String get() = "$appId.Session.$sessionId" override val modifyComparisonKey: String get() = "$appId.Session.$sessionId" - 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 - override val requiresJwt: Boolean get() = false constructor( appId: String, @@ -91,7 +61,6 @@ class UpdateSessionOperation() : Operation(SessionOperationExecutor.UPDATE_SESSI override fun translateIds(map: Map) { map[sessionId]?.let { sessionId = it } - map[onesignalId]?.let { onesignalId = it } - map[subscriptionId]?.let { subscriptionId = it } + super.translateIds(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 b04f4ad69e..37a8b2819f 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 @@ -2,26 +2,23 @@ package com.onesignal.session.internal.session.operations.impl import com.onesignal.common.NetworkUtils import com.onesignal.common.exceptions.BackendException -import com.onesignal.common.threading.OneSignalDispatchers 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.SessionOperation import com.onesignal.session.internal.session.operations.UpdateSessionOperation -import kotlinx.coroutines.withContext import java.util.concurrent.TimeUnit.MILLISECONDS internal class SessionOperationExecutor( private val sessionsBackend: ISessionsBackendService, private val deviceService: IDeviceService, - private val sessionModelStore: SessionModelStore, ) : IOperationExecutor { override val operations: List get() = listOf(CREATE_SESSION, UPDATE_SESSION) @@ -47,32 +44,18 @@ internal class SessionOperationExecutor( try { sessionsBackend.createSession(op.appId, request) } catch (ex: BackendException) { - return failure("create", ex) + return failure(op, "create", ex) } return if (serverSessionId == null) { // Retrying with the same idempotency key lets the backend return the session it already created. - Logging.warn("SessionOperationExecutor: create session response is missing data.session_id, retrying") - ExecutionResponse(ExecutionResult.FAIL_RETRY) + Logging.warn("SessionOperationExecutor: create session response is missing data.session_id") + retryOrDrop(op, "create", backendResponded = true, retryAfterSeconds = null) } else { - sessionCreated(op, serverSessionId) + // OperationRepo also writes this translation into the current session's pinned IDs. + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to serverSessionId)) } } - private suspend fun sessionCreated( - op: CreateSessionOperation, - serverSessionId: String, - ): ExecutionResponse { - // Persisted so updates enqueued after this point, including after a restart, use the backend ID directly. - // On SerialIO so the check and write can't interleave with SessionService starting a new session. - withContext(OneSignalDispatchers.SerialIO) { - val session = sessionModelStore.model - if (session.localSessionId == op.localSessionId) { - session.serverSessionId = serverSessionId - } - } - return ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to serverSessionId)) - } - private suspend fun updateSession(op: UpdateSessionOperation): ExecutionResponse { val request = UpdateSessionRequest( @@ -85,32 +68,52 @@ internal class SessionOperationExecutor( try { sessionsBackend.updateSession(op.appId, op.sessionId, request) } catch (ex: BackendException) { - return failure("update", ex) + return failure(op, "update", ex) } return ExecutionResponse(ExecutionResult.SUCCESS) } private fun failure( + op: SessionOperation, action: String, ex: BackendException, ): ExecutionResponse { val code = ex.statusCode - // RETRYABLE also covers 4xx codes NetworkUtils doesn't list, like 413 and 422. Those won't succeed on retry, - // and OperationRepo retries without a cap at the front of the queue, so they would block every other operation. + // RETRYABLE also covers 4xx codes NetworkUtils doesn't list, like 413 and 422. Those won't succeed on retry. val retryable = NetworkUtils.getResponseStatusType(code) == NetworkUtils.ResponseStatusType.RETRYABLE && (code !in CLIENT_ERROR_CODES || code in RETRYABLE_CLIENT_ERROR_CODES) - if (retryable) { - return ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = ex.retryAfterSeconds) + if (!retryable) { + Logging.warn("SessionOperationExecutor: $action session failed with $code, dropping") + return ExecutionResponse(ExecutionResult.FAIL_NORETRY) } - Logging.warn("SessionOperationExecutor: $action session failed with $code, dropping") - return ExecutionResponse(ExecutionResult.FAIL_NORETRY) + // A status of 0 or less means no response, e.g. offline, which must keep retrying until reconnect. + return retryOrDrop(op, action, backendResponded = code > 0, retryAfterSeconds = ex.retryAfterSeconds) + } + + /** + * OperationRepo retries at the front of the queue, so a backend that keeps failing blocks every + * other operation. Attempts it answered are capped; attempts that never reached it are not. + */ + private fun retryOrDrop( + op: SessionOperation, + action: String, + backendResponded: Boolean, + retryAfterSeconds: Int?, + ): ExecutionResponse { + if (backendResponded && ++op.failedAttempts >= MAX_FAILED_ATTEMPTS) { + Logging.warn("SessionOperationExecutor: $action session failed ${op.failedAttempts} times, dropping") + return ExecutionResponse(ExecutionResult.FAIL_NORETRY) + } + return ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = retryAfterSeconds) } companion object { const val CREATE_SESSION = "create-session" const val UPDATE_SESSION = "update-session" + internal const val MAX_FAILED_ATTEMPTS = 5 + private val CLIENT_ERROR_CODES = 400..499 // Request Timeout and Too Many Requests 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 61c5aef175..84fa082aa0 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 @@ -79,6 +79,9 @@ private class Mocks { var identityVerificationService = CoreInternalMocks.identityVerificationService() + // Created eagerly: operationRepo is lazy and may first be read inside an every { } block. + val sessionModelStore = MockHelper.sessionModelStore() + val operationRepo: OperationRepo by lazy { spyk( OperationRepo( @@ -89,6 +92,7 @@ private class Mocks { getNewRecordState(configModelStore), jwtTokenStore, identityVerificationService, + sessionModelStore, ), recordPrivateCalls = true, ) @@ -116,6 +120,7 @@ class OperationRepoTests : FunSpec({ getNewRecordState(mocks.configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(), + MockHelper.sessionModelStore(), ), ) 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 180f3cd573..8468498213 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 @@ -10,11 +10,13 @@ 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.SessionOperation 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 com.onesignal.user.internal.operations.LoginUserOperation import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.collections.shouldContainExactly import io.kotest.matchers.shouldBe @@ -42,6 +44,8 @@ private class SessionQueueMocks( it.useIdentityVerification = if (ivRequired) JwtRequirement.REQUIRED else JwtRequirement.NOT_REQUIRED } + val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "session" } + val storedOperations = mutableListOf() private val barriers = ConcurrentHashMap>() val operationModelStore: OperationModelStore = @@ -72,6 +76,7 @@ private class SessionQueueMocks( getNewRecordState(configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(newCodePathsRun = ivRequired, ivBehaviorActive = ivRequired), + sessionModelStore, ) /** The first execution suspends until [release] completes; [started] completes once it begins. */ @@ -368,4 +373,77 @@ class SessionOperationQueueTests : FunSpec({ mocks.queuedOperations shouldContainExactly listOf(createOp) verify(exactly = 0) { mocks.operationModelStore.remove(any()) } } + + test("session operations for an anonymous user are discarded at enqueue under identity verification") { + // Given + val mocks = SessionQueueMocks(ivRequired = true) + mocks.operationRepo.start() + mocks.operationRepo.awaitInitialized() + val identifiedLogin = LoginUserOperation(APP_ID, "local-identified", "external-id", existingOneSignalId = "local-pending") + val anonymousCreate = create(onesignalId = "local-anonymous") + val identifiedCreate = create(localSessionId = "local-other-session", onesignalId = "local-identified") + + // When + mocks.enqueueAll(identifiedLogin, anonymousCreate, identifiedCreate) + + // Then + mocks.queuedOperations shouldContainExactly listOf(identifiedCreate) + } + + test("session operations for an anonymous user are dropped with anonymous operations under identity verification") { + // Given + val mocks = SessionQueueMocks(ivRequired = true) + mocks.enqueueAll(create(onesignalId = "local-anonymous"), update(activeDuration = 10_000)) + mocks.operationRepo.start() + mocks.operationRepo.awaitInitialized() + + // When + mocks.operationRepo.removeOperationsWithoutExternalId() + + // Then + mocks.queuedOperations shouldBe emptyList() + mocks.storedOperations.filterIsInstance() shouldBe emptyList() + } + + test("translations applied to the queue are also applied to the current session's pinned IDs") { + // Given + val mocks = SessionQueueMocks() + val session = + mocks.sessionModelStore.model.apply { + onesignalId = "local-user" + subscriptionId = "local-subscription" + } + mocks.respondWith({ + ExecutionResponse( + ExecutionResult.SUCCESS, + idTranslations = mapOf(LOCAL_SESSION_ID to "server-session", "local-user" to "user", "local-subscription" to "subscription"), + ) + }) + mocks.enqueueAll(create()) + + // When + mocks.operationRepo.start() + withTimeout(2_000) { while (session.serverSessionId == null) delay(10) } + + // Then + session.serverSessionId shouldBe "server-session" + session.onesignalId shouldBe "user" + session.subscriptionId shouldBe "subscription" + } + + test("a create for an earlier session does not set the current session's server session ID") { + // Given + val mocks = SessionQueueMocks() + val session = mocks.sessionModelStore.model.apply { sessionId = "newer-session" } + mocks.respondWith({ ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(LOCAL_SESSION_ID to "server-session")) }) + mocks.enqueueAll(create()) + + // When + mocks.operationRepo.start() + withTimeout(2_000) { while (mocks.executed.isEmpty()) delay(10) } + delay(100) + + // Then + session.serverSessionId shouldBe null + } }) 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..a659195480 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,11 +1,7 @@ package com.onesignal.session.internal.session -import com.onesignal.common.modeling.ISingletonModelStoreChangeHandler -import com.onesignal.common.modeling.ModelChangeTags -import com.onesignal.common.modeling.ModelChangedArgs import com.onesignal.common.threading.OneSignalDispatchers import com.onesignal.common.threading.runOnSerialIO -import com.onesignal.core.internal.config.ConfigModel import com.onesignal.core.internal.features.IFeatureManager import com.onesignal.core.internal.time.ITime import com.onesignal.features.FeatureFlag @@ -13,8 +9,6 @@ import com.onesignal.mocks.IOMockHelper import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService import com.onesignal.session.internal.session.impl.SessionService -import com.onesignal.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.mockk.every @@ -23,7 +17,6 @@ import io.mockk.mockk import io.mockk.mockkObject import io.mockk.mockkStatic import io.mockk.runs -import io.mockk.slot import io.mockk.spyk import io.mockk.unmockkObject import io.mockk.unmockkStatic @@ -54,35 +47,6 @@ private class Mocks( val identityModelStore = MockHelper.identityModelStore { it.onesignalId = ONESIGNAL_ID } val configModelStore = MockHelper.configModelStore { it.pushSubscriptionId = SUBSCRIPTION_ID } - val identityChangeHandler = slot>() - val configChangeHandler = slot>() - - init { - every { identityModelStore.subscribe(capture(identityChangeHandler)) } just runs - every { configModelStore.subscribe(capture(configChangeHandler)) } just runs - } - - /** Mirrors the user-create executor replacing a local ID with the backend ID in place. */ - fun hydrateOnesignalId(backendId: String) { - val model = identityModelStore.model - val localId = model.onesignalId - model.onesignalId = backendId - identityChangeHandler.captured.onModelUpdated( - ModelChangedArgs(model, IdentityConstants.ONESIGNAL_ID, IdentityConstants.ONESIGNAL_ID, localId, backendId), - ModelChangeTags.HYDRATE, - ) - } - - fun hydratePushSubscriptionId(backendId: String) { - val model = configModelStore.model - val localId = model.pushSubscriptionId - model.pushSubscriptionId = backendId - configChangeHandler.captured.onModelUpdated( - ModelChangedArgs(model, ConfigModel::pushSubscriptionId.name, ConfigModel::pushSubscriptionId.name, localId, backendId), - ModelChangeTags.NORMAL, - ) - } - fun sessionModelStore(action: ((SessionModel) -> Unit)? = null): SessionModelStore { if (action != null) action(mockSessionModelStore.model) return mockSessionModelStore @@ -341,102 +305,6 @@ class SessionServiceTests : FunSpec({ sessionModelStore.model.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID } - test("pinned local IDs are replaced with backend IDs when the user is created") { - // Given - val mocks = Mocks() - mocks.identityModelStore.model.onesignalId = "local-user" - mocks.configModelStore.model.pushSubscriptionId = "local-subscription" - val sessionService = mocks.sessionService - sessionService.bootstrap() - sessionService.start() - val sessionModelStore = mocks.sessionModelStore { it.isValid = false } - sessionService.onFocus(false) - - // When - mocks.hydrateOnesignalId("backend-user") - mocks.hydratePushSubscriptionId("backend-subscription") - - // Then - sessionModelStore.model.onesignalId shouldBe "backend-user" - sessionModelStore.model.subscriptionId shouldBe "backend-subscription" - } - - test("pinned subscription ID is not replaced after login before the anonymous user is created") { - // Given - val mocks = Mocks() - mocks.identityModelStore.model.onesignalId = "local-anonymous" - mocks.configModelStore.model.pushSubscriptionId = "local-subscription" - val sessionService = mocks.sessionService - sessionService.bootstrap() - sessionService.start() - val sessionModelStore = mocks.sessionModelStore { it.isValid = false } - sessionService.onFocus(false) - - // When - val loggedInModel = IdentityModel().apply { onesignalId = "local-logged-in" } - every { mocks.identityModelStore.model } returns loggedInModel - mocks.identityChangeHandler.captured.onModelReplaced(loggedInModel, ModelChangeTags.NORMAL) - mocks.hydrateOnesignalId("backend-logged-in") - mocks.hydratePushSubscriptionId("backend-subscription") - - // Then - sessionModelStore.model.onesignalId shouldBe "local-anonymous" - sessionModelStore.model.subscriptionId shouldBe "local-subscription" - } - - test("pinned backend IDs are not replaced by later ID changes") { - // Given - val mocks = Mocks() - val sessionService = mocks.sessionService - sessionService.bootstrap() - sessionService.start() - val sessionModelStore = mocks.sessionModelStore { it.isValid = false } - sessionService.onFocus(false) - - // When - mocks.hydrateOnesignalId("other-user") - mocks.hydratePushSubscriptionId("other-subscription") - - // Then - sessionModelStore.model.onesignalId shouldBe Mocks.ONESIGNAL_ID - sessionModelStore.model.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID - } - - test("pinned local IDs are not replaced when a different local ID is translated") { - // Given - val mocks = Mocks() - mocks.identityModelStore.model.onesignalId = "local-user" - val sessionService = mocks.sessionService - sessionService.bootstrap() - sessionService.start() - val sessionModelStore = mocks.sessionModelStore { it.isValid = false } - sessionService.onFocus(false) - - // When - mocks.identityModelStore.model.onesignalId = "local-next-user" - mocks.hydrateOnesignalId("backend-next-user") - - // Then - sessionModelStore.model.onesignalId shouldBe "local-user" - } - - test("pinned local IDs are not replaced by another local ID") { - // Given - val mocks = Mocks() - mocks.identityModelStore.model.onesignalId = "local-user" - val sessionService = mocks.sessionService - sessionService.bootstrap() - sessionService.start() - val sessionModelStore = mocks.sessionModelStore { it.isValid = false } - sessionService.onFocus(false) - - // When - mocks.hydrateOnesignalId("local-next-user") - - // Then - sessionModelStore.model.onesignalId shouldBe "local-user" - } - test("sessions API active duration uses the monotonic clock") { // Given val mocks = Mocks(sessionsApiEnabled = true) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt index bcff2b93e2..2765c8146a 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 @@ -50,16 +50,14 @@ class SessionOperationExecutorTests : FunSpec({ // Given val backend = mockk() coEvery { backend.createSession(any(), any()) } returns "server-session" - val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "session-uuid" } - val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), sessionModelStore) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService()) // When - val response = executor.execute(listOf(createOp(localSessionId = sessionModelStore.model.localSessionId))) + val response = executor.execute(listOf(createOp())) // Then response.result shouldBe ExecutionResult.SUCCESS response.idTranslations shouldBe mapOf("local-session-uuid" to "server-session") - sessionModelStore.model.serverSessionId shouldBe "server-session" coVerify { backend.createSession( APP_ID, @@ -75,28 +73,13 @@ class SessionOperationExecutorTests : FunSpec({ } } - test("create for a session that is no longer current leaves the current session alone") { - // Given - val backend = mockk() - coEvery { backend.createSession(any(), any()) } returns "server-session" - val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "newer-session" } - val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), sessionModelStore) - - // When - val response = executor.execute(listOf(createOp(localSessionId = "local-older-session"))) - - // Then - response.idTranslations shouldBe mapOf("local-older-session" to "server-session") - sessionModelStore.model.serverSessionId shouldBe null - } - test("network errors, 408, 429, and 5xx are retried with Retry-After") { listOf(-1, 0, 408, 429, 500, 503).forEach { statusCode -> // Given val backend = mockk() coEvery { backend.createSession(any(), any()) } throws BackendException(statusCode, retryAfterSeconds = 30) coEvery { backend.updateSession(any(), any(), any()) } throws BackendException(statusCode, retryAfterSeconds = 30) - val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService()) // When val responses = listOf(executor.execute(listOf(createOp())), executor.execute(listOf(updateOp()))) @@ -115,7 +98,7 @@ class SessionOperationExecutorTests : FunSpec({ val backend = mockk() coEvery { backend.createSession(any(), any()) } throws BackendException(statusCode) coEvery { backend.updateSession(any(), any(), any()) } throws BackendException(statusCode) - val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService()) // When val responses = listOf(executor.execute(listOf(createOp())), executor.execute(listOf(updateOp()))) @@ -129,23 +112,56 @@ class SessionOperationExecutorTests : FunSpec({ // Given val backend = mockk() coEvery { backend.createSession(any(), any()) } returns null - val sessionModelStore = MockHelper.sessionModelStore { it.sessionId = "session-uuid" } - val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), sessionModelStore) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService()) // When - val response = executor.execute(listOf(createOp(localSessionId = sessionModelStore.model.localSessionId))) + val response = executor.execute(listOf(createOp())) // Then response.result shouldBe ExecutionResult.FAIL_RETRY response.idTranslations shouldBe null - sessionModelStore.model.serverSessionId shouldBe null + } + + test("an operation is dropped after the backend fails it the maximum number of times") { + listOf Unit>( + { coEvery { it.createSession(any(), any()) } returns null }, + { coEvery { it.createSession(any(), any()) } throws BackendException(503) }, + { coEvery { it.createSession(any(), any()) } throws BackendException(429, retryAfterSeconds = 30) }, + ).forEach { stub -> + // Given + val backend = mockk().also { stub(it) } + val executor = SessionOperationExecutor(backend, MockHelper.deviceService()) + val op = createOp() + + // When + val results = List(SessionOperationExecutor.MAX_FAILED_ATTEMPTS) { executor.execute(listOf(op)).result } + + // Then + results.dropLast(1).forEach { it shouldBe ExecutionResult.FAIL_RETRY } + results.last() shouldBe ExecutionResult.FAIL_NORETRY + } + } + + test("network errors do not count toward the attempt limit") { + // Given + val backend = mockk() + coEvery { backend.updateSession(any(), any(), any()) } throws BackendException(0) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService()) + val op = updateOp() + + // When + val results = List(SessionOperationExecutor.MAX_FAILED_ATTEMPTS * 2) { executor.execute(listOf(op)).result } + + // Then + results.forEach { it shouldBe ExecutionResult.FAIL_RETRY } + op.failedAttempts shouldBe 0 } test("update sends the duration and end time in seconds") { // Given val backend = mockk() coEvery { backend.updateSession(any(), any(), any()) } returns Unit - val executor = SessionOperationExecutor(backend, MockHelper.deviceService(), MockHelper.sessionModelStore()) + val executor = SessionOperationExecutor(backend, MockHelper.deviceService()) // When val response = executor.execute(listOf(updateOp(endTime = 1_700_000_100_999L))) @@ -198,6 +214,7 @@ class SessionOperationExecutorTests : FunSpec({ val prefs = MockPreferencesService() val create = createOp().apply { id = "create-op" } val duration = updateOp().apply { id = "duration-op" } + duration.failedAttempts = 2 val end = updateOp(endTime = 5_000L).apply { id = "end-op" } OperationModelStore(prefs).apply { loadOperations() @@ -218,6 +235,7 @@ class SessionOperationExecutorTests : FunSpec({ reloadedDuration.idempotencyKey shouldBe "update-key" reloadedDuration.endTime shouldBe null reloadedDuration.activeDuration shouldBe 42_900L + reloadedDuration.failedAttempts shouldBe 2 (reloaded[2] as UpdateSessionOperation).endTime shouldBe 5_000L } }) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt index cb16466bc2..676a5fda65 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 @@ -48,6 +48,7 @@ private class Mocks { ExecutorMocks.getNewRecordState(configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(), + MockHelper.sessionModelStore(), ), ) From 39e5d76d16743a31764a9de8c74f0cd3cbd655f5 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Thu, 8 Oct 2026 13:44:29 -0700 Subject: [PATCH 8/9] fix: [SDK-5378] address review: keep attempt count when coalescing, scope session subscription translation, join pin write - Coalesced updates keep the highest failedAttempts, so a session that keeps enqueueing updates still reaches the attempt limit. - The session's pinned subscription ID is only translated when the same response translates its pinned OneSignal ID, since login reuses a local push subscription for the new user. - The pin write is joined again so IDs are saved before a successful create is removed from the store, and it is skipped before the first session starts. - LoginUserFromSubscriptionOperation counts as a login that will create the user under identity verification. Co-Authored-By: Claude Opus 5.5 --- .../internal/operations/impl/OperationRepo.kt | 9 +-- .../impl/OperationRepoSessionExtensions.kt | 25 ++++--- .../session/internal/session/SessionModel.kt | 9 ++- .../operations/SessionOperationQueueTests.kt | 66 +++++++++++++++++++ 4 files changed, 94 insertions(+), 15 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 d3ac89301a..c99b6ae4d8 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 @@ -1,8 +1,8 @@ package com.onesignal.core.internal.operations.impl import com.onesignal.common.IDManager +import com.onesignal.common.threading.OneSignalDispatchers import com.onesignal.common.threading.WaiterWithValue -import com.onesignal.common.threading.runOnSerialIO import com.onesignal.common.threading.suspendifyOnIO import com.onesignal.core.internal.config.ConfigModelStore import com.onesignal.core.internal.config.impl.IdentityVerificationService @@ -26,6 +26,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.delay import kotlinx.coroutines.launch import kotlinx.coroutines.newSingleThreadContext +import kotlinx.coroutines.withContext import kotlinx.coroutines.withTimeoutOrNull import java.util.UUID import kotlin.math.max @@ -430,9 +431,9 @@ internal class OperationRepo( appliedIdTranslations.putAll(response.idTranslations) } response.idTranslations.values.forEach { _newRecordState.add(it) } - // On SerialIO, where SessionService starts sessions, so the session can't change mid-write. - // Updates built from the old IDs before this runs are still translated at enqueue. - runOnSerialIO { sessionModelStore.model.translateIds(response.idTranslations) } + // Joined so the IDs are saved before a successful create is removed from the store. On SerialIO, + // where SessionService starts sessions, so the session can't change mid-write. + withContext(OneSignalDispatchers.SerialIO) { sessionModelStore.model.translateIds(response.idTranslations) } } var highestRetries = 0 diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt index 593bc8d913..294c2f22bb 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepoSessionExtensions.kt @@ -6,6 +6,7 @@ import com.onesignal.debug.internal.logging.Logging import com.onesignal.session.internal.session.operations.CreateSessionOperation import com.onesignal.session.internal.session.operations.SessionOperation import com.onesignal.session.internal.session.operations.UpdateSessionOperation +import com.onesignal.user.internal.operations.LoginUserFromSubscriptionOperation import com.onesignal.user.internal.operations.LoginUserOperation /** @@ -54,8 +55,13 @@ private fun OperationRepo.isUnsendable(op: SessionOperation): Boolean = isIdentityVerificationRequired && IDManager.isLocalId(op.onesignalId) && (queue + inFlight).none { - val login = it.operation as? LoginUserOperation - login != null && login.onesignalId == op.onesignalId && login.externalId != null + val createsUser = + when (val login = it.operation) { + is LoginUserOperation -> login.onesignalId == op.onesignalId + is LoginUserFromSubscriptionOperation -> login.onesignalId == op.onesignalId + else -> false + } + createsUser && it.operation.externalId != null } /** @@ -83,19 +89,20 @@ private fun OperationRepo.admitSessionUpdate( return false } - val queued = queue.filter { (it.operation as? UpdateSessionOperation)?.isSameSession(op) == true } - queued.maxOfOrNull { (it.operation as UpdateSessionOperation).activeDuration }?.let { - op.activeDuration = maxOf(op.activeDuration, it) - } - queued.forEach { item -> + val queued = queue.mapNotNull { item -> (item.operation as? UpdateSessionOperation)?.takeIf { it.isSameSession(op) } } + queued.maxOfOrNull { it.activeDuration }?.let { op.activeDuration = maxOf(op.activeDuration, it) } + // Otherwise a session that keeps enqueueing updates would never reach the attempt limit. + queued.maxOfOrNull { it.failedAttempts }?.let { op.failedAttempts = maxOf(op.failedAttempts, it) } + val queuedItems = queue.filter { it.operation in queued } + queuedItems.forEach { item -> if (incoming.waiter == null) { incoming.waiter = item.waiter } else { item.waiter?.wake(true) } } - if (queued.isNotEmpty()) { - removeQueuedOperations(queued, operationModelStore) + if (queuedItems.isNotEmpty()) { + removeQueuedOperations(queuedItems, operationModelStore) Logging.debug("OperationRepo: coalesced ${queued.size} queued update(s) for session ${op.sessionId}") } return true 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 b70fe89928..a87f41d5ec 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 @@ -117,8 +117,13 @@ class SessionModel : Model() { * updates built from these fields, including after a restart, match them. */ internal fun translateIds(map: Map) { - onesignalId?.let { map[it] }?.let { onesignalId = it } - subscriptionId?.let { map[it] }?.let { subscriptionId = it } + // Unset until the first session starts. + if (!hasProperty(::sessionId.name)) return + map[localSessionId]?.let { serverSessionId = it } + onesignalId = onesignalId?.let { map[it] } ?: return + // Login reuses a local push subscription for the new user, so its ID only belongs to this + // session when the same response created this session's user. + subscriptionId?.let { map[it] }?.let { subscriptionId = it } } } 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 8468498213..42a2e48b75 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.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.SessionModel import com.onesignal.session.internal.session.operations.CreateSessionOperation import com.onesignal.session.internal.session.operations.SessionOperation import com.onesignal.session.internal.session.operations.UpdateSessionOperation @@ -16,6 +17,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.LoginUserFromSubscriptionOperation import com.onesignal.user.internal.operations.LoginUserOperation import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.collections.shouldContainExactly @@ -154,6 +156,20 @@ class SessionOperationQueueTests : FunSpec({ mocks.storedOperations shouldContainExactly listOf(second) } + test("coalescing keeps the highest failed attempt count") { + // Given + val mocks = SessionQueueMocks() + val retried = update(activeDuration = 10_000).apply { failedAttempts = 3 } + val next = update(activeDuration = 20_000) + + // When + mocks.enqueueAll(retried, next) + + // Then + mocks.queuedOperations shouldContainExactly listOf(next) + next.failedAttempts shouldBe 3 + } + test("an end replaces an unsent update for the same session") { // Given val mocks = SessionQueueMocks() @@ -446,4 +462,54 @@ class SessionOperationQueueTests : FunSpec({ // Then session.serverSessionId shouldBe null } + + test("a session operation is kept under identity verification while a subscription login creates its user") { + // Given + val mocks = SessionQueueMocks(ivRequired = true) + mocks.operationRepo.start() + mocks.operationRepo.awaitInitialized() + val createOp = create(onesignalId = "local-identified") + + // When + mocks.enqueueAll(LoginUserFromSubscriptionOperation(APP_ID, "local-identified", "external-id", "subscription-id"), createOp) + + // Then + mocks.queuedOperations shouldContainExactly listOf(createOp) + } + + test("a subscription translation without the session's user translation leaves the pinned subscription alone") { + // Given + val mocks = SessionQueueMocks() + val session = + mocks.sessionModelStore.model.apply { + onesignalId = "local-previous-user" + subscriptionId = "local-subscription" + } + mocks.respondWith({ + ExecutionResponse( + ExecutionResult.SUCCESS, + idTranslations = mapOf(LOCAL_SESSION_ID to "server-session", "local-subscription" to "subscription"), + ) + }) + mocks.enqueueAll(create()) + + // When + mocks.operationRepo.start() + withTimeout(2_000) { while (session.serverSessionId == null) delay(10) } + + // Then + session.onesignalId shouldBe "local-previous-user" + session.subscriptionId shouldBe "local-subscription" + } + + test("translations before the first session starts are ignored") { + // Given + val session = SessionModel() + + // When + session.translateIds(mapOf("local-user" to "user")) + + // Then + session.serverSessionId shouldBe null + } }) From a4da3514ab3df00f4fb91dccd6fe5adfca51f1b6 Mon Sep 17 00:00:00 2001 From: Fadi George Date: Thu, 8 Oct 2026 13:55:24 -0700 Subject: [PATCH 9/9] test: [SDK-5378] start the repo after the direct execution to avoid a race on queue order Co-Authored-By: Claude Opus 5.5 --- .../onesignal/core/internal/operations/OperationRepoTests.kt | 3 ++- 1 file changed, 2 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 84fa082aa0..1f8a968e41 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 @@ -709,11 +709,12 @@ class OperationRepoTests : FunSpec({ val secondOp = mockOperationNonGroupable() // When - mocks.operationRepo.start() mocks.operationRepo.enqueue(firstOp) mocks.operationRepo.executeOperations( listOf(OperationQueueItem(opWithResult, bucket = 0)), ) + // Started after the direct execution so the processing loop can't pick firstOp first. + mocks.operationRepo.start() mocks.operationRepo.enqueueAndWait(secondOp) // Then