-
Notifications
You must be signed in to change notification settings - Fork 29.8k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
http: invoke callback with ERR_STREAM_DESTROYED if the socket is destroyed #36692
Conversation
ping @ronag |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
commit description is wrong, should be invoke callback with ERR_STREAM_DESTROYED
a758806
to
9ebd6c3
Compare
9ebd6c3
to
2c2d3c3
Compare
2c2d3c3
to
c185bc3
Compare
This comment has been minimized.
This comment has been minimized.
c185bc3
to
92e2750
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I kind of think this check should live directly in .end() and .write() and not (only) in writeRaw or lot's of other side effects might apply before the check.
92e2750
to
110047e
Compare
The There is also a small question: should we move the |
I'm unsure whether we need the check in |
I need a little time to improve the code and add unit tests, the situation is a little more complicated than expected |
110047e
to
9024f44
Compare
From this PR (#31818), both Lines 725 to 730 in b12127a
We should only need to check |
I think something along these lines would be good and quite close to what writable.js does: diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js
index 50ef063241..6e6ab8eb92 100644
--- a/lib/_http_outgoing.js
+++ b/lib/_http_outgoing.js
@@ -51,10 +51,6 @@ const { Buffer } = require('buffer');
const common = require('_http_common');
const checkIsHttpToken = common._checkIsHttpToken;
const checkInvalidHeaderChar = common._checkInvalidHeaderChar;
-const {
- defaultTriggerAsyncIdScope,
- symbols: { async_id_symbol }
-} = require('internal/async_hooks');
const {
codes: {
ERR_HTTP_HEADERS_SENT,
@@ -689,23 +685,6 @@ OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {
return ret;
};
-function onError(msg, err, callback) {
- const triggerAsyncId = msg.socket ? msg.socket[async_id_symbol] : undefined;
- defaultTriggerAsyncIdScope(triggerAsyncId,
- process.nextTick,
- emitErrorNt,
- msg,
- err,
- callback);
-}
-
-function emitErrorNt(msg, err, callback) {
- callback(err);
- if (typeof msg.emit === 'function' && !msg._closed) {
- msg.emit('error', err);
- }
-}
-
function write_(msg, chunk, encoding, callback, fromEnd) {
if (typeof callback !== 'function')
callback = nop;
@@ -730,11 +709,8 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
}
if (err) {
- if (!msg.destroyed) {
- onError(msg, err, callback);
- } else {
- process.nextTick(callback, err);
- }
+ process.nextTick(callback, err);
+ msg.destroy(err);
return false;
}
@@ -823,31 +799,37 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
this.socket.cork();
}
- if (chunk) {
- if (this.finished) {
- onError(this,
- new ERR_STREAM_WRITE_AFTER_END(),
- typeof callback !== 'function' ? nop : callback);
- return this;
+ if (chunk !== null && chunk !== undefined)
+ this.write(chunk, encoding);
+
+ let err;
+ if (this.writableFinished) {
+ err = new ERR_STREAM_ALREADY_FINISHED('end');
+ } else if (this.destroyed) {
+ err = new ERR_STREAM_DESTROYED('end');
+ }
+
+ if (typeof callback === 'function') {
+ if (err || this.writableFinished) {
+ process.nextTick(callback, err);
+ } else {
+ // TODO (fix): What if error?
+ this.once('finish', 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'));
- }
+ }
+
+ if (err) {
+ if (this.socket) {
+ this.socket.uncork();
}
return this;
- } else if (!this._header) {
+ }
+
+ if (!this._header) {
this._contentLength = 0;
this._implicitHeader();
}
- if (typeof callback === 'function')
- this.once('finish', callback);
-
const finish = FunctionPrototypeBind(onFinish, undefined, this);
if (this._hasBody && this.chunkedEncoding) { |
A little more work: diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js
index 50ef063241..b2b3137235 100644
--- a/lib/_http_outgoing.js
+++ b/lib/_http_outgoing.js
@@ -51,10 +51,6 @@ const { Buffer } = require('buffer');
const common = require('_http_common');
const checkIsHttpToken = common._checkIsHttpToken;
const checkInvalidHeaderChar = common._checkInvalidHeaderChar;
-const {
- defaultTriggerAsyncIdScope,
- symbols: { async_id_symbol }
-} = require('internal/async_hooks');
const {
codes: {
ERR_HTTP_HEADERS_SENT,
@@ -689,23 +685,6 @@ OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {
return ret;
};
-function onError(msg, err, callback) {
- const triggerAsyncId = msg.socket ? msg.socket[async_id_symbol] : undefined;
- defaultTriggerAsyncIdScope(triggerAsyncId,
- process.nextTick,
- emitErrorNt,
- msg,
- err,
- callback);
-}
-
-function emitErrorNt(msg, err, callback) {
- callback(err);
- if (typeof msg.emit === 'function' && !msg._closed) {
- msg.emit('error', err);
- }
-}
-
function write_(msg, chunk, encoding, callback, fromEnd) {
if (typeof callback !== 'function')
callback = nop;
@@ -730,11 +709,8 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
}
if (err) {
- if (!msg.destroyed) {
- onError(msg, err, callback);
- } else {
- process.nextTick(callback, err);
- }
+ process.nextTick(callback, err);
+ msg.destroy(err);
return false;
}
@@ -809,13 +785,13 @@ function onFinish(outmsg) {
outmsg.emit('finish');
}
-OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
+OutgoingMessage.prototype.end = function end(chunk, encoding, cb) {
if (typeof chunk === 'function') {
- callback = chunk;
+ cb = chunk;
chunk = null;
encoding = null;
} else if (typeof encoding === 'function') {
- callback = encoding;
+ cb = encoding;
encoding = null;
}
@@ -823,38 +799,46 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
this.socket.cork();
}
- if (chunk) {
- if (this.finished) {
- onError(this,
- new ERR_STREAM_WRITE_AFTER_END(),
- typeof callback !== 'function' ? nop : callback);
- return this;
+ if (chunk !== null && chunk !== undefined)
+ this.write(chunk, encoding);
+
+ let err;
+ if (!this.finished) {
+ if (!this._header) {
+ this._contentLength = 0;
+ this._implicitHeader();
}
- 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'));
- }
+
+ const finish = FunctionPrototypeBind(onFinish, undefined, this);
+
+ if (this._hasBody && this.chunkedEncoding) {
+ this._send('0\r\n' + this._trailer + '\r\n', 'latin1', finish);
+ } else {
+ // Force a flush, HACK.
+ this._send('', 'latin1', finish);
}
- return this;
- } else if (!this._header) {
- this._contentLength = 0;
- this._implicitHeader();
- }
- if (typeof callback === 'function')
- this.once('finish', callback);
+ this.finished = true; // aka. WritableState.ended
+ } else if (this.writableFinished) {
+ err = new ERR_STREAM_ALREADY_FINISHED('end');
+ } else if (this.destroyed) {
+ err = new ERR_STREAM_DESTROYED('end');
+ }
- const finish = FunctionPrototypeBind(onFinish, undefined, this);
+ if (typeof cb === 'function') {
+ if (err || this.writableFinished) {
+ process.nextTick(cb, err);
+ } else {
+ // TODO (fix): What if error? See kOnFinished in writable.js.
+ this.once('finish', cb);
+ }
+ }
- if (this._hasBody && this.chunkedEncoding) {
- this._send('0\r\n' + this._trailer + '\r\n', 'latin1', finish);
- } else {
- // Force a flush, HACK.
- this._send('', 'latin1', finish);
+ if (err) {
+ if (this.socket) {
+ this.socket.uncork();
+ }
+ return this;
}
if (this.socket) {
@@ -864,14 +848,10 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
}
this[kCorked] = 0;
- this.finished = true;
-
// There is the first message on the outgoing queue, and we've sent
// everything to the socket.
debug('outgoing message end.');
- if (this.outputData.length === 0 &&
- this.socket &&
- this.socket._httpMessage === this) {
+ if (this.outputData.length === 0 && this.socket?._httpMessage === this) {
this._finish();
} |
Sorry if I made the scope on this task a lot bigger. But some of the question you had kind of lead to one thing and then another... let me know if you wish to collaborate on this. Basically our goal here is to make |
I wish to collaborate on this, I've learned a lot from this pull request and your guidance, and it's been a lot of fun.
Should we split it into multiple Pull Request, or just implement it in this pull request? |
Do you think you can try to apply my suggestion and see if and what test might fail? We should sort those out first. Then we need to add some new tests. Once the above is done I can take a look and make some TODO's for stuff that needs test and then you can try to make those? I'll help out if you get stuck with any. |
I think it's easier to do it here since a lot of the changes are dependent. |
5 unit tests failed at local. Detail
|
Can you push the changes? Do you want to take a try at fixing the tests? |
9024f44
to
4ebf579
Compare
The code has been pushed.
Yes. I would like to try fixing the tests |
4ebf579
to
bbafac6
Compare
Close this Pull Request due to outdated code |
Fixes: #36673
Refs: #29227 (comment)
Checklist
make -j4 test
(UNIX), orvcbuild test
(Windows) passes