diff --git a/lib/sdk/server/contract-tests/service/src/main/java/sdktest/Representations.java b/lib/sdk/server/contract-tests/service/src/main/java/sdktest/Representations.java index 99112355..09452e50 100644 --- a/lib/sdk/server/contract-tests/service/src/main/java/sdktest/Representations.java +++ b/lib/sdk/server/contract-tests/service/src/main/java/sdktest/Representations.java @@ -39,6 +39,11 @@ public static class SdkConfigParams { public static class SdkConfigStreamParams { URI baseUri; long initialRetryDelayMs; + // RETRY-conformance test knobs (per SDK-2789 / server-sdk-guide.md). Zero + // means "use SDK default"; positive values compress test timing below the + // 5-minute extended-regime default. + Long extendedInitialDelayMs; + Long resetThresholdMs; String filter; } @@ -177,6 +182,8 @@ public static class SdkConfigSynchronizerParams { public static class SdkConfigPollingParams { URI baseUri; Long pollIntervalMs; + // RETRY-conformance test knob (per SDK-2789 / server-sdk-guide.md). + Long extendedInitialDelayMs; String filter; } diff --git a/lib/sdk/server/contract-tests/service/src/main/java/sdktest/SdkClientEntity.java b/lib/sdk/server/contract-tests/service/src/main/java/sdktest/SdkClientEntity.java index f6bb9d96..16b3b281 100644 --- a/lib/sdk/server/contract-tests/service/src/main/java/sdktest/SdkClientEntity.java +++ b/lib/sdk/server/contract-tests/service/src/main/java/sdktest/SdkClientEntity.java @@ -397,6 +397,18 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) { if (params.streaming.initialRetryDelayMs > 0) { dataSource.initialReconnectDelay(Duration.ofMillis(params.streaming.initialRetryDelayMs)); } + // RETRY-conformance test knobs (SDK-2789 / server-sdk-guide.md). Reach the + // package-private impl setters via the SDK's internal bridge. + if (params.streaming.extendedInitialDelayMs != null) { + com.launchdarkly.sdk.server.internal.DataSourceInternalHelpers + .setStreamingExtendedInitialReconnectDelay(dataSource, + Duration.ofMillis(params.streaming.extendedInitialDelayMs.longValue())); + } + if (params.streaming.resetThresholdMs != null) { + com.launchdarkly.sdk.server.internal.DataSourceInternalHelpers + .setStreamingRetryResetInterval(dataSource, + Duration.ofMillis(params.streaming.resetThresholdMs.longValue())); + } dataSource.payloadFilter(params.streaming.filter); builder.dataSource(dataSource); } else if (params.polling != null && params.dataSystem == null) { @@ -405,6 +417,11 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) { if (params.polling.pollIntervalMs != null) { pollingDataSource.pollInterval(Duration.ofMillis(params.polling.pollIntervalMs)); } + if (params.polling.extendedInitialDelayMs != null) { + com.launchdarkly.sdk.server.internal.DataSourceInternalHelpers + .setPollingExtendedInitialDelay(pollingDataSource, + Duration.ofMillis(params.polling.extendedInitialDelayMs.longValue())); + } if (params.polling.filter != null && !params.polling.filter.isEmpty()) { pollingDataSource.payloadFilter(params.polling.filter); } 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..c413629b 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 @@ -126,6 +126,19 @@ public void close() throws IOException {} static final class StreamingDataSourceBuilderImpl extends StreamingDataSourceBuilder implements DiagnosticDescription { + // Package-private test-only setters. Follows the pollIntervalWithNoMinimum + // precedent below. Cross-module callers reach these via the bridge class in + // com.launchdarkly.sdk.server.internal. + StreamingDataSourceBuilderImpl extendedInitialReconnectDelay(Duration d) { + this.extendedInitialReconnectDelay = d; + return this; + } + + StreamingDataSourceBuilderImpl retryResetInterval(Duration d) { + this.retryResetInterval = d; + return this; + } + @Override public DataSource build(ClientContext context) { LDLogger baseLogger = context.getBaseLogger(); @@ -146,6 +159,8 @@ public DataSource build(ClientContext context) { streamUri, payloadFilter, initialReconnectDelay, + extendedInitialReconnectDelay, + retryResetInterval, logger); } @@ -170,6 +185,14 @@ PollingDataSourceBuilderImpl pollIntervalWithNoMinimum(Duration pollInterval) { this.pollInterval = pollInterval; return this; } + + // Package-private test-only setter for RETRY §1.6-driven extended-regime + // initial delay. Cross-module callers reach this via the bridge class in + // com.launchdarkly.sdk.server.internal. + PollingDataSourceBuilderImpl extendedInitialDelay(Duration d) { + this.extendedInitialDelay = d; + return this; + } @Override public DataSource build(ClientContext context) { @@ -196,6 +219,7 @@ public DataSource build(ClientContext context) { context.getDataSourceUpdateSink(), ClientContextImpl.get(context).sharedExecutor, pollInterval, + extendedInitialDelay, logger); } diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/ComponentsInternalBridge.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/ComponentsInternalBridge.java new file mode 100644 index 00000000..0db29fbb --- /dev/null +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/ComponentsInternalBridge.java @@ -0,0 +1,33 @@ +package com.launchdarkly.sdk.server; + +import com.launchdarkly.sdk.server.ComponentsImpl.PollingDataSourceBuilderImpl; +import com.launchdarkly.sdk.server.ComponentsImpl.StreamingDataSourceBuilderImpl; +import com.launchdarkly.sdk.server.integrations.PollingDataSourceBuilder; +import com.launchdarkly.sdk.server.integrations.StreamingDataSourceBuilder; + +import java.time.Duration; + +/** + * Package-private bridge from + * {@link com.launchdarkly.sdk.server.internal.DataSourceInternalHelpers} into + * the package-private test-only knobs on {@link ComponentsImpl}'s concrete + * builder impl classes. Public class-level visibility is required so the + * {@code .internal} bridge can call it; the intent is that consumers use the + * {@code .internal.DataSourceInternalHelpers} entry points, not this class + * directly. + */ +public final class ComponentsInternalBridge { + private ComponentsInternalBridge() {} + + public static void setStreamingExtendedInitialReconnectDelay(StreamingDataSourceBuilder b, Duration d) { + ((StreamingDataSourceBuilderImpl) b).extendedInitialReconnectDelay(d); + } + + public static void setStreamingRetryResetInterval(StreamingDataSourceBuilder b, Duration d) { + ((StreamingDataSourceBuilderImpl) b).retryResetInterval(d); + } + + public static void setPollingExtendedInitialDelay(PollingDataSourceBuilder b, Duration d) { + ((PollingDataSourceBuilderImpl) b).extendedInitialDelay(d); + } +} diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FailureClass.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FailureClass.java new file mode 100644 index 00000000..578a7613 --- /dev/null +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FailureClass.java @@ -0,0 +1,67 @@ +package com.launchdarkly.sdk.server; + +import javax.net.ssl.SSLException; + +import java.security.GeneralSecurityException; +import java.security.cert.CertificateException; + +/** + * Classifies a data-source failure per the RETRY specification (§1.5 / §1.6 / §1.7) + * into one of two regimes: {@link #NORMAL} or {@link #UNEXPECTED}. This is the + * SDK-side classification the streaming and polling data sources use to decide + * whether a failure should trigger extended-regime backoff (per SDK-2775 / + * server-sdk-guide.md). + */ +enum FailureClass { + /** + * Ordinary transient failure. Use the normal-regime backoff. Per RETRY §1.6.1, + * this includes HTTP 400 / 408 / 429, HTTP 5xx, any other HTTP status the SDK + * treats as a failure, and generic transport failures (connection refused, + * read timeout, DNS failure, etc.). + */ + NORMAL, + + /** + * Unexpected failure indicative of a longer-lived condition. Use the extended- + * regime backoff. Per RETRY §1.6 / §1.7, this includes HTTP 401 / 403 and any + * other 4xx not in the NORMAL list, plus TLS / certificate validation failures. + */ + UNEXPECTED; + + /** + * Classify a completed HTTP response by its status code, per RETRY §1.6. + */ + static FailureClass fromHttpStatus(int status) { + // 400, 408, 429 are NORMAL per §1.6.1. + if (status == 400 || status == 408 || status == 429) { + return NORMAL; + } + // 5xx is NORMAL. + if (status >= 500) { + return NORMAL; + } + // Any other 4xx (including 401 and 403) is UNEXPECTED per §1.6. + if (status >= 400 && status < 500) { + return UNEXPECTED; + } + // Any non-4xx / non-5xx status the SDK treats as a failure is NORMAL under + // the §1.6.1 catch-all. + return NORMAL; + } + + /** + * Classify a transport-level exception (network I/O, TLS, DNS, etc.), per + * RETRY §1.7. TLS / certificate validation failures are UNEXPECTED; all other + * transport failures are NORMAL. + */ + static FailureClass fromTransportException(Throwable t) { + for (Throwable c = t; c != null; c = c.getCause()) { + if (c instanceof SSLException + || c instanceof CertificateException + || c instanceof GeneralSecurityException) { + return UNEXPECTED; + } + } + return NORMAL; + } +} 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..75552dbd 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 @@ -21,9 +21,6 @@ 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"; @@ -32,7 +29,9 @@ final class PollingProcessor implements DataSource { private final DataSourceUpdateSink dataSourceUpdates; private final ScheduledExecutorService scheduler; @VisibleForTesting final Duration pollInterval; + @VisibleForTesting final PollingStrategy strategy; private final AtomicBoolean initialized = new AtomicBoolean(false); + private final AtomicBoolean closed = new AtomicBoolean(false); private final CompletableFuture initFuture; private volatile ScheduledFuture task; private final LDLogger logger; @@ -42,12 +41,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,12 +60,12 @@ public boolean isInitialized() { @Override public void close() throws IOException { + if (closed.getAndSet(true)) { + return; + } 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 (task != null) { task.cancel(true); @@ -73,20 +74,35 @@ public void close() throws IOException { } } + private final AtomicBoolean started = new AtomicBoolean(false); + @Override public Future start() { + if (started.getAndSet(true)) { + return initFuture; // idempotent per pre-existing contract + } logger.info("Starting LaunchDarkly polling client with interval: {} milliseconds", pollInterval.toMillis()); - + scheduleNext(Duration.ZERO); + return initFuture; + } + + // Per RETRY §1.2.1 no failure is permanently terminal. The polling loop drives + // itself via strategy.nextWait() after each poll (success or failure), replacing + // the pre-RETRY fixed-rate schedule. Extended-regime backoff is engaged by the + // strategy on classification of UNEXPECTED (RETRY §1.6 / §1.7). + private void scheduleNext(Duration delay) { + if (closed.get()) { + return; + } synchronized (this) { - if (task == null) { - task = scheduler.scheduleAtFixedRate(this::poll, 0L, pollInterval.toMillis(), TimeUnit.MILLISECONDS); + if (closed.get()) { + return; } + task = scheduler.schedule(this::poll, delay.toMillis(), TimeUnit.MILLISECONDS); } - - return initFuture; } - + private void poll() { try { // If we already obtained data earlier, and the poll request returns a cached response, then we don't @@ -101,35 +117,42 @@ private void poll() { if (dataSourceUpdates.init(allData)) { dataSourceUpdates.updateStatus(State.VALID, null); if (!initialized.getAndSet(true)) { - logger.info("Initialized LaunchDarkly client."); + logger.info("Initialized LaunchDarkly client."); initFuture.complete(null); } } } + 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 klass = FailureClass.fromHttpStatus(e.getStatus()); + logger.warn("Received HTTP {} error {} ({} - {})", e.getStatus(), + klass == FailureClass.UNEXPECTED ? "(unexpected regime)" : "(normal regime)", + ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE); + dataSourceUpdates.updateStatus(State.INTERRUPTED, errorInfo); + strategy.onFailure(klass); } catch (IOException e) { - checkIfErrorIsRecoverableAndLog(logger, e.toString(), ERROR_CONTEXT_MESSAGE, 0, WILL_RETRY_MESSAGE); + FailureClass klass = FailureClass.fromTransportException(e); + logger.warn("Transport error {} ({} - {}): {}", + klass == FailureClass.UNEXPECTED ? "(unexpected regime)" : "(normal regime)", + ERROR_CONTEXT_MESSAGE, WILL_RETRY_MESSAGE, e.toString()); dataSourceUpdates.updateStatus(State.INTERRUPTED, ErrorInfo.fromException(ErrorKind.NETWORK_ERROR, e)); + strategy.onFailure(klass); } 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. + // Per RETRY §1.2.1 the polling loop never permanently stops on a data-source + // failure — the extended-regime backoff carries the retry cadence instead. + 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..542b3103 --- /dev/null +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingStrategy.java @@ -0,0 +1,141 @@ +package com.launchdarkly.sdk.server; + +import java.time.Duration; +import java.util.Random; + +/** + * Encapsulates the retry-timing state machine for the polling data source per + * the RETRY specification (RETRY §1.4 / §1.5 / §1.8). Owns the two-counter state + * described in the LaunchDarkly server-SDK implementation guide (SDK-2775): + * + *

+ * The observability {@code attempts} counter from the guide's two-counter + * model is intentionally omitted (matches Go's polling implementation; see + * SDK-2788 retro §5). It can be added additively without disrupting this + * state machine. + *

+ * 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 { + /** + * Ceiling for the extended-regime backoff per the RETRY spec / server-SDK + * implementation guide: 1 hour. + */ + 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 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. On the transition from normal into + * extended regime (detected via {@code initialDelay == normalInterval}), + * set {@code n = 1} and swap in the extended bounds, so the first extended + * wait uses {@code extendedInitialInterval} directly per the "reset n when + * delays change" invariant. On any other failure, just increment n. + *

+ * Extended-regime bounds are clamped to be no smaller than the customer- + * configured {@code pollInterval} — the RETRY §1.4.4 override guarantees + * the wait never drops below that. + */ + void onFailure(FailureClass klass) { + this.priorPollWasSuccessful = false; + if (klass == FailureClass.UNEXPECTED && this.initialDelay.equals(this.normalInterval)) { + // transition from normal into extended regime + 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; + } + this.n++; + } + + /** + * Advance state after a poll success. Two-consecutive-successes reset per + * RETRY §1.8.1 (polling binding): after the second success in a row, n=0 + * and delay bounds revert to the normal regime. A single success sets a + * "prior succeeded" flag; any intervening failure clears it. + */ + void onSuccess() { + if (this.priorPollWasSuccessful) { + this.n = 0; + this.initialDelay = normalInterval; + this.maxDelay = normalInterval; + } + this.priorPollWasSuccessful = true; + } + + /** + * Compute the delay before the next poll attempt per RETRY §1.4: + * {@code T = initialDelay * 2^(n-1)}, clamped to {@code maxDelay}. + * Jitter {@code J} is uniform in {@code [0, T/2]} per RETRY §1.4.3. + * Final wait is {@code max(pollInterval, T - J)} per RETRY §1.4.4's + * polling override — 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..c4e7d155 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 @@ -16,6 +16,7 @@ import com.launchdarkly.eventsource.StreamException; import com.launchdarkly.eventsource.StreamHttpErrorException; import com.launchdarkly.eventsource.StreamIOException; +import com.launchdarkly.eventsource.StartedEvent; import com.launchdarkly.logging.LDLogger; import com.launchdarkly.logging.LogValues; import com.launchdarkly.sdk.internal.events.DiagnosticStore; @@ -89,6 +90,12 @@ final class StreamProcessor implements DataSource { final URI streamUri; @VisibleForTesting final Duration initialReconnectDelay; + @VisibleForTesting + final Duration extendedInitialReconnectDelay; + @VisibleForTesting + final Duration retryResetInterval; + private final Duration streamMaxRetryDelay = Duration.ofSeconds(30); + private final Duration streamExtendedMaxRetryDelay = Duration.ofHours(1); private final DiagnosticStore diagnosticAccumulator; private final int threadPriority; private final DataStoreStatusProvider.StatusListener statusListener; @@ -96,6 +103,15 @@ final class StreamProcessor implements DataSource { private final AtomicBoolean initialized = new AtomicBoolean(false); private final AtomicBoolean closed = new AtomicBoolean(false); private volatile long esStarted = 0; + // activeSince: monotonic timestamp of the most recent StartedEvent; used to + // gate the RETRY §1.8 healthy-operation reset that returns the stream from + // extended-regime backoff to normal-regime backoff. Zero when we've never + // observed a StartedEvent since the last regime switch. + private volatile long activeSince = 0; + // currentRegime: NORMAL until an UNEXPECTED failure engages extended-regime + // backoff via setInitialRetryDelayMillis / setMaxRetryDelayMillis; back to + // NORMAL after activeSince elapses retryResetInterval. + private volatile FailureClass currentRegime = FailureClass.NORMAL; private volatile boolean lastStoreUpdateFailed = false; private final LDLogger logger; @@ -107,12 +123,16 @@ final class StreamProcessor implements DataSource { URI streamUri, String payloadFilter, Duration initialReconnectDelay, + Duration extendedInitialReconnectDelay, + Duration retryResetInterval, LDLogger logger) { this.dataSourceUpdates = dataSourceUpdates; this.httpProperties = httpProperties; this.diagnosticAccumulator = diagnosticAccumulator; this.threadPriority = threadPriority; this.initialReconnectDelay = initialReconnectDelay; + this.extendedInitialReconnectDelay = extendedInitialReconnectDelay; + this.retryResetInterval = retryResetInterval; this.logger = logger; URI tempUri = HttpHelpers.concatenateUriPath(streamUri, StandardEndpoints.STREAMING_REQUEST_PATH); @@ -184,8 +204,16 @@ public Future start() { .logger(logger) .readBufferSize(5000) .streamEventData(true) - .expectFields("event") - .retryDelay(initialReconnectDelay.toMillis(), TimeUnit.MILLISECONDS); + .expectFields("event") + .retryDelay(initialReconnectDelay.toMillis(), TimeUnit.MILLISECONDS) + // Disable the library's built-in healthy-op reset. Per RETRY §1.8 the + // SDK owns the regime state; the library would otherwise silently revert + // to its baseRetryDelayStrategy on reset, which would lose the SDK's + // extended-regime max-delay setting. + .retryDelayResetThreshold(0, TimeUnit.MILLISECONDS) + .retryDelayStrategy( + com.launchdarkly.eventsource.RetryDelayStrategy.defaultStrategy() + .maxDelay(streamMaxRetryDelay.toMillis(), TimeUnit.MILLISECONDS)); es = builder.build(); Thread thread = new Thread(() -> { @@ -256,8 +284,13 @@ private boolean handleEvent(StreamEvent event, CompletableFuture initFutur if (closed.get()) { return false; } - logger.debug("Received StreamEvent: {}", event); - if (event instanceof MessageEvent) { + logger.debug("Received StreamEvent: {}", event); + if (event instanceof StartedEvent) { + // Successful (re)connect: begin the healthy-op activeSince clock. Used + // by the regime-swap decision in handleError to decide whether the + // extended regime should be released back to normal per RETRY §1.8. + activeSince = System.currentTimeMillis(); + } else if (event instanceof MessageEvent) { handleMessage((MessageEvent)event, initFuture); } else if (event instanceof FaultEvent) { return handleError(((FaultEvent)event).getCause(), initFuture); @@ -373,32 +406,63 @@ private boolean handleError(StreamException e, CompletableFuture initFutur // treat that as a failure in our analytics. streamFailed = false; } else { - logger.warn("Encountered EventSource error: {}", LogValues.exceptionSummary(e)); + logger.warn("Encountered EventSource error: {}", LogValues.exceptionSummary(e)); } recordStreamInit(streamFailed); - + + // Classify the failure per RETRY §1.6 / §1.7. Under RETRY §1.2.1 no failure + // is permanently terminal: everything triggers a retry, either at normal + // cadence or at extended-regime cadence. + FailureClass klass; + ErrorInfo errorInfo; if (e instanceof StreamHttpErrorException) { int status = ((StreamHttpErrorException)e).getCode(); - ErrorInfo errorInfo = ErrorInfo.fromHttpError(status); + klass = FailureClass.fromHttpStatus(status); + errorInfo = ErrorInfo.fromHttpError(status); + } else if (e instanceof StreamIOException || e instanceof StreamClosedByServerException) { + klass = FailureClass.fromTransportException(e); + errorInfo = ErrorInfo.fromException(ErrorKind.NETWORK_ERROR, e); + } else if (e instanceof StreamClosedByCallerException) { + klass = FailureClass.NORMAL; + errorInfo = ErrorInfo.fromException(ErrorKind.UNKNOWN, e); + } else { + klass = 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 + // Healthy-op reset gate: if we've been actively receiving events for at + // least retryResetInterval, return to normal-regime backoff. Per RETRY §1.8. + if (currentRegime == FailureClass.UNEXPECTED + && activeSince != 0 + && (System.currentTimeMillis() - activeSince) >= retryResetInterval.toMillis()) { + logger.info("Stream was healthy for at least {} ms; returning to normal-regime backoff.", + retryResetInterval.toMillis()); + EventSource stream = es; + if (stream != null) { + stream.setInitialRetryDelayMillis(initialReconnectDelay.toMillis()); + stream.setMaxRetryDelayMillis(streamMaxRetryDelay.toMillis()); + } + currentRegime = FailureClass.NORMAL; + } + // Reset the activeSince clock on any fault; it starts again on the next + // StartedEvent. + activeSince = 0; + + // Transition into extended regime on UNEXPECTED classification. Per SDK-2775 + // server-sdk-guide.md: single-profile with narrow setters on EventSource. + if (klass == FailureClass.UNEXPECTED && currentRegime == FailureClass.NORMAL) { + logger.info("Classified failure as UNEXPECTED per RETRY §1.6/§1.7; engaging extended-regime backoff."); + EventSource stream = es; + if (stream != null) { + stream.setInitialRetryDelayMillis(extendedInitialReconnectDelay.toMillis()); + stream.setMaxRetryDelayMillis(streamExtendedMaxRetryDelay.toMillis()); } + currentRegime = FailureClass.UNEXPECTED; } - 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 allow reconnect per RETRY §1.2.1 } private static T parseStreamJson(Function parser, Reader r) throws StreamInputException { diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/integrations/PollingDataSourceBuilder.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/integrations/PollingDataSourceBuilder.java index 7c73563e..ce63938b 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/integrations/PollingDataSourceBuilder.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/integrations/PollingDataSourceBuilder.java @@ -32,9 +32,20 @@ public abstract class PollingDataSourceBuilder implements ComponentConfigurer + * Package name {@code .internal} signals that these entry points are for + * LaunchDarkly-internal use (contract tests, sdk-test-harness). Consumers of the + * SDK should not use this class — its methods are not covered by semver + * guarantees and may change or disappear between minor versions. + *

+ * Extends the existing "package-private setter on the concrete impl" pattern + * used elsewhere in this SDK (see {@code ComponentsImpl.PollingDataSourceBuilderImpl#pollIntervalWithNoMinimum}) + * to cross-module callers. + */ +public final class DataSourceInternalHelpers { + private DataSourceInternalHelpers() {} + + /** + * Sets the extended-regime initial reconnect delay on a streaming data-source + * builder. Used by contract tests to compress the RETRY-spec extended-regime + * default (5 minutes) into millisecond-scale test intervals. + */ + public static void setStreamingExtendedInitialReconnectDelay(StreamingDataSourceBuilder b, Duration d) { + ComponentsInternalBridge.setStreamingExtendedInitialReconnectDelay(b, d); + } + + /** + * Sets the healthy-operation reset threshold on a streaming data-source + * builder (RETRY §1.8's activeSince mechanism). + */ + public static void setStreamingRetryResetInterval(StreamingDataSourceBuilder b, Duration d) { + ComponentsInternalBridge.setStreamingRetryResetInterval(b, d); + } + + /** + * Sets the extended-regime initial delay on a polling data-source builder. + * Effective delay is {@code max(this, pollInterval)} because polling's wait + * floor per RETRY §1.4.4 override. + */ + public static void setPollingExtendedInitialDelay(PollingDataSourceBuilder b, Duration d) { + ComponentsInternalBridge.setPollingExtendedInitialDelay(b, d); + } +} diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FailureClassTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FailureClassTest.java new file mode 100644 index 00000000..7bf685f8 --- /dev/null +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/FailureClassTest.java @@ -0,0 +1,78 @@ +package com.launchdarkly.sdk.server; + +import org.junit.Test; + +import javax.net.ssl.SSLHandshakeException; +import javax.net.ssl.SSLPeerUnverifiedException; + +import java.io.IOException; +import java.net.ConnectException; +import java.net.SocketTimeoutException; +import java.security.cert.CertificateException; +import java.security.cert.CertificateExpiredException; + +import static com.launchdarkly.sdk.server.FailureClass.NORMAL; +import static com.launchdarkly.sdk.server.FailureClass.UNEXPECTED; +import static org.junit.Assert.assertEquals; + +/** + * Unit coverage for {@link FailureClass} mapping per RETRY §1.6 (HTTP) and §1.7 + * (transport). + */ +@SuppressWarnings("javadoc") +public class FailureClassTest { + + // HTTP §1.6.1: 400, 408, 429 are NORMAL. + @Test public void http400IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(400)); } + @Test public void http408IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(408)); } + @Test public void http429IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(429)); } + + // HTTP §1.6: other 4xx (including 401, 403) is UNEXPECTED. + @Test public void http401IsUnexpected() { assertEquals(UNEXPECTED, FailureClass.fromHttpStatus(401)); } + @Test public void http403IsUnexpected() { assertEquals(UNEXPECTED, FailureClass.fromHttpStatus(403)); } + @Test public void http404IsUnexpected() { assertEquals(UNEXPECTED, FailureClass.fromHttpStatus(404)); } + @Test public void http418IsUnexpected() { assertEquals(UNEXPECTED, FailureClass.fromHttpStatus(418)); } + @Test public void http451IsUnexpected() { assertEquals(UNEXPECTED, FailureClass.fromHttpStatus(451)); } + + // HTTP 5xx is NORMAL. + @Test public void http500IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(500)); } + @Test public void http502IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(502)); } + @Test public void http503IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(503)); } + @Test public void http504IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(504)); } + @Test public void http599IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(599)); } + + // §1.6.1 catch-all: unusual non-4xx / non-5xx failure statuses are NORMAL. + @Test public void http300IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(300)); } + @Test public void http0IsNormal() { assertEquals(NORMAL, FailureClass.fromHttpStatus(0)); } + + // Transport §1.7: ordinary network I/O failures are NORMAL. + @Test public void connectExceptionIsNormal() { + assertEquals(NORMAL, FailureClass.fromTransportException(new ConnectException("connection refused"))); + } + @Test public void socketTimeoutIsNormal() { + assertEquals(NORMAL, FailureClass.fromTransportException(new SocketTimeoutException("timeout"))); + } + @Test public void ioExceptionIsNormal() { + assertEquals(NORMAL, FailureClass.fromTransportException(new IOException("something else"))); + } + + // Transport §1.7: TLS / certificate validation failures are UNEXPECTED. + @Test public void sslHandshakeIsUnexpected() { + assertEquals(UNEXPECTED, FailureClass.fromTransportException(new SSLHandshakeException("handshake failed"))); + } + @Test public void sslPeerUnverifiedIsUnexpected() { + assertEquals(UNEXPECTED, FailureClass.fromTransportException(new SSLPeerUnverifiedException("peer not verified"))); + } + @Test public void certificateExceptionIsUnexpected() { + assertEquals(UNEXPECTED, FailureClass.fromTransportException(new CertificateException("cert invalid"))); + } + @Test public void certificateExpiredIsUnexpected() { + assertEquals(UNEXPECTED, FailureClass.fromTransportException(new CertificateExpiredException("expired"))); + } + + // Transport §1.7: cause-chain walk finds TLS deep in wrapper exceptions. + @Test public void sslCauseWrappedIsUnexpected() { + IOException wrapper = new IOException("wrapped", new SSLHandshakeException("real cause")); + assertEquals(UNEXPECTED, FailureClass.fromTransportException(wrapper)); + } +} 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..e88bf38a 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,33 @@ public void clientStartsInPollingModeAfterRecoverableError() throws Exception { } } + // Post-RETRY: 401 no longer permanently stops polling; the SDK keeps retrying + // at extended-regime cadence. See SDK-2789 / server-sdk-guide.md and RETRY §1.2.1. @Test - public void clientFailsInPollingModeWith401Error() throws Exception { + public void clientInPollingModeKeepsRetryingOn401Error() throws Exception { try (HttpServer server = HttpServer.start(makeInvalidSdkKeyResponse())) { + // Use pollIntervalWithNoMinimum + small extendedInitialDelay so the + // extended-regime waits are observable at ms scale rather than the 5-min + // production default. + ComponentsImpl.PollingDataSourceBuilderImpl pollingBuilder = + Components.pollingDataSourceInternal() + .pollIntervalWithNoMinimum(Duration.ofMillis(5)); + pollingBuilder.extendedInitialDelay(Duration.ofMillis(20)); 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(pollingBuilder) + .startWait(Duration.ofMillis(500)) .events(noEvents()) .build(); - + try (LDClient client = new LDClient(sdkKey, config)) { + // Client init won't complete (never gets a successful response), but polling continues. assertFalse(client.isInitialized()); assertFalse(client.boolVariation(flagKey, user, false)); - + + // Multiple polls should have happened within the startWait window. + server.getRecorder().requireRequest(); server.getRecorder().requireRequest(); - server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); } } } @@ -174,35 +186,29 @@ public void clientStartsInStreamingModeAfterRecoverableError() throws Exception } } + // Post-RETRY: 401 no longer permanently stops streaming; the SDK engages + // extended-regime backoff and keeps retrying. See SDK-2789 / server-sdk-guide.md + // and RETRY §1.2.1. @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 post-RETRY; 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..c2741fe0 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 @@ -68,8 +68,14 @@ public void setup() { } private PollingProcessor makeProcessor(URI baseUri, Duration pollInterval) { + return makeProcessor(baseUri, pollInterval, + com.launchdarkly.sdk.server.integrations.PollingDataSourceBuilder.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 +264,17 @@ public void http400ErrorIsRecoverable() throws Exception { testRecoverableHttpError(400); } + // Per RETRY §1.2.1, 401/403 no longer trigger a permanent stop. The SDK + // engages the extended-regime backoff and keeps polling. See SDK-2789 / + // server-sdk-guide.md and the RETRY specification. @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,31 @@ public void http500ErrorIsRecoverable() throws Exception { testRecoverableHttpError(500); } - private void testUnrecoverableHttpError(int statusCode) throws Exception { + private void testUnexpectedHttpErrorKeepsPolling(int statusCode) throws Exception { + // Post-RETRY behavior: 401/403 (and other 4xx per §1.6) engage extended-regime + // backoff via PollingStrategy but never trigger a permanent State.OFF. Use a + // small extendedInitialDelay so the test wall clock stays short. 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(); - } - server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); + // State stays INITIALIZING (never got past init because every response + // was an error) with an ERROR_RESPONSE lastError. Pre-RETRY this would + // have flipped to OFF; post-RETRY it stays INITIALIZING and keeps retrying. + Status status = requireDataSourceStatus(statuses, State.INITIALIZING); + assertNotNull(status.getLastError()); + assertEquals(ErrorKind.ERROR_RESPONSE, status.getLastError().getKind()); + assertEquals(statusCode, status.getLastError().getStatusCode()); + assertFalse(pollingProcessor.isInitialized()); } } }); 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..65edde15 --- /dev/null +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingStrategyTest.java @@ -0,0 +1,146 @@ +package com.launchdarkly.sdk.server; + +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; + +/** + * Unit coverage for {@link PollingStrategy} state machine per RETRY §1.4 / §1.8. + * Uses ms-scale numbers so tests are fast; the ratios match the RETRY spec'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 per the §1.4.4 wait floor. + 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() { + // Mixed-classification transition test per server-sdk-guide §Testing considerations. + // 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() { + // RETRY §1.8.1 (polling binding). + 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)); + } +} 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..703079d1 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 @@ -478,14 +478,17 @@ public void http400ErrorIsRecoverable() throws Exception { testRecoverableHttpError(400); } + // Post-RETRY: 401/403 (and other 4xx per §1.6) no longer trigger a permanent + // State.OFF; the SDK engages extended-regime backoff and keeps retrying. + // See SDK-2789 / server-sdk-guide.md and RETRY §1.2.1. @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 @@ -771,25 +774,26 @@ 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. Pre-RETRY this would have transitioned to OFF; post-RETRY + // it stays INITIALIZING and keeps retrying. + 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()); } } } @@ -853,6 +857,8 @@ private StreamProcessor createStreamProcessor(LDConfig config, URI streamUri, Di streamUri, null, BRIEF_RECONNECT_DELAY, + com.launchdarkly.sdk.server.integrations.StreamingDataSourceBuilder.DEFAULT_EXTENDED_INITIAL_RECONNECT_DELAY, + com.launchdarkly.sdk.server.integrations.StreamingDataSourceBuilder.DEFAULT_RETRY_RESET_INTERVAL, testLogger ); }