Fix race condition in plugin resolution during disconnect (#6269)

* Update `DependencyResolverTest` to verify error handling when dependency resolution races with disconnection.

* Prevent race conditions during disconnects in `ChatClient`.

Author: andremion

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/ChatClient.kt

@@ -354,6 +354,7 @@ internal constructor(
      *
      * @see [Plugin]
      */
+    @Volatile
     @InternalStreamChatApi
     public var plugins: List<Plugin> = emptyList()
 
@@ -400,12 +401,16 @@ internal constructor(
     @Suppress("ThrowsCount")
     internal inline fun <reified P : DependencyResolver, reified T : Any> resolvePluginDependency(): T {
         StreamLog.v(TAG) { "[resolvePluginDependency] P: ${P::class.simpleName}, T: ${T::class.simpleName}" }
+        // Snapshot plugins BEFORE checking initializationState to avoid a race with disconnect().
+        // disconnect() sets initializationState to NOT_INITIALIZED before clearing plugins,
+        // so if we snapshot plugins first and then see COMPLETE, the snapshot is guaranteed valid.
+        val currentPlugins = plugins
         val initState = awaitInitializationState(RESOLVE_DEPENDENCY_TIMEOUT)
         if (initState != InitializationState.COMPLETE) {
             StreamLog.e(TAG) { "[resolvePluginDependency] failed (initializationState is not COMPLETE): $initState " }
             throw IllegalStateException("ChatClient::connectUser() must be called before resolving any dependency")
         }
-        val resolver = plugins.find { plugin ->
+        val resolver = currentPlugins.find { plugin ->
             plugin is P
         } ?: throw IllegalStateException(
             "Plugin '${P::class.qualifiedName}' was not found. Did you init it within ChatClient?",
@@ -1570,9 +1575,9 @@ internal constructor(
 
         notifications.onLogout()
         // Set initializationState to NOT_INITIALIZED BEFORE clearing plugins to prevent race condition.
-        // This ensures the StatePlugin extension methods don't access the plugin during disconnect.
+        // resolvePluginDependency() snapshots plugins before checking state, so if it sees COMPLETE
+        // here, the snapshot is guaranteed to still contain the plugins.
         mutableClientState.setInitializationState(InitializationState.NOT_INITIALIZED)
-
         plugins.forEach { it.onUserDisconnected() }
         plugins = emptyList()
         userStateService.onLogout()

stream-chat-android-client/src/test/java/io/getstream/chat/android/client/DependencyResolverTest.kt

@@ -30,7 +30,10 @@ import io.getstream.chat.android.models.NoOpMessageTransformer
 import io.getstream.chat.android.models.NoOpUserTransformer
 import io.getstream.chat.android.models.User
 import io.getstream.chat.android.test.TestCoroutineExtension
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.flow.FlowCollector
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.test.TestResult
 import kotlinx.coroutines.test.runTest
 import org.amshove.kluent.invoking
@@ -44,6 +47,7 @@ import org.junit.jupiter.params.provider.Arguments
 import org.junit.jupiter.params.provider.MethodSource
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.whenever
+import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.reflect.KClass
 
 public class DependencyResolverTest {
@@ -129,6 +133,22 @@ public class DependencyResolverTest {
         fResult `should be` expectedDependency
     }
 
+    @Test
+    public fun `Should resolve dependency when plugins are cleared during resolution`(): TestResult = runTest {
+        val expectedDependency = SomeDependency()
+        val fixture = Fixture()
+            .with(PluginDependency(mapOf(SomeDependency::class to expectedDependency)))
+
+        val client = fixture.get()
+
+        val racingFlow = DisconnectSimulatingStateFlow(client)
+        whenever(fixture.mutableClientState.initializationState).thenReturn(racingFlow)
+
+        val result = client.resolveDependency<PluginDependency, SomeDependency>()
+
+        result `should be` expectedDependency
+    }
+
     public companion object {
 
         @JvmField
@@ -219,4 +239,28 @@ public class DependencyResolverTest {
     }
 
     private class SomeDependency
+
+    private class DisconnectSimulatingStateFlow(
+        private val client: ChatClient,
+    ) : StateFlow<InitializationState> {
+
+        private val disconnected = AtomicBoolean(false)
+
+        override val value: InitializationState
+            get() {
+                if (disconnected.compareAndSet(false, true)) {
+                    client.plugins = emptyList()
+                    return InitializationState.COMPLETE
+                }
+                return InitializationState.NOT_INITIALIZED
+            }
+
+        override val replayCache: List<InitializationState>
+            get() = listOf(InitializationState.COMPLETE)
+
+        override suspend fun collect(collector: FlowCollector<InitializationState>): Nothing {
+            collector.emit(InitializationState.COMPLETE)
+            awaitCancellation()
+        }
+    }
 }