diff --git a/src/runtime/shell/Builtin.rs b/src/runtime/shell/Builtin.rs index 416dc4d3570a..b58578c16c75 100644 --- a/src/runtime/shell/Builtin.rs +++ b/src/runtime/shell/Builtin.rs @@ -726,6 +726,13 @@ impl Builtin { if redirect.stderr() { me.stderr = BuiltinIO::ArrayBuf { buf: mk(), i: 0 }; } + } else if crate::webcore::ReadableStream::is_readable_stream(jsval) { + let name = Self::of(interp, cmd).kind.as_str(); + let _ = global.throw(format_args!( + "ReadableStream cannot be redirected to a builtin command ('{name}'). \ + Use an external command or buffer the stream first", + )); + return Some(Yield::failed()); } else if let Some(body) = crate::webcore::body::Value::from_request_or_response(jsval) { diff --git a/src/runtime/shell/states/Cmd.rs b/src/runtime/shell/states/Cmd.rs index 9af29d8db51d..5088701eac07 100644 --- a/src/runtime/shell/states/Cmd.rs +++ b/src/runtime/shell/states/Cmd.rs @@ -758,8 +758,32 @@ impl Cmd { STDERR_NO as i32, )?; } - } else if crate::webcore::ReadableStream::from_js(jsval, global)?.is_some() { - panic!("TODO SHELL READABLE STREAM"); + } else if let Some(mut stream) = + crate::webcore::ReadableStream::from_js(jsval, global)? + { + if !flags.stdin() { + return Err(global.throw(format_args!( + "ReadableStream cannot be used for stdout or stderr; \ + only '< ${{...}}' (stdin) is supported" + ))); + } + if stream.is_locked(global) || stream.is_disturbed(global) { + return Err(global + .err( + crate::jsc::ErrorCode::INVALID_STATE, + format_args!( + "ReadableStream redirected to stdin has already been used" + ), + ) + .throw()); + } + // Fully-buffered / not-yet-started file-backed streams collapse + // to a blob and take the existing `StaticPipeWriter` path. + if let Some(blob) = stream.to_any_blob(global) { + stdio[STDIN_NO].extract_blob(global, blob, STDIN_NO as i32)?; + } else { + stdio[STDIN_NO] = Stdio::ReadableStream(stream); + } } else if let Some(req) = jsval.as_::() { // SAFETY: `as_` returns a live JSC-owned `*mut Response`; // `get_body_value` is `&self`. diff --git a/src/runtime/shell/subproc.rs b/src/runtime/shell/subproc.rs index a72a9cf4a09f..3d48384f9bf6 100644 --- a/src/runtime/shell/subproc.rs +++ b/src/runtime/shell/subproc.rs @@ -131,17 +131,35 @@ pub type StdioResult = Option; pub struct FileSinkPtr(core::ptr::NonNull); impl FileSinkPtr { - /// Adopt the +1 ref returned by `FileSink::create*`. - /// - /// # Safety - /// `ptr` is non-null, points to a live `FileSink` from - /// `FileSink::create*`, and the caller transfers its single owned ref to - /// this handle. + /// Create a `FileSink` writing to `fd` and adopt the create-ref. + #[cfg(not(windows))] + fn create(event_loop: EventLoopHandle, fd: Fd) -> Self { + Self( + core::ptr::NonNull::new(FileSink::create(event_loop, fd)) + .expect("FileSink::create returns non-null"), + ) + } + + /// Create a `FileSink` around `pipe` (ownership of the `Box` + /// transfers to the sink's writer) and adopt the create-ref. #[cfg(windows)] - #[inline] - unsafe fn adopt(ptr: *mut FileSink) -> Self { - // SAFETY: caller contract — `ptr` is non-null. - Self(unsafe { core::ptr::NonNull::new_unchecked(ptr) }) + fn create_with_pipe( + event_loop: EventLoopHandle, + pipe: *mut bun_sys::windows::libuv::Pipe, + ) -> Self { + Self( + core::ptr::NonNull::new(FileSink::create_with_pipe(event_loop, pipe)) + .expect("FileSink::create_with_pipe returns non-null"), + ) + } + + /// The attached process exited: cancel the assigned stream and close the + /// writer. + fn on_attached_process_exit(&self, status: &bun_process::Status) { + // SAFETY: `self.0` is the canonical `FileSink::create*` pointer (full + // write+dealloc provenance, never laundered through a reference) and + // the ref this handle owns keeps the sink live across the call. + unsafe { FileSink::on_attached_process_exit(self.0.as_ptr(), status) }; } } @@ -149,7 +167,7 @@ impl core::ops::Deref for FileSinkPtr { type Target = FileSink; #[inline] fn deref(&self) -> &FileSink { - // SAFETY: `adopt` contract — `self.0` is a live `FileSink` from + // SAFETY: constructor invariant — `self.0` is a live `FileSink` from // `FileSink::create*`; the held intrusive ref keeps it alive for `'_`. unsafe { self.0.as_ref() } } @@ -158,7 +176,7 @@ impl core::ops::Deref for FileSinkPtr { impl core::ops::DerefMut for FileSinkPtr { #[inline] fn deref_mut(&mut self) -> &mut FileSink { - // SAFETY: `adopt` contract — `self.0` is live; `&mut self` is exclusive + // SAFETY: constructor invariant — `self.0` is live; `&mut self` is exclusive // on this owning handle (FileSinkPtr is non-`Copy`, single-threaded // shell), so no other `&`/`&mut` to the `FileSink` overlaps. unsafe { self.0.as_mut() } @@ -168,7 +186,7 @@ impl core::ops::DerefMut for FileSinkPtr { impl Drop for FileSinkPtr { #[inline] fn drop(&mut self) { - // SAFETY: `adopt` contract — `self.0` is live with one owned intrusive + // SAFETY: constructor invariant — `self.0` is live with one owned intrusive // ref; `FileSink::deref` (CellRefCounted derive) frees on zero. unsafe { FileSink::deref(self.0.as_ptr()) }; } @@ -698,18 +716,26 @@ impl ShellSubprocess { // (they store it on StaticPipeWriter / PipeReader as a backref). let mut slot = Box::::new_uninit(); let subprocess: *mut Subprocess = slot.as_mut_ptr(); - // SAFETY: `out_subproc` points at the `SubprocExec.child` slot inside - // the heap-stable `Box` staged by the caller before this - // call; no `&` to that slot is live (the caller's `&mut Cmd` borrow - // ended before the call). Written *before* any callback below - // (`watch`/`start`/`read_all`) so re-entrant `Cmd` callbacks see a - // populated `exec.subproc.child`. - unsafe { *out_subproc = subprocess }; let stdin = match Writable::init(stdio0, event_loop, subprocess, spawn_stdin) { Ok(w) => w, - Err(WritableInitError::UnexpectedCreatingStdin) => { - panic!("unexpected error while creating stdin"); + Err(err) => { + #[cfg(not(windows))] + { + let _ = spawn_stdout.map(bun_sys::Fd::close); + let _ = spawn_stderr.map(bun_sys::Fd::close); + } + #[cfg(windows)] + { + // `WindowsSpawnResult::drop` uv_closes handed-back slots. + spawn_result.stdout = spawn_stdout; + spawn_result.stderr = spawn_stderr; + } + spawn_result.dispose_failed_spawn(event_loop); + return Err(match err { + WritableInitError::Sys(e) => ShellErr::Sys(e.to_shell_system_error()), + WritableInitError::StreamAssign(msg) => ShellErr::Custom(msg), + }); } }; let stdout = Readable::init( @@ -754,6 +780,14 @@ impl ShellSubprocess { // sound. // SAFETY: fully initialised by the `write` above. let _ = bun_core::heap::into_raw(unsafe { slot.assume_init() }); + // SAFETY: `out_subproc` points at the `SubprocExec.child` slot inside + // the heap-stable `Box` staged by the caller before this + // call; no `&` to that slot is live (the caller's `&mut Cmd` borrow + // ended before the call). Published now that the Subprocess is fully + // initialised and before any callback below (`watch`/`start`/ + // `read_all`) so re-entrant `Cmd` callbacks see a populated + // `exec.subproc.child`. + unsafe { *out_subproc = subprocess }; // SAFETY: `subprocess` is the just-allocated `ShellSubprocess`; the // owning `Cmd` outlives the `Process` exit callback. All accesses // below are scoped so no borrow of the Subprocess spans the @@ -768,37 +802,6 @@ impl ShellSubprocess { } let _ = scopeguard::ScopeGuard::into_inner(stdio_guard); - // Wire the FileSink's close-signal back to the enclosing `Writable` so - // `Writable::on_close` (drops the `Arc`) runs when the sink - // finishes. `stdin` lives inside the Box-allocated `Subprocess` at a - // stable address, so the self-referential raw pointer is sound for the - // life of the subprocess. Only reachable on Windows (POSIX - // `Writable::init` never returns `Pipe` for shell stdio). - { - // Derive `stdin_ptr` from the raw heap pointer (`subprocess`), not - // the local `subproc: &mut` reborrow — the pointer is stored - // long-term in `FileSink::source` and dereferenced from - // `Writable::on_close` after this frame returns. Under Stacked - // Borrows a child of `subproc`'s tag would be invalidated when - // that borrow ends; rooting in the allocation's provenance keeps - // it valid for the box's lifetime. - // SAFETY: `subprocess` is the live, fully-initialised heap alloc. - let stdin_ptr: *mut Writable = unsafe { &raw mut (*subprocess).stdin }; - // SAFETY: reborrow as a child of `stdin_ptr` so it does not - // invalidate the sibling we store in `source`. - if let Writable::Pipe(pipe) = unsafe { &mut *stdin_ptr } { - // SAFETY: shell is single-threaded; the FileSink allocation is - // disjoint from `*stdin_ptr`. `stdin_ptr` outlives the sink — - // the Subprocess owns both and `Writable::on_close` is the only - // path that drops the FileSinkPtr. - pipe.source - .set(webcore::streams::SourceHandle::ShellWritable( - // SAFETY: `stdin_ptr` is the live `&raw mut` writable (write provenance). - unsafe { bun_ptr::BackRef::from_raw_mut(stdin_ptr) }, - )); - } - } - // SAFETY: scoped access; `watch` does not re-enter the subprocess. match unsafe { (*subprocess).proc().watch() } { bun_sys::Result::Ok(()) => {} @@ -863,6 +866,19 @@ impl ShellSubprocess { pub(crate) fn on_process_exit(&mut self, _: &Process, status: &Status, _: &Rusage) { log!("onProcessExit({:x})", std::ptr::from_mut(self) as usize); + + // ReadableStream stdin (the only shell `Writable::Pipe`): close the + // FileSink and mark stdin closed. + let stdin_is_stream_pipe = if let Writable::Pipe(pipe) = &self.stdin { + pipe.on_attached_process_exit(status); + true + } else { + false + }; + if stdin_is_stream_pipe { + self.on_static_pipe_writer_done(); + } + let exit_code: Option = 'brk: { if let Status::Exited(exited) = &status { break 'brk Some(exited.code); @@ -887,8 +903,12 @@ impl ShellSubprocess { // through the node arena so it survives `Vec` reallocation. // `&mut self` is dead by NLL before `on_exit` re-enters interp. let cmd = unsafe { handle.cmd_mut() }; - if cmd.exit_code.is_none() { - cmd.on_exit(code.into()); + match cmd.exit_code { + None => cmd.on_exit(code.into()), + // Stdout/stderr closing set `exit_code` first; the stdin close + // above may have just satisfied `has_finished()`. + Some(prev) if cmd.has_finished() => cmd.on_exit(prev), + Some(_) => {} } } } @@ -900,8 +920,11 @@ impl ShellSubprocess { #[derive(thiserror::Error, Debug, strum::IntoStaticStr)] pub enum WritableInitError { - #[error("UnexpectedCreatingStdin")] - UnexpectedCreatingStdin, + #[error("{0}")] + Sys(bun_sys::Error), + /// `assign_to_stream` threw synchronously; carries the formatted message. + #[error("Failed to pipe ReadableStream to stdin")] + StreamAssign(Box<[u8]>), } pub enum Writable { @@ -913,20 +936,6 @@ pub enum Writable { Ignore, } -impl Writable { - // When the stream has closed we need to be notified to prevent a use-after-free - // We can test for this use-after-free by enabling hot module reloading on a file and then saving it twice - pub fn on_close(&mut self, _: Option) { - match self { - Writable::Buffer(_) | Writable::Pipe(_) => { - // Dropping the Arc on reassignment below derefs. - } - _ => {} - } - *self = Writable::Ignore; - } -} - impl Writable { pub(crate) fn init( stdio: Stdio, @@ -948,29 +957,19 @@ impl Writable { // Ownership of the `Box` transfers into the // FileSink's writer. let uv_pipe: *mut _ = bun_core::heap::into_raw(buf); - let pipe_ptr = FileSink::create_with_pipe(event_loop, uv_pipe); - - // SAFETY: `create_with_pipe` returns a freshly-boxed - // non-null FileSink with refcount 1; sole reference. - match unsafe { - (*pipe_ptr).writer.with_mut(|w| w.start_with_current_pipe()) - } { - bun_sys::Result::Ok(()) => {} - bun_sys::Result::Err(_err) => { - // SAFETY: pipe_ptr is live with refcount 1; - // deref frees it. - unsafe { FileSink::deref(pipe_ptr) }; - return Err(WritableInitError::UnexpectedCreatingStdin); - } - } + let pipe = FileSinkPtr::create_with_pipe(event_loop, uv_pipe); - // TODO: uncoment this when is ready, commented because was not compiling - // subprocess.weak_file_sink_stdin_ptr = pipe; - // subprocess.flags.has_stdin_destructor_called = false; + if let bun_sys::Result::Err(e) = + pipe.writer.with_mut(|w| w.start_with_current_pipe()) + { + // Dropping `pipe` derefs (and frees) the sink. + return Err(WritableInitError::Sys(e)); + } - // SAFETY: `create_with_pipe` returns non-null with one - // owned ref; `adopt` takes it over. - return Ok(Writable::Pipe(unsafe { FileSinkPtr::adopt(pipe_ptr) })); + if let Stdio::ReadableStream(rs) = &mut stdio { + return Self::assign_stream(pipe, rs, event_loop); + } + return Ok(Writable::Pipe(pipe)); } return Ok(Writable::Inherit); } @@ -1078,9 +1077,24 @@ impl Writable { Stdio::Inherit => Ok(Writable::Inherit), Stdio::Path(_) | Stdio::Ignore => Ok(Writable::Ignore), Stdio::Ipc | Stdio::Capture(_) => Ok(Writable::Ignore), - Stdio::ReadableStream(_) => { - // The shell never uses this - panic!("Unimplemented stdin readable_stream"); + Stdio::ReadableStream(rs) => { + let fd = result.unwrap(); + if let bun_sys::Result::Err(e) = bun_sys::set_nonblocking(fd) { + fd.close(); + return Err(WritableInitError::Sys(e)); + } + let pipe = FileSinkPtr::create(event_loop, fd); + if let bun_sys::Result::Err(e) = pipe.writer.with_mut(|w| w.start(fd, true)) { + // Dropping `pipe` derefs (and frees) the sink. + return Err(WritableInitError::Sys(e)); + } + // The fd is a socketpair half, same as `Bun.spawn` stdin. + pipe.writer.with_mut(|w| { + if let Some(poll) = w.handle.get_poll() { + poll.set_flag(bun_io::FilePollFlag::Socket); + } + }); + Self::assign_stream(pipe, rs, event_loop) } Stdio::SocketFd => { // The shell never uses this; rejected at i < 3 anyway. @@ -1090,6 +1104,59 @@ impl Writable { } } + /// Wire `stream` into the stdin `FileSink` (native `SinkHandle` wire + /// first, JS pump fallback). A synchronous throw consumes the sink. + fn assign_stream( + mut pipe: FileSinkPtr, + stream: &mut webcore::ReadableStream, + event_loop: EventLoopHandle, + ) -> Result { + // ReadableStream stdin only exists for JS-origin shells. + let global_ptr = event_loop.global_object(); + assert!( + !global_ptr.is_null(), + "ReadableStream stdin requires the JS event loop" + ); + let global = jsc::JSGlobalObject::opaque_ref(global_ptr.cast()); + let result = pipe.assign_to_stream(stream, global); + // Success shapes: undefined/null/empty (drained or natively wired) or + // a promise (pump in flight). Anything else is a synchronous throw — + // an `Error` instance or any other thrown value propagated as-is. + let thrown = if let Some(err) = result.to_error() { + Some(err) + } else if global.has_exception() { + global.clear_exception_except_termination(); + Some(jsc::JSValue::UNDEFINED) + } else if !result.is_empty_or_undefined_or_null() && result.as_any_promise().is_none() { + Some(result) + } else { + None + }; + if let Some(err) = thrown { + // `fmt::Write` (unlike `format!`/`io::Write`) propagates the + // formatter `Err` that `fmt_string` produces when the thrown + // value's `toString()` itself throws. + use core::fmt::Write as _; + let mut msg = String::new(); + if err.is_undefined() + || write!( + &mut msg, + "Failed to pipe ReadableStream to stdin: {}", + err.fmt_string(global) + ) + .is_err() + { + global.clear_exception_except_termination(); + msg.clear(); + msg.push_str("Failed to pipe ReadableStream to stdin"); + } + return Err(WritableInitError::StreamAssign( + msg.into_bytes().into_boxed_slice(), + )); + } + Ok(Writable::Pipe(pipe)) + } + // Note: there is intentionally no `Writable::toJS` here — the shell never // exposes its stdin Writable to JS. diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs index 5f5187732378..79b9d6ceb4db 100644 --- a/src/runtime/webcore/streams.rs +++ b/src/runtime/webcore/streams.rs @@ -907,7 +907,6 @@ pub enum SourceHandle { /// The `'static` bound erases the `&JSGlobalObject` borrow carried in /// `Subprocess<'a>`; the pointed-at allocation outlives this handle. Subprocess(BackRef>), - ShellWritable(BackRef), FetchResponseBody(BackRef), ServerRequestBody(crate::server::AnyRequestContext), S3DownloadBody(BackRef), @@ -947,8 +946,6 @@ impl SourceHandle { SourceHandle::FileReader(p) => p.on_close(err), SourceHandle::Subprocess(p) => p.on_close(err), // SAFETY: live backref; cleared before the pointee is freed. - SourceHandle::ShellWritable(mut p) => unsafe { p.get_mut() }.on_close(err), - // SAFETY: live backref; cleared before the pointee is freed. SourceHandle::FetchResponseBody(mut p) => unsafe { p.get_mut() }.on_stream_cancelled(), // SAFETY: live backref; cleared before the pointee is freed. SourceHandle::S3DownloadBody(mut p) => unsafe { p.get_mut() }.on_stream_cancelled(), @@ -978,9 +975,7 @@ impl SourceHandle { SourceHandle::ServerRequestBody(any) => any.on_request_body_stream_drained(), SourceHandle::HTMLRewriter(p) => p.on_ready(), // Remaining variants leave `on_ready` at the trait default (no-op). - SourceHandle::Subprocess(_) - | SourceHandle::ShellWritable(_) - | SourceHandle::S3DownloadBody(_) => {} + SourceHandle::Subprocess(_) | SourceHandle::S3DownloadBody(_) => {} } } @@ -994,7 +989,6 @@ impl SourceHandle { | SourceHandle::ByteStream(_) | SourceHandle::FileReader(_) | SourceHandle::Subprocess(_) - | SourceHandle::ShellWritable(_) | SourceHandle::S3DownloadBody(_) | SourceHandle::HTMLRewriter(_) => {} } diff --git a/src/spawn/process.rs b/src/spawn/process.rs index f4b58a10e691..127dbd8032d9 100644 --- a/src/spawn/process.rs +++ b/src/spawn/process.rs @@ -1473,6 +1473,19 @@ impl WindowsSpawnResult { pub fn to_process(&mut self, _event_loop: impl Sized) -> *mut Process { self.process.take().unwrap() } + + /// Kill and release a child whose post-spawn stdio setup failed. + /// Consumes the result so `Drop` also closes stdio pipes still held. + pub fn dispose_failed_spawn(mut self, _event_loop: impl Sized) { + let proc = self.to_process(()); + // SAFETY: `to_process` hands over the sole owned ref to the live + // heap-allocated `Process`; `deref` releases it after teardown. + unsafe { + let _ = (*proc).kill(bun_core::SignalCode::SIGKILL as u8); + (*proc).close(); + Process::deref(proc); + } + } } #[cfg(windows)] @@ -1632,6 +1645,9 @@ impl WindowsSpawnOptions { /// trait method so callers keep the `.to_process(loop_, sync)` spelling. pub trait SpawnResultExt { fn to_process(self, event_loop: EventLoopHandle) -> *mut Process; + + /// Kill, reap, and release a child whose post-spawn stdio setup failed. + fn dispose_failed_spawn(self, event_loop: EventLoopHandle); } #[cfg(unix)] @@ -1639,6 +1655,22 @@ impl SpawnResultExt for PosixSpawnResult { fn to_process(self, event_loop: EventLoopHandle) -> *mut Process { Process::init_posix(&self, event_loop) } + + fn dispose_failed_spawn(self, _event_loop: EventLoopHandle) { + // `Process::kill` no-ops while `Poller::Detached` (pre-`watch()`); + // signal the pid directly so the blocking reap terminates. + unsafe extern "C" { + #[link_name = "kill"] + safe fn libc_kill(pid: libc::pid_t, sig: c_int) -> c_int; + } + let _ = libc_kill(self.pid, bun_core::SignalCode::SIGKILL as c_int); + let _ = posix_spawn::wait4(self.pid, 0, None); + #[cfg(any(target_os = "linux", target_os = "android"))] + if let Some(pidfd) = self.pidfd { + use bun_sys::FdExt as _; + Fd::from_native(pidfd).close(); + } + } } #[cfg(unix)] diff --git a/test/js/bun/shell/bunshell.test.ts b/test/js/bun/shell/bunshell.test.ts index 95a990ba68e7..5fa7c82da696 100644 --- a/test/js/bun/shell/bunshell.test.ts +++ b/test/js/bun/shell/bunshell.test.ts @@ -3085,6 +3085,187 @@ test("stdin redirect from a Uint8Array sends the bytes captured when the command expect(result.exitCode).toBe(0); }, 60_000); +describe("redirect stdin from ReadableStream", () => { + // External command: on POSIX `cat` is external, on Windows it's a builtin, + // so use a spawned Bun child everywhere. + const childPump = `process.stdout.write(await Bun.stdin.text())`; + + test.concurrent("native source (subprocess stdout)", async () => { + // https://github.com/oven-sh/bun/issues/18262 + await using proc = Bun.spawn({ + cmd: [BUN, "-e", "process.stdout.write('hello from stream')"], + env: bunEnv, + stdout: "pipe", + }); + const out = await $`${BUN} -e ${childPump} < ${proc.stdout}`.env(bunEnv).nothrow().quiet(); + expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({ + stdout: "hello from stream", + exitCode: 0, + }); + }); + + test.concurrent("native source, large payload with backpressure", async () => { + const size = 256 * 1024; + await using proc = Bun.spawn({ + cmd: [BUN, "-e", `process.stdout.write(Buffer.alloc(${size}, 'x'))`], + env: bunEnv, + stdout: "pipe", + }); + const out = await $`${BUN} -e ${childPump} < ${proc.stdout}`.env(bunEnv).nothrow().quiet(); + expect(out.stdout.equals(Buffer.alloc(size, "x"))).toBe(true); + expect(out.exitCode).toBe(0); + }); + + test.concurrent("Bun.file().stream() collapses to the blob path", async () => { + using dir = tempDir("shell-stream-file", { "input.txt": "file-content" }); + const stream = Bun.file(join(String(dir), "input.txt")).stream(); + const out = await $`${BUN} -e ${childPump} < ${stream}`.env(bunEnv).nothrow().quiet(); + expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({ + stdout: "file-content", + exitCode: 0, + }); + }); + + test.concurrent("native ByteStream source (fetch response.body)", async () => { + await using server = Bun.serve({ + port: 0, + fetch: () => + new Response( + new ReadableStream({ + async pull(c) { + c.enqueue(Buffer.from("http-")); + await Bun.sleep(0); + c.enqueue(Buffer.from("body")); + c.close(); + }, + }), + ), + }); + const res = await fetch(server.url); + const out = await $`${BUN} -e ${childPump} < ${res.body}`.env(bunEnv).nothrow().quiet(); + expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({ + stdout: "http-body", + exitCode: 0, + }); + }); + + test.concurrent("JS-backed multi-chunk stream", async () => { + const stream = new ReadableStream({ + async pull(c) { + c.enqueue(new TextEncoder().encode("chunk1 ")); + await Bun.sleep(0); + c.enqueue(new TextEncoder().encode("chunk2")); + c.close(); + }, + }); + const out = await $`${BUN} -e ${childPump} < ${stream}`.env(bunEnv).nothrow().quiet(); + expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({ + stdout: "chunk1 chunk2", + exitCode: 0, + }); + }); + + test.concurrent("child exits while stream is still producing", async () => { + const { promise: cancelled, resolve: onCancel } = Promise.withResolvers(); + const stream = new ReadableStream({ + async pull(c) { + c.enqueue(new TextEncoder().encode("x")); + await Bun.sleep(0); + }, + cancel() { + onCancel(true); + }, + }); + // Child reads 3 bytes then exits; the shell must cancel the stream and + // finish the command with the child's exit status. + const child = ` + const b = Buffer.alloc(3); + let n = 0; + while (n < 3) { const r = require('fs').readSync(0, b, n, 3 - n); if (r <= 0) break; n += r; } + process.stdout.write(b.subarray(0, n)); + `; + const out = await $`${BUN} -e ${child} < ${stream}`.env(bunEnv).nothrow().quiet(); + expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({ + stdout: "xxx", + exitCode: 0, + }); + expect(await cancelled).toBe(true); + }); + + test("stdout/stderr redirect throws", async () => { + const s1 = new ReadableStream({ pull: c => c.close() }); + await expect(runWithErrorPromise(() => $`${BUN} -e 0 > ${s1}`)).resolves.toThrow( + /ReadableStream cannot be used for stdout or stderr/, + ); + const s2 = new ReadableStream({ pull: c => c.close() }); + await expect(runWithErrorPromise(() => $`${BUN} -e 0 2> ${s2}`)).resolves.toThrow( + /ReadableStream cannot be used for stdout or stderr/, + ); + }); + + test("locked stream throws", async () => { + const s = new ReadableStream({ pull: c => (c.enqueue(new Uint8Array([65])), c.close()) }); + const r = s.getReader(); + await expect(runWithErrorPromise(() => $`${BUN} -e 0 < ${s}`)).resolves.toThrow(/already been used/); + r.releaseLock(); + }); + + test("disturbed stream throws", async () => { + const s = new ReadableStream({ pull: c => (c.enqueue(new Uint8Array([65])), c.close()) }); + const r = s.getReader(); + await r.read(); + r.releaseLock(); + await expect(runWithErrorPromise(() => $`${BUN} -e 0 < ${s}`)).resolves.toThrow(/already been used/); + }); + + test("redirect to a builtin throws", async () => { + const s = new ReadableStream({ pull: c => (c.enqueue(new Uint8Array([65])), c.close()) }); + await expect(runWithErrorPromise(() => $`echo < ${s}`)).resolves.toThrow( + /ReadableStream cannot be redirected to a builtin command/, + ); + }); + + test.concurrent("direct stream whose pull() throws synchronously fails the command", async () => { + const s = new ReadableStream({ + type: "direct", + pull() { + throw new Error("boom-sync"); + }, + }); + const out = await $`${BUN} -e ${childPump} < ${s}`.env(bunEnv).nothrow().quiet(); + expect(out.stderr.toString()).toContain("Failed to pipe ReadableStream to stdin: Error: boom-sync"); + expect(out.exitCode).toBe(1); + }); + + test.concurrent("direct stream whose pull() throws a non-Error value fails the command", async () => { + const s = new ReadableStream({ + type: "direct", + pull() { + throw { foo: 1 }; + }, + }); + const out = await $`${BUN} -e ${childPump} < ${s}`.env(bunEnv).nothrow().quiet(); + expect(out.stderr.toString()).toContain("Failed to pipe ReadableStream to stdin"); + expect(out.exitCode).toBe(1); + }); + + test.concurrent("direct stream whose pull() throws a value with a throwing toString()", async () => { + const s = new ReadableStream({ + type: "direct", + pull() { + throw { + toString() { + throw 0; + }, + }; + }, + }); + const out = await $`${BUN} -e ${childPump} < ${s}`.env(bunEnv).nothrow().quiet(); + expect(out.stderr.toString()).toContain("Failed to pipe ReadableStream to stdin"); + expect(out.exitCode).toBe(1); + }); +}); + describe("stdin redirect from a zero-length buffer delivers EOF to the spawned command", () => { // A spawned command reading stdin (cat) must see EOF when the redirect // source is an empty ArrayBuffer/TypedArray, same as an empty Blob.