From ddbd8467ed8e80c8244c879610fe3c77543a0956 Mon Sep 17 00:00:00 2001 From: Tim Perry Date: Tue, 28 Jul 2026 16:27:33 +0200 Subject: [PATCH] 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 --- lib/_http_outgoing.js | 64 ++++++-- .../test-http-outgoing-end-multiple.js | 4 +- .../test-http-outgoing-writableFinished.js | 149 +++++++++++++++--- test/parallel/test-stream-pipeline.js | 6 +- 4 files changed, 183 insertions(+), 40 deletions(-) diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js index 7694fe4b5b3f..395ab86a95bb 100644 --- a/lib/_http_outgoing.js +++ b/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); diff --git a/test/parallel/test-http-outgoing-end-multiple.js b/test/parallel/test-http-outgoing-end-multiple.js index 696443f9390c..5463485f4061 100644 --- a/test/parallel/test-http-outgoing-end-multiple.js +++ b/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); diff --git a/test/parallel/test-http-outgoing-writableFinished.js b/test/parallel/test-http-outgoing-writableFinished.js index 6f84d91e7147..36257e0ea243 100644 --- a/test/parallel/test-http-outgoing-writableFinished.js +++ b/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'); +} diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js index 8852f24a7505..7b4052f2b47d 100644 --- a/test/parallel/test-stream-pipeline.js +++ b/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(() => {