fix: abort rejected response requests

.changeset/failed-stream-cleanup.md

@@ -0,0 +1,5 @@
+---
+'eventsource': patch
+---
+
+Abort the underlying request when a response fails EventSource validation.

src/EventSource.ts

@@ -668,9 +668,8 @@ class EventSourceImpl extends EventTarget implements EventSource {
   #failConnection(message?: string, code?: number) {
     // [spec] …if the readyState attribute is set to a value other than CLOSED,
     // [spec] sets the readyState attribute to CLOSED…
-    if (this.#readyState !== this.CLOSED) {
-      this.#readyState = this.CLOSED
-    }
+    // Release the request before reporting failure, including responses with an unread body.
+    this.close()
 
     // [spec] …and fires an event named `error` at the `EventSource` object.
     // [spec] Once the user agent has failed the connection, it does not attempt to reconnect.

test/client.test.ts

@@ -737,6 +737,42 @@ browserTest(
   },
 )
 
+test.each([200, 403])(
+  'aborts a rejected HTTP %i response before reporting failure',
+  async (status) => {
+    let signal: AbortSignal | undefined
+    let abortedOnError: boolean | undefined
+    const onError = getCallCounter<ErrorEvent>({name: 'connection failure'})
+    const es = new OurEventSource(`${serverUrl}/invalid-stream?status=${status}`, {
+      fetch(url, init) {
+        signal = init.signal
+        return request(url, init)
+      },
+    })
+    es.addEventListener('error', (event) => {
+      abortedOnError = signal?.aborted
+      onError.listener(event)
+    })
+
+    try {
+      await onError.waitForCallCount(1)
+      expect(es.readyState).toBe(OurEventSource.CLOSED)
+      expect(abortedOnError).toBe(true)
+      expect(signal?.aborted).toBe(true)
+      expect(onError.lastArg.code).toBe(status)
+      expect(onError.lastArg.message).toBe(
+        status === 200
+          ? 'Invalid content type, expected "text/event-stream"'
+          : 'Non-200 status code (403)',
+      )
+      es.close()
+      expect(onError.callCount).toBe(1)
+    } finally {
+      es.close()
+    }
+  },
+)
+
 test('throws on `fetch()` that does not return web-stream', async () => {
   const url = `${serverUrl}/`
 

test/helpers/server.ts

@@ -71,6 +71,8 @@ export function handleRequest(
       return writeSlowConnect(req, res)
     case '/debug':
       return writeDebug(req, res)
+    case '/invalid-stream':
+      return writeInvalidStream(req, res)
     case '/set-cookie':
       return writeCookies(req, res)
     case '/authed':
@@ -563,6 +565,14 @@ function writeAuthed(req: IncomingMessage, res: ServerResponse) {
   res.end()
 }
 
+function writeInvalidStream(req: IncomingMessage, res: ServerResponse) {
+  const status =
+    new URL(req.url || '/', 'http://localhost').searchParams.get('status') === '403' ? 403 : 200
+  res.writeHead(status, {'Content-Type': status === 200 ? 'text/plain' : 'text/event-stream'})
+  // Keep the response open so the client owns releasing the failed connection.
+  res.write('This response is not a usable event stream.')
+}
+
 function writeFallback(_req: IncomingMessage, res: ServerResponse) {
   res.writeHead(404, {
     'Content-Type': 'text/plain',