diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index e8ff33bd1778..cede360a7e60 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -449,6 +449,18 @@ impl FileReader { unsafe { (*self.parent()).increment_count() }; } } + #[cfg(windows)] + { + // Non-lazy fromPipe path (Bun.spawn stdout/stderr): hold a + // ref across the pending uv_read_start so the source is not + // finalized while IOCP has a read queued on it. + if !self.started.get() && self.reader().source.is_some() && !self.reader().is_done() + { + self.waiting_for_on_reader_done.set(true); + // SAFETY: see `parent()`. + unsafe { (*self.parent()).increment_count() }; + } + } } #[cfg(unix)] diff --git a/src/runtime/webcore/ReadableStream.rs b/src/runtime/webcore/ReadableStream.rs index ff7ecc724b26..17d9d76ab40a 100644 --- a/src/runtime/webcore/ReadableStream.rs +++ b/src/runtime/webcore/ReadableStream.rs @@ -577,6 +577,10 @@ pub trait SourceContext: Sized { fn js_on_drain_callback_set_cached(this: JSValue, global: &JSGlobalObject, value: JSValue); /// `js_${NAME}InternalReadableStreamSource::on_drain_callback_get_cached` fn js_on_drain_callback_get_cached(this: JSValue) -> Option; + /// `js_${NAME}InternalReadableStreamSource::on_close_callback_set_cached` + fn js_on_close_callback_set_cached(this: JSValue, global: &JSGlobalObject, value: JSValue); + /// `js_${NAME}InternalReadableStreamSource::on_close_callback_get_cached` + fn js_on_close_callback_get_cached(this: JSValue) -> Option; fn on_start(&mut self) -> streams::Start; fn on_pull(&mut self, buf: &mut [u8], view: JSValue) -> streams::Result; @@ -646,7 +650,6 @@ pub struct NewSource { /// owned/freed here). The JS path stores /// `on_js_close` and leaves this `None` — see [`Self::on_close`]. pub close_ctx: Option>, - pub close_jsvalue: bun_jsc::strong::Optional, /// R-2: cleared via `&self` from `FetchTasklet::clear_stream_cancel_handler` /// (through `ByteStream::parent_const`), so interior-mutable. pub cancel_handler: Cell)>>, @@ -655,9 +658,15 @@ pub struct NewSource { // `start()` from a fresh `&JSGlobalObject`; `BackRef` gives a safe `Deref` // projection without propagating a lifetime parameter into FFI codegen. pub global_this: Option>, - // SAFETY: this is the self-wrapper JSValue (points at the JSCell that owns this m_ctx). - // Kept alive by the wrapper itself; zeroed in finalize() before sweep. - pub this_jsvalue: JSValue, + /// Back-reference to the owning `JS{Blob,Bytes,File}InternalReadableStreamSource` + /// wrapper. Starts `Weak` (set in [`Self::to_readable_stream`]), is + /// [`JsRef::upgrade`]d to `Strong` in [`Self::increment_count`] while a + /// native I/O ref is held (FileReader `waiting_for_on_reader_done`), and + /// [`JsRef::downgrade`]d back to `Weak` in [`Self::decrement_count`] when + /// only the wrapper's own ref remains. [`Self::finalize`] flips it to + /// `Finalized` so [`Self::on_js_close`] reads `None` instead of a + /// dead-but-unswept cell. + pub this_jsvalue: jsc::JsRef, /// R-2: written by `&self` context methods (`ByteStream::to_any_blob`, /// `ByteBlobLoader::to_any_blob`) via `parent_const()`, so interior-mutable. pub is_closed: Cell, @@ -672,11 +681,10 @@ impl Default for NewSource { pending_err: None, close_handler: None, close_ctx: None, - close_jsvalue: bun_jsc::strong::Optional::empty(), cancel_handler: Cell::new(None), cancel_ctx: Cell::new(None), global_this: None, - this_jsvalue: JSValue::ZERO, + this_jsvalue: jsc::JsRef::empty(), is_closed: Cell::new(false), } } @@ -692,9 +700,11 @@ pub(crate) trait NewSourceCodegen { fn pending_promise_set_cached(this: JSValue, global: &JSGlobalObject, value: JSValue); fn on_drain_callback_set_cached(this: JSValue, global: &JSGlobalObject, value: JSValue); fn on_drain_callback_get_cached(this: JSValue) -> Option; + fn on_close_callback_set_cached(this: JSValue, global: &JSGlobalObject, value: JSValue); + fn on_close_callback_get_cached(this: JSValue) -> Option; } -/// Binds the four `SourceContext::js_*` accessors to the codegen'd +/// Binds the `SourceContext::js_*` accessors to the codegen'd /// `crate::generated_classes::js_${Name}InternalReadableStreamSource` module /// (one per `.classes.ts` entry: `Blob`, `File`, `Bytes`). The extern symbols /// are declared exactly once inside that module — no local `extern "C"` block. @@ -732,6 +742,20 @@ macro_rules! source_context_codegen { ) -> Option<$crate::webcore::jsc::JSValue> { $crate::generated_classes::$gen::on_drain_callback_get_cached(this) } + #[inline] + fn js_on_close_callback_set_cached( + this: $crate::webcore::jsc::JSValue, + global: &$crate::webcore::jsc::JSGlobalObject, + value: $crate::webcore::jsc::JSValue, + ) { + $crate::generated_classes::$gen::on_close_callback_set_cached(this, global, value) + } + #[inline] + fn js_on_close_callback_get_cached( + this: $crate::webcore::jsc::JSValue, + ) -> Option<$crate::webcore::jsc::JSValue> { + $crate::generated_classes::$gen::on_close_callback_get_cached(this) + } }; } @@ -754,6 +778,12 @@ impl NewSourceCodegen for NewSource { fn on_drain_callback_get_cached(this: JSValue) -> Option { C::js_on_drain_callback_get_cached(this) } + fn on_close_callback_set_cached(this: JSValue, global: &JSGlobalObject, value: JSValue) { + C::js_on_close_callback_set_cached(this, global, value) + } + fn on_close_callback_get_cached(this: JSValue) -> Option { + C::js_on_close_callback_get_cached(this) + } } // Enforce the layout invariant `from_js`/`Source` rely on. @@ -870,14 +900,37 @@ impl NewSource { fn on_js_close(ptr: Option<*mut c_void>) { // SAFETY: ptr was set to `self as *mut NewSource` in on_close()/set_on_close_from_js. let this = unsafe { &mut *(ptr.unwrap().cast::>()) }; - if let Some(cb) = this.close_jsvalue.try_swap() { - this.global_this().queue_microtask(cb, &[]); + // Reached from `FileReader::on_reader_done` off the event loop. While + // the across-read ref is held (`increment_count` upgraded to Strong), + // the wrapper is rooted and `try_get()` is `Some`. If the wrapper was + // already finalized, `try_get()` is `None` and there is no callback. + let Some(this_jsvalue) = this.this_jsvalue.try_get() else { + return; + }; + let global_this = this.global_this(); + if let Some(cb) = ::on_close_callback_get_cached(this_jsvalue) { + if !cb.is_undefined() { + global_this.queue_microtask(cb, &[]); + } } - this.close_jsvalue.deinit(); + ::on_close_callback_set_cached( + this_jsvalue, + global_this, + JSValue::UNDEFINED, + ); } pub fn increment_count(&mut self) { self.ref_count += 1; + // A ref beyond the JS wrapper's own is held (in practice a FileReader + // `waiting_for_on_reader_done` I/O ref). Root the wrapper so + // `on_js_close`, reached from `on_reader_done` off the event loop with + // no JS frame on the stack, never reads a dead-but-unswept cell. + if let Some(global) = self.global_this.as_deref() { + if self.this_jsvalue.is_not_empty() { + self.this_jsvalue.upgrade(global); + } + } } /// Release one reference. If the count hits zero, runs context teardown and @@ -901,10 +954,15 @@ impl NewSource { *r -= 1; *r }; + if remaining == 1 { + // Only the JS wrapper's own ref remains: drop the Strong root so + // the wrapper becomes collectable again. + // SAFETY: caller contract — `this` is live while remaining > 0. + unsafe { (*this).this_jsvalue.downgrade() }; + } if remaining == 0 { // SAFETY: still live; run side-effect teardown while fields are valid. unsafe { - (*this).close_jsvalue.deinit(); (*this).context.deinit_fn(); } // SAFETY: `this` originated from `Box::into_raw` in `Self::new`. No @@ -925,13 +983,15 @@ impl NewSource { } pub fn to_readable_stream(&mut self, global_this: &JSGlobalObject) -> JsResult { - let out_value = if self.this_jsvalue != JSValue::ZERO { - self.this_jsvalue + let out_value = if let Some(v) = self.this_jsvalue.try_get() { + v } else { ::to_js(self, global_this) }; out_value.ensure_still_alive(); - self.this_jsvalue = out_value; + if self.this_jsvalue.is_empty() { + self.this_jsvalue = jsc::JsRef::init_weak(out_value); + } ReadableStream::from_native(global_this, out_value) } @@ -983,7 +1043,6 @@ impl NewSource { let arguments = call_frame.arguments_old::<2>(); let view = arguments.ptr[0]; view.ensure_still_alive(); - self.this_jsvalue = this_jsvalue; let Some(mut buffer) = view.as_array_buffer(global_this) else { return Ok(JSValue::UNDEFINED); }; @@ -994,10 +1053,9 @@ impl NewSource { pub fn start_from_js( &mut self, global_this: &JSGlobalObject, - call_frame: &CallFrame, + _call_frame: &CallFrame, ) -> JsResult { self.global_this = Some(bun_ptr::BackRef::new(global_this)); - self.this_jsvalue = call_frame.this(); match self.on_start_from_js() { streams::Start::Empty => Ok(JSValue::js_number(0.0)), streams::Start::Ready => Ok(JSValue::js_number(16384.0)), @@ -1065,9 +1123,8 @@ impl NewSource { pub fn cancel_from_js( &mut self, _global_object: &JSGlobalObject, - call_frame: &CallFrame, + _call_frame: &CallFrame, ) -> JsResult { - self.this_jsvalue = call_frame.this(); self.cancel(); Ok(JSValue::UNDEFINED) } @@ -1084,7 +1141,13 @@ impl NewSource { self.global_this = Some(bun_ptr::BackRef::new(global_object)); if value.is_undefined() { - self.close_jsvalue.deinit(); + if let Some(this_jsvalue) = self.this_jsvalue.try_get() { + ::on_close_callback_set_cached( + this_jsvalue, + global_object, + JSValue::UNDEFINED, + ); + } return Ok(()); } @@ -1096,7 +1159,13 @@ impl NewSource { )); } let cb = value.with_async_context_if_needed(global_object); - self.close_jsvalue.set(global_object, cb); + if let Some(this_jsvalue) = self.this_jsvalue.try_get() { + ::on_close_callback_set_cached( + this_jsvalue, + global_object, + cb, + ); + } Ok(()) } @@ -1107,9 +1176,13 @@ impl NewSource { ) -> JsResult<()> { self.global_this = Some(bun_ptr::BackRef::new(global_object)); + let Some(this_jsvalue) = self.this_jsvalue.try_get() else { + return Ok(()); + }; + if value.is_undefined() { ::on_drain_callback_set_cached( - self.this_jsvalue, + this_jsvalue, global_object, JSValue::UNDEFINED, ); @@ -1124,20 +1197,25 @@ impl NewSource { )); } let cb = value.with_async_context_if_needed(global_object); - ::on_drain_callback_set_cached( - self.this_jsvalue, - global_object, - cb, - ); + ::on_drain_callback_set_cached(this_jsvalue, global_object, cb); Ok(()) } pub fn get_on_close_from_js(&mut self, _global_object: &JSGlobalObject) -> JSValue { - self.close_jsvalue.get().unwrap_or(JSValue::UNDEFINED) + if let Some(this_jsvalue) = self.this_jsvalue.try_get() { + if let Some(val) = + ::on_close_callback_get_cached(this_jsvalue) + { + return val; + } + } + JSValue::UNDEFINED } pub fn get_on_drain_from_js(&mut self, _global_object: &JSGlobalObject) -> JSValue { - ::on_drain_callback_get_cached(self.this_jsvalue) + self.this_jsvalue + .try_get() + .and_then(::on_drain_callback_get_cached) .unwrap_or(JSValue::UNDEFINED) } @@ -1146,7 +1224,6 @@ impl NewSource { _global_object: &JSGlobalObject, call_frame: &CallFrame, ) -> JsResult { - self.this_jsvalue = call_frame.this(); let ref_or_unref = call_frame.argument(0).to_boolean(); self.set_ref(ref_or_unref); Ok(JSValue::UNDEFINED) @@ -1158,7 +1235,7 @@ impl NewSource { // the raw refcount via a raw pointer (the call may free `*this`). let this = Box::into_raw(self); // SAFETY: `this` is live — just unwrapped from `Box`. - unsafe { (*this).this_jsvalue = JSValue::ZERO }; + unsafe { (*this).this_jsvalue.finalize() }; // SAFETY: `this` is live; the JS-wrapper ref below still pins the count. if unsafe { (*this).context.finalize_detach() } { // SAFETY: `this` is live; the JS-wrapper +1 (released below) keeps ref_count > 0. @@ -1171,9 +1248,8 @@ impl NewSource { pub fn drain_from_js( &mut self, global_this: &JSGlobalObject, - call_frame: &CallFrame, + _call_frame: &CallFrame, ) -> JsResult { - self.this_jsvalue = call_frame.this(); let mut list = self.drain(); if list.len() > 0 { // Ownership of the buffer transfers to JSC: `to_js` installs @@ -1192,10 +1268,9 @@ impl NewSource { fn to_buffered_value_from_js( &mut self, global_this: &JSGlobalObject, - call_frame: &CallFrame, + _call_frame: &CallFrame, action: streams::BufferActionTag, ) -> JsResult { - self.this_jsvalue = call_frame.this(); if let Some(r) = self.context.to_buffered_value(global_this, action) { return r; } diff --git a/test/js/web/streams/native-source-onclose-leak.test.ts b/test/js/web/streams/native-source-onclose-leak.test.ts new file mode 100644 index 000000000000..7d8f8f5285d8 --- /dev/null +++ b/test/js/web/streams/native-source-onclose-leak.test.ts @@ -0,0 +1,119 @@ +import { heapStats } from "bun:jsc"; +import { expect, test } from "bun:test"; + +// The JS onClose callback is installed on the native source wrapper when the +// lazy stream is first pulled. Storing it in a JSC::Strong rooted a cycle +// (source wrapper -> m_ctx NewSource -> Strong(onClose) -> bound fn -> +// NativeReadableStreamSource -> $stream -> source wrapper) that only broke +// when EOF ran the JS-side callClose or the cancel algorithm ran #cancel. A +// stream that is read partially and then dropped (releaseLock without cancel) +// never hits either path, so the source wrapper leaked forever. Storing +// onClose in the GC-traced onCloseCallback WriteBarrier slot (as onDrain +// already did) turns the cycle into an ordinary intra-heap cycle that +// mark-sweep collects. + +test("native ReadableStream source is collectable after partial read + releaseLock", async () => { + // Payload must exceed the native pull buffer (ByteBlobLoader caps + // chunk_size at 2MB) so the first pull does not return *AndDone and + // callClose is never queued. + const payload = Buffer.alloc(8 * 1024 * 1024, "x"); + async function once() { + const stream = new Blob([payload]).stream(); + const reader = stream.getReader(); + await reader.read(); + reader.releaseLock(); + } + + for (let i = 0; i < 5; i++) await once(); + Bun.gc(true); + const before = heapStats().objectTypeCounts.BlobInternalReadableStreamSource ?? 0; + + for (let i = 0; i < 30; i++) await once(); + Bun.gc(true); + await 1; + Bun.gc(true); + const after = heapStats().objectTypeCounts.BlobInternalReadableStreamSource ?? 0; + + // Pre-fix this grew by exactly 30 (one pinned wrapper per iteration). + expect(after - before).toBeLessThan(8); +}); + +test("fetch body native source is collectable after partial read + releaseLock", async () => { + const payload = Buffer.alloc(4 * 1024 * 1024, "x"); + using server = Bun.serve({ + port: 0, + fetch() { + return new Response(payload); + }, + }); + const url = `http://127.0.0.1:${server.port}/`; + + async function once() { + const res = await fetch(url); + const reader = res.body!.getReader(); + await reader.read(); + reader.releaseLock(); + } + + for (let i = 0; i < 5; i++) await once(); + Bun.gc(true); + const countSources = () => { + const c = heapStats().objectTypeCounts; + return (c.BlobInternalReadableStreamSource ?? 0) + (c.BytesInternalReadableStreamSource ?? 0); + }; + const before = countSources(); + + for (let i = 0; i < 30; i++) await once(); + Bun.gc(true); + await 1; + Bun.gc(true); + const after = countSources(); + + // Pre-fix this grew by ~30. + expect(after - before).toBeLessThan(8); +}); + +// Regression guard: the cycle-breaking paths that already worked must keep +// working after switching storage. +test("native ReadableStream source is collectable after full consumption", async () => { + const payload = Buffer.alloc(8 * 1024 * 1024, "x"); + async function once() { + const stream = new Blob([payload]).stream(); + for await (const _ of stream) { + } + } + + for (let i = 0; i < 5; i++) await once(); + Bun.gc(true); + const before = heapStats().objectTypeCounts.BlobInternalReadableStreamSource ?? 0; + + for (let i = 0; i < 30; i++) await once(); + Bun.gc(true); + await 1; + Bun.gc(true); + const after = heapStats().objectTypeCounts.BlobInternalReadableStreamSource ?? 0; + + expect(after - before).toBeLessThan(8); +}); + +test("native source onClose callback still fires after switching to cached slot", async () => { + // proc.stdout is a FileInternalReadableStreamSource; closing stdin makes + // cat exit, which EOFs stdout. FileReader.on_reader_done -> parent.on_close + // -> on_js_close must still find and invoke the JS onClose callback now + // that it lives in a WriteBarrier slot rather than a Strong. + await using proc = Bun.spawn({ + cmd: ["cat"], + stdin: "pipe", + stdout: "pipe", + env: { PATH: process.env.PATH }, + }); + const reader = proc.stdout.getReader(); + const pending = reader.read(); + proc.stdin.end(); + const { done } = await pending; + // If on_js_close did not fire, the pending pull promise would never resolve + // with done:true and this test would hang. + expect(done).toBe(true); + reader.releaseLock(); + await proc.exited; +});