feat(#2458): WebSocket through HTTP/2 (#4540)

* feat(#2458): WebSocket through HTTP/2

* refactor: adjust upgrade testing

* test: add testing

* fix: fixup

* fix: fixup!

* feat: implement busy for h2

* fix: fixup!

* fix: fixup

* feat: adjust upgrade for H2

* feat: extend Websocket to handle h2

* feat: implement experimental warning

* refactor: Update lib/web/websocket/connection.js

Co-authored-by: Khafra <[email protected]>

* refactor: Update lib/web/websocket/connection.js

Co-authored-by: Khafra <[email protected]>

* refactor: code review

* docs: add documentation warning for it

* fix: compatibility with backpressure

* fix: bad merge

* test: fix order of close

* docs: add documentation

* docs: Update docs/docs/api/WebSocket.md

Co-authored-by: Khafra <[email protected]>

* test: flakiness?

---------

Co-authored-by: Khafra <[email protected]>

Author: metcoder95

docs/docs/api/WebSocket.md

@@ -34,6 +34,27 @@ import { WebSocket } from 'undici'
 const ws = new WebSocket('wss://echo.websocket.events', ['echo', 'chat'])
 ```
 
+### Example with HTTP/2:
+
+> ⚠️ Warning: WebSocket over HTTP/2 is experimental, it is likely to change in the future.
+
+> 🗒️ Note: WebSocket over HTTP/2 may be enabled by default in a future version,
+> this will happen by enabling HTTP/2 connections as the default behavior of Undici's Agent as well the global dispatcher.
+> Stay tuned to the changelog for more information.
+
+This example will not work in browsers or other platforms that don't allow passing an object.
+
+```mjs
+import { Agent } from 'undici'
+
+const agent = new Agent({ allowH2: true })
+
+const ws = new WebSocket('wss://echo.websocket.events', {
+  dispatcher: agent,
+  protocols: ['echo', 'chat']
+})
+```
+
 # Class: WebSocketStream
 
 > ⚠️ Warning: the WebSocketStream API has not been finalized and is likely to change.

lib/api/api-upgrade.js

@@ -4,6 +4,7 @@ const { InvalidArgumentError, SocketError } = require('../core/errors')
 const { AsyncResource } = require('node:async_hooks')
 const assert = require('node:assert')
 const util = require('../core/util')
+const { kHTTP2Stream } = require('../core/symbols')
 const { addSignal, removeSignal } = require('./abort-signal')
 
 class UpgradeHandler extends AsyncResource {
@@ -50,7 +51,7 @@ class UpgradeHandler extends AsyncResource {
   }
 
   onUpgrade (statusCode, rawHeaders, socket) {
-    assert(statusCode === 101)
+    assert(socket[kHTTP2Stream] === true ? statusCode === 200 : statusCode === 101)
 
     const { callback, opaque, context } = this
 

lib/core/symbols.js

@@ -62,6 +62,9 @@ module.exports = {
   kListeners: Symbol('listeners'),
   kHTTPContext: Symbol('http context'),
   kMaxConcurrentStreams: Symbol('max concurrent streams'),
+  kEnableConnectProtocol: Symbol('http2session connect protocol'),
+  kRemoteSettings: Symbol('http2session remote settings'),
+  kHTTP2Stream: Symbol('http2session client stream'),
   kNoProxyAgent: Symbol('no proxy agent'),
   kHttpProxyAgent: Symbol('http proxy agent'),
   kHttpsProxyAgent: Symbol('https proxy agent')

lib/dispatcher/client-h2.js

@@ -7,7 +7,8 @@ const {
   RequestContentLengthMismatchError,
   RequestAbortedError,
   SocketError,
-  InformationalError
+  InformationalError,
+  InvalidArgumentError
 } = require('../core/errors.js')
 const {
   kUrl,
@@ -28,7 +29,10 @@ const {
   kSize,
   kHTTPContext,
   kClosed,
-  kBodyTimeout
+  kBodyTimeout,
+  kEnableConnectProtocol,
+  kRemoteSettings,
+  kHTTP2Stream
 } = require('../core/symbols.js')
 const { channels } = require('../core/diagnostics.js')
 
@@ -53,7 +57,10 @@ const {
     HTTP2_HEADER_SCHEME,
     HTTP2_HEADER_CONTENT_LENGTH,
     HTTP2_HEADER_EXPECT,
-    HTTP2_HEADER_STATUS
+    HTTP2_HEADER_STATUS,
+    HTTP2_HEADER_PROTOCOL,
+    NGHTTP2_REFUSED_STREAM,
+    NGHTTP2_CANCEL
   }
 } = http2
 
@@ -93,12 +100,21 @@ function connectH2 (client, socket) {
   session[kClient] = client
   session[kSocket] = socket
   session[kHTTP2Session] = null
+  // We set it to true by default in a best-effort; however once connected to an H2 server
+  // we will check if extended CONNECT protocol is supported or not
+  // and set this value accordingly.
+  session[kEnableConnectProtocol] = false
+  // States whether or not we have received the remote settings from the server
+  session[kRemoteSettings] = false
 
   util.addListener(session, 'error', onHttp2SessionError)
   util.addListener(session, 'frameError', onHttp2FrameError)
   util.addListener(session, 'end', onHttp2SessionEnd)
   util.addListener(session, 'goaway', onHttp2SessionGoAway)
   util.addListener(session, 'close', onHttp2SessionClose)
+  util.addListener(session, 'remoteSettings', onHttp2RemoteSettings)
+  // TODO (@metcoder95): implement SETTINGS support
+  // util.addListener(session, 'localSettings', onHttp2RemoteSettings)
 
   session.unref()
 
@@ -115,23 +131,67 @@ function connectH2 (client, socket) {
   return {
     version: 'h2',
     defaultPipelining: Infinity,
+    /**
+     * @param {import('../core/request.js')} request
+     * @returns {boolean}
+    */
     write (request) {
       return writeH2(client, request)
     },
+    /**
+     * @returns {void}
+     */
     resume () {
       resumeH2(client)
     },
+    /**
+     * @param {Error | null} err
+     * @param {() => void} callback
+     */
     destroy (err, callback) {
       if (socket[kClosed]) {
         queueMicrotask(callback)
       } else {
         socket.destroy(err).on('close', callback)
       }
     },
+    /**
+     * @type {boolean}
+     */
     get destroyed () {
       return socket.destroyed
     },
-    busy () {
+    /**
+     * @param {import('../core/request.js')} request
+     * @returns {boolean}
+    */
+    busy (request) {
+      if (request != null) {
+        if (client[kRunning] > 0) {
+          // We are already processing requests
+
+          // Non-idempotent request cannot be retried.
+          // Ensure that no other requests are inflight and
+          // could cause failure.
+          if (request.idempotent === false) return true
+          // Don't dispatch an upgrade until all preceding requests have completed.
+          // Possibly, we do not have remote settings confirmed yet.
+          if ((request.upgrade === 'websocket' || request.method === 'CONNECT') && session[kRemoteSettings] === false) return true
+          // Request with stream or iterator body can error while other requests
+          // are inflight and indirectly error those as well.
+          // Ensure this doesn't happen by waiting for inflight
+          // to complete before dispatching.
+
+          // Request with stream or iterator body cannot be retried.
+          // Ensure that no other requests are inflight and
+          // could cause failure.
+          if (util.bodyLength(request.body) !== 0 &&
+            (util.isStream(request.body) || util.isAsyncIterable(request.body) || util.isFormDataLike(request.body))) return true
+        } else {
+          return (request.upgrade === 'websocket' || request.method === 'CONNECT') && session[kRemoteSettings] === false
+        }
+      }
+
       return false
     }
   }
@@ -151,6 +211,27 @@ function resumeH2 (client) {
   }
 }
 
+function onHttp2RemoteSettings (settings) {
+  // Fallbacks are a safe bet, remote setting will always override
+  this[kClient][kMaxConcurrentStreams] = settings.maxConcurrentStreams ?? this[kClient][kMaxConcurrentStreams]
+  /**
+   * From RFC-8441
+   * A sender MUST NOT send a SETTINGS_ENABLE_CONNECT_PROTOCOL parameter
+   * with the value of 0 after previously sending a value of 1.
+   */
+  // Note: Cannot be tested in Node, it does not supports disabling the extended CONNECT protocol once enabled
+  if (this[kRemoteSettings] === true && this[kEnableConnectProtocol] === true && settings.enableConnectProtocol === false) {
+    const err = new InformationalError('HTTP/2: Server disabled extended CONNECT protocol against RFC-8441')
+    this[kSocket][kError] = err
+    this[kClient][kOnError](err)
+    return
+  }
+
+  this[kEnableConnectProtocol] = settings.enableConnectProtocol ?? this[kEnableConnectProtocol]
+  this[kRemoteSettings] = true
+  this[kClient][kResume]()
+}
+
 function onHttp2SessionError (err) {
   assert(err.code !== 'ERR_TLS_CERT_ALTNAME_INVALID')
 
@@ -282,8 +363,8 @@ function writeH2 (client, request) {
   const { method, path, host, upgrade, expectContinue, signal, protocol, headers: reqHeaders } = request
   let { body } = request
 
-  if (upgrade) {
-    util.errorRequest(client, request, new Error('Upgrade not supported for H2'))
+  if (upgrade != null && upgrade !== 'websocket') {
+    util.errorRequest(client, request, new InvalidArgumentError(`Custom upgrade "${upgrade}" not supported over HTTP/2`))
     return false
   }
 
@@ -364,13 +445,68 @@ function writeH2 (client, request) {
     return false
   }
 
-  if (method === 'CONNECT') {
+  if (upgrade || method === 'CONNECT') {
     session.ref()
-    // We are already connected, streams are pending, first request
+
+    if (upgrade === 'websocket') {
+      // We cannot upgrade to websocket if extended CONNECT protocol is not supported
+      if (session[kEnableConnectProtocol] === false) {
+        util.errorRequest(client, request, new InformationalError('HTTP/2: Extended CONNECT protocol not supported by server'))
+        session.unref()
+        return false
+      }
+
+      // We force the method to CONNECT
+      // as per RFC-8441
+      // https://datatracker.ietf.org/doc/html/rfc8441#section-4
+      headers[HTTP2_HEADER_METHOD] = 'CONNECT'
+      headers[HTTP2_HEADER_PROTOCOL] = 'websocket'
+      // :path and :scheme headers must be omitted when sending CONNECT but set if extended-CONNECT
+      headers[HTTP2_HEADER_PATH] = path
+
+      if (protocol === 'ws:' || protocol === 'wss:') {
+        headers[HTTP2_HEADER_SCHEME] = protocol === 'ws:' ? 'http' : 'https'
+      } else {
+        headers[HTTP2_HEADER_SCHEME] = protocol === 'http:' ? 'http' : 'https'
+      }
+
+      stream = session.request(headers, { endStream: false, signal })
+      stream[kHTTP2Stream] = true
+
+      stream.once('response', (headers, _flags) => {
+        const { [HTTP2_HEADER_STATUS]: statusCode, ...realHeaders } = headers
+
+        request.onUpgrade(statusCode, parseH2Headers(realHeaders), stream)
+
+        ++session[kOpenStreams]
+        client[kQueue][client[kRunningIdx]++] = null
+      })
+
+      stream.on('error', () => {
+        if (stream.rstCode === NGHTTP2_REFUSED_STREAM || stream.rstCode === NGHTTP2_CANCEL) {
+          // NGHTTP2_REFUSED_STREAM (7) or NGHTTP2_CANCEL (8)
+          // We do not treat those as errors as the server might
+          // not support websockets and refuse the stream
+          abort(new InformationalError(`HTTP/2: "stream error" received - code ${stream.rstCode}`))
+        }
+      })
+
+      stream.once('close', () => {
+        session[kOpenStreams] -= 1
+        if (session[kOpenStreams] === 0) session.unref()
+      })
+
+      stream.setTimeout(requestTimeout)
+      return true
+    }
+
+    // TODO: consolidate once we support CONNECT properly
+    // NOTE: We are already connected, streams are pending, first request
     // will create a new stream. We trigger a request to create the stream and wait until
     // `ready` event is triggered
     // We disabled endStream to allow the user to write to the stream
     stream = session.request(headers, { endStream: false, signal })
+    stream[kHTTP2Stream] = true
     stream.on('response', headers => {
       const { [HTTP2_HEADER_STATUS]: statusCode, ...realHeaders } = headers
 
@@ -389,7 +525,6 @@ function writeH2 (client, request) {
 
   // https://tools.ietf.org/html/rfc7540#section-8.3
   // :path and :scheme headers must be omitted when sending CONNECT
-
   headers[HTTP2_HEADER_PATH] = path
   headers[HTTP2_HEADER_SCHEME] = protocol === 'http:' ? 'http' : 'https'
 
@@ -469,13 +604,15 @@ function writeH2 (client, request) {
   if (expectContinue) {
     headers[HTTP2_HEADER_EXPECT] = '100-continue'
     stream = session.request(headers, { endStream: shouldEndStream, signal })
+    stream[kHTTP2Stream] = true
 
     stream.once('continue', writeBodyH2)
   } else {
     stream = session.request(headers, {
       endStream: shouldEndStream,
       signal
     })
+    stream[kHTTP2Stream] = true
 
     writeBodyH2()
   }

lib/dispatcher/h2c-client.js

@@ -85,7 +85,6 @@ class H2CClient extends DispatcherBase {
         .setNoDelay(true)
         .once('connect', function () {
           queueMicrotask(clearConnectTimeout)
-
           if (callback) {
             const cb = callback
             callback = null
@@ -94,7 +93,6 @@ class H2CClient extends DispatcherBase {
         })
         .on('error', function (err) {
           queueMicrotask(clearConnectTimeout)
-
           if (callback) {
             const cb = callback
             callback = null

lib/web/fetch/index.js

@@ -2228,8 +2228,10 @@ async function httpNetworkFetch (
         },
 
         onUpgrade (status, rawHeaders, socket) {
-          if (status !== 101) {
-            return
+          // We need to support 200 for websocket over h2 as per RFC-8441
+          // Absence of session means H1
+          if ((socket.session != null && status !== 200) || (socket.session == null && status !== 101)) {
+            return false
           }
 
           const headersList = new HeadersList()

lib/web/websocket/connection.js

@@ -14,6 +14,8 @@ const crypto = runtimeFeatures.has('crypto')
   ? require('node:crypto')
   : null
 
+let warningEmitted = false
+
 /**
  * @see https://websockets.spec.whatwg.org/#concept-websocket-establish
  * @param {URL} url
@@ -93,9 +95,25 @@ function establishWebSocketConnection (url, protocols, client, handler, options)
     processResponse (response) {
       // 1. If response is a network error or its status is not 101,
       //    fail the WebSocket connection.
+      // if (response.type === 'error' || ((response.socket?.session != null && response.status !== 200) && response.status !== 101)) {
       if (response.type === 'error' || response.status !== 101) {
-        failWebsocketConnection(handler, 1002, 'Received network error or non-101 status code.', response.error)
-        return
+        // The presence of a session property on the socket indicates HTTP2
+        // HTTP1
+        if (response.socket?.session == null) {
+          failWebsocketConnection(handler, 1002, 'Received network error or non-101 status code.', response.error)
+          return
+        }
+
+        // HTTP2
+        if (response.status !== 200) {
+          failWebsocketConnection(handler, 1002, 'Received network error or non-200 status code.', response.error)
+          return
+        }
+      }
+
+      if (warningEmitted === false && response.socket?.session != null) {
+        process.emitWarning('WebSocket over HTTP2 is experimental, and subject to change.', 'ExperimentalWarning')
+        warningEmitted = true
       }
 
       // 2. If protocols is not the empty list and extracting header
@@ -117,7 +135,8 @@ function establishWebSocketConnection (url, protocols, client, handler, options)
       //    header field contains a value that is not an ASCII case-
       //    insensitive match for the value "websocket", the client MUST
       //    _Fail the WebSocket Connection_.
-      if (response.headersList.get('Upgrade')?.toLowerCase() !== 'websocket') {
+      //    For H2, no upgrade header is expected.
+      if (response.socket.session == null && response.headersList.get('Upgrade')?.toLowerCase() !== 'websocket') {
         failWebsocketConnection(handler, 1002, 'Server did not set Upgrade header to "websocket".')
         return
       }
@@ -126,7 +145,8 @@ function establishWebSocketConnection (url, protocols, client, handler, options)
       //    |Connection| header field doesn't contain a token that is an
       //    ASCII case-insensitive match for the value "Upgrade", the client
       //    MUST _Fail the WebSocket Connection_.
-      if (response.headersList.get('Connection')?.toLowerCase() !== 'upgrade') {
+      //    For H2, no connection header is expected.
+      if (response.socket.session == null && response.headersList.get('Connection')?.toLowerCase() !== 'upgrade') {
         failWebsocketConnection(handler, 1002, 'Server did not set Connection header to "upgrade".')
         return
       }