From f484e35f39aaf8996fb37114f898c7f67469dda2 Mon Sep 17 00:00:00 2001 From: TseIan Date: Sun, 6 Sep 2026 09:46:33 +0800 Subject: [PATCH] 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: https://github.com/nodejs/node/issues/25131 --- doc/api/child_process.md | 5 + lib/internal/child_process.js | 13 ++- src/pipe_wrap.cc | 101 +++++++++++++++++- src/pipe_wrap.h | 12 +++ test/async-hooks/test-pipewrap.js | 7 +- .../test-child-process-stdin-close-event.js | 71 ++++++++++++ 6 files changed, 204 insertions(+), 5 deletions(-) create mode 100644 test/parallel/test-child-process-stdin-close-event.js diff --git a/doc/api/child_process.md b/doc/api/child_process.md index 138fb52f8612..2c9beedeebc3 100644 --- a/doc/api/child_process.md +++ b/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. diff --git a/lib/internal/child_process.js b/lib/internal/child_process.js index a12b2954db81..eb8ff9462012 100644 --- a/lib/internal/child_process.js +++ b/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++; diff --git a/src/pipe_wrap.cc b/src/pipe_wrap.cc index 48bd22e301d8..506c51317a09 100644 --- 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,7 @@ void PipeWrap::Initialize(Local 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& args) { args.GetReturnValue().Set(err); } +void PipeWrap::WatchPeerClose(const FunctionCallbackInfo& 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()->Value(); + Environment* env = wrap->env(); + Isolate* isolate = env->isolate(); + v8::HandleScope handle_scope(isolate); + v8::Context::Scope context_scope(env->context()); + Local obj = wrap->object(); + + // UnwatchPeerClose + if (!enable) { + if (obj->GetInternalField(kPeerCloseCallbackField) + .As() + ->IsUndefined()) { + return; + } + + obj->SetInternalField(kPeerCloseCallbackField, v8::Undefined(isolate)); + uv_read_stop(wrap->stream()); + return; + } + + if (!wrap->IsAlive()) { + return; + } + if (!obj->GetInternalField(kPeerCloseCallbackField) + .As() + ->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(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 obj = wrap->object(); + + // Check if callback is set + if (obj->GetInternalField(kPeerCloseCallbackField) + .As() + ->IsUndefined()) { + return; + } + Local cb_value = + obj->GetInternalField(kPeerCloseCallbackField).As(); + Local cb = cb_value.As(); + // 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& args) { Environment* env = Environment::GetCurrent(args); @@ -258,7 +358,6 @@ void PipeWrap::Connect(const FunctionCallbackInfo& 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 index c0722b63d853..f6191768e879 100644 --- a/src/pipe_wrap.h +++ b/src/pipe_wrap.h @@ -40,6 +40,11 @@ class PipeWrap : public ConnectionWrap { IPC }; + enum InternalFields { + kPeerCloseCallbackField = LibuvStreamWrap::kInternalFieldCount, + kInternalFieldCount + }; + static v8::MaybeLocal Instantiate(Environment* env, AsyncWrap* parent, SocketType type); @@ -64,6 +69,13 @@ class PipeWrap : public ConnectionWrap { static void Listen(const v8::FunctionCallbackInfo& args); static void Connect(const v8::FunctionCallbackInfo& args); static void Open(const v8::FunctionCallbackInfo& args); + static void WatchPeerClose(const v8::FunctionCallbackInfo& 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( diff --git a/test/async-hooks/test-pipewrap.js b/test/async-hooks/test-pipewrap.js index 7ea5f38adc85..2d3f95457dc8 100644 --- 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 new file mode 100644 index 000000000000..dbbf0078b39b --- /dev/null +++ b/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()); + })); +}));