From 865044c22625752157d35fed1319c1f06b67ab41 Mon Sep 17 00:00:00 2001 From: "Owen D. Goode" <86684779+owendavidgoode@users.noreply.github.com> Date: Thu, 29 Oct 2026 23:02:14 -0500 Subject: [PATCH 1/2] fix(utils): cancel late HTTP/2 response bodies --- src/utils.ts | 10 +++ test/listener.test.ts | 142 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 151 insertions(+), 1 deletion(-) diff --git a/src/utils.ts b/src/utils.ts index d9e8a444..5f486df2 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -1,4 +1,5 @@ import type { OutgoingHttpHeaders } from 'node:http' +import { Http2ServerResponse } from 'node:http2' import type { Writable } from 'node:stream' export async function readWithoutBlocking( @@ -16,6 +17,15 @@ export function writeFromReadableStreamDefaultReader( reader.cancel(error).catch(() => {}) } + // HTTP/2 responses expose lifecycle state on their backing stream, not the wrapper. + if ( + writable instanceof Http2ServerResponse && + (writable.stream.closed || writable.stream.destroyed) + ) { + cancel() + return reader.closed + } + writable.on('close', cancel) writable.on('error', cancel) ;(currentReadPromise ?? reader.read()).then(flow, handleStreamError) diff --git a/test/listener.test.ts b/test/listener.test.ts index 1dd17d00..002840ec 100644 --- a/test/listener.test.ts +++ b/test/listener.test.ts @@ -12,7 +12,7 @@ import { GlobalHeaders } from '../src/headers' import { getRequestListener } from '../src/listener' import { GlobalRequest, Request as LightweightRequest, RequestError } from '../src/request' import { GlobalResponse, Response as LightweightResponse } from '../src/response' -import { requestServer } from './helpers/request' +import { requestServer, requestServerHttp2 } from './helpers/request' const withTimeout = async (promise: Promise, message: string): Promise => { let timeoutId: ReturnType | undefined @@ -456,6 +456,146 @@ describe('Abort request', () => { ) }) +describe.each([ + ['native', GlobalResponse], + ['cached', LightweightResponse], +] as const)('HTTP/2 response body cancellation - %s', (_name, ResponseClass) => { + it.each(['stream', 'session'] as const)( + 'should cancel a body returned after the client closes the %s', + async (target) => { + let resolveReady!: () => void + const ready = new Promise((r) => { + resolveReady = r + }) + let resolveCancelled!: () => void + const cancelled = new Promise((r) => { + resolveCancelled = r + }) + let resolveHandled!: () => void + const handled = new Promise((r) => { + resolveHandled = r + }) + const cancel = vi.fn(resolveCancelled) + const requestListener = getRequestListener(async (_req, { outgoing }) => { + // Wait for the actual response close event, not just the client's reset. + const closed = new Promise((r) => outgoing.once('close', r)) + resolveReady() + await closed + return new ResponseClass( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('first chunk')) + }, + cancel, + }) + ) + }) + const server = createHttp2Server(async (req, res) => { + await requestListener(req, res) + resolveHandled() + }) + await new Promise((r) => server.listen(0, '127.0.0.1', r)) + const { port } = server.address() as AddressInfo + const client = http2Connect(`http://127.0.0.1:${port}`) + + try { + const stream = client.request({ ':path': '/' }) + stream.on('error', () => {}) + await withTimeout(ready, 'request did not reach the handler') + if (target === 'stream') { + stream.close(http2Constants.NGHTTP2_CANCEL) + } else { + client.destroy() + } + + await withTimeout(cancelled, 'late response body was not cancelled') + await withTimeout(handled, 'request listener did not finish') + expect(cancel).toHaveBeenCalledTimes(1) + } finally { + client.destroy() + await new Promise((r) => server.close(() => r())) + } + } + ) + + it('should send a complete body without cancelling it', async () => { + const cancel = vi.fn() + const server = createHttp2Server( + getRequestListener( + () => + new ResponseClass( + new ReadableStream({ + async pull(controller) { + await new Promise((resolve) => setImmediate(resolve)) + controller.enqueue(new TextEncoder().encode('complete body')) + controller.close() + }, + cancel, + }) + ) + ) + ) + + const response = await requestServerHttp2(server, { path: '/' }) + expect(response.status).toBe(200) + expect(await response.text()).toBe('complete body') + expect(cancel).not.toHaveBeenCalled() + }) +}) + +it('should cancel an HTTP/2 body when the client closes during the native pre-read', async () => { + let resolvePreRead!: () => void + const preRead = new Promise((r) => { + resolvePreRead = r + }) + let resolveClosed!: () => void + const closed = new Promise((r) => { + resolveClosed = r + }) + let resolveCancelled!: () => void + const cancelled = new Promise((r) => { + resolveCancelled = r + }) + const scheduleTimeout = global.setTimeout + const timer = vi.spyOn(global, 'setTimeout').mockImplementation((callback, delay, ...args) => { + if (delay === undefined) { + // Hold the native response's pre-read tick until the real close event has fired. + resolvePreRead() + return scheduleTimeout(() => void closed.then(() => callback(...args)), 0) + } + return scheduleTimeout(callback, delay, ...args) + }) + const server = createHttp2Server( + getRequestListener((_req, { outgoing }) => { + outgoing.once('close', resolveClosed) + return new GlobalResponse( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('first chunk')) + }, + cancel: resolveCancelled, + }) + ) + }) + ) + await new Promise((r) => server.listen(0, '127.0.0.1', r)) + const { port } = server.address() as AddressInfo + const client = http2Connect(`http://127.0.0.1:${port}`) + + try { + const stream = client.request({ ':path': '/' }) + stream.on('error', () => {}) + await withTimeout(preRead, 'native response did not reach the pre-read tick') + stream.close(http2Constants.NGHTTP2_CANCEL) + await withTimeout(closed, 'response did not close') + await withTimeout(cancelled, 'response body was not cancelled after the pre-read') + } finally { + timer.mockRestore() + client.destroy() + await new Promise((r) => server.close(() => r())) + } +}) + describe('Abort request - error path', () => { const runAbortDuringErrorHandlerCase = async (mode: 'sync' | 'async') => { let capturedReq: Request | undefined From c043d9cb29e4f9ac9d047769fc71d38a2f8c1ae8 Mon Sep 17 00:00:00 2001 From: "Owen D. Goode" <86684779+owendavidgoode@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:21:24 -0500 Subject: [PATCH 2/2] test(listener): tighten HTTP/2 cancellation coverage --- test/listener.test.ts | 139 ++++++++++++++++++++++++++++++++++-------- 1 file changed, 113 insertions(+), 26 deletions(-) diff --git a/test/listener.test.ts b/test/listener.test.ts index 002840ec..ae82f109 100644 --- a/test/listener.test.ts +++ b/test/listener.test.ts @@ -475,12 +475,19 @@ describe.each([ const handled = new Promise((r) => { resolveHandled = r }) + let assertListenersClean!: () => void const cancel = vi.fn(resolveCancelled) const requestListener = getRequestListener(async (_req, { outgoing }) => { // Wait for the actual response close event, not just the client's reset. const closed = new Promise((r) => outgoing.once('close', r)) resolveReady() await closed + const closeListeners = outgoing.listeners('close') + const errorListeners = outgoing.listeners('error') + assertListenersClean = () => { + expect(outgoing.listeners('close')).toEqual(closeListeners) + expect(outgoing.listeners('error')).toEqual(errorListeners) + } return new ResponseClass( new ReadableStream({ start(controller) { @@ -511,6 +518,8 @@ describe.each([ await withTimeout(cancelled, 'late response body was not cancelled') await withTimeout(handled, 'request listener did not finish') expect(cancel).toHaveBeenCalledTimes(1) + expect(cancel).toHaveBeenCalledWith(undefined) + assertListenersClean() } finally { client.destroy() await new Promise((r) => server.close(() => r())) @@ -518,6 +527,57 @@ describe.each([ } ) + it('should cancel after the client receives a prefix and resets the stream', async () => { + let resolveCancelled!: () => void + const cancelled = new Promise((r) => { + resolveCancelled = r + }) + let resolveHandled!: () => void + const handled = new Promise((r) => { + resolveHandled = r + }) + const cancel = vi.fn(resolveCancelled) + const requestListener = getRequestListener( + () => + new ResponseClass( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('prefix')) + }, + cancel, + }) + ) + ) + const server = createHttp2Server(async (req, res) => { + await requestListener(req, res) + resolveHandled() + }) + let client: ReturnType | undefined + + try { + await new Promise((r) => server.listen(0, '127.0.0.1', r)) + const { port } = server.address() as AddressInfo + client = http2Connect(`http://127.0.0.1:${port}`) + const stream = client.request({ ':path': '/' }) + stream.on('error', () => {}) + const prefix = new Promise((r) => stream.once('data', r)) + expect((await withTimeout(prefix, 'response prefix was not received')).toString()).toBe( + 'prefix' + ) + stream.close(http2Constants.NGHTTP2_CANCEL) + + await withTimeout(cancelled, 'streaming response body was not cancelled') + await withTimeout(handled, 'request listener did not finish') + expect(cancel).toHaveBeenCalledTimes(1) + expect(cancel).toHaveBeenCalledWith(undefined) + } finally { + client?.destroy() + if (server.listening) { + await new Promise((r) => server.close(() => r())) + } + } + }) + it('should send a complete body without cancelling it', async () => { const cancel = vi.fn() const server = createHttp2Server( @@ -556,43 +616,70 @@ it('should cancel an HTTP/2 body when the client closes during the native pre-re const cancelled = new Promise((r) => { resolveCancelled = r }) + let resolveHandled!: () => void + const handled = new Promise((r) => { + resolveHandled = r + }) + const cancel = vi.fn(resolveCancelled) const scheduleTimeout = global.setTimeout - const timer = vi.spyOn(global, 'setTimeout').mockImplementation((callback, delay, ...args) => { - if (delay === undefined) { - // Hold the native response's pre-read tick until the real close event has fired. - resolvePreRead() - return scheduleTimeout(() => void closed.then(() => callback(...args)), 0) - } - return scheduleTimeout(callback, delay, ...args) - }) - const server = createHttp2Server( - getRequestListener((_req, { outgoing }) => { - outgoing.once('close', resolveClosed) - return new GlobalResponse( - new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode('first chunk')) - }, - cancel: resolveCancelled, - }) - ) + let restoreTimer = () => {} + let assertListenersClean!: () => void + const requestListener = getRequestListener((_req, { outgoing }) => { + outgoing.once('close', () => { + const closeListeners = outgoing.listeners('close') + const errorListeners = outgoing.listeners('error') + assertListenersClean = () => { + expect(outgoing.listeners('close')).toEqual(closeListeners) + expect(outgoing.listeners('error')).toEqual(errorListeners) + } + resolveClosed() }) - ) - await new Promise((r) => server.listen(0, '127.0.0.1', r)) - const { port } = server.address() as AddressInfo - const client = http2Connect(`http://127.0.0.1:${port}`) + const response = new GlobalResponse( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('first chunk')) + }, + cancel, + }) + ) + // Install only after connection setup, and intercept just the first pre-read tick. + const timer = vi.spyOn(global, 'setTimeout').mockImplementation((callback, delay, ...args) => { + if (delay === undefined) { + timer.mockRestore() + resolvePreRead() + return scheduleTimeout(() => void closed.then(() => callback(...args)), 0) + } + return scheduleTimeout(callback, delay, ...args) + }) + restoreTimer = () => timer.mockRestore() + return response + }) + const server = createHttp2Server(async (req, res) => { + await requestListener(req, res) + resolveHandled() + }) + let client: ReturnType | undefined try { + await new Promise((r) => server.listen(0, '127.0.0.1', r)) + const { port } = server.address() as AddressInfo + client = http2Connect(`http://127.0.0.1:${port}`) const stream = client.request({ ':path': '/' }) stream.on('error', () => {}) await withTimeout(preRead, 'native response did not reach the pre-read tick') stream.close(http2Constants.NGHTTP2_CANCEL) await withTimeout(closed, 'response did not close') await withTimeout(cancelled, 'response body was not cancelled after the pre-read') + await withTimeout(handled, 'request listener did not finish') + expect(cancel).toHaveBeenCalledTimes(1) + expect(cancel).toHaveBeenCalledWith(undefined) + assertListenersClean() } finally { - timer.mockRestore() - client.destroy() - await new Promise((r) => server.close(() => r())) + restoreTimer() + client?.destroy() + if (server.listening) { + await new Promise((r) => server.close(() => r())) + } } })