diff --git a/src/io/PipeWriter.rs b/src/io/PipeWriter.rs index b21b6b70a339..a3dba390d2b8 100644 --- a/src/io/PipeWriter.rs +++ b/src/io/PipeWriter.rs @@ -634,6 +634,10 @@ pub struct PosixStreamingWriter { pub is_done: bool, pub closed_without_reporting: bool, pub force_sync: bool, + /// Mirrors the last `WriteStatus` reported to the parent, which is `Pending` + /// exactly when `write(2)` returned `EAGAIN`. `Cell` because the single site + /// that maintains it, `parent_on_write`, only holds `&self`. + backed_up: core::cell::Cell, } impl Default for PosixStreamingWriter { @@ -645,6 +649,7 @@ impl Default for PosixStreamingWriter PosixStreamingWriter { /// through this accessor. #[inline] fn parent_on_write(&self, amount: usize, status: WriteStatus) { + // Record before dispatching: `on_write` may re-enter `write()`, whose own + // `parent_on_write` then leaves the newer value in place. + self.backed_up.set(status == WriteStatus::Pending); // SAFETY: type invariant — set-once parent backref outlives writer. unsafe { Parent::on_write(self.parent(), amount, status) } } @@ -750,6 +758,14 @@ impl PosixStreamingWriter { self.outgoing.size() } + /// The destination refused bytes we offered it: `write(2)` returned `EAGAIN`, + /// so the remainder is sitting in `outgoing`. Not the same as + /// `has_pending_data()`, which is also true while writes below `CHUNK_SIZE` + /// coalesce in a buffer the kernel has not been shown yet. + pub fn is_backed_up(&self) -> bool { + self.backed_up.get() + } + pub fn should_buffer(&self, addition: usize) -> bool { !self.force_sync && self.outgoing.size() + addition < Self::CHUNK_SIZE } @@ -1030,6 +1046,9 @@ impl PosixStreamingWriter { self.outgoing.reset(); } } + // `drain_buffered_data` does not report to the parent, so maintain the + // flag here: anything left over is the kernel refusing the rest. + self.backed_up.set(self.outgoing.is_not_empty()); rc } @@ -2106,6 +2125,14 @@ impl WindowsStreamingWriter { self.outgoing.size() + self.current_payload.size() } + /// libuv refused bytes we offered it: `process_send` found a `uv_write` + /// already in flight and left them in `outgoing`. (`uv_write` is always + /// async, so a pending write on its own says nothing — `current_payload` is + /// the buffer the OS currently has, which is not backpressure.) + pub fn is_backed_up(&self) -> bool { + self.outgoing.is_not_empty() + } + fn on_write_complete(&mut self, status: uv::ReturnCode) { // PORT_NOTES_PLAN R-2: `&mut self` carries LLVM `noalias`, but // `Parent::on_write` (e.g. `FileSink::on_write`) re-enters JS via diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp index 86318aa2e8d2..9771f56e039b 100644 --- a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp +++ b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp @@ -307,15 +307,14 @@ static void startJSSinkController(JSC::VM& vm, JSGlobalObject* globalObject, JSO throwTypeError(globalObject, scope, "Unknown direct controller. This is a bug in Bun."_s); } -// ReadableStream.prototype.cancel semantics; the result promise is only ever markAsHandled'd. -static void publicStreamCancelIgnoringResult(JSC::VM& vm, JSGlobalObject* globalObject, JSReadableStream* stream, JSValue reason) +// The ReadableStreamCancel abstract op, as `pipeTo` invokes it when the destination +// errors. Not ReadableStream.prototype.cancel: a pump always holds the stream's reader, +// so the public method's lock check would reject every call and the underlying source's +// cancel() would never run. The result promise is only ever markAsHandled'd. +static void cancelStreamIgnoringResult(JSC::VM& vm, JSGlobalObject* globalObject, JSReadableStream* stream, JSValue reason) { auto catchScope = DECLARE_TOP_EXCEPTION_SCOPE(vm); - JSPromise* promise = nullptr; - if (isReadableStreamLocked(stream)) - promise = promiseRejectedWith(globalObject, createTypeError(globalObject, "ReadableStream is locked"_s)); - else - promise = readableStreamCancel(globalObject, stream, reason); + JSPromise* promise = readableStreamCancel(globalObject, stream, reason); if (catchScope.exception()) [[unlikely]] { takeAbruptCompletion(globalObject, catchScope); return; @@ -999,7 +998,7 @@ static void rsisAbrupt(JSC::VM& vm, JSGlobalObject* globalObject, JSReadStreamIn op->m_reader.clear(); auto* result = op->m_result.get(); if (auto* stream = op->m_stream.get()) - publicStreamCancelIgnoringResult(vm, globalObject, stream, error); + cancelStreamIgnoringResult(vm, globalObject, stream, error); JSValue rejectionValue = error; if (op->m_sink && !op->m_didClose) { op->m_didClose = true; @@ -1023,9 +1022,10 @@ static void rsisAbrupt(JSC::VM& vm, JSGlobalObject* globalObject, JSReadStreamIn RELEASE_AND_RETURN(scope, rejectPromise(globalObject, result, rejectionValue)); } -// One sink.write(chunk). `wrote < 0` = HTTP-sink backpressure: register the flush continuation -// (its context carries the unwritten batch tail) and suspend. A Promise `wrote` is -// deliberately NOT awaited, only marked as handled. +// One sink.write(chunk). `wrote < 0` = the sink is backed up: register the flush continuation +// (its context carries the unwritten batch tail) and suspend. A pending Promise `wrote` is +// deliberately NOT awaited, only marked as handled; an already-rejected one is the write +// failing outright, which aborts the pump. static std::optional rsisWriteChunk(JSC::VM& vm, JSGlobalObject* globalObject, JSReadStreamIntoSinkOperation* op, JSValue chunk, JSObject* batchValues, unsigned nextIndex, unsigned length) { auto scope = DECLARE_THROW_SCOPE(vm); @@ -1056,8 +1056,15 @@ static std::optional rsisWriteChunk(JSC::VM& vm, JSGlobalObject* globalObj flushPromise->performPromiseThenWithContext(vm, globalObject, runtime->onReadStreamIntoSinkFlushFulfilled(), runtime->onReadStreamIntoSinkRejected(), jsUndefined(), context); return false; } - if (auto* wrotePromise = dynamicDowncast(wrote)) + if (auto* wrotePromise = dynamicDowncast(wrote)) { markPromiseAsHandled(vm, wrotePromise); + // The destination went away mid-write (EPIPE once a subprocess closes its stdin). + // Without this the pump keeps reading the source forever into a dead sink. + if (wrotePromise->status() == JSPromise::Status::Rejected) { + throwException(globalObject, scope, wrotePromise->result()); + return std::nullopt; + } + } return true; } @@ -1331,7 +1338,7 @@ static void resumableHandleAbrupt(JSC::VM& vm, JSGlobalObject* globalObject, JSR op->m_error.set(vm, op, error); op->m_closed = true; if (auto* stream = op->m_stream.get()) - publicStreamCancelIgnoringResult(vm, globalObject, stream, error); + cancelStreamIgnoringResult(vm, globalObject, stream, error); queueStreamsMicrotask(globalObject, WebCore::JSStreamsRuntime::from(globalObject)->onResumableSinkEndMicrotask(), error, op); op->m_reading = false; } diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index 4a17e701fc50..3e5e929ede0f 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/src/runtime/webcore/FileSink.rs @@ -438,10 +438,15 @@ impl FileSink { FileSink::run_pending(this); + // `run_pending` resolves the flush promise and drains microtasks, so a + // suspended stream pump can buffer more bytes before this returns. POSIX + // `end()` closes the fd at once, so re-read instead of using the snapshot. + let has_pending_data = (*this).writer.get().has_pending_data(); + // this.done == true means ended was called let ended_and_done = (*this).done.get() && status == WriteStatus::EndOfFile; - if (*this).done.get() && status == WriteStatus::Drained { + if (*this).done.get() && status == WriteStatus::Drained && !has_pending_data { // if we call end/endFromJS and we have some pending returned from .flush() we should call writer.end() (*this).writer.with_mut(|w| w.end()); } else if ended_and_done && !has_pending_data { @@ -969,7 +974,7 @@ impl FileSink { // SAFETY(JsCell): `IOWriter::write` buffers/writes to fd; does not call JS. let rc = self.writer.with_mut(|w| w.write(data.slice())); let accepted = self.bytes_accepted(buffered_before, &rc); - self.to_result(rc, accepted) + self.write_result(rc, accepted) } #[inline] @@ -985,7 +990,7 @@ impl FileSink { // SAFETY(JsCell): `IOWriter::write_latin1` buffers/writes; no JS. let rc = self.writer.with_mut(|w| w.write_latin1(data.slice())); let accepted = self.bytes_accepted(buffered_before, &rc); - self.to_result(rc, accepted) + self.write_result(rc, accepted) } pub fn write_utf16(&self, data: &streams::Result) -> streams::Writable { @@ -996,7 +1001,7 @@ impl FileSink { // SAFETY(JsCell): `IOWriter::write_utf16` buffers/writes; no JS. let rc = self.writer.with_mut(|w| w.write_utf16(data.slice16())); let accepted = self.bytes_accepted(buffered_before, &rc); - self.to_result(rc, accepted) + self.write_result(rc, accepted) } pub fn end(&self, _err: Option) -> sys::Result<()> { @@ -1282,6 +1287,36 @@ impl FileSink { } } + /// `to_result`, plus the backpressure signal the `readStreamIntoSink` pump + /// needs. The destination itself decides: once it refuses bytes (`EAGAIN` + /// from `write(2)`, a `uv_write` already in flight on Windows) report + /// `Backpressure` so the pump awaits `flush(true)` instead of reading the + /// rest of the stream into memory. The writer then holds at most the one + /// chunk the kernel would not take. + /// + /// Only that pump understands the negative sentinel, so this must never reach + /// a write user JS issued. `readable_stream` is set by exactly one caller, + /// `assign_to_stream`, i.e. `Bun.spawn({ stdin: })` — and in + /// that case `proc.stdin` is the stream, not this sink (spawn caches it), so + /// the sink has no JS handle. Every sink that does have one (`proc.stdin` for + /// `stdin: "pipe"`, `Bun.file().writer()`) has no stream attached and keeps + /// `write()`'s number-or-Promise contract. + fn write_result(&self, write_result: WriteResult, accepted: u64) -> streams::Writable { + let result = self.to_result(write_result, accepted); + if !self.readable_stream.with_mut(|s| s.has()) || !self.writer.get().is_backed_up() { + return result; + } + match result { + // Finished or failed: `Backpressure` would park the pump on a drain + // that is never coming. + streams::Writable::Done + | streams::Writable::Err(_) + | streams::Writable::OwnedAndDone(_) + | streams::Writable::TemporaryAndDone(_) => result, + _ => streams::Writable::Backpressure(accepted), + } + } + // Helper for struct-init defaults. `EventLoopHandle` has // no `Default`, so `impl Default for FileSink` is not possible; kept private // to avoid exposing a half-initialized state. diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs index 0d86d23081bf..f741b158c687 100644 --- a/src/runtime/webcore/streams.rs +++ b/src/runtime/webcore/streams.rs @@ -388,11 +388,12 @@ pub enum Writable { Done, Owned(BlobSizeType), /// The bytes were accepted, but the transport is now backed up. `to_js()` - /// reports `-(len + 1)` so the JS write loop can detect backpressure - /// without conflating it with `Pending` (FileSink on Windows returns a - /// Promise on every write — `Promise < 0` is false, so `readStreamIntoSink` - /// keeps its main-branch behavior for non-HTTP sinks). The drain itself is - /// awaited via `flush(true)` → `pending_flush`. + /// reports `-(len + 1)` so the JS write loop can detect backpressure without + /// conflating it with `Pending` (a Promise, which FileSink returns for every + /// write on Windows, and `Promise < 0` is false). The drain itself is awaited + /// via `flush(true)` → `pending_flush`. Only `readStreamIntoSink` understands + /// the sentinel, so a sink reachable from user code must not report it for + /// writes that code issues directly. Backpressure(BlobSizeType), OwnedAndDone(BlobSizeType), TemporaryAndDone(BlobSizeType), diff --git a/test/js/bun/spawn/spawn-stdin-readable-stream.test.ts b/test/js/bun/spawn/spawn-stdin-readable-stream.test.ts index c79980878d9d..003cbef0027e 100644 --- a/test/js/bun/spawn/spawn-stdin-readable-stream.test.ts +++ b/test/js/bun/spawn/spawn-stdin-readable-stream.test.ts @@ -205,9 +205,9 @@ describe("spawn stdin ReadableStream", () => { expect(await proc.exited).toBe(0); }); - test.todo("ReadableStream cancellation when process exits early", async () => { - let cancelled = false; + test("ReadableStream cancellation when process exits early", async () => { let chunksEnqueued = 0; + const { promise: cancelled, resolve: onCancel } = Promise.withResolvers(); const stream = new ReadableStream({ async pull(controller) { @@ -217,7 +217,7 @@ describe("spawn stdin ReadableStream", () => { controller.enqueue(`chunk ${chunksEnqueued}\n`); }, cancel(_reason) { - cancelled = true; + onCancel(); }, }); @@ -243,15 +243,14 @@ describe("spawn stdin ReadableStream", () => { env: bunEnv, }); - const text = await proc.stdout.text(); - await proc.exited; + const [text] = await Promise.all([proc.stdout.text(), proc.exited]); - // Give some time for cancellation to happen - await Bun.sleep(100); + // The child is gone, so the sink is dead: the pump must cancel the source + // instead of pulling it forever. Hangs (and fails) if cancel() never runs. + await cancelled; - expect(cancelled).toBe(true); expect(chunksEnqueued).toBeGreaterThanOrEqual(2); - // head -n 2 should only output 2 lines + // The child exits after 2 lines. expect(text.trim().split("\n").length).toBe(2); }); @@ -355,16 +354,18 @@ describe("spawn stdin ReadableStream", () => { }); } - // The child never reads its stdin, so a 256 KiB write can never finish - // and the sink always holds an in-flight write. Once a few chunks have - // been handed to the sink, kill the child. How far the pump gets before - // the parent notices the death varies, so run several rounds. + // The child never reads its stdin, so the first 256 KiB write fills the + // pipe, gets EAGAIN on the rest, and leaves the sink holding an in-flight + // write. The pump parks right there, so no further chunk is ever produced: + // kill on the first one, from a macrotask, which runs after the microtask + // that issued the write. How far the pump gets before the parent notices + // the death varies, so run several rounds. function round() { let produced = 0; const child = Bun.spawn({ cmd: [process.execPath, "-e", "setTimeout(() => {}, 1e9)"], stdin: (${useIterator} ? iterate : readable)(() => { - if (++produced === 4) child.kill(); + if (++produced === 1) setTimeout(() => child.kill(), 0); }), stdout: "ignore", stderr: "ignore", @@ -411,7 +412,10 @@ describe("spawn stdin ReadableStream", () => { const chunk = Buffer.alloc(256 * 1024, "x"); let produced = 0; function producedOne() { - if (++produced === 4) child.kill(); + // The pump parks on the first refused write, so no second chunk is ever + // produced. Kill from a macrotask, which runs after the microtask that + // issued the write, so the sink is holding it when the child dies. + if (++produced === 1) setTimeout(() => child.kill(), 0); } async function* iterate() { while (true) { @@ -465,6 +469,165 @@ describe("spawn stdin ReadableStream", () => { await expectParentExitsAfterChildDies(false); }); + // The sink reports backpressure to the pump with a negative `write()` return, + // a sentinel only the pump understands. That is safe because a sink fed by a + // ReadableStream has no JS handle: spawn caches `proc.stdin` as the stream + // itself. If that ever changes, the sentinel reaches user code instead. + test("proc.stdin is the stream, not a writable sink, when stdin is a ReadableStream", async () => { + const stream = new ReadableStream({ + pull(controller) { + controller.enqueue("x"); + controller.close(); + }, + }); + + await using proc = spawn({ + cmd: [bunExe(), "-e", "process.stdin.resume()"], + stdin: stream, + stdout: "ignore", + env: bunEnv, + }); + + expect(proc.stdin).toBe(stream); + expect((proc.stdin as any).write).toBeUndefined(); + expect(await proc.exited).toBe(0); + }); + + // A synchronous pull() re-fills the stream's queue inside the pump's own + // microtask loop, so nothing but the sink refusing more data can stop it. + // Both fixtures bail out of pull() past a generous bound so they terminate + // instead of buffering the whole stream into memory. + const syncPullSource = (bound: number) => /* js */ ` + const chunk = Buffer.alloc(64 * 1024, "x"); + let pulls = 0; + const source = { + pull(controller) { + if (++pulls > ${bound}) { + console.log("pull() was never bounded by backpressure"); + process.exit(1); + } + controller.enqueue(chunk); + }, + }; + `; + + // Chunks larger than the pipe make the kernel refuse every write, so the pump + // parks on `sink.flush(true)` and resumes once per chunk. Resolving that promise + // re-enters JS from inside the sink's write-completion handler, which is where a + // stale "buffer is drained" snapshot once closed the pipe on top of bytes that + // had just been buffered. None of that may cost the child a byte. + test("a child behind backpressure still receives every byte", async () => { + const chunkSize = 256 * 1024; + const numChunks = 5; + const chunk = Buffer.alloc(chunkSize, "x"); + + // Where the final resume lands varies; a few rounds is enough to pin it. + for (let round = 0; round < 3; round++) { + let pushed = 0; + const stream = new ReadableStream({ + pull(controller) { + if (pushed < numChunks) { + controller.enqueue(chunk); + pushed++; + } else { + controller.close(); + } + }, + }); + + await using proc = spawn({ + cmd: [ + bunExe(), + "-e", + `let n = 0; + process.stdin.on("data", d => (n += d.length)); + process.on("beforeExit", () => console.log(n));`, + ], + stdin: stream, + stdout: "pipe", + env: bunEnv, + }); + + const received = parseInt(await proc.stdout.text()); + expect({ round, received, exitCode: await proc.exited }).toEqual({ + round, + received: chunkSize * numChunks, + exitCode: 0, + }); + } + }); + + test("a synchronous pull() does not starve the event loop", async () => { + await using proc = spawn({ + cmd: [ + bunExe(), + "-e", + /* js */ ` + ${syncPullSource(512)} + + // Never reads its stdin, so the pipe fills and the sink must push back. + const child = Bun.spawn({ + cmd: [process.execPath, "-e", "setTimeout(() => {}, 1e9)"], + stdin: new ReadableStream(source), + stdout: "ignore", + stderr: "ignore", + }); + + let ticks = 0; + const timer = setInterval(() => { + if (++ticks < 3) return; + clearInterval(timer); + child.kill(); + }, 1); + + await child.exited; + console.log("ticks=" + ticks); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout: stdout.trim(), exitCode }).toEqual({ stdout: "ticks=3", exitCode: 0 }); + expect(stderr).not.toContain("EPIPE"); + }); + + test("a synchronous pull() stops and cancels the source when the child exits early", async () => { + await using proc = spawn({ + cmd: [ + bunExe(), + "-e", + /* js */ ` + ${syncPullSource(512)} + const { promise: cancelled, resolve: onCancel } = Promise.withResolvers(); + source.cancel = () => onCancel(); + + const child = Bun.spawn({ + cmd: [process.execPath, "-e", "process.stdin.once('data', () => process.exit(0))"], + stdin: new ReadableStream(source), + stdout: "ignore", + stderr: "ignore", + }); + + console.log("exited=" + (await child.exited)); + // The sink is dead, so the pump must cancel the source rather than pull + // it forever. Hangs (and fails) if cancel() never runs. + await cancelled; + console.log("cancelled"); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout: stdout.trim(), exitCode }).toEqual({ stdout: "exited=0\ncancelled", exitCode: 0 }); + expect(stderr).not.toContain("EPIPE"); + }); + test("ReadableStream with process that exits immediately", async () => { const stream = new ReadableStream({ start(controller) {