fix(server): survive a late child-pipe ECONNRESET (upstream #16555 on nightly 2761, board #15806)

apps/server/src/process/spawnerPipeErrors.test.ts

@@ -0,0 +1,53 @@
+import { describe, it } from "@effect/vitest";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Stream from "effect/Stream";
+import * as ChildProcess from "effect/process/ChildProcess";
+import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
+import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
+
+// Covers patches/@effect__platform-node-shared@4.0.1.patch. A stopped child
+// never reads its input, so a chunk larger than the socket buffer leaves a
+// write queued. Closing the scope interrupts the writer and kills the child,
+// and the pipe then fails with EPIPE (stdin) or ECONNRESET (extra fds, which
+// Node also reads). Unpatched, nothing listens by then and the uncaught error
+// kills the server, as HtmlRender's Chrome pipe did on every torn-down capture.
+const killChildWithQueuedInput = (target: "stdin" | "fd3") =>
+  Effect.gen(function* () {
+    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
+    yield* Effect.scoped(
+      Effect.gen(function* () {
+        const child = yield* spawner.spawn(
+          ChildProcess.make(process.execPath, ["-e", "process.kill(process.pid, 'SIGSTOP')"], {
+            stdin: "pipe",
+            stdout: "ignore",
+            stderr: "ignore",
+            forceKillAfter: "100 millis",
+            additionalFds: { fd3: { type: "input" } },
+          }),
+        );
+        const pulled = yield* Deferred.make<void>();
+        const input = Stream.make(new Uint8Array(4 * 1024 * 1024)).pipe(
+          Stream.tap(() => Deferred.succeed(pulled, undefined)),
+        );
+        yield* Effect.forkScoped(
+          Stream.run(input, target === "stdin" ? child.stdin : child.getInputFd(3)),
+        );
+        yield* Deferred.await(pulled);
+      }),
+    );
+    // The scope ends once the killed child is reaped. Its pipe failure is
+    // already queued by then and lands in this same poll phase.
+    yield* Effect.promise(() => new Promise((resolve) => setImmediate(resolve)));
+  }).pipe(Effect.provide(NodeServices.layer));
+
+describe.skipIf(HostProcessPlatform.defaultValue() === "win32")("child process input pipes", () => {
+  it.live("survive a child killed with stdin still queued", () =>
+    killChildWithQueuedInput("stdin"),
+  );
+
+  it.live("survive a child killed with additional fd input still queued", () =>
+    killChildWithQueuedInput("fd3"),
+  );
+});

patches/@effect__platform-node-shared@4.0.1.patch

@@ -0,0 +1,69 @@
+diff --git a/dist/NodeChildProcessSpawner.js b/dist/NodeChildProcessSpawner.js
+index 822127c..d7d588d 100644
+--- a/dist/NodeChildProcessSpawner.js
++++ b/dist/NodeChildProcessSpawner.js
+@@ -170,6 +170,7 @@ const make = /*#__PURE__*/Effect.gen(function* () {
+             // Create a sink to write to for input file descriptors
+             let sink = Sink.drain;
+             if (nodeStream && "write" in nodeStream) {
++              observeLateErrors(nodeStream);
+               sink = NodeSink.fromWritable({
+                 evaluate: () => nodeStream,
+                 onError: error => toPlatformError(`fromWritable(fd${fd})`, toError(error), command)
+@@ -209,11 +210,20 @@ const make = /*#__PURE__*/Effect.gen(function* () {
+       getOutputFd: fd => outputStreams.get(fd) ?? Stream.empty
+     };
+   });
++  // A sink only listens for "error" while it is writing. Node child pipes are
++  // sockets, so a child that exits with unread input resets them and the
++  // parent sees EPIPE or ECONNRESET later, after the sink finished or was
++  // interrupted. With no listener left that event is an uncaught exception
++  // that kills this process. The exit code already reports what happened.
++  const observeLateErrors = stream => {
++    stream.on("error", () => {});
++  };
+   const setupChildStdin = (command, childProcess, config) => Effect.suspend(() => {
+     // If the child process has a standard input stream, connect it to the
+     // sink that will attached to the process handle
+     let sink = Sink.drain;
+     if (Predicate.isNotNull(childProcess.stdin)) {
++      observeLateErrors(childProcess.stdin);
+       sink = NodeSink.fromWritable({
+         evaluate: () => childProcess.stdin,
+         onError: error => toPlatformError("fromWritable(stdin)", toError(error), command),
+diff --git a/src/NodeChildProcessSpawner.ts b/src/NodeChildProcessSpawner.ts
+index 26143d5..3016d3f 100644
+--- a/src/NodeChildProcessSpawner.ts
++++ b/src/NodeChildProcessSpawner.ts
+@@ -243,6 +243,7 @@ const make = Effect.gen(function*() {
+           // Create a sink to write to for input file descriptors
+           let sink: Sink.Sink<void, Uint8Array, never, PlatformError.PlatformError> = Sink.drain
+           if (nodeStream && "write" in nodeStream) {
++            observeLateErrors(nodeStream)
+             sink = NodeSink.fromWritable({
+               evaluate: () => nodeStream,
+               onError: (error) => toPlatformError(`fromWritable(fd${fd})`, toError(error), command)
+@@ -289,6 +290,15 @@ const make = Effect.gen(function*() {
+     }
+   })
+
++  // A sink only listens for "error" while it is writing. Node child pipes are
++  // sockets, so a child that exits with unread input resets them and the
++  // parent sees EPIPE or ECONNRESET later, after the sink finished or was
++  // interrupted. With no listener left that event is an uncaught exception
++  // that kills this process. The exit code already reports what happened.
++  const observeLateErrors = (stream: NodeJS.WritableStream) => {
++    stream.on("error", () => {})
++  }
++
+   const setupChildStdin = (
+     command: ChildProcess.StandardCommand,
+     childProcess: NodeChildProcess.ChildProcess,
+@@ -299,6 +309,7 @@ const make = Effect.gen(function*() {
+       // sink that will attached to the process handle
+       let sink: Sink.Sink<void, unknown, never, PlatformError.PlatformError> = Sink.drain
+       if (Predicate.isNotNull(childProcess.stdin)) {
++        observeLateErrors(childProcess.stdin)
+         sink = NodeSink.fromWritable({
+           evaluate: () => childProcess.stdin!,
+           onError: (error) => toPlatformError("fromWritable(stdin)", toError(error), command),

pnpm-workspace.yaml

@@ -261,6 +261,9 @@ patchedDependencies:
   expo-blur@58.0.3: patches/expo-blur@58.0.3.patch
   expo-audio@58.0.4: patches/expo-audio@58.0.4.patch
   "@clerk/expo@4.6.8": patches/@clerk__expo@4.6.8.patch
+  # Child stdin and input fds keep an "error" listener after their writer stops.
+  # Otherwise a child that exits with unread input resets the pipe and crashes the server.
+  "@effect/platform-node-shared@4.0.1": patches/@effect__platform-node-shared@4.0.1.patch
   "@effect/vitest@4.0.1": patches/@effect__vitest@4.0.1.patch
   "@expo/metro-config@58.0.6": patches/@expo__metro-config@58.0.6.patch
   expo-sharing@58.0.13: patches/expo-sharing@58.0.13.patch