From 6d0c138ec91ff1298b0d24ee6feedb8826e1d16b Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Mon, 24 Aug 2026 09:24:15 -0400 Subject: [PATCH 1/4] feat(server): implement RETRY-spec conformance in FDv1 streaming and polling data sources (SDK-2789) Guided by the server-sdk-guide.md in sdk-scratchpad; analogous to the Go server SDK's reference implementation. The behavioral change: HTTP responses that today cause a data source to permanently stop (notably 401, 403, other 4xx) and TLS/certificate validation failures are no longer terminal. Streaming enters an extended backoff regime (5 min -> 1 hour, doubling); polling continues at its configured cadence with extended-regime waits between failing polls. Recovery from either regime uses a healthy-operation reset (60 s of continuous connectivity for streaming; two consecutive successful polls for polling). Scope: FDv1 streaming and polling data sources under `lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/`. FDv2 is out of scope for this epic and is deferred to a future one; nothing in `datasourcev2/` or the DataSystem-related code paths is touched. The classifier this depends on (`FailureClass` + `HttpErrors.classify*`) lives in `launchdarkly-java-sdk-internal` and ships in its own PR. Highlights: - PollingStrategy: new state-machine encapsulation with onFailure(class) / onSuccess() / nextWait() methods. State: n (formula input), initialDelay, maxDelay, priorPollWasSuccessful. Wait floor: max(pollInterval, T - J). Two-consecutive-successes returns from extended to normal regime. - PollingProcessor: rewired to a self-driven loop using strategy.nextWait(). Removed the State.OFF permanent-stop path entirely; state stays INITIALIZING/INTERRUPTED with a lastError. - StreamProcessor: consumes okhttp-eventsource's new multi-strategy retry API (see launchdarkly/okhttp-eventsource#110). On UNEXPECTED classification, activates the extended-regime RetryDelayStrategy on the underlying EventSource; the library's built-in healthy-op reset returns to normal-regime timing after 60 s of continuous connectivity. - Constructor plumbing: PollingProcessor and StreamProcessor take extendedInitialReconnectDelay, extendedStreamMaxRetryDelay, retryResetInterval, and extendedInitialDelay as constructor parameters; package-private defaults threaded through ComponentsImpl. - DataSourceStatusProvider Javadocs: State.INITIALIZING, State.OFF, State.INTERRUPTED, and getStateSince OFF-case updated to reflect the new semantics (no HTTP-error -> OFF transition). - LDClient constructor Javadoc: describes an SDK-key rejection as ongoing background retry rather than an "unsuccessful initialization" that reads as terminal. - Contract test service: declares retry-conformance-fdv1-streaming and retry-conformance-fdv1-polling capabilities. Tests: - Unit tests: full test suite green. New coverage for the strategy state machine (PollingStrategyTest) and extended-regime timing observation in StreamProcessorTest. Existing 401/403 tests rewritten to assert extended-regime retry rather than permanent stop. - Contract tests via sdk-test-harness PR #404 (RETRY-conformance tests): 7/7 parallel shards pass end-to-end at production timing (5-minute extended-initial-delay), ~12 min wall clock. CI: intentionally red on this PR until launchdarkly/okhttp-eventsource#110 releases okhttp-eventsource 5.0.0 and launchdarkly/java-core#204 releases launchdarkly-java-sdk-internal 1.11.0. The multi-strategy retry API this SDK relies on is only in that eventsource PR's branch, and the classifier helpers are only in that internal-artifact PR's branch. Once both are released, bump both versions in lib/sdk/server/build.gradle. --- .../src/main/java/sdktest/TestService.java | 4 +- .../sdk/server/ComponentsImpl.java | 4 + .../com/launchdarkly/sdk/server/LDClient.java | 2 +- .../sdk/server/PollingProcessor.java | 78 +++-- .../sdk/server/PollingStrategy.java | 147 +++++++++ .../sdk/server/StreamProcessor.java | 111 ++++--- .../interfaces/DataSourceStatusProvider.java | 16 +- .../sdk/server/LDClientEndToEndTest.java | 48 +-- .../sdk/server/PollingProcessorTest.java | 106 +++--- .../sdk/server/PollingStrategyTest.java | 185 +++++++++++ .../sdk/server/StreamProcessorTest.java | 304 +++++++++++++++++- 11 files changed, 846 insertions(+), 159 deletions(-) create mode 100644 lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingStrategy.java create mode 100644 lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingStrategyTest.java diff --git a/lib/sdk/server/contract-tests/service/src/main/java/sdktest/TestService.java b/lib/sdk/server/contract-tests/service/src/main/java/sdktest/TestService.java index faf35246..77d48cda 100644 --- a/lib/sdk/server/contract-tests/service/src/main/java/sdktest/TestService.java +++ b/lib/sdk/server/contract-tests/service/src/main/java/sdktest/TestService.java @@ -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(); diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/ComponentsImpl.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/ComponentsImpl.java index e861f733..f808ebf7 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/ComponentsImpl.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/ComponentsImpl.java @@ -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); } @@ -196,6 +199,7 @@ public DataSource build(ClientContext context) { context.getDataSourceUpdateSink(), ClientContextImpl.get(context).sharedExecutor, pollInterval, + PollingProcessor.DEFAULT_EXTENDED_INITIAL_DELAY, logger); } diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/LDClient.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/LDClient.java index ba0e4c93..b39b5cbe 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/LDClient.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/LDClient.java @@ -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 diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java index 99d63b55..17cea33c 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java @@ -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; @@ -21,20 +23,23 @@ 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 initFuture; - private volatile ScheduledFuture task; private final LDLogger logger; PollingProcessor( @@ -42,12 +47,14 @@ final class PollingProcessor implements DataSource { 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; } @@ -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 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 @@ -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); } } } diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingStrategy.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingStrategy.java new file mode 100644 index 00000000..bf9191f5 --- /dev/null +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingStrategy.java @@ -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: + *
    + *
  • Normal regime: successive attempts wait {@code pollInterval}. No backoff + * is applied because {@code initialDelay} and {@code maxDelay} both equal + * {@code pollInterval}.
    • + *
  • 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}.
    • + *
  • Healthy-op reset: two consecutive successful polls return the strategy + * to the normal regime.
    • + *
+ *

+ * 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. + *

+ * 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. + *

+ * 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. + *

+ * 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. + *

+ * 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); + } + + // Accessors for observability / testing. + + int getN() { return n; } + Duration getInitialDelay() { return initialDelay; } + Duration getMaxDelay() { return maxDelay; } + boolean getPriorPollWasSuccessful() { return priorPollWasSuccessful; } +} diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java index 773ae7e8..51413041 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java @@ -9,6 +9,7 @@ import com.launchdarkly.eventsource.FaultEvent; import com.launchdarkly.eventsource.HttpConnectStrategy; import com.launchdarkly.eventsource.MessageEvent; +import com.launchdarkly.eventsource.RetryDelayStrategy; import com.launchdarkly.eventsource.StreamClosedByCallerException; import com.launchdarkly.eventsource.StreamClosedByServerException; import com.launchdarkly.eventsource.StreamClosedWithIncompleteMessageException; @@ -19,7 +20,9 @@ import com.launchdarkly.logging.LDLogger; import com.launchdarkly.logging.LogValues; import com.launchdarkly.sdk.internal.events.DiagnosticStore; +import com.launchdarkly.sdk.internal.http.FailureClass; import com.launchdarkly.sdk.internal.http.HttpConsts; +import com.launchdarkly.sdk.internal.http.HttpErrors; import com.launchdarkly.sdk.internal.http.HttpHelpers; import com.launchdarkly.sdk.internal.http.HttpProperties; import com.launchdarkly.sdk.server.StreamProcessorEvents.DeleteData; @@ -45,9 +48,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; -import static com.launchdarkly.sdk.internal.http.HttpErrors.checkIfErrorIsRecoverableAndLog; -import static com.launchdarkly.sdk.internal.http.HttpErrors.httpErrorDescription; - import okhttp3.Headers; /** @@ -67,12 +67,14 @@ * 2b. If the data store doesn't support status notifications (which is normally only true of the in-memory store) * then we don't know the significance of the error, but we must assume that updates have been lost, so we'll * restart the stream. - * 3. If we receive an unrecoverable error like HTTP 401, we close the stream and don't retry, and set the state - * to OFF. Any other HTTP error or network error causes a retry with backoff, with a state of INTERRUPTED. - * 4. We set the Future returned by start() to tell the client initialization logic that initialization has either - * succeeded (we got an initial payload and successfully stored it) or permanently failed (we got a 401, etc.). - * Otherwise, the client initialization method may time out but we will still be retrying in the background, and - * if we succeed then the client can detect that we're initialized now by calling our Initialized method. + * 3. HTTP-level and transport-level failures do not permanently stop the stream processor. + * Any HTTP error or network error causes a retry with backoff, with a state of INTERRUPTED. Failures classified + * as unexpected engage the extended-regime backoff via activateRetryDelayStrategy on the underlying EventSource; + * the library reverts to normal-regime backoff automatically after a healthy-op reset threshold of continuous + * connectivity. + * 4. We set the Future returned by start() to tell the client initialization logic that initialization has + * succeeded. Initialization failures do not permanently fail the SDK, the stream keeps retrying + * in the background. */ final class StreamProcessor implements DataSource { private static final String PUT = "put"; @@ -82,6 +84,12 @@ final class StreamProcessor implements DataSource { private static final String ERROR_CONTEXT_MESSAGE = "in stream connection"; private static final String WILL_RETRY_MESSAGE = "will retry"; + private static final Duration STREAM_MAX_RETRY_DELAY = Duration.ofSeconds(30); + // Package-private defaults so others can pass them as constructor arguments + static final Duration DEFAULT_EXTENDED_INITIAL_RECONNECT_DELAY = Duration.ofMinutes(5); + static final Duration DEFAULT_EXTENDED_STREAM_MAX_RETRY_DELAY = Duration.ofHours(1); + static final Duration DEFAULT_RETRY_RESET_INTERVAL = Duration.ofSeconds(60); + private final DataSourceUpdateSink dataSourceUpdates; private final HttpProperties httpProperties; private final Headers headers; @@ -89,10 +97,19 @@ final class StreamProcessor implements DataSource { final URI streamUri; @VisibleForTesting final Duration initialReconnectDelay; + private final Duration extendedInitialReconnectDelay; + private final Duration extendedStreamMaxRetryDelay; + private final Duration retryResetInterval; private final DiagnosticStore diagnosticAccumulator; private final int threadPriority; private final DataStoreStatusProvider.StatusListener statusListener; private volatile EventSource es; + // extendedRegime is the retry-delay strategy the SDK activates on the underlying + // EventSource when a failure is classified as unexpected. + private volatile RetryDelayStrategy extendedRegime; + // loggedActivatedExtended gates the "engaging extended backoff" info log so it + // fires at most once. + private volatile boolean loggedActivatedExtended = false; private final AtomicBoolean initialized = new AtomicBoolean(false); private final AtomicBoolean closed = new AtomicBoolean(false); private volatile long esStarted = 0; @@ -107,12 +124,18 @@ final class StreamProcessor implements DataSource { URI streamUri, String payloadFilter, Duration initialReconnectDelay, + Duration extendedInitialReconnectDelay, + Duration extendedStreamMaxRetryDelay, + Duration retryResetInterval, LDLogger logger) { this.dataSourceUpdates = dataSourceUpdates; this.httpProperties = httpProperties; this.diagnosticAccumulator = diagnosticAccumulator; this.threadPriority = threadPriority; this.initialReconnectDelay = initialReconnectDelay; + this.extendedInitialReconnectDelay = extendedInitialReconnectDelay; + this.extendedStreamMaxRetryDelay = extendedStreamMaxRetryDelay; + this.retryResetInterval = retryResetInterval; this.logger = logger; URI tempUri = HttpHelpers.concatenateUriPath(streamUri, StandardEndpoints.STREAMING_REQUEST_PATH); @@ -177,6 +200,16 @@ public Future start() { // Set readTimeout last, to ensure that this hard-coded value overrides any other read // timeout that might have been set by httpProperties (see comment about readTimeout above). .readTimeout(DEAD_CONNECTION_INTERVAL.toMillis(), TimeUnit.MILLISECONDS); + + RetryDelayStrategy normalRegime = RetryDelayStrategy.defaultStrategy() + .initialDelay(initialReconnectDelay.toMillis(), TimeUnit.MILLISECONDS) + .maxDelay(STREAM_MAX_RETRY_DELAY.toMillis(), TimeUnit.MILLISECONDS); + RetryDelayStrategy extendedRegime = RetryDelayStrategy.defaultStrategy() + .initialDelay(extendedInitialReconnectDelay.toMillis(), TimeUnit.MILLISECONDS) + .maxDelay(extendedStreamMaxRetryDelay.toMillis(), TimeUnit.MILLISECONDS); + this.extendedRegime = extendedRegime; + loggedActivatedExtended = false; + EventSource.Builder builder = new EventSource.Builder(eventSourceHttpConfig) .errorStrategy(ErrorStrategy.alwaysContinue()) // alwaysContinue means we want EventSource to give us a FaultEvent rather @@ -184,8 +217,10 @@ public Future start() { .logger(logger) .readBufferSize(5000) .streamEventData(true) - .expectFields("event") - .retryDelay(initialReconnectDelay.toMillis(), TimeUnit.MILLISECONDS); + .expectFields("event") + .retryDelayStrategy(normalRegime) // first call sets default + .retryDelayStrategy(extendedRegime) // subsequent call adds extended-regime + .retryDelayResetThreshold(retryResetInterval.toMillis(), TimeUnit.MILLISECONDS); es = builder.build(); Thread thread = new Thread(() -> { @@ -250,13 +285,13 @@ public boolean isInitialized() { return initialized.get(); } - // Handles a single StreamEvent and returns true if we should keep the stream alive, - // or false if we should shut down permanently. + // Handles a single StreamEvent. Returns true to keep the stream alive; returns false + // only after this StreamProcessor has been closed. private boolean handleEvent(StreamEvent event, CompletableFuture initFuture) { if (closed.get()) { return false; } - logger.debug("Received StreamEvent: {}", event); + logger.debug("Received StreamEvent: {}", event); if (event instanceof MessageEvent) { handleMessage((MessageEvent)event, initFuture); } else if (event instanceof FaultEvent) { @@ -367,38 +402,40 @@ private void handleDelete(Reader eventData) throws StreamInputException, StreamS } private boolean handleError(StreamException e, CompletableFuture initFuture) { - boolean streamFailed = true; - if (e instanceof StreamClosedByCallerException) { - // This indicates that we ourselves deliberately restarted the stream, so we don't - // treat that as a failure in our analytics. - streamFailed = false; - } else { - logger.warn("Encountered EventSource error: {}", LogValues.exceptionSummary(e)); + boolean streamFailed = !(e instanceof StreamClosedByCallerException); + if (streamFailed) { + logger.warn("Encountered EventSource error: {}", LogValues.exceptionSummary(e)); } recordStreamInit(streamFailed); - + + FailureClass failureClass; + ErrorInfo errorInfo; if (e instanceof StreamHttpErrorException) { int status = ((StreamHttpErrorException)e).getCode(); - ErrorInfo errorInfo = ErrorInfo.fromHttpError(status); + failureClass = HttpErrors.classifyAndLogHTTPFailure(logger, status, ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE); + errorInfo = ErrorInfo.fromHttpError(status); + } else if (e instanceof StreamIOException || e instanceof StreamClosedByServerException) { + failureClass = HttpErrors.classifyAndLogTransportFailure(logger, e, ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE); + errorInfo = ErrorInfo.fromException(ErrorKind.NETWORK_ERROR, e); + } else { + // StreamClosedByCallerException or any other exception: classify NORMAL + // and don't emit a separate classify-and-log line (either self-inflicted or unknown). + failureClass = FailureClass.NORMAL; + errorInfo = ErrorInfo.fromException(ErrorKind.UNKNOWN, e); + } - boolean recoverable = checkIfErrorIsRecoverableAndLog(logger, httpErrorDescription(status), - ERROR_CONTEXT_MESSAGE, status, WILL_RETRY_MESSAGE); - if (recoverable) { - dataSourceUpdates.updateStatus(State.INTERRUPTED, errorInfo); - esStarted = System.currentTimeMillis(); - return true; // allow reconnect - } else { - dataSourceUpdates.updateStatus(State.OFF, errorInfo); - initFuture.complete(null); // if client is initializing, make it stop waiting; has no effect if already inited - return false; // don't reconnect + // Transition into extended regime on UNEXPECTED classification. + if (failureClass == FailureClass.UNEXPECTED) { + es.activateRetryDelayStrategy(extendedRegime); + if (!loggedActivatedExtended) { + logger.info("Classified failure as UNEXPECTED; engaging extended backoff."); + loggedActivatedExtended = true; } } - boolean isNetworkError = e instanceof StreamIOException || e instanceof StreamClosedByServerException; - checkIfErrorIsRecoverableAndLog(logger, e.toString(), ERROR_CONTEXT_MESSAGE, 0, WILL_RETRY_MESSAGE); - ErrorInfo errorInfo = ErrorInfo.fromException(isNetworkError ? ErrorKind.NETWORK_ERROR : ErrorKind.UNKNOWN, e); dataSourceUpdates.updateStatus(State.INTERRUPTED, errorInfo); - return true; // allow reconnect + esStarted = System.currentTimeMillis(); + return true; // always try reconnect } private static T parseStreamJson(Function parser, Reader r) throws StreamInputException { diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/interfaces/DataSourceStatusProvider.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/interfaces/DataSourceStatusProvider.java index a3eab8a3..17439a19 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/interfaces/DataSourceStatusProvider.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/interfaces/DataSourceStatusProvider.java @@ -91,8 +91,8 @@ public enum State { * The initial state of the data source when the SDK is being initialized. *

* If it encounters an error that requires it to retry initialization, the state will remain at - * {@link #INITIALIZING} until it either succeeds and becomes {@link #VALID}, or permanently fails and - * becomes {@link #OFF}. + * {@link #INITIALIZING} until it either succeeds and becomes {@link #VALID}, or the datasource is + * shut down and it becomes {@link #OFF}. */ INITIALIZING, @@ -110,17 +110,16 @@ public enum State { * Indicates that the data source encountered an error that it will attempt to recover from. *

* In streaming mode, this means that the stream connection failed, or had to be dropped due to some - * other error, and will be retried after a backoff delay. In polling mode, it means that the last poll - * request failed, and a new poll request will be made after the configured polling interval. + * other error, and will be retried after a backoff delay. In polling mode, it means that the last + * poll request failed; the next poll will be scheduled at the configured polling interval (or in + * rare cases, an extended backoff). */ INTERRUPTED, /** * Indicates that the data source has been permanently shut down. *

- * This could be because it encountered an unrecoverable error (for instance, the LaunchDarkly service - * rejected the SDK key; an invalid SDK key will never become valid), or because the SDK client was - * explicitly shut down. + * This could be because the SDK client was explicitly shut down. */ OFF; } @@ -326,8 +325,7 @@ public State getState() { * state, after previously having been either {@link State#INITIALIZING} or {@link State#INTERRUPTED}. *

  • For {@link State#INTERRUPTED}, it is the time that the data source most recently entered an * error state, after previously having been {@link State#VALID}. - *
  • For {@link State#OFF}, it is the time that the data source encountered an unrecoverable error - * or that the SDK was explicitly shut down. + *
  • For {@link State#OFF}, it is the time that the data source stopped operation. * * * @return the timestamp of the last state change diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/LDClientEndToEndTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/LDClientEndToEndTest.java index 4ea95f92..5ff4d237 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/LDClientEndToEndTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/LDClientEndToEndTest.java @@ -29,6 +29,7 @@ import static com.launchdarkly.testhelpers.httptest.Handlers.bodyJson; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.not; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -100,22 +101,32 @@ public void clientStartsInPollingModeAfterRecoverableError() throws Exception { } } + // A 401 does not permanently stop polling; the SDK keeps retrying. We can't + // observe multiple extended-regime polls in a fast test because the extended + // initial delay is a fixed 5-minute default -- the key assertion is that at + // least one poll happened and the SDK did not transition to a permanent-off + // state. @Test - public void clientFailsInPollingModeWith401Error() throws Exception { + public void clientInPollingModeKeepsRetryingOn401Error() throws Exception { try (HttpServer server = HttpServer.start(makeInvalidSdkKeyResponse())) { LDConfig config = baseConfig() .serviceEndpoints(Components.serviceEndpoints().polling(server.getUri())) - .dataSource(Components.pollingDataSourceInternal() - .pollIntervalWithNoMinimum(Duration.ofMillis(5))) // use small interval so we'll know if it does not stop permanently + .dataSource(Components.pollingDataSource()) + .startWait(Duration.ofMillis(500)) .events(noEvents()) .build(); - + try (LDClient client = new LDClient(sdkKey, config)) { assertFalse(client.isInitialized()); assertFalse(client.boolVariation(flagKey, user, false)); - + + // State should NOT be OFF; the data source is still trying (waiting + // out the extended-regime backoff between polls). + assertThat(client.getDataSourceStatusProvider().getStatus().getState(), + not(equalTo(DataSourceStatusProvider.State.OFF))); + + // At least one request should have been made. server.getRecorder().requireRequest(); - server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); } } } @@ -174,35 +185,28 @@ public void clientStartsInStreamingModeAfterRecoverableError() throws Exception } } + // A 401 does not permanently stop streaming; the SDK engages extended-regime + // backoff and keeps retrying. @Test - public void clientFailsInStreamingModeWith401Error() throws Exception { + public void clientInStreamingModeKeepsRetryingOn401Error() throws Exception { try (HttpServer server = HttpServer.start(makeInvalidSdkKeyResponse())) { LDConfig config = baseConfig() .serviceEndpoints(Components.serviceEndpoints().streaming(server.getUri())) .dataSource(Components.streamingDataSource().initialReconnectDelay(Duration.ZERO)) - // use zero reconnect delay so we'll know if it does not stop permanently + .startWait(Duration.ofMillis(200)) .events(noEvents()) .build(); - + try (LDClient client = new LDClient(sdkKey, config)) { assertFalse(client.isInitialized()); assertFalse(client.boolVariation(flagKey, user, false)); - - BlockingQueue statuses = new LinkedBlockingQueue<>(); - client.getDataSourceStatusProvider().addStatusListener(statuses::add); - Thread.sleep(100); // make sure it didn't retry the connection + // State should NOT be OFF; the data source is still trying. assertThat(client.getDataSourceStatusProvider().getStatus().getState(), - equalTo(DataSourceStatusProvider.State.OFF)); - while (!statuses.isEmpty()) { - // The status listener may or may not have been registered early enough to receive - // the OFF notification, but we should at least not see any *other* statuses. - assertThat(statuses.take().getState(), equalTo(DataSourceStatusProvider.State.OFF)); - } - assertThat(statuses.isEmpty(), equalTo(true)); - + not(equalTo(DataSourceStatusProvider.State.OFF))); + + // At least one request should have been made. server.getRecorder().requireRequest(); - server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); } } } diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java index 5d73e2ae..73112c68 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java @@ -1,5 +1,7 @@ package com.launchdarkly.sdk.server; +import com.launchdarkly.logging.LDLogLevel; +import com.launchdarkly.logging.LogCapture; import com.launchdarkly.sdk.server.DataModel.FeatureFlag; import com.launchdarkly.sdk.server.DataStoreTestTypes.DataBuilder; import com.launchdarkly.sdk.server.TestComponents.MockDataSourceUpdates; @@ -68,8 +70,13 @@ public void setup() { } private PollingProcessor makeProcessor(URI baseUri, Duration pollInterval) { + return makeProcessor(baseUri, pollInterval, PollingProcessor.DEFAULT_EXTENDED_INITIAL_DELAY); + } + + private PollingProcessor makeProcessor(URI baseUri, Duration pollInterval, Duration extendedInitialDelay) { FeatureRequestor requestor = new DefaultFeatureRequestor(defaultHttpProperties(), baseUri, null, testLogger); - return new PollingProcessor(requestor, dataSourceUpdates, sharedExecutor, pollInterval, testLogger); + return new PollingProcessor( + requestor, dataSourceUpdates, sharedExecutor, pollInterval, extendedInitialDelay, testLogger); } private static class TestPollHandler implements Handler { @@ -258,14 +265,16 @@ public void http400ErrorIsRecoverable() throws Exception { testRecoverableHttpError(400); } + // 401 / 403 engage the extended-regime backoff and keep polling instead of + // triggering a permanent stop. @Test - public void http401ErrorIsUnrecoverable() throws Exception { - testUnrecoverableHttpError(401); + public void http401TriggersExtendedRegimeAndKeepsPolling() throws Exception { + testUnexpectedHttpErrorKeepsPolling(401); } @Test - public void http403ErrorIsUnrecoverable() throws Exception { - testUnrecoverableHttpError(403); + public void http403TriggersExtendedRegimeAndKeepsPolling() throws Exception { + testUnexpectedHttpErrorKeepsPolling(403); } @Test @@ -283,51 +292,47 @@ public void http500ErrorIsRecoverable() throws Exception { testRecoverableHttpError(500); } - private void testUnrecoverableHttpError(int statusCode) throws Exception { + private void testUnexpectedHttpErrorKeepsPolling(int statusCode) throws Exception { + // 401 / 403 (and other UNEXPECTED 4xx) engage extended-regime backoff via + // PollingStrategy but never trigger a permanent State.OFF. Use a small + // extendedInitialDelay so the extended-regime waits are observable at ms + // scale rather than the 5-minute production default. TestPollHandler handler = new TestPollHandler(); - - // Test a scenario where the very first request gets this error handler.setError(statusCode); + Duration extendedInitial = Duration.ofMillis(30); withStatusQueue(statuses -> { try (HttpServer server = HttpServer.start(handler)) { - try (PollingProcessor pollingProcessor = makeProcessor(server.getUri(), BRIEF_INTERVAL)) { - long startTime = System.currentTimeMillis(); - Future initFuture = pollingProcessor.start(); - - assertFutureIsCompleted(initFuture, 2, TimeUnit.SECONDS); - assertTrue((System.currentTimeMillis() - startTime) < 9000); - assertTrue(initFuture.isDone()); - assertFalse(pollingProcessor.isInitialized()); - - verifyHttpErrorCausedShutdown(statuses, statusCode); - + try (PollingProcessor pollingProcessor = makeProcessor(server.getUri(), BRIEF_INTERVAL, extendedInitial)) { + pollingProcessor.start(); + + // Should observe multiple requests as extended-regime backoff continues to retry. + server.getRecorder().requireRequest(); + server.getRecorder().requireRequest(); server.getRecorder().requireRequest(); - server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); - } - } - }); - - // Now test a scenario where we have a successful startup, but a subsequent poll gets the error - handler.setError(0); - dataSourceUpdates = TestComponents.dataSourceUpdates(new InMemoryDataStore(), new MockDataStoreStatusProvider()); - withStatusQueue(statuses -> { - try (HttpServer server = HttpServer.start(handler)) { - try (PollingProcessor pollingProcessor = makeProcessor(server.getUri(), BRIEF_INTERVAL)) { - Future initFuture = pollingProcessor.start(); - - assertFutureIsCompleted(initFuture, 2, TimeUnit.SECONDS); - assertTrue(initFuture.isDone()); - assertTrue(pollingProcessor.isInitialized()); - requireDataSourceStatus(statuses, State.VALID); - // now make it so polls fail - handler.setError(statusCode); - - verifyHttpErrorCausedShutdown(statuses, statusCode); - while (server.getRecorder().count() > 0) { - server.getRecorder().requireRequest(); + // State stays INITIALIZING (never got past init because every response + // was an error) with an ERROR_RESPONSE lastError. The processor does + // not transition to OFF; it keeps retrying under extended-regime backoff. + Status status = requireDataSourceStatus(statuses, State.INITIALIZING); + assertNotNull(status.getLastError()); + assertEquals(ErrorKind.ERROR_RESPONSE, status.getLastError().getKind()); + assertEquals(statusCode, status.getLastError().getStatusCode()); + assertFalse(pollingProcessor.isInitialized()); + + // Unexpected classifications log at Error level. The SDK-emitted classify- + // and-log line is distinguished by the "Error on polling request" prefix. + boolean sawErrorForStatus = false; + for (LogCapture.Message m : logCapture.getMessages()) { + if (m.getText().startsWith("Error on polling request") + && m.getText().contains("HTTP error " + statusCode)) { + assertThat( + "unexpected-classification HTTP error should log at Error, not " + m.getLevel(), + m.getLevel(), equalTo(LDLogLevel.ERROR)); + sawErrorForStatus = true; + } } - server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); + assertTrue("expected an Error-level SDK log mentioning HTTP error " + statusCode, + sawErrorForStatus); } } }); @@ -365,6 +370,21 @@ private void testRecoverableHttpError(int statusCode) throws Exception { assertEquals(ErrorKind.ERROR_RESPONSE, status0.getLastError().getKind()); assertEquals(statusCode, status0.getLastError().getStatusCode()); + // Normal classifications log at Warn level (not Error). The SDK-emitted + // classify-and-log line is distinguished by the "Error on polling request" prefix. + boolean sawWarnForStatus = false; + for (LogCapture.Message m : logCapture.getMessages()) { + if (m.getText().startsWith("Error on polling request") + && m.getText().contains("HTTP error " + statusCode)) { + assertThat( + "normal-classification HTTP error should log at Warn, not " + m.getLevel(), + m.getLevel(), equalTo(LDLogLevel.WARN)); + sawWarnForStatus = true; + } + } + assertTrue("expected a Warn-level SDK log mentioning HTTP error " + statusCode, + sawWarnForStatus); + // and then that it succeeded requireDataSourceStatusEventually(statuses, State.VALID, State.INITIALIZING); } diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingStrategyTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingStrategyTest.java new file mode 100644 index 00000000..1ff16f86 --- /dev/null +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingStrategyTest.java @@ -0,0 +1,185 @@ +package com.launchdarkly.sdk.server; + +import com.launchdarkly.sdk.internal.http.FailureClass; + +import org.junit.Test; + +import java.time.Duration; +import java.util.Random; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.lessThanOrEqualTo; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Unit coverage for the {@link PollingStrategy} state machine. Uses ms-scale + * numbers so tests are fast; the ratios match production's minute-scale + * extended-regime targets. + */ +@SuppressWarnings("javadoc") +public class PollingStrategyTest { + private static final Duration NORMAL = Duration.ofMillis(100); + private static final Duration EXTENDED_INITIAL = Duration.ofMillis(500); + + private PollingStrategy strategy() { + // Deterministic seed so jitter is reproducible in tests. + return new PollingStrategy(NORMAL, EXTENDED_INITIAL, new Random(42L)); + } + + @Test + public void freshStrategyReturnsNormalIntervalOnFirstWait() { + PollingStrategy s = strategy(); + assertThat(s.nextWait(), equalTo(NORMAL)); + assertThat(s.getInitialDelay(), equalTo(NORMAL)); + assertThat(s.getMaxDelay(), equalTo(NORMAL)); + } + + @Test + public void normalFailuresDoNotChangeInitialDelay() { + PollingStrategy s = strategy(); + s.onFailure(FailureClass.NORMAL); + s.onFailure(FailureClass.NORMAL); + s.onFailure(FailureClass.NORMAL); + // initialDelay stays at pollInterval; maxDelay stays at pollInterval; + // so nextWait always == normalInterval. + assertThat(s.getInitialDelay(), equalTo(NORMAL)); + assertThat(s.getMaxDelay(), equalTo(NORMAL)); + assertThat(s.nextWait(), equalTo(NORMAL)); + } + + @Test + public void unexpectedFailureFromNormalRegimeTransitionsToExtended() { + PollingStrategy s = strategy(); + s.onFailure(FailureClass.UNEXPECTED); + // Transitioned: initialDelay swapped to extendedInitial; maxDelay to 1hr. + assertThat(s.getInitialDelay(), equalTo(EXTENDED_INITIAL)); + assertThat(s.getMaxDelay(), equalTo(PollingStrategy.EXTENDED_MAX_DELAY)); + // First extended wait must equal extendedInitial (n reset to 1 → T = initial * 2^0). + // Under jitter, actual wait is in [T/2, T]. + Duration w = s.nextWait(); + assertThat(w.toMillis(), lessThanOrEqualTo(EXTENDED_INITIAL.toMillis())); + assertThat(w.toMillis(), greaterThanOrEqualTo(EXTENDED_INITIAL.toMillis() / 2)); + } + + @Test + public void mixedClassificationNormalThenUnexpectedStartsAtExtendedInitial() { + // Two normal failures advance n; then an unexpected transition should reset n to 1 + // and use extendedInitial directly rather than extendedInitial * 2^currentN. + PollingStrategy s = strategy(); + s.onFailure(FailureClass.NORMAL); + s.onFailure(FailureClass.NORMAL); + // At this point still in normal regime; initialDelay unchanged. + assertThat(s.getInitialDelay(), equalTo(NORMAL)); + + s.onFailure(FailureClass.UNEXPECTED); + assertThat(s.getInitialDelay(), equalTo(EXTENDED_INITIAL)); + Duration w = s.nextWait(); + // T = extendedInitial * 2^0 = extendedInitial. Not extendedInitial * 2^3. + assertThat(w.toMillis(), lessThanOrEqualTo(EXTENDED_INITIAL.toMillis())); + assertThat(w.toMillis(), greaterThanOrEqualTo(EXTENDED_INITIAL.toMillis() / 2)); + } + + @Test + public void extendedRegimeProgressionClampsToMaxDelay() { + // With extendedInitial = 500ms and max = 1hr, doubling progression is + // 500ms, 1s, 2s, 4s, ... until clamped to 1hr. + // We use a smaller max via a custom construction to exercise the clamp + // quickly; see below. + Duration extInitial = Duration.ofMillis(50); + // Force max delay via a custom strategy. PollingStrategy.EXTENDED_MAX_DELAY + // is the 1hr default; not overridable, so we validate clamp indirectly by + // checking that many doublings never exceed max. + PollingStrategy s = new PollingStrategy(NORMAL, extInitial, new Random(1L)); + s.onFailure(FailureClass.UNEXPECTED); // enter extended, n=1 + // Advance n many times; verify T never exceeds max. + for (int i = 0; i < 40; i++) { + Duration w = s.nextWait(); + // Wait is T-J; T <= max; so wait <= max. + assertThat(w.compareTo(PollingStrategy.EXTENDED_MAX_DELAY) <= 0, equalTo(true)); + s.onFailure(FailureClass.NORMAL); // continue in extended regime, advance n + } + } + + @Test + public void firstSuccessDoesNotResetExtendedRegime() { + PollingStrategy s = strategy(); + s.onFailure(FailureClass.UNEXPECTED); // enter extended + assertThat(s.getInitialDelay(), equalTo(EXTENDED_INITIAL)); + + s.onSuccess(); // first success — sets flag but doesn't reset + assertThat(s.getInitialDelay(), equalTo(EXTENDED_INITIAL)); + assertThat(s.getMaxDelay(), equalTo(PollingStrategy.EXTENDED_MAX_DELAY)); + } + + @Test + public void twoConsecutiveSuccessesResetToNormalRegime() { + PollingStrategy s = strategy(); + s.onFailure(FailureClass.UNEXPECTED); + s.onSuccess(); + s.onSuccess(); + assertThat(s.getInitialDelay(), equalTo(NORMAL)); + assertThat(s.getMaxDelay(), equalTo(NORMAL)); + assertThat(s.getN(), equalTo(0)); + } + + @Test + public void failureBetweenSuccessesClearsPriorSuccessFlag() { + PollingStrategy s = strategy(); + s.onFailure(FailureClass.UNEXPECTED); + s.onSuccess(); // prior=success + s.onFailure(FailureClass.NORMAL); // clears prior=success + // Now a single success alone should NOT reset. + s.onSuccess(); + assertThat(s.getInitialDelay(), equalTo(EXTENDED_INITIAL)); + } + + @Test + public void extendedInitialClampedToPollInterval() { + // If extendedInitialInterval < pollInterval, effective floor is pollInterval. + Duration longPoll = Duration.ofMillis(1000); + Duration shortExt = Duration.ofMillis(200); + PollingStrategy s = new PollingStrategy(longPoll, shortExt, new Random(0)); + s.onFailure(FailureClass.UNEXPECTED); + assertThat(s.getInitialDelay(), equalTo(longPoll)); + } + + @Test + public void onFailureReturnsTrueOnlyOnTransitionIntoExtended() { + PollingStrategy s = new PollingStrategy(NORMAL, EXTENDED_INITIAL, new Random(0)); + assertFalse(s.onFailure(FailureClass.NORMAL)); // still in normal + assertTrue(s.onFailure(FailureClass.UNEXPECTED)); // transition -> extended + assertFalse(s.onFailure(FailureClass.UNEXPECTED)); // already in extended + assertFalse(s.onFailure(FailureClass.NORMAL)); // still in extended + } + + @Test + public void nInExtendedDoublesEvenWhenPollIntervalEqualsExtendedInitial() { + // Regression: an equality-based transition check (initialDelay == normalInterval) + // would hold n at 1 whenever pollInterval >= extendedInitial, since the extended + // clamp forces initialDelay back to normalInterval. The explicit inExtended flag + // keeps n doubling. + Duration equal = Duration.ofMillis(500); + PollingStrategy s = new PollingStrategy(equal, equal, new Random(0)); + s.onFailure(FailureClass.UNEXPECTED); // enter extended, n=1 + assertThat(s.getN(), equalTo(1)); + s.onFailure(FailureClass.UNEXPECTED); // still in extended, n=2 + assertThat(s.getN(), equalTo(2)); + s.onFailure(FailureClass.UNEXPECTED); // still in extended, n=3 + assertThat(s.getN(), equalTo(3)); + } + + @Test + public void twoConsecutiveSuccessesReArmExtendedTransition() { + // After healthy-op reset, a subsequent UNEXPECTED failure should re-transition + // into extended (onFailure returns true again). The inExtended flag must be + // cleared by the reset. + PollingStrategy s = new PollingStrategy(NORMAL, EXTENDED_INITIAL, new Random(0)); + assertTrue(s.onFailure(FailureClass.UNEXPECTED)); // -> extended + s.onSuccess(); + s.onSuccess(); // two consecutive successes -> reset to normal + assertTrue(s.onFailure(FailureClass.UNEXPECTED)); // re-transition -> extended + } +} diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamProcessorTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamProcessorTest.java index e79ab73d..b81a50b3 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamProcessorTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamProcessorTest.java @@ -40,6 +40,8 @@ import java.io.IOException; import java.net.URI; import java.time.Duration; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Future; @@ -478,14 +480,16 @@ public void http400ErrorIsRecoverable() throws Exception { testRecoverableHttpError(400); } + // 401 / 403 (and other UNEXPECTED 4xx) engage extended-regime backoff and + // keep retrying instead of transitioning to State.OFF. @Test - public void http401ErrorIsUnrecoverable() throws Exception { - testUnrecoverableHttpError(401); + public void http401TriggersExtendedRegimeAndKeepsRetrying() throws Exception { + testUnexpectedHttpErrorKeepsRetrying(401); } @Test - public void http403ErrorIsUnrecoverable() throws Exception { - testUnrecoverableHttpError(403); + public void http403TriggersExtendedRegimeAndKeepsRetrying() throws Exception { + testUnexpectedHttpErrorKeepsRetrying(403); } @Test @@ -502,7 +506,227 @@ public void http429ErrorIsRecoverable() throws Exception { public void http500ErrorIsRecoverable() throws Exception { testRecoverableHttpError(500); } - + + // Extended-regime timing observation tests. These use compressed extended-regime + // timing (via the StreamProcessor constructor seams) so we can observe strategy + // behavior at ms-scale. Delays are observed via the eventsource's + // "Waiting X milliseconds before reconnecting" INFO log, which emits the strategy's + // computed (jitter-applied) delay directly. Jitter is 0.5x, so observed delays fall + // in [preJitter/2, preJitter]. + + @Test + public void unexpectedErrorEngagesExtendedRegime() throws Exception { + // A: verifies that a 401 causes the SDK to emit the "engaging extended backoff" + // info log, indicating activateRetryDelayStrategy has been called on the eventsource. + Duration extendedInitial = Duration.ofMillis(50); + Duration extendedMax = Duration.ofMillis(1000); + Duration retryReset = Duration.ofSeconds(60); + try (HttpServer server = HttpServer.start(Handlers.status(401))) { + try (StreamProcessor sp = createStreamProcessor(null, server.getUri(), null, + extendedInitial, extendedMax, retryReset)) { + sp.start(); + LogCapture.Message engaged = awaitInfoMessageMatching( + "Classified failure as UNEXPECTED; engaging extended backoff.", 2000); + assertNotNull("expected 'engaging extended backoff' log", engaged); + } + } + } + + @Test + public void unexpectedAndRecoverableUseDifferentRegimes() throws Exception { + // B: 500 uses normal-regime timing (BRIEF_RECONNECT_DELAY = 10ms initial); 401 + // uses extended-regime timing (100ms initial). The observable difference in the + // "Waiting X ms" delays proves classification-drives-regime. + Duration extendedInitial = Duration.ofMillis(100); + Duration extendedMax = Duration.ofMillis(1000); + Duration retryReset = Duration.ofSeconds(60); + + // Phase 1: continuous 500s → normal-regime delays. + try (HttpServer server = HttpServer.start(Handlers.status(500))) { + try (StreamProcessor sp = createStreamProcessor(null, server.getUri(), null, + extendedInitial, extendedMax, retryReset)) { + sp.start(); + List normalDelays = awaitReconnectDelays(1, 2000); + assertFalse("expected some normal-regime reconnect delays", normalDelays.isEmpty()); + // Normal regime: initial=10ms, first delay pre-jitter=10, post-jitter [5, 10]. + // Second pre-jitter=20, post-jitter [10, 20]. Allow generous ceiling. + assertThat("first normal-regime delay should be <= 20ms; observed " + normalDelays.get(0), + normalDelays.get(0), lessThanOrEqualTo(20L)); + } + } + drainCapturedLogs(); + + // Phase 2: continuous 401s → extended-regime delays. + try (HttpServer server = HttpServer.start(Handlers.status(401))) { + try (StreamProcessor sp = createStreamProcessor(null, server.getUri(), null, + extendedInitial, extendedMax, retryReset)) { + sp.start(); + List extDelays = awaitReconnectDelays(1, 2000); + assertFalse("expected some extended-regime reconnect delays", extDelays.isEmpty()); + // Extended regime: initial=100ms, first delay pre-jitter=100, post-jitter [50, 100]. + assertThat("first extended-regime delay should be >= 40ms; observed " + extDelays.get(0), + extDelays.get(0), greaterThanOrEqualTo(40L)); + } + } + } + + @Test + public void healthyOpResetReturnsToNormalRegime() throws Exception { + // C: after an unexpected failure engages extended regime, a subsequent stream that + // stays open for >= retryResetInterval causes the eventsource library to revert to + // the normal-regime (default) strategy on the next reconnect. We observe the delay + // of the reconnect that follows the reset and expect it to be normal-regime-scale. + Duration extendedInitial = Duration.ofMillis(200); + Duration extendedMax = Duration.ofMillis(1000); + Duration retryReset = Duration.ofMillis(100); + + Semaphore closeSuccessfulStream = new Semaphore(0); + Handler seq = Handlers.sequential( + Handlers.status(401), // 1st: triggers extended + closableStreamResponse(EMPTY_DATA_EVENT, closeSuccessfulStream), // 2nd: healthy stream + Handlers.status(500) // 3rd: observe reconnect timing + ); + try (HttpServer server = HttpServer.start(seq)) { + try (StreamProcessor sp = createStreamProcessor(null, server.getUri(), null, + extendedInitial, extendedMax, retryReset)) { + sp.start(); + + // Wait for the SDK to reach VALID (2nd request succeeded, stream is open). + dataSourceUpdates.awaitInit(); + + // Sleep past retryReset while the stream is happily open. + Thread.sleep(retryReset.toMillis() + 50); + + // Drain the extended-regime reconnect delay log (from the 1st fault). + drainCapturedLogs(); + + // Close the successful stream → library computes reconnect delay. Because + // the stream was open >= retryReset, the library resets to default strategy. + closeSuccessfulStream.release(); + + // Observe the next "Waiting X ms" log: should be normal-regime timing. + List postResetDelays = awaitReconnectDelays(1, 2000); + assertFalse("expected a reconnect delay after healthy-op reset", + postResetDelays.isEmpty()); + assertThat("post-reset delay should be normal-regime (<= 20ms); observed " + + postResetDelays.get(0), + postResetDelays.get(0), lessThanOrEqualTo(20L)); + } + } + } + + @Test + public void extendedRegimeDoublesEachAttempt() throws Exception { + // D: repeated 401s under extended-regime should produce delays that double each + // attempt (10 → 20 → 40 → 80 ms pre-jitter). With jitter [x/2, x], the ratio of + // consecutive delays is loose, but the ratio of first-vs-later delays should show + // clear growth. + Duration extendedInitial = Duration.ofMillis(20); + Duration extendedMax = Duration.ofMillis(5000); // effectively no cap for this test + Duration retryReset = Duration.ofSeconds(60); + try (HttpServer server = HttpServer.start(Handlers.status(401))) { + try (StreamProcessor sp = createStreamProcessor(null, server.getUri(), null, + extendedInitial, extendedMax, retryReset)) { + sp.start(); + + // Collect 4 delays: pre-jitter should be 20, 40, 80, 160. + List delays = awaitReconnectDelays(4, 3000); + assertThat("expected at least 4 extended-regime delays; observed " + delays.size(), + delays.size(), greaterThanOrEqualTo(4)); + + // First delay pre-jitter=20, post-jitter [10, 20]; 4th delay pre-jitter=160, + // post-jitter [80, 160]. 4th should be at least 3x the first even under + // worst-case jitter (160/2 = 80, 20/1 = 20 → 4x). + long first = delays.get(0); + long fourth = delays.get(3); + assertThat( + "4th extended-regime delay should be significantly larger than 1st; " + + "observed 1st=" + first + " 4th=" + fourth, + fourth, greaterThanOrEqualTo(first * 3)); + } + } + } + + @Test + public void extendedRegimeClampsAtMax() throws Exception { + // E: repeated 401s under extended-regime with a tight extendedMax should show the + // doubling clamped at extendedMax. Pre-jitter: 10, 20, 40, 60 (clamped), 60, 60... + // Post-jitter [x/2, x]. After the clamp kicks in, all further delays fall in + // [max/2, max]. + Duration extendedInitial = Duration.ofMillis(10); + Duration extendedMax = Duration.ofMillis(60); + Duration retryReset = Duration.ofSeconds(60); + try (HttpServer server = HttpServer.start(Handlers.status(401))) { + try (StreamProcessor sp = createStreamProcessor(null, server.getUri(), null, + extendedInitial, extendedMax, retryReset)) { + sp.start(); + + // Collect several delays; last few should be at the clamp. + List delays = awaitReconnectDelays(6, 3000); + assertThat("expected at least 6 extended-regime delays; observed " + delays.size(), + delays.size(), greaterThanOrEqualTo(6)); + + // The last two delays should each be <= extendedMax (60) — that's the clamp. + // Under jitter, they should be >= extendedMax/2 (30). Assert both bounds + // on the last two collected delays. + long extMax = extendedMax.toMillis(); + long extHalf = extMax / 2; + for (int i = delays.size() - 2; i < delays.size(); i++) { + long d = delays.get(i); + assertThat("clamped delay index=" + i + " should be <= " + extMax + "; observed " + d, + d, lessThanOrEqualTo(extMax)); + assertThat("clamped delay index=" + i + " should be >= " + extHalf + "; observed " + d, + d, greaterThanOrEqualTo(extHalf)); + } + } + } + } + + // Helpers for the extended-regime timing observation tests. + + private static Long parseReconnectDelay(String logText) { + final String prefix = "Waiting "; + final String suffix = " milliseconds before reconnecting"; + if (!logText.startsWith(prefix) || !logText.endsWith(suffix)) { + return null; + } + String middle = logText.substring(prefix.length(), logText.length() - suffix.length()); + try { + return Long.parseLong(middle); + } catch (NumberFormatException e) { + return null; + } + } + + private List awaitReconnectDelays(int minCount, int waitBudgetMs) { + List delays = new ArrayList<>(); + long deadline = System.currentTimeMillis() + waitBudgetMs; + while (delays.size() < minCount) { + long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) break; + LogCapture.Message m = logCapture.awaitMessage(LDLogLevel.INFO, (int) remaining); + if (m == null) break; + Long d = parseReconnectDelay(m.getText()); + if (d != null) delays.add(d); + } + return delays; + } + + private LogCapture.Message awaitInfoMessageMatching(String expectedText, int waitBudgetMs) { + long deadline = System.currentTimeMillis() + waitBudgetMs; + while (true) { + long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) return null; + LogCapture.Message m = logCapture.awaitMessage(LDLogLevel.INFO, (int) remaining); + if (m == null) return null; + if (m.getText().equals(expectedText)) return m; + } + } + + private void drainCapturedLogs() { + while (logCapture.awaitMessage(1) != null) { } + } + @Test public void putEventWithInvalidJsonCausesStreamRestart() throws Exception { verifyEventCausesStreamRestart("put", "{sorry", ErrorKind.INVALID_DATA); @@ -771,25 +995,42 @@ public void streamFailingWithIncompleteEventDoesNotLogJsonError() throws Excepti } } - private void testUnrecoverableHttpError(int statusCode) throws Exception { + private void testUnexpectedHttpErrorKeepsRetrying(int statusCode) throws Exception { Handler errorResp = Handlers.status(statusCode); - + BlockingQueue statuses = new LinkedBlockingQueue<>(); dataSourceUpdates.statusBroadcaster.register(statuses::add); try (HttpServer server = HttpServer.start(errorResp)) { try (StreamProcessor sp = createStreamProcessor(null, server.getUri())) { - Future initFuture = sp.start(); - assertFutureIsCompleted(initFuture, 2, TimeUnit.SECONDS); - - assertFalse(sp.isInitialized()); - - Status newStatus = requireDataSourceStatus(statuses, State.OFF); + sp.start(); + + // Status stays INITIALIZING (never got past init) with an ERROR_RESPONSE + // lastError. The processor does not transition to OFF; it keeps + // retrying under extended-regime backoff. + Status newStatus = requireDataSourceStatus(statuses, State.INITIALIZING); assertEquals(ErrorKind.ERROR_RESPONSE, newStatus.getLastError().getKind()); assertEquals(statusCode, newStatus.getLastError().getStatusCode()); - + + // At least one request should have been made. server.getRecorder().requireRequest(); - server.getRecorder().requireNoRequests(50, TimeUnit.MILLISECONDS); + assertFalse(sp.isInitialized()); + + // Unexpected classifications log at Error level (even though the SDK + // will keep retrying). The SDK-emitted classify-and-log line is + // distinguished by the "Error in stream connection" prefix. + boolean sawErrorForStatus = false; + for (LogCapture.Message m : logCapture.getMessages()) { + if (m.getText().startsWith("Error in stream connection") + && m.getText().contains("HTTP error " + statusCode)) { + assertThat( + "unexpected-classification HTTP error should log at Error, not " + m.getLevel(), + m.getLevel(), equalTo(LDLogLevel.ERROR)); + sawErrorForStatus = true; + } + } + assertTrue("expected an Error-level SDK log mentioning HTTP error " + statusCode, + sawErrorForStatus); } } } @@ -836,15 +1077,43 @@ private void testRecoverableHttpError(int statusCode) throws Exception { // It tries again, and finally gets a valid response (stream2Resp). Status successStatus2 = requireDataSourceStatus(statuses, State.VALID); assertSame(failureStatus3.getLastError(), successStatus2.getLastError()); + + // Normal classifications log at Warn level (not Error). Verify the SDK-emitted + // classify-and-log line -- distinguished by the "Error in stream connection" + // prefix -- appears at Warn for this status. + boolean sawWarnForStatus = false; + for (LogCapture.Message m : logCapture.getMessages()) { + if (m.getText().startsWith("Error in stream connection") + && m.getText().contains("HTTP error " + statusCode)) { + assertThat( + "normal-classification HTTP error should log at Warn, not " + m.getLevel(), + m.getLevel(), equalTo(LDLogLevel.WARN)); + sawWarnForStatus = true; + } + } + assertTrue("expected a Warn-level SDK log mentioning HTTP error " + statusCode, + sawWarnForStatus); } } } - + private StreamProcessor createStreamProcessor(URI streamUri) { return createStreamProcessor(baseConfig().build(), streamUri, null); } private StreamProcessor createStreamProcessor(LDConfig config, URI streamUri, DiagnosticStore acc) { + return createStreamProcessor(config, streamUri, acc, + StreamProcessor.DEFAULT_EXTENDED_INITIAL_RECONNECT_DELAY, + StreamProcessor.DEFAULT_EXTENDED_STREAM_MAX_RETRY_DELAY, + StreamProcessor.DEFAULT_RETRY_RESET_INTERVAL); + } + + private StreamProcessor createStreamProcessor( + LDConfig config, URI streamUri, DiagnosticStore acc, + Duration extendedInitialReconnectDelay, + Duration extendedStreamMaxRetryDelay, + Duration retryResetInterval + ) { return new StreamProcessor( ComponentsImpl.toHttpProperties(clientContext(SDK_KEY, config == null ? baseConfig().build() : config).getHttp()), dataSourceUpdates, @@ -853,6 +1122,9 @@ private StreamProcessor createStreamProcessor(LDConfig config, URI streamUri, Di streamUri, null, BRIEF_RECONNECT_DELAY, + extendedInitialReconnectDelay, + extendedStreamMaxRetryDelay, + retryResetInterval, testLogger ); } From d3b56a64c9c06bfdd60fbee321427e20e12a63d5 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Mon, 24 Aug 2026 14:35:25 -0400 Subject: [PATCH 2/4] fix: transition PollingProcessor to State.OFF on close Aligns with StreamProcessor.close() and FDv2DataSource.close(), both of which call updateStatus(State.OFF, null) after closing their upstream I/O. Also aligns with the PR's updated Javadoc for State.OFF, which now describes it as the time the data source "stopped operation" (i.e., close), rather than the pre-PR meaning of "encountered an unrecoverable error". --- .../main/java/com/launchdarkly/sdk/server/PollingProcessor.java | 1 + 1 file changed, 1 insertion(+) diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java index 17cea33c..fb89b59e 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java @@ -78,6 +78,7 @@ public void close() throws IOException { } logger.info("Closing LaunchDarkly PollingProcessor"); requestor.close(); + dataSourceUpdates.updateStatus(State.OFF, null); } @Override From 14de006afaab30ba91f5772c936a27bf905b2b38 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Wed, 26 Aug 2026 16:36:56 -0400 Subject: [PATCH 3/4] fix: use renamed classifyAndLogHttpFailure from java-sdk-internal The classifier helper was renamed from classifyAndLogHTTPFailure to classifyAndLogHttpFailure in review of the internal-artifact PR. This branch was written against the earlier revision, so it would have failed compileJava once internal 1.11.x published -- independent of the unreleased-dependency pins, since it survives releasing both artifacts. Two call sites: StreamProcessor.handleError and PollingProcessor.poll. --- .../main/java/com/launchdarkly/sdk/server/PollingProcessor.java | 2 +- .../main/java/com/launchdarkly/sdk/server/StreamProcessor.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java index fb89b59e..d49c8a53 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java @@ -123,7 +123,7 @@ private void poll() { } strategy.onSuccess(); } catch (HttpErrorException e) { - FailureClass failureClass = HttpErrors.classifyAndLogHTTPFailure( + 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)) { diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java index 51413041..cdcc3c3a 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java @@ -412,7 +412,7 @@ private boolean handleError(StreamException e, CompletableFuture initFutur ErrorInfo errorInfo; if (e instanceof StreamHttpErrorException) { int status = ((StreamHttpErrorException)e).getCode(); - failureClass = HttpErrors.classifyAndLogHTTPFailure(logger, status, ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE); + failureClass = HttpErrors.classifyAndLogHttpFailure(logger, status, ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE); errorInfo = ErrorInfo.fromHttpError(status); } else if (e instanceof StreamIOException || e instanceof StreamClosedByServerException) { failureClass = HttpErrors.classifyAndLogTransportFailure(logger, e, ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE); From d345b089e02125727b21e82141841fd0603b5760 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Thu, 27 Aug 2026 09:21:56 -0400 Subject: [PATCH 4/4] test: supply a data source update sink in polling builder tests PollingProcessor.close() now reports State.OFF, so it dereferences the update sink. The two builder-configuration tests construct a processor from a bare ClientContext, whose getDataSourceUpdateSink() returns null, then dispose it via try-with-resources -- which NPE'd. A null sink is not a supported state: every SDK path that builds a data source calls withDataSourceUpdateSink() first (FDv1DataSystem, FDv2DataSystem), that method is package-private so callers outside the SDK cannot populate it, and poll() dereferences the sink unconditionally, so such a processor could never run anyway. These tests were relying on close() incidentally not touching the sink. Supplies the sink instead, matching the equivalent tests in StreamProcessorTest, which have always done this because StreamProcessor.close() has always reported OFF. --- .../com/launchdarkly/sdk/server/PollingProcessorTest.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java index 73112c68..d87a5d87 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java @@ -109,7 +109,8 @@ public void setError(int status) { @Test public void builderHasDefaultConfiguration() throws Exception { ComponentConfigurer f = Components.pollingDataSource(); - try (PollingProcessor pp = (PollingProcessor)f.build(clientContext(SDK_KEY, baseConfig().build()))) { + try (PollingProcessor pp = (PollingProcessor)f.build(clientContext(SDK_KEY, baseConfig().build()) + .withDataSourceUpdateSink(dataSourceUpdates))) { assertThat(((DefaultFeatureRequestor)pp.requestor).pollingUri.toString(), containsString(StandardEndpoints.DEFAULT_POLLING_BASE_URI.toString())); assertThat(pp.pollInterval, equalTo(PollingDataSourceBuilder.DEFAULT_POLL_INTERVAL)); } @@ -125,7 +126,8 @@ public void builderCanSpecifyConfiguration() throws Exception { try (PollingProcessor pp = (PollingProcessor) f.build( clientContext( SDK_KEY, - baseConfig().build()))) { + baseConfig().build()) + .withDataSourceUpdateSink(dataSourceUpdates))) { assertThat(pp.pollInterval, equalTo(LENGTHY_INTERVAL)); assertThat(((DefaultFeatureRequestor) pp.requestor).pollingUri.toString(), containsString("filter=myFilter")); }