http: prevent reuse after incomplete request destruction

Signed-off-by: Dayun <dlekdbs6530@gmail.com>

Author: dayun6530

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,131 @@
+'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;
+let firstResponseReceived = false;
+let firstResponseConnection;
+let firstResponseAborted = false;
+let firstResponseClosed = false;
+let secondStarted = false;
+
+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 port = server.address().port;
+
+  function startSecondRequest() {
+    if (secondStarted) {
+      return;
+    }
+
+    secondStarted = true;
+
+    process.nextTick(() => {
+      const second = http.request({
+        port,
+        method: 'GET',
+        agent,
+      }, common.mustCall((res) => {
+        second.setTimeout(0);
+
+        assert.strictEqual(second.reusedSocket, false);
+
+        if (firstResponseReceived) {
+          assert.strictEqual(firstResponseConnection, 'keep-alive');
+          assert.strictEqual(firstResponseAborted, true);
+          assert.strictEqual(firstResponseClosed, true);
+        }
+
+        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),
+        common.mustNotCall('second request timed out'),
+      );
+
+      second.end();
+    });
+  }
+
+  const first = http.request({
+    port,
+    method: 'POST',
+    agent,
+  }, (res) => {
+    firstResponseReceived = true;
+    firstResponseConnection = res.headers.connection;
+
+    res.on('end', common.mustNotCall());
+    res.on('aborted', () => {
+      firstResponseAborted = true;
+    });
+    res.on('error', common.expectsError({
+      code: 'ECONNRESET',
+      message: 'aborted',
+    }));
+
+    res.on('close', () => {
+      firstResponseClosed = true;
+      startSecondRequest();
+    });
+
+    res.resume();
+  });
+
+  first.on('error', (err) => {
+    switch (err.code) {
+      case 'ECONNRESET':
+      case 'ECONNABORTED':
+      case 'EPIPE':
+        break;
+      default:
+        throw err;
+    }
+
+    if (!firstResponseReceived) {
+      startSecondRequest();
+    }
+  });
+
+  first.end(Buffer.alloc(1_000_000));
+}));

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

@@ -0,0 +1,86 @@
+'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.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));
+}));