child_process: watch child_process stdin pipe peer close event

doc/api/child_process.md

@@ -1058,14 +1058,14 @@ pipes between the parent and child. The value is one of the following:
    corresponds to the index in the `stdio` array. The stream must have an
    underlying descriptor (file streams do not start until the `'open'` event has
    occurred).
-   **NOTE:** While it is technically possible to pass `stdin` as a writable or
-   `stdout`/`stderr` as readable, it is not recommended.
+   **NOTE:** While it is technically possible to pass `stdin` as a readable or
+   `stdout`/`stderr` as writable, it is not recommended.
    Readable and writable streams are designed with distinct behaviors, and using
    them incorrectly (e.g., passing a readable stream where a writable stream is
    expected) can lead to unexpected results or errors. This practice is discouraged
    as it may result in undefined behavior or dropped callbacks if the stream
-   encounters errors. Always ensure that `stdin` is used as readable and
-   `stdout`/`stderr` as writable to maintain the intended flow of data between
+   encounters errors. Always ensure that `stdin` is used as writable and
+   `stdout`/`stderr` as readable to maintain the intended flow of data between
    the parent and child processes.
 7. Positive integer: The integer value is interpreted as a file descriptor
    that is open in the parent process. It is shared with the child
@@ -2162,6 +2162,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;
 }
 
 
@@ -472,7 +479,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
@@ -213,6 +216,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);
 
@@ -252,7 +352,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,72 @@
+'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, setup }, 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();
+  }));
+
+  setup(child);
+}
+
+runTest({
+  script: 'setTimeout(() => require("fs").closeSync(0), 50); setTimeout(() => {}, 2000)',
+  setup: () => {},
+}, common.mustCall(() => {
+  runTest({
+    script: 'setTimeout(() => require("fs").closeSync(0), 200); setTimeout(() => {}, 2000)',
+    setup: (child) => {
+      const handle = child.stdin?._handle;
+      assert.strictEqual(typeof handle?.watchPeerClose, 'function');
+      handle.watchPeerClose(true, common.mustNotCall());
+    },
+  }, common.mustCall(() => {
+    runTest({
+      script: 'setTimeout(() => {}, 2000)',
+      setup: (child) => {
+        const handle = child.stdin?._handle;
+        assert.strictEqual(typeof handle?.watchPeerClose, 'function');
+
+        child.stdin.once('close', common.mustCall(() => {
+          assert.doesNotThrow(() => {
+            handle.watchPeerClose(true, common.mustNotCall());
+          });
+        }));
+        child.stdin.destroy();
+      },
+    }, common.mustCall(() => {}));
+  }));
+}));