FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

http: fix writableFinished and 'finish' after write errors · nodejs/node@4de7938 · GitHub

/ node Public

Commit 4de7938

Browse files
pimterry
authored and
aduh95
committed
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>
1 parent 0ba0911 commit 4de7938

4 files changed

Lines changed: 183 additions & 40 deletions

File tree

‎lib/_http_outgoing.js‎

Lines changed: 51 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,9 @@ const kChunkedLength = Symbol('kChunkedLength');
8585
const kUniqueHeaders = Symbol('kUniqueHeaders');
8686
const kBytesWritten = Symbol('kBytesWritten');
8787
const kErrored = Symbol('errored');
88+
const kWritableFinished = Symbol('kWritableFinished');
89+
const kEndCallbacks = Symbol('kEndCallbacks');
90+
const kFlushError = Symbol('kFlushError');
8891
const kHighWaterMark = Symbol('kHighWaterMark');
8992
const kRejectNonStandardBodyWrites = Symbol('kRejectNonStandardBodyWrites');
9093

@@ -153,6 +156,9 @@ function OutgoingMessage(options) {
153156
this._onPendingData = nop;
154157

155158
this[kErrored] = null;
159+
this[kWritableFinished] = false;
160+
this[kEndCallbacks] = null;
161+
this[kFlushError] = null;
156162
this[kHighWaterMark] = options?.highWaterMark ?? getDefaultHighWaterMark();
157163
this[kRejectNonStandardBodyWrites] = options?.rejectNonStandardBodyWrites ?? false;
158164
}
@@ -203,11 +209,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'closed', {
203209
ObjectDefineProperty(OutgoingMessage.prototype, 'writableFinished', {
204210
__proto__: null,
205211
get() {
206-
return (
207-
this.finished &&
208-
this.outputSize === 0 &&
209-
(!this[kSocket] || this[kSocket].writableLength === 0)
210-
);
212+
return this[kWritableFinished];
211213
},
212214
});
213215

@@ -1074,8 +1076,48 @@ OutgoingMessage.prototype.addTrailers = function addTrailers(headers) {
10741076
}
10751077
};
10761078

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);
10791121
outmsg.emit('finish');
10801122
}
10811123

@@ -1104,11 +1146,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
11041146
write_(this, chunk, encoding, null, true);
11051147
} else if (this.finished) {
11061148
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);
11121150
}
11131151
return this;
11141152
} else if (!this._header) {
@@ -1121,7 +1159,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
11211159
}
11221160

11231161
if (typeof callback === 'function')
1124-
this.once('finish', callback);
1162+
queueEndCallback(this, callback);
11251163

11261164
if (strictContentLength(this) && this[kBytesWritten] !== this._contentLength) {
11271165
throw new ERR_HTTP_CONTENT_LENGTH_MISMATCH(this[kBytesWritten], this._contentLength);

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,10 @@ const onWriteAfterEndError = common.mustCall((err) => {
1010
const server = http.createServer(common.mustCall(function(req, res) {
1111
res.end('testing ended state', common.mustCall());
1212
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.
1315
res.end(common.mustCall((err) => {
14-
assert.strictEqual(err.code, 'ERR_STREAM_ALREADY_FINISHED');
16+
assert.strictEqual(err, null);
1517
}));
1618
assert.strictEqual(res.writableCorked, 0);
1719
res.end('end', onWriteAfterEndError);

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

Lines changed: 124 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -2,31 +2,130 @@
22
const common = require('../common');
33
const assert = require('assert');
44
const http = require('http');
5+
const { Duplex } = require('stream');
56

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+
},
2348
});
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+
}));
24128

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+
}

‎test/parallel/test-stream-pipeline.js‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -270,7 +270,11 @@ tmpdir.refresh();
270270

271271
{
272272
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+
}));
274278
}));
275279

276280
server.listen(0, common.mustCall(() => {

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL