fix(h2): ensure every request settles

lib/dispatcher/client-h2.js

@@ -49,6 +49,13 @@ const kRequestStream = Symbol('request stream')
 const kRequestStreamCleanup = Symbol('request stream cleanup')
 const kRequestStreamState = Symbol('request stream state')
 const kReceivedGoAway = Symbol('received goaway')
+const kRefusedAttempts = Symbol('refused attempts')
+
+// How many times a single request may be requeued because the connection
+// refused to carry it (GOAWAY above lastStreamID). 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_REFUSED_ATTEMPTS = 3
 
 let extractBody
 
@@ -180,7 +187,17 @@ function completeRequest (client, request, resetPendingIdx = false) {
 function canRetryRequestAfterGoAway (request) {
   const { body } = request
 
-  return body == null || util.isBuffer(body) || util.isBlobLike(body)
+  if (body != null && !util.isBuffer(body) && !util.isBlobLike(body)) {
+    return false
+  }
+
+  // Count this refusal against the request's budget. A peer that refuses every
+  // connection must eventually surface an error to the caller rather than
+  // being retried forever.
+  const attempts = (request[kRefusedAttempts] ?? 0) + 1
+  request[kRefusedAttempts] = attempts
+
+  return attempts <= MAX_REFUSED_ATTEMPTS
 }
 
 function closeStream (stream, code = NGHTTP2_REFUSED_STREAM) {
@@ -195,10 +212,32 @@ 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)
+
+  if (stream[kHTTP2Session] != null) {
+    closeStreamSession(stream)
+  }
+
+  if (!stream.destroyed && !stream.closed) {
+    stream.once('error', noop)
+  }
+}
+
 function connectH2 (client, socket) {
   client[kSocket] = socket
 
@@ -221,6 +260,9 @@ function connectH2 (client, socket) {
   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,
     // 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
@@ -367,9 +409,76 @@ function resumeH2 (client) {
     } 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
   }
 }
 
+// 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]
 
@@ -550,6 +659,7 @@ function onHttp2SessionGoAway (errorCode, lastStreamID) {
   }
 
   clearHttp2IdleTimeout(this)
+  clearNoStreamsTimeout(this)
 
   if (!this.closed && !this.destroyed) {
     this.close()
@@ -574,6 +684,7 @@ function onHttp2SessionClose () {
   }
 
   clearHttp2IdleTimeout(this)
+  clearNoStreamsTimeout(this)
 
   if (state.ping.interval != null) {
     clearInterval(state.ping.interval)
@@ -685,6 +796,16 @@ function completeRequestStream () {
 
   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)

test/http2-goaway-refusal-storm.js

@@ -0,0 +1,208 @@
+'use strict'
+
+const assert = require('node:assert')
+const { test, after } = require('node:test')
+const { createServer } = require('node:tls')
+const { createSecureServer } = require('node:http2')
+const { once } = require('node:events')
+
+const pem = require('@metcoder95/https-pem')
+
+const { Client } = require('..')
+
+// A server that refuses a request with GOAWAY(lastStreamID = 0) is saying "I
+// processed nothing, retry elsewhere" (RFC 9113 §6.8) -- what a draining proxy
+// or an overloaded load balancer sends.
+//
+// onHttp2SessionGoAway() requeues the request and reconnects immediately. There
+// is no attempt counter, no backoff and no deadline, so while the condition
+// persists the client spins connect -> HEADERS -> GOAWAY -> requeue -> connect
+// several hundred times per second: the request never settles, a CPU core is
+// pinned and the event loop is starved for every other origin in the process.
+//
+// Node's http2 server clamps lastStreamID to the last stream it actually
+// received, so these tests drive the connection at the frame level.
+
+const PREFACE = Buffer.from('PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n')
+const FRAME_SETTINGS = 4
+const FRAME_PING = 6
+const FRAME_HEADERS = 1
+
+function frame (type, flags, streamId, payload = Buffer.alloc(0)) {
+  const head = Buffer.alloc(9)
+  head.writeUIntBE(payload.length, 0, 3)
+  head[3] = type
+  head[4] = flags
+  head.writeUInt32BE(streamId >>> 0, 5)
+  return Buffer.concat([head, payload])
+}
+
+function goawayFrame (lastStreamId, errorCode) {
+  const payload = Buffer.alloc(8)
+  payload.writeUInt32BE(lastStreamId >>> 0, 0)
+  payload.writeUInt32BE(errorCode >>> 0, 4)
+  return frame(7, 0, 0, payload)
+}
+
+// Minimal h2 peer: completes the handshake, then refuses whatever it is asked.
+async function refusingServer (errorCode) {
+  const { key, cert } = await pem.generate({ opts: { keySize: 2048 } })
+  let connections = 0
+
+  const sockets = new Set()
+  const server = createServer({ key, cert, ALPNProtocols: ['h2'] }, (socket) => {
+    connections++
+    sockets.add(socket)
+    socket.on('close', () => sockets.delete(socket))
+    let buf = Buffer.alloc(0)
+    let sawPreface = false
+
+    socket.on('error', () => {})
+    socket.write(frame(FRAME_SETTINGS, 0, 0))
+
+    socket.on('data', (chunk) => {
+      buf = Buffer.concat([buf, chunk])
+
+      if (!sawPreface) {
+        if (buf.length < PREFACE.length) return
+        buf = buf.subarray(PREFACE.length)
+        sawPreface = true
+      }
+
+      while (buf.length >= 9) {
+        const length = buf.readUIntBE(0, 3)
+        if (buf.length < 9 + length) break
+
+        const type = buf[3]
+        const flags = buf[4]
+        const payload = buf.subarray(9, 9 + length)
+        buf = buf.subarray(9 + length)
+
+        if (type === FRAME_SETTINGS && !(flags & 0x1)) socket.write(frame(FRAME_SETTINGS, 0x1, 0))
+        if (type === FRAME_PING && !(flags & 0x1)) socket.write(frame(FRAME_PING, 0x1, 0, payload))
+        if (type === FRAME_HEADERS) socket.write(goawayFrame(0, errorCode))
+      }
+    })
+  })
+
+  await once(server.listen(0), 'listening')
+  server.connections = () => connections
+  server.shutdown = () => {
+    for (const socket of sockets) socket.destroy()
+    server.close()
+  }
+  return server
+}
+
+// A wedged/spinning h2 client can leave the event loop without a ref'd handle.
+function holdEventLoop () {
+  const timer = setInterval(() => {}, 1000)
+  return () => clearInterval(timer)
+}
+
+async function measure (errorCode, label) {
+  const release = holdEventLoop()
+  const server = await refusingServer(errorCode)
+
+  const client = new Client(`https://localhost:${server.address().port}`, {
+    connect: { rejectUnauthorized: false },
+    allowH2: true,
+    headersTimeout: 1000,
+    bodyTimeout: 1000
+  })
+  after(async () => {
+    await client.destroy()
+    server.shutdown()
+    release()
+  })
+
+  const pending = client.request({ path: '/', method: 'GET' }).then(() => {}, () => {})
+
+  let timer
+  const outcome = await Promise.race([
+    pending.then(() => 'settled'),
+    new Promise((resolve) => { timer = setTimeout(() => resolve('stuck'), 3000) })
+  ]).finally(() => clearTimeout(timer))
+
+  const connections = server.connections()
+
+  await client.destroy()
+  await pending
+
+  assert.strictEqual(outcome, 'settled', `${label}: the request never settled`)
+  assert.ok(
+    connections <= 20,
+    `${label}: retried the refused request over ${connections} connections`
+  )
+}
+
+test('a repeatedly refused h2 request must not reconnect without bound', async () => {
+  await measure(0, 'GOAWAY(NO_ERROR)')
+})
+
+test('a GOAWAY carrying an error code must not be retried without bound', async () => {
+  // ENHANCE_YOUR_CALM is explicit backpressure; INTERNAL_ERROR is terminal.
+  // Neither is a reason to reconnect as fast as the event loop allows.
+  await measure(11, 'GOAWAY(ENHANCE_YOUR_CALM)')
+  await measure(2, 'GOAWAY(INTERNAL_ERROR)')
+})
+
+test('a refused h2 origin must not starve unrelated work in the process', async () => {
+  const release = holdEventLoop()
+
+  // A healthy h2 origin to measure against.
+  const healthy = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
+  healthy.on('stream', (stream) => {
+    stream.on('error', () => {})
+    stream.respond({ ':status': 200 })
+    stream.end('ok')
+  })
+  await once(healthy.listen(0), 'listening')
+
+  const healthyClient = new Client(`https://localhost:${healthy.address().port}`, {
+    connect: { rejectUnauthorized: false },
+    allowH2: true
+  })
+
+  const measureThroughput = async (ms) => {
+    let served = 0
+    const deadline = Date.now() + ms
+    while (Date.now() < deadline) {
+      const res = await healthyClient.request({ path: '/', method: 'GET' })
+      await res.body.dump()
+      served++
+    }
+    return served
+  }
+
+  const baseline = await measureThroughput(1000)
+
+  const refusing = await refusingServer(0)
+  const refusingClient = new Client(`https://localhost:${refusing.address().port}`, {
+    connect: { rejectUnauthorized: false },
+    allowH2: true,
+    headersTimeout: 1000,
+    bodyTimeout: 1000
+  })
+
+  after(async () => {
+    await refusingClient.destroy()
+    await healthyClient.destroy()
+    refusing.shutdown()
+    healthy.close()
+    release()
+  })
+
+  // One request to the refusing origin, then measure the healthy one again.
+  const pending = refusingClient.request({ path: '/', method: 'GET' }).then(() => {}, () => {})
+  const during = await measureThroughput(1000)
+
+  await refusingClient.destroy()
+  await pending
+
+  assert.ok(
+    during > baseline / 2,
+    `one refused h2 origin cut unrelated throughput from ${baseline}/s to ${during}/s ` +
+    `(${refusing.connections()} reconnects in 1s)`
+  )
+})