Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
18 changes: 7 additions & 11 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
Expand All @@ -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

Expand Down
101 changes: 94 additions & 7 deletions src/main/java/com/marketdata/sdk/HttpTransport.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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;
Expand All @@ -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
Expand All @@ -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() {
Expand All @@ -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);

Expand Down Expand Up @@ -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);
}
}

Expand Down
Loading
Loading