fix: Unhandled "WebSocket connection closed" when CDP connection is unstable

cli/CHANGELOG.md

@@ -5,6 +5,7 @@ _Released 7/16/2024 (PENDING)_
 
 **Bugfixes:**
 
+- Fixed an issue where unhandled `WebSocket connection closed` exceptions would be thrown when CDP connections rapidly connect, disconnect, and connect again while there are pending commands. Fixes [#29572](https://github.com/cypress-io/cypress/issues/29572).
 - CLI output properly displays non-JSON response bodies when a Test Replay upload attempt returns a non-JSON response body for a non-200 status code. Addressed in [#29801](https://github.com/cypress-io/cypress/pull/29801).
 - Fixed an issue where the ReadStream used to upload a Test Replay recording could erroneously be re-used when retrying in cases of retryable upload failures. Fixes [#29227](https://github.com/cypress-io/cypress/issues/29227).
 - Fixed an issue where command snapshots were not being captured within the `cy.origin()` command within Test Replay. Addressed in [#29828](https://github.com/cypress-io/cypress/pull/29828).

packages/server/lib/browsers/cdp-command-queue.ts

@@ -0,0 +1,86 @@
+import type ProtocolMapping from 'devtools-protocol/types/protocol-mapping'
+import pDefer, { DeferredPromise } from 'p-defer'
+import type { CdpCommand } from './cdp_automation'
+import Debug from 'debug'
+
+const debug = Debug('cypress:server:browsers:cdp-command-queue')
+const debugVerbose = Debug('cypress:server:browsers:cd-command-queue')
+
+type CommandReturn<T extends CdpCommand> = ProtocolMapping.Commands[T]['returnType']
+
+export type Command<T extends CdpCommand> = {
+  command: T
+  params?: object
+  deferred: DeferredPromise<CommandReturn<T>>
+  sessionId?: string
+}
+
+export class CDPCommandQueue {
+  private queue: Command<any>[] = []
+
+  public get entries () {
+    return [...this.queue]
+  }
+
+  public add <TCmd extends CdpCommand> (
+    command: TCmd,
+    params: ProtocolMapping.Commands[TCmd]['paramsType'][0],
+    sessionId?: string,
+  ): Promise<CommandReturn<TCmd>> {
+    debug('enqueing command %s', command)
+    debugVerbose('enqueing command %s with params %o', command, params)
+
+    const deferred = pDefer<CommandReturn<TCmd>>()
+
+    const commandPackage: Command<TCmd> = {
+      command,
+      params,
+      deferred,
+      sessionId,
+    }
+
+    this.queue.push(commandPackage)
+
+    debug('Command enqueued; new length: %d', this.queue.length)
+    debugVerbose('Queue Contents: %O', this.queue)
+
+    return deferred.promise
+  }
+
+  public clear () {
+    debug('clearing command queue')
+    this.queue = []
+  }
+
+  public extract<T extends CdpCommand> (search: Partial<Command<T>>): Command<T> | undefined {
+    // this should find, remove, and return if found a given command
+
+    const index = this.queue.findIndex((enqueued) => {
+      for (let k of Object.keys(search)) {
+        if (search[k] !== enqueued[k]) {
+          return false
+        }
+      }
+
+      return true
+    })
+
+    debug('extracting %o from commands at index %d', search, index)
+
+    if (index === -1) {
+      return undefined
+    }
+
+    const [extracted] = this.queue.splice(index, 1)
+
+    return extracted
+  }
+
+  public shift () {
+    return this.queue.shift()
+  }
+
+  public unshift (value: Command<any>) {
+    return this.queue.unshift(value)
+  }
+}