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/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt b/core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt index 76700e321..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 @@ -61,9 +61,13 @@ 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() { + // 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/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) } 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..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 @@ -1,9 +1,16 @@ 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.cancel +import kotlinx.coroutines.launch /** * Encrypted, user-scoped preference store for sensitive data that must @@ -19,35 +26,146 @@ 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) + + /** [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 + * 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 + private val 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 { + // 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, 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 + 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() } + } + + /** + * 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 + 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/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 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..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,10 +1,24 @@ 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 +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.amshove.kluent.shouldNotBeNull import org.junit.Test import org.junit.runner.RunWith import org.robolectric.RobolectricTestRunner @@ -14,8 +28,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 +58,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 +80,195 @@ 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() + // Superseded before the writer ran, so it never touched the Keystore. + verify(exactly = 0) { crypto.encrypt(any()) } + } + + @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_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) + 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") } + } + + @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() } + } } 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/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 + } +} 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/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..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,12 @@ 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 +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull /** * Coordinates all location state management: persistence, restoration, @@ -38,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 @@ -59,7 +66,23 @@ 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, 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. Past the deadline a stalled Keystore couldn't decrypt it anyway. + awaitRestoreBlocking() val location = trackedLocation ?: return emptyMap() return mapOf( "location_latitude" to location.latitude, @@ -75,24 +98,64 @@ internal class LocationTracker( * guaranteeing no stale data is available for a subsequent identify(). */ override fun resetContext() { - trackedLocation = null - lastKnownLocation = null - locationPreferenceStore.clearCachedLocation() + synchronized(stateLock) { + 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() + } locationSyncFilter.clearSyncedData() logger.debug("Location state reset") } + /** 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], until its deadline. */ + suspend fun awaitRestore() { + val job = restoreJob ?: return + withTimeoutOrNull(remainingRestoreWaitMillis()) { job.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 + 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 * 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 +171,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 +202,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 +261,11 @@ internal class LocationTracker( ) ) } + + private companion object { + // 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/main/kotlin/io/customer/location/ModuleLocation.kt b/location/src/main/kotlin/io/customer/location/ModuleLocation.kt index f365ab542..880c98df6 100644 --- a/location/src/main/kotlin/io/customer/location/ModuleLocation.kt +++ b/location/src/main/kotlin/io/customer/location/ModuleLocation.kt @@ -16,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. @@ -111,22 +112,29 @@ 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 { if (!it.userId.isNullOrEmpty()) { - 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() + } } } } diff --git a/location/src/test/java/io/customer/location/LocationTrackerTest.kt b/location/src/test/java/io/customer/location/LocationTrackerTest.kt index 6a6411b94..579a23d45 100644 --- a/location/src/test/java/io/customer/location/LocationTrackerTest.kt +++ b/location/src/test/java/io/customer/location/LocationTrackerTest.kt @@ -11,9 +11,17 @@ 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 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 import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test @@ -211,6 +219,100 @@ 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) + } + + @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 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..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 @@ -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,11 +118,18 @@ 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 } + if (awaitingIdentity) { + 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() @@ -177,10 +189,19 @@ internal class PollingLifecycleManager( 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() + } 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 e0adfa2ed..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 @@ -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,65 @@ 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 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() @@ -199,7 +259,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() } } }