From 9e14651af3673ef22ba6344f4062eb946ac56013 Mon Sep 17 00:00:00 2001 From: FZambia Date: Wed, 5 Aug 2026 08:23:10 +0300 Subject: [PATCH 1/6] Add transport selection and dependency resolution tests The dependency resolution and selection logic in _initializeTransport is about to be extracted into helpers. Two of the five transports, sockjs and webtransport, had zero function coverage, so a mistake in moving their construction or supported() wiring would not have been detected. Covers selection and construction for sockjs and webtransport via fakes, equivalence between resolving a dependency from config and from globalThis, unsupported handling for each transport driven through the real resolution path, skipping an unsupported entry, and connecting via a string endpoint. No test depends on which globals the running Node version provides: CI spans Node 18-25 and globalThis.WebSocket is not present across that whole range, so a dependency that must be present is passed via config and one that must be absent is deleted and restored explicitly. --- src/transport_selection.test.ts | 352 ++++++++++++++++++++++++++++++++ 1 file changed, 352 insertions(+) create mode 100644 src/transport_selection.test.ts diff --git a/src/transport_selection.test.ts b/src/transport_selection.test.ts new file mode 100644 index 0000000..6a70a65 --- /dev/null +++ b/src/transport_selection.test.ts @@ -0,0 +1,352 @@ +import { Centrifuge } from './centrifuge'; +import { TransportName, State } from './types'; + +import WebSocket from 'ws'; +import EventSource from 'eventsource'; +import { fetch } from 'undici'; +import { ReadableStream } from 'node:stream/web'; + +// Protection for the transport dependency-resolution and selection logic in +// _initializeTransport(). Those two concerns are about to be extracted into +// helpers, and two of the five transports (sockjs, webtransport) currently have +// zero function coverage, so a mistake in moving their construction or their +// supported() wiring would go undetected. +// +// Everything here passes against unmodified source: this file is the baseline +// the extraction has to preserve, not a specification of new behavior. +// +// Note deliberately absent from this file: nothing may depend on which globals +// the running Node version happens to provide. CI spans Node 18-25 and +// globalThis.WebSocket in particular is not present across that whole range, so +// a dependency that must be present is always passed via config, and one that +// must be absent is deleted and restored explicitly. + +const wsEndpoint = 'ws://localhost:8000/connection/websocket'; +const sseEndpoint = 'http://localhost:8000/connection/sse'; +const httpStreamEndpoint = 'http://localhost:8000/connection/http_stream'; +const sockjsEndpoint = 'http://localhost:8000/connection/sockjs'; +const wtEndpoint = 'https://localhost:8000/connection/webtransport'; + +/** Records construction and satisfies the shape SockjsTransport drives. */ +class FakeSockJS { + static instances: FakeSockJS[] = []; + url: string; + protocols: any; + options: any; + transport = 'fake-websocket'; + onopen: any = null; + onerror: any = null; + onclose: any = null; + onmessage: any = null; + closed = false; + + constructor(url: string, protocols: any, options: any) { + this.url = url; + this.protocols = protocols; + this.options = options; + FakeSockJS.instances.push(this); + } + close() { this.closed = true; } + send(_data: any) { /* no-op */ } +} + +/** Never resolves ready/closed, so initialize() parks after selection. */ +class FakeWebTransport { + static instances: FakeWebTransport[] = []; + url: string; + ready = new Promise(() => { /* never settles */ }); + closed = new Promise(() => { /* never settles */ }); + + constructor(url: string) { + this.url = url; + FakeWebTransport.instances.push(this); + } + close() { /* no-op */ } + createBidirectionalStream() { return new Promise(() => { /* never settles */ }); } +} + +const clients: Centrifuge[] = []; + +// Short, so the connect timeout below drains quickly. +const CONNECT_TIMEOUT = 100; + +function makeClient(endpoint: any, options: any): Centrifuge { + const c = new Centrifuge(endpoint, { timeout: CONNECT_TIMEOUT, ...options }); + clients.push(c); + c.on('error', () => { /* selection tests assert state, not events */ }); + return c; +} + +afterEach(async () => { + // Runs before any global-restoring afterEach registered later in a describe, + // so a still-cycling client never observes a dependency vanish mid-flight. + while (clients.length) { + const c = clients.pop(); + // disconnect() can throw against unmodified source when connect() gave up + // holding an unsupported transport; that leak is pinned in + // transport_init_error.test.ts, and must not mask assertions here. + try { c!.disconnect(); } catch { /* pinned elsewhere */ } + } + // disconnect() does not clear the connect timeout: it is a closure local of + // _initializeTransport, cleared only from onOpen/onClose. A transport that + // never opens therefore leaves an armed timer behind, which CI reports as an + // open handle since it runs jest --detectOpenHandles with no --forceExit. + // Waiting it out keeps this file honest against unmodified source; the fix + // moves the timeout onto the instance so _disconnect can clear it. + await new Promise(resolve => setTimeout(resolve, CONNECT_TIMEOUT + 50)); +}); + +/** Deletes globals for the duration of fn, restoring them afterwards. */ +function withoutGlobals(names: string[], fn: () => T): T { + const saved: Record = {}; + const had: Record = {}; + for (const n of names) { + had[n] = n in (globalThis as any); + saved[n] = (globalThis as any)[n]; + delete (globalThis as any)[n]; + } + try { + return fn(); + } finally { + for (const n of names) { + if (had[n]) { + (globalThis as any)[n] = saved[n]; + } + } + } +} + +function selectedTransportName(c: Centrifuge): string { + return (c as any)._transport.name(); +} + +describe('sockjs selection', () => { + beforeEach(() => { FakeSockJS.instances = []; }); + + test('is selected and constructed with the configured endpoint', () => { + const c = makeClient([{ transport: 'sockjs' as TransportName, endpoint: sockjsEndpoint }], { + sockjs: FakeSockJS, + sockjsOptions: { some: 'option' }, + }); + + c.connect(); + + expect(selectedTransportName(c)).toBe('sockjs'); + expect(FakeSockJS.instances).toHaveLength(1); + expect(FakeSockJS.instances[0].url).toBe(sockjsEndpoint); + expect(FakeSockJS.instances[0].options).toEqual({ some: 'option' }); + }); + + test('initialize() wires all four callbacks onto the instance', () => { + const c = makeClient([{ transport: 'sockjs' as TransportName, endpoint: sockjsEndpoint }], { + sockjs: FakeSockJS, + }); + + c.connect(); + + const fake = FakeSockJS.instances[0]; + expect(typeof fake.onopen).toBe('function'); + expect(typeof fake.onerror).toBe('function'); + expect(typeof fake.onclose).toBe('function'); + expect(typeof fake.onmessage).toBe('function'); + }); + + test('subName() reports the underlying sockjs transport once initialized', () => { + const c = makeClient([{ transport: 'sockjs' as TransportName, endpoint: sockjsEndpoint }], { + sockjs: FakeSockJS, + }); + + c.connect(); + + expect((c as any)._transport.subName()).toBe('sockjs-fake-websocket'); + }); + + test('is unsupported when no SockJS is available', () => { + withoutGlobals(['SockJS'], () => { + const c = makeClient([{ transport: 'sockjs' as TransportName, endpoint: sockjsEndpoint }], {}); + expect(() => c.connect()).toThrow(/no supported transport found/); + }); + }); +}); + +describe('webtransport selection', () => { + beforeEach(() => { FakeWebTransport.instances = []; }); + + test('is selected and constructed from globalThis.WebTransport', () => { + const saved = (globalThis as any).WebTransport; + (globalThis as any).WebTransport = FakeWebTransport; + try { + const c = makeClient([{ transport: 'webtransport' as TransportName, endpoint: wtEndpoint }], {}); + + c.connect(); + + expect(selectedTransportName(c)).toBe('webtransport'); + expect(FakeWebTransport.instances).toHaveLength(1); + expect(FakeWebTransport.instances[0].url).toBe(wtEndpoint); + } finally { + if (saved === undefined) { + delete (globalThis as any).WebTransport; + } else { + (globalThis as any).WebTransport = saved; + } + } + }); + + test('is unsupported when globalThis.WebTransport is absent', () => { + withoutGlobals(['WebTransport'], () => { + const c = makeClient([{ transport: 'webtransport' as TransportName, endpoint: wtEndpoint }], {}); + expect(() => c.connect()).toThrow(/no supported transport found/); + }); + }); +}); + +describe('dependency source equivalence: config vs globalThis', () => { + // The resolution at the top of _initializeTransport prefers an explicit config + // value and otherwise falls back to a global. Both routes must select the same + // transport — the invariant the extracted resolver has to preserve. + + test('websocket resolves from config', () => { + const c = makeClient([{ transport: 'websocket' as TransportName, endpoint: wsEndpoint }], { + websocket: WebSocket, + }); + c.connect(); + expect(selectedTransportName(c)).toBe('websocket'); + }); + + test('websocket resolves from globalThis', () => { + const saved = (globalThis as any).WebSocket; + (globalThis as any).WebSocket = WebSocket; + try { + const c = makeClient([{ transport: 'websocket' as TransportName, endpoint: wsEndpoint }], {}); + c.connect(); + expect(selectedTransportName(c)).toBe('websocket'); + } finally { + if (saved === undefined) { + delete (globalThis as any).WebSocket; + } else { + (globalThis as any).WebSocket = saved; + } + } + }); + + test('sockjs resolves from config', () => { + const c = makeClient([{ transport: 'sockjs' as TransportName, endpoint: sockjsEndpoint }], { + sockjs: FakeSockJS, + }); + c.connect(); + expect(selectedTransportName(c)).toBe('sockjs'); + }); + + test('sockjs resolves from globalThis', () => { + const saved = (globalThis as any).SockJS; + (globalThis as any).SockJS = FakeSockJS; + try { + const c = makeClient([{ transport: 'sockjs' as TransportName, endpoint: sockjsEndpoint }], {}); + c.connect(); + expect(selectedTransportName(c)).toBe('sockjs'); + } finally { + if (saved === undefined) { + delete (globalThis as any).SockJS; + } else { + (globalThis as any).SockJS = saved; + } + } + }); + + test('sse resolves from config', () => { + const c = makeClient([{ transport: 'sse' as TransportName, endpoint: sseEndpoint }], { + eventsource: EventSource, + fetch: fetch, + }); + c.connect(); + expect(selectedTransportName(c)).toBe('sse'); + }); + + test('http_stream resolves from config', () => { + const c = makeClient([{ transport: 'http_stream' as TransportName, endpoint: httpStreamEndpoint }], { + fetch: fetch, + readableStream: ReadableStream, + }); + c.connect(); + expect(selectedTransportName(c)).toBe('http_stream'); + }); +}); + +describe('unsupported dependencies are reported, not silently accepted', () => { + // supported() is checked with `!== null` only for sockjs, sse and http_stream, + // and with `!== undefined && !== null` for websocket and webtransport. The + // resolver must therefore yield null — not undefined — for an absent + // dependency, or supported() returns true and construction fails later on an + // undefined constructor. Each case here drives absence through the real + // resolution path rather than passing an explicit undefined. + + test('websocket: array config with no WebSocket anywhere', () => { + withoutGlobals(['WebSocket'], () => { + const c = makeClient([{ transport: 'websocket' as TransportName, endpoint: wsEndpoint }], {}); + expect(() => c.connect()).toThrow(/no supported transport found/); + }); + }); + + test('websocket: string endpoint with no WebSocket anywhere', () => { + withoutGlobals(['WebSocket'], () => { + const c = makeClient(wsEndpoint, {}); + expect(() => c.connect()).toThrow(/WebSocket constructor not found/); + }); + }); + + test('sse: no EventSource anywhere', () => { + withoutGlobals(['EventSource'], () => { + const c = makeClient([{ transport: 'sse' as TransportName, endpoint: sseEndpoint }], { + fetch: fetch, + }); + expect(() => c.connect()).toThrow(/no supported transport found/); + }); + }); + + test('http_stream: no fetch anywhere', () => { + withoutGlobals(['fetch'], () => { + const c = makeClient([{ transport: 'http_stream' as TransportName, endpoint: httpStreamEndpoint }], { + readableStream: ReadableStream, + }); + expect(() => c.connect()).toThrow(/no supported transport found/); + }); + }); + + test('http_stream: no ReadableStream anywhere', () => { + withoutGlobals(['ReadableStream'], () => { + const c = makeClient([{ transport: 'http_stream' as TransportName, endpoint: httpStreamEndpoint }], { + fetch: fetch, + }); + expect(() => c.connect()).toThrow(/no supported transport found/); + }); + }); +}); + +describe('selection order', () => { + test('skips an unsupported entry and selects the next supported one', () => { + withoutGlobals(['SockJS'], () => { + const c = makeClient([ + { transport: 'sockjs' as TransportName, endpoint: sockjsEndpoint }, + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + ], { + websocket: WebSocket, + }); + + c.connect(); + + expect(selectedTransportName(c)).toBe('websocket'); + }); + }); +}); + +describe('non-emulation string endpoint', () => { + test('connects successfully against the server', async () => { + const c = makeClient(wsEndpoint, { websocket: WebSocket }); + + c.connect(); + await c.ready(3000); + + expect(c.state).toBe(State.Connected); + expect(selectedTransportName(c)).toBe('websocket'); + }); +}); From f21519af592bb96fd56053ddc8b566840c0fe249 Mon Sep 17 00:00:00 2001 From: FZambia Date: Wed, 5 Aug 2026 08:33:09 +0300 Subject: [PATCH 2/6] Report transport initialize failures instead of dying silently (#268) new WebSocket(url) is not total. Browsers throw SecurityError for a ws:// URL on an https:// page and for a URL blocked by the CSP connect-src directive, SyntaxError for a malformed URL, and a replaced global WebSocket throws for its own reasons. The connect timeout was armed before initialize(), so a throw left the wrapper's inner transport null with the timeout already scheduled: it fired transport.close() against null seconds later, from a timer frame carrying no URL, transport name or connect context. That is the TypeError reported in the issue. Worse than the log line, the client died. _transportClosed is set false just before initialize(), and with no socket nothing ever delivered onClose to reset it, so every later reconnect attempt bailed on the "waiting for transport close" guard and the client sat in connecting for the life of the page. With getToken configured the cause was also misreported: initialize() runs inside a promise callback there, so the SecurityError surfaced as a connectToken error, pointing diagnosis at the token subsystem. Failures during a connection attempt now go through one path that resets _transportClosed, advances the transport index, and emits the existing transport error carrying the exception's own message - so a CSP block names the directive that caused it, and a client configured with several transports falls through to the next one. Under connect-src 'self' that succeeds, since same-origin https is permitted while wss is not. The connect timeout uses the same path rather than calling close() and waiting for the transport to report back, so a transport that accepts close() silently no longer wedges the client, and disconnect() now clears that timeout instead of leaving it armed. Configuration faults that no retry could fix still throw, but now from connect() itself, before any state change and before the getToken hop, so they behave the same however the client is configured. Also fixed while restructuring the selection loop: it assigned this._transport before testing supported(), leaving the client holding a never-initialized wrapper when it gave up, and it wrapped the transport index only on entry, so an unsupported entry after a failed one could read past the end of the list. --- src/centrifuge.ts | 328 +++++++---- src/transport_http_stream.ts | 3 +- src/transport_init_error.test.ts | 977 +++++++++++++++++++++++++++++++ src/transport_sockjs.ts | 10 +- src/transport_sse.ts | 9 +- src/transport_websocket.ts | 8 +- 6 files changed, 1219 insertions(+), 116 deletions(-) create mode 100644 src/transport_init_error.test.ts diff --git a/src/centrifuge.ts b/src/centrifuge.ts index 02469f1..588b85e 100644 --- a/src/centrifuge.ts +++ b/src/centrifuge.ts @@ -115,6 +115,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter = null; + private _connectTimeout: null | ReturnType = null; private _reconnectAttempts: number; private _client: null; private _session: string; @@ -163,6 +164,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter TypedEventEmitter TypedEventEmitter TypedEventEmitter= this._transports.length) { - this._triedAllTransports = true; - this._currentTransportIndex = 0; + if (!this._createTransport('websocket', this._endpoint as string, deps).supported()) { + throw new Error('WebSocket constructor not found, make sure it is available globally or passed as a dependency in Centrifuge options'); } + return; + } + + for (const transportConfig of this._transports) { + if (this._createTransport(transportConfig.transport, transportConfig.endpoint, deps).supported()) { + return; + } + } + throw new Error('no supported transport found'); + } + + private _initializeTransport() { + const deps = this._resolveTransportDeps(); + + if (!this._emulation) { + this._debug('client will use websocket'); + this._transport = this._createTransport('websocket', this._endpoint as string, deps); + } else { + let selected: any = null; let count = 0; - while (true) { - if (count >= this._transports.length) { - throw new Error('no supported transport found'); + while (count < this._transports.length) { + // Wrapped here rather than only on entry: the index is advanced by a + // failed attempt, so a later attempt can start past the end of the list. + if (this._currentTransportIndex >= this._transports.length) { + this._triedAllTransports = true; + this._currentTransportIndex = 0; } const transportConfig = this._transports[this._currentTransportIndex]; - const transportName = transportConfig.transport; - const transportEndpoint = transportConfig.endpoint; - - if (transportName === 'websocket') { - this._debug('trying websocket transport'); - this._transport = new WebsocketTransport(transportEndpoint, { - websocket: websocket - }); - if (!this._transport.supported()) { - this._debug('websocket transport not available'); - this._currentTransportIndex++; - count++; - continue; - } - } else if (transportName === 'webtransport') { - this._debug('trying webtransport transport'); - this._transport = new WebtransportTransport(transportEndpoint, { - webtransport: globalThis.WebTransport, - decoder: this._codec, - encoder: this._codec - }); - if (!this._transport.supported()) { - this._debug('webtransport transport not available'); - this._currentTransportIndex++; - count++; - continue; - } - } else if (transportName === 'http_stream') { - this._debug('trying http_stream transport'); - this._transport = new HttpStreamTransport(transportEndpoint, { - fetch: fetchFunc, - readableStream: readableStream, - emulationEndpoint: this._config.emulationEndpoint, - decoder: this._codec, - encoder: this._codec - }); - if (!this._transport.supported()) { - this._debug('http_stream transport not available'); - this._currentTransportIndex++; - count++; - continue; - } - } else if (transportName === 'sse') { - this._debug('trying sse transport'); - this._transport = new SseTransport(transportEndpoint, { - eventsource: eventsource, - fetch: fetchFunc, - emulationEndpoint: this._config.emulationEndpoint, - }); - if (!this._transport.supported()) { - this._debug('sse transport not available'); - this._currentTransportIndex++; - count++; - continue; - } - } else if (transportName === 'sockjs') { - this._debug('trying sockjs'); - this._transport = new SockjsTransport(transportEndpoint, { - sockjs: sockjs, - sockjsOptions: this._config.sockjsOptions - }); - if (!this._transport.supported()) { - this._debug('sockjs transport not available'); - this._currentTransportIndex++; - count++; - continue; - } - } else { - throw new Error('unknown transport ' + transportName); + this._debug('trying ' + transportConfig.transport + ' transport'); + const candidate = this._createTransport(transportConfig.transport, transportConfig.endpoint, deps); + // Assigned only once supported, so giving up never leaves the client + // holding a wrapper that was never initialized. + if (candidate.supported()) { + selected = candidate; + break; } - break; + this._debug(transportConfig.transport + ' transport not available'); + this._currentTransportIndex++; + count++; + } + if (selected === null) { + // connect() validated that at least one transport was supported, so + // getting here means a dependency disappeared under a running client. + // Retries run from a timer with no call site, so report it like any + // other transport failure rather than throwing. + this._debug('no supported transport found on reconnect'); + this._transportClosed = true; + this._reconnecting = false; + this.emit('error', { + type: 'transport', + error: { + code: errorCodes.transportClosed, + message: 'no supported transport found' + } + }); + this._disconnect(connectingCodes.transportClosed, 'no supported transport found', true); + return; } + this._transport = selected; } const self = this; @@ -991,17 +1029,57 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter { + if (self._transportId != transportId) { + self._debug('transport failure from non-actual transport', reason, e); + return; + } + self._clearConnectTimeout(); + // Nothing will ever deliver onClose for a transport that failed to + // initialize or never opened, so this reset is what keeps the client + // reconnecting instead of parking on the "waiting for transport close" + // guard for the rest of the page's life. + self._transportClosed = true; + self._debug(transport.name(), 'transport failed:', reason, e); + if (self._emulation && !self._transportWasOpen) { + self._currentTransportIndex++; + if (self._currentTransportIndex >= self._transports.length) { + self._triedAllTransports = true; + self._currentTransportIndex = 0; + } + } + if (self._isConnecting() && !wasOpen) { + self.emit('error', { + type: 'transport', + error: { + code: errorCodes.transportClosed, + // The exception message is the whole diagnostic payload here: it is + // what names the CSP directive or the mixed-content scheme. + message: (e && e.message) ? e.message : reason + }, + transport: transport.name() + }); + } + self._reconnecting = false; + self._disconnect(connectingCodes.transportClosed, reason, true); + }; + + // A verdict on the attempt, not a request for the transport to report back: + // a transport whose close() is a silent no-op would never deliver onClose, + // and the client would never reconnect. _disconnect closes the transport, so + // this must not close it as well. + this._clearConnectTimeout(); + this._connectTimeout = setTimeout(function () { + failTransport('connect timeout'); }, this._config.timeout); - this._transport.initialize(this._codecName(), { + try { + const initResult = this._transport.initialize(this._codecName(), { onOpen: function () { - if (connectTimeout) { - clearTimeout(connectTimeout); - connectTimeout = null; - } + self._clearConnectTimeout(); if (self._transportId != transportId) { self._debug('open callback from non-actual transport'); transport.close(); @@ -1029,10 +1107,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter TypedEventEmitter failTransport('transport initialize error', e)); + } + } catch (e) { + failTransport('transport initialize error', e); + return; + } //@ts-ignore must be used only for debug and test purposes. self.emit('__centrifuge_debug:transport_initialized', {}) } + private _clearConnectTimeout() { + if (this._connectTimeout !== null) { + clearTimeout(this._connectTimeout); + this._connectTimeout = null; + } + } + private _sendConnect(skipSending: boolean): any { const connectCommand = this._constructConnectCommand(); const self = this; @@ -1506,19 +1597,34 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter { /* no-op */ }; + if (closeBehavior === 'noop') { + // Accepts close() and reports nothing back - no onclose ever arrives. + this.close = () => { counter.closes++; }; + } else if (closeBehavior === 'throw') { + this.close = () => { counter.closes++; throw new Error('close is not available'); }; + } + // 'absent': no close method at all. + } as any; +} + +/** + * Tears a client down without letting an unrelated teardown bug mask the + * assertion a test is actually making. The throw itself is pinned by + * 'disconnect() after connect() gave up does not throw'. + */ +function safeDisconnect(c: Centrifuge) { + try { c.disconnect(); } catch { /* pinned separately */ } +} + +const liveClients: Centrifuge[] = []; + +/** + * Registers the client for teardown. CI runs jest --detectOpenHandles with no + * --forceExit, so a client still cycling at the end of a test would be reported + * as an open handle. + */ +function track(c: Centrifuge): Centrifuge { + liveClients.push(c); + return c; +} + +afterEach(() => { + while (liveClients.length) { + safeDisconnect(liveClients.pop()!); + } +}); + +function waitFor(predicate: () => boolean, timeout = 4000): Promise { + return new Promise((resolve, reject) => { + const started = Date.now(); + const tick = () => { + if (predicate()) return resolve(); + if (Date.now() - started > timeout) return reject(new Error('timeout waiting for condition')); + setTimeout(tick, 10); + }; + tick(); + }); +} + +/** + * Records unhandled rejections for the duration of a test. Uncaught exceptions + * thrown from timers are already surfaced by jest as test failures, so an + * orphaned connect timeout firing transport.close() on a null transport fails + * the test on its own. + */ +function collectUnhandledRejections() { + const seen: any[] = []; + const handler = (reason: any) => seen.push(reason); + process.on('unhandledRejection', handler); + return { + seen, + restore: () => process.off('unhandledRejection', handler), + }; +} + +const wsEndpoint = 'ws://localhost:8000/connection/websocket'; +const httpStreamEndpoint = 'http://localhost:8000/connection/http_stream'; +const sseEndpoint = 'http://localhost:8000/connection/sse'; +const emulationEndpoint = 'http://localhost:8000/emulation'; + +describe('transport initialize() throwing synchronously', () => { + test('connect() does not throw out of the client', () => { + const counter = { calls: 0 }; + const c = new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 50, + maxReconnectDelay: 50, + }); + c.on('error', () => { /* swallow, asserted elsewhere */ }); + + expect(() => c.connect()).not.toThrow(); + safeDisconnect(c); + }); + + test('reports the real reason as a transport error, not a token error', async () => { + const counter = { calls: 0 }; + const errors: ErrorContext[] = []; + + const c = new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 50, + maxReconnectDelay: 50, + }); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await waitFor(() => errors.length > 0); + + expect(errors[0].type).toBe('transport'); + expect(errors[0].transport).toBe('websocket'); + // The message is the whole diagnostic payload — a hardcoded 'transport + // closed' here leaves the developer with nothing to act on. + expect(errors[0].error.message).toContain('insecure WebSocket'); + + safeDisconnect(c); + }); + + test('misclassification guard: getToken path must not report connectToken', async () => { + // initialize() is called from inside a promise .then() when getToken is + // configured, so a synchronous throw lands in the getToken .catch() and is + // reported as a token problem — pointing diagnosis at the wrong subsystem. + const counter = { calls: 0 }; + const errors: ErrorContext[] = []; + + const c = new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 50, + maxReconnectDelay: 50, + getToken: () => Promise.resolve('token'), + }); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await waitFor(() => errors.length > 0); + + expect(errors.map(e => e.type)).not.toContain('connectToken'); + expect(errors[0].type).toBe('transport'); + + safeDisconnect(c); + }); + + test('keeps reconnecting instead of wedging forever', async () => { + // _transportClosed is set false right before initialize(); when initialize() + // throws there is no socket to ever deliver onClose, so without an explicit + // reset every later _startReconnecting() bails at the "waiting for transport + // close" guard and the client is dead for the life of the page. + const counter = { calls: 0 }; + const c = new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 50, + maxReconnectDelay: 50, + }); + c.on('error', () => { /* expected on every attempt */ }); + + c.connect(); + await waitFor(() => counter.calls >= 3); + + expect(counter.calls).toBeGreaterThanOrEqual(3); + expect(c.state).toBe(State.Connecting); + + safeDisconnect(c); + }); + + test('emits a transport error on every attempt, not only the first', async () => { + const counter = { calls: 0 }; + const errors: ErrorContext[] = []; + + const c = new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 50, + maxReconnectDelay: 50, + }); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await waitFor(() => errors.filter(e => e.type === 'transport').length >= 3); + + expect(errors.filter(e => e.type === 'transport').length).toBeGreaterThanOrEqual(3); + + safeDisconnect(c); + }); + + test('no orphaned connect timeout fires close() on a null transport', async () => { + // The connect timeout is armed before initialize(). If initialize() throws + // and the timeout is not cleared, it fires transport.close() seconds later + // against a null inner transport — the "null is not an object (evaluating + // 'this._transport.close')" report in the issue. jest fails this test if + // that exception escapes from the timer. + const counter = { calls: 0 }; + const rejections = collectUnhandledRejections(); + + const c = new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 50, // fires well within the wait below + minReconnectDelay: 10000, // keep a single attempt in flight + maxReconnectDelay: 10000, + }); + c.on('error', () => { /* expected */ }); + + c.connect(); + await new Promise(r => setTimeout(r, 400)); + + expect(rejections.seen).toEqual([]); + rejections.restore(); + safeDisconnect(c); + }); + + test('disconnect() after a failed initialize does not throw', async () => { + // _disconnect() nulls this._transport then calls close() on it with no + // try/catch — a throw there skips _scheduleReconnect() and leaves + // _transportClosed false, which is a second way to wedge the client. + const counter = { calls: 0 }; + const c = new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 50, + maxReconnectDelay: 50, + }); + c.on('error', () => { /* expected */ }); + + c.connect(); + await waitFor(() => counter.calls >= 1); + + expect(() => c.disconnect()).not.toThrow(); + expect(c.state).toBe(State.Disconnected); + }); +}); + +describe('transport fallback when initialize() throws', () => { + test('falls over from a blocked websocket to http_stream and connects', async () => { + // The realistic CSP shape: connect-src 'self' blocks the ws/wss scheme but + // permits same-origin https, so the emulation transports work. Note the + // selection loop only advances _currentTransportIndex on !supported(), and + // WebsocketTransport.supported() is true under a CSP block — the constructor + // exists, it just throws when invoked. Without an explicit advance the + // client retries the blocked transport forever and never reaches this one. + const counter = { calls: 0 }; + const errors: ErrorContext[] = []; + + const c = new Centrifuge([ + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + { transport: 'http_stream' as TransportName, endpoint: httpStreamEndpoint }, + ], { + websocket: throwingWebSocket(counter), + fetch: fetch, + readableStream: ReadableStream, + emulationEndpoint: emulationEndpoint, + timeout: 3000, + minReconnectDelay: 50, + maxReconnectDelay: 200, + }); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await c.ready(5000); + + expect(c.state).toBe(State.Connected); + expect(counter.calls).toBeGreaterThanOrEqual(1); + expect(errors.some(e => e.type === 'transport' && e.transport === 'websocket')).toBe(true); + + safeDisconnect(c); + }); + + test('falls over from a blocked sse to websocket and connects', async () => { + // Emulation transports call _sendConnect(true) before initialize(), which + // registers an outgoing command. When initialize() then throws, that + // callback is cleared by _clearOutgoingRequests() and its errback re-enters + // _disconnect() through _connectError — this asserts the reentrancy stays + // benign and the client still reaches the next transport. + const counter = { calls: 0 }; + const errors: ErrorContext[] = []; + + const c = new Centrifuge([ + { transport: 'sse' as TransportName, endpoint: sseEndpoint }, + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + ], { + websocket: WebSocket, + eventsource: throwingEventSource(counter), + fetch: fetch, + readableStream: ReadableStream, + emulationEndpoint: emulationEndpoint, + timeout: 3000, + minReconnectDelay: 50, + maxReconnectDelay: 200, + }); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await c.ready(5000); + + expect(c.state).toBe(State.Connected); + expect(counter.calls).toBeGreaterThanOrEqual(1); + expect(errors.some(e => e.type === 'transport' && e.transport === 'sse')).toBe(true); + + safeDisconnect(c); + }); + + test('all transports blocked: keeps cycling without wedging', async () => { + const wsCounter = { calls: 0 }; + const sseCounter = { calls: 0 }; + + const c = new Centrifuge([ + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + { transport: 'sse' as TransportName, endpoint: sseEndpoint }, + ], { + websocket: throwingWebSocket(wsCounter), + eventsource: throwingEventSource(sseCounter), + fetch: fetch, + readableStream: ReadableStream, + emulationEndpoint: emulationEndpoint, + timeout: 500, + minReconnectDelay: 50, + maxReconnectDelay: 100, + }); + c.on('error', () => { /* expected on every attempt */ }); + + c.connect(); + // Both must be tried, and the cycle must continue past a full sweep. + await waitFor(() => wsCounter.calls >= 2 && sseCounter.calls >= 2, 6000); + + expect(c.state).toBe(State.Connecting); + + safeDisconnect(c); + }); +}); + +// Static transport-configuration errors are a different class from the ones +// above: they depend only on the config object and the presence of globals, are +// decidable before any I/O, and are identical on every attempt. The client +// already throws them out of connect() -- but only when neither getToken nor +// getData is configured. With either one set, _initializeTransport() runs inside +// a promise callback, so `.then(f).catch(h)` hands the throw to the token/data +// handler and it is reported as a token problem instead. +// +// It then degrades: once _token is set, needTokenRefresh is false, so the next +// attempt takes the synchronous branch -- from inside a reconnect setTimeout, +// where the throw has no call site and escapes as an uncaught exception on +// every tick. +// +// Whether connect() should throw here or emit is the one open design choice; +// only 'throws synchronously ... with getToken set' below encodes it. Every +// other test in this block asserts an invariant that holds either way. +describe('static transport configuration errors', () => { + // No SockJS is available under Node, so the transport list is exhausted and + // _initializeTransport() throws 'no supported transport found'. + const unsupportedConfig = [ + { transport: 'sockjs' as TransportName, endpoint: 'http://localhost:8000/connection/sockjs' }, + ]; + + test('throws synchronously from connect() with no getToken (unchanged behavior)', () => { + const c = new Centrifuge(unsupportedConfig, {}); + expect(() => c.connect()).toThrow(/no supported transport found/); + safeDisconnect(c); + }); + + test('disconnect() after connect() gave up does not throw', () => { + // The selection loop assigns this._transport before testing supported(), so + // when it exhausts the list and throws 'no supported transport found' the + // client is left holding a wrapper whose inner transport was never created. + // _disconnect() then calls close() on it. This reaches the same null deref + // as issue #268 without any initialize() throw involved. + const c = new Centrifuge(unsupportedConfig, {}); + expect(() => c.connect()).toThrow(/no supported transport found/); + + expect(() => c.disconnect()).not.toThrow(); + expect(c.state).toBe(State.Disconnected); + }); + + test('throws synchronously from connect() for an http endpoint given as a string', () => { + const c = new Centrifuge('http://localhost:8000/connection/websocket', { + websocket: WebSocket, + }); + expect(() => c.connect()).toThrow(/explicit transport endpoints configuration/); + safeDisconnect(c); + }); + + test('throws synchronously from connect() with getToken set', () => { + // The design choice: a static config fault must surface the same way + // regardless of whether a token callback is configured. + const c = new Centrifuge(unsupportedConfig, { + getToken: () => Promise.resolve('token'), + minReconnectDelay: 20, + maxReconnectDelay: 20, + }); + c.on('error', () => { /* must not be the reporting channel for this class */ }); + + expect(() => c.connect()).toThrow(/no supported transport found/); + safeDisconnect(c); + }); + + test('never reported as a token error when getToken is set', async () => { + const errors: ErrorContext[] = []; + const c = new Centrifuge(unsupportedConfig, { + getToken: () => Promise.resolve('token'), + minReconnectDelay: 20, + maxReconnectDelay: 20, + }); + c.on('error', (ctx) => errors.push(ctx)); + + try { c.connect(); } catch { /* shape-dependent, asserted above */ } + await new Promise(r => setTimeout(r, 200)); + + expect(errors.map(e => e.type)).not.toContain('connectToken'); + safeDisconnect(c); + }); + + test('never reported as a data error when getData is set', async () => { + const errors: ErrorContext[] = []; + const c = new Centrifuge(unsupportedConfig, { + getData: () => Promise.resolve({}), + minReconnectDelay: 20, + maxReconnectDelay: 20, + }); + c.on('error', (ctx) => errors.push(ctx)); + + try { c.connect(); } catch { /* shape-dependent, asserted above */ } + await new Promise(r => setTimeout(r, 200)); + + expect(errors.map(e => e.type)).not.toContain('connectData'); + safeDisconnect(c); + }); + + test('never escapes as an uncaught exception from a reconnect timer', async () => { + // Second and later attempts run _startReconnecting() from a setTimeout. A + // throw there has no call site: it escapes the client entirely and repeats + // on every tick. jest fails this test if that happens. + const rejections = collectUnhandledRejections(); + const c = new Centrifuge(unsupportedConfig, { + getToken: () => Promise.resolve('token'), + minReconnectDelay: 20, + maxReconnectDelay: 20, + }); + c.on('error', () => { /* expected channel, if any */ }); + + try { c.connect(); } catch { /* shape-dependent, asserted above */ } + await new Promise(r => setTimeout(r, 400)); // ~20 reconnect ticks + + expect(rejections.seen).toEqual([]); + rejections.restore(); + safeDisconnect(c); + }); +}); + +describe('webtransport: initialize() is async', () => { + const originalWebTransport = (globalThis as any).WebTransport; + + afterEach(() => { + (globalThis as any).WebTransport = originalWebTransport; + }); + + test('a rejected initialize() falls over to websocket without an unhandled rejection', async () => { + // WebtransportTransport.initialize() is `async`, so a throw becomes a + // rejected promise on a return value nobody holds — a try/catch around the + // initialize() call catches nothing here. The failure must still be routed + // into the normal reconnect path. + const counter = { calls: 0 }; + const rejections = collectUnhandledRejections(); + + (globalThis as any).WebTransport = function () { + counter.calls++; + const e = new Error('WebTransport blocked'); + e.name = 'SecurityError'; + throw e; + }; + + const c = new Centrifuge([ + { transport: 'webtransport' as TransportName, endpoint: 'https://localhost:8000/connection/webtransport' }, + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + ], { + websocket: WebSocket, + fetch: fetch, + readableStream: ReadableStream, + emulationEndpoint: emulationEndpoint, + timeout: 3000, + minReconnectDelay: 50, + maxReconnectDelay: 200, + }); + c.on('error', () => { /* expected */ }); + + c.connect(); + // Deliberately below jest's default test timeout so a missing fallback + // reports as a failed assertion rather than a suite-level timeout. + await c.ready(2000); + + expect(c.state).toBe(State.Connected); + expect(counter.calls).toBeGreaterThanOrEqual(1); + expect(rejections.seen).toEqual([]); + + rejections.restore(); + safeDisconnect(c); + }); +}); + +// The fixtures above all throw from the constructor. This block covers the +// other half of the space: a transport that is constructed successfully and +// then misbehaves. It is the class the connect timeout belongs to, and it is +// what the original issue report described - a replaced global WebSocket that +// accepts close() and reports nothing back. +describe('transport constructed successfully but never usable', () => { + test('close() missing entirely is contained and the client keeps reconnecting', async () => { + const counter = { calls: 0, closes: 0 }; + const errors: ErrorContext[] = []; + + const c = track(new Centrifuge(wsEndpoint, { + websocket: stubWebSocket(counter, 'absent'), + timeout: 60, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await waitFor(() => counter.calls >= 3); + + expect(c.state).toBe(State.Connecting); + expect(errors.every(e => e.type === 'transport')).toBe(true); + expect(errors[0].error.message).toBe('connect timeout'); + }); + + test('a silent close() still lets the client recover', async () => { + // The guard on close() stops the TypeError, but on its own it would leave + // the client waiting forever for an onClose that this transport never + // sends. The connect timeout has to be the verdict. + const counter = { calls: 0, closes: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: stubWebSocket(counter, 'noop'), + timeout: 60, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* expected every attempt */ }); + + c.connect(); + await waitFor(() => counter.calls >= 3); + + expect(counter.closes).toBeGreaterThanOrEqual(1); + expect(c.state).toBe(State.Connecting); + }); + + test('a throwing close() does not stop reconnect scheduling', async () => { + // Guarded at two levels: the wrapper swallows it, and _disconnect schedules + // the reconnect from a finally so a throw could not skip it either way. + const counter = { calls: 0, closes: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: stubWebSocket(counter, 'throw'), + timeout: 60, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* expected every attempt */ }); + + c.connect(); + await waitFor(() => counter.calls >= 3); + + expect(counter.closes).toBeGreaterThanOrEqual(1); + expect(c.state).toBe(State.Connecting); + }); + + test('a hanging transport still falls through to the next one', async () => { + // The transport-index advance used to live only in onClose. A transport + // that hangs rather than closing must still yield to the next entry. + const counter = { calls: 0, closes: 0 }; + const c = track(new Centrifuge([ + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + { transport: 'http_stream' as TransportName, endpoint: httpStreamEndpoint }, + ], { + websocket: stubWebSocket(counter, 'noop'), + fetch: fetch, + readableStream: ReadableStream, + emulationEndpoint: emulationEndpoint, + timeout: 200, + minReconnectDelay: 20, + maxReconnectDelay: 50, + })); + c.on('error', () => { /* expected for the hanging websocket */ }); + + c.connect(); + await c.ready(4000); + + expect(c.state).toBe(State.Connected); + expect(counter.calls).toBeGreaterThanOrEqual(1); + }); + + test('disconnect() stops the attempt without a late error', async () => { + // disconnect() must also clear the connect timeout. If it did not, the + // timer would outlive the client, fire against a transport nobody uses, and + // hold the event loop open - which CI reports via --detectOpenHandles. + const counter = { calls: 0, closes: 0 }; + const errors: ErrorContext[] = []; + + const c = new Centrifuge(wsEndpoint, { + websocket: stubWebSocket(counter, 'noop'), + timeout: 60, + minReconnectDelay: 10000, + maxReconnectDelay: 10000, + }); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + c.disconnect(); + await new Promise(r => setTimeout(r, 200)); // well past the connect timeout + + expect(c.state).toBe(State.Disconnected); + expect(errors).toEqual([]); + }); +}); + +describe('remaining transports and selection edge cases', () => { + test('sockjs: an initialize throw falls over to websocket', async () => { + const counter = { calls: 0 }; + const errors: ErrorContext[] = []; + + const c = track(new Centrifuge([ + { transport: 'sockjs' as TransportName, endpoint: 'http://localhost:8000/connection/sockjs' }, + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + ], { + sockjs: throwingWebSocket(counter), + websocket: WebSocket, + timeout: 2000, + minReconnectDelay: 20, + maxReconnectDelay: 50, + })); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await c.ready(4000); + + expect(c.state).toBe(State.Connected); + expect(counter.calls).toBeGreaterThanOrEqual(1); + expect(errors.some(e => e.type === 'transport' && e.transport === 'sockjs')).toBe(true); + }); + + test('http_stream: a fetch that throws synchronously falls over to websocket', async () => { + const counter = { calls: 0 }; + + const c = track(new Centrifuge([ + { transport: 'http_stream' as TransportName, endpoint: httpStreamEndpoint }, + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + ], { + fetch: () => { counter.calls++; throw new Error('fetch blocked by policy'); }, + readableStream: ReadableStream, + websocket: WebSocket, + emulationEndpoint: emulationEndpoint, + timeout: 2000, + minReconnectDelay: 20, + maxReconnectDelay: 50, + })); + c.on('error', () => { /* expected for http_stream */ }); + + c.connect(); + await c.ready(4000); + + expect(c.state).toBe(State.Connected); + expect(counter.calls).toBeGreaterThanOrEqual(1); + }); + + test('an unsupported entry after a failing one does not crash the selection loop', async () => { + // The index is advanced by the failed websocket attempt, so the next pass + // starts at sockjs, finds it unsupported, and advances past the end of the + // list. The wrap has to happen inside the loop, not only on entry. + const saved = (globalThis as any).SockJS; + delete (globalThis as any).SockJS; + try { + const counter = { calls: 0 }; + const c = track(new Centrifuge([ + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + { transport: 'sockjs' as TransportName, endpoint: 'http://localhost:8000/connection/sockjs' }, + ], { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* expected every attempt */ }); + + c.connect(); + await waitFor(() => counter.calls >= 3); + + expect(c.state).toBe(State.Connecting); + } finally { + if (saved !== undefined) { (globalThis as any).SockJS = saved; } + } + }); + + test('recovers and connects once the transport stops failing', async () => { + // Proves the client resumes rather than merely looping. + let attempts = 0; + const flaky: any = function (this: any, url: string) { + attempts++; + if (attempts <= 2) { + const e = new Error(SECURITY_ERROR_MESSAGE); + e.name = 'SecurityError'; + throw e; + } + return new (WebSocket as any)(url); + }; + + const c = track(new Centrifuge(wsEndpoint, { + websocket: flaky, + timeout: 2000, + minReconnectDelay: 20, + maxReconnectDelay: 50, + })); + c.on('error', () => { /* expected for the first two attempts */ }); + + c.connect(); + await c.ready(4000); + + expect(c.state).toBe(State.Connected); + expect(attempts).toBeGreaterThanOrEqual(3); + }); +}); + +describe('reporting contract', () => { + test('getData path reports a transport error, not a data error', async () => { + const counter = { calls: 0 }; + const errors: ErrorContext[] = []; + + const c = track(new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 50, + maxReconnectDelay: 50, + getData: () => Promise.resolve({ some: 'data' }), + })); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await waitFor(() => errors.length > 0); + + expect(errors.map(e => e.type)).not.toContain('connectData'); + expect(errors[0].type).toBe('transport'); + + }); + + test('emulation transports report transport then connect', async () => { + // An emulation transport registers its connect command before initialize(). + // Clearing that command rejects a promise, so _connectError arrives on a + // microtask after the transport error rather than nested inside it - an + // assertion made synchronously would only ever see the first. + const counter = { calls: 0 }; + const errors: ErrorContext[] = []; + + const c = track(new Centrifuge([ + { transport: 'sse' as TransportName, endpoint: sseEndpoint }, + ], { + eventsource: throwingEventSource(counter), + fetch: fetch, + emulationEndpoint: emulationEndpoint, + timeout: 100, + minReconnectDelay: 10000, + maxReconnectDelay: 10000, + })); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await waitFor(() => errors.length >= 2); + + expect(errors.map(e => e.type)).toEqual(['transport', 'connect']); + }); + + test('does not re-emit connecting on every retry', async () => { + // The client is already Connecting, so a failed attempt must not restate it. + const counter = { calls: 0 }; + const connecting: any[] = []; + + const c = track(new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* expected */ }); + c.on('connecting', (ctx) => connecting.push(ctx)); + + c.connect(); + await waitFor(() => counter.calls >= 4); + + expect(connecting).toHaveLength(1); + }); + + test('disconnect() mid-loop stops the retries', async () => { + const counter = { calls: 0 }; + const c = new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 20, + maxReconnectDelay: 20, + }); + c.on('error', () => { /* expected */ }); + + c.connect(); + await waitFor(() => counter.calls >= 2); + c.disconnect(); + + const seen = counter.calls; + await new Promise(r => setTimeout(r, 200)); + + expect(c.state).toBe(State.Disconnected); + expect(counter.calls).toBe(seen); + }); +}); + +describe('nothing escapes a reconnect timer', () => { + // The umbrella invariant. Retries run from setTimeout, where a throw has no + // call site and escapes the client entirely. jest fails a test if that + // happens, so each scenario below simply has to survive several rounds. + + test('through an initialize-throw loop', async () => { + const rejections = collectUnhandledRejections(); + const counter = { calls: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: throwingWebSocket(counter), + timeout: 100, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* expected */ }); + + c.connect(); + await waitFor(() => counter.calls >= 5); + + expect(rejections.seen).toEqual([]); + rejections.restore(); + }); + + test('through a connect-timeout loop', async () => { + const rejections = collectUnhandledRejections(); + const counter = { calls: 0, closes: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: stubWebSocket(counter, 'absent'), + timeout: 40, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* expected */ }); + + c.connect(); + await waitFor(() => counter.calls >= 4); + + expect(rejections.seen).toEqual([]); + rejections.restore(); + }); + + test('when a dependency disappears under a running client', async () => { + // connect() validated that a transport was available; losing it afterwards + // is an environment change, so it must be reported rather than thrown from + // the timer that noticed it. + const rejections = collectUnhandledRejections(); + const errors: ErrorContext[] = []; + const saved = (globalThis as any).WebTransport; + (globalThis as any).WebTransport = function () { + throw new Error('webtransport blocked'); + }; + try { + const c = track(new Centrifuge([ + { transport: 'webtransport' as TransportName, endpoint: 'https://localhost:8000/connection/webtransport' }, + ], { + timeout: 100, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await waitFor(() => errors.length >= 1); + + // Now take it away entirely while the client is still cycling. + delete (globalThis as any).WebTransport; + await new Promise(r => setTimeout(r, 250)); + + expect(c.state).toBe(State.Connecting); + expect(rejections.seen).toEqual([]); + expect(errors.some(e => e.error.message === 'no supported transport found')).toBe(true); + } finally { + if (saved === undefined) { + delete (globalThis as any).WebTransport; + } else { + (globalThis as any).WebTransport = saved; + } + rejections.restore(); + } + }); +}); + +// Direct wrapper-level checks for the state that exists between constructing a +// transport and initializing it. The client no longer leaves a wrapper stranded +// in that state, but close() is reachable there from a stale reference, and the +// guards are cheap insurance rather than something a caller should have to +// reason about. +describe('transport wrappers before initialize()', () => { + test('WebsocketTransport.close() does not throw', () => { + const t = new WebsocketTransport(wsEndpoint, { websocket: WebSocket }); + expect(() => t.close()).not.toThrow(); + }); + + test('SockjsTransport.close() does not throw', () => { + const t = new SockjsTransport('http://localhost:8000/connection/sockjs', { sockjs: function () { /* fake */ } }); + expect(() => t.close()).not.toThrow(); + }); + + test('SockjsTransport.subName() does not throw', () => { + const t = new SockjsTransport('http://localhost:8000/connection/sockjs', { sockjs: function () { /* fake */ } }); + expect(() => t.subName()).not.toThrow(); + }); + + test('SseTransport.close() does not throw', () => { + const t = new SseTransport(sseEndpoint, { eventsource: function () { /* fake */ }, fetch: fetch }); + expect(() => t.close()).not.toThrow(); + }); + + test('HttpStreamTransport.close() does not throw', () => { + const t = new HttpStreamTransport(httpStreamEndpoint, { fetch: fetch, readableStream: ReadableStream }); + expect(() => t.close()).not.toThrow(); + }); +}); diff --git a/src/transport_sockjs.ts b/src/transport_sockjs.ts index e00b6ad..7c07ebc 100644 --- a/src/transport_sockjs.ts +++ b/src/transport_sockjs.ts @@ -15,7 +15,9 @@ export class SockjsTransport { } subName() { - return 'sockjs-' + this._transport.transport; + // Called from debug logging, which must not throw on a transport that was + // constructed but never initialized. + return 'sockjs-' + this._transport?.transport; } emulation() { @@ -47,7 +49,11 @@ export class SockjsTransport { } close() { - this._transport.close(); + try { + this._transport?.close(); + } catch (e) { + // already closed, or not closeable. + } } send(data: any) { diff --git a/src/transport_sse.ts b/src/transport_sse.ts index a84e729..24ee607 100644 --- a/src/transport_sse.ts +++ b/src/transport_sse.ts @@ -72,7 +72,14 @@ export class SseTransport { } close() { - this._transport.close(); + try { + this._transport?.close(); + } catch (e) { + // already closed, or not closeable. + } + // Deliberately outside the guard above: EventSource has no close event, so + // this synthesizes one, and errors raised downstream of it must not be + // mistaken for a transport-close failure. if (this._onClose !== null) { this._onClose(); } diff --git a/src/transport_websocket.ts b/src/transport_websocket.ts index 38ed4ea..ed73cab 100644 --- a/src/transport_websocket.ts +++ b/src/transport_websocket.ts @@ -58,7 +58,13 @@ export class WebsocketTransport { } close() { - this._transport.close(); + // _transport stays null when the constructor above threw, and a replaced + // global WebSocket may not provide close() at all. + try { + this._transport?.close(); + } catch (e) { + // already closed, or not closeable. + } } send(data: any) { From 01b963036ca9a3214ae8252245de489e1dad081f Mon Sep 17 00:00:00 2001 From: FZambia Date: Wed, 5 Aug 2026 08:34:49 +0300 Subject: [PATCH 3/6] Deliver SSE transport close asynchronously EventSource fires nothing when closed, so SseTransport has to synthesize the close notification itself. It delivered that inline from close(), which made SSE the only transport re-entering the client from inside its own close(): _disconnect -> close() -> onClose -> _disconnect again, on one stack. Every other transport reports asynchronously, via onclose, an aborted fetch, or the WebTransport closed promise. That reentrancy is why _disconnect has to null this._transport before closing it. Defer the synthesized close and consume it one-shot. The deferred callback now arrives after _disconnect has advanced the transport id, so it is ignored on the id guard - nothing is lost, because every path that closes a transport already resets the closed state itself. The synthesis is kept rather than removed so this change stays independent of the connect timeout handling it now overlaps with. --- src/transport_init_error.test.ts | 74 ++++++++++++++++++++++++++++++++ src/transport_sse.ts | 28 ++++++++---- 2 files changed, 93 insertions(+), 9 deletions(-) diff --git a/src/transport_init_error.test.ts b/src/transport_init_error.test.ts index a814639..9df802e 100644 --- a/src/transport_init_error.test.ts +++ b/src/transport_init_error.test.ts @@ -975,3 +975,77 @@ describe('transport wrappers before initialize()', () => { expect(() => t.close()).not.toThrow(); }); }); + +describe('SseTransport close is not reentrant', () => { + test('close() returns before onClose is delivered', async () => { + // Every other transport reports its close asynchronously. Delivering it + // inline made SSE the one transport that re-entered the client from inside + // its own close(), which is why _disconnect has to null this._transport + // before closing it. + const events: string[] = []; + const fakeEventSource: any = function () { + return { close: () => { events.push('inner close'); } }; + }; + + const t = new SseTransport(sseEndpoint, { eventsource: fakeEventSource, fetch: fetch }); + t.initialize('json', { + onOpen: () => { /* unused */ }, + onError: () => { /* unused */ }, + onClose: () => { events.push('onClose'); }, + onMessage: () => { /* unused */ }, + }, '{}'); + + t.close(); + events.push('close returned'); + + expect(events).toEqual(['inner close', 'close returned']); + + await new Promise(r => setTimeout(r, 10)); + expect(events).toEqual(['inner close', 'close returned', 'onClose']); + }); + + test('repeated close() synthesizes only one onClose', async () => { + let closes = 0; + const fakeEventSource: any = function () { + return { close: () => { /* no-op */ } }; + }; + + const t = new SseTransport(sseEndpoint, { eventsource: fakeEventSource, fetch: fetch }); + t.initialize('json', { + onOpen: () => { /* unused */ }, + onError: () => { /* unused */ }, + onClose: () => { closes++; }, + onMessage: () => { /* unused */ }, + }, '{}'); + + t.close(); + t.close(); + t.close(); + await new Promise(r => setTimeout(r, 10)); + + expect(closes).toBe(1); + }); + + test('an sse client still reconnects after its transport fails', async () => { + // The deferred close lands after _disconnect has advanced the transport id, + // so it is ignored on the id guard. The client must not depend on it: the + // connect timeout and initialize failures reset the state themselves. + const counter = { calls: 0 }; + const c = track(new Centrifuge([ + { transport: 'sse' as TransportName, endpoint: sseEndpoint }, + ], { + eventsource: throwingEventSource(counter), + fetch: fetch, + emulationEndpoint: emulationEndpoint, + timeout: 100, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* expected every attempt */ }); + + c.connect(); + await waitFor(() => counter.calls >= 3); + + expect(c.state).toBe(State.Connecting); + }); +}); diff --git a/src/transport_sse.ts b/src/transport_sse.ts index 24ee607..5b71d93 100644 --- a/src/transport_sse.ts +++ b/src/transport_sse.ts @@ -63,11 +63,18 @@ export class SseTransport { callbacks.onMessage(e.data); }; + // EventSource fires nothing when closed, so close() has to synthesize the + // notification. Deferred, because delivering it inline would make this the + // only transport that re-enters the client from its own close() - every + // other one reports asynchronously, via onclose, an aborted fetch or the + // WebTransport closed promise. self._onClose = function () { - callbacks.onClose({ - code: 4, - reason: 'connection closed' - }); + setTimeout(function () { + callbacks.onClose({ + code: 4, + reason: 'connection closed' + }); + }, 0); }; } @@ -77,11 +84,14 @@ export class SseTransport { } catch (e) { // already closed, or not closeable. } - // Deliberately outside the guard above: EventSource has no close event, so - // this synthesizes one, and errors raised downstream of it must not be - // mistaken for a transport-close failure. - if (this._onClose !== null) { - this._onClose(); + // Deliberately outside the guard above: errors raised downstream of the + // synthesized close must not be mistaken for a transport-close failure. + // Consumed one-shot, so repeated close() calls cannot synthesize repeated + // closes. + const onClose = this._onClose; + this._onClose = null; + if (onClose !== null) { + onClose(); } } From 43f03ac3ccb9b32b46ffb6a47079702567518a1d Mon Sep 17 00:00:00 2001 From: FZambia Date: Wed, 5 Aug 2026 08:38:40 +0300 Subject: [PATCH 4/6] Cover remaining transport guards and dependency sources Adds the WebtransportTransport close-before-initialize guard, and resolves eventsource, fetch and readableStream from globalThis as well as from config, so every transport is exercised through both resolution paths. --- src/transport_init_error.test.ts | 8 ++++++++ src/transport_selection.test.ts | 35 ++++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/src/transport_init_error.test.ts b/src/transport_init_error.test.ts index 9df802e..6e5830d 100644 --- a/src/transport_init_error.test.ts +++ b/src/transport_init_error.test.ts @@ -9,6 +9,7 @@ import { WebsocketTransport } from './transport_websocket'; import { SockjsTransport } from './transport_sockjs'; import { SseTransport } from './transport_sse'; import { HttpStreamTransport } from './transport_http_stream'; +import { WebtransportTransport } from './transport_webtransport'; // Regression tests for https://github.com/centrifugal/centrifuge-js/issues/268 // @@ -970,6 +971,13 @@ describe('transport wrappers before initialize()', () => { expect(() => t.close()).not.toThrow(); }); + test('WebtransportTransport.close() does not throw', () => { + const t = new WebtransportTransport('https://localhost:8000/connection/webtransport', { + webtransport: function () { /* fake */ }, + }); + expect(() => t.close()).not.toThrow(); + }); + test('HttpStreamTransport.close() does not throw', () => { const t = new HttpStreamTransport(httpStreamEndpoint, { fetch: fetch, readableStream: ReadableStream }); expect(() => t.close()).not.toThrow(); diff --git a/src/transport_selection.test.ts b/src/transport_selection.test.ts index 6a70a65..4fda274 100644 --- a/src/transport_selection.test.ts +++ b/src/transport_selection.test.ts @@ -350,3 +350,38 @@ describe('non-emulation string endpoint', () => { expect(selectedTransportName(c)).toBe('websocket'); }); }); + +describe('dependency source equivalence: emulation transports from globalThis', () => { + test('sse resolves eventsource from globalThis', () => { + const saved = (globalThis as any).EventSource; + (globalThis as any).EventSource = EventSource; + try { + const c = makeClient([{ transport: 'sse' as TransportName, endpoint: sseEndpoint }], { + fetch: fetch, + }); + c.connect(); + expect(selectedTransportName(c)).toBe('sse'); + } finally { + if (saved === undefined) { + delete (globalThis as any).EventSource; + } else { + (globalThis as any).EventSource = saved; + } + } + }); + + test('http_stream resolves fetch and readableStream from globalThis', () => { + const savedFetch = (globalThis as any).fetch; + const savedStream = (globalThis as any).ReadableStream; + (globalThis as any).fetch = fetch; + (globalThis as any).ReadableStream = ReadableStream; + try { + const c = makeClient([{ transport: 'http_stream' as TransportName, endpoint: httpStreamEndpoint }], {}); + c.connect(); + expect(selectedTransportName(c)).toBe('http_stream'); + } finally { + (globalThis as any).fetch = savedFetch; + (globalThis as any).ReadableStream = savedStream; + } + }); +}); From 0c185f53bccf0df72791666ddc7fef7ba10901d3 Mon Sep 17 00:00:00 2001 From: FZambia Date: Wed, 5 Aug 2026 08:47:34 +0300 Subject: [PATCH 5/6] Correct a stale comment about the connect timeout --- src/transport_selection.test.ts | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/transport_selection.test.ts b/src/transport_selection.test.ts index 4fda274..f0c635f 100644 --- a/src/transport_selection.test.ts +++ b/src/transport_selection.test.ts @@ -87,12 +87,13 @@ afterEach(async () => { // transport_init_error.test.ts, and must not mask assertions here. try { c!.disconnect(); } catch { /* pinned elsewhere */ } } - // disconnect() does not clear the connect timeout: it is a closure local of - // _initializeTransport, cleared only from onOpen/onClose. A transport that - // never opens therefore leaves an armed timer behind, which CI reports as an - // open handle since it runs jest --detectOpenHandles with no --forceExit. - // Waiting it out keeps this file honest against unmodified source; the fix - // moves the timeout onto the instance so _disconnect can clear it. + // disconnect() clears the connect timeout, so this wait is not needed for the + // current client. It is kept because this file has to pass against the source + // as it was before that was true - it is the baseline the transport selection + // rewrite had to preserve, and it must stay meaningful if that rewrite is + // reverted. Without it, a transport that never opens leaves an armed timer, + // which CI reports as an open handle: it runs jest --detectOpenHandles with + // no --forceExit. await new Promise(resolve => setTimeout(resolve, CONNECT_TIMEOUT + 50)); }); From 6558a9f87303d4a17319ee7e7a8c2183c317d946 Mon Sep 17 00:00:00 2001 From: FZambia Date: Wed, 5 Aug 2026 10:17:04 +0300 Subject: [PATCH 6/6] Add regression tests for recovering from a transport failure Recovering has to leave a connection that actually works, not merely one that reports connected. These were written while validating the transport initialize fixes on this branch, and cover ground the existing suite did not: subscriptions across a transport failure and after a dropped socket, publish, history and presence once recovered, map subscriptions, falling back to an emulation transport and then genuinely using it, rotating through a three transport list, token callbacks, network offline/online, disconnecting from inside an error listener, and timer accounting after repeated failures. --- src/centrifuge.ts | 16 +- src/transport_recovery.test.ts | 424 +++++++++++++++++++++++++++++++++ 2 files changed, 432 insertions(+), 8 deletions(-) create mode 100644 src/transport_recovery.test.ts diff --git a/src/centrifuge.ts b/src/centrifuge.ts index 588b85e..e64d2d7 100644 --- a/src/centrifuge.ts +++ b/src/centrifuge.ts @@ -1589,14 +1589,6 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter TypedEventEmitter { + while (clients.length) { + try { clients.pop()!.disconnect(); } catch { /* teardown must not mask assertions */ } + } +}); + +function waitFor(predicate: () => boolean, timeout = 5000): Promise { + return new Promise((resolve, reject) => { + const started = Date.now(); + const tick = () => { + if (predicate()) return resolve(); + if (Date.now() - started > timeout) return reject(new Error('timeout waiting for condition')); + setTimeout(tick, 10); + }; + tick(); + }); +} + +/** Throws for the first `failFor` attempts, then hands back a real socket. */ +function flakyWebSocket(state: { attempts: number }, failFor: number) { + return function (this: any, url: string) { + state.attempts++; + if (state.attempts <= failFor) { + throw new Error('blocked attempt ' + state.attempts); + } + return new (WebSocket as any)(url); + } as any; +} + +function alwaysThrowingWebSocket(state: { attempts: number }) { + return function () { + state.attempts++; + throw new Error('blocked'); + } as any; +} + +describe('subscriptions across a transport failure', () => { + test('a subscription made before connect still subscribes once the transport works', async () => { + const st = { attempts: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: flakyWebSocket(st, 3), + minReconnectDelay: 20, + maxReconnectDelay: 40, + })); + c.on('error', () => { /* expected for the failing attempts */ }); + + const sub = c.newSubscription('test'); + const subscribed: any[] = []; + sub.on('subscribed', (ctx) => subscribed.push(ctx)); + sub.on('error', () => { /* not expected, but must not throw as unhandled */ }); + sub.subscribe(); + + c.connect(); + await waitFor(() => subscribed.length > 0, 8000); + + expect(c.state).toBe(State.Connected); + expect(sub.state).toBe(SubscriptionState.Subscribed); + expect(st.attempts).toBeGreaterThanOrEqual(4); + }, 15000); + + test('a subscription resubscribes after the socket is dropped mid-session', async () => { + const c = track(new Centrifuge(wsEndpoint, { + websocket: WebSocket, + minReconnectDelay: 20, + maxReconnectDelay: 40, + })); + c.on('error', () => { /* ignore */ }); + + const sub = c.newSubscription('test'); + let subscribedCount = 0; + sub.on('subscribed', () => { subscribedCount++; }); + sub.on('error', () => { /* ignore */ }); + sub.subscribe(); + + c.connect(); + await waitFor(() => subscribedCount === 1, 8000); + + // Kill the underlying socket without telling the client. + (c as any)._transport._transport.close(); + + await waitFor(() => subscribedCount === 2, 8000); + expect(c.state).toBe(State.Connected); + expect(sub.state).toBe(SubscriptionState.Subscribed); + }, 15000); + + test('publish, receive and history all work after recovering', async () => { + const st = { attempts: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: flakyWebSocket(st, 2), + minReconnectDelay: 20, + maxReconnectDelay: 40, + })); + c.on('error', () => { /* expected */ }); + + const sub = c.newSubscription('test'); + const received: any[] = []; + sub.on('publication', (ctx) => received.push(ctx.data)); + sub.on('error', () => { /* ignore */ }); + sub.subscribe(); + + c.connect(); + await c.ready(8000); + await waitFor(() => sub.state === SubscriptionState.Subscribed, 8000); + + await sub.publish({ hello: 'after recovery' }); + await waitFor(() => received.length > 0, 5000); + expect(received[0]).toEqual({ hello: 'after recovery' }); + + const history = await sub.history({ limit: 10 }); + expect(history.publications.length).toBeGreaterThanOrEqual(1); + + const stats = await sub.presenceStats(); + expect(stats.numClients).toBeGreaterThanOrEqual(1); + }, 20000); + + test('a map subscription works after transport failures', async () => { + const st = { attempts: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: flakyWebSocket(st, 3), + minReconnectDelay: 20, + maxReconnectDelay: 40, + })); + c.on('error', () => { /* expected */ }); + + const sub = c.newMapSubscription('streamless:recovery'); + sub.on('error', () => { /* ignore */ }); + sub.subscribe(); + + c.connect(); + await c.ready(8000); + await waitFor(() => sub.state === SubscriptionState.Subscribed, 8000); + + const entries: any[] = []; + sub.on('publication', (ctx: any) => entries.push(ctx)); + await sub.publish('k1', { v: 1 }); + await waitFor(() => entries.length > 0, 5000); + + expect(st.attempts).toBeGreaterThanOrEqual(4); + }, 25000); +}); + +describe('falling back to an emulation transport yields a usable connection', () => { + test.each([ + ['http_stream', httpStreamEndpoint], + ['sse', sseEndpoint], + ])('%s', async (transport, endpoint) => { + const st = { attempts: 0 }; + const c = track(new Centrifuge([ + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + { transport: transport as TransportName, endpoint: endpoint }, + ], { + websocket: alwaysThrowingWebSocket(st), + eventsource: EventSource, + fetch: fetch, + readableStream: ReadableStream, + emulationEndpoint: emulationEndpoint, + minReconnectDelay: 20, + maxReconnectDelay: 50, + })); + c.on('error', () => { /* expected for the blocked websocket */ }); + + const sub = c.newSubscription('test'); + const received: any[] = []; + sub.on('publication', (ctx) => received.push(ctx.data)); + sub.on('error', () => { /* ignore */ }); + sub.subscribe(); + + c.connect(); + await c.ready(8000); + await waitFor(() => sub.state === SubscriptionState.Subscribed, 8000); + + await sub.publish({ via: transport }); + await waitFor(() => received.length > 0, 5000); + + expect(received[0]).toEqual({ via: transport }); + expect((c as any)._transport.name()).toBe(transport); + expect(st.attempts).toBeGreaterThanOrEqual(1); + }, 20000); + + test('rotates through a three transport list until one works', async () => { + const st = { attempts: 0 }; + const c = track(new Centrifuge([ + { transport: 'websocket' as TransportName, endpoint: wsEndpoint }, + { transport: 'sse' as TransportName, endpoint: sseEndpoint }, + { transport: 'http_stream' as TransportName, endpoint: httpStreamEndpoint }, + ], { + websocket: alwaysThrowingWebSocket(st), + eventsource: function () { throw new Error('sse blocked'); } as any, + fetch: fetch, + readableStream: ReadableStream, + emulationEndpoint: emulationEndpoint, + minReconnectDelay: 20, + maxReconnectDelay: 50, + })); + c.on('error', () => { /* expected for the first two */ }); + + c.connect(); + await c.ready(8000); + + expect(c.state).toBe(State.Connected); + expect((c as any)._transport.name()).toBe('http_stream'); + }, 20000); +}); + +describe('token callbacks are unaffected by transport error reporting', () => { + test('UnauthorizedError from getToken still terminates the client', async () => { + const disconnects: any[] = []; + const c = track(new Centrifuge(wsEndpoint, { + websocket: WebSocket, + minReconnectDelay: 20, + maxReconnectDelay: 40, + getToken: () => Promise.reject(new UnauthorizedError('nope')), + })); + c.on('error', () => { /* ignore */ }); + c.on('disconnected', (ctx) => disconnects.push(ctx)); + + c.connect(); + await waitFor(() => disconnects.length > 0, 5000); + + expect(c.state).toBe(State.Disconnected); + expect(disconnects[0].code).toBe(disconnectedCodes.unauthorized); + }, 15000); + + test('a failing getToken still reports connectToken and keeps retrying', async () => { + // Transport faults were moved off this reporting path; genuine token faults + // must still arrive on it. + const errors: any[] = []; + let calls = 0; + const c = track(new Centrifuge(wsEndpoint, { + websocket: WebSocket, + minReconnectDelay: 20, + maxReconnectDelay: 40, + getToken: () => { calls++; return Promise.reject(new Error('token backend down')); }, + })); + c.on('error', (ctx) => errors.push(ctx)); + + c.connect(); + await waitFor(() => calls >= 3, 5000); + + expect(errors.some(e => e.type === 'connectToken')).toBe(true); + expect(c.state).toBe(State.Connecting); + }, 15000); + + test('disconnect during an in-flight getToken leaves no transport behind', async () => { + let release: (v: string) => void = () => { /* assigned below */ }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: WebSocket, + minReconnectDelay: 20, + maxReconnectDelay: 40, + getToken: () => new Promise((res) => { release = res; }), + })); + c.on('error', () => { /* ignore */ }); + + c.connect(); + expect(c.state).toBe(State.Connecting); + c.disconnect(); + expect(c.state).toBe(State.Disconnected); + + release('a-token'); // token arrives after disconnect + await new Promise(r => setTimeout(r, 300)); + + expect(c.state).toBe(State.Disconnected); + expect((c as any)._transport).toBeNull(); + }, 15000); +}); + +describe('network events', () => { + test('offline then online reconnects', async () => { + const net = new EventTarget(); + const c = track(new Centrifuge(wsEndpoint, { + websocket: WebSocket, + networkEventTarget: net as any, + minReconnectDelay: 20, + maxReconnectDelay: 40, + })); + c.on('error', () => { /* ignore */ }); + + c.connect(); + await c.ready(6000); + expect(c.state).toBe(State.Connected); + + net.dispatchEvent(new Event('offline')); + expect(c.state).toBe(State.Connecting); + + net.dispatchEvent(new Event('online')); + await waitFor(() => c.state === State.Connected, 6000); + }, 20000); + + test('offline arriving during a failing attempt still recovers', async () => { + const net = new EventTarget(); + const st = { attempts: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: flakyWebSocket(st, 2), + networkEventTarget: net as any, + minReconnectDelay: 30, + maxReconnectDelay: 60, + })); + c.on('error', () => { /* expected */ }); + + c.connect(); + await waitFor(() => st.attempts >= 1, 4000); + net.dispatchEvent(new Event('offline')); + net.dispatchEvent(new Event('online')); + + await waitFor(() => c.state === State.Connected, 8000); + }, 20000); +}); + +// The SDK deliberately does no error handling for application callbacks - see +// "Errors in callbacks" in the README - so nothing here relies on a listener +// being allowed to throw. +describe('reacting to errors from a listener', () => { + test('disconnect() called from an error listener stops cleanly', async () => { + // The documented way to opt out of retrying an unrecoverable fault, so it + // has to survive being called re-entrantly from inside the error emit. + const st = { attempts: 0 }; + const c = track(new Centrifuge(wsEndpoint, { + websocket: alwaysThrowingWebSocket(st), + timeout: 100, + minReconnectDelay: 20, + maxReconnectDelay: 20, + })); + c.on('error', (ctx) => { + if (ctx.type === 'transport') { c.disconnect(); } + }); + + expect(() => c.connect()).not.toThrow(); + await waitFor(() => c.state === State.Disconnected, 4000); + + const seen = st.attempts; + await new Promise(r => setTimeout(r, 200)); + + expect(c.state).toBe(State.Disconnected); + expect(st.attempts).toBe(seen); + expect((c as any)._transport).toBeNull(); + }, 15000); +}); + +describe('client lifecycle', () => { + test('rapid connect and disconnect cycles leave a working client', async () => { + const c = track(new Centrifuge(wsEndpoint, { + websocket: WebSocket, + minReconnectDelay: 10, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* ignore */ }); + + for (let i = 0; i < 10; i++) { + c.connect(); + c.disconnect(); + } + + c.connect(); + await c.ready(6000); + expect(c.state).toBe(State.Connected); + }, 15000); + + test('connect() called synchronously after disconnect() reconnects', async () => { + const c = track(new Centrifuge(wsEndpoint, { + websocket: WebSocket, + minReconnectDelay: 10, + maxReconnectDelay: 20, + })); + c.on('error', () => { /* ignore */ }); + + c.connect(); + await c.ready(6000); + + c.disconnect(); + c.connect(); + await c.ready(6000); + + expect(c.state).toBe(State.Connected); + }, 15000); + + test('repeated failures leave no timers behind after disconnect', async () => { + const st = { attempts: 0 }; + const c = new Centrifuge(wsEndpoint, { + websocket: alwaysThrowingWebSocket(st), + timeout: 50, + minReconnectDelay: 10, + maxReconnectDelay: 10, + }); + c.on('error', () => { /* expected */ }); + + c.connect(); + await waitFor(() => st.attempts >= 15, 8000); + + c.disconnect(); + expect((c as any)._connectTimeout).toBeNull(); + expect((c as any)._reconnectTimeout).toBeNull(); + expect((c as any)._transport).toBeNull(); + + const seen = st.attempts; + await new Promise(r => setTimeout(r, 200)); + expect(st.attempts).toBe(seen); + }, 20000); +});