fix(interp): run function pipeline stages as background tasks, not inline

brush-core/src/commands.rs

@@ -504,6 +504,78 @@ impl<'a, SE: extensions::ShellExtensions> SimpleCommand<'a, SE> {
     async fn execute_via_function(
         self,
         func_registration: functions::Registration,
+    ) -> Result<ExecutionSpawnResult, error::Error> {
+        match self.shell {
+            ShellForCommand::OwnedShell { target, .. } => {
+                Ok(Self::execute_via_function_in_owned_shell(
+                    *target,
+                    self.params,
+                    func_registration,
+                    self.command_name,
+                    self.args,
+                ))
+            }
+            ShellForCommand::ParentShell(..) => {
+                self.execute_via_function_in_parent_shell(func_registration)
+                    .await
+            }
+        }
+    }
+
+    // As a non-last pipeline stage, a function must run as a background task
+    // (mirroring `execute_via_builtin_in_owned_shell` above) rather than
+    // being awaited inline in `spawn_pipeline_processes`'s own per-stage
+    // loop: awaiting it inline there blocks that loop until the function's
+    // body fully returns, so the *next* pipeline stage (the one that would
+    // actually drain this function's stdout pipe) is never even spawned
+    // until this one is done. A function that writes more to its stdout
+    // pipe than the OS pipe buffer holds (~64KiB on Linux) before returning
+    // then blocks on that `write()` forever, since nothing is reading the
+    // other end yet -- a plain deadlock, not a slow completion. Confirmed
+    // empirically with a two-line repro having nothing to do with any
+    // external command or builtin: `f() { for i in $(seq 1 5000); do echo
+    // line $i; done; }; f | cat` hangs indefinitely. Real bash always runs
+    // every pipeline stage as its own concurrently-running process, so this
+    // case never arises there.
+    fn execute_via_function_in_owned_shell(
+        mut shell: Shell<SE>,
+        params: ExecutionParameters,
+        func_registration: functions::Registration,
+        command_name: String,
+        args: Vec<CommandArg>,
+    ) -> ExecutionSpawnResult {
+        let last_arg = Self::take_last_arg(&args);
+        let join_handle = tokio::task::spawn_blocking(move || {
+            let cmd_context = ExecutionContext {
+                shell: &mut shell,
+                command_name,
+                params,
+            };
+
+            let rt = tokio::runtime::Handle::current();
+            let spawn_result =
+                rt.block_on(invoke_shell_function(func_registration, cmd_context, &args[1..]));
+
+            // Update $_ after command execution.
+            shell.update_last_arg_variable(last_arg);
+
+            match spawn_result? {
+                ExecutionSpawnResult::Completed(result) => Ok(result),
+                ExecutionSpawnResult::StartedProcess(_) | ExecutionSpawnResult::StartedTask(_) => {
+                    // invoke_shell_function() only ever returns Completed
+                    // today; this guards against a silent behavior change
+                    // rather than dropping already-started work.
+                    error::unimp("invoke_shell_function returned an unexpected spawn result")
+                }
+            }
+        });
+
+        ExecutionSpawnResult::StartedTask(join_handle)
+    }
+
+    async fn execute_via_function_in_parent_shell(
+        self,
+        func_registration: functions::Registration,
     ) -> Result<ExecutionSpawnResult, error::Error> {
         let mut shell = self.shell;
         let last_arg = Self::take_last_arg(&self.args);

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

@@ -57,4 +57,20 @@ cases:
   - name: "printf broken pipe returns 141 in PIPESTATUS"
     stdin: |
       printf '%s\n' {0..10000} | x=1
+
+  - name: "Function stage writing more than a pipe buffer before the next stage is spawned"
+    # Regression test: a shell function used as a non-last pipeline stage
+    # must run concurrently with the rest of the pipeline, not be awaited
+    # to completion before the next stage is even spawned. Otherwise, once
+    # the function writes more to its stdout pipe than the OS pipe buffer
+    # holds (commonly 64KiB on Linux) before returning, the write blocks
+    # forever because nothing is reading the other end yet.
+    stdin: |
+      big() {
+        local i
+        for i in $(seq 1 5000); do
+          echo "padding line number ${i} to exceed the pipe buffer size"
+        done
+      }
+      big | wc -l
       echo "Last: $?, PIPESTATUS: ${PIPESTATUS[*]}"