From 4f03dc45057c5fded1465e52d376d30eab68e94a Mon Sep 17 00:00:00 2001 From: Rohit <40559587+Rohit3523@users.noreply.github.com> Date: Tue, 14 Jul 2026 03:51:08 +0530 Subject: [PATCH 1/5] feat: add connection liveness check + forceReopen to ddp-client --- .changeset/fix-ddp-connection-liveness.md | 10 +++ packages/ddp-client/src/Connection.ts | 88 ++++++++++++++++++++++- packages/ddp-client/src/DDPSDK.ts | 8 +++ 3 files changed, 105 insertions(+), 1 deletion(-) create mode 100644 .changeset/fix-ddp-connection-liveness.md diff --git a/.changeset/fix-ddp-connection-liveness.md b/.changeset/fix-ddp-connection-liveness.md new file mode 100644 index 0000000000000..2e88ae8178b64 --- /dev/null +++ b/.changeset/fix-ddp-connection-liveness.md @@ -0,0 +1,10 @@ +--- +'@rocket.chat/ddp-client': patch +--- + +Adds connection liveness checking to recover from zombie WebSocket sockets (a `connected` socket whose transport is dead, common on mobile NAT timeouts that never fire `onclose`): + +- `Connection.probe(timeoutMs?)` — sends a DDP `ping` and resolves `true` if a `pong` arrives in time. +- `Connection.forceReopen()` — tears down and reconnects, deduplicating concurrent callers via a shared in-flight promise. +- `Connection.checkAndReopen(probeTimeoutMs?)` — probes when connected, force-reopens when not connected or on a dead probe. Also exposed on `DDPSDK`. +- `Connection.close()` now severs socket handlers before closing so a dying socket can't deliver late messages or clobber the live connection. diff --git a/packages/ddp-client/src/Connection.ts b/packages/ddp-client/src/Connection.ts index 2498fc1813313..246cac26fdea2 100644 --- a/packages/ddp-client/src/Connection.ts +++ b/packages/ddp-client/src/Connection.ts @@ -45,6 +45,26 @@ export interface Connection reconnect(): Promise; close(): void; + + /** + * Active liveness check: sends a DDP `ping` and resolves `true` if a `pong` + * arrives within `timeoutMs`. Detects zombie sockets that stay `connected` + * while their transport is actually dead (e.g. mobile NAT timeouts that + * never fire `onclose`). + */ + probe(timeoutMs?: number): Promise; + + /** + * Tear down the current socket and reconnect from scratch. Concurrent + * callers share one in-flight promise so parallel calls don't race. + */ + forceReopen(): Promise; + + /** + * Ensure the socket is usable: if not `connected`, force-reopens; if + * `connected`, probes for zombie sockets and force-reopens on a dead probe. + */ + checkAndReopen(probeTimeoutMs?: number): Promise; } interface WebSocketConstructor { @@ -78,6 +98,8 @@ export class ConnectionImpl public queue = new Set(); + private reopenInFlight: Promise | null = null; + constructor( url: string, private WS: WebSocketConstructor, @@ -270,10 +292,74 @@ export class ConnectionImpl close() { this.status = 'closed'; - this.ws?.close(); + if (this.ws) { + // Sever all handlers before closing so the dying socket cannot deliver + // late messages or fire `onclose` on the live connection (e.g. a replaced + // socket whose transport is still limping along). + try { + this.ws.onopen = null; + this.ws.onmessage = null; + this.ws.onerror = null; + this.ws.onclose = null; + } catch { + // some WebSocket implementations throw on null assignment — safe to ignore + } + this.ws.close(); + } this.emitStatus(); } + probe(timeoutMs = 2000): Promise { + return new Promise((resolve) => { + let settled = false; + let off: () => void = () => {}; + let timer: ReturnType; + + const finish = (alive: boolean) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timer); + off(); + resolve(alive); + }; + + timer = setTimeout(() => finish(false), timeoutMs); + off = this.client.onMessage((payload) => { + if (payload.msg === 'pong') { + finish(true); + } + }); + this.client.ping(); + }); + } + + forceReopen(): Promise { + if (this.reopenInFlight) { + return this.reopenInFlight; + } + this.close(); + const promise = this.reconnect() + .then((connected) => Boolean(connected)) + .catch(() => false); + this.reopenInFlight = promise; + void promise.finally(() => { + if (this.reopenInFlight === promise) { + this.reopenInFlight = null; + } + }); + return promise; + } + + async checkAndReopen(probeTimeoutMs = 2000): Promise { + if (this.status !== 'connected') { + return this.forceReopen(); + } + const alive = await this.probe(probeTimeoutMs); + return alive || this.forceReopen(); + } + static create( url: string, webSocketImpl: WebSocketConstructor, diff --git a/packages/ddp-client/src/DDPSDK.ts b/packages/ddp-client/src/DDPSDK.ts index 08e7b65116bcb..818566d52fa4f 100644 --- a/packages/ddp-client/src/DDPSDK.ts +++ b/packages/ddp-client/src/DDPSDK.ts @@ -40,6 +40,14 @@ export class DDPSDK implements SDK { return this.client.callAsync(method, ...params); } + /** + * Probe the connection for liveness and force-reopen if the socket is dead + * or zombie. See `Connection.checkAndReopen`. + */ + checkAndReopen(probeTimeoutMs?: number) { + return this.connection.checkAndReopen(probeTimeoutMs); + } + stream(name: string, data: unknown, cb: (...data: PublicationPayloads['fields']['args']) => void) { const [key, args] = Array.isArray(data) ? data : [data]; const subscription = this.client.subscribe(`stream-${name}`, key, { useCollection: false, args: [args] }); From f90d4b081dedc056b34b846127394943ff98dd53 Mon Sep 17 00:00:00 2001 From: Rohit <40559587+Rohit3523@users.noreply.github.com> Date: Tue, 14 Jul 2026 04:13:02 +0530 Subject: [PATCH 2/5] fix(ddp-client): resolve no-empty-function lint error in Connection probe --- packages/ddp-client/src/Connection.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/ddp-client/src/Connection.ts b/packages/ddp-client/src/Connection.ts index 246cac26fdea2..8637cb439eafe 100644 --- a/packages/ddp-client/src/Connection.ts +++ b/packages/ddp-client/src/Connection.ts @@ -1,6 +1,7 @@ import { Emitter } from '@rocket.chat/emitter'; import type { DDPClient } from './types/DDPClient'; +import type { RemoveListener } from './types/RemoveListener'; // type Subscription = { // name: string; @@ -312,7 +313,7 @@ export class ConnectionImpl probe(timeoutMs = 2000): Promise { return new Promise((resolve) => { let settled = false; - let off: () => void = () => {}; + let off: RemoveListener | undefined; let timer: ReturnType; const finish = (alive: boolean) => { @@ -321,7 +322,7 @@ export class ConnectionImpl } settled = true; clearTimeout(timer); - off(); + off?.(); resolve(alive); }; From 9a1eb8d7acd8a7e43101754e54ca4d385946cbd1 Mon Sep 17 00:00:00 2001 From: Rohit <40559587+Rohit3523@users.noreply.github.com> Date: Tue, 14 Jul 2026 17:27:38 +0530 Subject: [PATCH 3/5] test(ddp-client): cover probe/forceReopen/checkAndReopen liveness methods --- .../ddp-client/__tests__/Connection.spec.ts | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) diff --git a/packages/ddp-client/__tests__/Connection.spec.ts b/packages/ddp-client/__tests__/Connection.spec.ts index 99c77c7a8e6af..ade6a60e90231 100644 --- a/packages/ddp-client/__tests__/Connection.spec.ts +++ b/packages/ddp-client/__tests__/Connection.spec.ts @@ -334,3 +334,126 @@ it('should ignore a stale ws.onclose that fires after the socket has been replac expect(connection.status).toBe(statusBefore); expect((connection as unknown as { retryCount: number }).retryCount).toBe(retryBefore); }); + +// The existing tests above share the `ws://localhost:1234` URL. jest-websocket-mock +// is backed by `mock-socket`, whose server registry is keyed by URL; leftover closed +// servers accumulate across tests and can capture a later `new WebSocket(url)`. Run the +// liveness tests on a separate port so they never bind to a stale server. +const setupLiveness = (retryOptions: { retryCount: number; retryTime: number } = { retryCount: 0, retryTime: 0 }) => { + server = new WS('ws://localhost:4321/websocket'); + const client = new MinimalDDPClient(); + const connection = ConnectionImpl.create('ws://localhost:4321', WebSocket, client, retryOptions); + return { client, connection }; +}; + +it('probe resolves true when the server answers the ping with a pong', async () => { + const { connection } = setupLiveness(); + + await handleConnection(server, connection.connect()); + expect(connection.status).toBe('connected'); + + const probePromise = connection.probe(); + + await server.nextMessage.then((message) => { + expect(message).toBe('{"msg":"ping"}'); + server.send('{"msg":"pong"}'); + }); + + await expect(probePromise).resolves.toBe(true); +}); + +it('probe resolves false when no pong arrives before the timeout', async () => { + const { connection } = setupLiveness(); + + await handleConnection(server, connection.connect()); + expect(connection.status).toBe('connected'); + + jest.useFakeTimers(); + const probePromise = connection.probe(100); + await jest.advanceTimersByTimeAsync(100); + await expect(probePromise).resolves.toBe(false); + jest.useRealTimers(); +}); + +it('forceReopen closes the socket and re-establishes the connection', async () => { + const { connection } = setupLiveness(); + + await handleConnection(server, connection.connect()); + expect(connection.status).toBe('connected'); + + const reopened = connection.forceReopen(); + + await handleConnection(server, reopened); + + expect(connection.status).toBe('connected'); + await expect(reopened).resolves.toBe(true); + expect((connection as unknown as { reopenInFlight: unknown }).reopenInFlight).toBeNull(); +}); + +it('forceReopen dedupes concurrent callers onto the in-flight reopen', async () => { + const { connection } = setupLiveness(); + + await handleConnection(server, connection.connect()); + expect(connection.status).toBe('connected'); + + const first = connection.forceReopen(); + const second = connection.forceReopen(); + expect(second).toBe(first); + + await handleConnection(server, first, second); + + await expect(first).resolves.toBe(true); + expect((connection as unknown as { reopenInFlight: unknown }).reopenInFlight).toBeNull(); +}); + +it('checkAndReopen probes and returns true without a forced reopen when alive', async () => { + const { connection } = setupLiveness(); + + await handleConnection(server, connection.connect()); + const sessionBefore = connection.session; + + const resultPromise = connection.checkAndReopen(); + + await server.nextMessage.then((message) => { + expect(message).toBe('{"msg":"ping"}'); + server.send('{"msg":"pong"}'); + }); + + await expect(resultPromise).resolves.toBe(true); + expect(connection.status).toBe('connected'); + expect(connection.session).toBe(sessionBefore); +}); + +it('checkAndReopen force-reopens when the probe reports the connection dead', async () => { + const { connection } = setupLiveness(); + + await handleConnection(server, connection.connect()); + expect(connection.status).toBe('connected'); + + // Drive the "probe reports dead" branch directly so we don't depend on ping/ + // pong timing between two in-flight messages. + const probeSpy = jest.spyOn(connection, 'probe').mockResolvedValue(false); + + const resultPromise = connection.checkAndReopen(); + + await handleConnection(server, resultPromise); + + await expect(resultPromise).resolves.toBe(true); + expect(connection.status).toBe('connected'); + probeSpy.mockRestore(); +}); + +it('checkAndReopen force-reopens directly when the connection is not connected', async () => { + const { connection } = setupLiveness(); + + await handleConnection(server, connection.connect()); + connection.close(); + expect(connection.status).toBe('closed'); + + const resultPromise = connection.checkAndReopen(); + + await handleConnection(server, resultPromise); + + await expect(resultPromise).resolves.toBe(true); + expect(connection.status).toBe('connected'); +}); From 646cb6491de605bcb8d2a94b0034090f4f2d56ab Mon Sep 17 00:00:00 2001 From: Rohit <40559587+Rohit3523@users.noreply.github.com> Date: Sat, 25 Jul 2026 21:30:08 +0530 Subject: [PATCH 4/5] making coderabbit happy --- packages/ddp-client/__tests__/Connection.spec.ts | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/packages/ddp-client/__tests__/Connection.spec.ts b/packages/ddp-client/__tests__/Connection.spec.ts index ade6a60e90231..e16463f496b05 100644 --- a/packages/ddp-client/__tests__/Connection.spec.ts +++ b/packages/ddp-client/__tests__/Connection.spec.ts @@ -368,11 +368,14 @@ it('probe resolves false when no pong arrives before the timeout', async () => { await handleConnection(server, connection.connect()); expect(connection.status).toBe('connected'); - jest.useFakeTimers(); - const probePromise = connection.probe(100); - await jest.advanceTimersByTimeAsync(100); - await expect(probePromise).resolves.toBe(false); - jest.useRealTimers(); + try { + jest.useFakeTimers(); + const probePromise = connection.probe(100); + await jest.advanceTimersByTimeAsync(100); + await expect(probePromise).resolves.toBe(false); + } finally { + jest.useRealTimers(); + } }); it('forceReopen closes the socket and re-establishes the connection', async () => { From 42824efcbbc568c82ffd3231c4eee75012a6a020 Mon Sep 17 00:00:00 2001 From: Rohit <40559587+Rohit3523@users.noreply.github.com> Date: Sat, 25 Jul 2026 21:40:43 +0530 Subject: [PATCH 5/5] making cubic happy --- packages/ddp-client/__tests__/Connection.spec.ts | 10 ++++++---- packages/ddp-client/src/Connection.ts | 12 +++++++++--- 2 files changed, 15 insertions(+), 7 deletions(-) diff --git a/packages/ddp-client/__tests__/Connection.spec.ts b/packages/ddp-client/__tests__/Connection.spec.ts index e16463f496b05..98b9d67bbe028 100644 --- a/packages/ddp-client/__tests__/Connection.spec.ts +++ b/packages/ddp-client/__tests__/Connection.spec.ts @@ -355,8 +355,9 @@ it('probe resolves true when the server answers the ping with a pong', async () const probePromise = connection.probe(); await server.nextMessage.then((message) => { - expect(message).toBe('{"msg":"ping"}'); - server.send('{"msg":"pong"}'); + const { id } = JSON.parse(message); + expect(id).toEqual(expect.any(String)); + server.send(JSON.stringify({ msg: 'pong', id })); }); await expect(probePromise).resolves.toBe(true); @@ -418,8 +419,9 @@ it('checkAndReopen probes and returns true without a forced reopen when alive', const resultPromise = connection.checkAndReopen(); await server.nextMessage.then((message) => { - expect(message).toBe('{"msg":"ping"}'); - server.send('{"msg":"pong"}'); + const { id } = JSON.parse(message); + expect(id).toEqual(expect.any(String)); + server.send(JSON.stringify({ msg: 'pong', id })); }); await expect(resultPromise).resolves.toBe(true); diff --git a/packages/ddp-client/src/Connection.ts b/packages/ddp-client/src/Connection.ts index 8637cb439eafe..cffef7738001e 100644 --- a/packages/ddp-client/src/Connection.ts +++ b/packages/ddp-client/src/Connection.ts @@ -101,6 +101,8 @@ export class ConnectionImpl private reopenInFlight: Promise | null = null; + private onConnectionOff?: RemoveListener; + constructor( url: string, private WS: WebSocketConstructor, @@ -215,7 +217,8 @@ export class ConnectionImpl // If the server is willing to speak the version of the protocol specified in the connect message, it sends back a connected message. // Otherwise the server sends back a failed message with a version of DDP it would rather speak, informed by the connect message's support field, and closes the underlying transport. - this.client.onConnection((payload) => { + this.onConnectionOff?.(); + this.onConnectionOff = this.client.onConnection((payload) => { if (payload.msg === 'connected') { this.status = 'connected'; // Reset the retry budget on successful connection so a future @@ -293,6 +296,8 @@ export class ConnectionImpl close() { this.status = 'closed'; + this.onConnectionOff?.(); + this.onConnectionOff = undefined; if (this.ws) { // Sever all handlers before closing so the dying socket cannot deliver // late messages or fire `onclose` on the live connection (e.g. a replaced @@ -315,6 +320,7 @@ export class ConnectionImpl let settled = false; let off: RemoveListener | undefined; let timer: ReturnType; + const id = Math.random().toString(36).slice(2); const finish = (alive: boolean) => { if (settled) { @@ -328,11 +334,11 @@ export class ConnectionImpl timer = setTimeout(() => finish(false), timeoutMs); off = this.client.onMessage((payload) => { - if (payload.msg === 'pong') { + if (payload.msg === 'pong' && payload.id === id) { finish(true); } }); - this.client.ping(); + this.client.ping(id); }); }