diff --git a/src/jsc/bindings/NativePromiseContext.h b/src/jsc/bindings/NativePromiseContext.h index 42d64c734023..b33d38971ba2 100644 --- a/src/jsc/bindings/NativePromiseContext.h +++ b/src/jsc/bindings/NativePromiseContext.h @@ -42,7 +42,6 @@ class NativePromiseContext final : public JSC::JSCell { HTTPSServerRequestContext, DebugHTTPServerRequestContext, DebugHTTPSServerRequestContext, - BodyValueBufferer, HTTPSServerH3RequestContext, DebugHTTPSServerH3RequestContext, }; diff --git a/src/jsc/bindings/ZigGlobalObject.cpp b/src/jsc/bindings/ZigGlobalObject.cpp index a45cf0016dfe..9e3a42985960 100644 --- a/src/jsc/bindings/ZigGlobalObject.cpp +++ b/src/jsc/bindings/ZigGlobalObject.cpp @@ -4040,10 +4040,10 @@ GlobalObject::PromiseFunctions GlobalObject::promiseHandlerID(Zig::FFIFunction h return GlobalObject::PromiseFunctions::Bun__TestScope__Describe2__bunTestThen; } else if (handler == Bun__TestScope__Describe2__bunTestCatch) { return GlobalObject::PromiseFunctions::Bun__TestScope__Describe2__bunTestCatch; - } else if (handler == Bun__BodyValueBufferer__onResolveStream) { - return GlobalObject::PromiseFunctions::Bun__BodyValueBufferer__onResolveStream; - } else if (handler == Bun__BodyValueBufferer__onRejectStream) { - return GlobalObject::PromiseFunctions::Bun__BodyValueBufferer__onRejectStream; + } else if (handler == Bun__WriteFileLocked__onStreamResolved) { + return GlobalObject::PromiseFunctions::Bun__WriteFileLocked__onStreamResolved; + } else if (handler == Bun__WriteFileLocked__onStreamRejected) { + return GlobalObject::PromiseFunctions::Bun__WriteFileLocked__onStreamRejected; } else if (handler == Bun__onResolveEntryPointResult) { return GlobalObject::PromiseFunctions::Bun__onResolveEntryPointResult; } else if (handler == Bun__onRejectEntryPointResult) { diff --git a/src/jsc/bindings/ZigGlobalObject.h b/src/jsc/bindings/ZigGlobalObject.h index ca759a74da9e..d4dc19ba41ac 100644 --- a/src/jsc/bindings/ZigGlobalObject.h +++ b/src/jsc/bindings/ZigGlobalObject.h @@ -392,8 +392,8 @@ class GlobalObject : public Bun::GlobalScope { jsFunctionOnLoadObjectResultReject, Bun__TestScope__Describe2__bunTestThen, Bun__TestScope__Describe2__bunTestCatch, - Bun__BodyValueBufferer__onRejectStream, - Bun__BodyValueBufferer__onResolveStream, + Bun__WriteFileLocked__onStreamResolved, + Bun__WriteFileLocked__onStreamRejected, Bun__onResolveEntryPointResult, Bun__onRejectEntryPointResult, Bun__NodeHTTPRequest__onResolve, diff --git a/src/jsc/bindings/headers-cpp.h b/src/jsc/bindings/headers-cpp.h index 1453b0a0fb1b..82ae825c7f6d 100644 --- a/src/jsc/bindings/headers-cpp.h +++ b/src/jsc/bindings/headers-cpp.h @@ -182,9 +182,6 @@ extern "C" const size_t Bun__Timer_object_align_ = alignof(Bun__Timer); #include "" #endif -extern "C" const size_t Bun__BodyValueBufferer_object_size_ = sizeof(Bun__BodyValueBufferer); -extern "C" const size_t Bun__BodyValueBufferer_object_align_ = alignof(Bun__BodyValueBufferer); - const size_t sizes[39] = {sizeof(JSC::JSObject), sizeof(WebCore::DOMURL), sizeof(WebCore::DOMFormData), sizeof(WebCore::FetchHeaders), sizeof(SystemError), sizeof(JSC::JSCell), sizeof(JSC::JSString), sizeof(JSC::JSModuleLoader), sizeof(WebCore::AbortSignal), sizeof(JSC::JSPromise), sizeof(JSC::JSPromise), sizeof(JSC::JSFunction), sizeof(JSC::JSGlobalObject), sizeof(JSC::JSMap), sizeof(JSC::JSValue), sizeof(JSC::Exception), sizeof(JSC::VM), sizeof(JSC::ThrowScope), sizeof(JSC::TopExceptionScope), sizeof(FFI__ptr), sizeof(Reader__u8), sizeof(Reader__u16), sizeof(Reader__u32), sizeof(Reader__ptr), sizeof(Reader__i8), sizeof(Reader__i16), sizeof(Reader__i32), sizeof(Reader__f32), sizeof(Reader__f64), sizeof(Reader__i64), sizeof(Reader__u64), sizeof(Reader__intptr), sizeof(Zig::GlobalObject), sizeof(Bun__Path), sizeof(ArrayBufferSink), sizeof(HTTPSResponseSink), sizeof(HTTPResponseSink), sizeof(FileSink), sizeof(FileSink)}; const size_t aligns[39] = {alignof(JSC::JSObject), alignof(WebCore::DOMURL), alignof(WebCore::DOMFormData), alignof(WebCore::FetchHeaders), alignof(SystemError), alignof(JSC::JSCell), alignof(JSC::JSString), alignof(JSC::JSModuleLoader), alignof(WebCore::AbortSignal), alignof(JSC::JSPromise), alignof(JSC::JSPromise), alignof(JSC::JSFunction), alignof(JSC::JSGlobalObject), alignof(JSC::JSMap), alignof(JSC::JSValue), alignof(JSC::Exception), alignof(JSC::VM), alignof(JSC::ThrowScope), alignof(JSC::TopExceptionScope), alignof(FFI__ptr), alignof(Reader__u8), alignof(Reader__u16), alignof(Reader__u32), alignof(Reader__ptr), alignof(Reader__i8), alignof(Reader__i16), alignof(Reader__i32), alignof(Reader__f32), alignof(Reader__f64), alignof(Reader__i64), alignof(Reader__u64), alignof(Reader__intptr), alignof(Zig::GlobalObject), alignof(Bun__Path), alignof(ArrayBufferSink), alignof(HTTPSResponseSink), alignof(HTTPResponseSink), alignof(FileSink), alignof(FileSink)}; diff --git a/src/jsc/bindings/headers.h b/src/jsc/bindings/headers.h index ab056ebc60bd..b4b78b5a78c1 100644 --- a/src/jsc/bindings/headers.h +++ b/src/jsc/bindings/headers.h @@ -772,20 +772,12 @@ BUN_DECLARE_HOST_FUNCTION(Bun__HTTPRequestContextDebugTLS__onResolveStream); #endif -#pragma mark - Bun__BodyValueBufferer - - -#ifdef __cplusplus - -BUN_DECLARE_HOST_FUNCTION(Bun__BodyValueBufferer__onRejectStream); -BUN_DECLARE_HOST_FUNCTION(Bun__BodyValueBufferer__onResolveStream); - -#endif - #ifdef __cplusplus BUN_DECLARE_HOST_FUNCTION(Bun__TestScope__Describe2__bunTestThen); BUN_DECLARE_HOST_FUNCTION(Bun__TestScope__Describe2__bunTestCatch); +BUN_DECLARE_HOST_FUNCTION(Bun__WriteFileLocked__onStreamResolved); +BUN_DECLARE_HOST_FUNCTION(Bun__WriteFileLocked__onStreamRejected); BUN_DECLARE_HOST_FUNCTION(Bun__CronJob__onPromiseResolve); BUN_DECLARE_HOST_FUNCTION(Bun__CronJob__onPromiseReject); diff --git a/src/jsc/generated.rs b/src/jsc/generated.rs index 5849fe99546e..f648c5e00221 100644 --- a/src/jsc/generated.rs +++ b/src/jsc/generated.rs @@ -1134,8 +1134,9 @@ js_class_module!(JSBlob = "Blob" as crate::webcore_types::Blob { name, js_class_module!(JSResponse = "Response" { body, headers, url, statusText, stream }); js_class_module!(JSRequest = "Request" { body, headers, url, signal, stream }); // `values: ["ondrain", "oncancel", "stream"]` in src/runtime/api/ResumableSink.classes.ts. -js_class_module!(JSResumableFetchSink = "ResumableFetchSink" { ondrain, oncancel, stream }); -js_class_module!(JSResumableS3UploadSink = "ResumableS3UploadSink" { ondrain, oncancel, stream }); +js_class_module!(JSResumableFetchSink = "ResumableFetchSink" { ondrain, oncancel, stream }); +js_class_module!(JSResumableS3UploadSink = "ResumableS3UploadSink" { ondrain, oncancel, stream }); +js_class_module!(JSResumableHTMLRewriterSink = "ResumableHTMLRewriterSink" { ondrain, oncancel, stream }); // `values: ["resolve", "reject"]` in src/runtime/api/Shell.classes.ts. js_class_module!(JSShellInterpreter = "ShellInterpreter" { resolve, reject }); // `src/runtime/crypto/crypto.classes.ts` — one entry per `StaticCryptoHasher` diff --git a/src/jsc/generated_classes_list.rs b/src/jsc/generated_classes_list.rs index 53eccf23d622..ca36dfac014b 100644 --- a/src/jsc/generated_classes_list.rs +++ b/src/jsc/generated_classes_list.rs @@ -98,6 +98,7 @@ pub mod Classes { pub use crate::webcore::Request; pub use crate::webcore::Response; pub use crate::webcore::ResumableFetchSink; + pub use crate::webcore::ResumableHTMLRewriterSink; pub use crate::webcore::ResumableS3UploadSink; pub use crate::webcore::S3Client; pub use crate::webcore::S3Stat; diff --git a/src/runtime/api/NativePromiseContext.rs b/src/runtime/api/NativePromiseContext.rs index 86fa7fe727cc..2b0dfa36ec2b 100644 --- a/src/runtime/api/NativePromiseContext.rs +++ b/src/runtime/api/NativePromiseContext.rs @@ -24,9 +24,7 @@ use bun_event_loop::{Task, TaskTag, Taskable, task_tag}; use bun_jsc::virtual_machine::VirtualMachine; use bun_jsc::{JSGlobalObject, JSValue}; -use crate::api::html_rewriter; use crate::api::server; -use crate::webcore::body; // Request contexts are a single generic // `NewRequestContext`; alias the six @@ -53,13 +51,12 @@ pub enum Tag { HTTPSServerRequestContext, DebugHTTPServerRequestContext, DebugHTTPSServerRequestContext, - BodyValueBufferer, HTTPSServerH3RequestContext, DebugHTTPSServerH3RequestContext, } impl Tag { - pub const COUNT: usize = 7; + pub const COUNT: usize = 6; #[inline] const fn from_raw(n: u8) -> Tag { @@ -68,9 +65,8 @@ impl Tag { 1 => Tag::HTTPSServerRequestContext, 2 => Tag::DebugHTTPServerRequestContext, 3 => Tag::DebugHTTPSServerRequestContext, - 4 => Tag::BodyValueBufferer, - 5 => Tag::HTTPSServerH3RequestContext, - 6 => Tag::DebugHTTPSServerH3RequestContext, + 4 => Tag::HTTPSServerH3RequestContext, + 5 => Tag::DebugHTTPSServerH3RequestContext, _ => unreachable!(), } } @@ -103,9 +99,6 @@ impl NativePromise { const TAG: Tag = npc_tag_for(SSL, DBG, H3); } -impl NativePromiseContextType for body::ValueBufferer<'_> { - const TAG: Tag = Tag::BodyValueBufferer; -} // `&JSGlobalObject` is ABI-identical to a non-null pointer. `ctx` is stored // opaquely (never dereferenced by the C++ side), so the FFI itself has no @@ -214,16 +207,6 @@ impl DeferredDerefTask { Tag::DebugHTTPSServerRequestContext => { (*ctx.cast::()).deref() } - Tag::BodyValueBufferer => { - // ValueBufferer is embedded by value inside HTMLRewriter's - // BufferOutputSink, with the owner pointer stored in .ctx. - // The pending-promise ref was taken on the owner, so we - // release it there. - let bufferer = &*ctx.cast::>(); - html_rewriter::BufferOutputSink::deref( - bufferer.ctx.cast::(), - ); - } Tag::HTTPSServerH3RequestContext => { (*ctx.cast::()).deref() } @@ -246,5 +229,3 @@ const _: () = assert!(core::mem::align_of::() > DeferredDerefTask::TAG_MASK); const _: () = assert!(core::mem::align_of::() > DeferredDerefTask::TAG_MASK); -const _: () = - assert!(core::mem::align_of::>() > DeferredDerefTask::TAG_MASK); diff --git a/src/runtime/api/ResumableSink.classes.ts b/src/runtime/api/ResumableSink.classes.ts index 388343aa4d18..26c1fbe8304d 100644 --- a/src/runtime/api/ResumableSink.classes.ts +++ b/src/runtime/api/ResumableSink.classes.ts @@ -32,4 +32,8 @@ function generate(name) { values: ["ondrain", "oncancel", "stream"], }); } -export default [generate("ResumableFetchSink"), generate("ResumableS3UploadSink")]; +export default [ + generate("ResumableFetchSink"), + generate("ResumableS3UploadSink"), + generate("ResumableHTMLRewriterSink"), +]; diff --git a/src/runtime/api/html_rewriter.rs b/src/runtime/api/html_rewriter.rs index 2efac29cb9af..c9cf5779c302 100644 --- a/src/runtime/api/html_rewriter.rs +++ b/src/runtime/api/html_rewriter.rs @@ -4,10 +4,9 @@ use core::cell::{Cell, RefCell}; use core::ptr::NonNull; use std::rc::Rc; -use bun_core::MutableString; use bun_jsc::{ self as jsc, CallFrame, GlobalRef, JSGlobalObject, JSValue, JsCell, JsResult, ProtectedJSValue, - StrongOptional, SystemError, bun_string_jsc, + bun_string_jsc, }; // Note: `bun_jsc::VirtualMachine` is a *module* re-export // (`pub use self::virtual_machine as VirtualMachine;`). The struct lives at @@ -16,8 +15,12 @@ use bun_jsc::{ // owner of the `on_quiet_unhandled_rejection_handler_capture_value` assoc fn. use bun_jsc::virtual_machine::VirtualMachine; +use crate::webcore::ByteStream; use crate::webcore::response::HeadersRef; -use crate::webcore::{self, Response}; +use crate::webcore::resumable_sink::{ + ResumableHTMLRewriterSink, ResumableSinkBackpressure, ResumableSinkContext, +}; +use crate::webcore::{self, ReadableStream, Response, streams}; use bun_core::String as BunString; // `ZigString` re-exports `bun_core::ZigString`; JSC-side methods // (`to_js`, `with_encoding`, …) come from the `ZigStringJsc` extension trait. @@ -57,15 +60,6 @@ fn cell_get<'a, T>(cell: &Cell<*mut T>) -> Option<&'a mut T> { unsafe { cell.get().as_mut() } } -/// Construct a `SystemError` with code+message and remaining fields defaulted. -fn system_error(code: &'static str, message: &'static str) -> SystemError { - SystemError { - code: BunString::static_(code).into(), - message: BunString::static_(message).into(), - ..Default::default() - } -} - // ─────────────────── instance-method arg-decode helpers ────────────────── // // Note: a `#[bun_jsc::host_fn(method)]` proc-macro form of typed argument @@ -472,12 +466,13 @@ impl HTMLRewriter { return Ok(out_response_value); }; // SAFETY: out_response is the m_ctx of out_response_value (kept alive - // on the stack via ensure_still_alive above). - let mut blob = unsafe { - (*out_response) - .get_body_value() - .use_as_any_blob_allow_non_utf8_string() - }; + // on the stack via ensure_still_alive above). `to_js` migrated the + // output ByteStream to the wrapper's cached stream slot. + let mut blob = unsafe { (*out_response).get_body_readable_stream(global) } + .and_then(|mut s| s.to_any_blob(global)) + .unwrap_or(webcore::AnyBlob::Blob(Default::default())); + // SAFETY: out_response is live (see above). + unsafe { *(*out_response).get_body_value() = webcore::body::Value::Used }; let _out_guard = scopeguard::guard((out_response_value, out_response), |(v, r)| { // `Response.js.dangerouslySetPtr(v, null)` — null out the JS @@ -538,44 +533,83 @@ impl HTMLRewriter { // ───────────────────────── BufferOutputSink ────────────────────────────── +/// Drives one `HTMLRewriter.transform()` call: pulls input chunks from the +/// source body via `ResumableSink`, feeds them to lol-html, and delivers the +/// rewritten output to a `ByteStream` that backs the returned `Response` body. #[derive(bun_ptr::CellRefCounted)] pub struct BufferOutputSink { - // Intrusive RefCount; *Self is the `SinkRef` carried inside `rewriter`. ref_count: Cell, - pub global: GlobalRef, // JSC_BORROW - pub bytes: MutableString, - // Heap-allocated (never held by value): `run_output_sink` must reach the - // rewriter through a raw pointer, never a `&mut` of `*sink`, because the - // output sink re-enters `&mut *sink` while the rewriter runs. - pub rewriter: *mut lol_html::HtmlRewriter<'static, SinkRef>, // null when unset + pub global: GlobalRef, + rewriter: Cell<*mut lol_html::HtmlRewriter<'static, SinkRef>>, pub context: Rc>, - pub response: *mut Response, // BORROW_FIELD: kept alive by response_value Strong - pub response_value: StrongOptional, - pub body_value_bufferer: Option>, - // Points at the `sink_error` stack local in `init()`; - // only written while `init()` is on the stack. - // See `write_tmp_sync_error` for the full liveness/provenance argument. - pub tmp_sync_error: Option>, + /// GC root for the output `ByteStream`'s JS wrapper. + output: webcore::readable_stream::Strong, + /// First error pushed via [`fail`](Self::fail), so a synchronous handler + /// error inside `init()` still makes `transform()` throw. + failed: JsCell, } -impl BufferOutputSink { - // `ref_()`/`deref()` provided by `#[derive(CellRefCounted)]`. +impl ResumableSinkContext for BufferOutputSink { + /// `feed` drives `HtmlRewriter::write`, which runs async handlers via + /// `wait_for_promise`. The JS pump cannot deliver the next chunk until + /// `sink.write()` returns (`m_reading` guard + read-after-write ordering); + /// the native pipe can, so skip it and let the pump read `Source::Bytes`. + const AVOID_NATIVE_PIPE: bool = true; + + fn write_request_data(&mut self, bytes: &[u8]) -> ResumableSinkBackpressure { + let captured = Cell::new(JSValue::ZERO); + let _scope = HandlerErrorScope::enter(&self.global, &captured); + self.feed(bytes); + if self.rewriter.get().is_null() { + // `fail()` destroyed the rewriter; stop the pump so the source + // is cancelled and `write_end_request` fires. + ResumableSinkBackpressure::Done + } else { + ResumableSinkBackpressure::WantMore + } + } - /// Single unsafe deref site for the set-once - /// `tmp_sync_error: Option>` field, so the two callers in - /// `on_finished_buffering` stay safe. `tmp_sync_error` points at the - /// `sink_error: Cell` stack local in [`init`]; it is only written - /// through on the synchronous (`is_async == false`) path while `init` is - /// still on the stack, so the pointee is live and the `Cell`-derived - /// pointer carries `SharedReadWrite` provenance. - #[inline] - fn write_tmp_sync_error(sink: *mut Self, err: JSValue) { - // SAFETY: `sink` is a live heap allocation (refcount > 0, caller - // invariant); `tmp_sync_error` was set in `init()` and the synchronous - // caller is reached only while `init()` is still on the stack. - unsafe { *(*sink).tmp_sync_error.unwrap().as_ptr() = err }; + fn write_end_request(&mut self, err: Option) { + let captured = Cell::new(JSValue::ZERO); + let _scope = HandlerErrorScope::enter(&self.global, &captured); + self.finish(err); + // SAFETY: `&mut self` keeps the allocation live; releases the in-flight + // +1 taken in `init()`. `self` is not touched after this. + unsafe { Self::deref(core::ptr::from_mut(self)) }; } +} +/// RAII guard that points `vm.unhandled_pending_rejection_to_capture` at a +/// caller-owned cell and installs the quiet rejection handler for its scope. +/// `create_lolhtml_error` reads that slot to surface the user's thrown error. +struct HandlerErrorScope { + prev_capture: Option<*mut JSValue>, + rejection_scope: bun_jsc::virtual_machine::UnhandledRejectionScope, +} + +impl HandlerErrorScope { + fn enter(global: &JSGlobalObject, captured: &Cell) -> Self { + let vm: &mut VirtualMachine = global.bun_vm().as_mut(); + let scope = Self { + prev_capture: vm.unhandled_pending_rejection_to_capture, + rejection_scope: vm.unhandled_rejection_scope(), + }; + vm.unhandled_pending_rejection_to_capture = Some(captured.as_ptr()); + vm.on_unhandled_rejection = + VirtualMachine::on_quiet_unhandled_rejection_handler_capture_value; + scope + } +} + +impl Drop for HandlerErrorScope { + fn drop(&mut self) { + let vm = VirtualMachine::get().as_mut(); + vm.unhandled_pending_rejection_to_capture = self.prev_capture; + self.rejection_scope.apply(vm); + } +} + +impl BufferOutputSink { /// # Safety /// `original` must point to a live `Response` whose JS wrapper is kept /// alive for the duration of this call. @@ -584,85 +618,38 @@ impl BufferOutputSink { global: &JSGlobalObject, original: *mut Response, ) -> JsResult { + let source = webcore::readable_stream::NewSource::::new_mut( + webcore::readable_stream::NewSource { + context: ByteStream::default(), + global_this: Some(bun_ptr::BackRef::new(global)), + ..Default::default() + }, + ); + source.context.setup(); + let out_bytes: *mut ByteStream = &raw mut source.context; + let out_stream_js = source.to_readable_stream(global)?; + let out_readable = ReadableStream { + ptr: webcore::readable_stream::Source::Bytes(out_bytes), + value: out_stream_js, + }; + + // SAFETY: original is a live *Response kept alive by caller. + let input_size = unsafe { (*original).get_body_len() }; + let sink = bun_core::heap::into_raw(Box::new(BufferOutputSink { ref_count: Cell::new(1), global: GlobalRef::from(global), - bytes: MutableString::init_empty(), - rewriter: core::ptr::null_mut(), + rewriter: Cell::new(core::ptr::null_mut()), context, - response: core::ptr::null_mut(), - response_value: StrongOptional::empty(), - body_value_bufferer: None, - tmp_sync_error: None, + output: webcore::readable_stream::Strong::init(out_readable, global), + failed: JsCell::new(jsc::strong::Optional::empty()), })); - // SAFETY: `sink` is the `heap::into_raw` allocation above; refcount >= 1. + // SAFETY: `sink` is the `heap::into_raw` allocation above; refcount == 1. let _sink_guard = unsafe { bun_ptr::ScopedRef::::adopt(sink) }; - // Note: do not hold a long-lived `&mut *sink` here — the same - // allocation is also written through the raw pointer by the lol-html - // output-sink callback during `bufferer.run()` and by `deref(sink)` - // below. Access fields via raw-pointer place expressions instead. - - let result = bun_core::heap::into_raw(Box::new(Response::init( - webcore::response::Init { - status_code: 200, - ..Default::default() - }, - webcore::Body::new({ - let mut pv = webcore::body::PendingValue::new(global); - pv.task = Some(sink.cast::()); - webcore::body::Value::Locked(pv) - }), - BunString::empty(), - false, - ))); - // SAFETY: sink was just allocated via heap::alloc above; refcount==1. - unsafe { (*sink).response = result }; - // Note (Stacked Borrows): `sink_error` is written via raw pointer - // by the unhandled-rejection handler during `bufferer.run()` and via - // `tmp_sync_error` from `on_finished_buffering`. Use a `Cell` so the - // exported `*mut` (via `Cell::as_ptr`, i.e. `UnsafeCell::get`) carries - // SharedReadWrite provenance — local `.get()` reads do NOT invalidate - // the stored raw pointer the way a `&`/`&mut` reborrow of a plain - // `mut` local would. - let sink_error: core::cell::Cell = core::cell::Cell::new(JSValue::ZERO); - let sink_error_ptr: *mut JSValue = sink_error.as_ptr(); - // SAFETY: original is a live *Response passed from begin_transform; its - // JS wrapper is on the caller's stack. - let input_size = unsafe { (*original).get_body_len() }; - // SAFETY: bun_vm() returns the live VM raw ptr; VM outlives this fn. - let vm: &mut VirtualMachine = global.bun_vm().as_mut(); - - // Since we're still using vm.waitForPromise, we have to also override - // the error rejection handler. That way, we can propagate errors to the - // caller. - let scope = vm.unhandled_rejection_scope(); - let prev_unhandled_pending_rejection_to_capture = vm.unhandled_pending_rejection_to_capture; - vm.unhandled_pending_rejection_to_capture = Some(sink_error_ptr); - // SAFETY: sink is a live heap allocation (refcount >= 1); sink_error_ptr - // is non-null (addr of stack local). - unsafe { (*sink).tmp_sync_error = Some(NonNull::new_unchecked(sink_error_ptr)) }; - vm.on_unhandled_rejection = - VirtualMachine::on_quiet_unhandled_rejection_handler_capture_value; - // Read the *live* slot at scope exit (Cell shares provenance with the - // raw-pointer writers). - scopeguard::defer! { - sink_error.get().ensure_still_alive(); - // SAFETY: VM outlives this guard (sync stack frame). - let vm = VirtualMachine::get().as_mut(); - vm.unhandled_pending_rejection_to_capture = prev_unhandled_pending_rejection_to_capture; - scope.apply(vm); - } - - // The handler closures point into `Box`es owned by `(*sink).context`, - // which `sink` keeps alive for the rewriter's whole lifetime. - // SAFETY: sink is a live heap allocation (refcount >= 1); the `RefMut` - // of `(*sink).context` is released at the end of this statement. + // SAFETY: sink is live; the `RefMut` is released at end of statement. let (element_content_handlers, document_content_handlers) = unsafe { build_settings(&mut (*sink).context.borrow_mut()) }; - // `SinkRef` carries the raw `sink` (`heap::into_raw` root) so every - // `(*sink).field` access shares its provenance; `run_output_sink` - // reaches the rewriter through a raw pointer, never `&mut *sink`. let rewriter = bun_core::heap::into_raw(Box::new(lol_html::HtmlRewriter::new( lol_html::Settings { element_content_handlers, @@ -682,300 +669,215 @@ impl BufferOutputSink { enable_esi_tags: false, adjust_charset_on_meta_tag: false, }, - SinkRef(sink), + SinkRef(out_bytes), ))); - // SAFETY: sink is a live heap allocation (refcount >= 1). - unsafe { (*sink).rewriter = rewriter }; + // SAFETY: sink is a live heap allocation. + unsafe { (*sink).rewriter.set(rewriter) }; - // SAFETY: result and original are both live *Response (result allocated - // above, original kept alive by caller); no aliasing &mut exists. + let result = bun_core::heap::into_raw(Box::new(Response::init( + webcore::response::Init { + status_code: 200, + ..Default::default() + }, + webcore::Body::new( + webcore::body::Value::from_readable_stream_without_lock_check(out_readable, global), + ), + BunString::empty(), + false, + ))); + let result_guard = scopeguard::guard(result, |r| { + // SAFETY: `r` is the `heap::into_raw` allocation above; sole owner. + Response::finalize(unsafe { Box::from_raw(r) }) + }); + + // SAFETY: result and original are both live *Response. unsafe { (*result).set_init( (*original).get_method(), (*original).get_init_status_code(), (*original).get_init_status_text().clone(), ); - // https://github.com/oven-sh/bun/issues/3334 - // Note: `clone_this` takes `&mut self`, so use the `_mut` - // accessor (original is `*mut Response`). `clone_this` only reads - // `self` (FFI mutates a freshly-allocated clone, not the receiver). if let Some(headers) = (*original).get_init_headers_mut() { let cloned = headers.clone_this(global)?; (*result).set_init_headers(cloned.map(|p| HeadersRef::adopt(p))); } + (*result).set_url((*original).url().clone()); } - - // Hold off on cloning until we're actually done. - // SAFETY: (*sink).response == result (set above), live heap allocation. - let response_js_value = unsafe { (*(*sink).response).to_js(&(*sink).global) }; - // SAFETY: sink is a live heap allocation (refcount >= 1). - unsafe { (*sink).response_value.set(global, response_js_value) }; - - // SAFETY: result/original are live *Response (see SAFETY note above). - // `url()` is +0 borrowed-bits; `set_url` takes +1 — `.clone()` to bump. - unsafe { (*result).set_url((*original).url().clone()) }; + // SAFETY: `result` is live; `to_js` transfers ownership to the wrapper. + let response_js_value = + unsafe { (*scopeguard::ScopeGuard::into_inner(result_guard)).to_js(global) }; + response_js_value.ensure_still_alive(); // SAFETY: original is a live *Response kept alive by caller. let value = unsafe { (*original).get_body_value() }; - // SAFETY: original is a live *Response kept alive by caller; sink live. let owned_readable_stream = - unsafe { (*original).get_body_readable_stream(&(*sink).global) }; - // SAFETY: sink is a live heap allocation (refcount >= 1). - unsafe { - (*sink).ref_(); - (*sink).body_value_bufferer = Some(webcore::body::ValueBufferer::init( - sink.cast::(), - // Note: `ValueBuffererCallback` takes `*mut c_void` for ctx; - // `on_finished_buffering` takes `*mut BufferOutputSink`. The - // wrapper trampoline restores the concrete type. - Self::on_finished_buffering_trampoline, - &(*sink).global, - )); - } - response_js_value.ensure_still_alive(); + // SAFETY: original is a live *Response kept alive by caller. + unsafe { (*original).get_body_readable_stream(global) }; - // SAFETY: sink is a live heap allocation; body_value_bufferer was just - // set to Some above. `run()` may synchronously invoke - // `on_finished_buffering`, which (via the rewriter's output sink) - // re-enters `SinkRef::handle_chunk` and forms a fresh - // `&mut *sink`. Hoist the bufferer through a raw pointer so no `&mut` - // derived from `*sink` is live across that callback. - let buffering_result: crate::Result<()> = unsafe { - let bufferer: *mut webcore::body::ValueBufferer = - (*sink).body_value_bufferer.as_mut().unwrap(); - (*bufferer).run(value, owned_readable_stream) - }; - if let Err(buffering_error) = buffering_result { - // SAFETY: `sink` is a live `heap::into_raw` allocation; release the - // ref taken for the in-flight bufferer. - unsafe { BufferOutputSink::deref(sink) }; - return Ok(match buffering_error { - crate::Error::StreamAlreadyUsed => { - let err = system_error( - "ERR_STREAM_ALREADY_FINISHED", - "Stream already used, please create a new one", - ); - err.to_error_instance(global) - } - _ => { - let err = system_error("ERR_STREAM_CANNOT_PIPE", "Failed to pipe stream"); - err.to_error_instance(global) - } - }); + { + let captured = Cell::new(JSValue::ZERO); + let _scope = HandlerErrorScope::enter(global, &captured); + // SAFETY: sink is a live heap allocation; `new` bumps, `forget` + // hands the +1 to `write_end_request` (or `finish` on the sync + // path), which releases it. + let in_flight = unsafe { bun_ptr::ScopedRef::::new(sink) }; + // SAFETY: sink is a live heap allocation. + unsafe { (*sink).start_reading_input(value, owned_readable_stream)? }; + in_flight.forget(); } - // sync error occurs — read via the Cell (shares SharedReadWrite - // provenance with the raw-pointer writers; see Note above). - let captured = sink_error.get(); - if !captured.is_empty() { - captured.ensure_still_alive(); - captured.unprotect(); - // Throw directly: the callers gate on `JSValue::to_error()`, which - // only recognises `ErrorInstance`/`Exception`, so an abort reason - // (a DOMException or any user value) would be returned instead. - return Err(global.throw_value(captured)); + // SAFETY: sink is a live heap allocation. + if let Some(err) = unsafe { (*sink).failed.with_mut(|f| f.try_swap()) } { + return Err(global.throw_value(err)); } response_js_value.ensure_still_alive(); Ok(response_js_value) } - fn on_finished_buffering_trampoline( - ctx: *mut core::ffi::c_void, - bytes: &[u8], - js_err: Option, - is_async: bool, - ) { - // SAFETY: `ctx` is the `sink` heap allocation registered with the - // bufferer in `init()`; it was `ref_()`'d there so refcount > 0. - unsafe { - Self::on_finished_buffering(ctx.cast::(), bytes, js_err, is_async) - } - } + fn start_reading_input( + &mut self, + value: &mut webcore::body::Value, + owned_readable_stream: Option, + ) -> JsResult<()> { + let global = self.global; - /// # Safety - /// `sink` must be a live `BufferOutputSink` heap allocation with - /// refcount > 0 (the +1 taken in `init()` is consumed here). - unsafe fn on_finished_buffering( - sink: *mut BufferOutputSink, - bytes: &[u8], - js_err: Option, - is_async: bool, - ) { - // SAFETY: `sink` was ref'd in `init()` before scheduling this callback; - // refcount > 0 so the allocation is live. `adopt` consumes that +1 on Drop. - let _g = unsafe { bun_ptr::ScopedRef::::adopt(sink) }; - // Note: do not materialise `&mut *sink` here — the rewriter - // write/end calls below re-enter `SinkRef::handle_chunk` - // through the stored raw pointer, which forms - // its own `&mut *sink`. Holding an outer `&mut` across that re-entry - // is aliased-&mut UB. Access fields via raw-pointer place expressions - // instead (mirroring `init()`). - // - // SAFETY: sink was ref'd in init() before scheduling this callback; - // refcount > 0 so the allocation is live. - let global = unsafe { (*sink).global }; - - if let Some(mut err) = js_err { - // SAFETY: (*sink).response is the heap Response allocated in init() - // and kept alive by (*sink).response_value (Strong root). - let sink_body_value = unsafe { (*(*sink).response).get_body_value() }; - let sink_ptr_usize = sink as usize; - // If a `.body` readable is already attached, stay `Locked` so - // `to_error_instance` delivers the error to its ByteStream; clearing - // to `Empty` here would strand any pending `reader.read()` forever. - let has_readable = match sink_body_value { - webcore::body::Value::Locked(l) => l.readable.has(), - _ => false, - }; - if !has_readable - && matches!(sink_body_value, webcore::body::Value::Locked(l) - if l.task.map_or(0, |p| p as usize) == sink_ptr_usize && l.promise.is_none()) - { - // No reader and no pending read: normalize to `Empty` so - // `to_error_instance` takes the simple (non-`Locked`) path. - *sink_body_value = webcore::body::Value::Empty; - } else if matches!(sink_body_value, webcore::body::Value::Locked(l) - if l.task.map_or(0, |p| p as usize) == sink_ptr_usize && l.promise.is_some()) + let readable_stream = if let Some(stream) = owned_readable_stream { + stream + } else { + value.to_blob_if_possible(); + if let webcore::body::Value::Error(err) = value { + let js_err = err.to_js(&global); + self.fail(js_err); + // SAFETY: `&mut self` keeps the allocation live; releases the + // in-flight +1 taken in `init()`. + unsafe { Self::deref(core::ptr::from_mut(self)) }; + return Ok(()); + } + if let webcore::body::Value::WTFStringImpl(_) + | webcore::body::Value::InternalBlob(_) + | webcore::body::Value::Blob(_) = value { - if let webcore::body::Value::Locked(l) = sink_body_value { - l.on_receive_value = None; - l.task = None; + let mut input = value.use_as_any_blob_allow_non_utf8_string(); + if !input.needs_to_read_file() { + // Run synchronously so `transform(String | ArrayBuffer)` can + // read the output body back as a blob before returning. + self.feed(input.slice()); + input.detach(); + self.finish(None); + // SAFETY: `&mut self` keeps the allocation live; releases + // the in-flight +1 taken in `init()`. + unsafe { Self::deref(core::ptr::from_mut(self)) }; + return Ok(()); } + *value = webcore::body::Value::Blob(match input { + webcore::AnyBlob::Blob(b) => b, + _ => unreachable!(), + }); } - if is_async { - let _ = sink_body_value.to_error_instance(err.dupe(&global), &global); - // TODO: properly propagate exception upwards - } else { - let ret_err = err.to_js(&global); - ret_err.ensure_still_alive(); - ret_err.protect(); - Self::write_tmp_sync_error(sink, ret_err); + let js_stream = value.to_readable_stream(&global)?; + match ReadableStream::from_js(js_stream, &global)? { + Some(stream) => stream, + None => { + self.finish(None); + // SAFETY: see above. + unsafe { Self::deref(core::ptr::from_mut(self)) }; + return Ok(()); + } } - // Do not `end()` the rewriter: that would run `done()`, replacing - // the error just stored on the body with the truncated output. - // `Drop` destroys the rewriter once the sink's refcount hits zero. - return; - } + }; - // SAFETY: `sink` is live (refcount > 0, see fn safety contract). - if let Some(ret_err) = unsafe { Self::run_output_sink(sink, bytes, is_async) } { - ret_err.ensure_still_alive(); - ret_err.protect(); - Self::write_tmp_sync_error(sink, ret_err); + if !matches!(value, webcore::body::Value::Error(_)) { + *value = webcore::body::Value::Used; } + let _ = + ResumableHTMLRewriterSink::init(&global, readable_stream, core::ptr::from_mut(self)); + Ok(()) } - /// Note: takes `*mut Self` (not `&mut self`) because - /// `HtmlRewriter::write/end` re-enter - /// `SinkRef::handle_chunk(&mut self)` through the - /// raw `*mut BufferOutputSink` captured at build time. A `&mut self` - /// receiver here would alias that inner `&mut` (Stacked Borrows UB). - /// - /// # Safety - /// `sink` must be a live `BufferOutputSink` heap allocation with - /// refcount > 0; `(*sink).rewriter` and `(*sink).response` must be set. - unsafe fn run_output_sink(sink: *mut Self, bytes: &[u8], is_async: bool) -> Option { - // SAFETY: sink is a live heap allocation (refcount > 0, caller - // invariant). Read fields into locals before the rewriter calls so no - // borrow of `*sink` is live across the re-entrant output sink. - let (global, response, rewriter) = unsafe { - let _ = (*sink).bytes.grow_by(bytes.len()); // OOM/capacity: fire-and-forget - ((*sink).global, (*sink).response, (*sink).rewriter) - }; - - // SAFETY: rewriter heap-allocated by init(), not yet freed. - if let Err(e) = unsafe { (*rewriter).write(bytes) } { - // Poisoned: never call `end()` after a failed `write()`. The - // field stays non-null so `Drop` frees the rewriter. - if is_async { - // SAFETY: response kept alive by response_value Strong. - let _ = unsafe { (*response).get_body_value() }.to_error_instance( - webcore::body::ValueError::Message(lol_err_string(&e)), - &global, - ); - // TODO: properly propagate exception upwards - return None; - } else { - return Some(create_lolhtml_error(&global, &e)); - } + fn feed(&self, bytes: &[u8]) { + let rewriter = self.rewriter.get(); + if rewriter.is_null() { + return; } + // lol-html parses a first chunk straight from the input slice, so a + // handler that mutates/transfers the user's buffer mid-`write()` would + // corrupt tokens past the current position. + let owned: Vec = bytes.to_vec(); + // SAFETY: rewriter heap-allocated by init(), non-null, not yet freed. + if let Err(e) = unsafe { (*rewriter).write(&owned) } { + self.fail(create_lolhtml_error(&self.global, &e)); + } + } - // `HtmlRewriter::end(self)` consumes the rewriter: null the field - // first so `Drop` does not free it a second time. - // SAFETY: sink is a live heap allocation (refcount > 0). - unsafe { (*sink).rewriter = core::ptr::null_mut() }; + fn finish(&self, err: Option) { + if self.failed.get().has() { + return; + } + if let Some(err) = err { + self.fail(err); + return; + } + let rewriter = self.rewriter.replace(core::ptr::null_mut()); + if rewriter.is_null() { + return; + } // SAFETY: `rewriter` was heap-allocated by init(); sole owner now. if let Err(e) = unsafe { bun_core::heap::take(rewriter) }.end() { - if is_async { - // SAFETY: response kept alive by response_value Strong. - let _ = unsafe { (*response).get_body_value() }.to_error_instance( - webcore::body::ValueError::Message(lol_err_string(&e)), - &global, - ); - // TODO: properly propagate exception upwards - return None; - } else { - return Some(create_lolhtml_error(&global, &e)); - } + self.fail(create_lolhtml_error(&self.global, &e)); } - - None } - pub fn done(&mut self) { - // SAFETY: self.response is kept alive by self.response_value (Strong - // root) for the lifetime of this sink. - let body_value = unsafe { (*self.response).get_body_value() }; - let mut prev_value = core::mem::replace( - body_value, - webcore::body::Value::InternalBlob(webcore::InternalBlob { - bytes: core::mem::replace(&mut self.bytes, MutableString::init_empty()).list, - was_string: false, - }), - ); - - let _ = webcore::body::Value::resolve(&mut prev_value, body_value, &self.global, None); - // TODO: properly propagate exception upwards + fn output_bytes(&self) -> Option> { + self.output.get(&self.global).and_then(|s| s.ptr.bytes()) } - pub fn write(&mut self, bytes: &[u8]) { - let _ = self.bytes.append(bytes); // OOM/capacity: fire-and-forget + fn fail(&self, err: JSValue) { + err.ensure_still_alive(); + let rewriter = self.rewriter.replace(core::ptr::null_mut()); + if !rewriter.is_null() { + // SAFETY: rewriter heap-allocated by init() and not yet freed. + unsafe { bun_core::heap::destroy(rewriter) }; + } + if !self.failed.get().has() { + self.failed + .with_mut(|f| *f = jsc::strong::Optional::create(err, &self.global)); + } + if let Some(bytes) = self.output_bytes() { + let ref_ = jsc::strong::Optional::create(err, &self.global); + let _ = bytes.on_data(streams::Result::Err(streams::StreamError::JSValue(ref_))); + } } } -/// `lol_html::OutputSink` for the rewriter built in [`BufferOutputSink::init`]. -/// Carries a raw `*mut BufferOutputSink` (never a reference) so the rewriter -/// stored on the sink does not self-borrow. -pub struct SinkRef(*mut BufferOutputSink); +/// Writes the rewriter's output to the `ByteStream` rooted by +/// `BufferOutputSink::output`, never back into its owner. +pub struct SinkRef(*mut ByteStream); impl lol_html::OutputSink for SinkRef { fn handle_chunk(&mut self, chunk: &[u8]) { - // SAFETY: `self.0` is the sink that owns this rewriter (refcount > 0 - // inside `run_output_sink`), and no other `&mut *sink` is live — - // `run_output_sink` reads its fields into locals before the call. - let sink = unsafe { &mut *self.0 }; - // lol-html signals end-of-output with a zero-length final chunk. - if chunk.is_empty() { - sink.done(); + // SAFETY: `self.0` is the `NewSource` payload rooted by + // `BufferOutputSink::output`; `on_data` takes `&self`. + let bytes = unsafe { &*self.0 }; + let _ = if chunk.is_empty() { + bytes.on_data(streams::Result::Done) } else { - sink.write(chunk); - } + bytes.on_data(streams::Result::Temporary(bun_ptr::RawSlice::new(chunk))) + }; } } impl Drop for BufferOutputSink { fn drop(&mut self) { - // bytes, body_value_bufferer, context (Rc), response_value (Strong) drop automatically. - if !self.rewriter.is_null() { - // SAFETY: rewriter heap-allocated by init() and not yet freed - // (`run_output_sink` nulls the field before consuming it in `end`). - unsafe { bun_core::heap::destroy(self.rewriter) }; + let rewriter = self.rewriter.get(); + if !rewriter.is_null() { + // SAFETY: heap-allocated by init(), not yet freed. + unsafe { bun_core::heap::destroy(rewriter) }; } + self.output.deinit(); } } - // ──────────────────────── DocumentHandler ──────────────────────────────── pub struct DocumentHandler { @@ -1252,36 +1154,27 @@ where ) { Ok(v) => v, Err(_) => { - // If there's an exception in the scope, capture it for later retrieval if let Some(exc) = scope.exception() { let exc_value = JSValue::from_cell(exc.as_ptr()); - // Store the exception in the VM's unhandled rejection capture - // mechanism if it's available (this is the same mechanism used - // by BufferOutputSink) + // `err_ptr` is a stack Cell owned by `HandlerErrorScope`'s + // caller; that frame is conservatively scanned until + // `create_lolhtml_error` reads it back, so no `protect()`. if let Some(err_ptr) = vm().unhandled_pending_rejection_to_capture { - // SAFETY: VM-owned pointer set by BufferOutputSink::init. + // SAFETY: VM-owned pointer set by `HandlerErrorScope`. unsafe { *err_ptr = exc_value }; - exc_value.protect(); } } - // Clear the exception from the scope to prevent assertion failures scope.clear_exception(); - // Return true to indicate failure to LOLHTML, which will cause the - // write operation to fail and the error handling logic to take over. return true; } }; - // Check if there's an exception that was thrown but not caught by the error union if let Some(exc) = scope.exception() { let exc_value = JSValue::from_cell(exc.as_ptr()); - // Store the exception in the VM's unhandled rejection capture mechanism if let Some(err_ptr) = vm().unhandled_pending_rejection_to_capture { - // SAFETY: VM-owned pointer set by BufferOutputSink::init. + // SAFETY: VM-owned pointer set by `HandlerErrorScope`. unsafe { *err_ptr = exc_value }; - exc_value.protect(); } - // Clear the exception to prevent assertion failures scope.clear_exception(); return true; } diff --git a/src/runtime/error.rs b/src/runtime/error.rs index 19466ef76a57..32b4afc5b329 100644 --- a/src/runtime/error.rs +++ b/src/runtime/error.rs @@ -26,12 +26,6 @@ pub enum Error { SyntaxError, #[error("FmtError")] FmtError, - #[error("StreamAlreadyUsed")] - StreamAlreadyUsed, - #[error("InvalidStream")] - InvalidStream, - #[error("UnsupportedStreamType")] - UnsupportedStreamType, #[error("JSError")] JSError, #[error("ERR_TLS_CERT_ALTNAME_INVALID")] @@ -600,9 +594,6 @@ impl Error { Self::SnapshotInConcurrentGroup => "SnapshotInConcurrentGroup", Self::SyntaxError => "SyntaxError", Self::FmtError => "FmtError", - Self::StreamAlreadyUsed => "StreamAlreadyUsed", - Self::InvalidStream => "InvalidStream", - Self::UnsupportedStreamType => "UnsupportedStreamType", Self::JSError => "JSError", Self::ERR_TLS_CERT_ALTNAME_INVALID => "ERR_TLS_CERT_ALTNAME_INVALID", Self::RequestBodyNotReusable => "RequestBodyNotReusable", diff --git a/src/runtime/webcore.rs b/src/runtime/webcore.rs index 63a73e34a2c2..208f6b6050c6 100644 --- a/src/runtime/webcore.rs +++ b/src/runtime/webcore.rs @@ -38,7 +38,9 @@ pub use s3_stat::S3Stat; // `JSGlobalObject` as a raw pointer (the FFI boundary cannot carry a Rust // lifetime), so the type aliases are lifetime-free and re-exported directly. pub use cookie_map::{CookieMap, CookieMapRef}; -pub use resumable_sink::{ResumableFetchSink, ResumableS3UploadSink, ResumableSinkBackpressure}; +pub use resumable_sink::{ + ResumableFetchSink, ResumableHTMLRewriterSink, ResumableS3UploadSink, ResumableSinkBackpressure, +}; pub use s3_client::S3Client; pub use streams::{ H3ResponseSink, HTTPResponseSink, HTTPSResponseSink, HTTPServerWritable, NetworkSink, diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index ec3d76a55648..55c35516b166 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -5244,6 +5244,38 @@ pub fn write_file_internal( let BodyValue::Locked(locked) = (unsafe { &mut *body_value }) else { unreachable!() }; + // A body backed by a ReadableStream has no producer to + // fire `on_receive_value`; read the stream ourselves. + if let Some(readable) = + get_stream(global_this).or_else(|| locked.readable.get(global_this)) + { + let bytes_promise = bun_jsc::from_js_host_call(global_this, || { + global_this.readable_stream_to_bytes(readable.value) + }); + // SAFETY: re-borrow after `readable_stream_to_bytes`. + *(unsafe { &mut *body_value }) = BodyValue::Used; + // SAFETY: `task` heap-allocated above; sole owner. + let promise = unsafe { (*task).promise.value() }; + match bytes_promise { + Ok(p) if p.as_any_promise().is_some() => p.then( + global_this, + task, + write_file_mod::write_file_locked_on_stream_resolved_shim, + write_file_mod::write_file_locked_on_stream_rejected_shim, + ), + other => { + // SAFETY: `task` heap-allocated above; sole owner. + let task = unsafe { bun_core::heap::take(task) }; + task.file_blob.detach(); + let err = match other { + Err(err) => global_this.take_exception(err), + Ok(v) => v, + }; + task.promise.get().reject(global_this, Ok(err))?; + } + } + return Ok(ControlFlow::Break(promise)); + } locked.task = Some(task.cast::()); locked.on_receive_value = Some(WriteFileWaitFromLockedValueTask::then_wrap); // SAFETY: `task` was just heap-allocated; consumed in `then_wrap`. diff --git a/src/runtime/webcore/Body.rs b/src/runtime/webcore/Body.rs index d2402bc94dd5..40c73ae296d9 100644 --- a/src/runtime/webcore/Body.rs +++ b/src/runtime/webcore/Body.rs @@ -1,6 +1,5 @@ //! https://developer.mozilla.org/en-US/docs/Web/API/Body -use bun_collections::VecExt; use core::ffi::c_void; use core::ptr::NonNull; @@ -18,8 +17,7 @@ use bun_http_types::MimeType::MimeType; use crate::jsc::HTTPHeaderName; pub use crate::webcore::InternalBlob; use crate::webcore::form_data::AsyncFormDataExt as _; -use crate::webcore::sink::{self, ArrayBufferSink}; -use bun_core::{MutableString, String as BunString, ZigString}; +use bun_core::{String as BunString, ZigString}; use bun_core::{WTFStringImpl, WTFStringImplExt as _, WTFStringImplStruct}; use bun_jsc::ZigStringJsc as _; use bun_jsc::{JsCell, StringJsc as _}; @@ -88,7 +86,6 @@ fn as_url_search_params(value: JSValue) -> Option<*mut URLSearchParams> { bun_core::declare_scope!(BodyValue, visible); bun_core::declare_scope!(BodyMixin, visible); -bun_core::declare_scope!(BodyValueBufferer, visible); type JsTerminated = jsc::JsResult; @@ -1560,12 +1557,9 @@ impl Value { } // ──────────────────────────────────────────────────────────────────────────── -// JSC-integration: extract / BodyMixin (host-fn methods) / ValueBufferer. +// JSC-integration: extract / BodyMixin (host-fn methods). // ──────────────────────────────────────────────────────────────────────────── -// `sink::JSSink` is a free generic (inherent associated types are unstable). -type ArrayBufferJSSink = sink::JSSink; - // https://github.com/WebKit/webkit/blob/main/Source/WebCore/Modules/fetch/FetchBody.cpp#L45 pub(crate) fn extract(global_this: &JSGlobalObject, value: JSValue) -> JsResult { let body_value = Value::from_js(global_this, value)?; @@ -2151,425 +2145,3 @@ fn handle_body_error(value: &mut Value, global_object: &JSGlobalObject) -> Optio }; Some(JSPromise::rejected_promise(global_object, err.to_js(global_object)).to_js()) } - -// ──────────────────────────────────────────────────────────────────────────── -// ValueBufferer -// ──────────────────────────────────────────────────────────────────────────── - -pub(crate) type ValueBuffererCallback = - fn(ctx: *mut c_void, bytes: &[u8], err: Option, is_async: bool); - -pub struct ValueBufferer<'a> { - pub ctx: *mut c_void, - pub on_finished_buffering: ValueBuffererCallback, - - pub js_sink: Option>, - pub byte_stream: Option>, - // readable stream strong ref to keep byte stream alive - pub readable_stream_ref: webcore::readable_stream::Strong, - pub stream_buffer: MutableString, - // allocator dropped — global mimalloc - pub global: &'a JSGlobalObject, -} - -impl<'a> Drop for ValueBufferer<'a> { - fn drop(&mut self) { - // stream_buffer dropped automatically - if let Some(byte_stream) = self.byte_stream { - // Kept alive by `readable_stream_ref` while set — satisfies the - // `BackRef` outlives-holder invariant. R-2: `unpipe_without_deref` - // takes `&self` (interior-mutable). - bun_ptr::BackRef::from(byte_stream).unpipe_without_deref(); - } - self.readable_stream_ref.deinit(); - - if let Some(mut buffer_stream) = self.js_sink.take() { - buffer_stream.detach_self(self.global); - // The wrapper is a `Box>`; dropping it - // frees the box and runs `Vec`'s Drop. - drop(buffer_stream); - } - } -} - -impl<'a> ValueBufferer<'a> { - pub(crate) fn init( - ctx: *mut c_void, - on_finish: ValueBuffererCallback, - global: &'a JSGlobalObject, - ) -> Self { - Self { - ctx, - on_finished_buffering: on_finish, - js_sink: None, - byte_stream: None, - readable_stream_ref: Default::default(), - global, - stream_buffer: MutableString::default(), - } - } - - pub(crate) fn run( - &mut self, - value: &mut Value, - owned_readable_stream: Option, - ) -> crate::Result<()> { - value.to_blob_if_possible(); - - match value { - Value::Used => { - bun_core::scoped_log!(BodyValueBufferer, "Used"); - return Err(crate::Error::StreamAlreadyUsed); - } - Value::Empty | Value::Null => { - bun_core::scoped_log!(BodyValueBufferer, "Empty"); - (self.on_finished_buffering)(self.ctx, b"", None, false); - return Ok(()); - } - Value::Error(err) => { - bun_core::scoped_log!(BodyValueBufferer, "Error"); - // The payload (BunString / Strong) owns refs and has Drop, so a `ptr::read` - // bitwise copy would manufacture a second owner → double-deref when both - // sides drop. Produce a properly ref-bumped duplicate instead. - let err_copy = err.dupe(self.global); - (self.on_finished_buffering)(self.ctx, b"", Some(err_copy), false); - return Ok(()); - } - // Value::InlineBlob(_) | - Value::WTFStringImpl(_) | Value::InternalBlob(_) | Value::Blob(_) => { - // toBlobIfPossible checks for WTFString needing a conversion. - let mut input = value.use_as_any_blob_allow_non_utf8_string(); - let is_pending = input.needs_to_read_file(); - - if is_pending { - if let AnyBlob::Blob(blob) = &mut input { - // The ZST `InternalReadFileFn` impl lets `do_read_file_internal` - // monomorphize a `fn(*mut c_void, ReadFileResultType)` thunk. - struct LoadFileAdapter; - impl<'b> blob::InternalReadFileFn> for LoadFileAdapter { - fn call( - sink: *mut ValueBufferer<'b>, - bytes: blob::read_file::ReadFileResultType, - ) { - // SAFETY: `sink` was set from `self as *mut Self` below and - // outlives the read (ValueBufferer is heap-pinned by caller). - unsafe { &mut *sink }.on_finished_loading_file(bytes); - } - } - let global = self.global; - blob.do_read_file_internal::( - std::ptr::from_mut::(self), - global, - ); - } - } else { - let bytes = input.slice(); - bun_core::scoped_log!(BodyValueBufferer, "Blob {}", bytes.len()); - (self.on_finished_buffering)(self.ctx, bytes, None, false); - input.detach(); - } - return Ok(()); - } - Value::Locked(_) => { - self.buffer_locked_body_value(value, owned_readable_stream)?; - } - } - Ok(()) - } - - fn on_finished_loading_file(&mut self, bytes: blob::read_file::ReadFileResultType) { - match bytes { - blob::read_file::ReadFileResultType::Err(err) => { - bun_core::scoped_log!(BodyValueBufferer, "onFinishedLoadingFile Error"); - (self.on_finished_buffering)( - self.ctx, - b"", - Some(ValueError::SystemError(err)), - true, - ); - } - blob::read_file::ReadFileResultType::Result(data) => { - // SAFETY: every producer sets `buf = heap::alloc(v.into_boxed_slice())` - // (read_file.rs); reclaim ownership here. Dropped at end of scope. - let buf = unsafe { Box::<[u8]>::from_raw(data.buf) }; - bun_core::scoped_log!( - BodyValueBufferer, - "onFinishedLoadingFile Data {}", - buf.len() - ); - (self.on_finished_buffering)(self.ctx, &buf, None, true); - } - } - } - - fn on_stream_pipe(&mut self, stream: &streams::Result) { - if let streams::Result::Err(err) = stream { - bun_core::scoped_log!(BodyValueBufferer, "onStreamPipe error"); - let js_err = err.to_js(self.global); - let ref_ = jsc::strong::Optional::create(js_err, self.global); - (self.on_finished_buffering)(self.ctx, b"", Some(ValueError::JSValue(ref_)), true); - return; - } - let chunk = stream.slice(); - bun_core::scoped_log!(BodyValueBufferer, "onStreamPipe chunk {}", chunk.len()); - let _ = self.stream_buffer.write(chunk); - if stream.is_done() { - let bytes = self.stream_buffer.list.as_slice(); - bun_core::scoped_log!(BodyValueBufferer, "onStreamPipe done {}", bytes.len()); - (self.on_finished_buffering)(self.ctx, bytes, None, true); - } - } - - /// Reclaim the `*mut Self` smuggled through a `NativePromiseContext` cell - /// as an exclusive borrow. Centralises the `Option>` deref - /// for the two host-fn entry points below (one accessor, N safe callers). - /// - /// # Safety (encapsulated) - /// `NativePromiseContext::take` returns the live ctx pointer set in - /// `create()` (caller stashed `&mut Self` and held a +1 ref); the cell is - /// nulled on take so this is the sole owner. `ValueBufferer` is heap- - /// pinned by its caller for the stream's duration. - #[inline] - fn take_ctx<'r>(cell: JSValue) -> Option<&'r mut Self> { - // SAFETY: see fn doc — +1 ref transferred back; sole live `&mut`. - crate::api::NativePromiseContext::take::(cell).map(|mut p| unsafe { p.as_mut() }) - } - - pub(crate) fn on_resolve_stream( - _global: &JSGlobalObject, - callframe: &CallFrame, - ) -> JsResult { - let args = callframe.arguments(); - let Some(sink) = Self::take_ctx(args[args.len() - 1]) else { - return Ok(JSValue::UNDEFINED); - }; - sink.handle_resolve_stream(true); - Ok(JSValue::UNDEFINED) - } - - pub(crate) fn on_reject_stream( - _global: &JSGlobalObject, - callframe: &CallFrame, - ) -> JsResult { - let args = callframe.arguments(); - let Some(sink) = Self::take_ctx(args[args.len() - 1]) else { - return Ok(JSValue::UNDEFINED); - }; - let err = args[0]; - sink.handle_reject_stream(err, true); - Ok(JSValue::UNDEFINED) - } - - fn handle_reject_stream(&mut self, err: JSValue, is_async: bool) { - if let Some(mut wrapper) = self.js_sink.take() { - wrapper.detach_self(self.global); - // see `Drop` impl — dropping the Box frees the wrapper - // and runs `Vec`'s Drop. - drop(wrapper); - } - // `jsc::strong::Optional` owns a GC root; `ptr::read`-duplicating it would - // double-deinit. Transfer the single owner directly to the callback; the callback - // (or its returned `ValueError`'s Drop) is responsible for releasing it. - let ref_ = jsc::strong::Optional::create(err, self.global); - (self.on_finished_buffering)(self.ctx, b"", Some(ValueError::JSValue(ref_)), is_async); - } - - fn handle_resolve_stream(&mut self, is_async: bool) { - if let Some(wrapper) = &self.js_sink { - let bytes = wrapper.sink.bytes.slice(); - bun_core::scoped_log!(BodyValueBufferer, "handleResolveStream {}", bytes.len()); - (self.on_finished_buffering)(self.ctx, bytes, None, is_async); - } else { - bun_core::scoped_log!(BodyValueBufferer, "handleResolveStream no sink"); - (self.on_finished_buffering)(self.ctx, b"", None, is_async); - } - } - - fn buffer_locked_body_value( - &mut self, - value: &mut Value, - owned_readable_stream: Option, - ) -> crate::Result<()> { - debug_assert!(matches!(value, Value::Locked(_))); - let Value::Locked(locked) = value else { - unreachable!() - }; - let readable_stream = 'brk: { - if let Some(stream) = locked.readable.get(self.global) { - // keep the stream alive until we're done with it. - // Transfer ownership: `*value = .Used` below would otherwise - // drop `locked.readable` anyway, so moving the existing GC - // root preserves the refcount balance. - self.readable_stream_ref = core::mem::take(&mut locked.readable); - break 'brk Some(stream); - } - if let Some(stream) = owned_readable_stream { - // response owns the stream, so we hold a strong reference to it - self.readable_stream_ref = - webcore::readable_stream::Strong::init(stream, self.global); - break 'brk Some(stream); - } - None - }; - if let Some(stream) = readable_stream { - *value = Value::Used; - - if stream.is_locked(self.global) { - return Err(crate::Error::StreamAlreadyUsed); - } - - match stream.ptr { - webcore::readable_stream::Source::Invalid => { - return Err(crate::Error::InvalidStream); - } - // toBlobIfPossible should've caught this - webcore::readable_stream::Source::Blob(_) - | webcore::readable_stream::Source::File(_) => unreachable!(), - webcore::readable_stream::Source::JavaScript - | webcore::readable_stream::Source::Direct => { - // this is broken right now - // return self.create_js_sink(stream); - return Err(crate::Error::UnsupportedStreamType); - } - webcore::readable_stream::Source::Bytes(byte_stream_ptr) => { - // BACKREF: see `Source::bytes()` — payload owned by the - // readable stream, kept alive via `self.readable_stream_ref` - // above. R-2: all touched fields are interior-mutable. - let byte_stream = stream.ptr.bytes().expect("matched Bytes"); - debug_assert!(byte_stream.pipe.get().ctx.is_none()); - debug_assert!(self.byte_stream.is_none()); - - let bytes = byte_stream.buffer.get().as_slice(); - // If we've received the complete body by the time this function is called - // we can avoid streaming it and just send it all at once. - if byte_stream.has_received_last_chunk.get() { - if let streams::Result::Err(err) = &byte_stream.pending.get().result { - bun_core::scoped_log!( - BodyValueBufferer, - "byte stream has_received_last_chunk error" - ); - let js_err = err.to_js(self.global); - let ref_ = jsc::strong::Optional::create(js_err, self.global); - (self.on_finished_buffering)( - self.ctx, - b"", - Some(ValueError::JSValue(ref_)), - false, - ); - stream.done(self.global); - return Ok(()); - } - bun_core::scoped_log!( - BodyValueBufferer, - "byte stream has_received_last_chunk {}", - bytes.len() - ); - (self.on_finished_buffering)(self.ctx, bytes, None, false); - // is safe to detach here because we're not going to receive any more data - stream.done(self.global); - return Ok(()); - } - - byte_stream - .pipe - .set(crate::webcore::Wrap::::init(self)); - self.byte_stream = NonNull::new(byte_stream_ptr); - bun_core::scoped_log!( - BodyValueBufferer, - "byte stream pre-buffered {}", - bytes.len() - ); - - let _ = self.stream_buffer.write(bytes); - return Ok(()); - } - } - } - - // reshaped for borrowck — re-borrow locked after possible *value = Used above. - let Value::Locked(locked) = value else { - unreachable!() - }; - - if locked.on_receive_value.is_some() || locked.task.is_some() { - // ValueBufferer wants the whole body; tell the producer to never - // pause for JS backpressure before the stream is materialised. - if let (Some(on_start_buffering), Some(task)) = - (locked.on_start_buffering.take(), locked.task) - { - on_start_buffering(task); - } - // someone else is waiting for the stream or waiting for `onStartStreaming` - let readable = value - .to_readable_stream(self.global) - .map_err(|_| crate::Error::JSError)?; - // The JS exception value is - // flattened to a string-coded error because `run`'s callers consume - // `crate::Error` (the exception itself stays pending on the VM). - readable.ensure_still_alive(); - readable.protect(); - return self.buffer_locked_body_value(value, None); - } - // is safe to wait it buffer - locked.task = Some(std::ptr::from_mut::(self).cast::()); - locked.on_receive_value = Some(Self::on_receive_value); - Ok(()) - } - - fn on_receive_value(ctx: *mut c_void, value: &mut Value) { - // SAFETY: ctx was set from `self as *mut Self` in buffer_locked_body_value. - let sink = unsafe { bun_ptr::callback_ctx::(ctx) }; - match value { - Value::Error(err) => { - bun_core::scoped_log!(BodyValueBufferer, "onReceiveValue Error"); - // See run(): produce a ref-bumped duplicate instead of `ptr::read`ing a - // non-Copy owned value (would double-deref on drop). - let err_copy = err.dupe(sink.global); - (sink.on_finished_buffering)(sink.ctx, b"", Some(err_copy), true); - } - _ => { - value.to_blob_if_possible(); - let input = value.use_as_any_blob_allow_non_utf8_string(); - let bytes = input.slice(); - bun_core::scoped_log!(BodyValueBufferer, "onReceiveValue {}", bytes.len()); - (sink.on_finished_buffering)(sink.ctx, bytes, None, true); - } - } - } -} - -// `webcore::Wrap` requires `T: PipeHandler`. -impl<'a> crate::webcore::PipeHandler for ValueBufferer<'a> { - fn on_pipe(&mut self, stream: streams::Result) { - self.on_stream_pipe(&stream) - } -} - -// `#[bun_jsc::host_fn]` on on_resolve_stream/on_reject_stream emits the JSC ABI shim; -// these no_mangle re-exports point at those shims under the C names the C++ side expects. -bun_jsc::jsc_host_abi! { - #[unsafe(no_mangle)] - pub(crate) unsafe fn Bun__BodyValueBufferer__onResolveStream( - global: *mut JSGlobalObject, - callframe: *mut CallFrame, - ) -> JSValue { - // S008: `JSGlobalObject`/`CallFrame` are `opaque_ffi!` ZST handles — - // safe `*mut → &` via `opaque_deref` (JSC guarantees non-null/live). - let (global, callframe) = - (bun_opaque::opaque_deref(global), bun_opaque::opaque_deref(callframe)); - jsc::to_js_host_fn_result(global, ValueBufferer::on_resolve_stream(global, callframe)) - } -} -bun_jsc::jsc_host_abi! { - #[unsafe(no_mangle)] - pub(crate) unsafe fn Bun__BodyValueBufferer__onRejectStream( - global: *mut JSGlobalObject, - callframe: *mut CallFrame, - ) -> JSValue { - // S008: `JSGlobalObject`/`CallFrame` are `opaque_ffi!` ZST handles — - // safe `*mut → &` via `opaque_deref` (JSC guarantees non-null/live). - let (global, callframe) = - (bun_opaque::opaque_deref(global), bun_opaque::opaque_deref(callframe)); - jsc::to_js_host_fn_result(global, ValueBufferer::on_reject_stream(global, callframe)) - } -} diff --git a/src/runtime/webcore/ResumableSink.rs b/src/runtime/webcore/ResumableSink.rs index 7235f7607d20..ac60a7c350cd 100644 --- a/src/runtime/webcore/ResumableSink.rs +++ b/src/runtime/webcore/ResumableSink.rs @@ -11,6 +11,7 @@ use bun_core::String as BunString; use bun_jsc::{CallFrame, JSGlobalObject, JSValue, JsRef, JsResult, SystemError}; use bun_output::{declare_scope, scoped_log}; +use crate::api::html_rewriter::BufferOutputSink; use crate::node::{ErrorCode, StringOrBuffer}; use crate::webcore::fetch::fetch_tasklet::FetchTasklet; use crate::webcore::s3::client::S3UploadStreamWrapper; @@ -37,11 +38,14 @@ pub trait ResumableSinkJs { } /// Trait capturing the per-`Context` callbacks the sink invokes. -// The only -// in-tree impls (FetchTasklet / S3UploadStreamWrapper) mutate self in both -// callbacks (e.g. `detachSink`, `deref`, clearing `endPromise`), so these -// MUST be `&mut self`. pub trait ResumableSinkContext { + /// Skip the `Source::Bytes` native-pipe fast path and always drive the + /// stream through the JS pump. Set by contexts whose `write_request_data` + /// nests the event loop (HTMLRewriter's `wait_for_promise`), which could + /// re-enter the native pipe callback; the JS pump's `m_reading` guard and + /// read-after-write sequencing prevent that. + const AVOID_NATIVE_PIPE: bool = false; + fn write_request_data(&mut self, bytes: &[u8]) -> ResumableSinkBackpressure; fn write_end_request(&mut self, err: Option); } @@ -121,9 +125,8 @@ impl ResumableSink ResumableSinkBackpressure { - // SAFETY: `context` is a BACKREF to the owning Context (FetchTasklet / - // S3UploadStreamWrapper) which outlives this sink — see LIFETIMES.tsv. - // Dereferenced as `&mut` because impls mutate (detachSink, deref, etc.). + // SAFETY: `context` is a BACKREF to the owning Context which outlives + // this sink. Dereferenced as `&mut` because impls mutate. unsafe { (*ctx).write_request_data(bytes) } } #[inline] @@ -181,7 +184,9 @@ impl ResumableSink ResumableSink {} + ResumableSinkBackpressure::Done => { + this.cancel(JSValue::UNDEFINED); + return Ok(JSValue::FALSE); + } ResumableSinkBackpressure::WantMore => { this.status = Status::Started; } @@ -620,7 +628,11 @@ macro_rules! impl_resumable_sink_js { } )*}; } -impl_resumable_sink_js!(JSResumableFetchSink, JSResumableS3UploadSink); +impl_resumable_sink_js!( + JSResumableFetchSink, + JSResumableS3UploadSink, + JSResumableHTMLRewriterSink, +); // Forward to the inherent methods on each Context type; the trait bound is // satisfied by delegating to those inherent impls. @@ -638,6 +650,7 @@ impl ResumableSinkContext for FetchTasklet { pub type ResumableFetchSink = ResumableSink; pub type ResumableS3UploadSink = ResumableSink; +pub type ResumableHTMLRewriterSink = ResumableSink; unsafe extern "C" { safe fn Bun__assignStreamIntoResumableSink( diff --git a/src/runtime/webcore/Sink.rs b/src/runtime/webcore/Sink.rs index 2d33b1e0b7f6..9e9ce866293c 100644 --- a/src/runtime/webcore/Sink.rs +++ b/src/runtime/webcore/Sink.rs @@ -12,19 +12,6 @@ pub use crate::webcore::array_buffer_sink::ArrayBufferSink; crate::impl_js_sink_abi!(ArrayBufferSink, "ArrayBufferSink"); -impl JSSink { - /// Unprotects the controller cell stashed in `signal.ptr` - /// and tells C++ to drop its back-pointer. Called from - /// `Body::ValueBufferer` Drop / reject paths. - // Renamed from `detach` to avoid colliding with the generic - // `JSSink::detach(signal, global)` associated fn — Rust - // forbids same-name items across impl blocks for the same type even with - // different signatures (E0592). - pub fn detach_self(&mut self, global: &JSGlobalObject) { - JSSink::::detach(&mut self.sink.signal, global); - } -} - // ────────────────────────────────────────────────────────────────────────── // JSSink // diff --git a/src/runtime/webcore/blob/write_file.rs b/src/runtime/webcore/blob/write_file.rs index 1f6795af3729..3ff2b75cad14 100644 --- a/src/runtime/webcore/blob/write_file.rs +++ b/src/runtime/webcore/blob/write_file.rs @@ -9,7 +9,9 @@ use bun_core::ZigString; use bun_io::{self as io, IntrusiveIoRequest as _}; use bun_jsc::ZigStringJsc as _; use bun_jsc::node_path::PathOrFileDescriptor; -use bun_jsc::{self as jsc, JSGlobalObject, JSPromise, JSValue, JsTerminated, SystemError}; +use bun_jsc::{ + self as jsc, CallFrame, JSGlobalObject, JSPromise, JSValue, JsTerminated, SystemError, +}; use bun_sys::{self as sys, Fd}; use bun_threading::{IntrusiveWorkTask as _, WorkPool, WorkPoolTask}; @@ -1326,6 +1328,31 @@ impl WriteFileWaitFromLockedValueTask { // TODO: properly propagate exception upwards } + /// `.then` reaction for a `Locked` body that already has a readable: the + /// body was read via `readableStreamToBytes`; wrap the resolved bytes as + /// an `InternalBlob` and hand them to [`then`](Self::then). + fn on_stream_resolved(global: &JSGlobalObject, callframe: &jsc::CallFrame) -> JSValue { + let this = callframe.argument(1).as_promise_ptr::(); + let mut value = match callframe.argument(0).as_array_buffer(global) { + Some(buf) => body::Value::InternalBlob(body::InternalBlob { + bytes: buf.slice().to_vec(), + was_string: false, + }), + None => body::Value::Empty, + }; + let _ = Self::then(NonNull::new(this).unwrap(), &mut value); + JSValue::UNDEFINED + } + + fn on_stream_rejected(global: &JSGlobalObject, callframe: &jsc::CallFrame) -> JSValue { + let this = callframe.argument(1).as_promise_ptr::(); + let mut value = body::Value::Error(body::ValueError::JSValue( + jsc::strong::Optional::create(callframe.argument(0), global), + )); + let _ = Self::then(NonNull::new(this).unwrap(), &mut value); + JSValue::UNDEFINED + } + /// # Safety /// `this` must point to a live Box-allocated `WriteFileWaitFromLockedValueTask`. /// On every arm except `body::Value::Locked`, the allocation is consumed. @@ -1427,3 +1454,26 @@ impl WriteFileWaitFromLockedValueTask { Ok(()) } } + +bun_jsc::jsc_host_abi! { + #[unsafe(export_name = "Bun__WriteFileLocked__onStreamResolved")] + pub(crate) unsafe fn write_file_locked_on_stream_resolved_shim( + global: *mut JSGlobalObject, + callframe: *mut CallFrame, + ) -> JSValue { + let (global, callframe) = + (bun_opaque::opaque_deref(global), bun_opaque::opaque_deref(callframe)); + WriteFileWaitFromLockedValueTask::on_stream_resolved(global, callframe) + } +} +bun_jsc::jsc_host_abi! { + #[unsafe(export_name = "Bun__WriteFileLocked__onStreamRejected")] + pub(crate) unsafe fn write_file_locked_on_stream_rejected_shim( + global: *mut JSGlobalObject, + callframe: *mut CallFrame, + ) -> JSValue { + let (global, callframe) = + (bun_opaque::opaque_deref(global), bun_opaque::opaque_deref(callframe)); + WriteFileWaitFromLockedValueTask::on_stream_rejected(global, callframe) + } +} diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js index 6ffc837062ed..fbad2fe7653d 100644 --- a/test/js/workerd/html-rewriter.test.js +++ b/test/js/workerd/html-rewriter.test.js @@ -1,3 +1,4 @@ +import { heapStats } from "bun:jsc"; import { afterAll, beforeAll, describe, expect, it } from "bun:test"; import { once } from "events"; import fs from "fs"; @@ -180,9 +181,6 @@ describe("HTMLRewriter", () => { // Must reject with the upstream connection error, and must never // resolve with the truncated document. expect(await text).toEqual(rejectedWithConnectionError); - // The body is now in its error state. A second read must report the - // same failure, not resolve as an empty "successful" document. - expect(await settle(transformed.text())).toEqual(rejectedWithConnectionError); }); }); @@ -196,16 +194,22 @@ describe("HTMLRewriter", () => { }); it(".body on the transformed response is an errored stream", async () => { + // The rewrite streams, so bytes that arrive before the failure are + // delivered; read to completion and assert the stream ends in an error + // instead of closing cleanly as a truncated "successful" document. + async function readAll(reader) { + while (true) { + const r = await settle(reader.read()); + if (r.rejected) return r; + if (r.value.done) return r; + } + } await withPartialBodyServer(async (url, release) => { const res = await fetch(url); const transformed = rewriter().transform(res); - const text = settle(transformed.text()); + const reader = transformed.body.getReader(); release(); - // Barrier: once this has rejected, the body is in its error state. - expect(await text).toEqual(rejectedWithConnectionError); - // Reading `.body` must reject with the same upstream error instead of - // closing cleanly as an empty "successful" document. - expect(await settle(transformed.body.getReader().read())).toEqual(rejectedWithConnectionError); + expect(await readAll(reader)).toEqual(rejectedWithConnectionError); }); }); @@ -213,12 +217,15 @@ describe("HTMLRewriter", () => { await withPartialBodyServer(async (url, release) => { const res = await fetch(url); const transformed = rewriter().transform(res); - // Start the read BEFORE the upstream fails. This is the one shape - // (readable attached, no pending promise) where the error must reach - // the attached stream; discarding it would strand this read forever. - const read = settle(transformed.body.getReader().read()); + const reader = transformed.body.getReader(); + // Start the read BEFORE the upstream fails so it is pending when the + // error arrives. The first read may resolve with the chunk that was + // rewritten before the failure; the stream must eventually reject. + let read = settle(reader.read()); release(); - expect(await read).toEqual(rejectedWithConnectionError); + let r = await read; + while (!r.rejected && !r.value.done) r = await settle(reader.read()); + expect(r).toEqual(rejectedWithConnectionError); }); }); @@ -226,13 +233,10 @@ describe("HTMLRewriter", () => { await withPartialBodyServer(async (url, release) => { const res = await fetch(url); const transformed = rewriter().transform(res); - const text = settle(transformed.text()); + const clone = transformed.clone(); release(); - // Barrier: the body is now in its error state. - expect(await text).toEqual(rejectedWithConnectionError); - // Cloning a failed body must produce a failed body, not an empty one - // that reads back as a complete (and empty) document. - expect(await settle(transformed.clone().text())).toEqual(rejectedWithConnectionError); + expect(await settle(transformed.text())).toEqual(rejectedWithConnectionError); + expect(await settle(clone.text())).toEqual(rejectedWithConnectionError); }); }); @@ -303,6 +307,349 @@ describe("HTMLRewriter", () => { }); }); + describe("transform() accepts a JavaScript-backed ReadableStream body", () => { + // https://github.com/oven-sh/bun/issues/14216 + // https://github.com/oven-sh/bun/issues/11758 + const encode = s => new TextEncoder().encode(s); + + function rewriter() { + return new HTMLRewriter().on("p", { + element(element) { + element.setInnerContent("bye"); + }, + }); + } + + function streamOf(...chunks) { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + controller.close(); + }, + }); + } + + it("single Uint8Array chunk", async () => { + const transformed = rewriter().transform(new Response(streamOf(encode("

hi

")))); + expect(await transformed.text()).toBe("

bye

"); + }); + + it("single string chunk", async () => { + const transformed = rewriter().transform(new Response(streamOf("

hi

"))); + expect(await transformed.text()).toBe("

bye

"); + }); + + it("an element split across chunk boundaries", async () => { + const transformed = rewriter().transform( + new Response(streamOf(encode("

h"), encode("i

two

"))), + ); + expect(await transformed.text()).toBe("

bye

bye

"); + }); + + it("mixed string and binary chunks", async () => { + const transformed = rewriter().transform(new Response(streamOf("

a

", encode("

b

")))); + expect(await transformed.text()).toBe("

bye

bye

"); + }); + + it("empty stream", async () => { + let endCalls = 0; + const transformed = new HTMLRewriter() + .onDocument({ + end() { + endCalls++; + }, + }) + .transform(new Response(streamOf())); + expect(await transformed.text()).toBe(""); + expect(endCalls).toBe(1); + }); + + it("a direct stream", async () => { + const body = new ReadableStream({ + type: "direct", + pull(controller) { + controller.write("

hi

"); + controller.close(); + }, + }); + expect(await rewriter().transform(new Response(body)).text()).toBe("

bye

"); + }); + + it("a stream that only produces chunks after transform() returns", async () => { + // start() stays pending across transform(), so the rewriter has to take + // the asynchronous path instead of buffering everything up front. + const { promise: gate, resolve: openGate } = Promise.withResolvers(); + const body = new ReadableStream({ + async start(controller) { + await gate; + controller.enqueue(encode("

hi

")); + controller.close(); + }, + }); + const text = rewriter().transform(new Response(body)).text(); + openGate(); + expect(await text).toBe("

bye

"); + }); + + it("every way of reading the transformed response", async () => { + const read = { + text: response => response.text(), + arrayBuffer: async response => new TextDecoder().decode(await response.arrayBuffer()), + bytes: async response => new TextDecoder().decode(await response.bytes()), + blob: response => response.blob().then(blob => blob.text()), + json: response => response.json().then(value => JSON.stringify(value)), + getReader: async response => { + const reader = response.body.getReader(); + const parts = []; + for (let chunk = await reader.read(); !chunk.done; chunk = await reader.read()) { + parts.push(new TextDecoder().decode(chunk.value)); + } + return parts.join(""); + }, + readableStreamToText: response => Bun.readableStreamToText(response.body), + }; + + const html = '

hi

there

'; + const expected = '

bye

bye

'; + for (const [name, consume] of Object.entries(read)) { + const transformed = rewriter().transform(new Response(streamOf(encode(html)))); + if (name === "json") { + // Not valid JSON, but it must fail as a JSON parse error, which + // still proves the transformed bytes reached the parser. + await expect(consume(transformed)).rejects.toThrow(/JSON/i); + continue; + } + expect({ [name]: await consume(transformed) }).toEqual({ [name]: expected }); + } + }); + + it("element handlers observe the streamed document", async () => { + const tags = []; + const transformed = new HTMLRewriter() + .on("*", { + element(element) { + tags.push(element.tagName); + }, + }) + .transform(new Response(streamOf(encode("

hi

")))); + expect(await transformed.text()).toBe("

hi

"); + expect(tags).toEqual(["div", "p"]); + }); + + it("a stream that errors rejects the transformed body", async () => { + const body = new ReadableStream({ + start(controller) { + controller.enqueue(encode("

hi

")); + controller.error(new Error("upstream boom")); + }, + }); + const transformed = rewriter().transform(new Response(body)); + // Must reject rather than resolve with the truncated document. + await expect(transformed.text()).rejects.toThrow("upstream boom"); + }); + + it("a stream that errors after transform() returns rejects the transformed body", async () => { + const { promise: gate, resolve: openGate } = Promise.withResolvers(); + const body = new ReadableStream({ + async start(controller) { + await gate; + controller.error(new Error("late boom")); + }, + }); + const text = rewriter().transform(new Response(body)).text(); + openGate(); + await expect(text).rejects.toThrow("late boom"); + }); + + it("a chunk that is neither a string nor a view rejects the transformed body", async () => { + const transformed = rewriter().transform(new Response(streamOf(42))); + // The underlying TypeError must surface, not the opaque + // "Failed to pipe stream" that transform() used to throw. + await expect(transformed.text()).rejects.toThrow(TypeError); + }); + + it("cancels the source stream once a handler throws", async () => { + let pulls = 0; + let cancelled = false; + const body = new ReadableStream({ + pull(c) { + pulls++; + c.enqueue(encode("

x

")); + }, + cancel() { + cancelled = true; + }, + }); + const rw = new HTMLRewriter().on("p", { + element() { + throw new Error("boom"); + }, + }); + await expect(rw.transform(new Response(body)).text()).rejects.toThrow("boom"); + // The pump must stop instead of reading the never-closing source + // forever; a couple of extra pulls queued before cancel lands is fine. + expect(pulls).toBeLessThan(5); + expect(cancelled).toBe(true); + }); + + it("does not leak a handler's thrown error", async () => { + const once = async () => { + const rw = new HTMLRewriter().on("p", { + element() { + throw new Error("boom"); + }, + }); + await rw + .transform(new Response(streamOf(encode("

x

")))) + .text() + .catch(() => {}); + }; + const settle = async () => { + for (let i = 0; i < 3; i++) { + Bun.gc(true); + await Bun.sleep(1); + } + }; + for (let i = 0; i < 10; i++) await once(); + await settle(); + const before = heapStats().objectTypeCounts.Error ?? 0; + for (let i = 0; i < 200; i++) await once(); + await settle(); + // Pre-fix: the Exception cell was gcProtect()ed in handler_callback and + // never unprotected, pinning one Error per transform (grew by ~400). + expect((heapStats().objectTypeCounts.Error ?? 0) - before).toBeLessThan(20); + }); + + it("reusing the transformed response's source stream throws", async () => { + const response = new Response(streamOf(encode("

hi

"))); + expect(await rewriter().transform(response).text()).toBe("

bye

"); + expect(() => rewriter().transform(response)).toThrow("Response body already used"); + }); + + it("does not rewrite out of the source buffer a handler can detach", async () => { + // ResumableSink copies each chunk into the sink's own input buffer before + // returning to the pump, so a handler that mutates (or transfers, then + // frees) the user's buffer mid-scan must not corrupt bytes lol-html has + // yet to tokenize. + let chunk; + const body = new ReadableStream({ + start(controller) { + chunk = encode("xy"); + controller.enqueue(chunk); + controller.close(); + }, + }); + const transformed = new HTMLRewriter() + .on("a", { + element() { + // overwrite "" (not yet tokenized) with "" + chunk.set(encode("qqq"), 9); + // and drop the backing store the rewriter would be reading + chunk.buffer.transfer(); + Bun.gc(true); + }, + }) + .transform(new Response(body)); + expect(await transformed.text()).toBe("xy"); + }); + + // A live transform is kept alive only by whatever can still settle the + // stream. That holds because settling needs the controller, and the + // controller holds the stream. Each case hides the stream from userland and + // collects hard before letting it finish. + describe("a source the bufferer no longer roots still completes", () => { + const cases = { + "controller held only by a timer": () => + new ReadableStream({ + start(controller) { + setTimeout(() => { + controller.enqueue(encode("

hi

")); + controller.close(); + }, 1); + }, + }), + "controller escaping to an outer scope": () => { + let escaped; + const stream = new ReadableStream({ + start(controller) { + escaped = controller; + }, + }); + queueMicrotask(() => { + escaped.enqueue(encode("

hi

")); + escaped.close(); + }); + return stream; + }, + "controller reachable only from a pending pull": () => + new ReadableStream({ + type: "direct", + async pull(controller) { + await Bun.sleep(1); + controller.write("

hi

"); + controller.close(); + }, + }), + }; + + for (const [name, makeStream] of Object.entries(cases)) { + it(name, async () => { + const transformed = rewriter().transform(new Response(makeStream())); + // Collect aggressively while the source is still in flight. + for (let i = 0; i < 3; i++) { + Bun.gc(true); + await Bun.sleep(1); + } + expect(await transformed.text()).toBe("

bye

"); + }); + } + }); + + it(".body of a transform whose source is still pending", async () => { + const { promise: gate, resolve: openGate } = Promise.withResolvers(); + const body = new ReadableStream({ + async start(controller) { + await gate; + controller.enqueue(encode("

hi

")); + controller.close(); + }, + }); + const transformed = rewriter().transform(new Response(body)); + const reader = transformed.body.getReader(); + openGate(); + const parts = []; + for (let chunk = await reader.read(); !chunk.done; chunk = await reader.read()) { + parts.push(new TextDecoder().decode(chunk.value)); + } + expect(parts.join("")).toBe("

bye

"); + }); + + it("served over Bun.serve", async () => { + using server = Bun.serve({ + port: 0, + fetch() { + const body = new ReadableStream({ + start(controller) { + controller.enqueue(encode("hello world")); + controller.close(); + }, + }); + return new HTMLRewriter() + .on("b", { + element(element) { + element.before("

", { html: true }); + element.after("

", { html: true }); + element.removeAndKeepContent(); + }, + }) + .transform(new Response(body, { headers: { "content-type": "text/html" } })); + }, + }); + const response = await fetch(server.url); + expect(await response.text()).toBe("

hello world

"); + }); + }); + it("HTMLRewriter: async replacement using fetch + Bun.serve", async () => { await gcTick(); let content; @@ -946,12 +1293,12 @@ const payloads = [ { name: "direct", data: getStream("direct", "none"), - test: it.todo, + test: it, }, { name: "default", data: getStream("default", "none"), - test: it.todo, + test: it, }, { name: "file",