Apply backpressure when piping a ReadableStream into a subprocess

src/io/PipeWriter.rs

@@ -634,6 +634,10 @@ pub struct PosixStreamingWriter<Parent: PosixStreamingWriterParent> {
     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<bool>,
 }
 
 impl<Parent: PosixStreamingWriterParent> Default for PosixStreamingWriter<Parent> {
@@ -645,6 +649,7 @@ impl<Parent: PosixStreamingWriterParent> Default for PosixStreamingWriter<Parent
             is_done: false,
             closed_without_reporting: false,
             force_sync: false,
+            backed_up: core::cell::Cell::new(false),
         }
     }
 }
@@ -714,6 +719,9 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
     /// 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<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
         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<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
                 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<Parent: WindowsStreamingWriterParent> WindowsStreamingWriter<Parent> {
         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

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<bool> 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<bool> rsisWriteChunk(JSC::VM& vm, JSGlobalObject* globalObj
         flushPromise->performPromiseThenWithContext(vm, globalObject, runtime->onReadStreamIntoSinkFlushFulfilled(), runtime->onReadStreamIntoSinkRejected(), jsUndefined(), context);
         return false;
     }
-    if (auto* wrotePromise = dynamicDowncast<JSPromise>(wrote))
+    if (auto* wrotePromise = dynamicDowncast<JSPromise>(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;
 }

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::Error>) -> 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: <ReadableStream> })` — 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.

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),