http: fix the reverted commit which would swallow some errors

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') {

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);

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,

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();
+}));

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();
+}));

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();
+}));