fix: cancel a readable stream if a writable stream is closed before a readable stream is closed. (#280)

Author: usualoma

src/utils.ts

@@ -12,15 +12,17 @@ export function writeFromReadableStreamDefaultReader(
   writable: Writable,
   currentReadPromise?: Promise<ReadableStreamReadResult<Uint8Array>> | undefined
 ) {
-  const handleError = () => {
-    // ignore the error
+  const cancel = (error?: unknown) => {
+    reader.cancel(error).catch(() => {})
   }
 
-  writable.on('error', handleError)
+  writable.on('close', cancel)
+  writable.on('error', cancel)
   ;(currentReadPromise ?? reader.read()).then(flow, handleStreamError)
 
   return reader.closed.finally(() => {
-    writable.off('error', handleError)
+    writable.off('close', cancel)
+    writable.off('error', cancel)
   })
 
   // eslint-disable-next-line @typescript-eslint/no-explicit-any

test/utils.test.ts

@@ -78,14 +78,15 @@ describe('buildOutgoingHttpHeaders', () => {
 })
 
 describe('writeFromReadableStream', () => {
-  it('should handle client disconnection gracefully without canceling stream', async () => {
+  it('should handle client disconnection', async () => {
     let enqueueCalled = false
     let cancelCalled = false
+    let enqueueTimeout: NodeJS.Timeout | undefined
 
     // Create test ReadableStream
     const stream = new ReadableStream({
       start(controller) {
-        setTimeout(() => {
+        enqueueTimeout = setTimeout(() => {
           try {
             controller.enqueue(new TextEncoder().encode('test'))
             enqueueCalled = true
@@ -97,6 +98,7 @@ describe('writeFromReadableStream', () => {
       },
       cancel() {
         cancelCalled = true
+        clearTimeout(enqueueTimeout)
       },
     })
 
@@ -110,8 +112,8 @@ describe('writeFromReadableStream', () => {
 
     await writeFromReadableStream(stream, writable)
 
-    expect(enqueueCalled).toBe(true) // enqueue should succeed
-    expect(cancelCalled).toBe(false) // cancel should not be called
+    expect(enqueueCalled).toBe(false) // enqueue should not be called
+    expect(cancelCalled).toBe(true) // cancel should be called
   })
 })