Repository: denoland/deno
Pull request: fix(ext/node): preserve streaming body on stale keepalive retry

Streaming writes (e.g. pipeline, node-fetch) bypass outputData and go directly to _bodyWriter, so they were lost when a stale keepalive connection triggered a retry. This caused timeouts for libraries like node-fetch and gaxios that use pipeline() to send request bodies.

Capture body data written via _writeRaw() on reused sockets so it can be restored to outputData on retry, and re-call end() to set up body writer close logic for the new attempt.


Related: https://github.com/denoland/deno/issues/32006#issuecomment-3916719816

Changed files:
filename: ext/node/polyfills/_http_outgoing.ts
patch: @@ -530,7 +530,13 @@ Object.defineProperties(
       }
 
       const { data, encoding, callback } = this.outputData.shift();
-      const ret = this._writeRaw(data, encoding, callback);
+      this._flushingBuffer = true;
+      let ret;
+      try {
+        ret = this._writeRaw(data, encoding, callback);
+      } finally {
+        this._flushingBuffer = false;
+      }
       if (this.outputData.length > 0) {
         this.once("drain", this._flushBuffer);
       }
@@ -551,6 +557,21 @@ Object.defineProperties(
         data = new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
       }
       if (data.byteLength > 0) {
+        // Save body data for potential retry on stale keepAlive connections.
+        // Streaming writes (e.g. pipeline) bypass outputData and go directly
+        // to _bodyWriter, so _outputDataForRetry can't capture them.
+        // Skip when _flushingBuffer is true to avoid double-counting data
+        // already captured in _outputDataForRetry.
+        if (this.reusedSocket && !this._flushingBuffer) {
+          if (!this._bodyDataForRetry) {
+            this._bodyDataForRetry = [];
+          }
+          this._bodyDataForRetry.push({
+            data: new Uint8Array(data),
+            encoding: null,
+            callback: null,
+          });
+        }
         this._bodyWriter.ready.then(() => {
           if (this._bodyWriter.desiredSize > 0) {
             this._bodyWriter.write(data).then(() => {

filename: ext/node/polyfills/http.ts
patch: @@ -756,6 +756,21 @@ class ClientRequest extends OutgoingMessage {
               this.outputData = this._outputDataForRetry;
               this._outputDataForRetry = null;
             }
+            // Restore body data from streaming writes (e.g. pipeline)
+            // that bypassed outputData and went directly to _bodyWriter.
+            if (this._bodyDataForRetry) {
+              for (const entry of this._bodyDataForRetry) {
+                this.outputData.push(entry);
+              }
+              this._bodyDataForRetry = null;
+            }
+            // If the request body was already fully sent (e.g. pipeline
+            // called end()), re-call end() to set up body writer close
+            // logic for the retry attempt.
+            if (this.finished) {
+              this.finished = false;
+              this.end();
+            }
             this.agent.addRequest(this, this._agentOptions);
             return;
           }

filename: tests/specs/node/http_agent_keepalive_stale_streaming_post/__test__.jsonc
patch: @@ -0,0 +1,4 @@
+{
+  "args": "run -A main.ts",
+  "output": "main.out"
+}

filename: tests/specs/node/http_agent_keepalive_stale_streaming_post/main.out
patch: @@ -0,0 +1,3 @@
+Request 1: OK {"received":"{\"test\":true}"}
+Request 2: OK {"received":"{\"test\":true}"}
+PASS

filename: tests/specs/node/http_agent_keepalive_stale_streaming_post/main.ts
patch: @@ -0,0 +1,121 @@
+// Test that streaming POST requests (using pipeline) on stale keepAlive
+// connections are transparently retried with the request body intact.
+// Streaming writes bypass outputData and go directly to _bodyWriter,
+// so they need separate handling in the retry logic.
+import http from "node:http";
+import net from "node:net";
+import { Readable } from "node:stream";
+import { pipeline } from "node:stream/promises";
+
+let idleSocketClosed: Promise<void>;
+let resolveIdleSocketClosed: () => void;
+idleSocketClosed = new Promise((r) => (resolveIdleSocketClosed = r));
+
+const BODY = '{"test":true}';
+
+const server = net.createServer((socket) => {
+  let idleTimer: ReturnType<typeof setTimeout> | null = null;
+
+  const resetIdle = () => {
+    if (idleTimer) clearTimeout(idleTimer);
+    idleTimer = setTimeout(() => socket.destroy(), 50);
+  };
+
+  let buffer = "";
+  socket.on("data", (data) => {
+    buffer += data.toString();
+    const headerEnd = buffer.indexOf("\r\n\r\n");
+    if (headerEnd === -1) return;
+
+    const headers = buffer.substring(0, headerEnd);
+    const clMatch = headers.match(/content-length:\s*(\d+)/i);
+    const contentLength = clMatch ? parseInt(clMatch[1]) : 0;
+    const bodyStart = headerEnd + 4;
+    const body = buffer.substring(bodyStart, bodyStart + contentLength);
+
+    if (body.length < contentLength) return;
+
+    const responseBody = JSON.stringify({ received: body });
+    const response = [
+      "HTTP/1.1 200 OK",
+      `Content-Length: ${responseBody.length}`,
+      "Connection: keep-alive",
+      "",
+      responseBody,
+    ].join("\r\n");
+    socket.write(response);
+    buffer = "";
+    resetIdle();
+  });
+
+  socket.on("close", () => {
+    if (idleTimer) clearTimeout(idleTimer);
+    resolveIdleSocketClosed();
+  });
+});
+
+const agent = new http.Agent({
+  keepAlive: true,
+  maxSockets: 1,
+});
+
+function postStreaming(port: number): Promise<string> {
+  return new Promise((resolve, reject) => {
+    const timeout = setTimeout(() => reject(new Error("timeout")), 5000);
+    const req = http.request(
+      {
+        hostname: "127.0.0.1",
+        port,
+        method: "POST",
+        agent,
+        headers: { "Content-Length": String(BODY.length) },
+      },
+      (res) => {
+        let data = "";
+        res.on("data", (c: Buffer) => (data += c));
+        res.on("end", () => {
+          clearTimeout(timeout);
+          resolve(data);
+        });
+      },
+    );
+    req.on("error", (e: Error) => {
+      clearTimeout(timeout);
+      reject(e);
+    });
+
+    const bodyStream = Readable.from(Buffer.from(BODY));
+    pipeline(bodyStream, req).catch((e: Error) => {
+      clearTimeout(timeout);
+      reject(e);
+    });
+  });
+}
+
+server.listen(0, async () => {
+  const port = (server.address() as net.AddressInfo).port;
+  try {
+    // First request - establishes keepAlive connection
+    const first = await postStreaming(port);
+    console.log("Request 1: OK", first);
+
+    // Wait until the server closes the idle socket
+    await idleSocketClosed;
+    await new Promise((r) => setTimeout(r, 50));
+
+    // Reset the idle close promise for the next connection
+    idleSocketClosed = new Promise((r) => (resolveIdleSocketClosed = r));
+
+    // Second request on stale socket - should retry with body intact
+    const second = await postStreaming(port);
+    console.log("Request 2: OK", second);
+
+    console.log("PASS");
+  } catch (e) {
+    console.error("FAIL:", (e as Error).message);
+    Deno.exit(1);
+  } finally {
+    server.close();
+    agent.destroy();
+  }
+});