diff --git a/packages/fastify/src/response.test.ts b/packages/fastify/src/response.test.ts index c6fabd2..fe7749a 100644 --- a/packages/fastify/src/response.test.ts +++ b/packages/fastify/src/response.test.ts @@ -349,6 +349,48 @@ describe('sendStandardResponse', () => { }) }) + it('rejects and destroys the body when reply throws synchronously', async ({ onTestFinished }) => { + const cancelMock = vi.fn() + + const fastify = Fastify() + onTestFinished(() => fastify.close()) + + toNodeHttpBodySpy.mockReturnValueOnce([Readable.fromWeb(new ReadableStream({ + async pull(controller) { + controller.enqueue(new TextEncoder().encode('foo')) + await new Promise(r => setTimeout(r, 100)) + }, + cancel: cancelMock, + })), {}]) + + let thrownError: any + fastify.get('/', async (req, reply) => { + try { + await sendStandardResponse(reply, { + // status outside [100, 599] makes reply.status throw FST_ERR_BAD_STATUS_CODE + status: 999, + headers: {}, + async* body() { }, + }) + } + catch (err) { + thrownError = err + throw err + } + }) + + await fastify.ready() + const res = await request(fastify.server).get('/') + + // fastify's error handler can still send a response + expect(res.status).toBe(500) + expect(thrownError.code).toBe('FST_ERR_BAD_STATUS_CODE') + + await vi.waitFor(() => { + expect(cancelMock).toHaveBeenCalledTimes(1) + }) + }) + it('works with @fastify/cookie', async ({ onTestFinished }) => { const fastify = Fastify() onTestFinished(() => fastify.close()) diff --git a/packages/fastify/src/response.ts b/packages/fastify/src/response.ts index 230eb21..e2b7f89 100644 --- a/packages/fastify/src/response.ts +++ b/packages/fastify/src/response.ts @@ -35,17 +35,28 @@ export async function sendStandardResponse( reply.raw.once('error', reject) reply.raw.once('close', resolve) - reply.status(standardResponse.status) - - // DON'T pass headers with `undefined` value to fastify, it turns them into empty strings - for (const key in resHeaders) { - const value = resHeaders[key] - if (value !== undefined) { - reply.header(key, value) + try { + reply.status(standardResponse.status) + + // DON'T pass headers with `undefined` value to fastify, it turns them into empty strings + for (const key in resHeaders) { + const value = resHeaders[key] + if (value !== undefined) { + reply.header(key, value) + } } + + // fastify pipes and cleans up the stream body itself, no manual piping needed + reply.send(resBody) } + catch (error) { + if (typeof resBody === 'object' && !resBody.closed) { + resBody.on('error', reject) + resBody.destroy(error as any) + } - // fastify pipes and cleans up the stream body itself, no manual piping needed - reply.send(resBody) + // Don't destroy reply.raw: fastify's error handler can still send a response. + reject(error) + } }) } diff --git a/packages/node/src/response.test.ts b/packages/node/src/response.test.ts index e767a66..76bbe95 100644 --- a/packages/node/src/response.test.ts +++ b/packages/node/src/response.test.ts @@ -239,6 +239,68 @@ describe('sendStandardResponse', () => { expect(thrownError).toBe(undefined) }) + it('rejects, destroys the response and the body when applying headers throws', async () => { + let destroySpy: any + let thrownError: any + + const options = { } + await expect(request(async (req: IncomingMessage, res: ServerResponse) => { + destroySpy = vi.spyOn(res, 'destroy') + + try { + await sendStandardResponse(res, { + status: 207, + headers: { + 'x-invalid': 'bad\nvalue', + }, + body: (async function* () { + yield 1 + })(), + }, options) + } + catch (err) { + thrownError = err + } + }).get('/')).rejects.toThrow() + + expect(thrownError).toBeInstanceOf(Error) + expect(thrownError.code).toBe('ERR_INVALID_CHAR') + + expect(destroySpy).toHaveBeenCalledWith(thrownError) + + const [resBody] = toNodeHttpBodySpy.mock.results[0]!.value + expect((resBody as any).destroyed).toBe(true) + }) + + it('resolves without sending when headers were already flushed', async () => { + let sendError: any + + const res = await request(async (req: IncomingMessage, res: ServerResponse) => { + res.flushHeaders() + + try { + await sendStandardResponse(res, { + status: 207, + headers: { + 'x-custom-header': 'custom-value', + }, + body: undefined, + }) + } + catch (err) { + sendError = err + } + + res.end('flushed') + }).get('/') + + expect(sendError).toBeUndefined() + + expect(res.status).toBe(200) + expect(res.headers).not.toHaveProperty('x-custom-header') + expect(res.text).toEqual('flushed') + }) + describe('stream destroy while sending', () => { it('with error', async () => { let clean = false diff --git a/packages/node/src/response.ts b/packages/node/src/response.ts index 4f1d406..57bf07b 100644 --- a/packages/node/src/response.ts +++ b/packages/node/src/response.ts @@ -36,34 +36,47 @@ export async function sendStandardResponse( res.once('error', reject) res.once('close', resolve) - // DON'T use `res.writeHead` because it send response immediately in chunked mode - // while we only need chunked if the response body is stream - res.statusCode = standardResponse.status - for (const key in resHeaders) { - const value = resHeaders[key] - if (value !== undefined) { - res.setHeader(key, value) + try { + // DON'T use `res.writeHead` because it send response immediately in chunked mode + // while we only need chunked if the response body is stream + res.statusCode = standardResponse.status + for (const key in resHeaders) { + const value = resHeaders[key] + if (value !== undefined) { + res.setHeader(key, value) + } } - } - if (resBody === undefined) { - // NOTE: Lambda functions don't allow passing undefined to `res.end` - res.end() - } - else if (typeof resBody === 'string') { - res.end(resBody) - } - else { - res.once('close', () => { - if (!resBody.closed) { - resBody.destroy(getNodeResponseError(res) ?? undefined) - } - }) + if (resBody === undefined) { + // NOTE: Lambda functions don't allow passing undefined to `res.end` + res.end() + } + else if (typeof resBody === 'string') { + res.end(resBody) + } + else { + res.once('close', () => { + if (!resBody.closed) { + resBody.destroy(getNodeResponseError(res) ?? undefined) + } + }) + + // WARNING: errors that occur here are silently ignored and not reported to the Promise + resBody.once('error', error => res.destroy(error)) - // WARNING: errors that occur here are silently ignored and not reported to the Promise - resBody.once('error', error => res.destroy(error)) + resBody.pipe(res) + } + } + catch (error) { + if (typeof resBody === 'object' && !resBody.closed) { + resBody.on('error', reject) + resBody.destroy(error as any) + } - resBody.pipe(res) + // Destroy instead of leaving the response half-open: headers/status may be + // partially applied, so the connection is no longer safe to reuse. + res.destroy(error as any) + reject(error) } }) } diff --git a/packages/node/src/utils.test.ts b/packages/node/src/utils.test.ts index 3d25b90..34ae286 100644 --- a/packages/node/src/utils.test.ts +++ b/packages/node/src/utils.test.ts @@ -70,6 +70,38 @@ describe('canWriteToNodeResponse', () => { await handled }) + it('on http1 response with headers already flushed', async ({ onTestFinished }) => { + const server = http.createServer() + onTestFinished(() => new Promise(r => server.close(r))) + + const handled = new Promise((resolve, reject) => { + server.on('request', async (req, res) => { + try { + expect(canWriteToNodeResponse(res)).toBe(true) + + res.flushHeaders() + + expect(res.headersSent).toBe(true) + expect(canWriteToNodeResponse(res)).toBe(false) + + res.end() + + resolve() + } + catch (error) { + reject(error) + } + }) + }) + + await new Promise(r => server.listen(0, r)) + const port = (server.address() as any).port + + http.get(`http://localhost:${port}`, res => res.resume()) + + await handled + }) + it('on http2 response aborted by client', async ({ onTestFinished }) => { const server = http2.createServer() onTestFinished(() => new Promise(r => server.close(r))) diff --git a/packages/node/src/utils.ts b/packages/node/src/utils.ts index a342312..b55a717 100644 --- a/packages/node/src/utils.ts +++ b/packages/node/src/utils.ts @@ -5,6 +5,10 @@ import type { NodeHttpResponse } from './types' * Check both the response itself and its underlying stream (http2) are still writable. */ 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 }