http: prevent reuse after incomplete request destruction

lib/internal/streams/destroy.js

@@ -330,6 +330,18 @@ function destroyer(stream, err) {
 
   // TODO: Remove isRequest branches.
   if (isServerRequest(stream)) {
+    const socket = stream.socket;
+    const response = socket?._httpMessage;
+
+    if (response?.req === stream) {
+      if (response.headersSent) {
+        response.destroy();
+      } else {
+        response.shouldKeepAlive = false;
+        response._last = true;
+      }
+    }
+
     stream.socket = null;
     stream.destroy(err);
   } else if (isRequest(stream)) {

test/parallel/test-http-server-for-await-keepalive-headers-sent.js

@@ -0,0 +1,97 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+
+const agent = new http.Agent({
+  keepAlive: true,
+  maxSockets: 1,
+});
+
+let serverRequests = 0;
+
+const server = http.createServer(async (req, res) => {
+  serverRequests++;
+
+  if (serverRequests === 1) {
+    res.write('partial');
+
+    try {
+      for await (const chunk of req) {
+        throw new Error(`payload too large: ${chunk.length}`);
+      }
+    } catch {
+      res.end('payload too large');
+    }
+    return;
+  }
+
+  res.end('ok');
+});
+
+server.listen(0, common.mustCall(() => {
+  const first = http.request({
+    port: server.address().port,
+    method: 'POST',
+    agent,
+  }, common.mustCall((res) => {
+    assert.strictEqual(res.headers.connection, 'keep-alive');
+
+    res.on('end', common.mustNotCall());
+    res.on('aborted', common.mustCall());
+    res.on('error', common.expectsError({
+      code: 'ECONNRESET',
+      message: 'aborted',
+    }));
+
+    res.on('close', common.mustCall(() => {
+      process.nextTick(common.mustCall(() => {
+        const second = http.request({
+          port: server.address().port,
+          method: 'GET',
+          agent,
+        }, common.mustCall((res) => {
+          second.setTimeout(0);
+          assert.strictEqual(second.reusedSocket, false);
+          res.setEncoding('utf8');
+
+          let body = '';
+
+          res.on('data', (chunk) => {
+            body += chunk;
+          });
+
+          res.on('end', common.mustCall(() => {
+            assert.strictEqual(body, 'ok');
+            assert.strictEqual(serverRequests, 2);
+
+            agent.destroy();
+            server.close();
+          }));
+        }));
+
+        second.setTimeout(common.platformTimeout(1000), () => {
+          assert.fail('second request timed out');
+        });
+
+        second.end();
+      }));
+    }));
+
+    res.resume();
+  }));
+
+  first.on('error', (err) => {
+    switch (err.code) {
+      case 'ECONNRESET':
+      case 'ECONNABORTED':
+      case 'EPIPE':
+        break;
+      default:
+        assert.fail(`Unexpected error code ${err.code}`);
+    }
+  });
+
+  first.end(Buffer.alloc(1_000_000));
+}));

test/parallel/test-http-server-for-await-keepalive.js

@@ -0,0 +1,76 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+
+const agent = new http.Agent({
+  keepAlive: true,
+  maxSockets: 1,
+});
+
+let serverRequests = 0;
+
+const server = http.createServer(async (req, res) => {
+  serverRequests++;
+
+  if (serverRequests === 1) {
+    try {
+      for await (const chunk of req) {
+        throw new Error(`payload too large: ${chunk.length}`);
+      }
+    } catch {
+      res.end('payload too large');
+    }
+    return;
+  }
+
+  res.end('ok');
+});
+
+server.listen(0, common.mustCall(() => {
+  const first = http.request({
+    port: server.address().port,
+    method: 'POST',
+    agent,
+  }, common.mustCall((res) => {
+    assert.strictEqual(res.headers.connection, 'close');
+    res.resume();
+
+    res.on('end', common.mustCall(() => {
+      process.nextTick(common.mustCall(() => {
+        const second = http.request({
+          port: server.address().port,
+          method: 'GET',
+          agent,
+        }, common.mustCall((res) => {
+          second.setTimeout(0);
+          assert.strictEqual(second.reusedSocket, false);
+          res.setEncoding('utf8');
+
+          let body = '';
+
+          res.on('data', (chunk) => {
+            body += chunk;
+          });
+
+          res.on('end', common.mustCall(() => {
+            assert.strictEqual(body, 'ok');
+            assert.strictEqual(serverRequests, 2);
+
+            agent.destroy();
+            server.close();
+          }));
+        }));
+
+        second.setTimeout(common.platformTimeout(1000), () => {
+          assert.fail('second request timed out');
+        });
+
+        second.end();
+      }));
+    }));
+  }));
+
+  first.end(Buffer.alloc(1_000_000));
+}));