fix(listener): cancel the response body when the client has already disconnected

src/utils.ts

@@ -16,6 +16,12 @@ export function writeFromReadableStreamDefaultReader(
     reader.cancel(error).catch(() => {})
   }
 
+  if (writable.destroyed) {
+    // 'close' may already have been emitted, so cancel now instead of waiting for it.
+    cancel()
+    return reader.closed
+  }
+
   writable.on('close', cancel)
   writable.on('error', cancel)
   ;(currentReadPromise ?? reader.read()).then(flow, handleStreamError)
@@ -55,6 +61,7 @@ export function writeFromReadableStream(stream: ReadableStream<Uint8Array>, writ
   if (stream.locked) {
     throw new TypeError('ReadableStream is locked.')
   } else if (writable.destroyed) {
+    stream.cancel().catch(() => {})
     return
   }
 

test/listener.test.ts

@@ -412,6 +412,38 @@ describe('Abort request', () => {
     await expect(req).rejects.toThrow()
   })
 
+  it('should cancel the response body when the client disconnects before the response', async () => {
+    let resolveCancelled!: () => void
+    const cancelled = new Promise<void>((r) => {
+      resolveCancelled = r
+    })
+
+    const server = createServer(
+      getRequestListener(async (req: Request) => {
+        await new Promise((r) => req.signal.addEventListener('abort', r))
+        return new Response(
+          new ReadableStream({
+            start(controller) {
+              controller.enqueue(new TextEncoder().encode('data: first\n\n'))
+            },
+            cancel() {
+              resolveCancelled()
+            },
+          }),
+          { headers: { 'content-type': 'text/event-stream' } }
+        )
+      })
+    )
+
+    await requestServer(server, {
+      method: 'GET',
+      path: '/abort',
+      signal: AbortSignal.timeout(50),
+    }).catch(() => {})
+
+    await withTimeout(cancelled, 'response body was not cancelled')
+  })
+
   it.each(['stream', 'session'] as const)(
     'should emit an abort event when an HTTP/2 client closes the %s',
     async (target) => {

test/utils.test.ts

@@ -2,6 +2,7 @@ import { Writable } from 'node:stream'
 import {
   buildOutgoingHttpHeaders,
   writeFromReadableStream,
+  writeFromReadableStreamDefaultReader,
   readWithoutBlocking,
 } from '../src/utils'
 
@@ -130,6 +131,49 @@ describe('writeFromReadableStream', () => {
     expect(enqueueCalled).toBe(false) // enqueue should not be called
     expect(cancelCalled).toBe(true) // cancel should be called
   })
+
+  it('should cancel the stream if the client has already disconnected', async () => {
+    let cancelCalled = false
+    const stream = new ReadableStream({
+      cancel() {
+        cancelCalled = true
+      },
+    })
+
+    const writable = new Writable()
+    writable.destroy()
+    await new Promise((resolve) => writable.once('close', resolve))
+
+    await writeFromReadableStream(stream, writable)
+
+    expect(cancelCalled).toBe(true)
+  })
+})
+
+describe('writeFromReadableStreamDefaultReader', () => {
+  it('should cancel the reader if the client has already disconnected', async () => {
+    let cancelCalled = false
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('test'))
+      },
+      cancel() {
+        cancelCalled = true
+      },
+    })
+
+    const writable = new Writable({
+      write(_chunk, _encoding, callback) {
+        callback()
+      },
+    })
+    writable.destroy()
+    await new Promise((resolve) => writable.once('close', resolve))
+
+    await writeFromReadableStreamDefaultReader(stream.getReader(), writable)
+
+    expect(cancelCalled).toBe(true)
+  })
 })
 
 describe('readWithoutBlocking', () => {