From 513f7f5766ae43e9e7e84942997b480de3da68d8 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Tue, 28 Jul 2026 22:03:51 +0200 Subject: [PATCH 1/2] fix(h2): ensure every request settles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three ways an HTTP/2 request could stop making progress without the caller ever hearing back: 1. A stream that closed with no response and no error had its queue slot freed by completeRequestStream() without anyone calling request.onResponseError(), so the request simply vanished. This is reachable after a GOAWAY: detachRequestStreamForClose() releases the request-facing listeners and installs `once('error', noop)`, but leaves the 'close' listener and kRequestStreamState in place, so the abandoned stream later completes — and splices out — the request that has since been requeued onto another session. severRequestStream() now unbinds such a stream for good, and completeRequestStream() errors a request whose stream closed before the response was complete. 2. A peer that refuses a request with GOAWAY(lastStreamID = 0) had it requeued with no attempt budget, so a peer that keeps refusing turned one request into an unbounded connect/refuse/reconnect loop that never settled. Refusals now count against MAX_REFUSED_ATTEMPTS. 3. A peer may advertise SETTINGS_MAX_CONCURRENT_STREAMS = 0 to refuse new streams (RFC 9113 6.5.2). busy() then reports the client as permanently busy, and queued requests can neither open a stream — so no per-stream timeout covers them — nor trigger a reconnect, so the SETTINGS frame that would lift the limit can never arrive. headersTimeout now covers a request that cannot be sent at all, and the unusable session is dropped so the next request gets a fresh connection. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01A49JamgF2TkZHu5h58ChUM Signed-off-by: Matteo Collina --- lib/dispatcher/client-h2.js | 123 ++++++++++++- test/http2-goaway-refusal-storm.js | 208 ++++++++++++++++++++++ test/http2-max-concurrent-streams-zero.js | 162 +++++++++++++++++ test/http2-request-never-settles.js | 173 ++++++++++++++++++ 4 files changed, 665 insertions(+), 1 deletion(-) create mode 100644 test/http2-goaway-refusal-storm.js create mode 100644 test/http2-max-concurrent-streams-zero.js create mode 100644 test/http2-request-never-settles.js diff --git a/lib/dispatcher/client-h2.js b/lib/dispatcher/client-h2.js index 93531def59c..6319b8c3670 100644 --- a/lib/dispatcher/client-h2.js +++ b/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) diff --git a/test/http2-goaway-refusal-storm.js b/test/http2-goaway-refusal-storm.js new file mode 100644 index 00000000000..4e0dd28ac12 --- /dev/null +++ b/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)` + ) +}) diff --git a/test/http2-max-concurrent-streams-zero.js b/test/http2-max-concurrent-streams-zero.js new file mode 100644 index 00000000000..23889721959 --- /dev/null +++ b/test/http2-max-concurrent-streams-zero.js @@ -0,0 +1,162 @@ +'use strict' + +const assert = require('node:assert') +const { test, after } = require('node:test') +const { createSecureServer } = require('node:http2') +const { once } = require('node:events') + +const pem = require('@metcoder95/https-pem') + +const { Client, Agent } = require('..') +const { kMaxConcurrentStreams } = require('../lib/core/symbols') + +// RFC 9113 §6.5.2 lets a server advertise SETTINGS_MAX_CONCURRENT_STREAMS = 0 +// to temporarily refuse new streams (graceful drain, overload shedding, a load +// balancer draining a backend). +// +// The value is stored on the Client and makes client-h2's busy() degenerate to +// `running >= 0`, i.e. permanently busy. Nothing covers a request that is still +// queued: headersTimeout/bodyTimeout are armed on a stream that is never +// created, the h2 idle timeout bails while kSize !== 0, PING keeps succeeding, +// and _resume() returns at busy(request) so no new connection is ever made -- +// which means a later SETTINGS frame that would lift the limit can never +// arrive either. + +function settleOrTimeout (promise, ms) { + let timer + return Promise.race([ + promise.then(() => 'settled', () => 'settled'), + new Promise((resolve) => { timer = setTimeout(() => resolve('stuck'), ms) }) + ]).finally(() => clearTimeout(timer)) +} + +// A wedged h2 client unrefs its session and arms no timer, so the event loop +// can drain while a request is still outstanding. Hold it open for the +// duration of the test, otherwise the runner reports "Promise resolution is +// still pending but the event loop has already resolved" and hangs. +function holdEventLoop () { + const timer = setInterval(() => {}, 1000) + return () => clearInterval(timer) +} + +async function drainingServer () { + const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } })) + + // A wedged h2 session never ends on its own, so server.close() would hang. + const sessions = new Set() + server.on('session', (session) => { + sessions.add(session) + session.on('close', () => sessions.delete(session)) + }) + server.shutdown = () => { + for (const session of sessions) session.destroy() + server.close() + } + + server.on('stream', (stream) => { + stream.on('error', () => {}) + stream.respond({ ':status': 200 }) + stream.end('ok') + }) + + // Serve normally, then start draining. + server.on('session', (session) => { + session.once('stream', () => { + setTimeout(() => { + try { + session.settings({ maxConcurrentStreams: 0 }) + } catch {} + }, 50) + }) + }) + + await once(server.listen(0), 'listening') + return server +} + +test('a request queued behind maxConcurrentStreams: 0 must not hang forever', async () => { + const release = holdEventLoop() + const server = await drainingServer() + + const client = new Client(`https://localhost:${server.address().port}`, { + connect: { rejectUnauthorized: false }, + allowH2: true, + headersTimeout: 500, + bodyTimeout: 500 + }) + after(async () => { + await client.destroy() + server.shutdown() + release() + }) + + const warm = await client.request({ path: '/', method: 'GET' }) + assert.strictEqual(warm.statusCode, 200) + await warm.body.dump() + + // Let the drain SETTINGS land. + await new Promise(resolve => setTimeout(resolve, 300)) + assert.strictEqual(client[kMaxConcurrentStreams], 0, 'server has drained to 0 streams') + + // The request never reaches a stream, so no per-stream timeout can fire. + // Something must still settle it rather than queueing it indefinitely. + const pending = client.request({ path: '/after-drain', method: 'GET' }) + .then(res => res.body.dump(), () => {}) + const outcome = await settleOrTimeout(pending, 2000) + + // destroy() errors whatever is still queued, so the dangling promise settles + await client.destroy() + await pending + + assert.strictEqual(outcome, 'settled', 'request queued behind a 0-stream limit never settled') +}) + +test('a drained h2 origin must not accumulate wedged connections', async () => { + const release = holdEventLoop() + const server = await drainingServer() + + const origin = `https://localhost:${server.address().port}` + const agent = new Agent({ + connect: { rejectUnauthorized: false }, + allowH2: true, + headersTimeout: 500, + bodyTimeout: 500 + }) + after(async () => { + await agent.destroy() + server.shutdown() + release() + }) + + // Warm the pool and let the drain SETTINGS land, so every later request + // really does meet a wedged Client. + const warm = await agent.request({ origin, path: '/warm', method: 'GET' }) + 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. + let live = 0 + server.on('session', (session) => { + live++ + session.on('close', () => live--) + }) + + const unsettled = [] + for (let round = 0; round < 3; round++) { + const pending = [...Array(4)].map((_, i) => + agent.request({ origin, path: `/r${round}-${i}`, method: 'GET' }).then(res => res.body.dump(), () => {})) + const outcomes = await Promise.all(pending.map(p => settleOrTimeout(p, 3000))) + unsettled.push(...outcomes.filter(o => o === 'stuck')) + } + + // Give the drained sessions time to be torn down. + await new Promise(resolve => setTimeout(resolve, 1000)) + const stillOpen = live + + await agent.destroy() + + assert.strictEqual(unsettled.length, 0, 'no request may be left unsettled') + assert.ok(stillOpen <= 4, `${stillOpen} wedged connections were still open after 12 requests`) +}) diff --git a/test/http2-request-never-settles.js b/test/http2-request-never-settles.js new file mode 100644 index 00000000000..a5c2056473f --- /dev/null +++ b/test/http2-request-never-settles.js @@ -0,0 +1,173 @@ +'use strict' + +const assert = require('node:assert') +const { test, after } = require('node:test') +const { constants, createSecureServer } = require('node:http2') +const { once } = require('node:events') +const { Readable } = require('node:stream') + +const pem = require('@metcoder95/https-pem') + +const { Agent } = require('..') + +// completeRequestStream() runs on an h2 stream's 'close': +// +// releaseRequestStream(this) +// if (state.pendingEnd && !aborted && !completed) request.onResponseEnd(...) +// finalizeRequest(state) // frees the queue slot unconditionally +// +// There is no branch for "the stream closed with no response and no error". In +// that case the request is spliced out of the queue and nothing ever calls +// request.onResponseError(), so the caller's promise never settles -- while the +// client's own stats look perfectly healthy. +// +// The window is reachable once a GOAWAY has detached a stream: +// detachRequestStreamForClose() -> releaseRequestStream() removes the +// request-facing listeners and installs `once('error', noop)`, swallowing any +// later error, but leaves the 'close' listener and stream[kRequestStreamState] +// in place. +// +// Reproducing it needs several of those paths to interleave, so this drives a +// seeded mix of the churn a real h2 deployment produces and asserts the one +// invariant that must always hold: every request settles, one way or another. + +const SEEDS = [7, 42] +const ROUNDS = 12 +const CONCURRENCY = 8 + +function makeRandom (seed) { + let state = seed + return () => { + state = (state * 1103515245 + 12345) & 0x7fffffff + return state / 0x7fffffff + } +} + +function settleOrTimeout (promise, ms, label) { + let timer + return Promise.race([ + promise.then(() => null, () => null), + new Promise((resolve) => { timer = setTimeout(() => resolve(label), ms) }) + ]).finally(() => clearTimeout(timer)) +} + +async function churningServer (rnd) { + const server = createSecureServer( + await pem.generate({ opts: { keySize: 2048 } }), + { settings: { maxConcurrentStreams: 4 } } + ) + + const actions = [ + 'ok', 'ok', 'ok', 'ok', 'ok', + 'slow', 'rst-cancel', 'rst-refused', 'goaway-drain', 'goaway-refuse', 'kill-session', 'no-answer' + ] + + server.on('stream', (stream) => { + stream.on('error', () => {}) + const action = actions[Math.floor(rnd() * actions.length)] + + try { + switch (action) { + case 'ok': + stream.respond({ ':status': 200 }) + stream.end('ok') + break + case 'slow': + stream.respond({ ':status': 200 }) + setTimeout(() => { try { stream.end('slow') } catch {} }, 30) + break + case 'rst-cancel': + stream.close(constants.NGHTTP2_CANCEL) + break + case 'rst-refused': + stream.close(constants.NGHTTP2_REFUSED_STREAM) + break + case 'goaway-drain': + // finish this stream, refuse anything newer + stream.respond({ ':status': 200 }) + stream.end('draining') + stream.session.goaway(constants.NGHTTP2_NO_ERROR, stream.id) + break + case 'goaway-refuse': + stream.session.goaway(constants.NGHTTP2_NO_ERROR, 0) + break + case 'kill-session': + stream.session.destroy() + break + case 'no-answer': + break + } + } catch {} + }) + + const sessions = new Set() + server.on('session', (session) => { + sessions.add(session) + session.on('close', () => sessions.delete(session)) + }) + server.shutdown = () => { + for (const session of sessions) session.destroy() + server.close() + } + + await once(server.listen(0), 'listening') + return server +} + +for (const seed of SEEDS) { + test(`every h2 request settles under connection churn (seed ${seed})`, async () => { + const timer = setInterval(() => {}, 1000) + const rnd = makeRandom(seed) + const server = await churningServer(rnd) + + const origin = `https://localhost:${server.address().port}` + const agent = new Agent({ + connect: { rejectUnauthorized: false }, + allowH2: true, + headersTimeout: 500, + bodyTimeout: 500 + }) + after(async () => { + await agent.destroy() + server.shutdown() + clearInterval(timer) + }) + + const unsettled = [] + + for (let round = 0; round < ROUNDS && unsettled.length === 0; round++) { + const jobs = [] + + for (let i = 0; i < CONCURRENCY; i++) { + const kind = ['get', 'get', 'post-buffer', 'post-stream', 'abort'][Math.floor(rnd() * 5)] + const ac = new AbortController() + const opts = { origin, path: `/r${round}-${i}`, method: 'GET', signal: ac.signal } + + if (kind === 'post-buffer') { + opts.method = 'POST' + opts.body = Buffer.alloc(2048, 'a') + } else if (kind === 'post-stream') { + opts.method = 'POST' + opts.body = Readable.from([Buffer.alloc(512, 'b')]) + } else if (kind === 'abort') { + setTimeout(() => ac.abort(), Math.floor(rnd() * 8)) + } + + // A dropped request is removed from the client queue outright, so not + // even destroy() can reach it: never await these directly. + const pending = agent.request(opts).then(res => res.body.dump(), () => {}) + // A dropped request never settles at all, so a generous watchdog costs + // nothing in detection power and keeps this stable under a loaded CI. + jobs.push(settleOrTimeout(pending, 15000, `/r${round}-${i}`)) + } + + for (const label of await Promise.all(jobs)) { + if (label !== null) unsettled.push(label) + } + } + + await agent.destroy() + + assert.deepStrictEqual(unsettled, [], 'requests were dropped without a response and without an error') + }) +} From 09ef22f41df8fb7fb8fee530a06da626693d3a65 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 29 Jul 2026 01:06:19 +0200 Subject: [PATCH 2/2] test(h2): cover a stream cancelled before the response RST_STREAM(CANCEL) received before the response is the one reset code Node reports as a bare 'close' on the client stream -- no 'end' (unlike NO_ERROR) and no 'error' (unlike PROTOCOL_ERROR / INTERNAL_ERROR / REFUSED_STREAM) -- and destroying the stream unenrolls its timeout, so no 'timeout' follows either. That is the deterministic path into the case completeRequestStream() did not handle, and it is reachable against ordinary infrastructure: a proxy sends this upstream whenever its own downstream client goes away. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01A49JamgF2TkZHu5h58ChUM Signed-off-by: Matteo Collina --- test/http2-stream-cancel.js | 91 +++++++++++++++++++++++++++++++++++++ 1 file changed, 91 insertions(+) create mode 100644 test/http2-stream-cancel.js diff --git a/test/http2-stream-cancel.js b/test/http2-stream-cancel.js new file mode 100644 index 00000000000..69a90967f73 --- /dev/null +++ b/test/http2-stream-cancel.js @@ -0,0 +1,91 @@ +'use strict' + +const assert = require('node:assert') +const { test, after } = require('node:test') +const { constants, createSecureServer } = require('node:http2') +const { once } = require('node:events') + +const pem = require('@metcoder95/https-pem') + +const { Client } = require('..') + +// RST_STREAM(CANCEL) received before the response is the one reset code Node +// reports as a bare 'close' on the client stream: +// +// RST NO_ERROR -> 'end', 'close' +// RST PROTOCOL_ERROR -> 'error', 'close' +// RST INTERNAL_ERROR -> 'error', 'close' +// RST REFUSED_STREAM -> 'error', 'close' +// RST CANCEL -> 'close' <- nothing to act on +// +// CANCEL is the code Node uses when a stream is destroyed locally, so an +// incoming one is treated as a plain teardown rather than an error. Destroying +// the stream also unenrolls its timeout, so no 'timeout' follows either. +// +// completeRequestStream() therefore reached finalizeRequest() with no response +// delivered and no error reported, freeing the queue slot without ever calling +// request.onResponseError(): the caller's promise never settled, whatever +// headersTimeout and bodyTimeout were set to. Proxies send this upstream +// whenever their own downstream client goes away. + +test('a stream cancelled before the response settles the request', async () => { + const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } })) + + server.on('stream', (stream) => { + stream.on('error', () => {}) + setTimeout(() => { + try { + stream.close(constants.NGHTTP2_CANCEL) + } catch {} + }, 30) + }) + + after(() => server.close()) + await once(server.listen(0), 'listening') + + const client = new Client(`https://localhost:${server.address().port}`, { + connect: { rejectUnauthorized: false }, + allowH2: true, + // Long enough that only a real completion can settle this in time. + headersTimeout: 30000, + bodyTimeout: 30000 + }) + after(() => client.destroy()) + + let timer + const outcome = await Promise.race([ + client.request({ path: '/', method: 'GET' }).then(() => 'resolved', (err) => err.code), + new Promise((resolve) => { timer = setTimeout(() => resolve('never settled'), 5000) }) + ]).finally(() => clearTimeout(timer)) + + assert.strictEqual(outcome, 'UND_ERR_INFO', 'a cancelled stream must not strand its request') +}) + +test('a stream cancelled after the response headers still completes', async () => { + const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } })) + + server.on('stream', (stream) => { + stream.on('error', () => {}) + stream.respond({ ':status': 200 }) + setTimeout(() => { + try { + stream.close(constants.NGHTTP2_CANCEL) + } catch {} + }, 30) + }) + + after(() => server.close()) + await once(server.listen(0), 'listening') + + const client = new Client(`https://localhost:${server.address().port}`, { + connect: { rejectUnauthorized: false }, + allowH2: true, + headersTimeout: 30000, + bodyTimeout: 30000 + }) + after(() => client.destroy()) + + const res = await client.request({ path: '/', method: 'GET' }) + assert.strictEqual(res.statusCode, 200) + assert.strictEqual(await res.body.text(), '') +})