http: fix http client leaky with double response

lib/_http_client.js

@@ -47,6 +47,7 @@ const {
   HTTPParser,
   isLenient,
   prepareError,
+  kSkipPendingData,
 } = require('_http_common');
 const {
   kUniqueHeaders,
@@ -692,7 +693,14 @@ function parserOnIncomingClient(res, shouldKeepAlive) {
     // We already have a response object, this means the server
     // sent a double response.
     socket.destroy();
-    return 0;  // No special treatment.
+    if (socket.parser) {
+      // https://github.com/nodejs/node/issues/60025
+      // Now, parser.incoming is pointed to the new IncomingMessage,
+      // we need to rewrite it to the first one and skip all the pending IncomingMessage
+      socket.parser.incoming = req.res;
+      socket.parser.incoming[kSkipPendingData] = true;
+    }
+    return 0;
   }
   req.res = res;
 

lib/_http_common.js

@@ -41,6 +41,7 @@ const {
 } = incoming;
 
 const kIncomingMessage = Symbol('IncomingMessage');
+const kSkipPendingData = Symbol('SkipPendingData');
 const kOnMessageBegin = HTTPParser.kOnMessageBegin | 0;
 const kOnHeaders = HTTPParser.kOnHeaders | 0;
 const kOnHeadersComplete = HTTPParser.kOnHeadersComplete | 0;
@@ -126,7 +127,7 @@ function parserOnBody(b) {
   const stream = this.incoming;
 
   // If the stream has already been removed, then drop it.
-  if (stream === null)
+  if (stream === null || stream[kSkipPendingData])
     return;
 
   // Pretend this was the result of a stream._read call.
@@ -141,7 +142,7 @@ function parserOnMessageComplete() {
   const parser = this;
   const stream = parser.incoming;
 
-  if (stream !== null) {
+  if (stream !== null && !stream[kSkipPendingData]) {
     stream.complete = true;
     // Emit any trailing headers.
     const headers = parser._headers;
@@ -310,4 +311,5 @@ module.exports = {
   HTTPParser,
   isLenient,
   prepareError,
+  kSkipPendingData,
 };

test/parallel/test-http-client-leaky-with-double-response.js

@@ -0,0 +1,45 @@
+'use strict';
+// Flags: --expose-gc
+const common = require('../common');
+const http = require('http');
+const assert = require('assert');
+const { onGC } = require('../common/gc');
+
+function createServer() {
+  const server = http.createServer(common.mustCall((req, res) => {
+    res.setHeader('Content-Type', 'application/json');
+    res.end(JSON.stringify({ hello: 'world' }));
+    req.socket.write('HTTP/1.1 400 Bad Request\r\n\r\n');
+  }));
+
+  return new Promise((resolve) => {
+    server.listen(0, common.mustCall(() => {
+      resolve(server);
+    }));
+  });
+}
+
+async function main() {
+  const server = await createServer();
+  const req = http.get({
+    host: '127.0.0.1',
+    port: server.address().port,
+  }, common.mustCall((res) => {
+    const chunks = [];
+    res.on('data', (c) => chunks.push(c));
+    res.on('end', common.mustCall(() => {
+      const body = Buffer.concat(chunks).toString('utf8');
+      const data = JSON.parse(body);
+      assert.strictEqual(data.hello, 'world');
+    }));
+  }));
+  const timer = setInterval(global.gc, 300);
+  onGC(req, {
+    ongc: common.mustCall(() => {
+      clearInterval(timer);
+      server.close();
+    })
+  });
+}
+
+main();