diff --git a/lib/_http_client.js b/lib/_http_client.js index 9eb3c10547d520..484723b2634fe4 100644 --- a/lib/_http_client.js +++ b/lib/_http_client.js @@ -113,6 +113,8 @@ function emitErrorEvent(request, error) { } const { addAbortSignal, finished } = require('stream'); +const { setImmediate } = require('timers'); +const { kBeforeWriteError } = require('internal/streams/utils'); let debug = require('internal/util/debuglog').debuglog('http', (fn) => { debug = fn; @@ -752,10 +754,23 @@ function socketErrorListener(err) { debug('SOCKET ERROR:', err.message, err.stack); if (req) { + const res = req.res; + const exchangeComplete = req.writableFinished && res?.complete; + const isUserDestroyError = err === req[kError] || err === res?.errored; + // For Safety. Some additional errors might fire later on // and we need to make sure we don't double-fire the error event. socket._hadError = true; - emitErrorEvent(req, err); + // Once both messages are complete, a transport error cannot change the + // result of the exchange. User-provided destroy errors are always emitted. + if (!exchangeComplete || isUserDestroyError) { + emitErrorEvent(req, err); + } + + if (res) { + socket.destroy(); + return; + } } const parser = socket.parser; @@ -770,6 +785,21 @@ function socketErrorListener(err) { socket.destroy(); } +function deferSocketWriteError(socket, err, continueWriteError) { + const req = socket._httpMessage; + + if (!req || req.destroyed || socket.destroyed || !socket.parser || + err.syscall !== 'write' || + req.res?.complete || err === req[kError] || err === req.res?.errored) { + return false; + } + + // Let readable events from the current poll cycle reach the HTTP parser + // before the write error tears down both sides of the socket. + setImmediate(continueWriteError); + return true; +} + function socketOnEnd() { const socket = this; const req = this._httpMessage; @@ -1083,6 +1113,7 @@ function tickOnSocket(req, socket) { socket.parser = parser; socket._httpMessage = req; + socket[kBeforeWriteError] = deferSocketWriteError; // Propagate headers limit from request object to parser if (typeof req.maxHeadersCount === 'number') { diff --git a/lib/internal/stream_base_commons.js b/lib/internal/stream_base_commons.js index 6d144f8a0fa696..01c8d631c0b2f4 100644 --- a/lib/internal/stream_base_commons.js +++ b/lib/internal/stream_base_commons.js @@ -26,6 +26,7 @@ const { getTimerDuration, } = require('internal/timers'); const { isUint8Array } = require('internal/util/types'); +const { kBeforeWriteError } = require('internal/streams/utils'); const { clearTimeout } = require('timers'); const { validateFunction } = require('internal/validators'); @@ -86,7 +87,14 @@ function onWriteComplete(status) { if (status < 0) { const error = new ErrnoException(status, 'write', this.error); if (typeof this.callback === 'function') { - return this.callback(error); + const callback = this.callback; + const beforeWriteError = stream[kBeforeWriteError]; + if (this.error === undefined && + beforeWriteError?.(stream, error, () => callback(error))) { + return; + } + + return callback(error); } return stream.destroy(error); diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js index 45f55316104fe8..830bff9891dab0 100644 --- a/lib/internal/streams/utils.js +++ b/lib/internal/streams/utils.js @@ -18,6 +18,7 @@ const kIsWritable = SymbolFor('nodejs.stream.writable'); const kIsDisturbed = SymbolFor('nodejs.stream.disturbed'); const kOnConstructed = Symbol('kOnConstructed'); +const kBeforeWriteError = Symbol('kBeforeWriteError'); const kIsClosedPromise = SymbolFor('nodejs.webstream.isClosedPromise'); const kControllerErrorFunction = SymbolFor('nodejs.webstream.controllerErrorFunction'); @@ -316,6 +317,7 @@ function isErrored(stream) { } module.exports = { + kBeforeWriteError, kOnConstructed, isDestroyed, kIsDestroyed, diff --git a/test/parallel/test-http-client-complete-response-open-request-reset.js b/test/parallel/test-http-client-complete-response-open-request-reset.js new file mode 100644 index 00000000000000..7ce3d7ee063bc4 --- /dev/null +++ b/test/parallel/test-http-client-complete-response-open-request-reset.js @@ -0,0 +1,49 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); + +const BODY = Buffer.alloc(1024 * 1024); + +const server = http.createServer(common.mustCall((request, response) => { + response.writeHead(202, { 'content-length': 0 }); + response.end(); + + request.once('data', common.mustCall(() => { + request.socket.resetAndDestroy(); + })); +})); + +server.on('clientError', common.mustNotCall()); + +server.listen(0, common.mustCall(() => { + let response; + const req = http.request({ + method: 'POST', + port: server.address().port, + headers: { 'content-length': BODY.length }, + }, common.mustCall((res) => { + response = res; + assert.strictEqual(res.statusCode, 202); + assert.strictEqual(req.writableEnded, false); + req.write(BODY); + })); + + req.on('error', common.mustCall((err) => { + assert(response); + assert.strictEqual(response.complete, true); + assert.strictEqual(req.writableFinished, false); + switch (err.code) { + case 'ECONNRESET': + case 'ECONNABORTED': + case 'EPIPE': + break; + default: + assert.fail(`Unexpected error code ${err.code}`); + } + })); + + req.on('close', common.mustCall(() => server.close())); + req.flushHeaders(); +})); diff --git a/test/parallel/test-http-client-early-response-reset-after-request-end.js b/test/parallel/test-http-client-early-response-reset-after-request-end.js new file mode 100644 index 00000000000000..3479a6cb254f93 --- /dev/null +++ b/test/parallel/test-http-client-early-response-reset-after-request-end.js @@ -0,0 +1,64 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); + +const MAX_BODY_LENGTH = 1024 * 1024; +const BODY = JSON.stringify({ + a: 'x'.repeat(512 * 1024), + b: 'x'.repeat(512 * 1024), + c: 'x'.repeat(512 * 1024), + d: 'x'.repeat(512 * 1024), +}, null, '\t'); + +const server = http.createServer(common.mustCall((request, response) => { + let received = 0; + + request.setEncoding('utf8'); + request.on('data', function onData(chunk) { + received += chunk.length; + if (received > MAX_BODY_LENGTH) { + response.writeHead(413, { 'content-length': 0 }); + response.end(); + request.off('data', onData); + request.destroy(); + } + }); + + request.on('end', common.mustNotCall()); + request.on('close', common.mustCall(() => { + assert.ok(received > MAX_BODY_LENGTH); + assert.strictEqual(request.complete, false); + })); +})); + +server.on('clientError', common.mustNotCall()); + +server.listen(0, common.mustCall(() => { + let response; + const req = http.request({ + method: 'POST', + port: server.address().port, + headers: { + 'content-type': 'application/json', + }, + }, common.mustCall((res) => { + response = res; + assert.strictEqual(res.statusCode, 413); + assert.strictEqual(req.writableEnded, true); + })); + + req.on('error', common.mustNotCall()); + req.on('close', common.mustCall(() => { + assert(response); + assert.strictEqual(req.writableFinished, true); + assert.strictEqual(response.complete, true); + server.close(); + })); + + // Preserve the ordering from the reported regression: the request body is + // written and ended before the response is received. + req.write(BODY); + req.end(); +})); diff --git a/test/parallel/test-http-client-incomplete-response-reset.js b/test/parallel/test-http-client-incomplete-response-reset.js new file mode 100644 index 00000000000000..e48564c6151e25 --- /dev/null +++ b/test/parallel/test-http-client-incomplete-response-reset.js @@ -0,0 +1,57 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); + +// A response that is cut short by a connection reset must still surface the +// socket error on the request. The response exists, but it is not complete, so +// swallowing the error would leave the application with a silently truncated +// body it believes is intact. + +const LENGTH = 1024; + +let serverSocket; + +const server = http.createServer(common.mustCall((request, response) => { + serverSocket = request.socket; + // Promise more body than is ever sent. + response.writeHead(200, { 'content-length': LENGTH }); + response.write('hello'); +})); + +server.on('clientError', common.mustNotCall()); + +server.listen(0, common.mustCall(() => { + const req = http.request({ port: server.address().port }, common.mustCall((res) => { + assert.strictEqual(res.statusCode, 200); + + let received = 0; + let reset = false; + + res.on('data', common.mustCallAtLeast((chunk) => { + received += chunk.length; + if (reset) return; + reset = true; + // The headers and part of the body have arrived. Reset from the server + // side so the client sees a genuine inbound RST mid-body. + serverSocket.resetAndDestroy(); + }, 1)); + + res.on('end', common.mustNotCall()); + res.on('close', common.mustCall(() => { + assert.strictEqual(res.complete, false); + assert.strictEqual(res.errored.code, 'ECONNRESET'); + assert.strictEqual(res.errored.message, 'aborted'); + assert.ok(received > 0 && received < LENGTH, + `expected a truncated body, got ${received} of ${LENGTH}`); + server.close(); + })); + })); + + req.on('error', common.mustCall((err) => { + assert.strictEqual(err.code, 'ECONNRESET'); + })); + + req.end(); +}));