From 7a64696e4a6a649bae15c222f8f9296a4d34420b Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Thu, 10 Sep 2026 17:04:03 +0200 Subject: [PATCH] fix(h2): avoid pool fan-out during ALPN negotiation Signed-off-by: Matteo Collina --- docs/docs/api/Agent.md | 16 +-- docs/docs/api/Pool.md | 11 +- lib/dispatcher/pool-base.js | 61 ++++++++++- lib/dispatcher/pool.js | 97 ++++++++++++++-- test/http2-max-concurrent-streams-zero.js | 6 +- test/http2-pipelining-default.js | 128 ++++++++++++++++++++-- 6 files changed, 286 insertions(+), 33 deletions(-) diff --git a/docs/docs/api/Agent.md b/docs/docs/api/Agent.md index 0c59b6eac56..7d2a093634d 100644 --- a/docs/docs/api/Agent.md +++ b/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` diff --git a/docs/docs/api/Pool.md b/docs/docs/api/Pool.md index dc0bd2d0f68..871d4bf9bff 100644 --- a/docs/docs/api/Pool.md +++ b/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' diff --git a/lib/dispatcher/pool-base.js b/lib/dispatcher/pool-base.js index c8a91d42ba5..37487a458f5 100644 --- a/lib/dispatcher/pool-base.js +++ b/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 } diff --git a/lib/dispatcher/pool.js b/lib/dispatcher/pool.js index 39bf7403236..64b5ca2947a 100644 --- a/lib/dispatcher/pool.js +++ b/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,6 +99,8 @@ 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) { @@ -79,13 +108,42 @@ class Pool extends PoolBase { 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) @@ -93,9 +151,29 @@ class Pool extends PoolBase { 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 } diff --git a/test/http2-max-concurrent-streams-zero.js b/test/http2-max-concurrent-streams-zero.js index 23889721959..1c7e5a315ad 100644 --- a/test/http2-max-concurrent-streams-zero.js +++ b/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++ diff --git a/test/http2-pipelining-default.js b/test/http2-pipelining-default.js index 531dc1e01a6..80dc09a6a2b 100644 --- a/test/http2-pipelining-default.js +++ b/test/http2-pipelining-default.js @@ -1,7 +1,9 @@ 'use strict' const { test, after } = require('node:test') +const assert = require('node:assert') const { createSecureServer, createServer } = require('node:http2') +const { createServer: createHttpsServer } = require('node:https') const { once } = require('node:events') const { tspl } = require('@matteo.collina/tspl') const pem = require('@metcoder95/https-pem') @@ -64,6 +66,125 @@ test('h2 client multiplexes concurrent requests by default (#4143)', async t => await t.completed }) +test('Pool waits for ALPN instead of fanning out cold h2 bursts', async t => { + const N = 5 + const DELAY = 200 + const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } })) + const sessions = new Set() + const sessionClosed = [] + const sockets = new Set() + let inFlight = 0 + let peakInFlight = 0 + + server.on('session', session => { + sessions.add(session) + sessionClosed.push(once(session, 'close')) + }) + server.on('connection', socket => sockets.add(socket)) + server.on('stream', stream => { + inFlight++ + peakInFlight = Math.max(peakInFlight, inFlight) + setTimeout(() => { + inFlight-- + stream.respond({ ':status': 200 }) + stream.end('ok') + }, DELAY) + }) + + await once(server.listen(0), 'listening') + + const pool = new Pool(`https://localhost:${server.address().port}`, { + connect: { rejectUnauthorized: false }, + allowH2: true, + keepAliveTimeout: 50 + }) + + t.after(async () => { + await pool.destroy() + await new Promise(resolve => server.close(resolve)) + }) + + const disconnected = once(pool, 'disconnect') + const results = await Promise.all( + Array.from({ length: N }, () => + pool.request({ path: '/', method: 'GET' }) + .then(async response => { + await response.body.text() + return response.statusCode + }) + ) + ) + + assert.deepStrictEqual(results, new Array(N).fill(200)) + assert.strictEqual(sessions.size, 1) + assert.strictEqual(sockets.size, 1) + assert.strictEqual(peakInFlight, N) + + await Promise.all([sessionClosed[0], disconnected]) + peakInFlight = 0 + + const repeatedResults = await Promise.all( + Array.from({ length: N }, () => + pool.request({ path: '/', method: 'GET' }) + .then(async response => { + await response.body.text() + return response.statusCode + }) + ) + ) + + assert.deepStrictEqual(repeatedResults, new Array(N).fill(200)) + assert.strictEqual(sessions.size, 2) + assert.strictEqual(sockets.size, 2) + assert.strictEqual(peakInFlight, N) +}) + +test('Pool restores connection fan-out when ALPN selects h1', async t => { + const N = 5 + const DELAY = 200 + const sockets = new Set() + let inFlight = 0 + let peakInFlight = 0 + const server = createHttpsServer( + await pem.generate({ opts: { keySize: 2048 } }), + (request, response) => { + inFlight++ + peakInFlight = Math.max(peakInFlight, inFlight) + setTimeout(() => { + inFlight-- + response.end('ok') + }, DELAY) + } + ) + + server.on('connection', socket => sockets.add(socket)) + await once(server.listen(0), 'listening') + + const pool = new Pool(`https://localhost:${server.address().port}`, { + connect: { rejectUnauthorized: false }, + allowH2: true + }) + + t.after(async () => { + await pool.destroy() + await new Promise(resolve => server.close(resolve)) + }) + + const results = await Promise.all( + Array.from({ length: N }, () => + pool.request({ path: '/', method: 'GET' }) + .then(async response => { + await response.body.text() + return response.statusCode + }) + ) + ) + + assert.deepStrictEqual(results, new Array(N).fill(200)) + assert.strictEqual(sockets.size, N) + assert.strictEqual(peakInFlight, N) +}) + test('Pool with connections=1 multiplexes h2 streams on the single session (#4143)', async t => { const N = 5 const DELAY = 200 @@ -88,13 +209,6 @@ test('Pool with connections=1 multiplexes h2 streams on the single session (#414 // With connections=1 the Pool funnels every dispatch through a single // Client; that Client's h2 context lets the N concurrent requests // multiplex on one session. - // - // The unconstrained Pool case (no `connections` cap) is intentionally - // not asserted here: during the TLS/ALPN handshake the Client cannot - // yet know whether h2 will be negotiated, so the per-Client `kPending` - // gate still fans out and a cold burst opens one socket per request. - // Resolving that needs a Pool-level lazy-connect strategy and is left - // for a follow-up (see #4143 discussion). const pool = new Pool(`https://localhost:${server.address().port}`, { connect: { rejectUnauthorized: false }, allowH2: true,