perf: cri client memory leak (#29988)

* Refactor connection logic to cri-connection

CDPConnection class instantiates a persistent event emitter, and creates
a CDP Client instance via connect(). When this client emits a disconnect
event, the CDPConnection will automatically attempt to reconnect, if the
connection was not intentionally closed and the connection is created with
the reconnect flag. It only listens to 'event' and 'disconnect' events itself,
making it easy to unsubscribe from events in the event of a disconnection,
helping ease mental load when considering potential memory leaks.

Because it uses its own persistent event emitter, event listeners added
to the CDPConnection do not need to be re-added when the client reconnects.

The CriClient itself no longer handles reconnection logic: it only reacts to
connection state changes in order to re-send failed commands and enablements
when the CDPConnection restores the CDPClient.

* changelog

* clean up event listeners before reconnecting

* changelog

* changelog

* changelog

* unit tests

* lift out of the remote-interface dir

* complete lift from /remote-interface

* further fix imports

* import

* fix disconnect behavior to make integration test pass

* Update packages/server/test/integration/cdp_spec.ts comment

* fix snapgen

* improve debug

* further debug improvements

* do not attach crash handlers on browser level cri clients, add tests

* adds bindingCalled listeners discretely, rather than relying on 'event', which does not trigger for them

* back out of main `event` listener, use discrete listeners, add integration test for service worker bindingCalled

* Revert "back out of main `event` listener, use discrete listeners, add integration test for service worker bindingCalled"

This reverts commit 4855120.

* changelog

* adds integration test for various ways to add cdp event listeners

* note in changelog this is a cypress server fix

* ensure enablement promises resolve after reconnect

* Update cli/CHANGELOG.md

Co-authored-by: Matt Schile <mschile@cypress.io>

---------

Co-authored-by: Matt Schile <mschile@cypress.io>

cli/CHANGELOG.md

@@ -3,6 +3,10 @@
 
 _Released 8/27/2024 (PENDING)_
 
+**Performance:**
+
+- Fixed a potential memory leak in the Cypress server when re-connecting to an unintentionally disconnected CDP connection. Fixes [#29744](/~https://github.com/cypress-io/cypress/issues/29744). Addressed in [#29988](/~https://github.com/cypress-io/cypress/pull/29988)
+
 **Dependency Updates:**
 
 - Updated `detect-port` from `1.3.0` to `1.6.1`. Addressed in [#30038](/~https://github.com/cypress-io/cypress/pull/30038).

packages/server/lib/browsers/browser-cri-client.ts

@@ -5,6 +5,7 @@ import Debug from 'debug'
 import type { Protocol } from 'devtools-protocol'
 import { _connectAsync, _getDelayMsForRetry } from './protocol'
 import * as errors from '../errors'
+import type { CypressError } from '@packages/errors'
 import { CriClient, DEFAULT_NETWORK_ENABLE_OPTIONS } from './cri-client'
 import { serviceWorkerClientEventHandler, serviceWorkerClientEventHandlerName } from '@packages/proxy/lib/http/util/service-worker-manager'
 import type { ProtocolManagerShape } from '@packages/types'
@@ -23,7 +24,7 @@ type BrowserCriClientOptions = {
   host: string
   port: number
   browserName: string
-  onAsynchronousError: Function
+  onAsynchronousError: (err: CypressError) => void
   protocolManager?: ProtocolManagerShape
   fullyManageTabs?: boolean
   onServiceWorkerClientEvent: ServiceWorkerEventHandler
@@ -33,7 +34,7 @@ type BrowserCriClientCreateOptions = {
   browserName: string
   fullyManageTabs?: boolean
   hosts: string[]
-  onAsynchronousError: Function
+  onAsynchronousError: (err: CypressError) => void
   onReconnect?: (client: CriClient) => void
   port: number
   protocolManager?: ProtocolManagerShape
@@ -181,7 +182,7 @@ export class BrowserCriClient {
   private host: string
   private port: number
   private browserName: string
-  private onAsynchronousError: Function
+  private onAsynchronousError: (err: CypressError) => void
   private protocolManager?: ProtocolManagerShape
   private fullyManageTabs?: boolean
   onServiceWorkerClientEvent: ServiceWorkerEventHandler
@@ -479,7 +480,7 @@ export class BrowserCriClient {
       browserCriClient.onClose = resolve
 
       // or when the browser's CDP ws connection is closed
-      browserClient.ws.once('close', () => {
+      browserClient.ws?.once('close', () => {
         resolve(false)
       })
     })

packages/server/lib/browsers/cdp-connection.ts

@@ -0,0 +1,233 @@
+import type ProtocolMapping from 'devtools-protocol/types/protocol-mapping'
+import Debug from 'debug'
+import EventEmitter from 'events'
+import CDP from 'chrome-remote-interface'
+import type { CypressError } from '@packages/errors'
+import { debugCdpConnection, DebuggableCDPClient } from './debug-cdp-connection'
+import type { CdpEvent, CdpCommand } from './cdp_automation'
+import { CDPDisconnectedError, CDPTerminatedError, CDPAlreadyConnectedError } from './cri-errors'
+import { asyncRetry } from '../util/async_retry'
+import * as errors from '../errors'
+import type WebSocket from 'ws'
+
+const verboseDebugNs = 'cypress-verbose:server:browsers:cdp-connection'
+
+export type CDPListener<T extends keyof ProtocolMapping.Events> = (params: ProtocolMapping.Events[T][0], sessionId?: string) => void
+
+// CDPClient extends EventEmitter, but does not export that type information from its
+// definitelytyped module
+export type CdpClient = Exclude<EventEmitter, CDP.Client> & CDP.Client
+
+/**
+ * There are three error messages we can encounter which should not be re-thrown, but
+ * should trigger a reconnection attempt if one is not in progress, and enqueue the
+ * command that errored. This regex is used in client.send to check for:
+ * - WebSocket connection closed
+ * - WebSocket not open
+ * - WebSocket already in CLOSING or CLOSED state
+ */
+const isWebsocketClosedErrorMessage = (message: string) => {
+  return /^WebSocket (?:connection closed|is (?:not open|already in CLOSING or CLOSED state))/.test(message)
+}
+
+export type CDPConnectionOptions = {
+  automaticallyReconnect: boolean
+}
+
+type CDPConnectionEventListeners = {
+  'cdp-connection-reconnect-error': (err: CypressError) => void
+  'cdp-connection-reconnect': () => void
+  'cdp-connection-closed': () => void
+  'cdp-connection-reconnect-attempt': (attemptNumber: number) => void
+}
+export type CDPConnectionEvent = keyof CDPConnectionEventListeners
+
+type CDPConnectionEventListener<T extends CDPConnectionEvent> = CDPConnectionEventListeners[T]
+
+export class CDPConnection {
+  private _emitter: EventEmitter = new EventEmitter()
+  private _connection: CdpClient | undefined
+  private _autoReconnect: boolean
+  private _terminated: boolean = false
+  private _reconnection: Promise<void> | undefined
+  private debug: Debug.Debugger
+  private verboseDebug: Debug.Debugger
+
+  constructor (private readonly _options: CDP.Options, connectionOptions: CDPConnectionOptions) {
+    this._autoReconnect = connectionOptions.automaticallyReconnect
+    this.debug = Debug(`cypress:server:browsers:cdp-connection:${_options.target}`)
+    this.verboseDebug = Debug(`${verboseDebugNs}:${_options.target}`)
+  }
+
+  get terminated () {
+    return this._terminated
+  }
+
+  get ws () {
+    // this is reached into by browser-cri-client to detect close events - needs rethinking
+    return (this._connection as { _ws?: WebSocket})._ws
+  }
+
+  on<T extends CdpEvent> (event: T, callback: CDPListener<T>) {
+    this.debug('attaching event listener to cdp connection', event)
+
+    this._emitter.on(event, callback)
+  }
+  addConnectionEventListener<T extends CDPConnectionEvent> (event: T, callback: CDPConnectionEventListener<T>) {
+    this.debug('adding connection event listener for ', event)
+    this._emitter.on(event, callback)
+  }
+  off<T extends CdpEvent> (event: T, callback: CDPListener<T>) {
+    this._emitter.off(event, callback)
+  }
+  removeConnectionEventListener<T extends CDPConnectionEvent> (event: T, callback: CDPConnectionEventListener<T>) {
+    this._emitter.off(event, callback)
+  }
+
+  async connect (): Promise<void> {
+    if (this._terminated) {
+      throw new CDPTerminatedError(`Cannot connect to CDP. Client target ${this._options.target} has been terminated.`)
+    }
+
+    if (this._connection) {
+      throw new CDPAlreadyConnectedError(`Cannot connect to CDP. Client target ${this._options.target} is already connected. Did you disconnect first?`)
+    }
+
+    this._connection = await CDP(this._options) as CdpClient
+
+    debugCdpConnection(this.verboseDebug.namespace, this._connection as DebuggableCDPClient)
+
+    this._connection.on('event', this._broadcastEvent)
+
+    if (this._autoReconnect) {
+      this._connection.on('disconnect', this._reconnect)
+    }
+  }
+
+  async disconnect () {
+    this.debug('disconnect of target %s requested.', this._options.target, { terminated: this._terminated, connection: !!this._connection, reconnection: !!this._reconnection })
+    if (this._terminated && !this._connection) {
+      return
+    }
+
+    this._terminated = true
+
+    if (this._connection) {
+      await this._gracefullyDisconnect()
+      this._emitter.emit('cdp-connection-closed')
+    }
+  }
+
+  private _gracefullyDisconnect = async () => {
+    this._connection?.off('event', this._broadcastEvent)
+    this._connection?.off('disconnect', this._reconnect)
+
+    await this._connection?.close()
+    this._connection = undefined
+  }
+
+  async send<T extends CdpCommand> (
+    command: T,
+    data?: ProtocolMapping.Commands[T]['paramsType'][0],
+    sessionId?: string,
+  ): Promise<ProtocolMapping.Commands[T]['returnType']> {
+    if (this.terminated) {
+      throw new CDPDisconnectedError(`${command} will not run as the CRI connection to Target ${this._options.target} has been terminated.`)
+    }
+
+    if (!this._connection) {
+      throw new CDPDisconnectedError(`${command} will not run as the CRI connection to Target ${this._options.target} has not been established.`)
+    }
+
+    try {
+      return await this._connection.send(command, data, sessionId)
+    } catch (e) {
+      // Clients may wish to determine if the command should be enqueued
+      // should enqueue logic live in this class tho??
+      if (isWebsocketClosedErrorMessage(e.message)) {
+        throw new CDPDisconnectedError(`${command} failed due to the websocket being disconnected.`, e)
+      }
+
+      throw e
+    }
+  }
+
+  private _reconnect = async () => {
+    this.debug('Reconnection requested')
+    if (this._terminated) {
+      return
+    }
+
+    if (this._reconnection) {
+      return this._reconnection
+    }
+
+    if (this._connection) {
+      try {
+        await this._gracefullyDisconnect()
+      } catch (e) {
+        this.debug('Error cleaning up existing CDP connection before creating a new connection: ', e)
+      } finally {
+        this._connection = undefined
+      }
+    }
+
+    let attempt = 0
+
+    this._reconnection = asyncRetry(async () => {
+      attempt++
+
+      this.debug('Reconnection attempt %d for Target %s', attempt, this._options.target)
+
+      if (this._terminated) {
+        this.debug('Not reconnecting, connection to %s has been terminated', this._options.target)
+        throw new CDPTerminatedError(`Cannot reconnect to CDP. Client target ${this._options.target} has been terminated.`)
+      }
+
+      this._emitter.emit('cdp-connection-reconnect-attempt', attempt)
+
+      await this.connect()
+    }, {
+      maxAttempts: 20,
+      retryDelay: () => {
+        return 100
+      },
+      shouldRetry (err) {
+        return !(err && CDPTerminatedError.isCDPTerminatedError(err))
+      },
+    })()
+
+    try {
+      await this._reconnection
+      this._emitter.emit('cdp-connection-reconnect')
+    } catch (err) {
+      this.debug('error(s) on reconnecting: ', err)
+      const significantError: Error = err.errors ? (err as AggregateError).errors[err.errors.length - 1] : err
+
+      const retryHaltedDueToClosed = CDPTerminatedError.isCDPTerminatedError(err) ||
+        (err as AggregateError)?.errors?.find((predicate) => CDPTerminatedError.isCDPTerminatedError(predicate))
+
+      // if .disconnect() was called while trying to reconnect, there will be no active connection
+      // so the .disconnect() method will not emit the connection closed event. However, we do
+      // want to emit that once the reconnection attempts cease due to being closed.
+      if (retryHaltedDueToClosed) {
+        this._emitter.emit('cdp-connection-closed')
+      } else {
+        const cdpError = errors.get('CDP_COULD_NOT_RECONNECT', significantError)
+
+        cdpError.isFatalApiErr = true
+        this._emitter.emit('cdp-connection-reconnect-error', cdpError)
+      }
+    }
+
+    this._reconnection = undefined
+  }
+
+  private _broadcastEvent = ({ method, params, sessionId }: { method: CdpEvent, params: Record<string, any>, sessionId?: string }) => {
+    this.verboseDebug('rebroadcasting event', method, params, sessionId)
+
+    this._emitter.emit('event', { method, params, sessionId })
+    this._emitter.emit(method, params, sessionId)
+    this._emitter.emit(`${method}.${sessionId}`, params)
+  }
+}

packages/server/lib/browsers/chrome.ts

@@ -445,7 +445,9 @@ export = {
     const browserCriClient = this._getBrowserCriClient()
 
     // Handle chrome tab crashes.
+    debug('attaching crash handler to target ', pageCriClient.targetId)
     pageCriClient.on('Target.targetCrashed', async (event) => {
+      debug('target crashed!', event)
       if (event.targetId !== browserCriClient?.currentlyAttachedTarget?.targetId) {
         return
       }