fix(traffic-inspector): release WS subscriber and ping timer on a dead socket (#13155)

Attaching `close`/`error` before subscribing closes the window where a resource is held with no live cleanup path, and the destroyed-socket re-check after the handshake covers the in-flight case. `write()` not throwing synchronously is exactly why the old `try/catch` never fired.

---

Validated in one consolidated worktree cut from `release/v3.8.51`, boarded together with the other 13 PRs of this batch — zero merge conflicts between them.

- `typecheck:core` clean
- complexity 2799 / baseline 3218 and cognitive-complexity 1265 / baseline 1437 — both under baseline
- 71 focused assertions green across the 13 test files this batch adds or touches

⚠️ base-red inherited: #12732 — `Docs Gates (fast-path)`, `Merge integrity`, `No new ESLint warnings`, `Unit Tests fast-path` and `Fast Quality Gates` all reproduce on the pure `release/v3.8.51` tip (provider count 356 vs the 358 the modules define, SKILL.md drift, and `open-sse/utils/stream.ts` at 3115 > frozen 3098). None of them touch this diff.

Thanks @anhtahaylove — the root-cause write-up, the measured before/after numbers and the red-before-green proof on every one of these made the batch reviewable as a unit.

changelog.d/fixes/13152-traffic-inspector-ws-subscriber-leak.md

@@ -0,0 +1 @@
+- **fix(traffic-inspector):** the WebSocket route no longer leaks a traffic-buffer subscriber and a 30s ping timer when the client socket is already closed at handler time — listeners are attached before any resource is acquired, a destroyed socket bails out early, and the ping interval stops on a dead socket where `write()` never throws ([#13155](https://github.com/diegosouzapw/OmniRoute/pull/13155))

src/app/api/tools/traffic-inspector/ws/route.ts

@@ -96,6 +96,14 @@ export async function GET(request: Request): Promise<Response> {
   }
 
   const acceptHeader = acceptKey(clientKey);
+
+  // The client can vanish during the upgrade round trip. `close` has then
+  // ALREADY fired, so the listeners below would never run and every resource
+  // acquired past this point would be held with no path to release it.
+  if (socket.destroyed) {
+    return new Response(null, { status: 101 });
+  }
+
   socket.write(
     [
       "HTTP/1.1 101 Switching Protocols",
@@ -106,21 +114,17 @@ export async function GET(request: Request): Promise<Response> {
     ].join("\r\n")
   );
 
-  const unsubscribe = globalTrafficBuffer.subscribe((ev) => {
-    sendText(socket, ev);
-  });
-
-  const pingTimer = setInterval(() => {
-    try {
-      socket.write(encodeWsFrame(0x09)); // ping
-    } catch {
-      cleanup();
-    }
-  }, PING_INTERVAL_MS);
+  let unsubscribe: (() => void) | null = null;
+  let pingTimer: ReturnType<typeof setInterval> | null = null;
+  let cleanedUp = false;
 
   function cleanup(): void {
-    clearInterval(pingTimer);
-    unsubscribe();
+    if (cleanedUp) return;
+    cleanedUp = true;
+    if (pingTimer) clearInterval(pingTimer);
+    pingTimer = null;
+    unsubscribe?.();
+    unsubscribe = null;
     try {
       socket.destroy();
     } catch {
@@ -128,14 +132,43 @@ export async function GET(request: Request): Promise<Response> {
     }
   }
 
-  socket.once("close", cleanup);
-  socket.once("error", cleanup);
-
-  // Never resolve — the socket is the response channel.
-  await new Promise<void>((resolve) => {
+  // Attached BEFORE any resource is acquired, so there is no window in which a
+  // subscriber or timer exists without a live path to cleanup().
+  const settled = new Promise<void>((resolve) => {
     socket.once("close", resolve);
     socket.once("error", resolve);
   });
+  socket.once("close", cleanup);
+  socket.once("error", cleanup);
+
+  // Re-check: `close` may have fired while we were writing the handshake, in
+  // which case the listeners above already ran and cleanup() is a no-op we
+  // still must not skip.
+  if (socket.destroyed) {
+    cleanup();
+    return new Response(null, { status: 101 });
+  }
+
+  unsubscribe = globalTrafficBuffer.subscribe((ev) => {
+    sendText(socket, ev);
+  });
+
+  pingTimer = setInterval(() => {
+    // `socket.write()` does NOT throw synchronously on a destroyed socket, so
+    // the destroyed check — not the catch — is what stops a dead interval.
+    if (socket.destroyed) {
+      cleanup();
+      return;
+    }
+    try {
+      socket.write(encodeWsFrame(0x09)); // ping
+    } catch {
+      cleanup();
+    }
+  }, PING_INTERVAL_MS);
+
+  // Never resolve — the socket is the response channel.
+  await settled;
 
   cleanup();
   return new Response(null, { status: 101 });

tests/unit/traffic-inspector-ws-subscriber-leak-13152.test.ts

@@ -0,0 +1,134 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import net from "node:net";
+import type { AddressInfo } from "node:net";
+
+import { GET } from "@/app/api/tools/traffic-inspector/ws/route";
+import { globalTrafficBuffer } from "@/mitm/inspector/buffer";
+
+const DEAD_UPGRADES = 6;
+
+function armedTimers(): number {
+  return process.getActiveResourcesInfo().filter((r) => r === "Timeout").length;
+}
+
+function upgradeRequest(socket: net.Socket): Request {
+  const req = new Request("http://127.0.0.1/api/tools/traffic-inspector/ws", {
+    headers: {
+      upgrade: "websocket",
+      "sec-websocket-key": "dGhlIHNhbXBsZSBub25jZQ==",
+    },
+  });
+  Object.defineProperty(req, "socket", { value: socket, configurable: true });
+  return req;
+}
+
+async function deadSocket(port: number): Promise<net.Socket> {
+  const sock = net.connect(port, "127.0.0.1");
+  await new Promise<void>((r) => sock.once("connect", () => r()));
+  sock.on("error", () => {});
+  sock.destroy();
+  await new Promise((r) => setTimeout(r, 20));
+  return sock;
+}
+
+test("an already-closed socket leaves no subscriber and no ping timer", async () => {
+  const accepted: net.Socket[] = [];
+  const server = net.createServer((c) => {
+    accepted.push(c);
+    c.on("error", () => {});
+  });
+  await new Promise<void>((r) => server.listen(0, "127.0.0.1", () => r()));
+  const { port } = server.address() as AddressInfo;
+
+  try {
+    const timersBefore = armedTimers();
+    const subsBefore = globalTrafficBuffer.subscriberCount();
+
+    const handlers: Promise<unknown>[] = [];
+    for (let i = 0; i < DEAD_UPGRADES; i++) {
+      // Catch at creation time: the route answers a hijacked upgrade with a 101
+      // Response, which undici rejects off a real server. Left unattached, that
+      // rejection would sit through the next await and trip Node's unhandled
+      // rejection detection. Either settlement proves the handler released its
+      // resources instead of hanging, which is what this test measures.
+      handlers.push(GET(upgradeRequest(await deadSocket(port))).catch(() => undefined));
+    }
+
+    // Own the race timer so it can be cleared before measuring; otherwise the
+    // test's own armed timeout is counted as a leaked one.
+    let raceTimer: ReturnType<typeof setTimeout> | undefined;
+    const outcome = await Promise.race([
+      Promise.all(handlers).then(() => "settled"),
+      new Promise((r) => {
+        raceTimer = setTimeout(() => r("hung"), 2000);
+      }),
+    ]);
+    if (raceTimer) clearTimeout(raceTimer);
+    assert.equal(
+      outcome,
+      "settled",
+      "each handler must return instead of hanging forever on a dead socket"
+    );
+
+    const timersAfter = armedTimers();
+    assert.ok(
+      timersAfter <= timersBefore,
+      `${DEAD_UPGRADES} dead upgrades retained ${timersAfter - timersBefore} ping timer(s)`
+    );
+
+    // Measure the subscriber set directly; counting fan-out to our own probe
+    // says nothing about whether the dead sockets stayed subscribed.
+    assert.equal(
+      globalTrafficBuffer.subscriberCount(),
+      subsBefore,
+      `${DEAD_UPGRADES} dead upgrades left ${globalTrafficBuffer.subscriberCount() - subsBefore} subscriber(s) behind`
+    );
+  } finally {
+    // close() only fires once every accepted connection is gone.
+    for (const c of accepted) c.destroy();
+    await new Promise<void>((r) => server.close(() => r()));
+  }
+});
+
+test("a live socket keeps its subscription until the socket closes", async () => {
+  const accepted: net.Socket[] = [];
+  const server = net.createServer((c) => {
+    accepted.push(c);
+    c.on("error", () => {});
+  });
+  await new Promise<void>((r) => server.listen(0, "127.0.0.1", () => r()));
+  const { port } = server.address() as AddressInfo;
+
+  const sock = net.connect(port, "127.0.0.1");
+  await new Promise<void>((r) => sock.once("connect", () => r()));
+  sock.on("error", () => {});
+
+  try {
+    const subsBefore = globalTrafficBuffer.subscriberCount();
+
+    const handler = GET(upgradeRequest(sock)).catch(() => undefined);
+    await new Promise((r) => setTimeout(r, 100));
+
+    assert.equal(
+      globalTrafficBuffer.subscriberCount(),
+      subsBefore + 1,
+      "a live upgrade must register exactly one traffic subscriber"
+    );
+
+    // Closing the socket resolves the handler's `settled` promise, which is the
+    // only path that releases the subscriber.
+    sock.destroy();
+    await handler;
+
+    assert.equal(
+      globalTrafficBuffer.subscriberCount(),
+      subsBefore,
+      "closing the socket must release the subscriber"
+    );
+  } finally {
+    // close() only fires once every accepted connection is gone.
+    for (const c of accepted) c.destroy();
+    await new Promise<void>((r) => server.close(() => r()));
+  }
+});