http: cleanup pipeline queue

When a socket with pipelined requests is destroyed then some requests
will leak.

PR-URL: #62534
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>

Author: ronag

lib/_http_outgoing.js

@@ -386,14 +386,19 @@ OutgoingMessage.prototype.destroy = function destroy(error) {
   if (this[kSocket]) {
     this[kSocket].destroy(error);
   } else {
-    this.once('socket', function socketDestroyOnConnect(socket) {
-      socket.destroy(error);
-    });
+    process.nextTick(emitDestroyNT, this);
   }
 
   return this;
 };
 
+function emitDestroyNT(self) {
+  if (!self._closed) {
+    self._closed = true;
+    self.emit('close');
+  }
+}
+
 
 // This abstract either writing directly to the socket or buffering it.
 OutgoingMessage.prototype._send = function _send(data, encoding, callback, byteLength) {

lib/_http_server.js

@@ -833,14 +833,21 @@ function socketOnClose(socket, state) {
   debug('server socket close');
   freeParser(socket.parser, null, socket);
   abortIncoming(state.incoming);
+  abortOutgoing(state.outgoing);
 }
 
 function abortIncoming(incoming) {
   while (incoming.length) {
     const req = incoming.shift();
     req.destroy(new ConnResetException('aborted'));
   }
-  // Abort socket._httpMessage ?
+}
+
+function abortOutgoing(outgoing) {
+  while (outgoing.length) {
+    const req = outgoing.shift();
+    req.destroy(new ConnResetException('aborted'));
+  }
 }
 
 function socketOnEnd(server, socket, parser, state) {

test/parallel/test-http-outgoing-destroyed.js

@@ -2,6 +2,32 @@
 const common = require('../common');
 const http = require('http');
 const assert = require('assert');
+const { OutgoingMessage } = require('http');
+
+// OutgoingMessage.destroy() with no socket should emit 'close' and set closed.
+{
+  const msg = new OutgoingMessage();
+  assert.strictEqual(msg.destroyed, false);
+  assert.strictEqual(msg.closed, false);
+  msg.on('close', common.mustCall(() => {
+    assert.strictEqual(msg.destroyed, true);
+    assert.strictEqual(msg.closed, true);
+  }));
+  msg.destroy();
+  assert.strictEqual(msg.destroyed, true);
+}
+
+// OutgoingMessage.destroy(err) with no socket should set errored and emit 'close'.
+{
+  const msg = new OutgoingMessage();
+  const err = new Error('test destroy');
+  msg.on('close', common.mustCall(() => {
+    assert.strictEqual(msg.closed, true);
+    assert.strictEqual(msg.errored, err);
+  }));
+  msg.destroy(err);
+  assert.strictEqual(msg.errored, err);
+}
 
 {
   const server = http.createServer(common.mustCall((req, res) => {

test/parallel/test-http-pipeline-outgoing-destroy.js

@@ -0,0 +1,50 @@
+'use strict';
+// Test that queued (pipelined) outgoing responses are destroyed when the
+// socket closes before the first response has finished.  Previously,
+// socketOnClose only aborted state.incoming (pending requests) but left
+// state.outgoing responses with socket=null alive forever.
+
+const common = require('../common');
+const http = require('http');
+const net = require('net');
+const assert = require('assert');
+
+let requestCount = 0;
+
+const server = http.createServer(common.mustCall((req, res) => {
+  requestCount++;
+
+  if (requestCount === 1) {
+    // Keep the first response open so the second response is queued in
+    // state.outgoing with socket === null.
+    res.writeHead(200);
+    res.write('start');
+    // Intentionally do not call res.end().
+  } else {
+    // The second response should be queued — no socket assigned yet.
+    assert.strictEqual(res.socket, null);
+    assert.strictEqual(res.destroyed, false);
+    assert.strictEqual(res.closed, false);
+
+    res.on('close', common.mustCall(() => {
+      assert.strictEqual(res.destroyed, true);
+      assert.strictEqual(res.closed, true);
+      server.close();
+    }));
+
+    // Simulate client dying while first response is still in flight.
+    req.socket.destroy();
+  }
+}, 2));
+
+server.listen(0, common.mustCall(function() {
+  const port = this.address().port;
+  const client = net.connect(port);
+
+  // Send two pipelined HTTP/1.1 requests at once.
+  client.write(
+    `GET /1 HTTP/1.1\r\nHost: localhost:${port}\r\n\r\n` +
+    `GET /2 HTTP/1.1\r\nHost: localhost:${port}\r\n\r\n`,
+  );
+  client.resume();
+}));