From 2d305b0ef31e38dcc029458cede04cf93fc5c4e3 Mon Sep 17 00:00:00 2001
From: k-taro56 <121674121+k-taro56@users.noreply.github.com>
Date: Tue, 3 Nov 2026 04:01:56 +0900
Subject: [PATCH 1/2] fix(adapter): read the rest of a body its answer left
 unread, or close the connection

The native server handed the runtime `Readable.toWeb(req)`. A body the
application stopped reading left the request paused, or destroyed with its
connection detached, and in both cases Node.js stopped reading the connection.
That happened to a Server Action over its size limit, to a middleware's side of
a split body nobody read, and to a POST to a static page. The connection then
answered nothing more until its keep-alive timeout. A client that sent its next
request on it waited that long: ten minutes with `--keepAliveTimeout 600000`.
A slow upload stalled until `requestTimeout`.

The body is now read from the request directly. Cancelling it stops the
reading and leaves the request alone. Once the answer has gone out, the rest is
read and dropped within workerd's grace (64 KiB, 1 s). Past either, the
connection is ended and dropped 500 ms later. A read of the body after its
answer fails, as it does in workerd.
---
 packages/adapter/src/native/http.ts           |   6 +-
 packages/adapter/src/native/request-body.ts   | 117 ++++++++
 .../adapter/test/native-request-body.test.ts  | 255 ++++++++++++++++++
 3 files changed, 375 insertions(+), 3 deletions(-)
 create mode 100644 packages/adapter/src/native/request-body.ts
 create mode 100644 packages/adapter/test/native-request-body.test.ts

diff --git a/packages/adapter/src/native/http.ts b/packages/adapter/src/native/http.ts
index ece343e00..3f31ee215 100644
--- a/packages/adapter/src/native/http.ts
+++ b/packages/adapter/src/native/http.ts
@@ -6,6 +6,8 @@ import type { ReadableStream as NodeReadableStream } from 'node:stream/web';
 import { PLATFORM_REQUEST_HEADERS } from '@stayingupwind/core/paas';
 import { handedOnAsSent, PLATFORM_HEADER_PREFIX } from '@stayingupwind/core/request';
 
+import { nativeBody } from './request-body.ts';
+
 const HOP_HEADERS = new Set([
   'connection',
   'keep-alive',
@@ -87,9 +89,7 @@ export function nativeRequest(
     headers,
     signal: controller.signal,
     duplex: 'half',
-    ...(!(method === 'GET' || method === 'HEAD') && {
-      body: Readable.toWeb(req) as ReadableStream<Uint8Array>,
-    }),
+    ...(!(method === 'GET' || method === 'HEAD') && { body: nativeBody(req, res) }),
   };
   // Concatenation deliberately keeps a // pathname from becoming a new authority.
   return new Request(`${origin.origin}${target}`, init);
diff --git a/packages/adapter/src/native/request-body.ts b/packages/adapter/src/native/request-body.ts
new file mode 100644
index 000000000..9f1cfbcee
--- /dev/null
+++ b/packages/adapter/src/native/request-body.ts
@@ -0,0 +1,117 @@
+import type { IncomingMessage, ServerResponse } from 'node:http';
+import type { Socket } from 'node:net';
+import { finished } from 'node:stream';
+
+const KIB = 1024;
+/**
+ * What workerd's server allows a request whose answer left its body unread (KJ's
+ * `HttpServerSettings`, `canceledUploadGraceBytes` and `canceledUploadGracePeriod`): the client may
+ * finish sending it, up to this many more bytes within this long, and the connection then carries
+ * its next request. Past either, the connection is closed.
+ */
+const UNREAD_GRACE_KIB = 64;
+const UNREAD_GRACE_BYTES = UNREAD_GRACE_KIB * KIB;
+const UNREAD_GRACE_MS = 1000;
+/**
+ * How long a connection closed that way is kept once its end is sent. Bytes left unread make its
+ * close a reset, which can cost the client an answer it has not read yet (RFC 9112, section 9.6),
+ * so it is dropped this much later, or as soon as the client closes it.
+ */
+const CLOSING_LINGER_MS = 500;
+/** What a read of a body fails with once its answer has gone out, as it does in workerd. */
+const LATE_READ = "Can't read from request stream after response has been sent.";
+
+/** Read and drop what the answer left of a body, within the grace; past it, close the connection. */
+function dropUnread(req: IncomingMessage, socket: Socket): void {
+  let left = UNREAD_GRACE_BYTES;
+  let settled = false;
+  const close = (): void => {
+    if (settled) return;
+    settled = true;
+    // Read no further: the next bytes may already be another request, which this connection will
+    // not answer now. The request is paused too, as a read of it resumes its connection. Its end
+    // follows the answer out.
+    req.pause();
+    socket.pause();
+    socket.end();
+    const linger = setTimeout(() => {
+      socket.destroy();
+    }, CLOSING_LINGER_MS);
+    linger.unref();
+    socket.once('close', () => {
+      clearTimeout(linger);
+    });
+  };
+  const grace = setTimeout(close, UNREAD_GRACE_MS);
+  grace.unref();
+  const settle = (): void => {
+    settled = true;
+    clearTimeout(grace);
+  };
+  req.on('data', (chunk: Buffer) => {
+    left -= chunk.byteLength;
+    if (left < 0) close();
+  });
+  req.once('end', settle);
+  req.once('close', settle);
+  req.resume();
+}
+
+/**
+ * The request's body as the application reads it: from the connection, as fast as it is read.
+ *
+ * The connection is the answer's. Cancelling the body stops it being read, and nothing more:
+ * destroying the request, as `Readable.toWeb` does, has Node.js detach the connection from it and
+ * stop reading the connection for good. Once the answer has gone out, what the application left
+ * unread of the body is read and dropped, within workerd's grace, so that the connection carries
+ * its next request; past the grace, the connection is closed. Left as it was, it answered nothing
+ * more and stayed open until a timeout, and a client that sent its next request on it waited for
+ * the server's keep-alive timeout.
+ */
+export function nativeBody(req: IncomingMessage, res: ServerResponse): ReadableStream<Uint8Array> {
+  const { socket } = req;
+  let controller: ReadableStreamDefaultController<Uint8Array> | undefined;
+  let open = true;
+  const onData = (chunk: Buffer): void => {
+    // A copy: the parser's chunk may share its memory with the next one.
+    controller?.enqueue(new Uint8Array(chunk));
+    if ((controller?.desiredSize ?? 0) <= 0) req.pause();
+  };
+  const shut = (): void => {
+    open = false;
+    req.off('data', onData);
+  };
+  req.pause();
+  req.on('data', onData);
+  const cleanup = finished(req, (error) => {
+    cleanup();
+    // As `Readable.toWeb` does: a request that has failed once cannot fail its reader again.
+    req.on('error', () => {
+      // Already reported to the body's reader, or the body was let go of.
+    });
+    if (!open) return;
+    shut();
+    if (error === null || error === undefined) controller?.close();
+    else controller?.error(error);
+  });
+  res.once('finish', () => {
+    if (req.complete || req.destroyed) return;
+    if (open) {
+      shut();
+      controller?.error(new TypeError(LATE_READ));
+    }
+    dropUnread(req, socket);
+  });
+  return new ReadableStream<Uint8Array>(
+    {
+      start: (started) => {
+        controller = started;
+      },
+      pull: () => {
+        req.resume();
+      },
+      cancel: shut,
+    },
+    { highWaterMark: req.readableHighWaterMark, size: (chunk) => chunk.byteLength },
+  );
+}
diff --git a/packages/adapter/test/native-request-body.test.ts b/packages/adapter/test/native-request-body.test.ts
new file mode 100644
index 000000000..fc9d57df4
--- /dev/null
+++ b/packages/adapter/test/native-request-body.test.ts
@@ -0,0 +1,255 @@
+import { once } from 'node:events';
+import {
+  Agent,
+  type ClientRequest,
+  createServer,
+  request as httpRequest,
+  type Server,
+} from 'node:http';
+import type { Socket } from 'node:net';
+import { setTimeout as delay } from 'node:timers/promises';
+
+import { afterEach, describe, expect, it } from 'vitest';
+
+import { nativeRequest, writeNativeResponse } from '../src/native/http.ts';
+
+const KIB = 1024;
+
+interface Exchange {
+  readonly status: number;
+  readonly text: string;
+  readonly socket: Socket;
+}
+
+const servers: Server[] = [];
+const agents: Agent[] = [];
+
+/** A native boundary whose application answers what `answer` makes of the request. */
+async function boundary(
+  answer: (request: Request) => Promise<string> | string = (request) =>
+    `answered ${request.method}`,
+): Promise<string> {
+  const server = createServer((req, res) => {
+    const request = nativeRequest(req, res);
+    const respond = async (): Promise<void> => {
+      await writeNativeResponse(req, res, new Response(await answer(request)));
+    };
+    void respond().catch((error: unknown) => {
+      res.destroy(error instanceof Error ? error : undefined);
+    });
+  });
+  servers.push(server);
+  server.listen(0, '127.0.0.1');
+  await once(server, 'listening');
+  const address = server.address();
+  if (address === null || typeof address === 'string') throw new Error('no listener');
+  return `http://127.0.0.1:${String(address.port)}`;
+}
+
+/** One connection at most, kept between requests. */
+function keptAgent(): Agent {
+  const agent = new Agent({ keepAlive: true, maxSockets: 1 });
+  agents.push(agent);
+  return agent;
+}
+
+const PIECE = 8 * KIB;
+const PIECE_GAP_MS = 10;
+
+/**
+ * Write `bytes` in pieces a little apart, then end the request: each comes in a read of its own, so
+ * a server that stops reading once the request's buffer fills leaves the later ones on the socket.
+ */
+function sendInPieces(client: ClientRequest, bytes: number): void {
+  const piece = Math.min(PIECE, bytes);
+  if (bytes === piece) {
+    client.end(Buffer.alloc(piece));
+    return;
+  }
+  client.write(Buffer.alloc(piece));
+  setTimeout(() => {
+    sendInPieces(client, bytes - piece);
+  }, PIECE_GAP_MS);
+}
+
+/**
+ * Send `sent` bytes of a body declared `declared` bytes long, take the answer, then send `after`
+ * more bytes, if any, and end the request.
+ */
+async function exchange(
+  url: string,
+  agent: Agent,
+  body?: { declared: number; sent: number; after?: number },
+): Promise<Exchange> {
+  return new Promise<Exchange>((resolve, reject) => {
+    const client = httpRequest(url, {
+      agent,
+      method: body === undefined ? 'GET' : 'POST',
+      ...(body !== undefined && { headers: { 'content-length': String(body.declared) } }),
+    });
+    // A connection the server closes may fail the request after its answer has come: past then,
+    // that is the outcome under test, not a failure of the exchange.
+    client.on('error', reject);
+    let connection: Socket | undefined;
+    client.once('socket', (socket) => {
+      connection = socket;
+      socket.on('error', () => {
+        // Reported to the request above.
+      });
+    });
+    client.once('response', (response) => {
+      const chunks: Buffer[] = [];
+      response.on('data', (chunk: Buffer) => {
+        chunks.push(chunk);
+      });
+      response.once('end', () => {
+        if (connection === undefined) {
+          reject(new Error('answered before a connection was assigned'));
+          return;
+        }
+        if (body?.after !== undefined) sendInPieces(client, body.after);
+        resolve({
+          status: response.statusCode ?? 0,
+          text: Buffer.concat(chunks).toString(),
+          socket: connection,
+        });
+      });
+    });
+    if (body === undefined) client.end();
+    else if (body.after === undefined && body.sent === body.declared)
+      client.end(Buffer.alloc(body.sent));
+    else client.write(Buffer.alloc(body.sent));
+  });
+}
+
+/**
+ * Resolves once the connection is closed, at once if it already is. Not `once(socket, 'close')`,
+ * which rejects on the `EPIPE` of a body still being sent as the server closes the connection.
+ */
+async function closedOf(socket: Socket): Promise<void> {
+  if (socket.closed) return;
+  await new Promise<void>((resolve) => {
+    socket.once('close', () => {
+      resolve();
+    });
+  });
+}
+
+describe('a request body its answer leaves unread', () => {
+  afterEach(async () => {
+    for (const agent of agents.splice(0)) agent.destroy();
+    for (const server of servers.splice(0)) {
+      server.closeAllConnections();
+      await new Promise<void>((resolve) => {
+        server.close(() => {
+          resolve();
+        });
+      });
+    }
+  });
+
+  it('keeps the connection of a body read to its end', async () => {
+    const url = await boundary(async (request) => `read ${String((await request.text()).length)}`);
+    const agent = keptAgent();
+
+    const first = await exchange(url, agent, { declared: 256 * KIB, sent: 256 * KIB });
+    const next = await exchange(url, agent);
+
+    expect(first).toMatchObject({ status: 200, text: `read ${String(256 * KIB)}` });
+    expect(next).toMatchObject({ status: 200 });
+    expect(next.socket).toBe(first.socket);
+  });
+
+  it('reads the rest within the grace, and the connection carries the next request', async () => {
+    const url = await boundary();
+    const agent = keptAgent();
+
+    // The answer goes out with 48 KiB of the body still to come: the client sends it after.
+    const first = await exchange(url, agent, {
+      declared: 64 * KIB,
+      sent: 16 * KIB,
+      after: 48 * KIB,
+    });
+    // Asked while the body is still being sent, it waits for the connection, which the agent hands
+    // it without counting it reused (`reusedSocket` is set only for an idle one), so it is compared.
+    const next = await exchange(url, agent);
+
+    expect(first).toMatchObject({ status: 200, text: 'answered POST' });
+    expect(next).toMatchObject({ status: 200, text: 'answered GET' });
+    expect(next.socket).toBe(first.socket);
+  });
+
+  it('closes the connection after its answer when more than the grace is left', async () => {
+    const url = await boundary();
+    const agent = keptAgent();
+    const started = performance.now();
+
+    const first = await exchange(url, agent, { declared: 4096 * KIB, sent: 4096 * KIB });
+
+    expect(first).toMatchObject({ status: 200, text: 'answered POST' });
+
+    await closedOf(first.socket);
+
+    // Not after a timeout: closed once its answer is out, and a moment for the client to read it.
+    expect(performance.now() - started).toBeLessThan(3000);
+
+    const next = await exchange(url, agent);
+
+    expect(next).toMatchObject({ status: 200, text: 'answered GET' });
+    expect(next.socket).not.toBe(first.socket);
+  });
+
+  it('closes the connection when the rest does not come within the grace', async () => {
+    const url = await boundary();
+    const started = performance.now();
+
+    // The client sends 8 KiB of 64 KiB, and no more.
+    const stalled = await exchange(url, keptAgent(), { declared: 64 * KIB, sent: 8 * KIB });
+
+    expect(stalled).toMatchObject({ status: 200, text: 'answered POST' });
+
+    await closedOf(stalled.socket);
+
+    expect(performance.now() - started).toBeLessThan(5000);
+  });
+
+  it('answers on the connection when the application cancels the body', async () => {
+    const url = await boundary(async (request) => {
+      await request.body?.cancel('unused');
+      return 'answered without the body';
+    });
+    const agent = keptAgent();
+    const started = performance.now();
+
+    await expect(
+      exchange(url, agent, { declared: 4096 * KIB, sent: 4096 * KIB }),
+    ).resolves.toMatchObject({ status: 200, text: 'answered without the body' });
+    await expect(exchange(url, agent)).resolves.toMatchObject({
+      status: 200,
+      text: 'answered without the body',
+    });
+    // Not after a timeout: the unread body's connection is closed once its answer is out.
+    expect(performance.now() - started).toBeLessThan(3000);
+  });
+
+  it('fails a read of the body begun after its answer has gone out', async () => {
+    const late = Promise.withResolvers<unknown>();
+    const url = await boundary((request) => {
+      const readLate = async (): Promise<void> => {
+        await delay(50);
+        late.resolve(await request.text());
+      };
+      void readLate().catch((error: unknown) => {
+        late.resolve(error);
+      });
+      return 'answered first';
+    });
+
+    await expect(
+      exchange(url, keptAgent(), { declared: 64 * KIB, sent: 8 * KIB, after: 56 * KIB }),
+    ).resolves.toMatchObject({ status: 200, text: 'answered first' });
+    await expect(late.promise).resolves.toStrictEqual(
+      new TypeError("Can't read from request stream after response has been sent."),
+    );
+  });
+});

From 036044db923d132928bd0d0ed2b78373911320a6 Mon Sep 17 00:00:00 2001
From: k-taro56 <121674121+k-taro56@users.noreply.github.com>
Date: Tue, 3 Nov 2026 04:25:35 +0900
Subject: [PATCH 2/2] fix(adapter): fail a late read of any unread body, and
 pause a cancelled one

- The body stream queued up to 64 KiB ahead of the application. A small
  body could therefore be read whole into its queue and closed, and a read
  after the answer still succeeded. The stream now queues nothing ahead of
  the application's reads.
- At the answer's end, a body left open fails its readers whatever was
  received. A body received whole is only dropped, because the connection
  already reads its next request.
- `req.read(0)` marks the body as begun. Otherwise Node.js reads a body
  nothing has read to its end before the next request, with no bound.
- Cancelling now pauses the request. Without a 'data' listener, a flowing
  request kept reading its connection, past the grace.
- The test of a stalled body bounds its close at 3 s.
---
 packages/adapter/src/native/request-body.ts   | 26 +++++++---
 .../adapter/test/native-request-body.test.ts  | 49 +++++++++++++------
 2 files changed, 53 insertions(+), 22 deletions(-)

diff --git a/packages/adapter/src/native/request-body.ts b/packages/adapter/src/native/request-body.ts
index 9f1cfbcee..09fb5386f 100644
--- a/packages/adapter/src/native/request-body.ts
+++ b/packages/adapter/src/native/request-body.ts
@@ -62,11 +62,11 @@ function dropUnread(req: IncomingMessage, socket: Socket): void {
  *
  * The connection is the answer's. Cancelling the body stops it being read, and nothing more:
  * destroying the request, as `Readable.toWeb` does, has Node.js detach the connection from it and
- * stop reading the connection for good. Once the answer has gone out, what the application left
- * unread of the body is read and dropped, within workerd's grace, so that the connection carries
- * its next request; past the grace, the connection is closed. Left as it was, it answered nothing
- * more and stayed open until a timeout, and a client that sent its next request on it waited for
- * the server's keep-alive timeout.
+ * stop reading the connection for good. Once the answer has gone out, the body can no longer be
+ * read, as in workerd, and what the application left of it is read and dropped within workerd's
+ * grace, so that the connection carries its next request; past the grace, the connection is
+ * closed. Left as it was, it answered nothing more and stayed open until a timeout, and a client
+ * that sent its next request on it waited for the server's keep-alive timeout.
  */
 export function nativeBody(req: IncomingMessage, res: ServerResponse): ReadableStream<Uint8Array> {
   const { socket } = req;
@@ -79,10 +79,15 @@ export function nativeBody(req: IncomingMessage, res: ServerResponse): ReadableS
   };
   const shut = (): void => {
     open = false;
+    // Paused, not only unheard: a flowing request with no listener still reads its connection.
+    req.pause();
     req.off('data', onData);
   };
   req.pause();
   req.on('data', onData);
+  // Begun, so that the answer's end leaves what is unread to the grace below: Node.js reads a body
+  // no read began to its end, however long, before the connection takes its next request.
+  req.read(0);
   const cleanup = finished(req, (error) => {
     cleanup();
     // As `Readable.toWeb` does: a request that has failed once cannot fail its reader again.
@@ -95,11 +100,16 @@ export function nativeBody(req: IncomingMessage, res: ServerResponse): ReadableS
     else controller?.error(error);
   });
   res.once('finish', () => {
-    if (req.complete || req.destroyed) return;
     if (open) {
       shut();
       controller?.error(new TypeError(LATE_READ));
     }
+    if (req.readableEnded || req.destroyed) return;
+    // Received whole, the connection already reads its next request: what is left is only dropped.
+    if (req.complete) {
+      req.resume();
+      return;
+    }
     dropUnread(req, socket);
   });
   return new ReadableStream<Uint8Array>(
@@ -112,6 +122,8 @@ export function nativeBody(req: IncomingMessage, res: ServerResponse): ReadableS
       },
       cancel: shut,
     },
-    { highWaterMark: req.readableHighWaterMark, size: (chunk) => chunk.byteLength },
+    // Nothing is queued ahead of the application's reads, so none of what it has not asked for by
+    // its answer's end can still be read after it.
+    { highWaterMark: 0 },
   );
 }
diff --git a/packages/adapter/test/native-request-body.test.ts b/packages/adapter/test/native-request-body.test.ts
index fc9d57df4..929278098 100644
--- a/packages/adapter/test/native-request-body.test.ts
+++ b/packages/adapter/test/native-request-body.test.ts
@@ -14,6 +14,7 @@ import { afterEach, describe, expect, it } from 'vitest';
 import { nativeRequest, writeNativeResponse } from '../src/native/http.ts';
 
 const KIB = 1024;
+const LATE_READ = "Can't read from request stream after response has been sent.";
 
 interface Exchange {
   readonly status: number;
@@ -122,6 +123,23 @@ async function exchange(
   });
 }
 
+/** A boundary whose application answers first, and reads its body 50 ms later. */
+async function readingLate(): Promise<{ url: string; late: Promise<unknown> }> {
+  const late = Promise.withResolvers<unknown>();
+  const url = await boundary((request) => {
+    if (request.body === null) return 'answered GET';
+    const readLate = async (): Promise<void> => {
+      await delay(50);
+      late.resolve(await request.text());
+    };
+    void readLate().catch((error: unknown) => {
+      late.resolve(error);
+    });
+    return 'answered first';
+  });
+  return { url, late: late.promise };
+}
+
 /**
  * Resolves once the connection is closed, at once if it already is. Not `once(socket, 'close')`,
  * which rejects on the `EPIPE` of a body still being sent as the server closes the connection.
@@ -210,7 +228,7 @@ describe('a request body its answer leaves unread', () => {
 
     await closedOf(stalled.socket);
 
-    expect(performance.now() - started).toBeLessThan(5000);
+    expect(performance.now() - started).toBeLessThan(3000);
   });
 
   it('answers on the connection when the application cancels the body', async () => {
@@ -233,23 +251,24 @@ describe('a request body its answer leaves unread', () => {
   });
 
   it('fails a read of the body begun after its answer has gone out', async () => {
-    const late = Promise.withResolvers<unknown>();
-    const url = await boundary((request) => {
-      const readLate = async (): Promise<void> => {
-        await delay(50);
-        late.resolve(await request.text());
-      };
-      void readLate().catch((error: unknown) => {
-        late.resolve(error);
-      });
-      return 'answered first';
-    });
+    const { url, late } = await readingLate();
 
     await expect(
       exchange(url, keptAgent(), { declared: 64 * KIB, sent: 8 * KIB, after: 56 * KIB }),
     ).resolves.toMatchObject({ status: 200, text: 'answered first' });
-    await expect(late.promise).resolves.toStrictEqual(
-      new TypeError("Can't read from request stream after response has been sent."),
-    );
+    await expect(late).resolves.toStrictEqual(new TypeError(LATE_READ));
+  });
+
+  it('fails a late read of a body received whole, and keeps its connection', async () => {
+    const { url, late } = await readingLate();
+    const agent = keptAgent();
+
+    const first = await exchange(url, agent, { declared: 2 * KIB, sent: 2 * KIB });
+    const next = await exchange(url, agent);
+
+    expect(first).toMatchObject({ status: 200, text: 'answered first' });
+    await expect(late).resolves.toStrictEqual(new TypeError(LATE_READ));
+    expect(next).toMatchObject({ status: 200, text: 'answered GET' });
+    expect(next.socket).toBe(first.socket);
   });
 });