shell: support ReadableStream stdin redirect via SinkHandle

Closed
RO
19d ago
+1099-202

Fixes #18262

Problem

  • Bun.$`cmd < ${readableStream}` crashes with panic: TODO SHELL READABLE STREAM (Cmd::init_subproc_redirections, src/runtime/shell/states/Cmd.rs).
  • Shell stdin takes only a buffer that it already holds.

Fix

  • Writable::init (src/runtime/shell/subproc.rs) wraps the child's stdin pipe in a FileSink and calls assign_to_stream. A native source writes through SinkHandle::FileSink, with backpressure and no JS. Other streams use the JS pump.
  • on_process_exit tells the sink that the child exited, which cancels a stream that still produces.
  • A stream that fails fails the command: stderr gets bun: Failed to pipe ReadableStream to stdin: <error>, and the exit code is the child's own non-zero code, or 1. A stream that had already failed stops the spawn.
  • Verified: test/js/bun/shell/bunshell.test.ts (29 new tests, stock bun panics) and spawn-stdin-readable-stream.test.ts (2 new), on Linux ASAN and Windows. Notes lists the other suites.

Background

  • A FileSink is the native writer for an fd or a pipe. Bun.spawn({ stdin: stream }) uses it.
  • A SinkHandle is a native stream source's tagged pointer to its consumer. A SourceHandle points back.
  • Considered a shell-owned writer (the deleted SourceHandle::ShellWritable wiring). It repeats FileSink's backpressure and cancel logic.

Downsides

  • FileSink changes for all users: it records a synchronous write() or flush() error, so Bun.spawn({ stdin: fetchBody }) now aborts the download at that failure, and it marks an already rejected pump promise as handled. Notes lists each change.
  • Bun.spawn stays silent when its stdin stream fails. Only the shell fails the command.
  • Size: size_of::<FileSink>() grows from 344 to 352 bytes. Release bun text grows by 4,224 bytes (linux-x64).
Notes

Open PRs that touch the same code

  • Supersedes #40885, which throws at the same site (handle_template_value and the Cmd arm). If #40885 lands first, this PR rebases over it and removes its throw and its test. Earlier attempts: #30550, #33996.
  • #41209 fixes Stdio::extract_blob, which drops the window of a sliced file blob. Until it lands, a sliced file stream (Bun.file(p).slice(a, b).stream()) skips the blob collapse here (Cmd.rs, marked as a stop-gap). Whichever PR lands second removes the guard. Sites that still send the whole file and are left to #41209: stdio.rs:333, :511, :521, the direct Blob and Response arms of Cmd::init_subproc_redirections, and the path substitution in shell_body.rs handle_template_value.

The failure rule

  • The shell differs from Bun.spawn on purpose: a $ command that got a truncated stdin must not report success. Bun.spawn({ stdin: stream }) keeps exit 0 and no message (test "ReadableStream error handling").
  • FileSink keeps the failure itself (stream_source_failure), because the pipe cell lets go of a JS error when it is released. It is set when the first recorded error is not a write failure and the sink is not done: record_js_stream_error, or end_from_stream(Some(err)). That includes a chunk the sink rejects (TypeError [ERR_INVALID_ARG_TYPE]).
  • FileSink::attached_process_exited takes the failure before the exit notice. The exit notice cancels a stream that still produces, and a native source reports that cancel like a failure. Nothing is kept once the sink is done.
  • Native sources and the JS pump report a failed sink write back to the sink like an error of their own (sink.end(Some(StreamError::Error(e))), rsisAbrupt then close_with_error). One helper, FileSink::writer_io, runs all 7 writer ops that can fail and records the write error first. Without that record, a child that closes its stdin early is reported as a stream failure.
  • CmdState::WaitingWriteErr carries the exit code, like mv's state. A script that already failed reports nothing more.

Behaviour changes outside the shell (all from the FileSink record and the handled mark)

  • After a synchronous write failure, a ByteStream source such as a fetch body is cancelled. Before, the download stayed open with no consumer (new test in spawn-stdin-readable-stream.test.ts, which times out without the fix).
  • After a synchronous controller.write() or controller.flush() failure, a type: "direct" stream gets the write error as its cancel() reason. The asynchronous path already did that.
  • A failed controller.flush() now rejects Bun.write(dest, directStream). Before, it resolved with a byte count.
  • A stream that errors inside start() no longer surfaces an unhandledRejection from Bun.spawn({ stdin: stream }) (reproduced on 1.4.2, new test).
  • No effect: the error forwarded to ByteStream::cancel_from_sink, RewriterPipe::cancel_from_output, FileReader::on_close and Writable::on_close is discarded, and a pump stream already got the write error as its cancel reason.

Not part of this PR

  • < ${response} with a body that is not in memory still gives an empty stdin (#42785). Use < ${response.body} until then.
  • The output side (> ${buf} overflow) is #43196.
  • < ${Bun.file(missing).stream()} collapses to Stdio::Path, which the child opens with O_CREAT (#37641). A sliced stream of a missing file fails with ENOENT and the path.
  • The Writable::init error arm closes only the stdout and stderr ends that the spawn owns (!stdio.borrows_caller_fd(), as in js_bun_spawn_bindings.rs). The root cause, spawn_process_posix echoing the caller's fd into the spawn result, is left for a follow-up.
  • An async iterable is accepted as a < source, as Bun.spawn accepts one for stdin. The template step stores the stream that ReadableStream::from_js produced, so [Symbol.asyncIterator]() runs once.

Verification details

  • Each new test fails with its fix reverted (one build with the record, the handled mark, the exit-code rule, the done gate, the early take, the assign-time check and the script-failed gate removed: 8 shell tests and both Bun.spawn tests fail).
  • Leak probe over 9 paths, 8 iterations each: fd delta 0, live FileSink delta 0, no zombie child. Bun.spawn with a failing stream, 12 iterations: live FileSink delta 0.
  • Suites on Linux debug ASAN: all of bunshell.test.ts (465 pass), yield, shell-pipe-read-fault, epipe, shell-write-fault, throw, exec, bunshell-file, shell-hang, the spawn-stdin-* suites, filesink, bun-write, blob-write, fetch-backpressure, body-mixin-errors, fetch-abort-stream-body. Windows x64 debug: the 26 stream tests that apply and the spawn-stdin-readable-stream* suites. rust:check-all passes on 12 targets. clippy is clean.
  • Release text: 80,661,009 bytes against 80,656,785 at the merge base. A command with no stream pays one branch in on_process_exit and one in Cmd::next.
  • leak.test.ts memleak_Blob_* time out on a debug build with or without this PR: the test's own 128 KiB Array.join takes 1.3 s per iteration in debug JSC. On a Windows debug build, filesink.test.ts "Bun.spawn stdin pipe with an unref'd child" fails the same way on main (the canary release passes).

Earlier notes

  • A second panic, Unimplemented stdin readable_stream, was in Writable::init. A stream redirect is stdin only. A locked or disturbed stream throws ERR_INVALID_STATE. A builtin command throws. A shell Cmd finishes when it has an exit code and each piped stdio is closed, so on_process_exit also closes the Cmd's stdin.
  • Deleted: SourceHandle::ShellWritable, shell::Writable::on_close, and the block in spawn_maybe_sync_impl that wired them. Shell stdin is never Stdio::Pipe, so that block could not run. WritableInitError::UnexpectedCreatingStdin and its panic are replaced by Sys and StreamAssign.
  • *out_subproc is now written once the Subprocess is initialised. The stream assignment can run user JS. If that JS ends the process, the Cmd must not hold a pointer to memory that is not initialised.
  • FileSink::attached_process_exited(&RefPtr<FileSink>, ..) is a safe wrapper over on_attached_process_exit. The Bun.spawn call site that holds a RefPtr uses it too. The other call site holds a weak NonNull and is unchanged.
  • New unsafe in the shell code: one &mut *pipe.as_ptr() to call assign_to_stream(&mut self) on the sink that init just created (the idiom of subprocess::Writable::pipe_sink_mut), and one (*this).stdin.take_pipe() in on_process_exit, which takes a raw this on main.
  • Bun.file(path).stream() and a fully buffered stream collapse to a blob (to_any_blob) and take the StaticPipeWriter path.
  • A pipeline member whose stream fails to start: with stderr on a pipe the error line is written asynchronously, the next member took the closed fd number, the IOWriter close hit it, and the pipeline hung (test runs that case in a child process).
  • Rebased on main after #41436, #40516 and #40478, and again after #42288 (Yield::failed() is now Yield::Failed(NodeId)). The two stream rejections in Builtin::init_redirections name the failing command. A later pipeline member that throws while an earlier member reads a stream: Interpreter::fail kills the reader, and the child exit cancels the stream.
  • SourceHandle::ShellWritable was the last public edge to FileSink, SinkHandle, ReadableStream, the JS sinks and the S3 MultiPartUpload. With it deleted, the workspace unreachable_pub and dead_code lints apply to them. 61 items become pub(crate). MultiPartUpload.global_this (never read) is removed. Shell Writable::Fd drops the fd that nothing read. Shell Writable::Memfd is POSIX only.

no test proof · iteration 6 · platform-specific test(s) that do not run on this machine, deferring to CI, which covers all platforms: test/js/bun/spawn/spawn-stdin-readable-stream.test.ts, test/js/bun/shell/bunshell.test.ts