fix(node:http): retain corked responses in the socket write queue

packages/bun-uws/src/HttpContext.h

@@ -927,7 +927,7 @@ struct HttpContext {
         auto *httpContextData = getSocketContextDataS(s);
 
 
-        if (httpResponseData->isConnectRequest && httpResponseData->socketData && httpContextData->onSocketDrain) {
+        if ((IsNodeHttp || httpResponseData->isConnectRequest) && httpResponseData->socketData && httpContextData->onSocketDrain) {
             httpContextData->onSocketDrain(httpResponseData->socketData, SSL, (struct us_socket_t *) s);
         }
         /* Ask the developer to write data and return success (true) or failure (false), OR skip sending anything and return success (true). */

src/js/ahead-of-time/node/_http_outgoing.ts

@@ -1,5 +1,5 @@
 // What the compiler reads in place of src/js/node/_http_outgoing.ts, which `bun run` evaluates. The two do the same.
-// Written against that file with the SHA-256 7eb68379dcc85e37da343c90d1da52f21b0bd91937cea041c06b56a6db7551fc.
+// Written against that file with the SHA-256 844b8c71bb274a66021c5d1bf7478198b9f0155beb008519d5a1618352888462.
 // When that file changes, bring the change here and write the new hash: test/internal/second-texts-of-modules.test.ts.
 
 // Hardcoded module "node:_http_outgoing"
@@ -2427,6 +2427,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, "_headerNames", {
 });
 
 export default {
+  kCorked,
   kHighWaterMark,
   kUniqueHeaders,
   kOutHeaders,

src/js/internal/streams/utils.ts

@@ -32,6 +32,8 @@ const kIsReadable = Symbol.for("nodejs.stream.readable");
 const kIsWritable = Symbol.for("nodejs.stream.writable");
 const kIsDisturbed = Symbol.for("nodejs.stream.disturbed");
 
+const kWriteOperation = Symbol("kWriteOperation");
+const kWriteOperationEncoding = "bun:write-operation";
 const kIsClosedPromise = Symbol.for("nodejs.webstream.isClosedPromise");
 const kControllerErrorFunction = Symbol.for("nodejs.webstream.controllerErrorFunction");
 
@@ -390,6 +392,8 @@ function isErrored(stream) {
 }
 
 export default {
+  kWriteOperation,
+  kWriteOperationEncoding,
   isDestroyed,
   kIsDestroyed,
   isDisturbed,

src/js/internal/streams/writable.ts

@@ -9,6 +9,9 @@ const { Stream } = require("internal/streams/legacy");
 const destroyImpl = require("internal/streams/destroy");
 const eos = require("internal/streams/end-of-stream");
 const { addAbortSignal } = require("internal/streams/add-abort-signal");
+const streamUtils = require("internal/streams/utils");
+const kWriteOperation: typeof streamUtils.kWriteOperation = streamUtils.kWriteOperation;
+const kWriteOperationEncoding = streamUtils.kWriteOperationEncoding;
 const streamState = require("internal/streams/state");
 const { WritableState, resetBuffer, isReadableState, isWritableState, flagsOf, addFlags, removeFlags, internals, nop } =
   streamState;
@@ -230,6 +233,18 @@ function _write(stream: $Node.Writable, state: WritableState, chunk, encoding, c
   return writeOrBuffer(stream, state, chunk, encoding, cb, 1);
 }
 
+// Internal byte-counted operations share the byte stream's queue and lifecycle.
+Writable.prototype[kWriteOperation] = function (operation, callback) {
+  const state: WritableState = this._writableState;
+  const cb: WriteCallback = callback ?? nop;
+  if ((state[kState] & Flag.EndingOrDestroyed) !== 0) {
+    writeAfterEnd(this, state, cb);
+    return false;
+  }
+  state.pendingcb!++;
+  return writeOrBuffer(this, state, operation, kWriteOperationEncoding, cb, operation.length);
+};
+
 const writeBytes = function _write(stream: $Node.Writable, state: WritableState, chunk, encoding, cb: WriteCallback) {
   if (!encoding) {
     encoding = (state[kState] & Flag.DefaultUTF8Encoding) !== 0 ? "utf8" : state[kDefaultEncodingValue];

src/js/node/_http_outgoing.ts

@@ -1322,6 +1322,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, "_headerNames", {
 });
 
 export default {
+  kCorked,
   kHighWaterMark,
   kUniqueHeaders,
   kOutHeaders,