diff --git a/app/lib/services/__tests__/socketHealth.integration.test.ts b/app/lib/services/__tests__/socketHealth.integration.test.ts index edd77e267b..034e0606b0 100644 --- a/app/lib/services/__tests__/socketHealth.integration.test.ts +++ b/app/lib/services/__tests__/socketHealth.integration.test.ts @@ -1,5 +1,5 @@ import sdk from '../sdk'; -import { recoverSocket } from '../socketHealth'; +import { classifySocketHealth, recoverSocket } from '../socketHealth'; // eslint-disable-next-line @typescript-eslint/no-var-requires const { DDPDriver } = require('@rocket.chat/sdk/lib/drivers/ddp') as { @@ -26,10 +26,13 @@ interface WireFrame { interface PatchedDriver { userId: string; pingInterval: number; + lastPing: number; reopenNow(): Promise; + probe(timeoutMs: number): Promise; waitForNotifyUserMediaSubs(timeoutMs?: number): Promise; ddp: { lastPing: number; + lastPongAt: number; pingTimeout?: ReturnType; openTimeout?: ReturnType; open(): Promise; @@ -48,7 +51,9 @@ jest.mock('universal-websocket-client', () => if (message.msg === 'connect') { setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'connected', session: 'session-id' }) })); } else if (message.msg === 'ping') { - setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'pong' }) })); + // A DDP server echoes the ping's id on its pong, which is what lets the + // round-trip check tell its own answer from an unrelated frame. + setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'pong', id: message.id }) })); } else if (message.msg === 'sub') { setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'ready', subs: [message.id] }) })); } @@ -98,8 +103,13 @@ function addMediaSubs(driver: PatchedDriver) { }); } -function backdateLastPing(driver: PatchedDriver, ageMs: number) { +/** + * Age the socket's heartbeat. Both timestamps move: `lastPing` is what any inbound + * frame refreshes, `lastPongAt` is what the health classification ages against. + */ +function backdateHeartbeat(driver: PatchedDriver, ageMs: number) { driver.ddp.lastPing = Date.now() - ageMs; + driver.ddp.lastPongAt = Date.now() - ageMs; } /** Frames of a given `msg` sent over the wire on one connection. */ @@ -131,7 +141,7 @@ describe('recoverSocket against the real patched socket', () => { }); it('keeps a doubtful socket when the round trip gets a pong', async () => { - backdateLastPing(driver, PING_INTERVAL + 5000); + backdateHeartbeat(driver, PING_INTERVAL + 5000); const recovery = recoverSocket(); await jest.advanceTimersByTimeAsync(0); @@ -143,7 +153,7 @@ describe('recoverSocket against the real patched socket', () => { }); it('reopens a doubtful socket when the round trip gets no pong', async () => { - backdateLastPing(driver, PING_INTERVAL + 5000); + backdateHeartbeat(driver, PING_INTERVAL + 5000); // A zombie socket: still `readyState: 1`, but the server never answers. mockConnections[0].send.mockImplementation(() => undefined); @@ -177,7 +187,7 @@ describe('recoverSocket against the real patched socket', () => { }); it('reopens a known-dead socket without a round trip', async () => { - backdateLastPing(driver, PING_INTERVAL * 2 + 1000); + backdateHeartbeat(driver, PING_INTERVAL * 2 + 1000); const recovery = recoverSocket(); await jest.advanceTimersByTimeAsync(0); @@ -193,7 +203,7 @@ describe('recoverSocket against the real patched socket', () => { }); it('shares one reopen with a concurrent direct reopenNow', async () => { - backdateLastPing(driver, PING_INTERVAL * 3); + backdateHeartbeat(driver, PING_INTERVAL * 3); // The foreground path reopens the dead socket while recovery does the same. const directReopen = driver.reopenNow(); @@ -222,7 +232,7 @@ describe('recoverSocket against the real patched socket', () => { expect(rejected).toBe(false); // The socket dies silently after the call went out. - backdateLastPing(driver, PING_INTERVAL * 3); + backdateHeartbeat(driver, PING_INTERVAL * 3); const recovery = recoverSocket(); await jest.advanceTimersByTimeAsync(0); @@ -235,8 +245,89 @@ describe('recoverSocket against the real patched socket', () => { await expect(recovery).resolves.toBe('reopened'); }); + /** + * The Android-freeze case: the process is frozen past the ping deadline and the server + * drops the session, then the OS flushes the frames buffered during the freeze. Those + * frames must not vouch for a session that is already gone — if the socket is not + * reopened, no `close` and no `connecting` are emitted, the resume login never runs, + * and the server-side streams stay dead for the rest of the session. + * + * Each test isolates one way a flushed frame used to be mistaken for liveness. + */ + describe('a frozen socket whose buffered frames are flushed on resume', () => { + /** A frame that arrived during the freeze and is only delivered on resume. */ + const flush = (frame: Record) => { + mockConnections[0].onmessage({ data: JSON.stringify(frame) }); + }; + + beforeEach(() => { + // The server dropped this session during the freeze: nothing we send is answered. + mockConnections[0].send.mockImplementation(() => undefined); + addMediaSubs(driver); + }); + + it('does not let a flushed data frame make a long-frozen socket look young', async () => { + backdateHeartbeat(driver, 170_000); + + // The message the other user sent while we were frozen, delivered on resume. + // It refreshes `lastPing`, which is why the classification cannot age against it. + flush({ msg: 'changed', collection: 'stream-room-messages', fields: { args: [{ _id: 'm1' }] } }); + + expect(classifySocketHealth(driver)).toBe('reopen'); + + const recovery = recoverSocket(); + await jest.advanceTimersByTimeAsync(0); + + // Straight to a reopen: a socket this old is not worth a round trip. + expect(mockConnections).toHaveLength(2); + expect(framesOn(mockConnections[0], 'ping')).toHaveLength(0); + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + await expect(recovery).resolves.toBe('reopened'); + }); + + it('does not accept a pong that is not the answer to the round trip', async () => { + // In the gray zone, so recovery pays for a round trip rather than reopening blind. + backdateHeartbeat(driver, PING_INTERVAL + 5000); + expect(classifySocketHealth(driver)).toBe('round-trip-check'); + + const recovery = recoverSocket(); + // The OS drains the receive buffer over several ms, so the round trip is already + // waiting when the rest of the backlog lands. + await jest.advanceTimersByTimeAsync(50); + + // A pong buffered before the freeze — the answer to the periodic ping the ping + // loop sent on its way down, carrying no id, not the answer to our round trip. + flush({ msg: 'pong' }); + await jest.advanceTimersByTimeAsync(2000); + + expect(mockConnections).toHaveLength(2); + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + await expect(recovery).resolves.toBe('reopened'); + }); + + it('does not accept a pong answering a different round trip', async () => { + backdateHeartbeat(driver, PING_INTERVAL + 5000); + + const recovery = recoverSocket(); + await jest.advanceTimersByTimeAsync(50); + + // An id-carrying pong, but for a different probe — a real possibility once an + // earlier round trip has timed out and its ping is answered late. This round + // trip is `probe-0`, the first on this socket. + flush({ msg: 'pong', id: 'probe-7' }); + await jest.advanceTimersByTimeAsync(2000); + + expect(mockConnections).toHaveLength(2); + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + await expect(recovery).resolves.toBe('reopened'); + }); + }); + it('re-sends the media subscriptions on the new socket reusing their ids', async () => { - backdateLastPing(driver, PING_INTERVAL * 3); + backdateHeartbeat(driver, PING_INTERVAL * 3); addMediaSubs(driver); const recovery = recoverSocket(); diff --git a/app/lib/services/__tests__/socketHealth.test.ts b/app/lib/services/__tests__/socketHealth.test.ts index 031c7fe92a..8cfeb81cb3 100644 --- a/app/lib/services/__tests__/socketHealth.test.ts +++ b/app/lib/services/__tests__/socketHealth.test.ts @@ -15,6 +15,7 @@ const sdkMock = sdk as unknown as { current: { ddp: unknown } | undefined }; interface MockDdp { connected?: boolean; lastPing: number; + lastPongAt?: number; pingInterval?: number; config?: { ping?: number }; reopenNow: jest.Mock, []>; @@ -71,6 +72,23 @@ describe('classifySocketHealth', () => { const ddp = makeDdp({ connected: false, lastPing: now }); expect(classifySocketHealth(ddp)).toBe('reopen'); }); + + // `lastPing` is refreshed by every inbound frame, so it says "traffic arrived", not + // "the server answered us". Only the pong timestamp can age a socket. + it('ages against lastPongAt rather than a lastPing refreshed by an unrelated frame', () => { + const ddp = makeDdp({ lastPing: now, lastPongAt: now - 170000 }); + expect(classifySocketHealth(ddp)).toBe('reopen'); + }); + + it('keeps a socket whose pong is recent even if no other frame has arrived since', () => { + const ddp = makeDdp({ lastPing: now - 170000, lastPongAt: now - 5000 }); + expect(classifySocketHealth(ddp)).toBe('round-trip-check'); + }); + + it('falls back to lastPing on a driver without the patched pong timestamp', () => { + const ddp = makeDdp({ lastPing: now - 21000, lastPongAt: undefined }); + expect(classifySocketHealth(ddp)).toBe('reopen'); + }); }); describe('recoverSocket', () => { diff --git a/app/lib/services/connect.test.ts b/app/lib/services/connect.test.ts index 7f1ad5edd3..c7b84e0a54 100644 --- a/app/lib/services/connect.test.ts +++ b/app/lib/services/connect.test.ts @@ -1,7 +1,11 @@ +import { type Action } from 'redux'; + import { connect, determineAuthType, disconnect } from './connect'; import { mediaSessionInstance } from './voip/MediaSessionInstance'; import { pendingHangups } from './voip/pendingHangups'; import { setUser } from '../../actions/login'; +import { LOGIN } from '../../actions/actionsTypes'; +import connectReducer, { type IConnect } from '../../reducers/connect'; import database from '../database'; jest.mock('./voip/MediaSessionInstance', () => ({ @@ -415,6 +419,75 @@ describe('VoIP media session lifecycle (disconnect)', () => { }); }); +// A silent socket swap (`reopenNow`) emits `connecting` then `connected` and never `close`. +// The resume login only runs if `meteor.connected` is false by the time `connected` lands, +// so these drive the real connect reducer rather than hand-feeding the state. +describe('connect — resume login after a silent socket reopen', () => { + const TOKEN = 'resume-token'; + + const wireStoreToRealReducer = (initial: IConnect) => { + let meteor = initial; + mockStoreGetState.mockImplementation(() => ({ + meteor, + login: { user: { token: TOKEN }, isAuthenticated: true }, + settings: {} + })); + mockStoreDispatch.mockImplementation(action => { + meteor = connectReducer(meteor, action as Action); + return action; + }); + return () => meteor; + }; + + beforeEach(() => { + jest.clearAllMocks(); + mockOnStreamDataStops.length = 0; + mockStoreSubscribe.mockImplementation(noopUnsubscribe); + pendingHangups.clear(); + }); + + afterEach(() => { + mockStoreGetState.mockReset(); + mockStoreDispatch.mockReset(); + }); + + const dispatchedLoginRequests = () => + mockStoreDispatch.mock.calls + .map(([action]) => action as { type: string; credentials?: unknown }) + .filter(action => action.type === LOGIN.REQUEST); + + it('re-runs the resume login when connecting → connected replaces the socket', async () => { + const readMeteor = wireStoreToRealReducer({ connecting: false, connected: true }); + + await connect({ server: 'https://example.com' }); + + // The order `reopenNow` produces: createConnection emits `connecting`, then the + // server's handshake reply emits `connected`. No `close` is ever emitted. + getHandlersByEvent('connecting')[0](); + expect(readMeteor().connected).toBe(false); + + getHandlersByEvent('connected')[0](); + await flushMicrotasks(); + + expect(readMeteor().connected).toBe(true); + expect(dispatchedLoginRequests()).toHaveLength(1); + expect(dispatchedLoginRequests()[0].credentials).toEqual({ resume: TOKEN }); + }); + + it('does not re-run the resume login when connected repeats on the same socket', async () => { + wireStoreToRealReducer({ connecting: false, connected: true }); + + await connect({ server: 'https://example.com' }); + + // First `connected` logs in; a repeat with no intervening socket swap must not. + getHandlersByEvent('connected')[0](); + getHandlersByEvent('connected')[0](); + await flushMicrotasks(); + + expect(dispatchedLoginRequests()).toHaveLength(1); + }); +}); + describe('connect — pendingHangups drain on reconnect', () => { beforeEach(() => { jest.clearAllMocks(); diff --git a/app/lib/services/ddpSocket.test.ts b/app/lib/services/ddpSocket.test.ts index 44acd6e732..c2bd9403da 100644 --- a/app/lib/services/ddpSocket.test.ts +++ b/app/lib/services/ddpSocket.test.ts @@ -12,7 +12,8 @@ jest.mock('universal-websocket-client', () => { if (message.msg === 'connect') { setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'connected', session: 'session-id' }) })); } else if (message.msg === 'ping') { - setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'pong' }) })); + // A DDP server echoes the ping's id on its pong. + setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'pong', id: message.id }) })); } }), close: jest.fn(), @@ -69,11 +70,17 @@ describe('Socket.probe', () => { jest.useRealTimers(); }); - it('resolves true when pong arrives within deadline', async () => { - const { socket } = buildSocket(); + /** The id the probe put on the wire, so a test can answer that exact ping. */ + const probeIdFrom = (send: jest.Mock): string => { + const frame = JSON.parse(send.mock.calls.at(-1)[0]); + expect(frame.msg).toBe('ping'); + return frame.id; + }; + + it('resolves true when the pong answering its ping arrives within deadline', async () => { + const { socket, send } = buildSocket(); const probePromise = socket.probe(); - socket.lastPing += 1; - socket.emit('pong'); + socket.emit('pong', { msg: 'pong', id: probeIdFrom(send) }); await expect(probePromise).resolves.toBe(true); }); @@ -99,18 +106,62 @@ describe('Socket.probe', () => { await expect(socket.probe()).resolves.toBe(false); }); - it('ignores a stale pong that does not advance lastPing', async () => { + // A pong with no id cannot be attributed to anything. On resume the OS flushes the + // frames buffered during a freeze, so accepting one would vouch for a session the + // server has already dropped. + it('ignores an uncorrelated pong', async () => { jest.useFakeTimers(); const { socket } = buildSocket(); - const initialLastPing = Date.now() - 1000; - socket.lastPing = initialLastPing; const probePromise = socket.probe(); - socket.emit('pong'); + socket.emit('pong', { msg: 'pong' }); await jest.advanceTimersByTimeAsync(2000); await expect(probePromise).resolves.toBe(false); }); + + it('ignores a pong carrying a different probe id', async () => { + jest.useFakeTimers(); + const { socket, send } = buildSocket(); + + const probePromise = socket.probe(); + socket.emit('pong', { msg: 'pong', id: `${probeIdFrom(send)}-other` }); + + await jest.advanceTimersByTimeAsync(2000); + await expect(probePromise).resolves.toBe(false); + }); + + // `once` detaches before invoking its listener, so registering that way would let an + // uncorrelated pong consume the registration and strand the real answer. + it('still accepts its own pong after an uncorrelated one arrives first', async () => { + jest.useFakeTimers(); + const { socket, send } = buildSocket(); + + const probePromise = socket.probe(); + const probeId = probeIdFrom(send); + socket.emit('pong', { msg: 'pong' }); + socket.emit('pong', { msg: 'pong', id: probeId }); + + await expect(probePromise).resolves.toBe(true); + }); + + it('gives each round trip its own id', async () => { + jest.useFakeTimers(); + const { socket, send } = buildSocket(); + + const firstProbe = socket.probe(); + const first = probeIdFrom(send); + const secondProbe = socket.probe(); + const second = probeIdFrom(send); + + expect(second).not.toBe(first); + + // Both are held and awaited: neither socket is answered, so each must time out to + // false, and a rejection fails the test rather than surfacing as an unhandled one. + await jest.advanceTimersByTimeAsync(2000); + await expect(firstProbe).resolves.toBe(false); + await expect(secondProbe).resolves.toBe(false); + }); }); describe('Socket.reopenNow', () => { diff --git a/app/lib/services/socketHealth.ts b/app/lib/services/socketHealth.ts index aa7d4ceb5f..0394d23d51 100644 --- a/app/lib/services/socketHealth.ts +++ b/app/lib/services/socketHealth.ts @@ -9,6 +9,8 @@ import sdk from './sdk'; interface SocketHealthDdp { connected?: boolean; lastPing: number; + /** When the server last answered a ping. Absent on a driver without the patch. */ + lastPongAt?: number; pingInterval?: number; config?: { ping?: number }; reopenNow(): Promise; @@ -32,7 +34,11 @@ export function classifySocketHealth(ddp: SocketHealthDdp): SocketRecoveryPlan { return 'reopen'; } const pingInterval = (ddp.pingInterval ?? ddp.config?.ping) || 10000; - const age = Date.now() - ddp.lastPing; + // Age against the last pong, not the last frame. `lastPing` is refreshed by every + // inbound frame, so after the OS unfreezes a backgrounded app the flushed backlog + // makes a socket frozen for minutes look seconds old. `lastPing` stays as the + // fallback for a driver without the patched timestamp. + const age = Date.now() - (ddp.lastPongAt ?? ddp.lastPing); if (age > pingInterval * 2) { return 'reopen'; } diff --git a/patches/@rocket.chat+sdk+1.3.3-mobile.patch b/patches/@rocket.chat+sdk+1.3.3-mobile.patch index e0e2d0b446..8041392b28 100644 --- a/patches/@rocket.chat+sdk+1.3.3-mobile.patch +++ b/patches/@rocket.chat+sdk+1.3.3-mobile.patch @@ -1,8 +1,25 @@ diff --git a/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts b/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts -index 19d31ae..068b61e 100644 +index 19d31ae..a7d028a 100644 --- a/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts +++ b/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts -@@ -55,6 +55,7 @@ export class Socket extends EventEmitter { +@@ -45,8 +45,16 @@ const userDisconnectCloseCode = 4000; + /** Websocket handler class, manages connections and subscriptions by DDP */ + export class Socket extends EventEmitter { + sent = 0 ++ probed = 0 + host: string + lastPing = Date.now() ++ /** ++ * When the server last answered a ping. `lastPing` is refreshed by every inbound ++ * frame, so it measures traffic rather than heartbeat health: a single frame ++ * flushed from the OS buffer after a freeze makes a long-dead socket look young. ++ * This is advanced only by a pong, so it is safe to age a socket against. ++ */ ++ lastPongAt = Date.now() + subscriptions: { [id: string]: ISubscription } = {} + handlers: ISocketMessageHandler[] = [] + config: ISocketOptions | any +@@ -55,6 +63,7 @@ export class Socket extends EventEmitter { connection?: WebSocket session?: string logger: ILogger @@ -10,7 +27,16 @@ index 19d31ae..068b61e 100644 /** Create a websocket handler */ constructor ( -@@ -82,18 +83,13 @@ export class Socket extends EventEmitter { +@@ -77,23 +86,22 @@ export class Socket extends EventEmitter { + this.send({ msg: 'pong' }).then(this.logger.debug, this.logger.error) + }) + ++ this.on('pong', () => { ++ this.lastPongAt = Date.now() ++ }) ++ + this.on('result', (data: any) => this.emit(data.id, { id: data.id, result: data.result, error: data.error })) + this.on('ready', (data: any) => this.emit(data.subs[0], data)) } /** @@ -33,7 +59,7 @@ index 19d31ae..068b61e 100644 try { connection = new WebSocket(this.host, null, { headers: settings.customHeaders }) connection.onerror = reject -@@ -101,14 +97,53 @@ export class Socket extends EventEmitter { +@@ -101,14 +109,53 @@ export class Socket extends EventEmitter { this.logger.error(err) return reject(err) } @@ -88,7 +114,17 @@ index 19d31ae..068b61e 100644 /** Send handshake message to confirm connection, start pinging. */ onOpen = async (callback: Function) => { this.lastPing = Date.now() -@@ -125,7 +160,14 @@ export class Socket extends EventEmitter { +@@ -119,13 +166,24 @@ export class Socket extends EventEmitter { + support: ['1', 'pre2', 'pre1'] + }) + this.session = connected.session ++ // The handshake reply is a server round trip, so it proves liveness exactly as a ++ // pong does. Without this, a freshly opened socket would look stale until its ++ // first ping lands and any foreground event would reopen it for nothing. ++ this.lastPongAt = Date.now() + this.ping().catch((err) => this.logger.error(`[ddp] Unable to ping server: ${err.message}`)) + this.emit('open') + return callback(this.connection) } /** Emit close event so it can be used for promise resolve in close() */ @@ -104,7 +140,7 @@ index 19d31ae..068b61e 100644 this.emit('close', e) try { if (e?.code !== userDisconnectCloseCode) { -@@ -201,6 +243,85 @@ export class Socket extends EventEmitter { +@@ -201,6 +259,103 @@ export class Socket extends EventEmitter { }, this.config.reopen); } @@ -123,6 +159,9 @@ index 19d31ae..068b61e 100644 + this.reopenPromise = new Promise(resolve => { + this.openTimeout && clearTimeout(this.openTimeout as any) + this.lastPing = 0 ++ // The socket being replaced must not leave its liveness proof behind for the ++ // new one to inherit. ++ this.lastPongAt = 0 + this.emit('disconnected') + + let settled = false @@ -147,7 +186,19 @@ index 19d31ae..068b61e 100644 + + /** + * Bounded liveness check for a socket in the gray zone. Returns true only if -+ * the socket is open and the server answers the ping within the deadline. ++ * the socket is open and the server answers *this* ping within the deadline. ++ * ++ * The ping carries an id and only the pong echoing it counts. Without that ++ * correlation any inbound pong passed the check, and after the OS unfreezes a ++ * backgrounded app it flushes the frames buffered during the freeze — so a pong ++ * answering a pre-freeze ping vouched for a session the server had already ++ * dropped, and the socket was never reopened. A timestamp comparison cannot ++ * close this: `onMessage` refreshes `lastPing` for every inbound frame, so it ++ * only proves something arrived, not that the server answered us. ++ * ++ * Note `on`, not `once`: `once` detaches before invoking its listener, so an ++ * uncorrelated pong would consume the registration and the real answer would ++ * arrive with nothing listening. + */ + probe = (timeoutMs = 2000): Promise => { + return new Promise(resolve => { @@ -155,7 +206,10 @@ index 19d31ae..068b61e 100644 + return resolve(false) + } + -+ const lastPingAtStart = this.lastPing ++ // A dedicated counter: probe ids must never collide with the `ddp-N` ids ++ // `send` hands out to subscriptions and method calls. ++ const probeId = `probe-${this.probed}` ++ this.probed += 1 + + let settled = false + const cleanup = () => { @@ -165,13 +219,13 @@ index 19d31ae..068b61e 100644 + if (timeout) clearTimeout(timeout as any) + } + -+ const onPong = () => { -+ if (this.lastPing <= lastPingAtStart) return ++ const onPong = (data: any) => { ++ if (data?.id !== probeId) return + cleanup() + resolve(true) + } + -+ this.once('pong', onPong) ++ this.on('pong', onPong) + + const timeout = setTimeout(() => { + cleanup() @@ -179,7 +233,7 @@ index 19d31ae..068b61e 100644 + }, timeoutMs) + + try { -+ this.connection.send(JSON.stringify({ msg: 'ping' })) ++ this.connection.send(JSON.stringify({ msg: 'ping', id: probeId })) + } catch { + cleanup() + resolve(false) @@ -190,7 +244,7 @@ index 19d31ae..068b61e 100644 /** Check if websocket connected and ready. */ get connected () { return !!( -@@ -254,7 +375,7 @@ export class Socket extends EventEmitter { +@@ -254,7 +409,7 @@ export class Socket extends EventEmitter { return resolve() } this.once(listener, (result: any) => { @@ -199,7 +253,7 @@ index 19d31ae..068b61e 100644 return (result.error ? reject(result.error) : resolve({ ...(/connect|ping|pong/.test(obj.msg) ? {} : { id }) , ...result })) }) }) -@@ -447,7 +568,7 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { +@@ -447,7 +602,7 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { ...config, ...moreConfigs, host: host.replace(/(^\w+:|^)\/\//, ''), @@ -208,7 +262,7 @@ index 19d31ae..068b61e 100644 // reopen: number // ping: number // close: number -@@ -503,6 +624,22 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { +@@ -503,6 +658,26 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { return this.ddp.checkAndReopen() } @@ -224,6 +278,10 @@ index 19d31ae..068b61e 100644 + return this.ddp.lastPing + } + ++ get lastPongAt (): number { ++ return this.ddp.lastPongAt ++ } ++ + get pingInterval (): number { + return this.ddp.config.ping + } @@ -231,7 +289,7 @@ index 19d31ae..068b61e 100644 subscribe = (topic: string, eventname: string, ...args: any[]): Promise => { this.logger.info(`[DDP driver] Subscribing to ${topic} | ${JSON.stringify(args)}`) return this.ddp.subscribe(topic, [eventname, { 'useCollection': false, 'args': args }]) -@@ -549,10 +686,70 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { +@@ -549,10 +724,70 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { 'uiInteraction', 'e2ekeyRequest', 'userData',