Skip to content

Do not merge fix: stop the FDv2 synchronizer rotation from spinning (fixes testDebugUnitTest OOM) #384

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
abelonogov-ld wants to merge 2 commits into
base: main
Choose a base branch
from
+115 −24
Draft

Do not merge fix: stop the FDv2 synchronizer rotation from spinning (fixes testDebugUnitTest OOM) #384

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d28e121
fix: build a fresh synchronizer per factory call in recovery tests
abelonogov-ld Aug 12, 2026
69f18b0
fix: rate limit synchronizer rotation so dead sources cannot spin
abelonogov-ld Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 50 additions & 0 deletions ...hdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/FDv2DataSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

/**
Expand All @@ -54,6 +56,15 @@ public interface DataSourceFactory<T> {
private static final String INITIALIZER_CANCELLED = "Initializer cancelled: {}";
private static final String INITIALIZER_INTERRUPTED = "Initializer interrupted: {}";

/**
* A synchronizer session shorter than this never really connected to anything, so the rotation
* to the next synchronizer is paused rather than attempted immediately.
*/
private static final long MIN_SYNCHRONIZER_SESSION_MILLIS = 500;

/** Upper bound for the growing pause between synchronizer sessions that keep ending at once. */
private static final long MAX_ROTATION_PAUSE_MILLIS = 30_000;

private final List<DataSourceFactory<Initializer>> cacheInitializers;
private final SourceManager sourceManager;
private final long fallbackTimeoutSeconds;
Expand Down Expand Up @@ -510,13 +521,36 @@ private void maybeLogSynchronizerStatusChange(@Nullable String sourceName, @NonN
logger.info("Synchronizer '{}' reported status: {}.", sourceName, state.name());
}

/**
* Waits before building the next synchronizer. Waiting on {@link #shutdownCause} rather than
* sleeping means a {@link #stop(Callback)} during the pause is acted on right away.
*
* @return false if the data source shut down while waiting, in which case the caller must stop
*/
private boolean pauseBeforeRotation(long pauseMillis) {
logger.debug("Waiting {}ms before trying the next synchronizer.", pauseMillis);
try {
shutdownCause.get(pauseMillis, TimeUnit.MILLISECONDS);
return false;
} catch (TimeoutException e) {
return true;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
} catch (ExecutionException e) {
return false;
}
}

private void runSynchronizers(
@NonNull LDContext context,
@NonNull DataSourceUpdateSinkV2 sink
) {
long rotationPauseMillis = 0;
try {
Synchronizer synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive();
while (synchronizer != null) {
long sessionStartNanos = System.nanoTime();
String synchronizerName = synchronizer.name();
logger.info("Synchronizer '{}' is starting.", synchronizerName);
resetSynchronizerStatusDedupe();
Expand Down Expand Up @@ -651,6 +685,22 @@ private void runSynchronizers(
Thread.currentThread().interrupt();
return;
}

// A source that ends its session at once, such as one reporting SHUTDOWN as soon as
// it is built, must not be allowed to drive this rotation at CPU speed. The pause
// grows while sessions keep ending immediately and resets once one of them lasts.
if (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - sessionStartNanos)
< MIN_SYNCHRONIZER_SESSION_MILLIS) {
rotationPauseMillis = rotationPauseMillis == 0
? MIN_SYNCHRONIZER_SESSION_MILLIS
: Math.min(rotationPauseMillis * 2, MAX_ROTATION_PAUSE_MILLIS);
if (!pauseBeforeRotation(rotationPauseMillis)) {
return;
}
} else {
rotationPauseMillis = 0;
}

synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive();
}
if (!stopCalled.get()) {
Expand Down
89 changes: 65 additions & 24 deletions ...kly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/FDv2DataSourceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -517,22 +517,27 @@ public void emptyInitializerListSkipsToSynchronizers() throws Exception {

@Test
public void fallbackAndRecoveryTasksWellBehaved() throws Exception {
// First sync: changeset then INTERRUPTED; second sync: changeset; recovery brings back first
MockQueuedSynchronizer firstSync = new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false),
interrupted());
MockQueuedSynchronizer secondSync = new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false));

AtomicInteger firstCallCount = new AtomicInteger(0);
AtomicInteger secondCallCount = new AtomicInteger(0);
MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink();

// First sync: changeset then INTERRUPTED; second sync: changeset; recovery brings back first.
// Each factory must build a fresh synchronizer, because the data source closes the previous
// one when it switches, and a closed synchronizer only ever reports SHUTDOWN.
FDv2DataSource dataSource = buildDataSource(sink,
Collections.emptyList(),
Arrays.asList(
() -> { firstCallCount.incrementAndGet(); return firstSync; },
() -> { secondCallCount.incrementAndGet(); return secondSync; }),
() -> {
firstCallCount.incrementAndGet();
return new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false),
interrupted());
},
() -> {
secondCallCount.incrementAndGet();
return new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false));
}),
1, 2);

AwaitableCallback<Boolean> startCallback = startDataSource(dataSource);
Expand Down Expand Up @@ -663,17 +668,22 @@ public void recoveryResetsToFirstAvailableSynchronizer() throws Exception {
AtomicInteger firstCallCount = new AtomicInteger(0);
AtomicInteger secondCallCount = new AtomicInteger(0);

MockQueuedSynchronizer firstSync = new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false),
interrupted());
MockQueuedSynchronizer secondSync = new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false));

// Each factory must build a fresh synchronizer, because the data source closes the previous
// one when it switches, and a closed synchronizer only ever reports SHUTDOWN.
FDv2DataSource dataSource = buildDataSource(sink,
Collections.emptyList(),
Arrays.asList(
() -> { firstCallCount.incrementAndGet(); return firstSync; },
() -> { secondCallCount.incrementAndGet(); return secondSync; }),
() -> {
firstCallCount.incrementAndGet();
return new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false),
interrupted());
},
() -> {
secondCallCount.incrementAndGet();
return new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false));
}),
1, 2);

AwaitableCallback<Boolean> startCallback = startDataSource(dataSource);
Expand All @@ -687,6 +697,37 @@ public void recoveryResetsToFirstAvailableSynchronizer() throws Exception {
stopDataSource(dataSource);
}

@Test
public void synchronizersThatShutDownImmediatelyDoNotSpin() throws Exception {
MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink();
AtomicInteger buildCount = new AtomicInteger(0);

// Every session ends as soon as it starts, so the rotation has nothing to wait on and would
// run at CPU speed if it were not rate limited.
FDv2DataSource dataSource = buildDataSource(sink,
Collections.emptyList(),
Arrays.asList(
() -> {
buildCount.incrementAndGet();
return new MockQueuedSynchronizer(
FDv2SourceResult.status(FDv2SourceResult.Status.shutdown(), false));
},
() -> {
buildCount.incrementAndGet();
return new MockQueuedSynchronizer(
FDv2SourceResult.status(FDv2SourceResult.Status.shutdown(), false));
}));

startDataSource(dataSource);
Thread.sleep(1500);

// Pauses of 500ms, 1s, 2s and so on allow only a handful of attempts in this window.
int builds = buildCount.get();
assertTrue("expected rate limited rotation, but saw " + builds + " synchronizers built",
builds <= 6);
stopDataSource(dataSource);
}

@Test
public void fallbackMovesToNextSynchronizer() throws Exception {
MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink();
Expand Down Expand Up @@ -2024,16 +2065,16 @@ public void orchestrationLogging_fallback_logsInfo() throws Exception {
@Test
public void orchestrationLogging_recovery_logsInfo() throws Exception {
MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink();
MockQueuedSynchronizer firstSync = new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false),
interrupted());
MockQueuedSynchronizer secondSync = new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false));
// Fresh instances per build: the data source closes the previous synchronizer when it
// switches, and a closed synchronizer only ever reports SHUTDOWN.
FDv2DataSource dataSource = buildDataSource(sink,
Collections.emptyList(),
Arrays.asList(
() -> firstSync,
() -> secondSync),
() -> new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false),
interrupted()),
() -> new MockQueuedSynchronizer(
FDv2SourceResult.changeSet(makeChangeSet(false), false))),
1, 2);
AwaitableCallback<Boolean> startCallback = startDataSource(dataSource);
try {
Expand Down
Loading