http: fix writableFinished and 'finish' after write errors

Only emit 'finish' and set writableFinished once all data has actually
been flushed successfully. end() callbacks now report the outcome like
stream.Writable: called with null on finish, or with the error that
prevented the flush.
Signed-off-by: Tim Perry <pimterry@gmail.com>
PR-URL: #64847
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: pimterry

lib/_http_outgoing.js

@@ -85,6 +85,9 @@ const kChunkedLength = Symbol('kChunkedLength');
 constkUniqueHeaders=Symbol('kUniqueHeaders');
 constkBytesWritten=Symbol('kBytesWritten');
 constkErrored=Symbol('errored');
+constkWritableFinished=Symbol('kWritableFinished');
+constkEndCallbacks=Symbol('kEndCallbacks');
+constkFlushError=Symbol('kFlushError');
 constkHighWaterMark=Symbol('kHighWaterMark');
 constkRejectNonStandardBodyWrites=Symbol('kRejectNonStandardBodyWrites');
 
@@ -153,6 +156,9 @@ function OutgoingMessage(options) {
 this._onPendingData=nop;
 
 this[kErrored]=null;
+this[kWritableFinished]=false;
+this[kEndCallbacks]=null;
+this[kFlushError]=null;
 this[kHighWaterMark]=options?.highWaterMark??getDefaultHighWaterMark();
 this[kRejectNonStandardBodyWrites]=options?.rejectNonStandardBodyWrites??false;
 }
@@ -203,11 +209,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'closed', {
 ObjectDefineProperty(OutgoingMessage.prototype,'writableFinished',{
 __proto__: null,
 get(){
-return(
-this.finished&&
-this.outputSize===0&&
-(!this[kSocket]||this[kSocket].writableLength===0)
-);
+returnthis[kWritableFinished];
 },
 });
 
@@ -1074,8 +1076,48 @@ OutgoingMessage.prototype.addTrailers = function addTrailers(headers) {
 }
 };
 
-functiononFinish(outmsg){
-if(outmsg?.socket?._hadError)return;
+// Deliver end() callbacks, mirroring Writable: null on successful finish,
+// otherwise the error that prevented all data from being flushed.
+functionflushEndCallbacks(msg,err){
+constcallbacks=msg[kEndCallbacks];
+if(callbacks===null)
+return;
+msg[kEndCallbacks]=null;
+for(leti=0;i<callbacks.length;i++)
+callbacks[i](err);
+}
+
+functiongetEndCallbackError(msg){
+returnmsg[kErrored]??
+msg[kSocket]?.errored??
+newERR_STREAM_DESTROYED('end');
+}
+
+functionqueueEndCallback(msg,callback){
+if(msg[kWritableFinished]){
+callback(newERR_STREAM_ALREADY_FINISHED('end'));
+return;
+}
+if(msg[kFlushError]!==null){
+process.nextTick(callback,msg[kFlushError]);
+return;
+}
+msg[kEndCallbacks]??=[];
+msg[kEndCallbacks].push(callback);
+}
+
+functiononFinish(outmsg,err){
+if(err||
+outmsg[kErrored]||
+outmsg[kSocket]?.errored||
+outmsg[kSocket]?._hadError){
+outmsg[kFlushError]=err??getEndCallbackError(outmsg);
+flushEndCallbacks(outmsg,outmsg[kFlushError]);
+return;
+}
+
+outmsg[kWritableFinished]=true;
+flushEndCallbacks(outmsg,null);
 outmsg.emit('finish');
 }
 
@@ -1104,11 +1146,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
 write_(this,chunk,encoding,null,true);
 }elseif(this.finished){
 if(typeofcallback==='function'){
-if(!this.writableFinished){
-this.on('finish',callback);
-}else{
-callback(newERR_STREAM_ALREADY_FINISHED('end'));
-}
+queueEndCallback(this,callback);
 }
 returnthis;
 }elseif(!this._header){
@@ -1121,7 +1159,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
 }
 
 if(typeofcallback==='function')
-this.once('finish',callback);
+queueEndCallback(this,callback);
 
 if(strictContentLength(this)&&this[kBytesWritten]!==this._contentLength){
 thrownewERR_HTTP_CONTENT_LENGTH_MISMATCH(this[kBytesWritten],this._contentLength);

test/parallel/test-http-outgoing-end-multiple.js

@@ -10,8 +10,10 @@ const onWriteAfterEndError = common.mustCall((err) => {
 constserver=http.createServer(common.mustCall(function(req,res){
 res.end('testing ended state',common.mustCall());
 assert.strictEqual(res.writableCorked,0);
+// end() before 'finish' has been emitted queues the callback, which then
+// reports the outcome of the flush, matching stream.Writable.
 res.end(common.mustCall((err)=>{
-assert.strictEqual(err.code,'ERR_STREAM_ALREADY_FINISHED');
+assert.strictEqual(err,null);
 }));
 assert.strictEqual(res.writableCorked,0);
 res.end('end',onWriteAfterEndError);

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

@@ -2,31 +2,130 @@
 constcommon=require('../common');
 constassert=require('assert');
 consthttp=require('http');
+const{ Duplex }=require('stream');
 
-constserver=http.createServer(common.mustCall(function(req,res){
-assert.strictEqual(res.writableFinished,false);
-res
-.on('finish',common.mustCall(()=>{
-assert.strictEqual(res.writableFinished,true);
-server.close();
-}))
-.end();
-}));
-
-server.listen(0);
-
-server.on('listening',common.mustCall(function(){
-constclientRequest=http.request({
-port: server.address().port,
-method: 'GET',
-path: '/'
+// writableFinished becomes true once all data has been flushed, immediately
+// before 'finish' is emitted.
+{
+constserver=http.createServer(common.mustCall(function(req,res){
+assert.strictEqual(res.writableFinished,false);
+res
+.on('finish',common.mustCall(()=>{
+assert.strictEqual(res.writableFinished,true);
+server.close();
+}))
+.end();
+}));
+
+server.listen(0);
+
+server.on('listening',common.mustCall(function(){
+constclientRequest=http.request({
+port: server.address().port,
+method: 'GET',
+path: '/'
+});
+
+assert.strictEqual(clientRequest.writableFinished,false);
+clientRequest
+.on('finish',common.mustCall(()=>{
+assert.strictEqual(clientRequest.writableFinished,true);
+}))
+.end();
+assert.strictEqual(clientRequest.writableFinished,false);
+}));
+}
+
+// A request whose writes fail never becomes writableFinished and never emits
+// 'finish'; the end() callback receives the write error instead.
+{
+constwriteError=newError('forced write failure');
+constsocket=newDuplex({
+read(){},
+write(chunk,encoding,callback){
+callback(writeError);
+},
 });
+constfailedRequest=http.request({
+createConnection: common.mustCall(()=>socket),
+method: 'POST',
+});
+
+failedRequest.on('finish',common.mustNotCall());
+failedRequest.on('error',common.mustCall((err)=>{
+assert.strictEqual(err,writeError);
+}));
+failedRequest.on('close',common.mustCall(()=>{
+assert.strictEqual(failedRequest.writableFinished,false);
+}));
+
+failedRequest.write('body',common.mustCall((err)=>{
+assert.strictEqual(err,writeError);
+}));
+failedRequest.end(common.mustCall((err)=>{
+assert.ok(errinstanceofError);
+assert.strictEqual(failedRequest.writableFinished,false);
+
+// Ending again after the flush has failed still reports the failure.
+failedRequest.end(common.mustCall((endAgainErr)=>{
+assert.strictEqual(endAgainErr,err);
+}));
+}));
+}
+
+// The same for a server response whose flush fails (e.g. the connection is
+// reset mid-flush). Unlike the client case, the error here only ever
+// surfaces through the socket write callbacks.
+{
+constwriteError=newError('forced write failure');
+constsocket=newDuplex({
+read(){},
+write(chunk,encoding,callback){
+callback(writeError);
+},
+});
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+res.on('finish',common.mustNotCall());
+res.on('close',common.mustCall(()=>{
+assert.strictEqual(res.writableFinished,false);
+}));
+res.end('hello',common.mustCall((err)=>{
+assert.strictEqual(err,writeError);
+assert.strictEqual(res.writableFinished,false);
+}));
+}));
+
+server.emit('connection',socket);
+socket.push('GET / HTTP/1.1\r\nHost: example.com\r\n\r\n');
+}
+
+// The same when end() happens after the failed write, with no data left to
+// flush: the write failure must still be detected even though end() itself
+// has nothing to send.
+{
+constwriteError=newError('forced write failure');
+constsocket=newDuplex({
+read(){},
+write(chunk,encoding,callback){
+callback(writeError);
+},
+});
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+res.on('finish',common.mustNotCall());
+res.setHeader('Content-Length','5');
+res.write('hello',common.mustCall((err)=>{
+assert.strictEqual(err,writeError);
+}));
+setImmediate(common.mustCall(()=>{
+res.end(common.mustCall((err)=>{
+assert.strictEqual(err,writeError);
+assert.strictEqual(res.writableFinished,false);
+}));
+}));
+}));
 
-assert.strictEqual(clientRequest.writableFinished,false);
-clientRequest
-.on('finish',common.mustCall(()=>{
-assert.strictEqual(clientRequest.writableFinished,true);
-}))
-.end();
-assert.strictEqual(clientRequest.writableFinished,false);
-}));
+server.emit('connection',socket);
+socket.push('GET / HTTP/1.1\r\nHost: example.com\r\n\r\n');
+}

test/parallel/test-stream-pipeline.js

@@ -270,7 +270,11 @@ tmpdir.refresh();
 
 {
 constserver=http.createServer(common.mustCallAtLeast((req,res)=>{
-pipeline(req,res,common.mustSucceed());
+pipeline(req,res,common.mustCall((err)=>{
+// The client destroys the request body source before EOF below, so the
+// echoed response cannot finish successfully either.
+assert.strictEqual(err?.code,'ERR_STREAM_PREMATURE_CLOSE');
+}));
 }));
 
 server.listen(0,common.mustCall(()=>{