child_process: watch child_process stdin pipe peer close event

Tseian · Tseian · commit 4fbb171e53a0 · 2026-03-22T09:34:35.000+08:00
Fixes: #25131
diff --git a/lib/internal/child_process.js b/lib/internal/child_process.js
@@ -176,7 +176,7 @@ const handleConversion = {
           // waiting for the NODE_HANDLE_ACK of the current passing handle.
           assert(!target._pendingMessage);
           target._pendingMessage =
-              { callback, message, handle, options, retransmissions: 0 };
+            { callback, message, handle, options, retransmissions: 0 };
         } else {
           handle.close();
         }
@@ -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;
 }
 
 
@@ -368,7 +375,7 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
 
   validateOneOf(options.serialization, 'options.serialization',
-                [undefined, 'json', 'advanced']);
+    [undefined, 'json', 'advanced']);
   const serialization = options.serialization || 'json';
 
   if (ipc !== undefined) {
@@ -380,7 +387,7 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
     ArrayPrototypePush(options.envPairs, `NODE_CHANNEL_FD=${ipcFd}`);
     ArrayPrototypePush(options.envPairs,
-                       `NODE_CHANNEL_SERIALIZATION_MODE=${serialization}`);
+      `NODE_CHANNEL_SERIALIZATION_MODE=${serialization}`);
   }
 
   validateString(options.file, 'options.file');
@@ -401,10 +408,10 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
   // Run-time errors should emit an error, not throw an exception.
   if (err === UV_EACCES ||
-      err === UV_EAGAIN ||
-      err === UV_EMFILE ||
-      err === UV_ENFILE ||
-      err === UV_ENOENT) {
+    err === UV_EAGAIN ||
+    err === UV_EMFILE ||
+    err === UV_ENFILE ||
+    err === UV_ENOENT) {
     if (childProcessSpawn.hasSubscribers) {
       childProcessSpawn.error.publish({
         process: this,
@@ -472,9 +479,9 @@ 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) {
+        if (i > 0 && this.pid !== 0) {
         this._closesNeeded++;
         stream.socket.on('close', () => {
           maybeClose(this);
@@ -494,7 +501,7 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
   for (i = 0; i < stdio.length; i++)
     ArrayPrototypePush(this.stdio,
-                       stdio[i].socket === undefined ? null : stdio[i].socket);
+      stdio[i].socket === undefined ? null : stdio[i].socket);
 
   // Add .send() method and start listening for IPC data
   if (ipc !== undefined) setupChannel(this, ipc, serialization);
@@ -660,16 +667,16 @@ function setupChannel(target, channel, serializationMode) {
   target.on('internalMessage', function(message, handle) {
     // Once acknowledged - continue sending handles.
     if (message.cmd === 'NODE_HANDLE_ACK' ||
-        message.cmd === 'NODE_HANDLE_NACK') {
+      message.cmd === 'NODE_HANDLE_NACK') {
 
       if (target._pendingMessage) {
         if (message.cmd === 'NODE_HANDLE_ACK') {
           closePendingHandle(target);
         } else if (target._pendingMessage.retransmissions++ ===
-                   MAX_HANDLE_RETRANSMISSIONS) {
+          MAX_HANDLE_RETRANSMISSIONS) {
           closePendingHandle(target);
           process.emitWarning('Handle did not reach the receiving process ' +
-                              'correctly', 'SentHandleNotReceivedWarning');
+            'correctly', 'SentHandleNotReceivedWarning');
         }
       }
 
@@ -679,9 +686,9 @@ function setupChannel(target, channel, serializationMode) {
 
       if (target._pendingMessage) {
         target._send(target._pendingMessage.message,
-                     target._pendingMessage.handle,
-                     target._pendingMessage.options,
-                     target._pendingMessage.callback);
+          target._pendingMessage.handle,
+          target._pendingMessage.options,
+          target._pendingMessage.callback);
       }
 
       for (let i = 0; i < queue.length; i++) {
@@ -778,9 +785,9 @@ function setupChannel(target, channel, serializationMode) {
     // will result in error message that is weakly consumable.
     // So perform a final check on message prior to sending.
     if (typeof message !== 'string' &&
-        typeof message !== 'object' &&
-        typeof message !== 'number' &&
-        typeof message !== 'boolean') {
+      typeof message !== 'object' &&
+      typeof message !== 'number' &&
+      typeof message !== 'boolean') {
       throw new ERR_INVALID_ARG_TYPE(
         'message', ['string', 'object', 'number', 'boolean'], message);
     }
@@ -841,8 +848,8 @@ function setupChannel(target, channel, serializationMode) {
         handle.setSimultaneousAccepts(true);
       }
     } else if (this._handleQueue &&
-               !(message && (message.cmd === 'NODE_HANDLE_ACK' ||
-                             message.cmd === 'NODE_HANDLE_NACK'))) {
+      !(message && (message.cmd === 'NODE_HANDLE_ACK' ||
+        message.cmd === 'NODE_HANDLE_NACK'))) {
       // Queue request anyway to avoid out-of-order messages.
       ArrayPrototypePush(this._handleQueue, {
         callback: callback,
@@ -979,11 +986,11 @@ function setupChannel(target, channel, serializationMode) {
 const INTERNAL_PREFIX = 'NODE_';
 function isInternal(message) {
   return (message !== null &&
-          typeof message === 'object' &&
-          typeof message.cmd === 'string' &&
-          message.cmd.length > INTERNAL_PREFIX.length &&
-          StringPrototypeSlice(message.cmd, 0, INTERNAL_PREFIX.length) ===
-            INTERNAL_PREFIX);
+    typeof message === 'object' &&
+    typeof message.cmd === 'string' &&
+    message.cmd.length > INTERNAL_PREFIX.length &&
+    StringPrototypeSlice(message.cmd, 0, INTERNAL_PREFIX.length) ===
+    INTERNAL_PREFIX);
 }
 
 const nop = FunctionPrototype;
@@ -1021,7 +1028,7 @@ function getValidStdio(stdio, sync) {
     if (stdio === 'ignore') {
       ArrayPrototypePush(acc, { type: 'ignore' });
     } else if (stdio === 'pipe' || stdio === 'overlapped' ||
-               (typeof stdio === 'number' && stdio < 0)) {
+      (typeof stdio === 'number' && stdio < 0)) {
       const a = {
         type: stdio === 'overlapped' ? 'overlapped' : 'pipe',
         readable: i === 0,
@@ -1061,7 +1068,7 @@ function getValidStdio(stdio, sync) {
         fd: typeof stdio === 'number' ? stdio : stdio.fd,
       });
     } else if (getHandleWrapType(stdio) || getHandleWrapType(stdio.handle) ||
-               getHandleWrapType(stdio._handle)) {
+      getHandleWrapType(stdio._handle)) {
       const handle = getHandleWrapType(stdio) ?
         stdio :
         getHandleWrapType(stdio.handle) ? stdio.handle : stdio._handle;
diff --git a/src/pipe_wrap.cc b/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)
diff --git a/src/pipe_wrap.h b/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_;
 };
 
 
diff --git a/test/async-hooks/test-pipewrap.js b/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');
diff --git a/test/parallel/test-child-process-stdin-close-event.js b/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);
+}));
