feat(server): RETRY-spec conformance in FDv1 streaming and polling (SDK-2789)

lib/sdk/server/contract-tests/service/src/main/java/sdktest/Representations.java

@@ -39,6 +39,11 @@ public static class SdkConfigParams {
   public static class SdkConfigStreamParams {
     URI baseUri;
     long initialRetryDelayMs;
+    // RETRY-conformance test knobs (per SDK-2789 / server-sdk-guide.md). Zero
+    // means "use SDK default"; positive values compress test timing below the
+    // 5-minute extended-regime default.
+    Long extendedInitialDelayMs;
+    Long resetThresholdMs;
     String filter;
   }
 
@@ -177,6 +182,8 @@ public static class SdkConfigSynchronizerParams {
   public static class SdkConfigPollingParams {
     URI baseUri;
     Long pollIntervalMs;
+    // RETRY-conformance test knob (per SDK-2789 / server-sdk-guide.md).
+    Long extendedInitialDelayMs;
     String filter;
   }
 

lib/sdk/server/contract-tests/service/src/main/java/sdktest/SdkClientEntity.java

@@ -397,6 +397,18 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) {
       if (params.streaming.initialRetryDelayMs > 0) {
         dataSource.initialReconnectDelay(Duration.ofMillis(params.streaming.initialRetryDelayMs));
       }
+      // RETRY-conformance test knobs (SDK-2789 / server-sdk-guide.md). Reach the
+      // package-private impl setters via the SDK's internal bridge.
+      if (params.streaming.extendedInitialDelayMs != null) {
+        com.launchdarkly.sdk.server.internal.DataSourceInternalHelpers
+            .setStreamingExtendedInitialReconnectDelay(dataSource,
+                Duration.ofMillis(params.streaming.extendedInitialDelayMs.longValue()));
+      }
+      if (params.streaming.resetThresholdMs != null) {
+        com.launchdarkly.sdk.server.internal.DataSourceInternalHelpers
+            .setStreamingRetryResetInterval(dataSource,
+                Duration.ofMillis(params.streaming.resetThresholdMs.longValue()));
+      }
       dataSource.payloadFilter(params.streaming.filter);
       builder.dataSource(dataSource);
     } else if (params.polling != null && params.dataSystem == null) {
@@ -405,6 +417,11 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) {
       if (params.polling.pollIntervalMs != null) {
         pollingDataSource.pollInterval(Duration.ofMillis(params.polling.pollIntervalMs));
       }
+      if (params.polling.extendedInitialDelayMs != null) {
+        com.launchdarkly.sdk.server.internal.DataSourceInternalHelpers
+            .setPollingExtendedInitialDelay(pollingDataSource,
+                Duration.ofMillis(params.polling.extendedInitialDelayMs.longValue()));
+      }
       if (params.polling.filter != null && !params.polling.filter.isEmpty()) {
         pollingDataSource.payloadFilter(params.polling.filter);
       }

lib/sdk/server/contract-tests/service/src/main/java/sdktest/TestService.java

@@ -44,7 +44,9 @@ public class TestService {
     "server-side-polling",
     "polling-gzip",
     "fdv1-fallback",
-    "instance-id"
+    "instance-id",
+    "retry-conformance-fdv1-streaming",
+    "retry-conformance-fdv1-polling"
   };
 
   static final Gson gson = new GsonBuilder().serializeNulls().create();

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

@@ -126,6 +126,19 @@ public void close() throws IOException {}
 
   static final class StreamingDataSourceBuilderImpl extends StreamingDataSourceBuilder
       implements DiagnosticDescription {
+    // Package-private test-only setters. Follows the pollIntervalWithNoMinimum
+    // precedent below. Cross-module callers reach these via the bridge class in
+    // com.launchdarkly.sdk.server.internal.
+    StreamingDataSourceBuilderImpl extendedInitialReconnectDelay(Duration d) {
+      this.extendedInitialReconnectDelay = d;
+      return this;
+    }
+
+    StreamingDataSourceBuilderImpl retryResetInterval(Duration d) {
+      this.retryResetInterval = d;
+      return this;
+    }
+
     @Override
     public DataSource build(ClientContext context) {
       LDLogger baseLogger = context.getBaseLogger();
@@ -146,6 +159,8 @@ public DataSource build(ClientContext context) {
           streamUri,
           payloadFilter,
           initialReconnectDelay,
+          extendedInitialReconnectDelay,
+          retryResetInterval,
           logger);
     }
 
@@ -170,6 +185,14 @@ PollingDataSourceBuilderImpl pollIntervalWithNoMinimum(Duration pollInterval) {
       this.pollInterval = pollInterval;
       return this;
     }
+
+    // Package-private test-only setter for RETRY §1.6-driven extended-regime
+    // initial delay. Cross-module callers reach this via the bridge class in
+    // com.launchdarkly.sdk.server.internal.
+    PollingDataSourceBuilderImpl extendedInitialDelay(Duration d) {
+      this.extendedInitialDelay = d;
+      return this;
+    }
 
     @Override
     public DataSource build(ClientContext context) {
@@ -196,6 +219,7 @@ public DataSource build(ClientContext context) {
           context.getDataSourceUpdateSink(),
           ClientContextImpl.get(context).sharedExecutor,
           pollInterval,
+          extendedInitialDelay,
           logger);
     }
 

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

@@ -0,0 +1,33 @@
+package com.launchdarkly.sdk.server;
+
+import com.launchdarkly.sdk.server.ComponentsImpl.PollingDataSourceBuilderImpl;
+import com.launchdarkly.sdk.server.ComponentsImpl.StreamingDataSourceBuilderImpl;
+import com.launchdarkly.sdk.server.integrations.PollingDataSourceBuilder;
+import com.launchdarkly.sdk.server.integrations.StreamingDataSourceBuilder;
+
+import java.time.Duration;
+
+/**
+ * Package-private bridge from
+ * {@link com.launchdarkly.sdk.server.internal.DataSourceInternalHelpers} into
+ * the package-private test-only knobs on {@link ComponentsImpl}'s concrete
+ * builder impl classes. Public class-level visibility is required so the
+ * {@code .internal} bridge can call it; the intent is that consumers use the
+ * {@code .internal.DataSourceInternalHelpers} entry points, not this class
+ * directly.
+ */
+public final class ComponentsInternalBridge {
+  private ComponentsInternalBridge() {}
+
+  public static void setStreamingExtendedInitialReconnectDelay(StreamingDataSourceBuilder b, Duration d) {
+    ((StreamingDataSourceBuilderImpl) b).extendedInitialReconnectDelay(d);
+  }
+
+  public static void setStreamingRetryResetInterval(StreamingDataSourceBuilder b, Duration d) {
+    ((StreamingDataSourceBuilderImpl) b).retryResetInterval(d);
+  }
+
+  public static void setPollingExtendedInitialDelay(PollingDataSourceBuilder b, Duration d) {
+    ((PollingDataSourceBuilderImpl) b).extendedInitialDelay(d);
+  }
+}

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

@@ -0,0 +1,67 @@
+package com.launchdarkly.sdk.server;
+
+import javax.net.ssl.SSLException;
+
+import java.security.GeneralSecurityException;
+import java.security.cert.CertificateException;
+
+/**
+ * Classifies a data-source failure per the RETRY specification (§1.5 / §1.6 / §1.7)
+ * into one of two regimes: {@link #NORMAL} or {@link #UNEXPECTED}. This is the
+ * SDK-side classification the streaming and polling data sources use to decide
+ * whether a failure should trigger extended-regime backoff (per SDK-2775 /
+ * server-sdk-guide.md).
+ */
+enum FailureClass {
+  /**
+   * Ordinary transient failure. Use the normal-regime backoff. Per RETRY §1.6.1,
+   * this includes HTTP 400 / 408 / 429, HTTP 5xx, any other HTTP status the SDK
+   * treats as a failure, and generic transport failures (connection refused,
+   * read timeout, DNS failure, etc.).
+   */
+  NORMAL,
+
+  /**
+   * Unexpected failure indicative of a longer-lived condition. Use the extended-
+   * regime backoff. Per RETRY §1.6 / §1.7, this includes HTTP 401 / 403 and any
+   * other 4xx not in the NORMAL list, plus TLS / certificate validation failures.
+   */
+  UNEXPECTED;
+
+  /**
+   * Classify a completed HTTP response by its status code, per RETRY §1.6.
+   */
+  static FailureClass fromHttpStatus(int status) {
+    // 400, 408, 429 are NORMAL per §1.6.1.
+    if (status == 400 || status == 408 || status == 429) {
+      return NORMAL;
+    }
+    // 5xx is NORMAL.
+    if (status >= 500) {
+      return NORMAL;
+    }
+    // Any other 4xx (including 401 and 403) is UNEXPECTED per §1.6.
+    if (status >= 400 && status < 500) {
+      return UNEXPECTED;
+    }
+    // Any non-4xx / non-5xx status the SDK treats as a failure is NORMAL under
+    // the §1.6.1 catch-all.
+    return NORMAL;
+  }
+
+  /**
+   * Classify a transport-level exception (network I/O, TLS, DNS, etc.), per
+   * RETRY §1.7. TLS / certificate validation failures are UNEXPECTED; all other
+   * transport failures are NORMAL.
+   */
+  static FailureClass fromTransportException(Throwable t) {
+    for (Throwable c = t; c != null; c = c.getCause()) {
+      if (c instanceof SSLException
+          || c instanceof CertificateException
+          || c instanceof GeneralSecurityException) {
+        return UNEXPECTED;
+      }
+    }
+    return NORMAL;
+  }
+}

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

@@ -21,9 +21,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import static com.launchdarkly.sdk.internal.http.HttpErrors.checkIfErrorIsRecoverableAndLog;
-import static com.launchdarkly.sdk.internal.http.HttpErrors.httpErrorDescription;
-
 final class PollingProcessor implements DataSource {
   private static final String ERROR_CONTEXT_MESSAGE = "on polling request";
   private static final String WILL_RETRY_MESSAGE = "will retry at next scheduled poll interval";
@@ -32,7 +29,9 @@ final class PollingProcessor implements DataSource {
   private final DataSourceUpdateSink dataSourceUpdates;
   private final ScheduledExecutorService scheduler;
   @VisibleForTesting final Duration pollInterval;
+  @VisibleForTesting final PollingStrategy strategy;
   private final AtomicBoolean initialized = new AtomicBoolean(false);
+  private final AtomicBoolean closed = new AtomicBoolean(false);
   private final CompletableFuture<Void> initFuture;
   private volatile ScheduledFuture<?> task;
   private final LDLogger logger;
@@ -42,12 +41,14 @@ final class PollingProcessor implements DataSource {
       DataSourceUpdateSink dataSourceUpdates,
       ScheduledExecutorService sharedExecutor,
       Duration pollInterval,
+      Duration extendedInitialDelay,
       LDLogger logger
       ) {
     this.requestor = requestor; // note that HTTP configuration is applied to the requestor when it is created
     this.dataSourceUpdates = dataSourceUpdates;
     this.scheduler = sharedExecutor;
     this.pollInterval = pollInterval;
+    this.strategy = new PollingStrategy(pollInterval, extendedInitialDelay);
     this.initFuture = new CompletableFuture<>();
     this.logger = logger;
   }
@@ -59,12 +60,12 @@ public boolean isInitialized() {
 
   @Override
   public void close() throws IOException {
+    if (closed.getAndSet(true)) {
+      return;
+    }
     logger.info("Closing LaunchDarkly PollingProcessor");
     requestor.close();
-
-    // Even though the shared executor will be shut down when the LDClient is closed, it's still good
-    // behavior to remove our polling task now - especially because we might be running in a test
-    // environment where there isn't actually an LDClient.
+
     synchronized (this) {
       if (task != null) {
         task.cancel(true);
@@ -73,20 +74,35 @@ public void close() throws IOException {
     }
   }
 
+  private final AtomicBoolean started = new AtomicBoolean(false);
+
   @Override
   public Future<Void> start() {
+    if (started.getAndSet(true)) {
+      return initFuture; // idempotent per pre-existing contract
+    }
     logger.info("Starting LaunchDarkly polling client with interval: {} milliseconds",
         pollInterval.toMillis());
-
+    scheduleNext(Duration.ZERO);
+    return initFuture;
+  }
+
+  // Per RETRY §1.2.1 no failure is permanently terminal. The polling loop drives
+  // itself via strategy.nextWait() after each poll (success or failure), replacing
+  // the pre-RETRY fixed-rate schedule. Extended-regime backoff is engaged by the
+  // strategy on classification of UNEXPECTED (RETRY §1.6 / §1.7).
+  private void scheduleNext(Duration delay) {
+    if (closed.get()) {
+      return;
+    }
     synchronized (this) {
-      if (task == null) {
-        task = scheduler.scheduleAtFixedRate(this::poll, 0L, pollInterval.toMillis(), TimeUnit.MILLISECONDS);
+      if (closed.get()) {
+        return;
       }
+      task = scheduler.schedule(this::poll, delay.toMillis(), TimeUnit.MILLISECONDS);
     }
-
-    return initFuture;
   }
-  
+
   private void poll() {
     try {
       // If we already obtained data earlier, and the poll request returns a cached response, then we don't
@@ -101,35 +117,42 @@ private void poll() {
         if (dataSourceUpdates.init(allData)) {
           dataSourceUpdates.updateStatus(State.VALID, null);
           if (!initialized.getAndSet(true)) {
-            logger.info("Initialized LaunchDarkly client."); 
+            logger.info("Initialized LaunchDarkly client.");
             initFuture.complete(null);
           }
         }
       }
+      strategy.onSuccess();
     } catch (HttpErrorException e) {
       ErrorInfo errorInfo = ErrorInfo.fromHttpError(e.getStatus());
-      boolean recoverable = checkIfErrorIsRecoverableAndLog(logger, httpErrorDescription(e.getStatus()),
-          ERROR_CONTEXT_MESSAGE, e.getStatus(), WILL_RETRY_MESSAGE);
-      if (recoverable) {
-        dataSourceUpdates.updateStatus(State.INTERRUPTED, errorInfo);
-      } else {
-        dataSourceUpdates.updateStatus(State.OFF, errorInfo);
-        initFuture.complete(null); // if client is initializing, make it stop waiting; has no effect if already inited
-        if (task != null) {
-          task.cancel(true);
-          task = null;
-        }
-      }
+      FailureClass klass = FailureClass.fromHttpStatus(e.getStatus());
+      logger.warn("Received HTTP {} error {} ({} - {})", e.getStatus(),
+          klass == FailureClass.UNEXPECTED ? "(unexpected regime)" : "(normal regime)",
+          ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE);
+      dataSourceUpdates.updateStatus(State.INTERRUPTED, errorInfo);
+      strategy.onFailure(klass);
     } catch (IOException e) {
-      checkIfErrorIsRecoverableAndLog(logger, e.toString(), ERROR_CONTEXT_MESSAGE, 0, WILL_RETRY_MESSAGE);
+      FailureClass klass = FailureClass.fromTransportException(e);
+      logger.warn("Transport error {} ({} - {}): {}",
+          klass == FailureClass.UNEXPECTED ? "(unexpected regime)" : "(normal regime)",
+          ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE, e.toString());
       dataSourceUpdates.updateStatus(State.INTERRUPTED, ErrorInfo.fromException(ErrorKind.NETWORK_ERROR, e));
+      strategy.onFailure(klass);
     } catch (SerializationException e) {
       logger.error("Polling request received malformed data: {}", e.toString());
       dataSourceUpdates.updateStatus(State.INTERRUPTED, ErrorInfo.fromException(ErrorKind.INVALID_DATA, e));
+      strategy.onFailure(FailureClass.NORMAL);
     } catch (Exception e) {
       logger.error("Unexpected error from polling processor: {}", e.toString());
       logger.debug(e.toString(), e);
       dataSourceUpdates.updateStatus(State.INTERRUPTED, ErrorInfo.fromException(ErrorKind.UNKNOWN, e));
+      strategy.onFailure(FailureClass.NORMAL);
+    } finally {
+      // Regardless of poll outcome, schedule the next attempt per strategy.
+      // Per RETRY §1.2.1 the polling loop never permanently stops on a data-source
+      // failure — the extended-regime backoff carries the retry cadence instead.
+      Duration wait = strategy.nextWait();
+      scheduleNext(wait);
     }
   }
 }