fix: prevent frame abort listener leak

.github/workflows/ci.yml

@@ -146,6 +146,9 @@ jobs:
       - name: Node.js tests
         run: pnpm test:node
 
+      - name: Memory tests
+        run: pnpm test:memory
+
   test-node-22:
     name: test (node.js) (22)
     needs: build-22
@@ -179,6 +182,9 @@ jobs:
       - name: Node.js tests
         run: pnpm test:node
 
+      - name: Memory tests
+        run: pnpm test:memory
+
   test-e2e:
     name: test (e2e) (20)
     needs: build-20

package.json

@@ -183,7 +183,7 @@
     "build": "pnpm clean && cross-env NODE_ENV=production tsup && pnpm patch:dts",
     "patch:dts": "node \"./config/scripts/patch-ts.js\"",
     "publint": "publint",
-    "test": "pnpm test:unit && pnpm test:node && pnpm test:browser && pnpm test:native",
+    "test": "pnpm test:unit && pnpm test:node && pnpm test:browser && pnpm test:native && pnpm test:memory",
     "test:unit": "vitest",
     "test:node": "vitest --config=./test/node/vitest.config.ts",
     "test:native": "vitest --config=./test/native/vitest.config.ts",
@@ -192,6 +192,7 @@
     "test:modules:browser": "playwright test -c ./test/modules/browser/playwright.config.ts",
     "test:e2e": "vitest run --config=./test/e2e/vitest.config.ts",
     "test:ts": "vitest --config=./test/typings/vitest.config.ts",
+    "test:memory": "vitest --config=./test/memory/vitest.config.ts",
     "prepare": "pnpm simple-git-hooks init",
     "prepack": "pnpm build",
     "release": "release publish",
@@ -256,7 +257,7 @@
     "outvariant": "^1.4.3",
     "path-to-regexp": "^6.3.0",
     "picocolors": "^1.1.1",
-    "rettime": "^0.11.7",
+    "rettime": "^0.11.11",
     "statuses": "^2.0.2",
     "strict-event-emitter": "^0.5.1",
     "tough-cookie": "^6.0.1",

src/browser/sources/service-worker-source.ts

@@ -330,6 +330,7 @@ Please consider using a custom "serviceWorker.url" option to point to the actual
 
   async #handleResponse(event: WorkerChannelResponseEvent): Promise<void> {
     const { request, response, isMockedResponse } = event.data
+    const frame = this.#frames.get(request.id)
 
     /**
      * CORS requests with `mode: "no-cors"` result in "opaque" responses.
@@ -340,10 +341,10 @@ Please consider using a custom "serviceWorker.url" option to point to the actual
      */
     if (response.type?.includes('opaque')) {
       this.#frames.delete(request.id)
+      frame?.events.removeAllListeners()
       return
     }
 
-    const frame = this.#frames.get(request.id)
     this.#frames.delete(request.id)
 
     /**
@@ -379,17 +380,21 @@ Please consider using a custom "serviceWorker.url" option to point to the actual
             },
           )
 
-    frame.events.emit(
-      new ResponseEvent(
-        isMockedResponse ? 'response:mocked' : 'response:bypass',
-        {
-          requestId: frame.data.id,
-          request: fetchRequest,
-          response: fetchResponse,
-          isMockedResponse,
-        },
-      ),
-    )
+    try {
+      frame.events.emit(
+        new ResponseEvent(
+          isMockedResponse ? 'response:mocked' : 'response:bypass',
+          {
+            requestId: frame.data.id,
+            request: fetchRequest,
+            response: fetchResponse,
+            isMockedResponse,
+          },
+        ),
+      )
+    } finally {
+      frame.events.removeAllListeners()
+    }
   }
 
   #defaultFindWorker: FindWorker = (workerUrl, mockServiceWorkerUrl) => {

src/core/experimental/sources/interceptor-source.ts

@@ -87,16 +87,26 @@ export class InterceptorSource extends NetworkSource {
     }
 
     queueMicrotask(() => {
-      httpFrame.events.emit(
-        new ResponseEvent(
-          isMockedResponse ? 'response:mocked' : 'response:bypass',
-          {
-            requestId,
-            request,
-            response,
-          },
-        ),
-      )
+      try {
+        httpFrame.events.emit(
+          new ResponseEvent(
+            isMockedResponse ? 'response:mocked' : 'response:bypass',
+            {
+              requestId,
+              request,
+              response,
+            },
+          ),
+        )
+      } finally {
+        /**
+         * @note Remove any listeners from this frame.
+         * Past this point, it won't emit anything. The removal is crucial
+         * to prevent "rettime" from keeping the abort cleanup listeners internally.
+         * @see https://github.com/mswjs/msw/issues/2735
+         */
+        httpFrame.events.removeAllListeners()
+      }
     })
   }
 

test/memory/2735-memory-leak.test.ts

@@ -0,0 +1,153 @@
+// @vitest-environment node
+/**
+ * @see https://github.com/mswjs/msw/issues/2735
+ */
+import * as fs from 'node:fs'
+import * as os from 'node:os'
+import * as path from 'node:path'
+import { setTimeout as delay } from 'node:timers/promises'
+import { writeHeapSnapshot } from 'node:v8'
+import { http } from 'msw'
+import { setupServer } from 'msw/node'
+
+const TOTAL_REQUESTS = 5_000
+const CONCURRENCY = 20
+const URL = 'https://localhost/leak'
+
+async function fireBatch(count: number, concurrency: number): Promise<void> {
+  let inflight = 0
+  let started = 0
+  let failed = false
+
+  return new Promise((resolve, reject) => {
+    const next = () => {
+      if (failed) {
+        return
+      }
+      if (started >= count && inflight === 0) {
+        return resolve()
+      }
+
+      while (inflight < concurrency && started < count) {
+        started++
+        inflight++
+        fetch(URL)
+          .then((r) => r.text())
+          .catch((error) => {
+            failed = true
+            reject(error)
+          })
+          .finally(() => {
+            inflight--
+            next()
+          })
+      }
+    }
+    next()
+  })
+}
+
+async function forceGc(): Promise<void> {
+  for (let i = 0; i < 6; i++) {
+    global.gc?.()
+    await delay(30)
+  }
+}
+
+function countConstructors(snapPath: string): Map<string, number> {
+  const json = JSON.parse(fs.readFileSync(snapPath, 'utf8'))
+  const meta = json.snapshot.meta
+  const F = meta.node_fields.length
+  const fT = meta.node_fields.indexOf('type')
+  const fN = meta.node_fields.indexOf('name')
+  const objIdx = meta.node_types[0].indexOf('object')
+  const counts = new Map<string, number>()
+
+  for (let i = 0; i < json.snapshot.node_count; i++) {
+    const b = i * F
+    if (json.nodes[b + fT] !== objIdx) {
+      continue
+    }
+    const name = json.strings[json.nodes[b + fN]]
+    counts.set(name, (counts.get(name) ?? 0) + 1)
+  }
+
+  return counts
+}
+
+const server = setupServer()
+
+beforeAll(() => {
+  server.listen()
+})
+
+afterAll(() => {
+  server.close()
+})
+
+it('does not retain a per-request Emitter after the response is delivered', async () => {
+  server.use(
+    http.get('https://localhost/leak', () => {
+      return new Response()
+    }),
+  )
+
+  const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'msw-emitter-leak-'))
+
+  // Warm-up so JIT settles and we don't count startup allocation.
+  await fireBatch(200, CONCURRENCY)
+  await forceGc()
+
+  const baseSnap = path.join(tmp, 'baseline.heapsnapshot')
+  writeHeapSnapshot(baseSnap)
+  const baseCounts = countConstructors(baseSnap)
+
+  // The actual measurement burst.
+  await fireBatch(TOTAL_REQUESTS, CONCURRENCY)
+  await delay(2000)
+  await forceGc()
+  await delay(1000)
+  await forceGc()
+
+  const settledSnap = path.join(tmp, 'settled.heapsnapshot')
+  writeHeapSnapshot(settledSnap)
+  const settledCounts = countConstructors(settledSnap)
+
+  const watch = [
+    'Emitter',
+    'Listener',
+    'LensList',
+    'WeakMap',
+    'WeakSet',
+  ] as const
+  console.log(`\nConstructor count after ${TOTAL_REQUESTS} requests:\n`)
+  console.log(
+    '  ' +
+      'name'.padEnd(12) +
+      'baseline'.padStart(10) +
+      'settled'.padStart(10) +
+      'delta'.padStart(10),
+  )
+  console.log('  ' + '-'.repeat(42))
+  for (const name of watch) {
+    const b = baseCounts.get(name) ?? 0
+    const s = settledCounts.get(name) ?? 0
+    const d = s - b
+    const dStr = (d >= 0 ? '+' : '') + d
+    console.log(
+      '  ' +
+        name.padEnd(12) +
+        String(b).padStart(10) +
+        String(s).padStart(10) +
+        dStr.padStart(10),
+    )
+  }
+  console.log(`\nHeap snapshots written to: ${tmp}\n`)
+
+  const emitterDelta =
+    (settledCounts.get('Emitter') ?? 0) - (baseCounts.get('Emitter') ?? 0)
+
+  // After the fix, this delta is essentially 0. Today, it's ~= TOTAL.
+  // Threshold: 5% of request count to allow for noise.
+  expect(emitterDelta).toBeLessThan(TOTAL_REQUESTS * 0.05)
+})

test/memory/vitest.config.ts

@@ -0,0 +1,11 @@
+import { defineConfig } from 'vitest/config'
+
+export default defineConfig({
+  test: {
+    dir: './test/memory',
+    globals: true,
+    testTimeout: 120_000,
+    pool: 'forks',
+    execArgv: ['--expose-gc'],
+  },
+})