fix: address review feedback on close, clamps, and polling test coverage

Complete the start() future when a data source is closed. No failure
permanently stops either data source now, so shutdown is the only remaining
exit for a caller that waits on that future without a timeout; previously such
a caller waited forever, since close() never completed it and the error-path
completion went away with the permanent-stop paths. Both processors now
complete it in close(), after the OFF status is published so a released thread
observes a consistent state. isInitialized() still reports false, so this
signals "done waiting" rather than "initialized", and it gives the constructor
a fast exit if another thread closes the client during startWaitTime.
StreamProcessor's future was a local in start() and unreachable from close(),
so it becomes a field as PollingProcessor already had; repeated start() calls
consequently share one future, which is how PollingProcessor has always
behaved and LDClient only calls start() once.

Replace the paired compareTo clamps in PollingStrategy.onFailure with a max()
helper. Assign-then-maybe-overwrite twice obscured that both bounds are simply
floored at the poll interval, and compareTo at the call site made the operand
order hard to read. Behavior is unchanged.

Add polling coverage that distinguishes the extended regime from the normal
cadence. testUnexpectedHttpErrorKeepsPolling used a 30ms extended initial delay
against a 20ms poll interval, so after the wait floor the extended wait landed
in 20-30ms and every assertion in it passed whether or not the extended regime
engaged -- its comment claimed the waits were observable but nothing observed
them. The gap that matters is broader: PollingStrategy is covered in isolation,
but the wiring from poll() through onFailure and nextWait into scheduleNext had
no coverage at all, which is where a mis-wired classifier call would hide.

unexpectedFailureUsesExtendedCadence and oneSuccessfulPollDoesNotLeaveExtended-
Cadence assert observed request spacing with a 500ms extended initial delay,
giving a 250-500ms wait against a 20ms normal cadence. Both use a lower bound
only, so CI slowness cannot fail them. Verified non-vacuous by swapping the
injected 401 for a 500: the observed gap drops to 33ms and the assertion fails,
which is what the previous test could not detect. Also adds the missing
"engaging extended backoff" log assertion, which the streaming counterpart
already had.

Author: tanderson-ld

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java

@@ -79,6 +79,7 @@ public void close() throws IOException {
     logger.info("Closing LaunchDarkly PollingProcessor");
     requestor.close();
     dataSourceUpdates.updateStatus(State.OFF, null);
+    initFuture.complete(null);
   }
 
   @Override

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingStrategy.java

@@ -77,14 +77,8 @@ boolean onFailure(FailureClass failureClass) {
     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;
-      }
+      this.initialDelay = max(extendedInitialInterval, normalInterval);
+      this.maxDelay = max(EXTENDED_MAX_DELAY, normalInterval);
       return true;
     }
     this.n++;
@@ -138,6 +132,10 @@ Duration nextWait() {
     return Duration.ofMillis(waitMs);
   }
 
+  private static Duration max(Duration a, Duration b) {
+    return a.compareTo(b) >= 0 ? a : b;
+  }
+
   // Accessors for observability / testing.
 
   int getN() { return n; }

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java

@@ -114,6 +114,7 @@ final class StreamProcessor implements DataSource {
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private volatile long esStarted = 0;
   private volatile boolean lastStoreUpdateFailed = false;
+  private final CompletableFuture<Void> initFuture = new CompletableFuture<>();
   private final LDLogger logger;
 
   StreamProcessor(
@@ -178,8 +179,6 @@ private void onStoreStatusChanged(DataStoreStatusProvider.Status newStatus) {
 
   @Override
   public Future<Void> start() {
-    final CompletableFuture<Void> initFuture = new CompletableFuture<>();
-
     // Notes about the configuration of the EventSource below:
     //
     // 1. Setting streamEventData(true) is an optimization to let us read the event's data field directly
@@ -278,6 +277,7 @@ public void close() throws IOException {
       es.close();
     }
     dataSourceUpdates.updateStatus(State.OFF, null);
+    initFuture.complete(null);
   }
 
   @Override

lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingProcessorTest.java

@@ -49,6 +49,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -60,6 +61,8 @@ public class PollingProcessorTest extends BaseTest {
   private static final String SDK_KEY = "sdk-key";
   private static final Duration LENGTHY_INTERVAL = Duration.ofSeconds(60);
   private static final Duration BRIEF_INTERVAL = Duration.ofMillis(20);
+  private static final Duration OBSERVABLE_EXTENDED_INITIAL = Duration.ofMillis(500);
+  private static final long EXTENDED_GAP_FLOOR_MILLIS = 150L;
 
   private MockDataSourceUpdates dataSourceUpdates;
 
@@ -294,11 +297,54 @@ public void http500ErrorIsRecoverable() throws Exception {
     testRecoverableHttpError(500);
   }
 
+  @Test
+  public void unexpectedFailureUsesExtendedCadence() throws Exception {
+    TestPollHandler handler = new TestPollHandler();
+    handler.setError(401);
+    try (HttpServer server = HttpServer.start(handler)) {
+      try (PollingProcessor pp = makeProcessor(
+          server.getUri(), BRIEF_INTERVAL, OBSERVABLE_EXTENDED_INITIAL)) {
+        pp.start();
+
+        server.getRecorder().requireRequest(); // poll #1: 401, engages the extended regime
+        long afterFirst = System.currentTimeMillis();
+        server.getRecorder().requireRequest(); // poll #2: arrives after an extended-regime wait
+        long gap = System.currentTimeMillis() - afterFirst;
+
+        assertThat("gap between polls should be on the extended curve rather than the "
+            + BRIEF_INTERVAL.toMillis() + "ms normal cadence; observed " + gap + "ms",
+            gap, greaterThan(EXTENDED_GAP_FLOOR_MILLIS));
+      }
+    }
+  }
+
+  @Test
+  public void oneSuccessfulPollDoesNotLeaveExtendedCadence() throws Exception {
+    TestPollHandler handler = new TestPollHandler();
+    handler.setError(401);
+    try (HttpServer server = HttpServer.start(handler)) {
+      try (PollingProcessor pp = makeProcessor(
+          server.getUri(), BRIEF_INTERVAL, OBSERVABLE_EXTENDED_INITIAL)) {
+        pp.start();
+
+        server.getRecorder().requireRequest(); // poll #1: 401, engages the extended regime
+        handler.setError(0);                   // recover, so poll #2 succeeds
+
+        server.getRecorder().requireRequest(); // poll #2: first success
+        long afterFirstSuccess = System.currentTimeMillis();
+        server.getRecorder().requireRequest(); // poll #3: the confirming success
+        long gap = System.currentTimeMillis() - afterFirstSuccess;
+
+        assertThat("after a single success the SDK should still be on the extended curve; "
+            + "observed " + gap + "ms", gap, greaterThan(EXTENDED_GAP_FLOOR_MILLIS));
+      }
+    }
+  }
+
   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.
+    // 401 / 403 (and other UNEXPECTED 4xx) keep polling and never trigger a permanent
+    // State.OFF. Timing is not asserted here -- see unexpectedFailureUsesExtendedCadence
+    // for the extended-regime cadence.
     TestPollHandler handler = new TestPollHandler();
     handler.setError(statusCode);
     Duration extendedInitial = Duration.ofMillis(30);
@@ -335,6 +381,14 @@ private void testUnexpectedHttpErrorKeepsPolling(int statusCode) throws Exceptio
           }
           assertTrue("expected an Error-level SDK log mentioning HTTP error " + statusCode,
               sawErrorForStatus);
+
+          boolean sawEngagedLog = false;
+          for (LogCapture.Message m : logCapture.getMessages()) {
+            if (m.getText().contains("engaging extended backoff")) {
+              sawEngagedLog = true;
+            }
+          }
+          assertTrue("expected the extended-backoff engagement log", sawEngagedLog);
         }
       }
     });