use a reentrant lock

Author: typotter

features/dd-sdk-android-flags/src/main/kotlin/com/datadog/android/flags/internal/FlagsStateManager.kt

@@ -13,13 +13,15 @@ import com.datadog.android.flags.StateObservable
 import com.datadog.android.flags.model.FlagsClientState
 import com.datadog.android.internal.utils.DDCoreSubscription
 import java.util.concurrent.ExecutorService
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
 
 /**
  * Manages state transitions and notifications for a [com.datadog.android.flags.FlagsClient].
  *
  * This class handles state change notifications to registered listeners. All notification
  * methods are thread-safe and guarantee ordered delivery to listeners by using a
- * single-threaded executor service.
+ * single-threaded executor service and a fair ReentrantLock.
  *
  * The current state is stored and emitted to new listeners immediately upon registration,
  * ensuring every listener receives the current state.
@@ -33,19 +35,26 @@ internal class FlagsStateManager(
     private val executorService: ExecutorService,
     private val internalLogger: InternalLogger
 ) : StateObservable {
+    /**
+     * Fair lock to ensure FIFO ordering of state mutations and listener operations.
+     * The fair parameter ensures that threads acquire the lock in the order they requested it.
+     */
+    private val stateLock = ReentrantLock(true)
+
     /**
      * The current state of the client.
-     * Thread-safe: uses volatile for visibility across threads.
+     * Thread-safe: access is protected by stateLock.
      */
-    @Volatile
     private var currentState: FlagsClientState = FlagsClientState.NotReady
 
     /**
      * Returns the current state synchronously.
      *
      * @return The current [FlagsClientState].
      */
-    override fun getCurrentState(): FlagsClientState = currentState
+    override fun getCurrentState(): FlagsClientState = stateLock.withLock {
+        currentState
+    }
 
     /**
      * Updates the state and notifies all listeners.
@@ -56,8 +65,8 @@ internal class FlagsStateManager(
      * @param newState The new state to transition to.
      */
     internal fun updateState(newState: FlagsClientState) {
-        // lock the currentState until listener notifications are queued to prevent new listeners during state change.
-        synchronized(currentState) {
+        // Lock to ensure atomic state update and listener notification queueing
+        stateLock.withLock {
             currentState = newState
             subscription.notifyListeners {
                 executorService.executeSafe(
@@ -71,7 +80,7 @@ internal class FlagsStateManager(
     }
 
     override fun addListener(listener: FlagsStateListener) {
-        synchronized(currentState) {
+        stateLock.withLock {
             subscription.addListener(listener)
 
             val stateToEmit = currentState
@@ -85,7 +94,7 @@ internal class FlagsStateManager(
     }
 
     override fun removeListener(listener: FlagsStateListener) {
-        synchronized(currentState) {
+        stateLock.withLock {
             subscription.removeListener(listener)
         }
     }

features/dd-sdk-android-flags/src/test/kotlin/com/datadog/android/flags/internal/FlagsStateManagerTest.kt

@@ -169,6 +169,7 @@ internal class FlagsStateManagerTest {
         )
 
         val executionOrder = mutableListOf<String>()
+        val executionOrderLock = Any()
 
         // Creates a listener that adds start/end markers to the execution order and calls the additional block.
         fun createListener(
@@ -178,7 +179,7 @@ internal class FlagsStateManagerTest {
         ): FlagsStateListener = object : FlagsStateListener {
             override fun onStateChanged(newState: FlagsClientState) {
                 if (newState is FlagsClientState.Ready) {
-                    synchronized(executionOrder) {
+                    synchronized(executionOrderLock) {
                         executionOrder.add(name)
                         additionalBlock()
                         executionOrder.add("$name ended")
@@ -202,7 +203,7 @@ internal class FlagsStateManagerTest {
         managerWithRealExecutor.addListener(listener4)
 
         // When
-        synchronized(executionOrder) {
+        synchronized(executionOrderLock) {
             managerWithRealExecutor.updateState(FlagsClientState.Ready)
             executionOrder.add("updateState")
         }
@@ -215,7 +216,7 @@ internal class FlagsStateManagerTest {
         realExecutorService.awaitTermination(2, TimeUnit.SECONDS)
 
         // Then - all listeners should have been called in order, despite listener3 throwing
-        synchronized(executionOrder) {
+        synchronized(executionOrderLock) {
             assertThat(executionOrder).containsExactly(
                 "updateState",
                 "listener1",
@@ -229,6 +230,43 @@ internal class FlagsStateManagerTest {
         }
     }
 
+    @Test
+    fun `M maintain FIFO ordering W concurrent operations on fair lock`() {
+        // Given
+        realExecutorService = Executors.newSingleThreadExecutor()
+        val managerWithRealExecutor = FlagsStateManager(
+            DDCoreSubscription.create(),
+            realExecutorService,
+            mockInternalLogger
+        )
+
+        val notificationCount = java.util.concurrent.atomic.AtomicInteger(0)
+        val listeners = mutableListOf<FlagsStateListener>()
+
+        // When
+        repeat(10) {
+            val listener = object : FlagsStateListener {
+                override fun onStateChanged(newState: FlagsClientState) {
+                    if (newState is FlagsClientState.Ready) {
+                        notificationCount.incrementAndGet()
+                    }
+                }
+            }
+            listeners.add(listener)
+            managerWithRealExecutor.addListener(listener)
+        }
+
+        // Trigger state change
+        managerWithRealExecutor.updateState(FlagsClientState.Ready)
+
+        realExecutorService.shutdown()
+        realExecutorService.awaitTermination(2, TimeUnit.SECONDS)
+
+        // Then
+        assertThat(notificationCount.get()).isEqualTo(10)
+        assertThat(managerWithRealExecutor.getCurrentState()).isEqualTo(FlagsClientState.Ready)
+    }
+
     // endregion
 
     companion object {