Update after review

Primarily fixing error handling and improving the tests

Author: pimterry

doc/api/http.md

@@ -1713,7 +1713,7 @@ change.
 In the uncommon case that the incoming request has a body, this body will be
 parsed as normal separate to the upgrade stream, and the raw socket data will
 only begin after it has completed. To ensure that reading from the socket isn't
-blocked bv waiting for the request body to be read, any reads on the socket
+blocked by waiting for the request body to be read, any reads on the socket
 will start the request body flowing automatically. If you want to read the
 request body, ensure that you do so (i.e. you attach `'data'` listeners)
 before starting to read from the upgraded socket.

lib/_http_server.js

@@ -971,7 +971,6 @@ class UpgradeStream extends Duplex {
 
     // Proxy error, end & closure events immediately.
     socket.on('error', (err) => this.destroy(err));
-    this.on('error', (err) => socket.destroy(err));
 
     socket.on('close', () => this.destroy());
     this.on('close', () => socket.destroy());
@@ -1052,6 +1051,11 @@ class UpgradeStream extends Duplex {
   _write(chunk, encoding, callback) {
     this[kSocket].write(chunk, encoding, callback);
   }
+
+  _destroy(err, callback) {
+    this[kSocket].destroy(err);
+    callback(err);
+  }
 }
 
 function onParserExecuteCommon(server, socket, parser, state, ret, d) {
@@ -1095,7 +1099,7 @@ function onParserExecuteCommon(server, socket, parser, state, ret, d) {
         return;
       }
 
-      d ||= Buffer.from([]);
+      d ||= Buffer.alloc(0);
 
       upgradeStream = new UpgradeStream(socket, req);
       socket[kUpgradeStream] = upgradeStream;

test/parallel/test-http-upgrade-server-with-body-error.mjs

@@ -0,0 +1,46 @@
+import * as common from '../common/index.mjs';
+import * as assert from 'assert';
+import * as http from 'http';
+import * as net from 'net';
+
+const upgradeReceivedResolvers = Promise.withResolvers();
+
+const server = http.createServer();
+server.on('request', common.mustNotCall());
+server.on('upgrade', function(req, socket, upgradeHead) {
+  upgradeReceivedResolvers.resolve();
+
+  // As soon as the body starts arriving, simulate an error
+  req.on('data', () => {
+    req.socket.destroy(new Error('simulated body error'));
+  });
+});
+
+await new Promise((resolve) => server.listen(0, () => resolve()));
+
+const conn = net.createConnection(server.address().port);
+conn.setEncoding('utf8');
+
+await new Promise((resolve) => conn.on('connect', resolve));
+
+// Write request headers, leave the body pending:
+conn.write(
+  'POST / HTTP/1.1\r\n' +
+  'host: localhost\r\n' +
+  'Upgrade: custom-protocol\r\n' +
+  'Connection: Upgrade\r\n' +
+  'transfer-encoding: chunked\r\n' +
+  '\r\n'
+);
+
+// Make sure the server has processed the above & fired 'upgrade' before the body
+// data starts streaming:
+await upgradeReceivedResolvers.promise;
+
+conn.write('5\r\nhello\r\n');
+
+process.on('uncaughtException', common.mustCall((err) => {
+  assert.strictEqual(err.message, 'simulated body error');
+  conn.destroy();
+  server.close();
+}));

test/parallel/test-http-upgrade-server-with-body.mjs

@@ -3,207 +3,67 @@ import * as assert from 'assert';
 import * as http from 'http';
 import * as net from 'net';
 
-async function createServer({ expectedBodyLength }) {
-  const server = http.createServer();
-
-  server.on('request', common.mustNotCall());
-
-  server.on('upgrade', function(req, socket, upgradeHead) {
-    // Confirm the upgrade:
-    socket.write('HTTP/1.1 101 Switching Protocols\r\n' +
-                'Upgrade: custom-protocol\r\n' +
-                'Connection: Upgrade\r\n' +
-                '\r\n');
-
-    // Read and validate the request body, if and only if expectedBodyLength is set:
-    if (expectedBodyLength !== undefined) {
-      let reqBody = '';
-      req.on('data', (str) => { reqBody += str; });
-      req.on('end', common.mustCall(() => {
-        assert.strictEqual(reqBody.length, expectedBodyLength);
+const EXPECTED_BODY_LENGTH = 12;
+
+const server = http.createServer();
+server.on('request', common.mustNotCall());
+server.on('upgrade', function(req, socket, upgradeHead) {
+  // Confirm the upgrade:
+  socket.write('HTTP/1.1 101 Switching Protocols\r\n' +
+              'Upgrade: custom-protocol\r\n' +
+              'Connection: Upgrade\r\n' +
+              '\r\n');
+
+  // Read and validate the request body:
+  let reqBodyLength = 0;
+  req.on('data', (chunk) => { reqBodyLength += chunk.length; });
+  req.on('end', common.mustCall(() => {
+    assert.strictEqual(reqBodyLength, EXPECTED_BODY_LENGTH);
+
+    // Defer upgrade stream read slightly to make sure it doesn't start
+    // streaming along with the request body, until we actually read it:
+    setTimeout(common.mustCall(() => {
+      let socketData = upgradeHead;
+
+      socket.on('data', common.mustCall((d) => {
+        socketData = Buffer.concat([socketData, d]);
       }));
-    }
-
-    let socketData = upgradeHead;
-
-    // Collect the raw protocol data and check it:
-    socket.on('data', common.mustCall((d) => {
-      socketData = Buffer.concat([socketData, d]);
-    }));
-
-    socket.on('end', common.mustCall(() => {
-      assert.strictEqual(socketData.toString(), 'upgrade head\npost-upgrade message');
-      socket.end();
-    }));
-  });
-
-  await new Promise((resolve) => server.listen(0, () => resolve()));
-
-  return server;
-}
-
-async function testUpgradeWithBody() {
-  const server = await createServer({ expectedBodyLength: 12 });
-
-  const conn = net.createConnection(server.address().port);
-  conn.setEncoding('utf8');
-
-  await new Promise((resolve) => conn.on('connect', resolve));
-
-  // Write request headers, body & upgrade head all together:
-  conn.write(
-    'POST / HTTP/1.1\r\n' +
-    'host: localhost\r\n' +
-    'Upgrade: custom-protocol\r\n' +
-    'Connection: Upgrade\r\n' +
-    'transfer-encoding: chunked\r\n' +
-    '\r\n' +
-    'C\r\nrequest body\r\n' + // 12 byte body sent immediately
-    '0\r\n\r\n' +
-    'upgrade head'
-  );
-
-  const response = await new Promise((resolve) => conn.once('data', resolve));
-  assert.ok(response.startsWith('HTTP/1.1 101 Switching Protocols\r\n'));
-
-  // Send more data after connection is confirmed:
-  conn.write('\npost-upgrade message');
-  conn.end();
-
-  await new Promise((resolve) => conn.on('end', resolve));
-
-  server.close();
-}
-
-async function testUpgradeWithLargeBody() {
-  const server = await createServer({ expectedBodyLength: 100_000 });
-
-  const conn = net.createConnection(server.address().port);
-  conn.setEncoding('utf8');
-
-  await new Promise((resolve) => conn.on('connect', resolve));
-
-  // Write request headers, leave the body pending:
-  conn.write(
-    'POST / HTTP/1.1\r\n' +
-    'host: localhost\r\n' +
-    'Upgrade: custom-protocol\r\n' +
-    'Connection: Upgrade\r\n' +
-    'transfer-encoding: chunked\r\n' +
-    '\r\n'
-  );
-
-  await new Promise((resolve) => setTimeout(resolve, 10));
 
-  // Write the large body and part of the initial upgrade data.
-  for (let i = 0; i < 20_000; i++) {
-    conn.write('5\r\nhello\r\n'); // 10 byte chunk = 5 byte body
-  }
-
-  // End body and then send upgrade head
-  await new Promise((resolve) => setTimeout(resolve, 10));
-
-  conn.write(
-    '0\r\n\r\n' +
-    'upgrade head'
-  );
-
-  const response = await new Promise((resolve) => conn.once('data', resolve));
-  assert.ok(response.startsWith('HTTP/1.1 101 Switching Protocols\r\n'));
-
-  // Send more data after connection is confirmed:
-  conn.write('\npost-upgrade message');
-  conn.end();
-
-  await new Promise((resolve) => conn.on('end', resolve));
-
-  server.close();
-}
-
-async function testUpgradeWithUnreadBody() {
-  const server = await createServer({
-    expectedBodyLength: undefined // Don't read the request body at all
-  });
-
-  const conn = net.createConnection(server.address().port);
-  conn.setEncoding('utf8');
-
-  await new Promise((resolve) => conn.on('connect', resolve));
-
-  // Write request headers, body & upgrade head all together:
-  conn.write(
-    'POST / HTTP/1.1\r\n' +
-    'host: localhost\r\n' +
-    'Upgrade: custom-protocol\r\n' +
-    'Connection: Upgrade\r\n' +
-    'transfer-encoding: chunked\r\n' +
-    '\r\n' +
-    'C\r\nrequest body\r\n' + // 12 byte body sent immediately
-    '0\r\n\r\n' +
-    'upgrade head'
-  );
-
-  const response = await new Promise((resolve) => conn.once('data', resolve));
-  assert.ok(response.startsWith('HTTP/1.1 101 Switching Protocols\r\n'));
-
-  // Send more data after connection is confirmed:
-  conn.write('\npost-upgrade message');
-  conn.end();
-
-  await new Promise((resolve) => conn.on('end', resolve));
-
-  server.close();
-}
-
-async function testUpgradeWithUnreadLargeBody() {
-  const server = await createServer({
-    expectedBodyLength: undefined // Don't read the request body at all
-  });
-
-  const conn = net.createConnection(server.address().port);
-  conn.setEncoding('utf8');
-
-  await new Promise((resolve) => conn.on('connect', resolve));
-
-  // Write request headers, leave the body pending:
-  conn.write(
-    'POST / HTTP/1.1\r\n' +
-    'host: localhost\r\n' +
-    'Upgrade: custom-protocol\r\n' +
-    'Connection: Upgrade\r\n' +
-    'transfer-encoding: chunked\r\n' +
-    '\r\n'
-  );
-
-  await new Promise((resolve) => setTimeout(resolve, 10));
-
-  // Write the large body and part of the initial upgrade data.
-  for (let i = 0; i < 20_000; i++) {
-    conn.write('5\r\nhello\r\n'); // 10 byte chunk = 5 byte body
-  }
+      socket.on('end', common.mustCall(() => {
+        assert.strictEqual(socketData.toString(), 'upgrade head\npost-upgrade message');
+        socket.end();
+      }));
+    }), 10);
+  }));
+});
 
-  // End body and then send upgrade head
-  await new Promise((resolve) => setTimeout(resolve, 10));
+await new Promise((resolve) => server.listen(0, () => resolve()));
 
-  conn.write(
-    '0\r\n\r\n' +
-    'upgrade head'
-  );
+const conn = net.createConnection(server.address().port);
+conn.setEncoding('utf8');
 
-  const response = await new Promise((resolve) => conn.once('data', resolve));
-  assert.ok(response.startsWith('HTTP/1.1 101 Switching Protocols\r\n'));
+await new Promise((resolve) => conn.on('connect', resolve));
 
-  // Send more data after connection is confirmed:
-  conn.write('\npost-upgrade message');
-  conn.end();
+// Write request headers, body & upgrade head all together:
+conn.write(
+  'POST / HTTP/1.1\r\n' +
+  'host: localhost\r\n' +
+  'Upgrade: custom-protocol\r\n' +
+  'Connection: Upgrade\r\n' +
+  'transfer-encoding: chunked\r\n' +
+  '\r\n' +
+  'C\r\nrequest body\r\n' + // 12 byte body sent immediately
+  '0\r\n\r\n' +
+  'upgrade head'
+);
 
-  await new Promise((resolve) => conn.on('end', resolve));
+const response = await new Promise((resolve) => conn.once('data', resolve));
+assert.ok(response.startsWith('HTTP/1.1 101 Switching Protocols\r\n'));
 
-  server.close();
+// Send more data after connection is confirmed:
+conn.write('\npost-upgrade message');
+conn.end();
 
-}
+await new Promise((resolve) => conn.on('end', resolve));
 
-await testUpgradeWithBody();
-await testUpgradeWithLargeBody();
-await testUpgradeWithUnreadBody();
-await testUpgradeWithUnreadLargeBody();
+server.close();