GeezerVison is showing github.com — http: fix writableFinished and 'finish' after write errors · nodejs/node@e26891e
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: lib/_http_outgoing.js +51-13Lines changed: 51 additions & 13 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -85,6 +85,9 @@ const kChunkedLength = Symbol('kChunkedLength'); | 85 | 85 | const kUniqueHeaders = Symbol('kUniqueHeaders'); | 86 | 86 | const kBytesWritten = Symbol('kBytesWritten'); | 87 | 87 | const kErrored = Symbol('errored'); | | 88 | +const kWritableFinished = Symbol('kWritableFinished'); | | 89 | +const kEndCallbacks = Symbol('kEndCallbacks'); | | 90 | +const kFlushError = Symbol('kFlushError'); | 88 | 91 | const kHighWaterMark = Symbol('kHighWaterMark'); | 89 | 92 | const kRejectNonStandardBodyWrites = Symbol('kRejectNonStandardBodyWrites'); | 90 | 93 |
| @@ -153,6 +156,9 @@ function OutgoingMessage(options) { | 153 | 156 | this._onPendingData = nop; | 154 | 157 |
| 155 | 158 | this[kErrored] = null; | | 159 | + this[kWritableFinished] = false; | | 160 | + this[kEndCallbacks] = null; | | 161 | + this[kFlushError] = null; | 156 | 162 | this[kHighWaterMark] = options?.highWaterMark ?? getDefaultHighWaterMark(); | 157 | 163 | this[kRejectNonStandardBodyWrites] = options?.rejectNonStandardBodyWrites ?? false; | 158 | 164 | } | @@ -203,11 +209,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'closed', { | 203 | 209 | ObjectDefineProperty(OutgoingMessage.prototype, 'writableFinished', { | 204 | 210 | __proto__: null, | 205 | 211 | get() { | 206 | | - return ( | 207 | | - this.finished && | 208 | | - this.outputSize === 0 && | 209 | | - (!this[kSocket] || this[kSocket].writableLength === 0) | 210 | | - ); | | 212 | + return this[kWritableFinished]; | 211 | 213 | }, | 212 | 214 | }); | 213 | 215 |
| @@ -1074,8 +1076,48 @@ OutgoingMessage.prototype.addTrailers = function addTrailers(headers) { | 1074 | 1076 | } | 1075 | 1077 | }; | 1076 | 1078 |
| 1077 | | -function onFinish(outmsg) { | 1078 | | - if (outmsg?.socket?._hadError) return; | | 1079 | +// Deliver end() callbacks, mirroring Writable: null on successful finish, | | 1080 | +// otherwise the error that prevented all data from being flushed. | | 1081 | +function flushEndCallbacks(msg, err) { | | 1082 | + const callbacks = msg[kEndCallbacks]; | | 1083 | + if (callbacks === null) | | 1084 | + return; | | 1085 | + msg[kEndCallbacks] = null; | | 1086 | + for (let i = 0; i < callbacks.length; i++) | | 1087 | + callbacks[i](err); | | 1088 | +} | | 1089 | + | | 1090 | +function getEndCallbackError(msg) { | | 1091 | + return msg[kErrored] ?? | | 1092 | + msg[kSocket]?.errored ?? | | 1093 | + new ERR_STREAM_DESTROYED('end'); | | 1094 | +} | | 1095 | + | | 1096 | +function queueEndCallback(msg, callback) { | | 1097 | + if (msg[kWritableFinished]) { | | 1098 | + callback(new ERR_STREAM_ALREADY_FINISHED('end')); | | 1099 | + return; | | 1100 | + } | | 1101 | + if (msg[kFlushError] !== null) { | | 1102 | + process.nextTick(callback, msg[kFlushError]); | | 1103 | + return; | | 1104 | + } | | 1105 | + msg[kEndCallbacks] ??= []; | | 1106 | + msg[kEndCallbacks].push(callback); | | 1107 | +} | | 1108 | + | | 1109 | +function onFinish(outmsg, err) { | | 1110 | + if (err || | | 1111 | + outmsg[kErrored] || | | 1112 | + outmsg[kSocket]?.errored || | | 1113 | + outmsg[kSocket]?._hadError) { | | 1114 | + outmsg[kFlushError] = err ?? getEndCallbackError(outmsg); | | 1115 | + flushEndCallbacks(outmsg, outmsg[kFlushError]); | | 1116 | + return; | | 1117 | + } | | 1118 | + | | 1119 | + outmsg[kWritableFinished] = true; | | 1120 | + flushEndCallbacks(outmsg, null); | 1079 | 1121 | outmsg.emit('finish'); | 1080 | 1122 | } | 1081 | 1123 |
| @@ -1104,11 +1146,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) { | 1104 | 1146 | write_(this, chunk, encoding, null, true); | 1105 | 1147 | } else if (this.finished) { | 1106 | 1148 | if (typeof callback === 'function') { | 1107 | | - if (!this.writableFinished) { | 1108 | | - this.on('finish', callback); | 1109 | | - } else { | 1110 | | - callback(new ERR_STREAM_ALREADY_FINISHED('end')); | 1111 | | - } | | 1149 | + queueEndCallback(this, callback); | 1112 | 1150 | } | 1113 | 1151 | return this; | 1114 | 1152 | } else if (!this._header) { | @@ -1121,7 +1159,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) { | 1121 | 1159 | } | 1122 | 1160 |
| 1123 | 1161 | if (typeof callback === 'function') | 1124 | | - this.once('finish', callback); | | 1162 | + queueEndCallback(this, callback); | 1125 | 1163 |
| 1126 | 1164 | if (strictContentLength(this) && this[kBytesWritten] !== this._contentLength) { | 1127 | 1165 | throw new ERR_HTTP_CONTENT_LENGTH_MISMATCH(this[kBytesWritten], this._contentLength); | |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-http-outgoing-end-multiple.js +3-1Lines changed: 3 additions & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -10,8 +10,10 @@ const onWriteAfterEndError = common.mustCall((err) => { | 10 | 10 | const server = http.createServer(common.mustCall(function(req, res) { | 11 | 11 | res.end('testing ended state', common.mustCall()); | 12 | 12 | assert.strictEqual(res.writableCorked, 0); | | 13 | + // end() before 'finish' has been emitted queues the callback, which then | | 14 | + // reports the outcome of the flush, matching stream.Writable. | 13 | 15 | res.end(common.mustCall((err) => { | 14 | | - assert.strictEqual(err.code, 'ERR_STREAM_ALREADY_FINISHED'); | | 16 | + assert.strictEqual(err, null); | 15 | 17 | })); | 16 | 18 | assert.strictEqual(res.writableCorked, 0); | 17 | 19 | res.end('end', onWriteAfterEndError); | |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-http-outgoing-writableFinished.js +124-25Lines changed: 124 additions & 25 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 2 | 2 | const common = require('../common'); | 3 | 3 | const assert = require('assert'); | 4 | 4 | const http = require('http'); | | 5 | +const { Duplex } = require('stream'); | 5 | 6 |
| 6 | | -const server = http.createServer(common.mustCall(function(req, res) { | 7 | | - assert.strictEqual(res.writableFinished, false); | 8 | | - res | 9 | | - .on('finish', common.mustCall(() => { | 10 | | - assert.strictEqual(res.writableFinished, true); | 11 | | - server.close(); | 12 | | - })) | 13 | | - .end(); | 14 | | -})); | 15 | | - | 16 | | -server.listen(0); | 17 | | - | 18 | | -server.on('listening', common.mustCall(function() { | 19 | | - const clientRequest = http.request({ | 20 | | - port: server.address().port, | 21 | | - method: 'GET', | 22 | | - path: '/' | | 7 | +// writableFinished becomes true once all data has been flushed, immediately | | 8 | +// before 'finish' is emitted. | | 9 | +{ | | 10 | + const server = http.createServer(common.mustCall(function(req, res) { | | 11 | + assert.strictEqual(res.writableFinished, false); | | 12 | + res | | 13 | + .on('finish', common.mustCall(() => { | | 14 | + assert.strictEqual(res.writableFinished, true); | | 15 | + server.close(); | | 16 | + })) | | 17 | + .end(); | | 18 | + })); | | 19 | + | | 20 | + server.listen(0); | | 21 | + | | 22 | + server.on('listening', common.mustCall(function() { | | 23 | + const clientRequest = http.request({ | | 24 | + port: server.address().port, | | 25 | + method: 'GET', | | 26 | + path: '/' | | 27 | + }); | | 28 | + | | 29 | + assert.strictEqual(clientRequest.writableFinished, false); | | 30 | + clientRequest | | 31 | + .on('finish', common.mustCall(() => { | | 32 | + assert.strictEqual(clientRequest.writableFinished, true); | | 33 | + })) | | 34 | + .end(); | | 35 | + assert.strictEqual(clientRequest.writableFinished, false); | | 36 | + })); | | 37 | +} | | 38 | + | | 39 | +// A request whose writes fail never becomes writableFinished and never emits | | 40 | +// 'finish'; the end() callback receives the write error instead. | | 41 | +{ | | 42 | + const writeError = new Error('forced write failure'); | | 43 | + const socket = new Duplex({ | | 44 | + read() {}, | | 45 | + write(chunk, encoding, callback) { | | 46 | + callback(writeError); | | 47 | + }, | 23 | 48 | }); | | 49 | + const failedRequest = http.request({ | | 50 | + createConnection: common.mustCall(() => socket), | | 51 | + method: 'POST', | | 52 | + }); | | 53 | + | | 54 | + failedRequest.on('finish', common.mustNotCall()); | | 55 | + failedRequest.on('error', common.mustCall((err) => { | | 56 | + assert.strictEqual(err, writeError); | | 57 | + })); | | 58 | + failedRequest.on('close', common.mustCall(() => { | | 59 | + assert.strictEqual(failedRequest.writableFinished, false); | | 60 | + })); | | 61 | + | | 62 | + failedRequest.write('body', common.mustCall((err) => { | | 63 | + assert.strictEqual(err, writeError); | | 64 | + })); | | 65 | + failedRequest.end(common.mustCall((err) => { | | 66 | + assert.ok(err instanceof Error); | | 67 | + assert.strictEqual(failedRequest.writableFinished, false); | | 68 | + | | 69 | + // Ending again after the flush has failed still reports the failure. | | 70 | + failedRequest.end(common.mustCall((endAgainErr) => { | | 71 | + assert.strictEqual(endAgainErr, err); | | 72 | + })); | | 73 | + })); | | 74 | +} | | 75 | + | | 76 | +// The same for a server response whose flush fails (e.g. the connection is | | 77 | +// reset mid-flush). Unlike the client case, the error here only ever | | 78 | +// surfaces through the socket write callbacks. | | 79 | +{ | | 80 | + const writeError = new Error('forced write failure'); | | 81 | + const socket = new Duplex({ | | 82 | + read() {}, | | 83 | + write(chunk, encoding, callback) { | | 84 | + callback(writeError); | | 85 | + }, | | 86 | + }); | | 87 | + | | 88 | + const server = http.createServer(common.mustCall((req, res) => { | | 89 | + res.on('finish', common.mustNotCall()); | | 90 | + res.on('close', common.mustCall(() => { | | 91 | + assert.strictEqual(res.writableFinished, false); | | 92 | + })); | | 93 | + res.end('hello', common.mustCall((err) => { | | 94 | + assert.strictEqual(err, writeError); | | 95 | + assert.strictEqual(res.writableFinished, false); | | 96 | + })); | | 97 | + })); | | 98 | + | | 99 | + server.emit('connection', socket); | | 100 | + socket.push('GET / HTTP/1.1\r\nHost: example.com\r\n\r\n'); | | 101 | +} | | 102 | + | | 103 | +// The same when end() happens after the failed write, with no data left to | | 104 | +// flush: the write failure must still be detected even though end() itself | | 105 | +// has nothing to send. | | 106 | +{ | | 107 | + const writeError = new Error('forced write failure'); | | 108 | + const socket = new Duplex({ | | 109 | + read() {}, | | 110 | + write(chunk, encoding, callback) { | | 111 | + callback(writeError); | | 112 | + }, | | 113 | + }); | | 114 | + | | 115 | + const server = http.createServer(common.mustCall((req, res) => { | | 116 | + res.on('finish', common.mustNotCall()); | | 117 | + res.setHeader('Content-Length', '5'); | | 118 | + res.write('hello', common.mustCall((err) => { | | 119 | + assert.strictEqual(err, writeError); | | 120 | + })); | | 121 | + setImmediate(common.mustCall(() => { | | 122 | + res.end(common.mustCall((err) => { | | 123 | + assert.strictEqual(err, writeError); | | 124 | + assert.strictEqual(res.writableFinished, false); | | 125 | + })); | | 126 | + })); | | 127 | + })); | 24 | 128 |
| 25 | | - assert.strictEqual(clientRequest.writableFinished, false); | 26 | | - clientRequest | 27 | | - .on('finish', common.mustCall(() => { | 28 | | - assert.strictEqual(clientRequest.writableFinished, true); | 29 | | - })) | 30 | | - .end(); | 31 | | - assert.strictEqual(clientRequest.writableFinished, false); | 32 | | -})); | | 129 | + server.emit('connection', socket); | | 130 | + socket.push('GET / HTTP/1.1\r\nHost: example.com\r\n\r\n'); | | 131 | +} |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-stream-pipeline.js +5-1Lines changed: 5 additions & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -270,7 +270,11 @@ tmpdir.refresh(); | 270 | 270 |
| 271 | 271 | { | 272 | 272 | const server = http.createServer(common.mustCallAtLeast((req, res) => { | 273 | | - pipeline(req, res, common.mustSucceed()); | | 273 | + pipeline(req, res, common.mustCall((err) => { | | 274 | + // The client destroys the request body source before EOF below, so the | | 275 | + // echoed response cannot finish successfully either. | | 276 | + assert.strictEqual(err?.code, 'ERR_STREAM_PREMATURE_CLOSE'); | | 277 | + })); | 274 | 278 | })); | 275 | 279 |
| 276 | 280 | server.listen(0, common.mustCall(() => { | |
|