feat: Apply RETRY to FDv2 synchronizer orchestration

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,37 @@ 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));
                     }
                     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()
       );

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

@@ -176,8 +176,7 @@ static FDv2DataSystem create(
         fdv1FallbackFactory,
         dataSourceUpdates,
         config.threadPriority,
-        clientContext.getBaseLogger().subLogger(Loggers.DATA_SOURCE_LOGGER_NAME),
-        clientContext.sharedExecutor
+        clientContext.getBaseLogger().subLogger(Loggers.DATA_SOURCE_LOGGER_NAME)
       );
     }
 

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

@@ -1,6 +1,7 @@
 package com.launchdarkly.sdk.server;
 
 import com.launchdarkly.logging.LDLogger;
+import com.launchdarkly.sdk.internal.http.FailureClass;
 import com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event;
 import com.launchdarkly.sdk.internal.fdv2.sources.FDv2ProtocolHandler;
 import com.launchdarkly.sdk.fdv2.Selector;
@@ -65,7 +66,10 @@ protected CompletableFuture<FDv2SourceResult> poll(Selector selector, boolean on
                             e.toString(),
                             new Date().toInstant()
                     );
-                    return oneShot ? FDv2SourceResult.terminalError(info, fdv1Fallback) : FDv2SourceResult.interrupted(info, fdv1Fallback);
+                    boolean recoverable = classifyTransportFailure(e) == FailureClass.NORMAL;
+                    return oneShot || !recoverable
+                        ? FDv2SourceResult.terminalError(info, fdv1Fallback)
+                        : FDv2SourceResult.interrupted(info, fdv1Fallback);
                 } else if (ex instanceof SerializationException) {
                     SerializationException e = (SerializationException) ex;
                     logger.error("Polling request received malformed data: {}", e.toString());
@@ -101,7 +105,7 @@ protected CompletableFuture<FDv2SourceResult> poll(Selector selector, boolean on
             }
             if(!pollingResponse.isSuccess()) {
                     int statusCode = pollingResponse.getStatusCode();
-                    boolean recoverable = statusCode <= 0 || isHttpErrorRecoverable(statusCode);
+                    boolean recoverable = classifyHttpFailure(statusCode) == FailureClass.NORMAL;
                     DataSourceStatusProvider.ErrorInfo errorInfo = DataSourceStatusProvider.ErrorInfo.fromHttpError(statusCode);
                     logger.error("Polling request failed with HTTP error: {}", statusCode);
                     // For a one-shot request all errors are terminal.

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

@@ -10,13 +10,24 @@
 import java.util.concurrent.*;
 
 class PollingSynchronizerImpl extends PollingBase implements Synchronizer {
+    /**
+     * The number of consecutive successful polls that meets the reset threshold.
+     */
+    private static final int RESET_THRESHOLD_SUCCESSFUL_POLLS = 2;
+
     private final CompletableFuture<FDv2SourceResult> shutdownFuture = new CompletableFuture<>();
     private final SelectorSource selectorSource;
 
     private final ScheduledFuture<?> task;
 
     private final IterableAsyncQueue<FDv2SourceResult> resultQueue = new IterableAsyncQueue<>();
 
+    /**
+     * Only {@link #doPoll()} reads or writes this. {@code scheduleAtFixedRate} never starts a run of
+     * {@code doPoll} before the previous run has finished.
+     */
+    private int consecutiveSuccessfulPolls = 0;
+
     public PollingSynchronizerImpl(
             FDv2Requestor requestor,
             LDLogger logger,
@@ -47,10 +58,15 @@ private void doPoll() {
             boolean shouldShutdown = false;
             switch (res.getResultType()) {
                 case CHANGE_SET:
+                    consecutiveSuccessfulPolls++;
+                    if (consecutiveSuccessfulPolls == RESET_THRESHOLD_SUCCESSFUL_POLLS) {
+                        res = res.withResetThresholdMet();
+                    }
                     break;
                 case STATUS:
                     switch (res.getStatus().getState()) {
                         case INTERRUPTED:
+                            consecutiveSuccessfulPolls = 0;
                             break;
                         case SHUTDOWN:
                             // The base poller doesn't emit shutdown, we instead handle it at this level.