child_process: watch child_process stdin pipe peer close event

Fixes: #25131

Author: Tseian

doc/api/child_process.md

@@ -1066,14 +1066,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
@@ -2170,6 +2170,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(() => sock.destroy());
+    sock.once('close', () => pipe.unwatchPeerClose?.());
+  }
+  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,8 @@ 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);
+  SetProtoMethod(isolate, t, "unwatchPeerClose", UnwatchPeerClose);
 
 #ifdef _WIN32
   SetProtoMethod(isolate, t, "setPendingInstances", SetPendingInstances);
@@ -110,6 +113,8 @@ void PipeWrap::RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(Listen);
   registry->Register(Connect);
   registry->Register(Open);
+  registry->Register(WatchPeerClose);
+  registry->Register(UnwatchPeerClose);
 #ifdef _WIN32
   registry->Register(SetPendingInstances);
 #endif
@@ -159,6 +164,11 @@ PipeWrap::PipeWrap(Environment* env,
                    // Suggestion: uv_pipe_init() returns void.
 }
 
+PipeWrap::~PipeWrap() {
+  peer_close_watching_ = false;
+  peer_close_cb_.Reset();
+}
+
 void PipeWrap::Bind(const FunctionCallbackInfo<Value>& args) {
   PipeWrap* wrap;
   ASSIGN_OR_RETURN_UNWRAP(&wrap, args.This());
@@ -213,6 +223,96 @@ 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());
+
+  if (!wrap->IsAlive()) {
+    return args.GetReturnValue().Set(UV_EBADF);
+  }
+
+  if (wrap->peer_close_watching_) {
+    return args.GetReturnValue().Set(0);
+  }
+
+  CHECK_GT(args.Length(), 0);
+  CHECK(args[0]->IsFunction());
+
+  Environment* env = wrap->env();
+  Isolate* isolate = env->isolate();
+
+  // Store the JS callback securely so it isn't garbage collected.
+  wrap->peer_close_cb_.Reset(isolate, args[0].As<Function>());
+  wrap->peer_close_watching_ = true;
+
+  // 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) {
+    wrap->peer_close_watching_ = false;
+    wrap->peer_close_cb_.Reset();
+  }
+  args.GetReturnValue().Set(err);
+}
+
+void PipeWrap::UnwatchPeerClose(const FunctionCallbackInfo<Value>& args) {
+  PipeWrap* wrap;
+  ASSIGN_OR_RETURN_UNWRAP(&wrap, args.This());
+
+  if (!wrap->peer_close_watching_) {
+    wrap->peer_close_cb_.Reset();
+    return args.GetReturnValue().Set(0);
+  }
+
+  // Stop listening and release the JS callback to prevent memory leaks.
+  wrap->peer_close_watching_ = false;
+  wrap->peer_close_cb_.Reset();
+  args.GetReturnValue().Set(uv_read_stop(wrap->stream()));
+}
+
+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 || !wrap->peer_close_watching_) 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.
+  wrap->peer_close_watching_ = false;
+  uv_read_stop(stream);
+
+  if (wrap->peer_close_cb_.IsEmpty()) return;
+  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<Function> cb = wrap->peer_close_cb_.Get(isolate);
+  // Reset before calling to prevent re-entrancy issues
+  wrap->peer_close_cb_.Reset();
+
+  errors::TryCatchScope try_catch(env);
+  try_catch.SetVerbose(true);
+
+  // 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

@@ -54,6 +54,7 @@ class PipeWrap : public ConnectionWrap<PipeWrap, uv_pipe_t> {
   SET_SELF_SIZE(PipeWrap)
 
  private:
+  ~PipeWrap() override;
   PipeWrap(Environment* env,
            v8::Local<v8::Object> object,
            ProviderType provider,
@@ -64,12 +65,23 @@ 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 UnwatchPeerClose(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(
       const v8::FunctionCallbackInfo<v8::Value>& args);
 #endif
   static void Fchmod(const v8::FunctionCallbackInfo<v8::Value>& args);
+
+  bool peer_close_watching_ = false;
+  v8::Global<v8::Function> peer_close_cb_;
 };
 
 

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,27 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { spawn } = require('child_process');
+
+if (common.isWindows) {
+  common.skip('Not applicable on Windows');
+}
+
+const child = spawn(process.execPath, [
+  '-e',
+  'require("fs").closeSync(0); setTimeout(() => {}, 2000)',
+], { stdio: ['pipe', 'ignore', 'ignore'] });
+
+const timeout = setTimeout(() => {
+  assert.fail('stdin close event was not emitted');
+}, 1000);
+
+child.stdin.on('close', common.mustCall(() => {
+  clearTimeout(timeout);
+  child.kill();
+}));
+
+child.on('exit', common.mustCall(() => {
+  clearTimeout(timeout);
+}));