fix(ddp-client): never queue non-method frames behind a wait block (RocketChat#40307)

Author: ggazzo

.changeset/ddp-client-idempotent-reconnect.md

@@ -0,0 +1,5 @@
+---
+"@rocket.chat/ddp-client": patch
+---
+
+Make `Connection.connect()` and `Connection.reconnect()` idempotent. Previously they rejected with `Error('Connection in progress')` when called while a connection was already in flight or established. Because the internal retry timer (`ws.onclose` → `setTimeout(() => void this.reconnect(), …)`) fires with no `.catch`, that rejection surfaced as an unhandled rejection at the page level whenever an external caller (e.g. an SDK consumer's bootstrap path) won the race against the timer. While `status === 'connecting'`, both methods now return the in-flight handshake promise so a later `failed` payload still propagates to every caller instead of being masked by a synthesized success; while `status === 'connected'` they resolve with `true`. The timer also no-ops when the connection has already been re-established, and a stale `ws.onclose` from a replaced socket no longer clobbers the new socket's status or schedules a redundant retry.

.changeset/ddp-client-reset-retry-count.md

@@ -0,0 +1,5 @@
+---
+"@rocket.chat/ddp-client": patch
+---
+
+Reset `Connection.retryCount` to zero on a successful (re)connection. The counter was only ever incremented (in `ws.onclose`), never zeroed, so the retry budget was monotonically consumed across the connection's lifetime. With the default budget of `1`, any chain of two close events — for example the SDK reconnecting after a server force-logout, then the client running a follow-up `Meteor.logout()` whose server handler closes the WS again — drained the budget; the second close handler bailed at `retryCount >= retryOptions.retryCount` and the SDK stayed permanently disconnected. Method frames already in the dispatcher queue (e.g. a fresh `login` retry from the consumer) stayed queued forever.

.changeset/ddp-dispatcher-non-method-frames.md

@@ -0,0 +1,5 @@
+---
+"@rocket.chat/ddp-client": patch
+---
+
+Fix `DDPDispatcher` dropping non-method frames (connect, sub, unsub, ping, pong) when a `wait` block is at the head of the queue. Previously every payload flowed through the same wait-serialization path: a `connect` frame dispatched after a `wait: true` method (e.g. `login`) would be queued in a new non-wait block but never actually sent, wedging the DDP handshake — the socket stayed open, the server never replied `connected`, and any caller awaiting the connection hung. Non-method payloads now bypass the queue and emit immediately; wait-method serialization between methods is unchanged.

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

@@ -139,20 +139,198 @@ it('should queue messages if the connection is not ready', async () => {
 	await handleMethod(server, 'method', ['arg1', 'arg2'], '1');
 });
 
-it('should throw an error if a reconnect is called while a connection is in progress', async () => {
+it('should be idempotent if reconnect is called while already connected', async () => {
 	const client = new MinimalDDPClient();
 	const connection = ConnectionImpl.create('ws://localhost:1234', globalThis.WebSocket, client, { retryCount: 0, retryTime: 0 });
 
 	await handleConnection(server, connection.connect());
 
-	await expect(connection.reconnect()).rejects.toThrow('Connection in progress');
+	// Previous behavior was to throw "Connection in progress" — the consumer's
+	// `void this.reconnect()` paths (notably the ws.onclose retry timer)
+	// surfaced that as an unhandled rejection / pageError. Now a redundant
+	// reconnect is just a no-op resolving with the current state.
+	await expect(connection.reconnect()).resolves.toBe(true);
+	expect(connection.status).toBe('connected');
 });
 
-it('should throw an error if a connect is called while a connection is in progress', async () => {
+it('should be idempotent if connect is called while already connected', async () => {
 	const client = new MinimalDDPClient();
 	const connection = ConnectionImpl.create('ws://localhost:1234', globalThis.WebSocket, client, { retryCount: 0, retryTime: 0 });
 
 	await handleConnection(server, connection.connect());
 
-	await expect(connection.connect()).rejects.toThrow('Connection in progress');
+	await expect(connection.connect()).resolves.toBe(true);
+	expect(connection.status).toBe('connected');
+});
+
+it('should share the in-flight connect promise with a concurrent connect() caller', async () => {
+	// Regression: while status === 'connecting', a second connect() used to
+	// receive Promise.resolve(true) immediately, hiding any subsequent
+	// 'failed' payload from the in-flight handshake. Both callers must now
+	// observe the real outcome.
+	const client = new MinimalDDPClient();
+	const connection = new ConnectionImpl('ws://localhost:1234', WebSocket as any, client, { retryCount: 0, retryTime: 0 });
+
+	const first = connection.connect();
+	expect(connection.status).toBe('connecting');
+
+	const second = connection.connect();
+	expect(second).toBe(first);
+
+	await expect(handleConnectionAndRejects(server, first, second)).rejects.toBe('1');
+	expect(connection.status).toBe('failed');
+});
+
+it('should share the in-flight connect promise with a concurrent reconnect() caller', async () => {
+	// Same as above for the reconnect() entry point — the ws.onclose retry
+	// timer fires `reconnect()` and must piggyback on any handshake the
+	// consumer's bootstrap path already started.
+	const client = new MinimalDDPClient();
+	const connection = new ConnectionImpl('ws://localhost:1234', WebSocket as any, client, { retryCount: 0, retryTime: 0 });
+
+	const first = connection.connect();
+	expect(connection.status).toBe('connecting');
+
+	const second = connection.reconnect();
+	expect(second).toBe(first);
+
+	await handleConnection(server, first, second);
+	expect(connection.status).toBe('connected');
+});
+
+it('should not surface the retry timer rejection when an external connect won the race', async () => {
+	// Regression: ws.onclose schedules a `void this.reconnect()` timer; if the
+	// consumer (e.g. ddpSdk.ts startConnect) opens a fresh socket before that
+	// timer fires, the timer used to reject with "Connection in progress" and,
+	// because of the leading `void`, the rejection became an unhandled
+	// rejection on the page. The timer must now no-op silently when the
+	// connection has already been re-established.
+	const client = new MinimalDDPClient();
+	const connection = ConnectionImpl.create('ws://localhost:1234', WebSocket, client, { retryCount: 1, retryTime: 100 });
+
+	await handleConnection(server, connection.connect());
+	expect(connection.status).toBe('connected');
+
+	jest.useFakeTimers();
+
+	server.close();
+	WS.clean();
+	server = new WS('ws://localhost:1234/websocket');
+
+	expect(connection.status).toBe('disconnected');
+
+	// Track unhandled rejections on the timer's promise.
+	const unhandled = jest.fn();
+	process.on('unhandledRejection', unhandled);
+
+	// External code opens a new connection BEFORE the retry timer fires.
+	const externalConnect = handleConnection(server, connection.connect());
+
+	// Run the timer.
+	await jest.advanceTimersByTimeAsync(200);
+	await externalConnect;
+
+	// Drain any microtasks the timer might have queued.
+	await Promise.resolve();
+	await Promise.resolve();
+
+	expect(connection.status).toBe('connected');
+	expect(unhandled).not.toHaveBeenCalled();
+	process.off('unhandledRejection', unhandled);
+	jest.useRealTimers();
+});
+
+it('should reset retryCount on a successful connection so subsequent drops can retry', async () => {
+	// Regression: retryCount was only ever incremented on disconnect, never
+	// zeroed on successful (re)connect. With default retryCount=1 budget, a
+	// single force-logout cycle (server close → SDK reconnects → app calls
+	// `Meteor.logout()` → server's logout handler closes WS again) drained the
+	// budget, and the second close left the SDK permanently disconnected.
+	// Method frames queued on the SDK during that window stayed queued
+	// forever — observed in e2e-encryption/e2ee-passphrase-management as the
+	// next loginByUserState login frame never being delivered.
+	const client = new MinimalDDPClient();
+	const connection = ConnectionImpl.create('ws://localhost:1234', WebSocket, client, { retryCount: 1, retryTime: 100 });
+
+	await handleConnection(server, connection.connect());
+	expect(connection.status).toBe('connected');
+	expect((connection as unknown as { retryCount: number }).retryCount).toBe(0);
+
+	// First disconnect → schedules retry (retryCount: 0 → 1).
+	jest.useFakeTimers();
+	server.close();
+	WS.clean();
+	server = new WS('ws://localhost:1234/websocket');
+
+	expect(connection.status).toBe('disconnected');
+
+	await handleConnection(
+		server,
+		jest.advanceTimersByTimeAsync(200),
+		new Promise((resolve) => connection.once('reconnecting', () => resolve(undefined))),
+		new Promise((resolve) => connection.once('connection', (data) => resolve(data))),
+	);
+	jest.useRealTimers();
+	expect(connection.status).toBe('connected');
+
+	// Successful reconnect must zero the retry budget.
+	expect((connection as unknown as { retryCount: number }).retryCount).toBe(0);
+
+	// Second disconnect should still schedule a retry now that the budget reset.
+	jest.useFakeTimers();
+	server.close();
+	WS.clean();
+	server = new WS('ws://localhost:1234/websocket');
+
+	await handleConnection(
+		server,
+		jest.advanceTimersByTimeAsync(200),
+		new Promise((resolve) => connection.once('reconnecting', () => resolve(undefined))),
+		new Promise((resolve) => connection.once('connection', (data) => resolve(data))),
+	);
+	jest.useRealTimers();
+	expect(connection.status).toBe('connected');
+});
+
+it('should ignore a stale ws.onclose that fires after the socket has been replaced', async () => {
+	// Regression: ws.onclose handlers were closed over the original ws but
+	// mutated `this.status`/`this.retryCount` unconditionally. If a late close
+	// event from an old socket arrives after a new socket is connected, the
+	// handler would flip status back to 'disconnected' and schedule another
+	// retry timer.
+	const client = new MinimalDDPClient();
+	const connection = ConnectionImpl.create('ws://localhost:1234', WebSocket, client, { retryCount: 1, retryTime: 100 });
+
+	await handleConnection(server, connection.connect());
+	const firstWs = (connection as unknown as { ws: WebSocket }).ws;
+	expect(connection.status).toBe('connected');
+
+	jest.useFakeTimers();
+	server.close();
+	WS.clean();
+	server = new WS('ws://localhost:1234/websocket');
+
+	expect(connection.status).toBe('disconnected');
+
+	await handleConnection(
+		server,
+		jest.advanceTimersByTimeAsync(200),
+		new Promise((resolve) => connection.once('reconnecting', () => resolve(undefined))),
+		new Promise((resolve) => connection.once('connection', (data) => resolve(data))),
+	);
+
+	expect(connection.status).toBe('connected');
+	jest.useRealTimers();
+	const secondWs = (connection as unknown as { ws: WebSocket }).ws;
+	expect(secondWs).not.toBe(firstWs);
+
+	const statusBefore = connection.status;
+	const retryBefore = (connection as unknown as { retryCount: number }).retryCount;
+
+	// Synthesize a late `close` event on the original socket — the handler
+	// must short-circuit because `this.ws !== ws` for the closed-over ws.
+	(firstWs as unknown as { onclose?: () => void }).onclose?.();
+
+	expect(connection.status).toBe(statusBefore);
+	expect((connection as unknown as { retryCount: number }).retryCount).toBe(retryBefore);
 });

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

@@ -72,6 +72,34 @@ it('should send outstanding blocks if there is no block waiting and item is adde
 	expect(fn).toHaveBeenCalledTimes(1);
 });
 
+it('emits non-method payloads immediately, even when a wait block is at the head', () => {
+	// Regression: a connect frame dispatched while a wait `login` method is
+	// queued must still reach the server. Otherwise the DDP handshake never
+	// completes and the socket wedges open but unconnected.
+	const fn = jest.fn();
+	const ddpDispatcher = new DDPDispatcher();
+	ddpDispatcher.on('send', fn);
+
+	const login = ddp.call('login');
+	ddpDispatcher.dispatch(login, { wait: true });
+	expect(fn).toHaveBeenCalledTimes(1);
+	expect(fn).toHaveBeenNthCalledWith(1, login);
+
+	const connectPayload = { msg: 'connect' as const, version: '1', support: ['1'] };
+	ddpDispatcher.dispatch(connectPayload);
+	expect(fn).toHaveBeenCalledTimes(2);
+	expect(fn).toHaveBeenNthCalledWith(2, connectPayload);
+
+	const subPayload = { msg: 'sub' as const, id: 'a', name: 'foo', params: [] };
+	ddpDispatcher.dispatch(subPayload);
+	expect(fn).toHaveBeenCalledTimes(3);
+	expect(fn).toHaveBeenNthCalledWith(3, subPayload);
+
+	// Wait block remains pending — only the wait method is queued, the
+	// non-method frames bypassed it.
+	expect(ddpDispatcher.queue).toEqual([{ wait: true, items: [login] }]);
+});
+
 it('should send the next blocks if the outstanding block was completed', () => {
 	const fn = jest.fn();
 

packages/ddp-client/src/Connection.ts

@@ -74,6 +74,8 @@ export class ConnectionImpl
 
 	retryCount = 0;
 
+	private connectPromise?: Promise<boolean>;
+
 	public queue = new Set<string>();
 
 	constructor(
@@ -109,8 +111,23 @@ export class ConnectionImpl
 	}
 
 	reconnect(): Promise<boolean> {
-		if (this.status === 'connecting' || this.status === 'connected') {
-			return Promise.reject(new Error('Connection in progress'));
+		// Idempotent — if another caller already started (or finished) a connection
+		// since this reconnect was scheduled, we don't need to do anything. The
+		// retry timer enqueued by `ws.onclose` runs with no awareness of any
+		// concurrent `connect()` (e.g. the consumer's own bootstrap or
+		// resume-on-userId-change path), so without this guard a late timer
+		// rejected with "Connection in progress" — and because the timer fires
+		// from `void this.reconnect()` the rejection became an unhandled
+		// rejection at the page level.
+		if (this.status === 'connected') {
+			clearTimeout(this.retryOptions.retryTimer);
+			return Promise.resolve(true);
+		}
+		if (this.status === 'connecting') {
+			// Share the in-flight handshake promise so a `failed` payload
+			// later in the same attempt isn't masked by a synthesized success.
+			clearTimeout(this.retryOptions.retryTimer);
+			return this.connectPromise as Promise<boolean>;
 		}
 
 		clearTimeout(this.retryOptions.retryTimer);
@@ -123,19 +140,31 @@ export class ConnectionImpl
 	}
 
 	connect() {
-		if (this.status === 'connecting' || this.status === 'connected') {
-			return Promise.reject(new Error('Connection in progress'));
+		// Same idempotency guard as `reconnect()` — multiple call sites
+		// (`reconnect()`, ws.onclose retry timer, external `startConnect`) can
+		// race; rejecting forced every caller to wrap in `.catch(() => {})`
+		// just to silence noise, and the internal timer's `void this.reconnect()`
+		// path didn't have a catch at all.
+		if (this.status === 'connected') {
+			clearTimeout(this.retryOptions.retryTimer);
+			return Promise.resolve(true);
+		}
+		if (this.status === 'connecting') {
+			clearTimeout(this.retryOptions.retryTimer);
+			return this.connectPromise as Promise<boolean>;
 		}
 
 		this.status = 'connecting';
-		this.emit('connecting');
-		this.emitStatus();
 
 		const ws = new this.WS(`${this.ssl ? 'wss://' : 'ws://'}${this.url}/websocket`);
 
 		this.ws = ws;
 
-		return new Promise<boolean>((resolve, reject) => {
+		// Build the in-flight promise and publish it on `this.connectPromise`
+		// before emitting any status change, so a synchronous re-entrant
+		// caller (an event listener that calls `connect()`/`reconnect()`)
+		// hits the `'connecting'` guard and gets this same promise.
+		const connectPromise = new Promise<boolean>((resolve, reject) => {
 			ws.onopen = () => {
 				ws.onmessage = (event) => {
 					this.client.handleMessage(String(event.data));
@@ -166,6 +195,15 @@ export class ConnectionImpl
 				this.client.onConnection((payload) => {
 					if (payload.msg === 'connected') {
 						this.status = 'connected';
+						// Reset the retry budget on successful connection so a future
+						// disconnect can schedule reconnects again. Without this,
+						// long-lived connections that recover once would burn through
+						// `retryCount` permanently and stop reconnecting on subsequent
+						// drops — observed when a server-side ws.close (logout, force-
+						// logout) chained with a reconnect cycle saturated the
+						// budget; the next disconnect left frames stuck in the
+						// dispatcher queue forever because the socket never came back.
+						this.retryCount = 0;
 						this.emitStatus();
 						this.emit('connected', payload.session);
 						this.session = payload.session;
@@ -183,6 +221,15 @@ export class ConnectionImpl
 			};
 
 			ws.onclose = () => {
+				// If a newer ws has already taken over (this socket was closed
+				// after `connect()` opened a replacement), ignore the late
+				// onclose. Otherwise its handler would clobber `this.status` and
+				// `retryCount`, and could even schedule a redundant retry timer
+				// that fires while the new socket is healthy — observed as the
+				// "Connection in progress" pageError racing on every reconnect.
+				if (this.ws !== ws) {
+					return;
+				}
 				clearTimeout(this.retryOptions.retryTimer);
 				if (this.status === 'closed') {
 					return;
@@ -198,10 +245,27 @@ export class ConnectionImpl
 				this.retryCount += 1;
 
 				this.retryOptions.retryTimer = setTimeout(() => {
+					// Re-check the status when the timer actually fires. If the
+					// consumer bootstrapped a fresh `connect()` in the meantime
+					// (status flipped from 'disconnected' to 'connecting' or
+					// 'connected'), there's nothing for us to do. Without this
+					// the timer would call `this.reconnect()`, which (pre-this
+					// patch) rejected with "Connection in progress" and surfaced
+					// as an unhandled rejection.
+					if (this.status === 'connecting' || this.status === 'connected' || this.status === 'closed') {
+						return;
+					}
 					void this.reconnect();
 				}, this.retryOptions.retryTime * this.retryCount);
 			};
 		});
+
+		this.connectPromise = connectPromise;
+
+		this.emit('connecting');
+		this.emitStatus();
+
+		return connectPromise;
 	}
 
 	close() {