diff --git a/apps/docs/content/docs/clients.md b/apps/docs/content/docs/clients.md index a8d24523..7afbb446 100644 --- a/apps/docs/content/docs/clients.md +++ b/apps/docs/content/docs/clients.md @@ -818,3 +818,56 @@ exchange rejects a nonce it has already seen and the action cannot be applied tw The standalone `signAction`, `submitAction`, and `executeAction` functions in `@bloxwap/hyperliquid/actions/execution` accept the same exchange config. Existing client methods and `prepareRequest` / `submitPrepared` continue to work. + +## Bounded dispatch for remote signing + +By default the SDK signs concurrently but sends a signer's requests in nonce-issuance order, so one slow remote +signature delays every later request for that signer and network. The exchange does not require that order: it accepts +any unused nonce above the smallest of the signer's 100 highest nonces. A dedicated signer can opt into bounded +dispatch, which sends ready signatures ahead of a slower earlier one: + +```ts +const exchange = new ExchangeClient({ + transport, + wallet, + dispatchPolicy: { mode: "bounded", maxOvertakes: 99, maxPending: 1000 }, +}); +``` + +- `maxOvertakes` (0–99, default 99) caps how many newer requests may be sent while an earlier one is still + outstanding. The count is cumulative until the earlier request's response settles, so newer requests that finish + quickly cannot keep overtaking it and push its nonce out of the exchange's window, even if requests arrive in a + different order than they were sent. Among signed requests waiting for the window, the oldest goes first. `0` + reproduces ordered dispatch. +- `maxPending` (default 1000) caps signing, waiting, and in-flight requests. A call beyond it rejects with + `Bounded dispatch queue is full` without consuming a nonce. +- Before sending, a request that waited is checked again against the protocol timestamp window (its nonce must be + within two days before and one day after the current time) and its `expiresAfter`. A failed check rejects the call + without sending it. +- A wallet rejection, an abort, an expiry, or a transport failure (including a closed WebSocket) settles the call and + frees its slot; the skipped nonce is a gap the exchange tolerates. A signature that completes after its call was + aborted is never sent. A signature that never completes holds back further overtaking until it is aborted, so pass a + `signal` with a deadline when the signer can stall. +- WebSocket and HTTP transport limits are unchanged and apply separately from the nonce bound: `post` frames are + still charged to the shared WebSocket message budget without waiting for it. + +Bounded dispatch pays off when requests arrive steadily and some signatures are much slower than others: ready +requests no longer queue behind a slow one. It is not a throughput setting. A burst much larger than `maxOvertakes` +can finish later than under ordered dispatch, because ordered dispatch assumes requests arrive in the order they were +sent and keeps all of them in flight, while bounded dispatch counts each in-flight request against the window until +its response returns. Measure with your own signer and traffic before switching. + +The coordinator is shared per signer address and network across every `ExchangeClient` and standalone function in +the process, HTTP and WebSocket alike, and so is the default nonce source. Multi-sig requests are coordinated by the +leader signer, the same key their nonces are issued under; vault and sub-account requests share their signer's lane. + +Bounded mode supports managed calls only: ordinary exchange methods, `exchange.execute`, and `executeAction`. +`exchange.sign` / `exchange.submit`, `signAction` / `submitAction`, `prepareRequest`, and `submitPrepared` reject in +bounded mode, and also reject while bounded calls for that signer and network are outstanding, because a detached +submission's nonce would escape the bound. Ordered and bounded calls, or bounded calls with different limits, cannot be +active for the same signer and network at the same time. + +The coordinator only sees requests made through this process. Other processes, other SDK instances bundled separately, +direct `transport.request` calls, and anything that submits for the same signer elsewhere also consume nonces in its +100-highest window, and the bound cannot account for them. Use bounded dispatch only with a signer (typically an API +wallet) dedicated to this process. diff --git a/apps/docs/content/docs/transports.md b/apps/docs/content/docs/transports.md index de813513..b7bfa621 100644 --- a/apps/docs/content/docs/transports.md +++ b/apps/docs/content/docs/transports.md @@ -365,9 +365,11 @@ Outbound messages are **paced by default**: the shared quota runs a token bucket re-subscribes every held subscription at once, so 1000 subscriptions spend half the minute's budget instantly, and a flapping socket repeats the burst until the server refuses. -Pacing only ever delays `subscribe` and `unsubscribe` frames. **`post` requests and keep-alive pings never wait**: an -exchange action's wire order — and therefore per-wallet nonce ordering — depends on reaching the socket synchronously, -and delaying the keep-alive watchdog is how a half-open connection goes unnoticed. Both still *debit* the budget, so a +Pacing only ever delays `subscribe` and `unsubscribe` frames. **`post` requests and keep-alive pings never wait**: the +SDK's default ordered dispatch policy fixes an exchange action's wire order by reaching the socket synchronously (the +exchange itself only requires each nonce to be unused and within the signer's 100-highest window; see +[bounded dispatch](clients.md#bounded-dispatch-for-remote-signing)), and delaying the keep-alive watchdog is how a +half-open connection goes unnoticed. Both still *debit* the budget, so a burst of orders correctly slows subscription traffic rather than silently overrunning the shared limit. To opt out of pacing — or to resize the bucket — pass your own `quota`; constructed without `rateLimit`, it keeps the diff --git a/packages/hyperliquid/.dev/perf/remote_signing.ts b/packages/hyperliquid/.dev/perf/remote_signing.ts new file mode 100644 index 00000000..c2c333bb --- /dev/null +++ b/packages/hyperliquid/.dev/perf/remote_signing.ts @@ -0,0 +1,174 @@ +/** + * Ordered versus bounded dispatch under variable remote-signing latency. + * bun .dev/perf/remote_signing.ts [--filter stall] [--out /tmp/remote-signing.json] + * + * Orders for one signer arrive either open-loop, one per millisecond, or as a single burst. Each signature takes a seeded random + * latency from the scenario's profile, standing in for an `eth_signTypedData_v4` round trip to a + * remote wallet, and each request spends a fixed time on a mock transport. Reported per mode: + * - caller latency p50/p99, from arrival to settled response; + * - throughput, orders over the time from the first arrival to the last response; + * - outstanding work: peak signatures in progress, peak signed requests waiting to dispatch, and + * peak requests in flight on the transport. + * + * Both modes see the same arrival and latency sequence; modes alternate inside every round so + * machine drift hits each one equally, and each metric is the median over rounds. Numbers are + * wall-clock timings with real timers and mocks, not production latency. + * @module + */ +import { writeFile } from "node:fs/promises"; +import type { DispatchPolicy } from "../../src/api/exchange/_methods/_base/_dispatch.ts"; +import type { ExchangeConfig } from "../../src/api/exchange/_methods/_base/_config.ts"; +import { buildOrder } from "../../src/api/exchange/_methods/order.ts"; +import { executeAction } from "../../src/actions/execution.ts"; +import { delay } from "../../src/transport/runtime.ts"; + +const filterIndex = process.argv.indexOf("--filter"); +const filter = filterIndex >= 0 ? process.argv[filterIndex + 1] : undefined; +const outIndex = process.argv.indexOf("--out"); +const rounds = 5; +const orders = 300; +/** Gap between arrivals: `1` is open-loop at 1000 orders/s, `0` submits every order at once. */ +const arrivals = [1, 0]; +const transportMs = 5; +const signature = `0x${"11".repeat(64)}1b` as const; +const action = buildOrder({ + orders: [{ a: 0, b: true, p: "30000", s: "0.1", r: false, t: { limit: { tif: "Gtc" } } }], +}); +const response = { status: "ok", response: { type: "order", data: { statuses: [{ resting: { oid: 1 } }] } } }; + +/** Signing-latency profiles, as a function of a uniform random value and the order index. */ +const scenarios: Record number> = { + // One slow signature at the start, then 2-10 ms with an occasional 50 ms. + stall: (u, i) => (i === 0 ? 200 : i % 25 === 0 ? 50 : 2 + u * 8), + // Uniform 1-30 ms. + jitter: (u) => 1 + u * 29, + // 90% at 2 ms, 10% at 100 ms. + heavy_tail: (u) => (u < 0.9 ? 2 : 100), +}; + +const policies: Record = { + ordered: "ordered", + bounded: { mode: "bounded", maxOvertakes: 99 }, +}; + +/** Deterministic PRNG (mulberry32), so both modes see the same latency sequence. */ +function random(seed: number): () => number { + return () => { + seed = (seed + 0x6d2b79f5) | 0; + let t = Math.imul(seed ^ (seed >>> 15), 1 | seed); + t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +function percentile(values: number[], p: number): number { + const sorted = [...values].sort((a, b) => a - b); + return sorted[Math.min(sorted.length - 1, Math.ceil(sorted.length * p) - 1)]; +} + +const median = (values: number[]): number => percentile(values, 0.5); + +let walletId = 0x5000; + +interface Sample { + p50Ms: number; + p99Ms: number; + ordersPerSecond: number; + peakSigning: number; + peakWaiting: number; + peakInFlight: number; +} + +async function run( + profile: (u: number, i: number) => number, + policy: DispatchPolicy, + arrivalEveryMs: number, + seed: number, +): Promise { + const next = random(seed); + const latencies = Array.from({ length: orders }, (_, i) => profile(next(), i)); + let signing = 0; + let waiting = 0; + let inFlight = 0; + let peakSigning = 0; + let peakWaiting = 0; + let peakInFlight = 0; + let signed = 0; + let lastNonce = 0; + const config: ExchangeConfig = { + wallet: { + address: `0x${(++walletId).toString(16).padStart(40, "0")}`, + async signTypedData(): Promise<`0x${string}`> { + const i = signed++; + peakSigning = Math.max(peakSigning, ++signing); + await delay(latencies[i]); + signing--; + peakWaiting = Math.max(peakWaiting, ++waiting); + return signature; + }, + }, + // Wall-clock nonces: bounded dispatch checks them against the protocol timestamp window. + nonceManager: () => (lastNonce = Math.max(Date.now(), lastNonce + 1)), + dispatchPolicy: policy, + transport: { + isTestnet: true, + async request(): Promise { + waiting--; + peakInFlight = Math.max(peakInFlight, ++inFlight); + await delay(transportMs); + inFlight--; + return response as T; + }, + }, + }; + const callerLatencies: number[] = []; + const start = performance.now(); + const calls: Promise[] = []; + for (let i = 0; i < orders; i++) { + const arrived = performance.now(); + calls.push( + executeAction(config, action).then(() => { + callerLatencies.push(performance.now() - arrived); + }), + ); + if (arrivalEveryMs > 0) await delay(arrivalEveryMs); + } + await Promise.all(calls); + const elapsed = performance.now() - start; + return { + p50Ms: percentile(callerLatencies, 0.5), + p99Ms: percentile(callerLatencies, 0.99), + ordersPerSecond: (orders * 1000) / elapsed, + peakSigning, + peakWaiting, + peakInFlight, + }; +} + +const report: Record[] = []; +for (const [scenario, profile] of Object.entries(scenarios)) { + if (filter !== undefined && !scenario.includes(filter)) continue; + for (const arrivalEveryMs of arrivals) { + const samples: Record = { ordered: [], bounded: [] }; + for (let round = 0; round < rounds; round++) { + const modes = round % 2 === 0 ? ["ordered", "bounded"] : ["bounded", "ordered"]; + for (const mode of modes) samples[mode].push(await run(profile, policies[mode], arrivalEveryMs, round + 1)); + } + for (const [mode, rows] of Object.entries(samples)) { + const summary: Record = { scenario, arrival: arrivalEveryMs ? "1/ms" : "burst", mode }; + for (const key of Object.keys(rows[0]) as (keyof Sample)[]) { + summary[key] = Number(median(rows.map((row) => row[key])).toFixed(2)); + } + report.push(summary); + } + } +} + +console.log(`${orders} orders per run, ${transportMs} ms transport, median of ${rounds} rounds`); +console.table(report); +if (outIndex >= 0) { + await writeFile( + process.argv[outIndex + 1], + JSON.stringify({ runtime: process.versions, orders, transportMs, rounds, report }, null, 2), + ); +} diff --git a/packages/hyperliquid/src/actions/execution.ts b/packages/hyperliquid/src/actions/execution.ts index 01ec9bbd..2c1b7e5a 100644 --- a/packages/hyperliquid/src/actions/execution.ts +++ b/packages/hyperliquid/src/actions/execution.ts @@ -1,6 +1,7 @@ /** Explicit signing and execution of canonical actions. @module */ import { HyperliquidError } from "../_base.ts"; import type { ExchangeConfig } from "../api/exchange/_methods/_base/_config.ts"; +import { assertDetachedAllowed } from "../api/exchange/_methods/_base/_dispatch.ts"; import { executeL1Action, executeUserSignedAction } from "../api/exchange/_methods/_base/execute.ts"; import { getSigningContext, @@ -56,7 +57,7 @@ function run( * @param action Action produced by an SDK builder. * @param options Signing options and cancellation signal. * @return An immutable signed request with the action's response type. - * @throws {HyperliquidError} When action ownership is invalid. + * @throws {HyperliquidError} When action ownership or dispatch policy is invalid. */ export async function signAction( config: ExchangeConfig, @@ -84,7 +85,7 @@ export async function signAction( * @param signed Signed request produced by signAction. * @param options Cancellation signal for submission. * @return The action's response. - * @throws {HyperliquidError} When request ownership, network, is invalid. + * @throws {HyperliquidError} When request ownership, network, or dispatch policy is invalid. */ export async function submitAction( config: ExchangeConfig, @@ -95,6 +96,7 @@ export async function submitAction( if (!owner || owner.isTestnet !== config.transport.isTestnet) { throw new HyperliquidError("Signed request ownership or network mismatch; rebuild and sign for this network"); } + assertDetachedAllowed(owner.key, config.dispatchPolicy); const response = await config.transport.request("exchange", signed, options?.signal); assertSuccessResponse(response); return response; @@ -106,7 +108,7 @@ export async function submitAction( * @param action Action produced by an SDK builder. * @param options Signing options and cancellation signal. * @return The action's response. - * @throws {HyperliquidError} When action ownership is invalid. + * @throws {HyperliquidError} When action ownership or dispatch policy is invalid. */ export function executeAction( config: ExchangeConfig, @@ -121,3 +123,4 @@ export function executeAction( } export { createNonceManager, type NonceManager } from "../api/exchange/_methods/_base/_nonce.ts"; +export type { DispatchPolicy } from "../api/exchange/_methods/_base/_dispatch.ts"; diff --git a/packages/hyperliquid/src/api/exchange/_methods/_base/_config.ts b/packages/hyperliquid/src/api/exchange/_methods/_base/_config.ts index 8c14295b..bc8a193f 100644 --- a/packages/hyperliquid/src/api/exchange/_methods/_base/_config.ts +++ b/packages/hyperliquid/src/api/exchange/_methods/_base/_config.ts @@ -5,7 +5,9 @@ import type { AbstractWallet } from "../../../../signing/mod.ts"; import type { IRequestTransport } from "../../../../transport/mod.ts"; +import type { Runtime } from "../../../../transport/runtime.ts"; import type { SymbolConverter } from "../../../../utils/mod.ts"; +import type { DispatchPolicy } from "./_dispatch.ts"; // ============================================================ // Type Utilities @@ -27,6 +29,20 @@ interface BaseConfig { /** The transport used to connect to the Hyperliquid Exchange API. */ transport: T; + /** + * How signed requests for this signer/network reach the transport. Default: `"ordered"`, which + * dispatches in nonce issuance order. `{ mode: "bounded" }` lets a ready signature overtake a + * slower earlier one within the exchange's nonce window; it supports managed execution only and + * assumes this process is the signer's only nonce producer. + */ + dispatchPolicy?: DispatchPolicy; + + /** + * Wall clock used by bounded dispatch to re-check the nonce timestamp window and `expiresAfter` + * immediately before sending. Only `now` is read. Default: the platform clock. + */ + runtime?: Partial; + /** Signature chain ID for EIP-712 signing, defaults to wallet's chain ID. */ signatureChainId?: `0x${string}` | (() => MaybePromise<`0x${string}`>); diff --git a/packages/hyperliquid/src/api/exchange/_methods/_base/_dispatch.ts b/packages/hyperliquid/src/api/exchange/_methods/_base/_dispatch.ts new file mode 100644 index 00000000..20b6b30b --- /dev/null +++ b/packages/hyperliquid/src/api/exchange/_methods/_base/_dispatch.ts @@ -0,0 +1,193 @@ +/** + * Optional bounded dispatch coordination per `(signer × network)`. + * + * The exchange accepts any unused nonce above the smallest of the signer's 100 highest nonces, so + * a later nonce may reach it before an earlier one. What it will not accept is an earlier nonce + * that 100 newer ones have already pushed out of that set. A lane therefore counts, for every + * outstanding request, how many newer requests were dispatched ahead of it, and holds further + * newer requests once that count reaches `maxOvertakes` (at most 99). The count is cumulative + * until the older request settles — a cap on simultaneous requests alone would let newer requests + * keep finishing, releasing their slots, and overtaking it indefinitely. A request counts until its + * response settles, not just until it is sent, so the bound holds even when requests arrive in a + * different order than they were sent. + * + * @module + */ + +import { HyperliquidError } from "../../../../_base.ts"; +import { race } from "../../../../transport/_abort.ts"; + +/** + * How signed requests for one signer/network reach the transport. + * + * - `"ordered"` (default): requests are dispatched in nonce-issuance order; a slow signature delays + * every later request for the same signer. + * - `{ mode: "bounded" }`: a ready signature may be dispatched before a slower earlier one, while + * no outstanding request is overtaken by more newer requests than the exchange's nonce window + * tolerates. Supported by managed execution only, for a signer whose nonces this process owns. + */ +export type DispatchPolicy = + | "ordered" + | { + mode: "bounded"; + /** Maximum newer requests dispatched ahead of any outstanding request, in `[0, 99]`. Default: `99`. */ + maxOvertakes?: number; + /** Maximum signing, waiting, and in-flight requests for this signer/network. Default: `1000`. */ + maxPending?: number; + }; + +/** A bounded policy with its defaults applied. */ +export interface BoundedLimits { + maxOvertakes: number; + maxPending: number; +} + +/** One request from nonce reservation until its response (or failure) settles. */ +interface Slot { + nonce: number; + /** Newer requests dispatched while this one was outstanding. */ + overtakes: number; + /** Signed and waiting for its turn. */ + ready: boolean; + sent: boolean; +} + +/** Shared state of every bounded request outstanding for one signer/network. */ +interface Lane extends BoundedLimits { + slots: Set; + latestNonce: number; + /** Resolves on the next admission or release; waiters re-check their own condition. */ + changed: Promise; + wake: () => void; +} + +/** Active lanes per `(walletAddress × isTestnet)` key; a lane is dropped once its last slot releases. */ +const lanes = new Map(); + +/** Reservation of one nonce in a bounded lane. */ +export interface DispatchReservation { + /** Resolves once this request may be handed to the transport, then counts it as an overtake. */ + wait(signal?: AbortSignal): Promise; + /** Frees the slot; idempotent. Call once the request has settled or will never be sent. */ + release(): void; +} + +/** + * Whether a detached request under `policy` needs the signer-specific check: in bounded mode, or + * while any bounded lane is active. Lets callers skip resolving the signer address otherwise. + */ +export function needsDetachedCheck(policy?: DispatchPolicy): boolean { + return typeof policy === "object" || lanes.size > 0; +} + +/** Rejects detached signing and submission in bounded mode or while a bounded lane is active. */ +export function assertDetachedAllowed(key: string, policy?: DispatchPolicy): void { + if (typeof policy === "object" || lanes.has(key)) { + throw new HyperliquidError( + "Bounded dispatch supports managed execute calls only; detached signing/prepared submission is unavailable", + ); + } +} + +/** Rejects an ordered call while bounded calls for the same signer/network are outstanding. */ +export function assertOrderedAllowed(key: string): void { + if (lanes.size > 0 && lanes.has(key)) { + throw new HyperliquidError("All active calls for this signer/network must use the same bounded dispatch policy"); + } +} + +/** + * Validates a bounded policy and checks that the lane can take one more request, without changing + * any state. Called under the nonce lock BEFORE a nonce is allocated, so a rejected call consumes + * no nonce. + */ +export function admitBounded(key: string, policy: Exclude): BoundedLimits { + const { maxOvertakes = 99, maxPending = 1000 } = policy; + if ( + policy.mode !== "bounded" || + !Number.isSafeInteger(maxOvertakes) || + maxOvertakes < 0 || + maxOvertakes > 99 || + !Number.isSafeInteger(maxPending) || + maxPending < 1 + ) { + throw new RangeError("Bounded dispatch requires maxOvertakes in [0, 99] and a positive integer maxPending"); + } + const lane = lanes.get(key); + if (lane !== undefined) { + if (lane.maxOvertakes !== maxOvertakes || lane.maxPending !== maxPending) { + throw new HyperliquidError("Active signer/network dispatch policy does not match this call"); + } + if (lane.slots.size >= maxPending) throw new HyperliquidError("Bounded dispatch queue is full"); + } + return { maxOvertakes, maxPending }; +} + +function resetWake(lane: Lane): void { + lane.changed = new Promise((resolve) => { + lane.wake = resolve; + }); +} + +function notify(lane: Lane): void { + lane.wake(); + resetWake(lane); +} + +/** + * Reserves `nonce` in the lane, under the same nonce lock as {@linkcode admitBounded}. The slot is + * kept after dispatch until {@linkcode DispatchReservation.release}, because a sent request can + * still arrive after newer ones. + * + * @throws {HyperliquidError} When a custom nonce manager returned a nonce that does not exceed the + * lane's previous one; that nonce is consumed by the manager but never sent. + */ +export function reserveDispatch(key: string, nonce: number, limits: BoundedLimits): DispatchReservation { + let lane = lanes.get(key); + if (!Number.isSafeInteger(nonce) || nonce < 0 || (lane !== undefined && nonce <= lane.latestNonce)) { + throw new HyperliquidError("Bounded dispatch requires strictly increasing unique nonces"); + } + if (lane === undefined) { + lane = { ...limits, slots: new Set(), latestNonce: -1, changed: Promise.resolve(), wake: (): void => {} }; + resetWake(lane); + lanes.set(key, lane); + } + lane.latestNonce = nonce; + const slot: Slot = { nonce, overtakes: 0, ready: false, sent: false }; + lane.slots.add(slot); + const active = lane; + const { maxOvertakes } = limits; + let released = false; + const canSend = (): boolean => { + for (const older of active.slots) { + if (older.nonce >= nonce) continue; + // A sent request keeps counting until it settles, since it can still arrive after newer ones. + // With `maxOvertakes: 0` a sent predecessor no longer blocks, which reproduces ordered dispatch. + if (older.overtakes >= maxOvertakes && (!older.sent || maxOvertakes > 0)) return false; + // Among ready requests the older goes first. Anything blocking it also blocks this one, so this + // never delays this request past the wake-up that admits the older one, and it spares the + // older one an overtake. + if (older.ready && !older.sent) return false; + } + return true; + }; + return { + async wait(signal?: AbortSignal): Promise { + slot.ready = true; + while (!canSend()) await race(active.changed, signal); + if (signal?.aborted) throw signal.reason; + // Admission is synchronous from the check above to here, so concurrent ready requests cannot + // all observe the same remaining window and oversubscribe it. + slot.sent = true; + for (const older of active.slots) if (older.nonce < nonce) older.overtakes++; + notify(active); + }, + release(): void { + if (released) return; + released = true; + active.slots.delete(slot); + notify(active); + if (active.slots.size === 0 && lanes.get(key) === active) lanes.delete(key); + }, + }; +} diff --git a/packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts b/packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts index 5d57976e..a8697717 100644 --- a/packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts +++ b/packages/hyperliquid/src/api/exchange/_methods/_base/_nonce.ts @@ -57,11 +57,12 @@ export function createNonceManager(maxEntries: number = DEFAULT_MAX_ENTRIES, run // A throttled scan is chosen over evicting the oldest entry because // eviction is unsafe here: dropping a live wallet's entry resets its // nonce derivation to wall-clock time, which can go BACKWARDS relative - // to a nonce already sent, and the exchange rejects non-increasing - // nonces. Throttling only *delays* deletions that were already safe — - // an entry is still deleted only when `now > last`, i.e. when the next - // nonce for that key (at least `now`) is guaranteed to exceed the - // deleted value — so per-key monotonicity is preserved by construction. + // to a nonce already sent, risking a reused nonce or one already + // below the signer's 100-highest window. Throttling only *delays* + // deletions that were already safe — an entry is still deleted only + // when `now > last`, i.e. when the next nonce for that key (at least + // `now`) is guaranteed to exceed the deleted value — so per-key + // monotonicity is preserved by construction. // // The trade-off is memory, never correctness: stale entries may linger // up to `PRUNE_INTERVAL_MS` longer than before, and between scans the diff --git a/packages/hyperliquid/src/api/exchange/_methods/_base/_shell.ts b/packages/hyperliquid/src/api/exchange/_methods/_base/_shell.ts index 3c154a6a..dc1315a5 100644 --- a/packages/hyperliquid/src/api/exchange/_methods/_base/_shell.ts +++ b/packages/hyperliquid/src/api/exchange/_methods/_base/_shell.ts @@ -3,10 +3,12 @@ * @module */ +import { HyperliquidError } from "../../../../_base.ts"; import { getWalletAddress, type Signature } from "../../../../signing/mod.ts"; import { race } from "../../../../transport/_abort.ts"; import { registerExchangeWireRequest } from "../../../../transport/_wire.ts"; import type { ExchangeConfig } from "./_config.ts"; +import { admitBounded, assertDetachedAllowed, assertOrderedAllowed, reserveDispatch } from "./_dispatch.ts"; import { assertSuccessResponse } from "./errors.ts"; import { globalNonceManager } from "./_nonce.ts"; import { withLock } from "./_semaphore.ts"; @@ -100,6 +102,10 @@ export function getPreparedRequestState(prepared: PreparedExchangeRequest>(); * time; for a remote wallet, where signing is an `eth_signTypedData_v4` round trip, that is the * difference between one order in flight per wallet and all of them. * - * Wire order is preserved by {@linkcode dispatchChains} rather than by the lock: a request waits - * for its predecessor to reach `transport.request` before making its own call, so the default policy - * preserves nonce issuance order. Network responses resolve concurrently. + * Under the default ordered policy, wire order is preserved by {@linkcode dispatchChains} rather + * than by the lock: a request waits for its predecessor to reach `transport.request` before making + * its own call, so requests are dispatched in nonce issuance order. Under the bounded policy a ready + * request may be dispatched before a slower earlier one, within the lane's overtaking bound, after + * re-checking the protocol timestamp window and `expiresAfter`. Network responses resolve + * concurrently. * * @param config Exchange API configuration. * @param build Callback that, given the nonce, returns the action, signature, and any extras. @@ -147,6 +159,7 @@ const dispatchChains = new Map>(); * @return The validated API response, or the signed request when `prepareOnly` is set. * * @throws {ApiRequestError} If the API returns an error response. + * @throws {HyperliquidError} If the dispatch policy rejects the call (see `_dispatch.ts`). */ export async function executeWithShell( config: ExchangeConfig, @@ -157,7 +170,7 @@ export async function executeWithShell( const leader = "wallet" in config ? config.wallet : config.signers[0]; const walletAddress = await getWalletAddress(leader); - // Serialize nonce allocation per (wallet × testnet); managed execution preserves delivery order. + // Serialize nonce allocation per (wallet × testnet); the dispatch policy controls delivery order. // The key string is cached per (wallet × transport); it is rebuilt only when the wallet's // address or the transport's testnet flag no longer matches the cached entry. const isTestnet = config.transport.isTestnet; @@ -175,7 +188,23 @@ export async function executeWithShell( perTransport.set(config.transport, { walletAddress, isTestnet, key }); } if (signal?.aborted) throw signal.reason; + const policy = config.dispatchPolicy; const box = await withLock(key, async () => { + // --- Check the dispatch policy ----------------------------- + // Before nonce allocation, so a rejected call consumes no nonce. Under the lock, so the checks + // and the slot claimed below see the same lane and chain state. + let limits: ReturnType | undefined; + if (prepareOnly) { + assertDetachedAllowed(key, policy); + } else if (typeof policy === "object") { + if (dispatchChains.has(key)) { + throw new HyperliquidError("Cannot mix ordered and bounded calls for an active signer/network"); + } + limits = admitBounded(key, policy); + } else { + assertOrderedAllowed(key); + } + // --- Generate nonce -------------------------------------- // `globalNonceManager.getNonce` returns a plain number: skip the await (and its async hop) // unless a custom `nonceManager` actually handed back a promise. @@ -184,12 +213,14 @@ export async function executeWithShell( // --- Claim this nonce's slot in the dispatch order -------- // Taken under the lock, so slots are claimed in the same order nonces are issued. - const predecessor = prepareOnly ? undefined : dispatchChains.get(key); + const bounded = limits === undefined ? undefined : reserveDispatch(key, nonce, limits); + const ordered = !prepareOnly && bounded === undefined; + const predecessor = ordered ? dispatchChains.get(key) : undefined; let openGate!: () => void; const dispatched = new Promise((resolve) => { openGate = resolve; }); - if (!prepareOnly) dispatchChains.set(key, dispatched); + if (ordered) dispatchChains.set(key, dispatched); // --- Sign and dispatch, outside the lock ------------------ // Signing is a network round trip for a remote wallet; running it here rather than inside the @@ -207,17 +238,33 @@ export async function executeWithShell( return request as T; } if (predecessor !== undefined) await race(predecessor, signal); + if (bounded !== undefined) { + await bounded.wait(signal); + // Re-checked at dispatch: a request may have waited behind a stalled signature. + const now = config.runtime?.now ? config.runtime.now() : Date.now(); + if (nonce <= now - NONCE_MAX_AGE_MS || nonce >= now + NONCE_MAX_LEAD_MS) { + throw new HyperliquidError("Nonce is outside the protocol timestamp window"); + } + const expiresAfter = extras?.expiresAfter; + if (typeof expiresAfter === "number" && expiresAfter <= now) { + throw new HyperliquidError("Action expired before dispatch"); + } + } // `transport.request` runs synchronously up to its first await, so wire order is fixed // here. It is assigned rather than awaited so the gate below opens on dispatch, not on // the response. const request = { action, signature, nonce, ...extras }; registerExchangeWireRequest(request, action); response = config.transport.request("exchange", request, signal); + // A bounded slot stays reserved until the response settles: until then the request may + // still arrive after newer ones. + if (bounded !== undefined) return await response; } finally { // Wait for our turn even when this request never reached the wire. A rejected signature // or an abort burns its nonce, which the server tolerates as a gap — but opening the gate // early would let a later nonce overtake an earlier one that is still being signed. if (predecessor !== undefined) await predecessor; + bounded?.release(); openGate(); // Idle chain: drop the entry so the map does not grow one slot per key forever. Compared // by identity, so a successor that has already claimed the slot is left alone. diff --git a/packages/hyperliquid/src/api/exchange/_methods/_base/mod.ts b/packages/hyperliquid/src/api/exchange/_methods/_base/mod.ts index dc8ca81f..09a896d2 100644 --- a/packages/hyperliquid/src/api/exchange/_methods/_base/mod.ts +++ b/packages/hyperliquid/src/api/exchange/_methods/_base/mod.ts @@ -21,3 +21,4 @@ export { } from "./errors.ts"; export { executeL1Action, executeUserSignedAction } from "./execute.ts"; export type { PreparedExchangeRequest } from "./_shell.ts"; +export type { DispatchPolicy } from "./_dispatch.ts"; diff --git a/packages/hyperliquid/src/api/exchange/_methods/prepareRequest.ts b/packages/hyperliquid/src/api/exchange/_methods/prepareRequest.ts index fdb59a95..f0ef511c 100644 --- a/packages/hyperliquid/src/api/exchange/_methods/prepareRequest.ts +++ b/packages/hyperliquid/src/api/exchange/_methods/prepareRequest.ts @@ -3,8 +3,10 @@ // ============================================================ import { HyperliquidError } from "../../../_base.ts"; +import { getWalletAddress } from "../../../signing/mod.ts"; import type { IRequestTransport } from "../../../transport/mod.ts"; import type { ExchangeConfig, PreparedExchangeRequest } from "./_base/mod.ts"; +import { assertDetachedAllowed, needsDetachedCheck } from "./_base/_dispatch.ts"; import { linkPreparedRequestState, type PreparedRequestState } from "./_base/_shell.ts"; export type { PreparedExchangeRequest } from "./_base/mod.ts"; @@ -102,6 +104,7 @@ function toPlainSignedRequest(payload: unknown): PreparedExchangeRequest( config: ExchangeConfig, run: (config: ExchangeConfig) => Promise, ): Promise> { + // Prepared payloads bypass the dispatch coordinator, so they are unavailable in bounded mode or + // while bounded calls for this signer/network are outstanding. + if (needsDetachedCheck(config.dispatchPolicy)) { + const leader = "wallet" in config ? config.wallet : config.signers[0]; + assertDetachedAllowed(`${await getWalletAddress(leader)}:${config.transport.isTestnet}`, config.dispatchPolicy); + } + // Shared with the returned payload (see `linkPreparedRequestState`): invalid attempts recorded // after finalization — possible only for leaked callback work — poison the payload so // `submitPrepared` rejects it (best-effort). diff --git a/packages/hyperliquid/src/api/exchange/_methods/submitPrepared.ts b/packages/hyperliquid/src/api/exchange/_methods/submitPrepared.ts index b75bf405..78630697 100644 --- a/packages/hyperliquid/src/api/exchange/_methods/submitPrepared.ts +++ b/packages/hyperliquid/src/api/exchange/_methods/submitPrepared.ts @@ -3,9 +3,11 @@ // ============================================================ import { HyperliquidError } from "../../../_base.ts"; +import { getWalletAddress } from "../../../signing/mod.ts"; import type { ExchangeConfig, PreparedExchangeRequest } from "./_base/mod.ts"; import { assertSuccessResponse } from "./_base/errors.ts"; -import { getPreparedRequestState } from "./_base/_shell.ts"; +import { assertDetachedAllowed, needsDetachedCheck } from "./_base/_dispatch.ts"; +import { getPreparedRequestState, getSigningContext } from "./_base/_shell.ts"; /** Request options for the {@linkcode submitPrepared} function. */ export interface SubmitPreparedOptions { @@ -31,6 +33,7 @@ export interface SubmitPreparedOptions { * @return The API response. * * @throws {HyperliquidError} When the payload was poisoned by a request attempted after it was produced. + * @throws {HyperliquidError} In bounded dispatch mode, or while bounded calls for the signer are outstanding. * @throws {TransportError} When the transport layer throws an error. * @throws {ApiRequestError} When the API returns an unsuccessful response (e.g. a stale nonce). * @@ -53,6 +56,17 @@ export async function submitPrepared( prepared: PreparedExchangeRequest, opts?: SubmitPreparedOptions, ): Promise { + // Prepared payloads bypass the dispatch coordinator. Check the signer the payload was signed by + // when known, so routing it through another client's config cannot bypass its active bounded lane. + if (needsDetachedCheck(config.dispatchPolicy)) { + const context = getSigningContext(prepared); + const leader = "wallet" in config ? config.wallet : config.signers[0]; + assertDetachedAllowed( + context?.key ?? `${await getWalletAddress(leader)}:${config.transport.isTestnet}`, + config.dispatchPolicy, + ); + } + // Re-check the poison state synchronously, immediately before posting: a request attempted // (via the capture transport) after the payload was produced invalidates it, even though the // payload itself looks fine. This is a point-in-time check, not a happens-before guarantee: diff --git a/packages/hyperliquid/src/api/exchange/client.ts b/packages/hyperliquid/src/api/exchange/client.ts index a9a6b432..d1e89838 100644 --- a/packages/hyperliquid/src/api/exchange/client.ts +++ b/packages/hyperliquid/src/api/exchange/client.ts @@ -592,7 +592,7 @@ export class ExchangeClient(action: CanonicalAction, options?: ActionOptions): Promise> { return signAction(this.config, action, options); @@ -603,7 +603,7 @@ export class ExchangeClient(signed: SignedAction, options?: { signal?: AbortSignal }): Promise { return submitAction(this.config, signed, options); @@ -614,7 +614,7 @@ export class ExchangeClient(action: CanonicalAction, options?: ActionOptions): Promise { return executeAction(this.config, action, options); @@ -1784,6 +1784,7 @@ export class ExchangeClient; + constructor(first: number) { + // Pre-filled with 100 earlier requests, so the window is already full. + this.set = new Set(Array.from({ length: 100 }, (_, i) => first - 100 + i)); + } + /** Applies the exchange's acceptance rule, failing the test on a reused or out-of-window nonce. */ + accept(nonce: number): void { + expect(this.set.has(nonce)).toBe(false); + expect(nonce).toBeGreaterThan(Math.min(...this.set)); + this.set.add(nonce); + if (this.set.size > 100) this.set.delete(Math.min(...this.set)); + } +} + +const OK = { status: "ok", response: { type: "default" } }; +const SIGNATURE = `0x${"11".repeat(64)}1b` as const; +const built: BuildResult = { action: { type: "test" }, signature: { r: "0x1", s: "0x2", v: 27 } }; +const order = (): ReturnType => + buildOrder({ orders: [{ a: 0, b: true, p: "1", s: "1", r: false, t: { limit: { tif: "Gtc" } } }] }); + +let walletId = 100; +function nextAddress(): `0x${string}` { + return `0x${(++walletId).toString(16).padStart(40, "0")}`; +} + +interface Harness { + config: ExchangeConfig; + /** Nonces in the order they reached the transport. */ + nonces: number[]; + clock: FakeRuntime; + /** Nonces handed out by the nonce manager, sent or not. */ + allocated: () => number; +} + +function harness(maxOvertakes = 99): Harness { + const clock = new FakeRuntime(); + const nonces: number[] = []; + let nonce = clock.now(); + const first = nonce; + const server = new ServerNonces(nonce); + const config: ExchangeConfig = { + wallet: { address: nextAddress(), signTypedData: async () => SIGNATURE }, + runtime: clock, + nonceManager: () => nonce++, + dispatchPolicy: { mode: "bounded", maxOvertakes }, + transport: { + isTestnet: true, + request: async (_endpoint: "info" | "exchange", body: unknown): Promise => { + const received = (body as { nonce: number }).nonce; + server.accept(received); + nonces.push(received); + return OK as T; + }, + }, + }; + return { config, nonces, clock, allocated: () => nonce - first }; +} + +/** A build callback that resolves only when the returned `release` is called. */ +function stalled(): { build: () => Promise; release: (result?: BuildResult) => void } { + let resolve!: (result: BuildResult) => void; + const promise = new Promise((r) => { + resolve = r; + }); + return { build: () => promise, release: (result = built) => resolve(result) }; +} + +/** Deterministic PRNG (mulberry32), so the randomized schedules are reproducible. */ +function random(seed: number): () => number { + return () => { + seed = (seed + 0x6d2b79f5) | 0; + let t = Math.imul(seed ^ (seed >>> 15), 1 | seed); + t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +// ============================================================ +// Tests +// ============================================================ + +describe("bounded dispatch", () => { + test("ready signatures may overtake one stalled signature, but cumulative overtaking stops at 99", async () => { + const { config, nonces } = harness(); + const slow = stalled(); + const first = executeWithShell(config, slow.build); + await drain(); + // 120 newer requests against a full 100-nonce window: only 99 may go ahead of the stalled one. + const requests = Array.from({ length: 120 }, () => executeWithShell(config, async () => built)); + await Promise.all(requests.slice(0, 99)); + await drain(); + expect(nonces).toHaveLength(99); + slow.release(); + await Promise.all([first, ...requests]); + const firstNonce = Math.min(...nonces); + expect(nonces.indexOf(firstNonce)).toBe(99); + expect(new Set(nonces).size).toBe(121); + }); + + test("randomized signing and network latency never push an outstanding nonce out of the window", async () => { + for (const seed of [1, 2, 3]) { + const clock = new FakeRuntime(); + const next = random(seed); + let nonce = clock.now(); + const server = new ServerNonces(nonce); + let arrived = 0; + const config: ExchangeConfig = { + wallet: { address: nextAddress(), signTypedData: async () => SIGNATURE }, + runtime: clock, + nonceManager: () => nonce++, + dispatchPolicy: { mode: "bounded" }, + transport: { + isTestnet: true, + // The server sees each request after its own random network delay, so requests sent in + // one order can arrive in another: the bound must hold for arrival order too. + request: (_endpoint: "info" | "exchange", body: unknown): Promise => + new Promise((resolve) => { + clock.setTimeout( + () => { + server.accept((body as { nonce: number }).nonce); + arrived++; + resolve(OK as T); + }, + Math.floor(next() * 50), + ); + }), + }, + }; + // Mostly fast signatures, with a few that stall for far longer than 100 others take. + const requests = Array.from({ length: 400 }, () => + executeWithShell( + config, + () => + new Promise((resolve) => { + const roll = next(); + clock.setTimeout(() => resolve(built), roll < 0.02 ? 2_000 : Math.floor(roll * 20)); + }), + ), + ); + await drain(); + while (clock.nextTimer()) await drain(); + await Promise.all(requests); + expect(arrived).toBe(400); + } + }); + + test("transport completion does not erase earlier pending nonce overtaking history", async () => { + const { config, nonces } = harness(2); + const slow = stalled(); + const first = executeWithShell(config, slow.build); + await drain(); + await executeWithShell(config, async () => built); + await executeWithShell(config, async () => built); + const blocked = executeWithShell(config, async () => built); + await drain(); + expect(nonces).toHaveLength(2); + slow.release(); + await Promise.all([first, blocked]); + expect(nonces).toHaveLength(4); + }); + + test("maxOvertakes 0 dispatches in nonce order", async () => { + const { config, nonces } = harness(0); + const slow = stalled(); + const first = executeWithShell(config, slow.build); + await drain(); + const later = Array.from({ length: 3 }, () => executeWithShell(config, async () => built)); + await drain(); + expect(nonces).toHaveLength(0); + slow.release(); + await Promise.all([first, ...later]); + expect(nonces).toEqual([...nonces].sort((a, b) => a - b)); + }); + + test("ready requests waiting for the window are sent oldest first", async () => { + const { config, nonces } = harness(1); + const slow = stalled(); + const first = executeWithShell(config, slow.build); + await drain(); + await executeWithShell(config, async () => built); + // The window is spent; three more finish signing newest first and wait. + const signers = [stalled(), stalled(), stalled()]; + const waiting = signers.map((signer) => executeWithShell(config, signer.build)); + await drain(); + for (const signer of signers.toReversed()) { + signer.release(); + await drain(); + } + expect(nonces).toHaveLength(1); + slow.release(); + await Promise.all([first, ...waiting]); + const sent = nonces.slice(2); + expect(sent).toEqual([...sent].sort((a, b) => a - b)); + }); + + test("cancelling a stalled signer releases the window and late signing never posts", async () => { + const { config, nonces } = harness(1); + const abort = new AbortController(); + const slow = stalled(); + const first = executeWithShell(config, slow.build, abort.signal).catch((e: unknown) => e); + await drain(); + const second = executeWithShell(config, async () => built); + const third = executeWithShell(config, async () => built); + await drain(); + expect(nonces).toHaveLength(1); + abort.abort(new Error("stop signer")); + expect(await first).toBe(abort.signal.reason); + await Promise.all([second, third]); + slow.release(); + await drain(); + expect(nonces).toHaveLength(2); + }); + + test("cancelling a request waiting for its turn settles it without posting", async () => { + const { config, nonces } = harness(0); + const slow = stalled(); + const first = executeWithShell(config, slow.build); + await drain(); + const abort = new AbortController(); + const waiting = executeWithShell(config, async () => built, abort.signal).catch((e: unknown) => e); + await drain(); + abort.abort(new Error("stop waiting")); + expect(await waiting).toBe(abort.signal.reason); + slow.release(); + await first; + expect(nonces).toHaveLength(1); + }); + + test("the protocol timestamp window is checked again at dispatch, in both directions", async () => { + const { config, clock, nonces } = harness(); + const slow = stalled(); + const held = executeWithShell(config, slow.build).catch((e: unknown) => e); + await drain(); + // Waited past two days behind its own signature. + clock.advance(172_800_000); + slow.release(); + expect(String(await held)).toContain("timestamp window"); + + const future = { ...config, nonceManager: () => clock.now() + 86_400_000 }; + await expect(executeWithShell(future, async () => built)).rejects.toThrow("timestamp window"); + expect(nonces).toHaveLength(0); + // Rejections released their slots: a fresh, in-window request goes through. + await executeWithShell({ ...config, nonceManager: () => clock.now() }, async () => built); + expect(nonces).toHaveLength(1); + }); + + test("mixed policies and detached preparation are rejected", async () => { + const { config } = harness(); + const slow = stalled(); + const held = executeWithShell(config, slow.build); + await drain(); + await expect(executeWithShell({ ...config, dispatchPolicy: "ordered" }, async () => built)).rejects.toThrow( + "same bounded", + ); + await expect(executeWithShell({ ...config, dispatchPolicy: undefined }, async () => built)).rejects.toThrow( + "same bounded", + ); + await expect( + executeWithShell({ ...config, dispatchPolicy: { mode: "bounded", maxOvertakes: 5 } }, async () => built), + ).rejects.toThrow("does not match"); + await expect(prepareRequest(config, async () => {})).rejects.toThrow("managed execute"); + await expect(signAction(config, order())).rejects.toThrow("managed execute"); + await expect(submitAction(config, {} as never)).rejects.toThrow(); + slow.release(); + await held; + }); + + test("a bounded call is rejected while ordered calls for the signer are still being dispatched", async () => { + const { config } = harness(); + const slow = stalled(); + const ordered = executeWithShell({ ...config, dispatchPolicy: "ordered" }, slow.build); + await drain(); + await expect(executeWithShell(config, async () => built)).rejects.toThrow("Cannot mix"); + slow.release(); + await ordered; + // The ordered chain is idle again, so bounded calls are accepted. + await executeWithShell(config, async () => built); + }); + + test("detached submission checks the original signer even through another client's configuration", async () => { + const { config } = harness(); + const signed = await signAction({ ...config, dispatchPolicy: "ordered" }, order()); + const slow = stalled(); + const held = executeWithShell(config, slow.build); + await drain(); + const other = harness().config; + await expect(submitAction({ ...other, dispatchPolicy: "ordered" }, signed)).rejects.toThrow("managed execute"); + await expect(submitPrepared({ ...other, dispatchPolicy: "ordered" }, signed)).rejects.toThrow("managed execute"); + slow.release(); + await held; + // Once the lane has drained, the signed request may be submitted under the default policy. + await submitAction({ ...config, dispatchPolicy: "ordered" }, signed); + }); + + test("policy rejections are raised before a nonce is allocated", async () => { + const { config, allocated } = harness(0); + config.dispatchPolicy = { mode: "bounded", maxOvertakes: 0, maxPending: 1 }; + const slow = stalled(); + const held = executeWithShell(config, slow.build); + await drain(); + expect(allocated()).toBe(1); + await expect(executeWithShell(config, async () => built)).rejects.toThrow("queue is full"); + await expect(executeWithShell({ ...config, dispatchPolicy: "ordered" }, async () => built)).rejects.toThrow( + "same bounded", + ); + await expect(signAction({ ...config, dispatchPolicy: "ordered" }, order())).rejects.toThrow("managed execute"); + await expect( + executeWithShell({ ...config, dispatchPolicy: { mode: "bounded", maxOvertakes: 100 } }, async () => built), + ).rejects.toThrow(RangeError); + expect(allocated()).toBe(1); + slow.release(); + await held; + }); + + test("queue limits, failed signatures, and expiry release bounded state", async () => { + const { config, nonces, clock } = harness(0); + config.dispatchPolicy = { mode: "bounded", maxOvertakes: 0, maxPending: 1 }; + const slow = stalled(); + const held = executeWithShell(config, slow.build); + await drain(); + await expect(executeWithShell(config, async () => built)).rejects.toThrow("queue is full"); + slow.release(); + await held; + await expect( + executeWithShell(config, async () => { + throw new Error("wallet rejected"); + }), + ).rejects.toThrow("wallet rejected"); + await expect( + executeWithShell(config, async () => ({ ...built, extras: { expiresAfter: clock.now() - 1 } })), + ).rejects.toThrow("expired"); + await executeWithShell(config, async () => built); + expect(nonces).toHaveLength(2); + }); + + test("closing the transport settles signing, waiting, and in-flight requests and releases the lane", async () => { + const clock = new FakeRuntime(); + const posted: number[] = []; + // Never answers a post, so a sent request stays in flight until the socket closes. + const socket = new FakeSocket((data) => { + const frame = JSON.parse(data); + if (frame.method === "post") posted.push(frame.request.payload.nonce); + }); + const ws = new WebSocketTransport({ + isTestnet: true, + runtime: clock, + timeout: null, + reconnect: { maxRetries: 0 }, + webSocketFactory: () => socket as unknown as WebSocket, + }); + await drain(); + socket.open(); + await drain(); + let nonce = clock.now(); + const config: ExchangeConfig = { + wallet: { address: nextAddress(), signTypedData: async () => SIGNATURE }, + runtime: clock, + nonceManager: () => nonce++, + dispatchPolicy: { mode: "bounded", maxOvertakes: 1 }, + transport: ws, + }; + const slow = stalled(); + const signing = executeWithShell(config, slow.build).catch((e: unknown) => e); + await drain(); + const inFlight = executeWithShell(config, async () => built).catch((e: unknown) => e); + await drain(); + const waiting = executeWithShell(config, async () => built).catch((e: unknown) => e); + await drain(); + expect(posted).toHaveLength(1); + + ws.close(); + expect(await inFlight).toBeInstanceOf(Error); + slow.release(); + expect(await signing).toBeInstanceOf(Error); + expect(await waiting).toBeInstanceOf(Error); + expect(posted).toHaveLength(1); + + // Every slot was released: the same signer may switch back to the default policy. + const nonces: number[] = []; + await executeWithShell( + { + ...config, + dispatchPolicy: "ordered", + transport: { + isTestnet: true, + request: async (_endpoint: "info" | "exchange", body: unknown): Promise => { + nonces.push((body as { nonce: number }).nonce); + return OK as T; + }, + }, + }, + async () => built, + ); + expect(nonces).toHaveLength(1); + clock.advance(60_000); + expect(clock.pendingTimers).toBe(0); + }); + + test("the default policy is unchanged when dispatchPolicy is omitted", async () => { + const { config, nonces } = harness(); + const ordered = { ...config, dispatchPolicy: undefined }; + const slow = stalled(); + const first = executeWithShell(ordered, slow.build); + await drain(); + const later = Array.from({ length: 3 }, () => executeWithShell(ordered, async () => built)); + await drain(); + expect(nonces).toHaveLength(0); + slow.release(); + await Promise.all([first, ...later]); + expect(nonces).toEqual([...nonces].sort((a, b) => a - b)); + }); +}); diff --git a/packages/hyperliquid/tests/api/exchange/_dispatchOrder.test.ts b/packages/hyperliquid/tests/api/exchange/_dispatchOrder.test.ts index 86cb5990..eb7693ce 100644 --- a/packages/hyperliquid/tests/api/exchange/_dispatchOrder.test.ts +++ b/packages/hyperliquid/tests/api/exchange/_dispatchOrder.test.ts @@ -1,7 +1,9 @@ /** * Tests for the per-wallet dispatch order guaranteed by `executeWithShell`. * - * The server requires a wallet's exchange requests to reach it in strictly increasing nonce order. + * The SDK's default `"ordered"` dispatch policy sends a wallet's exchange requests in nonce-issuance + * order. The exchange does not require this — it accepts any unused nonce above the smallest of the + * wallet's 100 highest — but it is the compatibility default (see `_boundedDispatch.test.ts`). * The nonce lock only fixes the order nonces are ISSUED in; signing happens outside it so that * concurrent callers on one wallet — where signing is a network round trip for any remote wallet — * can sign at the same time. A per-wallet dispatch chain is what restores wire order afterwards. @@ -92,7 +94,7 @@ describe("executeWithShell dispatch order", () => { test("holds when signatures complete in reverse order", async () => { const harness = createHarness(); // The first nonce signs slowest and the last signs fastest, so completion order is the exact - // reverse of issue order — the case a naive "dispatch when signed" implementation gets wrong. + // reverse of issue order — the case that distinguishes the default ordered policy from bounded dispatch. await runConcurrent(harness, 10, (i) => (10 - i) * 10); assertEquals(harness.dispatched.length, 10); diff --git a/packages/hyperliquid/tests/api/exchange/_runtimeIntegration.test.ts b/packages/hyperliquid/tests/api/exchange/_runtimeIntegration.test.ts new file mode 100644 index 00000000..445a6af3 --- /dev/null +++ b/packages/hyperliquid/tests/api/exchange/_runtimeIntegration.test.ts @@ -0,0 +1,87 @@ +/** + * End-to-end bounded dispatch across real HTTP and WebSocket transports, driven by injected + * `fetch`, socket factory, and `FakeRuntime`: one signer's nonces stay unique across clients, and + * the nonce bound leaves the WebSocket message budget to the transport. + * @module + */ + +import { expect, test } from "bun:test"; + +import { buildOrder } from "../../../src/actions/order.ts"; +import { ExchangeClient } from "../../../src/api/exchange/client.ts"; +import { HttpTransport } from "../../../src/transport/http/mod.ts"; +import { WebSocketQuota, WebSocketTransport } from "../../../src/transport/websocket/mod.ts"; +import { drain, FakeRuntime } from "../../_fakeRuntime.ts"; +import { FakeSocket } from "../../_fakeSocket.ts"; + +test("HTTP and WebSocket clients share bounded coordination/nonces with multi-sig and vault options", async () => { + const clock = new FakeRuntime(); + clock.setWallTime(Date.now()); + const requests: { nonce: number; vaultAddress?: string; action: { type: string; signatures?: unknown[] } }[] = []; + const result = { status: "ok", response: { type: "order", data: { statuses: [{ resting: { oid: 1 } }] } } }; + const http = new HttpTransport({ + isTestnet: true, + runtime: clock, + timeout: null, + fetch: async (_url, init) => { + requests.push(JSON.parse(init!.body as string)); + return new Response(JSON.stringify(result), { headers: { "Content-Type": "application/json" } }); + }, + }); + // A one-message budget: every post after the first drives the bucket into debt. + const quota = new WebSocketQuota({ runtime: clock, rateLimit: { capacity: 1, refillPerMinute: 60 } }); + let posts = 0; + const socket = new FakeSocket((data) => { + const frame = JSON.parse(data); + if (frame.method !== "post") return; + posts++; + requests.push(frame.request.payload); + queueMicrotask(() => + socket.receive({ channel: "post", data: { id: frame.id, response: { type: "action", payload: result } } }), + ); + }); + const ws = new WebSocketTransport({ + isTestnet: true, + runtime: clock, + quota, + timeout: null, + reconnect: { stableTimeout: 0 }, + webSocketFactory: () => socket as unknown as WebSocket, + }); + try { + await drain(); + socket.open(); + await drain(); + const signer = (hex: string) => ({ + address: `0x${hex.repeat(20)}` as `0x${string}`, + signTypedData: async () => `0x${"11".repeat(64)}1b` as `0x${string}`, + }); + const signers = [signer("a1"), signer("a2")] as const; + const base = { + signers, + multiSigUser: `0x${"a3".repeat(20)}` as `0x${string}`, + dispatchPolicy: { mode: "bounded" as const, maxOvertakes: 1 }, + runtime: clock, + }; + const clients = [new ExchangeClient({ ...base, transport: http }), new ExchangeClient({ ...base, transport: ws })]; + const action = buildOrder({ orders: [{ a: 0, b: true, p: "1", s: "1", r: false, t: { limit: { tif: "Gtc" } } }] }); + const vaultAddress = `0x${"a4".repeat(20)}`; + // The clock never advances, so no post can have waited for a refill. + await Promise.all(Array.from({ length: 6 }, (_, i) => clients[i % 2].execute(action, { vaultAddress }))); + expect(requests).toHaveLength(6); + expect(new Set(requests.map((request) => request.nonce)).size).toBe(6); + for (const request of requests) { + expect(request.vaultAddress).toBe(vaultAddress); + expect(request.action.type).toBe("multiSig"); + expect(request.action.signatures).toHaveLength(2); + } + // Posts were charged to the shared budget without waiting: a subscription frame now has to. + expect(posts).toBe(3); + expect(quota.connections).toBe(1); + expect(quota.acquireSend()).toBeInstanceOf(Promise); + } finally { + ws.close(); + clock.advance(10_000); + } + expect(clock.pendingTimers).toBe(0); +});