diff --git a/lib/sdk/server/contract-tests/test-suppressions-fdv2.txt b/lib/sdk/server/contract-tests/test-suppressions-fdv2.txt index 9765a098..e69de29b 100644 --- a/lib/sdk/server/contract-tests/test-suppressions-fdv2.txt +++ b/lib/sdk/server/contract-tests/test-suppressions-fdv2.txt @@ -1,3 +0,0 @@ -streaming/fdv2/recoverable fallback to secondary synchronizer -streaming/fdv2/recoverable fallback with recovery -streaming/fdv2/permanent fallback with recovery 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..99269541 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 @@ -37,6 +37,12 @@ class FDv2DataSource implements DataSource { */ private static final long defaultRecoveryTimeout = 5 * 60; + /** + * Default time to wait for a synchronizer to deliver data, before any data has been applied, before falling back + * to the next synchronizer. The timeout is only configurable for testing. + */ + private static final long defaultInitializationFallbackTimeoutSeconds = 10; + private final SourceManager sourceManager; private final List conditionFactories; @@ -45,6 +51,7 @@ class FDv2DataSource implements DataSource { private final CompletableFuture startFuture = new CompletableFuture<>(); private final AtomicBoolean started = new AtomicBoolean(false); + private final AtomicBoolean dataApplied = new AtomicBoolean(false); private final int threadPriority; @@ -94,6 +101,31 @@ public FDv2DataSource( ScheduledExecutorService sharedExecutor, long fallbackTimeout, long recoveryTimeout + ) { + this(initializers, + synchronizers, + fdv1DataSourceFactory, + dataSourceUpdates, + threadPriority, + logger, + sharedExecutor, + fallbackTimeout, + recoveryTimeout, + defaultInitializationFallbackTimeoutSeconds + ); + } + + public FDv2DataSource( + ImmutableList> initializers, + ImmutableList> synchronizers, + DataSourceFactory fdv1DataSourceFactory, + DataSourceUpdateSinkV2 dataSourceUpdates, + int threadPriority, + LDLogger logger, + ScheduledExecutorService sharedExecutor, + long fallbackTimeout, + long recoveryTimeout, + long initializationFallbackTimeout ) { List synchronizerFactories = synchronizers .stream() @@ -118,7 +150,8 @@ public FDv2DataSource( this.threadPriority = threadPriority; this.logger = logger; this.conditionFactories = new ArrayList<>(); - this.conditionFactories.add(new FallbackCondition.Factory(sharedExecutor, fallbackTimeout)); + this.conditionFactories.add(new FallbackCondition.Factory( + sharedExecutor, fallbackTimeout, initializationFallbackTimeout, dataApplied::get)); this.conditionFactories.add(new RecoveryCondition.Factory(sharedExecutor, recoveryTimeout)); } @@ -210,7 +243,9 @@ private InitializerOutcome runInitializers() { DataSourceStatusProvider.ErrorInfo fallbackErrorInfo = null; switch (result.getResultType()) { case CHANGE_SET: - dataSourceUpdates.apply(result.getChangeSet()); + if (dataSourceUpdates.apply(result.getChangeSet())) { + dataApplied.set(true); + } anyDataReceived = true; logger.info("Initialized via '{}'.", initializerName); if (!result.getChangeSet().getSelector().isEmpty()) { @@ -405,7 +440,9 @@ private boolean runSynchronizers() { case CHANGE_SET: // A data update breaks the "in a row" streak for status deduplication. resetSynchronizerStatusDedupe(); - dataSourceUpdates.apply(result.getChangeSet()); + if (dataSourceUpdates.apply(result.getChangeSet())) { + dataApplied.set(true); + } dataSourceUpdates.updateStatus(DataSourceStatusProvider.State.VALID, null); // This could have been completed by any data source. But if it has not been completed before // now, then we complete it. 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..43e861ba 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 @@ -10,6 +10,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; /** * Container class for FDv2 data source conditions and related types. @@ -101,13 +102,35 @@ public Factory(ScheduledExecutorService sharedExecutor, long timeout) { */ static class FallbackCondition extends TimedCondition { static class Factory extends TimedCondition.Factory { + private final long initializationTimeoutSeconds; + private final BooleanSupplier dataApplied; + public Factory(ScheduledExecutorService sharedExecutor, long timeout) { + this(sharedExecutor, timeout, 0, () -> true); + } + + /** + * @param initializationTimeoutSeconds how long a synchronizer may go without delivering data, while no + * data has been applied yet, before falling back + * @param dataApplied reports whether the data source has applied any data yet + */ + public Factory( + ScheduledExecutorService sharedExecutor, + long timeout, + long initializationTimeoutSeconds, + BooleanSupplier dataApplied + ) { super(sharedExecutor, timeout); + this.initializationTimeoutSeconds = initializationTimeoutSeconds; + this.dataApplied = dataApplied; } @Override public Condition build() { - return new FallbackCondition(sharedExecutor, timeoutSeconds); + if (dataApplied.getAsBoolean()) { + return new FallbackCondition(sharedExecutor, timeoutSeconds); + } + return new FallbackCondition(sharedExecutor, timeoutSeconds, initializationTimeoutSeconds); } @Override @@ -120,6 +143,23 @@ public FallbackCondition(ScheduledExecutorService sharedExecutor, long timeoutSe super(sharedExecutor, timeoutSeconds); } + /** + * Creates a fallback condition for use before any data has been applied. A synchronizer that never connects + * may not report any status for a long time, so the timer starts immediately instead of waiting for an + * interruption. Receiving a change set cancels it. + */ + public FallbackCondition( + ScheduledExecutorService sharedExecutor, + long timeoutSeconds, + long initializationTimeoutSeconds + ) { + super(sharedExecutor, timeoutSeconds); + timerFuture = sharedExecutor.schedule(() -> { + resultFuture.complete(this); + return null; + }, initializationTimeoutSeconds, TimeUnit.SECONDS); + } + @Override public void inform(FDv2SourceResult sourceResult) { if (sourceResult.getResultType() == SourceResultType.CHANGE_SET) { diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv2DataSourceFallbackConditionTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv2DataSourceFallbackConditionTest.java index 94bcf3bf..632de556 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv2DataSourceFallbackConditionTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FDv2DataSourceFallbackConditionTest.java @@ -120,6 +120,47 @@ public void changeSetCancelsActiveTimer() throws Exception { assertFalse(resultFuture.isDone()); } + @Test + public void initializationTimerCompletesResultFutureWithoutStatus() throws Exception { + executor = Executors.newScheduledThreadPool(1); + FallbackCondition condition = new FallbackCondition(executor, 120, 1); + + CompletableFuture resultFuture = condition.execute(); + assertFalse(resultFuture.isDone()); + + assertSame(condition, resultFuture.get(2, TimeUnit.SECONDS)); + } + + @Test + public void changeSetCancelsInitializationTimer() throws Exception { + executor = Executors.newScheduledThreadPool(1); + FallbackCondition condition = new FallbackCondition(executor, 120, 1); + + CompletableFuture resultFuture = condition.execute(); + condition.inform(FDv2SourceResult.changeSet(makeChangeSet(), false)); + + Thread.sleep(1500); + + assertFalse(resultFuture.isDone()); + } + + @Test + public void factoryOnlyUsesInitializationTimerBeforeDataIsApplied() throws Exception { + executor = Executors.newScheduledThreadPool(1); + java.util.concurrent.atomic.AtomicBoolean dataApplied = new java.util.concurrent.atomic.AtomicBoolean(false); + FallbackCondition.Factory factory = new FallbackCondition.Factory(executor, 120, 1, dataApplied::get); + + Condition beforeData = factory.build(); + dataApplied.set(true); + Condition afterData = factory.build(); + + assertSame(beforeData, beforeData.execute().get(2, TimeUnit.SECONDS)); + assertFalse(afterData.execute().isDone()); + + beforeData.close(); + afterData.close(); + } + @Test public void changeSetWithoutActiveTimerDoesNothing() throws Exception { executor = Executors.newScheduledThreadPool(1); 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..64942067 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 @@ -1656,6 +1656,69 @@ public void fallbackMovesToNextSynchronizer() throws Exception { assertNotNull("Second synchronizer should be called after fallback", secondCalled); } + @Test + public void silentSynchronizerFallsBackBeforeDataIsApplied() throws Exception { + executor = Executors.newScheduledThreadPool(2); + MockDataSourceUpdateSink sink = new MockDataSourceUpdateSink(); + + ImmutableList> initializers = ImmutableList.of(); + + // First fails permanently, second never reports anything, third serves data. + BlockingQueue firstSyncResults = new LinkedBlockingQueue<>(); + firstSyncResults.add(makeTerminalErrorResult()); + BlockingQueue thirdSyncResults = new LinkedBlockingQueue<>(); + thirdSyncResults.add(FDv2SourceResult.changeSet(makeChangeSet(true), false)); + + BlockingQueue thirdCalledQueue = new LinkedBlockingQueue<>(); + + ImmutableList> synchronizers = ImmutableList.of( + () -> new MockQueuedSynchronizer(firstSyncResults), + () -> new MockQueuedSynchronizer(new LinkedBlockingQueue<>()), + () -> { + thirdCalledQueue.offer(true); + return new MockQueuedSynchronizer(thirdSyncResults); + } + ); + + // Long steady-state fallback timeout, short initialization fallback timeout. + FDv2DataSource dataSource = new FDv2DataSource(initializers, synchronizers, null, sink, Thread.NORM_PRIORITY, logger, executor, 120, 300, 1); + resourcesToClose.add(dataSource); + + Future startFuture = dataSource.start(); + startFuture.get(5, TimeUnit.SECONDS); + + assertTrue(dataSource.isInitialized()); + assertNotNull("Third synchronizer should be called", thirdCalledQueue.poll(1, TimeUnit.SECONDS)); + } + + @Test + public void silentSynchronizerDoesNotFallBackQuicklyAfterDataIsApplied() throws Exception { + executor = Executors.newScheduledThreadPool(2); + MockDataSourceUpdateSink sink = new MockDataSourceUpdateSink(); + + ImmutableList> initializers = ImmutableList.of( + () -> new MockInitializer(CompletableFuture.completedFuture(FDv2SourceResult.changeSet(makeChangeSet(true), false))) + ); + + BlockingQueue secondCalledQueue = new LinkedBlockingQueue<>(); + + ImmutableList> synchronizers = ImmutableList.of( + () -> new MockQueuedSynchronizer(new LinkedBlockingQueue<>()), + () -> { + secondCalledQueue.offer(true); + return new MockQueuedSynchronizer(new LinkedBlockingQueue<>()); + } + ); + + FDv2DataSource dataSource = new FDv2DataSource(initializers, synchronizers, null, sink, Thread.NORM_PRIORITY, logger, executor, 120, 300, 1); + resourcesToClose.add(dataSource); + + Future startFuture = dataSource.start(); + startFuture.get(2, TimeUnit.SECONDS); + + assertNull("Second synchronizer should not be called", secondCalledQueue.poll(2, TimeUnit.SECONDS)); + } + // ============================================================================ // Condition Lifecycle // ============================================================================