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
81 changes: 81 additions & 0 deletions apps/docs/content/docs/transports.md
Original file line number Diff line number Diff line change
Expand Up @@ -400,3 +400,84 @@ These settings apply independently of message pacing. Configure `maxConnections`
Each quota coordinates only transports in the current process. Separate processes and other hosts behind the same IP
must coordinate externally or choose lower limits to leave room for one another. Connected order/post dispatch is
unchanged; the connection gate only delays creation of sockets.

## Injectable runtimes

Both transports run on the platform by default: the global `fetch` and `WebSocket`, `Date.now`, `performance.now`,
`setTimeout`, and `Math.random`. Each of these can be replaced per transport, without touching the globals, which is
how tests exercise timeouts, retries, and reconnects deterministically and in isolation from one another.

- `HttpTransport` accepts `fetch`.
- `WebSocketTransport` (and `ReconnectingWebSocket`) accept `webSocketFactory`.
- Both, plus `WebSocketQuota` and `InfoCacheTransport`, accept `runtime`: any subset of `now`, `monotonicNow`,
`setTimeout`, `clearTimeout`, `setInterval`, `clearInterval`, and `random`. Omitted members fall back to the
platform, and the result is resolved once at construction, not per request.

The `Runtime` and `TimerHandle` types and `systemRuntime` are exported from `@bloxwap/hyperliquid/transport/runtime`
(and the `@bloxwap/hyperliquid/transport` barrel). Runtime methods keep their receiver, so a stateful fake clock can be
passed directly. A minimal one is enough to drive a 429 retry without sleeping:

```ts
import { expect, test } from "bun:test";
import { HttpTransport } from "@bloxwap/hyperliquid";
import type { Runtime, TimerHandle } from "@bloxwap/hyperliquid/transport/runtime";

/** Timers fire only when the test advances the clock. */
class FakeClock implements Runtime {
elapsed = 0;
private timers = new Map<number, { due: number; callback: () => void }>();
private nextId = 0;
now = () => 1_700_000_000_000 + this.elapsed;
monotonicNow = () => this.elapsed;
random = () => 0; // no jitter
setTimeout = (callback: () => void, ms: number): TimerHandle => {
this.timers.set(++this.nextId, { due: this.elapsed + ms, callback });
return this.nextId;
};
clearTimeout = (handle: TimerHandle | undefined) => void this.timers.delete(handle as number);
setInterval = this.setTimeout; // not needed by HttpTransport
clearInterval = this.clearTimeout;
advance(ms: number) {
this.elapsed += ms;
for (const [id, timer] of [...this.timers].sort((a, b) => a[1].due - b[1].due)) {
if (timer.due > this.elapsed) break;
this.timers.delete(id);
timer.callback();
}
}
}

test("retries a 429 after Retry-After", async () => {
const clock = new FakeClock();
let calls = 0;
const transport = new HttpTransport({
runtime: clock,
retryOnRateLimit: true,
fetch: async () =>
++calls === 1
? new Response("", { status: 429, headers: { "Retry-After": "2" } })
: Response.json({ BTC: "30000" }),
});

const pending = transport.request("info", { type: "allMids" });
await new Promise((resolve) => setImmediate(resolve)); // let the first attempt settle
clock.advance(1_999);
expect(calls).toBe(1); // still waiting out the server's 2 s
clock.advance(1);
expect(await pending).toEqual({ BTC: "30000" });
expect(calls).toBe(2);
});
```

Wall time (`now`) is used only for protocol timestamps: exchange nonces and HTTP-date `Retry-After` headers. Every
elapsed-time decision (request deadlines, rate-limit refills, cache TTLs, connection-attempt windows) reads
`monotonicNow`, so a wall-clock correction neither fires nor stalls them.

Injecting a runtime or factory never forks state that must stay shared:

- A `WebSocketTransport` without an explicit `quota` still uses the per-network shared `WebSocketQuota`, which keeps
the platform clock. To drive connection and message budgets on a fake clock too, pass
`quota: new WebSocketQuota({ runtime })`.
- Exchange nonces still come from the process-wide manager keyed by signer and network, on the platform wall clock, so
HTTP and WebSocket clients for the same signer never collide. To control nonces in a test, pass a `nonceManager`
function to the `ExchangeClient`.
3 changes: 3 additions & 0 deletions packages/hyperliquid/.dev/perf/info_cache_capacity.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,9 @@ try {
for (const mode of ["infinite", "fresh-finite", "mixed-expiry", "hit", "passthrough"]) {
clock = 0;
const cache = new InfoCacheTransport(inner, {
// Explicit elapsed time for the current SDK; the wall-clock shim above also
// keeps comparisons against older SDKs on the same simulated workload.
runtime: { now: () => clock, monotonicNow: () => clock },
maxSize,
ttl: mode === "fresh-finite" ? 60000 : Infinity,
ttlByType: mode === "mixed-expiry" ? { marginTable: 10 } : {},
Expand Down
1 change: 1 addition & 0 deletions packages/hyperliquid/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
"./transport": "./src/transport/mod.ts",
"./transport/http": "./src/transport/http/mod.ts",
"./transport/websocket": "./src/transport/websocket/mod.ts",
"./transport/runtime": "./src/transport/runtime.ts",
"./api/exchange": "./src/api/exchange/mod.ts",
"./api/exchange/client": "./src/api/exchange/client.ts",
"./api/explorer": "./src/api/explorer/mod.ts",
Expand Down
16 changes: 11 additions & 5 deletions packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
* @module
*/

import { resolveRuntime, type Runtime } from "../../../../transport/runtime.ts";

/** Default upper bound on map size before stale entries are pruned. */
const DEFAULT_MAX_ENTRIES = 10_000;

Expand All @@ -24,24 +26,28 @@ export interface NonceManager {
/**
* Creates a nonce manager that issues unique, monotonically increasing nonces per key.
*
* Uses `Date.now()` in ms; if the previous nonce for the key is greater than or equal to
* `Date.now()`, increments by 1 to maintain monotonicity.
* Uses the runtime's wall clock (`Date.now()` by default) in ms; if the previous nonce for the
* key is greater than or equal to it, increments by 1 to maintain monotonicity.
*
* To bound memory under high-cardinality workloads (e.g., a server proxying many wallets),
* stale entries are pruned when the internal map grows beyond `maxEntries`. An entry is
* considered stale if `Date.now()` has advanced past its last issued nonce. The prune scan is
* considered stale if the wall clock has advanced past its last issued nonce. The prune scan is
* throttled to at most one full pass per {@linkcode PRUNE_INTERVAL_MS}, so an over-capacity map
* does not turn every call into an O(n) sweep.
*
* @param maxEntries Upper bound on map size before stale entries are pruned. Default: `10000`.
* @param runtime Clock overrides; only `now` (wall time) is read, since nonces are timestamps.
* @return A {@linkcode NonceManager}.
*/
export function createNonceManager(maxEntries: number = DEFAULT_MAX_ENTRIES): NonceManager {
export function createNonceManager(maxEntries: number = DEFAULT_MAX_ENTRIES, runtime?: Partial<Runtime>): NonceManager {
// `undefined` keeps the default (process-wide) manager on a direct `Date.now()` call: this runs
// once per signed action, and the global manager is created before any test could inject one.
const clock = runtime === undefined ? undefined : resolveRuntime(runtime);
const map = new Map<string, number>();
let lastPruneAt = 0;
return {
getNonce(key: string): number {
const now = Date.now();
const now = clock === undefined ? Date.now() : clock.now();
// --- Bounded pruning: throttle the scan, never evict -----------------
// gh issue #7: under burst load nonces run AHEAD of the wall clock
// (`nonce = last + 1`), so once the map exceeds `maxEntries` the old
Expand Down
20 changes: 12 additions & 8 deletions packages/hyperliquid/src/transport/_abort.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
*/

import { DOMException_, Promise_ } from "./_polyfills.ts";
import { resolveRuntime, type Runtime, type TimerHandle } from "./runtime.ts";

/** Shared detach function for relays that need no cleanup. */
function noop(): void {
Expand Down Expand Up @@ -33,7 +34,7 @@ const DISABLED_TIMEOUT: { reason: Error; cancel: () => void } = Object.freeze({
/** A pending timeout: one node of the {@linkcode TimeoutWheel}'s deadline-sorted queue. */
interface TimeoutEntry {
/**
* Wall-clock due time, `Date.now() + ms` stamped at schedule time. A delay beyond 2^31-1 ms is
* Monotonic due time, `runtime.monotonicNow() + ms` stamped at schedule time. A delay beyond 2^31-1 ms is
* clamped to 1 ms up front, mirroring the runtime's own clamp (Bun/Node fire such timers almost
* immediately, with a `TimeoutOverflowWarning` the wheel does not reproduce): stamping the
* CLAMPED deadline is what keeps a covered re-arm from re-clamping into a 1 ms spin.
Expand All @@ -56,7 +57,7 @@ interface TimeoutEntry {
* timer, in place of a per-request `setTimeout`/`clearTimeout` pair on the HTTP transport's hot
* path.
*
* Per-request semantics are preserved exactly: the abort fires no earlier than `Date.now() + ms`
* Per-request semantics are preserved exactly: the abort fires no earlier than `runtime.monotonicNow() + ms`
* stamped at schedule time (the callback re-checks the deadline, so an early wake-up only
* re-arms), with the same lazily-created `TimeoutError` reason, and equal deadlines fire in
* insertion order, the way same-delay native timers would. What changes is the amortization: a
Expand All @@ -73,11 +74,14 @@ export class TimeoutWheel {
/** Live and spent entries, sorted by deadline ascending; equal deadlines keep insertion order. */
private readonly _entries: TimeoutEntry[];
/** The single armed native timer, or `undefined` when nothing is armed. */
private _timer: ReturnType<typeof setTimeout> | undefined;
private _timer: TimerHandle | undefined;
/** The deadline {@linkcode _timer} is armed for; never later than the head entry's deadline. */
private _timerDeadline: number;
/** Clock and scheduler, resolved once at construction; deadlines read its monotonic time. */
private readonly _runtime: Runtime;

constructor() {
constructor(runtime?: Partial<Runtime>) {
this._runtime = resolveRuntime(runtime);
this._entries = [];
this._timer = undefined;
this._timerDeadline = 0;
Expand All @@ -93,7 +97,7 @@ export class TimeoutWheel {
schedule(target: AbortController, ms: number | null): { reason: Error; cancel: () => void } {
if (ms === null || !Number.isFinite(ms)) return DISABLED_TIMEOUT;
const entry: TimeoutEntry = {
deadline: Date.now() + (ms > MAX_TIMEOUT_DELAY_MS ? 1 : ms),
deadline: this._runtime.monotonicNow() + (ms > MAX_TIMEOUT_DELAY_MS ? 1 : ms),
target,
reason: undefined,
cancelled: false,
Expand Down Expand Up @@ -135,7 +139,7 @@ export class TimeoutWheel {
/** Fires every overdue entry in deadline order and re-arms for what remains. */
private _onTimer(): void {
this._timer = undefined; // the timer that got us here has fired
const now = Date.now();
const now = this._runtime.monotonicNow();
const entries = this._entries;
for (;;) {
const head = entries[0];
Expand All @@ -156,13 +160,13 @@ export class TimeoutWheel {
/** (Re)arms the native timer for the current head's deadline; disarms when the queue is empty. */
private _rearm(): void {
if (this._timer !== undefined) {
clearTimeout(this._timer);
this._runtime.clearTimeout(this._timer);
this._timer = undefined;
}
const head = this._entries[0];
if (head === undefined) return;
this._timerDeadline = head.deadline;
this._timer = setTimeout(() => this._onTimer(), head.deadline - Date.now());
this._timer = this._runtime.setTimeout(() => this._onTimer(), head.deadline - this._runtime.monotonicNow());
// `unref` where the platform exposes it (Bun/Node timer objects); browser and fake timers
// return a plain number, hence the guarded call — typed through `unknown` because the DOM
// lib the TypeScript 7 gate compiles against types `setTimeout` as returning `number`.
Expand Down
11 changes: 10 additions & 1 deletion packages/hyperliquid/src/transport/_infoCache.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
*/

import type { IRequestTransport } from "./_base.ts";
import { resolveRuntime, type Runtime } from "./runtime.ts";

/**
* Info request types whose responses {@linkcode InfoCacheTransport} may cache.
Expand Down Expand Up @@ -59,6 +60,11 @@ export type InfoCacheableRequestType =

/** Configuration options for {@linkcode InfoCacheTransport}. */
export interface InfoCacheOptions {
/**
* Clock overrides for expiry, which reads {@linkcode Runtime.monotonicNow}. Missing members fall
* back to the platform; resolved once at construction.
*/
runtime?: Partial<Runtime>;
/**
* Time-to-live in ms applied to every allowlisted endpoint unless overridden in
* {@linkcode InfoCacheOptions.ttlByType}. Must be a non-negative number; `Infinity` caches
Expand Down Expand Up @@ -201,6 +207,8 @@ export class InfoCacheTransport<E extends "info" | "exchange" | "explorer" = "in
private readonly _coalesce: true | ReadonlySet<string> | undefined;
/** Coalesced requests still in flight, keyed like {@linkcode _entries}. */
private readonly _inFlight = new Map<string, InFlightEntry>();
/** Clock resolved once at construction; expiry reads its monotonic time. */
private readonly _runtime: Runtime;

/**
* Creates a caching wrapper around `inner`.
Expand All @@ -209,6 +217,7 @@ export class InfoCacheTransport<E extends "info" | "exchange" | "explorer" = "in
* @param options Cache configuration. See {@link InfoCacheOptions}.
*/
constructor(inner: IRequestTransport<E>, options?: InfoCacheOptions) {
this._runtime = resolveRuntime(options?.runtime);
const { ttl = 60_000, ttlByType = {}, maxSize = 1000, coalesce = false } = options ?? {};
if (
typeof ttl !== "number" ||
Expand Down Expand Up @@ -255,7 +264,7 @@ export class InfoCacheTransport<E extends "info" | "exchange" | "explorer" = "in
}

const key = stableKey(payload as Record<string, unknown>);
const now = Date.now();
const now = this._runtime.monotonicNow();
const hit = this._entries.get(key);
if (hit !== undefined && hit.expiresAt > now) return hit.promise as Promise<T>;

Expand Down
17 changes: 11 additions & 6 deletions packages/hyperliquid/src/transport/_rateLimiter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
*/

import { Promise_ } from "./_polyfills.ts";
import { resolveRuntime, type Runtime, type TimerHandle } from "./runtime.ts";

/**
* A queued acquisition waiting for enough tokens to accumulate.
Expand Down Expand Up @@ -68,15 +69,19 @@ export class TokenBucketRateLimiter {
/** Tail of the linked FIFO: the newest pending acquisition. */
private _tail: Waiter | undefined;
/** Wake-up timer for the head waiter; at most one is armed at a time. */
private _timer: ReturnType<typeof setTimeout> | undefined;
private _timer: TimerHandle | undefined;
/** Clock and scheduler, resolved once at construction; refills read its monotonic time. */
private readonly _runtime: Runtime;

/**
* Creates a token bucket.
*
* @param capacity Maximum burst size, in weight units. The bucket starts full.
* @param refillPerMinute Steady-state refill rate, in weight units per minute.
* @param runtime Clock and timer overrides; missing members fall back to the platform.
*/
constructor(capacity: number, refillPerMinute: number) {
constructor(capacity: number, refillPerMinute: number, runtime?: Partial<Runtime>) {
this._runtime = resolveRuntime(runtime);
// Infinity and NaN bypass or break the token arithmetic, non-positives stall the queue
// forever, and a refillPerMinute so small that the derived per-ms rate underflows to zero
// would leave every queued acquire with an infinite wait — all rejected up front.
Expand All @@ -89,7 +94,7 @@ export class TokenBucketRateLimiter {
this._capacity = capacity;
this._refillPerMs = refillPerMs;
this._tokens = capacity;
this._lastRefill = Date.now();
this._lastRefill = this._runtime.monotonicNow();
}

/**
Expand Down Expand Up @@ -126,7 +131,7 @@ export class TokenBucketRateLimiter {
// The wake-up was timed for this waiter as head; re-time it for the next in line,
// whose weight may accumulate sooner.
if (wasHead && this._timer !== undefined) {
clearTimeout(this._timer);
this._runtime.clearTimeout(this._timer);
this._timer = undefined;
if (this._head !== undefined) this._schedule();
}
Expand Down Expand Up @@ -173,7 +178,7 @@ export class TokenBucketRateLimiter {

/** Accrues the tokens earned since the last accounting, capped at the capacity. */
private _refill(): void {
const now = Date.now();
const now = this._runtime.monotonicNow();
const elapsed = now - this._lastRefill;
if (elapsed <= 0) return;
this._tokens = Math.min(this._capacity, this._tokens + elapsed * this._refillPerMs);
Expand Down Expand Up @@ -215,7 +220,7 @@ export class TokenBucketRateLimiter {
this._refill();
const deficit = this._needed(this._head!) - this._tokens;
const waitMs = Math.max(0, Math.ceil(deficit / this._refillPerMs));
this._timer = setTimeout(
this._timer = this._runtime.setTimeout(
() => {
this._timer = undefined;
this._drain();
Expand Down
Loading
Loading