diff --git a/docs/clients.md b/docs/clients.md index 2262bdf0..93dd5034 100644 --- a/docs/clients.md +++ b/docs/clients.md @@ -395,12 +395,13 @@ const subscription = await client.allMids((data) => { ### Errors Each subscription method takes an optional `options` argument — `{ signal?, onError? }`. The `onError` callback runs at -most once, when an already confirmed subscription fails: +most once per subscribe call, when an already confirmed subscription fails: - the server rejects a re-subscription after a [reconnect](transports.md#reconnection); - the connection is permanently terminated; - the connection goes down while [re-subscription](transports.md#resubscription) is disabled. +When several calls share one underlying subscription (see [unsubscribe](#unsubscribe)), every caller's `onError` fires. Failures before confirmation reject the subscribe promise instead. After `onError` fires, the subscription is removed and no further events arrive: @@ -418,6 +419,22 @@ const subscription = await client.allMids( ); ``` +The same failure is also exposed on the subscription handle as `failureSignal` — an `AbortSignal` that aborts with the +failure `TransportError` as its reason, and never on a voluntary `unsubscribe()`. It makes a dead feed observable even +when no `onError` was passed. The signal is always present on a `WebSocketTransport` subscription (it is optional only +for third-party transports): + +```ts +const subscription = await client.allMids((data) => { + console.log(data.mids); +}); + +subscription.failureSignal?.addEventListener("abort", () => { + // The subscription is gone — inspect the reason and re-subscribe if needed + console.error(subscription.failureSignal?.reason); +}); +``` + ### Unsubscribe Hyperliquid allows diff --git a/docs/transports.md b/docs/transports.md index 7722e51a..f2f85024 100644 --- a/docs/transports.md +++ b/docs/transports.md @@ -253,8 +253,8 @@ by hand. Delivery pauses while the connection is down and resumes once it's back const transport = new WebSocketTransport({ resubscribe: false }); ``` -If a subscription then fails to re-establish, its `onError` callback is invoked. Handle it as shown under -[subscription errors](clients.md#errors). +If a subscription then fails to re-establish, every subscriber's `onError` callback is invoked and each subscription +handle's `failureSignal` aborts. Handle it as shown under [subscription errors](clients.md#errors). ### WebSocket limits diff --git a/src/api/subscription/_methods/_base/_config.ts b/src/api/subscription/_methods/_base/_config.ts index 595c33c8..459f9d52 100644 --- a/src/api/subscription/_methods/_base/_config.ts +++ b/src/api/subscription/_methods/_base/_config.ts @@ -16,13 +16,15 @@ export interface SubscriptionOptions { /** Stops waiting for the confirmation and detaches the listener. */ signal?: AbortSignal; /** - * Callback invoked at most once, when an already confirmed subscription fails: + * Callback invoked at most once per subscribe call, when an already confirmed subscription fails: * - the server rejects a re-subscription after a reconnect; * - the connection is permanently terminated; * - the connection goes down while re-subscription is disabled. * + * When several calls share one underlying subscription, every caller's callback fires. * Failures before the confirmation reject the subscribe promise instead. * After the callback fires, the subscription is removed and no further events or errors follow. + * The same failure also aborts the resolved handle's `failureSignal`. */ onError?: (error: TransportError) => void; } diff --git a/src/transport/_base.ts b/src/transport/_base.ts index 869cf737..ae0d26a5 100644 --- a/src/transport/_base.ts +++ b/src/transport/_base.ts @@ -50,13 +50,15 @@ export interface ISubscriptionTransport { /** Stops waiting for the confirmation and detaches the listener. */ signal?: AbortSignal; /** - * Callback invoked at most once, when an already confirmed subscription fails: + * Callback invoked at most once per `subscribe()` call, when an already confirmed subscription fails: * - the server rejects a re-subscription after a reconnect; * - the connection is permanently terminated; * - the connection goes down while re-subscription is disabled. * + * When several calls share one underlying subscription, every caller's callback fires. * Failures before the confirmation reject the `subscribe()` promise instead. * After the callback fires, the subscription is removed and no further events or errors follow. + * The same failure also aborts the resolved handle's {@linkcode ISubscription.failureSignal}. */ onError?: (error: TransportError) => void; }, @@ -67,6 +69,15 @@ export interface ISubscriptionTransport { export interface ISubscription { /** Removes the event listener and unsubscribes from the event channel. */ unsubscribe(): Promise; + /** + * Aborts — with the failure {@linkcode TransportError} as its reason — when this already + * confirmed subscription fails, making a dying feed observable without passing `onError`. + * Never aborts on a voluntary {@linkcode ISubscription.unsubscribe | unsubscribe()}. + * + * Optional so that {@linkcode ISubscriptionTransport} implementations outside this package + * stay valid; the built-in WebSocket transport always provides it. + */ + readonly failureSignal?: AbortSignal; } /** diff --git a/src/transport/websocket/_subscriptionManager.ts b/src/transport/websocket/_subscriptionManager.ts index bf95dcd2..053fd941 100644 --- a/src/transport/websocket/_subscriptionManager.ts +++ b/src/transport/websocket/_subscriptionManager.ts @@ -22,12 +22,27 @@ interface RegistrationHandle { /** The call's error callback, invoked on subscription failure only while this lease is live. */ onError?: (error: WebSocketRequestError) => void; /** - * Whether this lease's `subscribe()` call resolved with the subscription live. A failure - * is reported through exactly one channel: the pending `subscribe()` promise before that - * (an unconfirmed lease never became a subscriber — its `onError` does not fire), the - * first live confirmed lease's `onError` after. + * Whether this lease's `subscribe()` call resolved with the subscription live. Each lease + * observes a failure through exactly one stage: the pending `subscribe()` promise rejects + * before that (an unconfirmed lease never became a subscriber — its `onError` does not fire, + * and no handle carrying a `failureSignal` was ever returned, so nothing dangles), while + * every live confirmed lease gets its `onError` call and its `failureSignal` abort after. */ confirmed: boolean; + /** + * Lazily created controller behind the returned handle's `failureSignal`: most subscribers + * never read the signal, so the common path allocates no AbortController. `_failSubscription` + * aborts it (with the failure as reason) for a live confirmed lease; a voluntary + * `unsubscribe()` leaves it untouched. + */ + failureController?: AbortController; + /** + * The failure this live confirmed lease was torn down with, recorded by `_failSubscription` + * so a `failureSignal` first accessed after the failure is created already aborted with the + * same reason a pre-failure access would have observed. Never set on a lease that retired + * voluntarily before the failure — its signal must stay inert. + */ + failure?: WebSocketRequestError; } /** Per-listener registration: its routed event type, live leases, and confirmation state. */ @@ -49,9 +64,9 @@ interface ListenerRegistration { * aborting waiter can therefore never roll back a registration an identical joiner still * awaits, and one handle's `unsubscribe()` cannot cut off another live handle. * - * Error ownership is first-live-owner: the first live confirmed lease in insertion order - * owns the failure callback, and when it retires the next live lease promotes — a failure - * fires exactly one `onError`, never a dead lease's. + * A failure notifies every live confirmed lease — each one's `onError` and `failureSignal`, + * once per `subscribe()` call — and never a dead or unconfirmed lease's; see + * `_failSubscription`. */ handles: Set; } @@ -146,13 +161,17 @@ export class WebSocketSubscriptionManager { * Subscribes to a Hyperliquid event channel. * * @param options.signal Stops waiting for the confirmation and detaches the listener. - * @param options.onError Callback invoked at most once, when an already confirmed subscription fails: + * @param options.onError Callback invoked at most once per `subscribe()` call, when an already confirmed subscription fails: * - the server rejects a re-subscription after a reconnect; * - the connection is permanently terminated; * - the connection goes down while re-subscription is disabled. * + * When several calls share one underlying subscription, every caller's callback fires. * Failures before the confirmation reject the `subscribe()` promise instead. * After the callback fires, the subscription is removed and no further events or errors follow. + * @return A handle whose `failureSignal` aborts with the same failure `onError` reports — and + * never on a voluntary `unsubscribe()` — so a subscriber without `onError` still + * observes a dying feed. * * @throws {WebSocketRequestError} When the subscription request fails or limits are exceeded. */ @@ -280,7 +299,21 @@ export class WebSocketSubscriptionManager { throw error; } - return { unsubscribe }; + return { + unsubscribe, + // Lazily materialized: most subscribers never read the signal, so the common path pays + // no AbortController per call. Repeat accesses return the same signal object; a first + // access after the failure creates the controller already aborted with the recorded + // reason, indistinguishable from one aborted while being observed. + get failureSignal(): AbortSignal { + let controller = handle.failureController; + if (controller === undefined) { + controller = handle.failureController = new AbortController(); + if (handle.failure !== undefined) controller.abort(handle.failure); + } + return controller.signal; + }, + }; } // =========================================================================== @@ -402,10 +435,11 @@ export class WebSocketSubscriptionManager { // =========================================================================== /** - * Removes the subscription with all its listeners, then notifies each - * confirmed listener's `onError` once. Sends nothing to the server: every - * caller deals with a subscription the server no longer serves — refused, - * or cut off by a close. + * Removes the subscription with all its listeners, then notifies every live + * confirmed lease: aborts its `failureSignal` and invokes its `onError`, each + * callback isolated so one throwing `onError` cannot silence the rest. + * Sends nothing to the server: every caller deals with a subscription the + * server no longer serves — refused, or cut off by a close. */ private _failSubscription(id: string, subscription: SubscriptionState, error: WebSocketRequestError): void { if (this._subscriptions.get(id) !== subscription) return; @@ -414,22 +448,39 @@ export class WebSocketSubscriptionManager { // with this failure instead of resolving a handle into the dead subscription. subscription.failure = error; + // Pass 1 — detach listeners and snapshot every live confirmed lease, recording the + // failure on each BEFORE any user code runs. Both notification channels invoke user + // code synchronously (`AbortController.abort` runs abort listeners, `onError` is a + // callback), and that code can re-enter `unsubscribe()` on a sibling lease — mutating + // `registration.handles` mid-iteration would then skip a lease that was live at + // failure time. The snapshot fixes the notified set at exactly that moment, and the + // pre-recorded failure means a `failureSignal` first accessed inside an earlier + // callback already observes the reason, whichever lease it belongs to. + const confirmed: RegistrationHandle[] = []; for (const [listener, registrations] of subscription.listeners) { for (const registration of registrations.values()) { this._hlEvents.removeEventListener(registration.eventType, listener); - // First-live-owner: exactly one live, confirmed lease — the first in insertion - // order — is notified. Unconfirmed leases observe the failure through their - // subscribe() rejection; dead leases were removed from the set already. + // Every live confirmed lease is notified, through both of its channels. Unconfirmed + // leases observe the failure through their subscribe() rejection; dead leases were + // removed from the set already. for (const handle of registration.handles) { if (!handle.confirmed) continue; - try { - handle.onError?.(error); - } catch { - // A throwing onError must not affect other listeners. - } - break; + handle.failure = error; + confirmed.push(handle); } } } + + // Pass 2 — notify off the snapshot. A lease another callback retired mid-teardown is + // still notified: it was live when the subscription died, and its `unsubscribe()` + // against the already-removed subscription was a no-op, not a disavowal of the failure. + for (const handle of confirmed) { + handle.failureController?.abort(error); + try { + handle.onError?.(error); + } catch { + // A throwing onError must not affect the other leases or listeners. + } + } } } diff --git a/src/transport/websocket/mod.ts b/src/transport/websocket/mod.ts index 7488bb7f..0de636d5 100644 --- a/src/transport/websocket/mod.ts +++ b/src/transport/websocket/mod.ts @@ -214,7 +214,10 @@ export class WebSocketTransport implements IRequestTransport<"info" | "exchange" * @param payload The payload to send with the subscription request. * @param listener The function to call when the event is dispatched. * @param options Subscription options; see {@linkcode WebSocketSubscriptionManager.subscribe}. - * @return A promise that resolves with a subscription handle once the server confirms the subscription. + * @return A promise that resolves with a subscription handle once the server confirms the + * subscription. On this transport the handle always carries a `failureSignal`, which + * aborts with the same failure `onError` reports — and never on a voluntary + * `unsubscribe()` — so a subscriber without `onError` still observes a dying feed. * * @throws {WebSocketRequestError} An error that occurs when the subscription request fails. * @@ -238,13 +241,15 @@ export class WebSocketTransport implements IRequestTransport<"info" | "exchange" /** Stops waiting for the confirmation and detaches the listener. */ signal?: AbortSignal; /** - * Callback invoked at most once, when an already confirmed subscription fails: + * Callback invoked at most once per `subscribe()` call, when an already confirmed subscription fails: * - the server rejects a re-subscription after a reconnect; * - the connection is permanently terminated; * - the connection goes down while re-subscription is disabled. * + * When several calls share one underlying subscription, every caller's callback fires. * Failures before the confirmation reject the `subscribe()` promise instead. * After the callback fires, the subscription is removed and no further events or errors follow. + * The same failure also aborts the resolved handle's `failureSignal`. */ onError?: (error: WebSocketRequestError) => void; }, diff --git a/tests/transport/websocket/_subscriptionManager.test.ts b/tests/transport/websocket/_subscriptionManager.test.ts index 5e53efea..887913f6 100644 --- a/tests/transport/websocket/_subscriptionManager.test.ts +++ b/tests/transport/websocket/_subscriptionManager.test.ts @@ -790,7 +790,7 @@ describe("WebSocketSubscriptionManager", () => { assertEquals((errors[0] as WebSocketRequestError).cause, socket.terminationSignal.reason); }); - test("onError: only the first live owner's callback fires on failure", async () => { + test("onError: every live confirmed lease on a shared registration fires on failure", async () => { const { socket, manager } = createManager(true); const payload = { channel: "test", extra: "data" }; @@ -810,9 +810,31 @@ describe("WebSocketSubscriptionManager", () => { socket.terminate(new Error("x")); await drain(); - // First-live-owner: the first live confirmed lease owns the failure callback. + // Both calls subscribed, so both observe the failure — once each, never more. assertEquals(first, 1); - assertEquals(second, 0); + assertEquals(second, 1); + }); + + test("onError: every subscriber fires on failure across different listeners", async () => { + const { socket, manager } = createManager(true); + const payload = { channel: "test", extra: "data" }; + + // Two subscribe calls with distinct listener callbacks: two registrations on one entry. + const seenA: unknown[] = []; + const seenB: unknown[] = []; + const p1 = manager.subscribe("test", payload, () => {}, { onError: (e) => seenA.push(e) }); + socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); + await p1; + await manager.subscribe("test", payload, () => {}, { onError: (e) => seenB.push(e) }); + + socket.terminate(new Error("gone for good")); + await drain(); + + assertEquals(seenA.length, 1); + assertEquals(seenB.length, 1); + assert(seenA[0] instanceof WebSocketRequestError); + assert(seenB[0] instanceof WebSocketRequestError); + assertEquals(manager._subscriptions.size, 0); }); test("onError: an unconfirmed joiner rejects on a synchronous terminate instead of resolving a zombie", async () => { @@ -832,13 +854,13 @@ describe("WebSocketSubscriptionManager", () => { socket.terminate(new Error("gone for good")); // No zombie resolution: B's subscribe() rejects with the failure; its onError never - // fires (it never became a live subscriber). A, the live confirmed owner, fires once. + // fires (it never became a live subscriber). A, the live confirmed lease, fires once. await assertRejects(() => subBPromise, WebSocketRequestError, "permanently terminated"); assertEquals(errorsB.length, 0); assertEquals(errorsA.length, 1); }); - test("onError: when the owner retires the next live lease promotes", async () => { + test("onError: a lease retired before the failure is not notified, the survivor is", async () => { const { socket, manager } = createManager(true); const payload = { type: "l2Book", coin: "BTC" }; const listener = () => {}; @@ -851,15 +873,15 @@ describe("WebSocketSubscriptionManager", () => { const subA = await subAPromise; await subBPromise; - // A retires while B is live: no teardown, and B becomes the owner. + // A retires while B is live: no teardown, and B alone remains a subscriber. await subA.unsubscribe(); assertEquals(manager._subscriptions.size, 1); socket.terminate(new Error("gone for good")); await drain(); - assertEquals(errorsA.length, 0); // retired owner: never notified - assertEquals(errorsB.length, 1); // promoted owner: exactly once + assertEquals(errorsA.length, 0); // retired lease: never notified + assertEquals(errorsB.length, 1); // live lease: exactly once }); test("onError: the same callback ref is safe across leases through the whole lifecycle", async () => { @@ -880,7 +902,7 @@ describe("WebSocketSubscriptionManager", () => { socket.mockMessage(RESPONSES.channelEvent("l2Book", { coin: "BTC", levels: [] })); assertEquals(events, 1); - // A retires; B's lease keeps the delivery and the ownership. + // A retires; B's lease keeps the delivery and the failure notification. await subA.unsubscribe(); socket.mockMessage(RESPONSES.channelEvent("l2Book", { coin: "BTC", levels: [] })); assertEquals(events, 2); @@ -1040,6 +1062,148 @@ describe("WebSocketSubscriptionManager", () => { }); }); + describe("failureSignal", () => { + /** Subscribes to `payload` and confirms it, returning the live handle. */ + async function subscribeConfirmed( + socket: MockWebSocket, + manager: ManagerWithInternals, + payload: unknown, + options?: { onError?: (error: WebSocketRequestError) => void }, + ): Promise { + const promise = manager.subscribe("test", payload, () => {}, options); + socket.mockMessage(RESPONSES.subscriptionResponse("subscribe", payload)); + return await promise; + } + + test("aborts with the WebSocketRequestError when a confirmed subscription fails", async () => { + const { socket, manager } = createManager(true); + const sub = await subscribeConfirmed(socket, manager, { channel: "test", extra: "data" }); + + // Observed before the failure: live and stable across accesses. + const signal = sub.failureSignal; + assert(signal !== undefined); + assert(signal === sub.failureSignal); // repeat access returns the same signal object + assertFalse(signal.aborted); + + let abortEvents = 0; + signal.addEventListener("abort", () => abortEvents++); + + socket.terminate(new Error("gone for good")); + await drain(); + + assert(signal.aborted); + assertEquals(abortEvents, 1); + assert(signal.reason instanceof WebSocketRequestError); + assertEquals((signal.reason as WebSocketRequestError).cause, socket.terminationSignal.reason); + }); + + test("every subscriber's signal aborts on a shared failure", async () => { + const { socket, manager } = createManager(true); + const payload = { channel: "test", extra: "data" }; + const subA = await subscribeConfirmed(socket, manager, payload); + const subB = await subscribeConfirmed(socket, manager, payload); + + const signalA = subA.failureSignal; + const signalB = subB.failureSignal; + assert(signalA !== undefined && signalB !== undefined); + assert(signalA !== signalB); // one signal per subscribe() call, not per subscription + + socket.terminate(new Error("gone for good")); + await drain(); + + assert(signalA.aborted); + assert(signalB.aborted); + assertEquals(signalA.reason, signalB.reason); // both carry the one shared failure + }); + + test("a callback that unsubscribes a sibling mid-failure does not rob it of its notification", async () => { + const { socket, manager } = createManager(true); + const payload = { channel: "test", extra: "data" }; + + // A's onError re-enters unsubscribe() on B while the manager is still notifying. + // The notified set must be the leases live at failure time — B was — so B's + // onError and failureSignal still fire even though A retired it mid-teardown. + const notified: string[] = []; + let subB: ISubscription | undefined; + const subA = await subscribeConfirmed(socket, manager, payload, { + onError: () => { + notified.push("A"); + void subB?.unsubscribe(); + }, + }); + subB = await subscribeConfirmed(socket, manager, payload, { onError: () => notified.push("B") }); + const signalB = subB.failureSignal; + assert(signalB !== undefined); + + socket.terminate(new Error("gone for good")); + await drain(); + + assertEquals(notified, ["A", "B"]); + assert(signalB.aborted); + assertFalse(subA.failureSignal === undefined); + }); + + test("does not abort on a voluntary unsubscribe()", async () => { + const { socket, manager } = createManager(true); + const payload = { channel: "test", extra: "data" }; + const sub = await subscribeConfirmed(socket, manager, payload); + const signal = sub.failureSignal; + assert(signal !== undefined); + + const unsubPromise = sub.unsubscribe(); + socket.mockMessage(RESPONSES.subscriptionResponse("unsubscribe", payload)); + await unsubPromise; + assertFalse(signal.aborted); + + // Even a later termination does not revive the retired lease's signal. + socket.terminate(new Error("gone for good")); + await drain(); + assertFalse(signal.aborted); + }); + + test("a lease retired before the failure keeps an inert signal, the survivor's aborts", async () => { + const { socket, manager } = createManager(true); + const payload = { channel: "test", extra: "data" }; + const subA = await subscribeConfirmed(socket, manager, payload); + const subB = await subscribeConfirmed(socket, manager, payload); + const signalA = subA.failureSignal; + assert(signalA !== undefined); + + // A retires voluntarily; the shared subscription stays live for B. + await subA.unsubscribe(); + assertEquals(manager._subscriptions.size, 1); + + socket.terminate(new Error("gone for good")); + await drain(); + + assertFalse(signalA.aborted); // voluntary unsubscribe: never aborts + assert(subB.failureSignal?.aborted); + }); + + test("accessed only after the failure it is already aborted with the same reason", async () => { + const { socket, manager } = createManager(true); + const payload = { channel: "test", extra: "data" }; + + const seen: unknown[] = []; + const sub = await subscribeConfirmed(socket, manager, payload, { onError: (e) => seen.push(e) }); + + // Reconnect, then the server refuses the re-subscription on the live socket. + socket.disconnect(); + socket.open(); + socket.mockMessage(RESPONSES.errorChannel(JSON.stringify({ method: "subscribe", subscription: payload }))); + await drain(); + assertEquals(seen.length, 1); + + // First access happens after the failure: the signal comes back already + // aborted, with the identical error object onError received as its reason. + const signal = sub.failureSignal; + assert(signal !== undefined); + assert(signal.aborted); + assertEquals(signal.reason, seen[0]); + assert(signal === sub.failureSignal); // and stays the same object afterwards + }); + }); + describe("unique user subscription limit", () => { // 14, not 15: a 2026-08-02 mainnet probe had the server refuse the 15th distinct user on // two independent connections, despite its own error frame saying "more than 15". See