child_process: watch child_process stdin pipe peer close event

Watch the child end of the stdin pipe for peer-close (EOF) events, only
supported on non-Windows platforms. When the child closes its end of the
pipe, subprocess.stdin is destroyed so that its 'close' event is emitted,
instead of only being reported as an EPIPE error on the next write.

Fixes: #25131

Author: Tseian

doc/api/child_process.md

@@ -2191,6 +2191,11 @@ until this stream has been closed via `end()`.
 If the child process was spawned with `stdio[0]` set to anything other than `'pipe'`,
 then this will be `null`.
 
+On non-Windows platforms, when `stdio[0]` is `'pipe'`, Node.js watches for the
+child process closing its end of the stdin pipe and destroys `subprocess.stdin`
+when that happens. This helps surface pipe peer-close semantics consistently for
+the writable side of the stream.
+
 `subprocess.stdin` is an alias for `subprocess.stdio[0]`. Both properties will
 refer to the same value.
 

lib/internal/child_process.js

@@ -332,8 +332,15 @@ function flushStdio(subprocess) {
 }
 
 
-function createSocket(pipe, readable) {
-  return net.Socket({ handle: pipe, readable });
+function createSocket(pipe, readable, watchPeerClose) {
+  const sock = net.Socket({ handle: pipe, readable });
+  if (watchPeerClose &&
+      process.platform !== 'win32' &&
+      typeof pipe?.watchPeerClose === 'function') {
+    pipe.watchPeerClose(true, () => sock.destroy());
+    sock.once('close', () => pipe.watchPeerClose(false));
+  }
+  return sock;
 }
 
 
@@ -489,7 +496,7 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
     if (stream.handle) {
       stream.socket = createSocket(this.pid !== 0 ?
-        stream.handle : null, i > 0);
+        stream.handle : null, i > 0, i === 0);
 
       if (i > 0 && this.pid !== 0) {
         this._closesNeeded++;

src/pipe_wrap.cc

@@ -28,6 +28,7 @@
 #include "handle_wrap.h"
 #include "node.h"
 #include "node_buffer.h"
+#include "node_errors.h"
 #include "node_external_reference.h"
 #include "stream_base-inl.h"
 #include "stream_wrap.h"
@@ -80,6 +81,7 @@ void PipeWrap::Initialize(Local<Object> target,
   SetProtoMethod(isolate, t, "listen", Listen);
   SetProtoMethod(isolate, t, "connect", Connect);
   SetProtoMethod(isolate, t, "open", Open);
+  SetProtoMethod(isolate, t, "watchPeerClose", WatchPeerClose);
 
 #ifdef _WIN32
   SetProtoMethod(isolate, t, "setPendingInstances", SetPendingInstances);
@@ -110,6 +112,7 @@ void PipeWrap::RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(Listen);
   registry->Register(Connect);
   registry->Register(Open);
+  registry->Register(WatchPeerClose);
 #ifdef _WIN32
   registry->Register(SetPendingInstances);
 #endif
@@ -219,6 +222,103 @@ void PipeWrap::Open(const FunctionCallbackInfo<Value>& args) {
   args.GetReturnValue().Set(err);
 }
 
+void PipeWrap::WatchPeerClose(const FunctionCallbackInfo<Value>& args) {
+  PipeWrap* wrap;
+  ASSIGN_OR_RETURN_UNWRAP(&wrap, args.This());
+
+  CHECK_GT(args.Length(), 0);
+  CHECK(args[0]->IsBoolean());
+  const bool enable = args[0].As<v8::Boolean>()->Value();
+  Environment* env = wrap->env();
+  Isolate* isolate = env->isolate();
+  v8::HandleScope handle_scope(isolate);
+  v8::Context::Scope context_scope(env->context());
+  Local<Object> obj = wrap->object();
+
+  // UnwatchPeerClose
+  if (!enable) {
+    if (obj->GetInternalField(kPeerCloseCallbackField)
+            .As<Value>()
+            ->IsUndefined()) {
+      return;
+    }
+
+    obj->SetInternalField(kPeerCloseCallbackField, v8::Undefined(isolate));
+    uv_read_stop(wrap->stream());
+    return;
+  }
+
+  if (!wrap->IsAlive()) {
+    return;
+  }
+  if (!obj->GetInternalField(kPeerCloseCallbackField)
+           .As<Value>()
+           ->IsUndefined()) {
+    return;
+  }
+
+  CHECK_GT(args.Length(), 1);
+  CHECK(args[1]->IsFunction());
+
+  // Store the JS callback in an internal field.
+  obj->SetInternalField(kPeerCloseCallbackField, args[1]);
+
+  // Start reading to detect EOF/ECONNRESET from the peer.
+  // We use our custom allocator and reader, ignoring actual data.
+  int err = uv_read_start(wrap->stream(), PeerCloseAlloc, PeerCloseRead);
+  if (err != 0) {
+    obj->SetInternalField(kPeerCloseCallbackField, v8::Undefined(isolate));
+  }
+}
+
+void PipeWrap::PeerCloseAlloc(uv_handle_t* handle,
+                              size_t suggested_size,
+                              uv_buf_t* buf) {
+  // We only care about EOF, not the actual data.
+  // Using a static 1-byte buffer avoids dynamic memory allocation overhead.
+  static char scratch;
+  *buf = uv_buf_init(&scratch, 1);
+}
+
+void PipeWrap::PeerCloseRead(uv_stream_t* stream,
+                             ssize_t nread,
+                             const uv_buf_t* buf) {
+  PipeWrap* wrap = static_cast<PipeWrap*>(stream->data);
+  if (wrap == nullptr) return;
+
+  // Ignore actual data reads or EAGAIN (0). We only watch for disconnects.
+  if (nread > 0 || nread == 0) return;
+
+  // Wait specifically for EOF or connection reset (peer closed).
+  if (nread != UV_EOF && nread != UV_ECONNRESET) return;
+
+  // Peer has closed the connection. Stop reading immediately.
+  uv_read_stop(stream);
+
+  Environment* env = wrap->env();
+  Isolate* isolate = env->isolate();
+
+  // Set up V8 context and handles to safely execute the JS callback.
+  v8::HandleScope handle_scope(isolate);
+  v8::Context::Scope context_scope(env->context());
+  Local<Object> obj = wrap->object();
+
+  // Check if callback is set
+  if (obj->GetInternalField(kPeerCloseCallbackField)
+          .As<Value>()
+          ->IsUndefined()) {
+    return;
+  }
+  Local<Value> cb_value =
+      obj->GetInternalField(kPeerCloseCallbackField).As<Value>();
+  Local<Function> cb = cb_value.As<Function>();
+  // Reset before calling to prevent re-entrancy issues
+  obj->SetInternalField(kPeerCloseCallbackField, v8::Undefined(isolate));
+
+  // MakeCallback properly tracks AsyncHooks context and flushes microtasks.
+  wrap->MakeCallback(cb, 0, nullptr);
+}
+
 void PipeWrap::Connect(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
 
@@ -258,7 +358,6 @@ void PipeWrap::Connect(const FunctionCallbackInfo<Value>& args) {
 
   args.GetReturnValue().Set(err);
 }
-
 }  // namespace node
 
 NODE_BINDING_CONTEXT_AWARE_INTERNAL(pipe_wrap, node::PipeWrap::Initialize)

src/pipe_wrap.h

@@ -40,6 +40,11 @@ class PipeWrap : public ConnectionWrap<PipeWrap, uv_pipe_t> {
     IPC
   };
 
+  enum InternalFields {
+    kPeerCloseCallbackField = LibuvStreamWrap::kInternalFieldCount,
+    kInternalFieldCount
+  };
+
   static v8::MaybeLocal<v8::Object> Instantiate(Environment* env,
                                                 AsyncWrap* parent,
                                                 SocketType type);
@@ -64,6 +69,13 @@ class PipeWrap : public ConnectionWrap<PipeWrap, uv_pipe_t> {
   static void Listen(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Connect(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Open(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void WatchPeerClose(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void PeerCloseAlloc(uv_handle_t* handle,
+                             size_t suggested_size,
+                             uv_buf_t* buf);
+  static void PeerCloseRead(uv_stream_t* stream,
+                            ssize_t nread,
+                            const uv_buf_t* buf);
 
 #ifdef _WIN32
   static void SetPendingInstances(

test/async-hooks/test-pipewrap.js

@@ -35,6 +35,7 @@ const processwrap = processes[0];
 const pipe1 = pipes[0];
 const pipe2 = pipes[1];
 const pipe3 = pipes[2];
+const pipe1ExpectedInvocations = process.platform === 'win32' ? 1 : 2;
 
 assert.strictEqual(processwrap.type, 'PROCESSWRAP');
 assert.strictEqual(processwrap.triggerAsyncId, 1);
@@ -83,7 +84,11 @@ function onexit() {
   // Usually it is just one event, but it can be more.
   assert.ok(ioEvents >= 3, `at least 3 stdout io events, got ${ioEvents}`);
 
-  checkInvocations(pipe1, { init: 1, before: 1, after: 1 },
+  checkInvocations(pipe1, {
+    init: 1,
+    before: pipe1ExpectedInvocations,
+    after: pipe1ExpectedInvocations,
+  },
                    'pipe wrap when sleep.spawn was called');
   checkInvocations(pipe2, { init: 1, before: ioEvents, after: ioEvents },
                    'pipe wrap when sleep.spawn was called');

test/parallel/test-child-process-stdin-close-event.js

@@ -0,0 +1,71 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { spawn } = require('child_process');
+
+if (common.isWindows) {
+  common.skip('Not applicable on Windows');
+}
+
+function spawnChild(script) {
+  return spawn(process.execPath, ['-e', script], {
+    stdio: ['pipe', 'ignore', 'ignore'],
+  });
+}
+
+function runTest({ script, onSpawn }, done) {
+  const child = spawnChild(script);
+
+  const timeout = setTimeout(() => {
+    assert.fail('stdin close event was not emitted');
+  }, 2000);
+
+  let closed = false;
+  let exited = false;
+  function maybeDone() {
+    if (!closed || !exited) return;
+    clearTimeout(timeout);
+    done();
+  }
+
+  child.stdin.once('close', common.mustCall(() => {
+    closed = true;
+    child.kill();
+    maybeDone();
+  }));
+
+  child.once('exit', common.mustCall(() => {
+    exited = true;
+    maybeDone();
+  }));
+
+  onSpawn?.(child);
+}
+
+runTest({
+  script: 'setTimeout(() => require("fs").closeSync(0), 50); setTimeout(() => {}, 2000)',
+}, common.mustCall(() => {
+  runTest({
+    script: 'setTimeout(() => require("fs").closeSync(0), 200); setTimeout(() => {}, 2000)',
+    onSpawn: common.mustCall((child) => {
+      const handle = child.stdin?._handle;
+      assert.strictEqual(typeof handle?.watchPeerClose, 'function');
+      handle.watchPeerClose(true, common.mustNotCall());
+    }),
+  }, common.mustCall(() => {
+    runTest({
+      script: 'setTimeout(() => {}, 2000)',
+      onSpawn: common.mustCall((child) => {
+        const handle = child.stdin?._handle;
+        assert.strictEqual(typeof handle?.watchPeerClose, 'function');
+
+        child.stdin.once('close', common.mustCall(() => {
+          // Calling watchPeerClose again after close must not throw.
+          handle.watchPeerClose(true, common.mustNotCall());
+        }));
+        child.stdin.destroy();
+      }),
+    }, common.mustCall());
+  }));
+}));