feat(h2): honor maxRequestsPerClient

docs/docs/api/Client.md

@@ -49,6 +49,8 @@ added: v1.0.0
   * `headersTimeout` {number|null} The timeout, in milliseconds, the parser
     waits to receive the complete HTTP headers before the request times out. Use
     `0` to disable it entirely. **Default:** `300e3`.
+    Over HTTP/2 this also bounds how long a session retired by
+    `maxRequestsPerClient` may wait for its active streams to close.
     HTTP/1.1 headers/body parser timeouts are not guaranteed to fire with exact
     millisecond precision: delays up to 1000ms use native timers, while larger
     delays use undici's lower-overhead fast timers with a target resolution
@@ -75,6 +77,12 @@ added: v1.0.0
   * `maxRequestsPerClient` {number|null} The maximum number of requests to send
     over a single connection before the socket is reset. Use `0` to disable this
     limit. **Default:** `null`.
+    Over HTTP/1.1 a request is counted per message and the socket is reset once
+    the limit is reached. Over HTTP/2 a request is counted per successfully
+    opened stream: the session is retired once the limit is reached, meaning it
+    accepts no further streams. Streams it already accepted are allowed to
+    complete for up to `headersTimeout` before the connection is reset. Queued
+    and subsequent requests are then dispatched on a new session.
   * `localAddress` {string|null} The local IP address the socket should connect
     from. **Default:** `null`.
   * `pipelining` {number|null} The number of concurrent requests sent over the

docs/docs/api/H2CClient.md

@@ -64,7 +64,9 @@ added: v7.7.0
   * `maxHeaderSize` {number} The maximum length of request headers in bytes.
     **Default:** Node.js' `--max-http-header-size` or `16384` (16 KiB).
   * `headersTimeout` {number} The amount of time, in milliseconds, the parser
-    waits to receive the complete HTTP headers. **Default:** `300e3`.
+    waits to receive the complete HTTP headers. This also bounds how long a
+    session retired by `maxRequestsPerClient` may wait for its active streams
+    to close. **Default:** `300e3`.
   * `connectTimeout` {number} The timeout for establishing a socket connection,
     in milliseconds. Use `0` to disable it entirely. **Default:** `10e3`.
   * `bodyTimeout` {number} The timeout, in milliseconds, after which a request
@@ -92,6 +94,11 @@ added: v7.7.0
   * `maxRequestsPerClient` {number} The maximum number of requests to send over
     a single connection before it is reset. Use `0` to disable this limit.
     **Default:** `null`.
+    A request is counted per successfully opened HTTP/2 stream. Once the limit
+    is reached the session is retired: it accepts no further streams. Streams
+    it already accepted are allowed to complete for up to `headersTimeout`
+    before the connection is reset. Queued and subsequent requests are
+    dispatched on a new session.
   * `localAddress` {string} The local IP address the socket should connect from.
   * `maxResponseSize` {number} The maximum allowed response body size in bytes.
     Use `-1` to disable. **Default:** `-1`.

lib/dispatcher/client-h2.js

@@ -41,7 +41,9 @@ const {
   kHTTP2Stream,
   kHTTP2SessionState,
   kHTTP2Options,
-  kMaxResponseSize
+  kMaxResponseSize,
+  kMaxRequests,
+  kCounter
 } = require('../core/symbols.js')
 const { channels } = require('../core/diagnostics.js')
 
@@ -53,6 +55,7 @@ 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
@@ -97,6 +100,8 @@ function resetHttp2Session (session, err) {
   const client = session[kClient]
   const socket = session[kSocket]
 
+  clearRetiredSessionTimeout(session)
+
   if (client[kHTTP2Session] === session) {
     client[kSocket] = null
     client[kHTTPContext] = null
@@ -271,6 +276,20 @@ function connectH2 (client, socket) {
     // 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
@@ -351,6 +370,13 @@ function connectH2 (client, socket) {
      * @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
       }
@@ -406,6 +432,13 @@ function resumeH2 (client) {
   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)
@@ -436,6 +469,51 @@ function clearNoStreamsTimeout (session) {
   }
 }
 
+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
@@ -531,6 +609,103 @@ function onHttp2SessionIdleTimeout (session) {
   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') {
@@ -669,6 +844,7 @@ function onHttp2SessionGoAway (errorCode, lastStreamID) {
 
   clearHttp2IdleTimeout(this)
   clearNoStreamsTimeout(this)
+  clearRetiredSessionTimeout(this)
 
   if (!this.closed && !this.destroyed) {
     this.close()
@@ -694,6 +870,7 @@ function onHttp2SessionClose () {
 
   clearHttp2IdleTimeout(this)
   clearNoStreamsTimeout(this)
+  clearRetiredSessionTimeout(this)
 
   if (state.ping.interval != null) {
     clearInterval(state.ping.interval)
@@ -721,11 +898,16 @@ function onHttp2SocketClose () {
   const client = session[kClient]
 
   if (client[kSocket] !== this) {
-    // Ignore stale socket closes from a detached GOAWAY session and from any
-    // session that has already been replaced. If the session was detached
-    // without a GOAWAY and there is no replacement yet, we still need the
+    // 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] || (client[kHTTP2Session] != null && client[kHTTP2Session] !== session)) {
+    if (
+      session[kReceivedGoAway] ||
+      session[kHTTP2SessionState].disconnectAnnounced === true ||
+      (client[kHTTP2Session] != null && client[kHTTP2Session] !== session)
+    ) {
       return
     }
   }
@@ -775,6 +957,13 @@ function closeStreamSession (stream) {
   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)
   }
@@ -963,8 +1152,7 @@ function setupUpgradeStream (stream, state) {
   stream.on('timeout', onUpgradeStreamTimeout)
   stream.once('close', onUpgradeStreamClose)
 
-  clearHttp2IdleTimeout(session)
-  ++session[kOpenStreams]
+  trackH2Stream(session, stream)
   stream.setTimeout(headersTimeout)
 }
 
@@ -1251,17 +1439,18 @@ function writeH2 (client, request) {
   stream[kRequestStreamState] = state
   state.stream = stream
 
-  // Increment counter as we have new streams open
-  clearHttp2IdleTimeout(session)
-  ++session[kOpenStreams]
-
   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)