From 98a9b49324e0d3cfc271a143e673cbbf2f9b0195 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Mon, 20 Jul 2026 09:43:10 +0200 Subject: [PATCH 01/16] feat(perps-controller): add AggregatedOrderBookConnection service Add AggregatedOrderBookConnection service with processAggregatedOrderBook helper and export its public types from the package index. --- packages/perps-controller/src/index.ts | 9 + .../services/AggregatedOrderBookConnection.ts | 304 +++++++++++++ .../AggregatedOrderBookConnection.test.ts | 410 ++++++++++++++++++ 3 files changed, 723 insertions(+) create mode 100644 packages/perps-controller/src/services/AggregatedOrderBookConnection.ts create mode 100644 packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts diff --git a/packages/perps-controller/src/index.ts b/packages/perps-controller/src/index.ts index 6e3e15352a5..da819c8cb2e 100644 --- a/packages/perps-controller/src/index.ts +++ b/packages/perps-controller/src/index.ts @@ -594,6 +594,15 @@ export { // Services (only externally consumed items) export { TradingReadinessCache } from './services/TradingReadinessCache'; export type { ServiceContext } from './services/ServiceContext'; +export { + AggregatedOrderBookConnection, + processAggregatedOrderBook, +} from './services/AggregatedOrderBookConnection'; +export type { + OrderBookConnectionStatus, + SubscribeAggregatedOrderBookParams, + AggregatedOrderBookConnectionOptions, +} from './services/AggregatedOrderBookConnection'; // Removed with Live Market Prices component: // - usePerpsPrices diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts new file mode 100644 index 00000000000..06732917665 --- /dev/null +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -0,0 +1,304 @@ +import { WebSocketTransport } from '@nktkas/hyperliquid'; +import type { ISubscription } from '@nktkas/hyperliquid'; + +import type { OrderBookData } from '../types'; + +/** + * A single L2 book price level as delivered by Hyperliquid's `l2Book` + * subscription. Declared locally to avoid coupling to the SDK's exported type + * names (and to keep this the only file that references the SDK's shapes). + */ +type HyperliquidL2BookLevel = { + /** Price. */ + px: string; + /** Total size resting at this price. */ + sz: string; + /** Number of individual orders. */ + n: number; +}; + +/** `l2Book` snapshot event (index 0 = bids, index 1 = asks). */ +type HyperliquidL2BookEvent = { + coin: string; + time: number; + levels: [bids: HyperliquidL2BookLevel[], asks: HyperliquidL2BookLevel[]]; + spread?: string; +}; + +/** + * Health of the dedicated order-book socket, surfaced to the UI so the panel + * can show a reconnect affordance. + * + * - `connecting`: socket opening or reconnecting after a transient drop. + * - `connected`: subscription is live. + * - `error`: dropped and automatic reconnection was exhausted; needs a manual reconnect. + */ +export type OrderBookConnectionStatus = 'connecting' | 'connected' | 'error'; + +export type SubscribeAggregatedOrderBookParams = { + /** Market symbol (e.g. 'BTC'). */ + symbol: string; + /** Number of levels per side to keep. */ + levels?: number; + /** Server-side aggregation significant figures. */ + nSigFigs?: 2 | 3 | 4 | 5; + /** Mantissa refinement when `nSigFigs` is 5. */ + mantissa?: 2 | 5; + /** Invoked with each processed snapshot. */ + callback: (data: OrderBookData) => void; + /** Invoked when the underlying socket's health changes. */ + onStatusChange?: (status: OrderBookConnectionStatus) => void; +}; + +export type AggregatedOrderBookConnectionOptions = { + /** Resolves the current network at subscribe time. */ + isTestnet: () => boolean; +}; + +// Fast mode streams 5 levels per side (slow mode streams 20). We run fast mode +// for lower-latency ladder updates, so the book never carries more than this. +const DEFAULT_LEVELS = 5; + +/** + * Transforms a raw Hyperliquid `l2Book` snapshot into the `OrderBookData` shape + * the UI consumes. Mirrors the subscription service's internal + * `processOrderBookData` so this dedicated connection is a drop-in replacement + * for `subscribeToOrderBook` on the aggregated channel. + * + * @param data - Raw `l2Book` event. + * @param levels - Number of levels per side to keep. + * @returns Processed order-book snapshot. + */ +export function processAggregatedOrderBook( + data: HyperliquidL2BookEvent, + levels: number, +): OrderBookData { + const bidsRaw = data?.levels?.[0] ?? []; + const asksRaw = data?.levels?.[1] ?? []; + + let bidCumulativeSize = 0; + let bidCumulativeNotional = 0; + const bids = bidsRaw.slice(0, levels).map((level) => { + const price = Number.parseFloat(level.px); + const size = Number.parseFloat(level.sz); + const notional = price * size; + bidCumulativeSize += size; + bidCumulativeNotional += notional; + return { + price: level.px, + size: level.sz, + total: bidCumulativeSize.toString(), + notional: notional.toFixed(2), + totalNotional: bidCumulativeNotional.toFixed(2), + }; + }); + + let askCumulativeSize = 0; + let askCumulativeNotional = 0; + const asks = asksRaw.slice(0, levels).map((level) => { + const price = Number.parseFloat(level.px); + const size = Number.parseFloat(level.sz); + const notional = price * size; + askCumulativeSize += size; + askCumulativeNotional += notional; + return { + price: level.px, + size: level.sz, + total: askCumulativeSize.toString(), + notional: notional.toFixed(2), + totalNotional: askCumulativeNotional.toFixed(2), + }; + }); + + const bestBid = bids[0]; + const bestAsk = asks[0]; + const bidPrice = bestBid ? Number.parseFloat(bestBid.price) : 0; + const askPrice = bestAsk ? Number.parseFloat(bestAsk.price) : 0; + const spread = askPrice > 0 && bidPrice > 0 ? askPrice - bidPrice : 0; + const midPrice = askPrice > 0 && bidPrice > 0 ? (askPrice + bidPrice) / 2 : 0; + const spreadPercentage = + midPrice > 0 ? ((spread / midPrice) * 100).toFixed(4) : '0'; + const maxTotal = Math.max(bidCumulativeSize, askCumulativeSize).toString(); + + return { + bids, + asks, + spread: spread.toFixed(5), + spreadPercentage, + midPrice: midPrice.toFixed(5), + lastUpdated: Date.now(), + maxTotal, + }; +} + +/** + * Owns a dedicated Hyperliquid WebSocket connection used solely for the + * order-book panel's server-aggregated `l2Book` subscription. + * + * The main connection (managed by the subscription service) multiplexes every + * subscription onto a single socket. The Hyperliquid SDK dispatches `l2Book` + * events by `coin` only, so running the raw (full-precision) and the aggregated + * (`nSigFigs`) subscriptions for the same coin on that shared socket + * cross-contaminates them — the coarse ladder and the precise spread/slippage + * clobber each other. Giving the aggregated subscription its own socket removes + * the collision entirely: this socket only ever carries a single `l2Book` + * stream, and the main socket is never touched by the panel's grouping. + * + * The socket is created lazily on the first subscription and torn down once the + * last subscription is removed, so it exists only while an order-book panel is + * open. Because network is a global setting, the transport is recreated if + * `isTestnet` changes between (re)subscriptions. + */ +export class AggregatedOrderBookConnection { + readonly #isTestnet: () => boolean; + + #transport: WebSocketTransport | null = null; + + #transportIsTestnet = false; + + #activeCount = 0; + + // Set when the socket's auto-reconnection is exhausted (SDK `terminate` + // event). A terminated socket cannot recover, so the next subscribe must + // build a fresh transport instead of reusing the dead one. + #terminated = false; + + constructor({ isTestnet }: AggregatedOrderBookConnectionOptions) { + this.#isTestnet = isTestnet; + } + + /** + * Opens an aggregated `l2Book` subscription on the dedicated socket. + * + * Mirrors the subscription service's synchronous-unsubscribe contract: the + * returned function can be called before the async subscribe resolves and + * will cancel the pending subscription. + * + * @param params - Subscription parameters. + * @returns An unsubscribe function. + */ + subscribe(params: SubscribeAggregatedOrderBookParams): () => void { + const levels = params.levels ?? DEFAULT_LEVELS; + const transport = this.#ensureTransport(this.#isTestnet()); + const { socket } = transport; + + let cancelled = false; + let subscription: ISubscription | null = null; + this.#activeCount += 1; + + const reportStatus = (status: OrderBookConnectionStatus): void => { + if (!cancelled) { + params.onStatusChange?.(status); + } + }; + + // Reflect the socket's live health. `terminate` fires only once automatic + // reconnection (maxRetries) is exhausted — that is the unrecoverable state + // the UI surfaces with a manual reconnect button. + const handleOpen = (): void => reportStatus('connected'); + const handleClose = (): void => reportStatus('connecting'); + const handleTerminate = (): void => { + this.#terminated = true; + reportStatus('error'); + }; + socket.addEventListener('open', handleOpen); + socket.addEventListener('close', handleClose); + socket.addEventListener('terminate', handleTerminate); + + const removeSocketListeners = (): void => { + socket.removeEventListener('open', handleOpen); + socket.removeEventListener('close', handleClose); + socket.removeEventListener('terminate', handleTerminate); + }; + + reportStatus('connecting'); + + // The SDK's typed `l2Book` subscription drops unknown fields, so it can't + // request `fast` mode. Send the raw subscription payload through the + // transport directly (this is exactly what the typed method does, minus the + // validation) so `fast: true` reaches the server. The listener receives the + // SDK's `CustomEvent`, whose `detail` is the `l2Book` snapshot. + transport + .subscribe( + 'l2Book', + { + type: 'l2Book', + coin: params.symbol, + nSigFigs: params.nSigFigs ?? null, + mantissa: params.mantissa ?? null, + fast: true, + }, + (event: CustomEvent) => { + const data = event.detail; + if (cancelled || data?.coin !== params.symbol || !data?.levels) { + return; + } + params.callback(processAggregatedOrderBook(data, levels)); + }, + ) + .then(async (sub) => { + if (cancelled) { + try { + await sub.unsubscribe(); + } catch { + // Ignore cleanup errors on an already-cancelled subscription. + } + return undefined; + } + subscription = sub; + reportStatus('connected'); + return undefined; + }) + .catch(() => { + reportStatus('error'); + }); + + return () => { + if (cancelled) { + return; + } + cancelled = true; + removeSocketListeners(); + if (subscription) { + subscription.unsubscribe().catch(() => undefined); + subscription = null; + } + this.#activeCount = Math.max(0, this.#activeCount - 1); + if (this.#activeCount === 0) { + this.#closeTransport(); + } + }; + } + + /** Closes the dedicated socket and drops all subscriptions. */ + close(): void { + this.#closeTransport(); + } + + #ensureTransport(isTestnet: boolean): WebSocketTransport { + if ( + this.#transport && + this.#transportIsTestnet === isTestnet && + !this.#terminated + ) { + return this.#transport; + } + // First use, the network changed, or the previous socket was terminated — + // (re)create the dedicated transport. + this.#closeTransport(); + const transport = new WebSocketTransport({ isTestnet }); + this.#transport = transport; + this.#transportIsTestnet = isTestnet; + return transport; + } + + #closeTransport(): void { + const transport = this.#transport; + this.#transport = null; + this.#activeCount = 0; + this.#terminated = false; + if (transport) { + transport.close(); + } + } +} diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts new file mode 100644 index 00000000000..750058f09a9 --- /dev/null +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -0,0 +1,410 @@ +import * as hl from '@nktkas/hyperliquid'; + +import { + AggregatedOrderBookConnection, + processAggregatedOrderBook, +} from '../../../src/services/AggregatedOrderBookConnection'; + +type MockTransport = { + options: { isTestnet: boolean }; + close: jest.Mock; + socket: EventTarget; + subscribe: jest.Mock; +}; +/** Invokes the connection's listener with the raw snapshot (wrapped as `detail`). */ +type L2Emit = (data: unknown) => void; + +type MockState = { + transports: MockTransport[]; + listeners: { channel: string; params: unknown; listener: L2Emit }[]; + unsubscribe: jest.Mock; + resolveSubscribe: boolean; + rejectSubscribe: boolean; +}; + +jest.mock('@nktkas/hyperliquid', () => { + const state: MockState = { + transports: [], + listeners: [], + unsubscribe: jest.fn().mockResolvedValue(undefined), + resolveSubscribe: true, + rejectSubscribe: false, + }; + + class WebSocketTransport { + options: { isTestnet: boolean }; + + close = jest.fn(); + + socket = new EventTarget(); + + subscribe = jest.fn( + ( + channel: string, + params: unknown, + listener: (event: { detail: unknown }) => void, + ) => { + // Store an emitter that mirrors the SDK's CustomEvent delivery so tests + // can push a raw snapshot via `listeners[i].listener(rawData)`. + state.listeners.push({ + channel, + params, + listener: (detail: unknown) => listener({ detail }), + }); + if (state.rejectSubscribe) { + return Promise.reject(new Error('subscribe failed')); + } + return state.resolveSubscribe + ? Promise.resolve({ unsubscribe: state.unsubscribe }) + : new Promise(() => undefined); + }, + ); + + constructor(options: { isTestnet: boolean }) { + this.options = options; + state.transports.push(this as unknown as MockTransport); + } + } + + return { WebSocketTransport, mockState: state }; +}); + +const { mockState } = hl as unknown as { mockState: MockState }; + +const flush = async (): Promise => { + // Drain a few microtasks so both the resolve (.then) and reject (.then skip + // → .catch) branches of the subscribe promise chain settle. + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +}; + +const l2Level = ( + px: string, + sz: string, + count = 1, +): { px: string; sz: string; n: number } => ({ px, sz, n: count }); + +describe('processAggregatedOrderBook', () => { + it('builds cumulative totals, spread, and mid price from an l2Book snapshot', () => { + const result = processAggregatedOrderBook( + { + coin: 'BTC', + time: 1, + levels: [ + [l2Level('64000', '2'), l2Level('63000', '1')], + [l2Level('65000', '1'), l2Level('66000', '3')], + ], + }, + 10, + ); + + expect(result.bids).toStrictEqual([ + { + price: '64000', + size: '2', + total: '2', + notional: '128000.00', + totalNotional: '128000.00', + }, + { + price: '63000', + size: '1', + total: '3', + notional: '63000.00', + totalNotional: '191000.00', + }, + ]); + expect(result.asks[0]).toStrictEqual({ + price: '65000', + size: '1', + total: '1', + notional: '65000.00', + totalNotional: '65000.00', + }); + // spread = 65000 - 64000, mid = (65000 + 64000) / 2 + expect(result.spread).toBe('1000.00000'); + expect(result.midPrice).toBe('64500.00000'); + expect(result.maxTotal).toBe('4'); + }); + + it('trims each side to the requested level count', () => { + const result = processAggregatedOrderBook( + { + coin: 'BTC', + time: 1, + levels: [ + [l2Level('3', '1'), l2Level('2', '1'), l2Level('1', '1')], + [l2Level('4', '1'), l2Level('5', '1'), l2Level('6', '1')], + ], + }, + 2, + ); + + expect(result.bids).toHaveLength(2); + expect(result.asks).toHaveLength(2); + }); + + it('reports a zero spread when a side is empty', () => { + const result = processAggregatedOrderBook( + { coin: 'BTC', time: 1, levels: [[], []] }, + 10, + ); + expect(result.spread).toBe('0.00000'); + expect(result.spreadPercentage).toBe('0'); + expect(result.midPrice).toBe('0.00000'); + }); +}); + +describe('AggregatedOrderBookConnection', () => { + beforeEach(() => { + mockState.transports.length = 0; + mockState.listeners.length = 0; + // `resetMocks: true` (repo jest config) clears the shared mock's + // implementation before each test, so re-apply the resolved value. + mockState.unsubscribe.mockReset().mockResolvedValue(undefined); + mockState.resolveSubscribe = true; + mockState.rejectSubscribe = false; + }); + + it('creates a mainnet transport and subscribes with the requested params', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const callback = jest.fn(); + + connection.subscribe({ + symbol: 'BTC', + levels: 20, + nSigFigs: 2, + callback, + }); + + expect(mockState.transports).toHaveLength(1); + expect(mockState.transports[0].options).toStrictEqual({ isTestnet: false }); + expect(mockState.listeners[0].channel).toBe('l2Book'); + // Runs in fast mode (5 levels) via the raw subscription payload. + expect(mockState.listeners[0].params).toStrictEqual({ + type: 'l2Book', + coin: 'BTC', + nSigFigs: 2, + mantissa: null, + fast: true, + }); + }); + + it('creates a testnet transport when isTestnet resolves true', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => true, + }); + connection.subscribe({ symbol: 'ETH', callback: jest.fn() }); + expect(mockState.transports[0].options).toStrictEqual({ isTestnet: true }); + }); + + it('transforms snapshots and forwards them to the callback', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const callback = jest.fn(); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback }); + + mockState.listeners[0].listener({ + coin: 'BTC', + time: 1, + levels: [[l2Level('64000', '1')], [l2Level('65000', '1')]], + }); + + expect(callback).toHaveBeenCalledTimes(1); + expect(callback.mock.calls[0][0].midPrice).toBe('64500.00000'); + }); + + it('ignores snapshots for a different coin', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const callback = jest.fn(); + connection.subscribe({ symbol: 'BTC', callback }); + + mockState.listeners[0].listener({ + coin: 'ETH', + time: 1, + levels: [[l2Level('1', '1')], [l2Level('2', '1')]], + }); + + expect(callback).not.toHaveBeenCalled(); + }); + + it('reuses the same transport for a second subscription on the same network', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + connection.subscribe({ symbol: 'ETH', callback: jest.fn() }); + + expect(mockState.transports).toHaveLength(1); + expect(mockState.transports[0].subscribe).toHaveBeenCalledTimes(2); + }); + + it('closes the transport once the last subscription is removed', async () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const unsubA = connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + const unsubB = connection.subscribe({ symbol: 'ETH', callback: jest.fn() }); + await flush(); + + const [transport] = mockState.transports; + + unsubA(); + expect(transport.close).not.toHaveBeenCalled(); + + unsubB(); + expect(transport.close).toHaveBeenCalledTimes(1); + // The pending SDK subscriptions are also cancelled. + expect(mockState.unsubscribe).toHaveBeenCalledTimes(2); + }); + + it('recreates the transport when the network changes between subscriptions', () => { + let testnet = false; + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => testnet, + }); + connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + expect(mockState.transports).toHaveLength(1); + + testnet = true; + connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + expect(mockState.transports).toHaveLength(2); + expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); + expect(mockState.transports[1].options).toStrictEqual({ isTestnet: true }); + }); + + it('cancels a subscription that resolves after unsubscribe', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const callback = jest.fn(); + const unsub = connection.subscribe({ symbol: 'BTC', callback }); + + // Unsubscribe before the snapshot listener is invoked. + unsub(); + + mockState.listeners[0].listener({ + coin: 'BTC', + time: 1, + levels: [[l2Level('1', '1')], [l2Level('2', '1')]], + }); + + expect(callback).not.toHaveBeenCalled(); + }); + + it('closes the active transport on close()', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + connection.close(); + expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); + }); + + describe('connection status', () => { + it('reports connecting immediately then connected once the subscription resolves', async () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + onStatusChange, + }); + + expect(onStatusChange).toHaveBeenNthCalledWith(1, 'connecting'); + await flush(); + expect(onStatusChange).toHaveBeenCalledWith('connected'); + }); + + it('reports connected on socket open and connecting on socket close', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + onStatusChange, + }); + const { socket } = mockState.transports[0]; + + socket.dispatchEvent(new Event('open')); + expect(onStatusChange).toHaveBeenCalledWith('connected'); + + socket.dispatchEvent(new Event('close')); + expect(onStatusChange).toHaveBeenLastCalledWith('connecting'); + }); + + it('reports error when the socket terminates (reconnection exhausted)', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + onStatusChange, + }); + + mockState.transports[0].socket.dispatchEvent(new Event('terminate')); + expect(onStatusChange).toHaveBeenLastCalledWith('error'); + }); + + it('reports error when the subscription request rejects', async () => { + mockState.rejectSubscribe = true; + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + onStatusChange, + }); + + await flush(); + expect(onStatusChange).toHaveBeenLastCalledWith('error'); + }); + + it('stops reporting status after unsubscribe', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + const unsub = connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + onStatusChange, + }); + const { socket } = mockState.transports[0]; + unsub(); + onStatusChange.mockClear(); + + socket.dispatchEvent(new Event('terminate')); + expect(onStatusChange).not.toHaveBeenCalled(); + }); + + it('builds a fresh transport when resubscribing after a terminated socket', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const unsub = connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + mockState.transports[0].socket.dispatchEvent(new Event('terminate')); + + // Reconnect flow: tear the dead subscription down, then resubscribe. + unsub(); + connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + + expect(mockState.transports).toHaveLength(2); + expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); + }); + }); +}); From 5a1e08fbb4c345eb7982da5957529fe93ccc7fa7 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Mon, 20 Jul 2026 09:58:50 +0200 Subject: [PATCH 02/16] fix(perps-controller): fix AggregatedOrderBookConnection refcount handling Release the subscription refcount when a subscribe request fails so a failed subscribe no longer leaks the dedicated socket, and scope teardown to the transport a subscription was created on so a stale unsubscribe after a network-driven transport recreate cannot tear down the live socket while other subscriptions remain active. --- .../services/AggregatedOrderBookConnection.ts | 46 +++++++++++++------ .../AggregatedOrderBookConnection.test.ts | 42 +++++++++++++++++ 2 files changed, 73 insertions(+), 15 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 06732917665..931eaae00c8 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -211,6 +211,32 @@ export class AggregatedOrderBookConnection { socket.removeEventListener('terminate', handleTerminate); }; + // Releases this subscription's refcount and tears down the socket once no + // subscriptions remain. Idempotent via `cancelled`, so it's safe whether it + // runs from the returned unsubscribe or from a failed subscribe. + const teardown = (): void => { + if (cancelled) { + return; + } + cancelled = true; + removeSocketListeners(); + if (subscription) { + subscription.unsubscribe().catch(() => undefined); + subscription = null; + } + // Only touch the refcount/current socket if this subscription still + // belongs to the active transport. If the transport was recreated (network + // change or terminate), this subscription's socket is already dead and + // `#activeCount` now tracks only the new transport's subscriptions — so an + // older unsubscribe must not decrement it and tear down the live socket. + if (transport === this.#transport) { + this.#activeCount = Math.max(0, this.#activeCount - 1); + if (this.#activeCount === 0) { + this.#closeTransport(); + } + } + }; + reportStatus('connecting'); // The SDK's typed `l2Book` subscription drops unknown fields, so it can't @@ -250,24 +276,14 @@ export class AggregatedOrderBookConnection { return undefined; }) .catch(() => { + // Report before teardown flips `cancelled` (which gates status updates), + // then release the refcount this subscription reserved so a failed + // subscribe doesn't keep the dedicated socket open. reportStatus('error'); + teardown(); }); - return () => { - if (cancelled) { - return; - } - cancelled = true; - removeSocketListeners(); - if (subscription) { - subscription.unsubscribe().catch(() => undefined); - subscription = null; - } - this.#activeCount = Math.max(0, this.#activeCount - 1); - if (this.#activeCount === 0) { - this.#closeTransport(); - } - }; + return teardown; } /** Closes the dedicated socket and drops all subscriptions. */ diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index 750058f09a9..d1fedf7bd44 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -279,6 +279,28 @@ describe('AggregatedOrderBookConnection', () => { expect(mockState.transports[1].options).toStrictEqual({ isTestnet: true }); }); + it('does not tear down the new transport when an old subscription unsubscribes after a network change', () => { + let testnet = false; + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => testnet, + }); + const unsubOld = connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + + // Network flips, so the next subscribe recreates the transport. + testnet = true; + const unsubNew = connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + expect(mockState.transports).toHaveLength(2); + const [, newTransport] = mockState.transports; + + // The stale subscription's unsubscribe must not touch the live socket. + unsubOld(); + expect(newTransport.close).not.toHaveBeenCalled(); + + // The remaining live subscription still owns the socket and closes it. + unsubNew(); + expect(newTransport.close).toHaveBeenCalledTimes(1); + }); + it('cancels a subscription that resolves after unsubscribe', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => false, @@ -374,6 +396,26 @@ describe('AggregatedOrderBookConnection', () => { expect(onStatusChange).toHaveBeenLastCalledWith('error'); }); + it('releases the refcount and closes the transport when the subscription request rejects', async () => { + mockState.rejectSubscribe = true; + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const unsub = connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + }); + + await flush(); + + // The failed subscribe must not leave the dedicated socket open. + expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); + + // A subsequent unsubscribe is a no-op and must not double-close. + unsub(); + expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); + }); + it('stops reporting status after unsubscribe', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => false, From a2916e1a61c1fdcca9cb0c2fad763c7462089e50 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Mon, 20 Jul 2026 11:13:14 +0200 Subject: [PATCH 03/16] style(perps-controller): fix formatting in AggregatedOrderBookConnection test --- .../AggregatedOrderBookConnection.test.ts | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index d1fedf7bd44..15324af0e09 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -284,11 +284,17 @@ describe('AggregatedOrderBookConnection', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => testnet, }); - const unsubOld = connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + const unsubOld = connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + }); // Network flips, so the next subscribe recreates the transport. testnet = true; - const unsubNew = connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + const unsubNew = connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + }); expect(mockState.transports).toHaveLength(2); const [, newTransport] = mockState.transports; @@ -438,7 +444,10 @@ describe('AggregatedOrderBookConnection', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => false, }); - const unsub = connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + const unsub = connection.subscribe({ + symbol: 'BTC', + callback: jest.fn(), + }); mockState.transports[0].socket.dispatchEvent(new Event('terminate')); // Reconnect flow: tear the dead subscription down, then resubscribe. From 205247e133365e3eb6604ae4c9868c44bd6abc1a Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Mon, 20 Jul 2026 11:46:12 +0200 Subject: [PATCH 04/16] docs(perps-controller): add changelog entry for AggregatedOrderBookConnection --- packages/perps-controller/CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/packages/perps-controller/CHANGELOG.md b/packages/perps-controller/CHANGELOG.md index 7ae168f6b84..33735953ce1 100644 --- a/packages/perps-controller/CHANGELOG.md +++ b/packages/perps-controller/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Add `AggregatedOrderBookConnection` service (with the `processAggregatedOrderBook` helper and the `OrderBookConnectionStatus`, `SubscribeAggregatedOrderBookParams`, and `AggregatedOrderBookConnectionOptions` types) for managing a dedicated, reference-counted aggregated order book subscription ([#9549](https://github.com/MetaMask/core/pull/9549)) + ### Changed - Bump `@metamask/account-tree-controller` from `^7.5.3` to `7.5.4` ([#9429](https://github.com/MetaMask/core/pull/9429)) From 220195cfa83d92203c7162071562d4d4952957bd Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 03:29:38 +0200 Subject: [PATCH 05/16] fix(perps-controller): require nSigFigs for aggregated order book Omitting `nSigFigs` sent `null`, which requested the raw full-precision book instead of an aggregated one, contradicting the service contract. Make `nSigFigs` required on `SubscribeAggregatedOrderBookParams`. --- .../services/AggregatedOrderBookConnection.ts | 10 +++-- .../AggregatedOrderBookConnection.test.ts | 43 ++++++++++++++----- 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 931eaae00c8..2547d58e8d8 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -40,8 +40,12 @@ export type SubscribeAggregatedOrderBookParams = { symbol: string; /** Number of levels per side to keep. */ levels?: number; - /** Server-side aggregation significant figures. */ - nSigFigs?: 2 | 3 | 4 | 5; + /** + * Server-side aggregation significant figures. Required: omitting it would + * request the raw, full-precision book instead of an aggregated one, which + * contradicts this service's contract. + */ + nSigFigs: 2 | 3 | 4 | 5; /** Mantissa refinement when `nSigFigs` is 5. */ mantissa?: 2 | 5; /** Invoked with each processed snapshot. */ @@ -250,7 +254,7 @@ export class AggregatedOrderBookConnection { { type: 'l2Book', coin: params.symbol, - nSigFigs: params.nSigFigs ?? null, + nSigFigs: params.nSigFigs, mantissa: params.mantissa ?? null, fast: true, }, diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index 15324af0e09..0365be7af10 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -197,7 +197,7 @@ describe('AggregatedOrderBookConnection', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => true, }); - connection.subscribe({ symbol: 'ETH', callback: jest.fn() }); + connection.subscribe({ symbol: 'ETH', nSigFigs: 2, callback: jest.fn() }); expect(mockState.transports[0].options).toStrictEqual({ isTestnet: true }); }); @@ -223,7 +223,7 @@ describe('AggregatedOrderBookConnection', () => { isTestnet: (): boolean => false, }); const callback = jest.fn(); - connection.subscribe({ symbol: 'BTC', callback }); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback }); mockState.listeners[0].listener({ coin: 'ETH', @@ -238,8 +238,8 @@ describe('AggregatedOrderBookConnection', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => false, }); - connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); - connection.subscribe({ symbol: 'ETH', callback: jest.fn() }); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); + connection.subscribe({ symbol: 'ETH', nSigFigs: 2, callback: jest.fn() }); expect(mockState.transports).toHaveLength(1); expect(mockState.transports[0].subscribe).toHaveBeenCalledTimes(2); @@ -249,8 +249,16 @@ describe('AggregatedOrderBookConnection', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => false, }); - const unsubA = connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); - const unsubB = connection.subscribe({ symbol: 'ETH', callback: jest.fn() }); + const unsubA = connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + }); + const unsubB = connection.subscribe({ + symbol: 'ETH', + nSigFigs: 2, + callback: jest.fn(), + }); await flush(); const [transport] = mockState.transports; @@ -269,11 +277,11 @@ describe('AggregatedOrderBookConnection', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => testnet, }); - connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); expect(mockState.transports).toHaveLength(1); testnet = true; - connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); expect(mockState.transports).toHaveLength(2); expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); expect(mockState.transports[1].options).toStrictEqual({ isTestnet: true }); @@ -286,6 +294,7 @@ describe('AggregatedOrderBookConnection', () => { }); const unsubOld = connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), }); @@ -293,6 +302,7 @@ describe('AggregatedOrderBookConnection', () => { testnet = true; const unsubNew = connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), }); expect(mockState.transports).toHaveLength(2); @@ -312,7 +322,11 @@ describe('AggregatedOrderBookConnection', () => { isTestnet: (): boolean => false, }); const callback = jest.fn(); - const unsub = connection.subscribe({ symbol: 'BTC', callback }); + const unsub = connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback, + }); // Unsubscribe before the snapshot listener is invoked. unsub(); @@ -330,7 +344,7 @@ describe('AggregatedOrderBookConnection', () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => false, }); - connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); connection.close(); expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); }); @@ -343,6 +357,7 @@ describe('AggregatedOrderBookConnection', () => { const onStatusChange = jest.fn(); connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), onStatusChange, }); @@ -359,6 +374,7 @@ describe('AggregatedOrderBookConnection', () => { const onStatusChange = jest.fn(); connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), onStatusChange, }); @@ -378,6 +394,7 @@ describe('AggregatedOrderBookConnection', () => { const onStatusChange = jest.fn(); connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), onStatusChange, }); @@ -394,6 +411,7 @@ describe('AggregatedOrderBookConnection', () => { const onStatusChange = jest.fn(); connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), onStatusChange, }); @@ -409,6 +427,7 @@ describe('AggregatedOrderBookConnection', () => { }); const unsub = connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), }); @@ -429,6 +448,7 @@ describe('AggregatedOrderBookConnection', () => { const onStatusChange = jest.fn(); const unsub = connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), onStatusChange, }); @@ -446,13 +466,14 @@ describe('AggregatedOrderBookConnection', () => { }); const unsub = connection.subscribe({ symbol: 'BTC', + nSigFigs: 2, callback: jest.fn(), }); mockState.transports[0].socket.dispatchEvent(new Event('terminate')); // Reconnect flow: tear the dead subscription down, then resubscribe. unsub(); - connection.subscribe({ symbol: 'BTC', callback: jest.fn() }); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); expect(mockState.transports).toHaveLength(2); expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); From 358ddeb51b402c275625c8c974d8f0e9752ff2e0 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 03:31:30 +0200 Subject: [PATCH 06/16] fix(perps-controller): use finite reconnection for aggregated order book The dedicated aggregated `l2Book` transport was created with the SDK's default `maxRetries` of Infinity, so a sustained outage never exhausted reconnection and never reached the documented `error`/manual-reconnect state. Reuse `HYPERLIQUID_TRANSPORT_CONFIG` so this socket shares the package's five-attempt policy and stays consistent with the main WebSocket transport. --- .../services/AggregatedOrderBookConnection.ts | 12 +++++++-- .../AggregatedOrderBookConnection.test.ts | 27 ++++++++++++++----- 2 files changed, 31 insertions(+), 8 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 2547d58e8d8..731896f705c 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -1,6 +1,7 @@ import { WebSocketTransport } from '@nktkas/hyperliquid'; import type { ISubscription } from '@nktkas/hyperliquid'; +import { HYPERLIQUID_TRANSPORT_CONFIG } from '../constants/hyperLiquidConfig'; import type { OrderBookData } from '../types'; /** @@ -304,9 +305,16 @@ export class AggregatedOrderBookConnection { return this.#transport; } // First use, the network changed, or the previous socket was terminated — - // (re)create the dedicated transport. + // (re)create the dedicated transport. Reuse the package's transport config + // so this socket shares the finite five-attempt reconnection policy; without + // it the SDK defaults `maxRetries` to Infinity and a sustained outage would + // never exhaust reconnection to reach the `error`/manual-reconnect state. this.#closeTransport(); - const transport = new WebSocketTransport({ isTestnet }); + const transport = new WebSocketTransport({ + isTestnet, + ...HYPERLIQUID_TRANSPORT_CONFIG, + reconnect: HYPERLIQUID_TRANSPORT_CONFIG.reconnect, + }); this.#transport = transport; this.#transportIsTestnet = isTestnet; return transport; diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index 0365be7af10..3b481b86bf0 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -1,12 +1,13 @@ import * as hl from '@nktkas/hyperliquid'; +import { HYPERLIQUID_TRANSPORT_CONFIG } from '../../../src/constants/hyperLiquidConfig'; import { AggregatedOrderBookConnection, processAggregatedOrderBook, } from '../../../src/services/AggregatedOrderBookConnection'; type MockTransport = { - options: { isTestnet: boolean }; + options: Record; close: jest.Mock; socket: EventTarget; subscribe: jest.Mock; @@ -32,7 +33,7 @@ jest.mock('@nktkas/hyperliquid', () => { }; class WebSocketTransport { - options: { isTestnet: boolean }; + options: Record; close = jest.fn(); @@ -60,7 +61,7 @@ jest.mock('@nktkas/hyperliquid', () => { }, ); - constructor(options: { isTestnet: boolean }) { + constructor(options: Record) { this.options = options; state.transports.push(this as unknown as MockTransport); } @@ -181,7 +182,13 @@ describe('AggregatedOrderBookConnection', () => { }); expect(mockState.transports).toHaveLength(1); - expect(mockState.transports[0].options).toStrictEqual({ isTestnet: false }); + // The dedicated transport reuses the package's transport config so it shares + // the finite five-attempt reconnection policy (not the SDK's Infinity). + expect(mockState.transports[0].options).toStrictEqual({ + isTestnet: false, + ...HYPERLIQUID_TRANSPORT_CONFIG, + reconnect: HYPERLIQUID_TRANSPORT_CONFIG.reconnect, + }); expect(mockState.listeners[0].channel).toBe('l2Book'); // Runs in fast mode (5 levels) via the raw subscription payload. expect(mockState.listeners[0].params).toStrictEqual({ @@ -198,7 +205,11 @@ describe('AggregatedOrderBookConnection', () => { isTestnet: (): boolean => true, }); connection.subscribe({ symbol: 'ETH', nSigFigs: 2, callback: jest.fn() }); - expect(mockState.transports[0].options).toStrictEqual({ isTestnet: true }); + expect(mockState.transports[0].options).toStrictEqual({ + isTestnet: true, + ...HYPERLIQUID_TRANSPORT_CONFIG, + reconnect: HYPERLIQUID_TRANSPORT_CONFIG.reconnect, + }); }); it('transforms snapshots and forwards them to the callback', () => { @@ -284,7 +295,11 @@ describe('AggregatedOrderBookConnection', () => { connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); expect(mockState.transports).toHaveLength(2); expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); - expect(mockState.transports[1].options).toStrictEqual({ isTestnet: true }); + expect(mockState.transports[1].options).toStrictEqual({ + isTestnet: true, + ...HYPERLIQUID_TRANSPORT_CONFIG, + reconnect: HYPERLIQUID_TRANSPORT_CONFIG.reconnect, + }); }); it('does not tear down the new transport when an old subscription unsubscribes after a network change', () => { From d82a3e1654d7ab4d843a519dd3ff146dff8869e1 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 03:34:42 +0200 Subject: [PATCH 07/16] fix(perps-controller): enforce one l2Book payload per asset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dedicated socket dispatches `l2Book` events by `coin` only, so two subscriptions for the same asset with different params (e.g. `nSigFigs`) would cross-contaminate — the exact collision this connection exists to avoid. Track the active payload per asset and reject a conflicting subscribe, while still allowing identical repeat subscriptions. --- .../services/AggregatedOrderBookConnection.ts | 59 +++++++++++++++-- .../AggregatedOrderBookConnection.test.ts | 65 +++++++++++++++++++ 2 files changed, 117 insertions(+), 7 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 731896f705c..9e562129472 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -163,6 +163,14 @@ export class AggregatedOrderBookConnection { #activeCount = 0; + // Tracks the single `l2Book` payload the dedicated socket carries per asset, + // keyed by symbol. The SDK dispatches `l2Book` events by `coin` only, so two + // subscriptions for the same asset with different params (e.g. `nSigFigs`) + // would cross-contaminate on this shared socket — exactly the collision this + // connection exists to avoid. `count` refcounts the (identical) subscriptions + // sharing a payload so the entry is dropped once the last one unsubscribes. + readonly #payloads = new Map(); + // Set when the socket's auto-reconnection is exhausted (SDK `terminate` // event). A terminated socket cannot recover, so the next subscribe must // build a fresh transport instead of reusing the dead one. @@ -179,17 +187,52 @@ export class AggregatedOrderBookConnection { * returned function can be called before the async subscribe resolves and * will cancel the pending subscription. * + * Only one `l2Book` payload per asset may be active at a time. Subscribing to + * an asset that already has a live subscription with different params (e.g. a + * different `nSigFigs` or `mantissa`) throws, because the shared socket + * dispatches by `coin` and the conflicting streams would clobber each other. + * * @param params - Subscription parameters. * @returns An unsubscribe function. + * @throws If the asset already has an active subscription with different params. */ subscribe(params: SubscribeAggregatedOrderBookParams): () => void { const levels = params.levels ?? DEFAULT_LEVELS; + // The `l2Book` payload the transport carries. `levels` is client-side only + // (it slices each snapshot), so it is deliberately excluded from the payload + // and its signature. + const payload = { + type: 'l2Book' as const, + coin: params.symbol, + nSigFigs: params.nSigFigs, + mantissa: params.mantissa ?? null, + fast: true, + }; + const signature = JSON.stringify(payload); + const transport = this.#ensureTransport(this.#isTestnet()); + + // Reject a conflicting payload for an asset already on this socket. A + // recreated transport (first use, network change, or terminate) starts with + // an empty payload map, so this can only trip on the reuse path — the shared + // socket that would actually suffer the collision. + const existingPayload = this.#payloads.get(params.symbol); + if (existingPayload && existingPayload.signature !== signature) { + throw new Error( + `AggregatedOrderBookConnection: "${params.symbol}" is already subscribed with different params; only one l2Book payload per asset is allowed on the dedicated socket.`, + ); + } + const { socket } = transport; let cancelled = false; let subscription: ISubscription | null = null; this.#activeCount += 1; + if (existingPayload) { + existingPayload.count += 1; + } else { + this.#payloads.set(params.symbol, { signature, count: 1 }); + } const reportStatus = (status: OrderBookConnectionStatus): void => { if (!cancelled) { @@ -236,6 +279,13 @@ export class AggregatedOrderBookConnection { // older unsubscribe must not decrement it and tear down the live socket. if (transport === this.#transport) { this.#activeCount = Math.max(0, this.#activeCount - 1); + const entry = this.#payloads.get(params.symbol); + if (entry) { + entry.count -= 1; + if (entry.count <= 0) { + this.#payloads.delete(params.symbol); + } + } if (this.#activeCount === 0) { this.#closeTransport(); } @@ -252,13 +302,7 @@ export class AggregatedOrderBookConnection { transport .subscribe( 'l2Book', - { - type: 'l2Book', - coin: params.symbol, - nSigFigs: params.nSigFigs, - mantissa: params.mantissa ?? null, - fast: true, - }, + payload, (event: CustomEvent) => { const data = event.detail; if (cancelled || data?.coin !== params.symbol || !data?.levels) { @@ -324,6 +368,7 @@ export class AggregatedOrderBookConnection { const transport = this.#transport; this.#transport = null; this.#activeCount = 0; + this.#payloads.clear(); this.#terminated = false; if (transport) { transport.close(); diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index 3b481b86bf0..b20de4ef97b 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -256,6 +256,71 @@ describe('AggregatedOrderBookConnection', () => { expect(mockState.transports[0].subscribe).toHaveBeenCalledTimes(2); }); + it('rejects a second subscription for the same asset with different params', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); + + expect(() => + connection.subscribe({ symbol: 'BTC', nSigFigs: 5, callback: jest.fn() }), + ).toThrow( + 'AggregatedOrderBookConnection: "BTC" is already subscribed with different params', + ); + // The conflicting subscribe must not reach the transport. + expect(mockState.transports[0].subscribe).toHaveBeenCalledTimes(1); + }); + + it('allows a repeat subscription for the same asset with identical params', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); + + expect(() => + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }), + ).not.toThrow(); + expect(mockState.transports).toHaveLength(1); + expect(mockState.transports[0].subscribe).toHaveBeenCalledTimes(2); + }); + + it('treats a different levels value as identical (levels is client-side only)', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + levels: 5, + callback: jest.fn(), + }); + + expect(() => + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + levels: 20, + callback: jest.fn(), + }), + ).not.toThrow(); + }); + + it('allows different params for the same asset once the prior subscription is removed', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const unsub = connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + }); + unsub(); + + expect(() => + connection.subscribe({ symbol: 'BTC', nSigFigs: 5, callback: jest.fn() }), + ).not.toThrow(); + }); + it('closes the transport once the last subscription is removed', async () => { const connection = new AggregatedOrderBookConnection({ isTestnet: (): boolean => false, From 67b3b13127eb2e1c9d45f002f98a51f14bd3ed45 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 03:39:57 +0200 Subject: [PATCH 08/16] fix(perps-controller): detect termination via terminationSignal The reconnecting socket never emits a `terminate` event, so the previous listener could never fire and the documented `error`/manual-reconnect state was unreachable. Permanent termination aborts the socket's `terminationSignal` right before the final `close`, so inspect that signal in the close handler (ignoring `TERMINATED_BY_USER` for our own `close()`). Tests now drive the real signal instead of dispatching a nonexistent `terminate` event. --- .../services/AggregatedOrderBookConnection.ts | 33 +++++++---- .../AggregatedOrderBookConnection.test.ts | 59 +++++++++++++++++-- 2 files changed, 75 insertions(+), 17 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 9e562129472..77addb5a8d0 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -171,9 +171,9 @@ export class AggregatedOrderBookConnection { // sharing a payload so the entry is dropped once the last one unsubscribes. readonly #payloads = new Map(); - // Set when the socket's auto-reconnection is exhausted (SDK `terminate` - // event). A terminated socket cannot recover, so the next subscribe must - // build a fresh transport instead of reusing the dead one. + // Set when the socket's auto-reconnection is exhausted (its + // `terminationSignal` aborts). A terminated socket cannot recover, so the next + // subscribe must build a fresh transport instead of reusing the dead one. #terminated = false; constructor({ isTestnet }: AggregatedOrderBookConnectionOptions) { @@ -240,23 +240,32 @@ export class AggregatedOrderBookConnection { } }; - // Reflect the socket's live health. `terminate` fires only once automatic - // reconnection (maxRetries) is exhausted — that is the unrecoverable state - // the UI surfaces with a manual reconnect button. + // Reflect the socket's live health. Every drop dispatches a `close` event; + // the reconnecting socket only exposes permanent termination through its + // `terminationSignal` (an `AbortSignal`), which it aborts *before* the final + // close. So an aborted signal on close — unless it was our own `close()` + // (`TERMINATED_BY_USER`) — means automatic reconnection is exhausted: the + // unrecoverable state the UI surfaces with a manual reconnect button. A + // still-live signal means a transient drop the socket will auto-reconnect. const handleOpen = (): void => reportStatus('connected'); - const handleClose = (): void => reportStatus('connecting'); - const handleTerminate = (): void => { - this.#terminated = true; - reportStatus('error'); + const handleClose = (): void => { + const { terminationSignal } = socket; + const terminatedByUser = + (terminationSignal.reason as { code?: string } | undefined)?.code === + 'TERMINATED_BY_USER'; + if (terminationSignal.aborted && !terminatedByUser) { + this.#terminated = true; + reportStatus('error'); + return; + } + reportStatus('connecting'); }; socket.addEventListener('open', handleOpen); socket.addEventListener('close', handleClose); - socket.addEventListener('terminate', handleTerminate); const removeSocketListeners = (): void => { socket.removeEventListener('open', handleOpen); socket.removeEventListener('close', handleClose); - socket.removeEventListener('terminate', handleTerminate); }; // Releases this subscription's refcount and tears down the socket once no diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index b20de4ef97b..e2dc699f983 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -6,10 +6,20 @@ import { processAggregatedOrderBook, } from '../../../src/services/AggregatedOrderBookConnection'; +/** + * Mirrors the reconnecting socket: a `terminationSignal` plus a helper that + * reproduces permanent termination (abort the signal, then a final `close`). + */ +type MockSocket = EventTarget & { + terminationSignal: AbortSignal; + /** Simulate permanent termination with the given `ReconnectingWebSocketError` code. */ + terminate: (code?: string) => void; +}; + type MockTransport = { options: Record; close: jest.Mock; - socket: EventTarget; + socket: MockSocket; subscribe: jest.Mock; }; /** Invokes the connection's listener with the raw snapshot (wrapped as `detail`). */ @@ -32,12 +42,30 @@ jest.mock('@nktkas/hyperliquid', () => { rejectSubscribe: false, }; + // Reproduces the reconnecting socket's termination model: permanent + // termination aborts `terminationSignal` (reason carries a `code`) *before* + // the final `close` event fires — never a standalone `terminate` event. + class MockSocket extends EventTarget { + readonly #abortController = new AbortController(); + + get terminationSignal(): AbortSignal { + return this.#abortController.signal; + } + + terminate(code = 'RECONNECTION_LIMIT'): void { + if (!this.#abortController.signal.aborted) { + this.#abortController.abort({ code }); + } + this.dispatchEvent(new Event('close')); + } + } + class WebSocketTransport { options: Record; close = jest.fn(); - socket = new EventTarget(); + socket = new MockSocket(); subscribe = jest.fn( ( @@ -479,10 +507,31 @@ describe('AggregatedOrderBookConnection', () => { onStatusChange, }); - mockState.transports[0].socket.dispatchEvent(new Event('terminate')); + // Reconnection exhausted: the socket aborts its `terminationSignal` then + // dispatches a final `close`. + mockState.transports[0].socket.terminate(); expect(onStatusChange).toHaveBeenLastCalledWith('error'); }); + it('reports connecting (not error) when close fires from an intentional close()', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + onStatusChange, + }); + + // A user-triggered close aborts the signal with `TERMINATED_BY_USER`; that + // must not be surfaced as the unrecoverable `error` state. + mockState.transports[0].socket.terminate('TERMINATED_BY_USER'); + expect(onStatusChange).toHaveBeenLastCalledWith('connecting'); + expect(onStatusChange).not.toHaveBeenCalledWith('error'); + }); + it('reports error when the subscription request rejects', async () => { mockState.rejectSubscribe = true; const connection = new AggregatedOrderBookConnection({ @@ -536,7 +585,7 @@ describe('AggregatedOrderBookConnection', () => { unsub(); onStatusChange.mockClear(); - socket.dispatchEvent(new Event('terminate')); + socket.terminate(); expect(onStatusChange).not.toHaveBeenCalled(); }); @@ -549,7 +598,7 @@ describe('AggregatedOrderBookConnection', () => { nSigFigs: 2, callback: jest.fn(), }); - mockState.transports[0].socket.dispatchEvent(new Event('terminate')); + mockState.transports[0].socket.terminate(); // Reconnect flow: tear the dead subscription down, then resubscribe. unsub(); From dba92a531bb286c85cdffc48381f7404e413161a Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 12:26:05 +0200 Subject: [PATCH 09/16] fix(perps-controller): keep terminal error status on late subscribe resolve Once the aggregated order book socket terminates (reconnection exhausted), the error status is terminal until teardown/resubscribe. Previously a subscribe promise resolving after termination could call reportStatus('connected'), flipping the UI back to a healthy state on a dead socket and hiding the manual-reconnect affordance. Suppress any late connected/connecting reports once terminated. --- .../services/AggregatedOrderBookConnection.ts | 13 +++++++++-- .../AggregatedOrderBookConnection.test.ts | 23 +++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 77addb5a8d0..744c969b710 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -234,10 +234,18 @@ export class AggregatedOrderBookConnection { this.#payloads.set(params.symbol, { signature, count: 1 }); } + // Set once this subscription's socket terminates (reconnection exhausted). + // The `error` state is terminal until teardown/resubscribe, so once set we + // suppress any late `connected`/`connecting` — e.g. from a subscribe promise + // that resolves *after* the socket died — which would otherwise flip the UI + // back to a healthy state on a dead socket and hide the manual-reconnect + // affordance. + let terminated = false; const reportStatus = (status: OrderBookConnectionStatus): void => { - if (!cancelled) { - params.onStatusChange?.(status); + if (cancelled || (terminated && status !== 'error')) { + return; } + params.onStatusChange?.(status); }; // Reflect the socket's live health. Every drop dispatches a `close` event; @@ -255,6 +263,7 @@ export class AggregatedOrderBookConnection { 'TERMINATED_BY_USER'; if (terminationSignal.aborted && !terminatedByUser) { this.#terminated = true; + terminated = true; reportStatus('error'); return; } diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index e2dc699f983..64005913a7f 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -532,6 +532,29 @@ describe('AggregatedOrderBookConnection', () => { expect(onStatusChange).not.toHaveBeenCalledWith('error'); }); + it('stays in error (not connected) when the subscribe promise resolves after the socket terminates', async () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + onStatusChange, + }); + + // Reconnection is exhausted before the pending subscribe promise settles. + mockState.transports[0].socket.terminate(); + expect(onStatusChange).toHaveBeenLastCalledWith('error'); + + // The pending subscribe promise now resolves; it must not flip the UI back + // to `connected` on the dead socket. + await flush(); + expect(onStatusChange).toHaveBeenLastCalledWith('error'); + expect(onStatusChange).not.toHaveBeenCalledWith('connected'); + }); + it('reports error when the subscription request rejects', async () => { mockState.rejectSubscribe = true; const connection = new AggregatedOrderBookConnection({ From 57b79b354edafd92b22591a98d6275a542db9aa6 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 12:28:55 +0200 Subject: [PATCH 10/16] fix(perps-controller): surface post-confirmation subscription failures The SDK reports failures of an already-confirmed subscription (e.g. the server rejecting a re-subscription after a reconnect) only through the subscribe options.onError callback, removing the listener without further events. The aggregated order book connection never passed onError, so such a failure left the panel showing connected with a frozen book while the refcount still tracked a live subscription. Pass an idempotent onError handler that reports error and tears down, reused from the initial-failure .catch path. --- .../services/AggregatedOrderBookConnection.ts | 28 +++++++++---- .../AggregatedOrderBookConnection.test.ts | 40 ++++++++++++++++++- 2 files changed, 59 insertions(+), 9 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 744c969b710..3c669fab103 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -310,6 +310,20 @@ export class AggregatedOrderBookConnection { } }; + // Surfaces a subscription failure the same way regardless of when it + // happens: report `error` (before teardown flips `cancelled`, which gates + // status updates) then release the refcount so the dead subscription doesn't + // keep the dedicated socket open. Used for both the initial subscribe + // rejection (`.catch`) and post-confirmation failures the SDK reports only + // through `onError` — e.g. the server rejecting the re-subscription after a + // reconnect, which removes the listener and stops all further events (a + // frozen order book that would otherwise still read as `connected`). + // Idempotent via `teardown`'s `cancelled` guard. + const handleSubscriptionError = (): void => { + reportStatus('error'); + teardown(); + }; + reportStatus('connecting'); // The SDK's typed `l2Book` subscription drops unknown fields, so it can't @@ -328,6 +342,12 @@ export class AggregatedOrderBookConnection { } params.callback(processAggregatedOrderBook(data, levels)); }, + // `onError` fires at most once for an *already confirmed* subscription + // that later fails (rejected re-subscription after reconnect, permanent + // termination, or a drop while re-subscription is disabled). The SDK + // removes the listener and emits nothing further, so treat it exactly + // like an initial failure. + { onError: handleSubscriptionError }, ) .then(async (sub) => { if (cancelled) { @@ -342,13 +362,7 @@ export class AggregatedOrderBookConnection { reportStatus('connected'); return undefined; }) - .catch(() => { - // Report before teardown flips `cancelled` (which gates status updates), - // then release the refcount this subscription reserved so a failed - // subscribe doesn't keep the dedicated socket open. - reportStatus('error'); - teardown(); - }); + .catch(handleSubscriptionError); return teardown; } diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index 64005913a7f..d97a5c35402 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -27,7 +27,13 @@ type L2Emit = (data: unknown) => void; type MockState = { transports: MockTransport[]; - listeners: { channel: string; params: unknown; listener: L2Emit }[]; + listeners: { + channel: string; + params: unknown; + listener: L2Emit; + /** The SDK's post-confirmation failure callback, if the caller passed one. */ + onError?: (error: Error) => void; + }[]; unsubscribe: jest.Mock; resolveSubscribe: boolean; rejectSubscribe: boolean; @@ -72,13 +78,17 @@ jest.mock('@nktkas/hyperliquid', () => { channel: string, params: unknown, listener: (event: { detail: unknown }) => void, + options?: { onError?: (error: Error) => void }, ) => { // Store an emitter that mirrors the SDK's CustomEvent delivery so tests - // can push a raw snapshot via `listeners[i].listener(rawData)`. + // can push a raw snapshot via `listeners[i].listener(rawData)`, plus the + // SDK's post-confirmation `onError` callback so tests can simulate a + // rejected re-subscription via `listeners[i].onError(err)`. state.listeners.push({ channel, params, listener: (detail: unknown) => listener({ detail }), + onError: options?.onError, }); if (state.rejectSubscribe) { return Promise.reject(new Error('subscribe failed')); @@ -555,6 +565,32 @@ describe('AggregatedOrderBookConnection', () => { expect(onStatusChange).not.toHaveBeenCalledWith('connected'); }); + it('errors and tears down when a confirmed subscription is rejected on resubscribe', async () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + onStatusChange, + }); + + // The subscription is confirmed first. + await flush(); + expect(onStatusChange).toHaveBeenLastCalledWith('connected'); + + // After a reconnect the server rejects the re-subscription: the SDK + // invokes `onError` and removes the listener (no further events follow). + // The service must surface `error` and tear down rather than keep + // reporting `connected` with a frozen order book. + mockState.listeners[0].onError?.(new Error('resubscribe rejected')); + + expect(onStatusChange).toHaveBeenLastCalledWith('error'); + expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); + }); + it('reports error when the subscription request rejects', async () => { mockState.rejectSubscribe = true; const connection = new AggregatedOrderBookConnection({ From 0dd848349bc1e020728af4b409e464015a94fa9e Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 12:33:36 +0200 Subject: [PATCH 11/16] refactor(perps-controller): subscribe via typed l2Book client @nktkas/hyperliquid@0.33.1 added `fast` to the typed l2Book schema and forwards subscription options (including onError), so the raw transport.subscribe workaround is no longer needed. Switch the aggregated order book connection to SubscriptionClient.l2Book to restore payload validation and drop the outdated comment about the typed API dropping unknown fields. --- .../services/AggregatedOrderBookConnection.ts | 33 ++++++++--------- .../AggregatedOrderBookConnection.test.ts | 36 ++++++++++++++++++- 2 files changed, 49 insertions(+), 20 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 3c669fab103..9e1d15d72d5 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -1,4 +1,4 @@ -import { WebSocketTransport } from '@nktkas/hyperliquid'; +import { SubscriptionClient, WebSocketTransport } from '@nktkas/hyperliquid'; import type { ISubscription } from '@nktkas/hyperliquid'; import { HYPERLIQUID_TRANSPORT_CONFIG } from '../constants/hyperLiquidConfig'; @@ -198,17 +198,16 @@ export class AggregatedOrderBookConnection { */ subscribe(params: SubscribeAggregatedOrderBookParams): () => void { const levels = params.levels ?? DEFAULT_LEVELS; - // The `l2Book` payload the transport carries. `levels` is client-side only - // (it slices each snapshot), so it is deliberately excluded from the payload - // and its signature. - const payload = { - type: 'l2Book' as const, + // The `l2Book` subscription params the socket carries. `levels` is + // client-side only (it slices each snapshot), so it is deliberately excluded + // from the params and their signature. + const l2BookParams = { coin: params.symbol, nSigFigs: params.nSigFigs, mantissa: params.mantissa ?? null, - fast: true, + fast: true as const, }; - const signature = JSON.stringify(payload); + const signature = JSON.stringify(l2BookParams); const transport = this.#ensureTransport(this.#isTestnet()); @@ -326,17 +325,13 @@ export class AggregatedOrderBookConnection { reportStatus('connecting'); - // The SDK's typed `l2Book` subscription drops unknown fields, so it can't - // request `fast` mode. Send the raw subscription payload through the - // transport directly (this is exactly what the typed method does, minus the - // validation) so `fast: true` reaches the server. The listener receives the - // SDK's `CustomEvent`, whose `detail` is the `l2Book` snapshot. - transport - .subscribe( - 'l2Book', - payload, - (event: CustomEvent) => { - const data = event.detail; + // Subscribe through the typed `l2Book` client so the params are validated + // before they reach the wire (`fast: true` requests fast mode — 5 levels at + // ~0.5s). The listener receives the decoded snapshot directly. + new SubscriptionClient({ transport }) + .l2Book( + l2BookParams, + (data: HyperliquidL2BookEvent) => { if (cancelled || data?.coin !== params.symbol || !data?.levels) { return; } diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index d97a5c35402..6d7c8b1a3ef 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -105,7 +105,41 @@ jest.mock('@nktkas/hyperliquid', () => { } } - return { WebSocketTransport, mockState: state }; + type SubscribingTransport = { + subscribe: ( + channel: string, + params: unknown, + listener: (event: { detail: unknown }) => void, + options?: { onError?: (error: Error) => void }, + ) => Promise; + }; + + // Mirrors the typed subscription client: `l2Book` validates/normalizes the + // params, prepends `type: 'l2Book'`, and delegates to `transport.subscribe`, + // unwrapping the CustomEvent so the caller's listener receives the snapshot + // directly. Forwards `options` (`onError`) unchanged. + class SubscriptionClient { + readonly #transport: SubscribingTransport; + + constructor({ transport }: { transport: SubscribingTransport }) { + this.#transport = transport; + } + + async l2Book( + params: Record, + listener: (data: unknown) => void, + options?: { onError?: (error: Error) => void }, + ): Promise { + return this.#transport.subscribe( + 'l2Book', + { type: 'l2Book', ...params }, + (event) => listener(event.detail), + options, + ); + } + } + + return { WebSocketTransport, SubscriptionClient, mockState: state }; }); const { mockState } = hl as unknown as { mockState: MockState }; From 9b33f370f9ec4a7562e6dfe1f05a9f6f536de572 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 12:52:43 +0200 Subject: [PATCH 12/16] fix(perps-controller): ignore stale transport on in-flight subscribe When the dedicated transport is replaced (network flip / recreate) before an in-flight l2Book subscribe settles, the promise success path stored the subscription and reported connected on the now-dead socket. Gate reportStatus and the success path on transport === this.#transport (mirroring teardown), unsubscribing the stale subscription instead of announcing connected. --- .../services/AggregatedOrderBookConnection.ts | 19 ++++++++-- .../AggregatedOrderBookConnection.test.ts | 35 +++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 9e1d15d72d5..c921b6fd1a5 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -241,7 +241,16 @@ export class AggregatedOrderBookConnection { // affordance. let terminated = false; const reportStatus = (status: OrderBookConnectionStatus): void => { - if (cancelled || (terminated && status !== 'error')) { + // Suppress reports from a subscription that no longer drives the UI: it + // was unsubscribed (`cancelled`), its transport was replaced by a network + // flip or recreate (`transport !== this.#transport`, so its socket is + // dead), or its socket permanently terminated and `error` is now sticky + // until teardown (`terminated`). + if ( + cancelled || + transport !== this.#transport || + (terminated && status !== 'error') + ) { return; } params.onStatusChange?.(status); @@ -345,11 +354,15 @@ export class AggregatedOrderBookConnection { { onError: handleSubscriptionError }, ) .then(async (sub) => { - if (cancelled) { + // Stale if this subscription was unsubscribed (`cancelled`) or its + // transport was replaced (network flip / recreate) before the subscribe + // settled — either way the captured socket is dead, so clean up the SDK + // subscription instead of storing it or announcing `connected`. + if (cancelled || transport !== this.#transport) { try { await sub.unsubscribe(); } catch { - // Ignore cleanup errors on an already-cancelled subscription. + // Ignore cleanup errors on an already-cancelled/stale subscription. } return undefined; } diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index 6d7c8b1a3ef..b52b7ee7598 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -625,6 +625,41 @@ describe('AggregatedOrderBookConnection', () => { expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); }); + it('does not report connected when the transport is replaced before an in-flight subscribe resolves', async () => { + let testnet = false; + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => testnet, + }); + const onStatusChangeOld = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + onStatusChange: onStatusChangeOld, + }); + + // Flip the network before the first subscribe settles: this recreates the + // transport, so the first (still-pending) subscription is now stale. + testnet = true; + const onStatusChangeNew = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + onStatusChange: onStatusChangeNew, + }); + expect(mockState.transports).toHaveLength(2); + + await flush(); + + // The stale subscription must not announce `connected` on the dead socket, + // and its SDK subscription is cleaned up rather than stored. + expect(onStatusChangeOld).not.toHaveBeenCalledWith('connected'); + expect(mockState.unsubscribe).toHaveBeenCalled(); + // The live subscription on the new transport still reports connected. + expect(onStatusChangeNew).toHaveBeenLastCalledWith('connected'); + }); + it('reports error when the subscription request rejects', async () => { mockState.rejectSubscribe = true; const connection = new AggregatedOrderBookConnection({ From eaab90be3cfb81e70326abd29ef7c8264bf94c80 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 17:20:43 +0200 Subject: [PATCH 13/16] fix(perps-controller): keep #9549 entry under Unreleased after 9.3.0 release main released perps-controller 9.3.0, moving the previously-unreleased entries into a versioned section. The merge auto-placed this PR's new AggregatedOrderBookConnection entry into 9.3.0; move it back under [Unreleased] so the merge-queue changelog check passes. --- packages/perps-controller/CHANGELOG.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/perps-controller/CHANGELOG.md b/packages/perps-controller/CHANGELOG.md index 6d36631864f..5848ba2da26 100644 --- a/packages/perps-controller/CHANGELOG.md +++ b/packages/perps-controller/CHANGELOG.md @@ -7,11 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Add `AggregatedOrderBookConnection` service (with the `processAggregatedOrderBook` helper and the `OrderBookConnectionStatus`, `SubscribeAggregatedOrderBookParams`, and `AggregatedOrderBookConnectionOptions` types) for managing a dedicated, reference-counted aggregated order book subscription ([#9549](https://github.com/MetaMask/core/pull/9549)) + ## [9.3.0] ### Added -- Add `AggregatedOrderBookConnection` service (with the `processAggregatedOrderBook` helper and the `OrderBookConnectionStatus`, `SubscribeAggregatedOrderBookParams`, and `AggregatedOrderBookConnectionOptions` types) for managing a dedicated, reference-counted aggregated order book subscription ([#9549](https://github.com/MetaMask/core/pull/9549)) - Add `proLayoutPreferences` state field (`orderBookExpanded`, `chartExpanded`, `orderBookPosition`, `orderFormPosition`) to `PerpsControllerState` for persisting Pro-mode layout across markets, along with the exported `ProLayoutPreferences` type and `DEFAULT_PRO_LAYOUT_PREFERENCES` constant, `getProLayoutPreferences()` / `setProLayoutPreferences(patch)` controller methods (exposed as messenger actions with exported `PerpsControllerGetProLayoutPreferencesAction` / `PerpsControllerSetProLayoutPreferencesAction` types), and a `selectProLayoutPreferences` selector; the getter and selector merge over defaults so callers always receive a fully-populated object ([#9550](https://github.com/MetaMask/core/pull/9550)) - Add a `PerpsMode` enum (`Lite`/`Pro`) and a persisted `mode` state field (defaulting to `PerpsMode.Lite`) to `PerpsControllerState`, along with an exported `DEFAULT_PERPS_MODE` constant, a `setPerpsMode(mode)` controller method (exposed as a messenger action with an exported `PerpsControllerSetPerpsModeAction` type), and a `selectPerpsMode` selector that falls back to the default mode ([#9550](https://github.com/MetaMask/core/pull/9550)) From 1cfd42cb1458c7e9703caf4f98ad09d0d937bb2a Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 17:32:16 +0200 Subject: [PATCH 14/16] fix(perps-controller): tear down orphaned subscriptions on transport rebuild When #closeTransport rebuilt the socket (network flip, post-termination resubscribe, or close()) it cleared refcounts without ending subscriptions still live on the old transport, leaking their SDK subscription/listeners and leaving callers frozen with no status update. Track a force-terminate callback per subscription and run them before detaching the transport so each orphaned caller is notified (error) and cleaned up. --- .../services/AggregatedOrderBookConnection.ts | 45 +++++++++++++++++++ .../AggregatedOrderBookConnection.test.ts | 36 +++++++++++++++ 2 files changed, 81 insertions(+) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index c921b6fd1a5..8029f16ebd1 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -171,6 +171,14 @@ export class AggregatedOrderBookConnection { // sharing a payload so the entry is dropped once the last one unsubscribes. readonly #payloads = new Map(); + // Force-terminate callback for every currently-active subscription. When a + // transport rebuild (`#closeTransport`) shuts the socket down out from under + // live subscriptions, these tear each one down — notifying the caller and + // releasing its SDK subscription / socket listeners — instead of orphaning + // them (stale handle, no more updates, and no further status because + // reporting is suppressed once `transport !== this.#transport`). + readonly #activeSubscriptions = new Set<() => void>(); + // Set when the socket's auto-reconnection is exhausted (its // `terminationSignal` aborts). A terminated socket cannot recover, so the next // subscribe must build a fresh transport instead of reusing the dead one. @@ -285,6 +293,33 @@ export class AggregatedOrderBookConnection { socket.removeEventListener('close', handleClose); }; + // Ends this subscription when its transport is torn down beneath it (network + // flip, post-termination resubscribe, or `close()`). Unlike `teardown` it + // leaves the shared refcount/payload state alone — `#closeTransport` clears + // those wholesale — but still notifies the caller and releases this + // subscription's resources so nothing leaks on the dead socket. Runs while + // `#transport` still points at the old transport, so the `error` report is + // not suppressed by the stale-transport guard. + const forceTerminate = (): void => { + if (cancelled) { + return; + } + // A subscription whose socket already terminated has reported `error`; + // a still-live one (e.g. abandoned by a network flip) needs the terminal + // signal so the caller stops trusting a now-dead book. + if (!terminated) { + reportStatus('error'); + } + cancelled = true; + removeSocketListeners(); + this.#activeSubscriptions.delete(forceTerminate); + if (subscription) { + subscription.unsubscribe().catch(() => undefined); + subscription = null; + } + }; + this.#activeSubscriptions.add(forceTerminate); + // Releases this subscription's refcount and tears down the socket once no // subscriptions remain. Idempotent via `cancelled`, so it's safe whether it // runs from the returned unsubscribe or from a failed subscribe. @@ -294,6 +329,7 @@ export class AggregatedOrderBookConnection { } cancelled = true; removeSocketListeners(); + this.#activeSubscriptions.delete(forceTerminate); if (subscription) { subscription.unsubscribe().catch(() => undefined); subscription = null; @@ -406,6 +442,15 @@ export class AggregatedOrderBookConnection { #closeTransport(): void { const transport = this.#transport; + // Tear down any subscriptions still live on this transport *before* + // detaching it, so orphaned callers are notified (`error`) and their SDK + // subscriptions / socket listeners are released instead of leaking on a + // dead socket. Snapshot the set since each callback mutates it. Subscriptions + // torn down normally have already removed themselves, so this is a no-op for + // them. + for (const forceTerminate of [...this.#activeSubscriptions]) { + forceTerminate(); + } this.#transport = null; this.#activeCount = 0; this.#payloads.clear(); diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index b52b7ee7598..bd3ae791de3 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -660,6 +660,42 @@ describe('AggregatedOrderBookConnection', () => { expect(onStatusChangeNew).toHaveBeenLastCalledWith('connected'); }); + it('terminates orphaned subscriptions when the transport is rebuilt on a network flip', async () => { + let testnet = false; + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => testnet, + }); + const onStatusChangeOld = jest.fn(); + const callbackOld = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: callbackOld, + onStatusChange: onStatusChangeOld, + }); + await flush(); + expect(onStatusChangeOld).toHaveBeenLastCalledWith('connected'); + + // Flip the network and subscribe again: this rebuilds the transport, + // orphaning the first subscription that was live on the old socket. + testnet = true; + connection.subscribe({ symbol: 'ETH', nSigFigs: 2, callback: jest.fn() }); + + // The orphaned subscription is notified with `error` and its SDK + // subscription is cleaned up rather than left dangling on the dead socket. + expect(onStatusChangeOld).toHaveBeenLastCalledWith('error'); + expect(mockState.unsubscribe).toHaveBeenCalled(); + expect(mockState.transports[0].close).toHaveBeenCalledTimes(1); + + // A late snapshot on the orphaned listener must be ignored. + mockState.listeners[0].listener({ + coin: 'BTC', + time: 1, + levels: [[l2Level('1', '1')], [l2Level('2', '1')]], + }); + expect(callbackOld).not.toHaveBeenCalled(); + }); + it('reports error when the subscription request rejects', async () => { mockState.rejectSubscribe = true; const connection = new AggregatedOrderBookConnection({ From 5bba037f5600e60604a8ccfab49c34b290f27407 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Tue, 21 Jul 2026 17:52:28 +0200 Subject: [PATCH 15/16] fix(perps-controller): handle reentrant subscribe during transport teardown forceTerminate notifies via onStatusChange, which can synchronously re-enter subscribe. #closeTransport ran callbacks while #transport still pointed at the dying socket, so a reentrant subscribe reused it and leaked its listeners and SDK subscription past close. Detach all shared state (set, #transport, refcounts) before invoking callbacks so reentrant subscribes build a fresh transport, and reuse it in #ensureTransport instead of orphaning it. --- .../services/AggregatedOrderBookConnection.ts | 49 +++++++++++++------ .../AggregatedOrderBookConnection.test.ts | 38 ++++++++++++++ 2 files changed, 71 insertions(+), 16 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 8029f16ebd1..1f4fc0f0189 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -297,18 +297,21 @@ export class AggregatedOrderBookConnection { // flip, post-termination resubscribe, or `close()`). Unlike `teardown` it // leaves the shared refcount/payload state alone — `#closeTransport` clears // those wholesale — but still notifies the caller and releases this - // subscription's resources so nothing leaks on the dead socket. Runs while - // `#transport` still points at the old transport, so the `error` report is - // not suppressed by the stale-transport guard. + // subscription's resources so nothing leaks on the dead socket. const forceTerminate = (): void => { if (cancelled) { return; } - // A subscription whose socket already terminated has reported `error`; - // a still-live one (e.g. abandoned by a network flip) needs the terminal - // signal so the caller stops trusting a now-dead book. + // Notify directly rather than via `reportStatus`: `#closeTransport` + // detaches `#transport` before invoking these callbacks (so a reentrant + // subscribe from this handler builds a fresh transport instead of binding + // to the dying one), which would otherwise trip `reportStatus`'s + // stale-transport guard. A subscription whose socket already terminated + // has reported `error`; a still-live one (abandoned by a network flip or + // `close()`) needs the terminal signal so the caller stops trusting a + // now-dead book. if (!terminated) { - reportStatus('error'); + params.onStatusChange?.('error'); } cancelled = true; removeSocketListeners(); @@ -430,6 +433,16 @@ export class AggregatedOrderBookConnection { // it the SDK defaults `maxRetries` to Infinity and a sustained outage would // never exhaust reconnection to reach the `error`/manual-reconnect state. this.#closeTransport(); + // `#closeTransport` notifies subscribers, which may synchronously re-enter + // `subscribe` and build a matching transport. Reuse it instead of orphaning + // it (which would leak the reentrant subscription on an unreferenced socket). + if ( + this.#transport && + this.#transportIsTestnet === isTestnet && + !this.#terminated + ) { + return this.#transport; + } const transport = new WebSocketTransport({ isTestnet, ...HYPERLIQUID_TRANSPORT_CONFIG, @@ -442,15 +455,16 @@ export class AggregatedOrderBookConnection { #closeTransport(): void { const transport = this.#transport; - // Tear down any subscriptions still live on this transport *before* - // detaching it, so orphaned callers are notified (`error`) and their SDK - // subscriptions / socket listeners are released instead of leaking on a - // dead socket. Snapshot the set since each callback mutates it. Subscriptions - // torn down normally have already removed themselves, so this is a no-op for - // them. - for (const forceTerminate of [...this.#activeSubscriptions]) { - forceTerminate(); - } + // Snapshot the subscriptions to force-terminate, then detach ALL shared + // state (the set, `#transport`, refcounts) *before* invoking any callback. + // Those callbacks notify subscribers via `onStatusChange`, which can + // synchronously re-enter `subscribe`; detaching first guarantees a reentrant + // subscribe builds a fresh transport (rather than reusing this dying one) + // and registers itself in a clean set (rather than being swept up by, or + // lingering past, this teardown). Subscriptions torn down normally have + // already removed themselves, so their entry is a no-op here. + const subscriptions = [...this.#activeSubscriptions]; + this.#activeSubscriptions.clear(); this.#transport = null; this.#activeCount = 0; this.#payloads.clear(); @@ -458,5 +472,8 @@ export class AggregatedOrderBookConnection { if (transport) { transport.close(); } + for (const forceTerminate of subscriptions) { + forceTerminate(); + } } } diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index bd3ae791de3..3c4a80930f2 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -696,6 +696,44 @@ describe('AggregatedOrderBookConnection', () => { expect(callbackOld).not.toHaveBeenCalled(); }); + it('binds a reentrant subscribe from an onStatusChange handler to a fresh transport', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + let reentered = false; + const onStatusChange = jest.fn((status: string) => { + // Re-subscribe synchronously from within the terminal `error` + // notification emitted while the transport is being torn down. + if (status === 'error' && !reentered) { + reentered = true; + connection.subscribe({ + symbol: 'ETH', + nSigFigs: 2, + callback: jest.fn(), + }); + } + }); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + onStatusChange, + }); + const [firstTransport] = mockState.transports; + + // Closing force-terminates BTC, whose `error` handler re-subscribes. The + // reentrant subscription must bind to a NEW transport, not the dying one. + connection.close(); + + expect(reentered).toBe(true); + expect(firstTransport.close).toHaveBeenCalledTimes(1); + expect(mockState.transports).toHaveLength(2); + const newTransport = mockState.transports[1]; + expect(newTransport).not.toBe(firstTransport); + // The reentrant subscription's transport is left live (not orphaned). + expect(newTransport.close).not.toHaveBeenCalled(); + }); + it('reports error when the subscription request rejects', async () => { mockState.rejectSubscribe = true; const connection = new AggregatedOrderBookConnection({ From 6951101dcfba1e07d99d5a4aa68d4d99d9eb2ad6 Mon Sep 17 00:00:00 2001 From: Michal Szorad Date: Wed, 22 Jul 2026 10:26:59 +0200 Subject: [PATCH 16/16] fix(perps-controller): detach listeners before closing dedicated transport Force-terminate subscriptions (detaching their socket listeners) before closing the transport in #closeTransport, so a final `close` on an already-exhausted socket can't re-set #terminated after it was cleared and cause the next subscribe to tear down the healthy replacement socket. Also guard handleClose against mutating shared state once its subscription is cancelled. --- .../services/AggregatedOrderBookConnection.ts | 18 +++++++-- .../AggregatedOrderBookConnection.test.ts | 37 ++++++++++++++++++- 2 files changed, 50 insertions(+), 5 deletions(-) diff --git a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts index 1f4fc0f0189..7ff4dc0ec52 100644 --- a/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts +++ b/packages/perps-controller/src/services/AggregatedOrderBookConnection.ts @@ -273,6 +273,13 @@ export class AggregatedOrderBookConnection { // still-live signal means a transient drop the socket will auto-reconnect. const handleOpen = (): void => reportStatus('connected'); const handleClose = (): void => { + // A torn-down subscription must not mutate shared connection state. Its + // listeners are normally detached before the socket closes, but guard + // anyway so a late `close` (e.g. from `transport.close()` racing listener + // removal) can't wrongly flip `#terminated`. + if (cancelled) { + return; + } const { terminationSignal } = socket; const terminatedByUser = (terminationSignal.reason as { code?: string } | undefined)?.code === @@ -469,11 +476,16 @@ export class AggregatedOrderBookConnection { this.#activeCount = 0; this.#payloads.clear(); this.#terminated = false; - if (transport) { - transport.close(); - } + // Force-terminate (which detaches each subscription's socket listeners) + // BEFORE closing the transport. `close()` on an already-exhausted socket + // dispatches a final `close`; if a stale `handleClose` were still attached + // it would re-set `#terminated` right after we cleared it, making the next + // `#ensureTransport` tear down the healthy replacement socket. for (const forceTerminate of subscriptions) { forceTerminate(); } + if (transport) { + transport.close(); + } } } diff --git a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts index 3c4a80930f2..84815b22ecc 100644 --- a/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts +++ b/packages/perps-controller/tests/src/services/AggregatedOrderBookConnection.test.ts @@ -69,10 +69,15 @@ jest.mock('@nktkas/hyperliquid', () => { class WebSocketTransport { options: Record; - close = jest.fn(); - socket = new MockSocket(); + // Mirror the SDK: closing aborts the termination signal (as + // `TERMINATED_BY_USER` if not already aborted) and dispatches a final + // `close` event. + close = jest.fn(() => { + this.socket.terminate('TERMINATED_BY_USER'); + }); + subscribe = jest.fn( ( channel: string, @@ -734,6 +739,34 @@ describe('AggregatedOrderBookConnection', () => { expect(newTransport.close).not.toHaveBeenCalled(); }); + it('does not leave the connection flagged terminated after closing an exhausted socket', () => { + const connection = new AggregatedOrderBookConnection({ + isTestnet: (): boolean => false, + }); + const onStatusChange = jest.fn(); + connection.subscribe({ + symbol: 'BTC', + nSigFigs: 2, + callback: jest.fn(), + onStatusChange, + }); + + // Reconnection is exhausted on the dedicated socket. + mockState.transports[0].socket.terminate(); + expect(onStatusChange).toHaveBeenLastCalledWith('error'); + + // Resubscribe rebuilds the transport. Closing the exhausted socket + // dispatches a final `close`; that must not re-flag the connection as + // terminated after it was cleared. + connection.subscribe({ symbol: 'BTC', nSigFigs: 2, callback: jest.fn() }); + expect(mockState.transports).toHaveLength(2); + + // A further subscribe on the same network must REUSE the healthy new + // transport rather than tearing it down as if it were dead. + connection.subscribe({ symbol: 'ETH', nSigFigs: 2, callback: jest.fn() }); + expect(mockState.transports).toHaveLength(2); + }); + it('reports error when the subscription request rejects', async () => { mockState.rejectSubscribe = true; const connection = new AggregatedOrderBookConnection({