diff --git a/packages/fastify/src/response.test.ts b/packages/fastify/src/response.test.ts index 12b4e12f..0405ba3f 100644 --- a/packages/fastify/src/response.test.ts +++ b/packages/fastify/src/response.test.ts @@ -1,4 +1,5 @@ import { Buffer } from 'node:buffer' +import http2 from 'node:http2' import { Readable, Writable } from 'node:stream' import FastifyCookie from '@fastify/cookie' import * as StandardServerNode from '@standard-server/node' @@ -590,4 +591,148 @@ describe('sendStandardResponse', () => { expect(reply.send).not.toHaveBeenCalled() }) }) + + describe('head request with a stream body', () => { + it('releases the body on an auto-exposed HEAD route', async ({ onTestFinished }) => { + const { body, isReleased } = createEndlessBody() + let sending: Promise | undefined + + const fastify = Fastify() + onTestFinished(() => fastify.close()) + + fastify.get('/', async (req, reply) => { + sending = sendStandardResponse(reply, { status: 201, headers: { 'x-custom-header': 'custom-value' }, body }) + await sending + }) + + await fastify.ready() + const res = await request(fastify.server).head('/') + + await expect(sending).resolves.toBeUndefined() + + expect(res.status).toBe(201) + expect(res.headers).toMatchObject({ 'x-custom-header': 'custom-value' }) + + await vi.waitFor(() => { + expect(isReleased()).toBe(true) + }) + }) + + it('releases the body on an auto-exposed http2 HEAD route', async ({ onTestFinished }) => { + const { body, isReleased } = createEndlessBody() + let sending: Promise | undefined + + const fastify = Fastify({ http2: true }) + onTestFinished(() => fastify.close()) + + fastify.get('/', async (req, reply) => { + sending = sendStandardResponse(reply, { status: 201, headers: { 'x-custom-header': 'custom-value' }, body }) + await sending + }) + + const response = await requestHttp2Head(await fastify.listen({ port: 0, host: '127.0.0.1' })) + + await expect(sending).resolves.toBeUndefined() + + expect(response.headers).toMatchObject({ ':status': 201, 'x-custom-header': 'custom-value' }) + expect(response.body).toBe('') + + await vi.waitFor(() => { + expect(isReleased()).toBe(true) + }) + }) + + it('rejects and releases the body on an explicit http2 HEAD route', async ({ onTestFinished }) => { + const { body, isReleased } = createEndlessBody() + let sending: Promise | undefined + + const fastify = Fastify({ http2: true }) + onTestFinished(() => fastify.close()) + + fastify.head('/', async (req, reply) => { + sending = sendStandardResponse(reply, { status: 201, headers: { 'x-custom-header': 'custom-value' }, body }) + await sending.catch(() => {}) + }) + + const response = await requestHttp2Head(await fastify.listen({ port: 0, host: '127.0.0.1' })) + + await expect(sending).rejects.toMatchObject({ code: 'ERR_STREAM_WRITE_AFTER_END' }) + + expect(response.headers).toMatchObject({ ':status': 201, 'x-custom-header': 'custom-value' }) + expect(response.body).toBe('') + + await vi.waitFor(() => { + expect(isReleased()).toBe(true) + }) + }) + }) + + it('rejects with the stream error when the client resets an http2 stream with an error code', async ({ onTestFinished }) => { + const { body, isReleased } = createEndlessBody() + let sending: Promise | undefined + + const fastify = Fastify({ http2: true }) + onTestFinished(() => fastify.close()) + + fastify.get('/', async (req, reply) => { + sending = sendStandardResponse(reply, { status: 200, headers: {}, body }) + await sending.catch(() => {}) + }) + + const client = http2.connect(await fastify.listen({ port: 0, host: '127.0.0.1' })) + onTestFinished(() => client.close()) + + const reqStream = client.request({ ':path': '/' }) + reqStream.once('error', () => {}) + + await new Promise(r => reqStream.once('data', r)) + reqStream.close(http2.constants.NGHTTP2_INTERNAL_ERROR) + + await expect(sending).rejects.toMatchObject({ code: 'ERR_HTTP2_STREAM_ERROR' }) + + await vi.waitFor(() => { + expect(isReleased()).toBe(true) + }) + }) }) + +function createEndlessBody() { + let released = false + + const body = (async function* () { + try { + while (true) { + yield 'foo' + await new Promise(r => setTimeout(r, 10)) + } + } + finally { + released = true + } + })() + + return { body, isReleased: () => released } +} + +async function requestHttp2Head(origin: string): Promise<{ headers: http2.IncomingHttpHeaders, body: string }> { + const client = http2.connect(origin) + + try { + const reqStream = client.request({ ':path': '/', ':method': 'HEAD' }) + + const chunks: Buffer[] = [] + reqStream.on('data', (chunk: Buffer) => chunks.push(chunk)) + + const headers = await new Promise((resolve, reject) => { + reqStream.once('response', resolve) + reqStream.once('error', reject) + }) + + await new Promise(r => reqStream.once('close', r)) + + return { headers, body: Buffer.concat(chunks).toString() } + } + finally { + client.close() + } +} diff --git a/packages/fastify/src/response.ts b/packages/fastify/src/response.ts index 4b304530..3e00e2f6 100644 --- a/packages/fastify/src/response.ts +++ b/packages/fastify/src/response.ts @@ -32,8 +32,16 @@ export async function sendStandardResponse( return } - reply.raw.once('error', reject) - reply.raw.once('close', resolve) + const connection = 'stream' in reply.raw ? reply.raw.stream : reply.raw + + connection.once('error', reject) + connection.once('close', () => { + if (typeof resBody === 'object' && !resBody.closed) { + resBody.destroy() + } + + resolve() + }) try { reply.status(standardResponse.status) @@ -46,7 +54,7 @@ export async function sendStandardResponse( } } - // fastify pipes and cleans up the stream body itself, no manual piping needed + // fastify pipes the stream body itself, no manual piping needed reply.send(resBody) } catch (error) { diff --git a/packages/node/src/response.test.ts b/packages/node/src/response.test.ts index 2801b6eb..209230b4 100644 --- a/packages/node/src/response.test.ts +++ b/packages/node/src/response.test.ts @@ -1,6 +1,9 @@ import type { StandardResponse } from '@standard-server/core' import type { IncomingMessage, ServerResponse } from 'node:http' +import type { Http2ServerRequest, Http2ServerResponse } from 'node:http2' +import type { AddressInfo } from 'node:net' import { Buffer } from 'node:buffer' +import http2 from 'node:http2' import Stream from 'node:stream' import request from 'supertest' import * as Body from './body' @@ -498,4 +501,141 @@ describe('sendStandardResponse', () => { expect((responseStream as any).setHeader).not.toHaveBeenCalled() }) }) + + describe('http2 HEAD request', () => { + it('sends the status and headers', async () => { + let sending: Promise | undefined + + const response = await requestHttp2Head((_, res) => { + sending = sendStandardResponse(res, { + status: 201, + headers: { 'x-custom-header': 'custom-value' }, + body: undefined, + }) + }) + + await expect(sending).resolves.toBeUndefined() + + expect(response.headers).toMatchObject({ + ':status': 201, + 'x-custom-header': 'custom-value', + }) + expect(response.body).toBe('') + }) + + it('rejects and releases a stream body it cannot send', async () => { + let clean = false + const body = (async function* () { + try { + while (true) { + yield 'foo' + await new Promise(r => setTimeout(r, 10)) + } + } + finally { + clean = true + } + })() + + let sending: Promise | undefined + + const response = await requestHttp2Head((_, res) => { + sending = sendStandardResponse(res, { status: 201, headers: {}, body }) + sending.catch(() => {}) + }) + + await expect(sending).rejects.toMatchObject({ code: 'ERR_STREAM_WRITE_AFTER_END' }) + + expect(response.headers).toMatchObject({ + ':status': 201, + 'content-type': 'text/event-stream', + }) + expect(response.body).toBe('') + + await vi.waitFor(() => { + expect(clean).toBe(true) + }) + }) + }) + + describe('http2 stream reset by the client', () => { + it.each([ + ['resolves on a cancel', http2.constants.NGHTTP2_CANCEL, undefined], + ['rejects with the stream error on an error code', http2.constants.NGHTTP2_INTERNAL_ERROR, 'ERR_HTTP2_STREAM_ERROR'], + ])('%s', async (_, code, errorCode) => { + let clean = false + const body = (async function* () { + try { + while (true) { + yield 'foo' + await new Promise(r => setTimeout(r, 10)) + } + } + finally { + clean = true + } + })() + + let sending: Promise | undefined + + const server = http2.createServer((req, res) => { + sending = sendStandardResponse(res, { status: 200, headers: {}, body }) + }) + await new Promise(r => server.listen(0, r)) + + const client = http2.connect(`http://localhost:${(server.address() as AddressInfo).port}`) + + try { + const reqStream = client.request({ ':path': '/' }) + reqStream.once('error', () => {}) + + await new Promise(r => reqStream.once('data', r)) + reqStream.close(code) + + if (errorCode) { + await expect(sending).rejects.toMatchObject({ code: errorCode }) + } + else { + await expect(sending).resolves.toBeUndefined() + } + + await vi.waitFor(() => { + expect(clean).toBe(true) + }) + } + finally { + client.close() + await new Promise(r => server.close(r)) + } + }) + }) }) + +async function requestHttp2Head( + listener: (req: Http2ServerRequest, res: Http2ServerResponse) => void, +): Promise<{ headers: http2.IncomingHttpHeaders, body: string }> { + const server = http2.createServer(listener) + await new Promise(r => server.listen(0, r)) + + const client = http2.connect(`http://localhost:${(server.address() as AddressInfo).port}`) + + try { + const reqStream = client.request({ ':path': '/', ':method': 'HEAD' }) + + const chunks: Buffer[] = [] + reqStream.on('data', (chunk: Buffer) => chunks.push(chunk)) + + const headers = await new Promise((resolve, reject) => { + reqStream.once('response', resolve) + reqStream.once('error', reject) + }) + + await new Promise(r => reqStream.once('close', r)) + + return { headers, body: Buffer.concat(chunks).toString() } + } + finally { + client.close() + await new Promise(r => server.close(r)) + } +} diff --git a/packages/node/src/response.ts b/packages/node/src/response.ts index 376add7d..d90a46ca 100644 --- a/packages/node/src/response.ts +++ b/packages/node/src/response.ts @@ -33,8 +33,10 @@ export async function sendStandardResponse( return } - res.once('error', reject) - res.once('close', resolve) + const connection = 'stream' in res ? res.stream : res + + connection.once('error', reject) + connection.once('close', resolve) try { // DON'T use `res.writeHead` because it send response immediately in chunked mode @@ -55,7 +57,7 @@ export async function sendStandardResponse( res.end(resBody) } else { - res.once('close', () => { + connection.once('close', () => { if (!resBody.closed) { resBody.destroy(getNodeResponseError(res) ?? undefined) } diff --git a/packages/node/src/signal.test.ts b/packages/node/src/signal.test.ts index a052f1b2..bd2e638f 100644 --- a/packages/node/src/signal.test.ts +++ b/packages/node/src/signal.test.ts @@ -171,4 +171,80 @@ describe('toAbortSignal', async () => { await handled }) + + describe('on http2 HEAD response, whose stream Node already ended', () => { + it('without abort', async ({ onTestFinished }) => { + const server = http2.createServer() + onTestFinished(() => new Promise(r => server.close(r))) + + const handled = new Promise((resolve, reject) => { + server.on('request', async (req, res) => { + try { + const signal = toAbortSignal(res) + + expect(signal.aborted).toBe(false) + + res.end() + await new Promise(r => res.stream.once('close', () => r())) + + expect(signal.aborted).toBe(false) + + resolve() + } + catch (error) { + reject(error) + } + }) + }) + + await new Promise(r => server.listen(0, r)) + const port = (server.address() as any).port + + const client = http2.connect(`http://localhost:${port}`) + const reqStream = client.request({ ':path': '/', ':method': 'HEAD' }) + reqStream.resume() + reqStream.once('close', () => client.close()) + + await handled + }) + + it('aborted by client before the response ends', async ({ onTestFinished }) => { + const server = http2.createServer() + onTestFinished(() => new Promise(r => server.close(r))) + + const handled = new Promise((resolve, reject) => { + server.on('request', async (req, res) => { + try { + const signal = toAbortSignal(res) + + expect(signal.aborted).toBe(false) + + await new Promise(r => res.stream.once('close', () => r())) + + expect(signal.aborted).toBe(true) + expect(signal.reason).toEqual(new AbortError('Writable stream closed before it finished writing')) + + resolve() + } + catch (error) { + reject(error) + } + }) + }) + + await new Promise(r => server.listen(0, r)) + const port = (server.address() as any).port + + const client = http2.connect(`http://localhost:${port}`) + const reqStream = client.request({ ':path': '/', ':method': 'HEAD' }) + reqStream.once('error', () => {}) + + setTimeout(() => { + reqStream.close(http2.constants.NGHTTP2_CANCEL) + client.close() + }, 50) + + await handled + }) + }) }) diff --git a/packages/node/src/signal.ts b/packages/node/src/signal.ts index 852d7e19..0a932fcf 100644 --- a/packages/node/src/signal.ts +++ b/packages/node/src/signal.ts @@ -1,29 +1,25 @@ import type Stream from 'node:stream' import type { NodeHttpResponse } from './types' import { AbortError } from '@standard-server/shared' -import { canWriteToNodeResponse, getNodeResponseError } from './utils' +import { getNodeResponseError } from './utils' -export function toAbortSignal(stream: Stream.Writable | NodeHttpResponse): AbortSignal { +export function toAbortSignal(res: Stream.Writable | NodeHttpResponse): AbortSignal { const controller = new AbortController() - const error = getNodeResponseError(stream) + const stream = 'stream' in res ? res.stream : res - if (error) { - controller.abort(error) - } - else if (!canWriteToNodeResponse(stream)) { - if (!stream.writableFinished || !stream.writableEnded) { - controller.abort(new AbortError('Writable stream closed before it finished writing')) + const onClose = () => { + if (!res.writableEnded) { + controller.abort(getNodeResponseError(res) ?? new AbortError('Writable stream closed before it finished writing')) } } + + if (stream.destroyed) { + onClose() + } else { stream.once('error', error => controller.abort(error)) - - stream.once('close', () => { - if (!stream.writableFinished || !stream.writableEnded) { - controller.abort(new AbortError('Writable stream closed before it finished writing')) - } - }) + stream.once('close', onClose) } return controller.signal diff --git a/packages/node/src/utils.test.ts b/packages/node/src/utils.test.ts index a8362410..45a65567 100644 --- a/packages/node/src/utils.test.ts +++ b/packages/node/src/utils.test.ts @@ -178,6 +178,39 @@ describe('canWriteToNodeResponse', () => { await handled }) + + it('on http2 HEAD response, whose stream Node already ended', async ({ onTestFinished }) => { + const server = http2.createServer() + onTestFinished(() => new Promise(r => server.close(r))) + + const handled = new Promise((resolve, reject) => { + server.on('request', async (req, res) => { + try { + expect(res.stream.writableEnded).toBe(true) + expect(canWriteToNodeResponse(res)).toBe(true) + + res.end() + + expect(canWriteToNodeResponse(res)).toBe(false) + + resolve() + } + catch (error) { + reject(error) + } + }) + }) + + await new Promise(r => server.listen(0, r)) + const port = (server.address() as any).port + + const client = http2.connect(`http://localhost:${port}`) + const reqStream = client.request({ ':path': '/', ':method': 'HEAD' }) + reqStream.resume() + reqStream.once('close', () => client.close()) + + await handled + }) }) describe('getNodeResponseError', () => { diff --git a/packages/node/src/utils.ts b/packages/node/src/utils.ts index 66def831..7ee27d39 100644 --- a/packages/node/src/utils.ts +++ b/packages/node/src/utils.ts @@ -67,22 +67,12 @@ async function _drainIterator(iterator: AsyncIterator): Promise { } /** - * Check both the response itself and its underlying stream (http2) are still writable. + * Check the response can still be sent. */ export function canWriteToNodeResponse(res: Stream.Writable | NodeHttpResponse): boolean { - if ('headersSent' in res && res.headersSent) { - return false - } - - if ('stream' in res && !_canWriteToStream(res.stream)) { - return false - } - - return _canWriteToStream(res) -} - -function _canWriteToStream(stream: Stream.Writable): boolean { - return !stream.closed && !stream.destroyed && !stream.writableFinished && !stream.writableEnded + return !('headersSent' in res && res.headersSent) + && !res.writableEnded + && !('stream' in res ? res.stream : res).destroyed } /** diff --git a/tests/data-transfer.test.ts b/tests/data-transfer.test.ts index a5174487..a1e865c1 100644 --- a/tests/data-transfer.test.ts +++ b/tests/data-transfer.test.ts @@ -56,6 +56,7 @@ describe.each([ ['/test', 'POST', {}, 200], ['/test?query=true', 'GET', { h1: 'v1', h2: 'v2' }, 201], ['/hi%2F', 'DELETE', { h3: 'v3' }, 500], + ['/test', 'HEAD', { h4: 'v4' }, 201], ])('url=$0, method=$1, headers=$2, status=$3', async (url, method, headers, status) => { clientServer.handler.mockResolvedValueOnce({ headers: { diff --git a/tests/signal-and-cancel.test.ts b/tests/signal-and-cancel.test.ts index 48fcc2de..3aba2e74 100644 --- a/tests/signal-and-cancel.test.ts +++ b/tests/signal-and-cancel.test.ts @@ -55,7 +55,7 @@ describe.each([ ] as const)('signal and cancel: $0', (adapter, createClientServer) => { const clientServer = createClientServer() - it('never aborted', async () => { + it.each(['GET', 'HEAD'])('never aborted: %s', async (method) => { let serverSignal!: AbortSignal clientServer.handler.mockImplementationOnce(async ({ signal }) => { @@ -66,14 +66,14 @@ describe.each([ return { headers: {}, status: 200, - body: 'Hello', + body: method === 'HEAD' ? undefined : 'Hello', } }) const response = await clientServer.request({ headers: {}, body: undefined, - method: 'GET', + method, url: '/', })