fix: Make FDv1 fallback TTL-based and recover to FDv2 after the TTL

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

@@ -0,0 +1,87 @@
+package com.launchdarkly.sdk.server;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event;
+
+import java.time.Duration;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Reads the TTL of an FDv1 fallback directive, which tells the SDK how long to stop using FDv2 before attempting to
+ * recover to it. The directive is carried by the {@code X-LD-FD-Fallback} response header, with its TTL in the
+ * {@code X-LD-FD-Fallback-TTL} header, or by the {@code protocolFallbackTTL} field of a {@code goodbye} event.
+ */
+final class FDv1FallbackDirective {
+    /**
+     * The maximum TTL, which is also the TTL used when the service supplies no usable TTL.
+     */
+    static final Duration DEFAULT_TTL = Duration.ofHours(1);
+
+    private static final String PROTOCOL_FALLBACK_TTL = "protocolFallbackTTL";
+
+    private FDv1FallbackDirective() {
+    }
+
+    /**
+     * Returns a service-supplied TTL as given if it is in the range (0, {@link #DEFAULT_TTL}]. Any other value,
+     * including null, is replaced by {@link #DEFAULT_TTL} minus a jitter of up to half of it, so that a fallback is
+     * never indefinite and SDKs that fell back together do not all attempt to recover at once.
+     *
+     * @param ttl the TTL supplied by the service, or null if none was supplied
+     * @return the TTL to use
+     */
+    static Duration ttlOrDefault(Duration ttl) {
+        if (ttl != null && !ttl.isNegative() && !ttl.isZero() && ttl.compareTo(DEFAULT_TTL) <= 0) {
+            return ttl;
+        }
+        long ttlSeconds = DEFAULT_TTL.getSeconds();
+        long jitterSeconds = ThreadLocalRandom.current().nextLong(ttlSeconds / 2 + 1);
+        return Duration.ofSeconds(ttlSeconds - jitterSeconds);
+    }
+
+    /**
+     * Reads a directive from the {@code X-LD-FD-Fallback} and {@code X-LD-FD-Fallback-TTL} response headers. The TTL
+     * header is a whole number of seconds.
+     *
+     * @param fallbackHeader the value of the {@code X-LD-FD-Fallback} header, or null
+     * @param ttlHeader the value of the {@code X-LD-FD-Fallback-TTL} header, or null
+     * @return the TTL to use, or null if the headers do not request fallback
+     */
+    static Duration fromHeaders(String fallbackHeader, String ttlHeader) {
+        if (fallbackHeader == null || !fallbackHeader.equalsIgnoreCase("true")) {
+            return null;
+        }
+        if (ttlHeader == null) {
+            return ttlOrDefault(null);
+        }
+        try {
+            return ttlOrDefault(Duration.ofSeconds(Long.parseLong(ttlHeader.trim())));
+        } catch (NumberFormatException e) {
+            return ttlOrDefault(null);
+        }
+    }
+
+    /**
+     * Reads a directive from a {@code goodbye} event. The presence of {@code protocolFallbackTTL} is itself the
+     * directive; its value is the TTL in seconds.
+     *
+     * @param goodbye the goodbye event
+     * @return the TTL to use, or null if the event has no {@code protocolFallbackTTL}
+     */
+    static Duration fromGoodbye(FDv2Event goodbye) {
+        JsonElement data = goodbye.getData();
+        if (data == null || !data.isJsonObject()) {
+            return null;
+        }
+        JsonObject object = data.getAsJsonObject();
+        if (!object.has(PROTOCOL_FALLBACK_TTL) || object.get(PROTOCOL_FALLBACK_TTL).isJsonNull()) {
+            return null;
+        }
+        JsonElement ttl = object.get(PROTOCOL_FALLBACK_TTL);
+        if (!ttl.isJsonPrimitive() || !ttl.getAsJsonPrimitive().isNumber()) {
+            return ttlOrDefault(null);
+        }
+        return ttlOrDefault(Duration.ofSeconds((long) ttl.getAsDouble()));
+    }
+}

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

@@ -6,6 +6,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -31,6 +32,7 @@ interface Condition extends Closeable {
         enum ConditionType {
             FALLBACK,
             RECOVERY,
+            FDV1_FALLBACK_EXPIRED,
         }
 
         CompletableFuture<Condition> execute();
@@ -180,4 +182,57 @@ public ConditionType getType() {
             return ConditionType.RECOVERY;
         }
     }
+
+    /**
+     * This condition is met when the TTL of an FDv1 fallback directive elapses, meaning the data source should
+     * return to its FDv2 synchronizers. A directive received while the condition is pending supersedes the TTL that
+     * was in effect.
+     * <p>
+     * This is package-private, instead of private, for ease of testing.
+     */
+    static class FDv1FallbackExpiryCondition implements Condition {
+        private final CompletableFuture<Condition> resultFuture = new CompletableFuture<>();
+        private final ScheduledExecutorService sharedExecutor;
+        private ScheduledFuture<?> timerFuture;
+
+        public FDv1FallbackExpiryCondition(ScheduledExecutorService sharedExecutor, Duration delay) {
+            this.sharedExecutor = sharedExecutor;
+            schedule(delay);
+        }
+
+        private synchronized void schedule(Duration delay) {
+            if (timerFuture != null) {
+                timerFuture.cancel(false);
+            }
+            timerFuture = sharedExecutor.schedule(
+                () -> resultFuture.complete(this),
+                Math.max(0, delay.toNanos()),
+                TimeUnit.NANOSECONDS);
+        }
+
+        @Override
+        public CompletableFuture<Condition> execute() {
+            return resultFuture;
+        }
+
+        @Override
+        public void inform(FDv2SourceResult sourceResult) {
+            if (sourceResult.isFdv1Fallback()) {
+                schedule(FDv1FallbackDirective.ttlOrDefault(sourceResult.getFdv1FallbackTtl()));
+            }
+        }
+
+        @Override
+        public synchronized void close() {
+            if (timerFuture != null) {
+                timerFuture.cancel(false);
+                timerFuture = null;
+            }
+        }
+
+        @Override
+        public ConditionType getType() {
+            return ConditionType.FDV1_FALLBACK_EXPIRED;
+        }
+    }
 }

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

@@ -2,7 +2,8 @@
 
 enum HeaderConstants {
   ENVIRONMENT_ID("x-ld-envid"),
-  FDV1_FALLBACK("x-ld-fd-fallback");
+  FDV1_FALLBACK("x-ld-fd-fallback"),
+  FDV1_FALLBACK_TTL("x-ld-fd-fallback-ttl");
 
   private final String headerName;
 

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

@@ -15,6 +15,7 @@
 import com.launchdarkly.sdk.json.SerializationException;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Date;
@@ -35,13 +36,14 @@ protected void internalShutdown() {
         requestor.close();
     }
 
-    private static boolean getFallback(FDv2Requestor.FDv2PayloadResponse response) {
+    private static Duration getFallback(FDv2Requestor.FDv2PayloadResponse response) {
         if (response != null && response.getHeaders() != null) {
-            String headerValue = response.getHeaders().get(HeaderConstants.FDV1_FALLBACK.getHeaderName());
-            return headerValue != null && headerValue.equalsIgnoreCase("true");
+            return FDv1FallbackDirective.fromHeaders(
+                response.getHeaders().get(HeaderConstants.FDV1_FALLBACK.getHeaderName()),
+                response.getHeaders().get(HeaderConstants.FDV1_FALLBACK_TTL.getHeaderName()));
         }
 
-        return false;
+        return null;
     }
 
     private static String getEnvironmentId(FDv2Requestor.FDv2PayloadResponse response) {
@@ -53,7 +55,7 @@ private static String getEnvironmentId(FDv2Requestor.FDv2PayloadResponse respons
 
     protected CompletableFuture<FDv2SourceResult> poll(Selector selector, boolean oneShot) {
         return requestor.Poll(selector).handle(((pollingResponse, ex) -> {
-            boolean fdv1Fallback = getFallback(pollingResponse);
+            Duration fdv1Fallback = getFallback(pollingResponse);
             String environmentId = getEnvironmentId(pollingResponse);
             if (ex != null) {
                 if (ex instanceof IOException) {
@@ -148,7 +150,8 @@ protected CompletableFuture<FDv2SourceResult> poll(Selector selector, boolean on
                     case GOODBYE:
                         String reason = ((FDv2ProtocolHandler.FDv2ActionGoodbye) res).getReason();
                         logger.info("Goodbye was received from the LaunchDarkly connection with reason: '{}'.", reason);
-                        return FDv2SourceResult.goodbye(reason, fdv1Fallback);
+                        Duration goodbyeFallback = FDv1FallbackDirective.fromGoodbye(event);
+                        return FDv2SourceResult.goodbye(reason, goodbyeFallback != null ? goodbyeFallback : fdv1Fallback);
                     case NONE:
                         break;
                     case INTERNAL_ERROR: {

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

@@ -4,6 +4,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 
 /**
@@ -36,6 +37,11 @@ class SourceManager implements Closeable {
      */
     private SynchronizerFactoryWithState currentSynchronizerFactory;
 
+    /**
+     * The FDv2 synchronizers that were available when FDv1 fallback began, so they can be restored on recovery.
+     */
+    private final List<SynchronizerFactoryWithState> suspendedForFDv1Fallback = new ArrayList<>();
+
     public SourceManager(List<SynchronizerFactoryWithState> synchronizers, List<FDv2DataSource.DataSourceFactory<Initializer>> initializers) {
         this.synchronizers = synchronizers;
         this.initializers = initializers;
@@ -61,15 +67,47 @@ public boolean hasFDv1Fallback() {
     }
 
     /**
-     * Block all synchronizers aside from the fdv1 fallback and unblock the fdv1 fallback.
+     * Block all synchronizers aside from the fdv1 fallback and unblock the fdv1 fallback. The FDv2 synchronizers
+     * that were available are remembered so that {@link #endFDv1Fallback()} can restore them.
      */
     public void fdv1Fallback() {
-        for (SynchronizerFactoryWithState factory : synchronizers) {
-            if(factory.isFDv1Fallback()) {
+        synchronized (activeSourceLock) {
+            for (SynchronizerFactoryWithState factory : synchronizers) {
+                if (factory.isFDv1Fallback()) {
+                    factory.unblock();
+                } else if (factory.getState() == SynchronizerFactoryWithState.State.Available) {
+                    factory.block();
+                    suspendedForFDv1Fallback.add(factory);
+                }
+            }
+        }
+    }
+
+    /**
+     * Restore the FDv2 synchronizers that were blocked by {@link #fdv1Fallback()}, block the fdv1 fallback, and
+     * start again from the first synchronizer.
+     */
+    public void endFDv1Fallback() {
+        synchronized (activeSourceLock) {
+            for (SynchronizerFactoryWithState factory : synchronizers) {
+                if (factory.isFDv1Fallback()) {
+                    factory.block();
+                }
+            }
+            for (SynchronizerFactoryWithState factory : suspendedForFDv1Fallback) {
                 factory.unblock();
-            } else {
-                factory.block();
             }
+            suspendedForFDv1Fallback.clear();
+            synchronizerIndex = -1;
+        }
+    }
+
+    /**
+     * @return true if {@link #fdv1Fallback()} suspended FDv2 synchronizers that have not yet been restored
+     */
+    public boolean hasSuspendedFDv2Synchronizers() {
+        synchronized (activeSourceLock) {
+            return !suspendedForFDv1Fallback.isEmpty();
         }
     }
 
@@ -133,6 +171,9 @@ public com.launchdarkly.sdk.server.datasources.Synchronizer getNextAvailableSync
             SynchronizerFactoryWithState factory = getNextAvailableSynchronizer();
             if (factory == null) {
                 currentSynchronizerFactory = null;
+                // Nothing replaces the previously active source, so stop it rather than leave it running.
+                safeClose(activeSource);
+                activeSource = null;
                 return null;
             }
 

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

@@ -6,6 +6,7 @@
 import com.launchdarkly.eventsource.FaultEvent;
 import com.launchdarkly.eventsource.HttpConnectStrategy;
 import com.launchdarkly.eventsource.MessageEvent;
+import com.launchdarkly.eventsource.ResponseHeaders;
 import com.launchdarkly.eventsource.RetryDelayStrategy;
 import com.launchdarkly.eventsource.StreamClosedByCallerException;
 import com.launchdarkly.eventsource.StreamEvent;
@@ -320,7 +321,8 @@ private void handleMessage(MessageEvent event) {
             case GOODBYE:
                 String reason = ((FDv2ProtocolHandler.FDv2ActionGoodbye) action).getReason();
                 logger.info("Goodbye was received from the LaunchDarkly connection with reason: '{}'.", reason);
-                result = FDv2SourceResult.goodbye(reason, getFallback(event));
+                Duration goodbyeFallback = FDv1FallbackDirective.fromGoodbye(fdv2Event);
+                result = FDv2SourceResult.goodbye(reason, goodbyeFallback != null ? goodbyeFallback : getFallback(event));
                 // We drop this current connection and attempt to restart the stream.
                 restartStream(false);  // Not a failure - deliberate server-initiated restart
                 break;
@@ -447,23 +449,25 @@ private FDv2Event parseFDv2Event(String eventName, Reader eventDataReader) throw
         }
     }
 
-    private static boolean getFallback(Exception ex) {
+    private static Duration getFallback(Exception ex) {
         if(ex instanceof StreamHttpErrorException) {
-            String headerValue = ((StreamHttpErrorException) ex).getHeaders()
-              .value(HeaderConstants.FDV1_FALLBACK.getHeaderName());
-            return headerValue != null && headerValue.equalsIgnoreCase("true");
+            return getFallback(((StreamHttpErrorException) ex).getHeaders());
         }
-        return false;
+        return null;
     }
 
-    private static boolean getFallback(StreamEvent event) {
-        String headerName = HeaderConstants.FDV1_FALLBACK.getHeaderName();
-        String headerValue = null;
+    private static Duration getFallback(StreamEvent event) {
         if(event instanceof FaultEvent) {
-            headerValue = ((FaultEvent) event).getHeaders().value(headerName);
+            return getFallback(((FaultEvent) event).getHeaders());
         } else if (event instanceof MessageEvent) {
-            headerValue = ((MessageEvent) event).getHeaders().value(headerName);
+            return getFallback(((MessageEvent) event).getHeaders());
         }
-        return headerValue != null && headerValue.equalsIgnoreCase("true");
+        return null;
+    }
+
+    private static Duration getFallback(ResponseHeaders headers) {
+        return FDv1FallbackDirective.fromHeaders(
+            headers.value(HeaderConstants.FDV1_FALLBACK.getHeaderName()),
+            headers.value(HeaderConstants.FDV1_FALLBACK_TTL.getHeaderName()));
     }
 }