diff --git a/CHANGELOG.md b/CHANGELOG.md index fbfd253..3b35e92 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/CLAUDE.md b/CLAUDE.md index 07cb5ca..4c89e15 100644 --- a/CLAUDE.md +++ b/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().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 diff --git a/src/main/java/com/marketdata/sdk/HttpTransport.java b/src/main/java/com/marketdata/sdk/HttpTransport.java index fc4692f..4549556 100644 --- a/src/main/java/com/marketdata/sdk/HttpTransport.java +++ b/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,6 +76,17 @@ 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; @@ -81,6 +94,7 @@ final class HttpTransport implements AutoCloseable { 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. - * - *

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. */ CompletableFuture executeAsync(RequestSpec spec, Class responseType) { + CompletableFuture 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> currentDispatched = new AtomicReference<>(); + result.whenComplete( + (r, t) -> { + if (t instanceof CancellationException) { + CompletableFuture inFlight = currentDispatched.get(); + if (inFlight != null && !inFlight.isDone()) { + inFlight.cancel(false); + } + } + }); + attempt(spec, responseType, 0, result, currentDispatched); + return result; + } + + private void attempt( + RequestSpec spec, + Class responseType, + int attemptIdx, + CompletableFuture result, + AtomicReference<@Nullable CompletableFuture> currentDispatched) { + if (result.isDone()) { + // Caller cancelled (or completed exceptionally from a previous attempt's whenComplete). + // Don't burn another HTTP request. + return; + } + CompletableFuture 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 CompletableFuture executeOnce(RequestSpec spec, Class responseType) { URI uri = buildUri(spec); HttpRequest request = buildRequest(uri); @@ -186,12 +266,19 @@ private CompletableFuture dispatch(URI uri, HttpRequest request, Class /** * Sync wrapper around {@link #executeAsync}. Per ADR-006, calls {@code .join()} and unwraps * {@link CompletionException} so callers see the underlying {@link MarketDataException} directly. + * + *

{@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 executeSync(RequestSpec spec, Class responseType) { try { return executeAsync(spec, responseType).join(); } catch (CompletionException e) { throw asRuntime(e.getCause()); + } catch (CancellationException e) { + throw asRuntime(e); } } diff --git a/src/main/java/com/marketdata/sdk/RetryPolicy.java b/src/main/java/com/marketdata/sdk/RetryPolicy.java new file mode 100644 index 0000000..9b3eb65 --- /dev/null +++ b/src/main/java/com/marketdata/sdk/RetryPolicy.java @@ -0,0 +1,101 @@ +package com.marketdata.sdk; + +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. Per SDK requirements + * §9.3: max 3 retries (yielding 4 total attempts) with exponential backoff {@code initial * + * 2^retry} 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. + * + *

Worst-case wall-clock per {@code executeAsync} call (defaults): 4 attempts × + * 99s per-request timeout + 1s + 2s + 4s backoff ≈ 6.75 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. + * + *

The constructor accepts custom values so tests can drive retries with sub-millisecond delays + * without waiting on real wall-clock backoffs. + */ +final class RetryPolicy { + + private final int maxAttempts; + private final Duration initialBackoff; + private final Duration maxBackoff; + + RetryPolicy(int maxAttempts, Duration initialBackoff, Duration maxBackoff) { + if (maxAttempts < 1) { + throw new IllegalArgumentException("maxAttempts must be >= 1, was " + maxAttempts); + } + this.maxAttempts = maxAttempts; + this.initialBackoff = initialBackoff; + this.maxBackoff = maxBackoff; + } + + /** Defaults: 4 attempts, 1s → 30s exponential. */ + static RetryPolicy defaults() { + return new RetryPolicy(4, Duration.ofSeconds(1), Duration.ofSeconds(30)); + } + + /** + * Whether the SDK should retry after {@code cause}, given that {@code attempt} attempts have + * already been spent (zero-indexed: {@code attempt == 0} means the original call just failed and + * we're considering the first retry). + */ + boolean shouldRetry(Throwable cause, int attempt) { + if (attempt + 1 >= maxAttempts) { + return false; + } + return isRetriable(cause); + } + + /** + * Backoff before the next attempt. {@code attempt == 0} means "before the first retry", i.e. the + * delay applied right after the original call failed. + */ + Duration backoffDelay(int attempt) { + long base = initialBackoff.toMillis(); + long max = maxBackoff.toMillis(); + // 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) { + return Duration.ofMillis(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) { + if (!(cause instanceof MarketDataException)) { + // Conservative: unknown failure types don't get retried. The caller sees the original + // exception rather than an amplified series of identical hits. + return false; + } + 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(); + // Spec §9: 500 is not retriable; 501–599 are. A null status means "we threw a ServerError + // without a real HTTP code" — that's only the synthetic-path of HttpStatusMapper today, so + // don't retry it. + return status != null && status >= 501 && status <= 599; + } + // AuthenticationError, BadRequestError, RateLimitError, NotFoundError, ParseError: §9 says + // never retry 4xx, and ParseError is deterministic. + return false; + } +} diff --git a/src/test/java/com/marketdata/sdk/HttpTransportRetryTest.java b/src/test/java/com/marketdata/sdk/HttpTransportRetryTest.java new file mode 100644 index 0000000..002a44c --- /dev/null +++ b/src/test/java/com/marketdata/sdk/HttpTransportRetryTest.java @@ -0,0 +1,548 @@ +package com.marketdata.sdk; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.fasterxml.jackson.annotation.JsonProperty; +import com.marketdata.sdk.exception.AuthenticationError; +import com.marketdata.sdk.exception.BadRequestError; +import com.marketdata.sdk.exception.NetworkError; +import com.marketdata.sdk.exception.RateLimitError; +import com.marketdata.sdk.exception.ServerError; +import java.io.IOException; +import java.net.Authenticator; +import java.net.CookieHandler; +import java.net.ProxySelector; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpHeaders; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.WebSocket; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.function.Supplier; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLParameters; +import org.junit.jupiter.api.Test; + +/** + * Exercises retry behavior. Uses a scripted {@link HttpClient} stub so a single test can drive a + * sequence of responses (e.g. 503, 503, 200) without spinning up an in-process HTTP server or + * waiting on real backoff durations. + */ +class HttpTransportRetryTest { + + /** Tiny response shape for body-decode assertions. */ + record Echo(@JsonProperty("value") String value) {} + + /** Retry policy with sub-millisecond delays so the suite stays under a second. */ + private static RetryPolicy fastPolicy(int maxAttempts) { + return new RetryPolicy(maxAttempts, Duration.ofMillis(1), Duration.ofMillis(5)); + } + + private static HttpTransport newTransport(MultiResponseHttpClient client, RetryPolicy policy) { + return new HttpTransport("http://stub.local", "v1", "test/0.0", null, client, policy); + } + + // ---------- happy paths ---------- + + @Test + void transientServer5xxRetriesAndEventuallySucceeds() { + MultiResponseHttpClient client = + new MultiResponseHttpClient( + response(503, "{}"), response(503, "{}"), response(200, "{\"value\":\"ok\"}")); + + Echo result = + newTransport(client, fastPolicy(3)) + .executeSync(RequestSpec.get("ping").build(), Echo.class); + + assertThat(result.value()).isEqualTo("ok"); + assertThat(client.callCount()).isEqualTo(3); + } + + @Test + void networkFailuresRetryAndEventuallySucceed() { + MultiResponseHttpClient client = + new MultiResponseHttpClient( + failedResponse(new IOException("connect refused")), + failedResponse(new IOException("connect refused")), + response(200, "{\"value\":\"ok\"}")); + + Echo result = + newTransport(client, fastPolicy(3)) + .executeSync(RequestSpec.get("ping").build(), Echo.class); + + assertThat(result.value()).isEqualTo("ok"); + assertThat(client.callCount()).isEqualTo(3); + } + + // ---------- non-retriable paths fail immediately ---------- + + @Test + void status500FailsImmediatelyWithoutRetry() { + MultiResponseHttpClient client = new MultiResponseHttpClient(response(500, "{}")); + + assertThatThrownBy( + () -> + newTransport(client, fastPolicy(3)) + .executeSync(RequestSpec.get("ping").build(), Echo.class)) + .isInstanceOf(ServerError.class); + + // Exactly one attempt — 500 is in the retriable status space but the spec specifically + // excludes it (see §9: "501-599 retry; 500 no retry"). + assertThat(client.callCount()).isEqualTo(1); + } + + @Test + void authenticationErrorFailsImmediately() { + MultiResponseHttpClient client = new MultiResponseHttpClient(response(401, "{}")); + + assertThatThrownBy( + () -> + newTransport(client, fastPolicy(3)) + .executeSync(RequestSpec.get("ping").build(), Echo.class)) + .isInstanceOf(AuthenticationError.class); + assertThat(client.callCount()).isEqualTo(1); + } + + @Test + void badRequestFailsImmediately() { + MultiResponseHttpClient client = new MultiResponseHttpClient(response(400, "{}")); + + assertThatThrownBy( + () -> + newTransport(client, fastPolicy(3)) + .executeSync(RequestSpec.get("ping").build(), Echo.class)) + .isInstanceOf(BadRequestError.class); + assertThat(client.callCount()).isEqualTo(1); + } + + @Test + void rateLimitErrorFailsImmediately() { + // Spec §9 explicitly says "Never retry rate limit errors." Even though the API may send + // Retry-After on 429, the SDK propagates immediately rather than blocking the caller. + MultiResponseHttpClient client = new MultiResponseHttpClient(response(429, "{}")); + + assertThatThrownBy( + () -> + newTransport(client, fastPolicy(3)) + .executeSync(RequestSpec.get("ping").build(), Echo.class)) + .isInstanceOf(RateLimitError.class); + assertThat(client.callCount()).isEqualTo(1); + } + + // ---------- exhaustion ---------- + + @Test + void exhaustedRetriesPropagatesLastError() { + // 4 stub responses — only 3 should be consumed before maxAttempts is hit and we give up. + MultiResponseHttpClient client = + new MultiResponseHttpClient( + response(503, "{}"), response(503, "{}"), response(503, "{}"), response(503, "{}")); + + assertThatThrownBy( + () -> + newTransport(client, fastPolicy(3)) + .executeSync(RequestSpec.get("ping").build(), Echo.class)) + .isInstanceOf(ServerError.class) + .satisfies(t -> assertThat(((ServerError) t).getStatusCode()).isEqualTo(503)); + + assertThat(client.callCount()) + .as("maxAttempts=3 must cap total calls — including the original attempt") + .isEqualTo(3); + } + + @Test + void exhaustedRetriesOnNetworkErrorsPropagatesLastError() { + MultiResponseHttpClient client = + new MultiResponseHttpClient( + failedResponse(new IOException("kaboom")), + failedResponse(new IOException("kaboom")), + failedResponse(new IOException("kaboom"))); + + assertThatThrownBy( + () -> + newTransport(client, fastPolicy(3)) + .executeSync(RequestSpec.get("ping").build(), Echo.class)) + .isInstanceOf(NetworkError.class); + + 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 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 + void permitsReturnToPoolAfterEveryAttemptRegardlessOfOutcome() throws Exception { + MultiResponseHttpClient client = + new MultiResponseHttpClient( + response(503, "{}"), response(503, "{}"), response(200, "{\"value\":\"ok\"}")); + HttpTransport transport = newTransport(client, fastPolicy(3)); + + transport.executeSync(RequestSpec.get("ping").build(), Echo.class); + + AsyncSemaphore permits = readSemaphore(transport); + assertThat(permits.availablePermits()) + .as("after a 3-attempt retry chain, every permit must be back in the pool") + .isEqualTo(HttpTransport.CONCURRENCY_LIMIT); + } + + // ---------- helpers ---------- + + private static AsyncSemaphore readSemaphore(HttpTransport t) throws Exception { + java.lang.reflect.Field f = HttpTransport.class.getDeclaredField("concurrencyPermits"); + f.setAccessible(true); + return (AsyncSemaphore) f.get(t); + } + + private static Supplier>> response(int code, String body) { + return response(code, body, Map.of()); + } + + private static Supplier>> response( + int code, String body, Map headers) { + return () -> CompletableFuture.completedFuture(new StubHttpResponse(code, body, headers)); + } + + private static Supplier>> failedResponse(Throwable t) { + return () -> CompletableFuture.failedFuture(t); + } + + /** + * {@link HttpClient} that returns scripted responses in order. Each invocation of {@code + * sendAsync} pops the next supplier and invokes it. Running out of script entries throws — that + * surfaces "we retried more times than the test expected" as a clear failure. + */ + private static final class MultiResponseHttpClient extends HttpClient { + private final Deque>>> script; + private int callCount = 0; + + @SafeVarargs + MultiResponseHttpClient(Supplier>>... responses) { + this.script = new ArrayDeque<>(List.of(responses)); + } + + int callCount() { + return callCount; + } + + @SuppressWarnings("unchecked") + @Override + public CompletableFuture> sendAsync( + HttpRequest request, HttpResponse.BodyHandler responseBodyHandler) { + callCount++; + Supplier>> next = script.pollFirst(); + if (next == null) { + return CompletableFuture.failedFuture( + new AssertionError("Retry overshot the test script — call #" + callCount)); + } + return (CompletableFuture>) (CompletableFuture) next.get(); + } + + @Override + public Optional cookieHandler() { + return Optional.empty(); + } + + @Override + public Optional connectTimeout() { + return Optional.empty(); + } + + @Override + public Redirect followRedirects() { + return Redirect.NEVER; + } + + @Override + public Optional proxy() { + return Optional.empty(); + } + + @Override + public SSLContext sslContext() { + throw new UnsupportedOperationException(); + } + + @Override + public SSLParameters sslParameters() { + throw new UnsupportedOperationException(); + } + + @Override + public Optional authenticator() { + return Optional.empty(); + } + + @Override + public Version version() { + return Version.HTTP_1_1; + } + + @Override + public Optional executor() { + return Optional.empty(); + } + + @Override + public HttpResponse send( + HttpRequest request, HttpResponse.BodyHandler responseBodyHandler) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture> sendAsync( + HttpRequest request, + HttpResponse.BodyHandler responseBodyHandler, + HttpResponse.PushPromiseHandler pushPromiseHandler) { + throw new UnsupportedOperationException(); + } + + @Override + public WebSocket.Builder newWebSocketBuilder() { + throw new UnsupportedOperationException(); + } + } + + /** + * 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 CompletableFuture> sendAsync( + HttpRequest request, HttpResponse.BodyHandler responseBodyHandler) { + callCount++; + throw new IllegalArgumentException("simulated synchronous throw from sendAsync"); + } + + @Override + public Optional cookieHandler() { + return Optional.empty(); + } + + @Override + public Optional connectTimeout() { + return Optional.empty(); + } + + @Override + public Redirect followRedirects() { + return Redirect.NEVER; + } + + @Override + public Optional proxy() { + return Optional.empty(); + } + + @Override + public SSLContext sslContext() { + throw new UnsupportedOperationException(); + } + + @Override + public SSLParameters sslParameters() { + throw new UnsupportedOperationException(); + } + + @Override + public Optional authenticator() { + return Optional.empty(); + } + + @Override + public Version version() { + return Version.HTTP_1_1; + } + + @Override + public Optional executor() { + return Optional.empty(); + } + + @Override + public HttpResponse send( + HttpRequest request, HttpResponse.BodyHandler responseBodyHandler) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture> sendAsync( + HttpRequest request, + HttpResponse.BodyHandler responseBodyHandler, + HttpResponse.PushPromiseHandler 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 { + private final int status; + private final byte[] body; + private final HttpHeaders headers; + + StubHttpResponse(int status, String body, Map headers) { + this.status = status; + this.body = body.getBytes(StandardCharsets.UTF_8); + Map> multi = new java.util.HashMap<>(); + headers.forEach((k, v) -> multi.put(k, new ArrayList<>(List.of(v)))); + this.headers = HttpHeaders.of(multi, (a, b) -> true); + } + + @Override + public int statusCode() { + return status; + } + + @Override + public HttpRequest request() { + return null; + } + + @Override + public Optional> previousResponse() { + return Optional.empty(); + } + + @Override + public HttpHeaders headers() { + return headers; + } + + @Override + public byte[] body() { + return body; + } + + @Override + public Optional sslSession() { + return Optional.empty(); + } + + @Override + public URI uri() { + return URI.create("http://stub.local/v1/ping/"); + } + + @Override + public HttpClient.Version version() { + return HttpClient.Version.HTTP_1_1; + } + } +} diff --git a/src/test/java/com/marketdata/sdk/HttpTransportTest.java b/src/test/java/com/marketdata/sdk/HttpTransportTest.java index fe94e71..bd69e6a 100644 --- a/src/test/java/com/marketdata/sdk/HttpTransportTest.java +++ b/src/test/java/com/marketdata/sdk/HttpTransportTest.java @@ -17,6 +17,9 @@ import java.util.ArrayList; import java.util.List; import java.util.Optional; +// These tests cover the SINGLE-ATTEMPT semantics of executeAsync. Retry behavior is exercised +// separately in HttpTransportRetryTest; here we explicitly disable retry so a permit-release +// assertion reflects exactly one HTTP call per executeAsync invocation. import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.Executor; @@ -26,6 +29,10 @@ class HttpTransportTest { + /** Policy with a single attempt — disables retry so each test asserts one HTTP call only. */ + private static final RetryPolicy NO_RETRY = + new RetryPolicy(1, Duration.ofMillis(1), Duration.ofMillis(1)); + /** * Regression for the synchronous-throw permit leak: if {@code httpClient.sendAsync(...)} throws * before returning a future (rare but possible — malformed request, internal NPE, OOM), the @@ -40,7 +47,8 @@ class HttpTransportTest { @Test void permitReleasedWhenSendAsyncThrowsSynchronously() throws Exception { HttpTransport transport = - new HttpTransport("http://localhost", "v1", "test/0.0", null, new SyncThrowingHttpClient()); + new HttpTransport( + "http://localhost", "v1", "test/0.0", null, new SyncThrowingHttpClient(), NO_RETRY); AsyncSemaphore permits = readSemaphore(transport); int initial = permits.availablePermits(); @@ -76,7 +84,7 @@ void permitReleasedWhenSendAsyncThrowsSynchronously() throws Exception { void errorThrownSynchronouslyIsPreservedAsRootCause() throws Exception { HttpTransport transport = new HttpTransport( - "http://localhost", "v1", "test/0.0", null, new ErrorThrowingHttpClient()); + "http://localhost", "v1", "test/0.0", null, new ErrorThrowingHttpClient(), NO_RETRY); AsyncSemaphore permits = readSemaphore(transport); int initial = permits.availablePermits(); @@ -112,7 +120,8 @@ void errorThrownSynchronouslyIsPreservedAsRootCause() throws Exception { @Test void permitsAreReleasedWhenSlowPathFuturesAreCancelled() throws Exception { ControllableHttpClient client = new ControllableHttpClient(); - HttpTransport transport = new HttpTransport("http://localhost", "v1", "test/0.0", null, client); + HttpTransport transport = + new HttpTransport("http://localhost", "v1", "test/0.0", null, client, NO_RETRY); AsyncSemaphore permits = readSemaphore(transport); int initial = permits.availablePermits(); diff --git a/src/test/java/com/marketdata/sdk/RetryPolicyTest.java b/src/test/java/com/marketdata/sdk/RetryPolicyTest.java new file mode 100644 index 0000000..dae21b5 --- /dev/null +++ b/src/test/java/com/marketdata/sdk/RetryPolicyTest.java @@ -0,0 +1,159 @@ +package com.marketdata.sdk; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.marketdata.sdk.exception.AuthenticationError; +import com.marketdata.sdk.exception.BadRequestError; +import com.marketdata.sdk.exception.ErrorContext; +import com.marketdata.sdk.exception.NetworkError; +import com.marketdata.sdk.exception.NotFoundError; +import com.marketdata.sdk.exception.ParseError; +import com.marketdata.sdk.exception.RateLimitError; +import com.marketdata.sdk.exception.ServerError; +import java.time.Duration; +import org.junit.jupiter.api.Test; + +class RetryPolicyTest { + + private static final RetryPolicy DEFAULTS = RetryPolicy.defaults(); + + // ---------- shouldRetry: which errors are retriable ---------- + + @Test + void networkErrorsWithIoCauseAreRetriable() { + // The canonical "real network failure" shape: NetworkError wraps an IOException (or + // subtype like ConnectException / HttpTimeoutException). + NetworkError err = + new NetworkError( + "connect refused", ErrorContext.empty(), new java.io.IOException("connect refused")); + assertThat(DEFAULTS.shouldRetry(err, 0)).isTrue(); + } + + @Test + void networkErrorsWithoutCauseAreNotRetriable() { + // A NetworkError with no cause has no signal that it was an actual network failure. + // Better to surface immediately than burn 3 attempts on a possibly-deterministic bug. + assertThat(DEFAULTS.shouldRetry(new NetworkError("boom", ErrorContext.empty()), 0)).isFalse(); + } + + @Test + void networkErrorsWrappingNonIoCauseAreNotRetriable() { + // Sync-throws from httpClient.sendAsync() (malformed request, internal NPE, + // IllegalArgumentException, etc.) get wrapped as NetworkError in `dispatch`, but they're + // deterministic — retrying just wastes the 1s+2s backoff for the same crash. + NetworkError syncThrow = + new NetworkError( + "Request failed before dispatch", + ErrorContext.empty(), + new IllegalArgumentException("malformed URI")); + assertThat(DEFAULTS.shouldRetry(syncThrow, 0)).isFalse(); + } + + @Test + void status500IsNotRetriable() { + ServerError err = new ServerError("500", new ErrorContext(null, "u", 500)); + assertThat(DEFAULTS.shouldRetry(err, 0)).isFalse(); + } + + @Test + void status501Through599AreRetriable() { + for (int code : new int[] {501, 502, 503, 504, 599}) { + ServerError err = new ServerError("err", new ErrorContext(null, "u", code)); + assertThat(DEFAULTS.shouldRetry(err, 0)).as("status %d should be retriable", code).isTrue(); + } + } + + @Test + void authenticationErrorIsNotRetriable() { + assertThat(DEFAULTS.shouldRetry(new AuthenticationError("a", ErrorContext.empty()), 0)) + .isFalse(); + } + + @Test + void badRequestErrorIsNotRetriable() { + assertThat(DEFAULTS.shouldRetry(new BadRequestError("b", ErrorContext.empty()), 0)).isFalse(); + } + + @Test + void rateLimitErrorIsNotRetriable() { + // Spec §9: "Never retry 4xx or rate limit errors." Even though 429 carries Retry-After in + // some protocols, the SDK contract is to surface RateLimitError to the caller immediately. + assertThat(DEFAULTS.shouldRetry(new RateLimitError("r", ErrorContext.empty()), 0)).isFalse(); + } + + @Test + void notFoundErrorIsNotRetriable() { + assertThat(DEFAULTS.shouldRetry(new NotFoundError("n", ErrorContext.empty()), 0)).isFalse(); + } + + @Test + void parseErrorIsNotRetriable() { + // A bad-shape body is deterministic — retrying produces the same broken decode. + assertThat(DEFAULTS.shouldRetry(new ParseError("p", ErrorContext.empty()), 0)).isFalse(); + } + + @Test + void unknownThrowableIsNotRetriable() { + // Conservative default for non-MarketDataException causes: don't retry. Better to surface + // the unknown failure than to silently hammer the API. + assertThat(DEFAULTS.shouldRetry(new RuntimeException("?"), 0)).isFalse(); + } + + // ---------- shouldRetry: respect max attempts ---------- + + @Test + void retriesStopAfterMaxAttempts() { + NetworkError retriable = + new NetworkError("net", ErrorContext.empty(), new java.io.IOException("transport down")); + // Defaults: maxAttempts = 4 → attempts 0, 1, 2 are eligible to be followed by a retry + // (attempt 3 was the fourth try; no fifth attempt allowed). + assertThat(DEFAULTS.shouldRetry(retriable, 0)).isTrue(); + assertThat(DEFAULTS.shouldRetry(retriable, 1)).isTrue(); + assertThat(DEFAULTS.shouldRetry(retriable, 2)).isTrue(); + assertThat(DEFAULTS.shouldRetry(retriable, 3)).isFalse(); + assertThat(DEFAULTS.shouldRetry(retriable, 99)).isFalse(); + } + + // ---------- backoffDelay: exponential with cap ---------- + + @Test + void backoffStartsAtInitialAndDoubles() { + assertThat(DEFAULTS.backoffDelay(0)).isEqualTo(Duration.ofSeconds(1)); + assertThat(DEFAULTS.backoffDelay(1)).isEqualTo(Duration.ofSeconds(2)); + assertThat(DEFAULTS.backoffDelay(2)).isEqualTo(Duration.ofSeconds(4)); + assertThat(DEFAULTS.backoffDelay(3)).isEqualTo(Duration.ofSeconds(8)); + } + + @Test + void backoffCapsAtMaxBackoff() { + // 2^5 = 32 > 30 cap; 2^10 way over. + assertThat(DEFAULTS.backoffDelay(5)).isEqualTo(Duration.ofSeconds(30)); + assertThat(DEFAULTS.backoffDelay(10)).isEqualTo(Duration.ofSeconds(30)); + } + + @Test + void backoffSaturatesOnExtremeAttemptIndices() { + // The shift `1L << attempt` is undefined for attempt >= 63 (the shift count is masked to + // the bottom 6 bits, wrapping silently); the implementation guards against this by + // capping at maxBackoff once the multiplier would overflow. + assertThat(DEFAULTS.backoffDelay(62)).isEqualTo(Duration.ofSeconds(30)); + assertThat(DEFAULTS.backoffDelay(70)).isEqualTo(Duration.ofSeconds(30)); + assertThat(DEFAULTS.backoffDelay(Integer.MAX_VALUE)).isEqualTo(Duration.ofSeconds(30)); + } + + // ---------- custom-tuned policy (used by tests that need fast retries) ---------- + + @Test + void customConstructorWiresValuesThrough() { + RetryPolicy tiny = + new RetryPolicy(/* maxAttempts */ 5, Duration.ofMillis(1), Duration.ofMillis(10)); + + NetworkError net = + new NetworkError("n", ErrorContext.empty(), new java.io.IOException("transport down")); + assertThat(tiny.shouldRetry(net, 3)).isTrue(); + assertThat(tiny.shouldRetry(net, 4)).isFalse(); + assertThat(tiny.backoffDelay(0)).isEqualTo(Duration.ofMillis(1)); + assertThat(tiny.backoffDelay(1)).isEqualTo(Duration.ofMillis(2)); + assertThat(tiny.backoffDelay(20)).isEqualTo(Duration.ofMillis(10)); + } +}