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>

Author: pimterry

lib/_http_outgoing.js

@@ -85,6 +85,9 @@ const kChunkedLength = Symbol('kChunkedLength');
 const kUniqueHeaders = Symbol('kUniqueHeaders');
 const kBytesWritten = Symbol('kBytesWritten');
 const kErrored = Symbol('errored');
+const kWritableFinished = Symbol('kWritableFinished');
+const kEndCallbacks = Symbol('kEndCallbacks');
+const kFlushError = Symbol('kFlushError');
 const kHighWaterMark = Symbol('kHighWaterMark');
 const kRejectNonStandardBodyWrites = 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)
-    );
+    return this[kWritableFinished];
   },
 });
 
@@ -1078,8 +1080,48 @@ OutgoingMessage.prototype.addTrailers = function addTrailers(headers) {
   }
 };
 
-function onFinish(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.
+function flushEndCallbacks(msg, err) {
+  const callbacks = msg[kEndCallbacks];
+  if (callbacks === null)
+    return;
+  msg[kEndCallbacks] = null;
+  for (let i = 0; i < callbacks.length; i++)
+    callbacks[i](err);
+}
+
+function getEndCallbackError(msg) {
+  return msg[kErrored] ??
+    msg[kSocket]?.errored ??
+    new ERR_STREAM_DESTROYED('end');
+}
+
+function queueEndCallback(msg, callback) {
+  if (msg[kWritableFinished]) {
+    callback(new ERR_STREAM_ALREADY_FINISHED('end'));
+    return;
+  }
+  if (msg[kFlushError] !== null) {
+    process.nextTick(callback, msg[kFlushError]);
+    return;
+  }
+  msg[kEndCallbacks] ??= [];
+  msg[kEndCallbacks].push(callback);
+}
+
+function onFinish(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');
 }
 
@@ -1108,11 +1150,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
     write_(this, chunk, encoding, null, true);
   } else if (this.finished) {
     if (typeof callback === 'function') {
-      if (!this.writableFinished) {
-        this.on('finish', callback);
-      } else {
-        callback(new ERR_STREAM_ALREADY_FINISHED('end'));
-      }
+      queueEndCallback(this, callback);
     }
     return this;
   } else if (!this._header) {
@@ -1125,7 +1163,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
   }
 
   if (typeof callback === 'function')
-    this.once('finish', callback);
+    queueEndCallback(this, callback);
 
   if (strictContentLength(this) && this[kBytesWritten] !== this._contentLength) {
     throw new ERR_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) => {
 const server = 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 @@
 const common = require('../common');
 const assert = require('assert');
 const http = require('http');
+const { Duplex } = require('stream');
 
-const server = 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() {
-  const clientRequest = http.request({
-    port: server.address().port,
-    method: 'GET',
-    path: '/'
+// writableFinished becomes true once all data has been flushed, immediately
+// before 'finish' is emitted.
+{
+  const server = 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() {
+    const clientRequest = 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.
+{
+  const writeError = new Error('forced write failure');
+  const socket = new Duplex({
+    read() {},
+    write(chunk, encoding, callback) {
+      callback(writeError);
+    },
   });
+  const failedRequest = 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(err instanceof Error);
+    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.
+{
+  const writeError = new Error('forced write failure');
+  const socket = new Duplex({
+    read() {},
+    write(chunk, encoding, callback) {
+      callback(writeError);
+    },
+  });
+
+  const server = 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.
+{
+  const writeError = new Error('forced write failure');
+  const socket = new Duplex({
+    read() {},
+    write(chunk, encoding, callback) {
+      callback(writeError);
+    },
+  });
+
+  const server = 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();
 
 {
   const server = 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(() => {