diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv1FallbackDirective.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv1FallbackDirective.java new file mode 100644 index 00000000..29121ac5 --- /dev/null +++ b/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())); + } +} diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSource.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSource.java index 360be672..121818fd 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSource.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSource.java @@ -11,6 +11,7 @@ import com.launchdarkly.sdk.server.subsystems.DataSource; import com.launchdarkly.sdk.server.subsystems.DataSourceUpdateSinkV2; +import java.time.Duration; import java.util.ArrayList; import java.util.Collections; import java.util.Date; @@ -23,6 +24,7 @@ import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.Condition; import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.ConditionFactory; +import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.FDv1FallbackExpiryCondition; import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.FallbackCondition; import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.RecoveryCondition; @@ -52,6 +54,16 @@ class FDv2DataSource implements DataSource { private volatile boolean closed = false; + private final CompletableFuture closedSignal = new CompletableFuture<>(); + + private final ScheduledExecutorService sharedExecutor; + + /** + * The {@link System#nanoTime()} at which the engaged FDv1 fallback expires, or null if FDv1 fallback is not + * engaged. Only accessed from the run thread. + */ + private Long fdv1FallbackExpiresAtNanos; + /** * Avoid duplicate orchestration logs for the same synchronizer and {@link SourceSignal}. */ @@ -117,6 +129,7 @@ public FDv2DataSource( this.dataSourceUpdates = dataSourceUpdates; this.threadPriority = threadPriority; this.logger = logger; + this.sharedExecutor = sharedExecutor; this.conditionFactories = new ArrayList<>(); this.conditionFactories.add(new FallbackCondition.Factory(sharedExecutor, fallbackTimeout)); this.conditionFactories.add(new RecoveryCondition.Factory(sharedExecutor, recoveryTimeout)); @@ -140,26 +153,19 @@ private void run() { initializerOutcome = runInitializers(); } - // If an initializer signalled FDv1 fallback, block every FDv2 synchronizer in - // one shot via fdv1Fallback() (which also unblocks the FDv1 fallback - // synchronizer, if one is configured). If FDv1 is configured we hand off to it; - // otherwise we halt the data system. This takes precedence over the standard - // synchronizer chain -- the FDv2 synchronizers are not given a chance to run. - if (initializerOutcome.fallbackToFDv1) { - sourceManager.fdv1Fallback(); + // If an initializer signalled FDv1 fallback, the FDv2 synchronizers are not given a + // chance to run until the directive's TTL elapses. If FDv1 is configured we hand off + // to it; otherwise no synchronizer runs until then. + if (initializerOutcome.fdv1FallbackTtl != null) { if (sourceManager.hasFDv1Fallback()) { logger.warn("Initializer requested fallback to FDv1; switching to FDv1 fallback synchronizer."); } else { logger.warn("Initializer requested fallback to FDv1, but no FDv1 fallback synchronizer is configured."); - dataSourceUpdates.updateStatus( - DataSourceStatusProvider.State.OFF, - initializerOutcome.errorInfo); - startFuture.complete(false); - return; } + engageFDv1Fallback(initializerOutcome.fdv1FallbackTtl, initializerOutcome.errorInfo); } - if(!sourceManager.hasAvailableSynchronizers()) { + if(!sourceManager.hasAvailableSynchronizers() && !sourceManager.hasSuspendedFDv2Synchronizers()) { // If already completed by the initializers, then this will have no effect. if (!isInitialized()) { // If we have no synchronizers, and we didn't manage to initialize, and we aren't shutting down, @@ -171,13 +177,11 @@ private void run() { return; } - boolean haltedByDirective = runSynchronizers(); + runSynchronizers(); - if (!haltedByDirective) { - // If we had synchronizers, and we ran out of them, and we aren't shutting down, then that was unexpected, - // and we will report it. - maybeReportUnexpectedExhaustion("All data source acquisition methods have been exhausted."); - } + // If we had synchronizers, and we ran out of them, and we aren't shutting down, then that was unexpected, + // and we will report it. + maybeReportUnexpectedExhaustion("All data source acquisition methods have been exhausted."); // If we had initialized at some point, then the future will already be complete and this will be ignored. startFuture.complete(false); @@ -223,7 +227,7 @@ private InitializerOutcome runInitializers() { dataSourceUpdates.updateStatus(DataSourceStatusProvider.State.VALID, null); startFuture.complete(true); if (result.isFdv1Fallback()) { - return InitializerOutcome.fallbackToFDv1(null); + return InitializerOutcome.fallbackToFDv1(FDv1FallbackDirective.ttlOrDefault(result.getFdv1FallbackTtl()), null); } return InitializerOutcome.completed(); } @@ -259,7 +263,7 @@ private InitializerOutcome runInitializers() { // to the FDv1 fallback synchronizer; the eventual VALID status will come // from the FDv1 synchronizer once it serves a selectorful payload. if (result.isFdv1Fallback()) { - return InitializerOutcome.fallbackToFDv1(fallbackErrorInfo); + return InitializerOutcome.fallbackToFDv1(FDv1FallbackDirective.ttlOrDefault(result.getFdv1FallbackTtl()), fallbackErrorInfo); } } } catch (ExecutionException | InterruptedException | CancellationException e) { @@ -296,20 +300,76 @@ private InitializerOutcome runInitializers() { * server-directed FDv1 fallback before the synchronizer phase begins. */ private static final class InitializerOutcome { - final boolean fallbackToFDv1; + final Duration fdv1FallbackTtl; final DataSourceStatusProvider.ErrorInfo errorInfo; - private InitializerOutcome(boolean fallbackToFDv1, DataSourceStatusProvider.ErrorInfo errorInfo) { - this.fallbackToFDv1 = fallbackToFDv1; + private InitializerOutcome(Duration fdv1FallbackTtl, DataSourceStatusProvider.ErrorInfo errorInfo) { + this.fdv1FallbackTtl = fdv1FallbackTtl; this.errorInfo = errorInfo; } static InitializerOutcome completed() { - return new InitializerOutcome(false, null); + return new InitializerOutcome(null, null); + } + + static InitializerOutcome fallbackToFDv1(Duration fdv1FallbackTtl, DataSourceStatusProvider.ErrorInfo errorInfo) { + return new InitializerOutcome(fdv1FallbackTtl, errorInfo); + } + } + + /** + * Stops the FDv2 synchronizers, retaining them so that they can be restored once the directive's TTL elapses, + * and enables the FDv1 fallback synchronizer if one is configured. If none is configured then no synchronizer + * runs until the TTL elapses, and the data source reports INTERRUPTED. + */ + private void engageFDv1Fallback(Duration ttl, DataSourceStatusProvider.ErrorInfo errorInfo) { + sourceManager.fdv1Fallback(); + fdv1FallbackExpiresAtNanos = System.nanoTime() + ttl.toNanos(); + logger.info("FDv2 will be retried in {} seconds.", ttl.getSeconds()); + if (!sourceManager.hasFDv1Fallback()) { + dataSourceUpdates.updateStatus( + DataSourceStatusProvider.State.INTERRUPTED, + errorInfo != null ? errorInfo : new DataSourceStatusProvider.ErrorInfo( + DataSourceStatusProvider.ErrorKind.UNKNOWN, + 0, + "FDv1 fallback directive received, but no FDv1 fallback synchronizer is configured", + new Date().toInstant())); } + } + + private void endFDv1Fallback() { + fdv1FallbackExpiresAtNanos = null; + sourceManager.endFDv1Fallback(); + logger.info("FDv1 fallback has expired; returning to FDv2 synchronizers."); + } + + private Duration remainingFDv1FallbackTtl() { + return Duration.ofNanos(Math.max(0, fdv1FallbackExpiresAtNanos - System.nanoTime())); + } - static InitializerOutcome fallbackToFDv1(DataSourceStatusProvider.ErrorInfo errorInfo) { - return new InitializerOutcome(true, errorInfo); + /** + * Gets the next synchronizer to run. If there is none because FDv1 fallback is engaged without an FDv1 + * fallback synchronizer that can run, this waits for the fallback to expire and then restores the FDv2 + * synchronizers. + * + * @return the next synchronizer, or null if there are no more synchronizers or the data source was closed + */ + private Synchronizer getNextSynchronizer() { + Synchronizer synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive(); + if (synchronizer != null || fdv1FallbackExpiresAtNanos == null || !sourceManager.hasSuspendedFDv2Synchronizers()) { + return synchronizer; + } + try { + closedSignal.get(remainingFDv1FallbackTtl().toNanos(), TimeUnit.NANOSECONDS); + return null; + } catch (TimeoutException e) { + endFDv1Fallback(); + return sourceManager.getNextAvailableSynchronizerAndSetActive(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } catch (ExecutionException e) { + return null; } } @@ -320,6 +380,14 @@ static InitializerOutcome fallbackToFDv1(DataSourceStatusProvider.ErrorInfo erro * @return a list of conditions to apply to the synchronizer */ private List getConditions() { + if (fdv1FallbackExpiresAtNanos != null) { + // Only the directive's TTL governs the return to FDv2. If there are no FDv2 synchronizers to + // return to, then the FDv1 fallback synchronizer keeps running. + if (!sourceManager.hasSuspendedFDv2Synchronizers()) { + return Collections.emptyList(); + } + return Collections.singletonList(new FDv1FallbackExpiryCondition(sharedExecutor, remainingFDv1FallbackTtl())); + } int availableSynchronizers = sourceManager.getAvailableSynchronizerCount(); boolean isPrimeSynchronizer = sourceManager.isPrimeSynchronizer(); @@ -339,18 +407,12 @@ private List getConditions() { /** * Runs the configured synchronizers, falling back / recovering as conditions allow, - * until the list is exhausted, the data source is closed, or a server-directed FDv1 - * fallback halts the data system. - * - * @return true when {@code runSynchronizers} halted in response to a server-directed - * FDv1 fallback directive that could not be satisfied (no FDv1 fallback - * synchronizer configured) -- the caller should NOT report exhaustion in - * that case because OFF has already been published with a specific error. + * until the list is exhausted or the data source is closed. */ - private boolean runSynchronizers() { + private void runSynchronizers() { // When runSynchronizers exists, no matter how it exits, the synchronizerStateManager will be closed. try { - Synchronizer synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive(); + Synchronizer synchronizer = getNextSynchronizer(); // We want to continue running synchronizers for as long as any are available. while (synchronizer != null) { @@ -386,6 +448,9 @@ private boolean runSynchronizers() { synchronizer.name() ); break; + case FDV1_FALLBACK_EXPIRED: + endFDv1Fallback(); + break; } // A running synchronizer will only have fallback and recovery conditions that it can act on. // So, if there are no synchronizers to recover to or fallback to, then we will not have @@ -431,7 +496,7 @@ private boolean runSynchronizers() { ); // We should be overall shutting down. logger.debug("Synchronizer shutdown."); - return false; + return; case TERMINAL_ERROR: maybeLogSynchronizerStatusChange( synchronizer.name(), @@ -453,34 +518,29 @@ private boolean runSynchronizers() { } break; } - // We have been requested to fall back to FDv1. Block every FDv2 - // synchronizer in one shot via fdv1Fallback() (which also unblocks the - // FDv1 fallback synchronizer, if one is configured). If FDv1 is - // configured we hand off to it; otherwise we halt the data system. - // An FDv1 fallback synchronizer asking to fall back again is ignored - // -- shouldn't happen in practice. - if (result.isFdv1Fallback() - && !sourceManager.isCurrentSynchronizerFDv1Fallback()) { - sourceManager.fdv1Fallback(); - if (sourceManager.hasFDv1Fallback()) { - logger.info("Falling back to an FDv1 fallback synchronizer."); - running = false; + // We have been requested to fall back to FDv1. Stop every FDv2 + // synchronizer until the directive's TTL elapses. If FDv1 is configured + // we hand off to it; otherwise no synchronizer runs until then. A + // directive received by the FDv1 fallback synchronizer itself replaces + // the TTL in effect without restarting it; the FDv1FallbackExpiryCondition + // reschedules itself when informed of the same result. + if (result.isFdv1Fallback()) { + if (sourceManager.isCurrentSynchronizerFDv1Fallback()) { + fdv1FallbackExpiresAtNanos = System.nanoTime() + + FDv1FallbackDirective.ttlOrDefault(result.getFdv1FallbackTtl()).toNanos(); } else { - // When the directive is signalled but no FDv1 fallback synchronizer - // is configured, halt the data system entirely. Surface OFF with - // the most recent error info (if any) and exit the synchronizer - // loop terminally. - logger.warn( - "Synchronizer '{}' requested FDv1 fallback, but no FDv1 fallback synchronizer is configured; halting the data system.", - synchronizer.name() - ); - DataSourceStatusProvider.ErrorInfo offError = - result.getStatus() != null ? result.getStatus().getErrorInfo() : null; - dataSourceUpdates.updateStatus( - DataSourceStatusProvider.State.OFF, - offError); - startFuture.complete(false); - return true; + if (sourceManager.hasFDv1Fallback()) { + logger.info("Falling back to an FDv1 fallback synchronizer."); + } else { + logger.warn( + "Synchronizer '{}' requested FDv1 fallback, but no FDv1 fallback synchronizer is configured.", + synchronizer.name() + ); + } + engageFDv1Fallback( + FDv1FallbackDirective.ttlOrDefault(result.getFdv1FallbackTtl()), + result.getStatus() != null ? result.getStatus().getErrorInfo() : null); + running = false; } } } @@ -501,7 +561,7 @@ private boolean runSynchronizers() { // Move to the next synchronizer. } // Get the next available synchronizer and set it active - synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive(); + synchronizer = getNextSynchronizer(); } if (!closed) { logger.warn("No more synchronizers available."); @@ -512,7 +572,6 @@ private boolean runSynchronizers() { } finally { sourceManager.close(); } - return false; } private static String detailForError(DataSourceStatusProvider.ErrorInfo errorInfo) { @@ -564,6 +623,7 @@ public boolean isInitialized() { @Override public void close() { closed = true; + closedSignal.complete(null); // If there is an active source, we will shut it down, and that will result in the loop handling that source // exiting. // If we do not have an active source, then the loop will check isShutdown when attempting to set one. When diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSourceConditions.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSourceConditions.java index 30f81b28..a8b5d954 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSourceConditions.java +++ b/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 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. + *

+ * This is package-private, instead of private, for ease of testing. + */ + static class FDv1FallbackExpiryCondition implements Condition { + private final CompletableFuture 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 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; + } + } } diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/HeaderConstants.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/HeaderConstants.java index 66ad1495..00e6095b 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/HeaderConstants.java +++ b/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; diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingBase.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingBase.java index c21bd5ba..e2d5adc9 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingBase.java +++ b/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 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 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: { diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/SourceManager.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/SourceManager.java index dd093d9a..cbf66190 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/SourceManager.java +++ b/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 suspendedForFDv1Fallback = new ArrayList<>(); + public SourceManager(List synchronizers, List> 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; } diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java index 5babb2d4..246378df 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java +++ b/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())); } } diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/datasources/FDv2SourceResult.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/datasources/FDv2SourceResult.java index 32b6c72b..d916bb28 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/datasources/FDv2SourceResult.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/datasources/FDv2SourceResult.java @@ -10,6 +10,7 @@ import com.launchdarkly.sdk.server.subsystems.DataStoreTypes.KeyedItems; import java.io.Closeable; +import java.time.Duration; import java.util.Map; import java.util.function.Function; @@ -72,6 +73,7 @@ public String getReason() { private final Status status; private final SourceResultType resultType; private final boolean fdv1Fallback; + private final Duration fdv1FallbackTtl; private final Function completionCallback; private FDv2SourceResult( @@ -79,11 +81,22 @@ private FDv2SourceResult( Status status, SourceResultType resultType, boolean fdv1Fallback, Function completionCallback + ) { + this(changeSet, status, resultType, fdv1Fallback, null, completionCallback); + } + + private FDv2SourceResult( + ChangeSet>>> changeSet, + Status status, SourceResultType resultType, + boolean fdv1Fallback, + Duration fdv1FallbackTtl, + Function completionCallback ) { this.changeSet = changeSet; this.status = status; this.resultType = resultType; this.fdv1Fallback = fdv1Fallback; + this.fdv1FallbackTtl = fdv1FallbackTtl; this.completionCallback = completionCallback; } @@ -100,6 +113,16 @@ public static FDv2SourceResult interrupted(DataSourceStatusProvider.ErrorInfo er completionCallback); } + /** + * @param errorInfo the error + * @param fdv1FallbackTtl the TTL of the FDv1 fallback directive carried by the result, or null if there is none + * @return the result + */ + public static FDv2SourceResult interrupted(DataSourceStatusProvider.ErrorInfo errorInfo, Duration fdv1FallbackTtl) { + return new FDv2SourceResult(null, Status.interrupted(errorInfo), SourceResultType.STATUS, + fdv1FallbackTtl != null, fdv1FallbackTtl, null); + } + public static FDv2SourceResult shutdown() { return shutdown(null); } @@ -124,6 +147,16 @@ public static FDv2SourceResult terminalError(DataSourceStatusProvider.ErrorInfo completionCallback); } + /** + * @param errorInfo the error + * @param fdv1FallbackTtl the TTL of the FDv1 fallback directive carried by the result, or null if there is none + * @return the result + */ + public static FDv2SourceResult terminalError(DataSourceStatusProvider.ErrorInfo errorInfo, Duration fdv1FallbackTtl) { + return new FDv2SourceResult(null, Status.terminalError(errorInfo), SourceResultType.STATUS, + fdv1FallbackTtl != null, fdv1FallbackTtl, null); + } + public static FDv2SourceResult changeSet(ChangeSet>>> changeSet, boolean fdv1Fallback) { return changeSet(changeSet, fdv1Fallback, null); } @@ -137,6 +170,16 @@ public static FDv2SourceResult changeSet(ChangeSet>>> changeSet, Duration fdv1FallbackTtl) { + return new FDv2SourceResult(changeSet, null, SourceResultType.CHANGE_SET, + fdv1FallbackTtl != null, fdv1FallbackTtl, null); + } + public static FDv2SourceResult goodbye(String reason, boolean fdv1Fallback) { return goodbye(reason, fdv1Fallback, null); } @@ -150,6 +193,16 @@ public static FDv2SourceResult goodbye(String reason, boolean fdv1Fallback, Func completionCallback); } + /** + * @param reason the reason for the goodbye + * @param fdv1FallbackTtl the TTL of the FDv1 fallback directive carried by the result, or null if there is none + * @return the result + */ + public static FDv2SourceResult goodbye(String reason, Duration fdv1FallbackTtl) { + return new FDv2SourceResult(null, Status.goodbye(reason), SourceResultType.STATUS, + fdv1FallbackTtl != null, fdv1FallbackTtl, null); + } + public SourceResultType getResultType() { return resultType; } @@ -166,6 +219,15 @@ public boolean isFdv1Fallback() { return fdv1Fallback; } + /** + * Returns the TTL of the FDv1 fallback directive carried by this result. + * + * @return the TTL, or null if there is no directive or the directive did not specify a TTL + */ + public Duration getFdv1FallbackTtl() { + return fdv1FallbackTtl; + } + /** * Creates a new result wrapping this one with an additional completion callback. *

@@ -183,7 +245,7 @@ public FDv2SourceResult withCompletion(Function newCallback) { } return null; }; - return new FDv2SourceResult(changeSet, status, resultType, fdv1Fallback, combinedCallback); + return new FDv2SourceResult(changeSet, status, resultType, fdv1Fallback, fdv1FallbackTtl, combinedCallback); } @Override diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv1FallbackDirectiveTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv1FallbackDirectiveTest.java new file mode 100644 index 00000000..1e95037b --- /dev/null +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv1FallbackDirectiveTest.java @@ -0,0 +1,95 @@ +package com.launchdarkly.sdk.server; + +import com.google.gson.JsonParser; +import com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event; + +import org.junit.Test; + +import java.time.Duration; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.allOf; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.lessThanOrEqualTo; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +@SuppressWarnings("javadoc") +public class FDv1FallbackDirectiveTest { + private static void assertJitteredDefault(Duration ttl) { + assertThat(ttl, allOf( + greaterThanOrEqualTo(Duration.ofMinutes(30)), + lessThanOrEqualTo(Duration.ofHours(1)))); + } + + private static FDv2Event goodbye(String json) { + return new FDv2Event("goodbye", JsonParser.parseString(json)); + } + + @Test + public void validServiceTtlIsUsedAsIs() { + assertEquals(Duration.ofMillis(1), FDv1FallbackDirective.ttlOrDefault(Duration.ofMillis(1))); + assertEquals(Duration.ofSeconds(90), FDv1FallbackDirective.ttlOrDefault(Duration.ofSeconds(90))); + assertEquals(Duration.ofHours(1), FDv1FallbackDirective.ttlOrDefault(Duration.ofHours(1))); + } + + @Test + public void invalidServiceTtlUsesJitteredDefault() { + assertJitteredDefault(FDv1FallbackDirective.ttlOrDefault(null)); + assertJitteredDefault(FDv1FallbackDirective.ttlOrDefault(Duration.ZERO)); + assertJitteredDefault(FDv1FallbackDirective.ttlOrDefault(Duration.ofSeconds(-1))); + assertJitteredDefault(FDv1FallbackDirective.ttlOrDefault(Duration.ofSeconds(3601))); + } + + @Test + public void defaultTtlIsJittered() { + for (int i = 0; i < 100; i++) { + assertJitteredDefault(FDv1FallbackDirective.ttlOrDefault(null)); + } + } + + @Test + public void headersWithoutFallbackHaveNoDirective() { + assertNull(FDv1FallbackDirective.fromHeaders(null, "60")); + assertNull(FDv1FallbackDirective.fromHeaders("false", "60")); + } + + @Test + public void headersWithFallbackUseValidTtlHeaderAsIs() { + assertEquals(Duration.ofSeconds(60), FDv1FallbackDirective.fromHeaders("true", "60")); + assertEquals(Duration.ofSeconds(60), FDv1FallbackDirective.fromHeaders("TRUE", "60")); + assertEquals(Duration.ofHours(1), FDv1FallbackDirective.fromHeaders("true", "3600")); + } + + @Test + public void headersWithFallbackAndInvalidTtlHeaderUseJitteredDefault() { + assertJitteredDefault(FDv1FallbackDirective.fromHeaders("true", null)); + assertJitteredDefault(FDv1FallbackDirective.fromHeaders("true", "")); + assertJitteredDefault(FDv1FallbackDirective.fromHeaders("true", "0")); + assertJitteredDefault(FDv1FallbackDirective.fromHeaders("true", "-5")); + assertJitteredDefault(FDv1FallbackDirective.fromHeaders("true", "3601")); + assertJitteredDefault(FDv1FallbackDirective.fromHeaders("true", "1.5")); + assertJitteredDefault(FDv1FallbackDirective.fromHeaders("true", "soon")); + assertJitteredDefault(FDv1FallbackDirective.fromHeaders("true", "99999999999999999999")); + } + + @Test + public void goodbyeWithoutProtocolFallbackTtlHasNoDirective() { + assertNull(FDv1FallbackDirective.fromGoodbye(goodbye("{\"reason\":\"bye\"}"))); + assertNull(FDv1FallbackDirective.fromGoodbye(goodbye("{\"reason\":\"bye\",\"protocolFallbackTTL\":null}"))); + } + + @Test + public void goodbyeWithValidProtocolFallbackTtlIsUsedAsIs() { + assertEquals(Duration.ofSeconds(120), + FDv1FallbackDirective.fromGoodbye(goodbye("{\"reason\":\"bye\",\"protocolFallbackTTL\":120}"))); + } + + @Test + public void goodbyeWithInvalidProtocolFallbackTtlUsesJitteredDefault() { + assertJitteredDefault(FDv1FallbackDirective.fromGoodbye(goodbye("{\"protocolFallbackTTL\":0}"))); + assertJitteredDefault(FDv1FallbackDirective.fromGoodbye(goodbye("{\"protocolFallbackTTL\":-1}"))); + assertJitteredDefault(FDv1FallbackDirective.fromGoodbye(goodbye("{\"protocolFallbackTTL\":7200}"))); + assertJitteredDefault(FDv1FallbackDirective.fromGoodbye(goodbye("{\"protocolFallbackTTL\":\"soon\"}"))); + } +} diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv2DataSourceTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv2DataSourceTest.java index d2015819..2aa570aa 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv2DataSourceTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv2DataSourceTest.java @@ -25,6 +25,7 @@ import org.junit.After; import org.junit.Test; +import java.time.Duration; import java.time.Instant; import java.util.ArrayList; import java.util.List; @@ -68,6 +69,10 @@ private ChangeSet>>> mak ); } + private static Duration shortFallbackTtl() { + return Duration.ofMillis(100); + } + private FDv2SourceResult makeInterruptedResult() { return FDv2SourceResult.interrupted( new DataSourceStatusProvider.ErrorInfo( @@ -2531,38 +2536,35 @@ public void fdv1FallbackWorksAfterInterruption() throws Exception { assertTrue("Should have at least 2 changesets", sink.getApplyCount() >= 2); } - // Synchronizer-phase analogue of the initializer halt path: when a synchronizer signals - // FDv1 fallback but no FDv1 fallback synchronizer is configured, the data system must halt - // (transition to OFF, stop building further synchronizers). + // When a synchronizer signals FDv1 fallback but no FDv1 fallback synchronizer is configured, no + // synchronizer runs until the directive's TTL elapses, the status is INTERRUPTED, and then the + // data source returns to the primary FDv2 synchronizer. @Test - public void fdv1FallbackOnSynchronizerWithoutFDv1ConfiguredHaltsDataSystem() throws Exception { + public void fdv1FallbackOnSynchronizerWithoutFDv1ConfiguredRecoversAfterTtl() throws Exception { executor = Executors.newScheduledThreadPool(2); MockDataSourceUpdateSink sink = new MockDataSourceUpdateSink(); ImmutableList> initializers = ImmutableList.of(); - // First synchronizer sends a normal payload, then a payload with the directive. - BlockingQueue firstSyncResults = new LinkedBlockingQueue<>(); - firstSyncResults.add(FDv2SourceResult.changeSet(makeChangeSet(false), false)); - firstSyncResults.add(FDv2SourceResult.changeSet(makeChangeSet(false), true)); + BlockingQueue primaryResults = new LinkedBlockingQueue<>(); + primaryResults.add(FDv2SourceResult.changeSet(makeChangeSet(true), false)); + primaryResults.add(FDv2SourceResult.changeSet(makeChangeSet(true), shortFallbackTtl())); - AtomicInteger firstSyncBuildCount = new AtomicInteger(0); - AtomicBoolean secondSyncCalled = new AtomicBoolean(false); + BlockingQueue primaryBuilds = new LinkedBlockingQueue<>(); + AtomicInteger primaryBuildCount = new AtomicInteger(0); + AtomicInteger secondaryBuildCount = new AtomicInteger(0); - // Two synchronizers configured; once the directive halts the data system, neither - // should be re-built. The second synchronizer in particular must never run. ImmutableList> synchronizers = ImmutableList.of( () -> { - firstSyncBuildCount.incrementAndGet(); - return new MockQueuedSynchronizer(firstSyncResults); + primaryBuilds.offer(primaryBuildCount.incrementAndGet()); + return new MockQueuedSynchronizer(primaryResults); }, () -> { - secondSyncCalled.set(true); + secondaryBuildCount.incrementAndGet(); return new MockQueuedSynchronizer(new LinkedBlockingQueue<>()); } ); - // No FDv1 fallback configured. FDv2DataSource dataSource = new FDv2DataSource( initializers, synchronizers, @@ -2577,33 +2579,15 @@ public void fdv1FallbackOnSynchronizerWithoutFDv1ConfiguredHaltsDataSystem() thr resourcesToClose.add(dataSource); dataSource.start().get(2, TimeUnit.SECONDS); + assertEquals(Integer.valueOf(1), primaryBuilds.poll(2, TimeUnit.SECONDS)); - // First payload is applied; the directive-bearing payload is also applied (the SDK - // surfaces the data) but no further synchronizers are dialled. - sink.awaitApplyCount(2, 2, TimeUnit.SECONDS); - - // Status must end up OFF because the directive could not be satisfied. - // Poll briefly for the OFF state in case the final transition is observed slightly - // after the apply count reaches 2; this keeps the assertion deterministic without - // relying on real-time delays. - DataSourceStatusProvider.State finalState = null; - for (int i = 0; i < 20 && finalState != DataSourceStatusProvider.State.OFF; i++) { - DataSourceStatusProvider.State next = sink.awaitStatus(100, TimeUnit.MILLISECONDS); - if (next != null) { - finalState = next; - } else if (sink.getLastState() == DataSourceStatusProvider.State.OFF) { - finalState = DataSourceStatusProvider.State.OFF; - } - } - assertEquals("Status should be OFF after directive halts the data system", - DataSourceStatusProvider.State.OFF, sink.getLastState()); + assertEquals(DataSourceStatusProvider.State.VALID, sink.awaitStatus(2, TimeUnit.SECONDS)); + assertEquals(DataSourceStatusProvider.State.VALID, sink.awaitStatus(2, TimeUnit.SECONDS)); + assertEquals(DataSourceStatusProvider.State.INTERRUPTED, sink.awaitStatus(2, TimeUnit.SECONDS)); + assertNotNull(sink.getLastError()); - // The second synchronizer must never be built. - assertFalse("Subsequent synchronizers must not run after directive-induced halt", - secondSyncCalled.get()); - // The first synchronizer should also not be re-built (we should stay halted). - assertEquals("First synchronizer should be built exactly once", - 1, firstSyncBuildCount.get()); + assertEquals(Integer.valueOf(2), primaryBuilds.poll(2, TimeUnit.SECONDS)); + assertEquals("Recovery should start from the primary synchronizer", 0, secondaryBuildCount.get()); } @Test @@ -2716,6 +2700,111 @@ public void fdv1FallbackOnlyCalledOncePerDataSource() throws Exception { assertNull("FDv1 fallback should only be called once", secondCall); } + @Test + public void fdv1FallbackRecoversToFDv2AfterTtl() throws Exception { + executor = Executors.newScheduledThreadPool(2); + MockDataSourceUpdateSink sink = new MockDataSourceUpdateSink(); + + ImmutableList> initializers = ImmutableList.of(); + + BlockingQueue primaryResults = new LinkedBlockingQueue<>(); + primaryResults.add(FDv2SourceResult.changeSet(makeChangeSet(true), shortFallbackTtl())); + + BlockingQueue primaryBuilds = new LinkedBlockingQueue<>(); + AtomicInteger primaryBuildCount = new AtomicInteger(0); + + ImmutableList> synchronizers = ImmutableList.of( + () -> { + primaryBuilds.offer(primaryBuildCount.incrementAndGet()); + return new MockQueuedSynchronizer(primaryResults); + } + ); + + BlockingQueue fdv1Built = new LinkedBlockingQueue<>(); + BlockingQueue fdv1Closed = new LinkedBlockingQueue<>(); + FDv2DataSource.DataSourceFactory fdv1Fallback = () -> { + fdv1Built.offer(true); + return new MockQueuedSynchronizer(new LinkedBlockingQueue<>()) { + @Override + public void close() { + fdv1Closed.offer(true); + super.close(); + } + }; + }; + + FDv2DataSource dataSource = new FDv2DataSource( + initializers, + synchronizers, + fdv1Fallback, + sink, + Thread.NORM_PRIORITY, + logger, + executor, + 120, + 300 + ); + resourcesToClose.add(dataSource); + + dataSource.start().get(2, TimeUnit.SECONDS); + + assertEquals(Integer.valueOf(1), primaryBuilds.poll(2, TimeUnit.SECONDS)); + assertNotNull("FDv1 fallback synchronizer should be activated", fdv1Built.poll(2, TimeUnit.SECONDS)); + assertEquals(Integer.valueOf(2), primaryBuilds.poll(2, TimeUnit.SECONDS)); + assertNotNull("FDv1 fallback synchronizer should be stopped on recovery", fdv1Closed.poll(2, TimeUnit.SECONDS)); + } + + // A directive received by the FDv1 fallback synchronizer replaces the TTL in effect, without + // restarting the FDv1 fallback synchronizer. + @Test + public void fdv1FallbackDirectiveFromFDv1SynchronizerReplacesTtl() throws Exception { + executor = Executors.newScheduledThreadPool(2); + MockDataSourceUpdateSink sink = new MockDataSourceUpdateSink(); + + ImmutableList> initializers = ImmutableList.of(); + + BlockingQueue primaryResults = new LinkedBlockingQueue<>(); + primaryResults.add(FDv2SourceResult.changeSet(makeChangeSet(true), Duration.ofHours(1))); + + BlockingQueue primaryBuilds = new LinkedBlockingQueue<>(); + AtomicInteger primaryBuildCount = new AtomicInteger(0); + + ImmutableList> synchronizers = ImmutableList.of( + () -> { + primaryBuilds.offer(primaryBuildCount.incrementAndGet()); + return new MockQueuedSynchronizer(primaryResults); + } + ); + + BlockingQueue fdv1Results = new LinkedBlockingQueue<>(); + fdv1Results.add(FDv2SourceResult.changeSet(makeChangeSet(true), shortFallbackTtl())); + + AtomicInteger fdv1BuildCount = new AtomicInteger(0); + FDv2DataSource.DataSourceFactory fdv1Fallback = () -> { + fdv1BuildCount.incrementAndGet(); + return new MockQueuedSynchronizer(fdv1Results); + }; + + FDv2DataSource dataSource = new FDv2DataSource( + initializers, + synchronizers, + fdv1Fallback, + sink, + Thread.NORM_PRIORITY, + logger, + executor, + 120, + 300 + ); + resourcesToClose.add(dataSource); + + dataSource.start().get(2, TimeUnit.SECONDS); + + assertEquals(Integer.valueOf(1), primaryBuilds.poll(2, TimeUnit.SECONDS)); + assertEquals(Integer.valueOf(2), primaryBuilds.poll(2, TimeUnit.SECONDS)); + assertEquals(1, fdv1BuildCount.get()); + } + // ============================================================================ // FDv1 Fallback (Initializer Phase) Tests // ============================================================================ @@ -2842,12 +2931,11 @@ public void fdv1FallbackOnInitializerErrorSwitchesToFDv1() throws Exception { fdv2SyncCalled.get()); } - // When an initializer signals FDv1 fallback but no FDv1 synchronizer is configured, the - // SDK must transition the data source status to OFF -- not stay stuck at INITIALIZING -- - // and surface the underlying initializer error so monitors can see why. + // When an initializer signals FDv1 fallback but no FDv1 synchronizer is configured, the status is + // INTERRUPTED with the initializer's error, and the FDv2 synchronizers run once the TTL elapses. @Test - public void fdv1FallbackOnInitializerWithoutFDv1ConfiguredTransitionsToOff() throws Exception { - executor = Executors.newScheduledThreadPool(1); + public void fdv1FallbackOnInitializerWithoutFDv1ConfiguredRecoversAfterTtl() throws Exception { + executor = Executors.newScheduledThreadPool(2); MockDataSourceUpdateSink sink = new MockDataSourceUpdateSink(); DataSourceStatusProvider.ErrorInfo initError = new DataSourceStatusProvider.ErrorInfo( @@ -2856,22 +2944,21 @@ public void fdv1FallbackOnInitializerWithoutFDv1ConfiguredTransitionsToOff() thr "fallback requested without fallback configured", Instant.now() ); + CompletableFuture initFuture = CompletableFuture.completedFuture( - FDv2SourceResult.terminalError(initError, true) + FDv2SourceResult.terminalError(initError, shortFallbackTtl()) ); ImmutableList> initializers = ImmutableList.of( () -> new MockInitializer(initFuture) ); - AtomicBoolean fdv2SyncCalled = new AtomicBoolean(false); + BlockingQueue syncResults = new LinkedBlockingQueue<>(); + syncResults.add(FDv2SourceResult.changeSet(makeChangeSet(true), false)); + ImmutableList> synchronizers = ImmutableList.of( - () -> { - fdv2SyncCalled.set(true); - return new MockQueuedSynchronizer(new LinkedBlockingQueue<>()); - } + () -> new MockQueuedSynchronizer(syncResults) ); - // No FDv1 fallback configured. FDv2DataSource dataSource = new FDv2DataSource( initializers, synchronizers, @@ -2885,16 +2972,15 @@ public void fdv1FallbackOnInitializerWithoutFDv1ConfiguredTransitionsToOff() thr ); resourcesToClose.add(dataSource); - dataSource.start().get(2, TimeUnit.SECONDS); + Future startFuture = dataSource.start(); - // Status must end up OFF (not INITIALIZING) so callers can observe the terminal state. - assertEquals(DataSourceStatusProvider.State.OFF, sink.getLastState()); - assertNotNull("Initializer error should be preserved on the OFF status", sink.getLastError()); - assertEquals(initError.getKind(), sink.getLastError().getKind()); - assertEquals(initError.getStatusCode(), sink.getLastError().getStatusCode()); + assertEquals(DataSourceStatusProvider.State.INTERRUPTED, sink.awaitStatus(2, TimeUnit.SECONDS)); + assertEquals(DataSourceStatusProvider.State.INTERRUPTED, sink.awaitStatus(2, TimeUnit.SECONDS)); + assertEquals(initError, sink.getLastError()); - // FDv2 synchronizers should not run. - assertFalse(fdv2SyncCalled.get()); + startFuture.get(2, TimeUnit.SECONDS); + assertTrue(dataSource.isInitialized()); + assertEquals(DataSourceStatusProvider.State.VALID, sink.getLastState()); } // When an initializer returns a successful payload-without-selector AND the FDv1 fallback diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingSynchronizerImplTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingSynchronizerImplTest.java index 5a8b5e79..a168f966 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingSynchronizerImplTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingSynchronizerImplTest.java @@ -919,6 +919,85 @@ public void fdv1FallbackFlagSetToTrueInSuccessResponse() throws Exception { } } + @Test + public void fdv1FallbackTtlReadFromHeader() throws Exception { + FDv2Requestor requestor = mockRequestor(); + SelectorSource selectorSource = mockSelectorSource(); + ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); + + okhttp3.Headers headers = new okhttp3.Headers.Builder() + .add("x-ld-fd-fallback", "true") + .add("x-ld-fd-fallback-ttl", "120") + .build(); + + FDv2Requestor.FDv2PayloadResponse response = FDv2Requestor.FDv2PayloadResponse.success( + com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray( + "{\"events\": [{\"event\": \"server-intent\", \"data\": {\"payloads\": [{\"id\": \"payload-1\", \"target\": 100, \"intentCode\": \"xfer-full\", \"reason\": \"payload-missing\"}]}}, {\"event\": \"payload-transferred\", \"data\": {\"state\": \"(p:payload-1:100)\", \"version\": 100}}]}" + ), + headers, + 200 + ); + + when(requestor.Poll(any(Selector.class))) + .thenReturn(CompletableFuture.completedFuture(response)); + + try { + PollingSynchronizerImpl synchronizer = new PollingSynchronizerImpl( + requestor, + testLogger, + selectorSource, + executor, + Duration.ofMillis(100) + ); + + FDv2SourceResult result = synchronizer.next().get(1, TimeUnit.SECONDS); + + assertEquals(SourceResultType.CHANGE_SET, result.getResultType()); + assertEquals(Duration.ofSeconds(120), result.getFdv1FallbackTtl()); + + synchronizer.close(); + } finally { + executor.shutdown(); + } + } + + @Test + public void fdv1FallbackTtlReadFromGoodbye() throws Exception { + FDv2Requestor requestor = mockRequestor(); + SelectorSource selectorSource = mockSelectorSource(); + ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); + + FDv2Requestor.FDv2PayloadResponse response = FDv2Requestor.FDv2PayloadResponse.success( + com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray( + "{\"events\": [{\"event\": \"goodbye\", \"data\": {\"reason\": \"fallback\", \"protocolFallbackTTL\": 120}}]}" + ), + new okhttp3.Headers.Builder().build(), + 200 + ); + + when(requestor.Poll(any(Selector.class))) + .thenReturn(CompletableFuture.completedFuture(response)); + + try { + PollingSynchronizerImpl synchronizer = new PollingSynchronizerImpl( + requestor, + testLogger, + selectorSource, + executor, + Duration.ofMillis(100) + ); + + FDv2SourceResult result = synchronizer.next().get(1, TimeUnit.SECONDS); + + assertEquals(SourceSignal.GOODBYE, result.getStatus().getState()); + assertEquals(Duration.ofSeconds(120), result.getFdv1FallbackTtl()); + + synchronizer.close(); + } finally { + executor.shutdown(); + } + } + @Test public void fdv1FallbackFlagSetToFalseWhenHeaderNotPresent() throws Exception { FDv2Requestor requestor = mockRequestor(); diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamingSynchronizerImplTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamingSynchronizerImplTest.java index 2ebd7a71..b1baddb2 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamingSynchronizerImplTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamingSynchronizerImplTest.java @@ -969,6 +969,76 @@ public void fdv1FallbackFlagSetToTrueInSuccessResponse() throws Exception { } } + @Test + public void fdv1FallbackTtlReadFromHeader() throws Exception { + String serverIntent = makeEvent("server-intent", "{\"payloads\":[{\"id\":\"payload-1\",\"target\":100,\"intentCode\":\"xfer-full\",\"reason\":\"payload-missing\"}]}"); + String payloadTransferred = makeEvent("payload-transferred", "{\"state\":\"(p:payload-1:100)\",\"version\":100}"); + + try (HttpServer server = HttpServer.start(Handlers.all( + Handlers.header("x-ld-fd-fallback", "true"), + Handlers.header("x-ld-fd-fallback-ttl", "120"), + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred), + Handlers.SSE.leaveOpen()))) { + + HttpProperties httpProperties = toHttpProperties(clientContext("sdk-key", baseConfig().build()).getHttp()); + SelectorSource selectorSource = mockSelectorSource(); + + StreamingSynchronizerImpl synchronizer = new StreamingSynchronizerImpl( + httpProperties, + server.getUri(), + "/stream", + testLogger, + selectorSource, + null, + Duration.ofMillis(100), + Thread.NORM_PRIORITY, + null + ); + + FDv2SourceResult result = synchronizer.next().get(5, TimeUnit.SECONDS); + + assertEquals(SourceResultType.CHANGE_SET, result.getResultType()); + assertEquals(Duration.ofSeconds(120), result.getFdv1FallbackTtl()); + + synchronizer.close(); + } + } + + @Test + public void fdv1FallbackTtlReadFromGoodbye() throws Exception { + String goodbyeEvent = makeEvent("goodbye", "{\"reason\":\"fallback\",\"protocolFallbackTTL\":120}"); + + try (HttpServer server = HttpServer.start(Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(goodbyeEvent), + Handlers.SSE.leaveOpen()))) { + + HttpProperties httpProperties = toHttpProperties(clientContext("sdk-key", baseConfig().build()).getHttp()); + SelectorSource selectorSource = mockSelectorSource(); + + StreamingSynchronizerImpl synchronizer = new StreamingSynchronizerImpl( + httpProperties, + server.getUri(), + "/stream", + testLogger, + selectorSource, + null, + Duration.ofMillis(100), + Thread.NORM_PRIORITY, + null + ); + + FDv2SourceResult result = synchronizer.next().get(5, TimeUnit.SECONDS); + + assertEquals(SourceSignal.GOODBYE, result.getStatus().getState()); + assertEquals(Duration.ofSeconds(120), result.getFdv1FallbackTtl()); + + synchronizer.close(); + } + } + @Test public void fdv1FallbackFlagSetToFalseWhenHeaderNotPresent() throws Exception { String serverIntent = makeEvent("server-intent", "{\"payloads\":[{\"id\":\"payload-1\",\"target\":100,\"intentCode\":\"xfer-full\",\"reason\":\"payload-missing\"}]}");