Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,9 @@ public class TestService {
"server-side-polling",
"polling-gzip",
"fdv1-fallback",
"instance-id"
"instance-id",
"retry-conformance-fdv1-streaming",
"retry-conformance-fdv1-polling"
};

static final Gson gson = new GsonBuilder().serializeNulls().create();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,9 @@ public DataSource build(ClientContext context) {
streamUri,
payloadFilter,
initialReconnectDelay,
StreamProcessor.DEFAULT_EXTENDED_INITIAL_RECONNECT_DELAY,
StreamProcessor.DEFAULT_EXTENDED_STREAM_MAX_RETRY_DELAY,
StreamProcessor.DEFAULT_RETRY_RESET_INTERVAL,
logger);
}

Expand Down Expand Up @@ -196,6 +199,7 @@ public DataSource build(ClientContext context) {
context.getDataSourceUpdateSink(),
ClientContextImpl.get(context).sharedExecutor,
pollInterval,
PollingProcessor.DEFAULT_EXTENDED_INITIAL_DELAY,
logger);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ private static DataModel.Segment getSegment(DataStore store, String key) {
* constructor will not throw an exception for any error condition that could only be
* detected after making a request to LaunchDarkly (such as an SDK key that is simply
* wrong despite being valid ASCII, so it is invalid but not illegal); those are logged
* and treated as an unsuccessful initialization, as described above.
* and the SDK will keep retrying in the background as described above.
*
* @param sdkKey the SDK key for your LaunchDarkly environment
* @param config a client configuration object
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

import com.google.common.annotations.VisibleForTesting;
import com.launchdarkly.logging.LDLogger;
import com.launchdarkly.sdk.internal.http.FailureClass;
import com.launchdarkly.sdk.internal.http.HttpErrors;
import com.launchdarkly.sdk.internal.http.HttpErrors.HttpErrorException;
import com.launchdarkly.sdk.server.interfaces.DataSourceStatusProvider.ErrorInfo;
import com.launchdarkly.sdk.server.interfaces.DataSourceStatusProvider.ErrorKind;
Expand All @@ -21,33 +23,38 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.launchdarkly.sdk.internal.http.HttpErrors.checkIfErrorIsRecoverableAndLog;
import static com.launchdarkly.sdk.internal.http.HttpErrors.httpErrorDescription;

final class PollingProcessor implements DataSource {
private static final String ERROR_CONTEXT_MESSAGE = "on polling request";
private static final String WILL_RETRY_MESSAGE = "will retry at next scheduled poll interval";
static final Duration DEFAULT_EXTENDED_INITIAL_DELAY = Duration.ofMinutes(5);

@VisibleForTesting final FeatureRequestor requestor;
private final DataSourceUpdateSink dataSourceUpdates;
private final ScheduledExecutorService scheduler;
@VisibleForTesting final Duration pollInterval;
private final PollingStrategy strategy;
private final AtomicBoolean initialized = new AtomicBoolean(false);
// task tracks the currently pending poll; null when we haven't started yet
// or when we've been closed.
private ScheduledFuture<?> task;
// isClosed is set once in close().
private volatile boolean isClosed = false;
private final CompletableFuture<Void> initFuture;
private volatile ScheduledFuture<?> task;
private final LDLogger logger;

PollingProcessor(
FeatureRequestor requestor,
DataSourceUpdateSink dataSourceUpdates,
ScheduledExecutorService sharedExecutor,
Duration pollInterval,
Duration extendedInitialDelay,
LDLogger logger
) {
this.requestor = requestor; // note that HTTP configuration is applied to the requestor when it is created
this.dataSourceUpdates = dataSourceUpdates;
this.scheduler = sharedExecutor;
this.pollInterval = pollInterval;
this.strategy = new PollingStrategy(pollInterval, extendedInitialDelay);
this.initFuture = new CompletableFuture<>();
this.logger = logger;
}
Expand All @@ -59,34 +66,41 @@ public boolean isInitialized() {

@Override
public void close() throws IOException {
logger.info("Closing LaunchDarkly PollingProcessor");
requestor.close();

// Even though the shared executor will be shut down when the LDClient is closed, it's still good
// behavior to remove our polling task now - especially because we might be running in a test
// environment where there isn't actually an LDClient.
synchronized (this) {
if (isClosed) {
return;
}
isClosed = true;
if (task != null) {
task.cancel(true);
task = null;
}
}
logger.info("Closing LaunchDarkly PollingProcessor");
requestor.close();
}

@Override
public Future<Void> start() {
logger.info("Starting LaunchDarkly polling client with interval: {} milliseconds",
pollInterval.toMillis());

synchronized (this) {
if (task == null) {
task = scheduler.scheduleAtFixedRate(this::poll, 0L, pollInterval.toMillis(), TimeUnit.MILLISECONDS);
if (!isClosed && task == null) {
logger.info("Starting LaunchDarkly polling client with interval: {} milliseconds",
pollInterval.toMillis());
task = scheduler.schedule(this::poll, 0L, TimeUnit.MILLISECONDS);
}
}

return initFuture;
}


private void scheduleNext(Duration delay) {
synchronized (this) {
if (isClosed) {
return;
}
task = scheduler.schedule(this::poll, delay.toMillis(), TimeUnit.MILLISECONDS);
}
}

private void poll() {
try {
// If we already obtained data earlier, and the poll request returns a cached response, then we don't
Expand All @@ -106,30 +120,34 @@ private void poll() {
}
}
}
strategy.onSuccess();
} catch (HttpErrorException e) {
ErrorInfo errorInfo = ErrorInfo.fromHttpError(e.getStatus());
boolean recoverable = checkIfErrorIsRecoverableAndLog(logger, httpErrorDescription(e.getStatus()),
ERROR_CONTEXT_MESSAGE, e.getStatus(), WILL_RETRY_MESSAGE);
if (recoverable) {
dataSourceUpdates.updateStatus(State.INTERRUPTED, errorInfo);
} else {
dataSourceUpdates.updateStatus(State.OFF, errorInfo);
initFuture.complete(null); // if client is initializing, make it stop waiting; has no effect if already inited
if (task != null) {
task.cancel(true);
task = null;
}
FailureClass failureClass = HttpErrors.classifyAndLogHTTPFailure(
logger, e.getStatus(), ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE);
dataSourceUpdates.updateStatus(State.INTERRUPTED, ErrorInfo.fromHttpError(e.getStatus()));
if (strategy.onFailure(failureClass)) {
logger.info("Classified failure as UNEXPECTED; engaging extended backoff.");
}
} catch (IOException e) {
checkIfErrorIsRecoverableAndLog(logger, e.toString(), ERROR_CONTEXT_MESSAGE, 0, WILL_RETRY_MESSAGE);
FailureClass failureClass = HttpErrors.classifyAndLogTransportFailure(
logger, e, ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE);
dataSourceUpdates.updateStatus(State.INTERRUPTED, ErrorInfo.fromException(ErrorKind.NETWORK_ERROR, e));
if (strategy.onFailure(failureClass)) {
logger.info("Classified failure as UNEXPECTED; engaging extended backoff.");
}
} catch (SerializationException e) {
logger.error("Polling request received malformed data: {}", e.toString());
dataSourceUpdates.updateStatus(State.INTERRUPTED, ErrorInfo.fromException(ErrorKind.INVALID_DATA, e));
strategy.onFailure(FailureClass.NORMAL);
} catch (Exception e) {
logger.error("Unexpected error from polling processor: {}", e.toString());
logger.debug(e.toString(), e);
dataSourceUpdates.updateStatus(State.INTERRUPTED, ErrorInfo.fromException(ErrorKind.UNKNOWN, e));
strategy.onFailure(FailureClass.NORMAL);
} finally {
// Regardless of poll outcome, schedule the next attempt per strategy.
Duration wait = strategy.nextWait();
scheduleNext(wait);
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
package com.launchdarkly.sdk.server;

import com.launchdarkly.sdk.internal.http.FailureClass;

import java.time.Duration;
import java.util.Random;

/**
* Retry-timing state machine for the polling data source. Selects a per-attempt
* delay based on prior outcomes:
* <ul>
* <li>Normal regime: successive attempts wait {@code pollInterval}. No backoff
* is applied because {@code initialDelay} and {@code maxDelay} both equal
* {@code pollInterval}.</li>
* <li>Extended regime: entered on an {@link FailureClass#UNEXPECTED} failure.
* Waits start at {@code extendedInitialInterval} (floored at
* {@code pollInterval}) and double each attempt, clamped to
* {@link #EXTENDED_MAX_DELAY}.</li>
* <li>Healthy-op reset: two consecutive successful polls return the strategy
* to the normal regime.</li>
* </ul>
* <p>
* The formula input {@code n} in {@code T = initialDelay * 2^(n-1)} resets to
* zero whenever the delay bounds change (regime transition), so the first
* attempt in the new regime uses the new initial delay directly.
* <p>
* All state is owned by the polling loop's own thread (currently the shared
* ScheduledExecutorService in {@link PollingProcessor}). No external synchronization
* is required as long as this invariant holds.
*/
final class PollingStrategy {
static final Duration EXTENDED_MAX_DELAY = Duration.ofHours(1);

private final Duration normalInterval;
private final Duration extendedInitialInterval;
private final Random rng;

private int n;
private boolean priorPollWasSuccessful;
private boolean inExtended;
private Duration initialDelay;
private Duration maxDelay;

PollingStrategy(Duration normalInterval, Duration extendedInitialInterval) {
this(normalInterval, extendedInitialInterval, new Random());
}

// Visible for testing; deterministic seed injectable so jitter is reproducible.
PollingStrategy(Duration normalInterval, Duration extendedInitialInterval, Random rng) {
this.normalInterval = normalInterval;
this.extendedInitialInterval = extendedInitialInterval;
this.rng = rng;
// Normal regime at construction: both initialDelay and maxDelay equal the
// customer-configured pollInterval (there's no backoff in the normal
// regime — successive normal-failure retries stay at pollInterval).
this.initialDelay = normalInterval;
this.maxDelay = normalInterval;
}

/**
* Advance state after a poll failure. Returns {@code true} exactly once per
* transition from the normal regime into the extended regime; the caller
* can use the return value to emit an operator-visible log at the moment of
* transition without re-firing on every subsequent UNEXPECTED failure while
* already in extended regime.
* <p>
* On the transition, set {@code n = 1} and swap in the extended bounds, so
* the first extended wait uses {@code extendedInitialInterval} directly
* (the "reset n when delays change" invariant). On any other failure,
* increment n so the delay doubles.
* <p>
* Extended-regime bounds are floored at the customer-configured
* {@code pollInterval} — the wait never drops below that.
*/
boolean onFailure(FailureClass failureClass) {
this.priorPollWasSuccessful = false;
if (failureClass == FailureClass.UNEXPECTED && !this.inExtended) {
this.inExtended = true;
this.n = 1;
this.initialDelay = extendedInitialInterval;
if (this.initialDelay.compareTo(normalInterval) < 0) {
this.initialDelay = normalInterval;
}
this.maxDelay = EXTENDED_MAX_DELAY;
if (this.maxDelay.compareTo(normalInterval) < 0) {
this.maxDelay = normalInterval;
}
return true;
}
this.n++;
return false;
}

/**
* Advance state after a poll success. After two successes in a row, n resets
* to zero and delay bounds revert to the normal regime. A single success
* sets a "prior succeeded" flag; any intervening failure clears it.
* <p>
* The reset also clears {@code inExtended} so a subsequent UNEXPECTED
* failure re-transitions into the extended regime (with the transition
* detected exactly once, per {@link #onFailure(FailureClass)}'s contract).
*/
void onSuccess() {
if (this.priorPollWasSuccessful) {
this.n = 0;
this.inExtended = false;
this.initialDelay = normalInterval;
this.maxDelay = normalInterval;
}
this.priorPollWasSuccessful = true;
}

/**
* Compute the delay before the next poll attempt:
* {@code T = initialDelay * 2^(n-1)}, clamped to {@code maxDelay}. Jitter
* {@code J} is uniform in {@code [0, T/2]}. Final wait is
* {@code max(pollInterval, T - J)} — the wait never drops below the
* customer-configured {@code pollInterval}.
*/
Duration nextWait() {
if (this.n <= 0) {
return normalInterval;
}
long initialMs = initialDelay.toMillis();
long maxMs = maxDelay.toMillis();
double factor = Math.pow(2, this.n - 1);
long tMs = (long) Math.min(initialMs * factor, (double) maxMs);
long jitterMs = 0;
long halfT = tMs / 2;
if (halfT > 0) {
jitterMs = (rng.nextLong() % halfT + halfT) % halfT;
}
long waitMs = tMs - jitterMs;
long floorMs = normalInterval.toMillis();
if (waitMs < floorMs) {
waitMs = floorMs;
}
return Duration.ofMillis(waitMs);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Extended wait after poll success

Medium Severity

After a successful poll while still in the extended regime, onSuccess leaves n unchanged, so nextWait keeps returning the large extended backoff. The confirming second success can be delayed by up to an hour, which stalls recovery long after the data source is healthy again and conflicts with returning to the configured poll cadence.

Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 9394a65. Configure here.

}

// Accessors for observability / testing.

int getN() { return n; }
Duration getInitialDelay() { return initialDelay; }
Duration getMaxDelay() { return maxDelay; }
boolean getPriorPollWasSuccessful() { return priorPollWasSuccessful; }
}
Loading
Loading