Merge pull request #6 from MarketDataApp/06_retry_policies

Retry policy (§9 partial) + `executeSync` cancellation handling

Author: MarketDataDev03

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+- Default retry attempts corrected from 3 to 4 (one initial + three retries) to
+  match SDK requirements §9.3 ("max 3 retries, yielding 4 total attempts").
+
 ### Added
 - Project scaffold per ADRs 001–007: Gradle Kotlin DSL build, JDK 17 toolchain,
   `integrationTest` source set, Spotless + JaCoCo, Vanniktech Maven Publish.

CLAUDE.md

@@ -65,15 +65,16 @@ The Java SDK must also satisfy the canonical, cross-language [SDK Requirements](
 - §4 configuration cascade — `Configuration.resolve(...)` does explicit → `MARKETDATA_*` env var → `.env` in CWD → default. Env var names live in `EnvVars` (package-private, in the SDK root package). The 4-arg constructor's parameters feed step 1; the no-arg constructor skips it and starts at step 2.
 - §5 demo mode + `validateOnStartup` parameter on the 4-arg constructor (defaults to `true` via the no-arg constructor); token redaction via `Tokens.redact` (matches the spec example `***…***YKT0`).
 - §6 sealed `MarketDataException` hierarchy with the 7 canonical subtypes and full support context (`requestId`, `requestUrl`, `statusCode`, `timestamp`, `exceptionType`) + `getSupportInfo()`.
-- §10 timeouts: `REQUEST_TIMEOUT = 99s` and `CONNECT_TIMEOUT = 2s` exposed as constants on `MarketDataClient`. Connect timeout is wired into the `HttpClient`; the per-request 99 s timeout is a constant ready to be applied to `HttpRequest.Builder#timeout` when the request layer lands.
-- §12 concurrency: `Semaphore(50)` field on `MarketDataClient` (wiring of acquire/release lands with the request layer).
+- §10 timeouts: `REQUEST_TIMEOUT = 99s` and `CONNECT_TIMEOUT = 2s` exposed as constants on `MarketDataClient`. Connect timeout is wired into the `HttpClient`; the per-request 99 s timeout is applied via `HttpRequest.Builder#timeout` in `HttpTransport.buildRequest`.
+- §12 concurrency: 50-permit `AsyncSemaphore` on `HttpTransport` with acquire/release wired around every dispatch. The custom semaphore replaces `java.util.concurrent.Semaphore` so `executeAsync` never parks the caller's thread on a full pool (ADR-007).
+- §9 retry/backoff: `RetryPolicy` (4 total attempts = 1 initial + 3 retries, exponential 1s→30s per §9.3) wired into `HttpTransport.executeAsync` via a per-attempt loop using `CompletableFuture.delayedExecutor` (no scheduled threads). Network errors and HTTP 501–599 retry; 500 and 4xx do not.
 - §15 packaging: SemVer, MIT `LICENSE`, `CHANGELOG.md` in Keep a Changelog format, version auto-detected via JAR manifest (`Implementation-Version`).
 - §16 security: tokens never logged verbatim (use `Tokens.redact`); TLS validated by default (`HttpClient` does not expose a skip-verify option).
 - ADR-002 CI: split into four workflows.
   - `.github/workflows/pull-request.yml` — runs on PR `opened`/`synchronize`/`reopened` (no pre-PR push trigger by design). JDK 17 only. Runs `./gradlew build` (unit tests + Spotless + JaCoCo) and uploads coverage to Codecov. **Does not** run integration tests — those are handled by the on-demand workflow below.
   - `.github/workflows/main.yml` — runs only on `push` to `main`. Two jobs: `verify` does the full forward-compat matrix `{17, 21, 25}` for unit tests via `-PtestJdk=N`; `integration-tests` does a parallel matrix `{17, 21, 25}` against the live API. Both are mandatory for the merge to be considered successful. The JDK 17 matrix entry of `verify` also uploads coverage to Codecov as the new baseline that PRs compare against. `integration-tests` fails the build if `MARKETDATA_TOKEN` secret is absent (it is required on main).
   - `.github/workflows/pr-matrix-on-demand.yml` — manually triggered on a PR by commenting `/run-all-jdks`, `/jdk-matrix`, or `/test-all`. Runs the **unit-test** matrix on JDK 21 and 25 (17 already ran via `pull-request.yml`). Gated to write/maintain/admin commenters. Reacts 👀 to the trigger comment and posts a result summary.
-  - `.github/workflows/pr-integration-on-demand.yml` — manually triggered on a PR by commenting `integrationtest` (JDK 17 only) or `integrationtestfull` (matrix `{17, 21, 25}`). Runs the **integration-test** suite against the live API. Same write+ permission gate as the matrix-on-demand workflow. Aggregates the matrix outcome into a single required check named **"Integration tests pass"** so branch protection can require it uniformly regardless of which command was used. Branch-protection rules on `main` should list this check as required for merge.
+  - `.github/workflows/pr-integration-on-demand.yml` — manually triggered on a PR by commenting `/integrationtest` (JDK 17 only) or `/integrationtestfull` (matrix `{17, 21, 25}`) on the **first line** of the comment body. Runs the **integration-test** suite against the live API. Same write+ permission gate as the matrix-on-demand workflow. The first-line + exact-match constraint prevents accidental triggers from quoted replies (`> /integrationtest`) or prose that mentions the command. Aggregates the matrix outcome into a single required check named **"Integration tests pass"** so branch protection can require it uniformly regardless of which command was used. Branch-protection rules on `main` should list this check as required for merge.
   - All four `issue_comment`-driven workflows execute from the default branch's copy of their YAML, not the PR's. Feature-branch edits to these workflows take effect only after merge to main.
   - `-PtestJdk=N` is wired to **all** `Test` tasks (`test` and `integrationTest`) via `tasks.withType<Test>().configureEach { javaLauncher.set(...) }` in `build.gradle.kts`, so the matrix flag works uniformly across unit and integration tests.
   - Coverage ratchet lives in `codecov.yml`: project status with `target: auto, threshold: 5%` (cannot drop >5 pp vs base branch) plus a patch-coverage requirement of 70 % on new code. Requires a `CODECOV_TOKEN` repo secret — without it the upload step fails because workflows pass `fail_ci_if_error: true`.
@@ -84,19 +85,14 @@ The Java SDK must also satisfy the canonical, cross-language [SDK Requirements](
 - §5 actual `/user/` startup validation call (the `validateOnStartup` flag is the seam; the call itself comes with the request layer).
 - §7 honoring `MARKETDATA_LOGGING_LEVEL` and the spec's exact `{timestamp} - {logger_name} - {level} - {message}` format. Currently the SDK uses `java.util.logging` with default formatting; consumers can attach their own handler.
 - §8 rate-limit header parsing, pre-flight check, request-scoped attachment.
-- §9 retry/backoff policy and `/status/` cache workflow.
-- §12 acquire/release of the concurrency semaphore around dispatched requests.
+- §9 `/status/` cache workflow and `Retry-After` header override (retry/backoff itself lives in `RetryPolicy` and is wired; what is missing is the `/status/` pre-check before retrying 501–599 and respecting the server-specified `Retry-After` over the calculated exponential backoff).
 - §13 100% coverage threshold via JaCoCo `violationRules`; deferred until there is functional code worth the threshold.
 
 When picking up new work, check this list before reaching for the SDK requirements doc — most foundational rules are already encoded in code; missing pieces are deferred deliberately, not by accident.
 
-**Known latent gaps to revisit when retry/timeout lands:**
-- `HttpTransport.executeSync` only catches `CompletionException` from `.join()`, not `CancellationException`. Today the latter is unreachable — the user can't cancel a future they never see (the future is local to `executeSync`), no internal code cancels it, and `dispatch`'s `handle((response, error) -> ...)` translates every upstream error (including a hypothetical `CancellationException` from `sendAsync`) into `CompletionException(NetworkError)`. The gap becomes real once we add:
-  - `dispatched.orTimeout(99s)` / `completeOnTimeout` to enforce the §10 timeout strictly (these produce `CancellationException` on the downstream future).
-  - A retry coordinator (§9) that cancels in-flight futures when aborting a retry chain.
-  - A bump to JDK 21+ where `HttpClient.close()` cancels in-flight futures.
-  When any of those land, extend the catch in `executeSync` (or fold it into `asRuntime`) so cancellations don't escape as raw `RuntimeException` to sync callers. Tracked as Issue #2 of the 2026-05-11 review (`REVIEW-2026-05-11-markets-status.md`).
+**Known latent gaps:**
 - `HttpTransport.buildUri` URL-encodes query-param values with `URLEncoder.encode(..., UTF_8)`, which is form-encoding semantics: spaces become `+`, not `%20`. Fine for today's typed params (dates, numerics) but a future endpoint that takes an arbitrary string (e.g. `symbol="BRK A"`) would round-trip differently against an RFC-3986-strict server. Switch to a path/query-segment-aware encoder when the first such param lands. Tracked as Issue #10 of the 2026-05-11 review.
+- `Retry-After` server header is parsed and respected by neither `RetryPolicy` nor `HttpTransport`. Today every retry uses the calculated exponential backoff (`min(1s × 2^N, 30s)`). Implementing the override needs the response headers to reach `RetryPolicy.backoffDelay`, which today only sees the attempt index — most natural path is to surface a `Duration` on `ServerError` (or thread it through a separate channel) when 5xx responses carry the header. Follow-up of the §9 work.
 
 ## Acceptance checklist
 

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

@@ -20,6 +20,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import org.jspecify.annotations.Nullable;
 
@@ -55,6 +56,7 @@ final class HttpTransport implements AutoCloseable {
   private final HttpClient httpClient;
   private final ObjectMapper jsonMapper;
   private final AsyncSemaphore concurrencyPermits;
+  private final RetryPolicy retryPolicy;
   private final AtomicReference<@Nullable RateLimits> latestRateLimits = new AtomicReference<>();
 
   private final String baseUrl;
@@ -63,7 +65,7 @@ final class HttpTransport implements AutoCloseable {
   private final @Nullable String token;
 
   HttpTransport(String baseUrl, String apiVersion, String userAgent, @Nullable String token) {
-    this(baseUrl, apiVersion, userAgent, token, defaultHttpClient());
+    this(baseUrl, apiVersion, userAgent, token, defaultHttpClient(), RetryPolicy.defaults());
   }
 
   // Package-private constructor used by tests to inject a stubbed HttpClient
@@ -74,13 +76,25 @@ final class HttpTransport implements AutoCloseable {
       String userAgent,
       @Nullable String token,
       HttpClient httpClient) {
+    this(baseUrl, apiVersion, userAgent, token, httpClient, RetryPolicy.defaults());
+  }
+
+  // Package-private constructor used by retry tests to drive sub-millisecond backoffs.
+  HttpTransport(
+      String baseUrl,
+      String apiVersion,
+      String userAgent,
+      @Nullable String token,
+      HttpClient httpClient,
+      RetryPolicy retryPolicy) {
     this.baseUrl = baseUrl;
     this.apiVersion = apiVersion;
     this.userAgent = userAgent;
     this.token = token;
     this.concurrencyPermits = new AsyncSemaphore(CONCURRENCY_LIMIT);
     this.jsonMapper = buildJsonMapper();
     this.httpClient = httpClient;
+    this.retryPolicy = retryPolicy;
   }
 
   private static HttpClient defaultHttpClient() {
@@ -102,14 +116,80 @@ private static HttpClient defaultHttpClient() {
   }
 
   /**
-   * Async-first request execution.
-   *
-   * <p>Acquires a concurrency permit, fires the request, parses rate-limit headers, decodes the
-   * body when the status is 200/203/404 (the API returns 404 with {@code {"s":"no_data"}} as a
-   * sentinel — see SDK requirements §9.1), and translates other status codes to the appropriate
-   * {@link MarketDataException} subtype.
+   * Async-first request execution with retry. Orchestrates one or more attempts according to {@link
+   * 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<>();
+    // 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,
+      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);
+    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) -> {
+          if (result.isDone()) {
+            return;
+          }
+          if (error == null) {
+            result.complete(value);
+            return;
+          }
+          Throwable cause = unwrap(error);
+          if (retryPolicy.shouldRetry(cause, attemptIdx)) {
+            long delayMs = retryPolicy.backoffDelay(attemptIdx).toMillis();
+            CompletableFuture.delayedExecutor(delayMs, TimeUnit.MILLISECONDS)
+                .execute(
+                    () -> attempt(spec, responseType, attemptIdx + 1, result, currentDispatched));
+          } else {
+            result.completeExceptionally(cause);
+          }
+        });
+  }
+
+  /**
+   * Single-shot dispatch — one HTTP request, one permit lease, one response decode. Public retry
+   * orchestration lives in {@link #executeAsync}.
+   */
+  private <T> CompletableFuture<T> executeOnce(RequestSpec spec, Class<T> responseType) {
     URI uri = buildUri(spec);
     HttpRequest request = buildRequest(uri);
 
@@ -186,12 +266,19 @@ private <T> CompletableFuture<T> dispatch(URI uri, HttpRequest request, Class<T>
   /**
    * Sync wrapper around {@link #executeAsync}. Per ADR-006, calls {@code .join()} and unwraps
    * {@link CompletionException} so callers see the underlying {@link MarketDataException} directly.
+   *
+   * <p>{@link CancellationException} can in principle escape {@code .join()} as a sibling of {@link
+   * CompletionException} (not nested), so it's caught explicitly. Today no internal code cancels
+   * the future {@code executeSync} owns, but covering it keeps the contract honest if a future
+   * change (timeout watchdog, retry coordinator) starts cancelling internally.
    */
   <T> T executeSync(RequestSpec spec, Class<T> responseType) {
     try {
       return executeAsync(spec, responseType).join();
     } catch (CompletionException e) {
       throw asRuntime(e.getCause());
+    } catch (CancellationException e) {
+      throw asRuntime(e);
     }
   }