node:http: make res._send() the stored head plus _writeRaw(), as in Node.js

packages/bun-uws/src/HttpResponse.h

@@ -570,11 +570,18 @@ struct HttpResponse : public AsyncSocket<SSL> {
         return this;
     }
 
-    /* Write a caller-built 1xx informational response line + headers. Shares
-     * the AsyncSocket write path (and buffer) writeStatus/end use, so a
-     * pipelined replay stays ordered ahead of the final response bytes. */
+    /* Write caller-built bytes as they are: a 1xx informational response line +
+     * headers, or the data of node:http's res._send(). Shares the AsyncSocket
+     * write path (and buffer) writeStatus/end use, so a pipelined replay stays
+     * ordered ahead of the final response bytes. */
     HttpResponse *writeRawInformational(std::string_view data) {
-        Super::write(data.data(), (int) data.length());
+        /* uSockets only deals with int sizes, so pass chunks of max signed int size.
+         * A write that is not optional buffers what the socket does not take. */
+        for (size_t written = 0; written < data.length();) {
+            int length = (int) std::min<size_t>(data.length() - written, INT_MAX);
+            Super::write(data.data() + written, length);
+            written += (size_t) length;
+        }
         return this;
     }
 

src/js/internal/http.ts

@@ -59,6 +59,10 @@ const http1ServerPipeline: {
   kMustCloseConnection?: symbol;
 } = {};
 
+// Set by node:_http_server. OutgoingMessage.prototype.write() and end() call them for a ServerResponse with a handle.
+type ServerResponseMethod = (this: unknown, ...args: unknown[]) => unknown;
+const serverResponseMethods: { write?: ServerResponseMethod; end?: ServerResponseMethod } = {};
+
 export const enum NodeHTTPResponseAbortEvent {
   none = 0,
   abort = 1,
@@ -558,6 +562,7 @@ export {
   parseProxyConfigFromEnv,
   parseProxyUrl,
   redactInvalidProxyUrl,
+  serverResponseMethods,
   serverSymbol,
   setMaxHTTPHeaderSize,
   setServerAppFlags,

src/js/internal/http1_server_fallback.ts

@@ -257,10 +257,14 @@ function createHttp1FallbackResponseHandle(socket, shouldKeepAlive, keepAliveTim
     writeContinue() {
       writeToSocket("HTTP/1.1 100 Continue\r\n\r\n");
     },
-    writeInformational(chunk, encoding) {
-      // _writeRaw hands the fully-rendered 1xx block here (writeEarlyHints /
-      // writeProcessing / writeInformation all route through it).
-      if (!socket.writableEnded) socket.write(chunk, encoding);
+    writeInformational(chunk, encoding, callback) {
+      // Raw bytes of _writeRaw(): a 1xx block, or the data of _send(). Like the native handle, -1 is for a socket that is gone.
+      if (this.aborted || socket.destroyed) return -1;
+      if (socket.writableEnded) {
+        if (typeof callback === "function") process.nextTick(callback);
+        return true;
+      }
+      return socket.write(chunk, encoding, callback) !== false;
     },
     writeHead(statusCode, statusMessage, headers, autoHeaderBits, keepAliveTimeoutSecs) {
       const originalStatusCode = statusCode;

src/js/node/_http_outgoing.ts

@@ -5,7 +5,14 @@ const { Stream } = require("internal/stream");
 const { isUint8Array, validateString } = require("internal/validators");
 const { deprecate } = require("internal/util/deprecate");
 const { getDefaultHighWaterMark } = require("internal/streams/state");
-const { kOutHeaders, kNeedDrain, utcDate } = require("internal/http");
+const {
+  kHandle,
+  kOutHeaders,
+  kNeedDrain,
+  kPendingCallbacks,
+  serverResponseMethods,
+  utcDate,
+} = require("internal/http");
 const {
   validateHeaderName,
   validateHeaderValue,
@@ -890,6 +897,16 @@ ObjectDefineProperty(OutgoingMessage.prototype, "writableNeedDrain", {
 
 const crlf_buf = Buffer.from("\r\n");
 OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {
+  // A ServerResponse with a handle has its own write(). The HTTP/1 fallback also gives a handle to other response classes.
+  if (this[kHandle] && this[kPendingCallbacks] !== undefined) {
+    // The checks of write_(), which this call does not reach.
+    if (chunk === null) throw $ERR_STREAM_NULL_VALUES();
+    if (typeof chunk !== "string" && !isUint8Array(chunk)) {
+      throw $ERR_INVALID_ARG_TYPE("chunk", ["string", "Buffer", "Uint8Array"], chunk);
+    }
+    return serverResponseMethods.write.$call(this, chunk, encoding, callback);
+  }
+
   if (typeof encoding === "function") {
     callback = encoding;
     encoding = null;
@@ -1064,6 +1081,14 @@ function onFinish(outmsg) {
 }
 
 OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
+  if (this[kHandle] && this[kPendingCallbacks] !== undefined) {
+    // Like below: after end(), the chunk is not looked at.
+    if (chunk && !this.finished && typeof chunk !== "function" && typeof chunk !== "string" && !isUint8Array(chunk)) {
+      throw $ERR_INVALID_ARG_TYPE("chunk", ["string", "Buffer", "Uint8Array"], chunk);
+    }
+    return serverResponseMethods.end.$call(this, chunk, encoding, callback);
+  }
+
   if (typeof chunk === "function") {
     callback = chunk;
     chunk = null;

src/js/node/_http_server.ts

@@ -62,6 +62,7 @@ const {
   kOutHeaders,
   onDataIncomingMessage,
   http1ServerPipeline,
+  serverResponseMethods,
 } = require("internal/http");
 const { FakeSocket } = require("internal/http/FakeSocket");
 const NumberIsNaN = Number.isNaN;
@@ -3001,10 +3002,7 @@ function advanceResponsePipeline(server, socket) {
         const op = ops[i];
         const kind = op[0];
         if (kind === "raw") {
-          // Buffered 1xx bytes: route through the same AsyncSocket buffer the
-          // response's own writeHead/end use so they precede the final response.
-          handle.writeInformational(op[1], op[2]);
-          if (typeof op[3] === "function") process.nextTick(op[3]);
+          if (!writeRawToHandle(res, handle, op[1], op[2], op[3])) hitBackpressure = true;
         } else if (kind === "write") {
           if (ServerResponsePrototypeWrite.$call(res, op[1], op[2], op[3]) === false) hitBackpressure = true;
         } else {
@@ -3080,21 +3078,35 @@ function accountQueuedHeaderBytes(res, queued) {
   addPipelineOutgoingData(queued, bytes);
 }
 
+// Like Node's write_(), the implicit head is stored before the throw: headersSent is true behind it.
+function refuseBodyChunk(res, fromEnd) {
+  if (res[headerStateSymbol] === NodeHTTPHeaderState.none) {
+    if (fromEnd) res[kImplicitHeaderFromEnd] = true;
+    try {
+      res._implicitHeader();
+    } finally {
+      if (fromEnd) res[kImplicitHeaderFromEnd] = false;
+    }
+  }
+  return $ERR_HTTP_BODY_NOT_ALLOWED();
+}
+
 function bufferPipelinedWrite(res, queued, chunk, encoding, callback) {
   callWriteHeadIfObservable(res, res[headerStateSymbol]);
   if (res[headerStateSymbol] === NodeHTTPHeaderState.none) {
     updateHasBody(res, res.statusCode);
   }
-  accountQueuedHeaderBytes(res, queued);
   if (!res._hasBody) {
     if (chunk && res[kRejectNonStandardBodyWrites]) {
-      throw $ERR_HTTP_BODY_NOT_ALLOWED();
+      throw refuseBodyChunk(res, false);
     }
+    accountQueuedHeaderBytes(res, queued);
     // Node's write_(): a response that cannot have a body discards the write, calls back and returns true.
     queued.ops.push(["write", undefined, encoding, undefined]);
     if (typeof callback === "function") process.nextTick(callback);
     return true;
   }
+  accountQueuedHeaderBytes(res, queued);
   queued.ops.push(["write", chunk, encoding, callback]);
   if (chunk) {
     const bytes = typeof chunk === "string" ? Buffer.byteLength(chunk, encoding) : chunk.length;
@@ -3116,13 +3128,13 @@ function bufferPipelinedEnd(res, queued, chunk, encoding, callback) {
   if (res[headerStateSymbol] === NodeHTTPHeaderState.none) {
     updateHasBody(res, res.statusCode);
   }
-  accountQueuedHeaderBytes(res, queued);
   if (chunk && !res._hasBody) {
     if (res[kRejectNonStandardBodyWrites]) {
-      throw $ERR_HTTP_BODY_NOT_ALLOWED();
+      throw refuseBodyChunk(res, true);
     }
     chunk = undefined;
   }
+  accountQueuedHeaderBytes(res, queued);
   queued.ops.push(["end", chunk, encoding, callback]);
   if (chunk) {
     const bytes = typeof chunk === "string" ? Buffer.byteLength(chunk, encoding) : chunk.length;
@@ -3339,33 +3351,71 @@ Object.defineProperty(ServerResponse.prototype, "headersSent", {
   },
 });
 
+// Writes a 1xx block or the data of _send() through the handle of the response, so the bytes keep their place among its own.
+function writeRawToHandle(res, handle, chunk, encoding, callback) {
+  // Native handle: the count of bytes that wait, or -1 for a socket that is gone. Fallback handle: what its socket.write() returned.
+  const result = handle.writeInformational(chunk, encoding, callback);
+  if (result === true) return true;
+  if (result !== false) {
+    // Like Node's _writeRaw(): a destroyed socket takes nothing, and nothing calls back.
+    if (result < 0) return false;
+    if (result === 0) {
+      // Nothing waits, so the callbacks of earlier writes are due too. They run first.
+      if (res[kPendingCallbacks].length !== 0) res._callPendingCallbacks();
+      if (typeof callback === "function") process.nextTick(callback);
+      return true;
+    }
+    // Like write(): the callback waits for the bytes.
+    if (typeof callback === "function") res[kPendingCallbacks].push(callback);
+  }
+  handle.onwritable ??= allowWritesToContinue.bind(res);
+  return false;
+}
+
 ServerResponse.prototype._writeRaw = function (chunk, encoding, callback) {
-  if (!this[kHandle]) {
+  const handle = this[kHandle];
+  if (!handle) {
     // Standalone path: OutgoingMessage._writeRaw buffers to outputData while
     // no socket is assigned yet (kSocket is null) and flushes the buffer
     // ahead of the chunk once one is - writing through the auto-creating
     // `socket` getter here would drop the bytes into a FakeSocket.
     return OutgoingMessagePrototype._writeRaw.$apply(this, arguments);
   }
+  // https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L408-L411
+  if (typeof encoding === "function") {
+    callback = encoding;
+    encoding = null;
+  }
+  // In Node.js the write() of the socket throws this: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/streams/writable.js#L462-L471
+  if (encoding && (encoding === "buffer" ? typeof chunk === "string" : !Buffer.isEncoding(encoding))) {
+    throw $ERR_UNKNOWN_ENCODING(encoding);
+  }
+  // The native handle knows fewer names than Buffer does ("utf-16le"), and writes utf8 for the others.
+  if (typeof chunk === "string" && encoding && encoding !== "utf8" && encoding !== "latin1" && encoding !== "ascii") {
+    chunk = Buffer.from(chunk, encoding);
+    encoding = null;
+  }
+  if (this.finished) {
+    // Node.js writes such bytes behind the message. Nothing goes out here: the callback runs behind 'finish', with an error for bytes.
+    const lost = chunk ? chunk.length !== 0 : false;
+    if (typeof callback === "function") {
+      const error = lost ? $ERR_STREAM_WRITE_AFTER_END() : undefined;
+      if (this.writableFinished) process.nextTick(callback, error);
+      else this.once("finish", () => process.nextTick(callback, error));
+    }
+    return !lost;
+  }
   const queued = this[kPipelinedQueuedState];
   if (queued !== undefined) {
-    // Queued pipelined response: like Node.js (which buffers to outputData
-    // while no socket is assigned and flushes on assignSocket), buffer the
-    // raw 1xx bytes and write them ahead of the buffered body once this
-    // response reaches the head of the pipeline.
+    // Queued pipelined response: like Node.js's outputData, the bytes are kept and written in their place at its turn.
     queued.ops.push(["raw", chunk, encoding, callback]);
     const bytes = typeof chunk === "string" ? Buffer.byteLength(chunk, encoding) : chunk.length;
     queued.bytes += bytes;
     this.outputSize += bytes;
     addPipelineOutgoingData(queued, bytes);
     return queued.bytes < this.writableHighWaterMark;
   }
-  // Write through the response handle's AsyncSocket buffer (same path as
-  // writeHead/end) so 1xx lines share ordering with the final response bytes;
-  // socket.write() would land in the socket handle's separate stream buffer.
-  this[kHandle].writeInformational(chunk, encoding);
-  if (typeof callback === "function") process.nextTick(callback);
-  return true;
+  return writeRawToHandle(this, handle, chunk, encoding, callback);
 };
 
 ServerResponse.prototype.writeEarlyHints = function (hints, cb) {
@@ -3517,7 +3567,7 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) {
   }
   if (chunk && !this._hasBody) {
     if (this[kRejectNonStandardBodyWrites]) {
-      throw $ERR_HTTP_BODY_NOT_ALLOWED();
+      throw refuseBodyChunk(this, true);
     } else {
       // node.js just ignores the write in this case
       chunk = undefined;
@@ -3727,7 +3777,7 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) {
   }
   if (chunk && !this._hasBody) {
     if (this[kRejectNonStandardBodyWrites]) {
-      throw $ERR_HTTP_BODY_NOT_ALLOWED();
+      throw refuseBodyChunk(this, false);
     } else {
       // node.js just ignores the write in this case
       chunk = undefined;
@@ -3771,16 +3821,17 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) {
     result = handle.write(chunk, encoding, onWritable, strict);
   }
 
-  if (result < 0 && hasBody) {
+  if (result >= 0) {
+    this._callPendingCallbacks();
+  } else if (hasBody) {
     if (callback) {
       // The write was buffered due to backpressure.
       // We need to defer the callback until the write actually goes through.
       this[kPendingCallbacks].push(callback);
     }
     return false;
   }
-
-  this._callPendingCallbacks();
+  // Else a response with no body, behind a raw write whose bytes wait: the callbacks that wait stay.
   if (callback) {
     // A write callback reports that its bytes reached the socket: uncork them before the callback can destroy the response.
     handle.flushHeaders();
@@ -3823,6 +3874,8 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) {
 // advanceResponsePipeline replays buffered ops through these; patched res.write/res.end (compression middleware) must not see them again.
 const ServerResponsePrototypeWrite = ServerResponse.prototype.write;
 const ServerResponsePrototypeEnd = ServerResponse.prototype.end;
+serverResponseMethods.write = ServerResponsePrototypeWrite;
+serverResponseMethods.end = ServerResponsePrototypeEnd;
 
 const kBytesBuffered = Symbol("kBytesBuffered");
 const kAccountingFlushScheduled = Symbol("kAccountingFlushScheduled");
@@ -3938,35 +3991,23 @@ Object.defineProperty(ServerResponse.prototype, "closed", {
   },
 });
 
+// Node's _send(): the stored head, then _writeRaw(). https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L374-L397
 ServerResponse.prototype._send = function (data, encoding, callback, _byteLength) {
-  const handle = this[kHandle];
-  if (!handle) {
+  if (!this[kHandle]) {
     return OutgoingMessagePrototype._send.$apply(this, arguments);
   }
 
-  const strict = strictContentLength(this, this[headerStateSymbol], false);
-  if (this[headerStateSymbol] !== NodeHTTPHeaderState.sent) {
-    handle.cork(() => {
-      const renderedHeaders = renderNativeHeaders(this);
-      try {
-        handle.writeHead(
-          this[kSnapshotStatusCode] ?? this.statusCode,
-          this[kSnapshotStatusMessage] ?? this.statusMessage,
-          renderedHeaders,
-          renderedAutoHeaders,
-          renderedKeepAliveSecs,
-        );
-      } finally {
-        // A throwing writeHead (status validation) must not leave the shared
-        // scratch array marked busy for the rest of the process.
-        releaseRenderedHeaders(renderedHeaders);
-      }
-      this[headerStateSymbol] = NodeHTTPHeaderState.sent;
-      handle.write(data, encoding, callback, strict);
-    });
-  } else {
-    handle.write(data, encoding, callback, strict);
+  if (this[headerStateSymbol] === NodeHTTPHeaderState.assigned) {
+    const queued = this[kPipelinedQueuedState];
+    if (queued === undefined) {
+      ServerResponsePrototypeFlushHeaders.$call(this);
+    } else if (queued.headerBytes === 0) {
+      // A kept write() with no chunk sends the head at the turn. Counted header bytes mean that a kept op sends it.
+      accountQueuedHeaderBytes(this, queued);
+      queued.ops.push(["write", undefined, undefined, undefined]);
+    }
   }
+  return this._writeRaw(data, encoding, callback);
 };
 
 const kSnapshotStatusCode = Symbol("kSnapshotStatusCode");
@@ -4082,8 +4123,8 @@ ServerResponse.prototype.flushHeaders = function () {
   if (this[headerStateSymbol] !== NodeHTTPHeaderState.assigned) this._implicitHeader();
 
   if (this[kPipelinedQueuedState] !== undefined) {
-    // Queued pipelined response: its headers go out when it is assigned the
-    // socket (advanceResponsePipeline) - nothing can be flushed before then.
+    // Queued pipelined response: like Node.js, _send("") keeps the head for the turn of the response.
+    this._send("");
     return;
   }
 
@@ -4114,6 +4155,7 @@ ServerResponse.prototype.flushHeaders = function () {
     this._send("");
   }
 };
+const ServerResponsePrototypeFlushHeaders = ServerResponse.prototype.flushHeaders;
 
 function updateHasBody(response, statusCode) {
   // RFC 2616, 10.2.5: