Add a proactive client-side rate-limiter pipeline step (token bucket)

[OmarAlJarrah]

Summary

The SDK can react to server-driven backpressure but cannot proactively pace its own outbound traffic. All pacing today is reactive: the retry stack parses Retry-After, retry-after-ms, x-ms-retry-after-ms, and X-RateLimit-Reset and waits only after a response tells it to (RetryAfterParser / BackoffCalculator / RetryRecovery under sdk-core/src/main/kotlin/org/dexpace/sdk/core/pipeline/step/retry/, and http/pipeline/steps/DefaultRetryStep.kt; specified by RETRY-15..RETRY-22 and RECOV-22..RECOV-25). No primitive in sdk-core/src/main limits the rate at which requests leave the client in the first place — there is no token-bucket, throttle, semaphore, or concurrency-limit type in the tree.

This proposes an optional, pure-JVM token-bucket rate-limiter step — an HttpStep with an AsyncHttpStep mirror — that a consumer can install to cap its own request rate. It ships but is not installed by default.

Problem

dexpace is a platform: its consumers are generated service clients, downstream SDKs, and application code built on top of sdk-core. Those consumers routinely talk to APIs that publish a hard request-rate quota (requests/second or requests/minute). Reactive pacing alone forces every one of them to earn that quota by tripping 429s and then backing off — burning quota, latency, and (on metered APIs) money on requests that were doomed the moment they left the client. During a burst, N concurrent callers all fire, all get 429, and all back off together; X-RateLimit-Reset jitter (RECOV-25) softens the retry stampede but does nothing about the initial overshoot.

A proactive throttle closes that gap: it smooths the outbound rate so the client stays under a known quota instead of discovering it by rejection. It composes with the reactive stack rather than replacing it — the token bucket keeps steady-state traffic beneath the ceiling; Retry-After / X-RateLimit-Reset handling remains the correction path for what the bucket cannot predict (server-side quota changes, quotas shared across processes).

Because this is a core capability gap, every consumer that needs it today has to build its own limiter and bolt it on outside the pipeline — losing per-call option propagation, the shared cancellation/interrupt semantics, and the deterministic stage ordering the pipeline already guarantees (PIPE-1, PIPE-11, PIPE-17). Solving it once as a first-class pipeline step means every generated client and downstream SDK inherits it, correctly and identically.

Proposed approach

Add a rate-limiter step to sdk-core implemented as a time-based token bucket (permits refill at a configured rate up to a burst capacity). Prefer this over a max-in-flight concurrency limiter: a concurrency cap largely duplicates the connection-pool bound that transports already own (OkHttp / java.net.http), whereas a rate-over-time bound is what quotas are actually expressed in and what nothing in the stack currently provides.

Constraints, all satisfiable without leaving the near-zero-dep core:

• Pure JVM, zero new dependencies. A token bucket is a timestamp plus a permit count guarded by a lock; no third-party runtime is required, so this belongs in sdk-core and stays consistent with NFR-1 (core depends only on the stdlib plus a compile-only logging facade). If a future variant ever wants a heavyweight limiter library, that variant — not this step — goes in a separate opt-in adapter module per NFR-2. This step needs no such module.
• ReentrantLock, not synchronized, per the repo convention, so the acquire path never pins a carrier thread under virtual threads.
• Interrupt-aware blocking. The synchronous acquire wait must honor the SDK's cancellation contract: catch InterruptedException, restore the interrupt flag via Thread.currentThread().interrupt(), and throw InterruptedIOException — the exact pattern in RetryRecovery.awaitDelay (sdk-core/src/main/kotlin/org/dexpace/sdk/core/pipeline/step/retry/RetryRecovery.kt:342-373), required by XCUT-1 and XCUT-3. A cancelled acquire must abort near-immediately and must not be reclassified as a timeout.
• Async mirror. Provide an AsyncHttpStep whose processAsync inserts the wait as a scheduled, non-blocking delay rather than parking a thread — composing Futures.delay(scheduler, delay) (sdk-core/src/main/kotlin/org/dexpace/sdk/core/util/Futures.kt:71, already used by DefaultAsyncRetryStep) with thenCompose. Per the AsyncHttpStep contract and PIPE-29, it must return an exceptionally-completed future for wait/cancel failures rather than throwing synchronously.
• Fits the existing step model. HttpStep.process(request, next) (sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/HttpStep.kt:38) and AsyncHttpStep.processAsync (.../AsyncHttpStep.kt:45): the step acquires a permit, then calls next (or the async equivalent). It is a non-pillar step, so consumers install it via HttpPipelineBuilder.append / prepend (.../HttpPipelineBuilder.kt:56, :59); it is deliberately absent from the appendStandardResilience() preset (.../HttpPipelineBuilder.kt:160), keeping it opt-in. The limiter instance holds shared mutable state (bucket level plus last-refill timestamp) guarded by its lock; it keeps no per-request state on the step, so it satisfies the rule that steps are shared across concurrent calls and must be thread-safe (PIPE-11).

Stage placement is the main design decision to settle in review. A limiter at Stage.PRE_SEND (order 1200) consumes one permit per actual wire send, so retried attempts and redirect hops each draw from the bucket — matching how a server counts requests against a quota. A limiter at Stage.PRE_REDIRECT (order 50) throttles once per logical call regardless of downstream retries/redirects. PRE_SEND is the better default (it is quota-accurate, and the reactive backoff still handles anything it under-counts); the alternative should be documented. A related question is whether the acquire wait should be clamped against the caller's remaining total-timeout budget the way retry pacing is (RECOV-22) — worth deciding rather than blocking indefinitely inside a deadline'd call.

Scope and non-goals

In scope:

• One token-bucket limiter primitive plus its sync HttpStep and async AsyncHttpStep, in sdk-core, with no new dependencies.
• Configurable refill rate and burst capacity; interrupt-/cancellation-aware acquire; a bounded (or explicitly documented) wait behavior.
• Shipped opt-in, not wired into appendStandardResilience().

Out of scope:

• Circuit breaker. Failure-ratio tracking, open/half-open/closed state, and probe logic are a distinct, larger concern; track separately.
• Max-in-flight / concurrency limiter and bulkheads — deliberately not chosen here (overlap with transport connection pools).
• Distributed / cross-process rate limiting (shared quota coordinated via an external store). Any such backend would be a separate opt-in adapter module (NFR-2), never core.
• Changes to the existing reactive pacing stack (RETRY-15..22, RECOV-22..25); the token bucket complements it and leaves it untouched.

Acceptance criteria

• A token-bucket limiter primitive lives in sdk-core with no new runtime dependency (NFR-1 preserved); refill rate and burst capacity are configurable.
• A synchronous HttpStep acquires a permit before invoking next; concurrent calls through the shared step are correctly serialized on the bucket (PIPE-11).
• The synchronous acquire wait is interrupt-aware: on InterruptedException it restores the interrupt flag and throws InterruptedIOException, aborts near-immediately, and is never reclassified as a timeout (XCUT-1, XCUT-3).
• An AsyncHttpStep mirror inserts the wait as a non-blocking scheduled delay (e.g. via Futures.delay) and surfaces wait/cancel failures through the returned future rather than throwing synchronously (PIPE-29); it reuses the same Stage identity as the sync step (PIPE-28).
• The step is not part of appendStandardResilience(); installing it is an explicit append / prepend opt-in.
• Stage placement is chosen and documented, including whether a permit is consumed per wire send or per logical call, and how the acquire wait interacts with a total-timeout deadline.
• The limiter uses ReentrantLock (no synchronized), so the acquire path does not pin a carrier thread under virtual threads.
• Tests cover: steady-state pacing under a configured rate; burst up to capacity then throttling; interrupt during a sync acquire (correct signal plus flag preserved); async cancellation during a pending delay; concurrent contention on one shared step instance; and composition with the existing retry / Retry-After stack.
• apiCheck snapshots regenerated for the new public surface; docs/pipelines.md updated to describe the step and its placement.

References

• Spec (docs/product-spec.md): NFR-1 (core depends only on the stdlib plus a compile-only logging facade), NFR-2 (each optional capability is a separately installable unit depending on the core plus at most one third-party library; third-party runtimes stay out of core); PIPE-11 (steps shared across calls must be thread-safe, with per-request state off the step), PIPE-12 (bidirectional step, may short-circuit), PIPE-28 / PIPE-29 (async runtime reuses stage identities; async steps signal failure via the future); XCUT-1 / XCUT-3 (cancellation is terminal / non-retryable with the flag preserved; inter-attempt waits promptly cancellable and implemented as a non-pinning scheduled timer); RETRY-15..RETRY-22 and RECOV-22..RECOV-25 (existing reactive pacing this complements, not replaces).
• Source touchpoints: sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/HttpStep.kt (process at :38), .../AsyncHttpStep.kt (processAsync at :45), .../Stage.kt (PRE_REDIRECT order 50, PRE_SEND order 1200), .../HttpPipelineBuilder.kt (append / prepend at :56 / :59, appendStandardResilience() at :160); sdk-core/src/main/kotlin/org/dexpace/sdk/core/pipeline/step/retry/RetryRecovery.kt:342 (interrupt-aware wait pattern to mirror); sdk-core/src/main/kotlin/org/dexpace/sdk/core/util/Futures.kt:71 (Futures.delay for the async mirror).
• Related: circuit breaker (separate, larger issue — out of scope here); the existing retry / pacing stack under sdk-core/src/main/kotlin/org/dexpace/sdk/core/pipeline/step/retry/.

Note: ReentrantLock-over-synchronized is an established codebase concurrency convention (to avoid pinning virtual-thread carriers under Loom), not a numbered spec requirement; the spec's non-pinning language lives in XCUT-3 (the scheduled-timer async wait), which the async mirror should follow.