feat(sse): invoke finalize on response stream end (#2741)

src/core/handlers/HttpHandler.test.ts

@@ -252,3 +252,92 @@ describe('run', () => {
     await expect(run()).resolves.toBe('complete')
   })
 })
+
+describe('finalize', () => {
+  it('returns the exact response instance if the resolver never accesses finalize', async () => {
+    const response = new HttpResponse(
+      new ReadableStream({
+        start(controller) {
+          controller.enqueue(new TextEncoder().encode('hello'))
+          controller.close()
+        },
+      }),
+    )
+    const handler = new HttpHandler('GET', '/resource', () => {
+      return response
+    })
+
+    const result = await handler.run({
+      request: new Request(new URL('/resource', location.href)),
+      requestId: createRequestId(),
+    })
+
+    expect(result?.response).toBe(response)
+  })
+
+  it('returns the exact response instance if finalize is accessed but never called', async () => {
+    const response = new HttpResponse(
+      new ReadableStream({
+        start(controller) {
+          controller.close()
+        },
+      }),
+    )
+    const handler = new HttpHandler('GET', '/resource', ({ finalize }) => {
+      expect(finalize).toBeInstanceOf(Function)
+      return response
+    })
+
+    const result = await handler.run({
+      request: new Request(new URL('/resource', location.href)),
+      requestId: createRequestId(),
+    })
+
+    expect(result?.response).toBe(response)
+  })
+
+  it('defers the cleanup until the response stream settles', async () => {
+    const cleanup = vi.fn()
+    const response = new HttpResponse(
+      new ReadableStream({
+        start(controller) {
+          controller.enqueue(new TextEncoder().encode('hello'))
+          controller.close()
+        },
+      }),
+    )
+    const handler = new HttpHandler('GET', '/resource', ({ finalize }) => {
+      finalize(cleanup)
+      return response
+    })
+
+    const result = await handler.run({
+      request: new Request(new URL('/resource', location.href)),
+      requestId: createRequestId(),
+    })
+
+    // The observed response is a copy of the original response.
+    expect(result?.response).not.toBe(response)
+    expect(cleanup).not.toHaveBeenCalled()
+
+    await expect(result?.response?.text()).resolves.toBe('hello')
+    await vi.waitFor(() => {
+      expect(cleanup).toHaveBeenCalledOnce()
+    })
+  })
+
+  it('runs the cleanup immediately for responses without a body', async () => {
+    const cleanup = vi.fn()
+    const handler = new HttpHandler('GET', '/resource', ({ finalize }) => {
+      finalize(cleanup)
+      return new HttpResponse(null)
+    })
+
+    await handler.run({
+      request: new Request(new URL('/resource', location.href)),
+      requestId: createRequestId(),
+    })
+
+    expect(cleanup).toHaveBeenCalledOnce()
+  })
+})

src/core/handlers/RequestHandler.ts

@@ -15,6 +15,7 @@ import {
 import type { GraphQLRequestBody } from './GraphQLHandler'
 import { devUtils } from '../utils/internal/devUtils'
 import { getRawSetCookie } from '../utils/HttpResponse/decorators'
+import { observeResponseBodyStream } from '../utils/internal/observe-response-body-stream'
 
 export type DefaultRequestMultipartBody = Record<
   string,
@@ -31,12 +32,7 @@ export type DefaultBodyType =
   | undefined
 
 export type JsonBodyType =
-  | Record<string, any>
-  | string
-  | number
-  | boolean
-  | null
-  | undefined
+  Record<string, any> | string | number | boolean | null | undefined
 
 export interface RequestHandlerDefaultInfo {
   header: string
@@ -94,6 +90,10 @@ export type ResponseResolverInfo<
   /**
    * Schedule a callback to run after this response resolver completes.
    * Handy for cleaning up the side effects introduced in the resolver.
+   *
+   * For responses with a `ReadableStream` body (including `sse()` handlers),
+   * the callback runs once the response stream settles: it is read to
+   * completion, errored, or canceled, or the request is aborted.
    * @example
    * sse('/', ({ client, finalize }) => {
    *   const interval = setInterval(() => client.send({ data: 'ping' }))
@@ -355,6 +355,20 @@ export abstract class RequestHandler<
 
     const listenerController = new AbortController()
 
+    /**
+     * @note Initialize the `finalize` machinery lazily, on the first
+     * access of the `finalize` property by the resolver. If the resolver
+     * never accesses it, the handler behaves as if `finalize` never
+     * existed: no abort listeners, no scheduled cleanups, and no response
+     * body stream observation (see `this.complete()`).
+     */
+    let finalizeFunction: ResponseResolverFinalizeFunction | undefined
+
+    const getFinalize = (): ResponseResolverFinalizeFunction => {
+      if (finalizeFunction == null) {
+        // Run any scheduled cleanups if the request gets aborted
+        // while the resolver is still executing.
+        if (!args.request.signal.aborted) {
           args.request.signal.addEventListener(
             'abort',
             () => this.runScheduledCleanups(args.requestId),
@@ -363,12 +377,32 @@ export abstract class RequestHandler<
               signal: listenerController.signal,
             },
           )
+        }
+
+        finalizeFunction = (callback) => {
+          this.scheduleCleanup(args.requestId, callback)
+
+          /**
+           * @note Run the cleanup immediately if the request has already
+           * been aborted. The "abort" listener above never fires for an
+           * already-aborted signal (and fires at most once), while
+           * long-lived resolvers (streams, generators) may never settle
+           * to run the cleanups on completion.
+           */
+          if (args.request.signal.aborted) {
+            void this.runScheduledCleanups(args.requestId)
+          }
+        }
+      }
+
+      return finalizeFunction
+    }
 
     const mockedResponsePromise = (
       executeResolver({
         ...resolverExtras,
-        finalize: (callback) => {
-          this.scheduleCleanup(args.requestId, callback)
+        get finalize(): ResponseResolverFinalizeFunction {
+          return getFinalize()
         },
         requestId: args.requestId,
         request: args.request,
@@ -423,9 +457,13 @@ export abstract class RequestHandler<
         }
 
         if (!isIterable(result)) {
-          // Otherwise, run the cleanup immediately if it's a plain response resolver.
-          await this.runScheduledCleanups(info.requestId)
-          return result
+          // Otherwise, run the cleanups if it's a plain response resolver
+          // (deferred until the response stream settles for streamed responses).
+          return this.complete({
+            request: info.request,
+            requestId: info.requestId,
+            response: result,
+          })
         }
 
         /**
@@ -465,11 +503,13 @@ export abstract class RequestHandler<
         // only after it's been completely exhausted.
         this.isUsed = true
 
-        await this.runScheduledCleanups(info.requestId)
-
         // Clone the previously stored response so it can be read
         // when receiving it repeatedly from the "done" generator.
-        return this.resolverIteratorResult?.clone()
+        return this.complete({
+          request: info.request,
+          requestId: info.requestId,
+          response: this.resolverIteratorResult?.clone(),
+        })
       }
 
       return nextResponse
@@ -532,28 +572,98 @@ export abstract class RequestHandler<
     }
   }
 
-  private async runScheduledCleanups(requestId: string): Promise<void> {
+  /**
+   * Remove and return the cleanups scheduled for the given request
+   * (or the pending iterator cleanups for generator resolvers).
+   */
+  private takeScheduledCleanups(
+    requestId: string,
+  ): Array<() => MaybePromise<void>> | undefined {
     if (
       this.resolverIterator &&
       this.resolverIteratorCleanups != null &&
       this.resolverIteratorCleanups.length > 0
     ) {
-      try {
-        await this.exhaustCleanups(this.resolverIteratorCleanups)
-      } finally {
+      const cleanups = this.resolverIteratorCleanups
       this.resolverIteratorCleanups = undefined
-      }
-      return
+      return cleanups
     }
 
     const cleanups = this.scheduledCleanups.get(requestId)
 
-    if (!cleanups || cleanups.length == 0) {
+    if (!cleanups || cleanups.length === 0) {
+      return undefined
+    }
+
+    this.scheduledCleanups.delete(requestId)
+    return cleanups
+  }
+
+  private async runScheduledCleanups(requestId: string): Promise<void> {
+    const cleanups = this.takeScheduledCleanups(requestId)
+
+    if (cleanups) {
+      await this.exhaustCleanups(cleanups)
+    }
+  }
+
+  /**
+   * Conclude the response resolution for the given request.
+   * Runs the scheduled cleanups immediately for responses without a
+   * `ReadableStream` body. For streamed responses, returns an observed
+   * copy of the response and defers the cleanups until its body settles
+   * (is read to completion, errored, or canceled) or the request is
+   * aborted, whichever comes first.
+   */
+  private async complete(args: {
+    request: StrictRequest<any>
+    requestId: string
+    response: Response | undefined | void
+  }): Promise<Response | undefined | void> {
+    const cleanups = this.takeScheduledCleanups(args.requestId)
+
+    if (!cleanups) {
+      return args.response
+    }
+
+    const observedResponse = args.response
+      ? observeResponseBodyStream(args.response)
+      : null
+
+    if (!observedResponse) {
+      await this.exhaustCleanups(cleanups)
+      return args.response
+    }
+
+    const listenerController = new AbortController()
+
+    const runCleanupsOnce = (): void => {
+      if (listenerController.signal.aborted) {
         return
       }
 
-    await this.exhaustCleanups(cleanups)
-    this.scheduledCleanups.delete(requestId)
+      listenerController.abort()
+      void this.exhaustCleanups(cleanups)
+    }
+
+    void observedResponse.settled.then(runCleanupsOnce)
+
+    /**
+     * @note Also run the cleanups when the request is aborted.
+     * Stream cancellation does not always propagate to the observed
+     * response (e.g. if an unconsumed clone of the response exists),
+     * while the request abort reliably means the response is unused.
+     */
+    if (args.request.signal.aborted) {
+      runCleanupsOnce()
+    } else {
+      args.request.signal.addEventListener('abort', runCleanupsOnce, {
+        once: true,
+        signal: listenerController.signal,
+      })
+    }
+
+    return observedResponse.response
   }
 }
 

src/core/sse.ts

@@ -1,6 +1,6 @@
 import { invariant } from 'outvariant'
 import { DeferredPromise } from '@open-draft/deferred-promise'
-import { Emitter } from 'strict-event-emitter'
+import { Emitter, TypedEvent } from 'rettime'
 import type { ResponseResolver } from './handlers/RequestHandler'
 import {
   HttpHandler,
@@ -105,11 +105,17 @@ class ServerSentEventHandler<
         this.#attachClientLogger(info.request, emitter)
       }
 
-      await resolver({
-        ...info,
+      /**
+       * @note Extend the resolver info via assignment instead of a spread.
+       * Spreading the info would invoke its lazy `finalize` getter,
+       * initializing the finalize machinery for resolvers that never use it.
+       */
+      await resolver(
+        Object.assign(info, {
           client,
           server,
-      })
+        }),
+      )
 
       return response
     })
@@ -151,7 +157,14 @@ class ServerSentEventHandler<
     return matches
   }
 
-  async log(_args: { request: Request; response: Response }): Promise<void> {
+  async log(args: { request: Request; response: Response }): Promise<void> {
+    /**
+     * @note Cancel the response stream because it's not needed for logging.
+     * Otherwise, this cloned response remains unconsumed and its original
+     * doesn't propagate stream cancelations at all.
+     */
+    args.response.body?.cancel()
+
     /**
      * @note Skip the default `this.log()` logic so that when this handler is logged
      * upon handling the request, nothing is printed (we log SSE requests early).
@@ -166,16 +179,14 @@ class ServerSentEventHandler<
     const publicUrl = toPublicUrl(request.url)
 
     /* eslint-disable no-console */
-    emitter.on('message', (payload) => {
+    emitter.on('message', ({ data }) => {
       console.groupCollapsed(
-        devUtils.formatMessage(
-          `${getTimestamp()} SSE %s %c⇣%c ${payload.event}`,
-        ),
+        devUtils.formatMessage(`${getTimestamp()} SSE %s %c⇣%c ${data.event}`),
         publicUrl,
         `color:${colors.mocked}`,
         'color:inherit',
       )
-      console.log(payload.frames)
+      console.log(data.frames)
       console.groupEnd()
     })
 
@@ -227,9 +238,9 @@ type ToEventDiscriminatedUnion<T> = Values<{
 }>
 
 type ServerSentEventClientEventMap = {
-  message: [payload: EventStreamMessage]
-  error: []
-  close: []
+  message: TypedEvent<EventStreamMessage>
+  error: TypedEvent
+  close: TypedEvent
 }
 
 const kClientEmitter = Symbol.for('kClientEmitter')
@@ -240,11 +251,13 @@ class ServerSentEventClient<
   private [kClientEmitter]?: Emitter<ServerSentEventClientEventMap>
 
   #encoder: TextEncoder
-  #writer: WritableStreamDefaultWriter
+  #controller: ReadableStreamDefaultController<Uint8Array>
+  #closed: DeferredPromise<void>
 
-  constructor(writable: WritableStream) {
+  constructor(controller: ReadableStreamDefaultController<Uint8Array>) {
     this.#encoder = new TextEncoder()
-    this.#writer = writable.getWriter()
+    this.#controller = controller
+    this.#closed = new DeferredPromise()
   }
 
   /**
@@ -300,37 +313,50 @@ class ServerSentEventClient<
    * error.
    */
   public error(): void {
-    this.#writer.abort().catch((error) => {
-      console.error(error)
-      devUtils.error(
-        'Failed to abort server-side EventSource. Please see the original error above.',
-      )
-    })
-    this[kClientEmitter]?.emit('error')
+    if (this.#closed.state !== 'pending') {
+      return
+    }
+
+    this.#controller.error()
+    this.#closed.resolve()
+    this[kClientEmitter]?.emit(new TypedEvent('error'))
   }
 
   /**
    * Closes the underlying `EventSource`, closing the connection.
    */
   public close(): void {
-    this.#writer.close().catch((error) => {
+    if (this.#closed.state !== 'pending') {
+      return
+    }
+
+    try {
+      this.#controller.close()
+      this.#closed.resolve()
+    } catch {
+      //
+    }
+
+    this[kClientEmitter]?.emit(new TypedEvent('close'))
+  }
+
+  #enqueue(chunk: Uint8Array): void {
+    if (this.#closed.state !== 'pending') {
+      return
+    }
+
+    try {
+      this.#controller.enqueue(chunk)
+    } catch (error) {
       console.error(error)
       devUtils.error(
-        'Failed to close server-side EventSource. Please see the original error above.',
+        'Failed to write to server-side EventSource. Please see the original error above.',
       )
-    })
-    this[kClientEmitter]?.emit('close')
+    }
   }
 
   #sendRetry(retry: number): void {
-    this.#writer
-      .write(this.#encoder.encode(`retry:${retry}\n\n`))
-      .catch((error) => {
-        console.error(error)
-        devUtils.error(
-          'Failed to send a retry packet to server-side EventSource. Please see the original error above.',
-        )
-      })
+    this.#enqueue(this.#encoder.encode(`retry:${retry}\n\n`))
   }
 
   #sendMessage(message: {
@@ -361,21 +387,18 @@ class ServerSentEventClient<
 
     frames.push('', '')
 
-    this.#writer
-      .write(this.#encoder.encode(frames.join('\n')))
-      .catch((error) => {
-        console.error(error)
-        devUtils.error(
-          'Failed to send a message to server-side EventSource. Please see the original error above.',
-        )
-      })
+    this.#enqueue(this.#encoder.encode(frames.join('\n')))
 
-    this[kClientEmitter]?.emit('message', {
+    this[kClientEmitter]?.emit(
+      new TypedEvent('message', {
+        data: {
           id: message.id,
           event: message.event?.toString() || 'message',
           data: message.data,
           frames,
-    })
+        },
+      }),
+    )
   }
 }
 
@@ -1002,9 +1025,18 @@ function createEventStream<EventMap extends EventMapConstraint>(
     request.url,
   )
 
-  const { readable, writable } = new TransformStream()
+  let controller!: ReadableStreamDefaultController<Uint8Array>
 
-  const client = new ServerSentEventClient<EventMap>(writable)
+  const readable = new ReadableStream<Uint8Array>({
+    start(defaultController) {
+      controller = defaultController
+    },
+    cancel() {
+      client.close()
+    },
+  })
+
+  const client = new ServerSentEventClient<EventMap>(controller)
   const server = new ServerSentEventServer({
     request,
     client,

src/core/utils/HttpResponse/decorators.ts

@@ -1,5 +1,5 @@
 import statuses from '../../../shims/statuses'
-import type { HttpResponseInit } from '../../HttpResponse'
+import type { HttpResponse, HttpResponseInit } from '../../HttpResponse'
 
 const { message } = statuses
 
@@ -27,7 +27,7 @@ export function normalizeResponseInit(
 }
 
 export function decorateResponse(
-  response: Response,
+  response: HttpResponse<any>,
   init: HttpResponseDecoratedInit,
 ): Response {
   // Allow mocking the response type.
@@ -59,3 +59,24 @@ export function decorateResponse(
 export function getRawSetCookie(response: Response): string | undefined {
   return Reflect.get(response, kSetCookie)
 }
+
+/**
+ * Copy the given response own properties, like internal symbols,
+ * onto another response. Used for faithful internal copying of responses.
+ */
+export function copyResponseOwnProperties(
+  source: Response,
+  target: Response,
+): void {
+  for (const propertyName of Reflect.ownKeys(source)) {
+    const descriptor = Object.getOwnPropertyDescriptor(source, propertyName)
+    const existingDescriptor = Object.getOwnPropertyDescriptor(
+      target,
+      propertyName,
+    )
+
+    if (descriptor && existingDescriptor == null) {
+      Object.defineProperty(target, propertyName, descriptor)
+    }
+  }
+}

src/core/utils/internal/observe-response-body-stream.test.ts

@@ -0,0 +1,64 @@
+import { observeResponseBodyStream } from './observe-response-body-stream'
+
+it('returns null for a response without a body', () => {
+  expect(observeResponseBodyStream(new Response(null))).toBeNull()
+})
+
+it('returns null for a response whose body is already used', async () => {
+  const response = new Response('hello')
+  await response.text()
+
+  expect(observeResponseBodyStream(response)).toBeNull()
+})
+
+it('returns null for a response whose body is locked', () => {
+  const response = new Response('hello')
+  response.body?.getReader()
+
+  expect(observeResponseBodyStream(response)).toBeNull()
+})
+
+it('resolves "settled" when the response body is read to completion', async () => {
+  const observed = observeResponseBodyStream(new Response('hello'))
+  expect(observed).not.toBeNull()
+
+  await expect(observed?.response.text()).resolves.toBe('hello')
+  await expect(observed?.settled).resolves.toBeUndefined()
+})
+
+it('resolves "settled" when the response body errors', async () => {
+  const stream = new ReadableStream({
+    start(controller) {
+      controller.error(new Error('stream error'))
+    },
+  })
+  const observed = observeResponseBodyStream(new Response(stream))
+  expect(observed).not.toBeNull()
+
+  await expect(observed?.response.text()).rejects.toThrow('stream error')
+  await expect(observed?.settled).resolves.toBeUndefined()
+})
+
+it('resolves "settled" when the response body is canceled', async () => {
+  const stream = new ReadableStream({
+    pull(controller) {
+      controller.enqueue(new TextEncoder().encode('ping'))
+    },
+  })
+  const observed = observeResponseBodyStream(new Response(stream))
+  const reader = observed?.response.body?.getReader()
+  expect(reader).toBeDefined()
+
+  await reader?.read()
+  await reader?.cancel(new Error('client disconnected'))
+
+  await expect(observed?.settled).resolves.toBeUndefined()
+})
+
+it('resolves "settled" when the response body is consumed via piping', async () => {
+  const observed = observeResponseBodyStream(new Response('hello'))
+  expect(observed).not.toBeNull()
+
+  await observed?.response.body?.pipeTo(new WritableStream())
+  await expect(observed?.settled).resolves.toBeUndefined()
+})

src/core/utils/internal/observe-response-body-stream.ts

@@ -0,0 +1,77 @@
+import { DeferredPromise } from '@open-draft/deferred-promise'
+import { FetchResponse } from '@mswjs/interceptors'
+import { copyResponseOwnProperties } from '../HttpResponse/decorators'
+
+export interface ObservedResponse {
+  response: Response
+  /**
+   * A promise that resolves once the response body stream has settled:
+   * it was closed, errored, or canceled, and will not be used anymore.
+   */
+  settled: Promise<void>
+}
+
+/**
+ * Observe the `ReadableStream` body of the given response.
+ * Returns a copy of that response whose body reports when it has
+ * settled (was read to completion, errored, or canceled by the consumer).
+ * Returns `null` for responses whose body cannot be observed
+ * (no body, already used, or locked).
+ */
+export function observeResponseBodyStream(
+  response: Response,
+): ObservedResponse | null {
+  if (response.body == null || response.bodyUsed || response.body.locked) {
+    return null
+  }
+
+  const settled = new DeferredPromise<void>()
+  const reader = response.body.getReader()
+
+  /**
+   * @note Relay the body through a manual underlying source instead of
+   * `.pipeThrough(new TransformStream({ flush, cancel }))`. The `cancel`
+   * transformer callback is not implemented in Chromium, which loses
+   * the stream error/cancelation signals there entirely.
+   */
+  const observedStream = new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      try {
+        const readResult = await reader.read()
+
+        if (readResult.done) {
+          settled.resolve()
+          controller.close()
+          return
+        }
+
+        controller.enqueue(readResult.value)
+      } catch (error) {
+        settled.resolve()
+        throw error
+      }
+    },
+    async cancel(reason) {
+      settled.resolve()
+      await reader.cancel(reason)
+    },
+  })
+
+  /**
+   * @note Reconstruct the response because the body of an existing
+   * response cannot be replaced. Use `FetchResponse` to support
+   * non-standard response status codes (e.g. 101).
+   */
+  const observedResponse = new FetchResponse(observedStream, {
+    status: response.status,
+    statusText: response.statusText,
+    headers: response.headers,
+  })
+
+  copyResponseOwnProperties(response, observedResponse)
+
+  return {
+    response: observedResponse,
+    settled,
+  }
+}

src/mockServiceWorker.js

@@ -137,8 +137,18 @@ async function handleRequest(event, requestId, requestInterceptedAt) {
   if (client && activeClientIds.has(client.id)) {
     const serializedRequest = await serializeRequest(requestCloneForEvents)
 
+    // Omit the body of server-sent event stream responses.
+    // Cloning such responses would prevent client-side stream cancelations
+    // from reaching the original stream (a teed stream only cancels its
+    // source once both of its branches cancel) and would buffer the
+    // entire stream into the unconsumed clone indefinitely.
+    const isEventStreamResponse = response.headers
+      .get('content-type')
+      ?.toLowerCase()
+      .startsWith('text/event-stream')
+
     // Clone the response so both the client and the library could consume it.
-    const responseClone = response.clone()
+    const responseClone = isEventStreamResponse ? null : response.clone()
 
     sendToClient(
       client,
@@ -151,15 +161,17 @@ async function handleRequest(event, requestId, requestInterceptedAt) {
             ...serializedRequest,
           },
           response: {
-            type: responseClone.type,
-            status: responseClone.status,
-            statusText: responseClone.statusText,
-            headers: Object.fromEntries(responseClone.headers.entries()),
-            body: responseClone.body,
+            type: response.type,
+            status: response.status,
+            statusText: response.statusText,
+            headers: Object.fromEntries(response.headers.entries()),
+            body: responseClone ? responseClone.body : null,
           },
         },
       },
-      responseClone.body ? [serializedRequest.body, responseClone.body] : [],
+      responseClone && responseClone.body
+        ? [serializedRequest.body, responseClone.body]
+        : [],
     )
   }
 

test/browser/sse-api/sse.finalize.test.ts

@@ -0,0 +1,214 @@
+import type { sse } from 'msw'
+import type { setupWorker } from 'msw/browser'
+import { DeferredPromise } from '@open-draft/deferred-promise'
+import { test, expect } from '../playwright.extend'
+
+declare namespace window {
+  export const msw: {
+    setupWorker: typeof setupWorker
+    sse: typeof sse
+  }
+}
+
+const EXAMPLE_URL = new URL('./sse.mocks.ts', import.meta.url)
+
+test('runs cleanup after the event source is closed by the client', async ({
+  loadExample,
+  page,
+}) => {
+  await loadExample(EXAMPLE_URL, {
+    skipActivation: true,
+  })
+
+  const finalizedAt = new DeferredPromise<number>()
+  await page.exposeFunction('notifyFinalized', () => {
+    finalizedAt.resolve(Date.now())
+  })
+
+  await page.evaluate(async () => {
+    const { setupWorker, sse } = window.msw
+
+    const worker = setupWorker(
+      sse('http://localhost/stream', ({ finalize }) => {
+        finalize(() => window.notifyFinalized())
+      }),
+    )
+    await worker.start()
+  })
+
+  const closedAt = await page.evaluate(() => {
+    const source = new EventSource('http://localhost/stream')
+
+    return new Promise<number>((resolve) => {
+      source.addEventListener('open', () => {
+        source.close()
+        resolve(Date.now())
+      })
+    })
+  })
+
+  await expect(finalizedAt).resolves.toBeGreaterThanOrEqual(closedAt)
+})
+
+test('runs cleanup after the event source is closed by the handler', async ({
+  loadExample,
+  page,
+}) => {
+  await loadExample(EXAMPLE_URL, {
+    skipActivation: true,
+  })
+
+  const finalizedAt = new DeferredPromise<number>()
+  await page.exposeFunction('notifyFinalized', () => {
+    finalizedAt.resolve(Date.now())
+  })
+
+  await page.evaluate(async () => {
+    const { setupWorker, sse } = window.msw
+
+    const worker = setupWorker(
+      sse('http://localhost/stream', ({ client, finalize }) => {
+        setTimeout(() => client.close(), 250)
+        finalize(() => window.notifyFinalized())
+      }),
+    )
+    await worker.start()
+  })
+
+  const closedAt = await page.evaluate(() => {
+    const source = new EventSource('http://localhost/stream')
+
+    return new Promise<number>((resolve) => {
+      source.addEventListener('open', () => {
+        resolve(Date.now())
+      })
+    })
+  })
+
+  await expect(finalizedAt).resolves.toBeGreaterThanOrEqual(closedAt)
+})
+
+test('runs cleanup after the event source is errored by the handler', async ({
+  loadExample,
+  page,
+}) => {
+  await loadExample(EXAMPLE_URL, {
+    skipActivation: true,
+  })
+
+  const finalizedAt = new DeferredPromise<number>()
+  await page.exposeFunction('notifyFinalized', () => {
+    finalizedAt.resolve(Date.now())
+  })
+
+  await page.evaluate(async () => {
+    const { setupWorker, sse } = window.msw
+
+    const worker = setupWorker(
+      sse('http://localhost/stream', ({ client, finalize }) => {
+        setTimeout(() => client.error(), 250)
+        finalize(() => window.notifyFinalized())
+      }),
+    )
+    await worker.start()
+  })
+
+  const closedAt = await page.evaluate(() => {
+    const source = new EventSource('http://localhost/stream')
+
+    return new Promise<number>((resolve) => {
+      source.addEventListener('open', () => {
+        resolve(Date.now())
+      })
+    })
+  })
+
+  await expect(finalizedAt).resolves.toBeGreaterThanOrEqual(closedAt)
+})
+
+test('runs independent cleanups for parallel event sources', async ({
+  loadExample,
+  page,
+}) => {
+  await loadExample(EXAMPLE_URL, {
+    skipActivation: true,
+  })
+
+  const finalizedIds: Array<string> = []
+  await page.exposeFunction('notifyFinalized', (id: string) => {
+    finalizedIds.push(id)
+  })
+
+  await page.evaluate(async () => {
+    const { setupWorker, sse } = window.msw
+
+    const worker = setupWorker(
+      sse('http://localhost/stream', ({ request, finalize }) => {
+        const url = new URL(request.url)
+        finalize(() => window.notifyFinalized(url.searchParams.get('id')))
+      }),
+    )
+    await worker.start()
+  })
+
+  await page.evaluate(() => {
+    const sourceOne = new EventSource('http://localhost/stream?id=one')
+    const sourceTwo = new EventSource('http://localhost/stream?id=two')
+    Object.assign(window, { sourceOne, sourceTwo })
+
+    return Promise.all([
+      new Promise((resolve) => {
+        sourceOne.addEventListener('open', resolve)
+      }),
+      new Promise((resolve) => {
+        sourceTwo.addEventListener('open', resolve)
+      }),
+    ])
+  })
+
+  // Closing one event source must only run the cleanups
+  // scheduled for that connection.
+  await page.evaluate(() => {
+    window.sourceOne.close()
+  })
+
+  await expect.poll(() => finalizedIds).toEqual(['one'])
+
+  await page.evaluate(() => {
+    window.sourceTwo.close()
+  })
+
+  await expect.poll(() => finalizedIds).toEqual(['one', 'two'])
+})
+
+test('runs cleanup when the handler closes the event source synchronously', async ({
+  loadExample,
+  page,
+}) => {
+  await loadExample(EXAMPLE_URL, {
+    skipActivation: true,
+  })
+
+  const finalized = new DeferredPromise<void>()
+  await page.exposeFunction('notifyFinalized', () => {
+    finalized.resolve()
+  })
+
+  await page.evaluate(async () => {
+    const { setupWorker, sse } = window.msw
+
+    const worker = setupWorker(
+      sse('http://localhost/stream', ({ client, finalize }) => {
+        finalize(() => window.notifyFinalized())
+        client.close()
+      }),
+    )
+    await worker.start()
+  })
+
+  await page.evaluate(() => {
+    new EventSource('http://localhost/stream')
+  })
+
+  await expect(finalized).resolves.toBeUndefined()
+})

test/node/msw-api/finalize.test.ts

@@ -1,7 +1,7 @@
 // @vitest-environment node
+import { setTimeout } from 'node:timers/promises'
 import { http, passthrough } from 'msw'
 import { setupServer } from 'msw/node'
-import { setTimeout } from 'node:timers/promises'
 
 const server = setupServer()
 
@@ -149,6 +149,66 @@ it('runs after the request has been aborted', async () => {
   expect(cleanup).toHaveBeenCalledOnce()
 })
 
+it('runs immediately when scheduled after the request has been aborted', async () => {
+  const cleanup = vi.fn()
+
+  server.use(
+    http.get('http://localhost/resource', async ({ finalize }) => {
+      // Await past the point where the request gets aborted so the
+      // first `finalize` access happens on an already-aborted signal.
+      await setTimeout(250)
+      finalize(cleanup)
+
+      // Simulate a long-lived resolver that never settles.
+      await new Promise(() => {})
+    }),
+  )
+
+  const controller = new AbortController()
+  const responsePromise = fetch('http://localhost/resource', {
+    signal: controller.signal,
+  })
+  await setTimeout(100)
+  controller.abort()
+
+  await expect(responsePromise).rejects.toThrow()
+  await vi.waitFor(() => {
+    expect(cleanup).toHaveBeenCalledOnce()
+  })
+})
+
+it('runs cleanups scheduled after the abort listener has fired', async () => {
+  const cleanupBeforeAbort = vi.fn()
+  const cleanupAfterAbort = vi.fn()
+
+  server.use(
+    http.get('http://localhost/resource', async ({ finalize }) => {
+      finalize(cleanupBeforeAbort)
+
+      // Await past the point where the request gets aborted
+      // (the "abort" listener fires and runs `cleanupBeforeAbort`).
+      await setTimeout(250)
+      finalize(cleanupAfterAbort)
+
+      // Simulate a long-lived resolver that never settles.
+      await new Promise(() => {})
+    }),
+  )
+
+  const controller = new AbortController()
+  const responsePromise = fetch('http://localhost/resource', {
+    signal: controller.signal,
+  })
+  await setTimeout(100)
+  controller.abort()
+
+  await expect(responsePromise).rejects.toThrow()
+  await vi.waitFor(() => {
+    expect(cleanupBeforeAbort).toHaveBeenCalledOnce()
+    expect(cleanupAfterAbort).toHaveBeenCalledOnce()
+  })
+})
+
 it('runs once the generator resolver is exhausted', async () => {
   const cleanup = vi.fn()