fix(h2): avoid pool fan-out during ALPN negotiation

docs/docs/api/Agent.md

@@ -56,16 +56,16 @@ changes:
     `Infinity`, no limit is enforced. Must be a number greater than `0`.
     **Default:** `Infinity`.
 
-`Agent` inherits all {PoolOptions} (and therefore all {ClientOptions}). The
-per-origin {Pool} it creates uses the default unlimited `connections`, so
-concurrent requests to the same origin are spread across separate {Client}
-instances on separate sockets.
+`Agent` inherits all {PoolOptions} (and therefore all {ClientOptions}). Each
+origin gets a separate {Pool}, with `connections` acting as the maximum number
+of clients that pool may create.
 
 > [!NOTE]
-> Because each concurrent request to an origin may use a different {Client},
-> HTTP/2 multiplexing on a shared session does not apply unless `connections` is
-> set to a small value (for example `connections: 1`). See {PoolOptions} and
-> {ClientOptions} for the full set of inherited options such as `allowH2`
+> For an h2-capable HTTPS origin, the per-origin pool waits for the first TLS
+> connection to finish ALPN negotiation. If the server selects h2, concurrent
+> requests share that session up to `maxConcurrentStreams`. If it selects
+> HTTP/1.1, normal connection fan-out resumes up to `connections`. See
+> {PoolOptions} and {ClientOptions} for inherited options such as `allowH2`
 > (default `true`) and `maxConcurrentStreams` (default `100`).
 
 ### `agent.closed`

docs/docs/api/Pool.md

@@ -73,12 +73,11 @@ connector shared by every pooled client.
 
 > [!NOTE]
 > `Pool` inherits all {ClientOptions}, including `allowH2` and
-> `maxConcurrentStreams`. With the default unlimited `connections`, the pool
-> opens a new client - and therefore a new TCP/TLS socket - per concurrent
-> dispatch, which defeats HTTP/2 multiplexing over a shared session. To benefit
-> from h2 multiplexing on a single session, cap `connections` (for example
-> `connections: 1`) so that concurrent requests share a session up to
-> `maxConcurrentStreams`.
+> `maxConcurrentStreams`. For an h2-capable HTTPS origin, the pool waits for
+> the first TLS connection to finish ALPN negotiation before opening more
+> clients. If the server selects h2, concurrent requests share that session up
+> to `maxConcurrentStreams`. If it selects HTTP/1.1, the pool resumes normal
+> connection fan-out up to `connections`.
 
 ```mjs
 import { Pool } from 'undici'

lib/dispatcher/pool-base.js

@@ -13,6 +13,9 @@ const kOnDrain = Symbol('onDrain')
 const kOnConnect = Symbol('onConnect')
 const kOnDisconnect = Symbol('onDisconnect')
 const kOnConnectionError = Symbol('onConnectionError')
+const kOnClientBusy = Symbol('on client busy')
+const kOnClientDrain = Symbol('on client drain')
+const kDrainQueue = Symbol('drain queue')
 const kGetDispatcher = Symbol('get dispatcher')
 const kHasDispatcher = Symbol('has dispatcher')
 const kAddClient = Symbol('add client')
@@ -29,9 +32,10 @@ class PoolBase extends DispatcherBase {
 
   [kOnDrain] (client, origin, targets) {
     const queue = this[kQueue]
-
     let needDrain = false
 
+    this[kOnClientDrain](client)
+
     while (!needDrain) {
       const item = queue.shift()
       if (!item) {
@@ -42,6 +46,9 @@ class PoolBase extends DispatcherBase {
     }
 
     client[kNeedDrain] = needDrain
+    if (needDrain) {
+      this[kOnClientBusy](client)
+    }
 
     if (!needDrain && this[kNeedDrain]) {
       this[kNeedDrain] = false
@@ -61,6 +68,52 @@ class PoolBase extends DispatcherBase {
     }
   }
 
+  [kOnClientBusy] () {}
+
+  [kOnClientDrain] () {}
+
+  [kDrainQueue] (origin, targets) {
+    const queue = this[kQueue]
+    let hasDispatcher = true
+
+    while (!queue.isEmpty()) {
+      const dispatcher = this[kGetDispatcher]()
+      if (!dispatcher) {
+        hasDispatcher = false
+        break
+      }
+
+      const item = queue.shift()
+      this[kQueued]--
+
+      if (!dispatcher.dispatch(item.opts, item.handler)) {
+        dispatcher[kNeedDrain] = true
+        this[kOnClientBusy](dispatcher)
+        hasDispatcher = this[kHasDispatcher]()
+        if (!hasDispatcher) {
+          break
+        }
+      }
+    }
+
+    if (hasDispatcher && this[kNeedDrain]) {
+      this[kNeedDrain] = false
+      this.emit('drain', origin, [this, ...targets])
+    }
+
+    if (this[kClosedResolve] && queue.isEmpty()) {
+      const closeAll = []
+      for (let i = 0; i < this[kClients].length; i++) {
+        const client = this[kClients][i]
+        if (!client.destroyed) {
+          closeAll.push(client.close())
+        }
+      }
+      return Promise.all(closeAll)
+        .then(this[kClosedResolve])
+    }
+  }
+
   [kOnConnect] = (origin, targets) => {
     this.emit('connect', origin, [this, ...targets])
   };
@@ -163,6 +216,7 @@ class PoolBase extends DispatcherBase {
       this[kQueued]++
     } else if (!dispatcher.dispatch(opts, handler)) {
       dispatcher[kNeedDrain] = true
+      this[kOnClientBusy](dispatcher)
       this[kNeedDrain] = !this[kHasDispatcher]()
     }
 
@@ -196,7 +250,7 @@ class PoolBase extends DispatcherBase {
 
     if (this[kNeedDrain]) {
       queueMicrotask(() => {
-        if (this[kNeedDrain]) {
+        if (this[kNeedDrain] && !client[kNeedDrain]) {
           this[kOnDrain](client, client[kUrl], [client, this])
         }
       })
@@ -227,6 +281,9 @@ module.exports = {
   kNeedDrain,
   kAddClient,
   kRemoveClient,
+  kDrainQueue,
+  kOnClientBusy,
+  kOnClientDrain,
   kGetDispatcher,
   kHasDispatcher
 }

lib/dispatcher/pool.js

@@ -5,6 +5,9 @@ const {
   kClients,
   kNeedDrain,
   kAddClient,
+  kDrainQueue,
+  kOnClientBusy,
+  kOnClientDrain,
   kGetDispatcher,
   kHasDispatcher,
   kRemoveClient
@@ -14,17 +17,41 @@ const {
   InvalidArgumentError
 } = require('../core/errors')
 const util = require('../core/util')
-const { kUrl } = require('../core/symbols')
+const { kConnecting, kHTTPContext, kUrl } = require('../core/symbols')
 const buildConnector = require('../core/connect')
 
 const kOptions = Symbol('options')
 const kConnections = Symbol('connections')
 const kFactory = Symbol('factory')
+const kProtocol = Symbol('protocol')
+const kProtocolProbe = Symbol('protocol probe')
 
 function defaultFactory (origin, opts) {
   return new Client(origin, opts)
 }
 
+function shouldCreateProtocolProbe (pool, dispatcher) {
+  return dispatcher instanceof Client &&
+    pool[kProtocol] !== 'h1' &&
+    (pool[kOptions].useH2c === true || (pool[kUrl].protocol === 'https:' && pool[kOptions].allowH2 !== false))
+}
+
+function createClient (pool) {
+  const dispatcher = pool[kFactory](pool[kUrl], pool[kOptions])
+
+  // HTTPS does not reveal whether the peer selected h1 or h2 until ALPN
+  // completes. While h2 is still possible, let one Client probe the protocol
+  // and keep later requests in the Pool queue instead of opening one TLS
+  // connection per request. A confirmed h1 connection disables this gate and
+  // restores the usual Pool fan-out.
+  if (shouldCreateProtocolProbe(pool, dispatcher)) {
+    pool[kProtocolProbe] = dispatcher
+  }
+
+  pool[kAddClient](dispatcher)
+  return dispatcher
+}
+
 class Pool extends PoolBase {
   constructor (origin, {
     connections,
@@ -72,30 +99,81 @@ class Pool extends PoolBase {
     this[kUrl] = util.parseOrigin(origin)
     this[kOptions] = { ...util.deepClone(options), connect, allowH2, useH2c, clientTtl, socketPath }
     this[kFactory] = factory
+    this[kProtocol] = null
+    this[kProtocolProbe] = null
 
     this.on('connect', (origin, targets) => {
       if (clientTtl != null && clientTtl > 0) {
         for (const target of targets) {
           Object.assign(target, { ttl: Date.now() })
         }
       }
+
+      const client = targets[targets.length - 1]
+      if (client instanceof Client) {
+        this[kProtocol] = client[kHTTPContext]?.version
+      }
+
+      if (client === this[kProtocolProbe]) {
+        // An h2 Client's drain event releases the requests accumulated during
+        // negotiation onto that Client. If ALPN selected h1, release the probe
+        // immediately and restore normal Pool fan-out instead.
+        if (this[kProtocol] !== 'h2') {
+          this[kProtocolProbe] = null
+          this[kDrainQueue](origin, targets.slice(1))
+        }
+      }
+    })
+
+    this.on('disconnect', (origin, targets) => {
+      if (targets.includes(this[kProtocolProbe])) {
+        this[kProtocolProbe] = null
+        this[kDrainQueue](origin, targets.slice(1))
+      }
     })
 
-    this.on('connectionError', (origin, targets, error) => {
+    this.on('connectionError', (origin, targets) => {
+      let resumeQueued = false
+
       // If a connection error occurs, we remove the client from the pool,
       // and emit a connectionError event. They will not be re-used.
       // Fixes https://github.com/nodejs/undici/issues/3895
       for (const target of targets) {
+        if (target === this[kProtocolProbe]) {
+          this[kProtocolProbe] = null
+          resumeQueued = true
+        }
+
         // Do not use kRemoveClient here, as it will close the client,
         // but the client cannot be closed in this state.
         const idx = this[kClients].indexOf(target)
         if (idx !== -1) {
           this[kClients].splice(idx, 1)
         }
       }
+
+      if (resumeQueued) {
+        this[kDrainQueue](origin, targets.slice(1))
+      }
     })
   }
 
+  [kOnClientBusy] (client) {
+    if (
+      this[kProtocolProbe] === null &&
+      client[kConnecting] &&
+      shouldCreateProtocolProbe(this, client)
+    ) {
+      this[kProtocolProbe] = client
+    }
+  }
+
+  [kOnClientDrain] (client) {
+    if (client === this[kProtocolProbe]) {
+      this[kProtocolProbe] = null
+    }
+  }
+
   [kGetDispatcher] () {
     const clientTtlOption = this[kOptions].clientTtl
     for (let i = 0; i < this[kClients].length; i++) {
@@ -110,10 +188,12 @@ class Pool extends PoolBase {
       }
     }
 
+    if (this[kProtocolProbe] !== null) {
+      return
+    }
+
     if (!this[kConnections] || this[kClients].length < this[kConnections]) {
-      const dispatcher = this[kFactory](this[kUrl], this[kOptions])
-      this[kAddClient](dispatcher)
-      return dispatcher
+      return createClient(this)
     }
   }
 
@@ -130,9 +210,12 @@ class Pool extends PoolBase {
       }
     }
 
+    if (this[kProtocolProbe] !== null) {
+      return false
+    }
+
     if (!this[kConnections] || this[kClients].length < this[kConnections]) {
-      const dispatcher = this[kFactory](this[kUrl], this[kOptions])
-      this[kAddClient](dispatcher)
+      createClient(this)
       return true
     }
 

test/http2-max-concurrent-streams-zero.js

@@ -134,9 +134,9 @@ test('a drained h2 origin must not accumulate wedged connections', async () => {
   await warm.body.dump()
   await new Promise(resolve => setTimeout(resolve, 300))
 
-  // Every request that finds the existing Clients wedged makes the Pool build
-  // another Client + socket, which serves one request and wedges in turn. Those
-  // sockets must not pile up for the lifetime of the process.
+  // Requests that find the existing Clients wedged can make the Pool build a
+  // replacement Client and socket. Those sockets must not pile up for the
+  // lifetime of the process.
   let live = 0
   server.on('session', (session) => {
     live++