http: defer write errors for pending responses

A transport write error can be delivered before a readable event from
the same poll cycle. Writable error handling then destroys both sides of
the socket before the HTTP parser can consume an already-sent response.

Defer native write errors that do not carry protocol-specific details.
After pending reads run, suppress the error only when the request write
and response parse are both complete. Continue reporting open writes,
truncated responses, user destroy errors, and TLS protocol errors.

Follow-up to: #64507
Original PR Refs: #64278
Fixes: #64272
Refs:#64511
Refs: libuv/libuv#5196
Refs: #64507 (comment)
Refs: #64511 (comment)

Signed-off-by: Archkon <180910180+Archkon@users.noreply.github.com>

Author: Archkon

lib/_http_client.js

@@ -116,6 +116,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;
@@ -767,10 +769,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;
@@ -785,6 +800,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;
@@ -1100,6 +1130,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();
+}));