chore: fixing fdv2 race condition (#343)

The root of the bug is that OFF is reported in stop(), but the worker
thread may still be running and making its own calls.

The fix is to update stop() to behave exactly like start, which is that
the request is made only once and the callback is queued into a pending
list. The worker thread is now the only thread that makes data and
status callbacks, which eliminates the race risk.


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Touches FDv2 data source lifecycle/concurrency and changes when OFF is
reported; bugs here could affect connection shutdown behavior and
callback ordering under contention.
> 
> **Overview**
> Fixes a race where `stop()` could report `DataSourceState.OFF` while
the FDv2 worker thread was still emitting later status updates.
> 
> `FDv2DataSource.stop()` is reworked to be idempotent and
callback-queued (like `start()`), with a shared `shutdownCause` used to
coordinate shutdown so **only the worker thread** emits the final `OFF`
status (and includes an error only for unintentional shutdowns). Tests
update mock synchronizers to unblock pending `next()` calls on `close()`
and add coverage ensuring **no status events occur after `OFF`**.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
0228cef. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: tanderson-ld

launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/FDv2DataSource.java

@@ -6,7 +6,6 @@
 import com.launchdarkly.logging.LDLogger;
 import com.launchdarkly.sdk.LDContext;
 import com.launchdarkly.sdk.android.subsystems.Callback;
-import com.launchdarkly.sdk.android.DataModel;
 import com.launchdarkly.sdk.fdv2.ChangeSet;
 import com.launchdarkly.sdk.fdv2.SourceResultType;
 import com.launchdarkly.sdk.android.subsystems.DataSourceState;
@@ -52,15 +51,17 @@ public interface DataSourceFactory<T> {
     private final long recoveryTimeoutSeconds;
     private final ScheduledExecutorService sharedExecutor;
 
-    private final AtomicBoolean started = new AtomicBoolean(false);
+    private final AtomicBoolean startCalled = new AtomicBoolean(false);
     private final AtomicBoolean startCompleted = new AtomicBoolean(false);
-    private final AtomicBoolean stopped = new AtomicBoolean(false);
-    /** Result of the first start (null = not yet completed). Used so second start() gets the same result. */
     private volatile Boolean startResult = null;
     private volatile Throwable startError = null;
-    private final Object startResultLock = new Object();
     private final List<Callback<Boolean>> pendingStartCallbacks = new ArrayList<>();
+    private final AtomicBoolean stopCalled = new AtomicBoolean(false);
+    private final AtomicBoolean stopCompleted = new AtomicBoolean(false);
+    private final List<Callback<Void>> pendingStopCallbacks = new ArrayList<>();
 
+    // This future is set by either the worker thread terminating or stop() being called.
+    private final LDAwaitFuture<Throwable> shutdownCause = new LDAwaitFuture<>();
     /**
      * Convenience constructor using default fallback and recovery timeouts.
      * See {@link #FDv2DataSource(LDContext, List, List, DataSourceFactory, DataSourceUpdateSinkV2,
@@ -131,7 +132,7 @@ public interface DataSourceFactory<T> {
 
     @Override
     public void start(@NonNull Callback<Boolean> resultCallback) {
-        synchronized (startResultLock) {
+        synchronized (pendingStartCallbacks) {
             // Late caller: the first start already finished, so replay its result immediately.
             if (startResult != null) {
                 if (startResult) {
@@ -147,7 +148,7 @@ public void start(@NonNull Callback<Boolean> resultCallback) {
         }
 
         // Only the first caller spawns the background thread; subsequent callers just queued above.
-        if (!started.compareAndSet(false, true)) {
+        if (!startCalled.compareAndSet(false, true)) {
             return;
         }
         // Do not reset stopped here: it is initialized false and start() runs once. Resetting would
@@ -160,7 +161,7 @@ public void start(@NonNull Callback<Boolean> resultCallback) {
                     logger.info("No initializers or synchronizers; data source will not connect.");
                     dataSourceUpdateSink.setStatus(DataSourceState.VALID, null);
                     tryCompleteStart(true, null);
-                    return;
+                    return; // this will go to the finally block and block until stop sets shutdownCause
                 }
 
                 if (sourceManager.hasInitializers()) {
@@ -169,34 +170,41 @@ public void start(@NonNull Callback<Boolean> resultCallback) {
 
                 if (!sourceManager.hasAvailableSynchronizers()) {
                     if (!startCompleted.get()) {
-                        LDFailure failure = maybeReportUnexpectedExhaustion("All initializers exhausted and there are no available synchronizers.");
-                        tryCompleteStart(false, failure);
+                        // try to claim this is the cause of the shutdown, but it might have already been set by an intentional stop().
+                        shutdownCause.set(new LDFailure("All initializers exhausted and there are no available synchronizers.", LDFailure.FailureType.UNKNOWN_ERROR));
                     }
                     return;
                 }
 
                 runSynchronizers(context, dataSourceUpdateSink);
-                LDFailure failure = maybeReportUnexpectedExhaustion("All data source acquisition methods have been exhausted.");
-                tryCompleteStart(false, failure);
+                // try to claim this is the cause of the shutdown, but it might have already been set by an intentional stop().
+                shutdownCause.set(new LDFailure("All data source acquisition methods have been exhausted.", LDFailure.FailureType.UNKNOWN_ERROR));
             } catch (Throwable t) {
                 logger.warn("FDv2DataSource error: {}", t.toString());
-                tryCompleteStart(false, t);
+                shutdownCause.set(t);
+            } finally {
+
+                // Here we grab the cause of shutdown to report with the OFF status. This is done to ensure that
+                // all status callbacks are handled by the worker thread. This future may have been set
+                // by this thread itself, but it may have also been set by the stop() call via another thread.
+                //
+                // This intentionally blocks on this future in certain configurations and that may seem 
+                // inefficient, but it simplifies the implementation. Such cases are rare in practice.
+                Throwable cause;
+                try {
+                    cause = shutdownCause.get();
+                } catch (Exception e) {
+                    cause = e;
+                }
+
+                boolean intentional = cause instanceof CancellationException;
+                dataSourceUpdateSink.setStatus(DataSourceState.OFF, intentional ? null : cause);
+                tryCompleteStart(false, cause); // must always provide cause with false success
+                tryCompleteStop();
             }
         });
     }
 
-    /**
-     * If not stopped, reports OFF with the given message (e.g. exhaustion).
-     * Returns the failure so callers can forward it to {@link #tryCompleteStart}.
-     */
-    private LDFailure maybeReportUnexpectedExhaustion(String message) {
-        LDFailure failure = new LDFailure(message, LDFailure.FailureType.UNKNOWN_ERROR);
-        if (!stopped.get()) {
-            dataSourceUpdateSink.setStatus(DataSourceState.OFF, failure);
-        }
-        return failure;
-    }
-
     /**
      * Records the start result and notifies all callbacks (first and any subsequent start() callers).
      * No-op if start has already completed. If success is false, error is not null.
@@ -208,7 +216,7 @@ private void tryCompleteStart(boolean success, Throwable error) {
             return;
         }
         List<Callback<Boolean>> toNotify;
-        synchronized (startResultLock) {
+        synchronized (pendingStartCallbacks) {
             startResult = success;
             startError = error;
             toNotify = new ArrayList<>(pendingStartCallbacks);
@@ -225,11 +233,50 @@ private void tryCompleteStart(boolean success, Throwable error) {
 
     @Override
     public void stop(@NonNull Callback<Void> completionCallback) {
-        stopped.set(true);
-        sourceManager.close();
-        // Caller owns sharedExecutor; we do not shut it down.
-        dataSourceUpdateSink.setStatus(DataSourceState.OFF, null);
-        completionCallback.onSuccess(null);
+        synchronized (pendingStopCallbacks) {
+            if (stopCompleted.get()) {
+                // we have already stopped
+                completionCallback.onSuccess(null);
+                return;
+            }
+
+            // stopping is still in progress; queue the callback to be fired by tryCompleteStop.
+            pendingStopCallbacks.add(completionCallback);
+        }
+
+        // Only the first call to stop does anything
+        if (!stopCalled.compareAndSet(false, true)) {
+            return;
+        }
+
+        shutdownCause.set(new CancellationException("Data source was stopped intentionally."));
+        sourceManager.close(); // unblocks worker thread so it can shutdown
+
+        // If the data source had never started, we need to complete the stop here
+        if (!startCalled.get()) {
+            tryCompleteStop();
+        }
+    }
+
+    /**
+     * Notifies all stop callbacks (if there are any).
+     * No-op if stop has already completed.
+     */
+    private void tryCompleteStop() {
+        // Idempotent: only the first call wins.
+        if (!stopCompleted.compareAndSet(false, true)) {
+            return;
+        }
+
+        List<Callback<Void>> toNotify;
+        synchronized (pendingStopCallbacks) {
+            toNotify = new ArrayList<>(pendingStopCallbacks);
+            pendingStopCallbacks.clear();
+        }
+
+        for (Callback<Void> c : toNotify) {
+            c.onSuccess(null);
+        }
     }
 
     @Override
@@ -246,9 +293,6 @@ private void runInitializers(
         boolean anyDataReceived = false;
         Initializer initializer = sourceManager.getNextInitializerAndSetActive();
         while (initializer != null) {
-            if (stopped.get()) {
-                return;
-            }
             try {
                 FDv2SourceResult result = initializer.run().get();
 
@@ -345,9 +389,6 @@ private void runSynchronizers(
         try {
             Synchronizer synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive();
             while (synchronizer != null) {
-                if (stopped.get()) {
-                    return;
-                }
                 int synchronizerCount = sourceManager.getAvailableSynchronizerCount();
                 boolean isPrime = sourceManager.isPrimeSynchronizer();
                 try {

launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/FDv2DataSourceTest.java

@@ -196,6 +196,7 @@ private static class MockSynchronizer implements Synchronizer {
         private final LDAwaitFuture<FDv2SourceResult> firstResult;
         private volatile boolean closed = false;
         private volatile boolean resultReturned = false;
+        private volatile LDAwaitFuture<FDv2SourceResult> pendingFuture;
 
         MockSynchronizer(FDv2SourceResult result) {
             this.firstResult = new LDAwaitFuture<>();
@@ -218,12 +219,17 @@ public LDAwaitFuture<FDv2SourceResult> next() {
                 resultReturned = true;
                 return firstResult;
             }
-            return new LDAwaitFuture<>(); // never completes; simulates waiting for next event
+            pendingFuture = new LDAwaitFuture<>();
+            return pendingFuture;
         }
 
         @Override
         public void close() {
             closed = true;
+            LDAwaitFuture<FDv2SourceResult> f = pendingFuture;
+            if (f != null) {
+                f.set(FDv2SourceResult.status(FDv2SourceResult.Status.shutdown(), false));
+            }
         }
     }
 
@@ -258,6 +264,11 @@ void addResult(FDv2SourceResult result) {
         @Override
         public LDAwaitFuture<FDv2SourceResult> next() {
             synchronized (lock) {
+                if (closed) {
+                    LDAwaitFuture<FDv2SourceResult> f = new LDAwaitFuture<>();
+                    f.set(FDv2SourceResult.status(FDv2SourceResult.Status.shutdown(), false));
+                    return f;
+                }
                 if (!queue.isEmpty()) {
                     LDAwaitFuture<FDv2SourceResult> f = new LDAwaitFuture<>();
                     f.set(queue.poll());
@@ -823,6 +834,66 @@ public void stopInterruptsConditionWaiting() throws Exception {
         stopDataSource(dataSource); // should not hang while condition is waiting for fallback timeout
     }
 
+    @Test
+    public void noStatusEventsEmittedAfterOff() throws Exception {
+        MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink();
+        CountDownLatch workerBlocked = new CountDownLatch(1);
+        LDAwaitFuture<FDv2SourceResult> controlledFuture = new LDAwaitFuture<>();
+
+        Synchronizer sync = new Synchronizer() {
+            private final AtomicBoolean firstReturned = new AtomicBoolean(false);
+
+            @Override
+            public LDAwaitFuture<FDv2SourceResult> next() {
+                if (!firstReturned.getAndSet(true)) {
+                    LDAwaitFuture<FDv2SourceResult> f = new LDAwaitFuture<>();
+                    f.set(FDv2SourceResult.changeSet(makeChangeSet(false), false));
+                    return f;
+                }
+                workerBlocked.countDown();
+                return controlledFuture;
+            }
+
+            @Override
+            public void close() {
+                // Intentionally does NOT complete controlledFuture.
+                // The test triggers it manually to control timing.
+            }
+        };
+
+        FDv2DataSource dataSource = buildDataSource(sink,
+                Collections.emptyList(),
+                Collections.singletonList(() -> sync));
+
+        AwaitableCallback<Boolean> startCallback = startDataSource(dataSource);
+        assertTrue(startCallback.await(AWAIT_TIMEOUT_SECONDS * 1000));
+        assertTrue(workerBlocked.await(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS));
+
+        // Stop the data source. On old code that had bug, this reports OFF synchronously.
+        AwaitableCallback<Void> stopCb = new AwaitableCallback<>();
+        dataSource.stop(stopCb);
+
+        // Simulate a delayed error arriving after stop (e.g. network close exception).
+        // Old code with the bug: OFF was already reported by stop(), so the INTERRUPTED comes after OFF.
+        // New code with the fix: the worker handles INTERRUPTED first, then reports OFF.
+        controlledFuture.setException(new RuntimeException("connection closed"));
+
+        stopCb.await(AWAIT_TIMEOUT_SECONDS * 1000);
+        Thread.sleep(200);
+
+        List<MockComponents.MockDataSourceUpdateSink.StatusEvent> events = sink.statusEvents;
+        int offIndex = -1;
+        for (int i = 0; i < events.size(); i++) {
+            if (events.get(i).state == DataSourceState.OFF) {
+                offIndex = i;
+                break;
+            }
+        }
+        assertTrue("Expected OFF status to be reported", offIndex >= 0);
+        assertEquals("OFF should be the last status event, but found events after it",
+                events.size() - 1, offIndex);
+    }
+
     // ============================================================================
     // Multiple Start Calls / Concurrency
     // ============================================================================