feat: ddp-client connection liveness check and forceReopen

.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.

packages/ddp-client/__tests__/Connection.spec.ts

@@ -334,3 +334,131 @@ 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) => {
+		const { id } = JSON.parse(message);
+		expect(id).toEqual(expect.any(String));
+		server.send(JSON.stringify({ msg: 'pong', id }));
+	});
+
+	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');
+
+	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 () => {
+	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) => {
+		const { id } = JSON.parse(message);
+		expect(id).toEqual(expect.any(String));
+		server.send(JSON.stringify({ msg: 'pong', id }));
+	});
+
+	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');
+});

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;
@@ -45,6 +46,26 @@ export interface Connection
 	reconnect(): Promise<boolean>;
 
 	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<boolean>;
+
+	/**
+	 * Tear down the current socket and reconnect from scratch. Concurrent
+	 * callers share one in-flight promise so parallel calls don't race.
+	 */
+	forceReopen(): Promise<boolean>;
+
+	/**
+	 * 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<boolean>;
 }
 
 interface WebSocketConstructor {
@@ -78,6 +99,10 @@ export class ConnectionImpl
 
 	public queue = new Set<string>();
 
+	private reopenInFlight: Promise<boolean> | null = null;
+
+	private onConnectionOff?: RemoveListener;
+
 	constructor(
 		url: string,
 		private WS: WebSocketConstructor,
@@ -192,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
@@ -270,10 +296,77 @@ export class ConnectionImpl
 
 	close() {
 		this.status = 'closed';
-		this.ws?.close();
+		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
+			// 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<boolean> {
+		return new Promise<boolean>((resolve) => {
+			let settled = false;
+			let off: RemoveListener | undefined;
+			let timer: ReturnType<typeof setTimeout>;
+			const id = Math.random().toString(36).slice(2);
+
+			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' && payload.id === id) {
+					finish(true);
+				}
+			});
+			this.client.ping(id);
+		});
+	}
+
+	forceReopen(): Promise<boolean> {
+		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<boolean> {
+		if (this.status !== 'connected') {
+			return this.forceReopen();
+		}
+		const alive = await this.probe(probeTimeoutMs);
+		return alive || this.forceReopen();
+	}
+
 	static create(
 		url: string,
 		webSocketImpl: WebSocketConstructor,

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] });