diff --git a/src/centrifuge.ts b/src/centrifuge.ts index 02469f1..e64d2d7 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; @@ -1498,6 +1589,15 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter 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('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(); + }); +}); + +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_recovery.test.ts b/src/transport_recovery.test.ts new file mode 100644 index 0000000..d94decf --- /dev/null +++ b/src/transport_recovery.test.ts @@ -0,0 +1,424 @@ +import { Centrifuge, UnauthorizedError } from './centrifuge'; +import { State, SubscriptionState, TransportName } from './types'; +import { disconnectedCodes } from './codes'; + +import WebSocket from 'ws'; +import EventSource from 'eventsource'; +import { fetch } from 'undici'; +import { ReadableStream } from 'node:stream/web'; + +// Recovering from a transport failure has to leave a connection that actually +// works, not merely one that reports Connected. These exercise the paths a real +// application takes across a failure - subscriptions, publishes, history, +// network events, token callbacks - against a live server. + +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'; + +const clients: Centrifuge[] = []; + +function track(c: Centrifuge): Centrifuge { + clients.push(c); + return c; +} + +afterEach(() => { + 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); +}); diff --git a/src/transport_selection.test.ts b/src/transport_selection.test.ts new file mode 100644 index 0000000..f0c635f --- /dev/null +++ b/src/transport_selection.test.ts @@ -0,0 +1,388 @@ +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() 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)); +}); + +/** 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'); + }); +}); + +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; + } + }); +}); 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..5b71d93 100644 --- a/src/transport_sse.ts +++ b/src/transport_sse.ts @@ -63,18 +63,35 @@ 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); }; } close() { - this._transport.close(); - if (this._onClose !== null) { - this._onClose(); + try { + this._transport?.close(); + } catch (e) { + // already closed, or not closeable. + } + // 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(); } } 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) {