Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit 39af61f

Browse filesBrowse files
addaleaxBethGriggs
addaleax
authored and
BethGriggs
committed
stream: fix end-of-stream for HTTP/2
HTTP/2 streams call `.end()` on themselves from their `.destroy()` method, which might be queued (e.g. due to network congestion) and not processed before the stream itself is destroyed. In that case, the `_writableState.ended` property could be set before the stream emits its `'close'` event, and never actually emits the `'finished'` event, confusing the end-of-stream implementation so that it wouldn’t call its callback. This can be fixed by watching for the end events themselves using the existing `'finish'` and `'end'` listeners rather than relying on the `.ended` properties of the `_...State` objects. These properties still need to be checked to know whether stream closure was premature – My understanding is that ideally, streams should not emit `'close'` before `'end'` and/or `'finished'`, so this might be another bug, but changing this would require modifying tests and almost certainly be a breaking change. Fixes: #24456 PR-URL: #24926 Reviewed-By: Rich Trott <rtrott@gmail.com> Reviewed-By: Luigi Pinca <luigipinca@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Franziska Hinkelmann <franziska.hinkelmann@gmail.com>
1 parent b2e6cbd commit 39af61f
Copy full SHA for 39af61f

File tree

Expand file treeCollapse file tree

3 files changed

+53
-9
lines changed
Open diff view settings
Filter options
Expand file treeCollapse file tree

3 files changed

+53
-9
lines changed
Open diff view settings
Collapse file

‎lib/internal/streams/end-of-stream.js‎

Copy file name to clipboardExpand all lines: lib/internal/streams/end-of-stream.js
+14-7Lines changed: 14 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -36,22 +36,24 @@ function eos(stream, opts, callback) {
3636

3737
callback = once(callback);
3838

39-
const ws = stream._writableState;
40-
const rs = stream._readableState;
4139
let readable = opts.readable || (opts.readable !== false && stream.readable);
4240
let writable = opts.writable || (opts.writable !== false && stream.writable);
4341

4442
const onlegacyfinish = () => {
4543
if (!stream.writable) onfinish();
4644
};
4745

46+
var writableEnded = stream._writableState && stream._writableState.finished;
4847
const onfinish = () => {
4948
writable = false;
49+
writableEnded = true;
5050
if (!readable) callback.call(stream);
5151
};
5252

53+
var readableEnded = stream._readableState && stream._readableState.endEmitted;
5354
const onend = () => {
5455
readable = false;
56+
readableEnded = true;
5557
if (!writable) callback.call(stream);
5658
};
5759

@@ -60,11 +62,16 @@ function eos(stream, opts, callback) {
6062
};
6163

6264
const onclose = () => {
63-
if (readable && !(rs && rs.ended)) {
64-
return callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
65+
let err;
66+
if (readable && !readableEnded) {
67+
if (!stream._readableState || !stream._readableState.ended)
68+
err = new ERR_STREAM_PREMATURE_CLOSE();
69+
return callback.call(stream, err);
6570
}
66-
if (writable && !(ws && ws.ended)) {
67-
return callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
71+
if (writable && !writableEnded) {
72+
if (!stream._writableState || !stream._writableState.ended)
73+
err = new ERR_STREAM_PREMATURE_CLOSE();
74+
return callback.call(stream, err);
6875
}
6976
};
7077

@@ -77,7 +84,7 @@ function eos(stream, opts, callback) {
7784
stream.on('abort', onclose);
7885
if (stream.req) onrequest();
7986
else stream.on('request', onrequest);
80-
} else if (writable && !ws) { // legacy streams
87+
} else if (writable && !stream._writableState) { // legacy streams
8188
stream.on('end', onlegacyfinish);
8289
stream.on('close', onlegacyfinish);
8390
}
Collapse file

‎test/parallel/parallel.status‎

Copy file name to clipboardExpand all lines: test/parallel/parallel.status
-2Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,6 @@ test-net-connect-options-port: PASS,FLAKY
1212
test-http2-pipe: PASS,FLAKY
1313
test-worker-syntax-error: PASS,FLAKY
1414
test-worker-syntax-error-file: PASS,FLAKY
15-
# https://github.com/nodejs/node/issues/24456
16-
test-stream-pipeline-http2: PASS,FLAKY
1715

1816
[$system==linux]
1917

Collapse file

‎test/parallel/test-stream-pipeline-queued-end-in-destroy.js‎

Copy file name to clipboard
+39Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
'use strict';
2+
const common = require('../common');
3+
const assert = require('assert');
4+
const { Readable, Duplex, pipeline } = require('stream');
5+
6+
// Test that the callback for pipeline() is called even when the ._destroy()
7+
// method of the stream places an .end() request to itself that does not
8+
// get processed before the destruction of the stream (i.e. the 'close' event).
9+
// Refs: https://github.com/nodejs/node/issues/24456
10+
11+
const readable = new Readable({
12+
read: common.mustCall(() => {})
13+
});
14+
15+
const duplex = new Duplex({
16+
write(chunk, enc, cb) {
17+
// Simulate messages queueing up.
18+
},
19+
read() {},
20+
destroy(err, cb) {
21+
// Call end() from inside the destroy() method, like HTTP/2 streams
22+
// do at the time of writing.
23+
this.end();
24+
cb(err);
25+
}
26+
});
27+
28+
duplex.on('finished', common.mustNotCall());
29+
30+
pipeline(readable, duplex, common.mustCall((err) => {
31+
assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
32+
}));
33+
34+
// Write one chunk of data, and destroy the stream later.
35+
// That should trigger the pipeline destruction.
36+
readable.push('foo');
37+
setImmediate(() => {
38+
readable.destroy();
39+
});

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.