fix(node, fastify, aws-lambda): resolve sendStandardResponse when the client disconnects

packages/aws-lambda/README.md

@@ -86,7 +86,7 @@ export const handler = awslambda.streamifyResponse(async (event, responseStream,
 })
 ```
 
-`sendStandardResponse()` sends the status, headers, and cookies as the response stream metadata prelude via `awslambda.HttpResponseStream.from()`, then streams the body. It resolves once the response is fully flushed, and rejects if the stream errors.
+`sendStandardResponse()` sends the status, headers, and cookies as the response stream metadata prelude via `awslambda.HttpResponseStream.from()`, then streams the body. It resolves once the response is fully flushed or the client disconnects, and rejects if the stream fails otherwise.
 
 > [!TIP]
 > When sending responses, you can pass additional options such as event-stream keep-alive.

packages/aws-lambda/src/response.test.ts

@@ -48,6 +48,10 @@ function createResponseStream(): HttpResponseStream & { chunks: Buffer[] } {
   return stream
 }
 
+function createDisconnectError(): Error {
+  return Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' })
+}
+
 function metadataOf(stream: HttpResponseStream & { chunks: Buffer[] }): Record<string, unknown> {
   return JSON.parse(stream.chunks[0]!.toString())
 }
@@ -253,40 +257,46 @@ describe('sendStandardResponse', () => {
     await sendPromise
   })
 
-  it('destroys the body without error when the response stream closes cleanly while streaming', async () => {
-    const responseStream = createResponseStream()
+  it.each([
+    ['cleanly', undefined],
+    ['on a client disconnect', createDisconnectError()],
+  ])(
+    'resolves and destroys the body when the response stream closes %s while streaming',
+    async (_, error) => {
+      const responseStream = createResponseStream()
 
-    let clean = false
-    const standardResponse: StandardResponse = {
-      body: (async function* () {
-        try {
-          yield 1
-          await new Promise((r) => setTimeout(r, 100))
-          yield 2
-          await new Promise((r) => setTimeout(r, 9999999))
-          yield 3
-        } finally {
-          clean = true
-        }
-      })(),
-      headers: {},
-      status: 200,
-    }
+      let clean = false
+      const standardResponse: StandardResponse = {
+        body: (async function* () {
+          try {
+            yield 1
+            await new Promise((r) => setTimeout(r, 100))
+            yield 2
+            await new Promise((r) => setTimeout(r, 9999999))
+            yield 3
+          } finally {
+            clean = true
+          }
+        })(),
+        headers: {},
+        status: 200,
+      }
 
-    const sendPromise = sendStandardResponse(responseStream, standardResponse)
+      const sendPromise = sendStandardResponse(responseStream, standardResponse)
 
-    await vi.waitFor(() => {
-      expect(bodyOf(responseStream)).toContain('data: 1')
-    })
+      await vi.waitFor(() => {
+        expect(bodyOf(responseStream)).toContain('data: 1')
+      })
 
-    responseStream.destroy()
+      responseStream.destroy(error)
 
-    await vi.waitFor(() => {
-      expect(clean).toBe(true)
-    })
+      await vi.waitFor(() => {
+        expect(clean).toBe(true)
+      })
 
-    await sendPromise
-  })
+      await expect(sendPromise).resolves.toBeUndefined()
+    },
+  )
 
   it.each([
     [
@@ -329,9 +339,13 @@ describe('sendStandardResponse', () => {
   })
 
   describe('response stream closed before sending', () => {
-    it('resolves and destroys the body', async () => {
+    it.each([
+      ['closed cleanly', undefined],
+      ['destroyed by a client disconnect', createDisconnectError()],
+    ])('resolves and destroys the body when the response stream was %s', async (_, error) => {
       const responseStream = createResponseStream()
-      responseStream.destroy()
+      responseStream.once('error', () => {})
+      responseStream.destroy(error)
 
       await vi.waitFor(() => {
         expect(responseStream.closed).toBe(true)

packages/aws-lambda/src/response.ts

@@ -4,6 +4,7 @@ import {
   canWriteToNodeResponse,
   destroyNodeHttpBody,
   getNodeResponseError,
+  isNodeResponseDisconnectError,
   toNodeHttpBody,
 } from '@standard-server/node'
 
@@ -40,7 +41,7 @@ export async function sendStandardResponse(
 
       destroyNodeHttpBody(resBody, error, reject)
 
-      if (error) {
+      if (error && !isNodeResponseDisconnectError(error)) {
         reject(error)
       } else {
         resolve()
@@ -49,7 +50,11 @@ export async function sendStandardResponse(
       return
     }
 
-    responseStream.once('error', reject)
+    responseStream.once('error', (error) => {
+      if (!isNodeResponseDisconnectError(error)) {
+        reject(error)
+      }
+    })
     responseStream.once('close', resolve)
 
     try {

packages/fastify/README.md

@@ -90,7 +90,7 @@ fastify.all('/*', async (req, reply) => {
 await fastify.listen({ port: 3000 })
 ```
 
-`sendStandardResponse()` resolves once the response is fully flushed, and rejects if the underlying connection errors. Do not return a value from the route handler afterwards — Fastify would try to send a second response.
+`sendStandardResponse()` resolves once the response is fully flushed or the client disconnects (an HTTP/2 stream reset included), and rejects if sending fails otherwise. Do not return a value from the route handler afterwards — Fastify would try to send a second response.
 
 > [!TIP]
 > When sending responses, you can pass additional options such as event-stream keep-alive.

packages/fastify/src/response.test.ts

@@ -733,34 +733,86 @@ describe('sendStandardResponse', () => {
     })
   })
 
-  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<void> | undefined
+  describe('http2 client disconnect', () => {
+    const disconnects: [
+      string,
+      (client: http2.ClientHttp2Session, reqStream: http2.ClientHttp2Stream) => void,
+    ][] = [
+      ['an error reset', (_, reqStream) => reqStream.close(http2.constants.NGHTTP2_INTERNAL_ERROR)],
+      ['an error goaway', (client) => client.goaway(http2.constants.NGHTTP2_PROTOCOL_ERROR)],
+    ]
+
+    it.for(disconnects)(
+      'resolves and releases the body on %s while sending',
+      async ([, disconnect], { onTestFinished }) => {
+        const { body, isReleased } = createEndlessBody()
+        let sending: Promise<void> | undefined
+
+        const fastify = Fastify({ http2: true })
+        onTestFinished(() => fastify.close())
+
+        fastify.get('/', async (req, reply) => {
+          sending = sendStandardResponse(reply, { status: 200, headers: {}, body })
+          await sending
+        })
 
-    const fastify = Fastify({ http2: true })
-    onTestFinished(() => fastify.close())
+        const client = http2.connect(await fastify.listen({ port: 0, host: '127.0.0.1' }))
+        client.once('error', () => {})
+        onTestFinished(() => client.close())
 
-    fastify.get('/', async (req, reply) => {
-      sending = sendStandardResponse(reply, { status: 200, headers: {}, body })
-      await sending.catch(() => {})
-    })
+        const reqStream = client.request({ ':path': '/' })
+        reqStream.once('error', () => {})
+
+        await new Promise((r) => reqStream.once('data', r))
+        disconnect(client, reqStream)
 
-    const client = http2.connect(await fastify.listen({ port: 0, host: '127.0.0.1' }))
-    onTestFinished(() => client.close())
+        await expect(sending).resolves.toBeUndefined()
 
-    const reqStream = client.request({ ':path': '/' })
-    reqStream.once('error', () => {})
+        await vi.waitFor(() => {
+          expect(isReleased()).toBe(true)
+        })
+      },
+    )
 
-    await new Promise((r) => reqStream.once('data', r))
-    reqStream.close(http2.constants.NGHTTP2_INTERNAL_ERROR)
+    it.for(disconnects)(
+      'resolves and releases the body on %s before sending',
+      async ([, disconnect], { onTestFinished }) => {
+        const { body, isReleased } = createEndlessBody()
 
-    await expect(sending).rejects.toMatchObject({ code: 'ERR_HTTP2_STREAM_ERROR' })
+        let received!: () => void
+        const receivedPromise = new Promise<void>((r) => (received = r))
+        let send!: (sending: Promise<void>) => void
+        const sending = new Promise<void>((r) => (send = r))
 
-    await vi.waitFor(() => {
-      expect(isReleased()).toBe(true)
-    })
+        const fastify = Fastify({ http2: true })
+        onTestFinished(() => fastify.close())
+
+        fastify.get('/', async (req, reply) => {
+          const closed = new Promise((r) => reply.raw.stream.once('close', r))
+          received()
+          await closed
+
+          send(sendStandardResponse(reply, { status: 200, headers: {}, body }))
+          await sending
+        })
+
+        const client = http2.connect(await fastify.listen({ port: 0, host: '127.0.0.1' }))
+        client.once('error', () => {})
+        onTestFinished(() => client.close())
+
+        const reqStream = client.request({ ':path': '/' })
+        reqStream.once('error', () => {})
+
+        await receivedPromise
+        disconnect(client, reqStream)
+
+        await expect(sending).resolves.toBeUndefined()
+
+        await vi.waitFor(() => {
+          expect(isReleased()).toBe(true)
+        })
+      },
+    )
   })
 })
 

packages/fastify/src/response.ts

@@ -4,6 +4,7 @@ import {
   canWriteToNodeResponse,
   destroyNodeHttpBody,
   getNodeResponseError,
+  isNodeResponseDisconnectError,
   toNodeHttpBody,
 } from '@standard-server/node'
 
@@ -28,7 +29,7 @@ export async function sendStandardResponse(
 
       destroyNodeHttpBody(resBody, error, reject)
 
-      if (error) {
+      if (error && !isNodeResponseDisconnectError(error)) {
         reject(error)
       } else {
         resolve()
@@ -39,7 +40,11 @@ export async function sendStandardResponse(
 
     const connection = 'stream' in reply.raw ? reply.raw.stream : reply.raw
 
-    connection.once('error', reject)
+    connection.once('error', (error) => {
+      if (!isNodeResponseDisconnectError(error)) {
+        reject(error)
+      }
+    })
     connection.once('close', () => {
       if (typeof resBody === 'object' && !resBody.closed) {
         resBody.destroy()

packages/node/README.md

@@ -85,6 +85,8 @@ createServer(async (req, res) => {
 }).listen(3000)
 ```
 
+`sendStandardResponse()` resolves once the response is fully flushed or the client disconnects (an HTTP/2 stream reset included), and rejects if sending fails otherwise.
+
 > [!TIP]
 > When sending responses, you can pass additional options such as event-stream keep-alive.