fix(MockHttpSocket): handle response stream errors

src/interceptors/ClientRequest/MockHttpSocket.ts

@@ -239,34 +239,47 @@ export class MockHttpSocket extends MockSocket {
     }
 
     if (response.body) {
-      const reader = response.body.getReader()
-
-      while (true) {
-        const { done, value } = await reader.read()
-
-        if (done) {
-          break
+      try {
+        const reader = response.body.getReader()
+
+        while (true) {
+          const { done, value } = await reader.read()
+
+          if (done) {
+            break
+          }
+
+          // Flush the headers upon the first chunk in the stream.
+          // This ensures the consumer will start receiving the response
+          // as it streams in (subsequent chunks are pushed).
+          if (httpHeaders.length > 0) {
+            flushHeaders(value)
+            continue
+          }
+
+          // Subsequent body chukns are push to the stream.
+          this.push(value)
         }
-
-        // Flush the headers upon the first chunk in the stream.
-        // This ensures the consumer will start receiving the response
-        // as it streams in (subsequent chunks are pushed).
-        if (httpHeaders.length > 0) {
-          flushHeaders(value)
-          continue
+      } catch (error) {
+        if (error instanceof Error) {
+          // Flush the headers on response stream errors.
+          // This way, the client still receives the "response" event,
+          // and the actual stream error is forwarded as the "error"
+          // event on the http.IncomingMessage instance.
+          flushHeaders()
+
+          // Forward response streaming errors as response errors.
+          /** @todo This doesn't quite do it. */
+          // this.destroy(error)
         }
-
-        // Subsequent body chukns are push to the stream.
-        this.push(value)
+        return
       }
     }
 
     // If the headers were not flushed up to this point,
     // this means the response either had no body or had
     // an empty body stream. Flush the headers.
-    if (httpHeaders.length > 0) {
-      flushHeaders()
-    }
+    flushHeaders()
 
     // Close the socket if the connection wasn't marked as keep-alive.
     if (!this.shouldKeepAlive) {

test/modules/http/compliance/http-res-readable.test.ts

@@ -0,0 +1,87 @@
+/**
+ * @vitest-environment node
+ */
+import { vi, it, expect, beforeAll, afterAll } from 'vitest'
+import http from 'node:http'
+import { Readable } from 'node:stream'
+import { HttpServer } from '@open-draft/test-server/http'
+import { ClientRequestInterceptor } from '../../../../src/interceptors/ClientRequest'
+import { waitForClientRequest } from '../../../helpers'
+
+function createErrorStream() {
+  return new ReadableStream({
+    start(controller) {
+      controller.enqueue(new TextEncoder().encode('hello'))
+      controller.error(new Error('stream error'))
+    },
+  })
+}
+
+const httpServer = new HttpServer((app) => {
+  app.get('/resource', (req, res) => {
+    res.pipe(Readable.fromWeb(createErrorStream()))
+  })
+})
+
+const interceptor = new ClientRequestInterceptor()
+
+beforeAll(async () => {
+  interceptor.apply()
+  await httpServer.listen()
+})
+
+afterAll(async () => {
+  interceptor.dispose()
+  await httpServer.close()
+})
+
+it('supports ReadableStream as a mocked response', async () => {
+  const encoder = new TextEncoder()
+  interceptor.once('request', ({ request }) => {
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(encoder.encode('hello'))
+        controller.enqueue(encoder.encode(' '))
+        controller.enqueue(encoder.encode('world'))
+        controller.close()
+      },
+    })
+    request.respondWith(new Response(stream))
+  })
+
+  const request = http.get('http://example.com/resource')
+  const { text } = await waitForClientRequest(request)
+  expect(await text()).toBe('hello world')
+})
+
+it('forwards ReadableStream errors to the request', async () => {
+  const requestErrorListener = vi.fn()
+  const responseErrorListener = vi.fn()
+
+  interceptor.once('request', ({ request }) => {
+    request.respondWith(new Response(createErrorStream()))
+  })
+
+  const request = http.get(httpServer.http.url('/resource'))
+  request.on('error', requestErrorListener)
+  request.on('response', (response) => {
+    response.on('error', responseErrorListener)
+  })
+
+  await vi.waitFor(() => {
+    return new Promise<http.IncomingMessage>((resolve) => {
+      request.on('response', resolve)
+    })
+  })
+
+  // Response stream errors are not request errors.
+  expect(requestErrorListener).not.toHaveBeenCalled()
+  expect(request.destroyed).toBe(false)
+
+  // Response stream errors are "error" events on the response.
+  await vi.waitFor(() => {
+    expect(responseErrorListener).toHaveBeenCalledWith(
+      new Error('stream error')
+    )
+  })
+})