feat: Apply RETRY to FDv2 synchronizer orchestration

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

@@ -141,8 +141,9 @@ public Synchronizer build(DataSourceBuildInputs inputs) {
                     StandardEndpoints.FDV2_STREAMING_REQUEST_BASE_PATH,
                     requestor,
                     initialReconnectDelayMillis,
+                    FDv2StreamingSynchronizer.RESET_THRESHOLD_MILLIS,
                     inputs.isEvaluationReasons(), inputs.getHttp().shouldUsePost(),
-                    httpProps, inputs.getSharedExecutor(),
+                    httpProps, inputs.getSharedExecutor(), inputs.getSharedExecutor(),
                     inputs.getBaseLogger(), null);
         }
     }

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

@@ -31,6 +31,11 @@
  * objects so it can be used as a drop-in synchronizer within the FDv2 data source pipeline.
  */
 final class FDv1PollingSynchronizer implements Synchronizer {
+    /**
+     * The number of consecutive successful polls that meets the reset threshold, the same as for an FDv2 polling
+     * synchronizer.
+     */
+    private static final int RESET_THRESHOLD_SUCCESSFUL_POLLS = 2;
 
     private final LDContext evaluationContext;
     private final FeatureFetcher fetcher;
@@ -42,6 +47,12 @@ final class FDv1PollingSynchronizer implements Synchronizer {
     private volatile ScheduledFuture<?> scheduledTask;
     private final Object taskLock = new Object();
 
+    /**
+     * Only {@link #pollAndEnqueue()} reads or writes this. {@code scheduleWithFixedDelay} never starts a run of
+     * {@code pollAndEnqueue} before the previous run has finished.
+     */
+    private int consecutiveSuccessfulPolls = 0;
+
     /**
      * @param evaluationContext  the context to evaluate flags for
      * @param fetcher            the HTTP transport for FDv1 polling requests
@@ -74,8 +85,18 @@ private void pollAndEnqueue() {
         try {
             FDv2SourceResult result = doPoll();
 
+            if (result.getResultType() == SourceResultType.CHANGE_SET) {
+                consecutiveSuccessfulPolls++;
+                if (consecutiveSuccessfulPolls == RESET_THRESHOLD_SUCCESSFUL_POLLS) {
+                    result = result.withResetThresholdMet();
+                }
+            }
+
             if (result.getResultType() == SourceResultType.STATUS) {
                 FDv2SourceResult.Status status = result.getStatus();
+                if (status != null && status.getState() == SourceSignal.INTERRUPTED) {
+                    consecutiveSuccessfulPolls = 0;
+                }
                 if (status != null && status.getState() == SourceSignal.TERMINAL_ERROR) {
                     synchronized (taskLock) {
                         if (scheduledTask != null) {
@@ -91,6 +112,7 @@ private void pollAndEnqueue() {
             resultQueue.put(result);
         } catch (RuntimeException e) {
             LDUtil.logExceptionAtErrorLevel(logger, e, "Unexpected exception in FDv1 polling synchronizer task");
+            consecutiveSuccessfulPolls = 0;
             resultQueue.put(FDv2SourceResult.status(FDv2SourceResult.Status.interrupted(e), false));
         }
     }

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

@@ -26,6 +26,11 @@
  * that subsequent {@link #next()} calls also return the terminal result immediately.
  */
 final class FDv2PollingSynchronizer extends FDv2PollingBase implements Synchronizer {
+    /**
+     * The number of consecutive successful polls that meets the reset threshold.
+     */
+    private static final int RESET_THRESHOLD_SUCCESSFUL_POLLS = 2;
+
     private final SelectorSource selectorSource;
 
     private final LDAsyncQueue<FDv2SourceResult> resultQueue = new LDAsyncQueue<>();
@@ -34,6 +39,12 @@ final class FDv2PollingSynchronizer extends FDv2PollingBase implements Synchroni
     private volatile ScheduledFuture<?> scheduledTask;
     private final Object taskLock = new Object();
 
+    /**
+     * Only {@link #pollAndEnqueue()} reads or writes this. {@code scheduleWithFixedDelay} never starts a run of
+     * {@code pollAndEnqueue} before the previous run has finished.
+     */
+    private int consecutiveSuccessfulPolls = 0;
+
     /**
      * @param requestor           the FDv2 requestor used to perform each poll
      * @param selectorSource      source of the current selector, sent as the {@code basis} param
@@ -70,8 +81,18 @@ private void pollAndEnqueue() {
         try {
             FDv2SourceResult result = doPoll(selectorSource.getSelector(), false);
 
+            if (result.getResultType() == SourceResultType.CHANGE_SET) {
+                consecutiveSuccessfulPolls++;
+                if (consecutiveSuccessfulPolls == RESET_THRESHOLD_SUCCESSFUL_POLLS) {
+                    result = result.withResetThresholdMet();
+                }
+            }
+
             if (result.getResultType() == SourceResultType.STATUS) {
                 FDv2SourceResult.Status status = result.getStatus();
+                if (status != null && status.getState() == SourceSignal.INTERRUPTED) {
+                    consecutiveSuccessfulPolls = 0;
+                }
                 if (status != null && status.getState() == SourceSignal.TERMINAL_ERROR) {
                     synchronized (taskLock) {
                         if (scheduledTask != null) {
@@ -93,6 +114,7 @@ private void pollAndEnqueue() {
             // cancels the task on any unchecked exception, ending all future polls with no
             // error signal. Log and enqueue an INTERRUPTED result so the consumer is notified
             LDUtil.logExceptionAtErrorLevel(logger, e, "Unexpected exception in polling synchronizer task");
+            consecutiveSuccessfulPolls = 0;
             resultQueue.put(FDv2SourceResult.status(FDv2SourceResult.Status.interrupted(e), false));
         }
     }