From c628ec931163abee24003b6d8af9cc3324e83d6f Mon Sep 17 00:00:00 2001 From: Joe Blau Date: Mon, 3 Aug 2026 21:44:16 -0700 Subject: [PATCH] fix(transport): pace WebSocket messages by default and charge flushed posts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The default WebSocket transport shipped with the 2000 msg/min per-IP budget unenforced: sharedWebSocketQuota() constructed its per-network instance without rateLimit, so acquireSend/chargeSend were no-ops and a reconnect at the 1000-subscription cap re-sent every subscribe frame instantly — half the minute's server budget in one burst, repeated by a flapping socket until the server refused. The shared quota is now created with pacing enabled ({ rateLimit: {} }: capacity 2000, refilling 2000/minute, the server's own budget). To keep acquireSend's load-bearing contract — undefined, never a resolved promise, whenever no wait is needed — TokenBucketRateLimiter gains a synchronous tryAcquire(weight) that deducts inline only when no waiter is queued (the FIFO is never bypassed) and the bucket covers the cost; acquireSend probes it before falling back to the queued acquire. Direct new WebSocketQuota() construction stays accounting-only, which is the documented opt-out via WebSocketTransportOptions.quota. Also fixes a charging gap in the dispatcher: posts queued while disconnected and flushed by the open handler never debited the message budget, since only the send-immediately branch charged them. The flush now debits post entries (numeric id) exactly once, where their frame actually reaches the socket; subscription entries already paid in acquireSend at request() time and are not double-charged. Docs (quota/rate-limit JSDoc and docs/transports.md) updated from "pacing is opt-in" to the new default, with the opt-out spelled out. Fixes #90 Co-Authored-By: Claude Opus 5 (1M context) --- docs/transports.md | 34 +++-- src/transport/_rateLimiter.ts | 22 ++- src/transport/websocket/_dispatcher.ts | 67 +++++++-- src/transport/websocket/_quota.ts | 56 +++++-- src/transport/websocket/mod.ts | 16 +- tests/transport/_rateLimiter.test.ts | 35 +++++ tests/transport/websocket/_quota.test.ts | 181 ++++++++++++++++++++++- 7 files changed, 366 insertions(+), 45 deletions(-) diff --git a/docs/transports.md b/docs/transports.md index 7722e51a..4599bb13 100644 --- a/docs/transports.md +++ b/docs/transports.md @@ -280,31 +280,41 @@ Reservations are released when a subscription is unsubscribed or when its connec when the server frees them too. Call `transport.close()` on a transport you are done with. Pass your own `quota` when the default's assumption does not hold — a process behind several egress IPs needs one per -IP, and tests usually want isolation: +IP, and tests usually want isolation. Pass `rateLimit` to keep the default's [message pacing](#websocket-rate-limiting) +on the replacement; a `WebSocketQuota` constructed without it is accounting-only: ```ts import { WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; -const transport = new WebSocketTransport({ quota: new WebSocketQuota() }); +const transport = new WebSocketTransport({ quota: new WebSocketQuota({ rateLimit: {} }) }); ``` ### WebSocket rate limiting -Outbound messages are budgeted but **not paced by default**. Opt into pacing when you approach the 2000/minute ceiling -— most easily by holding many subscriptions, since one reconnect re-subscribes all of them at once and 1000 -subscriptions spend half the minute's budget instantly: - -```ts -import { WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; - -const quota = new WebSocketQuota({ rateLimit: { capacity: 2000, refillPerMinute: 2000 } }); -const transport = new WebSocketTransport({ quota }); -``` +Outbound messages are **paced by default**: the shared quota runs a token bucket sized to the server's budget (capacity +2000, refilling 2000/minute), because the default transport is exactly the one that trips the limit — one reconnect +re-subscribes every held subscription at once, so 1000 subscriptions spend half the minute's budget instantly, and a +flapping socket repeats the burst until the server refuses. Pacing only ever delays `subscribe` and `unsubscribe` frames. **`post` requests and keep-alive pings never wait**: an exchange action's wire order — and therefore per-wallet nonce ordering — depends on reaching the socket synchronously, and delaying the keep-alive watchdog is how a half-open connection goes unnoticed. Both still *debit* the budget, so a burst of orders correctly slows subscription traffic rather than silently overrunning the shared limit. +To opt out of pacing — or to resize the bucket — pass your own `quota`; constructed without `rateLimit`, it keeps the +subscription and unique-user guards but never delays a frame client-side: + +```ts +import { WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; + +// Accounting only: nothing waits client-side. +const transport = new WebSocketTransport({ quota: new WebSocketQuota() }); + +// Or keep pacing with a custom burst size and refill rate. +const paced = new WebSocketTransport({ + quota: new WebSocketQuota({ rateLimit: { capacity: 1000, refillPerMinute: 2000 } }), +}); +``` + As with [HTTP rate limiting](#rate-limiting), the budget is client-side bookkeeping: other processes, other machines behind the same IP, and traffic the SDK cannot see all draw on the same server-side bucket. diff --git a/src/transport/_rateLimiter.ts b/src/transport/_rateLimiter.ts index cc91fd01..bba62306 100644 --- a/src/transport/_rateLimiter.ts +++ b/src/transport/_rateLimiter.ts @@ -10,7 +10,9 @@ * The WebSocket transport budgets a different quantity against the same machinery — outbound * *messages*, capped at 2000/minute per IP across every connection — through * `WebSocketQuota`. Both callers need the identical FIFO-with-debt semantics, so the - * bucket lives at the transport root rather than under either one. + * bucket lives at the transport root rather than under either one. `WebSocketQuota` + * additionally probes the bucket synchronously ({@linkcode TokenBucketRateLimiter.tryAcquire}) + * so an uncontended send never pays for a promise it did not need. * @module */ @@ -139,6 +141,24 @@ export class TokenBucketRateLimiter { return promise; } + /** + * Deducts `weight` tokens synchronously when that costs no queued acquisition its turn: + * after a refill, the deduction happens only when no waiter is pending and the bucket + * covers the cost. Returns whether the tokens were deducted. + * + * A pending FIFO always wins: even a bucket that covers `weight` refuses while any + * acquisition is queued, because serving the newcomer first would let a later arrival + * overtake an earlier one — exactly the reordering the queue in + * {@linkcode TokenBucketRateLimiter.acquire} exists to prevent. A refused caller falls + * back to `acquire` and takes its place at the tail. + */ + tryAcquire(weight: number): boolean { + this._refill(); + if (this._head !== undefined || this._tokens < weight) return false; + this._tokens -= weight; + return true; + } + /** * Debits `weight` tokens without waiting, possibly driving the bucket into debt. * diff --git a/src/transport/websocket/_dispatcher.ts b/src/transport/websocket/_dispatcher.ts index 594a907b..7359ebfa 100644 --- a/src/transport/websocket/_dispatcher.ts +++ b/src/transport/websocket/_dispatcher.ts @@ -186,10 +186,11 @@ export class WebSocketDispatcher { /** * The per-IP outbound message budget, or `undefined` when this dispatcher is not budgeted. * - * Hyperliquid caps messages sent at 2000/minute per IP across every connection, and the - * SDK can overrun it in one burst: at the 1000-subscription cap a single reconnect - * re-subscribes everything at once, spending half the minute's budget instantly, and a - * socket flapping under `maxRetries: Infinity` repeats that. + * Hyperliquid caps messages sent at 2000/minute per IP across every connection, and + * without pacing the SDK overruns it in one burst: at the 1000-subscription cap a single + * reconnect re-subscribes everything at once, spending half the minute's budget instantly, + * and a socket flapping under `maxRetries: Infinity` repeats that — which is why the + * shared default quota paces outbound messages (see `sharedWebSocketQuota` in `_quota.ts`). */ private readonly _quota: WebSocketQuota | undefined; @@ -234,6 +235,12 @@ export class WebSocketDispatcher { for (const entry of this._queue) { entry.sent = true; this._socket.send(entry.frame); + // Only `post` entries (numeric id) debit the message budget here: their frame is + // reaching the socket for the first time and the send-immediately charge in + // `request` never ran. Subscription entries already paid a token in `acquireSend` + // at request() time; charging them again would double-count every flushed frame. + // See the send-immediately branch in `request` for the full charging story. + if (typeof entry.id === "number") this._quota?.chargeSend(); } }); @@ -277,14 +284,41 @@ export class WebSocketDispatcher { // // `post` is deliberately excluded: `_shell.ts` fixes the wire order of an exchange // action on `transport.request` reaching `send` synchronously, so awaiting here would - // let a later nonce overtake an earlier one. Posts debit the budget without waiting - // (below), which still slows subscription traffic when orders are heavy but can never - // delay or reorder an order. `acquireSend` returns `undefined` rather than a resolved - // promise when nothing needs waiting on, keeping this function synchronous to `send` - // for every request that is not actually being throttled. - if (method !== "post") { - const paced = this._quota?.acquireSend(signal); - if (paced !== undefined) await paced; + // let a later nonce overtake an earlier one. Posts debit the budget without waiting — + // when their frame reaches the socket, in the send-immediately branch below or in the + // `open` flush — which still slows subscription traffic when orders are heavy but can + // never delay or reorder an order. `acquireSend` returns `undefined` rather than a + // resolved promise when nothing needs waiting on, keeping this function synchronous to + // `send` for every request that is not actually being throttled. + if (method !== "post" && this._quota !== undefined) { + // The wait must not outlive the connection: the socket's termination signal abandons + // the wait when the transport closes for good. Otherwise each paced waiter of a + // closed transport would still consume a token for a frame that can never send — and + // hold its FIFO slot ahead of live transports sharing the quota. A caller signal + // composes with it via `AbortSignal.any`, with the caller's reason winning when both + // are already aborted — the same precedence the controller relay below reproduces. + // No in-repo paced caller passes a signal today (the subscription manager never + // does), so the common path allocates no composite. + const pacingSignal = + signal === undefined + ? this._socket.terminationSignal + : AbortSignal.any([signal, this._socket.terminationSignal]); + const paced = this._quota.acquireSend(pacingSignal); + if (paced !== undefined) { + try { + await paced; + } catch (error) { + // Wrapped here because this await runs before the main try — whose catch would + // never see this rejection — and the dispatcher's contract is to reject only + // with WebSocketRequestError. `payload` is safe to attach: only the subscription + // manager reaches this path, and it always hands over its own plain-data snapshot. + const terminated = this._socket.terminationSignal.aborted && error === this._socket.terminationSignal.reason; + throw new WebSocketRequestError( + terminated ? "WebSocket connection permanently terminated" : "Request aborted", + { cause: error, request: payload }, + ); + } + } } // One controller per request: the timeout timer, the user signal, and the @@ -346,9 +380,12 @@ export class WebSocketDispatcher { const sent = this._socket.readyState === ReconnectingWebSocket.OPEN; if (sent) { this._socket.send(frame); - // A `post` never waited above, so it debits the shared budget here instead — - // driving it into debt when orders outpace the refill, which later `subscribe` - // frames wait off. Subscribes already paid in `acquireSend`. + // The charging story, stated once: every frame debits the message budget exactly + // one time. `subscribe`/`unsubscribe` paid a token in `acquireSend` above, whether + // the frame goes out here or from the `open` flush. A `post` never waited there, so + // it debits when its frame reaches the socket — here when connected, in the `open` + // flush when queued while disconnected. Post debits can drive the bucket into debt + // when orders outpace the refill, which later `subscribe` frames wait off. if (method === "post") this._quota?.chargeSend(); } diff --git a/src/transport/websocket/_quota.ts b/src/transport/websocket/_quota.ts index d7ff1918..f0405d70 100644 --- a/src/transport/websocket/_quota.ts +++ b/src/transport/websocket/_quota.ts @@ -20,7 +20,11 @@ * * One instance of this class is shared by every transport pointed at the same deployment * (see {@linkcode sharedWebSocketQuota}), so the counts the guards read are the counts the - * server keeps. + * server keeps. That shared instance also paces outbound messages against the 2000/minute + * budget by default: the default transport is exactly the one that overruns it, since at the + * 1000-subscription cap one reconnect re-sends every subscribe frame instantly and a flapping + * socket repeats the burst. A directly constructed `new WebSocketQuota()` stays + * accounting-only, which is the opt-out (pass it via `WebSocketTransportOptions.quota`). * * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits * @module @@ -94,14 +98,18 @@ export interface WebSocketRateLimitOptions { /** Configuration options for a {@linkcode WebSocketQuota}. */ export interface WebSocketQuotaOptions { /** - * Opt-in pacing of outbound messages against Hyperliquid's 2000-per-minute per-IP budget. + * Pacing of outbound messages against Hyperliquid's 2000-per-minute per-IP budget. * * When set, `subscribe` and `unsubscribe` frames wait for a token before going out. * `post` frames and keep-alive pings never wait — they debit the bucket without blocking * (see {@linkcode WebSocketQuota.chargeSend}) — so enabling this can only ever delay * subscription traffic, never an order. * - * Default: `undefined` (accounting only; nothing waits) + * Default: `undefined` (accounting only; nothing waits). That default governs direct + * construction only — the {@linkcode sharedWebSocketQuota} instance every transport uses + * unless given its own is created with pacing enabled (`{}`: capacity 2000, refilling + * 2000/minute). Opting out of pacing therefore means passing an accounting-only + * `new WebSocketQuota()` via `WebSocketTransportOptions.quota`. */ rateLimit?: WebSocketRateLimitOptions; /** @@ -230,11 +238,21 @@ export class WebSocketQuota { /** * Waits until the outbound budget covers one message, then deducts it. * - * Returns `undefined` — not a resolved promise — whenever the wait is unnecessary, so the - * caller can skip the `await` entirely and stay synchronous. That distinction is load - * bearing: `_shell.ts` fixes the wire order of an exchange action on `transport.request` - * running synchronously up to its first await, so a promise handed back here where none - * was needed would let a later nonce overtake an earlier one. + * Returns `undefined` — not a resolved promise — whenever the wait is unnecessary: when + * pacing is off, and on the {@linkcode TokenBucketRateLimiter.tryAcquire} fast path when + * the bucket covers the message with no queue in front of it. The caller can then skip + * the `await` entirely and stay synchronous. That distinction is load bearing: + * `_shell.ts` fixes the wire order of an exchange action on `transport.request` running + * synchronously up to its first await, so a promise handed back here where none was + * needed would let a later nonce overtake an earlier one — and with pacing on by default + * (see {@linkcode sharedWebSocketQuota}), a plain `acquire` would hand back exactly such + * a promise for every uncontended send. + * + * "Unnecessary" is judged by the synchronous probe: in the sliver between a failed probe + * and the queued `acquire`, the clock can cross a refill boundary and hand back an + * already-resolved promise instead of `undefined`. That costs the subscribe frame one + * microtask and nothing else — a `post` never enters this path, so no order can be + * delayed or reordered by it. * * Only `subscribe` / `unsubscribe` frames go through this path. See * {@linkcode WebSocketQuota.chargeSend} for the frames that must never wait. @@ -243,6 +261,15 @@ export class WebSocketQuota { */ acquireSend(signal?: AbortSignal): Promise | undefined { if (this._limiter === null) return undefined; + // An already-aborted caller must not spend budget: its frame will never be sent, and + // repeated aborted attempts would drain the shared bucket and throttle the legitimate + // subscription traffic behind it. Mirrors `acquire`'s own already-aborted contract, + // which the synchronous probe below would otherwise bypass. + if (signal?.aborted) return Promise.reject(signal.reason); + // The synchronous probe deducts the token inline whenever it can do so without stealing + // a queued waiter's turn; only a genuinely empty bucket — or one with a queue in front + // of it — costs the caller a promise. + if (this._limiter.tryAcquire(1)) return undefined; return this._limiter.acquire(1, signal); } @@ -279,8 +306,17 @@ export class WebSocketQuota { */ const SHARED: { mainnet?: WebSocketQuota; testnet?: WebSocketQuota } = {}; -/** The quota every transport on `isTestnet` shares unless it was given its own. */ +/** + * The quota every transport on `isTestnet` shares unless it was given its own. + * + * Created with pacing enabled ({@linkcode WebSocketQuotaOptions.rateLimit} `{}`: capacity + * 2000, refilling 2000/minute — the server's own budget), because the default transport is + * exactly the one that trips the limit: at the 1000-subscription cap a single reconnect + * re-sends every subscribe frame instantly, spending half the minute's budget in one burst, + * and a socket flapping under `maxRetries: Infinity` repeats it. To opt out of pacing, pass + * an accounting-only `new WebSocketQuota()` via `WebSocketTransportOptions.quota`. + */ export function sharedWebSocketQuota(isTestnet: boolean): WebSocketQuota { const key = isTestnet ? "testnet" : "mainnet"; - return (SHARED[key] ??= new WebSocketQuota()); + return (SHARED[key] ??= new WebSocketQuota({ rateLimit: {} })); } diff --git a/src/transport/websocket/mod.ts b/src/transport/websocket/mod.ts index 7488bb7f..761cbf2a 100644 --- a/src/transport/websocket/mod.ts +++ b/src/transport/websocket/mod.ts @@ -78,15 +78,21 @@ export interface WebSocketTransportOptions { * * Hyperliquid scopes every WebSocket limit to the client IP rather than to the connection, * so by default all transports on the same network share one {@linkcode WebSocketQuota} and - * the guards count what the server counts. Pass an instance to override that — a process - * behind several egress IPs needs one quota per IP, and a test usually wants isolation. + * the guards count what the server counts. The shared default also paces outbound messages + * against the documented 2000/minute budget: `subscribe`/`unsubscribe` frames wait for a + * token when the budget is spent, while `post` frames and keep-alive pings only ever debit + * it, so pacing can delay subscription traffic but never an order. * - * @example Pace outbound messages against the documented 2000/minute budget + * Pass an instance to override the default — a process behind several egress IPs needs one + * quota per IP, a test usually wants isolation, and a directly constructed + * `new WebSocketQuota()` (no `rateLimit`) is the opt-out from pacing: it keeps the + * subscription and unique-user guards but never delays a frame client-side. + * + * @example Opt out of outbound message pacing (accounting only; nothing waits) * ```ts * import { WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; * - * const quota = new WebSocketQuota({ rateLimit: { capacity: 2000, refillPerMinute: 2000 } }); - * const transport = new WebSocketTransport({ quota }); + * const transport = new WebSocketTransport({ quota: new WebSocketQuota() }); * ``` * * Default: {@linkcode sharedWebSocketQuota} for this network diff --git a/tests/transport/_rateLimiter.test.ts b/tests/transport/_rateLimiter.test.ts index 4f8780f3..317e5591 100644 --- a/tests/transport/_rateLimiter.test.ts +++ b/tests/transport/_rateLimiter.test.ts @@ -292,6 +292,41 @@ describe("TokenBucketRateLimiter", () => { assert(next.resolved); }); + test("tryAcquire deducts synchronously while the bucket covers the weight", () => { + const bucket = new TokenBucketRateLimiter(2, 60); + + assert(bucket.tryAcquire(1)); + assert(bucket.tryAcquire(1)); + // The bucket is now empty: the probe refuses without queueing anything. + 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 + 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 + + 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 + // 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 + await flush(); + assert(large.resolved); + + // With the queue drained, the same probe succeeds again. + time.tick(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"); diff --git a/tests/transport/websocket/_quota.test.ts b/tests/transport/websocket/_quota.test.ts index 69bb670d..586ab1c4 100644 --- a/tests/transport/websocket/_quota.test.ts +++ b/tests/transport/websocket/_quota.test.ts @@ -1,11 +1,13 @@ /** * Tests for the per-IP WebSocket budget: subscription and unique-user reservations shared - * across connections, and outbound message pacing that must never delay or reorder a `post`. + * across connections, outbound message pacing that must never delay or reorder a `post`, + * and the reconnect flush's exactly-once charging of held-back frames. * @module */ import { describe, test } from "bun:test"; import { assert, assertEquals, assertRejects } from "@jsr/std__assert"; +import { TokenBucketRateLimiter } from "../../../src/transport/_rateLimiter.ts"; import type { ReconnectingWebSocket } from "../../../src/transport/websocket/_reconnectingSocket.ts"; import { WebSocketDispatcher, WebSocketRequestError } from "../../../src/transport/websocket/_dispatcher.ts"; import { HyperliquidEventTarget } from "../../../src/transport/websocket/_events.ts"; @@ -190,13 +192,115 @@ describe("WebSocketQuota", () => { }); describe("outbound message budget", () => { - test("accounting-only by default: nothing waits", () => { + test("direct construction is accounting-only: nothing waits", () => { + // The pacing default belongs to `sharedWebSocketQuota` alone; a directly constructed + // quota without `rateLimit` is the documented opt-out and must never delay a frame. const quota = new WebSocketQuota(); // `undefined` rather than a resolved promise, so the caller can stay synchronous. assertEquals(quota.acquireSend(), undefined); quota.chargeSend(); }); + test("pacing on: acquireSend stays synchronous until the bucket drains", async () => { + // `rateLimit: {}` is exactly how the shared default enables pacing (there it means + // capacity 2000, refilling 2000/minute); a tiny bucket with a glacial refill keeps + // the drain cheap and the token arithmetic immune to elapsed wall time. + const quota = new WebSocketQuota({ rateLimit: { capacity: 2, refillPerMinute: 1 } }); + + // Sync fast path: the bucket covers the message, so no promise is handed back — a + // resolved one would cost every uncontended subscribe its synchronicity, the + // load-bearing property documented on `acquireSend`. + assertEquals(quota.acquireSend(), undefined); + + // Drain the last token the way a post would. + quota.chargeSend(); + + // Now the wait is real: a promise comes back and the caller must await it. + const controller = new AbortController(); + const paced = quota.acquireSend(controller.signal); + assert(paced instanceof Promise); + // Abandon the wait rather than letting a refill timer resolve it after the test. + controller.abort(new Error("drained")); + await assertRejects(() => paced, Error, "drained"); + }); + + test("an already-aborted signal rejects without spending a token", async () => { + // Capacity 1 with a glacial refill: if the aborted call below spent the only token, + // the follow-up probe could not stay synchronous. The synchronous fast path must not + // bypass `acquire`'s already-aborted contract — an aborted request's frame is never + // sent, so budget spent on it would only throttle the legitimate traffic behind it. + const quota = new WebSocketQuota({ rateLimit: { capacity: 1, refillPerMinute: 1 } }); + const aborted = AbortSignal.abort(new Error("gone before pacing")); + + const paced = quota.acquireSend(aborted); + assert(paced instanceof Promise); + await assertRejects(() => paced, Error, "gone before pacing"); + + // The token is still there: the fast path answers synchronously. + assertEquals(quota.acquireSend(), undefined); + }); + + test("terminating the connection abandons a paced wait and frees its queue slot", async () => { + const quota = new WebSocketQuota({ rateLimit: { capacity: 1, refillPerMinute: 1 } }); + const { socket, manager } = createManager(quota); + + await subscribeConfirmed(socket, manager, { channel: "one" }); // spends the only token + assertEquals(socket.sentMessages.length, 1); + + // Parked in the limiter FIFO: the bucket is empty and the refill is glacial. A closed + // transport must not leave this waiter behind — it would eventually spend a token on a + // frame that can never send, and hold its FIFO slot ahead of live transports sharing + // the quota. + const parked = manager.subscribe("test", { channel: "two" }, () => {}); + parked.catch(() => {}); + assertEquals(socket.sentMessages.length, 1); // never reached the socket + + socket.terminate(); + await assertRejects(() => parked, WebSocketRequestError, "permanently terminated"); + + // The waiter left the FIFO without spending: no head remains to block other callers. + const limiter = (quota as unknown as { _limiter: { _head?: unknown } })._limiter; + assertEquals(limiter._head, undefined); + }); + + test("termination is observed even while a live caller signal rides the paced wait", async () => { + // A caller-supplied signal must compose with — not replace — the termination signal: + // were it the only abort source, a terminated transport's paced request would sit in + // the FIFO until the caller aborted or a refill granted it a token for a dead frame. + const quota = new WebSocketQuota({ rateLimit: { capacity: 1, refillPerMinute: 1 } }); + const socket = new MockWebSocket() as ReconnectingWebSocket & MockWebSocket; + const hlEvents = new HyperliquidEventTarget(socket); + const dispatcher = new WebSocketDispatcher(socket, hlEvents, 10_000, quota); + + quota.chargeSend(); // drain the only token, so the request below parks + const caller = new AbortController(); // live for the whole test, never aborted + const parked = dispatcher.request("subscribe", { channel: "late" }, caller.signal); + parked.catch(() => {}); + + socket.terminate(); + await assertRejects(() => parked, WebSocketRequestError, "permanently terminated"); + + // The waiter left the FIFO without spending its token. + const limiter = (quota as unknown as { _limiter: { _head?: unknown } })._limiter; + assertEquals(limiter._head, undefined); + }); + + test("the shared default quota paces outbound messages", () => { + const testnet = sharedWebSocketQuota(true); + + // One synchronous probe, spending a single token of 2000: the fast path holds on the + // shared instance without disturbing the process-wide budget other tests draw on + // (every default-quota transport in this run shares these instances). + assertEquals(testnet.acquireSend(), undefined); + + // Proving the limiter exists *behaviourally* would mean draining 2000 tokens of + // process-wide state and throttling every offline test still to run, so pacing is + // asserted structurally here; the drain behaviour is covered above on isolated + // instances built with the same `rateLimit` shape. + const limiter = (testnet as unknown as { _limiter: TokenBucketRateLimiter | null })._limiter; + assert(limiter instanceof TokenBucketRateLimiter); + }); + test("subscribe waits once the bucket is empty", async () => { // Capacity 1 refilling at 60/minute: the second subscribe cannot go out for ~1 s. const quota = new WebSocketQuota({ rateLimit: { capacity: 1, refillPerMinute: 60 } }); @@ -291,4 +395,77 @@ describe("WebSocketQuota", () => { assertEquals(quota.subscriptions, 1); }); }); + + describe("reconnect flush charging", () => { + /** A dispatcher over a mock socket that has already lost its connection. */ + function createDisconnectedDispatcher(quota: WebSocketQuota): { + socket: MockWebSocket; + dispatcher: WebSocketDispatcher; + } { + const socket = new MockWebSocket() as ReconnectingWebSocket & MockWebSocket; + const hlEvents = new HyperliquidEventTarget(socket); + const dispatcher = new WebSocketDispatcher(socket, hlEvents, 10_000, quota); + socket.disconnect(); // the drop happens before any request below is made + return { socket, dispatcher }; + } + + /** + * Asserts the bucket holds exactly `tokens` more whole tokens, by probing: that many + * synchronous acquires must succeed and the next must park. The glacial refills the + * tests below configure keep the count a step function of the charges made. + */ + async function assertRemainingTokens(quota: WebSocketQuota, tokens: number): Promise { + for (let i = 0; i < tokens; i++) assertEquals(quota.acquireSend(), undefined); + const controller = new AbortController(); + const parked = quota.acquireSend(controller.signal); + assert(parked instanceof Promise); + // Abandon the wait rather than letting a refill timer resolve it after the test. + controller.abort(new Error("probe done")); + await assertRejects(() => parked, Error, "probe done"); + } + + test("a post queued while disconnected debits the budget exactly once when flushed", async () => { + const quota = new WebSocketQuota({ rateLimit: { capacity: 3, refillPerMinute: 1 } }); + const { socket, dispatcher } = createDisconnectedDispatcher(quota); + + const post = dispatcher.request("post", { type: "test" }); + post.catch(() => {}); + // Held back while disconnected: nothing on the wire, and — posts pay only when their + // frame reaches the socket — nothing charged yet. + assertEquals(socket.sentMessages.length, 0); + + socket.open(); // reconnect: the `open` flush sends the held-back frame + assertEquals(socket.sentMessages.length, 1); + + // Exactly one token of three is spent. A double charge would leave one; a missed + // charge (the original bug: the flush skipped the debit entirely) would leave three. + await assertRemainingTokens(quota, 2); + + // Settle the flushed post so its request timeout does not outlive the test. + socket.mockMessage(RESPONSES.info(1, "ok")); + assertEquals(await post, "ok"); + }); + + test("a subscribe flushed on open is not charged again — it paid at request() time", async () => { + const quota = new WebSocketQuota({ rateLimit: { capacity: 2, refillPerMinute: 1 } }); + const { socket, dispatcher } = createDisconnectedDispatcher(quota); + + // The token is acquired synchronously in `acquireSend` at request() time, even + // though the frame itself is held back until the connection returns. + const pending = dispatcher.request("subscribe", { channel: "one" }); + pending.catch(() => {}); + assertEquals(socket.sentMessages.length, 0); + + socket.open(); // reconnect: the `open` flush sends the held-back frame + assertEquals(socket.sentMessages.length, 1); + + // Still exactly one token of two spent: the flush charges only posts, so the + // subscribe was not double-charged for reaching the socket late. + await assertRemainingTokens(quota, 1); + + // Settle the flushed subscribe. + socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", { channel: "one" })); + await pending; + }); + }); });