From 0ef6f05d0be2d8b50a1483dbc0ce4c22150a16ee Mon Sep 17 00:00:00 2001 From: Joe Blau Date: Mon, 3 Aug 2026 21:40:30 -0700 Subject: [PATCH] fix(transport): notify every subscriber and expose failureSignal when a confirmed subscription dies _failSubscription used to notify exactly one confirmed lease per listener registration (first-live-owner) and only when that caller had passed options.onError - every other subscriber sharing the payload, and any subscriber without onError, lost its feed silently while unsubscribe() still appeared to work. - _failSubscription now notifies EVERY live confirmed lease's onError, each call wrapped in its own try/catch so one throwing callback cannot silence the rest; unconfirmed leases still observe the failure solely through their subscribe() rejection, and voluntarily retired leases are never notified. - ISubscription gains an optional failureSignal?: AbortSignal that aborts with the failure TransportError as its reason when an already confirmed subscription fails, and never on a voluntary unsubscribe(). It is optional for back-compat with external ISubscriptionTransport implementations, but the WebSocket transport always provides it: the handle returned by WebSocketSubscriptionManager.subscribe() carries a lazily materialized per-call AbortController - accessed before the failure it aborts at failure time, first accessed after the failure it comes back already aborted with the same recorded reason, and the common path that never reads the signal allocates nothing. Subscription and explorer API methods return the transport handle unchanged, so the signal reaches the public API without threading. - Updated the invalidated comments (first-live-owner ownership text, RegistrationHandle.confirmed, _failSubscription, subscribe() option docs in the manager, transport, and ISubscriptionTransport) plus docs/clients.md and docs/transports.md failure semantics. - Tests: shared-payload failures fan out to every onError (same and different listeners), failureSignal aborts with the WebSocketRequestError, stays inert on voluntary unsubscribe() and for leases retired before the failure, and post-failure first access returns an already-aborted signal with the identical reason. Fixes #89 --- docs/clients.md | 19 +- docs/transports.md | 4 +- .../subscription/_methods/_base/_config.ts | 4 +- src/transport/_base.ts | 13 +- .../websocket/_subscriptionManager.ts | 95 ++++++--- src/transport/websocket/mod.ts | 9 +- .../websocket/_subscriptionManager.test.ts | 182 +++++++++++++++++- 7 files changed, 288 insertions(+), 38 deletions(-) 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