fix(MockHttpSocket): handle response stream errors

src/interceptors/ClientRequest/MockHttpSocket.ts

@@ -248,34 +248,53 @@ 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)
         }
+      } catch (error) {
+        if (error instanceof Error) {
+          // Coerce response stream errors to 500 responses.
+          // Don't flush the original response headers because
+          // unhandled errors translate to 500 error responses forcefully.
+          this.respondWith(
+            new Response(
+              JSON.stringify({
+                name: error.name,
+                message: error.message,
+                stack: error.stack,
+              }),
+              { status: 500, statusText: 'Unhandled Exception' }
+            )
+          )
 
-        // 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
+          return
         }
-
-        // 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) {

src/interceptors/ClientRequest/index.ts

@@ -12,6 +12,7 @@ import { MockAgent, MockHttpsAgent } from './agents'
 import { emitAsync } from '../../utils/emitAsync'
 import { toInteractiveRequest } from '../../utils/toInteractiveRequest'
 import { normalizeClientRequestArgs } from './utils/normalizeClientRequestArgs'
+import { isNodeLikeError } from '../../utils/isNodeLikeError'
 
 export class ClientRequestInterceptor extends Interceptor<HttpRequestEventMap> {
   static symbol = Symbol('client-request-interceptor')
@@ -144,13 +145,38 @@ export class ClientRequestInterceptor extends Interceptor<HttpRequestEventMap> {
     })
 
     if (listenerResult.error) {
-      socket.errorWith(listenerResult.error)
+      // Allow mocking Node-like errors.
+      if (isNodeLikeError(listenerResult.error)) {
+        socket.errorWith(listenerResult.error)
+        return
+      }
+
+      socket.respondWith(
+        new Response(
+          JSON.stringify({
+            name: listenerResult.error.name,
+            message: listenerResult.error.message,
+            stack: listenerResult.error.stack,
+          }),
+          {
+            status: 500,
+            statusText: 'Unhandled Exception',
+            headers: {
+              'Content-Type': 'application/json',
+            },
+          }
+        )
+      )
       return
     }
 
     const mockedResponse = listenerResult.data
 
     if (mockedResponse) {
+      /**
+       * @note The `.respondWith()` method will handle "Response.error()".
+       * Maybe we should make all interceptors do that?
+       */
       socket.respondWith(mockedResponse)
       return
     }

test/modules/http/compliance/http-req-write.test.ts

@@ -186,7 +186,6 @@ it('calls all write callbacks before the mocked response', async () => {
     method: 'POST',
   })
   request.write('one', () => {
-    console.log('write callback!')
     request.end()
   })
 

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

@@ -0,0 +1,88 @@
+/**
+ * @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('original'))
+      queueMicrotask(() => {
+        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)
+  })
+
+  const response = await vi.waitFor(() => {
+    return new Promise<http.IncomingMessage>((resolve) => {
+      request.on('response', resolve)
+    })
+  })
+
+  // Response stream errors are translated to unhandled exceptions,
+  // and then the server decides how to handle them. This is often
+  // done as returning a 500 response.
+  expect(response.statusCode).toBe(500)
+  expect(response.statusMessage).toBe('Unhandled Exception')
+
+  // Response stream errors are not request errors.
+  expect(requestErrorListener).not.toHaveBeenCalled()
+  expect(request.destroyed).toBe(false)
+})

test/third-party/miniflare.test.ts

@@ -15,17 +15,12 @@ const interceptor = new BatchInterceptor({
   ],
 })
 
-const requestListener = vi.fn().mockImplementation(({ request }) => {
-  request.respondWith(new Response('mocked-body'))
-})
-
-interceptor.on('request', requestListener)
-
 beforeAll(() => {
   interceptor.apply()
 })
 
 afterEach(() => {
+  interceptor.removeAllListeners()
   vi.clearAllMocks()
 })
 
@@ -34,26 +29,35 @@ afterAll(() => {
 })
 
 test('responds to fetch', async () => {
+  interceptor.once('request', ({ request }) => {
+    request.respondWith(new Response('mocked-body'))
+  })
+
   const response = await fetch('https://example.com')
   expect(response.status).toEqual(200)
   expect(await response.text()).toEqual('mocked-body')
-  expect(requestListener).toHaveBeenCalledTimes(1)
 })
 
 test('responds to http.get', async () => {
+  interceptor.once('request', ({ request }) => {
+    request.respondWith(new Response('mocked-body'))
+  })
+
   const { resBody } = await httpGet('http://example.com')
   expect(resBody).toEqual('mocked-body')
-  expect(requestListener).toHaveBeenCalledTimes(1)
 })
 
 test('responds to https.get', async () => {
+  interceptor.once('request', ({ request }) => {
+    request.respondWith(new Response('mocked-body'))
+  })
+
   const { resBody } = await httpsGet('https://example.com')
   expect(resBody).toEqual('mocked-body')
-  expect(requestListener).toHaveBeenCalledTimes(1)
 })
 
 test('throws when responding with a network error', async () => {
-  requestListener.mockImplementationOnce(({ request }) => {
+  interceptor.once('request', ({ request }) => {
     /**
      * @note "Response.error()" static method is NOT implemented in Miniflare.
      * This expression will throw.