From 2d63ae9a1e383d39a17b2c5dc36bce5fce4299f9 Mon Sep 17 00:00:00 2001 From: Joe Blau Date: Sun, 2 Aug 2026 16:04:01 -0700 Subject: [PATCH] perf: per-IP WebSocket quota, O(n) termination fan-out, narrow entrypoints (release 0.2.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four models diagnosed this SDK independently and every candidate finding was adversarially verified by a skeptic who reproduced it from scratch. This lands the findings that survived. Minor, not patch, per docs/README.md#versioning: two changes alter the behaviour of working 0.1.6 programs. Both convert a silent server-side failure into an immediate local one — the server already refused this traffic, as a 10 s timeout with no echoed request to match it to. WebSocket per-IP quota (src/transport/websocket/_quota.ts) Hyperliquid scopes every documented WebSocket limit to the client IP, not the connection, and says so for two of them ("across all websocket connections"). Subscription and unique-user budgets were tracked per manager, so N transports admitted N x 1000 subscriptions against a limit of 1000. They now share one quota per network by default; pass `quota` to opt out. MAX_UNIQUE_USERS is 14, not 15. A live mainnet probe on 2026-08-02 subscribed distinct users one at a time with the 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 own message states, so the previous value let the 15th subscription through to be dropped without an echo — the exact failure the guard exists to prevent. The same probe established the scope: a second connection was refused a 15th distinct user while still being allowed one the first connection already held. Also adds an opt-in outbound message limiter for the documented 2000/min budget. It paces subscribe/unsubscribe only: `post` frames and keep-alive pings debit the budget without ever waiting, because _shell.ts fixes an exchange action's wire order on transport.request reaching send synchronously, and delaying the watchdog is how a half-open socket goes unnoticed. O(n) termination fan-out (src/transport/websocket/_dispatcher.ts) Every request relayed the socket's single shared terminationSignal, putting one listener per in-flight request on one AbortSignal. EventTarget scans that list linearly on both add and remove, so a burst was O(n^2): measured at 195 ns per add/remove pair with the list empty and 13.1 us with 5000 resident on Bun. One listener plus a Set makes it O(1). Measured 2000 in-flight requests 9.2-10.3 ms -> 4.4-5.0 ms (-50%, 5/5 interleaved pairs, no overlap), and a real 1000-subscription reconnect -18.9%. Reason precedence is preserved deliberately: the caller's signal is relayed first and the termination branch is gated on the controller still being unaborted, so a caller's own abort still outranks the socket's when both are already aborted. A mutant with those two reversed fails exactly the new test for it and nothing else. Narrow entrypoints (package.json) and the /utils import graph The root barrel evaluates all four clients plus both transports, and ./transport was not exported at all. Seven additive keys let an info-only consumer enter through ./api/info/client + ./transport: 69.5 -> 41.0 ms on Node, 22.0 -> 8.3 ms on Bun. src/utils/_symbolConverter.ts imported four functions from api/info/mod.ts — the only value import of that barrel in src/ — dragging 80+ method modules into @bloxwap/hyperliquid/utils. Importing the four _methods/* modules directly takes the built /utils entry from 91 to 10 modules: 23.0-31.5 -> 6.2-6.3 ms on Node, 6.2-8.2 -> 3.5-3.7 ms on Bun. .dev/import_graph_check.ts gates it; run against the previous tree it reports OVER 90/20 and exits 1. Perf suite corrections order_100_concurrent runs at 20 ms latency while order_sequential runs at 0, and the harness divides burst wall time by 100 — so 200 us/order of that figure is amortized RTT, and four separate audits each "found" a phantom regression in it. Adds order_100_concurrent_instant (same shape, 0 ms) which measures 106 us against order_sequential's 111: concurrency is cheaper, not 3x worse. Added as a sibling, not a rename, so the gate's join key survives. The 20 ms scenario now reports latencyMs and rttPerOrderUs. Documents the harness's strictly-sequential execution, whose peak in-flight of 1 is why the dispatcher's O(n^2) was invisible to ws_request_round_trip. Adds a --record warning for baseline entries above 15% rme; the committed baseline has nine, including the 41% rme entry that generated a phantom 7.7 us "validation cost" three audits then spent effort refuting. Test harness runTestWithExchange gated only on OFFLINE while funding a throwaway account, so without PRIVATE_KEY four subscription tests reached ExchangeClient with an undefined wallet (a `MAIN_WALLET!` assertion hid it) and failed with "TypeError: wallet is not an Object" instead of skipping. Now gated on CAN_FUND_TEMP_ACCOUNT, matching the exchange harness. Note: bun run perf:gate fails closed on this commit because the suite fingerprint changed. That is the documented behaviour for an intentional tests/perf edit and needs an explicit merge decision; there is no override flag. Interleaved runs show no regression (HEAD 0/0, this tree 1/0 across two rounds). Co-Authored-By: Claude Opus 5 (1M context) --- .dev/import_graph_check.ts | 136 ++++++++ .dev/perf/gate.ts | 31 ++ docs/clients.md | 59 +++- docs/reference/known-drift.md | 45 ++- docs/transports.md | 53 ++++ package.json | 16 +- src/api/exchange/client.ts | 21 ++ src/transport/{http => }/_rateLimiter.ts | 11 +- src/transport/http/mod.ts | 2 +- src/transport/websocket/_dispatcher.ts | 94 +++++- src/transport/websocket/_keepAlive.ts | 19 +- src/transport/websocket/_quota.ts | 286 +++++++++++++++++ .../websocket/_subscriptionManager.ts | 87 ++---- src/transport/websocket/mod.ts | 36 ++- src/utils/_symbolConverter.ts | 26 +- tests/api/exchange/_t.ts | 28 +- tests/api/subscription/_t.ts | 12 +- tests/perf/README.md | 16 + tests/perf/_harness.ts | 14 + tests/perf/scenarios/transaction.ts | 62 +++- tests/perf/scenarios/user_account_channels.ts | 12 +- .../transport/{http => }/_rateLimiter.test.ts | 2 +- tests/transport/websocket/_dispatcher.test.ts | 68 ++++ tests/transport/websocket/_quota.test.ts | 294 ++++++++++++++++++ .../websocket/_subscriptionManager.test.ts | 44 +-- 25 files changed, 1351 insertions(+), 123 deletions(-) create mode 100644 .dev/import_graph_check.ts rename src/transport/{http => }/_rateLimiter.ts (95%) create mode 100644 src/transport/websocket/_quota.ts rename tests/transport/{http => }/_rateLimiter.test.ts (99%) create mode 100644 tests/transport/websocket/_quota.test.ts 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; }); });