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

@@ -328,14 +328,19 @@ OutgoingMessage.prototype.destroy = function destroy(error) {
 if(this[kSocket]){
 this[kSocket].destroy(error);
 }else{
-this.once('socket',functionsocketDestroyOnConnect(socket){
-socket.destroy(error);
-});
+process.nextTick(emitDestroyNT,this);
 }
 
 returnthis;
 };
 
+functionemitDestroyNT(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

@@ -841,14 +841,21 @@ function socketOnClose(socket, state) {
 debug('server socket close');
 freeParser(socket.parser,null,socket);
 abortIncoming(state.incoming);
+abortOutgoing(state.outgoing);
 }
 
 functionabortIncoming(incoming){
 while(incoming.length){
 constreq=incoming.shift();
 req.destroy(newConnResetException('aborted'));
 }
-// Abort socket._httpMessage ?
+}
+
+functionabortOutgoing(outgoing){
+while(outgoing.length){
+constreq=outgoing.shift();
+req.destroy(newConnResetException('aborted'));
+}
 }
 
 functionsocketOnEnd(server,socket,parser,state){

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

@@ -2,6 +2,32 @@
 constcommon=require('../common');
 consthttp=require('http');
 constassert=require('assert');
+const{ OutgoingMessage }=require('http');
+
+// OutgoingMessage.destroy() with no socket should emit 'close' and set closed.
+{
+constmsg=newOutgoingMessage();
+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'.
+{
+constmsg=newOutgoingMessage();
+consterr=newError('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);
+}
 
 {
 constserver=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.
+
+constcommon=require('../common');
+consthttp=require('http');
+constnet=require('net');
+constassert=require('assert');
+
+letrequestCount=0;
+
+constserver=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(){
+constport=this.address().port;
+constclient=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();
+}));