diff --git a/apps/docs/content/docs/transports.md b/apps/docs/content/docs/transports.md index 923ff473..de813513 100644 --- a/apps/docs/content/docs/transports.md +++ b/apps/docs/content/docs/transports.md @@ -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 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`. diff --git a/packages/hyperliquid/.dev/perf/info_cache_capacity.ts b/packages/hyperliquid/.dev/perf/info_cache_capacity.ts index a6203291..a9b5380f 100644 --- a/packages/hyperliquid/.dev/perf/info_cache_capacity.ts +++ b/packages/hyperliquid/.dev/perf/info_cache_capacity.ts @@ -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 } : {}, diff --git a/packages/hyperliquid/package.json b/packages/hyperliquid/package.json index 1a5aecd1..c8535b6f 100644 --- a/packages/hyperliquid/package.json +++ b/packages/hyperliquid/package.json @@ -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", diff --git a/packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts b/packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts index ca2f90b3..5d57976e 100644 --- a/packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts +++ b/packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts @@ -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; @@ -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): 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(); 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 diff --git a/packages/hyperliquid/src/transport/_abort.ts b/packages/hyperliquid/src/transport/_abort.ts index cb797709..139667ee 100644 --- a/packages/hyperliquid/src/transport/_abort.ts +++ b/packages/hyperliquid/src/transport/_abort.ts @@ -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 { @@ -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. @@ -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 @@ -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 | 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) { + this._runtime = resolveRuntime(runtime); this._entries = []; this._timer = undefined; this._timerDeadline = 0; @@ -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, @@ -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]; @@ -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`. diff --git a/packages/hyperliquid/src/transport/_infoCache.ts b/packages/hyperliquid/src/transport/_infoCache.ts index 3f83798e..4fe30a0a 100644 --- a/packages/hyperliquid/src/transport/_infoCache.ts +++ b/packages/hyperliquid/src/transport/_infoCache.ts @@ -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. @@ -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; /** * Time-to-live in ms applied to every allowlisted endpoint unless overridden in * {@linkcode InfoCacheOptions.ttlByType}. Must be a non-negative number; `Infinity` caches @@ -201,6 +207,8 @@ export class InfoCacheTransport | 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) { + 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. @@ -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(); } /** @@ -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(); } @@ -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); @@ -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(); diff --git a/packages/hyperliquid/src/transport/http/mod.ts b/packages/hyperliquid/src/transport/http/mod.ts index f2961cd5..7f289db2 100644 --- a/packages/hyperliquid/src/transport/http/mod.ts +++ b/packages/hyperliquid/src/transport/http/mod.ts @@ -34,9 +34,22 @@ import * as abort from "../_abort.ts"; import { redactSignature, UNSERIALIZABLE_REQUEST } from "../_redact.ts"; import { TokenBucketRateLimiter } from "../_rateLimiter.ts"; import { exchangeWireJSON } from "../_wire.ts"; +import { delay, resolveRuntime, type Runtime } from "../runtime.ts"; /** Configuration options for the HTTP transport layer. */ export interface HttpTransportOptions { + /** + * Sends this transport's requests in place of the global `fetch`, which is left untouched. + * + * Default: the global `fetch`, read at request time (so a polyfill installed later applies). + */ + fetch?: (input: RequestInfo | URL, init?: RequestInit) => Promise; + /** + * Clock, timer, and jitter overrides for request timeouts, the {@linkcode rateLimit} bucket, + * and {@linkcode retryOnRateLimit} waits. Missing members fall back to the platform; resolved + * once at construction. + */ + runtime?: Partial; /** * Indicates this transport uses testnet endpoint. * @@ -331,8 +344,14 @@ export class HttpTransport implements IRequestTransport<"info" | "exchange" | "e private readonly _timeouts: abort.TimeoutWheel; /** Memoized endpoint URLs, keyed by base and endpoint; mutating `apiUrl`/`rpcUrl` simply misses the cache. */ private readonly _urlCache: Map; + /** Clock, scheduler, and jitter source, resolved once at construction. */ + private readonly _runtime: Runtime; + /** The injected fetch; `undefined` calls the global `fetch` at request time (so late polyfills apply). */ + private readonly _fetch: ((input: RequestInfo | URL, init?: RequestInit) => Promise) | undefined; constructor(options?: HttpTransportOptions) { + this._runtime = resolveRuntime(options?.runtime); + this._fetch = options?.fetch; this.isTestnet = options?.isTestnet ?? false; this.timeout = options?.timeout === undefined ? 10_000 : options.timeout; this.exchangeTimeout = options?.exchangeTimeout; @@ -342,9 +361,13 @@ export class HttpTransport implements IRequestTransport<"info" | "exchange" | "e this._rateLimit = options?.rateLimit === undefined ? null - : new TokenBucketRateLimiter(options.rateLimit.capacity ?? 1200, options.rateLimit.refillPerMinute ?? 1200); + : new TokenBucketRateLimiter( + options.rateLimit.capacity ?? 1200, + options.rateLimit.refillPerMinute ?? 1200, + this._runtime, + ); this._retryOnRateLimit = normalizeRetryOnRateLimit(options?.retryOnRateLimit); - this._timeouts = new abort.TimeoutWheel(); + this._timeouts = new abort.TimeoutWheel(this._runtime); this._urlCache = new Map(); } @@ -447,9 +470,12 @@ export class HttpTransport implements IRequestTransport<"info" | "exchange" | "e // opted-in retry can never outlive the caller's timeout; the wait itself races the // request's signal, so caller aborts interrupt it too. const retry = this._retryOnRateLimit; + // Called through a local, never as `this._fetch(...)`: platform fetches (browsers' + // `window.fetch`) throw "Illegal invocation" when their receiver is not the global. + const fetchImpl = this._fetch; for (let attempt = 0; ; attempt++) { try { - const response = await fetch(url, init); + const response = await (fetchImpl === undefined ? fetch(url, init) : fetchImpl(url, init)); if (!response.ok || !response.headers.get("Content-Type")?.includes("application/json")) { const clone = response.clone(); const text = await response.text().catch(() => undefined); // releases connection, clone stays readable @@ -505,13 +531,19 @@ export class HttpTransport implements IRequestTransport<"info" | "exchange" | "e // Only a 429 is retried, and only while attempts remain; every other failure — // a non-429 HttpRequestError included — propagates to the outer classifier as-is. if (!(error instanceof HttpRateLimitError) || retry === null || attempt >= retry.maxRetries) throw error; - const delayMs = retryDelayMs(error.retryAfter, attempt, retry.maxDelayMs); + // Re-parsed against the runtime's wall clock: an HTTP-date header is relative to `now`. + const delayMs = retryDelayMs( + parseRetryAfter(error.response?.headers.get("Retry-After") ?? null, this._runtime.now()), + attempt, + retry.maxDelayMs, + this._runtime.random, + ); if (delayMs === undefined) throw error; // the asked wait exceeds the configured bound // A retry is another billed attempt (the server bills attempts, not logical // requests): debit the bucket WITHOUT waiting, so later requests pace off the real // cost while this retry waits out the server's own delay rather than the refill. if (rateLimit !== null) rateLimit.charge(weight); - await abort.race(sleep(delayMs), controller?.signal); + await delay(delayMs, this._runtime, controller?.signal); } } } catch (error) { @@ -859,7 +891,7 @@ function buildDate( * Callers scheduling from this value must apply their own bounds — the SDK's only internal * scheduler (the rate-limit bucket) slices timer waits at 2^31-1 ms for exactly that reason. */ -function parseRetryAfter(value: string | null): number | undefined { +function parseRetryAfter(value: string | null, now: number = Date.now()): number | undefined { if (value === null) return undefined; const trimmed = value.trim(); if (/^\d+$/.test(trimmed)) { @@ -869,7 +901,7 @@ function parseRetryAfter(value: string | null): number | undefined { } const date = parseHttpDate(trimmed); if (date === undefined) return undefined; - return Math.max(0, (date - Date.now()) / 1000); + return Math.max(0, (date - now) / 1000); } /** @@ -902,22 +934,18 @@ function normalizeRetryOnRateLimit( * server allowed would just burn attempts on another 429, so the error surfaces instead. A * near-`MAX_SAFE_INTEGER` header lands here too, its millisecond conversion dwarfing any bound. */ -function retryDelayMs(retryAfter: number | undefined, attempt: number, maxDelayMs: number): number | undefined { +function retryDelayMs( + retryAfter: number | undefined, + attempt: number, + maxDelayMs: number, + random: () => number = Math.random, +): number | undefined { if (retryAfter !== undefined) { const askedMs = retryAfter * 1000; if (askedMs > maxDelayMs) return undefined; - return askedMs + Math.random() * 1000; + return askedMs + random() * 1000; } - return Math.random() * Math.min(1000 * 2 ** attempt, maxDelayMs); -} - -/** Resolves after `ms`; the timer is `unref`'d where supported, so an abandoned wait never holds the process open. */ -function sleep(ms: number): Promise { - return new Promise((resolve) => { - const timer = setTimeout(resolve, ms); - // Same guarded call as the TimeoutWheel: browser and fake timers return a plain number. - (timer as unknown as { unref?: () => void }).unref?.(); - }); + return random() * Math.min(1000 * 2 ** attempt, maxDelayMs); } /** Resolves an endpoint against a base URL without dropping the base path or query. */ diff --git a/packages/hyperliquid/src/transport/mod.ts b/packages/hyperliquid/src/transport/mod.ts index 5f1b10f8..e73533ae 100644 --- a/packages/hyperliquid/src/transport/mod.ts +++ b/packages/hyperliquid/src/transport/mod.ts @@ -17,4 +17,5 @@ export * from "./_base.ts"; export * from "./_infoCache.ts"; export * from "./http/mod.ts"; +export * from "./runtime.ts"; export * from "./websocket/mod.ts"; diff --git a/packages/hyperliquid/src/transport/runtime.ts b/packages/hyperliquid/src/transport/runtime.ts new file mode 100644 index 00000000..35cf3d63 --- /dev/null +++ b/packages/hyperliquid/src/transport/runtime.ts @@ -0,0 +1,105 @@ +/** + * Injectable platform dependencies (clocks, timers, and the random source) used by the transports + * and their schedulers, so tests can run them on a virtual clock without patching globals. + * + * @example + * ```ts + * import { HttpTransport } from "@bloxwap/hyperliquid"; + * + * // Retry jitter pinned to zero; every other member falls back to the platform. + * const transport = new HttpTransport({ retryOnRateLimit: true, runtime: { random: () => 0 } }); + * ``` + * + * @module + */ + +/** Timer handles may be numeric (browsers/fakes) or platform timer objects. */ +export type TimerHandle = ReturnType | number; + +/** + * Clock, scheduler, and random source. + * + * Wall time ({@linkcode Runtime.now}) is used only where the protocol needs a timestamp: nonces and + * HTTP-date `Retry-After` headers. Every elapsed-time decision (request deadlines, rate-limit + * refills, cache expiry, connection-attempt windows) reads {@linkcode Runtime.monotonicNow}, so a + * wall-clock correction neither fires nor stalls them. + */ +export interface Runtime { + /** Unix milliseconds. */ + now(): number; + /** Elapsed milliseconds, independent of wall-clock corrections. */ + monotonicNow(): number; + /** Schedule a callback. */ + setTimeout(callback: () => void, ms: number): TimerHandle; + /** Cancel a callback. */ + clearTimeout(handle: TimerHandle | undefined): void; + /** Schedule a repeating callback. */ + setInterval(callback: () => void, ms: number): TimerHandle; + /** Cancel a repeating callback. */ + clearInterval(handle: TimerHandle | undefined): void; + /** Uniform random value in [0, 1). */ + random(): number; +} + +/** + * The platform runtime. Each member reads its global at call time, so polyfills installed after + * import still apply. + */ +export const systemRuntime: Runtime = { + now: (): number => Date.now(), + monotonicNow: (): number => performance.now(), + setTimeout: (callback: () => void, ms: number): TimerHandle => setTimeout(callback, ms), + clearTimeout: (handle: TimerHandle | undefined): void => clearTimeout(handle as ReturnType), + setInterval: (callback: () => void, ms: number): TimerHandle => setInterval(callback, ms), + clearInterval: (handle: TimerHandle | undefined): void => clearInterval(handle as ReturnType), + random: (): number => Math.random(), +}; + +/** Runtimes produced by {@linkcode resolveRuntime}, returned as-is when resolved again. */ +const resolved = new WeakSet(); +resolved.add(systemRuntime); + +/** + * Fills the members `overrides` omits from {@linkcode systemRuntime}. Called once per component at + * construction, never per request. Returns `systemRuntime` itself when nothing is overridden, and + * an already-resolved runtime unchanged, so components sharing one runtime share one object. + * Overriding methods keep `overrides` as their receiver, so a stateful fake clock can be passed + * directly. + */ +export function resolveRuntime(overrides?: Partial): Runtime { + if (overrides === undefined) return systemRuntime; + if (resolved.has(overrides as Runtime)) return overrides as Runtime; + const result: Runtime = { ...systemRuntime }; + for (const key of Object.keys(systemRuntime) as (keyof Runtime)[]) { + const override = overrides[key]; + if (override !== undefined) Object.assign(result, { [key]: override.bind(overrides) }); + } + resolved.add(result); + return result; +} + +/** + * Resolves after `ms` on `runtime`'s scheduler, or rejects with `signal`'s reason once it aborts. + * The timer and the abort listener are released on every settlement path, and the timer is + * `unref`'d where the platform supports it, so an abandoned wait never holds the process open. + */ +export function delay(ms: number, runtime: Runtime = systemRuntime, signal?: AbortSignal): Promise { + if (signal?.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + const cleanup = (): void => { + runtime.clearTimeout(timer); + signal?.removeEventListener("abort", onAbort); + }; + const onAbort = (): void => { + cleanup(); + reject(signal?.reason); + }; + const timer = runtime.setTimeout(() => { + cleanup(); + resolve(); + }, ms); + // Same guarded call as the TimeoutWheel: browser and fake timers return a plain number. + (timer as unknown as { unref?: () => void }).unref?.(); + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} diff --git a/packages/hyperliquid/src/transport/websocket/_dispatcher.ts b/packages/hyperliquid/src/transport/websocket/_dispatcher.ts index ea7c03f7..83d6c868 100644 --- a/packages/hyperliquid/src/transport/websocket/_dispatcher.ts +++ b/packages/hyperliquid/src/transport/websocket/_dispatcher.ts @@ -14,6 +14,7 @@ import type { HyperliquidEventTarget, PostResponse, SubscribeUnsubscribeResponse import { isSubset, normalize, requestToId, specificity } from "./_id.ts"; import type { WebSocketQuota } from "./_quota.ts"; import { exchangeWireJSON } from "../_wire.ts"; +import type { Runtime } from "../runtime.ts"; // ============================================================================= // Errors @@ -202,10 +203,11 @@ export class WebSocketDispatcher { hlEvents: HyperliquidEventTarget, timeout: number | null, quota?: WebSocketQuota, + runtime?: Partial, ) { this.timeout = timeout; this._socket = socket; - this._timeouts = new abort.TimeoutWheel(); + this._timeouts = new abort.TimeoutWheel(runtime); this._quota = quota; // --- Hyperliquid event handlers ------------------------------------------ diff --git a/packages/hyperliquid/src/transport/websocket/_keepAlive.ts b/packages/hyperliquid/src/transport/websocket/_keepAlive.ts index b056fa4d..d4399f43 100644 --- a/packages/hyperliquid/src/transport/websocket/_keepAlive.ts +++ b/packages/hyperliquid/src/transport/websocket/_keepAlive.ts @@ -8,6 +8,7 @@ import type { ReconnectingWebSocket } from "./_reconnectingSocket.ts"; import type { HyperliquidEventTarget } from "./_events.ts"; import type { WebSocketQuota } from "./_quota.ts"; +import { resolveRuntime, type Runtime, type TimerHandle } from "../runtime.ts"; /** Configuration options for the keep-alive watchdog. */ export interface WebSocketKeepAliveOptions { @@ -41,15 +42,19 @@ export class WebSocketKeepAlive { * unnoticed — so heavy ping traffic slows subscribes instead of itself. */ private readonly _quota: WebSocketQuota | undefined; - private _pingInterval: ReturnType | undefined; - private _pongTimeout: ReturnType | undefined; + private _pingInterval: TimerHandle | undefined; + private _pongTimeout: TimerHandle | undefined; + /** Scheduler for the ping interval and pong deadline, resolved once at construction. */ + private readonly _runtime: Runtime; constructor( socket: ReconnectingWebSocket, hlEvents: HyperliquidEventTarget, options?: WebSocketKeepAliveOptions, quota?: WebSocketQuota, + runtime?: Partial, ) { + this._runtime = resolveRuntime(runtime); this._socket = socket; this._interval = options?.interval ?? 5_000; this._timeout = options?.timeout ?? 3_000; @@ -63,22 +68,22 @@ export class WebSocketKeepAlive { private _start(): void { if (this._pingInterval) return; - this._pingInterval = setInterval(() => { + this._pingInterval = this._runtime.setInterval(() => { this._socket.send('{"method":"ping"}'); this._quota?.chargeSend(); // A half-open connection never answers: reconnect once a ping stays unanswered. - this._pongTimeout ??= setTimeout(() => this._socket.reconnect(), this._timeout); + this._pongTimeout ??= this._runtime.setTimeout(() => this._socket.reconnect(), this._timeout); }, this._interval); } private _stop(): void { - clearInterval(this._pingInterval); + this._runtime.clearInterval(this._pingInterval); this._pingInterval = undefined; this._disarm(); } private _disarm(): void { - clearTimeout(this._pongTimeout); + this._runtime.clearTimeout(this._pongTimeout); this._pongTimeout = undefined; } } diff --git a/packages/hyperliquid/src/transport/websocket/_quota.ts b/packages/hyperliquid/src/transport/websocket/_quota.ts index 19e7a050..06010d5e 100644 --- a/packages/hyperliquid/src/transport/websocket/_quota.ts +++ b/packages/hyperliquid/src/transport/websocket/_quota.ts @@ -31,6 +31,7 @@ */ import { TokenBucketRateLimiter } from "../_rateLimiter.ts"; +import { resolveRuntime, type Runtime, type TimerHandle } from "../runtime.ts"; // ============================================================================= // Documented limits @@ -97,6 +98,11 @@ export interface WebSocketRateLimitOptions { /** Configuration options for a {@linkcode WebSocketQuota}. */ export interface WebSocketQuotaOptions { + /** + * Clock and timer overrides for the rolling connection-attempt window and the message pacer. + * Missing members fall back to the platform; resolved once at construction. + */ + runtime?: Partial; /** Maximum connecting/open/closing sockets sharing this quota; null disables. Default: 10. */ maxConnections?: number | null; /** Maximum new socket attempts in any rolling minute; null disables. Default: 30. */ @@ -158,7 +164,7 @@ export class WebSocketQuota { signal?: AbortSignal; abort: () => void; }[] = []; - private _connectionTimer: ReturnType | undefined; + private _connectionTimer: TimerHandle | undefined; /** Maximum concurrent subscriptions, or `null` when the guard is disabled. */ readonly maxSubscriptions: number | null; /** Maximum unique users across user-specific subscriptions, or `null` when the guard is disabled. */ @@ -176,8 +182,11 @@ export class WebSocketQuota { private readonly _users: Map = new Map(); /** Outbound message pacer, or `null` when pacing is off and only accounting runs. */ private readonly _limiter: TokenBucketRateLimiter | null; + /** Clock and scheduler, resolved once at construction; attempt windows read its monotonic time. */ + private readonly _runtime: Runtime; constructor(options?: WebSocketQuotaOptions) { + this._runtime = resolveRuntime(options?.runtime); this.maxConnections = options?.maxConnections === undefined ? 10 : options.maxConnections; this.maxConnectionAttemptsPerMinute = options?.maxConnectionAttemptsPerMinute === undefined ? 30 : options.maxConnectionAttemptsPerMinute; @@ -193,6 +202,7 @@ export class WebSocketQuota { : new TokenBucketRateLimiter( options.rateLimit.capacity ?? MAX_MESSAGES_PER_MINUTE, options.rateLimit.refillPerMinute ?? MAX_MESSAGES_PER_MINUTE, + this._runtime, ); } @@ -232,7 +242,7 @@ export class WebSocketQuota { } private _pruneAttempts(): void { - const cutoff = Date.now() - 60_000; + const cutoff = this._runtime.monotonicNow() - 60_000; while (this._attempts.length && this._attempts[0] <= cutoff) this._attempts.shift(); } @@ -245,7 +255,7 @@ export class WebSocketQuota { private _takeConnection(): () => void { this._connections++; - if (this.maxConnectionAttemptsPerMinute !== null) this._attempts.push(Date.now()); + if (this.maxConnectionAttemptsPerMinute !== null) this._attempts.push(this._runtime.monotonicNow()); let released = false; return (): void => { if (released) return; @@ -256,7 +266,7 @@ export class WebSocketQuota { } private _drainConnections(): void { - clearTimeout(this._connectionTimer); + this._runtime.clearTimeout(this._connectionTimer); this._connectionTimer = undefined; this._pruneAttempts(); while (this._connectionWaiters.length && this._canConnect()) { @@ -269,9 +279,9 @@ export class WebSocketQuota { (this.maxConnections === null || this._connections < this.maxConnections) && this._attempts.length ) { - this._connectionTimer = setTimeout( + this._connectionTimer = this._runtime.setTimeout( () => this._drainConnections(), - Math.max(1, this._attempts[0] + 60_000 - Date.now()), + Math.max(1, this._attempts[0] + 60_000 - this._runtime.monotonicNow()), ); } } diff --git a/packages/hyperliquid/src/transport/websocket/_reconnectingSocket.ts b/packages/hyperliquid/src/transport/websocket/_reconnectingSocket.ts index d48cdcf0..1bdb8067 100644 --- a/packages/hyperliquid/src/transport/websocket/_reconnectingSocket.ts +++ b/packages/hyperliquid/src/transport/websocket/_reconnectingSocket.ts @@ -10,6 +10,7 @@ */ import { DOMException_ } from "../_polyfills.ts"; +import { resolveRuntime, type Runtime, type TimerHandle } from "../runtime.ts"; /** Specifies the type of binary data being received over a `WebSocket` connection. */ export type BinaryType = "blob" | "arraybuffer"; @@ -45,6 +46,18 @@ export type ReconnectingWebSocketErrorCode = /** Configuration options for the {@linkcode ReconnectingWebSocket}. */ export interface ReconnectingWebSocketOptions { + /** + * Creates each underlying socket in place of the global `WebSocket`, which is left untouched. + * Fixed at construction, unlike the other options. + * + * Default: `new WebSocket(url, protocols)`, read from the global at connection time. + */ + webSocketFactory?: (url: string, protocols?: string | string[]) => WebSocket; + /** + * Timer and jitter overrides for retry delays, connection timeouts, and the stable-connection + * reset. Missing members fall back to the platform. Fixed at construction. + */ + runtime?: Partial; /** Admission hook for shared connection budgets. Its release runs on socket close or construction failure. */ acquireConnection?: (signal: AbortSignal) => (() => void) | Promise<() => void>; /** @@ -171,9 +184,9 @@ const MAX_RECONNECTION_DELAY = 10_000; * at {@linkcode MAX_RECONNECTION_DELAY}, with equal jitter — half the capped * delay is fixed, the other half is uniform random. */ -function defaultReconnectionDelay(attempt: number): number { +function defaultReconnectionDelay(attempt: number, random: () => number = Math.random): number { const capped = Math.min(2 ** attempt * 150, MAX_RECONNECTION_DELAY); - return capped / 2 + Math.random() * (capped / 2); + return capped / 2 + random() * (capped / 2); } /** Validates `close()` / `reconnect()` arguments the way the `WebSocket` spec requires. */ @@ -239,11 +252,11 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { /** Current reconnection attempt number. */ private _retryCount = 0; /** Pending retry delay; `undefined` when not sleeping between attempts. */ - private _retryTimer: ReturnType | undefined; + private _retryTimer: TimerHandle | undefined; /** Pending retry-counter reset, armed on every `open`. */ - private _stableTimer: ReturnType | undefined; + private _stableTimer: TimerHandle | undefined; /** Pending connection timeout for the in-flight attempt. */ - private _connectionTimer: ReturnType | undefined; + private _connectionTimer: TimerHandle | undefined; /** Bumped per connection attempt, so a superseded in-flight factory resolution is discarded. */ private _connectGeneration = 0; /** Set while a `close` event is being dispatched, so `close()` treats it as the final one. */ @@ -256,10 +269,14 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { private readonly _abortController = new AbortController(); /** Reconnection configuration options. Read on every attempt, so changes apply live. */ - reconnectOptions: Required> & + reconnectOptions: Required> & Pick; private _attemptController: AbortController | undefined; private _releaseConnection: (() => void) | undefined; + /** Scheduler and jitter source, resolved once at construction. */ + private readonly _runtime: Runtime; + /** Creates each underlying socket; the global `WebSocket` unless one was injected. */ + private readonly _webSocketFactory: (url: string, protocols?: string | string[]) => WebSocket; /** * AbortSignal that is aborted when the instance is permanently terminated. @@ -275,7 +292,7 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { * @param url URL or factory function for the WebSocket connection. * @param options Configuration options. * - * @throws {TypeError} If no WebSocket implementation is available. + * @throws {TypeError} If no `webSocketFactory` is given and no global `WebSocket` is available. */ constructor(url: UrlProvider, options?: ReconnectingWebSocketOptions); /** @@ -285,7 +302,7 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { * @param protocols Subprotocol(s) or factory function. * @param options Configuration options. * - * @throws {TypeError} If no WebSocket implementation is available. + * @throws {TypeError} If no `webSocketFactory` is given and no global `WebSocket` is available. */ constructor(url: UrlProvider, protocols?: ProtocolsProvider, options?: ReconnectingWebSocketOptions); constructor( @@ -294,24 +311,32 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { options?: ReconnectingWebSocketOptions, ) { super(); - if (typeof globalThis.WebSocket !== "function") { - throw new TypeError("No WebSocket implementation is available in this environment"); - } this._urlProvider = url; // Distinguish the (url, options) call from (url, protocols, options): protocols is a string, // a string array, a function, or undefined — an options bag is the only plain object. const isOptions = typeof protocolsOrOptions === "object" && protocolsOrOptions !== null && !Array.isArray(protocolsOrOptions); + // Injected dependencies are fixed at construction; everything else stays a live option. + const { runtime, webSocketFactory, ...rest } = (isOptions ? protocolsOrOptions : options) ?? {}; + if (webSocketFactory === undefined && typeof globalThis.WebSocket !== "function") { + throw new TypeError("No WebSocket implementation is available in this environment"); + } + this._runtime = resolveRuntime(runtime); + this._webSocketFactory = + webSocketFactory ?? + ((url: string, protocols?: string | string[]): WebSocket => + protocols === undefined ? new WebSocket(url) : new WebSocket(url, protocols)); this._protocolsProvider = isOptions ? undefined : protocolsOrOptions; + const random = this._runtime.random; this.reconnectOptions = { maxRetries: Infinity, maxEnqueuedMessages: Infinity, connectionTimeout: 10_000, stableTimeout: 3_000, - reconnectionDelay: defaultReconnectionDelay, + reconnectionDelay: (attempt: number): number => defaultReconnectionDelay(attempt, random), shouldReconnect: (): boolean => true, - ...(isOptions ? protocolsOrOptions : options), + ...rest, }; void this._connect(); @@ -364,7 +389,7 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { let socket: WebSocket; try { this._url = String(url); - socket = protocols === undefined ? new WebSocket(this._url) : new WebSocket(this._url, protocols); + socket = this._webSocketFactory(this._url, protocols); socket.binaryType = this._binaryType; } catch (error) { release?.(); @@ -377,11 +402,11 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { socket.addEventListener("open", () => { if (this._socket !== socket) return; - clearTimeout(this._connectionTimer); + this._runtime.clearTimeout(this._connectionTimer); this._connectionTimer = undefined; // The retry counter only resets once the connection has proven stable. - clearTimeout(this._stableTimer); - this._stableTimer = setTimeout(() => { + this._runtime.clearTimeout(this._stableTimer); + this._stableTimer = this._runtime.setTimeout(() => { this._retryCount = 0; }, this.reconnectOptions.stableTimeout); this._flushBuffer(); @@ -405,9 +430,9 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { if (this._socket !== socket) return; this._releaseConnection = undefined; this._socket = undefined; - clearTimeout(this._connectionTimer); + this._runtime.clearTimeout(this._connectionTimer); this._connectionTimer = undefined; - clearTimeout(this._stableTimer); + this._runtime.clearTimeout(this._stableTimer); this._stableTimer = undefined; if (this._abortController.signal.aborted) return; // A fresh event: the incoming one is mid-dispatch on the underlying socket @@ -420,7 +445,7 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { // The clock starts once the socket exists; url/protocols factories are not bounded. const connectionTimeout = this.reconnectOptions.connectionTimeout; if (connectionTimeout !== null) { - this._connectionTimer = setTimeout(() => { + this._connectionTimer = this._runtime.setTimeout(() => { this._connectionTimer = undefined; if (this._socket !== socket) return; this._socket = undefined; @@ -515,7 +540,7 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { // reconnected it: the pending retry belongs to a superseded connection plan. if (this._abortController.signal.aborted) return; if (this._connectGeneration !== generation) return; - this._retryTimer = setTimeout(() => { + this._retryTimer = this._runtime.setTimeout(() => { this._retryTimer = undefined; void this._connect(); }, delay); @@ -547,11 +572,11 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { this._abortController.abort(error); this._attemptController?.abort(error); - clearTimeout(this._retryTimer); + this._runtime.clearTimeout(this._retryTimer); this._retryTimer = undefined; - clearTimeout(this._stableTimer); + this._runtime.clearTimeout(this._stableTimer); this._stableTimer = undefined; - clearTimeout(this._connectionTimer); + this._runtime.clearTimeout(this._connectionTimer); this._connectionTimer = undefined; this._connectGeneration++; // discard any in-flight factory resolution this._socket = undefined; // a still-open socket is closed by the caller @@ -799,7 +824,7 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { assertValidCloseParams(code, reason); if (this._abortController.signal.aborted) return; - clearTimeout(this._retryTimer); + this._runtime.clearTimeout(this._retryTimer); this._retryTimer = undefined; const socket = this._socket; @@ -807,9 +832,9 @@ export class ReconnectingWebSocket extends EventTarget implements WebSocket { const release = this._releaseConnection; this._releaseConnection = undefined; this._socket = undefined; // its events are ignored from here on - clearTimeout(this._connectionTimer); + this._runtime.clearTimeout(this._connectionTimer); this._connectionTimer = undefined; - clearTimeout(this._stableTimer); + this._runtime.clearTimeout(this._stableTimer); this._stableTimer = undefined; // Close before dispatching the synthetic close: a close listener that terminates the // instance (e.g. transport.close()) aborts the rest of this method, and the abandoned diff --git a/packages/hyperliquid/src/transport/websocket/mod.ts b/packages/hyperliquid/src/transport/websocket/mod.ts index 6ab82626..af2b183f 100644 --- a/packages/hyperliquid/src/transport/websocket/mod.ts +++ b/packages/hyperliquid/src/transport/websocket/mod.ts @@ -34,6 +34,7 @@ import { type WebSocketRateLimitOptions, } from "./_quota.ts"; import { WebSocketSubscriptionManager } from "./_subscriptionManager.ts"; +import { resolveRuntime, type Runtime } from "../runtime.ts"; export { WebSocketQuota, type WebSocketQuotaOptions, type WebSocketRateLimitOptions, WebSocketRequestError }; export { @@ -44,6 +45,20 @@ export { /** Configuration options for the WebSocket transport layer. */ export interface WebSocketTransportOptions { + /** + * Clock, timer, and jitter overrides for this transport's reconnects, request timeouts, and + * keep-alive. Missing members fall back to the platform; resolved once at construction. + * + * The default shared {@linkcode WebSocketQuota} keeps the platform clock: pass a + * {@linkcode quota} built with the same runtime to drive connection and message budgets too. + */ + runtime?: Partial; + /** + * Creates the underlying sockets in place of the global `WebSocket`, which is left untouched. + * + * Default: `new WebSocket(url, protocols)`, read from the global at connection time. + */ + webSocketFactory?: (url: string, protocols?: string | string[]) => WebSocket; /** * Indicates this transport uses testnet endpoint. * @@ -184,11 +199,15 @@ export class WebSocketTransport implements IRequestTransport<"info" | "exchange" constructor(options?: WebSocketTransportOptions) { this.isTestnet = options?.isTestnet ?? false; this.quota = options?.quota ?? sharedWebSocketQuota(this.isTestnet); + // Resolved once and shared by the socket, dispatcher, and keep-alive. + const runtime = resolveRuntime(options?.runtime ?? options?.reconnect?.runtime); this.socket = new ReconnectingWebSocket( options?.url ?? (this.isTestnet ? TESTNET_API_WS_URL : MAINNET_API_WS_URL), { ...options?.reconnect, + runtime, + webSocketFactory: options?.webSocketFactory ?? options?.reconnect?.webSocketFactory, acquireConnection: (signal: AbortSignal): (() => void) | Promise<() => void> => this.quota.acquireConnection(signal), }, @@ -201,10 +220,11 @@ export class WebSocketTransport implements IRequestTransport<"info" | "exchange" this._hlEvents, options?.timeout === undefined ? 10_000 : options.timeout, this.quota, + runtime, ); // The keep-alive watchdog is fully self-contained: it exposes no API and drives itself from the // socket's own "open"/"close"/"error" events, which keep it reachable. Nothing to hold on to. - new WebSocketKeepAlive(this.socket, this._hlEvents, options?.keepAlive, this.quota); + new WebSocketKeepAlive(this.socket, this._hlEvents, options?.keepAlive, this.quota, runtime); this._subscriptionManager = new WebSocketSubscriptionManager( this.socket, this._dispatcher, diff --git a/packages/hyperliquid/tests/_fakeRuntime.ts b/packages/hyperliquid/tests/_fakeRuntime.ts new file mode 100644 index 00000000..386b0f94 --- /dev/null +++ b/packages/hyperliquid/tests/_fakeRuntime.ts @@ -0,0 +1,87 @@ +/** + * Virtual clock and timers for transport tests, injected through the `runtime` options. + * @module + */ + +import type { Runtime, TimerHandle } from "../src/transport/runtime.ts"; + +/** + * Isolated virtual clock and timers; never patches platform globals. Wall and monotonic time + * advance together through {@linkcode FakeRuntime.advance}; {@linkcode FakeRuntime.setWallTime} + * moves only the wall clock, to simulate corrections. `random()` is 0 unless overridden. + */ +export class FakeRuntime implements Runtime { + private _wall = 1_700_000_000_000; + private _elapsed = 0; + private _id = 0; + private readonly _timers = new Map void; interval?: number }>(); + private readonly _random: () => number; + /** @param options.random Jitter source; defaults to always 0. */ + constructor(options?: { random?: () => number }) { + this._random = options?.random ?? ((): number => 0); + } + now(): number { + return this._wall; + } + monotonicNow(): number { + return this._elapsed; + } + random(): number { + return this._random(); + } + /** Timers and intervals scheduled and not yet fired or cancelled. */ + get pendingTimers(): number { + return this._timers.size; + } + /** Moves only the wall clock, as an NTP correction would; monotonic time is unaffected. */ + setWallTime(value: number): void { + this._wall = value; + } + setTimeout(callback: () => void, ms: number): TimerHandle { + const id = ++this._id; + this._timers.set(id, { due: this._elapsed + Math.max(0, ms), callback }); + return id; + } + clearTimeout(handle: TimerHandle | undefined): void { + this._timers.delete(handle as number); + } + setInterval(callback: () => void, ms: number): TimerHandle { + const id = this.setTimeout(callback, ms) as number; + this._timers.get(id)!.interval = Math.max(1, ms); + return id; + } + clearInterval(handle: TimerHandle | undefined): void { + this.clearTimeout(handle); + } + /** Advances to the earliest pending timer and fires it; `false` when none is pending. */ + nextTimer(): boolean { + const due = Math.min(...[...this._timers.values()].map((timer) => timer.due)); + if (!Number.isFinite(due)) return false; + this.advance(due - this._elapsed); + return true; + } + /** Advances both clocks by `ms`, firing every timer that falls due, in deadline order. */ + advance(ms: number): void { + const end = this._elapsed + ms; + for (let count = 0; ; count++) { + if (count > 100_000) throw new Error("Virtual timer did not make progress"); + const next = [...this._timers] + .filter(([, timer]) => timer.due <= end) + .sort((a, b) => a[1].due - b[1].due || a[0] - b[0])[0]; + if (!next) break; + const [id, timer] = next; + this._wall += timer.due - this._elapsed; + this._elapsed = timer.due; + if (timer.interval === undefined) this._timers.delete(id); + else timer.due += timer.interval; + timer.callback(); + } + this._wall += end - this._elapsed; + this._elapsed = end; + } +} + +/** Drain nested nonce-lock/signing reactions without sleeping. */ +export async function drain(): Promise { + for (let i = 0; i < 60; i++) await Promise.resolve(); +} diff --git a/packages/hyperliquid/tests/_fakeSocket.ts b/packages/hyperliquid/tests/_fakeSocket.ts new file mode 100644 index 00000000..8e01a69e --- /dev/null +++ b/packages/hyperliquid/tests/_fakeSocket.ts @@ -0,0 +1,34 @@ +/** + * In-memory socket for transport tests, injected through the `webSocketFactory` options. + * @module + */ + +import { RealCloseEvent } from "./_noCloseEvent.ts"; + +/** Injectable, in-memory platform socket; no global WebSocket replacement. */ +export class FakeSocket extends EventTarget { + readyState = 0; + binaryType = "blob"; + bufferedAmount = 0; + protocol = ""; + extensions = ""; + readonly sent: string[] = []; + constructor(readonly onSend?: (data: string) => void) { + super(); + } + open(): void { + this.readyState = 1; + this.dispatchEvent(new Event("open")); + } + send(data: string): void { + this.sent.push(data); + this.onSend?.(data); + } + receive(data: unknown): void { + this.dispatchEvent(new MessageEvent("message", { data: JSON.stringify(data) })); + } + close(): void { + this.readyState = 3; + this.dispatchEvent(new RealCloseEvent("close", { code: 1000 })); + } +} diff --git a/packages/hyperliquid/tests/api/exchange/_nonce.test.ts b/packages/hyperliquid/tests/api/exchange/_nonce.test.ts index bcffa275..a462b89d 100644 --- a/packages/hyperliquid/tests/api/exchange/_nonce.test.ts +++ b/packages/hyperliquid/tests/api/exchange/_nonce.test.ts @@ -9,6 +9,7 @@ import { describe, test } from "bun:test"; import { assert, assertEquals } from "@jsr/std__assert"; import { createNonceManager, type NonceManager } from "../../../src/api/exchange/_methods/_base/_nonce.ts"; +import { FakeRuntime } from "../../_fakeRuntime.ts"; // ============================================================ // Helpers @@ -20,24 +21,14 @@ function walletKey(i: number): string { } /** - * Runs `fn` with `Date.now` replaced by a controllable clock, restoring it afterwards. - * A deterministic clock is what lets these tests assert prune behaviour exactly, with no - * wall-clock timing assertions (which would be flaky in CI). + * Runs `fn` against a virtual wall clock injected into the managers it creates. A + * deterministic clock is what lets these tests assert prune behaviour exactly, with no + * wall-clock timing assertions (which would be flaky in CI) and without patching `Date.now`. */ -function withMockClock(start: number, fn: (clock: { now: () => number; advance: (ms: number) => void }) => void): void { - let now = start; - const realDateNow = Date.now; - Date.now = () => now; - try { - fn({ - now: () => now, - advance: (ms: number) => { - now += ms; - }, - }); - } finally { - Date.now = realDateNow; - } +function withMockClock(start: number, fn: (clock: FakeRuntime) => void): void { + const clock = new FakeRuntime(); + clock.setWallTime(start); + fn(clock); } // The prune throttle in `_nonce.ts` rescans at most once per this interval. @@ -59,7 +50,10 @@ interface PruneCounters { * no other map in the process is affected. Counting scans (rather than timing them) is what * keeps the bounded-work assertions deterministic instead of flaky in CI. */ -function managerWithCounters(maxEntries: number): { manager: NonceManager; counters: PruneCounters } { +function managerWithCounters( + maxEntries: number, + clock: FakeRuntime, +): { manager: NonceManager; counters: PruneCounters } { const counters: PruneCounters = { scans: 0, deletes: 0 }; class CountingMap extends Map { override [Symbol.iterator](): MapIterator<[K, V]> { @@ -75,7 +69,7 @@ function managerWithCounters(maxEntries: number): { manager: NonceManager; count const realMap = globalThis.Map; globalThis.Map = CountingMap as unknown as MapConstructor; try { - return { manager: createNonceManager(maxEntries), counters }; + return { manager: createNonceManager(maxEntries, clock), counters }; } finally { globalThis.Map = realMap; } @@ -88,7 +82,7 @@ function managerWithCounters(maxEntries: number): { manager: NonceManager; count describe("createNonceManager", () => { test("nonces stay strictly increasing per key beyond maxEntries, ahead of wall clock", () => { withMockClock(1_700_000_000_000, (clock) => { - const manager = createNonceManager(); // default maxEntries: 10_000 + const manager = createNonceManager(undefined, clock); // default maxEntries: 10_000 const keyCount = 12_000; // over capacity, so the prune path is active throughout const lastNonce = new Map(); @@ -123,7 +117,7 @@ describe("createNonceManager", () => { withMockClock(1_700_000_000_000, (clock) => { const maxEntries = 100; const keyCount = 500; // 5x over capacity - const { manager, counters } = managerWithCounters(maxEntries); + const { manager, counters } = managerWithCounters(maxEntries, clock); // Fill past capacity with a frozen clock. The first over-capacity call pays for one // scan (which frees nothing: every entry's nonce equals `now`, none are stale); @@ -164,7 +158,7 @@ describe("createNonceManager", () => { withMockClock(1_700_000_000_000, (clock) => { const maxEntries = 100; const keyCount = 500; - const { manager, counters } = managerWithCounters(maxEntries); + const { manager, counters } = managerWithCounters(maxEntries, clock); for (let i = 0; i < keyCount; i++) manager.getNonce(walletKey(i)); assertEquals(counters.scans, 1); @@ -172,7 +166,7 @@ describe("createNonceManager", () => { // An NTP correction steps wall time back an hour. A throttle comparing // `now - lastPruneAt >= interval` alone would now wait an hour before pruning again, // leaving the map to grow unchecked for the whole excursion. - clock.advance(-3_600_000); + clock.setWallTime(clock.now() - 3_600_000); const nonce = manager.getNonce(walletKey(0)); assertEquals(counters.scans, 2, "a backwards clock step must re-arm the prune scan immediately"); diff --git a/packages/hyperliquid/tests/transport/_abort.test.ts b/packages/hyperliquid/tests/transport/_abort.test.ts index 7c24b44d..5c3499c4 100644 --- a/packages/hyperliquid/tests/transport/_abort.test.ts +++ b/packages/hyperliquid/tests/transport/_abort.test.ts @@ -6,9 +6,9 @@ * @module */ -import { afterEach, beforeEach, describe, test } from "bun:test"; +import { beforeEach, describe, test } from "bun:test"; import { assert, assertEquals, assertIsError, assertStrictEquals } from "@jsr/std__assert"; -import { FakeTime } from "@jsr/std__testing/time"; +import { FakeRuntime } from "../_fakeRuntime.ts"; import { relay, TimeoutWheel } from "../../src/transport/_abort.ts"; /** A live AbortController's signal, aborted with `reason` when given. */ @@ -400,51 +400,66 @@ describe("abort.TimeoutWheel", () => { }); }); - describe("fake time", () => { - // The wheel reads Date.now() for deadlines and arms plain setTimeout/clearTimeout — the - // pair FakeTime patches together — so tick-driven tests behave exactly like real ones. - let time: FakeTime; + describe("injected runtime", () => { + // The wheel reads monotonic time for deadlines and arms its timer through the runtime, so a + // virtual clock drives it without patching Date.now or the global timer functions. + let clock: FakeRuntime; beforeEach(() => { - time = new FakeTime(); - }); - afterEach(() => { - time.restore(); + clock = new FakeRuntime(); }); - test("deadlines track the patched clock", () => { - const wheel = new TimeoutWheel(); + test("deadlines track the virtual clock", () => { + const wheel = new TimeoutWheel(clock); const controller = new AbortController(); wheel.schedule(controller, 5); - time.tick(4); + clock.advance(4); assert(!controller.signal.aborted); - time.tick(1); + clock.advance(1); assert(controller.signal.aborted); assertEquals(controller.signal.reason.name, "TimeoutError"); + assertEquals(clock.pendingTimers, 0); }); test("a not-yet-due entry survives an early wake-up and is re-armed", () => { - const wheel = new TimeoutWheel(); + const wheel = new TimeoutWheel(clock); const early = new AbortController(); const late = new AbortController(); wheel.schedule(early, 5); wheel.schedule(late, 100); - time.tick(5); // the shared timer fires for `early`; `late` is not due + clock.advance(5); // the shared timer fires for `early`; `late` is not due assert(early.signal.aborted); assert(!late.signal.aborted); + assertEquals(clock.pendingTimers, 1); - time.tick(95); // the re-armed timer serves `late` at its own deadline + clock.advance(95); // the re-armed timer serves `late` at its own deadline assert(late.signal.aborted); + assertEquals(clock.pendingTimers, 0); }); - test("cancel under fake time still prevents the abort", () => { - const wheel = new TimeoutWheel(); + test("cancel on the virtual clock still prevents the abort", () => { + const wheel = new TimeoutWheel(clock); const controller = new AbortController(); wheel.schedule(controller, 5).cancel(); - time.tick(1_000); + clock.advance(1_000); + assert(!controller.signal.aborted); + assertEquals(clock.pendingTimers, 0); + }); + + test("deadlines are monotonic: a wall-clock rollback neither delays nor hastens them", () => { + const wheel = new TimeoutWheel(clock); + const controller = new AbortController(); + wheel.schedule(controller, 5); + + clock.setWallTime(clock.now() - 60_000); + clock.advance(4); + assert(!controller.signal.aborted); + clock.setWallTime(clock.now() + 3_600_000); assert(!controller.signal.aborted); + clock.advance(1); + assert(controller.signal.aborted); }); }); }); diff --git a/packages/hyperliquid/tests/transport/_infoCache.test.ts b/packages/hyperliquid/tests/transport/_infoCache.test.ts index aa5d458d..64938cb7 100644 --- a/packages/hyperliquid/tests/transport/_infoCache.test.ts +++ b/packages/hyperliquid/tests/transport/_infoCache.test.ts @@ -5,9 +5,9 @@ * @module */ -import { afterEach, beforeEach, describe, test } from "bun:test"; +import { beforeEach, describe, test } from "bun:test"; import { assertEquals, assertRejects, assertThrows } from "@jsr/std__assert"; -import { FakeTime } from "@jsr/std__testing/time"; +import { FakeRuntime } from "../_fakeRuntime.ts"; import { InfoCacheTransport } from "../../src/transport/_infoCache.ts"; import type { IRequestTransport } from "../../src/transport/_base.ts"; @@ -33,19 +33,15 @@ function echoHandler(_endpoint: Endpoint, payload: Record): unk } describe("InfoCacheTransport", () => { - let time: FakeTime; + let time: FakeRuntime; beforeEach(() => { - time = new FakeTime(); - }); - - afterEach(() => { - time.restore(); + time = new FakeRuntime(); }); test("serves a repeated allowlisted request from cache without a second network call", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 60_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000 }); const first = await transport.request("info", { type: "meta" }); const second = await transport.request("info", { type: "meta" }); @@ -56,28 +52,25 @@ describe("InfoCacheTransport", () => { test("refetches once the TTL expires", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 1_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 1_000 }); await transport.request("info", { type: "spotMeta" }); - time.tick(999); + time.advance(999); await transport.request("info", { type: "spotMeta" }); assertEquals(mock.calls.length, 1); // one ms short of expiry: still cached - time.tick(1); + time.advance(1); await transport.request("info", { type: "spotMeta" }); assertEquals(mock.calls.length, 2); // exactly at expiry: stale, refetched }); test("applies the default TTL unless a per-type override exists", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { - ttl: 60_000, - ttlByType: { marginTable: 500 }, - }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000, ttlByType: { marginTable: 500 } }); await transport.request("info", { type: "marginTable", id: 1 }); await transport.request("info", { type: "meta" }); - time.tick(600); + time.advance(600); await transport.request("info", { type: "marginTable", id: 1 }); // override expired: refetch await transport.request("info", { type: "meta" }); // default TTL still fresh: cache hit @@ -86,7 +79,7 @@ describe("InfoCacheTransport", () => { test("keys entries by request params, so distinct params never collide", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 60_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000 }); await transport.request("info", { type: "marginTable", id: 1 }); await transport.request("info", { type: "marginTable", id: 2 }); @@ -105,7 +98,7 @@ describe("InfoCacheTransport", () => { test("passes non-allowlisted info requests straight through", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 60_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000 }); await transport.request("info", { type: "allMids" }); await transport.request("info", { type: "allMids" }); @@ -121,7 +114,7 @@ describe("InfoCacheTransport", () => { test("passes exchange and explorer requests straight through", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 60_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000 }); await transport.request("exchange", { action: { type: "order" } }); await transport.request("exchange", { action: { type: "order" } }); @@ -139,7 +132,7 @@ describe("InfoCacheTransport", () => { test("ttl: 0 disables caching", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 0 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 0 }); await transport.request("info", { type: "meta" }); await transport.request("info", { type: "meta" }); @@ -150,7 +143,7 @@ describe("InfoCacheTransport", () => { let release: (value: unknown) => void; const gate = new Promise((resolve) => (release = resolve)); const mock = new MockTransport(() => gate); - const transport = new InfoCacheTransport(mock, { ttl: 60_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000 }); const first = transport.request("info", { type: "perpDexs" }); const second = transport.request("info", { type: "perpDexs" }); @@ -170,7 +163,7 @@ describe("InfoCacheTransport", () => { } return { echo: payload }; }); - const transport = new InfoCacheTransport(mock, { ttl: 60_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000 }); await assertRejects(() => transport.request("info", { type: "outcomeMeta" })); const retried = await transport.request("info", { type: "outcomeMeta" }); @@ -183,7 +176,7 @@ describe("InfoCacheTransport", () => { test("clear() drops every cached entry", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 60_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000 }); await transport.request("info", { type: "meta" }); await transport.request("info", { type: "tokenDetails", tokenId: "0x1" }); @@ -197,12 +190,12 @@ describe("InfoCacheTransport", () => { test("evicts the oldest entries beyond maxSize", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 60_000, maxSize: 2 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000, maxSize: 2 }); await transport.request("info", { type: "tokenDetails", tokenId: "0x1" }); - time.tick(1); // distinct insertion order for oldest-first eviction + time.advance(1); // distinct insertion order for oldest-first eviction await transport.request("info", { type: "tokenDetails", tokenId: "0x2" }); - time.tick(1); + time.advance(1); await transport.request("info", { type: "tokenDetails", tokenId: "0x3" }); // evicts 0x1 await transport.request("info", { type: "tokenDetails", tokenId: "0x3" }); // cached @@ -215,7 +208,7 @@ describe("InfoCacheTransport", () => { test("caches every allowlisted type", async () => { const mock = new MockTransport(echoHandler); - const transport = new InfoCacheTransport(mock, { ttl: 60_000 }); + const transport = new InfoCacheTransport(mock, { runtime: time, ttl: 60_000 }); const types = [ "meta", "spotMeta", @@ -234,11 +227,11 @@ describe("InfoCacheTransport", () => { test("rejects invalid options", () => { const mock = new MockTransport(echoHandler); - assertThrows(() => new InfoCacheTransport(mock, { ttl: -1 }), RangeError); - assertThrows(() => new InfoCacheTransport(mock, { ttl: Number.NaN }), RangeError); - assertThrows(() => new InfoCacheTransport(mock, { ttlByType: { meta: -5 } }), RangeError); - assertThrows(() => new InfoCacheTransport(mock, { maxSize: 0 }), RangeError); - assertThrows(() => new InfoCacheTransport(mock, { maxSize: 1.5 }), RangeError); + assertThrows(() => new InfoCacheTransport(mock, { runtime: time, ttl: -1 }), RangeError); + assertThrows(() => new InfoCacheTransport(mock, { runtime: time, ttl: Number.NaN }), RangeError); + assertThrows(() => new InfoCacheTransport(mock, { runtime: time, ttlByType: { meta: -5 } }), RangeError); + assertThrows(() => new InfoCacheTransport(mock, { runtime: time, maxSize: 0 }), RangeError); + assertThrows(() => new InfoCacheTransport(mock, { runtime: time, maxSize: 1.5 }), RangeError); }); test("mirrors the wrapped transport's isTestnet flag", () => { diff --git a/packages/hyperliquid/tests/transport/_infoCacheExpiry.test.ts b/packages/hyperliquid/tests/transport/_infoCacheExpiry.test.ts index 66d20d29..5d35a6b0 100644 --- a/packages/hyperliquid/tests/transport/_infoCacheExpiry.test.ts +++ b/packages/hyperliquid/tests/transport/_infoCacheExpiry.test.ts @@ -1,5 +1,5 @@ import { expect, test } from "bun:test"; -import { FakeTime } from "@jsr/std__testing/time"; +import { FakeRuntime } from "../_fakeRuntime.ts"; import { InfoCacheTransport } from "../../src/transport/_infoCache.ts"; import type { IRequestTransport } from "../../src/transport/_base.ts"; interface Entry { @@ -35,29 +35,30 @@ function assertBoundedIndex(cache: InfoCacheTransport, bound: number): void { } } test("mixed TTL capacity misses evict expired entries before older infinite-TTL entries", async () => { - const time = new FakeTime(); + const time = new FakeRuntime(); let calls = 0; const inner: IRequestTransport = { isTestnet: false, request: async (): Promise => ++calls as T }; - const cache = new InfoCacheTransport(inner, { maxSize: 3, ttl: Infinity, ttlByType: { marginTable: 10 } }); - try { - const retained = await cache.request("info", { type: "meta" }); - await cache.request("info", { type: "marginTable", id: 1 }); - await cache.request("info", { type: "marginTable", id: 2 }); - time.tick(10); - await cache.request("info", { type: "tokenDetails", tokenId: "new" }); - expect(await cache.request("info", { type: "meta" })).toBe(retained); - expect(calls).toBe(4); - assertBoundedIndex(cache, 3); - await cache.request("info", { type: "tokenDetails", tokenId: "next" }); - expect(await cache.request("info", { type: "meta" })).toBe(retained); - await cache.request("info", { type: "tokenDetails", tokenId: "final" }); - expect(await cache.request("info", { type: "meta" })).not.toBe(retained); - } finally { - time.restore(); - } + const cache = new InfoCacheTransport(inner, { + runtime: time, + maxSize: 3, + ttl: Infinity, + ttlByType: { marginTable: 10 }, + }); + const retained = await cache.request("info", { type: "meta" }); + await cache.request("info", { type: "marginTable", id: 1 }); + await cache.request("info", { type: "marginTable", id: 2 }); + time.advance(10); + await cache.request("info", { type: "tokenDetails", tokenId: "new" }); + expect(await cache.request("info", { type: "meta" })).toBe(retained); + expect(calls).toBe(4); + assertBoundedIndex(cache, 3); + await cache.request("info", { type: "tokenDetails", tokenId: "next" }); + expect(await cache.request("info", { type: "meta" })).toBe(retained); + await cache.request("info", { type: "tokenDetails", tokenId: "final" }); + expect(await cache.request("info", { type: "meta" })).not.toBe(retained); }); test("expiry bookkeeping stays bounded through heterogeneous churn, rejection, overwrite and clear", async () => { - const time = new FakeTime(); + const time = new FakeRuntime(); let count = 0; const inner: IRequestTransport = { isTestnet: false, @@ -67,29 +68,26 @@ test("expiry bookkeeping stays bounded through heterogeneous churn, rejection, o }, }; const cache = new InfoCacheTransport(inner, { + runtime: time, maxSize: 31, ttl: 100, ttlByType: { meta: 10, marginTable: 1000, tokenDetails: Infinity }, }); - try { - let random = 12345; - for (let i = 0; i < 2000; i++) { - random = (random * 1664525 + 1013904223) >>> 0; - const type = ["meta", "marginTable", "tokenDetails"][random % 3]; - await cache.request("info", { type, id: random % 79 }).catch(() => {}); - time.tick(random % 5); - if (i % 127 === 0) cache.clear(); - if (i % 13 === 0) assertBoundedIndex(cache, 31); - } - assertBoundedIndex(cache, 31); - cache.clear(); - expect(indexes(cache)._expiry).toHaveLength(0); - } finally { - time.restore(); + let random = 12345; + for (let i = 0; i < 2000; i++) { + random = (random * 1664525 + 1013904223) >>> 0; + const type = ["meta", "marginTable", "tokenDetails"][random % 3]; + await cache.request("info", { type, id: random % 79 }).catch(() => {}); + time.advance(random % 5); + if (i % 127 === 0) cache.clear(); + if (i % 13 === 0) assertBoundedIndex(cache, 31); } + assertBoundedIndex(cache, 31); + cache.clear(); + expect(indexes(cache)._expiry).toHaveLength(0); }); test("a stale pending rejection cannot remove the replacement entry or its expiry node", async () => { - const time = new FakeTime(); + const time = new FakeRuntime(); const pending: ReturnType>[] = []; const inner: IRequestTransport = { isTestnet: false, @@ -99,22 +97,18 @@ test("a stale pending rejection cannot remove the replacement entry or its expir return request.promise as Promise; }, }; - const cache = new InfoCacheTransport(inner, { ttl: 10, maxSize: 2 }); + const cache = new InfoCacheTransport(inner, { runtime: time, ttl: 10, maxSize: 2 }); const payload = { type: "meta" }; - try { - const first = cache.request("info", payload); - const failed = first.catch((error: unknown) => error); - time.tick(11); - const second = cache.request("info", payload); - pending[0].reject(new Error("old failed")); - expect(((await failed) as Error).message).toBe("old failed"); - expect(cache.request("info", payload)).toBe(second); - assertBoundedIndex(cache, 2); - pending[1].resolve("new value"); - expect(await second).toBe("new value"); - cache.clear(); - assertBoundedIndex(cache, 2); - } finally { - time.restore(); - } + const first = cache.request("info", payload); + const failed = first.catch((error: unknown) => error); + time.advance(11); + const second = cache.request("info", payload); + pending[0].reject(new Error("old failed")); + expect(((await failed) as Error).message).toBe("old failed"); + expect(cache.request("info", payload)).toBe(second); + assertBoundedIndex(cache, 2); + pending[1].resolve("new value"); + expect(await second).toBe("new value"); + cache.clear(); + assertBoundedIndex(cache, 2); }); diff --git a/packages/hyperliquid/tests/transport/_rateLimiter.test.ts b/packages/hyperliquid/tests/transport/_rateLimiter.test.ts index 317e5591..39083a50 100644 --- a/packages/hyperliquid/tests/transport/_rateLimiter.test.ts +++ b/packages/hyperliquid/tests/transport/_rateLimiter.test.ts @@ -4,9 +4,9 @@ * @module */ -import { afterEach, beforeEach, describe, test } from "bun:test"; +import { beforeEach, describe, test } from "bun:test"; import { assert, assertEquals, assertRejects, assertThrows } from "@jsr/std__assert"; -import { FakeTime } from "@jsr/std__testing/time"; +import { FakeRuntime } from "../_fakeRuntime.ts"; import { TokenBucketRateLimiter } from "../../src/transport/_rateLimiter.ts"; /** Waits until queued promise reactions have settled. */ @@ -22,20 +22,14 @@ function track(promise: Promise): { resolved: boolean } { } describe("TokenBucketRateLimiter", () => { - // The npm build of `@std/testing/time` drops the `[Symbol.dispose]` member the Deno version - // declares, so the clock is installed and restored through hooks instead of a `using` binding. - let time: FakeTime; + let time: FakeRuntime; beforeEach(() => { - time = new FakeTime(); - }); - - afterEach(() => { - time.restore(); + time = new FakeRuntime(); }); test("resolves without waiting while the bucket covers the weight", async () => { - const bucket = new TokenBucketRateLimiter(2, 60); + const bucket = new TokenBucketRateLimiter(2, 60, time); const first = track(bucket.acquire(1)); const second = track(bucket.acquire(1)); @@ -49,7 +43,7 @@ describe("TokenBucketRateLimiter", () => { }); test("defaults to a weight of 1 per acquisition", async () => { - const bucket = new TokenBucketRateLimiter(3, 60); + const bucket = new TokenBucketRateLimiter(3, 60, time); bucket.acquire(2); // 2 of 3 tokens const immediate = track(bucket.acquire(1)); // the last token @@ -60,28 +54,28 @@ describe("TokenBucketRateLimiter", () => { }); test("refills at the configured per-minute rate", async () => { - const bucket = new TokenBucketRateLimiter(1, 60); // 1 token per second + const bucket = new TokenBucketRateLimiter(1, 60, time); // 1 token per second bucket.acquire(1); // empties the bucket const waiting = track(bucket.acquire(1)); await flush(); assertEquals(waiting.resolved, false); - time.tick(999); // 1 ms short of the next token + time.advance(999); // 1 ms short of the next token await flush(); assertEquals(waiting.resolved, false); - time.tick(1); // exactly one second: one token + time.advance(1); // exactly one second: one token await flush(); assert(waiting.resolved); }); test("caps the refill at the capacity", async () => { - const bucket = new TokenBucketRateLimiter(2, 1200); // 20 tokens per second + const bucket = new TokenBucketRateLimiter(2, 1200, time); // 20 tokens per second bucket.acquire(1); bucket.acquire(1); // bucket empty - time.tick(10_000); // would earn 200 uncapped; the bucket holds at most 2 + time.advance(10_000); // would earn 200 uncapped; the bucket holds at most 2 await flush(); const first = track(bucket.acquire(1)); @@ -93,48 +87,48 @@ describe("TokenBucketRateLimiter", () => { }); test("serves waiting acquisitions in arrival order", async () => { - const bucket = new TokenBucketRateLimiter(1, 60); // 1 token per second + const bucket = new TokenBucketRateLimiter(1, 60, time); // 1 token per second bucket.acquire(1); // empties the bucket const first = track(bucket.acquire(1)); const second = track(bucket.acquire(1)); await flush(); - time.tick(1_000); // one token: only the earlier waiter may proceed + time.advance(1_000); // one token: only the earlier waiter may proceed await flush(); assert(first.resolved); assertEquals(second.resolved, false); - time.tick(1_000); + time.advance(1_000); await flush(); assert(second.resolved); }); test("a small later request never overtakes a large earlier one", async () => { - const bucket = new TokenBucketRateLimiter(5, 300); // 5 tokens per second + const bucket = new TokenBucketRateLimiter(5, 300, time); // 5 tokens per second bucket.acquire(5); // empties the bucket const large = track(bucket.acquire(4)); // needs 800 ms const small = track(bucket.acquire(1)); // could proceed after 200 ms on its own await flush(); - time.tick(600); // 3 tokens: enough for `small`, not for `large` + time.advance(600); // 3 tokens: enough for `small`, not for `large` await flush(); assertEquals(large.resolved, false); assertEquals(small.resolved, false); - time.tick(200); // 4 tokens: `large` proceeds, consuming all of them + time.advance(200); // 4 tokens: `large` proceeds, consuming all of them await flush(); assert(large.resolved); assertEquals(small.resolved, false); - time.tick(200); // 1 token: now `small` + time.advance(200); // 1 token: now `small` await flush(); assert(small.resolved); }); test("an over-capacity weight is served out of a full bucket and drives it into debt", async () => { - const bucket = new TokenBucketRateLimiter(10, 600); // 10 tokens per second + const bucket = new TokenBucketRateLimiter(10, 600, time); // 10 tokens per second // 15 > capacity: the acquisition can only be served by the full bucket (10), // leaving a debt of 5 tokens. @@ -142,7 +136,7 @@ describe("TokenBucketRateLimiter", () => { await flush(); assertEquals(large.resolved, false); - time.tick(1); // the bucket was already full; the wake-up fires immediately + time.advance(1); // the bucket was already full; the wake-up fires immediately await flush(); assert(large.resolved); @@ -151,17 +145,17 @@ describe("TokenBucketRateLimiter", () => { await flush(); assertEquals(next.resolved, false); - time.tick(500); // 501 ms of the 600 ms the debt needs + time.advance(500); // 501 ms of the 600 ms the debt needs await flush(); assertEquals(next.resolved, false); - time.tick(200); // 701 ms: comfortably past the debt + time.advance(200); // 701 ms: comfortably past the debt await flush(); assert(next.resolved); }); test("an already-aborted signal rejects without spending tokens", async () => { - const bucket = new TokenBucketRateLimiter(1, 60); + const bucket = new TokenBucketRateLimiter(1, 60, time); const reason = new Error("aborted"); await assertRejects(() => bucket.acquire(1, AbortSignal.abort(reason)), Error, "aborted"); @@ -173,7 +167,7 @@ describe("TokenBucketRateLimiter", () => { }); test("aborting a queued acquisition removes it without spending tokens", async () => { - const bucket = new TokenBucketRateLimiter(1, 60); // 1 token per second + const bucket = new TokenBucketRateLimiter(1, 60, time); // 1 token per second bucket.acquire(1); // empties the bucket const controller = new AbortController(); @@ -186,7 +180,7 @@ describe("TokenBucketRateLimiter", () => { await flush(); assertEquals(next.resolved, false); - time.tick(1_000); + time.advance(1_000); await flush(); assert(next.resolved); }); @@ -194,9 +188,9 @@ describe("TokenBucketRateLimiter", () => { test("schedules year-scale waits in bounded slices and serves at the real due time", async () => { // 1 token per 60_000_000_000 ms: the wake-up for the next token is ~1.9 years out, far // beyond the 2^31-1 ms a single setTimeout can express without clamping to ~1 ms. - const bucket = new TokenBucketRateLimiter(1, 1e-6); + const bucket = new TokenBucketRateLimiter(1, 1e-6, time); bucket.acquire(1); // consumes the only token - const start = time.now; + const start = time.now(); const waiting = track(bucket.acquire(1)); await flush(); @@ -206,7 +200,7 @@ describe("TokenBucketRateLimiter", () => { // reach the real due time — not a clamped early one, and never a spin. let advances = 0; while (!waiting.resolved) { - assert(time.next(), "a wake-up must always be armed while a waiter is pending"); + assert(time.nextTimer(), "a wake-up must always be armed while a waiter is pending"); advances++; assert(advances <= 100, "the scheduler must not spin: at most a handful of slices per wait"); await flush(); @@ -214,12 +208,12 @@ describe("TokenBucketRateLimiter", () => { assert(advances > 1, "a year-scale wait must be sliced, not scheduled as one overflowing timer"); const expected = Math.ceil(1 / (1e-6 / 60_000)); // the same arithmetic the bucket performs - const elapsed = time.now - start; + const elapsed = time.now() - start; assert(Math.abs(elapsed - expected) <= 1, `served after ${elapsed} ms, expected ~${expected}`); }); test("abort removes waiters at middle and tail; survivors keep FIFO order", async () => { - const bucket = new TokenBucketRateLimiter(1, 60); // 1 token per second + const bucket = new TokenBucketRateLimiter(1, 60, time); // 1 token per second bucket.acquire(1); // empties the bucket const order: string[] = []; @@ -239,25 +233,25 @@ describe("TokenBucketRateLimiter", () => { // Drop the middle: the head's wake-up is untouched, the tail moves up one slot. middle.abort(new Error("drop middle")); - time.tick(1_000); + time.advance(1_000); await flush(); assertEquals(order, ["rejected:middle", "first"]); // Drop the tail (now also the head): the queue empties without serving it. tail.abort(new Error("drop tail")); - time.tick(5_000); + time.advance(5_000); await flush(); assertEquals(order, ["rejected:middle", "first", "rejected:last"]); // The bucket still works: a fresh acquisition is served one refill later. named("after"); - time.tick(1_000); + time.advance(1_000); await flush(); assertEquals(order, ["rejected:middle", "first", "rejected:last", "after"]); }); test("aborting the head re-times the wake-up for the next waiter", async () => { - const bucket = new TokenBucketRateLimiter(5, 300); // 5 tokens per second + const bucket = new TokenBucketRateLimiter(5, 300, time); // 5 tokens per second bucket.acquire(5); // empties the bucket const controller = new AbortController(); @@ -268,13 +262,13 @@ describe("TokenBucketRateLimiter", () => { await assertRejects(() => large, Error, "skip"); // `small` becomes the head and goes at 200 ms, not at the stale 800 ms wake-up. - time.tick(200); + time.advance(200); await flush(); assert(small.resolved); }); test("charge debits tokens without waiting, driving the bucket into debt", async () => { - const bucket = new TokenBucketRateLimiter(5, 300); // 5 tokens per second + const bucket = new TokenBucketRateLimiter(5, 300, time); // 5 tokens per second bucket.acquire(3); // 2 tokens left bucket.charge(4); // post-hoc debit: -2 tokens of debt @@ -283,17 +277,17 @@ describe("TokenBucketRateLimiter", () => { await flush(); assertEquals(next.resolved, false); - time.tick(599); + time.advance(599); await flush(); assertEquals(next.resolved, false); - time.tick(1); + time.advance(1); await flush(); assert(next.resolved); }); test("tryAcquire deducts synchronously while the bucket covers the weight", () => { - const bucket = new TokenBucketRateLimiter(2, 60); + const bucket = new TokenBucketRateLimiter(2, 60, time); assert(bucket.tryAcquire(1)); assert(bucket.tryAcquire(1)); @@ -301,54 +295,54 @@ describe("TokenBucketRateLimiter", () => { assertEquals(bucket.tryAcquire(1), false); // A refused probe spent nothing and armed no timer: the refill alone re-covers it. - time.tick(1_000); // 1 token at 60/minute + time.advance(1_000); // 1 token at 60/minute assert(bucket.tryAcquire(1)); }); test("tryAcquire never bypasses queued waiters, even when the tokens would cover it", async () => { - const bucket = new TokenBucketRateLimiter(5, 300); // 5 tokens per second + const bucket = new TokenBucketRateLimiter(5, 300, time); // 5 tokens per second bucket.acquire(5); // empties the bucket const large = track(bucket.acquire(4)); // queued: needs 800 ms await flush(); - time.tick(600); // 3 tokens: enough for a weight-1 probe on its own, not for the head + time.advance(600); // 3 tokens: enough for a weight-1 probe on its own, not for the head // FIFO is not bypassed: the head waiter keeps its turn, so the probe refuses even // though the bucket covers its weight. assertEquals(bucket.tryAcquire(1), false); // And the refusal deducted nothing: the head is served on its original schedule. - time.tick(200); // 4 tokens + time.advance(200); // 4 tokens await flush(); assert(large.resolved); // With the queue drained, the same probe succeeds again. - time.tick(200); // 1 token + time.advance(200); // 1 token assert(bucket.tryAcquire(1)); }); test("rejects non-positive, non-finite, or underflowing configuration", () => { // Zero and negatives would stall the queue forever. - assertThrows(() => new TokenBucketRateLimiter(0, 60), RangeError, "capacity=0"); - assertThrows(() => new TokenBucketRateLimiter(60, 0), RangeError, "refillPerMinute=0"); - assertThrows(() => new TokenBucketRateLimiter(-1, 60), RangeError, "capacity=-1"); - assertThrows(() => new TokenBucketRateLimiter(60, -1), RangeError, "refillPerMinute=-1"); + assertThrows(() => new TokenBucketRateLimiter(0, 60, time), RangeError, "capacity=0"); + assertThrows(() => new TokenBucketRateLimiter(60, 0, time), RangeError, "refillPerMinute=0"); + assertThrows(() => new TokenBucketRateLimiter(-1, 60, time), RangeError, "capacity=-1"); + assertThrows(() => new TokenBucketRateLimiter(60, -1, time), RangeError, "refillPerMinute=-1"); // NaN and Infinity break the token arithmetic. - assertThrows(() => new TokenBucketRateLimiter(Number.NaN, 60), RangeError, "capacity=NaN"); - assertThrows(() => new TokenBucketRateLimiter(60, Number.NaN), RangeError, "refillPerMinute=NaN"); - assertThrows(() => new TokenBucketRateLimiter(Number.POSITIVE_INFINITY, 60), RangeError, "capacity=Infinity"); + assertThrows(() => new TokenBucketRateLimiter(Number.NaN, 60, time), RangeError, "capacity=NaN"); + assertThrows(() => new TokenBucketRateLimiter(60, Number.NaN, time), RangeError, "refillPerMinute=NaN"); + assertThrows(() => new TokenBucketRateLimiter(Number.POSITIVE_INFINITY, 60, time), RangeError, "capacity=Infinity"); assertThrows( - () => new TokenBucketRateLimiter(60, Number.POSITIVE_INFINITY), + () => new TokenBucketRateLimiter(60, Number.POSITIVE_INFINITY, time), RangeError, "refillPerMinute=Infinity", ); // The derived per-ms refill underflows to zero: a queued acquire would wait forever. - assertThrows(() => new TokenBucketRateLimiter(1, Number.MIN_VALUE), RangeError, "refillPerMinute=5e-324"); + assertThrows(() => new TokenBucketRateLimiter(1, Number.MIN_VALUE, time), RangeError, "refillPerMinute=5e-324"); }); test("accepts the smallest refill that still yields a nonzero per-ms rate", async () => { // 60_000 * MIN_VALUE is about the smallest refillPerMinute whose refillPerMs stays nonzero. - const bucket = new TokenBucketRateLimiter(1, 60_000 * Number.MIN_VALUE); + const bucket = new TokenBucketRateLimiter(1, 60_000 * Number.MIN_VALUE, time); const immediate = track(bucket.acquire(1)); await flush(); diff --git a/packages/hyperliquid/tests/transport/_runtime.test.ts b/packages/hyperliquid/tests/transport/_runtime.test.ts new file mode 100644 index 00000000..532a5d16 --- /dev/null +++ b/packages/hyperliquid/tests/transport/_runtime.test.ts @@ -0,0 +1,256 @@ +/** + * Tests for injectable transport runtimes: isolation from globals, deterministic jitter, timer + * and abort-listener cleanup, and wall-clock versus monotonic semantics. + * @module + */ + +import { expect, test } from "bun:test"; +import { getEventListeners } from "node:events"; +import { HttpTransport } from "../../src/transport/http/mod.ts"; +import { InfoCacheTransport } from "../../src/transport/_infoCache.ts"; +import { TimeoutWheel } from "../../src/transport/_abort.ts"; +import { TokenBucketRateLimiter } from "../../src/transport/_rateLimiter.ts"; +import { ReconnectingWebSocket } from "../../src/transport/websocket/_reconnectingSocket.ts"; +import { createNonceManager } from "../../src/api/exchange/_methods/_base/_nonce.ts"; +import { delay, resolveRuntime, systemRuntime } from "../../src/transport/runtime.ts"; +import { FakeRuntime, drain } from "../_fakeRuntime.ts"; +import { FakeSocket } from "../_fakeSocket.ts"; + +test("HTTP fetch/clock injection isolates simultaneous transports and retry cancellation releases timers", async () => { + const globalFetch = globalThis.fetch; + const clock = new FakeRuntime(); + let firstCalls = 0; + const first = new HttpTransport({ + runtime: clock, + retryOnRateLimit: true, + timeout: 1000, + fetch: async () => { + firstCalls++; + return new Response("limited", { status: 429, headers: { "Retry-After": "10" } }); + }, + }); + const second = new HttpTransport({ + runtime: new FakeRuntime(), + timeout: null, + fetch: async () => new Response('{"BTC":"1"}', { headers: { "Content-Type": "application/json" } }), + }); + const abort = new AbortController(); + const pending = first.request("info", {}, abort.signal).catch((e: unknown) => e); + await drain(); + expect(firstCalls).toBe(1); + expect(await second.request>("info", {})).toEqual({ BTC: "1" }); + abort.abort(new Error("stop")); + expect(await pending).toBeInstanceOf(Error); + // TimeoutWheel may retain its covered unref timer until its deadline; the retry delay is gone. + clock.advance(1000); + expect(clock.pendingTimers).toBe(0); + expect(getEventListeners(abort.signal, "abort")).toHaveLength(0); + expect(globalThis.fetch).toBe(globalFetch); +}); + +test("monotonic deadlines/refill ignore wall-clock rollback; nonce timestamps remain wall based", async () => { + const clock = new FakeRuntime(); + const wheel = new TimeoutWheel(clock); + const abort = new AbortController(); + wheel.schedule(abort, 10); + const bucket = new TokenBucketRateLimiter(1, 60, clock); + await bucket.acquire(1); + const waiting = bucket.acquire(1); + const nonce = createNonceManager(10, clock); + const before = nonce.getNonce("wallet"); + clock.setWallTime(before - 100_000); + clock.advance(10); + expect(abort.signal.aborted).toBe(true); + expect(nonce.getNonce("wallet")).toBe(before + 1); + clock.advance(990); + await waiting; + expect(clock.pendingTimers).toBe(0); +}); + +test("cache expiry and abortable delays run on an isolated clock", async () => { + const clock = new FakeRuntime(); + let calls = 0; + const inner = { isTestnet: true, request: async (): Promise => ++calls as T }; + const cache = new InfoCacheTransport(inner, { runtime: clock, ttl: 10 }); + await cache.request("info", { type: "meta" }); + await cache.request("info", { type: "meta" }); + expect(calls).toBe(1); + clock.setWallTime(clock.now() - 100_000); + clock.advance(10); + await cache.request("info", { type: "meta" }); + expect(calls).toBe(2); + const abort = new AbortController(); + const pending = delay(10, clock, abort.signal).catch((e: unknown) => e); + abort.abort("stop"); + expect(await pending).toBe("stop"); + expect(clock.pendingTimers).toBe(0); + expect(getEventListeners(abort.signal, "abort")).toHaveLength(0); +}); + +test("injected socket factories and reconnect timers do not replace global WebSocket", async () => { + const original = globalThis.WebSocket; + const clock = new FakeRuntime(); + const sockets: EventTarget[] = []; + const ws = new ReconnectingWebSocket("wss://test.invalid", { + runtime: clock, + webSocketFactory: () => { + const socket = new FakeSocket(); + sockets.push(socket); + return socket as unknown as WebSocket; + }, + connectionTimeout: 10, + reconnectionDelay: 5, + }); + await drain(); + expect(sockets).toHaveLength(1); + clock.advance(10); + await drain(); + expect(sockets).toHaveLength(1); + clock.advance(5); + await drain(); + expect(sockets).toHaveLength(2); + ws.close(); + expect(clock.pendingTimers).toBe(0); + expect(globalThis.WebSocket).toBe(original); +}); + +test("HTTP retry jitter, timeouts, success, and failure clean up without sleeping", async () => { + const clock = new FakeRuntime(); + let attempts = 0; + const transport = new HttpTransport({ + runtime: clock, + timeout: null, + retryOnRateLimit: true, + fetch: async () => { + attempts++; + return attempts === 1 + ? new Response("limited", { status: 429 }) + : new Response('{"ok":true}', { headers: { "Content-Type": "application/json" } }); + }, + }); + const abort = new AbortController(); + const retried = transport.request("info", {}, abort.signal); + await drain(); + expect(attempts).toBe(1); + expect(clock.pendingTimers).toBe(1); + clock.nextTimer(); + await retried; + expect(attempts).toBe(2); + expect(clock.pendingTimers).toBe(0); + expect(getEventListeners(abort.signal, "abort")).toHaveLength(0); + const failing = new HttpTransport({ + runtime: clock, + timeout: 10, + fetch: async () => { + throw new Error("offline"); + }, + }); + await expect(failing.request("info", {}, abort.signal)).rejects.toThrow("offline"); + clock.advance(10); + expect(clock.pendingTimers).toBe(0); + expect(getEventListeners(abort.signal, "abort")).toHaveLength(0); + const hanging = new HttpTransport({ + runtime: clock, + timeout: 10, + fetch: async (_url, init) => + new Promise((_resolve, reject) => { + const signal = init?.signal; + signal?.addEventListener("abort", () => reject(signal.reason), { once: true }); + }), + }); + const timedOut = hanging.request("info", {}, abort.signal).catch((e: unknown) => e); + clock.advance(10); + const error = await timedOut; + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain("timed out"); + expect(clock.pendingTimers).toBe(0); + expect(getEventListeners(abort.signal, "abort")).toHaveLength(0); +}); + +test("resolveRuntime returns the platform runtime when nothing is overridden and is idempotent", () => { + expect(resolveRuntime()).toBe(systemRuntime); + expect(resolveRuntime(systemRuntime)).toBe(systemRuntime); + const clock = new FakeRuntime(); + const resolved = resolveRuntime(clock); + expect(resolveRuntime(resolved)).toBe(resolved); // shared by a transport's components as one object + // Overrides keep their receiver; omitted members fall back to the platform. + const partial = resolveRuntime({ random: () => 0.25 }); + expect(partial.random()).toBe(0.25); + expect(partial.now).toBe(systemRuntime.now); + clock.advance(7); + expect(resolved.monotonicNow()).toBe(7); +}); + +test("injected random makes backoff jitter exact", async () => { + const clock = new FakeRuntime({ random: () => 0.5 }); + let attempts = 0; + const transport = new HttpTransport({ + runtime: clock, + timeout: null, + retryOnRateLimit: true, + fetch: async () => { + attempts++; + return attempts < 3 + ? new Response("limited", { status: 429 }) + : new Response("{}", { headers: { "Content-Type": "application/json" } }); + }, + }); + const pending = transport.request("info", {}); + await drain(); + clock.advance(499); // attempt 0 waits 0.5 * 1 s + await drain(); + expect(attempts).toBe(1); + clock.advance(1); + await drain(); + expect(attempts).toBe(2); + clock.advance(999); // attempt 1 waits 0.5 * 2 s + await drain(); + expect(attempts).toBe(2); + clock.advance(1); + await pending; + expect(attempts).toBe(3); + expect(clock.pendingTimers).toBe(0); +}); + +test("HTTP-date Retry-After is measured against the injected wall clock", async () => { + const clock = new FakeRuntime(); + let attempts = 0; + const retryAt = new Date(clock.now() + 3_000).toUTCString(); // whole seconds: up to 1 s sooner + const transport = new HttpTransport({ + runtime: clock, + timeout: null, + retryOnRateLimit: true, + fetch: async () => + ++attempts === 1 + ? new Response("limited", { status: 429, headers: { "Retry-After": retryAt } }) + : new Response("{}", { headers: { "Content-Type": "application/json" } }), + }); + const pending = transport.request("info", {}); + await drain(); + clock.advance(1_999); // at least 2 s remain on the virtual wall clock, whatever the real date + await drain(); + expect(attempts).toBe(1); + clock.advance(1_001); + await pending; + expect(attempts).toBe(2); +}); + +test("the platform delay timer is unref'd, so a pending retry wait never holds the process open", async () => { + const original = globalThis.setTimeout; + let timer: { hasRef?: () => boolean } | undefined; + globalThis.setTimeout = ((callback: () => void, ms: number) => { + const handle = original(callback, ms); + timer = handle as unknown as { hasRef?: () => boolean }; + return handle; + }) as typeof setTimeout; + const abort = new AbortController(); + let pending: Promise; + try { + pending = delay(60_000, systemRuntime, abort.signal).catch((e: unknown) => e); + } finally { + globalThis.setTimeout = original; + } + expect(timer?.hasRef?.()).toBe(false); + abort.abort("done"); + expect(await pending).toBe("done"); +}); diff --git a/packages/hyperliquid/tests/transport/_runtimeIntegration.test.ts b/packages/hyperliquid/tests/transport/_runtimeIntegration.test.ts new file mode 100644 index 00000000..0bb41acb --- /dev/null +++ b/packages/hyperliquid/tests/transport/_runtimeIntegration.test.ts @@ -0,0 +1,127 @@ +/** + * End-to-end checks for injected transport runtimes: HTTP and WebSocket transports built on + * their own virtual clocks, fetch, and socket factories run side by side without touching any + * global, while the coordination that must stay shared (per-signer nonces and the default + * per-network WebSocket quota) is not silently forked by the injection. + * @module + */ + +import { expect, test } from "bun:test"; +import { getEventListeners } from "node:events"; +import { ExchangeClient, HttpTransport, WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; +import { FakeRuntime, drain } from "../_fakeRuntime.ts"; +import { FakeSocket } from "../_fakeSocket.ts"; + +const OK = { status: "ok", response: { type: "default" } }; + +/** Minimal viem local-account shape: callable `signTypedData` plus a string address. */ +const wallet = { + address: "0x5f3c2a4b6d8e0f1a3c5e7b9d1f2a4c6e8b0d2f4a" as const, + signTypedData: async (): Promise<`0x${string}`> => `0x${"11".repeat(64)}1b` as `0x${string}`, +}; + +/** A WebSocket transport on `clock` whose socket answers every `post` with `OK`. */ +function webSocketOn(clock: FakeRuntime, quota?: WebSocketQuota, nonces?: number[]) { + const socket = new FakeSocket((data) => { + const frame = JSON.parse(data); + if (frame.method !== "post") return; + nonces?.push(frame.request.payload.nonce); + queueMicrotask(() => + socket.receive({ channel: "post", data: { id: frame.id, response: { type: "action", payload: OK } } }), + ); + }); + const transport = new WebSocketTransport({ + isTestnet: true, + runtime: clock, + quota, + webSocketFactory: () => socket as unknown as WebSocket, + }); + return { transport, socket }; +} + +test("HTTP and WebSocket clients on separate fake runtimes still share per-signer nonces", async () => { + const globalFetch = globalThis.fetch; + const globalWebSocket = globalThis.WebSocket; + const nonces: number[] = []; + + const httpClock = new FakeRuntime(); + const http = new HttpTransport({ + isTestnet: true, + runtime: httpClock, + fetch: async (_url, init) => { + nonces.push(JSON.parse(init!.body as string).nonce); + return new Response(JSON.stringify(OK), { headers: { "Content-Type": "application/json" } }); + }, + }); + + const wsClock = new FakeRuntime(); + const quota = new WebSocketQuota({ runtime: wsClock }); + const { transport: ws, socket } = webSocketOn(wsClock, quota, nonces); + await drain(); + socket.open(); + + try { + const clients = [new ExchangeClient({ transport: http, wallet }), new ExchangeClient({ transport: ws, wallet })]; + await Promise.all( + Array.from({ length: 6 }, (_, i) => clients[i % 2].updateLeverage({ asset: 0, isCross: true, leverage: 5 })), + ); + // One process-wide nonce source per signer/network: the injected clocks never fork it. + expect(nonces).toHaveLength(6); + expect(new Set(nonces).size).toBe(6); + } finally { + await ws.close(); + } + + // Close released the keep-alive and reconnect timers. A TimeoutWheel may keep its one covered + // (unref'd) timer until the default 10 s request deadline passes, then nothing remains. + httpClock.advance(10_000); + wsClock.advance(10_000); + expect(httpClock.pendingTimers).toBe(0); + expect(wsClock.pendingTimers).toBe(0); + expect(globalThis.fetch).toBe(globalFetch); + expect(globalThis.WebSocket).toBe(globalWebSocket); +}); + +test("keep-alive pings and request timeouts run on the injected WebSocket clock", async () => { + const clock = new FakeRuntime(); + const quota = new WebSocketQuota({ runtime: clock }); + const socket = new FakeSocket(); + const ws = new WebSocketTransport({ + runtime: clock, + quota, + timeout: 1_000, + webSocketFactory: () => socket as unknown as WebSocket, + }); + await drain(); + socket.open(); + + clock.advance(5_000); // default keep-alive interval + expect(JSON.parse(socket.sent.at(-1)!)).toEqual({ method: "ping" }); + + const signal = new AbortController().signal; + const pending = ws.request("info", { type: "allMids" }, signal).catch((e: unknown) => e); + await drain(); + clock.advance(999); + await drain(); + socket.receive({ channel: "pong" }); + clock.advance(1); // the request timeout fires on the virtual clock, not a real one + const error = await pending; + expect((error as Error).message).toContain("timed out"); + expect(getEventListeners(signal, "abort")).toHaveLength(0); + + await ws.close(); + expect(clock.pendingTimers).toBe(0); +}); + +test("injecting a runtime or socket factory keeps the default shared quota", async () => { + const first = webSocketOn(new FakeRuntime()); + const second = webSocketOn(new FakeRuntime()); + const isolated = webSocketOn(new FakeRuntime(), new WebSocketQuota()); + try { + // Same network, no explicit quota: both draw from one per-IP budget despite separate runtimes. + expect(first.transport.quota).toBe(second.transport.quota); + expect(isolated.transport.quota).not.toBe(first.transport.quota); + } finally { + await Promise.all([first, second, isolated].map(({ transport }) => transport.close())); + } +}); diff --git a/packages/hyperliquid/tests/transport/http/mod.test.ts b/packages/hyperliquid/tests/transport/http/mod.test.ts index 20f976e3..c32ad9b2 100644 --- a/packages/hyperliquid/tests/transport/http/mod.test.ts +++ b/packages/hyperliquid/tests/transport/http/mod.test.ts @@ -1,14 +1,16 @@ /** - * Tests for HttpTransport against a mocked global fetch: URL routing, - * error wrapping, fetch options merging, and abort/timeout handling. + * Tests for HttpTransport: URL routing, error wrapping, fetch options merging, and + * abort/timeout handling against the default (global) fetch, which is what an omitted + * `fetch` option resolves to; retry and rate-limit timing against an injected fetch and + * virtual clock. * @module */ -import { afterEach, beforeEach, describe, test } from "bun:test"; +import { beforeEach, describe, test } from "bun:test"; import { getEventListeners } from "node:events"; import { assert, assertEquals, assertIsError, assertRejects } from "@jsr/std__assert"; -import { FakeTime } from "@jsr/std__testing/time"; import { HttpRateLimitError, HttpRequestError, HttpTransport } from "@bloxwap/hyperliquid"; +import { FakeRuntime } from "../../_fakeRuntime.ts"; // ============================================================================= // Helpers @@ -63,6 +65,21 @@ function stubFetch(handler: (input: FetchArgs[0], init?: FetchArgs[1]) => Respon return stub; } +/** Injectable fetch, scoped to one transport, counting the requests it answers. */ +function fakeFetch(handler: (input: FetchArgs[0], init?: FetchArgs[1]) => Response | Promise): { + calls: number; + fetch: (input: FetchArgs[0], init?: FetchArgs[1]) => Promise; +} { + const stub = { + calls: 0, + fetch: async (input: FetchArgs[0], init?: FetchArgs[1]): Promise => { + stub.calls++; + return await handler(input, init); + }, + }; + return stub; +} + /** Waits until queued promise reactions have settled. */ async function flush(): Promise { for (let i = 0; i < 10; i++) await Promise.resolve(); @@ -312,6 +329,22 @@ describe("HttpTransport", () => { }); }); + describe("fetch option", () => { + test("calls the injected fetch without the transport as its receiver", async () => { + // Mirrors browsers' `window.fetch`, which throws "Illegal invocation" for any other receiver. + let calls = 0; + const receiverSensitiveFetch = function (this: unknown, _input: FetchArgs[0], _init?: FetchArgs[1]) { + if (this !== undefined && this !== globalThis) throw new TypeError("Illegal invocation"); + calls++; + return Promise.resolve(jsonResponse({ ok: true })); + }; + + const transport = new HttpTransport({ fetch: receiverSensitiveFetch }); + assertEquals(await transport.request("info", {}), { ok: true }); + assertEquals(calls, 1); + }); + }); + describe("fetchOptions", () => { test("headers as object", async () => { mockFetch((_req, init) => { @@ -783,16 +816,12 @@ describe("HttpTransport", () => { }); describe("retryOnRateLimit", () => { - // Same FakeTime hook pattern as the rateLimit block: retry waits are real timers, - // ticked deterministically here. - let time: FakeTime; + // Retry waits and timeouts run on an injected virtual clock (jitter pinned to 0) and the + // fetch is injected per transport, so nothing global is patched or slept on. + let clock: FakeRuntime; beforeEach(() => { - time = new FakeTime(); - }); - - afterEach(() => { - time.restore(); + clock = new FakeRuntime(); }); /** A 429 response carrying the given Retry-After value (or none). */ @@ -803,210 +832,169 @@ describe("HttpTransport", () => { }); test("disabled by default: a 429 throws on the first attempt", async () => { - const stub = stubFetch(() => rateLimited("0")); - try { - const transport = new HttpTransport(); - await assertRejects(() => transport.request("info", {}), HttpRateLimitError); - assertEquals(stub.calls, 1); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => rateLimited("0")); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch }); + await assertRejects(() => transport.request("info", {}), HttpRateLimitError); + assertEquals(stub.calls, 1); }); test("retries a 429 and resolves once the server recovers", async () => { - const stub = stubFetch(() => (stub.calls === 1 ? rateLimited("0") : jsonResponse({ ok: true }))); - try { - const transport = new HttpTransport({ retryOnRateLimit: true }); - const pending = transport.request("info", {}); - await flush(); - assertEquals(stub.calls, 1); + const stub = fakeFetch(() => (stub.calls === 1 ? rateLimited("0") : jsonResponse({ ok: true }))); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, retryOnRateLimit: true }); + const pending = transport.request("info", {}); + await flush(); + assertEquals(stub.calls, 1); - time.tick(1_000); // Retry-After 0 plus up to 1 s of jitter - assertEquals(await pending, { ok: true }); - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + clock.advance(1_000); // Retry-After 0 plus up to 1 s of jitter + assertEquals(await pending, { ok: true }); + assertEquals(stub.calls, 2); }); test("honors Retry-After: no retry before the asked delay (the jitter only ever adds wait)", async () => { - const stub = stubFetch(() => (stub.calls === 1 ? rateLimited("2") : jsonResponse())); - try { - const transport = new HttpTransport({ retryOnRateLimit: true }); - const pending = transport.request("info", {}); - await flush(); - assertEquals(stub.calls, 1); + const stub = fakeFetch(() => (stub.calls === 1 ? rateLimited("2") : jsonResponse())); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, retryOnRateLimit: true }); + const pending = transport.request("info", {}); + await flush(); + assertEquals(stub.calls, 1); - time.tick(1_999); // 1 ms short of the asked 2 s - await flush(); - assertEquals(stub.calls, 1); + clock.advance(1_999); // 1 ms short of the asked 2 s + await flush(); + assertEquals(stub.calls, 1); - time.tick(1_001); // past 2 s plus the worst-case 1 s of jitter - await pending; - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + clock.advance(1_001); // past 2 s plus the worst-case 1 s of jitter + await pending; + assertEquals(stub.calls, 2); }); test("without Retry-After falls back to bounded exponential backoff", async () => { - const stub = stubFetch(() => (stub.calls < 3 ? rateLimited() : jsonResponse())); - try { - const transport = new HttpTransport({ retryOnRateLimit: true }); - const pending = transport.request("info", {}); - await flush(); - assertEquals(stub.calls, 1); + const stub = fakeFetch(() => (stub.calls < 3 ? rateLimited() : jsonResponse())); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, retryOnRateLimit: true }); + const pending = transport.request("info", {}); + await flush(); + assertEquals(stub.calls, 1); - time.tick(1_000); // first backoff: a random wait within [0, 1 s) - await flush(); - assertEquals(stub.calls, 2); + clock.advance(1_000); // first backoff: a random wait within [0, 1 s) + await flush(); + assertEquals(stub.calls, 2); - time.tick(2_000); // second backoff: within [0, 2 s) - await pending; - assertEquals(stub.calls, 3); - } finally { - stub.restore(); - } + clock.advance(2_000); // second backoff: within [0, 2 s) + await pending; + assertEquals(stub.calls, 3); }); test("gives up after maxRetries and throws the last 429", async () => { - const stub = stubFetch(() => rateLimited()); - try { - const transport = new HttpTransport({ retryOnRateLimit: { maxRetries: 2 } }); - const pending = transport.request("info", {}); - const rejection = assertRejects(() => pending, HttpRateLimitError); - await flush(); - assertEquals(stub.calls, 1); - - time.tick(1_000); // retry 1 - await flush(); - assertEquals(stub.calls, 2); - - time.tick(2_000); // retry 2 — the next 429 is final - const error = await rejection; - assertEquals(error.status, 429); - assertEquals(stub.calls, 3); // maxRetries + 1 attempts in total - } finally { - stub.restore(); - } + const stub = fakeFetch(() => rateLimited()); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, retryOnRateLimit: { maxRetries: 2 } }); + const pending = transport.request("info", {}); + const rejection = assertRejects(() => pending, HttpRateLimitError); + await flush(); + assertEquals(stub.calls, 1); + + clock.advance(1_000); // retry 1 + await flush(); + assertEquals(stub.calls, 2); + + clock.advance(2_000); // retry 2 — the next 429 is final + const error = await rejection; + assertEquals(error.status, 429); + assertEquals(stub.calls, 3); // maxRetries + 1 attempts in total }); test("a Retry-After beyond maxDelayMs surfaces the 429 instead of waiting", async () => { - const stub = stubFetch(() => rateLimited("60")); - try { - const transport = new HttpTransport({ retryOnRateLimit: true }); // default maxDelayMs: 30 s - await assertRejects(() => transport.request("info", {}), HttpRateLimitError); - assertEquals(stub.calls, 1); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => rateLimited("60")); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, retryOnRateLimit: true }); // default maxDelayMs: 30 s + await assertRejects(() => transport.request("info", {}), HttpRateLimitError); + assertEquals(stub.calls, 1); }); test("maxDelayMs is configurable", async () => { - const stub = stubFetch(() => (stub.calls === 1 ? rateLimited("60") : jsonResponse())); - try { - // timeout: null — otherwise the default 10 s timeout (which spans retry waits) fires first. - const transport = new HttpTransport({ timeout: null, retryOnRateLimit: { maxDelayMs: 120_000 } }); - const pending = transport.request("info", {}); - await flush(); - assertEquals(stub.calls, 1); + const stub = fakeFetch(() => (stub.calls === 1 ? rateLimited("60") : jsonResponse())); + // timeout: null — otherwise the default 10 s timeout (which spans retry waits) fires first. + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + timeout: null, + retryOnRateLimit: { maxDelayMs: 120_000 }, + }); + const pending = transport.request("info", {}); + await flush(); + assertEquals(stub.calls, 1); - time.tick(61_000); // the asked 60 s plus up to 1 s of jitter - await pending; - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + clock.advance(61_000); // the asked 60 s plus up to 1 s of jitter + await pending; + assertEquals(stub.calls, 2); }); test("the overall request timeout spans every retry wait", async () => { - const stub = stubFetch(() => (stub.calls === 1 ? rateLimited("5") : jsonResponse())); - try { - const transport = new HttpTransport({ timeout: 100, retryOnRateLimit: true }); - const pending = transport.request("info", {}); - const rejection = assertRejects(() => pending, HttpRequestError, "Request timed out after 100 ms"); - await flush(); - assertEquals(stub.calls, 1); // first attempt 429'd; the ~5 s retry wait is pending + const stub = fakeFetch(() => (stub.calls === 1 ? rateLimited("5") : jsonResponse())); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, timeout: 100, retryOnRateLimit: true }); + const pending = transport.request("info", {}); + const rejection = assertRejects(() => pending, HttpRequestError, "Request timed out after 100 ms"); + await flush(); + assertEquals(stub.calls, 1); // first attempt 429'd; the ~5 s retry wait is pending - time.tick(100); // the timeout fires mid-wait — the retry never goes out - await rejection; - assertEquals(stub.calls, 1); - } finally { - stub.restore(); - } + clock.advance(100); // the timeout fires mid-wait — the retry never goes out + await rejection; + assertEquals(stub.calls, 1); }); test("a caller abort during the retry wait cancels the request", async () => { - const stub = stubFetch(() => (stub.calls === 1 ? rateLimited("5") : jsonResponse())); - try { - const controller = new AbortController(); - const reason = new DOMException("user cancel", "AbortError"); - const transport = new HttpTransport({ retryOnRateLimit: true }); - const pending = transport.request("info", {}, controller.signal); - const rejection = assertRejects(() => pending, HttpRequestError, "Request aborted"); - await flush(); - assertEquals(stub.calls, 1); + const stub = fakeFetch(() => (stub.calls === 1 ? rateLimited("5") : jsonResponse())); + const controller = new AbortController(); + const reason = new DOMException("user cancel", "AbortError"); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, retryOnRateLimit: true }); + const pending = transport.request("info", {}, controller.signal); + const rejection = assertRejects(() => pending, HttpRequestError, "Request aborted"); + await flush(); + assertEquals(stub.calls, 1); - controller.abort(reason); - const error = await rejection; - assertEquals(error.cause, reason); - assertEquals(stub.calls, 1); // the retry never went out - } finally { - stub.restore(); - } + controller.abort(reason); + const error = await rejection; + assertEquals(error.cause, reason); + assertEquals(stub.calls, 1); // the retry never went out }); test("only a 429 is retried: other failures surface on the first attempt", async () => { - const stub = stubFetch(() => new Response("nope", { status: 500 })); - try { - const transport = new HttpTransport({ retryOnRateLimit: true }); - const error = await assertRejects(() => transport.request("info", {}), HttpRequestError); - assert(!(error instanceof HttpRateLimitError)); - assertEquals(stub.calls, 1); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => new Response("nope", { status: 500 })); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, retryOnRateLimit: true }); + const error = await assertRejects(() => transport.request("info", {}), HttpRequestError); + assert(!(error instanceof HttpRateLimitError)); + assertEquals(stub.calls, 1); }); test("a 200-OK error envelope is not retried either", async () => { - const stub = stubFetch(() => jsonResponse({ type: "error", message: "server-side failure" })); - try { - const transport = new HttpTransport({ retryOnRateLimit: true }); - await assertRejects(() => transport.request("info", {}), HttpRequestError, "server-side failure"); - assertEquals(stub.calls, 1); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => jsonResponse({ type: "error", message: "server-side failure" })); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch, retryOnRateLimit: true }); + await assertRejects(() => transport.request("info", {}), HttpRequestError, "server-side failure"); + assertEquals(stub.calls, 1); }); test("composed with rateLimit, each retry debits the bucket again", async () => { - const stub = stubFetch(() => (stub.calls === 1 ? rateLimited("0") : jsonResponse())); - try { - // 1 weight per second: the first attempt acquires 1 (bucket 0), the retry charges 1 - // more (bucket -1), and the ~1 s retry wait refills exactly that token (bucket 0). - const transport = new HttpTransport({ - rateLimit: { capacity: 1, refillPerMinute: 60 }, - retryOnRateLimit: true, - }); - const pending = transport.request("exchange", { action: { type: "noop" } }); - await flush(); - assertEquals(stub.calls, 1); - - time.tick(1_000); // Retry-After 0 plus jitter — the retry goes out and succeeds - await pending; - assertEquals(stub.calls, 2); - - // The retry's debit left the bucket empty: the next request waits one refill second. - // (Had the retry not been charged, the refilled token would send it immediately.) - const followUp = transport.request("exchange", { action: { type: "noop" } }); - await flush(); - assertEquals(stub.calls, 2); - time.tick(1_000); - await followUp; - assertEquals(stub.calls, 3); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => (stub.calls === 1 ? rateLimited("0") : jsonResponse())); + // 1 weight per second: the first attempt acquires 1 (bucket 0), the retry charges 1 + // more (bucket -1), and the ~1 s retry wait refills exactly that token (bucket 0). + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + rateLimit: { capacity: 1, refillPerMinute: 60 }, + retryOnRateLimit: true, + }); + const pending = transport.request("exchange", { action: { type: "noop" } }); + await flush(); + assertEquals(stub.calls, 1); + + clock.advance(1_000); // Retry-After 0 plus jitter — the retry goes out and succeeds + await pending; + assertEquals(stub.calls, 2); + + // The retry's debit left the bucket empty: the next request waits one refill second. + // (Had the retry not been charged, the refilled token would send it immediately.) + const followUp = transport.request("exchange", { action: { type: "noop" } }); + await flush(); + assertEquals(stub.calls, 2); + clock.advance(1_000); + await followUp; + assertEquals(stub.calls, 3); }); }); @@ -1226,87 +1214,79 @@ describe("HttpTransport", () => { }); describe("rateLimit", () => { - // The npm build of `@std/testing/time` drops the `[Symbol.dispose]` member the Deno version - // declares, so the clock is installed and restored through hooks instead of a `using` binding. - let time: FakeTime; + // Bucket refills run on an injected virtual clock and an injected fetch, as in retryOnRateLimit. + let clock: FakeRuntime; beforeEach(() => { - time = new FakeTime(); - }); - - afterEach(() => { - time.restore(); + clock = new FakeRuntime(); }); test("waits for weight before sending instead of throwing", async () => { - const stub = stubFetch(() => jsonResponse()); - try { - const transport = new HttpTransport({ rateLimit: { capacity: 1, refillPerMinute: 60 } }); // 1 weight/second + const stub = fakeFetch(() => jsonResponse()); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + rateLimit: { capacity: 1, refillPerMinute: 60 }, + }); // 1 weight/second - await transport.request("exchange", { action: { type: "noop" } }); // consumes the only token - assertEquals(stub.calls, 1); + await transport.request("exchange", { action: { type: "noop" } }); // consumes the only token + assertEquals(stub.calls, 1); - const pending = transport.request("exchange", { action: { type: "noop" } }); // bucket empty: waits, no fetch yet - await flush(); - assertEquals(stub.calls, 1); + const pending = transport.request("exchange", { action: { type: "noop" } }); // bucket empty: waits, no fetch yet + await flush(); + assertEquals(stub.calls, 1); - time.tick(1_000); // one token refilled - await pending; - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + clock.advance(1_000); // one token refilled + await pending; + assertEquals(stub.calls, 2); }); test("exchange batches cost 1 + floor(batchLength / 40) weight", async () => { - const stub = stubFetch(() => jsonResponse()); - try { - const transport = new HttpTransport({ rateLimit: { capacity: 2, refillPerMinute: 60 } }); // 1 weight/second + const stub = fakeFetch(() => jsonResponse()); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + rateLimit: { capacity: 2, refillPerMinute: 60 }, + }); // 1 weight/second - // 41 orders cost 2 weight: the whole bucket. - await transport.request("exchange", { action: { type: "order", orders: Array.from({ length: 41 }) } }); - assertEquals(stub.calls, 1); + // 41 orders cost 2 weight: the whole bucket. + await transport.request("exchange", { action: { type: "order", orders: Array.from({ length: 41 }) } }); + assertEquals(stub.calls, 1); - const pending = transport.request("exchange", { action: { type: "noop" } }); // weight 1: waits 1 s - await flush(); - assertEquals(stub.calls, 1); + const pending = transport.request("exchange", { action: { type: "noop" } }); // weight 1: waits 1 s + await flush(); + assertEquals(stub.calls, 1); - time.tick(1_000); - await pending; - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + clock.advance(1_000); + await pending; + assertEquals(stub.calls, 2); }); test("throttled waits never trip the request timeout", async () => { - const stub = stubFetch(() => jsonResponse()); - try { - const transport = new HttpTransport({ timeout: 100, rateLimit: { capacity: 1, refillPerMinute: 60 } }); + const stub = fakeFetch(() => jsonResponse()); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + timeout: 100, + rateLimit: { capacity: 1, refillPerMinute: 60 }, + }); - await transport.request("exchange", { action: { type: "noop" } }); - const pending = transport.request("exchange", { action: { type: "noop" } }); // waits 1 s, far beyond the 100 ms timeout - await flush(); + await transport.request("exchange", { action: { type: "noop" } }); + const pending = transport.request("exchange", { action: { type: "noop" } }); // waits 1 s, far beyond the 100 ms timeout + await flush(); - time.tick(1_000); - await pending; // resolves: the timeout starts only once the request goes out - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + clock.advance(1_000); + await pending; // resolves: the timeout starts only once the request goes out + assertEquals(stub.calls, 2); }); test("disabled by default: requests are never delayed client-side", async () => { - const stub = stubFetch(() => jsonResponse()); - try { - const transport = new HttpTransport(); + const stub = fakeFetch(() => jsonResponse()); + const transport = new HttpTransport({ runtime: clock, fetch: stub.fetch }); - await transport.request("info", {}); - await transport.request("exchange", { action: { type: "order", orders: Array.from({ length: 100 }) } }); - assertEquals(stub.calls, 2); // both sent without ticking the clock - } finally { - stub.restore(); - } + await transport.request("info", {}); + await transport.request("exchange", { action: { type: "order", orders: Array.from({ length: 100 }) } }); + assertEquals(stub.calls, 2); // both sent without ticking the clock }); describe("documented weights", () => { @@ -1316,27 +1296,27 @@ describe("HttpTransport", () => { * exactly one second at the 1 token/second refill. */ async function assertWeight(endpoint: "info" | "exchange" | "explorer", payload: unknown, weight: number) { - const stub = stubFetch(() => jsonResponse()); - try { - const transport = new HttpTransport({ rateLimit: { capacity: weight, refillPerMinute: 60 } }); - - await transport.request(endpoint, payload); // empties the bucket exactly - assertEquals(stub.calls, 1); - - const pending = transport.request("exchange", { action: { type: "noop" } }); // weight 1 - await flush(); - assertEquals(stub.calls, 1); - - time.tick(999); // 1 ms short of the next token - await flush(); - assertEquals(stub.calls, 1); - - time.tick(1); - await pending; - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => jsonResponse()); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + rateLimit: { capacity: weight, refillPerMinute: 60 }, + }); + + await transport.request(endpoint, payload); // empties the bucket exactly + assertEquals(stub.calls, 1); + + const pending = transport.request("exchange", { action: { type: "noop" } }); // weight 1 + await flush(); + assertEquals(stub.calls, 1); + + clock.advance(999); // 1 ms short of the next token + await flush(); + assertEquals(stub.calls, 1); + + clock.advance(1); + await pending; + assertEquals(stub.calls, 2); } // The classes from https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits @@ -1475,32 +1455,32 @@ describe("HttpTransport", () => { }); test("billing derives from the serialized form, never from live getters or toJSON", async () => { - const stub = stubFetch(() => jsonResponse()); - try { - const transport = new HttpTransport({ rateLimit: { capacity: 2, refillPerMinute: 60 } }); - - // The live object says "userRole" (60 weight); the serialized form says "allMids" (2). - const payload = { - get type(): string { - return "userRole"; - }, - toJSON() { - return { type: "allMids" }; - }, - }; - await transport.request("info", payload); // billed 2 — the wire's weight: empties the bucket - assertEquals(stub.calls, 1); - - const pending = transport.request("exchange", { action: { type: "noop" } }); // weight 1: waits 1 s - await flush(); - assertEquals(stub.calls, 1); - - time.tick(1_000); - await pending; - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => jsonResponse()); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + rateLimit: { capacity: 2, refillPerMinute: 60 }, + }); + + // The live object says "userRole" (60 weight); the serialized form says "allMids" (2). + const payload = { + get type(): string { + return "userRole"; + }, + toJSON() { + return { type: "allMids" }; + }, + }; + await transport.request("info", payload); // billed 2 — the wire's weight: empties the bucket + assertEquals(stub.calls, 1); + + const pending = transport.request("exchange", { action: { type: "noop" } }); // weight 1: waits 1 s + await flush(); + assertEquals(stub.calls, 1); + + clock.advance(1_000); + await pending; + assertEquals(stub.calls, 2); }); }); @@ -1517,27 +1497,27 @@ describe("HttpTransport", () => { baseWeight: number, surcharge: number, ) { - const stub = stubFetch(() => jsonResponse(Array.from({ length: itemCount }))); - try { - const transport = new HttpTransport({ rateLimit: { capacity: baseWeight, refillPerMinute: 60 } }); - - await transport.request(endpoint, payload); // bucket: 0 after the acquire, -surcharge after the debit - assertEquals(stub.calls, 1); - - const pending = transport.request("exchange", { action: { type: "noop" } }); // deficit: surcharge + 1 - await flush(); - assertEquals(stub.calls, 1); - - time.tick(surcharge * 1_000); // one token still short - await flush(); - assertEquals(stub.calls, 1); - - time.tick(1_000); // debt paid off plus the one token needed - await pending; - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => jsonResponse(Array.from({ length: itemCount }))); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + rateLimit: { capacity: baseWeight, refillPerMinute: 60 }, + }); + + await transport.request(endpoint, payload); // bucket: 0 after the acquire, -surcharge after the debit + assertEquals(stub.calls, 1); + + const pending = transport.request("exchange", { action: { type: "noop" } }); // deficit: surcharge + 1 + await flush(); + assertEquals(stub.calls, 1); + + clock.advance(surcharge * 1_000); // one token still short + await flush(); + assertEquals(stub.calls, 1); + + clock.advance(1_000); // debt paid off plus the one token needed + await pending; + assertEquals(stub.calls, 2); } test("userFills: 1 extra weight per 20 returned items", async () => { @@ -1559,76 +1539,77 @@ describe("HttpTransport", () => { describe("abort", () => { test("a pre-aborted signal rejects instantly without consuming weight", async () => { - const stub = stubFetch(() => jsonResponse()); - try { - const transport = new HttpTransport({ rateLimit: { capacity: 1, refillPerMinute: 60 } }); - const reason = new Error("cancelled before sending"); - - const error = await assertRejects( - () => transport.request("exchange", { action: { type: "noop" } }, AbortSignal.abort(reason)), - HttpRequestError, - "Request aborted", - ); - assertEquals(error.cause, reason); - assertEquals(stub.calls, 0); // never reached fetch - - // The token was not consumed: the next request proceeds without waiting. - await transport.request("exchange", { action: { type: "noop" } }); - assertEquals(stub.calls, 1); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => jsonResponse()); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + rateLimit: { capacity: 1, refillPerMinute: 60 }, + }); + const reason = new Error("cancelled before sending"); + + const error = await assertRejects( + () => transport.request("exchange", { action: { type: "noop" } }, AbortSignal.abort(reason)), + HttpRequestError, + "Request aborted", + ); + assertEquals(error.cause, reason); + assertEquals(stub.calls, 0); // never reached fetch + + // The token was not consumed: the next request proceeds without waiting. + await transport.request("exchange", { action: { type: "noop" } }); + assertEquals(stub.calls, 1); }); test("aborting during a queued wait rejects without consuming weight", async () => { - const stub = stubFetch(() => jsonResponse()); - try { - const transport = new HttpTransport({ rateLimit: { capacity: 1, refillPerMinute: 60 } }); - - await transport.request("exchange", { action: { type: "noop" } }); // consumes the only token - assertEquals(stub.calls, 1); - - const controller = new AbortController(); - const pending = transport.request("exchange", { action: { type: "noop" } }, controller.signal); - await flush(); // queued in the bucket, no fetch yet - assertEquals(stub.calls, 1); - - controller.abort(new DOMException("user cancel", "AbortError")); - const error = await assertRejects(() => pending, HttpRequestError, "Request aborted"); - assertIsError(error.cause, DOMException); - assertEquals(stub.calls, 1); // never reached fetch - - // The aborted request consumed nothing: the survivor is served on the normal refill. - const followUp = transport.request("exchange", { action: { type: "noop" } }); - await flush(); - assertEquals(stub.calls, 1); - time.tick(1_000); - await followUp; - assertEquals(stub.calls, 2); - } finally { - stub.restore(); - } + const stub = fakeFetch(() => jsonResponse()); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + rateLimit: { capacity: 1, refillPerMinute: 60 }, + }); + + await transport.request("exchange", { action: { type: "noop" } }); // consumes the only token + assertEquals(stub.calls, 1); + + const controller = new AbortController(); + const pending = transport.request("exchange", { action: { type: "noop" } }, controller.signal); + await flush(); // queued in the bucket, no fetch yet + assertEquals(stub.calls, 1); + + controller.abort(new DOMException("user cancel", "AbortError")); + const error = await assertRejects(() => pending, HttpRequestError, "Request aborted"); + assertIsError(error.cause, DOMException); + assertEquals(stub.calls, 1); // never reached fetch + + // The aborted request consumed nothing: the survivor is served on the normal refill. + const followUp = transport.request("exchange", { action: { type: "noop" } }); + await flush(); + assertEquals(stub.calls, 1); + clock.advance(1_000); + await followUp; + assertEquals(stub.calls, 2); }); test("the timeout still fires once the request is in flight", async () => { - const stub = stubFetch( + const stub = fakeFetch( (_req, init) => new Promise((_resolve, reject) => { init?.signal?.addEventListener("abort", () => reject(init.signal!.reason)); }), ); - try { - const transport = new HttpTransport({ timeout: 5, rateLimit: { capacity: 1, refillPerMinute: 60 } }); + const transport = new HttpTransport({ + runtime: clock, + fetch: stub.fetch, + timeout: 5, + rateLimit: { capacity: 1, refillPerMinute: 60 }, + }); - const pending = transport.request("exchange", { action: { type: "noop" } }); - await flush(); // bucket full: acquired and sent, now hanging in fetch - assertEquals(stub.calls, 1); + const pending = transport.request("exchange", { action: { type: "noop" } }); + await flush(); // bucket full: acquired and sent, now hanging in fetch + assertEquals(stub.calls, 1); - time.tick(5); // fires the request timeout armed after the acquire - await assertRejects(() => pending, HttpRequestError, "Request timed out after 5 ms"); - } finally { - stub.restore(); - } + clock.advance(5); // fires the request timeout armed after the acquire + await assertRejects(() => pending, HttpRequestError, "Request timed out after 5 ms"); }); }); }); diff --git a/packages/hyperliquid/tests/transport/websocket/_connectionQuota.test.ts b/packages/hyperliquid/tests/transport/websocket/_connectionQuota.test.ts index c4737c6e..60ab908d 100644 --- a/packages/hyperliquid/tests/transport/websocket/_connectionQuota.test.ts +++ b/packages/hyperliquid/tests/transport/websocket/_connectionQuota.test.ts @@ -1,6 +1,6 @@ import { afterEach, beforeEach, expect, test } from "bun:test"; -import { FakeTime } from "@jsr/std__testing/time"; -import { WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; +import { type WebSocketQuotaOptions, WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; +import { FakeRuntime } from "../../_fakeRuntime.ts"; import { RealCloseEvent } from "../../_noCloseEvent.ts"; class Socket extends EventTarget { @@ -27,7 +27,8 @@ class Socket extends EventTarget { } } const original = globalThis.WebSocket; -let time: FakeTime; +// Sockets come from an injected factory and attempt windows run on a virtual clock: nothing global is patched. +let clock: FakeRuntime; const transports: WebSocketTransport[] = []; async function settle(): Promise { for (let i = 0; i < 8; i++) await Promise.resolve(); @@ -37,24 +38,27 @@ function transport(quota: WebSocketQuota): WebSocketTransport { url: "wss://test.local/ws", quota, reconnect: { connectionTimeout: null, reconnectionDelay: 0 }, + runtime: clock, + webSocketFactory: (url) => new Socket(url) as unknown as WebSocket, }); transports.push(result); return result; } +function sharedQuota(options: WebSocketQuotaOptions): WebSocketQuota { + return new WebSocketQuota({ ...options, runtime: clock }); +} beforeEach(() => { - time = new FakeTime(); + clock = new FakeRuntime(); Socket.instances = []; Socket.fail = false; - globalThis.WebSocket = Socket as unknown as typeof WebSocket; }); afterEach(() => { for (const t of transports.splice(0)) t.close(); for (const socket of Socket.instances) socket.finishClose(); - time.restore(); - globalThis.WebSocket = original; + expect(globalThis.WebSocket).toBe(original); }); test("shared transports queue behind the active cap, including closing sockets", async () => { - const quota = new WebSocketQuota({ maxConnections: 1 }); + const quota = sharedQuota({ maxConnections: 1 }); const first = transport(quota); transport(quota); expect(Socket.instances.length).toBe(1); @@ -69,26 +73,26 @@ test("shared transports queue behind the active cap, including closing sockets", expect(quota.connections).toBe(1); }); test("initial attempts and retry storms share a strict rolling minute", async () => { - const quota = new WebSocketQuota({ maxConnections: 10, maxConnectionAttemptsPerMinute: 3 }); + const quota = sharedQuota({ maxConnections: 10, maxConnectionAttemptsPerMinute: 3 }); transport(quota); transport(quota); transport(quota); transport(quota); expect(Socket.instances.length).toBe(3); for (const socket of [...Socket.instances]) socket.finishClose(); - time.tick(0); + clock.advance(0); await settle(); expect(quota.connections).toBe(0); - time.tick(59_999); + clock.advance(59_999); await settle(); expect(Socket.instances.length).toBe(3); - time.tick(1); + clock.advance(1); await settle(); expect(Socket.instances.length).toBe(6); expect(quota.connections).toBe(3); }); test("cancelling queued transports never constructs a socket or spends an attempt", async () => { - const quota = new WebSocketQuota({ maxConnections: 1, maxConnectionAttemptsPerMinute: 2 }); + const quota = sharedQuota({ maxConnections: 1, maxConnectionAttemptsPerMinute: 2 }); const first = transport(quota); const waiting = transport(quota); waiting.close(); @@ -101,7 +105,7 @@ test("cancelling queued transports never constructs a socket or spends an attemp expect(Socket.instances.length).toBe(2); }); test("superseding a queued reconnect cancels its old waiter", async () => { - const quota = new WebSocketQuota({ maxConnections: 1 }); + const quota = sharedQuota({ maxConnections: 1 }); const first = transport(quota); const waiting = transport(quota); waiting.socket.reconnect(); @@ -113,21 +117,21 @@ test("superseding a queued reconnect cancels its old waiter", async () => { expect(quota.connections).toBe(1); }); test("construction failures release active reservations but count against the attempt window", async () => { - const quota = new WebSocketQuota({ maxConnections: 1, maxConnectionAttemptsPerMinute: 1 }); + const quota = sharedQuota({ maxConnections: 1, maxConnectionAttemptsPerMinute: 1 }); Socket.fail = true; transport(quota); expect(quota.connections).toBe(0); Socket.fail = false; - time.tick(0); + clock.advance(0); await settle(); expect(Socket.instances.length).toBe(0); - time.tick(60_000); + clock.advance(60_000); await settle(); expect(Socket.instances.length).toBe(1); expect(quota.connections).toBe(1); }); test("connection guards can be disabled and reservations release idempotently", async () => { - const quota = new WebSocketQuota({ maxConnections: null, maxConnectionAttemptsPerMinute: null }); + const quota = sharedQuota({ maxConnections: null, maxConnectionAttemptsPerMinute: null }); for (let i = 0; i < 40; i++) transport(quota); expect(Socket.instances.length).toBe(40); const release = await quota.acquireConnection(); @@ -137,5 +141,5 @@ test("connection guards can be disabled and reservations release idempotently", const abort = new AbortController(); abort.abort("cancel"); expect(() => quota.acquireConnection(abort.signal)).toThrow(); - expect(() => new WebSocketQuota({ maxConnections: 0 })).toThrow(TypeError); + expect(() => sharedQuota({ maxConnections: 0 })).toThrow(TypeError); }); diff --git a/packages/hyperliquid/tests/transport/websocket/_keepAlive.test.ts b/packages/hyperliquid/tests/transport/websocket/_keepAlive.test.ts index 3f3e6d2e..e31190af 100644 --- a/packages/hyperliquid/tests/transport/websocket/_keepAlive.test.ts +++ b/packages/hyperliquid/tests/transport/websocket/_keepAlive.test.ts @@ -4,101 +4,98 @@ * @module */ -import { afterEach, beforeEach, describe, test } from "bun:test"; +import { beforeEach, describe, test } from "bun:test"; import { assertEquals } from "@jsr/std__assert"; -import { FakeTime } from "@jsr/std__testing/time"; import type { ReconnectingWebSocket } from "../../../src/transport/websocket/_reconnectingSocket.ts"; import { WebSocketKeepAlive, type WebSocketKeepAliveOptions } from "../../../src/transport/websocket/_keepAlive.ts"; import { HyperliquidEventTarget } from "../../../src/transport/websocket/_events.ts"; +import { FakeRuntime } from "../../_fakeRuntime.ts"; import { getLastSent, MockWebSocket } from "./_mock.ts"; -/** Creates a keep-alive watchdog over a mock socket. */ -function createKeepAlive(options?: WebSocketKeepAliveOptions): { socket: MockWebSocket } { +/** Creates a keep-alive watchdog over a mock socket, scheduling on the given virtual clock. */ +function createKeepAlive(clock: FakeRuntime, options?: WebSocketKeepAliveOptions): { socket: MockWebSocket } { const socket = new MockWebSocket() as ReconnectingWebSocket & MockWebSocket; const hlEvents = new HyperliquidEventTarget(socket); - new WebSocketKeepAlive(socket, hlEvents, options); + new WebSocketKeepAlive(socket, hlEvents, options, undefined, clock); return { socket }; } describe("WebSocketKeepAlive", () => { - // The npm build of `@std/testing/time` drops the `[Symbol.dispose]` member the Deno version - // declares, so the clock is installed and restored through hooks instead of a `using` binding. - let time: FakeTime; + // Ping interval and pong deadline run on an injected virtual clock; no global timer is patched. + let clock: FakeRuntime; beforeEach(() => { - time = new FakeTime(); - }); - - afterEach(() => { - time.restore(); + clock = new FakeRuntime(); }); test("reconnects when a ping stays unanswered", () => { - const { socket } = createKeepAlive(); + const { socket } = createKeepAlive(clock); socket.open(); - time.tick(5_000); + clock.advance(5_000); assertEquals(getLastSent(socket).method, "ping"); - time.tick(3_000); + clock.advance(3_000); assertEquals(socket.reconnectCalls, 1); }); test("a pong in time keeps the connection", () => { - const { socket } = createKeepAlive(); + const { socket } = createKeepAlive(clock); socket.open(); - time.tick(5_000); + clock.advance(5_000); socket.mockMessage({ channel: "pong" }); - time.tick(3_000); + clock.advance(3_000); assertEquals(socket.reconnectCalls, 0); }); test("disconnect clears the watchdog", () => { - const { socket } = createKeepAlive(); + const { socket } = createKeepAlive(clock); socket.open(); - time.tick(5_000); + clock.advance(5_000); socket.disconnect(); + assertEquals(clock.pendingTimers, 0); // interval and pong deadline both released const sentBeforeTick = socket.sentMessages.length; - time.tick(60_000); + clock.advance(60_000); assertEquals(socket.reconnectCalls, 0); assertEquals(socket.sentMessages.length, sentBeforeTick); }); test("a socket error also clears the watchdog", () => { - const { socket } = createKeepAlive(); + const { socket } = createKeepAlive(clock); socket.open(); - time.tick(5_000); + clock.advance(5_000); // The error listener is a distinct arrow from the close handler — both call `_stop`. socket.dispatchEvent(new Event("error")); + assertEquals(clock.pendingTimers, 0); // interval and pong deadline both released const sentBeforeTick = socket.sentMessages.length; - time.tick(60_000); + clock.advance(60_000); assertEquals(socket.reconnectCalls, 0); assertEquals(socket.sentMessages.length, sentBeforeTick); }); test("a second open while the interval is armed is a no-op", () => { - const { socket } = createKeepAlive(); + const { socket } = createKeepAlive(clock); socket.open(); // Re-fire open: `_start` early-returns when the interval is already set. socket.dispatchEvent(new Event("open")); - time.tick(5_000); + clock.advance(5_000); assertEquals(getLastSent(socket).method, "ping"); }); test("honors custom interval and timeout", () => { - const { socket } = createKeepAlive({ interval: 5_000, timeout: 1_000 }); + const { socket } = createKeepAlive(clock, { interval: 5_000, timeout: 1_000 }); socket.open(); - time.tick(5_000); + clock.advance(5_000); assertEquals(getLastSent(socket).method, "ping"); - time.tick(1_000); + clock.advance(1_000); assertEquals(socket.reconnectCalls, 1); }); }); diff --git a/packages/hyperliquid/tests/transport/websocket/_reconnectingSocket.test.ts b/packages/hyperliquid/tests/transport/websocket/_reconnectingSocket.test.ts index 794ae009..09a33d5d 100644 --- a/packages/hyperliquid/tests/transport/websocket/_reconnectingSocket.test.ts +++ b/packages/hyperliquid/tests/transport/websocket/_reconnectingSocket.test.ts @@ -1,7 +1,8 @@ /** * Tests for the reconnecting WebSocket: retry policy, backoff scheduling, - * offline buffering, and termination semantics, driven by a fake global - * `WebSocket` the tests steer by hand. The test process runs without a global + * offline buffering, and termination semantics, driven by a fake `WebSocket` the + * tests steer by hand: installed over the global for the default-construction path, + * injected through `webSocketFactory` (with a virtual clock) for the timer-driven tests. The test process runs without a global * `CloseEvent` (see `tests/_noCloseEvent.ts`), so every close event the wrapper * dispatches also exercises the module's local `CloseEvent` fallback class. * @module @@ -9,7 +10,6 @@ import { afterEach, beforeEach, describe, test } from "bun:test"; import { assert, assertEquals, assertFalse, assertInstanceOf, assertRejects } from "@jsr/std__assert"; -import { FakeTime } from "@jsr/std__testing/time"; import { ReconnectingWebSocket, ReconnectingWebSocketError, @@ -17,6 +17,7 @@ import { } from "../../../src/transport/websocket/_reconnectingSocket.ts"; import { WebSocketDispatcher, WebSocketRequestError } from "../../../src/transport/websocket/_dispatcher.ts"; import { HyperliquidEventTarget } from "../../../src/transport/websocket/_events.ts"; +import { FakeRuntime } from "../../_fakeRuntime.ts"; import { RealCloseEvent } from "../../_noCloseEvent.ts"; import { drain } from "./_mock.ts"; @@ -108,9 +109,9 @@ function createSocket( return ws; } -/** Sleeps for real milliseconds (no fake timers installed). */ -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); +/** Injected socket factory: the timer-driven tests construct sockets without touching the global. */ +function webSocketFactory(url: string, protocols?: string | string[]): WebSocket { + return new FakeWebSocket(url, protocols) as unknown as WebSocket; } // ============================================================================= @@ -326,36 +327,47 @@ describe("ReconnectingWebSocket", () => { }); test("default delay is exponential backoff with equal jitter, capped at 10s", () => { - const time = new FakeTime(); - try { - const ws = new ReconnectingWebSocket("ws://localhost/ws"); + // random() pinned to 0 and to just under 1 brackets the equal-jitter window exactly. + for (const [random, attempt0, attempt1] of [ + [0, 75, 150], + [0.999, 149.925, 299.85], + ] as const) { + FakeWebSocket.instances = []; + const clock = new FakeRuntime({ random: () => random }); + const ws = new ReconnectingWebSocket("ws://localhost/ws", { runtime: clock, webSocketFactory }); lastSocket().serverClose(); // attempt 0: 2**0 * 150 = 150 → delay in [75, 150) - time.tick(74); + clock.advance(Math.ceil(attempt0) - 1); assertEquals(FakeWebSocket.instances.length, 1); - time.tick(76); + clock.advance(1); assertEquals(FakeWebSocket.instances.length, 2); lastSocket().serverClose(); // attempt 1: 2**1 * 150 = 300 → delay in [150, 300) - time.tick(149); + clock.advance(Math.ceil(attempt1) - 1); assertEquals(FakeWebSocket.instances.length, 2); - time.tick(151); + clock.advance(1); assertEquals(FakeWebSocket.instances.length, 3); ws.close(); - } finally { - time.restore(); + assertEquals(clock.pendingTimers, 0); } }); - test("retry counter resets after the connection stays open for stableTimeout", async () => { - const ws = createSocket("ws://localhost/ws", { stableTimeout: 5 }); + test("retry counter resets after the connection stays open for stableTimeout", () => { + const clock = new FakeRuntime(); + const ws = createSocket("ws://localhost/ws", { stableTimeout: 5, runtime: clock, webSocketFactory }); lastSocket().serverOpen(); - await sleep(10); + clock.advance(4); + lastSocket().serverClose(); + assertEquals(ws.retryCount, 1); // dropped 1 ms early: the streak was never reset + clock.advance(0); // the 0 ms retry reconnects + lastSocket().serverOpen(); + clock.advance(5); lastSocket().serverClose(); assertEquals(ws.retryCount, 1); // first failure of a fresh streak ws.close(); + assertEquals(clock.pendingTimers, 0); }); test("retry counter keeps counting when connections drop before stableTimeout", async () => { @@ -368,23 +380,21 @@ describe("ReconnectingWebSocket", () => { ws.close(); }); - test("connection timeout recycles a stuck handshake", async () => { - const ws = createSocket("ws://localhost/ws", { connectionTimeout: 5 }); + test("connection timeout recycles a stuck handshake", () => { + const clock = new FakeRuntime(); + const ws = createSocket("ws://localhost/ws", { connectionTimeout: 5, runtime: clock, webSocketFactory }); assertEquals(FakeWebSocket.instances.length, 1); - // Every attempt that stays stuck past the timeout fails and is retried. Timer scheduling - // is unbounded on a loaded CI runner, so poll for the recycle (generous deadline) instead - // of asserting after a fixed sleep — what matters is that fresh attempts happen at all. - const deadline = Date.now() + 5_000; - while (FakeWebSocket.instances.length < 2 && Date.now() < deadline) { - await sleep(5); - } - assert(FakeWebSocket.instances.length >= 2); - assert(ws.retryCount >= 1); + clock.advance(4); // still within the handshake budget + assertEquals(FakeWebSocket.instances.length, 1); + clock.advance(1); // the stuck attempt fails; the 0 ms retry is due at the same instant + assertEquals(FakeWebSocket.instances.length, 2); + assertEquals(ws.retryCount, 1); lastSocket().serverOpen(); assertEquals(ws.readyState, ReconnectingWebSocket.OPEN); ws.close(); + assertEquals(clock.pendingTimers, 0); }); test("retries without bound by default (maxRetries: Infinity)", async () => {