feat: Apply RETRY to FDv2 synchronizer orchestration

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

@@ -0,0 +1,65 @@
+package com.launchdarkly.sdk.server;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.WeakHashMap;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A latch that is released once, and that callers wait on through futures.
+ * <p>
+ * This exists so that a loop can wait for "a result or the latch, whichever comes first" on every iteration without
+ * a memory leak. The obvious way to write that wait is {@code CompletableFuture.anyOf(result, latchFuture)}. But
+ * {@code anyOf} attaches a completion node to each future it is given, and a pending {@link CompletableFuture} keeps
+ * those nodes until it completes. If the latch used one shared future, every iteration that ended with a result would
+ * leave one more node on that future. The latch can stay pending for the life of the process, so the nodes would
+ * accumulate without limit.
+ * <p>
+ * To prevent this, {@link #newFuture()} returns a new future for each wait. The latch holds these futures weakly, so a
+ * future that nobody waits on any more is garbage collected, together with the nodes that {@code anyOf} attached to
+ * it. When the latch is released, it completes each future that is still waiting. After the release,
+ * {@link #newFuture()} returns a future that is already complete.
+ */
+final class AsyncLatch {
+    private final Object lock = new Object();
+
+    // The futures that wait for the release, or null after the release.
+    private Set<CompletableFuture<Void>> pending = Collections.newSetFromMap(new WeakHashMap<>());
+
+    /**
+     * A new future that completes when the latch is released.
+     *
+     * @return the future, which is already complete if the latch has been released
+     */
+    CompletableFuture<Void> newFuture() {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        synchronized (lock) {
+            if (pending != null) {
+                pending.add(future);
+                return future;
+            }
+        }
+        future.complete(null);
+        return future;
+    }
+
+    /**
+     * Releases the latch, which completes every future from {@link #newFuture()}. Releasing it again has no effect.
+     */
+    void release() {
+        List<CompletableFuture<Void>> waiting;
+        synchronized (lock) {
+            if (pending == null) {
+                return;
+            }
+            // The list holds the futures strongly, so none is garbage collected before it completes.
+            waiting = new ArrayList<>(pending);
+            pending = null;
+        }
+        for (CompletableFuture<Void> future : waiting) {
+            future.complete(null);
+        }
+    }
+}

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

@@ -1,6 +1,8 @@
 package com.launchdarkly.sdk.server;
 
 import com.launchdarkly.sdk.internal.collections.IterableAsyncQueue;
+import com.launchdarkly.sdk.internal.http.FailureClass;
+import com.launchdarkly.sdk.internal.http.HttpErrors;
 import com.launchdarkly.sdk.fdv2.Selector;
 import com.launchdarkly.sdk.server.datasources.FDv2SourceResult;
 import com.launchdarkly.sdk.server.datasources.Synchronizer;
@@ -130,9 +132,14 @@ public void close()  {
      * This sink doesn't delegate to any other sink - it exists solely to convert FDv1 updates to FDv2 results.
      */
     private static class ConvertingUpdateSink implements DataSourceUpdateSink {
+        // The number of VALID statuses in a row that meet the reset threshold, the same as for a polling synchronizer.
+        private static final int RESET_THRESHOLD_SUCCESSFUL_RESULTS = 2;
+
         private final IterableAsyncQueue<FDv2SourceResult> resultQueue;
         // Only a full data set carries an environment ID; it is retained for subsequent partial updates.
         private volatile String environmentId = null;
+        // The number of VALID statuses the data source has reported since its last failure.
+        private int consecutiveSuccessfulResults = 0;
 
         public ConvertingUpdateSink(IterableAsyncQueue<FDv2SourceResult> resultQueue) {
             this.resultQueue = resultQueue;
@@ -189,16 +196,39 @@ public void updateStatus(DataSourceStatusProvider.State newState, DataSourceStat
             // Convert state changes to FDv2SourceResult status events
             switch (newState) {
                 case INTERRUPTED:
-                    resultQueue.put(FDv2SourceResult.interrupted(newError, false));
+                    consecutiveSuccessfulResults = 0;
+                    if (newError != null
+                        && newError.getKind() == DataSourceStatusProvider.ErrorKind.ERROR_RESPONSE
+                        && HttpErrors.classifyHttpFailure(newError.getStatusCode()) == FailureClass.UNEXPECTED) {
+                        resultQueue.put(FDv2SourceResult.terminalError(newError, false));
+                    } else {
+                        resultQueue.put(FDv2SourceResult.interrupted(newError, false));
+                    }
                     break;
                 case OFF:
+                    consecutiveSuccessfulResults = 0;
                     if (newError != null) {
                         resultQueue.put(FDv2SourceResult.terminalError(newError, false));
+                    } else {
+                        resultQueue.put(FDv2SourceResult.shutdown());
                     }
                     break;
                 case VALID:
+                    consecutiveSuccessfulResults++;
+                    if (consecutiveSuccessfulResults == RESET_THRESHOLD_SUCCESSFUL_RESULTS) {
+                        // Any data that came with this status is already queued, so report the reset with no changes.
+                        ChangeSet<Iterable<Map.Entry<DataKind, KeyedItems<ItemDescriptor>>>> noChanges =
+                                new ChangeSet<>(
+                                        ChangeSetType.None,
+                                        Selector.EMPTY,
+                                        Collections.emptyList(),
+                                        environmentId,
+                                        /* shouldPersist= */ true);
+                        resultQueue.put(FDv2SourceResult.changeSet(noChanges, false).withResetThresholdMet());
+                    }
+                    break;
                 case INITIALIZING:
-                    // These states don't map to FDv2SourceResult status events
+                    // This state doesn't map to an FDv2SourceResult status event
                     break;
             }
         }

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

@@ -105,8 +105,10 @@ public Synchronizer build(DataSourceBuildInputs context) {
               StandardEndpoints.FDV2_STREAMING_REQUEST_PATH,
               context.getBaseLogger(),
               context.getSelectorSource(),
+              context.getSharedExecutor(),
               payloadFilter,
               initialReconnectDelay,
+              StreamingSynchronizerImpl.RESET_THRESHOLD,
               context.getThreadPriority(),
               context.getDiagnosticStore()
       );