Repository: vercel/next.js
Pull request: Turbopack: handle webpack loader event backpressure

## Improve IPC socket handling in Turbopack Node.js communication

### What?
Enhances the IPC communication between Turbopack and Node.js processes by implementing proper socket backpressure handling.

### Why?
The current implementation doesn't properly handle backpressure when writing to sockets, which can lead to hanging where Node.js doesn't write the rest of the data.

### How?
- Added a `Writable` stream wrapper around the socket to properly handle backpressure
- Implemented drain event handling to ensure data is written correctly
- Enabled noDelay on both the Node.js client and Rust server sides
- Moved stringify into the promise to make the stringify + write happen in the same tick.

Part of PACK-4409

Fixes #78407

Closes PACK-4452

Changed files:
filename: turbopack/crates/turbopack-node/js/src/ipc/index.ts
patch: @@ -1,4 +1,5 @@
 import { createConnection } from "node:net";
+import { Writable } from "node:stream";
 import type { StackFrame } from "../compiled/stacktrace-parser";
 import { parse as parseStackTrace } from "../compiled/stacktrace-parser";
 import { getProperError } from "./error";
@@ -40,7 +41,30 @@ export type Ipc<TIncoming, TOutgoing> = {
 function createIpc<TIncoming, TOutgoing>(
   port: number
 ): Ipc<TIncoming, TOutgoing> {
-  const socket = createConnection(port, "127.0.0.1");
+  const socket = createConnection({
+    port,
+    host: "127.0.0.1",
+  });
+
+  /**
+   * A writable stream that writes to the socket.
+   * We don't write directly to the socket because we need to
+   * handle backpressure and wait for the socket to be drained
+   * before writing more data.
+   */
+  const socketWritable = new Writable({
+    write(chunk, _enc, cb) {
+      if (socket.write(chunk)) {
+        cb();
+      } else {
+        socket.once('drain', cb);
+      }
+    },
+    final(cb) {
+      socket.end(cb);
+    }
+  });
+
   const packetQueue: Buffer[] = [];
   const recvPromiseResolveQueue: Array<(message: TIncoming) => void> = [];
 
@@ -97,13 +121,12 @@ function createIpc<TIncoming, TOutgoing>(
 
   // TODO(lukesandberg): some of the messages being sent are very large and contain lots
   //  of redundant information.  Consider adding gzip compression to our stream.
-  function send(message: any): Promise<void> {
-    // Reserve 4 bytes for our length prefix, we will over-write after encoding.
-    const packet = Buffer.from("0000" + JSON.stringify(message), "utf8");
-    packet.writeUInt32BE(packet.length - 4, 0);
-
+  function doSend(message: any): Promise<void> {
     return new Promise((resolve, reject) => {
-      socket.write(packet, (err) => {
+      // Reserve 4 bytes for our length prefix, we will over-write after encoding.
+      const packet = Buffer.from("0000" + message, "utf8");
+      packet.writeUInt32BE(packet.length - 4, 0);
+      socketWritable.write(packet, (err) => {
         process.stderr.write(`TURBOPACK_OUTPUT_D\n`);
         process.stdout.write(`TURBOPACK_OUTPUT_D\n`);
         if (err != null) {
@@ -115,21 +138,12 @@ function createIpc<TIncoming, TOutgoing>(
     });
   }
 
-  function sendReady(): Promise<void> {
-    const length = Buffer.from([0, 0, 0, 0]);
-    return new Promise((resolve, reject) => {
-      socket.write(length, (err) => {
-        process.stderr.write(`TURBOPACK_OUTPUT_D\n`);
-        process.stdout.write(`TURBOPACK_OUTPUT_D\n`);
-
-        if (err != null) {
-          reject(err);
-        } else {
-          resolve();
-        }
-      });
-    });
-  }
+  function send(message: any): Promise<void> {
+    return doSend(JSON.stringify(message));
+   }
+   function sendReady(): Promise<void> {
+      return doSend("");
+   }
 
   return {
     async recv() {