fix(core): harden ProviderRepository concurrency (#185)

mstepien · mstepien · commit a63d85e9d66e · 2026-04-15T15:28:30.000+02:00
harden ProviderRepository concurrency, lock inversions, and backpressure testing
Signed-off-by: Marcin Stepien &lt;marcin.stepien@fluxon.com&gt;
diff --git a/kotlin-sdk/api/android/kotlin-sdk.api b/kotlin-sdk/api/android/kotlin-sdk.api
@@ -39,7 +39,7 @@ public final class dev/openfeature/kotlin/sdk/DomainState {
 	public final fun getStatus ()Ldev/openfeature/kotlin/sdk/OpenFeatureStatus;
 	public final fun getStatusFlow ()Lkotlinx/coroutines/flow/Flow;
 	public final fun get_statusFlow ()Lkotlinx/coroutines/flow/MutableSharedFlow;
-	public final fun initializeListener (Lkotlinx/coroutines/CoroutineDispatcher;)V
+	public final fun initializeListener (Lkotlinx/coroutines/CoroutineDispatcher;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public final fun setContext (Ldev/openfeature/kotlin/sdk/EvaluationContext;)V
 	public final fun setSetEvaluationContextJob (Lkotlinx/coroutines/Job;)V
 	public final fun setSetProviderJob (Lkotlinx/coroutines/Job;)V
diff --git a/kotlin-sdk/api/jvm/kotlin-sdk.api b/kotlin-sdk/api/jvm/kotlin-sdk.api
@@ -39,7 +39,7 @@ public final class dev/openfeature/kotlin/sdk/DomainState {
 	public final fun getStatus ()Ldev/openfeature/kotlin/sdk/OpenFeatureStatus;
 	public final fun getStatusFlow ()Lkotlinx/coroutines/flow/Flow;
 	public final fun get_statusFlow ()Lkotlinx/coroutines/flow/MutableSharedFlow;
-	public final fun initializeListener (Lkotlinx/coroutines/CoroutineDispatcher;)V
+	public final fun initializeListener (Lkotlinx/coroutines/CoroutineDispatcher;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public final fun setContext (Ldev/openfeature/kotlin/sdk/EvaluationContext;)V
 	public final fun setSetEvaluationContextJob (Lkotlinx/coroutines/Job;)V
 	public final fun setSetProviderJob (Lkotlinx/coroutines/Job;)V
diff --git a/kotlin-sdk/src/commonMain/kotlin/dev/openfeature/kotlin/sdk/ProviderRepository.kt b/kotlin-sdk/src/commonMain/kotlin/dev/openfeature/kotlin/sdk/ProviderRepository.kt
@@ -2,23 +2,29 @@ package dev.openfeature.kotlin.sdk
 
 import dev.openfeature.kotlin.sdk.events.OpenFeatureProviderEvents
 import dev.openfeature.kotlin.sdk.events.toOpenFeatureStatusError
+import dev.openfeature.kotlin.sdk.exceptions.ErrorCode
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.distinctUntilChanged
-import kotlinx.coroutines.flow.firstOrNull
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.retryWhen
 import kotlinx.coroutines.flow.update
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
+import kotlin.concurrent.Volatile
 
 class DomainState {
     var setProviderJob: Job? = null
@@ -30,35 +36,64 @@ class DomainState {
     var context: EvaluationContext? = null
 
     val _statusFlow: MutableSharedFlow<OpenFeatureStatus> =
-        MutableSharedFlow<OpenFeatureStatus>(replay = 1, extraBufferCapacity = 5).apply {
+        MutableSharedFlow<OpenFeatureStatus>(
+            replay = 1,
+            extraBufferCapacity = 5,
+            onBufferOverflow = BufferOverflow.DROP_OLDEST
+        ).apply {
             tryEmit(OpenFeatureStatus.NotReady)
         }
 
     val statusFlow: Flow<OpenFeatureStatus> get() = _statusFlow.distinctUntilChanged()
 
     private var domainScope: CoroutineScope? = null
 
-    fun initializeListener(dispatcher: CoroutineDispatcher) {
-        if (domainScope != null) return
-        domainScope = CoroutineScope(SupervisorJob() + dispatcher).apply {
-            launch {
-                providersFlow.collectLatest { currentProvider ->
-                    currentProvider.observe().collect { providerEvent ->
-                        when (providerEvent) {
-                            is OpenFeatureProviderEvents.ProviderReady -> {
-                                emitStatus(OpenFeatureStatus.Ready)
-                            }
-                            is OpenFeatureProviderEvents.ProviderStale -> {
-                                emitStatus(OpenFeatureStatus.Stale)
-                            }
-                            is OpenFeatureProviderEvents.ProviderError -> {
-                                emitStatus(providerEvent.toOpenFeatureStatusError())
-                            }
-                            else -> { // All other states should not be emitted from here
-                            }
+    @Volatile
+    private var currentDispatcher: CoroutineDispatcher? = null
+
+    suspend fun initializeListener(dispatcher: CoroutineDispatcher) {
+        providerMutex.withLock {
+            currentDispatcher = dispatcher
+
+            if (domainScope != null) return@withLock
+            val scope = CoroutineScope(SupervisorJob() + dispatcher)
+            domainScope = scope
+
+            providersFlow
+                .flatMapLatest { currentProvider ->
+                    currentProvider.observe()
+                        .retryWhen { cause, _ ->
+                            emit(
+                                OpenFeatureProviderEvents.ProviderError(
+                                    OpenFeatureProviderEvents.EventDetails(
+                                        message = cause.message ?: "Provider observe() crashed",
+                                        errorCode = ErrorCode.GENERAL
+                                    )
+                                )
+                            )
+                            delay(3000L)
+                            true
+                        }
+                        .map { event -> currentProvider to event }
+                }
+                .onEach { (currentProvider, providerEvent) ->
+                    val activeDispatcher = currentDispatcher ?: dispatcher
+                    withContext(activeDispatcher) {
+                        if (providersFlow.value === currentProvider) {
+                            processProviderEvent(providerEvent)
                         }
                     }
                 }
+                .launchIn(scope)
+        }
+    }
+
+    private suspend fun processProviderEvent(event: OpenFeatureProviderEvents) {
+        when (event) {
+            is OpenFeatureProviderEvents.ProviderReady -> emitStatus(OpenFeatureStatus.Ready)
+            is OpenFeatureProviderEvents.ProviderStale -> emitStatus(OpenFeatureStatus.Stale)
+            is OpenFeatureProviderEvents.ProviderError -> emitStatus(event.toOpenFeatureStatusError())
+            else -> { // All other states should not be emitted from here
             }
         }
     }
@@ -72,11 +107,14 @@ class DomainState {
     suspend fun shutdown() {
         setProviderJob?.cancel(CancellationException("Provider set job was cancelled due to shutdown"))
         setEvaluationContextJob?.cancel(CancellationException("Set context job was cancelled due to shutdown"))
-        domainScope?.cancel(CancellationException("DomainScope was cancelled due to shutdown"))
-        domainScope = null
+        providerMutex.withLock {
+            domainScope?.cancel(CancellationException("DomainScope was cancelled due to shutdown"))
+            domainScope = null
+            currentDispatcher = null
+        }
         provider.shutdown()
         providersFlow.value = NoOpProvider()
-        _statusFlow.emit(OpenFeatureStatus.NotReady)
+        _statusFlow.tryEmit(OpenFeatureStatus.NotReady)
     }
 }
 
@@ -110,12 +148,16 @@ class ProviderRepository {
     }
 
     suspend fun clearAll() {
-        repositoryMutex.withLock {
+        // Evaluate and detach existing states safely
+        val allStatesToShutdown = repositoryMutex.withLock {
             val all = getAllStates()
-            all.forEach { state ->
-                state.shutdown()
-            }
             domainsFlow.value = emptyMap()
+            all
+        }
+
+        // Shutdown cleanly outside the repository lock to prevent lock-inversion deadlocks!
+        allStatesToShutdown.forEach { state ->
+            state.shutdown()
         }
     }
 }
diff --git a/kotlin-sdk/src/commonTest/kotlin/dev/openfeature/kotlin/sdk/DomainE2ETest.kt b/kotlin-sdk/src/commonTest/kotlin/dev/openfeature/kotlin/sdk/DomainE2ETest.kt
@@ -5,11 +5,11 @@ import dev.openfeature.kotlin.sdk.helpers.BrokenInitProvider
 import dev.openfeature.kotlin.sdk.helpers.DoSomethingProvider
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.currentTime
 import kotlinx.coroutines.test.runTest
-import kotlinx.coroutines.delay
 import kotlin.test.AfterTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
diff --git a/kotlin-sdk/src/commonTest/kotlin/dev/openfeature/kotlin/sdk/ProviderRepositoryTest.kt b/kotlin-sdk/src/commonTest/kotlin/dev/openfeature/kotlin/sdk/ProviderRepositoryTest.kt
@@ -3,16 +3,20 @@ package dev.openfeature.kotlin.sdk
 import dev.openfeature.kotlin.sdk.events.OpenFeatureProviderEvents
 import dev.openfeature.kotlin.sdk.exceptions.OpenFeatureError
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertNotEquals
 import kotlin.test.assertSame
+import kotlin.test.assertTrue
 
 @OptIn(ExperimentalCoroutinesApi::class)
 class ProviderRepositoryTest {
@@ -157,4 +161,146 @@ class ProviderRepositoryTest {
         // Verify the newly mapped listener successfully overrides state
         assertEquals(OpenFeatureStatus.Stale, state.getStatus())
     }
+
+    @Test
+    fun `DomainState should restart event listener if dispatcher changes`() = runTest {
+        val state = DomainState()
+
+        val dispatcher1 = StandardTestDispatcher(testScheduler)
+        val dispatcher2 = StandardTestDispatcher(testScheduler)
+
+        val providerEvents = MutableSharedFlow<OpenFeatureProviderEvents>()
+        class MockEventProvider : FeatureProvider by NoOpProvider() {
+            override fun observe() = providerEvents
+        }
+        val provider = MockEventProvider()
+        state.providersFlow.value = provider
+
+        // Step 1: Initialize with dispatcher1
+        state.initializeListener(dispatcher1)
+        testScheduler.advanceUntilIdle()
+
+        // Emit event to verify dispatcher1 listener is working
+        providerEvents.emit(OpenFeatureProviderEvents.ProviderReady())
+        testScheduler.advanceUntilIdle()
+        assertEquals(OpenFeatureStatus.Ready, state.getStatus())
+
+        // Step 2: Initialize with the EXACT SAME dispatcher, shouldn't disrupt anything
+        state.initializeListener(dispatcher1)
+        testScheduler.advanceUntilIdle()
+
+        // Step 3: Initialize with a DIFFERENT dispatcher (dispatcher2)
+        state.initializeListener(dispatcher2)
+        testScheduler.advanceUntilIdle()
+
+        // Fire another event, it should be processed by the NEW listener that was just hot-swapped
+        providerEvents.emit(OpenFeatureProviderEvents.ProviderStale())
+        testScheduler.advanceUntilIdle()
+        assertEquals(OpenFeatureStatus.Stale, state.getStatus())
+    }
+
+    @Test
+    fun `DomainState should automatically retry and emit error on provider unhandled exception`() = runTest {
+        val state = DomainState()
+        val errorMsg = "Simulated internal crash"
+        var observeCallCount = 0
+
+        class UnstableProvider : FeatureProvider by NoOpProvider() {
+            override fun observe(): kotlinx.coroutines.flow.Flow<OpenFeatureProviderEvents> = flow {
+                observeCallCount++
+                throw RuntimeException(errorMsg)
+            }
+        }
+
+        state.providersFlow.value = UnstableProvider()
+        val testDispatcher = StandardTestDispatcher(testScheduler)
+        state.initializeListener(testDispatcher)
+
+        testScheduler.advanceTimeBy(100L) // Process first crash without entering infinite loop
+
+        val status = state.getStatus()
+        assertTrue(status is OpenFeatureStatus.Error)
+        assertEquals(errorMsg, status.error.message)
+        assertEquals(1, observeCallCount)
+
+        // Then, advance by 3000ms. It should trigger retryWhen delay and retry
+        testScheduler.advanceTimeBy(3500L)
+        // Ensure it re-observed!
+        assertEquals(2, observeCallCount)
+
+        // Cancel scope explicitly to avoid hanging the `runTest` finalizer loop
+        state.shutdown()
+        testScheduler.advanceUntilIdle()
+    }
+
+    @Test
+    fun `DomainState should drop oldest status seamlessly and avoid suspending backpressure when blasted`() = runTest {
+        val state = DomainState()
+        val eventsFlow = MutableSharedFlow<OpenFeatureProviderEvents>()
+
+        class SpammyProvider : FeatureProvider by NoOpProvider() {
+            override fun observe() = eventsFlow
+        }
+
+        state.providersFlow.value = SpammyProvider()
+        val testDispatcher = StandardTestDispatcher(testScheduler)
+        state.initializeListener(testDispatcher)
+        testScheduler.advanceUntilIdle()
+
+        // Mimic a slow processor that purposefully does not collect from statusFlow.
+        val slowSubscriber = launch(StandardTestDispatcher(testScheduler)) {
+            state.statusFlow.collect {
+                delay(10000L) // Extremely slow
+            }
+        }
+        testScheduler.advanceUntilIdle()
+
+        // Blast 50 items simultaneously into eventsFlow -> emitStatus.
+        // If it was BufferOverflow.SUSPEND this wouldn't finish.
+        val job = launch(StandardTestDispatcher(testScheduler)) {
+            for (i in 1..50) {
+                eventsFlow.emit(
+                    if (i % 2 == 0) {
+                        OpenFeatureProviderEvents.ProviderReady()
+                    } else {
+                        OpenFeatureProviderEvents.ProviderStale()
+                    }
+                )
+            }
+        }
+        testScheduler.advanceUntilIdle()
+
+        // Assert the job actually finished and didn't hang
+        assertTrue(job.isCompleted)
+        slowSubscriber.cancel()
+
+        // Assert the state correctly processed them
+        val finalStatus = state.getStatus()
+        assertTrue(finalStatus is OpenFeatureStatus.Ready || finalStatus is OpenFeatureStatus.Stale)
+    }
+
+    @Test
+    fun `ProviderRepository clearAll should not deadlock against busy domain state shutdowns`() = runTest {
+        val repository = ProviderRepository()
+        val state = repository.getOrCreateState("deadlock-domain")
+
+        // Thread A: Holds providerMutex and tries to get repositoryMutex
+        val threadA = launch(StandardTestDispatcher(testScheduler)) {
+            state.providerMutex.withLock {
+                delay(50L) // Wait to ensure Thread B traps repositoryMutex
+                repository.getOrCreateState("new-domain") // Wants repositoryMutex
+            }
+        }
+
+        // Thread B: Runs clearAll
+        val threadB = launch(StandardTestDispatcher(testScheduler)) {
+            repository.clearAll() // Holds repositoryMutex initially, then wants providerMutex (via shutdown)
+        }
+
+        testScheduler.advanceUntilIdle()
+
+        // If the vulnerability exists, both threads will be permanently blocked!
+        assertTrue(threadA.isCompleted)
+        assertTrue(threadB.isCompleted)
+    }
 }
