shell: support ReadableStream stdin redirect via SinkHandle
Closed
RO
19d ago+1099+1099+0123456789012345678901234567890123456789-202-202-012345678901234567890123456789
Fixes #18262
Problem
Bun.$`cmd < ${readableStream}`crashes withpanic: 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 aFileSinkand callsassign_to_stream. A native source writes throughSinkHandle::FileSink, with backpressure and no JS. Other streams use the JS pump.on_process_exittells 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) andspawn-stdin-readable-stream.test.ts(2 new), on Linux ASAN and Windows. Notes lists the other suites.
Background
- A
FileSinkis the native writer for an fd or a pipe.Bun.spawn({ stdin: stream })uses it. - A
SinkHandleis a native stream source's tagged pointer to its consumer. ASourceHandlepoints back. - Considered a shell-owned writer (the deleted
SourceHandle::ShellWritablewiring). It repeatsFileSink's backpressure and cancel logic.
Downsides
FileSinkchanges for all users: it records a synchronouswrite()orflush()error, soBun.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.spawnstays silent when its stdin stream fails. Only the shell fails the command.- Size:
size_of::<FileSink>()grows from 344 to 352 bytes. Releasebuntext 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_valueand theCmdarm). 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 directBlobandResponsearms ofCmd::init_subproc_redirections, and the path substitution inshell_body.rshandle_template_value.
The failure rule
- The shell differs from
Bun.spawnon 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"). FileSinkkeeps 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, orend_from_stream(Some(err)). That includes a chunk the sink rejects (TypeError [ERR_INVALID_ARG_TYPE]).FileSink::attached_process_exitedtakes 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))),rsisAbruptthenclose_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::WaitingWriteErrcarries the exit code, likemv'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
ByteStreamsource such as a fetch body is cancelled. Before, the download stayed open with no consumer (new test inspawn-stdin-readable-stream.test.ts, which times out without the fix). - After a synchronous
controller.write()orcontroller.flush()failure, atype: "direct"stream gets the write error as itscancel()reason. The asynchronous path already did that. - A failed
controller.flush()now rejectsBun.write(dest, directStream). Before, it resolved with a byte count. - A stream that errors inside
start()no longer surfaces anunhandledRejectionfromBun.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_closeandWritable::on_closeis 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 toStdio::Path, which the child opens withO_CREAT(#37641). A sliced stream of a missing file fails withENOENTand the path.- The
Writable::initerror arm closes only the stdout and stderr ends that the spawn owns (!stdio.borrows_caller_fd(), as injs_bun_spawn_bindings.rs). The root cause,spawn_process_posixechoing the caller's fd into the spawn result, is left for a follow-up. - An async iterable is accepted as a
<source, asBun.spawnaccepts one forstdin. The template step stores the stream thatReadableStream::from_jsproduced, 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.spawntests fail). - Leak probe over 9 paths, 8 iterations each: fd delta 0, live
FileSinkdelta 0, no zombie child.Bun.spawnwith a failing stream, 12 iterations: liveFileSinkdelta 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, thespawn-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 thespawn-stdin-readable-stream*suites.rust:check-allpasses 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_exitand one inCmd::next. leak.test.tsmemleak_Blob_*time out on a debug build with or without this PR: the test's own 128 KiBArray.jointakes 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 inWritable::init. A stream redirect is stdin only. A locked or disturbed stream throwsERR_INVALID_STATE. A builtin command throws. A shellCmdfinishes when it has an exit code and each piped stdio is closed, soon_process_exitalso closes theCmd's stdin. - Deleted:
SourceHandle::ShellWritable,shell::Writable::on_close, and the block inspawn_maybe_sync_implthat wired them. Shell stdin is neverStdio::Pipe, so that block could not run.WritableInitError::UnexpectedCreatingStdinand its panic are replaced bySysandStreamAssign. *out_subprocis now written once theSubprocessis initialised. The stream assignment can run user JS. If that JS ends the process, theCmdmust not hold a pointer to memory that is not initialised.FileSink::attached_process_exited(&RefPtr<FileSink>, ..)is a safe wrapper overon_attached_process_exit. TheBun.spawncall site that holds aRefPtruses it too. The other call site holds a weakNonNulland is unchanged.- New
unsafein the shell code: one&mut *pipe.as_ptr()to callassign_to_stream(&mut self)on the sink thatinitjust created (the idiom ofsubprocess::Writable::pipe_sink_mut), and one(*this).stdin.take_pipe()inon_process_exit, which takes a rawthison main. Bun.file(path).stream()and a fully buffered stream collapse to a blob (to_any_blob) and take theStaticPipeWriterpath.- 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
IOWriterclose 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 nowYield::Failed(NodeId)). The two stream rejections inBuiltin::init_redirectionsname the failing command. A later pipeline member that throws while an earlier member reads a stream:Interpreter::failkills the reader, and the child exit cancels the stream. SourceHandle::ShellWritablewas the last public edge toFileSink,SinkHandle,ReadableStream, the JS sinks and the S3MultiPartUpload. With it deleted, the workspaceunreachable_pubanddead_codelints apply to them. 61 items becomepub(crate).MultiPartUpload.global_this(never read) is removed. ShellWritable::Fddrops the fd that nothing read. ShellWritable::Memfdis 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