http: coalesce chunked writes during auto-corking

lib/_http_outgoing.js

@@ -67,6 +67,7 @@ const {
     ERR_STREAM_DESTROYED,
     ERR_STREAM_NULL_VALUES,
     ERR_STREAM_WRITE_AFTER_END,
+    ERR_UNKNOWN_ENCODING,
   },
   hideStackFrames,
 } = require('internal/errors');
@@ -228,7 +229,11 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableObjectMode', {
 ObjectDefineProperty(OutgoingMessage.prototype, 'writableLength', {
   __proto__: null,
   get() {
-    return this.outputSize + this[kChunkedLength] + (this[kSocket] ? this[kSocket].writableLength : 0);
+    let bufferedLength = this[kChunkedLength];
+    if (bufferedLength !== 0) {
+      bufferedLength += bufferedLength.toString(16).length + 4 + (!this._headerSent && this._header !== null ? this._header.length : 0);
+    }
+    return this.outputSize + bufferedLength + (this[kSocket] ? this[kSocket].writableLength : 0);
   },
 });
 
@@ -299,37 +304,48 @@ OutgoingMessage.prototype.cork = function cork() {
 
 OutgoingMessage.prototype.uncork = function uncork() {
   this[kCorked]--;
-  if (this[kSocket]) {
-    this[kSocket].uncork();
-  }
-
-  if (this[kCorked] || this[kChunkedBuffer].length === 0) {
-    return;
-  }
-
-  const len = this[kChunkedLength];
-  const buf = this[kChunkedBuffer];
-
-  assert(this.chunkedEncoding);
+  const socket = this[kSocket];
+  try {
+    if (this[kCorked] || this[kChunkedBuffer].length === 0) {
+      return;
+    }
 
-  let callbacks;
-  this._send(len.toString(16), 'latin1', null);
-  this._send(crlf_buf, null, null);
-  for (let n = 0; n < buf.length; n += 3) {
-    this._send(buf[n + 0], buf[n + 1], null);
-    if (buf[n + 2]) {
-      callbacks ??= [];
-      callbacks.push(buf[n + 2]);
+    const len = this[kChunkedLength];
+    const buf = this[kChunkedBuffer];
+    if (this.destroyed || socket?.destroyed) {
+      const error = this[kErrored] || socket?._writableState?.errored || new ERR_STREAM_DESTROYED('write');
+      for (let n = 2; n < buf.length; n += 3) {
+        if (buf[n] !== nop) {
+          process.nextTick(buf[n], error);
+        }
+      }
+      buf.length = 0;
+      this[kChunkedLength] = 0;
+      return;
     }
-  }
-  this._send(crlf_buf, null, callbacks.length ? (err) => {
-    for (const callback of callbacks) {
-      callback(err);
+    assert(this.chunkedEncoding);
+
+    let callbacks;
+    this._send(len.toString(16), 'latin1', null);
+    this._send(crlf_buf, null, null);
+    for (let n = 0; n < buf.length; n += 3) {
+      this._send(buf[n + 0], buf[n + 1], null);
+      if (buf[n + 2]) {
+        callbacks ??= [];
+        callbacks.push(buf[n + 2]);
+      }
     }
-  } : null);
+    this._send(crlf_buf, null, callbacks.length ? (err) => {
+      for (const callback of callbacks) {
+        callback(err);
+      }
+    } : null);
 
-  this[kChunkedBuffer].length = 0;
-  this[kChunkedLength] = 0;
+    this[kChunkedBuffer].length = 0;
+    this[kChunkedLength] = 0;
+  } finally {
+    socket?.uncork();
+  }
 
   // If we had a pending drain and flushed all data, emit the drain event.
   if (this[kNeedDrain] && this.writableLength === 0) {
@@ -365,6 +381,10 @@ OutgoingMessage.prototype.destroy = function destroy(error) {
   this.destroyed = true;
 
   this[kErrored] = error;
+  if (this[kChunkedBuffer].length !== 0) {
+    this[kCorked] = 1;
+    this.uncork();
+  }
 
   if (this[kSocket]) {
     this[kSocket].destroy(error);
@@ -1004,24 +1024,34 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
     }
   }
 
+  if (msg.chunkedEncoding && typeof chunk === 'string' && encoding && !Buffer.isEncoding(encoding)) {
+    throw new ERR_UNKNOWN_ENCODING(encoding);
+  }
+
   if (!fromEnd && msg.socket && !msg.socket.writableCorked) {
-    msg.socket.cork();
-    process.nextTick(connectionCorkNT, msg.socket);
+    msg.cork();
+    process.nextTick(connectionCorkNT, msg, msg.socket);
   }
 
   let ret;
   if (msg.chunkedEncoding && chunk.length !== 0) {
     len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength;
-    if (msg[kCorked] && msg._headerSent) {
+    if (msg[kCorked]) {
+      if (typeof chunk !== 'string' && !(chunk instanceof Buffer)) {
+        chunk = Stream._uint8ArrayToBuffer(chunk);
+      }
       msg[kChunkedBuffer].push(chunk, encoding, callback);
       msg[kChunkedLength] += len;
-      ret = msg[kChunkedLength] < msg[kHighWaterMark];
+      ret = msg.writableLength < msg.writableHighWaterMark;
     } else {
       msg._send(len.toString(16), 'latin1', null);
       msg._send(crlf_buf, null, null);
       msg._send(chunk, encoding, null, len);
       ret = msg._send(crlf_buf, null, callback);
     }
+  } else if (msg.chunkedEncoding && msg[kCorked] && msg[kChunkedBuffer].length !== 0) {
+    msg[kChunkedBuffer].push(chunk, encoding, callback);
+    ret = msg.writableLength < msg.writableHighWaterMark;
   } else {
     ret = msg._send(chunk, encoding, callback, len);
   }
@@ -1031,8 +1061,13 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
 }
 
 
-function connectionCorkNT(conn) {
-  conn.uncork();
+function connectionCorkNT(msg, conn) {
+  if (msg[kCorked]) {
+    msg.uncork();
+  }
+  if (msg[kSocket] !== conn) {
+    conn.uncork();
+  }
 }
 
 OutgoingMessage.prototype.addTrailers = function addTrailers(headers) {
@@ -1138,6 +1173,13 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
     throw new ERR_HTTP_CONTENT_LENGTH_MISMATCH(this[kBytesWritten], this._contentLength);
   }
 
+  // Flush buffered chunks before terminating the response.
+  if (this[kChunkedBuffer].length !== 0) {
+    this[kSocket]?.cork();
+    this[kCorked] = 1;
+    this.uncork();
+  }
+
   const finish = onFinish.bind(undefined, this);
 
   if (this._hasBody && this.chunkedEncoding) {

test/parallel/test-http-1.0.js

@@ -127,9 +127,12 @@ function test(handler, request_generator, response_validator) {
     assert.strictEqual(req.httpVersionMinor, 1);
     res.sendDate = false;
     res.writeHead(200, { 'Content-Type': 'text/plain' });
-    res.write('Hello, '); res._send('');
+    assert.throws(() => res.write('ignored', 'invalid'), {
+      code: 'ERR_UNKNOWN_ENCODING',
+    });
+    res.write('Hello, ', common.mustCall()); res._send('');
     res.write('world!'); res._send('');
-    res.end();
+    process.nextTick(() => res.end('X'));
   }
 
   function request_generator() {
@@ -148,10 +151,10 @@ function test(handler, request_generator, response_validator) {
                               'Connection: close\r\n' +
                               'Transfer-Encoding: chunked\r\n' +
                               '\r\n' +
-                              '7\r\n' +
-                              'Hello, \r\n' +
-                              '6\r\n' +
-                              'world!\r\n' +
+                              'd\r\n' +
+                              'Hello, world!\r\n' +
+                              '1\r\n' +
+                              'X\r\n' +
                               '0\r\n' +
                               '\r\n';
 

test/parallel/test-http-outgoing-destroyed.js

@@ -66,6 +66,9 @@ const { OutgoingMessage } = require('http');
       });
     }));
     const err = new Error('Destroy test');
+    res.write('x', common.mustCall((writeErr) => {
+      assert.strictEqual(writeErr, err);
+    }));
     res.destroy(err);
     assert.strictEqual(res.errored, err);
   })).listen(0, common.mustCall(() => {