fix(node,fastify): handle synchronous throws on the response-send path

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())

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)
+    }
   })
 }

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

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)
     }
   })
 }

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<any>(r => server.close(r)))
+
+    const handled = new Promise<void>((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<void>(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<any>(r => server.close(r)))

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
   }