diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 916c259..d97571f 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -17,3 +17,4 @@ jobs: node-version-file: '.nvmrc' - run: npm install - run: npm run build + - run: npm test diff --git a/README.md b/README.md index 5b06129..65a5a53 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,44 @@ async function someBrokerizeActions() { someBrokerizeActions().then(console.log, console.error) ``` +# websocket error handling + +Every error reported by the websocket client is a `BrokerizeWebSocketError`, a subclass of +`BrokerizeError`, so it can be handled like any other error of this library. Errors reach an +application through two separate channels: + +- **Subscription callbacks** only receive _terminal_ errors: the subscription is over and will not + receive any further data. This maps directly onto e.g. an RxJS `subscriber.error()`. +- **`subscribeConnectionState`** reports the state of the underlying connection, including the + errors that the client recovers from on its own. Use it to tell the user that live updates are + currently unavailable, instead of tearing down subscriptions on every connection hiccup. + +```js +const client = ctx.createWebSocketClient(); + +client.subscribeConnectionState(({ status, error }) => { + /* 'closed' | 'connecting' | 'open' | 'reconnecting' | 'fatal' */ + if (status == "fatal") { + /* the client is unusable, all subscriptions were terminated with `error`. */ + console.error("live updates stopped", error); + client.reconnect(); /* ... and subscribe again */ + } +}); + +client.subscribeInvalidate({ entity: "sessions" }, (err) => { + if (err) { + /* err is a BrokerizeWebSocketError with err.terminal === true. */ + console.error("subscription ended", err.code, err.message); + return; + } + reloadSessions(); +}); +``` + +A connection that cannot be authenticated is retried once with a freshly refreshed token, because +the usual cause is an id token that expired in flight. Only if that retry is rejected as well does +the client become `fatal`. + # development & maintenance The code in `src/swagger` is generated by the npm script `npm run generate` from the OpenAPI spec and will be regenerated whenever API changes are made. No manual changes should be made in the `src/swagger` directory. `npm run download-spec` will download the current spec from `api-preview.brokerize.com`. diff --git a/etc/client.api.md b/etc/client.api.md index 61d604a..c60d842 100644 --- a/etc/client.api.md +++ b/etc/client.api.md @@ -42,7 +42,7 @@ interface Amount { // @public (undocumented) export interface Auth { // (undocumented) - getToken: () => Promise<{ + getToken: (opts?: GetTokenOptions) => Promise<{ idToken: string; }>; } @@ -346,6 +346,11 @@ export class Brokerize { refreshGuestUser(refreshToken: string): Promise; } +// @public +export class BrokerizeAbortError extends Error { + constructor(msg: string); +} + // @public (undocumented) export interface BrokerizeConfig { acceptLanguage?: string | (() => string | null | undefined | Promise); @@ -388,15 +393,41 @@ export class BrokerizeTimeoutError extends BrokerizeError { // @public (undocumented) export interface BrokerizeWebSocketClient { + // (undocumented) + getConnectionState: () => WebSocketConnectionState; + // (undocumented) + reconnect: () => void; + // (undocumented) + subscribeConnectionState: (callback: ConnectionStateCallback) => Subscription; // (undocumented) subscribeDecoupledOperation: (subscribe: Pick, callback: Callback) => Subscription; // (undocumented) subscribeInvalidate: (subscribe: SubscribeInvalidateDetails, callback: Callback) => Subscription; } -// @public (undocumented) -type BrokerizeWebSocketError = { +// @public +export class BrokerizeWebSocketError extends BrokerizeError { + constructor(input: BrokerizeWebSocketErrorParams); + readonly cause?: unknown; + readonly terminal: boolean; +} + +// @public +type BrokerizeWebSocketError_2 = { message: string; + code?: string; +}; + +// @public +export type BrokerizeWebSocketErrorCode = "AUTH" | "WEBSOCKET_CONNECTION_FAILED" | "WEBSOCKET_PROTOCOL_ERROR" | "WEBSOCKET_ERROR"; + +// @public (undocumented) +export type BrokerizeWebSocketErrorParams = { + msg: string; + code: BrokerizeWebSocketErrorCode; + httpStatusCode?: number; + terminal?: boolean; + cause?: unknown; }; // @public @@ -442,7 +473,7 @@ interface BrokerMetaImages { lightSquare: string; } -// @public (undocumented) +// @public export type Callback = (err: any, data: T) => void; // @public @@ -590,7 +621,7 @@ export type CognitoConfig = { // @public (undocumented) export type CognitoFacade = { createSession: (cognitoPoolConfig: CognitoPoolConfig, authCfg: RegisteredUserAuthContextConfiguration) => { - getToken: () => Promise<{ + getToken: (opts?: GetTokenOptions) => Promise<{ idToken: string; }>; }; @@ -616,6 +647,9 @@ interface ConfirmOAuthResponse { sessionId: string; } +// @public (undocumented) +export type ConnectionStateCallback = (state: WebSocketConnectionState) => void; + // @public interface CostDetailsLink { footerText?: string; @@ -1109,6 +1143,11 @@ interface GetRecoveryPhrasesResponse { // @public function getSizeUnitsFromConstraints(preparedTrade: PreparedTrade, orderModel: OrderModel, direction: Direction, cashAccountId: string): string[]; +// @public (undocumented) +export type GetTokenOptions = { + forceRefresh?: boolean; +}; + // @public interface GetUserResponse { createdAt: Date; @@ -1128,6 +1167,9 @@ type InvalidateMessage = { subscriptionId: number; }; +// @public +export function isBrokerizeWebSocketError(x: unknown): x is BrokerizeWebSocketError; + // @public interface LegalTermConfirmItem { html: string; @@ -2202,9 +2244,18 @@ type WebSocketCommandUnsubscribe = { subscriptionId: number; }; +// @public (undocumented) +export type WebSocketConnectionState = { + status: WebSocketConnectionStatus; + error?: BrokerizeWebSocketError; +}; + +// @public +export type WebSocketConnectionStatus = "closed" | "connecting" | "open" | "reconnecting" | "fatal"; + // @public (undocumented) type WebSocketError = { - error: BrokerizeWebSocketError; + error: BrokerizeWebSocketError_2; }; // @public (undocumented) @@ -2213,7 +2264,7 @@ type WebSocketMessage = WebSocketSubscriptionMessage | WebSocketError | WebSocke // @public (undocumented) type WebSocketMessageErrorOnSubscription = { subscriptionId: number; - error: BrokerizeWebSocketError; + error: BrokerizeWebSocketError_2; }; // @public (undocumented) @@ -2243,7 +2294,7 @@ declare namespace WebSocketTypes { WebSocketMessageErrorOnSubscription, InvalidateMessage, UpdateDecoupledOperationMessage, - BrokerizeWebSocketError, + BrokerizeWebSocketError_2 as BrokerizeWebSocketError, WebSocketError, DecoupledOperationState_2 as DecoupledOperationState, DecoupledOperationStatus_2 as DecoupledOperationStatus diff --git a/package.json b/package.json index 768f5fd..3813590 100644 --- a/package.json +++ b/package.json @@ -52,7 +52,8 @@ "openapi-preprocess": "node openapi-preprocess.js", "prepack": "node scripts/prepack-package-json.cjs strip", "postpack": "node scripts/prepack-package-json.cjs restore", - "prepare": "husky" + "prepare": "husky", + "test": "node --test \"test/**/*.test.js\"" }, "lint-staged": { "*.{js,ts}": [ diff --git a/src/apiCtx.ts b/src/apiCtx.ts index cf253fb..3a8621b 100644 --- a/src/apiCtx.ts +++ b/src/apiCtx.ts @@ -92,8 +92,17 @@ export interface GuestAuthContextConfiguration { tokens: { updatedAt: number; response: Models.CreateGuestUserResponse }; } +export type GetTokenOptions = { + /** + * Request a fresh token even if the currently known one is not expired yet. Used by the + * websocket client to retry once after the server rejected a token, before giving up. + * Implementations that cannot refresh on demand may ignore this. + */ + forceRefresh?: boolean; +}; + export interface Auth { - getToken: () => Promise<{ idToken: string }>; + getToken: (opts?: GetTokenOptions) => Promise<{ idToken: string }>; } export function createConfiguration(cfg: BrokerizeConfig) { @@ -123,7 +132,7 @@ export function createAuth({ JSON.stringify(authCfg), ) as GuestAuthContextConfiguration; // clone it return { - async getToken() { + async getToken(opts?: GetTokenOptions) { if (guestAuthCfg.tokens) { let expiresIn = guestAuthCfg.tokens.response.expiresIn; if (expiresIn == null) { @@ -136,7 +145,8 @@ export function createAuth({ /* modern tokens */ const tokenExpiresAt = guestAuthCfg.tokens.updatedAt + expiresIn * 1000 - 10000; - const needsRefresh = Date.now() > tokenExpiresAt; + const needsRefresh = + opts?.forceRefresh || Date.now() > tokenExpiresAt; if (needsRefresh && guestAuthCfg.tokens.response.refreshToken) { if (!cfg.fetch) { throw new Error( @@ -219,8 +229,8 @@ export function createAuth({ authCfg, ); return { - async getToken() { - return session.getToken(); + async getToken(opts?: GetTokenOptions) { + return session.getToken(opts); }, }; } else { @@ -244,7 +254,7 @@ export type CognitoFacade = { cognitoPoolConfig: CognitoPoolConfig, authCfg: RegisteredUserAuthContextConfiguration, ) => { - getToken: () => Promise<{ idToken: string }>; + getToken: (opts?: GetTokenOptions) => Promise<{ idToken: string }>; }; }; diff --git a/src/authorizedApiContext.ts b/src/authorizedApiContext.ts index 1876793..19307ab 100644 --- a/src/authorizedApiContext.ts +++ b/src/authorizedApiContext.ts @@ -5,7 +5,7 @@ import { createConfiguration, withAcceptLanguage, } from "./apiCtx"; -import { createErrorFromResponse } from "./errors"; +import { BrokerizeAbortError, createErrorFromResponse } from "./errors"; import { createPollingSubscription } from "./pollingSubscription"; import * as openApiClient from "./swagger"; import { @@ -45,6 +45,11 @@ export class AuthorizedApiContext { private readonly _orderApiCryptoService: openApiClient.OrderApi; private readonly _logoutSubject: Subject; private readonly _wsClient: BrokerizeWebSocketClient; + /** + * `true` if this context created the websocket client itself and therefore has to destroy it. + * A client passed in by the application is owned by the application. + */ + private readonly _ownsWsClient: boolean; private readonly _cache: { getBrokers?: Promise; }; @@ -144,6 +149,7 @@ export class AuthorizedApiContext { ); } this._abortController = cfg.createAbortController(); + this._ownsWsClient = !wsClient; this._wsClient = wsClient || this._initInternalWebSocketClient(); if ( (this._wsClient as BrokerizeWebSocketClientImpl)._setAuthorizedApiContext @@ -171,7 +177,7 @@ export class AuthorizedApiContext { } private async _initRequestInit() { if (this._isDestroyed) { - throw new Error("AuthorizedApiContext is destroyed"); + throw new BrokerizeAbortError("AuthorizedApiContext is destroyed"); } const tok = await this._auth.getToken(); const headers = await withAcceptLanguage(this._cfg, { @@ -765,12 +771,24 @@ export class AuthorizedApiContext { subscribeInvalidate: this._wsClient.subscribeInvalidate.bind( this._wsClient, ), + subscribeConnectionState: this._wsClient.subscribeConnectionState.bind( + this._wsClient, + ), + getConnectionState: this._wsClient.getConnectionState.bind( + this._wsClient, + ), + reconnect: this._wsClient.reconnect.bind(this._wsClient), }; return wrappedClient; } destroy() { this._isDestroyed = true; this._abortController.abort(); + if (this._ownsWsClient) { + /* otherwise its reconnect timer would keep running and keep reconnecting with the + tokens of a context that is gone. */ + (this._wsClient as BrokerizeWebSocketClientImpl).destroy(); + } } subscribeLogout(callback: Callback): Subscription { const s = this._logoutSubject.subscribe({ diff --git a/src/errors.ts b/src/errors.ts index d3df664..56d48bc 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -155,3 +155,104 @@ export async function createErrorFromResponse( }, ); } + +/** + * Error codes used by {@link BrokerizeWebSocketError}. + * + * - `AUTH`: the websocket connection could not be authenticated. Same code as a failed token + * refresh on the REST API, so applications can treat both identically. + * - `WEBSOCKET_CONNECTION_FAILED`: the connection could not be established or was lost + * (network error, failed handshake, closed socket). The client retries these automatically. + * - `WEBSOCKET_PROTOCOL_ERROR`: a message from the server could not be parsed. + * - `WEBSOCKET_ERROR`: any other error reported by the server. + */ +export type BrokerizeWebSocketErrorCode = + | "AUTH" + | "WEBSOCKET_CONNECTION_FAILED" + | "WEBSOCKET_PROTOCOL_ERROR" + | "WEBSOCKET_ERROR"; + +export type BrokerizeWebSocketErrorParams = { + msg: string; + code: BrokerizeWebSocketErrorCode; + httpStatusCode?: number; + /** + * See {@link BrokerizeWebSocketError.terminal}. + */ + terminal?: boolean; + /** + * The underlying error or event, if any (e.g. the rejection of `Auth.getToken` or the + * `error` event of the websocket). + */ + cause?: unknown; +}; + +const DEFAULT_STATUS_CODES: Record = { + AUTH: 401, + WEBSOCKET_CONNECTION_FAILED: 503, + WEBSOCKET_PROTOCOL_ERROR: 500, + WEBSOCKET_ERROR: 500, +}; + +/** + * Every error reported by the websocket client is an instance of this class, so applications can + * handle it like any other error (`instanceof Error`, `err.message`, stack trace) and can use the + * same error handling as for the REST API, which reports `BrokerizeError`. + */ +export class BrokerizeWebSocketError extends BrokerizeError { + /** + * If `true`, the affected subscription is over and will not receive any further data. Errors + * that are passed to a subscription callback are always terminal; non-terminal errors are only + * reported via `BrokerizeWebSocketClient.subscribeConnectionState`, because the client + * recovers from them on its own. + */ + readonly terminal: boolean; + /** + * The underlying error or event, if any. + */ + readonly cause?: unknown; + + constructor({ + msg, + code, + httpStatusCode, + terminal, + cause, + }: BrokerizeWebSocketErrorParams) { + super(httpStatusCode ?? DEFAULT_STATUS_CODES[code], { msg, code }); + Object.setPrototypeOf(this, new.target.prototype); // restore prototype chain (https://stackoverflow.com/questions/41102060/typescript-extending-error-class) + this.name = "BrokerizeWebSocketError"; + this.terminal = terminal ?? false; + this.cause = cause; + } +} + +/** + * Check whether the given value is a `BrokerizeWebSocketError`. Prefer this over `instanceof`, + * which fails if more than one copy of this package ends up in a dependency tree. + */ +export function isBrokerizeWebSocketError( + x: unknown, +): x is BrokerizeWebSocketError { + return ( + !!x && + typeof x === "object" && + (x as Error).name === "BrokerizeWebSocketError" + ); +} + +/** + * Thrown when a request is made on an `AuthorizedApiContext` that was already destroyed. + * + * `destroy()` aborts the context's `AbortController`, so a request that is already in flight + * rejects with a `DOMException` named `AbortError`. This error carries the same `name`, so that + * both cases can be handled with a single check (`err.name === "AbortError"`), the way `fetch` + * itself rejects when handed an already-aborted signal. + */ +export class BrokerizeAbortError extends Error { + constructor(msg: string) { + super(msg); + Object.setPrototypeOf(this, new.target.prototype); // restore prototype chain (https://stackoverflow.com/questions/41102060/typescript-extending-error-class) + this.name = "AbortError"; + } +} diff --git a/src/index.ts b/src/index.ts index f62be3a..cd060ed 100644 --- a/src/index.ts +++ b/src/index.ts @@ -9,6 +9,7 @@ import { CognitoPoolConfig, createAuth, createConfiguration, + GetTokenOptions, GuestAuthContextConfiguration, RegisteredUserAuthContextConfiguration, TokenSet, @@ -19,9 +20,14 @@ import { getWebSocketURLByBasePath, } from "./authorizedApiContext"; import { + BrokerizeAbortError, BrokerizeError, BrokerizeTimeoutError, + BrokerizeWebSocketError, + BrokerizeWebSocketErrorCode, + BrokerizeWebSocketErrorParams, createErrorFromResponse, + isBrokerizeWebSocketError, } from "./errors"; import * as openApiClient from "./swagger"; import * as Models from "./modelExports"; @@ -29,7 +35,10 @@ import { BrokerizeWebSocketClient, BrokerizeWebSocketClientImpl, Callback, + ConnectionStateCallback, Subscription, + WebSocketConnectionState, + WebSocketConnectionStatus, } from "./websocketClient"; import * as WebSocketTypes from "./websocketTypes"; import * as Utils from "./utils"; @@ -46,6 +55,15 @@ export { Utils, BrokerizeError, BrokerizeTimeoutError, + BrokerizeWebSocketError, + BrokerizeWebSocketErrorCode, + BrokerizeWebSocketErrorParams, + BrokerizeAbortError, + GetTokenOptions, + isBrokerizeWebSocketError, + ConnectionStateCallback, + WebSocketConnectionState, + WebSocketConnectionStatus, Auth, TokenSet, }; diff --git a/src/websocketClient.ts b/src/websocketClient.ts index 4899467..c7f5a2c 100644 --- a/src/websocketClient.ts +++ b/src/websocketClient.ts @@ -2,7 +2,9 @@ import { Auth } from "./apiCtx"; import { AuthorizedApiContext } from "./authorizedApiContext"; import { WebSocket } from "./dependencyDefinitions/webSocket"; +import { BrokerizeWebSocketError } from "./errors"; import type { + BrokerizeWebSocketError as WebSocketErrorPayload, InvalidateMessage, SubscribeDecoupledOperation, SubscribeInvalidateDetails, @@ -32,11 +34,22 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { private _auth: Auth; private _createWebsocket: (url: string) => WebSocket; - private _fatalError: WebSocketError | null = null; - private _lastNonFatalError: WebSocketError | null = null; + private _fatalError: BrokerizeWebSocketError | null = null; + private _lastNonFatalError: BrokerizeWebSocketError | null = null; private _errorCount = 0; private _apiCtx: AuthorizedApiContext | null = null; + /** + * Set after the server rejected our token once. We then retry with a forced token refresh and + * only treat a second rejection as fatal. Reset on every successful authentication. + */ + private _authRetried = false; + private _forceTokenRefresh = false; + + private _destroyed = false; + private _connectionState: WebSocketConnectionState = { status: "closed" }; + private _connectionStateCallbacks: ConnectionStateCallback[] = []; + constructor( websocketUrl: string, auth: Auth, @@ -54,8 +67,9 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { private _updateReconnectInterval() { if (this._reconnectIntvl) { clearInterval(this._reconnectIntvl); + this._reconnectIntvl = null; } - if (this._fatalError) { + if (this._fatalError || this._destroyed) { return; } this._reconnectIntvl = setInterval(() => { @@ -88,7 +102,7 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { } private _shouldConnect() { - if (this._fatalError) { + if (this._fatalError || this._destroyed) { return false; } @@ -100,10 +114,41 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { return false; } + /** + * Invoke a subscriber callback. A callback that throws must not prevent the remaining + * subscribers from being notified, so everything is contained here. + */ + private _invokeCallback(cb: Callback, err: any, data: any) { + try { + cb(err, data); + } catch (e) { + console.warn( + LOG_PREFIX + "a subscription callback threw an error unexpectedly.", + e, + ); + } + } + private subscribe(cmd: WebSocketCommandSubscribe, callback: Callback) { + if (this._destroyed) { + const err = new BrokerizeWebSocketError({ + msg: "the websocket client was destroyed.", + code: "WEBSOCKET_CONNECTION_FAILED", + terminal: true, + }); + Promise.resolve().then(() => this._invokeCallback(callback, err, null)); + return { + unsubscribe: () => {}, + } as Subscription; + } + if (this._fatalError) { - // no more subscriptions on a websocket client that has a fatal error - callback(this._fatalError, null); + // no more subscriptions on a websocket client that has a fatal error. Report it + // asynchronously so the caller holds the Subscription before the callback runs. + const fatalError = this._fatalError; + Promise.resolve().then(() => + this._invokeCallback(callback, fatalError, null), + ); return { unsubscribe: () => {}, } as Subscription; @@ -138,10 +183,13 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { return; } unsub = true; - this._map[key].callbacks = this._map[key].callbacks.filter( - (cb) => cb != wrappedCb, - ); - if (this._map[key].callbacks.length == 0) { + const entry = this._map[key]; + if (!entry) { + /* the subscription was already dropped, e.g. because of a fatal error. */ + return; + } + entry.callbacks = entry.callbacks.filter((cb) => cb != wrappedCb); + if (entry.callbacks.length == 0) { this._endSubscription(key); } this._startOrStopDisconnectTimeout(); @@ -178,6 +226,123 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { return this.subscribe(cmd, callback); } + /** + * The current connection state. See `subscribeConnectionState`. + */ + getConnectionState(): WebSocketConnectionState { + return this._connectionState; + } + + /** + * Observe the state of the underlying websocket connection. The callback is invoked immediately + * with the current state and afterwards on every change. + * + * This is the only place where non-terminal errors are reported: the client recovers from them + * on its own, so they are not passed to the subscription callbacks, which would force + * applications to tear down subscriptions on every connection hiccup. Use this to show that + * live updates are currently unavailable. + */ + subscribeConnectionState(callback: ConnectionStateCallback): Subscription { + this._connectionStateCallbacks.push(callback); + try { + callback(this._connectionState); + } catch (e) { + console.warn( + LOG_PREFIX + "a connection state callback threw an error unexpectedly.", + e, + ); + } + let unsub = false; + return { + unsubscribe: () => { + if (unsub) { + return; + } + unsub = true; + this._connectionStateCallbacks = this._connectionStateCallbacks.filter( + (cb) => cb != callback, + ); + }, + }; + } + + /** + * Recover from a fatal error (see `WebSocketConnectionState`). The client drops the fatal error + * state and connects again, retrieving a fresh token. Existing subscriptions were already + * terminated when the fatal error occurred, so applications have to subscribe again. + * + * Calling this without a fatal error forces a reconnect of the current connection. + */ + reconnect() { + if (this._destroyed) { + return; + } + this._fatalError = null; + this._errorCount = 0; + this._lastNonFatalError = null; + this._authRetried = false; + this._forceTokenRefresh = true; + this._closeSocket(); + this._updateReconnectInterval(); + if (this._shouldConnect()) { + this._connect(); + } else { + this._setConnectionState("closed"); + } + } + + /** + * Release all resources: close the connection, drop all subscriptions and stop the reconnect and + * ping timers. A destroyed client cannot be revived with `reconnect()`. + * + * This is called by `AuthorizedApiContext.destroy()` for the client it created itself. A client + * that was passed in from outside is owned by the application and is not destroyed with the + * context. + */ + destroy() { + if (this._destroyed) { + return; + } + this._destroyed = true; + if (this._reconnectIntvl) { + clearInterval(this._reconnectIntvl); + this._reconnectIntvl = null; + } + if (this._disconnectTimeout) { + clearTimeout(this._disconnectTimeout); + this._disconnectTimeout = null; + } + this._closeSocket(); + this._map = {}; + this._setConnectionState("closed"); + this._connectionStateCallbacks = []; + } + + private _setConnectionState( + status: WebSocketConnectionStatus, + error?: BrokerizeWebSocketError, + ) { + if ( + this._connectionState.status == status && + this._connectionState.error == error + ) { + return; + } + this._connectionState = error ? { status, error } : { status }; + const state = this._connectionState; + this._connectionStateCallbacks.slice().forEach((cb) => { + try { + cb(state); + } catch (e) { + console.warn( + LOG_PREFIX + + "a connection state callback threw an error unexpectedly.", + e, + ); + } + }); + } + private _startSubscription(key: string) { if (this._fatalError) { this._notifySubscribersAboutFatalError(key); @@ -216,7 +381,9 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { LOG_PREFIX + "connection was interrupted. filling in assumed invalidates that may have happened while socket was not available.", ); - callbacks.forEach((cb) => cb(null, assumedInvalidate)); + callbacks.forEach((cb) => + this._invokeCallback(cb, null, assumedInvalidate), + ); } else if (cmd.type == "decoupledOperationStatus") { if (this._apiCtx) { this._apiCtx @@ -234,7 +401,7 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { LOG_PREFIX + "sending virtual decoupled operation status updates to subscribers due to connection interruption.", ); - callbacks.forEach((cb) => cb(null, invMsg)); + callbacks.forEach((cb) => this._invokeCallback(cb, null, invMsg)); }, (err) => { console.log( @@ -249,15 +416,13 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { } private _endSubscription(key: string) { - if (this._map[key].idOnSocket != null) { - this._sendWs( - { - cmd: "unsubscribe", - subscriptionId: this._map[key].idOnSocket as number, - }, - true, - ); - this._map[key].idOnSocket = null; + const entry = this._map[key]; + if (entry?.idOnSocket != null) { + this._sendWs({ + cmd: "unsubscribe", + subscriptionId: entry.idOnSocket as number, + }); + entry.idOnSocket = null; } } @@ -278,10 +443,12 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { } else if (!this._disconnectTimeout) { // if there is no subscription open for 3s, disconnect from WebSocket this._disconnectTimeout = setTimeout(() => { + this._disconnectTimeout = null; if (!this._shouldConnect()) { - this._socket?.close(); - this._socket = null; - this._disconnectTimeout = null; + this._closeSocket(); + if (!this._fatalError) { + this._setConnectionState("closed"); + } } }, 3000); } @@ -303,103 +470,191 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { } } + /** + * Close the current socket, if any, and detach its handlers so that a late `close` event of an + * outdated socket cannot interfere with a connection established in the meantime. + */ + private _closeSocket() { + const socket = this._socket; + this._socket = null; + this._isOpen = false; + this._authenticatedCallback = null; + if (this._pingIntvl) { + clearInterval(this._pingIntvl); + this._pingIntvl = null; + } + if (!socket) { + return; + } + socket.onmessage = null; + socket.onopen = null; + /* keep no-op handlers: some implementations (e.g. the `ws` package) throw on unhandled error events. */ + socket.onerror = () => {}; + socket.onclose = () => {}; + try { + socket.close(); + } catch (err) { + console.warn(LOG_PREFIX + "closing the socket failed.", err); + } + } + private _connect() { - if (this._socket) { - this._isOpen = false; - this._socket.close(); + if (this._fatalError || this._destroyed) { + // no more connections on a websocket client that has a fatal error or was destroyed + return; } - if (this._fatalError) { - // no more connections on a websocket client that has a fatal error + this._closeSocket(); + this._setConnectionState( + this._errorCount > 0 ? "reconnecting" : "connecting", + this._lastNonFatalError || undefined, + ); + + let socket: WebSocket; + try { + socket = this._createWebsocket(this._url); + } catch (err) { + this._handleNonFatalError( + new BrokerizeWebSocketError({ + msg: "the websocket could not be created.", + code: "WEBSOCKET_CONNECTION_FAILED", + cause: err, + }), + ); return; } + this._socket = socket; + + /* Per-socket bookkeeping: a failed connection attempt must be counted exactly once, no matter + whether we learn about it from an error event, from a close event or from both. */ + let failureReported = false; + let authenticated = false; + const reportConnectionFailure = (msg: string, cause?: unknown) => { + if (failureReported) { + return; + } + failureReported = true; + this._handleNonFatalError( + new BrokerizeWebSocketError({ + msg, + code: "WEBSOCKET_CONNECTION_FAILED", + cause, + }), + ); + }; - this._authenticatedCallback = null; + socket.onmessage = (msg) => { + if (this._socket !== socket) { + return; + } + let message: WebSocketMessage; + try { + message = JSON.parse(msg.data) as WebSocketMessage; + } catch (err) { + this._handleNonFatalError( + new BrokerizeWebSocketError({ + msg: "a message from the server could not be parsed.", + code: "WEBSOCKET_PROTOCOL_ERROR", + cause: err, + }), + ); + return; + } + this._handleMessage(message); + }; - this._socket = this._createWebsocket(this._url); - this._socket.onmessage = (msg) => { - const message = JSON.parse(msg.data) as WebSocketMessage; - if ((message as WebSocketSubscriptionMessage).subscriptionId) { - /* message concerns a subscription */ - if ((message as InvalidateMessage).cmd == "invalidate") { - const entry = this._findSubscriptionEntry( - (message as WebSocketSubscriptionMessage).subscriptionId, - ); - entry?.entry.callbacks.forEach((cb) => cb(null, message)); - } else if ( - (message as UpdateDecoupledOperationMessage).cmd == - "updateDecoupledOperationStatus" - ) { - const entry = this._findSubscriptionEntry( - (message as WebSocketSubscriptionMessage).subscriptionId, - ); - entry?.entry.callbacks.forEach((cb) => cb(null, message)); - } else if ((message as WebSocketMessageErrorOnSubscription).error) { - /* error on subscription */ - const entry = this._findSubscriptionEntry( - (message as WebSocketSubscriptionMessage).subscriptionId, - ); - entry?.entry.callbacks.forEach((cb) => - cb(message as WebSocketMessageErrorOnSubscription, null), - ); - } - } else if ( - (message as WebSocketAuthenticatedMessage).cmd == "authenticated" - ) { - if (this._authenticatedCallback) this._authenticatedCallback(); - this._authenticatedCallback = null; - } else if ((message as WebSocketPingMessage).cmd == "ping") { - // NOP - } else if (message as WebSocketError) { - const e = message as WebSocketError; - if (e.error?.message == "authentication failed") { - // we treet authentication failed as a fatal error, this should just not happen if the client creates - // the websocket client correctly. If it happens, applications will have to create a new instance. - this._handleFatalError(e); - } else { - this._handleNonFatalError(e); - } + socket.onerror = (ev) => { + if (this._socket !== socket) { + return; } + /* The error event carries no usable detail; the following close event triggers the retry. */ + reportConnectionFailure("the websocket connection failed.", ev); }; - this._socket.onopen = () => { - Promise.resolve(this._auth.getToken()).then( + socket.onopen = () => { + if (this._socket !== socket) { + return; + } + const forceRefresh = this._forceTokenRefresh; + this._forceTokenRefresh = false; + Promise.resolve(this._auth.getToken({ forceRefresh })).then( (token) => { + if (this._socket !== socket) { + return; + } const _authCb = () => { + authenticated = true; this._isOpen = true; this._resetErrorState(); + this._setConnectionState("open"); for (const k in this._map) { this._startSubscription(k); } }; if (token) { - this._sendWs( - { - cmd: "authorize", - idToken: token.idToken, - }, - true, - ); + this._sendWs({ + cmd: "authorize", + idToken: token.idToken, + }); this._authenticatedCallback = _authCb; } else { _authCb(); } }, (err) => { - console.error(LOG_PREFIX + " connection failed", err); + if (this._socket !== socket) { + return; + } + /* Without a token we cannot use this connection. Report it and close, so that the + reconnect interval retries instead of leaving an unauthenticated socket open. */ + failureReported = true; + this._handleNonFatalError( + new BrokerizeWebSocketError({ + msg: + err?.message || + "a token for the websocket connection could not be retrieved.", + code: "AUTH", + cause: err, + }), + ); + this._closeSocket(); }, ); }; - this._socket.onclose = () => { + socket.onclose = () => { + if (this._socket !== socket) { + return; + } // mark all subscriptions interrupted so they can issue synthetic messages on reconnect for (const k in this._map) { this._map[k].interrupted = true; - this._endSubscription(k); + this._map[k].idOnSocket = null; } this._socket = null; + this._isOpen = false; + this._authenticatedCallback = null; + if (this._pingIntvl) { + clearInterval(this._pingIntvl); + this._pingIntvl = null; + } + if (!authenticated) { + /* the connection never became usable, so this attempt counts towards the backoff. */ + reportConnectionFailure( + "the websocket connection was closed before it could be used.", + ); + } + if (this._shouldConnect()) { + this._setConnectionState( + "reconnecting", + this._lastNonFatalError || undefined, + ); + } else if (!this._fatalError) { + this._setConnectionState("closed"); + } }; + if (this._pingIntvl) clearInterval(this._pingIntvl); this._pingIntvl = setInterval(() => { // 2: CLOSING, 3: CLOSED @@ -420,25 +675,111 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { }, 30000); } + private _handleMessage(message: WebSocketMessage) { + const subscriptionId = (message as WebSocketSubscriptionMessage) + .subscriptionId; + if (subscriptionId != null) { + /* message concerns a subscription */ + const entry = this._findSubscriptionEntry(subscriptionId); + if ( + (message as InvalidateMessage).cmd == "invalidate" || + (message as UpdateDecoupledOperationMessage).cmd == + "updateDecoupledOperationStatus" + ) { + entry?.entry.callbacks.forEach((cb) => + this._invokeCallback(cb, null, message), + ); + } else if ((message as WebSocketMessageErrorOnSubscription).error) { + /* error on a single subscription. Nothing retries it, so it ends this subscription. */ + const err = toWebSocketError( + (message as WebSocketMessageErrorOnSubscription).error, + true, + ); + entry?.entry.callbacks.forEach((cb) => + this._invokeCallback(cb, err, null), + ); + } else { + console.warn( + LOG_PREFIX + "received an unknown subscription message.", + message, + ); + } + return; + } + + if ((message as WebSocketAuthenticatedMessage).cmd == "authenticated") { + if (this._authenticatedCallback) this._authenticatedCallback(); + this._authenticatedCallback = null; + } else if ((message as WebSocketPingMessage).cmd == "ping") { + // NOP + } else if ((message as WebSocketError).error) { + const e = message as WebSocketError; + if (isAuthenticationError(e.error)) { + this._handleAuthenticationFailure(e.error); + } else { + this._handleNonFatalError(toWebSocketError(e.error, false)); + } + } else { + console.warn(LOG_PREFIX + "received an unknown message.", message); + } + } + + /** + * The server rejected our token. The usual cause is an id token that expired between retrieving + * it and the server validating it, which a fresh token fixes, so retry once before giving up on + * this client. + */ + private _handleAuthenticationFailure(payload: WebSocketErrorPayload) { + if (!this._authRetried) { + this._authRetried = true; + this._forceTokenRefresh = true; + console.warn( + LOG_PREFIX + + "authentication failed. retrying once with a refreshed token.", + ); + this._handleNonFatalError(toWebSocketError(payload, false)); + this._closeSocket(); + this._setConnectionState( + "reconnecting", + this._lastNonFatalError || undefined, + ); + return; + } + + // authentication failed again with a fresh token: this client is unusable. + this._handleFatalError(toWebSocketError(payload, true)); + } + /** * Error happend when trying to connect, but it may be solved by trying a reconnect later. */ - private _handleNonFatalError(e: WebSocketError) { + private _handleNonFatalError(e: BrokerizeWebSocketError) { this._errorCount++; this._lastNonFatalError = e; this._updateReconnectInterval(); + if (this._shouldConnect()) { + this._setConnectionState("reconnecting", e); + } } /** - * Fatal error: this means the client is unusable from now on and must be recreated. + * Fatal error: this means the client is unusable from now on and must be recreated or revived + * with `reconnect()`. */ - private _handleFatalError(e: WebSocketError) { + private _handleFatalError(e: BrokerizeWebSocketError) { this._fatalError = e; this._updateReconnectInterval(); - if (this._pingIntvl) clearInterval(this._pingIntvl); + if (this._pingIntvl) { + clearInterval(this._pingIntvl); + this._pingIntvl = null; + } + this._closeSocket(); for (const key in this._map) { this._notifySubscribersAboutFatalError(key); } + /* the subscriptions are terminated, their callbacks must not be called again. */ + this._map = {}; + this._setConnectionState("fatal", e); } /** @@ -446,11 +787,7 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { */ private _notifySubscribersAboutFatalError(key: string) { this._map[key].callbacks.forEach((cb) => { - try { - cb(this._fatalError, null); - } catch (err) { - console.warn(LOG_PREFIX + "error in callback", err); - } + this._invokeCallback(cb, this._fatalError, null); }); } @@ -460,6 +797,7 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { private _resetErrorState() { this._lastNonFatalError = null; this._errorCount = 0; + this._authRetried = false; this._updateReconnectInterval(); } @@ -474,6 +812,24 @@ export class BrokerizeWebSocketClientImpl implements BrokerizeWebSocketClient { } } +function isAuthenticationError(payload: WebSocketErrorPayload | undefined) { + return payload?.code == "AUTH" || payload?.message == "authentication failed"; +} + +/** + * Turn an error message received over the websocket into a proper `Error`. + */ +function toWebSocketError( + payload: WebSocketErrorPayload | undefined, + terminal: boolean, +) { + return new BrokerizeWebSocketError({ + msg: payload?.message || "an unknown websocket error occurred.", + code: isAuthenticationError(payload) ? "AUTH" : "WEBSOCKET_ERROR", + terminal, + }); +} + export interface BrokerizeWebSocketClient { subscribeInvalidate: ( subscribe: SubscribeInvalidateDetails, @@ -487,8 +843,39 @@ export interface BrokerizeWebSocketClient { >, callback: Callback, ) => Subscription; + + subscribeConnectionState: (callback: ConnectionStateCallback) => Subscription; + + getConnectionState: () => WebSocketConnectionState; + + reconnect: () => void; } +/** + * - `closed`: not connected, because nothing is subscribed. + * - `connecting`: the first connection attempt is in progress. + * - `open`: connected and authenticated, live updates are being delivered. + * - `reconnecting`: the connection failed or was lost and the client is retrying. `error` holds + * the error that caused it. Subscriptions stay alive and receive a synthetic invalidate once + * the connection is back. + * - `fatal`: the client is unusable. All subscriptions were terminated with `error`. Call + * `reconnect()` or create a new client. + */ +export type WebSocketConnectionStatus = + "closed" | "connecting" | "open" | "reconnecting" | "fatal"; + +export type WebSocketConnectionState = { + status: WebSocketConnectionStatus; + error?: BrokerizeWebSocketError; +}; + +export type ConnectionStateCallback = (state: WebSocketConnectionState) => void; + +/** + * Callback of a subscription. `err` is always a `BrokerizeWebSocketError` with + * `terminal === true`: the subscription is over and will not receive any further data. Errors the + * client recovers from on its own are reported via `subscribeConnectionState` instead. + */ export type Callback = (err: any, data: T) => void; export type Subscription = { diff --git a/src/websocketTypes.ts b/src/websocketTypes.ts index 32dcff9..90cd76c 100644 --- a/src/websocketTypes.ts +++ b/src/websocketTypes.ts @@ -86,8 +86,18 @@ export type UpdateDecoupledOperationMessage = { status: DecoupledOperationStatus; }; +/** + * The error payload as it is sent over the websocket. Applications do not see this type: the + * websocket client converts it into a `BrokerizeWebSocketError` (the exported error class) before + * passing it on. + */ export type BrokerizeWebSocketError = { message: string; + /** + * A stable error code, if the server sends one. Preferred over matching `message`, which is + * meant to be human-readable and may change. + */ + code?: string; }; export type WebSocketError = { diff --git a/test/websocketClient.test.js b/test/websocketClient.test.js new file mode 100644 index 0000000..6c97ea1 --- /dev/null +++ b/test/websocketClient.test.js @@ -0,0 +1,363 @@ +import assert from "node:assert/strict"; +import { after, describe, it } from "node:test"; + +import { BrokerizeWebSocketClientImpl } from "../dist/websocketClient.js"; +import { + BrokerizeWebSocketError, + isBrokerizeWebSocketError, +} from "../dist/errors.js"; + +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * The backoff of the client retries after ~100-1100ms on the first error and after ~1-2s + * afterwards, so waiting this long is enough to observe a reconnect. + */ +const RECONNECT_WAIT = 2500; + +class FakeSocket { + constructor(url) { + this.url = url; + this.readyState = 0; + this.sent = []; + } + send(data) { + this.sent.push(JSON.parse(data)); + } + close() { + this.readyState = 3; + } + + /* test helpers */ + open() { + this.readyState = 1; + this.onopen?.({}); + } + receive(obj) { + this.onmessage?.({ data: JSON.stringify(obj) }); + } + receiveRaw(data) { + this.onmessage?.({ data }); + } + fail(ev) { + this.onerror?.(ev || { type: "error" }); + } + serverClose() { + this.readyState = 3; + this.onclose?.({}); + } + get subscriptionId() { + return this.sent.find((m) => m.cmd == "subscribe")?.subscriptionId; + } +} + +const createdClients = []; + +/** + * Every client gets its own socket list so that a client that is still reconnecting from an + * earlier test cannot interfere. + */ +function createClient(auth) { + const sockets = []; + const client = new BrokerizeWebSocketClientImpl( + "wss://example.invalid/websocket", + auth || { getToken: async () => ({ idToken: "token" }) }, + (url) => { + const socket = new FakeSocket(url); + sockets.push(socket); + return socket; + }, + ); + createdClients.push(client); + return { client, sockets, last: () => sockets[sockets.length - 1] }; +} + +/** Open the connection and let the server confirm the authorization. */ +async function authenticate(socket) { + socket.open(); + await sleep(5); + socket.receive({ cmd: "authenticated" }); +} + +/** Open the connection and let `getToken()` settle, without the server confirming it. */ +async function openOnly(socket) { + socket.open(); + await sleep(5); +} + +after(() => createdClients.forEach((c) => c.destroy())); + +describe("BrokerizeWebSocketClient error handling", () => { + it("reports an error on a subscription as a terminal BrokerizeWebSocketError", async () => { + const { client, sockets } = createClient(); + const seen = []; + client.subscribeInvalidate({ entity: "sessions" }, (err, data) => + seen.push({ err, data }), + ); + await authenticate(sockets[0]); + sockets[0].receive({ + subscriptionId: sockets[0].subscriptionId, + error: { message: "no access to portfolio" }, + }); + + const err = seen[0].err; + assert.ok(err instanceof Error); + assert.ok(isBrokerizeWebSocketError(err)); + assert.equal(err.message, "no access to portfolio"); + assert.equal(err.code, "WEBSOCKET_ERROR"); + assert.equal(err.terminal, true); + assert.ok(err.stack); + }); + + it("retries a rejected token once with a forced refresh before giving up", async () => { + const tokenCalls = []; + const { client, sockets, last } = createClient({ + getToken: async (opts) => { + tokenCalls.push(opts); + return { idToken: "token" }; + }, + }); + const errors = []; + client.subscribeInvalidate({ entity: "sessions" }, (err) => { + if (err) errors.push(err); + }); + await authenticate(sockets[0]); + + sockets[0].receive({ error: { message: "authentication failed" } }); + await sleep(10); + assert.deepEqual( + errors, + [], + "the first rejection must not reach subscribers", + ); + assert.equal(client.getConnectionState().status, "reconnecting"); + + await sleep(RECONNECT_WAIT); + assert.ok(sockets.length >= 2, "it must connect again on its own"); + await openOnly(last()); + assert.ok( + tokenCalls.some((opts) => opts?.forceRefresh === true), + "the retry must ask for a fresh token", + ); + + last().receive({ error: { message: "authentication failed" } }); + await sleep(10); + assert.equal( + errors.length, + 1, + "the second rejection terminates subscriptions", + ); + assert.ok(isBrokerizeWebSocketError(errors[0])); + assert.equal(errors[0].code, "AUTH"); + assert.equal(errors[0].httpStatusCode, 401); + assert.equal(errors[0].terminal, true); + assert.equal(client.getConnectionState().status, "fatal"); + assert.equal(client.getConnectionState().error, errors[0]); + }); + + it("restores the retry budget after a successful authentication", async () => { + const { client, sockets, last } = createClient(); + const errors = []; + client.subscribeInvalidate({ entity: "sessions" }, (err) => { + if (err) errors.push(err); + }); + await authenticate(sockets[0]); + + /* the token expired mid-session */ + sockets[0].receive({ error: { message: "authentication failed" } }); + await sleep(RECONNECT_WAIT); + await authenticate(last()); + assert.deepEqual(errors, [], "recovering must not terminate subscriptions"); + assert.equal(client.getConnectionState().status, "open"); + + /* ... and if it happens again later, it is retried again rather than being fatal */ + last().receive({ error: { message: "authentication failed" } }); + await sleep(10); + assert.equal(client.getConnectionState().status, "reconnecting"); + }); + + it("revives a client with a fatal error via reconnect()", async () => { + const { client, sockets, last } = createClient(); + const errors = []; + client.subscribeInvalidate({ entity: "sessions" }, (err) => { + if (err) errors.push(err); + }); + await authenticate(sockets[0]); + sockets[0].receive({ error: { message: "authentication failed" } }); + await sleep(RECONNECT_WAIT); + await openOnly(last()); + last().receive({ error: { message: "authentication failed" } }); + await sleep(10); + assert.equal(client.getConnectionState().status, "fatal"); + + const socketsBefore = sockets.length; + client.reconnect(); + const errorsAfterReconnect = []; + client.subscribeInvalidate({ entity: "sessions" }, (err) => { + if (err) errorsAfterReconnect.push(err); + }); + await sleep(RECONNECT_WAIT); + assert.ok(sockets.length > socketsBefore); + await authenticate(last()); + assert.equal(client.getConnectionState().status, "open"); + assert.equal( + errors.length, + 1, + "terminated subscribers must not be called again", + ); + assert.deepEqual(errorsAfterReconnect, []); + }); + + it("reports the fatal error asynchronously when subscribing on a dead client", async () => { + const { client, sockets, last } = createClient(); + client.subscribeInvalidate({ entity: "sessions" }, () => {}); + await authenticate(sockets[0]); + sockets[0].receive({ error: { message: "authentication failed" } }); + await sleep(RECONNECT_WAIT); + await openOnly(last()); + last().receive({ error: { message: "authentication failed" } }); + await sleep(10); + + let subscription = null; + let calledBeforeReturn = false; + let seen = null; + subscription = client.subscribeInvalidate({ entity: "sessions" }, (err) => { + if (subscription === null) calledBeforeReturn = true; + seen = err; + }); + assert.equal( + calledBeforeReturn, + false, + "the caller must hold the Subscription before the callback runs", + ); + assert.equal(typeof subscription.unsubscribe, "function"); + await sleep(5); + assert.ok(isBrokerizeWebSocketError(seen)); + assert.equal(seen.code, "AUTH"); + }); + + it("keeps notifying the other subscribers when one callback throws", async () => { + const { client, sockets } = createClient(); + let reached = false; + client.subscribeInvalidate({ entity: "sessions" }, () => { + throw new Error("boom from a subscriber"); + }); + client.subscribeInvalidate({ entity: "sessions" }, () => { + reached = true; + }); + await authenticate(sockets[0]); + sockets[0].receive({ + cmd: "invalidate", + subscriptionId: sockets[0].subscriptionId, + }); + assert.ok(reached); + }); + + it("does not throw out of the message handler on an unparsable message", async () => { + const { client, sockets } = createClient(); + const states = []; + client.subscribeConnectionState((s) => states.push(s)); + client.subscribeInvalidate({ entity: "sessions" }, () => {}); + await authenticate(sockets[0]); + + assert.doesNotThrow(() => + sockets[0].receiveRaw("502 Bad Gateway"), + ); + assert.equal( + states[states.length - 1].error?.code, + "WEBSOCKET_PROTOCOL_ERROR", + ); + }); + + it("closes and retries instead of hanging when getToken() rejects", async () => { + let calls = 0; + const { client, sockets, last } = createClient({ + getToken: async () => { + calls++; + if (calls == 1) { + throw new BrokerizeWebSocketError({ + msg: "refresh token expired", + code: "AUTH", + }); + } + return { idToken: "token" }; + }, + }); + const states = []; + client.subscribeConnectionState((s) => states.push(s)); + client.subscribeInvalidate({ entity: "sessions" }, () => {}); + sockets[0].open(); + await sleep(20); + assert.equal(states[states.length - 1].error?.code, "AUTH"); + + await sleep(RECONNECT_WAIT); + assert.ok( + sockets.length >= 2, + "it must not stay on the unauthenticated socket", + ); + await authenticate(last()); + assert.equal(client.getConnectionState().status, "open"); + }); + + it("counts a failed connection attempt once for the backoff", async () => { + const { client, sockets } = createClient(); + client.subscribeInvalidate({ entity: "sessions" }, () => {}); + await sleep(5); + const states = []; + client.subscribeConnectionState((s) => states.push(s)); + + /* both an error and a close event for a single failed attempt */ + sockets[0].fail(); + sockets[0].serverClose(); + await sleep(10); + + const state = states[states.length - 1]; + assert.equal(state.status, "reconnecting"); + assert.ok(state.error instanceof Error); + assert.equal(state.error.code, "WEBSOCKET_CONNECTION_FAILED"); + assert.equal(client._errorCount, 1); + }); + + it("does not let the close event of an outdated socket drop the new connection", async () => { + const { client, sockets, last } = createClient(); + client.subscribeInvalidate({ entity: "sessions" }, () => {}); + await sleep(5); + const first = sockets[0]; + await authenticate(first); + assert.equal(client.getConnectionState().status, "open"); + + client.reconnect(); + await sleep(5); + const second = last(); + assert.notEqual(second, first); + + first.onclose?.({}); /* the late close event of the replaced socket */ + await authenticate(second); + assert.equal(client.getConnectionState().status, "open"); + }); + + it("stops reconnecting when destroyed", async () => { + const { client, sockets } = createClient(); + const subscription = client.subscribeInvalidate( + { entity: "sessions" }, + () => {}, + ); + await authenticate(sockets[0]); + + client.destroy(); + assert.equal(sockets[0].readyState, 3); + const socketsBefore = sockets.length; + await sleep(RECONNECT_WAIT); + assert.equal(sockets.length, socketsBefore); + assert.equal(client.getConnectionState().status, "closed"); + + let seen = null; + client.subscribeInvalidate({ entity: "sessions" }, (err) => { + seen = err; + }); + await sleep(5); + assert.ok(isBrokerizeWebSocketError(seen)); + assert.equal(seen.terminal, true); + assert.doesNotThrow(() => subscription.unsubscribe()); + }); +});