feat: expose narrow setters on EventSource for RETRY-spec regime switching

src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java

@@ -114,6 +114,23 @@ private DefaultRetryDelayStrategy(
     this.backoffMultiplier = backoffMultiplier;
     this.jitterMultiplier = jitterMultiplier;
   }
+
+  // Package-private helper used by EventSource.setInitialRetryDelayMillis. Returns a
+  // copy with the exponent counter reset to 0 but the max delay, backoff multiplier,
+  // and jitter multiplier preserved. Ensures the next apply() uses the base delay
+  // directly, matching the "first attempt in a new regime uses the initial delay"
+  // invariant from the RETRY specification and the server-SDK implementation guide.
+  DefaultRetryDelayStrategy withResetCounter() {
+    return new DefaultRetryDelayStrategy(0, this.maxDelayMillis, this.backoffMultiplier, this.jitterMultiplier);
+  }
+
+  // Package-private helper used by EventSource.setMaxRetryDelayMillis. Returns a
+  // copy with a new max delay AND the exponent counter reset to 0. Preserves the
+  // backoff multiplier and jitter multiplier. Same reset-on-delay-change invariant
+  // as withResetCounter().
+  DefaultRetryDelayStrategy withMaxDelayMillisAndResetCounter(long newMaxDelayMillis) {
+    return new DefaultRetryDelayStrategy(0, newMaxDelayMillis, this.backoffMultiplier, this.jitterMultiplier);
+  }
 
   @Override
   public Result apply(long baseDelayMillis) {

src/main/java/com/launchdarkly/eventsource/EventSource.java

@@ -88,11 +88,24 @@ public class EventSource implements Closeable {
   // accessed from the thread that is reading from EventSource.
   private EventParser eventParser;
   ErrorStrategy currentErrorStrategy;
-  RetryDelayStrategy currentRetryDelayStrategy;
   private long connectedTime;
   private long disconnectedTime;
   private StreamEvent nextEvent;
 
+  // currentRetryDelayStrategy is volatile because it can be updated via the public
+  // setMaxRetryDelayMillis / setInitialRetryDelayMillis setters from any thread
+  // (e.g., an SDK error handler that runs on the reading thread, or a caller that
+  // wants to swap regimes from a different thread). Updates are always full-object
+  // replacements of the immutable strategy; volatile provides the necessary
+  // publication semantics.
+  volatile RetryDelayStrategy currentRetryDelayStrategy;
+
+  // Set to true by computeReconnectDelay() (following a fault), and cleared by
+  // tryStart() on successful reconnect. Used by setInitialRetryDelayMillis /
+  // setMaxRetryDelayMillis to know whether they should recompute the pending
+  // reconnect delay (SDK-driven mid-flight regime change).
+  private volatile boolean pendingReconnectWait = false;
+
   // These fields are set by the thread that is reading the stream, but can
   // be modified from other threads if they call stop() or interrupt(). We
   // use AtomicReference because we need atomicity in updates.
@@ -212,7 +225,113 @@ public long getBaseRetryDelayMillis() {
   public long getNextRetryDelayMillis() {
     return nextReconnectDelayMillis;
   }
-
+
+  /**
+   * Updates the base retry delay used for computing subsequent reconnect delays.
+   * <p>
+   * This is the SDK-side entry point for RETRY-spec regime switching: for example,
+   * a data source that has classified a failure as "unexpected" per the RETRY
+   * specification and wants to transition into an extended-regime backoff calls
+   * this method with the extended-regime initial delay (e.g., 5 minutes).
+   * <p>
+   * Semantics match the SSE {@code retry:} field: the new value becomes the base
+   * used by the current {@link RetryDelayStrategy} on each subsequent {@code apply()}
+   * call, and the strategy's internal exponent counter is reset so the first
+   * subsequent reconnect uses the new base directly (rather than the new base
+   * multiplied by the current backoff exponent). If the current strategy is a
+   * {@link DefaultRetryDelayStrategy}, the counter reset preserves the strategy's
+   * max delay, backoff multiplier, and jitter multiplier. For any other custom
+   * {@link RetryDelayStrategy} implementation, only the base delay field is
+   * updated; the strategy's own state is left untouched (custom strategies can
+   * observe the new base via the {@code baseDelayMillis} argument to their
+   * {@code apply()} method).
+   * <p>
+   * This method is thread-safe.
+   *
+   * @param millis the new base retry delay in milliseconds
+   * @since 4.1.0
+   * @see #setMaxRetryDelayMillis(long)
+   * @see #getBaseRetryDelayMillis()
+   */
+  public void setInitialRetryDelayMillis(long millis) {
+    baseRetryDelayMillis = millis;
+    RetryDelayStrategy current = currentRetryDelayStrategy;
+    if (current instanceof DefaultRetryDelayStrategy) {
+      currentRetryDelayStrategy = ((DefaultRetryDelayStrategy) current).withResetCounter();
+    }
+    // For non-Default strategies we can't force a counter reset; the strategy's
+    // apply() will see the new baseDelayMillis on the next call and behave accordingly.
+    // If a reconnect delay was already computed by the immediately-prior fault
+    // (which used the OLD strategy), recompute it now so the pending wait
+    // reflects the new regime. Without this, an SDK-driven regime transition
+    // from inside an error handler would take effect only on the NEXT fault.
+    recomputeNextReconnectDelayForRegimeSwitch();
+  }
+
+  /**
+   * Updates the maximum retry delay used by the current retry delay strategy.
+   * <p>
+   * This is the SDK-side entry point for RETRY-spec regime switching: paired with
+   * {@link #setInitialRetryDelayMillis(long)}, a data source calls this method with
+   * the extended-regime maximum (e.g., 1 hour) when transitioning into the extended
+   * regime, and with the normal-regime maximum when transitioning back after a
+   * healthy-operation reset.
+   * <p>
+   * Internally: constructs a new {@link DefaultRetryDelayStrategy} instance with the
+   * given max delay and the exponent counter reset to 0, preserving the current
+   * backoff multiplier and jitter multiplier, and atomically swaps the reference.
+   * The counter reset ensures the first subsequent reconnect uses the current base
+   * delay directly rather than the base multiplied by the current backoff exponent
+   * (matching the "reset {@code n} when delays change" invariant from the RETRY
+   * specification's server-SDK implementation guide).
+   * <p>
+   * Currently only supported when the active strategy is a
+   * {@link DefaultRetryDelayStrategy} — the standard case for callers that
+   * configured retry via {@link Builder#retryDelay(long, TimeUnit)} or the
+   * default strategy. If the active strategy is a custom implementation, this
+   * method is a silent no-op (custom strategies define their own timing shape
+   * and don't expose a "max delay" concept via the abstract
+   * {@link RetryDelayStrategy} interface).
+   * <p>
+   * This method is thread-safe.
+   *
+   * @param millis the new max retry delay in milliseconds
+   * @since 4.1.0
+   * @see #setInitialRetryDelayMillis(long)
+   */
+  public void setMaxRetryDelayMillis(long millis) {
+    RetryDelayStrategy current = currentRetryDelayStrategy;
+    if (current instanceof DefaultRetryDelayStrategy) {
+      currentRetryDelayStrategy =
+          ((DefaultRetryDelayStrategy) current).withMaxDelayMillisAndResetCounter(millis);
+    }
+    // else: silent no-op for custom strategies (see Javadoc).
+    // Same rationale as setInitialRetryDelayMillis: recompute the pending
+    // reconnect delay so an SDK-driven regime transition takes effect on the
+    // upcoming reconnect (not the NEXT fault's).
+    recomputeNextReconnectDelayForRegimeSwitch();
+  }
+
+  // Recompute nextReconnectDelayMillis using the current strategy + base delay.
+  // Called from the SDK-facing setters so a regime transition inside an error
+  // handler affects the pending reconnect wait, not just the one after. Only
+  // meaningful when a reconnect wait is pending (nextReconnectDelayMillis > 0).
+  // Deliberately does NOT advance the strategy: this recompute overwrites the
+  // just-computed delay from the fault-handling path; if we advanced the
+  // strategy here it would double-advance the counter for the next fault.
+  private void recomputeNextReconnectDelayForRegimeSwitch() {
+    // Only meaningful when we're in the between-fault-and-reconnect window.
+    if (!pendingReconnectWait) {
+      return;
+    }
+    RetryDelayStrategy.Result result = currentRetryDelayStrategy.apply(baseRetryDelayMillis);
+    nextReconnectDelayMillis = result.getDelayMillis();
+    // Intentionally do NOT assign result.getNext() to currentRetryDelayStrategy.
+    // The prior fault's computeReconnectDelay() already advanced the counter
+    // for the impending reconnect; our job here is just to overwrite the
+    // stored delay value with what it should have been under the new regime.
+  }
+
   /**
    * Attempts to start the stream if it is not already active.
    * <p>
@@ -321,6 +440,11 @@ private FaultEvent tryStart(boolean canReturnFaultEvent) throws StreamException
       connectionCloser.set(clientResult.getCloser());
       origin = clientResult.getOrigin() == null ? client.getOrigin() : clientResult.getOrigin();
       connectedTime = System.currentTimeMillis();
+      // Clear the pending-reconnect flag now that the reconnect succeeded. The
+      // SDK-facing setInitialRetryDelayMillis / setMaxRetryDelayMillis methods
+      // use this flag to detect an in-flight reconnect wait and re-run the delay
+      // computation under a mid-flight regime change.
+      pendingReconnectWait = false;
       logger.debug("Connected to SSE stream");
 
       ResponseHeaders headers = clientResult.getHeaders();
@@ -665,6 +789,7 @@ private void computeReconnectDelay() {
     if (result.getNext() != null) {
       currentRetryDelayStrategy = result.getNext();
     }
+    pendingReconnectWait = true;
   }
 
   private boolean closeCurrentStream(boolean deliberatelyInterrupted, boolean shouldStopIterating) {