fix(runtime): drop a bridge connection whose server left its request unread

packages/runtime/src/native/bridge.ts

@@ -8,6 +8,7 @@ import {
   type Server,
   type ServerResponse,
 } from 'node:http';
+import type { Socket } from 'node:net';
 import { Readable } from 'node:stream';
 import { pipeline } from 'node:stream/promises';
 import type { ReadableStream as NodeReadableStream } from 'node:stream/web';
@@ -80,6 +81,8 @@ interface Invocation {
   accepted: boolean;
   handlerFinished: boolean;
   transportFinished: boolean;
+  /** The request as the server read it, and the server's end of its connection. */
+  served?: { readonly request: IncomingMessage; readonly socket: Socket } | undefined;
 }
 
 function boundedInteger(value: number, name: string): number {
@@ -209,6 +212,7 @@ class NativeBridge implements NodeBridge {
       });
       this.#clients.add(client);
       let upload: Readable | undefined;
+      let connection: Socket | undefined;
       let answered = false;
       const timer = setTimeout(() => {
         client.destroy(
@@ -225,6 +229,22 @@ class NativeBridge implements NodeBridge {
       client.on('error', (error) => {
         if (!answered) reject(error);
       });
+      client.once('socket', (socket) => {
+        connection = socket;
+      });
+      /**
+       * A server may also stop reading a request before its end, as Next.js does with a Server
+       * Action over its size limit, even where all of it was sent. The rest stays on the connection,
+       * which reads nothing more: a call the Agent sent on it from its pool waited for its headers
+       * deadline, and the server's end, its idle connections being kept, was never closed. So both
+       * ends are dropped, once the answer has been read whole or the call has ended without it.
+       */
+      const dropUnread = (): void => {
+        const { served } = invocation;
+        if (served === undefined || served.request.complete) return;
+        connection?.destroy();
+        served.socket.destroy();
+      };
       client.once('close', () => {
         if (!answered) reject(new Error('native transport closed before response headers'));
         clearTimeout(timer);
@@ -233,6 +253,7 @@ class NativeBridge implements NodeBridge {
         // A server may finish its answer before consuming an upload. A pooled socket must never
         // retain that producer, nor be handed to the next call with unfinished request framing.
         if (upload !== undefined && !upload.readableEnded) upload.destroy();
+        dropUnread();
         invocation.transportFinished = true;
         this.#release(id, invocation);
       });
@@ -251,6 +272,8 @@ class NativeBridge implements NodeBridge {
             upload.destroy();
             client.destroy();
           }
+          // Before the Agent pools the connection, which it does on the next tick.
+          dropUnread();
         });
         clearTimeout(timer);
         try {
@@ -318,6 +341,7 @@ class NativeBridge implements NodeBridge {
     // would otherwise wait for this one until the drain's deadline. Ended, not destroyed, so an
     // answer still in the socket's buffer reaches the caller before the connection closes.
     const { socket } = req;
+    invocation.served = { request: req, socket };
     let answered = false;
     let received = false;
     const endIfIdle = (): void => {

packages/runtime/test/native/bridge.test.ts

@@ -3,6 +3,8 @@ import assert from 'node:assert/strict';
 import { AsyncLocalStorage } from 'node:async_hooks';
 import { once } from 'node:events';
 import { type IncomingMessage, request as httpRequest, type ServerResponse } from 'node:http';
+import { Transform, Writable } from 'node:stream';
+import { pipeline } from 'node:stream/promises';
 import { setTimeout as delay } from 'node:timers/promises';
 import { gzipSync } from 'node:zlib';
 
@@ -84,6 +86,37 @@ async function answerUpload(req: IncomingMessage, res: ServerResponse): Promise<
   res.end(await textOf(req));
 }
 
+/**
+ * A handler that reads its body as Next.js reads a Server Action's: through a size limit. Past the
+ * limit, the pipeline's failure destroys the request, and Node.js reads its connection no further.
+ */
+async function overLimit(req: IncomingMessage, res: ServerResponse): Promise<void> {
+  let size = 0;
+  const limit = new Transform({
+    transform(chunk: Buffer, _encoding, callback) {
+      size += chunk.byteLength;
+      callback(size > 64 * 1024 ? new Error('Body exceeded the limit') : null, chunk);
+    },
+  });
+  const sink = new Writable({
+    write(_chunk, _encoding, callback) {
+      callback();
+    },
+  });
+  try {
+    await pipeline(req, limit, sink);
+    res.end('read');
+  } catch {
+    res.statusCode = 413;
+    res.end('too large');
+  }
+}
+
+/** A body the loopback's buffers hold whole, so the bridge sends all of it. */
+function megabyteUpload(): Request {
+  return upload(new Blob([new Uint8Array(1024 * 1024)]).stream());
+}
+
 describe('native Node bridge', () => {
   it('restores rewritten URL and public headers on real Node request/response objects', async () => {
     const bridge = fixture();
@@ -207,6 +240,52 @@ describe('native Node bridge', () => {
     assert.ok(performance.now() - closingAt < 2000);
   });
 
+  it('does not pool a connection whose server left its request unread', async () => {
+    const bridge = fixture({ headersTimeoutMs: 3000 });
+    const sockets: unknown[] = [];
+    const first = await bridge.invokeNodeHandler(
+      input(
+        async (req, res) => {
+          sockets.push(req.socket);
+          await overLimit(req, res);
+        },
+        { request: megabyteUpload() },
+      ),
+    );
+    assert.equal(first.status, 413);
+    assert.equal(await first.text(), 'too large');
+
+    const second = await bridge.invokeNodeHandler(
+      input(async (req, res) => {
+        sockets.push(req.socket);
+        res.end('second');
+      }),
+    );
+
+    assert.equal(await second.text(), 'second');
+    assert.equal(sockets.length, 2);
+    assert.notEqual(sockets[1], sockets[0]);
+  });
+
+  it('closes the server end of a connection whose request was left unread', async () => {
+    const bridge = fixture();
+    const served = Promise.withResolvers<NodeJS.EventEmitter>();
+    const response = await bridge.invokeNodeHandler(
+      input(
+        async (req, res) => {
+          served.resolve(req.socket);
+          await overLimit(req, res);
+        },
+        { request: megabyteUpload() },
+      ),
+    );
+    const socket = await served.promise;
+    const closed = once(socket, 'close');
+
+    assert.equal(await response.text(), 'too large');
+    await closed;
+  });
+
   it('preserves an explicit Host instead of replacing it with the loopback host', async () => {
     const bridge = fixture();
     const response = await bridge.invokeNodeHandler(