chore: Add support for FDv2 fallback and recovery. (#113)

Adds support for FDv2 fallback and recovery.

This doesn't add support for fallback to FDv1.
It also does not finish data source status support.

It does some refactoring to limit the complexity of the FDv2DataSource
by moving some implementation details regarding synchronizer state
handling and conditions out of its implementation.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Implements FDv2 fallback/recovery and refactors synchronizer
orchestration.
> 
> - Adds `FDv2DataSourceConditions` with timed `FallbackCondition` (on
prolonged `INTERRUPTED`) and `RecoveryCondition` (time-based), and wires
them into `FDv2DataSource` run loop
> - Extracts synchronizer lifecycle into `SynchronizerStateManager` and
`SynchronizerFactoryWithState` for active-source tracking, rotation, and
blocking on terminal errors
> - `FDv2DataSource` now supports configurable timeouts (defaults:
fallback 2m, recovery 5m), applies thread priority, logs via injected
`LDLogger`, and returns FDv1 fallback signal (execution still TODO)
> - `FDv2DataSystem` constructs `FDv2DataSource` with `threadPriority`,
`DataSource` sub-logger, and shared executor; logging refined for
polling/streaming components via new names in `Loggers`
> - Extensive unit tests added for conditions, data source behavior
(fallback/recovery/closure), and synchronizer state manager
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
e6b032e. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: kinyoklion

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSource.java

@@ -0,0 +1,181 @@
+package com.launchdarkly.sdk.server;
+
+import com.launchdarkly.sdk.server.datasources.FDv2SourceResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Container class for FDv2 data source conditions and related types.
+ * <p>
+ * This class is non-constructable and serves only as a namespace for condition-related types.
+ * Package-private for internal use and testing.
+ */
+class FDv2DataSourceConditions {
+    /**
+     * Private constructor to prevent instantiation.
+     */
+    private FDv2DataSourceConditions() {
+    }
+
+    /**
+     * Package-private for testing.
+     */
+    interface Condition extends Closeable {
+        enum ConditionType {
+            FALLBACK,
+            RECOVERY,
+        }
+
+        CompletableFuture<Condition> execute();
+
+        void inform(FDv2SourceResult sourceResult);
+
+        void close();
+
+        ConditionType getType();
+    }
+
+    interface ConditionFactory {
+        Condition build();
+
+        Condition.ConditionType getType();
+    }
+
+    static abstract class TimedCondition implements Condition {
+        protected final CompletableFuture<Condition> resultFuture = new CompletableFuture<>();
+
+        protected final ScheduledExecutorService sharedExecutor;
+
+        /**
+         * Future for the timeout task, if any. Will be null when no timeout is active.
+         */
+        protected ScheduledFuture<Void> timerFuture;
+
+        /**
+         * Timeout duration for the fallback operation.
+         */
+        protected final long timeoutSeconds;
+
+        public TimedCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
+            this.sharedExecutor = sharedExecutor;
+            this.timeoutSeconds = timeoutSeconds;
+        }
+
+        @Override
+        public CompletableFuture<Condition> execute() {
+            return resultFuture;
+        }
+
+        @Override
+        public void close() {
+            if (timerFuture != null) {
+                timerFuture.cancel(false);
+                timerFuture = null;
+            }
+        }
+
+        static abstract class Factory implements ConditionFactory {
+            protected final ScheduledExecutorService sharedExecutor;
+            protected final long timeoutSeconds;
+
+            public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
+                this.sharedExecutor = sharedExecutor;
+                this.timeoutSeconds = timeout;
+            }
+        }
+    }
+
+    /**
+     * This condition is used to determine if a fallback should be performed. It is informed of each data source result
+     * via {@link #inform(FDv2SourceResult)}. Based on the results, it updates its internal state. When the fallback
+     * condition is met, then the {@link java.util.concurrent.Future} returned by {@link #execute()} will complete.
+     * <p>
+     * This is package-private, instead of private, for ease of testing.
+     */
+    static class FallbackCondition extends TimedCondition {
+        static class Factory extends TimedCondition.Factory {
+            public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
+                super(sharedExecutor, timeout);
+            }
+
+            @Override
+            public Condition build() {
+                return new FallbackCondition(sharedExecutor, timeoutSeconds);
+            }
+
+            @Override
+            public ConditionType getType() {
+                return ConditionType.FALLBACK;
+            }
+        }
+
+        public FallbackCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
+            super(sharedExecutor, timeoutSeconds);
+        }
+
+        @Override
+        public void inform(FDv2SourceResult sourceResult) {
+            if (sourceResult.getResultType() == FDv2SourceResult.ResultType.CHANGE_SET) {
+                if (timerFuture != null) {
+                    timerFuture.cancel(false);
+                    timerFuture = null;
+                }
+            }
+            if (sourceResult.getResultType() == FDv2SourceResult.ResultType.STATUS && sourceResult.getStatus().getState() == FDv2SourceResult.State.INTERRUPTED) {
+                if (timerFuture == null) {
+                    timerFuture = sharedExecutor.schedule(() -> {
+                        resultFuture.complete(this);
+                        return null;
+                    }, timeoutSeconds, TimeUnit.SECONDS);
+                }
+            }
+        }
+
+        @Override
+        public ConditionType getType() {
+            return ConditionType.FALLBACK;
+        }
+    }
+
+    static class RecoveryCondition extends TimedCondition {
+
+        static class Factory extends TimedCondition.Factory {
+            public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
+                super(sharedExecutor, timeout);
+            }
+
+            @Override
+            public Condition build() {
+                return new RecoveryCondition(sharedExecutor, timeoutSeconds);
+            }
+
+            @Override
+            public ConditionType getType() {
+                return ConditionType.RECOVERY;
+            }
+        }
+
+        public RecoveryCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
+            super(sharedExecutor, timeoutSeconds);
+            timerFuture = sharedExecutor.schedule(() -> {
+                resultFuture.complete(this);
+                return null;
+            }, timeoutSeconds, TimeUnit.SECONDS);
+        }
+
+        @Override
+        public void inform(FDv2SourceResult sourceResult) {
+            // Time-based recovery.
+        }
+
+        @Override
+        public ConditionType getType() {
+            return ConditionType.RECOVERY;
+        }
+    }
+}

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSourceConditions.java

@@ -157,7 +157,10 @@ static FDv2DataSystem create(
     DataSource dataSource = new FDv2DataSource(
       initializerFactories,
       synchronizerFactories,
-      dataSourceUpdates
+      dataSourceUpdates,
+      config.threadPriority,
+      clientContext.getBaseLogger().subLogger(Loggers.DATA_SOURCE_LOGGER_NAME),
+      clientContext.sharedExecutor
     );
     DataSourceStatusProvider dataSourceStatusProvider = new DataSourceStatusProviderImpl(
       dataSourceStatusBroadcaster,

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSystem.java

@@ -24,4 +24,7 @@ private Loggers() {}
   static final String EVALUATION_LOGGER_NAME = "Evaluation";
   static final String EVENTS_LOGGER_NAME = "Events";
   static final String HOOKS_LOGGER_NAME = "Hooks";
+  static final String STREAMING_SYNCHRONIZER = "StreamingSynchronizer";
+  static final String POLLING_SYNCHRONIZER = "PollingSynchronizer";
+  static final String POLLING_INITIALIZER = "PollingInitializer";
 }

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/Loggers.java

@@ -12,7 +12,7 @@ class PollingInitializerImpl extends PollingBase implements Initializer {
     private final SelectorSource selectorSource;
 
     public PollingInitializerImpl(FDv2Requestor requestor, LDLogger logger, SelectorSource selectorSource) {
-        super(requestor, logger);
+        super(requestor, logger.subLogger(Loggers.POLLING_INITIALIZER));
         this.selectorSource = selectorSource;
     }
 

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingInitializerImpl.java

@@ -23,7 +23,7 @@ public PollingSynchronizerImpl(
             ScheduledExecutorService sharedExecutor,
             Duration pollInterval
     ) {
-        super(requestor, logger);
+        super(requestor, logger.subLogger(Loggers.POLLING_SYNCHRONIZER));
         this.selectorSource = selectorSource;
 
         synchronized (this) {

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingSynchronizerImpl.java

@@ -71,7 +71,7 @@ public StreamingSynchronizerImpl(
     ) {
         this.httpProperties = httpProperties;
         this.selectorSource = selectorSource;
-        this.logger = logger;
+        this.logger = logger.subLogger(Loggers.STREAMING_SYNCHRONIZER);
         this.payloadFilter = payloadFilter;
         this.streamUri = HttpHelpers.concatenateUriPath(baseUri, requestPath);
         this.initialReconnectDelay = initialReconnectDelaySeconds;

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java

@@ -0,0 +1,38 @@
+package com.launchdarkly.sdk.server;
+
+import com.launchdarkly.sdk.server.datasources.Synchronizer;
+
+class SynchronizerFactoryWithState {
+    public enum State {
+        /**
+         * This synchronizer is available to use.
+         */
+        Available,
+
+        /**
+         * This synchronizer is no longer available to use.
+         */
+        Blocked
+    }
+
+    private final FDv2DataSource.DataSourceFactory<Synchronizer> factory;
+
+    private State state = State.Available;
+
+
+    public SynchronizerFactoryWithState(FDv2DataSource.DataSourceFactory<Synchronizer> factory) {
+        this.factory = factory;
+    }
+
+    public State getState() {
+        return state;
+    }
+
+    public void block() {
+        state = State.Blocked;
+    }
+
+    public Synchronizer build() {
+        return factory.build();
+    }
+}