fix(openfiles): share descriptors across cloned shell contexts

brush-core/src/commands.rs

@@ -221,7 +221,7 @@ pub fn compose_std_command<S: AsRef<OsStr>, SE: extensions::ShellExtensions>(
     match context.try_fd(OpenFiles::STDIN_FD) {
         Some(OpenFile::Stdin(_)) | None => (),
         Some(stdin_file) => {
-            let as_stdio: Stdio = stdin_file.into();
+            let as_stdio: Stdio = stdin_file.try_into()?;
             cmd.stdin(as_stdio);
         }
     }
@@ -230,7 +230,7 @@ pub fn compose_std_command<S: AsRef<OsStr>, SE: extensions::ShellExtensions>(
     match context.try_fd(OpenFiles::STDOUT_FD) {
         Some(OpenFile::Stdout(_)) | None => (),
         Some(stdout_file) => {
-            let as_stdio: Stdio = stdout_file.into();
+            let as_stdio: Stdio = stdout_file.try_into()?;
             cmd.stdout(as_stdio);
         }
     }
@@ -239,7 +239,7 @@ pub fn compose_std_command<S: AsRef<OsStr>, SE: extensions::ShellExtensions>(
     match context.try_fd(OpenFiles::STDERR_FD) {
         Some(OpenFile::Stderr(_)) | None => {}
         Some(stderr_file) => {
-            let as_stdio: Stdio = stderr_file.into();
+            let as_stdio: Stdio = stderr_file.try_into()?;
             cmd.stderr(as_stdio);
         }
     }
@@ -714,9 +714,6 @@ pub(crate) async fn invoke_shell_function(
 
     let positional_args = args.iter().map(|a| a.to_string());
 
-    // Pass through open files.
-    let params = context.params.clone();
-
     // Note that we're going deeper. Once we do this, we need to make sure we don't bail early
     // before "exiting" the function.
     context.shell.enter_function(
@@ -726,11 +723,11 @@ pub(crate) async fn invoke_shell_function(
         &context.params,
     )?;
 
-    // Invoke the function.
-    let result = body.execute(context.shell, &params).await;
-
-    // Clean up parameters so any owned files are closed.
-    drop(params);
+    // A function executes within the current shell process and shares its caller's open files,
+    // so the parameters are passed through by shared reference rather than cloned. This prevents
+    // direct mutation of the caller's `ExecutionParameters` open-file table, though the function
+    // may still change the shell's persistent open files via builtins (e.g. `exec`).
+    let result = body.execute(context.shell, &context.params).await;
 
     // We've come back out, reflect it.
     context.shell.leave_function()?;

brush-core/src/interp.rs

@@ -1736,9 +1736,7 @@ pub(crate) async fn setup_redirect(
 
                     let fd_num = specified_fd_num.unwrap_or(default_fd_if_unspecified);
 
-                    if let Some(f) = params.try_fd(shell, *fd) {
-                        let target_file = f.try_clone()?;
-
+                    if let Some(target_file) = params.try_fd(shell, *fd) {
                         params.open_files.set_fd(fd_num, target_file);
                     } else {
                         return Err(error::ErrorKind::BadFileDescriptor(*fd).into());
@@ -1779,10 +1777,8 @@ pub(crate) async fn setup_redirect(
                             .parse::<ShellFd>()
                             .map_err(|_| error::ErrorKind::InvalidRedirection)?;
 
-                        // Duplicate the fd.
-                        let target_file = if let Some(f) = params.try_fd(shell, source_fd_num) {
-                            f.try_clone()?
-                        } else {
+                        // Reference the same open file as the source fd (shared handle; no OS-level duplication).
+                        let Some(target_file) = params.try_fd(shell, source_fd_num) else {
                             return Err(error::ErrorKind::BadFileDescriptor(source_fd_num).into());
                         };
 
@@ -1817,7 +1813,7 @@ pub(crate) async fn setup_redirect(
                                 subshell_cmd,
                             )?;
 
-                            let target_file = substitution_file.try_clone()?;
+                            let target_file = substitution_file.clone();
                             params.open_files.set_fd(substitution_fd, substitution_file);
 
                             let fd_num = specified_fd_num
@@ -1895,7 +1891,7 @@ fn setup_redirect_output_and_error_to(
             )
         })?;
 
-    let stderr_file = stdout_file.try_clone()?;
+    let stderr_file = stdout_file.clone();
 
     params.open_files.set_fd(OpenFiles::STDOUT_FD, stdout_file);
     params.open_files.set_fd(OpenFiles::STDERR_FD, stderr_file);

brush-core/src/openfiles.rs

@@ -3,10 +3,10 @@
 use std::collections::HashMap;
 use std::io::IsTerminal;
 use std::process::Stdio;
+use std::sync::Arc;
 
 use crate::ShellFd;
 use crate::error;
-use crate::ioutils;
 use crate::sys;
 
 /// A trait representing a stream that can be read from and written to.
@@ -30,6 +30,15 @@ pub trait Stream: std::io::Read + std::io::Write + Send + Sync {
 }
 
 /// Represents a file open in a shell context.
+///
+/// File- and pipe-backed variants hold their handle behind an [`Arc`], so cloning an `OpenFile`
+/// shares the underlying descriptor by reference count. The shell opens a fresh context for each
+/// subshell, command substitution, background job, and function call and runs them as in-process
+/// tasks against one process-wide descriptor table (rather than via `fork(2)` like a traditional
+/// shell); sharing the descriptor keeps deeply nested or highly concurrent execution from
+/// exhausting that table. A descriptor is duplicated for real only when an independently owned
+/// copy is needed to hand to an external child process — see [`OpenFile::try_clone_to_owned`] and
+/// the `Stdio` conversion below.
 pub enum OpenFile {
     /// The original standard input this process was started with.
     Stdin(std::io::Stdin),
@@ -38,11 +47,11 @@ pub enum OpenFile {
     /// The original standard error this process was started with.
     Stderr(std::io::Stderr),
     /// A file open for reading or writing.
-    File(std::fs::File),
+    File(Arc<std::fs::File>),
     /// A read end of a pipe.
-    PipeReader(std::io::PipeReader),
+    PipeReader(Arc<std::io::PipeReader>),
     /// A write end of a pipe.
-    PipeWriter(std::io::PipeWriter),
+    PipeWriter(Arc<std::io::PipeWriter>),
     /// A custom stream.
     Stream(Box<dyn Stream>),
 }
@@ -90,16 +99,22 @@ impl<'de> serde::Deserialize<'de> for OpenFile {
 /// Returns an open file that will discard all I/O.
 pub fn null() -> Result<OpenFile, error::Error> {
     let file = sys::fs::open_null_file()?;
-    Ok(OpenFile::File(file))
+    Ok(file.into())
 }
 
 impl Clone for OpenFile {
     fn clone(&self) -> Self {
-        // If we fail to clone the open file for any reason, we return a special file
-        // that discards all I/O. This allows us to avoid fatally erroring out.
-        self.try_clone().unwrap_or_else(|_err| {
-            ioutils::FailingReaderWriter::new("failed to duplicate open file").into()
-        })
+        match self {
+            Self::Stdin(_) => std::io::stdin().into(),
+            Self::Stdout(_) => std::io::stdout().into(),
+            Self::Stderr(_) => std::io::stderr().into(),
+            // File and pipe handles are shared by reference count; cloning never issues a
+            // syscall and so cannot fail.
+            Self::File(f) => Self::File(Arc::clone(f)),
+            Self::PipeReader(r) => Self::PipeReader(Arc::clone(r)),
+            Self::PipeWriter(w) => Self::PipeWriter(Arc::clone(w)),
+            Self::Stream(s) => Self::Stream(s.clone_box()),
+        }
     }
 }
 
@@ -118,22 +133,8 @@ impl std::fmt::Display for OpenFile {
 }
 
 impl OpenFile {
-    /// Tries to duplicate the open file.
-    pub fn try_clone(&self) -> Result<Self, std::io::Error> {
-        let result = match self {
-            Self::Stdin(_) => std::io::stdin().into(),
-            Self::Stdout(_) => std::io::stdout().into(),
-            Self::Stderr(_) => std::io::stderr().into(),
-            Self::File(f) => f.try_clone()?.into(),
-            Self::PipeReader(f) => f.try_clone()?.into(),
-            Self::PipeWriter(f) => f.try_clone()?.into(),
-            Self::Stream(s) => Self::Stream(s.clone_box()),
-        };
-
-        Ok(result)
-    }
-
-    /// Converts the open file into an `OwnedFd`.
+    /// Converts the open file into an `OwnedFd`. For shared file/pipe handles this materializes
+    /// a real duplicate via `dup(2)` so the caller receives an independently owned descriptor.
     #[cfg(unix)]
     pub(crate) fn try_clone_to_owned(self) -> Result<std::os::fd::OwnedFd, error::Error> {
         use std::os::fd::AsFd as _;
@@ -142,9 +143,9 @@ impl OpenFile {
             Self::Stdin(f) => Ok(f.as_fd().try_clone_to_owned()?),
             Self::Stdout(f) => Ok(f.as_fd().try_clone_to_owned()?),
             Self::Stderr(f) => Ok(f.as_fd().try_clone_to_owned()?),
-            Self::File(f) => Ok(f.into()),
-            Self::PipeReader(r) => Ok(std::os::fd::OwnedFd::from(r)),
-            Self::PipeWriter(w) => Ok(std::os::fd::OwnedFd::from(w)),
+            Self::File(f) => Ok(f.as_fd().try_clone_to_owned()?),
+            Self::PipeReader(r) => Ok(r.as_fd().try_clone_to_owned()?),
+            Self::PipeWriter(w) => Ok(w.as_fd().try_clone_to_owned()?),
             Self::Stream(s) => s.try_clone_to_owned(),
         }
     }
@@ -212,34 +213,37 @@ impl From<std::io::Stderr> for OpenFile {
 
 impl From<std::fs::File> for OpenFile {
     fn from(file: std::fs::File) -> Self {
-        Self::File(file)
+        Self::File(Arc::new(file))
     }
 }
 
 impl From<std::io::PipeReader> for OpenFile {
     fn from(reader: std::io::PipeReader) -> Self {
-        Self::PipeReader(reader)
+        Self::PipeReader(Arc::new(reader))
     }
 }
 
 impl From<std::io::PipeWriter> for OpenFile {
     fn from(writer: std::io::PipeWriter) -> Self {
-        Self::PipeWriter(writer)
+        Self::PipeWriter(Arc::new(writer))
     }
 }
 
-impl From<OpenFile> for Stdio {
-    fn from(open_file: OpenFile) -> Self {
+impl TryFrom<OpenFile> for Stdio {
+    type Error = error::Error;
+
+    fn try_from(open_file: OpenFile) -> Result<Self, Self::Error> {
+        // File and pipe handles are shared behind an `Arc`, so the descriptor cannot be moved
+        // out; duplicate it to give the child an independently owned descriptor. Duplication can
+        // fail (e.g. under descriptor exhaustion), so the conversion is fallible and the error is
+        // surfaced to the caller rather than silently degrading the child's streams.
         match open_file {
-            OpenFile::Stdin(_) => Self::inherit(),
-            OpenFile::Stdout(_) => Self::inherit(),
-            OpenFile::Stderr(_) => Self::inherit(),
-            OpenFile::File(f) => f.into(),
-            OpenFile::PipeReader(f) => f.into(),
-            OpenFile::PipeWriter(f) => f.into(),
-            // NOTE: Custom streams cannot be converted to `Stdio`; we do our best here
-            // and return a null device instead.
-            OpenFile::Stream(_) => Self::null(),
+            OpenFile::Stdin(_) | OpenFile::Stdout(_) | OpenFile::Stderr(_) => Ok(Self::inherit()),
+            OpenFile::File(f) => Ok(f.try_clone()?.into()),
+            OpenFile::PipeReader(r) => Ok(r.try_clone()?.into()),
+            OpenFile::PipeWriter(w) => Ok(w.try_clone()?.into()),
+            // Custom streams have no descriptor to hand to a child process.
+            OpenFile::Stream(_) => Ok(Self::null()),
         }
     }
 }
@@ -254,8 +258,10 @@ impl std::io::Read for OpenFile {
             Self::Stderr(_) => Err(std::io::Error::other(
                 error::ErrorKind::OpenFileNotReadable("stderr"),
             )),
-            Self::File(f) => f.read(buf),
-            Self::PipeReader(reader) => reader.read(buf),
+            // The handle is shared behind an `Arc`; read through a shared reference (`&File`
+            // and `&PipeReader` both implement `Read`).
+            Self::File(f) => f.as_ref().read(buf),
+            Self::PipeReader(reader) => reader.as_ref().read(buf),
             Self::PipeWriter(_) => Err(std::io::Error::other(
                 error::ErrorKind::OpenFileNotReadable("pipe writer"),
             )),
@@ -272,11 +278,13 @@ impl std::io::Write for OpenFile {
             )),
             Self::Stdout(f) => f.write(buf),
             Self::Stderr(f) => f.write(buf),
-            Self::File(f) => f.write(buf),
+            // The handle is shared behind an `Arc`; write through a shared reference (`&File`
+            // and `&PipeWriter` both implement `Write`).
+            Self::File(f) => f.as_ref().write(buf),
             Self::PipeReader(_) => Err(std::io::Error::other(
                 error::ErrorKind::OpenFileNotWritable("pipe reader"),
             )),
-            Self::PipeWriter(writer) => writer.write(buf),
+            Self::PipeWriter(writer) => writer.as_ref().write(buf),
             Self::Stream(s) => s.write(buf),
         }
     }
@@ -286,9 +294,9 @@ impl std::io::Write for OpenFile {
             Self::Stdin(_) => Ok(()),
             Self::Stdout(f) => f.flush(),
             Self::Stderr(f) => f.flush(),
-            Self::File(f) => f.flush(),
+            Self::File(f) => f.as_ref().flush(),
             Self::PipeReader(_) => Ok(()),
-            Self::PipeWriter(writer) => writer.flush(),
+            Self::PipeWriter(writer) => writer.as_ref().flush(),
             Self::Stream(s) => s.flush(),
         }
     }

brush-core/src/shell/fs.rs

@@ -204,7 +204,7 @@ impl<SE: crate::extensions::ShellExtensions> crate::Shell<SE> {
             && let Ok(fd_num) = filename.to_string_lossy().to_string().parse::<ShellFd>()
             && let Some(open_file) = params.try_fd(self, fd_num)
         {
-            return open_file.try_clone();
+            return Ok(open_file);
         }
 
         Ok(options.open(path_to_open)?.into())

brush-core/src/shell/io.rs

@@ -64,9 +64,7 @@ impl<SE: extensions::ShellExtensions> crate::Shell<SE> {
         };
 
         // If we have a valid trace file, write to it.
-        if let Some(trace_file) = trace_file
-            && let Ok(mut trace_file) = trace_file.try_clone()
-        {
+        if let Some(mut trace_file) = trace_file {
             let _ = writeln!(trace_file, "{prefix}{}", command.as_ref());
         }
     }

brush-shell/tests/cases/compat/fd_management.yaml

@@ -0,0 +1,72 @@
+name: "File descriptor management"
+# Regression tests for issue #1173: nested/concurrent shell constructs must not
+# exhaust the process-wide file-descriptor table.
+#
+# brush runs subshells, command substitutions, background jobs, and function
+# call contexts as in-process tasks rather than via fork(2), so they all share a
+# single process-wide descriptor table. Each context inherits the caller's open
+# files; sharing those descriptors -- rather than giving each context its own
+# copy -- is what keeps deeply nested or highly concurrent scripts within the
+# limit. A real shell gives each forked subshell its own descriptor table and so
+# stays well within the limit too.
+#
+# Each case lowers the descriptor limit with `ulimit -n` and runs a workload that
+# stays within the limit only when inherited descriptors are shared across
+# contexts; if each context copied them, the workload would exceed the limit and
+# fail with "Too many open files" and truncated output. The limits leave the
+# correct (sharing) behavior several-fold headroom. These cases assume Unix
+# `ulimit -n` semantics.
+cases:
+  # Concurrent fan-out: every backgrounded job is alive simultaneously (they are
+  # all launched before `wait`), so copying the inherited descriptors per job
+  # would multiply the live descriptor count by the number of jobs.
+  - name: "Many concurrent background jobs each using command substitution"
+    incompatible_platforms: ["wasi"]
+    args:
+      - "-c"
+      - |
+        ulimit -n 256 || exit 1
+        emit() {
+          local v
+          v=$(echo "$1")
+          echo "$v"
+        }
+        ( for i in $(seq 1 60); do emit "$i" & done; wait ) | wc -l | tr -d ' '
+
+  # nvm-style: backgrounded jobs in a subshell, piped to a filter, each touching
+  # /dev/null via redirects -- the precise shape that exhausted descriptors in
+  # `nvm ls`.
+  - name: "Subshell of backgrounded jobs with redirects piped to a filter"
+    incompatible_platforms: ["wasi"]
+    args:
+      - "-c"
+      - |
+        ulimit -n 256 || exit 1
+        work() {
+          local v
+          v=$(echo "alias-$1" 2>/dev/null)
+          echo "$v" 2>/dev/null
+        }
+        ( for i in $(seq 1 50); do work "$i" 2>/dev/null & done; wait ) | sort | wc -l | tr -d ' '
+
+  # Nested depth: each command-substitution level keeps its ancestors' contexts
+  # alive while it runs, so copying inherited descriptors would grow the live
+  # count with nesting depth.
+  - name: "Deeply nested command substitution with redirects stays within the fd limit"
+    incompatible_platforms: ["wasi"]
+    args:
+      - "-c"
+      - |
+        ulimit -n 256 || exit 1
+        rec() {
+          local n=$1
+          if [ "$n" -le 0 ]; then
+            printf 'x' 2>/dev/null
+            return
+          fi
+          local inner
+          inner=$(rec $((n - 1)) 2>/dev/null)
+          printf '%s' "$inner" 2>/dev/null
+        }
+        out=$(rec 40)
+        echo "${#out}"