fix test

Author: MarketDataDev03

src/main/java/com/marketdata/sdk/HttpTransport.java

@@ -117,36 +117,52 @@ private static HttpClient defaultHttpClient() {
 
   /**
    * Async-first request execution with retry. Orchestrates one or more attempts according to {@link
-   * RetryPolicy}: retries 501–599 and {@link NetworkError} with exponential backoff, surfaces every
-   * other failure immediately. Cancellation of the returned future bails out of any pending backoff
-   * and skips further attempts.
+   * RetryPolicy}: retries 501–599 and IOException-shaped {@link NetworkError}s with exponential
+   * backoff, surfaces every other failure immediately. Cancellation of the returned future bails
+   * out of any pending backoff and propagates to the current in-flight attempt.
    */
   <T> CompletableFuture<T> executeAsync(RequestSpec spec, Class<T> responseType) {
     CompletableFuture<T> result = new CompletableFuture<>();
-    attempt(spec, responseType, 0, result);
+    // One cascade-cancel handler installed once: whichever attempt is currently in flight is
+    // tracked in `currentDispatched`; cancelling `result` cancels that. Previous attempts in
+    // the chain are already done by the time the next one updates the reference, so this
+    // avoids accumulating a handler per attempt.
+    AtomicReference<@Nullable CompletableFuture<T>> currentDispatched = new AtomicReference<>();
+    result.whenComplete(
+        (r, t) -> {
+          if (t instanceof CancellationException) {
+            CompletableFuture<T> inFlight = currentDispatched.get();
+            if (inFlight != null && !inFlight.isDone()) {
+              inFlight.cancel(false);
+            }
+          }
+        });
+    attempt(spec, responseType, 0, result, currentDispatched);
     return result;
   }
 
   private <T> void attempt(
-      RequestSpec spec, Class<T> responseType, int attemptIdx, CompletableFuture<T> result) {
+      RequestSpec spec,
+      Class<T> responseType,
+      int attemptIdx,
+      CompletableFuture<T> result,
+      AtomicReference<@Nullable CompletableFuture<T>> currentDispatched) {
     if (result.isDone()) {
       // Caller cancelled (or completed exceptionally from a previous attempt's whenComplete).
       // Don't burn another HTTP request.
       return;
     }
     CompletableFuture<T> dispatched = executeOnce(spec, responseType);
-
-    // Cascade cancel from `result` (returned to caller) to `dispatched` (single-attempt internal
-    // future). Without this, a caller cancelling `result` would leave a slow-path waiter alive
-    // in the semaphore queue: a later release() would transfer the permit but the dispatch
-    // function would never run, leaking the permit. The cascade re-enters executeOnce's own
-    // CancellationException handler, which cancels the waiter so release() skips it.
-    result.whenComplete(
-        (r, t) -> {
-          if (t instanceof CancellationException && !dispatched.isDone()) {
-            dispatched.cancel(false);
-          }
-        });
+    currentDispatched.set(dispatched);
+
+    // If the caller cancelled `result` between attempts (during a backoff window), the handler
+    // installed in executeAsync has fired but `currentDispatched` was either null or pointing
+    // to the previous (already-done) attempt — so the new one was never cancelled. Check here
+    // and propagate immediately.
+    if (result.isCancelled() && !dispatched.isDone()) {
+      dispatched.cancel(false);
+      return;
+    }
 
     dispatched.whenComplete(
         (value, error) -> {
@@ -161,7 +177,8 @@ private <T> void attempt(
           if (retryPolicy.shouldRetry(cause, attemptIdx)) {
             long delayMs = retryPolicy.backoffDelay(attemptIdx).toMillis();
             CompletableFuture.delayedExecutor(delayMs, TimeUnit.MILLISECONDS)
-                .execute(() -> attempt(spec, responseType, attemptIdx + 1, result));
+                .execute(
+                    () -> attempt(spec, responseType, attemptIdx + 1, result, currentDispatched));
           } else {
             result.completeExceptionally(cause);
           }

src/main/java/com/marketdata/sdk/RetryPolicy.java

@@ -3,13 +3,20 @@
 import com.marketdata.sdk.exception.MarketDataException;
 import com.marketdata.sdk.exception.NetworkError;
 import com.marketdata.sdk.exception.ServerError;
+import java.io.IOException;
 import java.time.Duration;
 
 /**
  * Decides which failures get retried and how long to wait between attempts. SDK requirements §9
  * fixes the parameters: max 3 attempts total (one initial + two retries), exponential backoff
- * starting at 1s, capped at 30s. Network errors and HTTP 501–599 are retriable; 500 specifically is
- * not. Everything 4xx (including 401/429) surfaces immediately.
+ * starting at 1s, capped at 30s. Network errors (only when wrapping an {@link IOException}-shaped
+ * cause — see {@link #shouldRetry}) and HTTP 501–599 are retriable; 500 specifically is not, and
+ * 4xx (including 401/429) surfaces immediately.
+ *
+ * <p><strong>Worst-case wall-clock per {@code executeAsync} call (defaults):</strong> 3 attempts ×
+ * 99s per-request timeout + 1s + 2s backoff ≈ 5 minutes. SDK requirements §10 only mandates the
+ * per-request timeout, not an overall deadline, so this is compliant — but callers in
+ * latency-sensitive contexts may want to wrap calls with their own {@code orTimeout} cap.
  *
  * <p>The constructor accepts custom values so tests can drive retries with sub-millisecond delays
  * without waiting on real wall-clock backoffs.
@@ -53,17 +60,18 @@ boolean shouldRetry(Throwable cause, int attempt) {
   Duration backoffDelay(int attempt) {
     long base = initialBackoff.toMillis();
     long max = maxBackoff.toMillis();
-    // attempt 0 → base * 1, attempt 1 → base * 2, attempt N → base * 2^N. Long arithmetic with
-    // an explicit cap because shifting >= 63 bits is undefined in Java; also avoids overflow if
-    // a misconfigured policy used a huge attempt count.
-    long delay;
+    // Two saturation points: (1) for large attempt indices, the shift `1L << N` would silently
+    // wrap once N >= 63 (Java masks the shift count to its low 6 bits), and (2) for moderate
+    // indices, `base * 2^attempt` can overflow Long before we get a chance to cap. (1) is
+    // handled by the early return; (2) by the rearranged inequality
+    // `base > max / multiplier ⇔ base * multiplier > max`, which detects overflow without
+    // actually overflowing.
     if (attempt >= 62) {
-      delay = max;
-    } else {
-      long multiplier = 1L << attempt;
-      delay = (base > max / multiplier) ? max : base * multiplier;
+      return Duration.ofMillis(max);
     }
-    return Duration.ofMillis(Math.min(delay, max));
+    long multiplier = 1L << Math.max(attempt, 0);
+    long delay = (base > max / multiplier) ? max : base * multiplier;
+    return Duration.ofMillis(delay);
   }
 
   private static boolean isRetriable(Throwable cause) {
@@ -72,8 +80,12 @@ private static boolean isRetriable(Throwable cause) {
       // exception rather than an amplified series of identical hits.
       return false;
     }
-    if (cause instanceof NetworkError) {
-      return true;
+    if (cause instanceof NetworkError net) {
+      // NetworkError wraps two shapes: actual transport failures (IOException + subtypes:
+      // ConnectException, HttpTimeoutException, ...) and sync-throws from httpClient.sendAsync
+      // (NPE, IllegalArgumentException — bugs, not network). Retry only the former; the latter
+      // is deterministic and just burns the backoff for the same crash.
+      return net.getCause() instanceof IOException;
     }
     if (cause instanceof ServerError server) {
       Integer status = server.getStatusCode();

src/test/java/com/marketdata/sdk/HttpTransportRetryTest.java

@@ -178,6 +178,101 @@ void exhaustedRetriesOnNetworkErrorsPropagatesLastError() {
     assertThat(client.callCount()).isEqualTo(3);
   }
 
+  // ---------- sync-throw bugs do NOT retry ----------
+
+  /**
+   * If {@code httpClient.sendAsync} throws synchronously (malformed request, internal NPE, {@code
+   * IllegalArgumentException}), the failure is wrapped as {@code NetworkError} but its cause is not
+   * an {@link IOException}. {@link RetryPolicy} treats that as non-retriable: a deterministic bug
+   * doesn't get better with 1s+2s of backoff.
+   */
+  @Test
+  void synchronousThrowDoesNotRetry() {
+    SyncThrowingHttpClient client = new SyncThrowingHttpClient();
+    HttpTransport transport =
+        new HttpTransport("http://stub.local", "v1", "test/0.0", null, client, fastPolicy(3));
+
+    assertThatThrownBy(() -> transport.executeSync(RequestSpec.get("ping").build(), Echo.class))
+        .isInstanceOf(NetworkError.class)
+        .hasMessageContaining("before dispatch")
+        .hasCauseInstanceOf(IllegalArgumentException.class);
+
+    assertThat(client.callCount())
+        .as("a sync-throw is deterministic — retrying just burns backoff for the same crash")
+        .isEqualTo(1);
+  }
+
+  // ---------- rate-limit snapshot consistency under retry ----------
+
+  /**
+   * If attempt 1 returns 503 with rate-limit headers and attempt 2 returns 200 without them, the
+   * snapshot must reflect attempt 1's values (Issue #4 conservation rule applies cross-attempt, not
+   * just cross-request).
+   */
+  @Test
+  void rateLimitSnapshotPreservedAcrossRetryAttempts() {
+    MultiResponseHttpClient client =
+        new MultiResponseHttpClient(
+            response(
+                503,
+                "{}",
+                Map.of(
+                    "x-api-ratelimit-limit", "50000",
+                    "x-api-ratelimit-remaining", "12345",
+                    "x-api-ratelimit-reset", "1735689600",
+                    "x-api-ratelimit-consumed", "37655")),
+            response(200, "{\"value\":\"ok\"}", Map.of()));
+
+    HttpTransport transport = newTransport(client, fastPolicy(3));
+    transport.executeSync(RequestSpec.get("ping").build(), Echo.class);
+
+    RateLimits snapshot = transport.getLatestRateLimits();
+    assertThat(snapshot).isNotNull();
+    assertThat(snapshot.remaining())
+        .as("the snapshot must keep the headers from the 503 attempt, not be cleared by the 200")
+        .isEqualTo(12345L);
+  }
+
+  // ---------- mid-backoff cancellation ----------
+
+  /**
+   * Cancelling the returned future while a backoff is pending must (a) skip the next attempt and
+   * (b) leave the permit pool intact. The cascade-cancel chain is the trickiest piece of {@link
+   * HttpTransport}; this test is the explicit regression for it.
+   */
+  @Test
+  void cancellationMidBackoffSkipsRemainingAttempts() throws Exception {
+    // Use a slow policy so we have a real backoff window to cancel into. 200 ms is short enough
+    // to keep the test fast but long enough to reliably interleave the cancel.
+    RetryPolicy slowPolicy = new RetryPolicy(3, Duration.ofMillis(200), Duration.ofSeconds(1));
+    MultiResponseHttpClient client =
+        new MultiResponseHttpClient(
+            response(503, "{}"), response(503, "{}"), response(200, "{\"value\":\"ok\"}"));
+    HttpTransport transport = newTransport(client, slowPolicy);
+
+    java.util.concurrent.CompletableFuture<Echo> future =
+        transport.executeAsync(RequestSpec.get("ping").build(), Echo.class);
+
+    // Let attempt 1 run and fail (503 → schedule retry with 200 ms backoff). Then cancel before
+    // the delayedExecutor fires the second attempt.
+    Thread.sleep(50);
+    boolean cancelled = future.cancel(false);
+    assertThat(cancelled).isTrue();
+
+    // Give the would-be next attempt plenty of time to fire if cancellation didn't stop it.
+    Thread.sleep(400);
+
+    assertThat(client.callCount())
+        .as("after mid-backoff cancellation, no further attempts may run")
+        .isEqualTo(1);
+
+    AsyncSemaphore permits = readSemaphore(transport);
+    assertThat(permits.availablePermits())
+        .as("permit lent to attempt 1 must have come back to the pool")
+        .isEqualTo(HttpTransport.CONCURRENCY_LIMIT);
+    assertThat(permits.queueLength()).isZero();
+  }
+
   // ---------- permits are still conserved across retries ----------
 
   @Test
@@ -204,7 +299,12 @@ private static AsyncSemaphore readSemaphore(HttpTransport t) throws Exception {
   }
 
   private static Supplier<CompletableFuture<HttpResponse<byte[]>>> response(int code, String body) {
-    return () -> CompletableFuture.completedFuture(new StubHttpResponse(code, body, Map.of()));
+    return response(code, body, Map.of());
+  }
+
+  private static Supplier<CompletableFuture<HttpResponse<byte[]>>> response(
+      int code, String body, Map<String, String> headers) {
+    return () -> CompletableFuture.completedFuture(new StubHttpResponse(code, body, headers));
   }
 
   private static Supplier<CompletableFuture<HttpResponse<byte[]>>> failedResponse(Throwable t) {
@@ -307,6 +407,90 @@ public WebSocket.Builder newWebSocketBuilder() {
     }
   }
 
+  /**
+   * Stub {@link HttpClient} whose {@code sendAsync} throws {@link IllegalArgumentException}
+   * synchronously. Used by {@link #synchronousThrowDoesNotRetry()} to drive the pre-dispatch-fault
+   * path.
+   */
+  private static final class SyncThrowingHttpClient extends HttpClient {
+    private int callCount = 0;
+
+    int callCount() {
+      return callCount;
+    }
+
+    @Override
+    public <T> CompletableFuture<HttpResponse<T>> sendAsync(
+        HttpRequest request, HttpResponse.BodyHandler<T> responseBodyHandler) {
+      callCount++;
+      throw new IllegalArgumentException("simulated synchronous throw from sendAsync");
+    }
+
+    @Override
+    public Optional<CookieHandler> cookieHandler() {
+      return Optional.empty();
+    }
+
+    @Override
+    public Optional<Duration> connectTimeout() {
+      return Optional.empty();
+    }
+
+    @Override
+    public Redirect followRedirects() {
+      return Redirect.NEVER;
+    }
+
+    @Override
+    public Optional<ProxySelector> proxy() {
+      return Optional.empty();
+    }
+
+    @Override
+    public SSLContext sslContext() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public SSLParameters sslParameters() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Optional<Authenticator> authenticator() {
+      return Optional.empty();
+    }
+
+    @Override
+    public Version version() {
+      return Version.HTTP_1_1;
+    }
+
+    @Override
+    public Optional<Executor> executor() {
+      return Optional.empty();
+    }
+
+    @Override
+    public <T> HttpResponse<T> send(
+        HttpRequest request, HttpResponse.BodyHandler<T> responseBodyHandler) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public <T> CompletableFuture<HttpResponse<T>> sendAsync(
+        HttpRequest request,
+        HttpResponse.BodyHandler<T> responseBodyHandler,
+        HttpResponse.PushPromiseHandler<T> pushPromiseHandler) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public WebSocket.Builder newWebSocketBuilder() {
+      throw new UnsupportedOperationException();
+    }
+  }
+
   /** Minimal {@link HttpResponse} stub — just the bits {@code HttpTransport} reads. */
   private static final class StubHttpResponse implements HttpResponse<byte[]> {
     private final int status;