Streams: one PipeReader loop with owned chunks, hold-not-adopt buffer pins, right-sized native pulls

src/event_loop/MiniEventLoop.rs

@@ -47,9 +47,6 @@ unsafe extern "Rust" {
 }
 // ────────────────────────────────────────────────────────────────────────────
 
-const PIPE_READ_BUFFER_SIZE: usize = 256 * 1024;
-pub type PipeReadBuffer = [u8; PIPE_READ_BUFFER_SIZE];
-
 /// Intrusive MPSC queue over `AnyTaskWithExtraContext` linked via its `.next` field.
 type ConcurrentTaskQueue = UnboundedQueue<AnyTaskWithExtraContext>;
 
@@ -82,7 +79,7 @@ pub struct MiniEventLoop {
     // Opaque ctx assigned externally; only read/cleared here.
     pub(crate) after_event_loop_callback_ctx: Option<NonNull<c_void>>,
     pub(crate) after_event_loop_callback: Option<unsafe extern "C" fn(*mut c_void)>,
-    pub pipe_read_buffer: Option<Box<PipeReadBuffer>>,
+    pub pipe_read_scratch: bun_io::PipeReadScratch,
 }
 
 thread_local! {
@@ -215,14 +212,6 @@ impl MiniEventLoop {
         self.env
     }
 
-    pub fn pipe_read_buffer(&mut self) -> &mut [u8] {
-        // `boxed_zeroed` avoids the 256 KiB stack temporary `Box::new([0u8; N])`
-        // would create in debug builds.
-        &mut self
-            .pipe_read_buffer
-            .get_or_insert_with(bun_core::boxed_zeroed::<PipeReadBuffer>)[..]
-    }
-
     pub fn on_after_event_loop(&mut self) {
         if let Some(cb) = self.after_event_loop_callback {
             let ctx = self.after_event_loop_callback_ctx;
@@ -277,7 +266,7 @@ impl MiniEventLoop {
             top_level_dir: Box::default(),
             after_event_loop_callback_ctx: None,
             after_event_loop_callback: None,
-            pipe_read_buffer: None,
+            pipe_read_scratch: bun_io::PipeReadScratch::new(),
         }
     }
 
@@ -438,7 +427,7 @@ bun_io::link_impl_EventLoopCtx! {
             (*this).after_event_loop_callback = cb;
             (*this).after_event_loop_callback_ctx = ctx;
         },
-        pipe_read_buffer() => core::ptr::from_mut::<[u8]>((*this).pipe_read_buffer()),
+        pipe_read_scratch() => &raw const (*this).pipe_read_scratch,
     }
 }
 

src/event_loop/lib.rs

@@ -35,10 +35,10 @@ pub use ConcurrentTask::{Task, TaskTag, Taskable, task_tag};
 // the type/module namespace collision on the PascalCase form.
 pub use DeferredTaskQueue as deferred_task_queue;
 
-pub use MiniEventLoop::PipeReadBuffer;
 pub use any_event_loop::{
     AnyEventLoop, EventLoopHandle, EventLoopTask, JsPoster, JsPosterVTable, Posted,
 };
+pub use bun_io::PipeReadScratch;
 
 // JS-event-loop arm of `AnyEventLoop` / `EventLoopHandle`. `bun_event_loop` is
 // a lower tier than `bun_jsc`, so it cannot name `jsc::EventLoop` /
@@ -50,7 +50,6 @@ bun_dispatch::link_interface! {
         fn file_polls() -> *mut bun_io::file_poll::Store;
         fn put_file_poll(poll: *mut bun_io::FilePoll, was_ever_registered: bool);
         fn uws_loop() -> *mut bun_uws::Loop;
-        fn pipe_read_buffer() -> *mut [u8];
         fn tick();
         fn auto_tick();
         fn auto_tick_active();

src/io/PipeReader.rs

@@ -148,30 +148,6 @@ impl Drop for ParentKeepAlive {
     }
 }
 
-// The per-loop `pipe_read_buffer` scratch is handed to `on_read_chunk` as the
-// chunk itself, and a consumer may keep parsing it while running user code
-// that starts a *second* reader synchronously (HTMLRewriter handlers do). A
-// nested read loop must not refill the scratch under the outer one, so only
-// the outermost loop on the thread borrows it; nested ones read into their
-// own `_buffer`.
-thread_local! {
-    static READ_SCRATCH_IN_USE: core::cell::Cell<bool> = const { core::cell::Cell::new(false) };
-}
-
-struct ReadScratchClaim;
-
-impl ReadScratchClaim {
-    fn try_claim() -> Option<Self> {
-        READ_SCRATCH_IN_USE.with(|in_use| (!in_use.replace(true)).then_some(Self))
-    }
-}
-
-impl Drop for ReadScratchClaim {
-    fn drop(&mut self) {
-        READ_SCRATCH_IN_USE.with(|in_use| in_use.set(false));
-    }
-}
-
 // ──────────────────────────────────────────────────────────────────────────
 // PosixBufferedReader
 // ──────────────────────────────────────────────────────────────────────────
@@ -821,20 +797,17 @@ impl PosixBufferedReader {
         // SAFETY: caller contract — `this` is live.
         let vtable = unsafe { (*this).vtable };
         let mut received_hup = received_hup_initially;
-        let scratch = ReadScratchClaim::try_claim();
+        let mut scratch = vtable.event_loop().claim_pipe_read_scratch();
         loop {
             let streaming = vtable.is_streaming_enabled();
             let mut got_retry = false;
 
             // SAFETY: caller contract; borrow ends at `;`.
-            let unbuffered = scratch.is_some() && unsafe { (*this)._buffer.is_empty() };
-            if unbuffered {
-                // Use stack buffer for streaming — per-loop scratch buffer;
-                // single-threaded event loop (see `EventLoopCtx::pipe_read_buffer_mut`).
+            let buffer_empty = unsafe { (*this)._buffer.is_empty() };
+            if let (true, Some(scratch)) = (buffer_empty, scratch.as_mut()) {
                 // SAFETY: caller contract; `maxbuf` is Copy.
                 let maxbuf = unsafe { (*this).maxbuf };
-                let stack_buffer = vtable.event_loop().pipe_read_buffer_mut();
-                let stack_buffer = MaxBuf::clamp_read_buf(maxbuf, stack_buffer);
+                let stack_buffer = MaxBuf::clamp_read_buf(maxbuf, &mut scratch[..]);
 
                 match sys::read_nonblocking(fd, stack_buffer) {
                     sys::Result::Ok(bytes_read) => {
@@ -1062,13 +1035,10 @@ impl PosixBufferedReader {
         // SAFETY: caller contract — `this` is live.
         let vtable = unsafe { (*this).vtable };
         let streaming = vtable.is_streaming_enabled();
-        let scratch = ReadScratchClaim::try_claim();
+        let mut scratch = vtable.event_loop().claim_pipe_read_scratch();
 
-        if streaming && scratch.is_some() {
-            // Per-loop scratch buffer; single-threaded event loop (see
-            // `EventLoopCtx::pipe_read_buffer_mut`).
-            let event_loop = vtable.event_loop();
-            let stack_buffer_len = event_loop.pipe_read_buffer_mut().len();
+        if let (true, Some(scratch)) = (streaming, scratch.as_mut()) {
+            let stack_buffer_len = scratch.len();
             // SAFETY: caller contract; borrow ends at the loop test.
             while unsafe { (*this)._buffer.is_empty() } {
                 let stack_buffer_cutoff = stack_buffer_len / 2;
@@ -1078,7 +1048,7 @@ impl PosixBufferedReader {
                     // before the syscall's buffer borrow (event-loop scratch,
                     // not `*this`).
                     let (maxbuf, offset) = unsafe { ((*this).maxbuf, (*this)._offset) };
-                    let buf = &mut event_loop.pipe_read_buffer_mut()[head_start..];
+                    let buf = &mut scratch[head_start..];
                     let buf = MaxBuf::clamp_read_buf(maxbuf, buf);
 
                     match sys_fn(fd, buf, offset) {
@@ -1097,10 +1067,8 @@ impl PosixBufferedReader {
                                 // SAFETY: caller contract; borrow ends at `;`.
                                 unsafe { (*this).close_without_reporting() };
                                 if head_start > 0 {
-                                    let _ = vtable.on_read_chunk(
-                                        &event_loop.pipe_read_buffer_mut()[..head_start],
-                                        ReadState::Eof,
-                                    );
+                                    let _ = vtable
+                                        .on_read_chunk(&scratch[..head_start], ReadState::Eof);
                                 }
                                 // SAFETY: caller contract; `done()` is the tail.
                                 unsafe {
@@ -1126,7 +1094,7 @@ impl PosixBufferedReader {
                                 // returns the remaining bytes then 0, so
                                 // draining to `bytes_read == 0` is bounded.
                                 let keep_going = vtable.on_read_chunk(
-                                    &event_loop.pipe_read_buffer_mut()[..head_start],
+                                    &scratch[..head_start],
                                     if received_hup {
                                         ReadState::Eof
                                     } else {
@@ -1164,19 +1132,15 @@ impl PosixBufferedReader {
                                 }
 
                                 if head_start > 0 {
-                                    let _ = vtable.on_read_chunk(
-                                        &event_loop.pipe_read_buffer_mut()[..head_start],
-                                        ReadState::Drained,
-                                    );
+                                    let _ = vtable
+                                        .on_read_chunk(&scratch[..head_start], ReadState::Drained);
                                 }
                                 return;
                             }
 
                             if head_start > 0 {
-                                let _ = vtable.on_read_chunk(
-                                    &event_loop.pipe_read_buffer_mut()[..head_start],
-                                    ReadState::Progress,
-                                );
+                                let _ = vtable
+                                    .on_read_chunk(&scratch[..head_start], ReadState::Progress);
                             }
                             // SAFETY: caller contract; `on_error` is the tail.
                             unsafe { Self::on_error(this, err) };
@@ -1187,7 +1151,7 @@ impl PosixBufferedReader {
 
                 if head_start > 0 {
                     let keep_going = vtable.on_read_chunk(
-                        &event_loop.pipe_read_buffer_mut()[..head_start],
+                        &scratch[..head_start],
                         if received_hup {
                             ReadState::Eof
                         } else {
@@ -1212,14 +1176,11 @@ impl PosixBufferedReader {
             let take_stack_path = !streaming
                 && scratch.is_some()
                 && unsafe { (*this)._buffer.capacity() == 0 && (*this)._offset == 0 };
-            if take_stack_path {
+            if let (true, Some(scratch)) = (take_stack_path, scratch.as_mut()) {
                 // Avoid a 16 KB dynamic memory allocation when the buffer might very well be empty.
-                // Per-loop scratch buffer; single-threaded event loop (see
-                // `EventLoopCtx::pipe_read_buffer_mut`).
                 // SAFETY: caller contract; `maxbuf` is Copy.
                 let maxbuf = unsafe { (*this).maxbuf };
-                let stack_buffer = vtable.event_loop().pipe_read_buffer_mut();
-                let stack_buffer = MaxBuf::clamp_read_buf(maxbuf, stack_buffer);
+                let stack_buffer = MaxBuf::clamp_read_buf(maxbuf, &mut scratch[..]);
 
                 // Unlike the block of code following this one, only handle the non-streaming case.
                 debug_assert!(!streaming);

src/io/lib.rs

@@ -326,7 +326,7 @@ bun_dispatch::link_interface! {
             cb: Option<OpaqueCallback>,
             ctx: Option<core::ptr::NonNull<core::ffi::c_void>>,
         );
-        fn pipe_read_buffer() -> *mut [u8];
+        fn pipe_read_scratch() -> *const PipeReadScratch;
     }
 }
 
@@ -381,22 +381,11 @@ impl EventLoopCtx {
         // discipline above — see block comment.
         unsafe { &mut *self.file_polls_ptr() }
     }
-    /// Single nonnull-asref accessor for the per-loop pipe-read scratch
-    /// buffer. Same contract as [`loop_mut`]: `pub(crate)`, the buffer is a
-    /// per-thread set-once allocation owned by the VM/Mini loop, and the
-    /// event loop is single-threaded, so no second `&mut [u8]` to it can be
-    /// live. Every in-crate caller (`PipeReader::read_*`) uses it for one
-    /// blocking syscall and drops the borrow before re-entering the loop.
-    /// `'static` matches the unbounded lifetime the inline raw-ptr derefs at
-    /// the call sites already produced; collapses their N identical
-    /// `&mut *ctx.pipe_read_buffer()` derefs into this one block.
+    /// Claims the per-loop pipe-read scratch; `None` while a read further up the stack holds it.
     #[inline]
-    fn pipe_read_buffer_mut(&self) -> &'static mut [u8] {
-        // SAFETY: per-thread set-once scratch buffer (`BackRef`-shaped); the
-        // event loop is single-threaded so this is the sole live `&mut`, and
-        // every crate-internal caller drops the borrow before any path that
-        // could re-derive it — see doc comment above.
-        unsafe { &mut *self.pipe_read_buffer() }
+    fn claim_pipe_read_scratch(&self) -> Option<PipeReadScratchGuard> {
+        // SAFETY: per-thread scratch owned by the VM/Mini loop, which outlives every read.
+        unsafe { (*self.pipe_read_scratch()).claim() }
     }
     #[inline]
     pub(crate) fn loop_ref(&self) {
@@ -471,12 +460,14 @@ pub mod heap;
 pub mod max_buf;
 #[path = "openForWriting.rs"]
 pub mod open_for_writing_mod;
+pub mod pipe_read_scratch;
 #[path = "PipeReader.rs"]
 pub mod pipe_reader;
 #[path = "PipeWriter.rs"]
 pub mod pipe_writer;
 #[path = "pipes.rs"]
 pub mod pipes;
+pub use pipe_read_scratch::{PipeReadScratch, PipeReadScratchGuard};
 #[cfg(windows)]
 #[path = "source.rs"]
 pub mod source;

src/io/pipe_read_scratch.rs

@@ -0,0 +1,76 @@
+use core::cell::{Cell, UnsafeCell};
+use core::ops::{Deref, DerefMut};
+use core::ptr::NonNull;
+
+pub const PIPE_READ_BUFFER_SIZE: usize = 256 * 1024;
+type PipeReadBuffer = [u8; PIPE_READ_BUFFER_SIZE];
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum State {
+    Available,
+    Used,
+}
+
+/// Per-loop scratch for blocking pipe/file reads. Chunks are delivered straight out of it, and a consumer may run user code that starts a nested read while still parsing the chunk, so only one borrower on the thread may hold it at a time. Interior-mutable so a refused nested `claim` never forms a `&mut` over the outer guard's pointers.
+pub struct PipeReadScratch {
+    state: Cell<State>,
+    buffer: UnsafeCell<Option<Box<PipeReadBuffer>>>,
+}
+
+impl PipeReadScratch {
+    pub const fn new() -> Self {
+        Self {
+            state: Cell::new(State::Available),
+            buffer: UnsafeCell::new(None),
+        }
+    }
+
+    /// `None` while a borrower further up the stack still holds the guard.
+    pub fn claim(&self) -> Option<PipeReadScratchGuard> {
+        if self.state.replace(State::Used) == State::Used {
+            return None;
+        }
+        // SAFETY: `state` was just moved to `Used`, so no guard (the only other accessor of `buffer`) exists; the borrow ends at `;`.
+        let buffer = unsafe { &mut *self.buffer.get() }.get_or_insert_with(bun_core::boxed_zeroed);
+        Some(PipeReadScratchGuard {
+            buffer: NonNull::from(&mut buffer[..]),
+            state: NonNull::from(&self.state),
+        })
+    }
+}
+
+impl Default for PipeReadScratch {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Exclusive claim on the scratch; released on drop. The owner (VM rare data / mini loop) outlives every guard.
+pub struct PipeReadScratchGuard {
+    buffer: NonNull<[u8]>,
+    state: NonNull<Cell<State>>,
+}
+
+impl Deref for PipeReadScratchGuard {
+    type Target = [u8];
+    #[inline]
+    fn deref(&self) -> &[u8] {
+        // SAFETY: the buffer is a separate heap allocation only reachable through this guard while `state` is `Used`.
+        unsafe { self.buffer.as_ref() }
+    }
+}
+
+impl DerefMut for PipeReadScratchGuard {
+    #[inline]
+    fn deref_mut(&mut self) -> &mut [u8] {
+        // SAFETY: as in `deref`.
+        unsafe { self.buffer.as_mut() }
+    }
+}
+
+impl Drop for PipeReadScratchGuard {
+    fn drop(&mut self) {
+        // SAFETY: the owner outlives the guard; `Cell` needs only a shared ref.
+        unsafe { self.state.as_ref() }.set(State::Available);
+    }
+}

src/jsc/VirtualMachine.rs

@@ -2250,9 +2250,7 @@
             vm.after_event_loop_callback = cb;
             vm.after_event_loop_callback_ctx = ctx.map(|p| p.as_ptr());
         },
-        pipe_read_buffer() => {
-            core::ptr::from_mut::<[u8]>(vm_from_owner(this.cast()).rare_data().pipe_read_buffer())
-        },
+        pipe_read_scratch() => core::ptr::from_ref(vm_from_owner(this.cast()).rare_data().pipe_read_scratch()),
     }
 }
 

src/jsc/event_loop.rs

@@ -332,14 +332,6 @@ impl EventLoop {
         result
     }
 
-    /// SAFETY: returns `&mut` into VM-owned scratch; two calls alias the same
-    /// buffer. Caller must not hold another live `&mut` to it.
-    pub unsafe fn pipe_read_buffer(&mut self) -> &mut [u8] {
-        // SAFETY: vm() is the live owning VM; rare_data() lazily inits the
-        // per-VM scratch buffer. Caller contract (see doc): no concurrent &mut.
-        unsafe { &mut (*self.vm()).rare_data().pipe_read_buffer()[..] }
-    }
-
     pub fn drain_microtasks_with_global(
         &mut self,
         global_object: &JSGlobalObject,
@@ -1377,7 +1369,6 @@ bun_event_loop::link_impl_JsEventLoop! {
             (*store).put(core::ptr::NonNull::new_unchecked(poll), ctx, was_ever_registered);
         },
         uws_loop() => (*this).usockets_loop(),
-        pipe_read_buffer() => core::ptr::from_mut::<[u8]>((*this).pipe_read_buffer()),
         tick() => (*this).tick(),
         auto_tick() => (*this).auto_tick(),
         auto_tick_active() => (*this).auto_tick_active(),