diff --git a/.dev/import_graph_check.ts b/.dev/import_graph_check.ts new file mode 100644 index 00000000..5a97a29c --- /dev/null +++ b/.dev/import_graph_check.ts @@ -0,0 +1,136 @@ +/** + * Import Graph Checker + * + * Budgets the number of modules each public entry point pulls in at RUNTIME, so a barrel import + * cannot quietly reattach a large subgraph to a small entry point again. + * + * This exists because of a measured regression: one value import of `../api/info/mod.ts` in + * `src/utils/_symbolConverter.ts` — four functions from a barrel that re-exports every Info + * method — made `@bloxwap/hyperliquid/utils` load **91 modules instead of 10**, costing 22.7 ms + * on Node and 6.9 ms on Bun per process. Nothing in the test suite, the type gates or the export + * gate noticed, because the import is perfectly valid and the package still behaves identically. + * Only a budget catches it. + * + * Runtime edges are resolved syntactically, which is exact here because `verbatimModuleSyntax` + * is on: a whole-statement `import type` / `export type` is erased and carries no runtime edge, + * and anything else does — including `import { foo, type Bar }`, which keeps the statement. That + * makes this check cheap enough to run on every `bun run check`, with no build step. + * + * Usage: bun run .dev/import_graph_check.ts + * + * @module + */ + +import { readFileSync } from "node:fs"; +import { dirname, relative, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import process from "node:process"; +import ts from "typescript"; + +// ============================================================================= +// CONFIGURATION +// ============================================================================= + +/** Repository root, derived from this file's location so the script runs from any directory. */ +const ROOT_DIR: string = resolve(fileURLToPath(import.meta.url), "../.."); + +/** + * Runtime module budget per entry point. + * + * `limit` is the ceiling, set a little above the measured closure so ordinary growth does not + * trip it. `why` explains what the budget protects, and is printed on failure — a budget whose + * rationale is not obvious gets raised by the next person who trips it. + */ +const BUDGETS: readonly { entry: string; limit: number; why: string }[] = [ + { + entry: "src/utils/mod.ts", + limit: 20, + why: "Formatting and symbol helpers must not drag in the Info API surface. Importing four functions from `api/info/mod.ts` instead of their `_methods/*` modules took this from 10 to 91 and cost 22.7 ms on Node.", + }, + { + entry: "src/transport/mod.ts", + limit: 30, + why: "The transports are the entry point for a consumer that wants no API surface at all.", + }, + { + entry: "src/signing/mod.ts", + limit: 20, + why: "The signing helpers stand alone; they must not reach into the API or transport layers.", + }, + { + entry: "src/api/info/client.ts", + limit: 110, + why: "The narrow read-only entry point. It legitimately pulls the Info methods it wraps, but must not also pull the exchange, subscription or signing graphs.", + }, +]; + +// ============================================================================= +// GRAPH +// ============================================================================= + +/** + * Module specifiers of `file` that survive compilation. + * + * Skips whole-statement `import type` / `export type` — erased under `verbatimModuleSyntax` — + * and keeps everything else, including a mixed `import { foo, type Bar }` whose statement is + * emitted. Bare specifiers (`valibot`, `@noble/hashes`) are ignored: the budget is about this + * package's own graph, and a dependency's internals are not ours to police. + */ +function runtimeEdges(file: string): string[] { + const source = ts.createSourceFile(file, readFileSync(file, "utf8"), ts.ScriptTarget.ESNext, true); + const out: string[] = []; + for (const statement of source.statements) { + let specifier: ts.Expression | undefined; + if (ts.isImportDeclaration(statement)) { + if (statement.importClause?.isTypeOnly) continue; + specifier = statement.moduleSpecifier; + } else if (ts.isExportDeclaration(statement)) { + if (statement.isTypeOnly || statement.moduleSpecifier === undefined) continue; + specifier = statement.moduleSpecifier; + } + if (specifier === undefined || !ts.isStringLiteral(specifier)) continue; + if (!specifier.text.startsWith(".")) continue; + out.push(specifier.text); + } + return out; +} + +/** Every module reachable from `entry` through runtime edges, including `entry` itself. */ +function closure(entry: string): Set { + const seen = new Set(); + const stack = [entry]; + while (stack.length > 0) { + const file = stack.pop()!; + if (seen.has(file)) continue; + seen.add(file); + for (const specifier of runtimeEdges(file)) { + stack.push(resolve(dirname(file), specifier)); + } + } + return seen; +} + +// ============================================================================= +// MAIN +// ============================================================================= + +let failed = false; +for (const { entry, limit, why } of BUDGETS) { + const size = closure(resolve(ROOT_DIR, entry)).size; + const status = size <= limit ? "ok" : "OVER"; + console.log(`${status.padEnd(5)} ${entry.padEnd(28)} ${String(size).padStart(4)} / ${limit}`); + if (size > limit) { + failed = true; + console.error(`\n ${entry} now loads ${size} modules at runtime, over its budget of ${limit}.`); + console.error(` ${why}`); + console.error( + " Import the specific modules you need rather than a `mod.ts` barrel, or raise the budget deliberately.\n", + ); + } +} + +if (failed) { + console.error("Import graph budgets exceeded."); + process.exit(1); +} +console.log(`All ${BUDGETS.length} import graph budgets are within limits.`); diff --git a/.dev/perf/gate.ts b/.dev/perf/gate.ts index 0bf9fc8a..194403c1 100644 --- a/.dev/perf/gate.ts +++ b/.dev/perf/gate.ts @@ -78,6 +78,36 @@ function flag(args: readonly string[], name: string): string | undefined { return i >= 0 && i + 1 < args.length ? args[i + 1] : undefined; } +/** Sampling noise above which a recorded entry is not worth comparing against, in percent. */ +const NOISY_RME_PCT = 15; + +/** + * Warns about entries recorded with too much sampling noise to be a useful baseline. + * + * A high-`rme` entry is worse than no entry: the gate compares against it for months, and + * anyone reading the report treats it as the truth. The recorded + * `signing/order_e2e_no_ecdsa_unchecked` sat at 3031.9 ns with **41.0% rme** — a figure that + * does not reproduce (the same scenario measures 8–12 µs across a dozen fresh processes) — and + * three separate performance audits each spent effort explaining a 7.7 µs "validation cost" + * that only existed because that one number was noise. + * + * This warns rather than fails: a noisy machine is a reason to re-record, not to block the + * person doing it, and some scenarios are legitimately jittery. + */ +export async function warnOnNoisyEntries(path: string): Promise { + const report = (await Bun.file(path).json()) as { scenarios?: { name: string; nsPerUnit: number; rme: number }[] }; + const noisy = (report.scenarios ?? []).filter((s) => s.rme > NOISY_RME_PCT).sort((a, b) => b.rme - a.rme); + if (noisy.length === 0) return; + + console.warn( + `\n${noisy.length} scenario(s) recorded above ${NOISY_RME_PCT}% rme. A baseline entry this noisy will produce` + + ` false regressions and false all-clears for as long as it is committed:`, + ); + for (const s of noisy) + console.warn(` ${s.rme.toFixed(1).padStart(5)}% ${s.nsPerUnit.toFixed(1).padStart(10)} ns ${s.name}`); + console.warn("Re-record on an idle machine before committing, or accept these entries deliberately.\n"); +} + if (import.meta.main) { const args = Bun.argv.slice(2); @@ -85,6 +115,7 @@ if (import.meta.main) { if (args.includes("--record")) { await runSuite(BASELINE, "baseline"); console.log(`\nRecorded baseline: ${BASELINE}`); + await warnOnNoisyEntries(BASELINE); process.exit(0); } diff --git a/docs/clients.md b/docs/clients.md index 873c964a..2262bdf0 100644 --- a/docs/clients.md +++ b/docs/clients.md @@ -283,6 +283,54 @@ Beyond callback settle, enforcement is **best-effort**: > later nonce does NOT invalidate it. The payload goes **stale** only once 100 newer nonces have been consumed. > Prepare immediately before use anyway. +### Placing many orders at once + +An `order` action carries an **array** of orders, and the whole array is covered by **one signature** and costs +**one request**. Fanning the same orders out into N concurrent `order()` calls pays N signatures and N requests for +the same result, and secp256k1 is ~90% of the cost of an order — so this is by a wide margin the largest performance +decision available to a caller. + +```ts +// One action, one signature, one request. +await client.order({ + orders: [ + { a: 0, b: true, p: "30000", s: "0.1", r: false, t: { limit: { tif: "Gtc" } } }, + { a: 1, b: false, p: "2000", s: "1.5", r: false, t: { limit: { tif: "Gtc" } } }, + // ... up to the venue's per-action limit + ], + grouping: "na", +}); +``` + +Measured on this tree for 100 orders (Bun 1.4.0, Apple M3 Max, zero-latency in-memory transport; +median of 9, so the numbers isolate SDK CPU from the network): + +| Approach | Wall time | Rate-limit weight | +| -------------------------------------------- | -------------- | ----------------- | +| One batched action (either wallet) | **0.3–0.4 ms** | **3** | +| 100 concurrent `order()` calls, fast wallet | 7.2 ms | 100 | +| 100 concurrent `order()` calls, viem account | 12.1 ms | 100 | + +Batching is ~20–35× less CPU, and it makes the wallet choice stop mattering: one signature +amortized over 100 orders leaves the curve implementation contributing nothing measurable +(0.43 ms fast vs 0.32 ms viem, ranges overlapping). Choosing +[`createFastLocalWallet`](signing.md#fast-local-wallet-wasm-secp256k1) matters most for orders you *cannot* batch. + +The weight column is the part that bites first in production: the exchange endpoint charges +`1 + floor(batchLength / 40)`, so 100 orders in one action cost **3** of your +[1200 weight/minute per IP](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits), +while 100 separate actions cost **100**. Batching raises the ceiling on how often you can act by ~33×, independently +of any CPU saving. + +> [!NOTE] +> +> `grouping: "na"` is **not** atomic — it is documented as "standard order without grouping", and the orders in the +> array are accepted or rejected individually. You do not need separate actions to avoid all-or-nothing behavior. +> Use `"normalTpsl"` or `"positionTpsl"` only when you actually want the take-profit/stop-loss grouping semantics. + +Separate actions are genuinely required only when the orders differ in a field the action carries once rather than +per order — `vaultAddress`, `expiresAfter`, `builder`, or `grouping` itself. + ### Orders over WebSocket (low latency) Every `ExchangeClient` method also works over [`WebSocketTransport`](transports.md#websocket) — the server accepts @@ -372,11 +420,12 @@ const subscription = await client.allMids( ### Unsubscribe -A single connection supports up to -[1000 active subscriptions and 15 unique users](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits) -(the official docs, updated ~17 days earlier, say 10 — but a live mainnet probe on 2026-07-26 observed the server -accepting 15 users and rejecting the 16th with "Cannot track more than 15 total users."; the server is the authority -here, the docs lag). +Hyperliquid allows +[1000 active subscriptions and 14 unique users](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits) +**per IP address** — not per connection, so every `WebSocketTransport` on a network shares one budget (see +[WebSocket limits](transports.md#websocket-limits)). The official docs say 10 unique users and the server's own +refusal says 15; a live mainnet probe on 2026-08-02 measured the enforced ceiling at 14 on two independent +connections, which is what the SDK guards against — see [known drift](reference/known-drift.md). Call `unsubscribe()` to remove a listener and free these slots: ```ts diff --git a/docs/reference/known-drift.md b/docs/reference/known-drift.md index 804f53d5..0415d664 100644 --- a/docs/reference/known-drift.md +++ b/docs/reference/known-drift.md @@ -53,17 +53,23 @@ When an entry is resolved upstream (docs fixed, or server aligned with docs), mo listeners separate even though they share one server-side subscription (`src/transport/websocket/_subscriptionManager.ts`). -### 6. Unique users per connection — docs say 10, server allows 15 +### 6. Unique users — docs say 10, the server's error says 15, the server enforces 14 -- **Observed:** 2026-07-26 (live mainnet). +- **Observed:** 2026-08-02 (live mainnet), superseding a 2026-07-26 observation. - **Docs claim:** maximum of 10 unique users across user-specific WebSocket subscriptions (updated ~2026-07; [rate limits](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits)). -- **Server reality:** the 15th unique user was acknowledged; the 16th was rejected with - `Cannot track more than 15 total users.` -- **SDK behavior:** matches the server — the subscription manager enforces 15 unique users per connection - client-side (`MAX_UNIQUE_USERS = 15`, `src/transport/websocket/_subscriptionManager.ts`) and throws the same - message before wasting a round trip. Note [Clients → Unsubscribe](../clients.md#unsubscribe) still quotes the - docs' "10 unique users". +- **Server reality:** **14** distinct users are accepted; the **15th** is refused with an `error` frame reading + `Cannot track more than 15 total users.` — the message is off by one from the enforcement. Measured by subscribing + distinct users one at a time with the client-side guard disabled, twice, on two independent connections; both runs + stopped at 14. +- **Scope:** **per IP, not per connection.** With one connection holding 14 users, a second connection from the same + host was refused a 15th distinct user, while still being allowed to subscribe a user the first connection already + held. Sharding user channels across sockets therefore buys no additional user slots. +- **SDK behavior:** enforces the measured 14 (`MAX_UNIQUE_USERS = 14`, `src/transport/websocket/_quota.ts`), counted + against a per-IP [`WebSocketQuota`](../transports.md#websocket-limits) shared by every transport on the network. + The earlier value of 15 was taken from the server's error text and was one too high — the 15th subscription passed + the client guard, and because the server's refusal carries no echoed request, it could not be matched to the + pending subscribe and surfaced only as a request timeout ~10 s later. ### 7. `TwapState` frames gained `trigger` / `stopPx` @@ -88,6 +94,29 @@ When an entry is resolved upstream (docs fixed, or server aligned with docs), mo resolves outcome asset IDs (`100000000 + outcomeId * 10 + sideIndex`) from `outcomeMeta` alone. Format prices and sizes for outcome markets with the spot-like model above, at your own risk. +### 9. `outcomeMeta` outcomes gained a `deployer` field + +- **Observed:** 2026-08-02 (live mainnet), via the `outcomeMeta` schema-coverage test. +- **Docs claim:** each entry of `outcomes` carries no `deployer`. +- **Server reality:** every outcome in the response now includes `deployer`; the schema-coverage check reports + `additionalProperty: "deployer"` across the whole `outcomes` array (observed at indices 0 through 157+). +- **SDK behavior:** runtime unaffected — Info responses are delivered to callers as received, so the field is present + on the objects you get. The `OutcomeMetaResponse` **type** in `src/api/info/_methods/outcomeMeta.ts` does not + declare it yet, so it is invisible to TypeScript and `tests/api/info/outcomeMeta.test.ts` fails online until the + type is widened. Per [Versioning](../README.md#versioning) that type change ships in a patch release. + +### 10. `validatorL1Votes` actions gained `registerTemplate` + +- **Observed:** 2026-08-02 (live mainnet), via the `validatorL1Votes` schema-coverage test. +- **Docs claim:** the validator action union covers `registerTokensAndStandaloneOutcome` among its variants. +- **Server reality:** a live vote carried an action whose `O` object holds `registerTemplate` and omits + `registerTokensAndStandaloneOutcome`, so it matches no variant of the documented union — the check reports both + `missingProperty: "registerTokensAndStandaloneOutcome"` and `additionalProperty: "registerTemplate"` for the same + sample. +- **SDK behavior:** runtime unaffected for the same reason as #9; the union in + `src/api/info/_methods/validatorL1Votes.ts` needs a `registerTemplate` variant, and + `tests/api/info/validatorL1Votes.test.ts` fails online until it has one. + ## Resolved _None yet._ diff --git a/docs/transports.md b/docs/transports.md index c6903bbb..7722e51a 100644 --- a/docs/transports.md +++ b/docs/transports.md @@ -255,3 +255,56 @@ const transport = new WebSocketTransport({ resubscribe: false }); If a subscription then fails to re-establish, its `onError` callback is invoked. Handle it as shown under [subscription errors](clients.md#errors). + +### WebSocket limits + +Hyperliquid scopes every documented WebSocket limit to your **IP address**, not to the connection — two of them say so +in their own text ("across all websocket connections"): + +| Limit | Value | Scope | +| -------------------------------------- | ----------- | ------------------------------ | +| Connections | 10 | per IP | +| New connections | 30/minute | per IP | +| Subscriptions | 1000 | per IP | +| Unique users across user-specific subs | 14 | per IP | +| Messages sent to Hyperliquid | 2000/minute | per IP, across all connections | +| Simultaneous inflight post requests | 100 | per IP, across all connections | + +Because that scope is the IP and not the socket, every `WebSocketTransport` on a network **shares one budget** by +default, and the subscription and unique-user guards count what the server counts. Two transports no longer admit 2000 +subscriptions against a limit of 1000 — the excess is refused locally with a clear +[`WebSocketRequestError`](error-handling.md) instead of by the server, whose refusal carries no echoed request and +therefore surfaces only as a request timeout ten seconds later. + +Reservations are released when a subscription is unsubscribed or when its connection is permanently closed, which is +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: + +```ts +import { WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; + +const transport = new WebSocketTransport({ quota: new WebSocketQuota() }); +``` + +### 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 }); +``` + +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. + +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/package.json b/package.json index c431f7f8..b0c8ac29 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@bloxwap/hyperliquid", - "version": "0.1.6", + "version": "0.2.0", "description": "Blazing fast TypeScript Hyperliquid SDK.", "license": "MIT", "type": "module", @@ -31,18 +31,26 @@ "node": ">=22.12.0" }, "//exports": "Points at TypeScript sources: Bun consumes them directly, and self-referential imports (`@bloxwap/hyperliquid/...`) in tests, docs and JSDoc examples resolve without a build. The publishable package is emitted by `bun run build`, which writes dist/ with its own package.json whose exports point at .js/.d.ts.", + "//exports:narrow": "The `.` barrel re-exports all four clients plus the transports, so importing it evaluates the exchange and signing graph even for a read-only process: 69.5 ms on Node, 22.0 ms on Bun, against 41.0 / 8.3 ms for an info-only consumer entering through `./api/info/client` + `./transport`. The `./transport*` and `./api/*/client` keys below exist for that consumer — `.` stays the compatible default and nothing under src/ changed to add them.", "exports": { ".": "./src/mod.ts", "./signing": "./src/signing/mod.ts", "./utils": "./src/utils/mod.ts", + "./transport": "./src/transport/mod.ts", + "./transport/http": "./src/transport/http/mod.ts", + "./transport/websocket": "./src/transport/websocket/mod.ts", "./api/exchange": "./src/api/exchange/mod.ts", + "./api/exchange/client": "./src/api/exchange/client.ts", "./api/explorer": "./src/api/explorer/mod.ts", + "./api/explorer/client": "./src/api/explorer/client.ts", "./api/info": "./src/api/info/mod.ts", - "./api/subscription": "./src/api/subscription/mod.ts" + "./api/info/client": "./src/api/info/client.ts", + "./api/subscription": "./src/api/subscription/mod.ts", + "./api/subscription/client": "./src/api/subscription/client.ts" }, "scripts": { "//check": "Everything CI gates on.", - "check": "bun run check:format && bun run check:lint && bun run check:docs && bun run check:types && bun run check:ts7 && bun run check:jsdoc && bun run check:export", + "check": "bun run check:format && bun run check:lint && bun run check:docs && bun run check:types && bun run check:ts7 && bun run check:jsdoc && bun run check:export && bun run check:imports", "check:format": "biome format .", "check:lint": "biome lint .", "//check:docs": "Verifies Markdown formatting, the docs index, and GitHub-compatible syntax. External links are checked by the Documentation workflow.", @@ -55,6 +63,8 @@ "check:jsdoc": "bun run .dev/jsdoc_sync_check.ts", "//check:export": "Verifies every module is reachable from the package exports.", "check:export": "bun run .dev/export_sync_check.ts", + "//check:imports": "Budgets how many modules each entry point loads at runtime, so a barrel import cannot silently reattach a large subgraph to a small entry point.", + "check:imports": "bun run .dev/import_graph_check.ts", "format": "biome format --write .", "lint": "biome lint --write .", "test": "bun test tests/", diff --git a/src/api/exchange/client.ts b/src/api/exchange/client.ts index 6975c274..1b857065 100644 --- a/src/api/exchange/client.ts +++ b/src/api/exchange/client.ts @@ -351,6 +351,27 @@ export class ExchangeClient = new Map(); /** Shared request-timeout scheduler: at most one armed native timer, however many requests are in flight. */ private readonly _timeouts: abort.TimeoutWheel; - - constructor(socket: ReconnectingWebSocket, hlEvents: HyperliquidEventTarget, timeout: number | null) { + /** + * Controllers to abort when the socket terminates permanently: every in-flight request that + * was not already settled by its own signal. + * + * The socket's `terminationSignal` is ONE `AbortSignal` shared by every request. Relaying it + * per request — `abort.relay([signal, terminationSignal], controller)` — put one listener per + * in-flight request on that single signal, and `AbortSignal` is an `EventTarget` whose + * per-type listener list is scanned linearly by both `addEventListener` and + * `removeEventListener`. Each request's attach/detach pair was therefore O(in-flight), making + * a burst O(n^2): measured at 195 ns per pair with the list empty and 13.1 µs with 5000 + * listeners resident on Bun (40.4 µs on Node). A reconnect at the 1000-subscription cap — + * exactly the storm the {@linkcode PendingRequest.echo} comment describes — paid it worst. + * + * One listener on the signal plus a `Set` of controllers makes attach and detach O(1), and + * costs ~300 ns less per request even at an in-flight count of one. + */ + private readonly _terminating: Set = new Set(); + /** + * 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. + */ + private readonly _quota: WebSocketQuota | undefined; + + constructor( + socket: ReconnectingWebSocket, + hlEvents: HyperliquidEventTarget, + timeout: number | null, + quota?: WebSocketQuota, + ) { this.timeout = timeout; this._socket = socket; this._timeouts = new abort.TimeoutWheel(); + this._quota = quota; // --- Hyperliquid event handlers ------------------------------------------ hlEvents.addEventListener("subscriptionResponse", (event) => this._handleSubscriptionResponse(event.detail)); @@ -203,6 +236,23 @@ export class WebSocketDispatcher { this._socket.send(entry.frame); } }); + + // --- Termination fan-out ------------------------------------------------- + // ONE listener on the socket's shared termination signal for this dispatcher's whole life, + // fanning out to the in-flight requests. See {@linkcode _terminating} for why a per-request + // listener here was quadratic. + socket.terminationSignal.addEventListener( + "abort", + () => { + const reason = socket.terminationSignal.reason; + // Snapshot before aborting: each `abort` runs its request's `finally`, which deletes + // from this set while it is being iterated. + const pending = [...this._terminating]; + this._terminating.clear(); + for (const controller of pending) controller.abort(reason); + }, + { once: true }, + ); } // =========================================================================== @@ -221,13 +271,41 @@ export class WebSocketDispatcher { signal?: AbortSignal, hint?: RequestHint, ): Promise { + // --- Outbound budget ------------------------------------------------------ + // Paced before the timeout is armed, so deliberate throttling never trips the request + // timeout — the same ordering `HttpTransport` uses. + // + // `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; + } + // One controller per request: the timeout timer, the user signal, and the // socket termination relay into it, and `finally` detaches everything, so // no listener or timer outlives the request. const controller = new AbortController(); const timeoutMs = this.timeout; // for correct error message after user changes const timeout = this._timeouts.schedule(controller, timeoutMs); - const detachRelay = abort.relay([signal, this._socket.terminationSignal], controller); + // The caller's signal is relayed first, and the termination branch below runs only while the + // controller is still unaborted. Together those reproduce `relay([signal, terminationSignal])` + // exactly, INCLUDING its reason precedence: `relay` aborts with the first already-aborted + // source in argument order, so when the caller's signal and the socket's termination are both + // aborted before the request starts, the caller's reason is the one that surfaces. Reversing + // these two lines silently changes the error a caller sees in that case. + const detachRelay = abort.relay([signal], controller); + if (!controller.signal.aborted) { + // A shared listener with an O(1) Set membership, rather than a listener per request on a + // signal every request shares — see {@linkcode _terminating}. + if (this._socket.terminationSignal.aborted) controller.abort(this._socket.terminationSignal.reason); + else this._terminating.add(controller); + } let entry: PendingRequest | undefined; // The one serialization of the request envelope — wire form and error snapshot source. @@ -266,7 +344,13 @@ export class WebSocketDispatcher { // --- Send or queue ----------------------------------------------------- frame ??= JSON.stringify(request); const sent = this._socket.readyState === ReconnectingWebSocket.OPEN; - if (sent) this._socket.send(frame); + 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`. + if (method === "post") this._quota?.chargeSend(); + } const { promise, resolve, reject } = Promise_.withResolvers(); const pending = (entry = { id, frame, sent, echo, resolve, reject }); @@ -312,6 +396,8 @@ export class WebSocketDispatcher { if (entry) this._dequeue(entry); timeout.cancel(); detachRelay(); + // O(1) — this is the detach that used to scan the shared signal's listener list. + this._terminating.delete(controller); } } diff --git a/src/transport/websocket/_keepAlive.ts b/src/transport/websocket/_keepAlive.ts index 1ac70ca8..b056fa4d 100644 --- a/src/transport/websocket/_keepAlive.ts +++ b/src/transport/websocket/_keepAlive.ts @@ -7,6 +7,7 @@ import type { ReconnectingWebSocket } from "./_reconnectingSocket.ts"; import type { HyperliquidEventTarget } from "./_events.ts"; +import type { WebSocketQuota } from "./_quota.ts"; /** Configuration options for the keep-alive watchdog. */ export interface WebSocketKeepAliveOptions { @@ -31,13 +32,28 @@ export class WebSocketKeepAlive { private readonly _socket: ReconnectingWebSocket; private readonly _interval: number; private readonly _timeout: number; + /** + * The per-IP outbound message budget pings are billed to, or `undefined` when unbudgeted. + * + * Pings are not free: at the 5 s default each open socket spends 12 of the 2000 messages + * a minute allows, and that budget is shared by every connection from this host. They are + * charged rather than paced — delaying the watchdog is how a half-open socket goes + * unnoticed — so heavy ping traffic slows subscribes instead of itself. + */ + private readonly _quota: WebSocketQuota | undefined; private _pingInterval: ReturnType | undefined; private _pongTimeout: ReturnType | undefined; - constructor(socket: ReconnectingWebSocket, hlEvents: HyperliquidEventTarget, options?: WebSocketKeepAliveOptions) { + constructor( + socket: ReconnectingWebSocket, + hlEvents: HyperliquidEventTarget, + options?: WebSocketKeepAliveOptions, + quota?: WebSocketQuota, + ) { this._socket = socket; this._interval = options?.interval ?? 5_000; this._timeout = options?.timeout ?? 3_000; + this._quota = quota; hlEvents.addEventListener("pong", () => this._disarm()); socket.addEventListener("open", () => this._start()); @@ -49,6 +65,7 @@ export class WebSocketKeepAlive { if (this._pingInterval) return; this._pingInterval = 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._interval); diff --git a/src/transport/websocket/_quota.ts b/src/transport/websocket/_quota.ts new file mode 100644 index 00000000..d7ff1918 --- /dev/null +++ b/src/transport/websocket/_quota.ts @@ -0,0 +1,286 @@ +/** + * The per-IP budget every WebSocket connection to one Hyperliquid deployment shares. + * + * Hyperliquid scopes **every** documented WebSocket limit to the client IP, not to the + * connection — two of them say so in their own text ("across all websocket connections"): + * + * | Limit | Value | Scope | + * | ------------------------------ | -------------- | ------------------------------ | + * | Subscriptions | 1000 | per IP | + * | Unique users across user subs | 14 (see below) | per IP | + * | Messages sent to Hyperliquid | 2000/minute | per IP, across all connections | + * | Simultaneous inflight posts | 100 | per IP, across all connections | + * + * A budget tracked per {@linkcode WebSocketSubscriptionManager} therefore counts the wrong + * thing: two transports in one process each admitted 1000 subscriptions against a limit of + * 1000 total, and the excess was refused by the server rather than by the guard. That + * refusal is the expensive kind — the server's `error` frame carries no echoed request, so + * it cannot be matched to the pending subscribe and surfaces only when the request times + * out, ten seconds later, as a timeout rather than as a limit. + * + * 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. + * + * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits + * @module + */ + +import { TokenBucketRateLimiter } from "../_rateLimiter.ts"; + +// ============================================================================= +// Documented limits +// ============================================================================= + +/** + * Maximum concurrent subscriptions per IP; the server rejects the excess without echoing + * the request, so the guard must run client-side. + */ +const MAX_SUBSCRIPTIONS = 1000; + +/** + * Maximum unique users across user-specific subscriptions per IP; the server rejects the + * excess without echoing the request, so the guard must run client-side. + * + * **14, not 15 — and the server's own error message is what misleads here.** A live mainnet + * probe on 2026-08-02 subscribed distinct users one at a time on a fresh connection with this + * guard disabled, twice, on two independent connections: both accepted exactly **14** and had + * the 15th refused by an `error` frame reading `Cannot track more than 15 total users.` The + * server enforces one fewer than its message states, so taking that message at face value + * (as the 2026-07-26 probe recorded in this file's history did) sets the guard one too high — + * and the 15th subscription is then the worst case of all: it passes the client check, the + * server drops it *without echoing the request*, nothing can be matched to the pending + * subscribe, and the caller waits out the full request timeout for what should have been an + * instant local error. + * + * The same probe established the scope: after one connection held 14 users, a **second** + * connection from the same host was refused a 15th distinct user while still being allowed to + * subscribe a user the first connection already held. The cap is per IP, not per connection — + * so sharding user channels across sockets buys no additional user slots. + * + * Erring low is the safe direction: refusing one subscription the server might have taken + * costs a slot, while admitting one it refuses costs a 10 s timeout. + * + * (The official docs say 10, which matches neither the message nor the enforcement.) + */ +const MAX_UNIQUE_USERS = 14; + +/** Maximum messages sent to Hyperliquid per minute per IP, across every connection. */ +const MAX_MESSAGES_PER_MINUTE = 2000; + +/** Maximum simultaneous inflight `post` requests per IP, across every connection. */ +const MAX_INFLIGHT_POSTS = 100; + +// ============================================================================= +// Options +// ============================================================================= + +/** Pacing configuration for outbound WebSocket messages. */ +export interface WebSocketRateLimitOptions { + /** + * Maximum burst size, in messages. The bucket starts full. + * + * Default: `2000` + */ + capacity?: number; + /** + * Steady-state refill rate, in messages per minute. + * + * Default: `2000` + */ + refillPerMinute?: number; +} + +/** Configuration options for a {@linkcode WebSocketQuota}. */ +export interface WebSocketQuotaOptions { + /** + * Opt-in 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) + */ + rateLimit?: WebSocketRateLimitOptions; + /** + * Maximum concurrent subscriptions, or `null` to disable the guard. + * + * Default: `1000` + */ + maxSubscriptions?: number | null; + /** + * Maximum unique users across user-specific subscriptions, or `null` to disable the guard. + * + * Default: `14` + */ + maxUniqueUsers?: number | null; +} + +/** Why a subscription reservation was refused. */ +export type QuotaRefusal = "subscriptions" | "users"; + +// ============================================================================= +// Quota +// ============================================================================= + +/** + * The subscription, unique-user and outbound-message budget for one Hyperliquid deployment. + * + * Every counter here mirrors a limit the server enforces per IP, so one instance must be + * shared by every connection that reaches that deployment from this host. Sharing is the + * default; {@linkcode WebSocketTransportOptions.quota} exists for the cases the default + * cannot see — a process behind several egress IPs, or a test that wants isolation. + */ +export class WebSocketQuota { + /** 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. */ + readonly maxUniqueUsers: number | null; + /** Maximum simultaneous inflight `post` requests the server accepts across all connections. */ + readonly maxInflightPosts: number = MAX_INFLIGHT_POSTS; + + /** Live subscriptions across every connection sharing this quota. */ + private _subscriptions = 0; + /** + * Live subscription count per tracked user, maintained incrementally so the unique-user + * guard stays O(1) per subscribe instead of rescanning every registered id. The map's + * size — not its summed values — is what the limit applies to. + */ + private readonly _users: Map = new Map(); + /** Outbound message pacer, or `null` when pacing is off and only accounting runs. */ + private readonly _limiter: TokenBucketRateLimiter | null; + + constructor(options?: WebSocketQuotaOptions) { + this.maxSubscriptions = options?.maxSubscriptions === undefined ? MAX_SUBSCRIPTIONS : options.maxSubscriptions; + this.maxUniqueUsers = options?.maxUniqueUsers === undefined ? MAX_UNIQUE_USERS : options.maxUniqueUsers; + this._limiter = + options?.rateLimit === undefined + ? null + : new TokenBucketRateLimiter( + options.rateLimit.capacity ?? MAX_MESSAGES_PER_MINUTE, + options.rateLimit.refillPerMinute ?? MAX_MESSAGES_PER_MINUTE, + ); + } + + // =========================================================================== + // Subscription budget + // =========================================================================== + + /** Live subscriptions counted against this quota. */ + get subscriptions(): number { + return this._subscriptions; + } + + /** Distinct users counted against this quota. */ + get uniqueUsers(): number { + return this._users.size; + } + + /** + * Reserves one subscription slot, and a user slot when `user` is set. + * + * Checking and reserving are one step on purpose: a split check-then-add lets two + * subscribes interleave between the two halves and both pass a guard only one of them + * should have. Returns the limit that refused the reservation, or `undefined` when it + * succeeded — the caller builds the error, so this module stays free of a dependency on + * the dispatcher's error type. + * + * @param user Lowercased user address this subscription tracks, or `undefined` when it tracks none. + */ + reserveSubscription(user: string | undefined): QuotaRefusal | undefined { + if (this.maxSubscriptions !== null && this._subscriptions >= this.maxSubscriptions) return "subscriptions"; + // Only a user not already tracked consumes a slot: N subscriptions for one user cost one. + if ( + user !== undefined && + this.maxUniqueUsers !== null && + !this._users.has(user) && + this._users.size >= this.maxUniqueUsers + ) { + return "users"; + } + + this._subscriptions++; + if (user !== undefined) this._users.set(user, (this._users.get(user) ?? 0) + 1); + return undefined; + } + + /** + * Releases a slot taken by {@linkcode WebSocketQuota.reserveSubscription}. + * + * @param user The same value passed to the matching reservation. + */ + releaseSubscription(user: string | undefined): void { + if (this._subscriptions > 0) this._subscriptions--; + if (user === undefined) return; + + const count = this._users.get(user); + // The count is only ever absent if a reservation was lost, in which case forgetting the + // user is the safe direction: the server rejects a genuine overflow, a stuck count would + // wedge the guard closed against subscriptions the server would have accepted. + if (count === undefined || count <= 1) this._users.delete(user); + else this._users.set(user, count - 1); + } + + // =========================================================================== + // Outbound message budget + // =========================================================================== + + /** + * 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. + * + * Only `subscribe` / `unsubscribe` frames go through this path. See + * {@linkcode WebSocketQuota.chargeSend} for the frames that must never wait. + * + * @param signal Abandons the wait; the caller's request fails rather than sending late. + */ + acquireSend(signal?: AbortSignal): Promise | undefined { + if (this._limiter === null) return undefined; + return this._limiter.acquire(1, signal); + } + + /** + * Deducts one message from the outbound budget without waiting. + * + * For frames that cannot be delayed without breaking something else: + * - `post` frames, because an exchange action's wire order is fixed by reaching the socket + * synchronously (see {@linkcode WebSocketQuota.acquireSend}); + * - keep-alive pings, because delaying the watchdog is how a half-open socket goes unnoticed. + * + * The deduction can drive the bucket into debt, which later `subscribe` frames wait off — + * so an order burst correctly slows subscription traffic instead of silently overrunning + * the shared budget. + */ + chargeSend(): void { + this._limiter?.charge(1); + } +} + +// ============================================================================= +// Shared instances +// ============================================================================= + +/** + * The process-wide quota per deployment, created on first use. + * + * Keyed by network rather than by URL: mainnet and testnet are different servers keeping + * different per-IP counters, while a custom `url` almost always points at the same + * deployment through a proxy. Collapsing a custom URL onto its network's quota therefore + * errs toward counting *together* — the safe direction, since over-counting refuses a + * subscription the server would have taken (a clear client-side error) while under-counting + * produces the unmatched server refusal this whole module exists to avoid. + */ +const SHARED: { mainnet?: WebSocketQuota; testnet?: WebSocketQuota } = {}; + +/** The quota every transport on `isTestnet` shares unless it was given its own. */ +export function sharedWebSocketQuota(isTestnet: boolean): WebSocketQuota { + const key = isTestnet ? "testnet" : "mainnet"; + return (SHARED[key] ??= new WebSocketQuota()); +} diff --git a/src/transport/websocket/_subscriptionManager.ts b/src/transport/websocket/_subscriptionManager.ts index aff02b8b..bf95dcd2 100644 --- a/src/transport/websocket/_subscriptionManager.ts +++ b/src/transport/websocket/_subscriptionManager.ts @@ -1,6 +1,9 @@ /** * Subscription lifecycle manager: tracks listeners per subscription payload, - * resubscribes on reconnect, and enforces per-connection subscription limits. + * resubscribes on reconnect, and enforces Hyperliquid's per-IP subscription limits. + * + * The limits are enforced against a {@linkcode WebSocketQuota} shared with every other + * connection to the same deployment, because that is the scope the server counts in. * * @module */ @@ -11,6 +14,7 @@ import type { ISubscription } from "../_base.ts"; import type { HyperliquidEventTarget } from "./_events.ts"; import { type WebSocketDispatcher, WebSocketRequestError } from "./_dispatcher.ts"; import { normalize } from "./_id.ts"; +import { WebSocketQuota } from "./_quota.ts"; import { payloadEventType } from "./_routing.ts"; /** A live reference to a registration: one per subscribing call, until its waiter settles or its handle unsubscribes. */ @@ -91,30 +95,6 @@ interface SubscriptionState { failure?: WebSocketRequestError; } -/** - * Maximum number of subscriptions; the server rejects the excess without - * echoing the request, so the guard must run client-side. - * - * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits - */ -const MAX_SUBSCRIPTIONS = 1000; - -/** - * Maximum number of unique users across subscriptions; the server rejects the - * excess without echoing the request, so the guard must run client-side. - * - * The official docs say 10 (updated ~17 days before the probe), but a live - * mainnet probe on 2026-07-26 observed the server accepting 15 users and - * rejecting the 16th with an `error` frame "Cannot track more than 15 total - * users." — the server is the authority here and the docs lag. The rejection - * carries no echoed request, so it cannot be matched to the pending subscribe - * and would surface only via the request timeout — another reason the - * client-side guard must fire first. - * - * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/rate-limits-and-user-limits - */ -const MAX_UNIQUE_USERS = 15; - /** Lowercased `user` of a subscription payload, or `undefined` when the payload tracks no user. */ function userOf(payload: unknown): string | undefined { return typeof payload === "object" && payload !== null && "user" in payload && typeof payload.user === "string" @@ -132,21 +112,26 @@ export class WebSocketSubscriptionManager { private readonly _hlEvents: HyperliquidEventTarget; private _subscriptions: Map = new Map(); /** - * Live subscription count per tracked user, maintained incrementally so the unique-user guard - * stays O(1) per subscribe instead of rescanning (and re-parsing) every registered id. + * The per-IP budget this connection draws from. + * + * Subscription and unique-user counts live here rather than on the manager because the + * server counts them per IP, across every connection from this host — a manager-local + * count admitted N x 1000 subscriptions against a limit of 1000. See {@linkcode WebSocketQuota}. */ - private readonly _users: Map = new Map(); + private readonly _quota: WebSocketQuota; constructor( socket: ReconnectingWebSocket, dispatcher: WebSocketDispatcher, hlEvents: HyperliquidEventTarget, resubscribe: boolean, + quota: WebSocketQuota = new WebSocketQuota(), ) { this._socket = socket; this._dispatcher = dispatcher; this._hlEvents = hlEvents; this.resubscribe = resubscribe; + this._quota = quota; socket.addEventListener("open", () => this._handleOpen()); socket.addEventListener("close", () => this._handleClose()); @@ -197,13 +182,18 @@ export class WebSocketSubscriptionManager { // --- Subscription state -------------------------------------------------- let subscription = this._subscriptions.get(id); if (!subscription) { - if (this._subscriptions.size >= MAX_SUBSCRIPTIONS) { - throw new WebSocketRequestError(`Cannot subscribe to more than ${MAX_SUBSCRIPTIONS} channels.`, { + // Reserved against the shared per-IP budget before the request goes out. The reservation + // is released again by `_deleteSubscription` on every exit path — unsubscribe, refusal, + // disconnect — so an abandoned subscribe never leaks a slot to the other connections. + const user = userOf(snapshot); + const refusal = this._quota.reserveSubscription(user); + if (refusal === "subscriptions") { + throw new WebSocketRequestError(`Cannot subscribe to more than ${this._quota.maxSubscriptions} channels.`, { request: payload, }); } - if (this._exceedsUserLimit(payload)) { - throw new WebSocketRequestError(`Cannot track more than ${MAX_UNIQUE_USERS} total users.`, { + if (refusal === "users") { + throw new WebSocketRequestError(`Cannot track more than ${this._quota.maxUniqueUsers} total users.`, { request: payload, }); } @@ -215,7 +205,7 @@ export class WebSocketSubscriptionManager { .finally(() => (created.promiseFinished = true)); const created: SubscriptionState = { payload: snapshot, - user: userOf(snapshot), + user, listeners: new Map(), promise, promiseFinished: false, @@ -357,26 +347,22 @@ export class WebSocketSubscriptionManager { // Registry // =========================================================================== - /** Registers a subscription and counts its user towards the unique-user limit. */ + /** + * Registers a subscription whose quota slot the caller has already reserved. + * + * The reservation happens in `subscribe()` rather than here, because it must run — and be + * able to refuse — before the subscribe request is put on the wire. + */ private _addSubscription(id: string, subscription: SubscriptionState): void { this._subscriptions.set(id, subscription); - const user = subscription.user; - if (user !== undefined) this._users.set(user, (this._users.get(user) ?? 0) + 1); } - /** Removes a subscription and releases its user from the unique-user count. */ + /** Removes a subscription and returns its slot to the shared per-IP budget. */ private _deleteSubscription(id: string): void { const subscription = this._subscriptions.get(id); if (subscription === undefined) return; this._subscriptions.delete(id); - - const user = subscription.user; - if (user === undefined) return; - const count = this._users.get(user); - // The count is only ever absent if a registration was lost, in which case forgetting the user - // is the safe direction: the server rejects a genuine overflow, a stuck count would not. - if (count === undefined || count <= 1) this._users.delete(user); - else this._users.set(user, count - 1); + this._quota.releaseSubscription(subscription.user); } /** @@ -446,15 +432,4 @@ export class WebSocketSubscriptionManager { } } } - - // =========================================================================== - // Subscription limit checks - // =========================================================================== - - /** True when subscribing `payload` would track one user above the limit. */ - private _exceedsUserLimit(payload: unknown): boolean { - const user = userOf(payload); - if (user === undefined) return false; - return !this._users.has(user) && this._users.size >= MAX_UNIQUE_USERS; - } } diff --git a/src/transport/websocket/mod.ts b/src/transport/websocket/mod.ts index 0f6e7d18..7488bb7f 100644 --- a/src/transport/websocket/mod.ts +++ b/src/transport/websocket/mod.ts @@ -26,9 +26,15 @@ import type { IRequestTransport, ISubscription, ISubscriptionTransport } from ". import { WebSocketDispatcher, WebSocketRequestError } from "./_dispatcher.ts"; import { HyperliquidEventTarget } from "./_events.ts"; import { WebSocketKeepAlive, type WebSocketKeepAliveOptions } from "./_keepAlive.ts"; +import { + sharedWebSocketQuota, + WebSocketQuota, + type WebSocketQuotaOptions, + type WebSocketRateLimitOptions, +} from "./_quota.ts"; import { WebSocketSubscriptionManager } from "./_subscriptionManager.ts"; -export { WebSocketRequestError }; +export { WebSocketQuota, type WebSocketQuotaOptions, type WebSocketRateLimitOptions, WebSocketRequestError }; /** Configuration options for the WebSocket transport layer. */ export interface WebSocketTransportOptions { @@ -66,6 +72,26 @@ export interface WebSocketTransportOptions { * Default: `true` */ resubscribe?: boolean; + /** + * The per-IP budget this transport draws from: subscriptions, unique users, and outbound + * messages. + * + * 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. + * + * @example Pace outbound messages against the documented 2000/minute budget + * ```ts + * import { WebSocketQuota, WebSocketTransport } from "@bloxwap/hyperliquid"; + * + * const quota = new WebSocketQuota({ rateLimit: { capacity: 2000, refillPerMinute: 2000 } }); + * const transport = new WebSocketTransport({ quota }); + * ``` + * + * Default: {@linkcode sharedWebSocketQuota} for this network + */ + quota?: WebSocketQuota; } /** Mainnet API WebSocket URL. */ @@ -113,6 +139,9 @@ export class WebSocketTransport implements IRequestTransport<"info" | "exchange" this._dispatcher.timeout = value; } + /** The per-IP budget this transport draws from; shared with every transport that was not given its own. */ + readonly quota: WebSocketQuota; + private readonly _hlEvents: HyperliquidEventTarget; private readonly _dispatcher: WebSocketDispatcher; private readonly _subscriptionManager: WebSocketSubscriptionManager; @@ -120,6 +149,7 @@ export class WebSocketTransport implements IRequestTransport<"info" | "exchange" /** Creates the transport and immediately starts connecting. */ constructor(options?: WebSocketTransportOptions) { this.isTestnet = options?.isTestnet ?? false; + this.quota = options?.quota ?? sharedWebSocketQuota(this.isTestnet); this.socket = new ReconnectingWebSocket( options?.url ?? (this.isTestnet ? TESTNET_API_WS_URL : MAINNET_API_WS_URL), @@ -131,15 +161,17 @@ export class WebSocketTransport implements IRequestTransport<"info" | "exchange" this.socket, this._hlEvents, options?.timeout === undefined ? 10_000 : options.timeout, + this.quota, ); // 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); + new WebSocketKeepAlive(this.socket, this._hlEvents, options?.keepAlive, this.quota); this._subscriptionManager = new WebSocketSubscriptionManager( this.socket, this._dispatcher, this._hlEvents, options?.resubscribe ?? true, + this.quota, ); } diff --git a/src/utils/_symbolConverter.ts b/src/utils/_symbolConverter.ts index ef909f67..de716c1d 100644 --- a/src/utils/_symbolConverter.ts +++ b/src/utils/_symbolConverter.ts @@ -1,13 +1,19 @@ -import { - meta, - type MetaResponse, - outcomeMeta, - type OutcomeMetaResponse, - perpDexs, - type PerpDexsResponse, - spotMeta, - type SpotMetaResponse, -} from "../api/info/mod.ts"; +// Imported from the four method modules directly, NOT from `../api/info/mod.ts`. +// +// That barrel re-exports every Info method, and this was the only value import of it anywhere in +// `src/` — so importing four functions pulled the whole Info surface, and with it valibot's +// schema graph for ~90 endpoints, into anyone who imported `@bloxwap/hyperliquid/utils`. Measured +// on the built package: 91 modules loaded and 22.71 ms on Node / 6.88 ms on Bun, against 10 +// modules and 5.50 / 4.06 ms importing the same four modules directly. The type-only imports +// below cost nothing either way and are listed here for locality. +// +// `.dev/import_graph_check.ts` gates this: it fails if `dist/utils/mod.js` ever pulls in more +// than a handful of modules again, which is the only thing standing between this file and a +// future `import { … } from "../api/info/mod.ts"` quietly restoring the 22 ms. +import { meta, type MetaResponse } from "../api/info/_methods/meta.ts"; +import { outcomeMeta, type OutcomeMetaResponse } from "../api/info/_methods/outcomeMeta.ts"; +import { perpDexs, type PerpDexsResponse } from "../api/info/_methods/perpDexs.ts"; +import { spotMeta, type SpotMetaResponse } from "../api/info/_methods/spotMeta.ts"; import type { IRequestTransport } from "../transport/mod.ts"; import { type DecimalParts, FormatError, stripTrailingZeros, toDecimal, toFixed } from "./_decimal.ts"; diff --git a/tests/api/exchange/_t.ts b/tests/api/exchange/_t.ts index 6488dfd1..38680ae1 100644 --- a/tests/api/exchange/_t.ts +++ b/tests/api/exchange/_t.ts @@ -30,6 +30,16 @@ const TIMEOUT = 120_000; const PRIVATE_KEY = process.env.PRIVATE_KEY as `0x${string}` | undefined; const MAIN_WALLET = PRIVATE_KEY ? privateKeyToAccount(PRIVATE_KEY) : undefined; +/** + * Whether {@linkcode createTempExchangeClient} can run, i.e. whether a funded `PRIVATE_KEY` is + * configured to pay for the throwaway account. + * + * Exported as a predicate rather than exporting `MAIN_WALLET` itself: every caller outside this + * module wants the guard, not the key. Harnesses in other API families that create a temporary + * account MUST skip on it — see the note on {@linkcode createTempExchangeClient}. + */ +export const CAN_FUND_TEMP_ACCOUNT: boolean = MAIN_WALLET !== undefined; + // ============================================================ // Preparation // ============================================================ @@ -97,11 +107,25 @@ export function runTest(options: { // Helpers // ============================================================ -/** Funds a throwaway testnet account and returns a client for it, optionally as a multi-sig user. */ +/** + * Funds a throwaway testnet account and returns a client for it, optionally as a multi-sig user. + * + * Requires a funded `PRIVATE_KEY`. Callers must skip on {@linkcode CAN_FUND_TEMP_ACCOUNT} rather + * than calling this without one — an unguarded call used to reach `ExchangeClient` with + * `MAIN_WALLET!` as `undefined`, where the non-null assertion silenced the type error and the + * test failed several frames later with `TypeError: wallet is not an Object`, which names + * neither the missing key nor the harness that forgot to check. + */ export async function createTempExchangeClient( type: "user" | "multisig", ): Promise> { - const mainExchClient = new ExchangeClient({ wallet: MAIN_WALLET!, transport }); + if (MAIN_WALLET === undefined) { + throw new Error( + "createTempExchangeClient() needs a funded PRIVATE_KEY to activate the throwaway account. " + + "Set PRIVATE_KEY, or skip the test on CAN_FUND_TEMP_ACCOUNT.", + ); + } + const mainExchClient = new ExchangeClient({ wallet: MAIN_WALLET, transport }); // Create temporary account const tempWallet = privateKeyToAccount(generatePrivateKey()); diff --git a/tests/api/subscription/_t.ts b/tests/api/subscription/_t.ts index 5f8bcd29..af7d8a77 100644 --- a/tests/api/subscription/_t.ts +++ b/tests/api/subscription/_t.ts @@ -14,7 +14,7 @@ import { } from "@bloxwap/hyperliquid"; import { OFFLINE } from "../../_offline.ts"; import { createTestContext, type TestContext } from "../../_testContext.ts"; -import { cleanupTempExchangeClient, createTempExchangeClient } from "../exchange/_t.ts"; +import { CAN_FUND_TEMP_ACCOUNT, cleanupTempExchangeClient, createTempExchangeClient } from "../exchange/_t.ts"; // ============================================================= // Arguments @@ -70,6 +70,14 @@ export function runTest(options: { /** * Runs a subscription test that also needs to place real actions from a temporary funded account. * + * Skipped without a funded `PRIVATE_KEY` as well as offline, matching + * `runTest` in `../exchange/_t.ts`: this harness funds a throwaway account through + * {@linkcode createTempExchangeClient}, so it has exactly the same precondition the exchange + * suite guards on. Before that guard existed these four tests did not skip — they ran, reached + * `ExchangeClient` with an `undefined` wallet, and failed with `TypeError: wallet is not an + * Object`, which is why an online run with no key reported four failures that looked like + * subscription bugs. + * * @param options Test options. * @param options.name Name of the subscription under test. * @param options.fn Test body; receives a test context plus subscription, exchange and info clients. @@ -87,7 +95,7 @@ export function runTestWithExchange(options: { }): void { const { name, fn } = options; - test.skipIf(OFFLINE)( + test.skipIf(OFFLINE || !CAN_FUND_TEMP_ACCOUNT)( name, async () => { await new Promise((r) => setTimeout(r, WAIT)); // delay to avoid rate limits diff --git a/tests/perf/README.md b/tests/perf/README.md index 0ffe4d93..676a1aa6 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -71,6 +71,22 @@ is where per-wallet lock scope shows up: | Pre-fix | 1 (per-wallet semaphore wraps signing **and** the request) | ~20 ms | | Post-fix (this tree)| ~100 | ~hundreds of µs | +> [!IMPORTANT] +> +> **Do not compare `order_100_concurrent`'s per-order figure against `order_sequential`.** It runs at +> `LATENCY_MS = 20` while `order_sequential` runs at 0, and the harness divides the whole burst's wall time by 100 — +> so **200 µs/order of the figure is amortized round trip** before any SDK cost is counted. The scenario now reports +> `latencyMs` and `rttPerOrderUs` in its `extra` column so the arithmetic is visible without opening the file. +> +> Read `order_100_concurrent` for `maxInFlight` — the wire-overlap guard it was written to be — and read +> **`transaction/order_100_concurrent_instant`** (same shape, 0 ms latency) for per-order SDK CPU. That one is +> directly comparable to `order_sequential`, and measures *lower*: concurrency is slightly cheaper per order, not 3× +> more expensive. Four independent audits have each "discovered" a phantom 200 µs regression here. +> +> `order_100_concurrent_instant` reports `maxInFlight=1`, which is also correct rather than a lock regression: a +> zero-latency transport resolves on a microtask, so each order settles before the next signature finishes. The calls +> are still concurrent through validate → lock → nonce → sign, which is what that scenario measures. + ### `subscription` — WebSocket fan-out cost `subscription/l2book_dispatch_50_coins` subscribes 50 coins on one channel, injects 500 frames for **one** coin, and diff --git a/tests/perf/_harness.ts b/tests/perf/_harness.ts index 5815897a..edfaee85 100644 --- a/tests/perf/_harness.ts +++ b/tests/perf/_harness.ts @@ -208,6 +208,20 @@ export async function runScenario(def: Scenario): Promise { const ctx = (await def.setup?.()) as never; // One sample: `iterations` calls, timed as a block, reduced to per-unit nanoseconds. + // + // Iterations are STRICTLY SEQUENTIAL — each `run` is awaited before the next begins — so a + // scenario's peak concurrency is whatever a single `run` body creates, and is 1 for any + // scenario whose body performs one request. That is a deliberate property (it isolates + // per-operation CPU from queueing effects), but it has a sharp consequence worth knowing + // before reading any result: **these scenarios cannot surface a cost that scales with the + // number of in-flight operations.** + // + // A real instance: the WebSocket dispatcher registered one abort listener per in-flight + // request on a single `AbortSignal` shared by the whole socket, which is O(n^2) across a + // burst. `transport/ws_request_round_trip` never saw it — at an in-flight count of 1 the + // defect is a ~300 ns constant — and it went unnoticed until a burst was measured outside + // the suite. A scenario that needs to catch that class of defect has to build the + // concurrency inside its own `run` body, the way `transaction/order_100_concurrent` does. const sample = async (): Promise<{ nsPerUnit: number; extra?: ExtraMetrics }> => { let extra: ExtraMetrics | undefined; const start = performance.now(); diff --git a/tests/perf/scenarios/transaction.ts b/tests/perf/scenarios/transaction.ts index f0cf44b3..c2fba46a 100644 --- a/tests/perf/scenarios/transaction.ts +++ b/tests/perf/scenarios/transaction.ts @@ -71,6 +71,24 @@ scenario({ // stays at 1. If the lock covers only nonce issuance and signing, requests overlap and // `maxInFlight` approaches 100. `maxInFlight` is reported so the shape of the win is // visible in the report, not just the wall time. +// +// ┌─ READ THIS BEFORE COMPARING THE NUMBER TO `order_sequential` ─────────────────────────┐ +// │ This scenario runs at LATENCY_MS = 20 while `order_sequential` runs at 0, and the │ +// │ harness divides the whole burst's wall time by CONCURRENT_ORDERS. One round trip │ +// │ cannot be amortized below itself, so 20 ms / 100 = **200 µs of pure waiting is │ +// │ charged to every order** before any SDK cost is counted. │ +// │ │ +// │ Wall time is `RTT + N x C`, where C is the serialized per-order CPU (secp256k1 is │ +// │ single-threaded, so the N signatures queue). At N=100 that predicts │ +// │ 20 ms + 100 x ~110 µs = ~31 ms, i.e. ~310 µs/order — which is what the suite reports. │ +// │ The gap to `order_sequential` is arithmetic, NOT lock or queue contention: with │ +// │ LATENCY_MS flipped to 0 the same scenario measures ~142 µs/order against │ +// │ `order_sequential`'s ~157 µs, so concurrency is measurably CHEAPER per order. │ +// │ │ +// │ Four separate audits have "discovered" a phantom 200 µs regression here. Compare │ +// │ against `order_100_concurrent_instant` below, which is the apples-to-apples number, │ +// │ and read this one for `maxInFlight` — the overlap guard it was written to be. │ +// └────────────────────────────────────────────────────────────────────────────────────────┘ const LATENCY_MS = 20; const CONCURRENT_ORDERS = 100; @@ -80,7 +98,8 @@ scenario({ group: "transaction", description: `${CONCURRENT_ORDERS} concurrent ExchangeClient.order() calls at ${LATENCY_MS} ms transport latency; ` + - `reports peak in-flight requests`, + `reports peak in-flight requests. ${(LATENCY_MS * 1000) / CONCURRENT_ORDERS} µs/order of the figure is ` + + `amortized RTT — see order_100_concurrent_instant for SDK CPU`, unit: "order", unitsPerIteration: CONCURRENT_ORDERS, iterations: 1, @@ -96,6 +115,47 @@ scenario({ Array.from({ length: CONCURRENT_ORDERS }, (_, i) => client.order({ orders: [order(i)], grouping: "na" })), ); + // `latencyMs` and `rttPerOrderUs` are reported so the report itself carries the arithmetic + // above — a reader who never opens this file still sees how much of the figure is waiting. + return { + maxInFlight: transport.maxInFlight, + calls: transport.calls.length, + latencyMs: LATENCY_MS, + rttPerOrderUs: Math.round((LATENCY_MS * 1000) / CONCURRENT_ORDERS), + }; + }, +}); + +// The apples-to-apples counterpart: identical shape, zero transport latency, so the figure is +// the SDK's own per-order CPU under 100-way concurrency and is directly comparable to +// `order_sequential`. Added alongside the 20 ms scenario rather than replacing it — the perf +// gate joins baselines by scenario name, so renaming or re-parameterizing the existing one +// would silently invalidate every recorded baseline. +// +// `maxInFlight` is 1 here, and that is correct rather than a lock regression: a zero-latency +// transport resolves on a microtask, so each order settles before the next one's signature +// finishes. The 100 calls are still genuinely concurrent through validate → lock → nonce → +// sign, which is the part this scenario measures; overlap ON THE WIRE is what the 20 ms +// sibling exists to assert, and only its `maxInFlight` carries that meaning. +scenario({ + name: "transaction/order_100_concurrent_instant", + group: "transaction", + description: + `${CONCURRENT_ORDERS} concurrent ExchangeClient.order() calls at 0 ms transport latency ` + + `(SDK CPU per order under concurrency; compare directly against order_sequential)`, + unit: "order", + unitsPerIteration: CONCURRENT_ORDERS, + iterations: 1, + samples: 5, + warmupSamples: 1, + run: async () => { + const transport = new MockExchangeTransport(0); + const client = new ExchangeClient({ transport, wallet: privateKeyToAccount(TEST_PRIVATE_KEY) }); + + await Promise.all( + Array.from({ length: CONCURRENT_ORDERS }, (_, i) => client.order({ orders: [order(i)], grouping: "na" })), + ); + return { maxInFlight: transport.maxInFlight, calls: transport.calls.length }; }, }); diff --git a/tests/perf/scenarios/user_account_channels.ts b/tests/perf/scenarios/user_account_channels.ts index 1ff9a4e2..6e09aca2 100644 --- a/tests/perf/scenarios/user_account_channels.ts +++ b/tests/perf/scenarios/user_account_channels.ts @@ -28,10 +28,10 @@ const POSITIONS = 10; const BALANCES = 20; /** * Users sharing the channel in the BY_USER routing scenario, in addition to the target user. - * The server tracks at most 15 unique users per connection (mirrored client-side by - * `MAX_UNIQUE_USERS`), so 14 + the target saturates the allowed crowd exactly. + * The server tracks at most 14 unique users per IP (mirrored client-side by + * `MAX_UNIQUE_USERS`), so 13 + the target saturates the allowed crowd exactly. */ -const USER_SUBSCRIPTIONS = 14; +const USER_SUBSCRIPTIONS = 13; const USER = "0x1111111111111111111111111111111111111111" as const; @@ -176,11 +176,15 @@ frameDispatchScenario("spotState", (client, listener) => client.spotState({ user frameDispatchScenario("webData3", (client, listener) => client.webData3({ user: USER }, listener)); +// The name says 15 because that is what the crowd was when the scenario was written and the +// perf gate joins baselines by scenario name — renaming it would silently invalidate every +// recorded baseline. The crowd is now `USER_SUBSCRIPTIONS + 1` = the real per-IP cap; the +// description below is the accurate figure. scenario({ name: "subscription/user_dispatch_15_users", group: "subscription", description: - `clearinghouseState dispatch with ${USER_SUBSCRIPTIONS} users on one channel; ` + + `clearinghouseState dispatch with ${USER_SUBSCRIPTIONS + 1} users on one channel (the per-IP cap); ` + `a frame for one user must run exactly one listener (BY_USER route)`, unit: "frame", unitsPerIteration: FRAMES, diff --git a/tests/transport/http/_rateLimiter.test.ts b/tests/transport/_rateLimiter.test.ts similarity index 99% rename from tests/transport/http/_rateLimiter.test.ts rename to tests/transport/_rateLimiter.test.ts index adff4087..4f8780f3 100644 --- a/tests/transport/http/_rateLimiter.test.ts +++ b/tests/transport/_rateLimiter.test.ts @@ -7,7 +7,7 @@ import { afterEach, beforeEach, describe, test } from "bun:test"; import { assert, assertEquals, assertRejects, assertThrows } from "@jsr/std__assert"; import { FakeTime } from "@jsr/std__testing/time"; -import { TokenBucketRateLimiter } from "../../../src/transport/http/_rateLimiter.ts"; +import { TokenBucketRateLimiter } from "../../src/transport/_rateLimiter.ts"; /** Waits until queued promise reactions have settled. */ async function flush(): Promise { diff --git a/tests/transport/websocket/_dispatcher.test.ts b/tests/transport/websocket/_dispatcher.test.ts index 0c2ecc09..1550aaa9 100644 --- a/tests/transport/websocket/_dispatcher.test.ts +++ b/tests/transport/websocket/_dispatcher.test.ts @@ -613,6 +613,74 @@ describe("WebSocketDispatcher", () => { assertEquals(err.cause, socket.terminationSignal.reason); }); + test("rejects every in-flight request when permanently closed, not just the first", async () => { + // The termination signal is fanned out from ONE listener to a Set of in-flight + // controllers, rather than relayed per request. A fan-out that aborts while iterating + // its own backing set — each abort runs a `finally` that deletes from it — drops + // requests; this asserts all of them settle. + const { socket, requester } = createRequester(); + + const promises = Array.from({ length: 50 }, (_, i) => + assertRejects( + () => requester.request("post", { seq: i }), + WebSocketRequestError, + "WebSocket connection permanently terminated", + ), + ); + socket.terminate(new Error("Permanently closed")); + + await Promise.all(promises); + }); + + test("a caller's own abort reason outranks the socket's when both are already aborted", async () => { + // Reason precedence. `relay([signal, terminationSignal])` aborted with the first + // already-aborted source in argument order, so the caller's reason won. Splitting the + // relay from the termination branch must preserve that: gating the termination branch on + // the controller still being unaborted is what keeps the caller's reason on top. Reversing + // the two silently changes the error a caller sees — and nothing else in this suite covers it. + const { socket, requester } = createRequester(); + + socket.terminate(new Error("Permanently closed")); + const controller = new AbortController(); + controller.abort(new Error("Caller gave up first")); + + const err = await assertRejects( + () => requester.request("post", { foo: "bar" }, controller.signal), + WebSocketRequestError, + "Request aborted", + ); + assertEquals((err.cause as Error).message, "Caller gave up first"); + }); + + test("the socket's reason is used when only the socket is already aborted", async () => { + const { socket, requester } = createRequester(); + + socket.terminate(new Error("Permanently closed")); + const controller = new AbortController(); // live, never aborted + + const err = await assertRejects( + () => requester.request("post", { foo: "bar" }, controller.signal), + WebSocketRequestError, + "WebSocket connection permanently terminated", + ); + assertEquals(err.cause, socket.terminationSignal.reason); + }); + + test("a settled request is not aborted by a later termination", async () => { + // The `finally` must remove the controller from the fan-out set; otherwise a terminate + // after the response would abort an already-resolved request's controller. + const { socket, requester } = createRequester(); + + const promise = requester.request("post", { foo: "bar" }); + socket.mockMessage(RESPONSES.info(1, { ok: true })); + await promise; + + socket.terminate(new Error("Permanently closed")); + await drain(); + // Resolving twice or rejecting after resolve would surface as an unhandled rejection. + assertEquals(await promise, { ok: true }); + }); + describe("AbortSignal", () => { test("rejects if aborted before call", async () => { const { requester } = createRequester(); diff --git a/tests/transport/websocket/_quota.test.ts b/tests/transport/websocket/_quota.test.ts new file mode 100644 index 00000000..69bb670d --- /dev/null +++ b/tests/transport/websocket/_quota.test.ts @@ -0,0 +1,294 @@ +/** + * 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`. + * @module + */ + +import { describe, test } from "bun:test"; +import { assert, assertEquals, assertRejects } from "@jsr/std__assert"; +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"; +import { sharedWebSocketQuota, WebSocketQuota } from "../../../src/transport/websocket/_quota.ts"; +import { WebSocketSubscriptionManager } from "../../../src/transport/websocket/_subscriptionManager.ts"; +import { drain, MockWebSocket, RESPONSES } from "./_mock.ts"; + +// ============================================================================= +// Helpers +// ============================================================================= + +/** A manager over a mock socket, optionally sharing an explicit quota. */ +function createManager(quota?: WebSocketQuota): { socket: MockWebSocket; manager: WebSocketSubscriptionManager } { + const socket = new MockWebSocket() as ReconnectingWebSocket & MockWebSocket; + const hlEvents = new HyperliquidEventTarget(socket); + const dispatcher = new WebSocketDispatcher(socket, hlEvents, 10_000, quota); + const manager = new WebSocketSubscriptionManager(socket, dispatcher, hlEvents, true, quota); + return { socket, manager }; +} + +/** + * Subscribes and confirms it against the mock server, so the reservation settles. + * + * The `drain()` before the response is required once pacing is on: `subscribe` awaits its + * token before the frame reaches the socket, so a confirmation sent in the same tick would + * arrive before the request it confirms and never match. + */ +async function subscribeConfirmed( + socket: MockWebSocket, + manager: WebSocketSubscriptionManager, + payload: Record, +): Promise<{ unsubscribe: () => Promise }> { + const pending = manager.subscribe("test", payload, () => {}); + await drain(); + socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); + return await pending; +} + +/** Unsubscribes and answers the wire `unsubscribe` request the manager sends. */ +async function unsubscribeConfirmed( + socket: MockWebSocket, + sub: { unsubscribe: () => Promise }, + payload: Record, +): Promise { + const pending = sub.unsubscribe(); + await drain(); + socket.mockMessage(RESPONSES.subscriptionResponse("unsubscribe", payload)); + await pending; +} + +// ============================================================================= +// Tests +// ============================================================================= + +describe("WebSocketQuota", () => { + describe("subscription budget", () => { + test("counts reservations and releases them", () => { + const quota = new WebSocketQuota(); + + assertEquals(quota.reserveSubscription(undefined), undefined); + assertEquals(quota.subscriptions, 1); + quota.releaseSubscription(undefined); + assertEquals(quota.subscriptions, 0); + }); + + test("refuses past maxSubscriptions, naming the limit that refused", () => { + const quota = new WebSocketQuota({ maxSubscriptions: 2 }); + + assertEquals(quota.reserveSubscription(undefined), undefined); + assertEquals(quota.reserveSubscription(undefined), undefined); + assertEquals(quota.reserveSubscription(undefined), "subscriptions"); + // A refused reservation must not consume a slot. + assertEquals(quota.subscriptions, 2); + }); + + test("several subscriptions for one user cost one user slot", () => { + const quota = new WebSocketQuota({ maxUniqueUsers: 1 }); + + assertEquals(quota.reserveSubscription("0xaaa"), undefined); + assertEquals(quota.reserveSubscription("0xaaa"), undefined); + assertEquals(quota.uniqueUsers, 1); + // A second distinct user is one too many. + assertEquals(quota.reserveSubscription("0xbbb"), "users"); + // The slot is freed only once the last subscription for that user goes. + quota.releaseSubscription("0xaaa"); + assertEquals(quota.uniqueUsers, 1); + quota.releaseSubscription("0xaaa"); + assertEquals(quota.uniqueUsers, 0); + assertEquals(quota.reserveSubscription("0xbbb"), undefined); + }); + + test("null disables a guard", () => { + const quota = new WebSocketQuota({ maxSubscriptions: null, maxUniqueUsers: null }); + for (let i = 0; i < 2_000; i++) assertEquals(quota.reserveSubscription(`0x${i}`), undefined); + assertEquals(quota.subscriptions, 2_000); + }); + }); + + describe("sharing across connections", () => { + test("two managers on one quota share the subscription budget", async () => { + // The regression this whole module exists for: before the quota, each manager kept its + // own count and two transports admitted 2x the limit against a budget the server + // scopes per IP. + const quota = new WebSocketQuota({ maxSubscriptions: 2 }); + const a = createManager(quota); + const b = createManager(quota); + + await subscribeConfirmed(a.socket, a.manager, { channel: "one" }); + await subscribeConfirmed(b.socket, b.manager, { channel: "two" }); + assertEquals(quota.subscriptions, 2); + + // The third subscription is refused client-side even though it is the first on a + // third connection — the server would have refused it without echoing the request. + await assertRejects( + () => b.manager.subscribe("test", { channel: "three" }, () => {}), + WebSocketRequestError, + "Cannot subscribe to more than 2 channels.", + ); + }); + + test("two managers on one quota share the unique-user budget", async () => { + const quota = new WebSocketQuota({ maxUniqueUsers: 1 }); + const a = createManager(quota); + const b = createManager(quota); + + await subscribeConfirmed(a.socket, a.manager, { channel: "c", user: "0xAAA" }); + await assertRejects( + () => b.manager.subscribe("test", { channel: "c", user: "0xBBB" }, () => {}), + WebSocketRequestError, + "Cannot track more than 1 total users.", + ); + }); + + test("unsubscribing on one connection frees the slot for another", async () => { + const quota = new WebSocketQuota({ maxSubscriptions: 1 }); + const a = createManager(quota); + const b = createManager(quota); + + const sub = await subscribeConfirmed(a.socket, a.manager, { channel: "one" }); + await assertRejects(() => b.manager.subscribe("test", { channel: "two" }, () => {}), WebSocketRequestError); + + await unsubscribeConfirmed(a.socket, sub, { channel: "one" }); + assertEquals(quota.subscriptions, 0); + await subscribeConfirmed(b.socket, b.manager, { channel: "two" }); + assertEquals(quota.subscriptions, 1); + }); + + test("a permanently terminated connection returns its slots", async () => { + const quota = new WebSocketQuota({ maxSubscriptions: 1 }); + const a = createManager(quota); + const b = createManager(quota); + + await subscribeConfirmed(a.socket, a.manager, { channel: "one" }); + assertEquals(quota.subscriptions, 1); + + // Terminal close fails every subscription, which is exactly when the server frees them. + a.socket.terminate(); + assertEquals(quota.subscriptions, 0); + await subscribeConfirmed(b.socket, b.manager, { channel: "two" }); + assertEquals(quota.subscriptions, 1); + }); + + test("a refused subscription leaves the budget untouched", async () => { + const quota = new WebSocketQuota({ maxSubscriptions: 1 }); + const { socket, manager } = createManager(quota); + + await subscribeConfirmed(socket, manager, { channel: "one" }); + await assertRejects(() => manager.subscribe("test", { channel: "two" }, () => {}), WebSocketRequestError); + assertEquals(quota.subscriptions, 1); + }); + + test("sharedWebSocketQuota keys by network", () => { + const mainnet = sharedWebSocketQuota(false); + const testnet = sharedWebSocketQuota(true); + + // Every transport on one network draws from one budget, which is the whole point. + assert(sharedWebSocketQuota(false) === mainnet); + assert(sharedWebSocketQuota(true) === testnet); + // Mainnet and testnet are different servers keeping different per-IP counters. + assert(mainnet !== testnet); + }); + }); + + describe("outbound message budget", () => { + test("accounting-only by default: nothing waits", () => { + const quota = new WebSocketQuota(); + // `undefined` rather than a resolved promise, so the caller can stay synchronous. + assertEquals(quota.acquireSend(), undefined); + quota.chargeSend(); + }); + + 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 } }); + const { socket, manager } = createManager(quota); + + await subscribeConfirmed(socket, manager, { channel: "one" }); + assertEquals(socket.sentMessages.length, 1); + + // Deliberately left parked: this call is never confirmed, so it must be caught here. + // A floating rejection outlives this file — the parked subscribe eventually sends, waits + // out its 10 s request timeout, and rejects long after the test that created it has + // passed, surfacing as an "Unhandled error between tests" against whichever file is + // running by then and failing the whole suite with 0 reported failures. + const parked = manager.subscribe("test", { channel: "two" }, () => {}); + parked.catch(() => {}); + + await drain(); + // Still parked in the bucket: the frame has not reached the socket. + assertEquals(socket.sentMessages.length, 1); + + // Terminate rather than leaving the request pending: the rejection lands now, on the + // handler above, instead of on a timer after this test returns. + socket.terminate(); + await parked.catch(() => {}); + }); + + test("post never waits, even with an empty bucket", async () => { + // The load-bearing guarantee: `_shell.ts` fixes the wire order of an exchange action on + // `transport.request` reaching `send` synchronously. If pacing ever awaited on the post + // path, a later nonce could overtake an earlier one. + const quota = new WebSocketQuota({ rateLimit: { capacity: 0.5, refillPerMinute: 1 } }); + const socket = new MockWebSocket() as ReconnectingWebSocket & MockWebSocket; + const hlEvents = new HyperliquidEventTarget(socket); + const dispatcher = new WebSocketDispatcher(socket, hlEvents, 10_000, quota); + + // Not awaited and no microtask drained: the frame must already be on the socket. + dispatcher.request("post", { type: "action" }).catch(() => {}); + assertEquals(socket.sentMessages.length, 1); + + // And a burst stays in issue order, which is what nonce ordering depends on. + for (let i = 0; i < 20; i++) dispatcher.request("post", { seq: i }).catch(() => {}); + const seqs = socket.sentMessages + .slice(1) + .map((frame) => (JSON.parse(frame) as { request: { seq: number } }).request.seq); + assertEquals( + seqs, + Array.from({ length: 20 }, (_, i) => i), + ); + + // Settle the 21 pending posts now; otherwise each carries a live 10 s request timeout + // into the rest of the run. + socket.terminate(); + }); + + test("posts spend the budget subscribes then wait off", async () => { + // One socket, one quota, both a dispatcher and a manager on it — the production shape. + const quota = new WebSocketQuota({ rateLimit: { capacity: 2, refillPerMinute: 60 } }); + const socket = new MockWebSocket() as ReconnectingWebSocket & MockWebSocket; + const hlEvents = new HyperliquidEventTarget(socket); + const dispatcher = new WebSocketDispatcher(socket, hlEvents, 10_000, quota); + const manager = new WebSocketSubscriptionManager(socket, dispatcher, hlEvents, true, quota); + + // Two posts drain the bucket without waiting... + dispatcher.request("post", { a: 1 }).catch(() => {}); + dispatcher.request("post", { a: 2 }).catch(() => {}); + const afterPosts = socket.sentMessages.length; + assertEquals(afterPosts, 2); + + // ...so the next subscribe has to wait for a refill. + manager.subscribe("test", { channel: "one" }, () => {}).catch(() => {}); + await drain(); + assertEquals(socket.sentMessages.length, afterPosts); + + socket.terminate(); // settle the parked subscribe and the two posts before returning + }); + + test("an aborted wait never sends and never spends a token", async () => { + const quota = new WebSocketQuota({ rateLimit: { capacity: 1, refillPerMinute: 60 } }); + const { socket, manager } = createManager(quota); + + await subscribeConfirmed(socket, manager, { channel: "one" }); + const sent = socket.sentMessages.length; + + const controller = new AbortController(); + const pending = manager.subscribe("test", { channel: "two" }, () => {}, { signal: controller.signal }); + await drain(); + controller.abort(); + + await assertRejects(() => pending, WebSocketRequestError); + assertEquals(socket.sentMessages.length, sent); + // The abandoned reservation is returned rather than stranded. + assertEquals(quota.subscriptions, 1); + }); + }); +}); diff --git a/tests/transport/websocket/_subscriptionManager.test.ts b/tests/transport/websocket/_subscriptionManager.test.ts index 88e7e298..5e53efea 100644 --- a/tests/transport/websocket/_subscriptionManager.test.ts +++ b/tests/transport/websocket/_subscriptionManager.test.ts @@ -276,19 +276,19 @@ describe("WebSocketSubscriptionManager", () => { test("limit errors carry the request payload", async () => { const { socket, manager } = createManager(); - for (let i = 0; i < 15; i++) { + for (let i = 0; i < 14; i++) { const payload = { type: "userEvents", user: `0x${i.toString().padStart(40, "0")}` }; const promise = manager.subscribe("userEvents", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); await promise; } - const payload16 = { type: "userEvents", user: "0x000000000000000000000000000000000000000f" }; + const payload15 = { type: "userEvents", user: "0x000000000000000000000000000000000000000f" }; const err = await assertRejects( - () => manager.subscribe("userEvents", payload16, () => {}), + () => manager.subscribe("userEvents", payload15, () => {}), WebSocketRequestError, ); - assertEquals(err.request, payload16); + assertEquals(err.request, payload15); }); }); @@ -1041,21 +1041,24 @@ describe("WebSocketSubscriptionManager", () => { }); describe("unique user subscription limit", () => { - test("rejects when exceeding 15 unique users", async () => { + // 14, not 15: a 2026-08-02 mainnet probe had the server refuse the 15th distinct user on + // two independent connections, despite its own error frame saying "more than 15". See + // MAX_UNIQUE_USERS in _quota.ts. + test("rejects when exceeding 14 unique users", async () => { const { socket, manager } = createManager(); - for (let i = 0; i < 15; i++) { + for (let i = 0; i < 14; i++) { const payload = { type: "userEvents", user: `0x${i.toString().padStart(40, "0")}` }; const promise = manager.subscribe("userEvents", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); await promise; } - const payload16 = { type: "userEvents", user: "0x000000000000000000000000000000000000000f" }; + const payload15 = { type: "userEvents", user: "0x000000000000000000000000000000000000000f" }; await assertRejects( - () => manager.subscribe("userEvents", payload16, () => {}), + () => manager.subscribe("userEvents", payload15, () => {}), WebSocketRequestError, - "Cannot track more than 15 total users.", + "Cannot track more than 14 total users.", ); }); @@ -1063,7 +1066,7 @@ describe("WebSocketSubscriptionManager", () => { const { socket, manager } = createManager(); const subs = []; - for (let i = 0; i < 15; i++) { + for (let i = 0; i < 14; i++) { const payload = { type: "userEvents", user: `0x${i.toString().padStart(40, "0")}` }; const promise = manager.subscribe("userEvents", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); @@ -1079,7 +1082,8 @@ describe("WebSocketSubscriptionManager", () => { socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", newUserPayload)); await promise; - assertEquals(manager._subscriptions.size, 15); + // 14 subscribed, one unsubscribed, one added back into the freed slot. + assertEquals(manager._subscriptions.size, 14); }); test("does not count subscriptions without user parameter", async () => { @@ -1101,14 +1105,14 @@ describe("WebSocketSubscriptionManager", () => { test("allows a new channel of an already tracked user at the limit", async () => { const { socket, manager } = createManager(); - for (let i = 0; i < 15; i++) { + for (let i = 0; i < 14; i++) { const payload = { type: "userEvents", user: `0x${i.toString().padStart(40, "0")}` }; const promise = manager.subscribe("userEvents", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); await promise; } - // A new channel of user 0 does not add a 16th user. + // A new channel of user 0 does not add a 15th user. const payload = { type: "userFills", user: `0x${"0".padStart(40, "0")}` }; const promise = manager.subscribe("userFills", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); @@ -1119,14 +1123,14 @@ describe("WebSocketSubscriptionManager", () => { const { socket, manager } = createManager(); const mixedCase = "0x00000000000000000000000000000000000000AB"; - for (const user of [mixedCase, ...Array.from({ length: 14 }, (_, i) => `0x${`${i}`.padStart(40, "0")}`)]) { + for (const user of [mixedCase, ...Array.from({ length: 13 }, (_, i) => `0x${`${i}`.padStart(40, "0")}`)]) { const payload = { type: "userEvents", user }; const promise = manager.subscribe("userEvents", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); await promise; } - // The same address in a different case is not a 16th user. + // The same address in a different case is not a 15th user. const payload = { type: "userFills", user: mixedCase.toLowerCase() }; const promise = manager.subscribe("userFills", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); @@ -1136,7 +1140,7 @@ describe("WebSocketSubscriptionManager", () => { test("allows multiple listeners on same user subscription", async () => { const { socket, manager } = createManager(); - for (let i = 0; i < 15; i++) { + for (let i = 0; i < 14; i++) { const payload = { type: "userEvents", user: `0x${i.toString().padStart(40, "0")}` }; const promise = manager.subscribe("userEvents", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); @@ -1171,7 +1175,7 @@ describe("WebSocketSubscriptionManager", () => { const { socket, manager } = createManager(); const subs = []; - for (let i = 0; i < 15; i++) { + for (let i = 0; i < 14; i++) { const payload = { type: "userEvents", user: `0x${i.toString().padStart(40, "0")}` }; const promise = manager.subscribe("userEvents", payload, () => {}); socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); @@ -1196,9 +1200,9 @@ describe("WebSocketSubscriptionManager", () => { await unsubPromise; // The original user's slot was released: a 16th unique user fits again. - const payload16 = { type: "userEvents", user: "0x000000000000000000000000000000000000000f" }; - const promise = manager.subscribe("userEvents", payload16, () => {}); - socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload16)); + const payload15 = { type: "userEvents", user: "0x000000000000000000000000000000000000000f" }; + const promise = manager.subscribe("userEvents", payload15, () => {}); + socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload15)); await promise; }); });