fix: keep Keystore work off the SDK init thread

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)
             }
         }
     }

core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt

@@ -61,9 +61,13 @@ class AndroidSDKComponentImpl(
     override val globalPreferenceStore: GlobalPreferenceStore
         get() = singleton<GlobalPreferenceStore> { GlobalPreferenceStoreImpl(applicationContext) }
     override val secureUserStore: SecureUserStore
-        get() = singleton<SecureUserStore> { SecureUserStoreImpl(applicationContext, SDKComponent.logger) }
+        get() = singleton<SecureUserStore> {
+            SecureUserStoreImpl(applicationContext, SDKComponent.logger, SDKComponent.dispatchersProvider.background)
+        }
 
     override fun reset() {
+        // A background userId write started before the reset must not land after it.
+        (getOrNull<SecureUserStore>() as? SecureUserStoreImpl)?.cancelPendingWrites()
         super.reset()
 
         SDKComponent.activityLifecycleCallbacks.unregister(application)

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)
 }

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"

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<Unit>()
+        val slowEvents = mutableListOf<Event>()
+        val otherEvents = mutableListOf<Event>()
+        val slowJob = eventBus.subscribe(Event.ScreenViewedEvent::class) { event ->
+            releaseSlowSubscriber.await()
+            slowEvents.add(event)
+        }
+        val otherJob = eventBus.subscribe<Event.ScreenViewedEvent> { 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