node:http: emit 'pause' on req.socket once an unread body fills the IncomingMessage buffer

src/js/node/_http_incoming.ts

@@ -424,9 +424,15 @@ function onDataIncomingMessage(
   // Incoming request-body bytes are socket activity: push the connection's
   // inactivity timeout (socket.setTimeout / server.timeout) further out, like
   // Node.js does for reads on the socket.
-  this.socket?._unrefTimer?.();
+  const socket = this.socket;
+  socket?._unrefTimer?.();
 
-  if (chunk && !this._dumped) this.push(chunk);
+  if (chunk && !this._dumped) {
+    // Like Node's parserOnBody: pause the connection once the buffer fills.
+    // Skipped for upgrade (the listener owns socket flow) and once writable
+    // has ended (socketHandle.end() resumed the poll so kqueue sees EV_EOF).
+    if (!this.push(chunk) && !this.upgrade && !socket?.writableEnded) readStop(socket);
+  }
 
   if (isLast) {
     emitEOFIncomingMessage(this);
@@ -800,4 +806,4 @@ function onError(self, error, cb) {
   }
 }
 
-export { IncomingMessage, kReqShouldKeepAlive, readStart, readStop };
+export { IncomingMessage, kReqShouldKeepAlive, onDataIncomingMessage, readStart, readStop };

src/js/node/_http_server.ts

@@ -71,7 +71,7 @@ const NumberIsNaN = Number.isNaN;
 
 const { format } = require("internal/util/inspect");
 
-const { IncomingMessage, kReqShouldKeepAlive } = require("node:_http_incoming");
+const { IncomingMessage, kReqShouldKeepAlive, onDataIncomingMessage } = require("node:_http_incoming");
 const {
   OutgoingMessage,
   kErrored,
@@ -865,9 +865,12 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
 
         setIsNextIncomingMessageHTTPS(prevIsNextIncomingMessageHTTPS);
         handle.onabort = socket[kBoundOnAbort] ??= onServerRequestEvent.bind(socket);
-        // start buffering data if any, the user will need to resume() or .on("data") to read it
+        // Like Node's connectionListener -> parserOnBody: body bytes flow into
+        // the IncomingMessage as they arrive, and the push callback readStop()s
+        // the socket (which emits 'pause' on it) once the buffer fills.
         if (hasBody) {
-          handle.pause();
+          handle.ondata = onDataIncomingMessage.bind(http_req);
+          handle.hasCustomOnData = false;
         }
         drainMicrotasks();
 

src/runtime/server/NodeHTTPResponse.rs

@@ -1354,8 +1354,13 @@
         {
             return Ok(JSValue::FALSE);
         }
-        self.update_flags(|f| f.insert(Flags::IS_DATA_BUFFERED_DURING_PAUSE));
-        raw.on_data(on_buffer_paused_shim, self.as_ctx_ptr());
+        // Body already delivered: nothing to buffer, and re-arming onData would
+        // overwrite a pipelined request's userData on the shared HttpResponseData.
+        // pause_socket() still runs so pausePipelineReads can gate the fd.
+        if self.body_read_state.get() == BodyReadState::Pending {
+            self.update_flags(|f| f.insert(Flags::IS_DATA_BUFFERED_DURING_PAUSE));
+            raw.on_data(on_buffer_paused_shim, self.as_ctx_ptr());
+        }
 
         // TODO: figure out why windows is not emitting EOF with UV_DISCONNECT
         #[cfg(not(windows))]
@@ -1416,6 +1421,9 @@
             || flags.contains(Flags::SOCKET_CLOSED)
             || flags.contains(Flags::ENDED)
             || flags.contains(Flags::UPGRADED)
+            // Body already delivered: re-arming onData/onTimeout would overwrite
+            // a pipelined request's userData on the shared HttpResponseData.
+            || self.body_read_state.get() != BodyReadState::Pending
         {
             return JSValue::FALSE;
         }

test/js/node/http/node-http-req-socket-pause.test.ts

@@ -0,0 +1,75 @@
+/**
+ * All tests in this file should also run in Node.js.
+ */
+import { expect, it } from "bun:test";
+import { once } from "node:events";
+import { createServer, request } from "node:http";
+import type { AddressInfo } from "node:net";
+
+it("req.socket emits 'pause' once an unread request body fills the IncomingMessage buffer", async () => {
+  // Node's test-http-no-read-no-dump: a handler that never reads the body sees
+  // 'pause' on req.connection once the IncomingMessage push() backpressures.
+  const { promise: paused, resolve: onPause, reject } = Promise.withResolvers<number>();
+  const server = createServer((req, res) => {
+    req.connection!.on("pause", () => {
+      onPause((req as any).readableLength);
+      res.end("ok");
+    });
+    res.writeHead(200);
+    res.flushHeaders();
+  });
+  try {
+    await once(server.listen(0), "listening");
+    const port = (server.address() as AddressInfo).port;
+    const post = request({ method: "POST", port });
+    post.on("error", reject);
+    post.flushHeaders();
+    // One body chunk at the default highWaterMark: the first parserOnBody push
+    // returns false and Node's readStop pauses the socket.
+    post.write(Buffer.alloc(64 * 1024, "X"));
+    const buffered = await paused;
+    expect(buffered).toBeGreaterThan(0);
+    post.destroy();
+  } finally {
+    server.closeAllConnections();
+    server.close();
+  }
+});
+
+it("body reading from 'pause' still delivers every byte and 'end'", async () => {
+  // A handler that keys its slow-reader flow off the socket's 'pause' event
+  // (the pattern from Node's test-http-no-read-no-dump) must still be able to
+  // drain the full body once it attaches a 'data' listener.
+  const { promise: done, resolve, reject } = Promise.withResolvers<{ pauses: number; received: number }>();
+  const server = createServer((req, res) => {
+    let pauses = 0;
+    let received = 0;
+    req.connection!.on("pause", () => {
+      pauses++;
+      if (pauses > 1) return;
+      req.on("data", chunk => (received += chunk.length));
+      req.on("end", () => {
+        res.end("ok");
+        resolve({ pauses, received });
+      });
+    });
+    res.writeHead(200);
+    res.flushHeaders();
+  });
+  try {
+    await once(server.listen(0), "listening");
+    const port = (server.address() as AddressInfo).port;
+    const payload = 256 * 1024;
+    const post = request({ method: "POST", port });
+    post.on("error", reject);
+    post.flushHeaders();
+    await once(post, "response");
+    post.end(Buffer.alloc(payload, "X"));
+    const { pauses, received } = await done;
+    expect(received).toBe(payload);
+    expect(pauses).toBeGreaterThanOrEqual(1);
+  } finally {
+    server.closeAllConnections();
+    server.close();
+  }
+});

test/js/node/test/parallel/test-http-no-read-no-dump.js

@@ -0,0 +1,56 @@
+'use strict';
+const common = require('../common');
+const http = require('http');
+
+let onPause = null;
+
+const server = http.createServer(common.mustCallAtLeast((req, res) => {
+  if (req.method === 'GET')
+    return res.end();
+
+  res.writeHead(200);
+  res.flushHeaders();
+
+  req.on('close', common.mustCall(() => {
+    req.on('end', common.mustNotCall());
+  }));
+
+  req.connection.on('pause', () => {
+    res.end();
+    onPause();
+  });
+})).listen(0, common.mustCall(() => {
+  const agent = new http.Agent({
+    maxSockets: 1,
+    keepAlive: true
+  });
+
+  const port = server.address().port;
+
+  const post = http.request({
+    agent,
+    method: 'POST',
+    port,
+  }, common.mustCall((res) => {
+    res.resume();
+
+    post.write(Buffer.alloc(64 * 1024).fill('X'));
+    onPause = () => {
+      post.end('something');
+    };
+  }));
+
+  // What happens here is that the server `end`s the response before we send
+  // `something`, and the client thought that this is a green light for sending
+  // next GET request
+  post.write('initial');
+
+  http.request({
+    agent,
+    method: 'GET',
+    port,
+  }, common.mustCall((res) => {
+    server.close();
+    res.connection.end();
+  })).end();
+}));