client-h2.js

'use strict'

const assert = require('node:assert')
const { pipeline } = require('node:stream')
const util = require('../core/util.js')
const {
  RequestContentLengthMismatchError,
  RequestAbortedError,
  SocketError,
  InformationalError,
  InvalidArgumentError,
  HeadersTimeoutError,
  BodyTimeoutError,
  ResponseExceededMaxSizeError
} = require('../core/errors.js')
const {
  kUrl,
  kReset,
  kClient,
  kRunning,
  kPending,
  kQueue,
  kPendingIdx,
  kRunningIdx,
  kError,
  kSocket,
  kStrictContentLength,
  kOnError,
  kMaxConcurrentStreams,
  kHTTP2Session,
  kHostAuthority,
  kResume,
  kSize,
  kHTTPContext,
  kClosed,
  kKeepAliveDefaultTimeout,
  kHeadersTimeout,
  kBodyTimeout,
  kEnableConnectProtocol,
  kRemoteSettings,
  kHTTP2Stream,
  kHTTP2SessionState,
  kHTTP2Options,
  kMaxResponseSize,
  kMaxRequests,
  kCounter
} = require('../core/symbols.js')
const { channels } = require('../core/diagnostics.js')

const kOpenStreams = Symbol('open streams')
const kRequestStreamId = Symbol('request stream id')
const kRequestStream = Symbol('request stream')
const kRequestStreamCleanup = Symbol('request stream cleanup')
const kRequestStreamState = Symbol('request stream state')
const kReceivedGoAway = Symbol('received goaway')
const kGoAwayReplayAttempts = Symbol('goaway replay attempts')
const kRefusedStreamRetry = Symbol('refused stream retry')
const kRetiringSession = Symbol('retiring session')

// RFC 9113 section 8.7: a client SHOULD NOT automatically retry a request more
// than once. Without a budget a peer that keeps refusing turns one request into
// an unbounded connect/refuse/reconnect loop that never settles and starves the
// event loop.
const MAX_GOAWAY_REPLAY_ATTEMPTS = 1

let extractBody

/** @type {import('http2')} */
let http2
try {
  http2 = require('node:http2')
} catch {
  // @ts-ignore
  http2 = { constants: {} }
}

const {
  constants: {
    HTTP2_HEADER_AUTHORITY,
    HTTP2_HEADER_METHOD,
    HTTP2_HEADER_PATH,
    HTTP2_HEADER_SCHEME,
    HTTP2_HEADER_CONTENT_LENGTH,
    HTTP2_HEADER_EXPECT,
    HTTP2_HEADER_STATUS,
    HTTP2_HEADER_PROTOCOL,
    NGHTTP2_NO_ERROR,
    NGHTTP2_REFUSED_STREAM
  }
} = http2

function getGoAwayError (session, errorCode) {
  return session[kError] ||
    (errorCode === NGHTTP2_NO_ERROR
      ? new InformationalError(`HTTP/2: "GOAWAY" frame received with code ${errorCode}`)
      : new SocketError(`HTTP/2: "GOAWAY" frame received with code ${errorCode}`, util.getSocketInfo(session[kSocket])))
}

function resetHttp2Session (session, err) {
  const client = session[kClient]
  const socket = session[kSocket]

  clearRetiredSessionTimeout(session)

  if (client[kHTTP2Session] === session) {
    client[kSocket] = null
    client[kHTTPContext] = null
    client[kHTTP2Session] = null
  }

  if (socket != null && socket[kError] == null) {
    socket[kError] = err
  }

  if (!session.closed && !session.destroyed) {
    try {
      session.destroy(err)
    } catch {}
  }

  util.destroy(socket, err)
}

function getGoAwayPendingIdx (client, lastStreamID) {
  const maxAcceptedStreamID = Number.isInteger(lastStreamID) ? lastStreamID : Number.MAX_SAFE_INTEGER

  for (let i = client[kRunningIdx]; i < client[kPendingIdx]; i++) {
    const request = client[kQueue][i]

    if (request == null) {
      continue
    }

    if (typeof request[kRequestStreamId] !== 'number' || request[kRequestStreamId] > maxAcceptedStreamID) {
      return i
    }
  }

  return client[kPendingIdx]
}

function detachRequestFromStream (request) {
  request[kRequestStreamId] = null
  request[kRequestStream] = null
  request[kRequestStreamCleanup] = null
}

function bindRequestToStream (request, stream, cleanup) {
  const previousCleanup = request[kRequestStreamCleanup]
  const previousStream = request[kRequestStream]
  detachRequestFromStream(request)
  previousCleanup?.(previousStream)
  request[kRequestStreamId] = stream.id
  request[kRequestStream] = stream
  request[kRequestStreamCleanup] = cleanup
}

function clearRequestStream (request) {
  const cleanup = request[kRequestStreamCleanup]
  const stream = request[kRequestStream]
  detachRequestFromStream(request)
  cleanup?.(stream)
}

function requeueUnsentRequest (client, request) {
  client[kQueue].splice(client[kPendingIdx] + 1, 0, request)
}

function completeRequest (client, request, resetPendingIdx = false) {
  const queue = client[kQueue]
  const runningIdx = client[kRunningIdx]

  // In-order completion: clear the request and advance without splicing.
  // The client's resume loop compacts cleared slots once the index grows.
  if (runningIdx < client[kPendingIdx] && queue[runningIdx] === request) {
    queue[runningIdx] = null
    client[kRunningIdx] = runningIdx + 1
    return
  }

  const index = queue.indexOf(request, runningIdx)

  if (index === -1 || index >= client[kPendingIdx]) {
    return
  }

  queue.splice(index, 1)
  client[kPendingIdx]--

  if (resetPendingIdx && client[kPendingIdx] < client[kRunningIdx]) {
    client[kPendingIdx] = client[kRunningIdx]
  }
}

function canReplayRequest (request) {
  const { body } = request

  return body == null || util.isBuffer(body) || util.isBlobLike(body)
}

// Count a GOAWAY refusal against the request's replay budget. A peer that
// refuses every connection must eventually surface an error to the caller
// rather than being retried forever. Kept separate from canReplayRequest so
// that the REFUSED_STREAM retry, which has its own single-attempt limit, does
// not consume this budget just by asking whether the body can be replayed.
function registerGoAwayRefusal (request) {
  const attempts = (request[kGoAwayReplayAttempts] ?? 0) + 1
  request[kGoAwayReplayAttempts] = attempts

  return attempts <= MAX_GOAWAY_REPLAY_ATTEMPTS
}

function closeStream (stream, code = NGHTTP2_REFUSED_STREAM) {
  if (stream != null && !stream.destroyed && !stream.closed) {
    try {
      stream.close(code)
    } catch {}
  }
}

function detachRequestStreamForClose (request) {
  const stream = request[kRequestStream]

  clearRequestStream(request)
  severRequestStream(stream)

  return stream
}

// Unbind a stream from its request for good. releaseRequestStream() alone
// leaves the 'close' listener attached and kRequestStreamState populated, so a
// stream abandoned here would still run completeRequestStream() later — and
// splice out the request that has since been requeued onto another session.
function severRequestStream (stream) {
  if (stream == null || stream[kRequestStreamState] == null) {
    return
  }

  stream[kRequestStreamState] = null
  stream.off('close', completeRequestStream)
  // Upgrade streams use their own close cleanup, which would otherwise release
  // the session a second time after the stream has been severed for GOAWAY.
  stream.off('close', onUpgradeStreamClose)

  if (stream[kHTTP2Session] != null) {
    closeStreamSession(stream)
  }

  if (!stream.destroyed && !stream.closed) {
    stream.once('error', noop)
  }
}

function connectH2 (client, socket) {
  client[kSocket] = socket

  const http2InitialWindowSize = client[kHTTP2Options].sessionOptions?.initialWindowSize
  const http2ConnectionWindowSize = client[kHTTP2Options].connectionWindowSize

  const session = http2.connect(client[kUrl], {
    createConnection: () => socket,
    peerMaxConcurrentStreams: client[kHTTP2Options].maxConcurrentStreams,
    settings: {
      // TODO(metcoder95): add support for PUSH
      enablePush: false,
      ...(http2InitialWindowSize != null ? { initialWindowSize: http2InitialWindowSize } : null)
    }
  })

  client[kSocket] = socket
  session[kOpenStreams] = 0
  session[kClient] = client
  session[kSocket] = socket
  session[kHTTP2SessionState] = {
    idleTimeout: null,
    // Armed while the peer advertises MAX_CONCURRENT_STREAMS = 0 and we have
    // work that cannot start. See setNoStreamsTimeout.
    noStreamsTimeout: null,
    // Set once the session has opened maxRequestsPerClient streams. A retired
    // session never accepts another stream; it drains and is then torn down.
    // See trackH2Stream.
    retired: false,
    // Bounds how long a retired session may wait for its accepted streams to
    // close before they are destroyed so queued work can reconnect.
    retiredSessionTimeout: null,
    // Number of streams counted against maxRequestsPerClient that have not
    // physically closed yet. Only maintained when the limit is enabled.
    countedStreams: 0,
    // Set when closeRetiredH2Session has already emitted 'disconnect' and
    // resumed the queue, so that the trailing socket close does not do it
    // again. See onHttp2SocketClose.
    disconnectAnnounced: false,
    // Sockets start out ref'd. Session ref/unref proxies to the socket, so a
    // single cached flag lets us skip redundant uv ref/unref calls, provided
    // every ref/unref of the session or its socket goes through
    // refH2Session/unrefH2Session.
    refed: true,
    ping: {
      interval: client[kHTTP2Options].pingInterval === 0 ? null : setInterval(onHttp2SendPing, client[kHTTP2Options].pingInterval, session).unref()
    }
  }
  session[kReceivedGoAway] = false
  // 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

  // Apply connection-level flow control once connected (if supported).
  if (http2ConnectionWindowSize) {
    util.addListener(session, 'connect', applyConnectionWindowSize.bind(session, http2ConnectionWindowSize))
  }

  util.addListener(session, 'error', onHttp2SessionError)
  util.addListener(session, 'frameError', onHttp2FrameError)
  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)

  unrefH2Session(session)

  client[kHTTP2Session] = session
  socket[kHTTP2Session] = session

  util.addListener(socket, 'error', onHttp2SocketError)
  util.addListener(socket, 'end', onHttp2SocketEnd)
  util.addListener(socket, 'close', onHttp2SocketClose)

  socket[kClosed] = false
  socket.on('close', onSocketClose)

  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
    },
    /**
     * @param {import('../core/request.js')} request
     * @returns {boolean}
    */
    busy (request) {
      // A session that has reached maxRequestsPerClient is draining: it must
      // not receive another stream, so queued work waits here until the session
      // is torn down and a fresh one replaces it.
      if (session[kHTTP2SessionState].retired === true) {
        return true
      }

      if (session[kRemoteSettings] === false && client[kRunning] > 0) {
        return true
      }

      if (client[kRunning] >= client[kMaxConcurrentStreams]) {
        return true
      }

      if (request != null) {
        if (client[kRunning] > 0) {
          // We are already processing requests

          // Unlike HTTP/1.1 pipelining, HTTP/2 multiplexes requests on
          // independent streams, so non-idempotent requests can be dispatched
          // concurrently. Retry eligibility is handled by stream/session error
          // handling instead of by serializing all non-idempotent requests.
          // 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
        } else {
          return (request.upgrade === 'websocket' || request.method === 'CONNECT') && session[kRemoteSettings] === false
        }
      }

      return false
    }
  }
}

// Session ref/unref proxies to the underlying socket, so refH2Session and
// unrefH2Session cover both and can skip the call when the cached ref state
// already matches.
function refH2Session (session) {
  const state = session[kHTTP2SessionState]

  if (state.refed === false) {
    state.refed = true
    session.ref()
  }
}

function unrefH2Session (session) {
  const state = session[kHTTP2SessionState]

  if (state.refed === true) {
    state.refed = false
    session.unref()
  }
}

function resumeH2 (client) {
  const socket = client[kSocket]
  const session = client[kHTTP2Session]

  if (socket?.destroyed === false) {
    if (session[kHTTP2SessionState].retired === true) {
      // A retired session must not be unref'ed or given an idle timeout: the
      // streams it already accepted may still be uploading. Teardown happens
      // in onCountedStreamClose once the last of them is physically closed.
      return
    }

    // After an upgrade the queue is empty but its stream is still in use, so never unref while a stream is open.
    if (session[kOpenStreams] === 0 && (client[kSize] === 0 || client[kMaxConcurrentStreams] === 0)) {
      unrefH2Session(session)
    } else {
      refH2Session(session)
    }

    if (client[kSize] === 0 && session[kOpenStreams] === 0) {
      setHttp2IdleTimeout(session)
    } else {
      clearHttp2IdleTimeout(session)
    }

    if (client[kMaxConcurrentStreams] === 0 && client[kRunning] === 0 && client[kPending] > 0) {
      setNoStreamsTimeout(session)
    } else {
      clearNoStreamsTimeout(session)
    }
  }
}

function clearNoStreamsTimeout (session) {
  const state = session[kHTTP2SessionState]

  if (state?.noStreamsTimeout != null) {
    clearTimeout(state.noStreamsTimeout)
    state.noStreamsTimeout = null
  }
}

function clearRetiredSessionTimeout (session) {
  const state = session[kHTTP2SessionState]

  if (state?.retiredSessionTimeout != null) {
    clearTimeout(state.retiredSessionTimeout)
    state.retiredSessionTimeout = null
  }
}

function setRetiredSessionTimeout (session) {
  const client = session[kClient]
  const state = session[kHTTP2SessionState]
  const timeout = client[kHeadersTimeout]

  if (!timeout || state.retiredSessionTimeout != null) {
    return
  }

  state.retiredSessionTimeout = setTimeout(onRetiredSessionTimeout, timeout, session).unref()
}

function onRetiredSessionTimeout (session) {
  const client = session[kClient]
  const state = session[kHTTP2SessionState]

  state.retiredSessionTimeout = null

  if (
    client[kHTTP2Session] !== session ||
    state.retired !== true ||
    state.countedStreams === 0 ||
    session.closed ||
    session.destroyed
  ) {
    return
  }

  closeRetiredH2Session(
    session,
    new InformationalError(
      `HTTP/2: retired session did not drain within ${client[kHeadersTimeout]}`
    )
  )
}

// A peer is allowed to advertise SETTINGS_MAX_CONCURRENT_STREAMS = 0 to refuse
// new streams (RFC 9113 §6.5.2), and is expected to raise it again later. Until
// it does, busy() reports the client as permanently busy and queued requests
// cannot open a stream — which means no per-stream timeout covers them, and no
// reconnect can happen either, so the SETTINGS frame that would lift the limit
// can never arrive. Give the peer headersTimeout to start honouring requests
// before failing them; a request that cannot even be sent has missed the same
// deadline as one whose headers never arrive.
function setNoStreamsTimeout (session) {
  const client = session[kClient]
  const state = session[kHTTP2SessionState]
  const timeout = client[kHeadersTimeout]

  if (!timeout || state.noStreamsTimeout != null) {
    return
  }

  state.noStreamsTimeout = setTimeout(onNoStreamsTimeout, timeout, session).unref()
}

function onNoStreamsTimeout (session) {
  const client = session[kClient]
  const state = session[kHTTP2SessionState]

  state.noStreamsTimeout = null

  if (
    client[kHTTP2Session] !== session ||
    client[kMaxConcurrentStreams] !== 0 ||
    client[kRunning] !== 0 ||
    client[kPending] === 0
  ) {
    return
  }

  const err = new HeadersTimeoutError(
    `HTTP/2: server did not accept a new stream within ${client[kHeadersTimeout]}`
  )

  const requests = client[kQueue].splice(client[kPendingIdx])
  for (let i = 0; i < requests.length; i++) {
    if (requests[i] != null) {
      util.errorRequest(client, requests[i], err)
    }
  }

  // Drop the unusable session so the next request gets a fresh connection,
  // whose SETTINGS may well allow streams again.
  session[kError] = err
  resetHttp2Session(session, err)
}

function clearHttp2IdleTimeout (session) {
  const state = session[kHTTP2SessionState]

  if (state?.idleTimeout != null) {
    clearTimeout(state.idleTimeout)
    state.idleTimeout = null
  }
}

function setHttp2IdleTimeout (session) {
  const client = session[kClient]

  if (client[kHTTP2Session] !== session || session.closed || session.destroyed) {
    return
  }

  if (session[kOpenStreams] !== 0 || client[kSize] !== 0) {
    clearHttp2IdleTimeout(session)
    return
  }

  const state = session[kHTTP2SessionState]
  if (state.idleTimeout == null) {
    state.idleTimeout = setTimeout(onHttp2SessionIdleTimeout, client[kKeepAliveDefaultTimeout], session).unref()
  }
}

function onHttp2SessionIdleTimeout (session) {
  const client = session[kClient]
  const socket = session[kSocket]
  const state = session[kHTTP2SessionState]

  state.idleTimeout = null

  if (client[kHTTP2Session] !== session || session[kOpenStreams] !== 0 || client[kSize] !== 0 || session.closed || session.destroyed) {
    return
  }

  const err = new InformationalError('socket idle timeout')
  socket[kError] = err
  util.destroy(socket, err)
}

// Called once for every stream that session.request() successfully opened.
// Besides the shared bookkeeping (an open stream cancels the idle timeout and
// keeps the session alive) this is where maxRequestsPerClient is enforced: for
// HTTP/2 one request is one opened stream, so a stream that never made it past
// session.request() must not consume the budget.
function trackH2Stream (session, stream) {
  clearHttp2IdleTimeout(session)
  ++session[kOpenStreams]

  const maxRequests = session[kClient][kMaxRequests]

  // `null`, `undefined` and `0` all disable the limit, as in HTTP/1.1.
  if (!maxRequests) {
    return
  }

  // kOpenStreams is released as soon as the response is complete, which for
  // HTTP/2 can happen while the request body is still being uploaded. Track the
  // physical stream separately so that retiring the session cannot truncate
  // that upload.
  session[kHTTP2SessionState].countedStreams += 1
  stream[kRetiringSession] = session
  stream.once('close', onCountedStreamClose)

  const socket = session[kSocket]
  const counter = (socket[kCounter] ?? 0) + 1
  socket[kCounter] = counter

  // `>=` mirrors client-h1.js: the stream that reaches the limit is served,
  // the next one is not.
  if (counter >= maxRequests) {
    // Retire synchronously so that a resume loop dispatching several requests
    // in one pass cannot slip stream N+1 onto this session: busy() reads the
    // flag before every write. Teardown is left to onCountedStreamClose, which
    // runs once every stream this session accepted has physically closed.
    session[kHTTP2SessionState].retired = true
    setRetiredSessionTimeout(session)
  }
}

function onCountedStreamClose () {
  const session = this[kRetiringSession]

  if (session == null) {
    return
  }

  this[kRetiringSession] = null

  const state = session[kHTTP2SessionState]
  state.countedStreams -= 1

  if (state.retired === true && state.countedStreams === 0) {
    closeRetiredH2Session(session)
  }
}

// Tearing the session down is deliberately deferred until it has drained.
// Doing it in the same tick as session.request() makes nghttp2 refuse the very
// stream that triggered retirement, and a locally initiated graceful
// session.close() can leave the socket half-open indefinitely when the peer
// keeps its side open (for example when it sent a GOAWAY of its own). Once no
// stream is left there is nothing to be graceful about, so reuse the standard
// reset path. The error is informational so queued requests remain available
// to replay on a fresh session, including when the drain deadline expires.
function closeRetiredH2Session (
  session,
  err = new InformationalError('HTTP/2: session retired after reaching maxRequestsPerClient')
) {
  clearHttp2IdleTimeout(session)
  clearNoStreamsTimeout(session)
  clearRetiredSessionTimeout(session)

  if (session.destroyed) {
    return
  }

  const client = session[kClient]
  const state = session[kHTTP2SessionState]

  // Only the attached session owns the client's connection state, so only it
  // may announce the disconnect. The flag tells onHttp2SocketClose that this
  // has already been taken care of; it is deliberately not `state.retired`,
  // because a retired session can also be torn down by the peer, and that path
  // still needs onHttp2SocketClose to flush the client state.
  const announce = client[kHTTP2Session] === session
  state.disconnectAnnounced = announce

  session[kError] = err
  resetHttp2Session(session, err)

  if (announce) {
    client.emit('disconnect', client[kUrl], [client], err)
    client[kResume]()
  }
}

function applyConnectionWindowSize (connectionWindowSize) {
  try {
    if (typeof this.setLocalWindowSize === 'function') {
      this.setLocalWindowSize(connectionWindowSize)
    }
  } catch {
    // Best-effort only.
  }
}

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 onHttp2SendPing (session) {
  const state = session[kHTTP2SessionState]
  if ((session.closed || session.destroyed) && state.ping.interval != null) {
    clearInterval(state.ping.interval)
    state.ping.interval = null
    return
  }

  // If no ping sent, do nothing
  session.ping(onPing.bind(session))

  function onPing (err, duration) {
    const client = this[kClient]
    const socket = this[kSocket]

    if (err != null) {
      const error = new InformationalError(`HTTP/2: "PING" errored - type ${err.message}`)
      socket[kError] = error
      client[kOnError](error)
    } else {
      client.emit('ping', duration)
    }
  }
}

function onHttp2SessionError (err) {
  assert(err.code !== 'ERR_TLS_CERT_ALTNAME_INVALID')

  this[kSocket][kError] = err

  if (this[kReceivedGoAway]) {
    return
  }

  this[kClient][kOnError](err)
}

function onHttp2FrameError (type, code, id) {
  if (id === 0) {
    if (this[kReceivedGoAway]) {
      return
    }

    const err = new InformationalError(`HTTP/2: "frameError" received - type ${type}, code ${code}`)
    this[kSocket][kError] = err
    this[kClient][kOnError](err)
  }
}

/**
 * This is the root cause of #3011
 * We need to handle GOAWAY frames properly, and trigger the session close
 * along with the socket right away
 *
 * @this {import('http2').ClientHttp2Session}
 * @param {number} errorCode
 * @param {number} lastStreamID
 */
function onHttp2SessionGoAway (errorCode, lastStreamID) {
  if (this[kReceivedGoAway]) {
    return
  }

  this[kReceivedGoAway] = true

  const err = getGoAwayError(this, errorCode)
  const client = this[kClient]
  const previousPendingIdx = client[kPendingIdx]
  const pendingIdx = getGoAwayPendingIdx(client, lastStreamID)
  const retriableRequests = []
  const streamsToClose = []

  // Closing one stream after GOAWAY can synchronously emit frameError on
  // sibling streams. Detach all affected requests first so those errors do
  // not fail requests that are about to be requeued.
  for (let i = pendingIdx; i < previousPendingIdx; i++) {
    const request = client[kQueue][i]

    if (request != null) {
      streamsToClose.push(detachRequestStreamForClose(request))

      if (canReplayRequest(request) && registerGoAwayRefusal(request)) {
        retriableRequests.push(request)
      } else {
        util.errorRequest(client, request, err)
      }
    }
  }

  for (let i = 0; i < streamsToClose.length; i++) {
    closeStream(streamsToClose[i])
  }

  if (pendingIdx !== previousPendingIdx) {
    const remainingPendingRequests = client[kQueue].slice(previousPendingIdx)
    client[kQueue].length = pendingIdx
    client[kQueue].push(...retriableRequests, ...remainingPendingRequests)
  }

  if (client[kHTTP2Session] === this) {
    client[kSocket] = null
    client[kHTTPContext] = null
    client[kHTTP2Session] = null
  }

  clearHttp2IdleTimeout(this)
  clearNoStreamsTimeout(this)
  clearRetiredSessionTimeout(this)

  if (!this.closed && !this.destroyed) {
    this.close()
  }

  client[kPendingIdx] = pendingIdx

  client.emit('disconnect', client[kUrl], [client], err)

  client[kResume]()
}

function onHttp2SessionClose () {
  const { [kClient]: client, [kHTTP2SessionState]: state, [kSocket]: socket } = this

  const err = socket[kError] || this[kError] || new SocketError('closed', util.getSocketInfo(socket))

  if (client[kHTTP2Session] === this) {
    client[kSocket] = null
    client[kHTTPContext] = null
    client[kHTTP2Session] = null
  }

  clearHttp2IdleTimeout(this)
  clearNoStreamsTimeout(this)
  clearRetiredSessionTimeout(this)

  if (state.ping.interval != null) {
    clearInterval(state.ping.interval)
    state.ping.interval = null
  }

  if (client.destroyed) {
    assert(client[kPending] === 0)

    // Fail entire queue.
    const requests = client[kQueue].splice(client[kRunningIdx])
    for (let i = 0; i < requests.length; i++) {
      const request = requests[i]
      if (request != null) {
        util.errorRequest(client, request, err)
      }
    }
  }
}

function onHttp2SocketClose () {
  const err = this[kError] || new SocketError('closed', util.getSocketInfo(this))

  const session = this[kHTTP2Session]
  const client = session[kClient]

  if (client[kSocket] !== this) {
    // Ignore stale socket closes from a detached GOAWAY session, from a
    // retired session that has already announced its own disconnect and from
    // any session that has already been replaced. If the session was detached
    // without any of those and there is no replacement yet, we still need the
    // close event to flush the client state.
    if (
      session[kReceivedGoAway] ||
      session[kHTTP2SessionState].disconnectAnnounced === true ||
      (client[kHTTP2Session] != null && client[kHTTP2Session] !== session)
    ) {
      return
    }
  }

  client[kSocket] = null
  client[kHTTPContext] = null
  if (client[kHTTP2Session] === session) {
    client[kHTTP2Session] = null
  }

  session.destroy(err)

  client[kPendingIdx] = client[kRunningIdx]

  assert(client[kRunning] === 0)

  client.emit('disconnect', client[kUrl], [client], err)

  client[kResume]()
}

function onHttp2SocketError (err) {
  assert(err.code !== 'ERR_TLS_CERT_ALTNAME_INVALID')

  this[kError] = err

  if (this[kHTTP2Session]?.[kReceivedGoAway]) {
    return
  }

  this[kHTTP2Session]?.[kClient]?.[kOnError](err)
}

function onHttp2SocketEnd () {
  util.destroy(this, new SocketError('other side closed', util.getSocketInfo(this)))
}

function onSocketClose () {
  this[kClosed] = true
}

function noop () {}

function closeStreamSession (stream) {
  const session = stream[kHTTP2Session]

  stream[kHTTP2Session] = null
  session[kOpenStreams] -= 1
  if (session[kOpenStreams] === 0) {
    if (session[kHTTP2SessionState].retired === true) {
      // Teardown is driven by onCountedStreamClose once every counted stream
      // has physically closed. Until then keep the session ref'd and leave the
      // idle timeout disarmed so it cannot cut an in-progress upload short.
      return
    }

    unrefH2Session(session)
    setHttp2IdleTimeout(session)
  }
}

function onUpgradeStreamClose () {
  this.off('error', noop)

  const state = this[kRequestStreamState]
  this[kRequestStreamState] = null

  failUpgradeStream(state, new InformationalError('HTTP/2: stream closed before response headers'))
  closeStreamSession(this)
}

// Idempotent terminal cleanup, called from both 'end' and 'close': the
// null-state guard no-ops the later call.
function completeRequestStream () {
  const state = this[kRequestStreamState]

  if (state == null) {
    return
  }

  // Release the stream first so request references are cleared,
  // then complete the response with trailers if available.
  releaseRequestStream(this)

  if (state.pendingEnd && !state.request.aborted && !state.request.completed) {
    state.request.onResponseEnd(state.trailers || {})
  } else if (!state.request.aborted && !state.request.completed) {
    // The stream closed without a complete response and without reporting an
    // error. finalizeRequest() below frees the queue slot either way, so
    // without this the request would simply vanish and its caller would never
    // hear back.
    util.errorRequest(
      state.client,
      state.request,
      new InformationalError('HTTP/2: stream closed before the response was complete')
    )
  }

  finalizeRequest(state)
  closeStreamSession(this)
  this[kRequestStreamState] = null
}

// https://www.rfc-editor.org/rfc/rfc7230#section-3.3.2
function shouldSendContentLength (method) {
  return method !== 'GET' && method !== 'HEAD' && method !== 'OPTIONS' && method !== 'TRACE' && method !== 'CONNECT'
}

function buildRequestHeaders (reqHeaders) {
  const headers = {}

  for (let n = 0; n < reqHeaders.length; n += 2) {
    const key = reqHeaders[n + 0]
    const val = reqHeaders[n + 1]
    const current = headers[key]

    if (key === 'cookie') {
      if (current != null) {
        headers[key] = Array.isArray(current) ? (current.push(val), current) : [current, val]
      } else {
        headers[key] = val
      }

      continue
    }

    if (typeof val === 'string') {
      headers[key] = current ? `${current}, ${val}` : val
      continue
    }

    for (let i = 0; i < val.length; i++) {
      headers[key] = headers[key] ? `${headers[key]}, ${val[i]}` : val[i]
    }
  }

  return headers
}

function removeUpgradeStreamListeners (stream) {
  stream.off('response', onUpgradeResponse)
  stream.off('error', onUpgradeStreamError)
  stream.off('end', onUpgradeStreamEnd)
  stream.off('timeout', onUpgradeStreamTimeout)
  stream.off('error', noop)
}

function releaseUpgradeStream (stream) {
  if (stream == null) {
    return
  }

  const state = stream[kRequestStreamState]
  if (state == null) {
    return
  }

  const { request } = state

  if (request[kRequestStream] === stream) {
    detachRequestFromStream(request)
  }

  removeUpgradeStreamListeners(stream)

  if (!stream.destroyed && !stream.closed) {
    stream.once('error', noop)
  }
}

function failUpgradeStream (state, err) {
  if (state == null) {
    return
  }

  const { request } = state
  if (state.responseReceived || request.aborted || request.completed) {
    return
  }

  releaseUpgradeStream(state.stream)
  state.abort(err, true)
}

function onUpgradeStreamError () {
  const state = this[kRequestStreamState]

  if (typeof this.rstCode === 'number' && this.rstCode !== 0) {
    failUpgradeStream(state, new InformationalError(`HTTP/2: "stream error" received - code ${this.rstCode}`))
  } else {
    failUpgradeStream(state, new InformationalError('HTTP/2: stream errored before response headers'))
  }
}

function onUpgradeStreamEnd () {
  failUpgradeStream(this[kRequestStreamState], new InformationalError('HTTP/2: stream half-closed (remote)'))
}

function onUpgradeStreamTimeout () {
  const state = this[kRequestStreamState]
  failUpgradeStream(state, new InformationalError(`HTTP/2: "stream timeout after ${state.headersTimeout}"`))
}

function onUpgradeResponse (headers, _flags) {
  const stream = this
  const state = stream[kRequestStreamState]
  const { request } = state

  state.responseReceived = true

  const statusCode = headers[HTTP2_HEADER_STATUS]
  delete headers[HTTP2_HEADER_STATUS]

  try {
    request.onRequestUpgrade(statusCode, headers, stream)
  } catch (err) {
    state.abort(err)
    return
  }

  if (request.aborted) {
    return
  }

  removeUpgradeStreamListeners(stream)
  detachRequestFromStream(request)
  finalizeRequest(state)
}

function setupUpgradeStream (stream, state) {
  const { request, headersTimeout, session } = state

  stream[kHTTP2Stream] = true
  stream[kHTTP2Session] = session
  stream[kRequestStreamState] = state
  state.stream = stream

  bindRequestToStream(request, stream, releaseUpgradeStream)
  stream.once('response', onUpgradeResponse)
  stream.on('error', onUpgradeStreamError)
  stream.once('end', onUpgradeStreamEnd)
  stream.on('timeout', onUpgradeStreamTimeout)
  stream.once('close', onUpgradeStreamClose)

  trackH2Stream(session, stream)
  stream.setTimeout(headersTimeout)
}

function finalizeRequest (state, resetPendingIdx = false) {
  if (state.requestFinalized) {
    return
  }

  state.requestFinalized = true
  completeRequest(state.client, state.request, resetPendingIdx)

  state.client[kResume]()
}

function openStream (client, request, session, abort, headers, options) {
  try {
    return session.request(headers, options)
  } catch (err) {
    // A GOAWAY'd session rejects new streams, same as an invalid session:
    // reset and requeue on a fresh connection rather than the destroy + abort
    // below, whose destroy(socket, err) can crash via an unhandled 'error'.
    if (err?.code === 'ERR_HTTP2_INVALID_SESSION' || err?.code === 'ERR_HTTP2_GOAWAY_SESSION') {
      const wrappedErr = new SocketError(err.message, util.getSocketInfo(session[kSocket]))
      wrappedErr.cause = err
      session[kError] = wrappedErr
      resetHttp2Session(session, wrappedErr)
      requeueUnsentRequest(client, request)

      return null
    }

    const wrappedErr = new InformationalError(err.message, { cause: err })
    session[kError] = wrappedErr
    session[kSocket][kError] = wrappedErr

    session.destroy(wrappedErr)
    util.destroy(session[kSocket], wrappedErr)
    abort(wrappedErr)

    return null
  }
}

function writeH2 (client, request) {
  const headersTimeout = request.headersTimeout ?? client[kHeadersTimeout]
  const bodyTimeout = request.bodyTimeout ?? client[kBodyTimeout]
  const session = client[kHTTP2Session]
  const { method, path, host, upgrade, expectContinue, signal, protocol, headers: reqHeaders } = request

  if (upgrade != null && upgrade !== 'websocket') {
    util.errorRequest(client, request, new InvalidArgumentError(`Custom upgrade "${upgrade}" not supported over HTTP/2`))
    return false
  }

  const headers = buildRequestHeaders(reqHeaders)

  headers[HTTP2_HEADER_AUTHORITY] = host || client[kHostAuthority]
  headers[HTTP2_HEADER_METHOD] = method

  // Single pre-shaped state object shared by all stream event handlers.
  // All fields are declared up-front so the object keeps a stable hidden
  // class for the whole request lifetime.
  const state = {
    abort: null,
    body: request.body,
    bytesRead: 0,
    client,
    contentLength: null,
    expectsPayload: false,
    maxResponseSize: client[kMaxResponseSize],
    request,
    headersTimeout,
    bodyTimeout,
    requestFinalized: false,
    responseReceived: false,
    bodySent: false,
    pendingEnd: false,
    trailers: null,
    session,
    stream: null
  }

  const abort = (err, resetPendingIdx = false) => {
    if (request.aborted || request.completed) {
      return
    }

    err = err || new RequestAbortedError()

    util.errorRequest(client, request, err)

    if (state.stream != null) {
      clearRequestStream(request)

      // On Abort, we close the stream to send RST_STREAM frame.
      const stream = state.stream
      stream.close()

      // close() alone leaves cleanup waiting on the 'close' event; on a busy,
      // long-lived multiplexed session that event can fail to fire, leaving the
      // native Http2Stream (and the whole request graph it pins) alive for the
      // session's life. Destroy the stream synchronously to release the handle
      // deterministically. Deferring the destroy (e.g. via setImmediate) leaks
      // the same way when the event loop is stalled and the callback never runs
      // under abort churn (#5558); close() has already queued the RST_STREAM
      // frame on the native session, so a synchronous destroy still sends it.
      if (!stream.destroyed) {
        util.destroy(stream)
      }

      // We move the running index to the next request
      client[kOnError](err)
      finalizeRequest(state, resetPendingIdx)
    }

    // We do not destroy the socket as we can continue using the session
    // the stream gets destroyed and the session remains to create new streams
    util.destroy(state.body, err)
  }

  state.abort = abort

  /** @type {import('node:http2').ClientHttp2Stream} */
  let stream = null

  try {
    // We are already connected, streams are pending.
    // We can call on connect, and wait for abort
    request.onRequestStart(abort, null)
  } catch (err) {
    util.errorRequest(client, request, err)
  }

  if (request.aborted) {
    return false
  }

  if (upgrade || method === 'CONNECT') {
    refH2Session(session)

    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'))
        unrefH2Session(session)
        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 = openStream(client, request, session, abort, headers, { endStream: false, signal })
      if (stream == null) {
        unrefH2Session(session)
        return false
      }
      setupUpgradeStream(stream, state)
      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 = openStream(client, request, session, abort, headers, { endStream: false, signal })
    if (stream == null) {
      unrefH2Session(session)
      return false
    }
    setupUpgradeStream(stream, state)

    return true
  }

  // 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'

  // https://tools.ietf.org/html/rfc7231#section-4.3.1
  // https://tools.ietf.org/html/rfc7231#section-4.3.2
  // https://tools.ietf.org/html/rfc7231#section-4.3.5

  // Sending a payload body on a request that does not
  // expect it can cause undefined behavior on some
  // servers and corrupt connection state. Do not
  // re-use the connection for further requests.

  const expectsPayload = (
    method === 'PUT' ||
    method === 'POST' ||
    method === 'PATCH' ||
    method === 'QUERY' ||
    method === 'PROPFIND' ||
    method === 'PROPPATCH'
  )

  let body = state.body

  if (body && typeof body.read === 'function') {
    // Try to read EOF in order to get length.
    body.read(0)
  }

  let contentLength = util.bodyLength(body)

  if (util.isFormDataLike(body)) {
    extractBody ??= require('../web/fetch/body.js').extractBody

    const [bodyStream, contentType] = extractBody(body)
    headers['content-type'] = contentType

    body = bodyStream.stream
    contentLength = bodyStream.length
  }

  if (contentLength == null) {
    contentLength = request.contentLength
  }

  if (contentLength === 0 && !expectsPayload) {
    // https://tools.ietf.org/html/rfc7230#section-3.3.2
    // A user agent SHOULD NOT send a Content-Length header field when
    // the request message does not contain a payload body and the method
    // semantics do not anticipate such a body.
    // And for methods that don't expect a payload, omit Content-Length.
    contentLength = null
  }

  // https://github.com/nodejs/undici/issues/2046
  // A user agent may send a Content-Length header with 0 value, this should be allowed.
  if (shouldSendContentLength(method) && contentLength > 0 && request.contentLength != null && request.contentLength !== contentLength) {
    if (client[kStrictContentLength]) {
      util.errorRequest(client, request, new RequestContentLengthMismatchError())
      return false
    }

    process.emitWarning(new RequestContentLengthMismatchError())
  }

  if (contentLength != null) {
    assert(body || contentLength === 0, 'no body must not have content length')
    headers[HTTP2_HEADER_CONTENT_LENGTH] = `${contentLength}`
  }

  refH2Session(session)

  if (channels.sendHeaders.hasSubscribers) {
    let header = ''
    for (const key in headers) {
      header += `${key}: ${headers[key]}\r\n`
    }
    channels.sendHeaders.publish({ request, headers: header, socket: session[kSocket] })
  }

  // TODO(metcoder95): add support for sending trailers
  const shouldEndStream = body === null || contentLength === 0

  state.body = body
  state.contentLength = contentLength
  state.expectsPayload = expectsPayload

  if (expectContinue) {
    headers[HTTP2_HEADER_EXPECT] = '100-continue'
  }

  stream = openStream(client, request, session, abort, headers, { endStream: shouldEndStream, signal })
  if (stream == null) {
    return false
  }
  stream[kHTTP2Stream] = true
  stream[kRequestStreamState] = state
  state.stream = stream

  if (headersTimeout) {
    stream.setTimeout(headersTimeout)
  }

  stream[kHTTP2Session] = session
  stream.on('close', completeRequestStream)

  // Registered after the request's own 'close' handler so that request
  // completion always runs before a session retirement can tear the
  // connection down.
  trackH2Stream(session, stream)

  bindRequestToStream(request, stream, releaseRequestStream)
  if (expectContinue) {
    stream.once('continue', writeBodyH2)
  }
  // The handlers below either remove themselves on first invocation or
  // become unreachable once the stream closes, so plain `on` avoids the
  // per-listener `once` wrapper allocation.
  stream.on('response', onResponse)
  stream.on('headers', onInterimResponse)
  stream.on('end', onEnd)
  stream.on('error', onError)
  stream.on('frameError', onFrameError)
  stream.on('aborted', onAborted)
  if (headersTimeout || bodyTimeout) {
    stream.on('timeout', onTimeout)
  }
  stream.on('trailers', onTrailers)

  if (!expectContinue) {
    writeBodyH2.call(stream)
  }

  return true
}

function removeRequestStreamListeners (stream) {
  stream.off('error', noop)
  stream.off('continue', writeBodyH2)
  stream.off('response', onResponse)
  stream.off('headers', onInterimResponse)
  stream.off('end', onEnd)
  stream.off('error', onError)
  stream.off('frameError', onFrameError)
  stream.off('aborted', onAborted)
  stream.off('timeout', onTimeout)
  stream.off('trailers', onTrailers)
  stream.off('data', onData)
}

function releaseRequestStream (stream) {
  if (stream == null) {
    return
  }

  const state = stream[kRequestStreamState]
  if (state == null) {
    return
  }

  const { request } = state

  if (request[kRequestStream] === stream) {
    detachRequestFromStream(request)
  }

  // A closed or destroyed stream cannot emit further events; leaving the
  // listeners in place saves the removal scans (they are collected with
  // the stream). All handlers bail out when the stream state is gone.
  if (!stream.destroyed && !stream.closed) {
    removeRequestStreamListeners(stream)
    stream.once('error', noop)
  }
}

function onData (chunk) {
  const stream = this
  const state = stream[kRequestStreamState]

  if (state == null) {
    return
  }

  const { request, maxResponseSize } = state

  if (request.aborted || request.completed) {
    return
  }

  if (maxResponseSize > -1 && state.bytesRead + chunk.length > maxResponseSize) {
    // Unlike HTTP/1.1, which destroys the socket because it cannot abandon one
    // response without losing framing, resetting the offending stream leaves
    // the session usable for its siblings.
    state.abort(new ResponseExceededMaxSizeError())
    return
  }

  state.bytesRead += chunk.length

  if (request.onResponseData(chunk) === false) {
    stream.pause()
  }
}

function onInterimResponse (headers) {
  const stream = this
  const state = stream[kRequestStreamState]

  if (state == null) {
    return
  }

  const { request } = state

  if (request.aborted || request.completed) {
    return
  }

  // node http2 emits 'headers' for interim (1xx) informational responses,
  // while the final response arrives via 'response'. Forward these to the
  // handler so that onInfo is invoked, matching the HTTP/1 behaviour and the
  // documented onInfo contract.
  const statusCode = headers[HTTP2_HEADER_STATUS]
  delete headers[HTTP2_HEADER_STATUS]

  request.onResponseStart(Number(statusCode), headers, noop, '')
}

function onResponse (headers) {
  const stream = this
  const state = stream[kRequestStreamState]

  if (state == null) {
    return
  }

  const { request } = state

  stream.off('response', onResponse)

  // Final response received while still awaiting 100 (Continue): the body won't
  // be sent, so close our half or the stream stays open and never completes.
  if (state.body != null && !state.bodySent && !stream.writableEnded) {
    stream.removeListener('continue', writeBodyH2)
    stream.end()
  }

  const statusCode = headers[HTTP2_HEADER_STATUS]
  delete headers[HTTP2_HEADER_STATUS]
  request.onResponseStarted()
  state.responseReceived = true

  if (state.headersTimeout || state.bodyTimeout) {
    stream.setTimeout(state.bodyTimeout)
  }

  // Due to the stream nature, it is possible we face a race condition
  // where the stream has been assigned, but the request has been aborted
  // or already completed and headers hasn't been received yet. A late
  // 'response' delivered after completion would call request.onResponseStart
  // post-completion, tripping its `assert(!this.completed)` (an uncatchable
  // throw on the http2 event tick). Guard `completed` here as onEnd/onTrailers
  // already do; best effort is to release the stream immediately as there's
  // no value to keep it open.
  if (request.aborted || request.completed) {
    releaseRequestStream(stream)
    return
  }

  if (request.onResponseStart(Number(statusCode), headers, stream.resume.bind(stream), '') === false) {
    stream.pause()
  }

  stream.on('data', onData)
}

function onEnd () {
  const stream = this
  const state = stream[kRequestStreamState]

  if (state == null) {
    return
  }

  const { request } = state

  stream.off('end', onEnd)

  // onTrailers (which may fire after 'end' on Windows) has already stored
  // trailers on the state by now, so completing here still delivers them.
  if (state.responseReceived) {
    if (!request.aborted && !request.completed) {
      state.pendingEnd = true

      // Complete on 'end': a blocked event loop can keep the stream's 'close'
      // from firing, stranding its buffers until OOM. Idempotent, so a later
      // 'close' no-ops.
      completeRequestStream.call(stream)
    }
  } else {
    // Stream ended without receiving a response - this is an error
    // (e.g., server destroyed the stream before sending headers)
    state.abort(new InformationalError('HTTP/2: stream half-closed (remote)'), true)
  }
}

function retryRefusedStream (stream, state) {
  const { client, request } = state

  if (
    state.responseReceived ||
    request.aborted ||
    request.completed ||
    request[kRefusedStreamRetry] ||
    !canReplayRequest(request)
  ) {
    return false
  }

  // RFC 9113 section 8.7 permits retrying REFUSED_STREAM, but says clients
  // SHOULD NOT automatically retry the same request more than once.
  request[kRefusedStreamRetry] = true

  // Detach the failed attempt before moving the request back to the pending
  // queue. The peer only reset this stream, so the HTTP/2 session remains
  // usable for the retry. Severing also drops the 'close' listener, so the
  // abandoned stream cannot later complete the retried request.
  detachRequestStreamForClose(request)
  state.stream = null
  state.requestFinalized = true

  completeRequest(client, request)
  client[kQueue].splice(client[kPendingIdx], 0, request)
  client[kResume]()

  return true
}

function onError (err) {
  const stream = this
  const state = stream[kRequestStreamState]

  if (state == null) {
    return
  }

  stream.off('error', onError)

  if (typeof stream.rstCode === 'number' && stream.rstCode !== NGHTTP2_NO_ERROR) {
    err.http2ErrorCode = stream.rstCode
  }

  if (
    stream.rstCode === NGHTTP2_REFUSED_STREAM &&
    retryRefusedStream(stream, state)
  ) {
    return
  }

  state.abort(err)
}

function onFrameError (type, code) {
  const stream = this
  const state = stream[kRequestStreamState]

  if (state == null) {
    return
  }

  stream.off('frameError', onFrameError)
  state.abort(new InformationalError(`HTTP/2: "frameError" received - type ${type}, code ${code}`))
}

function onAborted () {
  this.off('data', onData)
}

function onTimeout () {
  const stream = this
  const state = stream[kRequestStreamState]

  if (state == null) {
    return
  }

  // Remove self so timeout doesn't fire again after we handle it
  stream.off('timeout', onTimeout)

  const err = state.responseReceived
    ? new BodyTimeoutError(`HTTP/2: "stream timeout after ${state.bodyTimeout}"`)
    : new HeadersTimeoutError(`HTTP/2: "headers timeout after ${state.headersTimeout}"`)
  state.abort(err)
}

function onTrailers (trailers) {
  const stream = this
  const state = stream[kRequestStreamState]

  if (state == null) {
    return
  }

  const { request } = state

  stream.off('trailers', onTrailers)
  stream.off('data', onData)

  if (request.aborted || request.completed) {
    return
  }

  // Store trailers for completeRequestStream to use when completing
  state.trailers = trailers
}

function writeBodyH2 () {
  const stream = this
  const state = stream[kRequestStreamState]
  state.bodySent = true
  const { abort, body, client, contentLength, expectsPayload, request } = state

  if (!body || contentLength === 0) {
    writeBuffer(
      abort,
      stream,
      null,
      client,
      request,
      client[kSocket],
      contentLength,
      expectsPayload
    )
  } else if (util.isBuffer(body)) {
    writeBuffer(
      abort,
      stream,
      body,
      client,
      request,
      client[kSocket],
      contentLength,
      expectsPayload
    )
  } else if (util.isBlobLike(body)) {
    if (typeof body.stream === 'function') {
      writeIterable(
        abort,
        stream,
        body.stream(),
        client,
        request,
        client[kSocket],
        contentLength,
        expectsPayload
      )
    } else {
      writeBlob(
        abort,
        stream,
        body,
        client,
        request,
        client[kSocket],
        contentLength,
        expectsPayload
      )
    }
  } else if (util.isStream(body)) {
    writeStream(
      abort,
      client[kSocket],
      expectsPayload,
      stream,
      body,
      client,
      request,
      contentLength
    )
  } else if (util.isIterable(body)) {
    writeIterable(
      abort,
      stream,
      body,
      client,
      request,
      client[kSocket],
      contentLength,
      expectsPayload
    )
  } else {
    assert(false)
  }
}

function writeBuffer (abort, h2stream, body, client, request, socket, contentLength, expectsPayload) {
  try {
    if (body != null && util.isBuffer(body)) {
      assert(contentLength === body.byteLength, 'buffer body must have content length')
      h2stream.cork()
      h2stream.write(body)
      h2stream.uncork()
      h2stream.end()

      request.onBodySent(body)
    }

    if (!expectsPayload) {
      socket[kReset] = true
    }

    request.onRequestSent()
    client[kResume]()
  } catch (error) {
    abort(error)
  }
}

function writeStream (abort, socket, expectsPayload, h2stream, body, client, request, contentLength) {
  assert(contentLength !== 0 || client[kRunning] === 0, 'stream body cannot be pipelined')

  // For HTTP/2, is enough to pipe the stream
  const pipe = pipeline(
    body,
    h2stream,
    (err) => {
      if (err) {
        util.destroy(pipe, err)
        abort(err)
      } else {
        util.removeAllListeners(pipe)
        request.onRequestSent()

        if (!expectsPayload) {
          socket[kReset] = true
        }

        client[kResume]()
      }
    }
  )

  util.addListener(pipe, 'data', onPipeData)

  function onPipeData (chunk) {
    request.onBodySent(chunk)
  }
}

async function writeBlob (abort, h2stream, body, client, request, socket, contentLength, expectsPayload) {
  try {
    if (contentLength != null && contentLength !== body.size) {
      throw new RequestContentLengthMismatchError()
    }

    const buffer = Buffer.from(await body.arrayBuffer())

    h2stream.cork()
    h2stream.write(buffer)
    h2stream.uncork()
    h2stream.end()

    request.onBodySent(buffer)
    request.onRequestSent()

    if (!expectsPayload) {
      socket[kReset] = true
    }

    client[kResume]()
  } catch (err) {
    abort(err)
  }
}

async function writeIterable (abort, h2stream, body, client, request, socket, contentLength, expectsPayload) {
  assert(contentLength !== 0 || client[kRunning] === 0, 'iterator body cannot be pipelined')

  let callback = null
  function onDrain () {
    if (callback) {
      const cb = callback
      callback = null
      cb()
    }
  }

  const waitForDrain = () => new Promise((resolve, reject) => {
    assert(callback === null)

    if (socket[kError]) {
      reject(socket[kError])
    } else {
      callback = resolve
    }
  })

  h2stream
    .on('close', onDrain)
    .on('drain', onDrain)

  try {
    // It's up to the user to somehow abort the async iterable.
    for await (const chunk of body) {
      if (socket[kError]) {
        throw socket[kError]
      }

      const res = h2stream.write(chunk)
      request.onBodySent(chunk)
      if (!res) {
        await waitForDrain()
      }
    }

    h2stream.end()

    request.onRequestSent()

    if (!expectsPayload) {
      socket[kReset] = true
    }

    client[kResume]()
  } catch (err) {
    abort(err)
  } finally {
    h2stream
      .off('close', onDrain)
      .off('drain', onDrain)
  }
}

module.exports = connectH2