From 629333a53717a3ddc31b28bd8cae80160488c13a Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 15:23:14 +0400 Subject: [PATCH 01/12] fix(core): persist the restored userId off the init thread Init encrypted the persisted userId on the caller's thread, so a slow or stalled Keystore could ANR an identified cold start. The save now runs in the background and UserChangedEvent is published once it lands, so every module still finds the userId on disk when notified. A version counter orders it against identify and clearIdentify; a same-user identify skips the Keystore and publishes after the pending write. In-app polling waits for an identity instead of fetching without a user token. Co-Authored-By: Claude Opus 5.5 --- .../sdk/core/di/AndroidSDKComponent.kt | 4 +- .../sdk/data/store/SecureUserStore.kt | 104 ++++++++++- .../sdk/data/store/SecureUserStoreTest.kt | 162 +++++++++++++++++- .../main/kotlin/io/customer/sdk/CustomerIO.kt | 17 +- .../DataPipelineEventBusInteractionTest.kt | 2 +- .../UserChangedEventPersistenceTest.kt | 79 +++++++++ .../presentation/PollingLifecycleManager.kt | 6 + .../PollingLifecycleManagerTest.kt | 35 +++- 8 files changed, 386 insertions(+), 23 deletions(-) create mode 100644 datapipelines/src/test/java/io/customer/datapipelines/UserChangedEventPersistenceTest.kt diff --git a/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt b/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt index 76700e321..52f67efea 100644 --- a/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt +++ b/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt @@ -61,7 +61,9 @@ class AndroidSDKComponentImpl( override val globalPreferenceStore: GlobalPreferenceStore get() = singleton { GlobalPreferenceStoreImpl(applicationContext) } override val secureUserStore: SecureUserStore - get() = singleton { SecureUserStoreImpl(applicationContext, SDKComponent.logger) } + get() = singleton { + SecureUserStoreImpl(applicationContext, SDKComponent.logger, SDKComponent.dispatchersProvider.background) + } override fun reset() { super.reset() diff --git a/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt b/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt index 08713aafb..5cbb60525 100644 --- a/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt +++ b/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt @@ -1,9 +1,15 @@ package io.customer.sdk.data.store import android.content.Context +import android.content.SharedPreferences import androidx.core.content.edit import io.customer.base.internal.InternalCustomerIOApi import io.customer.sdk.core.util.Logger +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.CoroutineExceptionHandler +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.launch /** * Encrypted, user-scoped preference store for sensitive data that must @@ -19,35 +25,115 @@ import io.customer.sdk.core.util.Logger */ @InternalCustomerIOApi interface SecureUserStore { - fun saveUserId(userId: String?) + /** + * Persists [userId] (null removes it), then calls [onPersisted], so anything notified from + * [onPersisted] finds the value through [getUserId]. Encrypts on the caller's thread, unless + * the same userId is already stored or being stored: then no Keystore call is made, and + * [onPersisted] runs now or once that write lands. A later save or [clearAll] that supersedes + * this one drops both the write and [onPersisted]. + */ + fun saveUserId(userId: String?, onPersisted: () -> Unit = {}) + + /** + * Same as [saveUserId], but encrypts on a background thread so the caller never waits on the + * Keystore. Until the write lands, [getUserId] returns the previously stored value. + */ + fun saveUserIdInBackground(userId: String?, onPersisted: () -> Unit) + fun getUserId(): String? fun clearAll() } internal class SecureUserStoreImpl( context: Context, - logger: Logger + logger: Logger, + backgroundDispatcher: CoroutineDispatcher, + private val crypto: PreferenceCrypto = PreferenceCrypto(KEY_ALIAS, logger) ) : PreferenceStore(context), SecureUserStore { - private val crypto = PreferenceCrypto(KEY_ALIAS, logger) + private val writeScope = CoroutineScope( + backgroundDispatcher + SupervisorJob() + CoroutineExceptionHandler { _, throwable -> + logger.error("Secure user store write failed: $throwable", throwable = throwable) + } + ) + + // Orders saves and clears without holding a lock across a Keystore call: each one that changes + // the stored user bumps [version], and an encrypted write lands only if its version is current. + private val lock = Any() + private var version = 0L + + // Written by this process, so saving the same user again needs no encryption. + private var persistedUserId: String? = null + private var pendingWrite: PendingWrite? = null + + private class PendingWrite(val userId: String, val version: Long) { + val onPersisted = mutableListOf<() -> Unit>() + } override val prefsName: String by lazy { "io.customer.sdk.secure_user.${context.packageName}" } - override fun saveUserId(userId: String?) = prefs.edit { - if (userId != null) { - putString(KEY_USER_ID, crypto.encrypt(userId)) - } else { - remove(KEY_USER_ID) + override fun saveUserId(userId: String?, onPersisted: () -> Unit) { + val write = startWrite(userId, onPersisted) ?: return + finishWrite(write, crypto.encrypt(write.userId)) + } + + override fun saveUserIdInBackground(userId: String?, onPersisted: () -> Unit) { + val write = startWrite(userId, onPersisted) ?: return + writeScope.launch { finishWrite(write, crypto.encrypt(write.userId)) } + } + + /** Returns the write to encrypt, or null when [userId] needs no Keystore call. */ + private fun startWrite(userId: String?, onPersisted: () -> Unit): PendingWrite? = synchronized(lock) { + val pending = pendingWrite + when { + userId == null -> { + resetLocked { remove(KEY_USER_ID) } + onPersisted() + null + } + + pending?.userId == userId -> { + pending.onPersisted += onPersisted + null + } + + pending == null && persistedUserId == userId -> { + onPersisted() + null + } + + else -> PendingWrite(userId, ++version).also { + it.onPersisted += onPersisted + pendingWrite = it + } } } + // Callbacks run under the lock, so a clear can't slip between this write and its event. + private fun finishWrite(write: PendingWrite, encrypted: String) = synchronized(lock) { + if (write.version != version) return@synchronized + pendingWrite = null + prefs.edit { putString(KEY_USER_ID, encrypted) } + persistedUserId = write.userId + write.onPersisted.forEach { it() } + } + override fun getUserId(): String? = prefs.read { getString(KEY_USER_ID, null)?.let { crypto.decrypt(it) } } - override fun clearAll() = prefs.edit { clear() } + override fun clearAll() = synchronized(lock) { + resetLocked { clear() } + } + + private fun resetLocked(edit: SharedPreferences.Editor.() -> Unit) { + version++ + pendingWrite = null + persistedUserId = null + prefs.edit(action = edit) + } companion object { private const val KEY_ALIAS = "cio_secure_user_key" diff --git a/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt b/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt index 796579bc7..00ab4be0e 100644 --- a/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt +++ b/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt @@ -2,9 +2,19 @@ package io.customer.sdk.data.store import io.customer.commontest.core.RobolectricTest import io.customer.sdk.core.util.Logger +import io.mockk.every import io.mockk.mockk +import io.mockk.verify +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.UnconfinedTestDispatcher import org.amshove.kluent.shouldBeEqualTo +import org.amshove.kluent.shouldBeFalse import org.amshove.kluent.shouldBeNull +import org.amshove.kluent.shouldBeTrue import org.junit.Test import org.junit.runner.RunWith import org.robolectric.RobolectricTestRunner @@ -14,8 +24,27 @@ class SecureUserStoreTest : RobolectricTest() { private val mockLogger: Logger = mockk(relaxed = true) + // Background writes stay pending until the test advances this dispatcher. + private val writer = StandardTestDispatcher() + + private val crypto: PreferenceCrypto = fakeCrypto() + + private fun fakeCrypto(onEncrypt: () -> Unit = {}): PreferenceCrypto = mockk { + every { encrypt(any()) } answers { + onEncrypt() + "enc:${firstArg()}" + } + every { decrypt(any()) } answers { firstArg().removePrefix("enc:") } + } + private fun newStore(): SecureUserStoreImpl = - SecureUserStoreImpl(contextMock, mockLogger).also { it.clearAll() } + SecureUserStoreImpl(contextMock, mockLogger, UnconfinedTestDispatcher()).also { it.clearAll() } + + private fun newStoreWithPendingWrites( + crypto: PreferenceCrypto = this.crypto, + dispatcher: CoroutineDispatcher = writer + ): SecureUserStoreImpl = + SecureUserStoreImpl(contextMock, mockLogger, dispatcher, crypto).also { it.clearAll() } @Test fun saveUserId_thenGetUserId_expectRoundTrip() { @@ -25,7 +54,7 @@ class SecureUserStoreTest : RobolectricTest() { store.getUserId() shouldBeEqualTo "user-42" // A fresh instance reads the same persisted value (receiver/worker cold start). - SecureUserStoreImpl(contextMock, mockLogger).getUserId() shouldBeEqualTo "user-42" + SecureUserStoreImpl(contextMock, mockLogger, UnconfinedTestDispatcher()).getUserId() shouldBeEqualTo "user-42" } @Test @@ -47,4 +76,133 @@ class SecureUserStoreTest : RobolectricTest() { store.getUserId().shouldBeNull() } + + @Test + fun saveUserIdInBackground_whenWriteLands_expectCallbackFindsUserIdPersisted() { + val store = newStoreWithPendingWrites() + store.saveUserId("previous-user") + var seenInCallback: String? = null + + store.saveUserIdInBackground("user-42") { seenInCallback = store.getUserId() } + + // Not encrypted on the caller's thread; the old value stays until the write lands. + verify(exactly = 0) { crypto.encrypt("user-42") } + seenInCallback.shouldBeNull() + store.getUserId() shouldBeEqualTo "previous-user" + + writer.scheduler.advanceUntilIdle() + + seenInCallback shouldBeEqualTo "user-42" + store.getUserId() shouldBeEqualTo "user-42" + } + + @Test + fun saveUserIdInBackground_givenNull_expectClearedAndCallbackRunsNow() { + val store = newStoreWithPendingWrites() + store.saveUserId("user-42") + var callbackRan = false + + store.saveUserIdInBackground(null) { callbackRan = true } + + callbackRan.shouldBeTrue() + store.getUserId().shouldBeNull() + } + + @Test + fun clearAll_givenBackgroundWritePending_expectWriteAndCallbackDropped() { + val store = newStoreWithPendingWrites() + var callbackRan = false + store.saveUserIdInBackground("user-42") { callbackRan = true } + + store.clearAll() + writer.scheduler.advanceUntilIdle() + + callbackRan.shouldBeFalse() + store.getUserId().shouldBeNull() + } + + @Test + fun saveUserId_givenOtherUserWhileBackgroundWritePending_expectNewerSaveWins() { + val store = newStoreWithPendingWrites() + val published = mutableListOf() + store.saveUserIdInBackground("user-a") { published += "user-a" } + + store.saveUserId("user-b") { published += "user-b" } + writer.scheduler.advanceUntilIdle() + + published shouldBeEqualTo listOf("user-b") + store.getUserId() shouldBeEqualTo "user-b" + } + + @Test + fun saveUserId_givenSameUserWhileBackgroundWritePending_expectCallbackAfterWriteWithoutEncryptingAgain() { + val store = newStoreWithPendingWrites() + val seenInCallbacks = mutableListOf() + store.saveUserIdInBackground("user-42") { seenInCallbacks += store.getUserId() } + + store.saveUserId("user-42") { seenInCallbacks += store.getUserId() } + + // Waits for the pending write rather than publishing before the userId is on disk. + seenInCallbacks shouldBeEqualTo emptyList() + + writer.scheduler.advanceUntilIdle() + + seenInCallbacks shouldBeEqualTo listOf("user-42", "user-42") + verify(exactly = 1) { crypto.encrypt("user-42") } + } + + @Test + fun saveUserId_givenSameUserAlreadyPersisted_expectCallbackNowWithoutEncryptingAgain() { + val store = newStoreWithPendingWrites() + store.saveUserIdInBackground("user-42") {} + writer.scheduler.advanceUntilIdle() + var callbackRan = false + + store.saveUserId("user-42") { callbackRan = true } + + callbackRan.shouldBeTrue() + verify(exactly = 1) { crypto.encrypt("user-42") } + } + + @Test + fun saveUserId_givenSameUserAfterClear_expectEncryptedAgain() { + val store = newStoreWithPendingWrites() + store.saveUserId("user-42") + store.clearAll() + + store.saveUserId("user-42") + + store.getUserId() shouldBeEqualTo "user-42" + verify(exactly = 2) { crypto.encrypt("user-42") } + } + + @Test + fun saveUserIdInBackground_whileEncryptionBlocked_expectSaveAndClearNotToWaitOnIt() { + val encryptEntered = CountDownLatch(1) + val releaseEncrypt = CountDownLatch(1) + val blockedCrypto = fakeCrypto { + encryptEntered.countDown() + releaseEncrypt.await() + } + val store = newStoreWithPendingWrites(crypto = blockedCrypto, dispatcher = Dispatchers.IO) + val published = CountDownLatch(1) + store.saveUserIdInBackground("user-42") { published.countDown() } + encryptEntered.await(5, TimeUnit.SECONDS).shouldBeTrue() + + // The lock is never held across encryption, so these return while the Keystore is stuck. + val caller = Thread { + store.saveUserId("user-42") { published.countDown() } + store.clearAll() + } + caller.start() + caller.join(5_000) + caller.isAlive.shouldBeFalse() + + releaseEncrypt.countDown() + + // The clear superseded the blocked write, so neither the write nor its callbacks land. + published.await(1, TimeUnit.SECONDS).shouldBeFalse() + store.getUserId().shouldBeNull() + verify(exactly = 1) { blockedCrypto.encrypt("user-42") } + } } diff --git a/datapipelines/src/main/kotlin/io/customer/sdk/CustomerIO.kt b/datapipelines/src/main/kotlin/io/customer/sdk/CustomerIO.kt index 1034f0948..ddca1db9e 100644 --- a/datapipelines/src/main/kotlin/io/customer/sdk/CustomerIO.kt +++ b/datapipelines/src/main/kotlin/io/customer/sdk/CustomerIO.kt @@ -155,15 +155,20 @@ class CustomerIO private constructor( // subscribe to journey events emitted from push/in-app module to send them via data pipelines subscribeToJourneyEvents() - // republish profile/anonymous events for late-added modules - publishUserChanged() + // republish profile/anonymous events for late-added modules. Encrypts in the background so a + // slow Keystore can't block init (ANR); modules get the event once the userId is persisted. + publishUserChanged(userId = analytics.userId(), persistInBackground = true) } - /** Persists the userId before publishing so subscribers that read [secureUserStore] don't race the event. */ - private fun publishUserChanged(userId: String? = analytics.userId()) { + /** Publishes once the userId is persisted, so subscribers that read [secureUserStore] don't race the event. */ + private fun publishUserChanged(userId: String?, persistInBackground: Boolean = false) { val anonymousId = analytics.anonymousId() - secureUserStore.saveUserId(userId) - eventBus.publish(Event.UserChangedEvent(userId = userId, anonymousId = anonymousId)) + val publish = { eventBus.publish(Event.UserChangedEvent(userId = userId, anonymousId = anonymousId)) } + if (persistInBackground) { + secureUserStore.saveUserIdInBackground(userId, onPersisted = publish) + } else { + secureUserStore.saveUserId(userId, onPersisted = publish) + } } private fun subscribeToJourneyEvents() { diff --git a/datapipelines/src/test/java/io/customer/datapipelines/DataPipelineEventBusInteractionTest.kt b/datapipelines/src/test/java/io/customer/datapipelines/DataPipelineEventBusInteractionTest.kt index e3dd56422..8ca0a67f5 100644 --- a/datapipelines/src/test/java/io/customer/datapipelines/DataPipelineEventBusInteractionTest.kt +++ b/datapipelines/src/test/java/io/customer/datapipelines/DataPipelineEventBusInteractionTest.kt @@ -52,7 +52,7 @@ class DataPipelineEventBusInteractionTest : JUnitTest() { // passes only if the write happens in identify() itself, not via event dispatch. sdkInstance.identify("user-sync") - verify(exactly = 1) { mockSecureUserStore.saveUserId("user-sync") } + verify(exactly = 1) { mockSecureUserStore.saveUserId("user-sync", any()) } } @Test diff --git a/datapipelines/src/test/java/io/customer/datapipelines/UserChangedEventPersistenceTest.kt b/datapipelines/src/test/java/io/customer/datapipelines/UserChangedEventPersistenceTest.kt new file mode 100644 index 000000000..a7d80d4f1 --- /dev/null +++ b/datapipelines/src/test/java/io/customer/datapipelines/UserChangedEventPersistenceTest.kt @@ -0,0 +1,79 @@ +package io.customer.datapipelines + +import io.customer.commontest.config.TestConfig +import io.customer.datapipelines.testutils.core.JUnitTest +import io.customer.datapipelines.testutils.core.testConfiguration +import io.customer.sdk.communication.Event +import io.customer.sdk.communication.EventBus +import io.customer.sdk.data.store.SecureUserStore +import io.mockk.Runs +import io.mockk.every +import io.mockk.just +import io.mockk.mockk +import io.mockk.slot +import io.mockk.spyk +import io.mockk.verify +import org.amshove.kluent.shouldBeEmpty +import org.amshove.kluent.shouldBeEqualTo +import org.junit.jupiter.api.Test + +/** + * [Event.UserChangedEvent] is published only once the userId is persisted, so modules that read + * [SecureUserStore] when notified find it. At init the save runs in the background, keeping the + * Keystore off the caller's thread. + */ +class UserChangedEventPersistenceTest : JUnitTest() { + private val restoredUserId = "restored-user" + private val publishedEvents = mutableListOf() + + // 'mock' prefix to avoid shadowing the real androidSDKComponent.secureUserStore. + private val mockSecureUserStore: SecureUserStore = mockk(relaxed = true) + private val initSaveCallback = slot<() -> Unit>() + + override fun setup(testConfig: TestConfig) { + every { mockSecureUserStore.saveUserIdInBackground(restoredUserId, capture(initSaveCallback)) } just Runs + + super.setup( + testConfiguration { + // A user identified in a previous run, restored by analytics before CustomerIO init. + analytics { spyk(this).also { every { it.userId() } returns restoredUserId } } + diGraph { + sdk { + val mockEventBus = mockk(relaxed = true) + every { mockEventBus.publish(any()) } answers { publishedEvents += firstArg() } + overrideDependency(mockEventBus) + } + android { overrideDependency(mockSecureUserStore) } + } + } + ) + } + + private fun userChangedEvents() = publishedEvents.filterIsInstance() + + @Test + fun givenIdentifiedUserAtInit_expectSaveInBackgroundAndEventOnlyAfterItLands() { + verify(exactly = 0) { mockSecureUserStore.saveUserId(any(), any()) } + userChangedEvents().shouldBeEmpty() + + initSaveCallback.captured.invoke() + + userChangedEvents() shouldBeEqualTo listOf( + Event.UserChangedEvent(userId = restoredUserId, anonymousId = analytics.anonymousId()) + ) + } + + @Test + fun givenIdentify_expectEventOnlyAfterSaveReportsPersisted() { + val identifySaveCallback = slot<() -> Unit>() + every { mockSecureUserStore.saveUserId("user-b", capture(identifySaveCallback)) } just Runs + + sdkInstance.identify("user-b") + + userChangedEvents().shouldBeEmpty() + + identifySaveCallback.captured.invoke() + + userChangedEvents().map { it.userId } shouldBeEqualTo listOf("user-b") + } +} diff --git a/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt b/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt index fff61dfa7..0660f9565 100644 --- a/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt +++ b/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt @@ -118,6 +118,12 @@ internal class PollingLifecycleManager( logger.debug("[Polling] Skipping polling - SSE is active (sseEnabled=${currentState.sseEnabled}, isUserIdentified=${currentState.isUserIdentified})") return } + // No identity yet: an identified cold start delivers it once the userId is persisted, and a + // fetch now would go out without a user token. The fetch on UserChangedEvent starts polling. + if (currentState.userId == null && currentState.anonymousId == null) { + logger.debug("[Polling] Skipping polling - no user identity yet") + return + } logger.debug("[Polling] Starting polling (sseEnabled=${currentState.sseEnabled}, isUserIdentified=${currentState.isUserIdentified}, interval=${duration}ms, initialDelay=${initialDelay}ms)") timer?.cancel() diff --git a/messaginginapp/src/test/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManagerTest.kt b/messaginginapp/src/test/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManagerTest.kt index e0adfa2ed..86c37a884 100644 --- a/messaginginapp/src/test/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManagerTest.kt +++ b/messaginginapp/src/test/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManagerTest.kt @@ -30,8 +30,9 @@ class PollingLifecycleManagerTest : JUnitTest() { private val logger = mockk(relaxed = true) private val mainThreadPoster = mockk(relaxed = true) - // SSE inactive by default (sseEnabled=false), so polling is the active transport. - private var state = InAppMessagingState(pollInterval = 600000L) + // SSE inactive by default (sseEnabled=false), so polling is the active transport. An anonymous + // identity is present, as it is once the SDK publishes the first UserChangedEvent. + private var state = InAppMessagingState(pollInterval = 600000L, anonymousId = "anon-123") // Captured attribute-subscription listeners, in registration order: // 0 = pollInterval, 1 = sseEnabled, 2 = isUserIdentified. They are captured (not auto-invoked) @@ -89,7 +90,7 @@ class PollingLifecycleManagerTest : JUnitTest() { @Test fun init_whenAppStartedAndSseInactive_startsPollingAndFetches() { every { lifecycle.currentState } returns Lifecycle.State.STARTED - state = InAppMessagingState(pollInterval = 600000L, sseEnabled = false) + state = InAppMessagingState(pollInterval = 600000L, sseEnabled = false, anonymousId = "anon-123") createManager() @@ -119,6 +120,32 @@ class PollingLifecycleManagerTest : JUnitTest() { verify { logger.debug(match { it.contains("Not starting polling") }) } } + @Test + fun onStart_whenNoIdentityYet_doesNotPoll() { + // Identified cold start: UserChangedEvent arrives only once the userId is persisted. + state = InAppMessagingState(pollInterval = 600000L) + createManager() + val observer = captureObserver() + + observer.onStart(processLifecycleOwner) + + verify(exactly = 0) { gistQueue.fetchUserMessages() } + verify { logger.debug(match { it.contains("no user identity yet") }) } + } + + @Test + fun fetchInAppMessages_whenIdentityArrivesAfterSkippedForeground_startsPolling() { + state = InAppMessagingState(pollInterval = 600000L) + val manager = createManager() + captureObserver().onStart(processLifecycleOwner) + + // ModuleMessagingInApp sets the identity, then fetches, on UserChangedEvent. + state = state.copy(userId = "user-123", anonymousId = "anon-123") + manager.fetchInAppMessages() + + verify(exactly = 1, timeout = 2000) { gistQueue.fetchUserMessages() } + } + @Test fun onStart_whenAlreadyForegrounded_skipsSecondTime() { createManager() @@ -199,7 +226,7 @@ class PollingLifecycleManagerTest : JUnitTest() { // Timer alive after reset(). A fast interval surfaces orphans quickly. val fetchCount = AtomicInteger(0) every { gistQueue.fetchUserMessages() } answers { fetchCount.incrementAndGet() } - state = InAppMessagingState(pollInterval = 30L) + state = InAppMessagingState(pollInterval = 30L, anonymousId = "anon-123") val manager = createManager() val threads = (1..8).map { Thread { repeat(25) { manager.fetchInAppMessages() } } } From 33d1abeec9a44f44e6cc1aac9f97ac6faea393ad Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 15:23:14 +0400 Subject: [PATCH 02/12] fix(location): restore the persisted location off the init thread The restore decrypted the cached location on the caller's thread during initialize(). It now runs on the location scope; identify context, the UserChangedEvent handler and getLastKnownLocation wait for it off main. A restore that lands after a newer fix or a sign-out leaves state alone. Co-Authored-By: Claude Opus 5.5 --- .../io/customer/location/LocationServices.kt | 3 +- .../customer/location/LocationServicesImpl.kt | 2 +- .../io/customer/location/LocationTracker.kt | 72 +++++++++++++++-- .../io/customer/location/ModuleLocation.kt | 9 ++- .../customer/location/LocationTrackerTest.kt | 81 +++++++++++++++++++ 5 files changed, 155 insertions(+), 12 deletions(-) diff --git a/location/src/main/kotlin/io/customer/location/LocationServices.kt b/location/src/main/kotlin/io/customer/location/LocationServices.kt index 3c951e557..1b5317b71 100644 --- a/location/src/main/kotlin/io/customer/location/LocationServices.kt +++ b/location/src/main/kotlin/io/customer/location/LocationServices.kt @@ -72,7 +72,8 @@ interface LocationServices { * Returns the most recent location the SDK observed from any source — host * [setLastKnownLocation], [requestLocationUpdate], or a silent * [requestLocationUpdateSilently] fix — or null if none yet. Internal API for - * other SDK modules. + * other SDK modules. Waits for the persisted location restored at init, so call + * it off the main thread. */ @InternalCustomerIOApi fun getLastKnownLocation(): LocationCoordinates? diff --git a/location/src/main/kotlin/io/customer/location/LocationServicesImpl.kt b/location/src/main/kotlin/io/customer/location/LocationServicesImpl.kt index 13de0125f..22a905e91 100644 --- a/location/src/main/kotlin/io/customer/location/LocationServicesImpl.kt +++ b/location/src/main/kotlin/io/customer/location/LocationServicesImpl.kt @@ -93,7 +93,7 @@ internal class LocationServicesImpl( } @OptIn(InternalCustomerIOApi::class) - override fun getLastKnownLocation(): LocationCoordinates? = locationTracker.lastKnownLocation + override fun getLastKnownLocation(): LocationCoordinates? = locationTracker.awaitLastKnownLocation() /** * Cancels any in-flight location request. diff --git a/location/src/main/kotlin/io/customer/location/LocationTracker.kt b/location/src/main/kotlin/io/customer/location/LocationTracker.kt index a9afae838..035fdd9b1 100644 --- a/location/src/main/kotlin/io/customer/location/LocationTracker.kt +++ b/location/src/main/kotlin/io/customer/location/LocationTracker.kt @@ -10,6 +10,11 @@ import io.customer.sdk.core.pipeline.IdentifyHook import io.customer.sdk.core.util.Logger import io.customer.sdk.util.EventNames import java.lang.ref.WeakReference +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull /** * Coordinates all location state management: persistence, restoration, @@ -59,7 +64,20 @@ internal class LocationTracker( internal var lastKnownLocation: LocationCoordinates? = null private set + // Guards the two locations against a background restore landing after a newer fix or a + // sign-out. Never held across a Keystore call. + private val stateLock = Any() + private var resetCount = 0 + + // The background restore started at init; readers that need its result wait on it. + @Volatile + private var restoreJob: Job? = null + override fun getIdentifyContext(): Map { + // Runs on Segment's pipeline thread, so an identify right after init still gets the + // persisted location. Capped like Segment's own WaitingPlugin; a stalled Keystore + // couldn't decrypt it anyway. + awaitRestoreBlocking() val location = trackedLocation ?: return emptyMap() return mapOf( "location_latitude" to location.latitude, @@ -75,24 +93,55 @@ internal class LocationTracker( * guaranteeing no stale data is available for a subsequent identify(). */ override fun resetContext() { - trackedLocation = null - lastKnownLocation = null + synchronized(stateLock) { + resetCount++ + trackedLocation = null + lastKnownLocation = null + } locationPreferenceStore.clearCachedLocation() locationSyncFilter.clearSyncedData() logger.debug("Location state reset") } + /** Runs [restorePersistedLocation] in [scope], keeping its Keystore decrypts off the caller's thread. */ + fun restorePersistedLocationInBackground(scope: CoroutineScope) { + restoreJob = scope.launch { restorePersistedLocation() } + } + + /** Waits for the restore started by [restorePersistedLocationInBackground], if any. */ + suspend fun awaitRestore() { + restoreJob?.join() + } + + /** [lastKnownLocation] once the init restore has landed. Blocks, so never call it on the main thread. */ + fun awaitLastKnownLocation(): LocationCoordinates? { + awaitRestoreBlocking() + return lastKnownLocation + } + + private fun awaitRestoreBlocking() { + val job = restoreJob ?: return + if (job.isCompleted) return + runBlocking { withTimeoutOrNull(RESTORE_WAIT_MILLIS) { job.join() } } + } + /** * Reads persisted cached location from the preference store and sets the * in-memory cache so that identify events have location context * immediately after SDK restart. */ fun restorePersistedLocation() { + val resetsAtStart = synchronized(stateLock) { resetCount } val lat = locationPreferenceStore.getCachedLatitude() ?: return val lng = locationPreferenceStore.getCachedLongitude() ?: return val restored = LocationCoordinates(latitude = lat, longitude = lng) - trackedLocation = restored - lastKnownLocation = restored + synchronized(stateLock) { + // A sign-out while decrypting must not bring the old user's location back, and a + // fix that arrived meanwhile is newer than the persisted one. + if (resetCount != resetsAtStart) return + if (trackedLocation == null) trackedLocation = restored + if (lastKnownLocation == null) lastKnownLocation = restored + } logger.debug("Restored persisted location: lat=$lat, lng=$lng") } @@ -108,8 +157,10 @@ internal class LocationTracker( logger.debug("Location update received: lat=$latitude, lng=$longitude") val location = LocationCoordinates(latitude = latitude, longitude = longitude) - trackedLocation = location - lastKnownLocation = location + synchronized(stateLock) { + trackedLocation = location + lastKnownLocation = location + } locationPreferenceStore.saveCachedLocation(latitude, longitude) trySendLocationTrack(latitude, longitude) @@ -137,7 +188,9 @@ internal class LocationTracker( ) { logger.debug("Location update received (geofence-only, not tracked): lat=$latitude, lng=$longitude") - lastKnownLocation = LocationCoordinates(latitude = latitude, longitude = longitude) + synchronized(stateLock) { + lastKnownLocation = LocationCoordinates(latitude = latitude, longitude = longitude) + } eventBus.publish( Event.LocationAcquired(latitude = latitude, longitude = longitude) ) @@ -194,6 +247,11 @@ internal class LocationTracker( ) ) } + + private companion object { + // Segment's WaitingPlugin default. + const val RESTORE_WAIT_MILLIS = 30_000L + } } /** diff --git a/location/src/main/kotlin/io/customer/location/ModuleLocation.kt b/location/src/main/kotlin/io/customer/location/ModuleLocation.kt index f365ab542..1ddc0eaeb 100644 --- a/location/src/main/kotlin/io/customer/location/ModuleLocation.kt +++ b/location/src/main/kotlin/io/customer/location/ModuleLocation.kt @@ -8,7 +8,6 @@ import io.customer.location.store.LocationPreferenceStoreImpl import io.customer.location.sync.LocationSyncFilter import io.customer.location.sync.LocationSyncStoreImpl import io.customer.sdk.communication.Event -import io.customer.sdk.communication.subscribe import io.customer.sdk.core.di.SDKComponent import io.customer.sdk.core.module.CustomerIOModule import io.customer.sdk.core.pipeline.DataPipeline @@ -111,13 +110,17 @@ class ModuleLocation @JvmOverloads constructor( val services = _locationServices ?: return - locationTracker.restorePersistedLocation() + // Off the caller's thread: decrypting is a Keystore call that can stall for seconds (ANR + // at init). Readers of the restored location wait for it instead. + locationTracker.restorePersistedLocationInBackground(locationScope) // On identify, attempt to send a supplementary "CIO Location Update" track event. // The identify event itself already carries location via context enrichment — // this track event is for journey/segment triggers in the user's timeline. - eventBus.subscribe { + eventBus.subscribe(Event.UserChangedEvent::class) { if (!it.userId.isNullOrEmpty()) { + // The ON_APP_START check below must see the restored location. + locationTracker.awaitRestore() locationTracker.onUserIdentified() // ON_APP_START's lifecycle one-shot fires per process, but resetContext() diff --git a/location/src/test/java/io/customer/location/LocationTrackerTest.kt b/location/src/test/java/io/customer/location/LocationTrackerTest.kt index 6a6411b94..a019d3231 100644 --- a/location/src/test/java/io/customer/location/LocationTrackerTest.kt +++ b/location/src/test/java/io/customer/location/LocationTrackerTest.kt @@ -11,9 +11,15 @@ import io.mockk.every import io.mockk.mockk import io.mockk.slot import io.mockk.verify +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.runBlocking import org.amshove.kluent.shouldBeEmpty import org.amshove.kluent.shouldBeEqualTo import org.amshove.kluent.shouldBeNull +import org.amshove.kluent.shouldBeTrue import org.amshove.kluent.shouldNotBeEmpty import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test @@ -211,6 +217,81 @@ class LocationTrackerTest { tracker.getIdentifyContext().shouldBeEmpty() } + // -- restorePersistedLocationInBackground -- + + private val restoreScope = CoroutineScope(Dispatchers.IO) + + // Holds the restore inside its Keystore read until released. + private fun blockRestoreAt(latitude: Double, longitude: Double): Pair { + val restoreEntered = CountDownLatch(1) + val releaseRestore = CountDownLatch(1) + every { store.getCachedLatitude() } answers { + restoreEntered.countDown() + releaseRestore.await() + latitude + } + every { store.getCachedLongitude() } returns longitude + return restoreEntered to releaseRestore + } + + @Test + fun givenRestoreInBackground_expectIdentifyContextWaitsForRestoredLocation() { + val (restoreEntered, releaseRestore) = blockRestoreAt(37.7749, -122.4194) + tracker.restorePersistedLocationInBackground(restoreScope) + restoreEntered.await(5, TimeUnit.SECONDS).shouldBeTrue() + + Thread { + Thread.sleep(200) + releaseRestore.countDown() + }.start() + + tracker.getIdentifyContext()["location_latitude"] shouldBeEqualTo 37.7749 + } + + @Test + fun givenFixArrivesWhileRestoring_expectRestoreDoesNotOverwriteIt() { + val (restoreEntered, releaseRestore) = blockRestoreAt(1.0, 2.0) + tracker.restorePersistedLocationInBackground(restoreScope) + restoreEntered.await(5, TimeUnit.SECONDS).shouldBeTrue() + + tracker.onLocationReceived(37.7749, -122.4194) + releaseRestore.countDown() + runBlocking { tracker.awaitRestore() } + + val fresh = LocationCoordinates(latitude = 37.7749, longitude = -122.4194) + tracker.trackedLocation shouldBeEqualTo fresh + tracker.lastKnownLocation shouldBeEqualTo fresh + } + + @Test + fun givenResetWhileRestoring_expectRestoreDoesNotBringLocationBack() { + val (restoreEntered, releaseRestore) = blockRestoreAt(37.7749, -122.4194) + tracker.restorePersistedLocationInBackground(restoreScope) + restoreEntered.await(5, TimeUnit.SECONDS).shouldBeTrue() + + tracker.resetContext() + releaseRestore.countDown() + runBlocking { tracker.awaitRestore() } + + tracker.trackedLocation.shouldBeNull() + tracker.lastKnownLocation.shouldBeNull() + tracker.getIdentifyContext().shouldBeEmpty() + } + + @Test + fun givenSilentFixWhileRestoring_expectTrackedRestoredAndLastKnownKeepsFix() { + val (restoreEntered, releaseRestore) = blockRestoreAt(1.0, 2.0) + tracker.restorePersistedLocationInBackground(restoreScope) + restoreEntered.await(5, TimeUnit.SECONDS).shouldBeTrue() + + tracker.onLocationReceivedWithoutTracking(37.7749, -122.4194) + releaseRestore.countDown() + runBlocking { tracker.awaitRestore() } + + tracker.trackedLocation shouldBeEqualTo LocationCoordinates(latitude = 1.0, longitude = 2.0) + tracker.lastKnownLocation shouldBeEqualTo LocationCoordinates(latitude = 37.7749, longitude = -122.4194) + } + // -- getIdentifyContext -- @Test From 8d6a9e6baa2ff5f1cd6a716e255f068b76434aed Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 15:33:49 +0400 Subject: [PATCH 03/12] fix(core): run event bus, location and geofence work on the IO dispatcher These scopes block on disk and Keystore calls. On Default, which has as many threads as the device has cores (2 on low-end devices), a slow Keystore in geofence or location work stalled event delivery for every module, including the push and in-app metrics carried by the bus. Co-Authored-By: Claude Opus 5.5 --- .../kotlin/io/customer/sdk/core/util/ScopeProvider.kt | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/core/src/main/kotlin/io/customer/sdk/core/util/ScopeProvider.kt b/core/src/main/kotlin/io/customer/sdk/core/util/ScopeProvider.kt index c6a3aa0bf..b886eec3c 100644 --- a/core/src/main/kotlin/io/customer/sdk/core/util/ScopeProvider.kt +++ b/core/src/main/kotlin/io/customer/sdk/core/util/ScopeProvider.kt @@ -23,14 +23,17 @@ class SdkScopeProvider(private val dispatchers: DispatchersProvider) : ScopeProv SDKComponent.logger.error("Uncaught exception in SDK coroutine: $throwable", throwable = throwable) } + // Event bus, location and geofence work blocks on disk and Keystore calls, so it runs on the + // background (IO) dispatcher. On Default, which has as few threads as the device has cores, a + // slow Keystore would stall event delivery, including the metrics carried by the bus. override val eventBusScope: CoroutineScope - get() = CoroutineScope(dispatchers.default + SupervisorJob() + exceptionHandler) + get() = CoroutineScope(dispatchers.background + SupervisorJob() + exceptionHandler) override val lifecycleListenerScope: CoroutineScope get() = CoroutineScope(dispatchers.default + SupervisorJob() + exceptionHandler) override val inAppLifecycleScope: CoroutineScope get() = CoroutineScope(dispatchers.default + SupervisorJob() + exceptionHandler) override val locationScope: CoroutineScope - get() = CoroutineScope(dispatchers.default + SupervisorJob() + exceptionHandler) + get() = CoroutineScope(dispatchers.background + SupervisorJob() + exceptionHandler) override val geofenceScope: CoroutineScope - get() = CoroutineScope(dispatchers.default + SupervisorJob() + exceptionHandler) + get() = CoroutineScope(dispatchers.background + SupervisorJob() + exceptionHandler) } From 309aa75917c9702b34407f86bc54ab155154b5f3 Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 15:33:49 +0400 Subject: [PATCH 04/12] fix(messaginginapp): start polling once when the identity arrives late When foreground polling was skipped for lack of identity, the identification-change path and the UserChangedEvent fetch both started polling, sending two fetches. The identification-change path now leaves the start to the identity fetch. Co-Authored-By: Claude Opus 5.5 --- .../presentation/PollingLifecycleManager.kt | 16 +++++++-- .../PollingLifecycleManagerTest.kt | 33 +++++++++++++++++++ 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt b/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt index 0660f9565..077ea9889 100644 --- a/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt +++ b/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt @@ -35,6 +35,11 @@ internal class PollingLifecycleManager( @Volatile private var timer: Timer? = null + // Set when a start was skipped for lack of identity. The fetch ModuleMessagingInApp runs when the + // identity arrives starts polling, so the identification-change path must not start a second one. + // Guarded by this instance's monitor, like [timer] writes. + private var awaitingIdentity = false + // Guards the poll-interval subscription's initial replay emission. The initial/foreground start // is owned by handleForegrounded()/fetchInAppMessages(); reacting to the replay would race with // and could cancel that catch-up fetch. Only genuine interval changes should restart polling. @@ -113,14 +118,15 @@ internal class PollingLifecycleManager( @Synchronized private fun startPolling(duration: Long, initialDelay: Long = 0) { val currentState = state + // No identity yet: an identified cold start delivers it once the userId is persisted, and a + // fetch now would go out without a user token. The fetch on UserChangedEvent starts polling. + awaitingIdentity = currentState.userId == null && currentState.anonymousId == null // Only skip polling if SSE should be used (both flag enabled AND user identified) if (currentState.shouldUseSse) { logger.debug("[Polling] Skipping polling - SSE is active (sseEnabled=${currentState.sseEnabled}, isUserIdentified=${currentState.isUserIdentified})") return } - // No identity yet: an identified cold start delivers it once the userId is persisted, and a - // fetch now would go out without a user token. The fetch on UserChangedEvent starts polling. - if (currentState.userId == null && currentState.anonymousId == null) { + if (awaitingIdentity) { logger.debug("[Polling] Skipping polling - no user identity yet") return } @@ -177,6 +183,8 @@ internal class PollingLifecycleManager( } } + // Synchronized with startPolling, so the identity fetch can't start polling between the checks below. + @Synchronized private fun updatePollingForSseAvailability(reason: String) { // Only manage polling when app is foregrounded if (!isForegrounded.get()) { @@ -187,6 +195,8 @@ internal class PollingLifecycleManager( if (currentState.shouldUseSse) { logger.debug("[Polling] $reason - SSE now active, stopping polling") resetTimer() + } else if (awaitingIdentity) { + logger.debug("[Polling] $reason - waiting for the identity fetch to start polling") } else if (timer == null) { // Resume polling only when it isn't already running, i.e. we're transitioning away // from SSE (which had stopped the timer). When polling is already active this branch diff --git a/messaginginapp/src/test/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManagerTest.kt b/messaginginapp/src/test/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManagerTest.kt index 86c37a884..3fdfe85cc 100644 --- a/messaginginapp/src/test/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManagerTest.kt +++ b/messaginginapp/src/test/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManagerTest.kt @@ -146,6 +146,39 @@ class PollingLifecycleManagerTest : JUnitTest() { verify(exactly = 1, timeout = 2000) { gistQueue.fetchUserMessages() } } + @Test + fun identificationChange_whenIdentityArrivesBeforeItsFetch_expectExactlyOneFetch() { + state = InAppMessagingState(pollInterval = 600000L) + val manager = createManager() + captureObserver().onStart(processLifecycleOwner) + + // The isUserIdentified subscription can fire before ModuleMessagingInApp's own fetch. + state = state.copy(userId = "user-123", anonymousId = "anon-123") + userIdentifiedListener(true) + manager.fetchInAppMessages() + + verify(exactly = 1, timeout = 2000) { gistQueue.fetchUserMessages() } + Thread.sleep(200) // a second immediate tick would land well within this + verify(exactly = 1) { gistQueue.fetchUserMessages() } + } + + @Test + fun sseAvailabilityChange_whenIdentityArrivedWhileSseActive_expectPollingResumesWhenSseStops() { + state = InAppMessagingState(pollInterval = 600000L) + val manager = createManager() + captureObserver().onStart(processLifecycleOwner) + + state = state.copy(userId = "user-123", anonymousId = "anon-123", sseEnabled = true) + userIdentifiedListener(true) + manager.fetchInAppMessages() + verify(exactly = 0) { gistQueue.fetchUserMessages() } + + state = state.copy(sseEnabled = false) + sseEnabledListener(false) + + verify(timeout = 2000) { gistQueue.fetchUserMessages() } + } + @Test fun onStart_whenAlreadyForegrounded_skipsSecondTime() { createManager() From b6f6c9958fe4bf29a9030ff721e37e03408b174b Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 15:58:20 +0400 Subject: [PATCH 05/12] fix(core): never drop bus events behind a slow subscriber The bus forwarded with tryEmit, which fails once the slowest subscriber is a full replay buffer behind, losing the event for every subscriber, metrics included. emit suspends the forwarder instead; the unlimited channel holds what is published meanwhile. Co-Authored-By: Claude Opus 5.5 --- .../io/customer/sdk/communication/EventBus.kt | 4 ++- .../sdk/communication/EventBusTest.kt | 25 +++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/core/src/main/kotlin/io/customer/sdk/communication/EventBus.kt b/core/src/main/kotlin/io/customer/sdk/communication/EventBus.kt index ae0779dc4..25f291c16 100644 --- a/core/src/main/kotlin/io/customer/sdk/communication/EventBus.kt +++ b/core/src/main/kotlin/io/customer/sdk/communication/EventBus.kt @@ -47,7 +47,9 @@ class EventBusImpl( init { scope.launch { eventChannel.consumeAsFlow().collect { event -> - sharedFlow.tryEmit(event) + // Suspends while a slow subscriber is a full buffer behind; tryEmit would drop the + // event for every subscriber. The unlimited channel holds what's published meanwhile. + sharedFlow.emit(event) } } } diff --git a/core/src/test/java/io/customer/sdk/communication/EventBusTest.kt b/core/src/test/java/io/customer/sdk/communication/EventBusTest.kt index 9aa03f4c7..512b254cb 100644 --- a/core/src/test/java/io/customer/sdk/communication/EventBusTest.kt +++ b/core/src/test/java/io/customer/sdk/communication/EventBusTest.kt @@ -7,6 +7,7 @@ import io.customer.commontest.util.ScopeProviderStub import io.customer.sdk.core.di.SDKComponent import io.customer.sdk.core.util.ScopeProvider import io.customer.sdk.events.Metric +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.runBlocking @@ -273,6 +274,30 @@ class EventBusTest : JUnit5Test() { job.cancel() } + @Test + fun givenSubscriberStalledFullBufferBehind_expectNoEventDropped() = runBlocking { + val releaseSlowSubscriber = CompletableDeferred() + val slowEvents = mutableListOf() + val otherEvents = mutableListOf() + val slowJob = eventBus.subscribe(Event.ScreenViewedEvent::class) { event -> + releaseSlowSubscriber.await() + slowEvents.add(event) + } + val otherJob = eventBus.subscribe { otherEvents.add(it) } + + // More than the 100-event buffer while one subscriber is stuck (e.g. on a slow Keystore). + repeat(150) { index -> eventBus.publish(Event.ScreenViewedEvent(name = "screen$index")) } + yield() + releaseSlowSubscriber.complete(Unit) + yield() + + otherEvents.map { (it as Event.ScreenViewedEvent).name } shouldBeEqualTo (0 until 150).map { "screen$it" } + slowEvents.size shouldBeEqualTo 150 + + slowJob.cancel() + otherJob.cancel() + } + @Test fun givenMoreThan100EventsExpectReplayBufferLimited() = runBlocking { // Publish 150 events without any subscribers to test replay buffer limit From e65d01c80f1867f96c8d97895459b3f6848d8962 Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 15:58:20 +0400 Subject: [PATCH 06/12] fix(core): skip the Keystore for a superseded background write Co-Authored-By: Claude Opus 5.5 --- .../kotlin/io/customer/sdk/data/store/SecureUserStore.kt | 6 +++++- .../java/io/customer/sdk/data/store/SecureUserStoreTest.kt | 2 ++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt b/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt index 5cbb60525..97cebdb71 100644 --- a/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt +++ b/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt @@ -81,7 +81,11 @@ internal class SecureUserStoreImpl( override fun saveUserIdInBackground(userId: String?, onPersisted: () -> Unit) { val write = startWrite(userId, onPersisted) ?: return - writeScope.launch { finishWrite(write, crypto.encrypt(write.userId)) } + writeScope.launch { + // Superseded before it started: skip the Keystore call rather than hold the key ahead of the newer save. + if (synchronized(lock) { write.version != version }) return@launch + finishWrite(write, crypto.encrypt(write.userId)) + } } /** Returns the write to encrypt, or null when [userId] needs no Keystore call. */ diff --git a/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt b/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt index 00ab4be0e..a7bf39d00 100644 --- a/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt +++ b/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt @@ -119,6 +119,8 @@ class SecureUserStoreTest : RobolectricTest() { callbackRan.shouldBeFalse() store.getUserId().shouldBeNull() + // Superseded before the writer ran, so it never touched the Keystore. + verify(exactly = 0) { crypto.encrypt(any()) } } @Test From 45b436fc3d399dc5f15b369716808161d2ad3a5b Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 15:58:20 +0400 Subject: [PATCH 07/12] fix(location): close the restore race with sign-out, keep waits off the bus resetContext cleared the persisted location after releasing its lock, so a restore could still read the signed-out user's location. The identify handler now waits for the restore in its own coroutine, not in the bus collector. Co-Authored-By: Claude Opus 5.5 --- .../io/customer/location/LocationTracker.kt | 4 ++- .../io/customer/location/ModuleLocation.kt | 29 +++++++++++-------- 2 files changed, 20 insertions(+), 13 deletions(-) diff --git a/location/src/main/kotlin/io/customer/location/LocationTracker.kt b/location/src/main/kotlin/io/customer/location/LocationTracker.kt index 035fdd9b1..bd459047d 100644 --- a/location/src/main/kotlin/io/customer/location/LocationTracker.kt +++ b/location/src/main/kotlin/io/customer/location/LocationTracker.kt @@ -97,8 +97,10 @@ internal class LocationTracker( resetCount++ trackedLocation = null lastKnownLocation = null + // Under the lock (no Keystore call), so a restore can't read the old values after the + // count it checks has already moved on. + locationPreferenceStore.clearCachedLocation() } - locationPreferenceStore.clearCachedLocation() locationSyncFilter.clearSyncedData() logger.debug("Location state reset") } diff --git a/location/src/main/kotlin/io/customer/location/ModuleLocation.kt b/location/src/main/kotlin/io/customer/location/ModuleLocation.kt index 1ddc0eaeb..880c98df6 100644 --- a/location/src/main/kotlin/io/customer/location/ModuleLocation.kt +++ b/location/src/main/kotlin/io/customer/location/ModuleLocation.kt @@ -8,6 +8,7 @@ import io.customer.location.store.LocationPreferenceStoreImpl import io.customer.location.sync.LocationSyncFilter import io.customer.location.sync.LocationSyncStoreImpl import io.customer.sdk.communication.Event +import io.customer.sdk.communication.subscribe import io.customer.sdk.core.di.SDKComponent import io.customer.sdk.core.module.CustomerIOModule import io.customer.sdk.core.pipeline.DataPipeline @@ -15,6 +16,7 @@ import io.customer.sdk.core.pipeline.identifyHookRegistry import io.customer.sdk.core.util.HandlerMainThreadPoster import io.customer.sdk.core.util.Logger import io.customer.sdk.core.util.MainThreadPoster +import kotlinx.coroutines.launch /** * Location module for Customer.io SDK. @@ -117,19 +119,22 @@ class ModuleLocation @JvmOverloads constructor( // On identify, attempt to send a supplementary "CIO Location Update" track event. // The identify event itself already carries location via context enrichment — // this track event is for journey/segment triggers in the user's timeline. - eventBus.subscribe(Event.UserChangedEvent::class) { + eventBus.subscribe { if (!it.userId.isNullOrEmpty()) { - // The ON_APP_START check below must see the restored location. - locationTracker.awaitRestore() - locationTracker.onUserIdentified() - - // ON_APP_START's lifecycle one-shot fires per process, but resetContext() - // wipes trackedLocation on logout — a subsequent identify in the same process - // needs a fresh fetch. MANUAL deliberately doesn't auto-fetch. - if (locationTracker.trackedLocation == null && - moduleConfig.trackingMode == LocationTrackingMode.ON_APP_START - ) { - services.requestLocationUpdate() + // Off the bus collector, so waiting on a slow restore doesn't hold up other events. + locationScope.launch { + // The ON_APP_START check below must see the restored location. + locationTracker.awaitRestore() + locationTracker.onUserIdentified() + + // ON_APP_START's lifecycle one-shot fires per process, but resetContext() + // wipes trackedLocation on logout — a subsequent identify in the same process + // needs a fresh fetch. MANUAL deliberately doesn't auto-fetch. + if (locationTracker.trackedLocation == null && + moduleConfig.trackingMode == LocationTrackingMode.ON_APP_START + ) { + services.requestLocationUpdate() + } } } } From 7c62d1314b53b5a86e9349bb56ecc017fc8c257c Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 15:58:20 +0400 Subject: [PATCH 08/12] fix(messaginginapp): read in-app state before taking the polling lock Co-Authored-By: Claude Opus 5.5 --- .../gist/presentation/PollingLifecycleManager.kt | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt b/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt index 077ea9889..ce0609bbb 100644 --- a/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt +++ b/messaginginapp/src/main/java/io/customer/messaginginapp/gist/presentation/PollingLifecycleManager.kt @@ -183,15 +183,20 @@ internal class PollingLifecycleManager( } } - // Synchronized with startPolling, so the identity fetch can't start polling between the checks below. - @Synchronized private fun updatePollingForSseAvailability(reason: String) { // Only manage polling when app is foregrounded if (!isForegrounded.get()) { return } + // Read before taking this manager's monitor: dispatch holds the store's monitor while it + // calls into startPolling, so reading state under ours would invert that lock order. val currentState = state + // Synchronized with startPolling, so the identity fetch can't start polling between the checks below. + synchronized(this) { updatePollingLocked(currentState, reason) } + } + + private fun updatePollingLocked(currentState: InAppMessagingState, reason: String) { if (currentState.shouldUseSse) { logger.debug("[Polling] $reason - SSE now active, stopping polling") resetTimer() From 698e4496cb5ccd55aecbe8201fe515b6efebf12b Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 17:26:45 +0400 Subject: [PATCH 09/12] fix(core): keep the one-argument saveUserId, finish a background write on Error Modules built against the one-argument saveUserId keep linking when an app mixes module versions. A background write whose encryption throws an Error now takes the same plaintext fallback PreferenceCrypto uses for exceptions, so the write lands and its callbacks are never stranded. Co-Authored-By: Claude Opus 5.5 --- .../sdk/data/store/SecureUserStore.kt | 20 ++++++++++++--- .../sdk/data/store/SecureUserStoreTest.kt | 25 +++++++++++++++++++ 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt b/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt index 97cebdb71..51dd41546 100644 --- a/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt +++ b/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt @@ -32,7 +32,10 @@ interface SecureUserStore { * [onPersisted] runs now or once that write lands. A later save or [clearAll] that supersedes * this one drops both the write and [onPersisted]. */ - fun saveUserId(userId: String?, onPersisted: () -> Unit = {}) + fun saveUserId(userId: String?, onPersisted: () -> Unit) + + /** [saveUserId] with no callback. Kept so modules built against this signature still link. */ + fun saveUserId(userId: String?) = saveUserId(userId) {} /** * Same as [saveUserId], but encrypts on a background thread so the caller never waits on the @@ -46,7 +49,7 @@ interface SecureUserStore { internal class SecureUserStoreImpl( context: Context, - logger: Logger, + private val logger: Logger, backgroundDispatcher: CoroutineDispatcher, private val crypto: PreferenceCrypto = PreferenceCrypto(KEY_ALIAS, logger) ) : PreferenceStore(context), SecureUserStore { @@ -84,10 +87,21 @@ internal class SecureUserStoreImpl( writeScope.launch { // Superseded before it started: skip the Keystore call rather than hold the key ahead of the newer save. if (synchronized(lock) { write.version != version }) return@launch - finishWrite(write, crypto.encrypt(write.userId)) + finishWrite(write, encryptOrFallback(write.userId)) } } + // PreferenceCrypto already stores plaintext when the Keystore throws an Exception. An Error from + // a broken Keystore gets the same fallback, so the write and its callbacks are never stranded. + private fun encryptOrFallback(userId: String): String = try { + crypto.encrypt(userId) + } catch (e: VirtualMachineError) { + throw e + } catch (t: Throwable) { + logger.error("Keystore encryption failed, storing without encryption: $t", throwable = t) + userId + } + /** Returns the write to encrypt, or null when [userId] needs no Keystore call. */ private fun startWrite(userId: String?, onPersisted: () -> Unit): PendingWrite? = synchronized(lock) { val pending = pendingWrite diff --git a/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt b/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt index a7bf39d00..8d7ec0be8 100644 --- a/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt +++ b/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt @@ -15,6 +15,7 @@ import org.amshove.kluent.shouldBeEqualTo import org.amshove.kluent.shouldBeFalse import org.amshove.kluent.shouldBeNull import org.amshove.kluent.shouldBeTrue +import org.amshove.kluent.shouldNotBeNull import org.junit.Test import org.junit.runner.RunWith import org.robolectric.RobolectricTestRunner @@ -178,6 +179,30 @@ class SecureUserStoreTest : RobolectricTest() { verify(exactly = 2) { crypto.encrypt("user-42") } } + @Test + fun saveUserIdInBackground_whenEncryptionThrowsError_expectWriteLandsAndCallbacksRun() { + val brokenCrypto: PreferenceCrypto = mockk { + every { encrypt(any()) } throws ExceptionInInitializerError("broken Keystore provider") + every { decrypt(any()) } answers { firstArg() } + } + val store = newStoreWithPendingWrites(crypto = brokenCrypto) + val published = mutableListOf() + store.saveUserIdInBackground("user-42") { published += "init" } + // Attaches to the pending write; must not be stranded when that write fails. + store.saveUserId("user-42") { published += "identify" } + + writer.scheduler.advanceUntilIdle() + + published shouldBeEqualTo listOf("init", "identify") + store.getUserId() shouldBeEqualTo "user-42" + } + + @Test + fun saveUserId_expectSingleArgumentFormStillOnTheInterface() { + // Modules built against the one-argument form call it directly; removing it breaks linking. + SecureUserStore::class.java.getMethod("saveUserId", String::class.java).shouldNotBeNull() + } + @Test fun saveUserIdInBackground_whileEncryptionBlocked_expectSaveAndClearNotToWaitOnIt() { val encryptEntered = CountDownLatch(1) From cf05863acbeb2c4a4f32602ae68e8b20936071f3 Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 17:26:45 +0400 Subject: [PATCH 10/12] fix(location): bound the total wait for the init restore Readers waited up to 30 s again on every call while a stalled restore was running: each identify and each geofence anchor read. The 30 s now runs from the start of the restore, shared by all readers. Co-Authored-By: Claude Opus 5.5 --- .../io/customer/location/LocationTracker.kt | 28 +++++++++++++------ .../customer/location/LocationTrackerTest.kt | 21 ++++++++++++++ 2 files changed, 41 insertions(+), 8 deletions(-) diff --git a/location/src/main/kotlin/io/customer/location/LocationTracker.kt b/location/src/main/kotlin/io/customer/location/LocationTracker.kt index bd459047d..663caaa4e 100644 --- a/location/src/main/kotlin/io/customer/location/LocationTracker.kt +++ b/location/src/main/kotlin/io/customer/location/LocationTracker.kt @@ -10,6 +10,7 @@ import io.customer.sdk.core.pipeline.IdentifyHook import io.customer.sdk.core.util.Logger import io.customer.sdk.util.EventNames import java.lang.ref.WeakReference +import java.util.concurrent.TimeUnit import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.launch @@ -43,7 +44,8 @@ internal class LocationTracker( private val locationPreferenceStore: LocationPreferenceStore, private val locationSyncFilter: LocationSyncFilter, private val eventBus: EventBus, - private val logger: Logger + private val logger: Logger, + private val restoreWaitMillis: Long = RESTORE_WAIT_MILLIS ) : IdentifyHook { // WeakReference prevents this hook (retained by IdentifyHookRegistry @@ -69,14 +71,17 @@ internal class LocationTracker( private val stateLock = Any() private var resetCount = 0 - // The background restore started at init; readers that need its result wait on it. + // The background restore started at init; readers that need its result wait on it, but only + // until the deadline. A stalled Keystore then costs one bounded wait, not one per reader. @Volatile private var restoreJob: Job? = null + @Volatile + private var restoreDeadlineNanos = 0L + override fun getIdentifyContext(): Map { // Runs on Segment's pipeline thread, so an identify right after init still gets the - // persisted location. Capped like Segment's own WaitingPlugin; a stalled Keystore - // couldn't decrypt it anyway. + // persisted location. Past the deadline a stalled Keystore couldn't decrypt it anyway. awaitRestoreBlocking() val location = trackedLocation ?: return emptyMap() return mapOf( @@ -107,12 +112,14 @@ internal class LocationTracker( /** Runs [restorePersistedLocation] in [scope], keeping its Keystore decrypts off the caller's thread. */ fun restorePersistedLocationInBackground(scope: CoroutineScope) { + restoreDeadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(restoreWaitMillis) restoreJob = scope.launch { restorePersistedLocation() } } - /** Waits for the restore started by [restorePersistedLocationInBackground], if any. */ + /** Waits for the restore started by [restorePersistedLocationInBackground], until its deadline. */ suspend fun awaitRestore() { - restoreJob?.join() + val job = restoreJob ?: return + withTimeoutOrNull(remainingRestoreWaitMillis()) { job.join() } } /** [lastKnownLocation] once the init restore has landed. Blocks, so never call it on the main thread. */ @@ -124,9 +131,14 @@ internal class LocationTracker( private fun awaitRestoreBlocking() { val job = restoreJob ?: return if (job.isCompleted) return - runBlocking { withTimeoutOrNull(RESTORE_WAIT_MILLIS) { job.join() } } + val remaining = remainingRestoreWaitMillis() + if (remaining <= 0) return + runBlocking { withTimeoutOrNull(remaining) { job.join() } } } + private fun remainingRestoreWaitMillis(): Long = + TimeUnit.NANOSECONDS.toMillis(restoreDeadlineNanos - System.nanoTime()) + /** * Reads persisted cached location from the preference store and sets the * in-memory cache so that identify events have location context @@ -251,7 +263,7 @@ internal class LocationTracker( } private companion object { - // Segment's WaitingPlugin default. + // Total time readers wait for the init restore, from its start. Segment's WaitingPlugin default. const val RESTORE_WAIT_MILLIS = 30_000L } } diff --git a/location/src/test/java/io/customer/location/LocationTrackerTest.kt b/location/src/test/java/io/customer/location/LocationTrackerTest.kt index a019d3231..579a23d45 100644 --- a/location/src/test/java/io/customer/location/LocationTrackerTest.kt +++ b/location/src/test/java/io/customer/location/LocationTrackerTest.kt @@ -13,11 +13,13 @@ import io.mockk.slot import io.mockk.verify import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit +import kotlin.system.measureTimeMillis import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.runBlocking import org.amshove.kluent.shouldBeEmpty import org.amshove.kluent.shouldBeEqualTo +import org.amshove.kluent.shouldBeLessThan import org.amshove.kluent.shouldBeNull import org.amshove.kluent.shouldBeTrue import org.amshove.kluent.shouldNotBeEmpty @@ -292,6 +294,25 @@ class LocationTrackerTest { tracker.lastKnownLocation shouldBeEqualTo LocationCoordinates(latitude = 37.7749, longitude = -122.4194) } + @Test + fun givenRestoreStalledPastDeadline_expectOneBoundedWaitNotOnePerReader() { + val stalledTracker = LocationTracker(dataPipeline, store, syncFilter, mockEventBus, logger, restoreWaitMillis = 300) + val (restoreEntered, releaseRestore) = blockRestoreAt(37.7749, -122.4194) + stalledTracker.restorePersistedLocationInBackground(restoreScope) + restoreEntered.await(5, TimeUnit.SECONDS).shouldBeTrue() + + val firstWait = measureTimeMillis { stalledTracker.getIdentifyContext().shouldBeEmpty() } + val laterWaits = measureTimeMillis { + stalledTracker.getIdentifyContext().shouldBeEmpty() + stalledTracker.awaitLastKnownLocation().shouldBeNull() + runBlocking { stalledTracker.awaitRestore() } + } + + firstWait shouldBeLessThan 2_000L + laterWaits shouldBeLessThan 100L + releaseRestore.countDown() + } + // -- getIdentifyContext -- @Test From 44233b66a64c2dbe7310e95ad177b12fb0ab3475 Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 17:26:45 +0400 Subject: [PATCH 11/12] test(datapipelines): metrics keep flowing while the init write is pending Co-Authored-By: Claude Opus 5.5 --- .../PendingInitWriteNoLossTest.kt | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) create mode 100644 datapipelines/src/test/java/io/customer/datapipelines/PendingInitWriteNoLossTest.kt diff --git a/datapipelines/src/test/java/io/customer/datapipelines/PendingInitWriteNoLossTest.kt b/datapipelines/src/test/java/io/customer/datapipelines/PendingInitWriteNoLossTest.kt new file mode 100644 index 000000000..29c7ed068 --- /dev/null +++ b/datapipelines/src/test/java/io/customer/datapipelines/PendingInitWriteNoLossTest.kt @@ -0,0 +1,123 @@ +package io.customer.datapipelines + +import com.segment.analytics.kotlin.core.Analytics +import com.segment.analytics.kotlin.core.BaseEvent +import com.segment.analytics.kotlin.core.TrackEvent +import com.segment.analytics.kotlin.core.platform.Plugin +import io.customer.commontest.config.TestConfig +import io.customer.datapipelines.testutils.core.JUnitTest +import io.customer.datapipelines.testutils.core.testConfiguration +import io.customer.sdk.communication.Event +import io.customer.sdk.communication.EventBus +import io.customer.sdk.communication.subscribe +import io.customer.sdk.core.di.SDKComponent +import io.customer.sdk.data.store.SecureUserStore +import io.customer.sdk.events.Metric +import io.customer.sdk.util.EventNames +import io.mockk.every +import io.mockk.spyk +import java.util.Collections +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.concurrent.thread +import org.amshove.kluent.shouldBeEmpty +import org.amshove.kluent.shouldBeEqualTo +import org.amshove.kluent.shouldBeTrue +import org.junit.jupiter.api.Test + +/** + * While the identified user's init write is still encrypting on a background thread, metrics keep + * flowing to analytics, and modules get UserChangedEvent only once the userId is persisted. + */ +class PendingInitWriteNoLossTest : JUnitTest() { + private val restoredUserId = "restored-user" + private val releaseInitWrite = CountDownLatch(1) + + // Stands in for the Keystore-backed store: the init write lands on its own thread when released. + private val pendingWriteStore = object : SecureUserStore { + @Volatile + var persistedUserId: String? = null + + override fun saveUserId(userId: String?, onPersisted: () -> Unit) { + persistedUserId = userId + onPersisted() + } + + override fun saveUserIdInBackground(userId: String?, onPersisted: () -> Unit) { + thread(name = "slow-keystore-write") { + releaseInitWrite.await() + persistedUserId = userId + onPersisted() + } + } + + override fun getUserId(): String? = persistedUserId + override fun clearAll() { + persistedUserId = null + } + } + + private lateinit var eventBus: EventBus + + // Analytics runs plugins on its own threads, so count with an atomic rather than a plain list. + private val deliveryMetricsTracked = AtomicInteger() + private val deliveryMetricCounter = object : Plugin { + override val type: Plugin.Type = Plugin.Type.After + override lateinit var analytics: Analytics + + override fun execute(event: BaseEvent): BaseEvent { + if ((event as? TrackEvent)?.event == EventNames.METRIC_DELIVERY) deliveryMetricsTracked.incrementAndGet() + return event + } + } + + override fun setup(testConfig: TestConfig) { + super.setup( + testConfiguration { + analytics { spyk(this).also { every { it.userId() } returns restoredUserId } } + diGraph { + android { overrideDependency(pendingWriteStore) } + } + } + ) + eventBus = SDKComponent.eventBus + analytics.add(deliveryMetricCounter) + } + + override fun teardown() { + releaseInitWrite.countDown() + super.teardown() + } + + // Bus delivery hops threads, so wait for it rather than assume it's synchronous. + private fun waitUntil(condition: () -> Boolean): Boolean { + val deadline = System.currentTimeMillis() + 5_000 + while (!condition() && System.currentTimeMillis() < deadline) Thread.sleep(10) + return condition() + } + + @Test + fun givenInitWritePending_expectEveryMetricTrackedAndUserEventOnlyAfterPersist() { + val userIdOnDiskWhenNotified = Collections.synchronizedList(mutableListOf()) + val userEventReceived = CountDownLatch(1) + eventBus.subscribe { + userIdOnDiskWhenNotified += pendingWriteStore.getUserId() + userEventReceived.countDown() + } + + // More than the bus's 100-event replay buffer, all while the Keystore write is stuck. + repeat(150) { index -> + eventBus.publish(Event.TrackPushMetricEvent(deliveryId = "delivery-$index", event = Metric.Delivered, deviceToken = "token")) + } + + waitUntil { deliveryMetricsTracked.get() == 150 }.shouldBeTrue() + userIdOnDiskWhenNotified.shouldBeEmpty() + + releaseInitWrite.countDown() + + userEventReceived.await(5, TimeUnit.SECONDS).shouldBeTrue() + userIdOnDiskWhenNotified.toList() shouldBeEqualTo listOf(restoredUserId) + deliveryMetricsTracked.get() shouldBeEqualTo 150 + } +} From e495b5f00b83cfe8cfe73ff91a8688a30e8e3ac3 Mon Sep 17 00:00:00 2001 From: Mahmoud Elmorabea Date: Mon, 19 Oct 2026 22:43:40 +0400 Subject: [PATCH 12/12] fix(core): drop a pending userId write when the SDK is reset A background init write still inside the Keystore when clearInstance resets the SDK could land afterwards and store the old user. Reset now invalidates the pending write and its callbacks and stops the store's writer. Co-Authored-By: Claude Opus 5.5 --- .../sdk/core/di/AndroidSDKComponent.kt | 2 + .../sdk/data/store/SecureUserStore.kt | 14 +++++++ .../sdk/data/store/SecureUserStoreTest.kt | 39 +++++++++++++++++++ 3 files changed, 55 insertions(+) diff --git a/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt b/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt index 52f67efea..48f4902d5 100644 --- a/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt +++ b/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt @@ -66,6 +66,8 @@ class AndroidSDKComponentImpl( } override fun reset() { + // A background userId write started before the reset must not land after it. + (getOrNull() as? SecureUserStoreImpl)?.cancelPendingWrites() super.reset() SDKComponent.activityLifecycleCallbacks.unregister(application) diff --git a/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt b/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt index 51dd41546..33cdb6e58 100644 --- a/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt +++ b/core/src/main/kotlin/io/customer/sdk/data/store/SecureUserStore.kt @@ -9,6 +9,7 @@ import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel import kotlinx.coroutines.launch /** @@ -146,6 +147,19 @@ internal class SecureUserStoreImpl( resetLocked { clear() } } + /** + * Drops the pending write and its callbacks and stops the writer, for when the SDK is reset and + * this instance is discarded. A write already inside the Keystore can't be interrupted, but it + * no longer lands, so it can't overwrite what a later instance stores. + */ + fun cancelPendingWrites() { + synchronized(lock) { + version++ + pendingWrite = null + } + writeScope.cancel() + } + private fun resetLocked(edit: SharedPreferences.Editor.() -> Unit) { version++ pendingWrite = null diff --git a/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt b/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt index 8d7ec0be8..fd8ee45c7 100644 --- a/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt +++ b/core/src/test/java/io/customer/sdk/data/store/SecureUserStoreTest.kt @@ -1,9 +1,12 @@ package io.customer.sdk.data.store import io.customer.commontest.core.RobolectricTest +import io.customer.sdk.core.di.SDKComponent +import io.customer.sdk.core.di.setupAndroidComponent import io.customer.sdk.core.util.Logger import io.mockk.every import io.mockk.mockk +import io.mockk.spyk import io.mockk.verify import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit @@ -232,4 +235,40 @@ class SecureUserStoreTest : RobolectricTest() { store.getUserId().shouldBeNull() verify(exactly = 1) { blockedCrypto.encrypt("user-42") } } + + @Test + fun cancelPendingWrites_whileEncryptionBlocked_expectWriteAndCallbackDropped() { + val encryptEntered = CountDownLatch(1) + val releaseEncrypt = CountDownLatch(1) + var blockEncrypt = false + val blockedCrypto = fakeCrypto { + if (blockEncrypt) { + encryptEntered.countDown() + releaseEncrypt.await() + } + } + val store = newStoreWithPendingWrites(crypto = blockedCrypto, dispatcher = Dispatchers.IO) + store.saveUserId("previous-user") + blockEncrypt = true + val published = CountDownLatch(1) + store.saveUserIdInBackground("user-42") { published.countDown() } + encryptEntered.await(5, TimeUnit.SECONDS).shouldBeTrue() + + store.cancelPendingWrites() + releaseEncrypt.countDown() + + // The Keystore call finishes, but the write it was for no longer lands. + published.await(1, TimeUnit.SECONDS).shouldBeFalse() + store.getUserId() shouldBeEqualTo "previous-user" + } + + @Test + fun sdkReset_expectPendingWritesOfTheGraphStoreCancelled() { + val store = spyk(newStoreWithPendingWrites()) + SDKComponent.setupAndroidComponent(contextMock).overrideDependency(store) + + SDKComponent.reset() + + verify(exactly = 1) { store.cancelPendingWrites() } + } }