From 3f9006ac718a8501597d41f7a23300fcf8e0acd8 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Thu, 27 Aug 2026 07:11:41 +0000 Subject: [PATCH 1/4] bun_ptr: RefPtr::new_cyclic + SelfRoot; BackRef::dangling only for Shared/Mut new_cyclic builds a refcounted value that stores its own root pointer as an opaque SelfRoot; the token can only be turned into a ThisPtr through the constructed &T, so it cannot be followed early or late. A Root back-reference can hand out ThisPtrs, so BackRef::dangling() is limited to Shared/Mut. --- src/ptr/lib.rs | 43 +++++++++++++++++++++++++++- src/ptr/ref_count.rs | 68 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 1 deletion(-) diff --git a/src/ptr/lib.rs b/src/ptr/lib.rs index 1c47c9bcbc11..7fbf2f1908be 100644 --- a/src/ptr/lib.rs +++ b/src/ptr/lib.rs @@ -162,7 +162,48 @@ impl BackRef { } } -impl BackRef { +/// The root pointer of a value built by [`RefPtr::new_cyclic`], stored inside +/// that value. It has no accessors of its own: the only way to use it is +/// [`SelfRoot::this_ptr`], which takes the enclosing `&T` — so it cannot be +/// followed before the value exists or after it is gone. +#[repr(transparent)] +pub struct SelfRoot(pub(crate) core::ptr::NonNull); + +impl SelfRoot { + /// The enclosing value as a [`ThisPtr`]. `owner` must be the value this + /// token is stored in (checked). + #[inline] + pub fn this_ptr(&self, owner: &T) -> ThisPtr { + assert!( + core::ptr::eq(self.0.as_ptr().cast_const(), owner), + "SelfRoot used from a value it does not belong to" + ); + // SAFETY: `owner` is a live `&T` at `self.0` (asserted), so the value is + // constructed; `self.0` keeps the allocation-root provenance a last + // release needs. + unsafe { ThisPtr::new(self.0.as_ptr()) } + } + + /// The enclosing value as a root back-reference. + #[inline] + pub fn backref(&self, owner: &T) -> BackRef { + self.this_ptr(owner).into() + } +} + +/// Provenance markers a placeholder [`BackRef::dangling`] may carry. Not +/// [`Root`]: a `Root` back-reference can hand out a [`ThisPtr`], so it is +/// only ever minted from a real one (see [`RefPtr::new_cyclic`]). +pub trait DanglingOk: sealed::Sealed {} +impl DanglingOk for Shared {} +impl DanglingOk for Mut {} +mod sealed { + pub trait Sealed {} + impl Sealed for super::Shared {} + impl Sealed for super::Mut {} +} + +impl BackRef { #[inline] pub const fn dangling() -> Self { BackRef(core::ptr::NonNull::dangling(), core::marker::PhantomData) diff --git a/src/ptr/ref_count.rs b/src/ptr/ref_count.rs index 646a5f68866e..8d098be53ddc 100644 --- a/src/ptr/ref_count.rs +++ b/src/ptr/ref_count.rs @@ -545,6 +545,21 @@ impl RefPtr { /// /// # Safety /// `raw_ptr` must point to a live `T`. + /// [`new`](Self::new) for a `T` that stores its own root pointer (to hand + /// out [`ThisPtr`](crate::ThisPtr)s from `&self` entry points). `init` + /// receives a [`SelfRoot`](crate::SelfRoot) to store in the value; the + /// token cannot be dereferenced on its own, only through the `&T` that + /// exists once construction is done. + pub fn new_cyclic(init: impl FnOnce(crate::SelfRoot) -> T) -> Self { + let raw: NonNull = bun_core::heap::into_raw_nn(Box::::new_uninit()).cast::(); + let value = init(crate::SelfRoot(raw)); + // SAFETY: `raw` is a live, uninitialized, properly aligned `T` slot. + unsafe { raw.as_ptr().write(value) }; + // SAFETY: freshly written, so live. + debug_assert!(unsafe { T::rc_has_one_ref(raw.as_ptr()) }); + Self(raw) + } + #[inline] pub unsafe fn init_ref(raw_ptr: *mut T) -> Self { // SAFETY: caller contract @@ -975,4 +990,57 @@ mod tests { assert_eq!(type_base_name("a::b::Foo"), "Foo"); assert_eq!(type_base_name("Foo"), "Foo"); } + + struct Cyclic { + ref_count: RefCount, + self_ref: crate::SelfRoot, + payload: Box, + } + + impl Drop for Cyclic { + fn drop(&mut self) { + DROPS.fetch_add(1, Ordering::SeqCst); + } + } + + impl RefCounted for Cyclic { + unsafe fn get_ref_count(this: *mut Self) -> *mut RefCount { + // SAFETY: caller contract — field projection on a live allocation. + unsafe { &raw mut (*this).ref_count } + } + unsafe fn destructor(this: *mut Self) { + // SAFETY: caller contract — refcount hit zero, sole owner. + drop(unsafe { bun_core::heap::take(this) }); + } + } + + #[test] + fn new_cyclic_self_root_round_trip() { + let _serial = serial(); + let before = drops(); + let p = RefPtr::new_cyclic(|self_ref| Cyclic { + ref_count: RefCount::init(), + self_ref, + payload: Box::new(9), + }); + assert_eq!(*p.payload, 9); + { + // The token hands back a `ThisPtr` to the same allocation; a ref + // taken through it bumps and releases the count. + let this = p.self_ref.this_ptr(&*p); + assert_eq!(this.as_ptr().cast_const(), p.as_ptr().cast_const()); + let _guard = RefPtr::from_this(this); + assert_eq!(*this.payload, 9); + assert_eq!(p.ref_count.get(), 2); + } + assert_eq!(p.ref_count.get(), 1); + assert_eq!(drops(), before); + // Last release goes through the root the token stored. + let this = p.self_ref.this_ptr(&*p); + let guard = RefPtr::from_this(this); + drop(p); + assert_eq!(drops(), before); + drop(guard); + assert_eq!(drops(), before + 1); + } } From 5acfd15877da86245652011edd979bd03a16e5fd Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Thu, 27 Aug 2026 07:20:24 +0000 Subject: [PATCH 2/4] ref_count: init_ref keeps its own doc block --- src/ptr/ref_count.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/ptr/ref_count.rs b/src/ptr/ref_count.rs index 8d098be53ddc..7c530fc3500c 100644 --- a/src/ptr/ref_count.rs +++ b/src/ptr/ref_count.rs @@ -541,10 +541,6 @@ impl RefPtr { Self(ptr) } - /// Take a new ref on `*raw_ptr`. - /// - /// # Safety - /// `raw_ptr` must point to a live `T`. /// [`new`](Self::new) for a `T` that stores its own root pointer (to hand /// out [`ThisPtr`](crate::ThisPtr)s from `&self` entry points). `init` /// receives a [`SelfRoot`](crate::SelfRoot) to store in the value; the @@ -560,6 +556,10 @@ impl RefPtr { Self(raw) } + /// Take a new ref on `*raw_ptr`. + /// + /// # Safety + /// `raw_ptr` must point to a live `T`. #[inline] pub unsafe fn init_ref(raw_ptr: *mut T) -> Self { // SAFETY: caller contract From 3bef0fe51374329f26c9b283cd118a42a4d8bc22 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Thu, 27 Aug 2026 07:23:43 +0000 Subject: [PATCH 3/4] cron: CronJob stores a SelfRoot (RefPtr::new_cyclic) instead of a patched-in Root back-reference --- src/runtime/api/cron.rs | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/src/runtime/api/cron.rs b/src/runtime/api/cron.rs index 899400542c95..9088b44c4009 100644 --- a/src/runtime/api/cron.rs +++ b/src/runtime/api/cron.rs @@ -28,7 +28,7 @@ use bun_jsc::{ JSGlobalObject, JSObject, JSValue, JsCell, JsRef, JsResult, }; use bun_paths as path; -use bun_ptr::{BackRef, RefPtr, ThisPtr}; +use bun_ptr::{RefPtr, ThisPtr}; use bun_resolver::fs::FileSystem; #[cfg(not(target_os = "macos"))] use bun_resolver::fs::RealFS; @@ -1372,7 +1372,7 @@ pub struct CronJob { ref_count: Cell, /// Set from the allocating `RefPtr` so `&self` host fns can reach the /// `ThisPtr`-taking paths that may release refs. - self_ref: Cell>, + self_ref: bun_ptr::SelfRoot, // pub: `bun_core::from_field_ptr!(CronJob, event_loop_timer)` needs `offset_of!` visibility. // `JsCell` is `#[repr(transparent)]`, so the byte offset of the inner // `EventLoopTimer` is identical and the dispatch.rs `owner!` macro works @@ -1655,7 +1655,7 @@ impl CronJob { #[bun_jsc::host_fn(method)] pub(crate) fn stop(&self, _global: &JSGlobalObject, frame: &CallFrame) -> JsResult { - Self::self_stop(self.self_ref.get().this_ptr(), self.global.bun_vm()); + Self::self_stop(self.self_ref.this_ptr(self), self.global.bun_vm()); Ok(frame.this()) } @@ -1709,9 +1709,9 @@ impl CronJob { let vm = global.bun_vm().as_mut(); - let job = RefPtr::new(CronJob { + let job = RefPtr::new_cyclic(|self_ref| CronJob { ref_count: Cell::new(1), - self_ref: Cell::new(BackRef::dangling()), + self_ref, event_loop_timer: JsCell::new(EventLoopTimer::init_paused(EventLoopTimerTag::CronJob)), global: GlobalRef::from(global), parsed, @@ -1723,7 +1723,6 @@ impl CronJob { pending_ref: JsCell::new(None), in_fire: Cell::new(false), }); - job.self_ref.set(BackRef::from(job.this_ptr())); let Some(next_time) = job.compute_next_timespec() else { return Err(global.throw_invalid_arguments(format_args!( From a292a10fdaffd8f0428237f5292820eb893d1596 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Sat, 29 Aug 2026 07:30:00 +0000 Subject: [PATCH 4/4] node:http2: remove the remaining unsafe from H2FrameParser Squash of the branch's H2FrameParser commits for the rebase over #40516: - node:http2: remove the remaining unsafe from H2FrameParser - hand off the cork slot's ref across the re-entrant write in uncork() - cork() takes the slot in two steps; displace a re-corked owner without writing; retire the auto-flush of a displaced cork owner once its bytes are out - keep a latched compression / transport write error's auto-flush; expose the registration to tests; poll for the auto-flush retirement instead of a fixed sleep - cork/auto-flush/writeonly/keepalive refs release on drop; keepalive refs live in a typed slot - release a latched session error's auto-flush registration when the VM exits before it runs --- src/js/internal-for-testing.ts | 8 + src/jsc/AbortSignal.rs | 52 + src/jsc/H2Headers.rs | 37 + src/jsc/bindings/FfiSlice.h | 19 + src/jsc/bindings/H2HeadersMaterializer.cpp | 12 +- src/jsc/bindings/webcore/WebSocket.cpp | 5 +- src/jsc/bindings/webcore/WebSocket.h | 8 +- src/jsc/lib.rs | 2 + src/runtime/api.rs | 1 + src/runtime/api/bun/h2_frame_parser.rs | 1625 +++++++++----------- src/runtime/socket/socket_body.rs | 6 +- src/runtime/webcore.rs | 16 +- test/js/node/http2/node-http2.test.js | 216 +++ 13 files changed, 1105 insertions(+), 902 deletions(-) create mode 100644 src/jsc/H2Headers.rs create mode 100644 src/jsc/bindings/FfiSlice.h diff --git a/src/js/internal-for-testing.ts b/src/js/internal-for-testing.ts index 592b4921c8ff..cfc9f289333a 100644 --- a/src/js/internal-for-testing.ts +++ b/src/js/internal-for-testing.ts @@ -707,6 +707,14 @@ export const memoryPressureWatcherHasOsBackend: () => boolean = $newRustFunction // null where there is no PSI backend (everything except Linux). export const memoryPressurePsiTrigger: () => Buffer | null = $newRustFunction("memory_pressure.rs", "jsPsiTrigger", 0); +// Whether an H2FrameParser (a session's `Symbol.for("::bunhttp2native::")`) still has its +// end-of-tick auto-flush task registered. +export const h2AutoFlushRegistered: (parser: unknown) => boolean = $newRustFunction( + "h2_frame_parser.rs", + "jsAutoFlushRegistered", + 1, +); + export const getEventLoopStats: () => { activeTasks: number; tasks: number; diff --git a/src/jsc/AbortSignal.rs b/src/jsc/AbortSignal.rs index 5e5e64bb4e05..5e22918680bb 100644 --- a/src/jsc/AbortSignal.rs +++ b/src/jsc/AbortSignal.rs @@ -73,6 +73,34 @@ pub trait AbortListener { fn on_abort(&mut self, reason: JSValue); } +/// A native `abort` listener registered with [`AbortSignal::listen_native`]; +/// called on the signal's JS thread. +pub trait NativeAbortListener: Sized { + fn on_abort(this: bun_ptr::ThisPtr, reason: JSValue); +} + +/// A native listener's registration on an [`AbortSignal`]: holds a reference +/// on the signal and one unit of its pending activity, and removes the +/// listener when dropped. The listener keeps this for as long as it lives at +/// the address it registered. +pub struct AbortListenerRegistration { + signal: AbortSignalRef, + ctx: *mut c_void, +} + +impl AbortListenerRegistration { + pub fn signal(&self) -> &AbortSignal { + &self.signal + } +} + +impl Drop for AbortListenerRegistration { + fn drop(&mut self) { + self.signal.clean_native_bindings(self.ctx); + self.signal.pending_activity_unref(); + } +} + impl AbortSignal { pub fn listen(&self, ctx: *mut C) -> &AbortSignal { extern "C" fn callback(ptr: *mut c_void, reason: JSValue) { @@ -84,6 +112,25 @@ impl AbortSignal { self.add_listener(ctx.cast::(), callback::) } + /// Call `C::on_abort` on `listener`'s pointee when the signal aborts, for + /// as long as the returned registration is held. + pub fn listen_native( + &self, + listener: bun_ptr::BackRef, + ) -> AbortListenerRegistration { + extern "C" fn callback(ptr: *mut c_void, reason: JSValue) { + // SAFETY: `ptr` is the root pointer `listen_native` registered; its + // pointee holds the registration, which unregisters this callback + // before the pointee goes away. + C::on_abort(unsafe { bun_ptr::ThisPtr::new(ptr.cast::()) }, reason); + } + let ctx = listener.this_ptr().as_ptr().cast::(); + let signal = self.ref_(); + self.pending_activity_ref(); + self.add_listener(ctx, callback::); + AbortListenerRegistration { signal, ctx } + } + pub fn add_listener( &self, ctx: *mut c_void, @@ -157,6 +204,11 @@ impl AbortSignal { )) } + /// Node's `AbortError` (`code === "ABORT_ERR"`) carrying `cause`. + pub fn wrap_abort_error(global: &JSGlobalObject, cause: JSValue) -> JSValue { + Bun__wrapAbortError(global, cause) + } + /// Take a ref on the signal. pub fn ref_(&self) -> AbortSignalRef { // SAFETY: `WebCore__AbortSignal__ref` bumps the intrusive refcount and diff --git a/src/jsc/H2Headers.rs b/src/jsc/H2Headers.rs new file mode 100644 index 000000000000..c6153e2cabc7 --- /dev/null +++ b/src/jsc/H2Headers.rs @@ -0,0 +1,37 @@ +//! `Bun__h2__materializeHeaders` (H2HeadersMaterializer.cpp): one-call +//! materialization of a decoded HTTP/2 header block. + +use bun_core::ffi::FfiSlice; + +use crate::{JSGlobalObject, JSValue, JsResult}; + +unsafe extern "C" { + /// `packed` must hold at least the bytes `meta`'s lengths describe. + fn Bun__h2__materializeHeaders( + global_object: &JSGlobalObject, + packed: FfiSlice<'_, u8>, + meta: FfiSlice<'_, u32>, + ) -> JSValue; +} + +/// Build the `[rawHeadersArray, headersObject, sensitiveArray | undefined]` +/// tuple for one header block. `packed` is every name then value, back to back; +/// `meta` holds two u32s per field (`[nameLen | sensitive << 31, valueLen]`). +#[track_caller] +pub fn materialize(global: &JSGlobalObject, packed: &[u8], meta: &[u32]) -> JsResult { + let described: usize = meta + .as_chunks::<2>() + .0 + .iter() + .map(|m| (m[0] & 0x7fff_ffff) as usize + m[1] as usize) + .sum(); + assert!( + described <= packed.len(), + "h2 header block shorter than its field lengths" + ); + crate::call_zero_is_throw(global, || { + // SAFETY: every field `meta` describes lies inside `packed` (checked above); both are + // live borrows for the call. + unsafe { Bun__h2__materializeHeaders(global, FfiSlice::new(packed), FfiSlice::new(meta)) } + }) +} diff --git a/src/jsc/bindings/FfiSlice.h b/src/jsc/bindings/FfiSlice.h new file mode 100644 index 000000000000..eb0a2592ed0c --- /dev/null +++ b/src/jsc/bindings/FfiSlice.h @@ -0,0 +1,19 @@ +#pragma once + +#include +#include +#include + +namespace Bun { + +/// `bun_core::ffi::FfiSlice<'_, T>` — a borrowed Rust `&[T]` passed by value +/// (`struct { const T* ptr; size_t len; }`). Valid for the duration of the call. +template +struct FfiSlice { + const T* ptr; + size_t len; + + std::span span() const { return { ptr, len }; } +}; + +} // namespace Bun diff --git a/src/jsc/bindings/H2HeadersMaterializer.cpp b/src/jsc/bindings/H2HeadersMaterializer.cpp index 7e7265b5ee8c..49f932bd8bd4 100644 --- a/src/jsc/bindings/H2HeadersMaterializer.cpp +++ b/src/jsc/bindings/H2HeadersMaterializer.cpp @@ -18,6 +18,7 @@ #include #include "HTTPHeaderIdentifiers.h" #include "HTTPHeaderNames.h" +#include "FfiSlice.h" using namespace JSC; using namespace WebCore; @@ -61,14 +62,16 @@ static JSString* h2ValueToJS(VM& vm, const uint8_t* ptr, size_t length) // packed layout: name bytes then value bytes, in field order. // Returns [rawHeadersArray, headersObject, sensitiveArray | undefined], or 0 // with an exception pending. -extern "C" [[ZIG_EXPORT(zero_is_throw)]] JSC::EncodedJSValue Bun__h2__materializeHeaders( +extern "C" JSC::EncodedJSValue Bun__h2__materializeHeaders( JSC::JSGlobalObject* globalObject, - const uint8_t* packed, - const uint32_t* meta, - size_t fieldCount) + FfiSlice packedSlice, + FfiSlice metaSlice) { auto& vm = JSC::getVM(globalObject); auto scope = DECLARE_THROW_SCOPE(vm); + const uint8_t* packed = packedSlice.ptr; + const uint32_t* meta = metaSlice.ptr; + const size_t fieldCount = metaSlice.len / 2; JSC::JSArray* raw = JSC::constructEmptyArray(globalObject, nullptr, static_cast(fieldCount * 2)); RETURN_IF_EXCEPTION(scope, {}); @@ -85,6 +88,7 @@ extern "C" [[ZIG_EXPORT(zero_is_throw)]] JSC::EncodedJSValue Bun__h2__materializ const size_t nameLen = packedNameLen & 0x7fffffffu; const size_t valueLen = meta[i * 2 + 1]; + ASSERT(offset + nameLen + valueLen <= packedSlice.len); const uint8_t* nameBytes = packed + offset; offset += nameLen; const uint8_t* valueBytes = packed + offset; diff --git a/src/jsc/bindings/webcore/WebSocket.cpp b/src/jsc/bindings/webcore/WebSocket.cpp index 8ce8300a1ea7..66db2aaaa87d 100644 --- a/src/jsc/bindings/webcore/WebSocket.cpp +++ b/src/jsc/bindings/webcore/WebSocket.cpp @@ -1737,10 +1737,7 @@ extern "C" void WebSocket__didConnectWithTunnel(WebCore::WebSocket* webSocket, v webSocket->didConnectWithTunnel(tunnel, bufferedData, deflate_params); } -struct FfiRawHeaderSlice { - const WebCore::WebSocket::HandshakeRawHeader* ptr; - size_t len; -}; +using FfiRawHeaderSlice = Bun::FfiSlice; extern "C" void WebSocket__didReceiveHandshakeResponse(WebCore::WebSocket* webSocket, uint16_t statusCode, WebCore::WebSocket::FfiSlice statusMessage, FfiRawHeaderSlice headers, WebCore::WebSocket::FfiSlice body) { diff --git a/src/jsc/bindings/webcore/WebSocket.h b/src/jsc/bindings/webcore/WebSocket.h index 6af322b184a3..e4512fef22ca 100644 --- a/src/jsc/bindings/webcore/WebSocket.h +++ b/src/jsc/bindings/webcore/WebSocket.h @@ -39,6 +39,7 @@ #include #include "FetchHeaders.h" #include "WebSocketErrorCode.h" +#include "FfiSlice.h" namespace WebCore { class JSBlob; @@ -215,12 +216,7 @@ class WebSocket final : public RefCounted, public EventTargetWithInli void didReceiveMessage(String&& message); void didReceiveBinaryData(const AtomString& eventName, const std::span binaryData); - /// `bun_core::ffi::FfiSlice` — a borrowed `&[u8]` passed by value. - struct FfiSlice { - const uint8_t* ptr; - size_t len; - std::span span() const { return { ptr, len }; } - }; + using FfiSlice = Bun::FfiSlice; struct HandshakeRawHeader { FfiSlice name; FfiSlice value; diff --git a/src/jsc/lib.rs b/src/jsc/lib.rs index 03faa8dec0e7..b576754f98c0 100644 --- a/src/jsc/lib.rs +++ b/src/jsc/lib.rs @@ -408,6 +408,8 @@ pub mod btjs; pub mod cpp_task; #[path = "fmt_jsc.rs"] pub mod fmt_jsc; +#[path = "H2Headers.rs"] +pub mod h2_headers; #[path = "HTTPServerAgent.rs"] pub mod http_server_agent; #[path = "JSSecrets.rs"] diff --git a/src/runtime/api.rs b/src/runtime/api.rs index 3a3c08eef699..32009bb37004 100644 --- a/src/runtime/api.rs +++ b/src/runtime/api.rs @@ -135,6 +135,7 @@ pub mod bun { pub use crate::api::h2_frame_parser_body::H2FrameParser; // js2native thunk (`$rust(h2_frame_parser.rs, …)` in generated_js2native.rs). pub(crate) use crate::api::h2_frame_parser_body::h2_frame_parser_constructor; + pub(crate) use crate::api::h2_frame_parser_body::js_auto_flush_registered; } } diff --git a/src/runtime/api/bun/h2_frame_parser.rs b/src/runtime/api/bun/h2_frame_parser.rs index 00f065396660..292147b113f4 100644 --- a/src/runtime/api/bun/h2_frame_parser.rs +++ b/src/runtime/api/bun/h2_frame_parser.rs @@ -6,29 +6,27 @@ )] use core::cell::{Cell, RefCell}; -use core::ffi::c_void; -use core::mem::ManuallyDrop; -use core::ptr::NonNull; use std::borrow::Cow; +use std::rc::Rc; use crate::api::socket::{TCPSocket, TLSSocket}; use crate::node::{Encoding, StringOrBuffer}; use crate::socket::NativeCallbacks; use crate::webcore::AutoFlusher; use bstr::BStr; -use bun_collections::{ByteVecExt, HashMap as BunHashMap, HiveArrayFallback, VecExt}; +use bun_collections::{ByteVecExt, HashMap as BunHashMap, VecExt}; use bun_core::strings; use bun_http::lshpack; -use bun_jsc::AbortSignal; use bun_jsc::ErrorCode as JscErrorCode; -use bun_jsc::abort_signal::AbortListener; +use bun_jsc::abort_signal::{AbortListenerRegistration, NativeAbortListener}; use bun_jsc::array_buffer::BinaryType; use bun_jsc::bun_string_jsc; use bun_jsc::virtual_machine::VirtualMachine; use bun_jsc::{ - CallFrame, GlobalRef, JSGlobalObject, JSValue, JsCell, JsRef, JsResult, StrongOptional, + AbortSignal, CallFrame, GlobalRef, JSGlobalObject, JSValue, JsCell, JsRef, JsResult, + StrongOptional, }; -use bun_ptr::RefPtr; +use bun_ptr::{BackRef, RefPtr, ThisPtr}; bun_output::declare_scope!(H2FrameParser, visible); @@ -91,15 +89,17 @@ pub mod JSH2FrameParser { const MAX_PAYLOAD_SIZE_WITHOUT_FRAME: usize = 16384 - FrameHeader::BYTE_SIZE - 1; -/// `Copy` view of [`NativeSocket`] for call sites to snapshot across -/// re-entrant writes. BACKREF — the socket strictly outlives the attachment: -/// `Tls`/`Tcp` are kept alive by the `RefPtr` stored in the -/// socket's `native_callback` slot, `*Writeonly` by the ref [`NativeSocket`] -/// holds on them. #[derive(Default, Clone, Copy)] enum BunSocket { #[default] None, + // BACKREF — the socket strictly outlives the H2FrameParser while attached: + // `Tls`/`Tcp` are kept alive by the parser ref stored in the socket's + // `native_callback` slot (released in `detach_native_socket`), + // and `*Writeonly` by the ref `H2FrameParser::writeonly_socket_ref` holds + // between `attach_to_native_socket` and `detach_native_socket`. `BackRef` + // makes the shared-only deref safe at every read site (all `NewSocket` + // methods used here take `&self`). Tls(bun_ptr::BackRef), TlsWriteonly(bun_ptr::BackRef), Tcp(bun_ptr::BackRef), @@ -109,81 +109,8 @@ enum BunSocket { /// The ref a `BunSocket::*Writeonly` attachment holds on its socket; held /// only to be dropped on detach. enum WriteonlySocketRef { - Tls(#[allow(dead_code)] RefPtr), - Tcp(#[allow(dead_code)] RefPtr), -} - -/// The parser's attachment to a native socket, and its owner: attaching either -/// installs the parser as the socket's native callback (`Tls`/`Tcp`) or, when -/// that slot is taken, holds a ref on the socket (`*Writeonly`); `detach` / -/// `Drop` undo whichever one it was. Readers take a [`BunSocket`] snapshot. -#[derive(Default)] -struct NativeSocket { - socket: Cell, - /// Held while `socket` is a `*Writeonly`. - writeonly_ref: Cell>, -} - -impl NativeSocket { - #[inline] - fn get(&self) -> BunSocket { - self.socket.get() - } - - /// `attach_native_callback` stores `h2` (a ref on the parser) in the - /// socket, dropped in `NewSocket::detach_native_callback` (or inside - /// `attach_native_callback` when rejected); when rejected we fall back to - /// write-only mode and hold a ref on the socket instead. - fn attach( - &self, - socket: bun_ptr::ThisPtr>, - h2: RefPtr, - attached: fn(bun_ptr::BackRef, bun_ptr::Root>) -> BunSocket, - writeonly: fn(bun_ptr::BackRef, bun_ptr::Root>) -> BunSocket, - writeonly_ref: fn(RefPtr>) -> WriteonlySocketRef, - ) { - debug_assert!(matches!(self.socket.get(), BunSocket::None)); - // BACKREF: `socket` is the live `m_ctx` of the JS wrapper rooted by the - // caller's `socket_js`. - self.socket - .set(if socket.attach_native_callback(NativeCallbacks::H2(h2)) { - attached(socket.into()) - } else { - self.writeonly_ref - .set(Some(writeonly_ref(RefPtr::from_this(socket)))); - writeonly(socket.into()) - }); - } - - fn detach(&self) { - match self.socket.replace(BunSocket::None) { - BunSocket::Tcp(socket) => socket.detach_native_callback(), - BunSocket::Tls(socket) => socket.detach_native_callback(), - BunSocket::TcpWriteonly(_) | BunSocket::TlsWriteonly(_) => { - drop(self.writeonly_ref.take()); - } - BunSocket::None => {} - } - } -} - -impl Drop for NativeSocket { - fn drop(&mut self) { - self.detach(); - } -} - -unsafe extern "C" { - safe fn Bun__wrapAbortError(global_object: &JSGlobalObject, cause: JSValue) -> JSValue; - /// One-call materialization of a decoded header block: returns the - /// [rawHeadersArray, headersObject, sensitiveArray|undefined] tuple, or a - /// zero JSValue with a JS exception pending. See H2HeadersMaterializer.cpp. - fn Bun__h2__materializeHeaders( - global_object: &JSGlobalObject, - packed: *const u8, - meta: *const u32, - field_count: usize, - ) -> JSValue; + Tls(#[allow(dead_code)] bun_ptr::RefPtr), + Tcp(#[allow(dead_code)] bun_ptr::RefPtr), } // ────────────────────────────────────────────────────────────────────────── @@ -365,26 +292,20 @@ impl UInt31WithReserved { } } -// packed struct(u40): streamIdentifier: u32, weight: u8 -#[repr(C, packed)] +// Wire layout (u40): streamIdentifier: u32 (big-endian), weight: u8. #[derive(Clone, Copy, Default)] struct StreamPriority { stream_identifier: u32, weight: u8, } -// SAFETY: `#[repr(C, packed)]` with `u32 + u8` fields — no padding, no niches, -// every 5-byte pattern is a valid value. -unsafe impl bytemuck::Zeroable for StreamPriority {} -// SAFETY: see `Zeroable` impl above; additionally `Copy + 'static`. -unsafe impl bytemuck::Pod for StreamPriority {} -const _: () = assert!(core::mem::size_of::() == StreamPriority::BYTE_SIZE); impl StreamPriority { pub(crate) const BYTE_SIZE: usize = 5; #[inline] fn write(self, writer: &mut impl WireWriter) -> bool { - let mut swap = self; - swap.stream_identifier = swap.stream_identifier.swap_bytes(); - writer.write_all(bytemuck::bytes_of(&swap)).is_ok() + let mut bytes = [0u8; Self::BYTE_SIZE]; + bytes[..4].copy_from_slice(&self.stream_identifier.to_be_bytes()); + bytes[4] = self.weight; + writer.write_all(&bytes).is_ok() } } @@ -847,14 +768,8 @@ use bun_io::FixedBufferStream; // ────────────────────────────────────────────────────────────────────────── const ENABLE_AUTO_CORK: bool = true; -const ENABLE_ALLOCATOR_POOL: bool = true; // ENABLE HIVE ALLOCATOR OPTIMIZATION const MAX_BUFFER_SIZE: u32 = 32768; -/// `bun.HiveArray(H2FrameParser, 256).Fallback` — per-thread slab of 256 -/// parser slots with heap fallback. Lazily boxed on first use (the inline -/// array is ~tens of KB and would otherwise sit in every thread's TLS). -type H2FrameParserHiveAllocator = HiveArrayFallback; - // Exactly one max-size TLS record of plaintext: every full-buffer flush is one SSL_write // producing one full 16 KB record (uSockets' BIO sends per record, so corking beyond a // record adds memcpy without saving syscalls). write() fills to this boundary and keeps @@ -875,23 +790,16 @@ thread_local! { static BATCH_BUFFER: RefCell> = const { RefCell::new(Vec::new()) }; // Wire-order segments of the in-progress send_data batch: frame headers (and any // padded frames) live in BATCH_BUFFER and are referenced by offset; payload slices - // of plain-TCP writes are referenced directly so flush can writev them without - // copying 16 KB per frame into the batch. + // of plain-TCP writes are referenced by offset into send_data's payload so flush can + // writev them without copying 16 KB per frame into the batch. static BATCH_SEGMENTS: RefCell> = const { RefCell::new(Vec::new()) }; // Reused iovec scratch for the vectored flush. static BATCH_IOVECS: RefCell> = const { RefCell::new(Vec::new()) }; - /// The parser whose frames `CORK_BUFFER` holds; the slot keeps a ref on it. - /// `ManuallyDrop` so the slot needs no TLS destructor (a bare - /// `#[thread_local]`, and nothing derefs a parser at thread exit). - static CORKED_H2: Cell>>> = - const { Cell::new(ManuallyDrop::new(None)) }; - // `ManuallyDrop` inside the `Box`: the TLS destructor runs after - // `WebWorker::destroy` has raw-deallocated the VM, so `HiveArray::Drop` - // on any leaked parser would touch freed JSC/uws state. Skip slot - // teardown (a leaked parser is a bug anyway) while still freeing the - // pool allocation itself. - static POOL: RefCell>>> = - const { RefCell::new(None) }; + /// The parser whose bytes CORK_BUFFER currently holds. Invariant, kept by `cork`, `uncork`, + /// `displace_cork` and `release_refs_stranded_by_exit` (the only writers): `p.cork_ref.is_some()` ⇔ + /// `CORKED_H2 == Some(p)` — so the owner is alive while it is here and the back-reference + /// is always live. + static CORKED_H2: Cell>> = const { Cell::new(None) }; } /// One wire-order piece of a multi-frame send_data batch (see BATCH_SEGMENTS). @@ -899,16 +807,16 @@ thread_local! { enum BatchSegment { /// Bytes inside BATCH_BUFFER (frame headers, padded frames, corked prefix). Batch { off: u32, len: u32 }, - /// A borrowed payload slice, valid for the duration of the send_data call. - Ext { ptr: *const u8, len: u32 }, + /// Bytes inside the `payload` of the send_data call that recorded it. + Payload { off: u32, len: u32 }, } impl BatchSegment { #[inline(always)] - fn raw_parts(self, batch: &[u8]) -> (*const u8, usize) { + fn bytes<'a>(self, batch: &'a [u8], payload: &'a [u8]) -> &'a [u8] { match self { - BatchSegment::Batch { off, len } => (batch[off as usize..].as_ptr(), len as usize), - BatchSegment::Ext { ptr, len } => (ptr, len as usize), + BatchSegment::Batch { off, len } => &batch[off as usize..][..len as usize], + BatchSegment::Payload { off, len } => &payload[off as usize..][..len as usize], } } } @@ -988,42 +896,35 @@ impl TxFrameTracker { } } -/// The `+1` a native frame holds on the parser while it runs code that can free it (an inbound -/// dispatch, a write that re-enters JS). Live guards are counted in -/// `H2FrameParser::native_keepalives` so `finalize` can release the ones whose frame will never +/// The `+1` a native frame holds on the parser while it runs code that can release every other +/// ref (an inbound dispatch, a write that re-enters JS). The refs live in +/// `H2FrameParser::keepalive_refs` so `finalize` can release the ones whose frame will never /// return — see `release_refs_stranded_by_exit`. -struct Keepalive<'a>(&'a H2FrameParser); +struct Keepalive<'a> { + parser: &'a H2FrameParser, +} impl Drop for Keepalive<'_> { fn drop(&mut self) { - let parser = self.0; - debug_assert!(parser.native_keepalives.get() > 0); - // Decrement first: this `deref()` can be the last one and free `parser`. - parser - .native_keepalives - .set(parser.native_keepalives.get() - 1); - parser.deref(); + // Taken out first, dropped after: this release can be the last one. + let ours = self.parser.keepalive_refs.with_mut(Vec::pop); + debug_assert!(ours.is_some()); + drop(ours); } } -/// A `&mut Stream` that only exists inside an armed dispatch scope (`enter_stream_dispatch`): -/// while it is live, rewrite_read defers stream frees, so user JS that re-enters `read()` -/// (option getters, header-value `toString`) cannot free the stream out from under the borrow. +/// A stream handle held inside an armed dispatch scope (`enter_stream_dispatch`): while it is +/// live, rewrite_read defers evicting finished streams, so user JS that re-enters `read()` +/// (option getters, header-value `toString`) cannot retire the stream mid-operation. struct GuardedStream<'a> { - stream: &'a mut Stream, + stream: Rc, _dispatch: DispatchGuard<'a>, } impl core::ops::Deref for GuardedStream<'_> { type Target = Stream; fn deref(&self) -> &Stream { - self.stream - } -} - -impl core::ops::DerefMut for GuardedStream<'_> { - fn deref_mut(&mut self) -> &mut Stream { - self.stream + &self.stream } } @@ -1033,13 +934,17 @@ impl core::ops::DerefMut for GuardedStream<'_> { // `&mut T` auto-derefs to `&T` so the impls below compile against either. #[bun_jsc::JsClass] #[derive(bun_ptr::RefCounted)] -#[ref_count(destroy = Self::release)] pub struct H2FrameParser { + /// This allocation's root pointer, so `&self` methods can mint the refs the cork slot, the + /// auto-flush registration, a keepalive or an abort listener hold. + self_ref: bun_ptr::SelfRoot, strong_this: JsCell, global_this: GlobalRef, // JSC_BORROW — read-only after construction // allocator field dropped — global mimalloc handlers: JsCell, - native_socket: NativeSocket, + native_socket: Cell, + /// Held while `native_socket` is a `*Writeonly`; released by `detach_native_socket`. + writeonly_socket_ref: Cell>, local_settings: Cell, /// Bitmask (SETTING_BIT_*) of standard SETTINGS explicitly provided by JS. node only /// serializes explicitly submitted settings — defaults are never put on the wire, so the @@ -1131,7 +1036,7 @@ pub struct H2FrameParser { // TODO: this will be removed when I re-add header and data priorization outbound_queue_size: Cell, - streams: JsCell>, + streams: JsCell>>, hpack: JsCell>, @@ -1153,12 +1058,18 @@ pub struct H2FrameParser { /// Where the bytes emitted through `write()` over a JS-backed transport stand relative to /// frame and header-block boundaries. tx_tracker: Cell, - ref_count: bun_ptr::RefCount, // intrusive — bun.ptr.RefCount(@This(), "ref_count", deinit, .{}) - /// Number of live `Keepalive` guards: the `+1`s held by native frames currently on the stack. - /// Read only by `release_refs_stranded_by_exit()`. - native_keepalives: Cell, - - auto_flusher: JsCell, + ref_count: bun_ptr::RefCount, + /// The `+1`s held by the `Keepalive` guards of native frames currently on the stack; only + /// `release_refs_stranded_by_exit()` releases them out of turn. + keepalive_refs: JsCell>>, + + pub(crate) auto_flusher: JsCell, + /// Held while `auto_flusher` is registered with the deferred-task queue. Registered ⇒ this + /// parser owns the cork slot, or has displaced/unsent bytes in `write_buffer`, or has a + /// latched transport/compression error to report. + auto_flush_ref: Cell>>, + /// Held exactly while this parser owns `CORKED_H2` (see the invariant there). + cork_ref: Cell>>, padding_strategy: Cell, // ---- from-scratch rewrite engine (src/runtime/api/bun/h2) ---- @@ -1181,69 +1092,38 @@ pub struct H2FrameParser { } impl H2FrameParser { - /// `RefCounted` destructor (and the constructor's error path): drop in - /// place, then return the slot to the pool (or free the Box). - fn release(this: *mut Self) { - if ENABLE_ALLOCATOR_POOL { - POOL.with_borrow_mut(|pool| { - // SAFETY: `this` is a live, fully-initialised allocation we exclusively - // own; `put` drops it in place and recycles the storage. - unsafe { - pool.as_mut() - .expect("H2FrameParser released before constructor initialised pool") - .put(this) - } - }); - } else { - // SAFETY: `this` was `heap::alloc`'d in `constructor`. - unsafe { bun_core::heap::destroy(this) }; - } - } - /// Safe accessor for the JSC_BORROW global. #[inline] fn global(&self) -> GlobalRef { self.global_this } - /// `self`'s address as `*mut Self` for uSockets / deferred-task ctx slots. - /// The callbacks deref it as `&*const` (shared) — see `on_auto_flush_trampoline` - /// — so no write provenance is required; the `*mut` spelling is purely to - /// match the C signature. All mutation goes through `Cell`/`JsCell` fields. - #[inline] - fn as_ctx_ptr(&self) -> *mut Self { - std::ptr::from_ref::(self).cast_mut() + fn this_ptr(&self) -> ThisPtr { + self.self_ref.this_ptr(self) + } + + /// A new `+1` on this parser. + fn new_ref(&self) -> RefPtr { + RefPtr::from_this(self.this_ptr()) } - /// Hold a `+1` for the extent of a native frame that can re-enter JS (and therefore free the - /// parser). Counted, so `finalize` can release it if `process.exit()` strands the frame. + fn is_corked(&self) -> bool { + CORKED_H2 + .get() + .is_some_and(|p| std::ptr::eq(p.as_const_ptr(), self)) + } + + /// Hold a `+1` for the extent of a native frame that can re-enter JS (and therefore release + /// every other ref). Counted, so `finalize` can release it if `process.exit()` strands the + /// frame. fn keepalive(&self) -> Keepalive<'_> { - self.ref_(); - self.native_keepalives.set(self.native_keepalives.get() + 1); - Keepalive(self) + let ours = self.new_ref(); + self.keepalive_refs.with_mut(|refs| refs.push(ours)); + Keepalive { parser: self } } - /// Hold a ref on `self` for the guard's lifetime (across re-entrant calls). - #[inline] - pub(crate) fn ref_guard(&self) -> RefPtr { - // SAFETY: `self` is the live heap allocation. - unsafe { RefPtr::init_ref(self.as_ctx_ptr()) } - } - - pub(crate) fn ref_(&self) { - // SAFETY: `self` is live; `RefCount::ref_` only reads/writes the - // embedded `ref_count` Cell (interior-mutable), so `&self`→`*mut` - // is sound for that single field access. - unsafe { bun_ptr::RefCount::::ref_(self.as_ctx_ptr()) }; - } - // R-2: `&self` — `RefCount` is `Cell`-backed and every other field is - // `Cell`/`JsCell`, so `Drop` writes only through `UnsafeCell`-derived - // pointers; the `*mut` cast is signature-only. - pub(crate) fn deref(&self) { - // SAFETY: `self` is live; `deref` decrements the intrusive count and, - // on zero, drops and releases the slot. The caller must not touch - // `self` after this returns when count was 1. - unsafe { bun_ptr::RefCount::::deref(self.as_ctx_ptr()) }; + fn stream(&self, id: u32) -> Option> { + self.streams.get().get(&id).cloned() } } @@ -1259,7 +1139,7 @@ impl H2FrameParser { pub(crate) struct StreamResumableIterator { // Note: `streams` // is `JsCell`-backed, so a shared backref suffices and the in-loop - // body can keep its own `&H2FrameParser` without provenance gymnastics. + // body can keep its own `&H2FrameParser`. // `ParentRef` encapsulates the back-pointer invariant (parser outlives the // iterator — every call site constructs the iterator from a live `&Self` // and drains it in the same scope) so `next()` derefs through safe `Deref`. @@ -1276,14 +1156,14 @@ impl StreamResumableIterator { index: 0, } } - pub(crate) fn next(&mut self) -> Option<*mut Stream> { + pub(crate) fn next(&mut self) -> Option> { // R-2: `streams` is `JsCell`-backed (UnsafeCell), so the shared backref // read here coexists soundly with the loop body's own `&self` accesses. let streams = self.parser.streams.get(); while let Some(&id) = self.ids.get(self.index) { self.index += 1; - if let Some(&stream) = streams.get(&id) { - return Some(stream); + if let Some(stream) = streams.get(&id) { + return Some(Rc::clone(stream)); } } None @@ -1307,67 +1187,68 @@ enum StreamState { CLOSED = 7, } +/// Shared through `Rc`: host functions, engine callbacks and the abort listener all reach a +/// stream by id while user JS they run can re-enter and finish it, so every field is a cell and +/// the map entry is only one of the owners. pub struct Stream { id: u32, - state: StreamState, - js_context: StrongOptional, // jsc.Strong.Optional - wait_for_trailers: bool, - end_after_headers: bool, - padding_strategy: PaddingStrategy, - rst_code: u32, - stream_dependency: u32, - exclusive: bool, - weight: u16, + state: Cell, + js_context: JsCell, + wait_for_trailers: Cell, + end_after_headers: Cell, + padding_strategy: Cell, + rst_code: Cell, + stream_dependency: Cell, + exclusive: Cell, + weight: Cell, // current window size for the stream - window_size: u64, + window_size: Cell, // used window size for the stream - used_window_size: u64, + used_window_size: Cell, // remote window size for the stream - remote_window_size: u64, + remote_window_size: Cell, // remote used window size for the stream - remote_used_window_size: u64, - signal: Option>, + remote_used_window_size: Cell, + signal: JsCell>>, // The JS readable for this stream is paused (setStreamReading(id, false)): the engine defers // replenishing the stream's receive window until reading resumes, backpressuring the peer. - reading_paused: bool, + reading_paused: Cell, // when we have backpressure we queue the data e round robin the Streams - data_frame_queue: PendingQueue, + data_frame_queue: JsCell, } +/// The abort listener a stream registers on its request's `AbortSignal` (`attach_signal`); +/// owned by the stream, dropped by `free_resources`. pub(crate) struct SignalRef { - signal: bun_jsc::AbortSignalRef, - // TODO: We should not need this ref counting here, since Parser owns Stream + registration: JsCell>, + /// Released on drop. parser: RefPtr, stream_id: u32, } -impl SignalRef { - pub(crate) fn is_aborted(&self) -> bool { - self.signal.aborted() - } - - pub(crate) fn abort_listener(this: &mut SignalRef, reason: JSValue) { +impl NativeAbortListener for SignalRef { + fn on_abort(this: ThisPtr, reason: JSValue) { bun_output::scoped_log!(H2FrameParser, "abortListener"); reason.ensure_still_alive(); - let parser = &*this.parser; - let Some(stream) = parser.streams.get().get(&this.stream_id).copied() else { - return; - }; - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the map entry exists - let stream = unsafe { &mut *stream }; - if stream.state != StreamState::CLOSED { - let wrapped = Bun__wrapAbortError(&parser.global_this, reason); - parser.abort_stream(stream, wrapped); + // `abort_stream` frees this listener (via `free_resources`) before it dispatches, so + // nothing of `this` is touched past that call and the parser is held on its own. + let parser = this.parser.this_ptr(); + let stream_id = this.stream_id; + let _keepalive = parser.keepalive(); + if let Some(stream) = parser.stream(stream_id) + && stream.state.get() != StreamState::CLOSED + { + let wrapped = AbortSignal::wrap_abort_error(&parser.global_this, reason); + parser.abort_stream(&stream, wrapped); } } } impl Drop for SignalRef { fn drop(&mut self) { - // Release our listener; dropping `signal` then unrefs it. - let this = std::ptr::from_mut(self).cast::(); - self.signal.clean_native_bindings(this); + // `parser` is released after this. + drop(self.registration.replace(None)); } } @@ -1443,7 +1324,7 @@ impl PendingFrame { impl Stream { pub fn get_padding(&self, frame_len: usize, max_len: usize) -> u8 { - match self.padding_strategy { + match self.padding_strategy.get() { PaddingStrategy::None => 0, PaddingStrategy::Aligned => { let diff = (frame_len + 9) % 8; @@ -1460,82 +1341,104 @@ impl Stream { } } - pub fn flush_queue(&mut self, client: &H2FrameParser, written: &mut usize) -> FlushState { + pub fn flush_queue(&self, client: &H2FrameParser, written: &mut usize) -> FlushState { if !self.can_send_data() { // empty or cannot send data return FlushState::NoAction; } - // try to flush one frame - let Some(front) = self.data_frame_queue.peek_front() else { + // What the head of the queue can put on the wire right now. Decided (and dequeued / + // sliced) in one queue borrow; written and dispatched after it. + enum Next { + /// A zero-length frame, dequeued. + Empty(PendingFrame), + /// Flow control leaves no room; `Backpressure` when the connection window is the limit. + Limited(FlushState), + /// The first `max_size` bytes of the head frame (which stays queued, advanced). + Partial { bytes: Vec, max_size: usize }, + /// The whole head frame, dequeued. + Whole { + frame: PendingFrame, + max_size: usize, + }, + } + let stream_window = (self + .remote_window_size + .get() + .saturating_sub(self.remote_used_window_size.get())) + as usize; + let conn_window = (client + .remote_window_size + .get() + .saturating_sub(client.remote_used_window_size.get())) + as usize; + let next = self.data_frame_queue.with_mut(|q| { + let front = q.peek_front()?; + if front.len == 0 { + return q.dequeue().map(Next::Empty); + } + let frame_remaining = front.slice().len(); + let max_size = frame_remaining + .min(stream_window) + .min(conn_window) + .min(MAX_PAYLOAD_SIZE_WITHOUT_FRAME); + if max_size == 0 { + bun_output::scoped_log!( + H2FrameParser, + "dataFrame flow control limited {} {} {} {} {} {}", + frame_remaining, + self.remote_window_size.get(), + self.remote_used_window_size.get(), + client.remote_window_size.get(), + client.remote_used_window_size.get(), + max_size + ); + // we are flow control limited lets return backpressure if is limited in the connection so we short circuit the flush + return Some(Next::Limited( + if client.remote_window_size.get() == client.remote_used_window_size.get() { + FlushState::Backpressure + } else { + FlushState::NoAction + }, + )); + } + if max_size < frame_remaining { + // we need to break the frame into smaller chunks + let bytes = front.slice()[0..max_size].to_vec(); + front.offset += u32::try_from(max_size).expect("int cast"); + Some(Next::Partial { bytes, max_size }) + } else { + q.dequeue().map(|frame| Next::Whole { frame, max_size }) + } + }); + let Some(next) = next else { return FlushState::NoAction; }; - let frame_len = front.len; - let frame_remaining = front.slice().len(); let mut owned_frame: Option = None; let no_backpressure: bool = 'brk: { let mut writer = client.to_writer(); - if frame_len == 0 { - // flush a zero payload frame - let Some(frame) = self.data_frame_queue.dequeue() else { - return FlushState::NoAction; - }; - let data_header = FrameHeader { - type_: FrameType::HTTP_FRAME_DATA as u8, - flags: if frame.end_stream && !self.wait_for_trailers { - DataFrameFlags::END_STREAM as u8 - } else { - 0 - }, - stream_identifier: self.id, - length: 0, - }; - owned_frame = Some(frame); - break 'brk data_header.write(&mut writer, &client.frames_sent_legacy); - } else { - let max_size = frame_remaining - .min( - (self - .remote_window_size - .saturating_sub(self.remote_used_window_size)) - as usize, - ) - .min( - (client - .remote_window_size - .get() - .saturating_sub(client.remote_used_window_size.get())) - as usize, - ) - .min(MAX_PAYLOAD_SIZE_WITHOUT_FRAME); - if max_size == 0 { - bun_output::scoped_log!( - H2FrameParser, - "dataFrame flow control limited {} {} {} {} {} {}", - frame_remaining, - self.remote_window_size, - self.remote_used_window_size, - client.remote_window_size.get(), - client.remote_used_window_size.get(), - max_size - ); - // we are flow control limited lets return backpressure if is limited in the connection so we short circuit the flush - return if client.remote_window_size.get() - == client.remote_used_window_size.get() - { - FlushState::Backpressure - } else { - FlushState::NoAction + match next { + Next::Limited(state) => return state, + Next::Empty(frame) => { + // flush a zero payload frame + let data_header = FrameHeader { + type_: FrameType::HTTP_FRAME_DATA as u8, + flags: if frame.end_stream && !self.wait_for_trailers.get() { + DataFrameFlags::END_STREAM as u8 + } else { + 0 + }, + stream_identifier: self.id, + length: 0, }; + owned_frame = Some(frame); + break 'brk data_header.write(&mut writer, &client.frames_sent_legacy); } - if max_size < frame_remaining { - // we need to break the frame into smaller chunks - let Some(frame) = self.data_frame_queue.peek_front() else { - return FlushState::NoAction; - }; - let able_to_send = frame.slice()[0..max_size].to_vec(); - frame.offset += u32::try_from(max_size).expect("int cast"); + Next::Partial { + bytes: able_to_send, + max_size, + } => { client .queued_data_size .set(client.queued_data_size.get() - able_to_send.len() as u64); @@ -1556,7 +1459,8 @@ impl Stream { max_size, payload_size ); - self.remote_used_window_size += payload_size as u64; + self.remote_used_window_size + .set(self.remote_used_window_size.get() + payload_size as u64); client .remote_used_window_size .set(client.remote_used_window_size.get() + payload_size as u64); @@ -1578,12 +1482,10 @@ impl Stream { } else { break 'brk writer.write_all(&able_to_send).is_ok(); } - } else { + } + Next::Whole { frame, max_size } => { // flush with some payload - owned_frame = self.data_frame_queue.dequeue(); - let Some(frame) = owned_frame.as_ref() else { - return FlushState::NoAction; - }; + let frame = owned_frame.insert(frame); let frame_slice: &[u8] = frame.slice(); client .queued_data_size @@ -1605,12 +1507,13 @@ impl Stream { max_size, payload_size ); - self.remote_used_window_size += payload_size as u64; + self.remote_used_window_size + .set(self.remote_used_window_size.get() + payload_size as u64); client .remote_used_window_size .set(client.remote_used_window_size.get() + payload_size as u64); client.note_engine_send_consumed(self.id, payload_size as u64); - let mut flags: u8 = if frame.end_stream && !self.wait_for_trailers { + let mut flags: u8 = if frame.end_stream && !self.wait_for_trailers.get() { DataFrameFlags::END_STREAM as u8 } else { 0 @@ -1643,23 +1546,23 @@ impl Stream { if let Some(callback_value) = _frame.callback.get() { client.dispatch_write_callback(callback_value); } - if self.data_frame_queue.is_empty() { + if self.data_frame_queue.get().is_empty() { if _frame.end_stream { - if self.wait_for_trailers { + if self.wait_for_trailers.get() { client.dispatch(JSH2FrameParser::Gc::onWantTrailers, self.get_identifier()); } else { let identifier = self.get_identifier(); identifier.ensure_still_alive(); - if self.state == StreamState::HALF_CLOSED_REMOTE { - self.state = StreamState::CLOSED; + if self.state.get() == StreamState::HALF_CLOSED_REMOTE { + self.state.set(StreamState::CLOSED); self.free_resources::(client); } else { - self.state = StreamState::HALF_CLOSED_LOCAL; + self.state.set(StreamState::HALF_CLOSED_LOCAL); } client.dispatch_with_extra( JSH2FrameParser::Gc::onStreamEnd, identifier, - JSValue::js_number(self.state as u8 as f64), + JSValue::js_number(self.state.get() as u8 as f64), ); } } @@ -1675,7 +1578,7 @@ impl Stream { } pub fn queue_frame( - &mut self, + &self, client: &H2FrameParser, bytes: &[u8], callback: JSValue, @@ -1683,94 +1586,70 @@ impl Stream { ) { let global_this = client.global_this; - // Note: `dispatch_write_callback()` below re-enters JS, which can - // call back into `H2FrameParser` host-fns (e.g. `writeStream`) that - // look this `Stream` up by id from `client.streams` and reach - // `queue_frame()` again with a fresh `&mut Stream` aliasing this one. - // R-2: `client` is now `&H2FrameParser` (UnsafeCell-backed fields), so - // the parser-side noalias miscompile is structurally impossible. The - // `Stream`-side `&mut self` alias across re-entry remains; keep the - // `black_box` launder on `self`/`last_frame` as defense-in-depth until - // `Stream` itself is celled. - let this: *mut Self = core::hint::black_box(core::ptr::from_mut(self)); - // SAFETY: `this` is the live `&mut self` payload; no other `&` to - // `*this` exists between here and the dispatch call. - if let Some(last_frame_ref) = unsafe { (*this).data_frame_queue.peek_last() } { - // Raw, opaque-provenance pointer for post-dispatch accesses. - let last_frame: *mut PendingFrame = - core::hint::black_box(core::ptr::from_mut(last_frame_ref)); - // SAFETY: helper for the pre-dispatch accesses below; `last_frame` - // is the unique tail slot in `self.data_frame_queue.data`, valid - // until the dispatch call (after which we re-`black_box` before - // every access — see note above). - macro_rules! lf { - () => { - // SAFETY: `last_frame` points at the live tail slot of - // `self.data_frame_queue`; provenance is re-laundered via - // `black_box` before each post-dispatch expansion so no - // other `&mut` to the slot is live here (see note). - unsafe { &mut *last_frame } - }; - } + // What happened to `bytes` against the queue's tail frame. The queue borrow ends before + // the replaced callback is dispatched: that JS can re-enter `writeStream` and reach + // `queue_frame()` for this same stream. + enum Merge { + /// No tail frame, or it is full: enqueue a new one. + NewFrame, + /// Everything was merged into the tail frame; its previous callback is concluded now + /// (one pending callback per frame — per-stream corking at the frame level). + Done(StrongOptional), + /// The first `usize` bytes were merged; the rest starts a new frame. + Partial(usize), + } + let merge = self.data_frame_queue.with_mut(|queue| { + let Some(last_frame) = queue.peek_last() else { + return Merge::NewFrame; + }; if bytes.is_empty() { // just merge the end_stream - lf!().end_stream = end_stream; - // we can only hold 1 callback at a time so we conclude the last one, and keep the last one as pending - // this is fine is like a per-stream CORKING in a frame level - let old_callback = core::mem::replace( - &mut lf!().callback, + last_frame.end_stream = end_stream; + return Merge::Done(core::mem::replace( + &mut last_frame.callback, StrongOptional::create(callback, &global_this), - ); + )); + } + if last_frame.len == 0 { + // we have an empty frame with means we can just use this frame with a new buffer + last_frame.buffer = Vec::with_capacity(MAX_PAYLOAD_SIZE_WITHOUT_FRAME); + } + let max_size = MAX_PAYLOAD_SIZE_WITHOUT_FRAME as u32; + let remaining = max_size - last_frame.len; + if remaining == 0 { + return Merge::NewFrame; + } + // ok we can cork frames + let consumed_len = (remaining as usize).min(bytes.len()); + last_frame.buffer.extend_from_slice(&bytes[0..consumed_len]); + last_frame.len += u32::try_from(consumed_len).expect("int cast"); + bun_output::scoped_log!(H2FrameParser, "dataFrame merged {}", consumed_len); + + client + .queued_data_size + .set(client.queued_data_size.get() + consumed_len as u64); + if consumed_len == bytes.len() { + last_frame.end_stream = end_stream; + return Merge::Done(core::mem::replace( + &mut last_frame.callback, + StrongOptional::create(callback, &global_this), + )); + } + // we keep the old callback because the new will be part of another frame + Merge::Partial(consumed_len) + }); + match merge { + Merge::Done(old_callback) => { if let Some(old_callback_value) = old_callback.get() { - // Escape `this` so a self-derived address is observable - // across the opaque JS call (belt-and-suspenders; either - // launder alone defeats the caching). - core::hint::black_box(this); client.dispatch_write_callback(old_callback_value); } drop(old_callback); return; } - if lf!().len == 0 { - // we have an empty frame with means we can just use this frame with a new buffer - lf!().buffer = Vec::with_capacity(MAX_PAYLOAD_SIZE_WITHOUT_FRAME); - } - let max_size = MAX_PAYLOAD_SIZE_WITHOUT_FRAME as u32; - let remaining = max_size - lf!().len; - if remaining > 0 { - // ok we can cork frames - let consumed_len = (remaining as usize).min(bytes.len()); - let merge = &bytes[0..consumed_len]; - lf!().buffer.extend_from_slice(merge); - lf!().len += u32::try_from(consumed_len).expect("int cast"); - bun_output::scoped_log!(H2FrameParser, "dataFrame merged {}", consumed_len); - - client - .queued_data_size - .set(client.queued_data_size.get() + consumed_len as u64); - // lets fallthrough if we still have some data - let more_data = &bytes[consumed_len..]; - if more_data.is_empty() { - lf!().end_stream = end_stream; - // we can only hold 1 callback at a time so we conclude the last one, and keep the last one as pending - // this is fine is like a per-stream CORKING in a frame level - let old_callback = core::mem::replace( - &mut lf!().callback, - StrongOptional::create(callback, &global_this), - ); - if let Some(old_callback_value) = old_callback.get() { - core::hint::black_box(this); - client.dispatch_write_callback(old_callback_value); - } - drop(old_callback); - return; - } - // we keep the old callback because the new will be part of another frame - // SAFETY: `this` is the live `&mut self`; no borrow of `*this` - // is held here (the `last_frame` raw pointer is unused past - // this point). - return unsafe { (*this).queue_frame(client, more_data, callback, end_stream) }; + Merge::Partial(consumed_len) => { + return self.queue_frame(client, &bytes[consumed_len..], callback, end_stream); } + Merge::NewFrame => {} } bun_output::scoped_log!( H2FrameParser, @@ -1807,7 +1686,7 @@ impl Stream { global_this.vm().deprecated_report_extra_memory(bytes.len()); } bun_output::scoped_log!(H2FrameParser, "dataFrame enqueued {}", frame.len); - self.data_frame_queue.enqueue(frame); + self.data_frame_queue.with_mut(|q| q.enqueue(frame)); client .outbound_queue_size .set(client.outbound_queue_size.get() + 1); @@ -1824,24 +1703,24 @@ impl Stream { ) -> Stream { Stream { id: stream_identifier, - state: StreamState::OPEN, - js_context: StrongOptional::empty(), - wait_for_trailers: false, - end_after_headers: false, - padding_strategy, - rst_code: 0, - stream_dependency: 0, - exclusive: false, + state: Cell::new(StreamState::OPEN), + js_context: JsCell::new(StrongOptional::empty()), + wait_for_trailers: Cell::new(false), + end_after_headers: Cell::new(false), + padding_strategy: Cell::new(padding_strategy), + rst_code: Cell::new(0), + stream_dependency: Cell::new(0), + exclusive: Cell::new(false), // RFC 7540 §5.3.5 / nghttp2 NGHTTP2_DEFAULT_WEIGHT: streams default to weight 16, // which is what stream.state.weight reports when no priority was signaled. - weight: 16, - window_size: initial_window_size as u64, - used_window_size: 0, - remote_window_size: remote_window_size as u64, - remote_used_window_size: 0, - signal: None, - reading_paused: false, - data_frame_queue: PendingQueue::default(), + weight: Cell::new(16), + window_size: Cell::new(initial_window_size as u64), + used_window_size: Cell::new(0), + remote_window_size: Cell::new(remote_window_size as u64), + remote_used_window_size: Cell::new(0), + signal: JsCell::new(None), + reading_paused: Cell::new(false), + data_frame_queue: JsCell::new(PendingQueue::default()), } } @@ -1853,55 +1732,54 @@ impl Stream { /// - CLOSED: stream is finished pub fn can_receive_data(&self) -> bool { matches!( - self.state, + self.state.get(), StreamState::IDLE | StreamState::OPEN | StreamState::HALF_CLOSED_LOCAL ) } pub fn can_send_data(&self) -> bool { matches!( - self.state, + self.state.get(), StreamState::IDLE | StreamState::OPEN | StreamState::HALF_CLOSED_REMOTE ) } - pub fn set_context(&mut self, value: JSValue, global_object: &JSGlobalObject) { - let old = core::mem::replace( - &mut self.js_context, - StrongOptional::create(value, global_object), - ); - drop(old); + pub fn set_context(&self, value: JSValue, global_object: &JSGlobalObject) { + self.js_context + .set(StrongOptional::create(value, global_object)); } pub fn get_identifier(&self) -> JSValue { self.js_context + .get() .get() .unwrap_or_else(|| JSValue::js_number(self.id as f64)) } - pub fn attach_signal(&mut self, parser: &H2FrameParser, signal: &mut AbortSignal) { + pub fn attach_signal(&self, parser: &H2FrameParser, signal: &AbortSignal) { // we need a stable pointer to know what signal points to what stream_id + parser - let mut signal_ref = Box::new(SignalRef { - signal: signal.ref_(), - parser: parser.ref_guard(), + let listener = bun_ptr::OwnedThis::new(SignalRef { + registration: JsCell::new(None), + // TODO: We should not need this ref counting here, since Parser owns Stream + parser: parser.new_ref(), stream_id: self.id, }); - // `signal_ref` is heap-allocated and outlives the listener registration - // (cleared via `detach` in `Drop for SignalRef`). - signal.listen(&raw mut *signal_ref); - self.signal = Some(signal_ref); + listener.registration.set(Some( + signal.listen_native(BackRef::from(listener.this_ptr())), + )); + self.signal.set(Some(listener)); } - pub fn detach_context(&mut self) { - self.js_context.deinit(); + pub fn detach_context(&self) { + self.js_context.with_mut(|c| c.deinit()); } - fn clean_queue(&mut self, client: &H2FrameParser) { + fn clean_queue(&self, client: &H2FrameParser) { bun_output::scoped_log!( H2FrameParser, "cleanQueue len: {} front: {} outboundQueueSize: {}", - self.data_frame_queue.len, - self.data_frame_queue.front, + self.data_frame_queue.get().len, + self.data_frame_queue.get().front, client.outbound_queue_size.get() ); @@ -1909,7 +1787,7 @@ impl Stream { // socket's ref and free `client` between iterations. Not during // finalize: refcount is already 0 and a ref/deref would re-destroy. let _keepalive = (!FINALIZING).then(|| client.keepalive()); - let mut queue = core::mem::take(&mut self.data_frame_queue); + let mut queue = self.data_frame_queue.replace(PendingQueue::default()); while let Some(item) = queue.dequeue() { let frame = item; let len = frame.slice().len(); @@ -1931,13 +1809,13 @@ impl Stream { } /// this can be called multiple times - pub fn free_resources(&mut self, client: &H2FrameParser) { + pub fn free_resources(&self, client: &H2FrameParser) { // The rewrite engine only sees inbound traffic, so a completed request would leave - // its engine entry as HalfClosedRemote and its legacy slot + Box behind forever — + // its engine entry as HalfClosedRemote and its legacy map entry behind forever — // one entry per request. Queue the id; the next rewrite_read batch evicts the engine - // entry and frees the legacy slot. Always deferred: every caller still holds - // `&mut Stream` into the map entry, and the engine cell may be mutably borrowed - // (stream completing synchronously inside receive()). + // entry and drops the legacy one. Always deferred: callers are mid-operation on this + // stream, and the engine cell may be mutably borrowed (stream completing + // synchronously inside receive()). if !FINALIZING { client .pending_engine_stream_closes @@ -1950,17 +1828,7 @@ impl Stream { } self.detach_context(); self.clean_queue::(client); - if let Some(signal) = self.signal.take() { - drop(signal); - } - } -} - -// Route AbortSignal callbacks through the trait — -// `bun_jsc::abort_signal::listen` expects `*mut C: AbortListener`. -impl AbortListener for SignalRef { - fn on_abort(&mut self, reason: JSValue) { - SignalRef::abort_listener(self, reason); + drop(self.signal.replace(None)); } } @@ -2106,7 +1974,7 @@ impl H2FrameParser { true } - pub(crate) fn abort_stream(&self, stream: &mut Stream, abort_reason: JSValue) { + pub(crate) fn abort_stream(&self, stream: &Stream, abort_reason: JSValue) { bun_output::scoped_log!( H2FrameParser, "HTTP_FRAME_RST_STREAM id: {} code: CANCEL", @@ -2125,11 +1993,10 @@ impl H2FrameParser { }; let _ = frame.write(&mut writer_stream, &self.frames_sent_legacy); let mut value: u32 = ErrorCode::CANCEL.0; - stream.rst_code = value; + stream.rst_code.set(value); value = value.swap_bytes(); let _ = writer_stream.write_all(&value.to_ne_bytes()); - let old_state = stream.state; - stream.state = StreamState::CLOSED; + let old_state = stream.state.replace(StreamState::CLOSED); let identifier = stream.get_identifier(); identifier.ensure_still_alive(); stream.free_resources::(self); @@ -2142,14 +2009,14 @@ impl H2FrameParser { let _ = self.write(&buffer); } - pub(crate) fn end_stream(&self, stream: &mut Stream, rst_code: ErrorCode) { + pub(crate) fn end_stream(&self, stream: &Stream, rst_code: ErrorCode) { bun_output::scoped_log!( H2FrameParser, "HTTP_FRAME_RST_STREAM id: {} code: {}", stream.id, rst_code.0 ); - if stream.state == StreamState::CLOSED { + if stream.state.get() == StreamState::CLOSED { return; } let mut buffer = [0u8; FrameHeader::BYTE_SIZE + 4]; @@ -2163,11 +2030,11 @@ impl H2FrameParser { }; let _ = frame.write(&mut writer_stream, &self.frames_sent_legacy); let mut value: u32 = rst_code.0; - stream.rst_code = value; + stream.rst_code.set(value); value = value.swap_bytes(); let _ = writer_stream.write_all(&value.to_ne_bytes()); - stream.state = StreamState::CLOSED; + stream.state.set(StreamState::CLOSED); let identifier = stream.get_identifier(); identifier.ensure_still_alive(); stream.free_resources::(self); @@ -2175,7 +2042,7 @@ impl H2FrameParser { self.dispatch_with_extra( JSH2FrameParser::Gc::onStreamEnd, identifier, - JSValue::js_number(stream.state as u8 as f64), + JSValue::js_number(stream.state.get() as u8 as f64), ); } else { self.dispatch_with_extra( @@ -2366,26 +2233,20 @@ impl H2FrameParser { let _ = self.write(&buffer); } - /// Armed across every JS dispatch wrapper AND every section that holds a `&mut Stream` - /// while user JS can run (property getters, iteration, string coercion), so - /// rewrite_read's deferred stream free (pending_engine_stream_closes) only runs at depth 0. + /// Armed across every JS dispatch wrapper AND every section that operates on a stream + /// while user JS can run (property getters, iteration, string coercion), so rewrite_read's + /// deferred stream eviction (pending_engine_stream_closes) only runs at depth 0. fn enter_dispatch(&self) -> DispatchGuard<'_> { self.dispatch_depth.set(self.dispatch_depth.get() + 1); DispatchGuard(&self.dispatch_depth) } - /// Reborrows a host fn's `*mut Stream` with the dispatch guard armed for the borrow's whole - /// lifetime: user JS the caller runs while holding it (option getters, `toString`) can - /// re-enter `read()` without freeing the stream. Use this instead of a raw `&mut *ptr`. - fn enter_stream_dispatch(&self, stream_ptr: *mut Stream) -> GuardedStream<'_> { + /// A host fn's stream with the dispatch guard armed for as long as it is held: user JS the + /// caller runs meanwhile (option getters, `toString`) can re-enter `read()` without the + /// stream being evicted mid-operation. + fn enter_stream_dispatch(&self, stream: Rc) -> GuardedStream<'_> { let _dispatch = self.enter_dispatch(); - GuardedStream { - // SAFETY: stream_ptr is the heap::alloc'd *mut Stream stored in self.streams; the - // map entry outlives the returned borrow because the armed dispatch depth defers - // the only free path (rewrite_read's pending close drain) while the guard is live. - stream: unsafe { &mut *stream_ptr }, - _dispatch, - } + GuardedStream { stream, _dispatch } } pub(crate) fn dispatch(&self, event: JSH2FrameParser::Gc, value: JSValue) { @@ -2485,31 +2346,55 @@ impl H2FrameParser { self.register_auto_flush(); } - fn set_corked(parser: Option>) -> Option> { - CORKED_H2.with(|c| ManuallyDrop::into_inner(c.replace(ManuallyDrop::new(parser)))) - } - - /// The parser holding the cork slot, if any. - fn corked() -> Option<*mut H2FrameParser> { - // SAFETY: JS-thread-local; peeks without moving the ref out. - CORKED_H2.with(|c| unsafe { (**c.as_ptr()).as_ref().map(RefPtr::as_ptr) }) + /// Take the cork slot for this parser, first flushing whoever holds it. + fn cork(&self) { + match CORKED_H2.get() { + // already corked + Some(corked) if std::ptr::eq(corked.as_const_ptr(), self) => return, + // force uncork: flush the foreign owner once (may run JS) + Some(corked) => { + corked.uncork(); + } + None => {} + } + // That flush can re-enter JS and cork this or yet another session before we resume. + match CORKED_H2.get() { + Some(corked) if std::ptr::eq(corked.as_const_ptr(), self) => return, + Some(corked) => corked.displace_cork(), + None => {} + } + debug_assert!(CORKED_H2.get().is_none()); + CORKED_H2.set(Some(self.this_ptr().into())); + let prev = self.cork_ref.replace(Some(self.new_ref())); + debug_assert!(prev.is_none(), "cork_ref already held"); + self.register_auto_flush(); + bun_output::scoped_log!(H2FrameParser, "cork {:p}", self); + CORK_OFFSET.with(|c| c.set(0)); } - fn cork(&self) { - if let Some(corked) = Self::corked() { - if std::ptr::eq(corked, self.as_ctx_ptr()) { - // already corked - return; + /// Give up the cork slot without writing (so without running JS): the corked bytes move to + /// the end of this parser's own `write_buffer` — where `uncork()`'s `_write` would have queued + /// them under backpressure — and its own flush (auto-flush stays registered) sends them. + fn displace_cork(&self) { + debug_assert!(self.is_corked()); + // Only ever non-empty between a producer's back-to-back chunks, never across sessions. + debug_assert!(BATCH_BUFFER.with_borrow(|b| b.is_empty())); + bun_output::scoped_log!(H2FrameParser, "displace cork {:p}", self); + let off = CORK_OFFSET.with(|c| c.get()) as usize; + if off != 0 { + CORK_BUFFER.with_borrow(|buf| { + let _ = self.write_buffer.with_mut(|wb| wb.write(&buf[0..off])); + }); + self.global().vm().deprecated_report_extra_memory(off); + if matches!(self.native_socket.get(), BunSocket::None) { + // Later `_write`s queue behind these bytes instead of overtaking them in JS. + self.has_nonnative_backpressure.set(true); } - // force uncork - // SAFETY: CORKED_H2 holds a ref()'d *mut H2FrameParser; valid until matching deref() in uncork - unsafe { (*corked.cast_const()).uncork() }; + self.register_auto_flush(); } - // cork - Self::set_corked(Some(self.ref_guard())); - self.register_auto_flush(); - bun_output::scoped_log!(H2FrameParser, "cork {:p}", self); + CORKED_H2.set(None); CORK_OFFSET.with(|c| c.set(0)); + drop(self.cork_ref.take()); } pub(crate) fn generic_flush(&self, mut socket: S) -> usize { @@ -2654,9 +2539,6 @@ impl H2FrameParser { let mut it = StreamResumableIterator::init(self); something_was_flushed = false; while let Some(stream) = it.next() { - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the - // map entry exists. Separate heap allocation from `self`, so no aliasing. - let stream = unsafe { &mut *stream }; // reach backpressure let result = stream.flush_queue(self, &mut written); match result { @@ -2845,11 +2727,9 @@ impl H2FrameParser { /// `resize()` the buffer: under this session's own transport writes, or when taking the /// cork slot first flushes another such session's corked bytes through its transport. fn stable_payload<'a>(&self, bytes: &'a [u8]) -> Cow<'a, [u8]> { - let foreign_cork_runs_js = || match Self::corked() { - Some(other) if !std::ptr::eq(other, self.as_ctx_ptr()) => { - CORK_OFFSET.with(|c| c.get()) > 0 - // SAFETY: CORKED_H2 holds a ref()'d parser until that parser's uncork(). - && unsafe { (*other).transport_write_runs_js() } + let foreign_cork_runs_js = || match CORKED_H2.get() { + Some(other) if !std::ptr::eq(other.as_const_ptr(), self) => { + CORK_OFFSET.with(|c| c.get()) > 0 && other.transport_write_runs_js() } _ => false, }; @@ -2861,10 +2741,10 @@ impl H2FrameParser { } fn uncork(&self) -> usize { - let Some(corked_ptr) = Self::corked() else { + let Some(corked) = CORKED_H2.get() else { return 0; }; - if !std::ptr::eq(corked_ptr, self.as_ctx_ptr()) { + if !std::ptr::eq(corked.as_const_ptr(), self) { // Another parser owns the cork slot; its own auto_flush / // on_native_writable will drain it. Draining it here writes to a // foreign fd from inside self's writable callback and tears down @@ -2874,9 +2754,11 @@ impl H2FrameParser { return 0; } self.unregister_auto_flush(); - bun_output::scoped_log!(H2FrameParser, "uncork {:p}", corked_ptr); - // The slot's ref on `self`, released once the corked bytes are written. - let _slot_ref = Self::set_corked(None); + bun_output::scoped_log!(H2FrameParser, "uncork {:p}", self); + CORKED_H2.set(None); + // The slot is free from here, so a `cork()` re-entered from `_write` below takes a fresh + // `cork_ref`; ours is released once the corked bytes are out. + let held = self.cork_ref.take(); // _write can re-enter JS (JS-stream-backed sockets, h2-over-h2 tunnels), // so no thread-local borrow may be held across it: move the corked bytes @@ -2894,6 +2776,7 @@ impl H2FrameParser { *b = data; } }); + drop(held); n } @@ -2901,19 +2784,10 @@ impl H2FrameParser { if self.auto_flusher.get().registered.get() { return; } - self.ref_(); - // R-2: inlined so the path is `&self` + extra `self.ref_()` (matches - // NodeHTTPResponse f1e506c8). `HasAutoFlusher` is now `&self` too. - debug_assert!(!self.auto_flusher.get().registered.get()); - self.auto_flusher.get().registered.set(true); - let ctx = NonNull::new(self.as_ctx_ptr().cast::()); - let found_existing = self - .global_this - .bun_vm() - .event_loop_mut() - .deferred_tasks - .post_task(ctx, on_auto_flush_trampoline); - debug_assert!(!found_existing); + let prev = self.auto_flush_ref.replace(Some(self.new_ref())); + debug_assert!(prev.is_none(), "auto_flush_ref already held"); + + AutoFlusher::register_deferred_microtask_with_type(self, self.global_this.bun_vm()); } fn unregister_auto_flush(&self) { @@ -2925,17 +2799,8 @@ impl H2FrameParser { if self.pending_header_compression_error.get() { return; } - debug_assert!(self.auto_flusher.get().registered.get()); - let ctx = NonNull::new(self.as_ctx_ptr().cast::()); - let removed = self - .global_this - .bun_vm() - .event_loop_mut() - .deferred_tasks - .unregister_task(ctx); - debug_assert!(removed); - self.auto_flusher.get().registered.set(false); - self.deref(); + AutoFlusher::unregister_deferred_microtask_with_type(self, self.global_this.bun_vm()); + drop(self.auto_flush_ref.take()); } /// A `write_maybe_corked` in `generic_write`/`generic_flush` returned a fatal @@ -3010,7 +2875,7 @@ impl H2FrameParser { // (detach -> unregister_auto_flush) see an unregistered flusher and // early-return instead of removing a map entry run() still owns. self.auto_flusher.get().registered.set(false); - self.deref(); + drop(self.auto_flush_ref.take()); // An empty write buffer here means a later write in the same flush() // cycle already drained the bytes the failing send left behind (racy // one-off errnos, e.g. macOS EPROTOTYPE) - the transport recovered. @@ -3038,12 +2903,24 @@ impl H2FrameParser { // map entry it owns. if self.auto_flusher.get().registered.get() { self.auto_flusher.get().registered.set(false); - self.deref(); + drop(self.auto_flush_ref.take()); } return false; } let _ = self.flush(); - // we will unregister ourselves when the buffer is empty + // flush() unregisters through uncork() once our cork is out. A parser whose cork was + // displaced (`displace_cork`) stays registered only for the bytes moved to write_buffer; + // once those are out too, nothing is left for this task. + if self.auto_flusher.get().registered.get() + && !self.is_corked() + && self.write_buffer.get().len() <= self.write_buffer_offset.get() + && !self.pending_header_compression_error.get() + && !self.transport_write_fatal.get() + { + self.auto_flusher.get().registered.set(false); + drop(self.auto_flush_ref.take()); + return false; + } true } @@ -3055,8 +2932,8 @@ impl H2FrameParser { // send_data()'s multi-frame path reaches here without having called cork(), and // prepending another session's corked frames to this one's batch sends them to // the wrong peer. uncork() clears CORKED_H2 before calling this, so None passes. - if let Some(corked) = Self::corked() - && !std::ptr::eq(corked, self.as_ctx_ptr()) + if let Some(corked) = CORKED_H2.get() + && !std::ptr::eq(corked.as_const_ptr(), self) { return; } @@ -3069,14 +2946,15 @@ impl H2FrameParser { } /// Send the accumulated multi-frame batch in one socket write. No-op when empty. - fn flush_batch_buffer(&self) { + /// `payload` is the send_data payload any `BatchSegment::Payload` refers to. + fn flush_batch_buffer(&self, payload: &[u8]) { // Take the Vecs out of the thread-locals before writing: _write can re-enter JS // (JS-stream-backed sockets), and a re-entrant send_data must not hit a borrowed // RefCell. The re-entrant call sees an empty batch and flushes independently. let mut data = BATCH_BUFFER.with_borrow_mut(core::mem::take); let mut segments = BATCH_SEGMENTS.with_borrow_mut(core::mem::take); if !segments.is_empty() { - self.flush_batch_vectored(&data, &segments); + self.flush_batch_vectored(&data, &segments, payload); } else if !data.is_empty() { let _ = self._write(&data); } @@ -3103,7 +2981,7 @@ impl H2FrameParser { /// Vectored flush for plain-TCP batches: frame headers from the batch scratch and /// payload slices straight from the caller's buffer, one writev. A partial write /// copies the unwritten tail into write_buffer, which engages backpressure. - fn flush_batch_vectored(&self, batch: &[u8], segments: &[BatchSegment]) { + fn flush_batch_vectored(&self, batch: &[u8], segments: &[BatchSegment], payload: &[u8]) { let mut total: usize = 0; let total_written = match self.native_socket.get() { BunSocket::Tcp(socket) | BunSocket::TcpWriteonly(socket) => BATCH_IOVECS @@ -3111,14 +2989,14 @@ impl H2FrameParser { iov.clear(); iov.reserve(segments.len()); for seg in segments { - let (ptr, len) = seg.raw_parts(batch); - if len == 0 { + let bytes = seg.bytes(batch, payload); + if bytes.is_empty() { continue; } - total += len; + total += bytes.len(); iov.push(bun_uws_sys::UsIoVec { - base: ptr.cast(), - len, + base: bytes.as_ptr().cast(), + len: bytes.len(), }); } if total == 0 { @@ -3132,10 +3010,7 @@ impl H2FrameParser { // to preserve order. let mut all: Vec = Vec::new(); for seg in segments { - let (ptr, len) = seg.raw_parts(batch); - // SAFETY: Batch ranges were recorded inside `batch`, and Ext slices are - // valid for the send_data call duration, which is still running. - all.extend_from_slice(unsafe { core::slice::from_raw_parts(ptr, len) }); + all.extend_from_slice(seg.bytes(batch, payload)); } let _ = self._write(&all); return; @@ -3146,13 +3021,12 @@ impl H2FrameParser { let mut skip = total_written; let mut buffered: usize = 0; for seg in segments { - let (ptr, len) = seg.raw_parts(batch); - if skip >= len { - skip -= len; + let bytes = seg.bytes(batch, payload); + if skip >= bytes.len() { + skip -= bytes.len(); continue; } - // SAFETY: same as the copy path above; skip < len - let rest = unsafe { core::slice::from_raw_parts(ptr.add(skip), len - skip) }; + let rest = &bytes[skip..]; skip = 0; let _ = self.write_buffer.with_mut(|wb| wb.write(rest)); buffered += rest.len(); @@ -3289,6 +3163,19 @@ impl H2FrameParser { } } +/// bun:internal-for-testing `h2AutoFlushRegistered(parser)`: whether the parser still has a +/// deferred auto-flush task registered. +#[bun_jsc::host_fn] +pub(crate) fn js_auto_flush_registered( + global: &JSGlobalObject, + frame: &CallFrame, +) -> JsResult { + let Some(parser) = frame.argument(0).as_class_ref::() else { + return Err(global.throw_invalid_arguments(format_args!("Expected an H2FrameParser"))); + }; + Ok(JSValue::from(parser.auto_flusher.get().registered.get())) +} + /// Trait to abstract over TLSSocket / TCPSocket for `generic_flush`/`generic_write`. pub(crate) trait NativeSocketWrite { fn write_maybe_corked(&mut self, buf: &[u8]) -> i32; @@ -3306,21 +3193,6 @@ impl NativeSocketWrite for &TCPSocket { } } -// ────────────────────────────────────────────────────────────────────────── -// R-2: `HasAutoFlusher` (which requires `fn auto_flusher(&mut self)`) is no -// longer implemented here — the deferred-task registration is inlined in -// `register_auto_flush` / `unregister_auto_flush` so the whole path is `&self`. -// The `DeferredRepeatingTask` trampoline that the trait would have generated. -// Body discharges its own preconditions; a safe `extern "C" fn` coerces to the -// `DeferredRepeatingTask` pointer at `post_task` (matches NodeHTTPResponse.rs). -// ────────────────────────────────────────────────────────────────────────── -extern "C" fn on_auto_flush_trampoline(ctx: *mut c_void) -> bool { - // SAFETY: `ctx` is the `*const H2FrameParser` registered by - // `register_auto_flush`; `DeferredTaskQueue::run` feeds it back unchanged - // on the JS thread. `on_auto_flush` takes `&self`. - unsafe { (*(ctx.cast_const().cast::())).on_auto_flush() } -} - // (`JsValueArrayPush` / `VmReportExtraMemory` shims removed — // `bun_jsc::JSValue::push` and `bun_jsc::VM::deprecated_report_extra_memory` // are inherent methods now.) @@ -3350,15 +3222,14 @@ impl H2FrameParser { bun_string_jsc::create_utf8_for_js(&global, payload) } - /// Returned *Stream is heap-allocated and stable for the lifetime of this H2FrameParser. - fn handle_received_stream_id(&self, stream_identifier: u32) -> Option<*mut Stream> { + fn handle_received_stream_id(&self, stream_identifier: u32) -> Option> { // connection stream if stream_identifier == 0 { return None; } // already exists - if let Some(stream) = self.streams.get().get(&stream_identifier).copied() { + if let Some(stream) = self.stream(stream_identifier) { return Some(stream); } @@ -3378,7 +3249,7 @@ impl H2FrameParser { } else { self.local_settings.get().initial_window_size }; - let stream = bun_core::heap::into_raw(Box::new(Stream::init( + let stream = Rc::new(Stream::init( stream_identifier, local_window_size, self.remote_settings @@ -3386,9 +3257,9 @@ impl H2FrameParser { .map(|s| s.initial_window_size) .unwrap_or(DEFAULT_WINDOW_SIZE as u32), self.padding_strategy.get(), - ))); + )); self.streams - .with_mut(|s| s.insert(stream_identifier, stream)); + .with_mut(|s| s.insert(stream_identifier, Rc::clone(&stream))); let Some(this_value) = self.strong_this.get().try_get() else { return Some(stream); @@ -3402,9 +3273,7 @@ impl H2FrameParser { let global = self.handlers.get().global(); // The callback runs arbitrary JS while `stream` is held (here and by every - // caller): arm the dispatch guard so a reentrant read() cannot free the box at - // depth 0. Bare guard, not enter_stream_dispatch — rst_stream reached from the - // callback takes its own `&mut` to this stream, so ours must wait for the return. + // caller): arm the dispatch guard so a reentrant read() cannot evict it at depth 0. let _dispatch = self.enter_dispatch(); // A top-level call of its own: a throwing `streamStart` is reported and // yields no stream object. Called bare (no scope of its own, so no @@ -3437,15 +3306,14 @@ impl H2FrameParser { self.sctx.with_mut(|m| { m.insert(stream_identifier, StrongOptional::create(returned, &global)); }); - self.enter_stream_dispatch(stream) - .set_context(returned, &global); + stream.set_context(returned, &global); } Some(stream) } fn to_writer(&self) -> DirectWriterStruct { DirectWriterStruct { - writer: bun_ptr::BackRef::new(self), + writer: BackRef::new(self), } } } @@ -3501,9 +3369,8 @@ impl H2FrameParser { if let Some(ctx) = self.sctx.get().get(&stream_id).and_then(|s| s.get()) { return ctx; } - if let Some(stream) = self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - return unsafe { (*stream).get_identifier() }; + if let Some(stream) = self.streams.get().get(&stream_id) { + return stream.get_identifier(); } JSValue::UNDEFINED } @@ -3602,23 +3469,15 @@ impl H2FrameParser { } }); // Streams whose legacy lifecycle finished since the last batch: evict the engine - // entry and free the legacy slot. free_resources already ran for these (it is the + // entry and drop the legacy one. free_resources already ran for these (it is the // only producer of this queue); duplicate ids are fine — remove() yields None. + // dispatch_depth == 0 means no caller below us on the stack is mid-operation on + // a stream (every such site arms enter_dispatch). if self.dispatch_depth.get() == 0 { self.pending_engine_stream_closes.with_mut(|v| { for id in v.drain(..) { engine.close_stream(id); - if let Some(stream) = self.streams.with_mut(|m| m.remove(&id)) { - // SAFETY: stream is the heap::alloc'd *mut Stream owned by the - // map entry just removed; free_resources ran when it was queued, - // dispatch_depth == 0 means no caller below us on the stack holds - // a `&mut Stream` across anything that can run user JS (every - // such site arms enter_dispatch), ids never repeat within a - // session, so this frees exactly once. - unsafe { - drop(bun_core::heap::take(stream)); - } - } + self.streams.with_mut(|m| m.remove(&id)); } }); } @@ -3806,11 +3665,11 @@ impl crate::api::h2::connection::Sink for H2FrameParser { // send window of streams opened before its SETTINGS arrived (a client's first request is // typically sent before the server's SETTINGS lands), then resume queued sends. let mut window_grew = false; - for (_, item) in self.streams.get().iter() { - // SAFETY: item is &*mut Stream from streams.iter(); the boxed Stream outlives the iteration - let stream = unsafe { &mut **item }; - if (settings.initial_window_size as u64) > stream.remote_window_size { - stream.remote_window_size = settings.initial_window_size as u64; + for stream in self.streams.get().values() { + if (settings.initial_window_size as u64) > stream.remote_window_size.get() { + stream + .remote_window_size + .set(settings.initial_window_size as u64); window_grew = true; } } @@ -3900,9 +3759,10 @@ impl crate::api::h2::connection::Sink for H2FrameParser { if stream_id == 0 { self.remote_window_size .set(self.remote_window_size.get() + increment as u64); - } else if let Some(stream) = self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - unsafe { (*stream).remote_window_size += increment as u64 }; + } else if let Some(stream) = self.streams.get().get(&stream_id) { + stream + .remote_window_size + .set(stream.remote_window_size.get() + increment as u64); } let _ = self.flush(); } @@ -3941,9 +3801,8 @@ impl crate::api::h2::connection::Sink for H2FrameParser { } fn is_stream_reading(&self, stream_id: u32) -> bool { - match self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - Some(stream) => unsafe { !(*stream).reading_paused }, + match self.streams.get().get(&stream_id) { + Some(stream) => !stream.reading_paused.get(), None => true, } } @@ -4027,9 +3886,8 @@ impl crate::api::h2::connection::Sink for H2FrameParser { fn on_headers_complete(&self, stream_id: u32, end_stream: bool, flags: u8) { // Bridge: the JS endAfterHeaders getter reads the legacy stream's end_after_headers flag. - if let Some(stream) = self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - unsafe { (*stream).end_after_headers = end_stream }; + if let Some(stream) = self.streams.get().get(&stream_id) { + stream.end_after_headers.set(end_stream); } // Materialize the accumulated block in a single native pass: the raw array, the // node-shaped headers object, and the sensitive list (a zero-field block yields an @@ -4037,15 +3895,9 @@ impl crate::api::h2::connection::Sink for H2FrameParser { let g = self.global(); let tuple = self.hdr_block.with_mut(|block| { self.hdr_meta.with_mut(|meta| { - let field_count = meta.len() / 2; - // `call_zero_is_throw` performs the exception-presence check the JSC - // validator requires at this FFI boundary (zero return == throw). - // SAFETY: block/meta are live Vec borrows for the call duration; the - // returned tuple is rooted by the conservative stack scan until - // dispatched below. - let v = bun_jsc::call_zero_is_throw(&g, || unsafe { - Bun__h2__materializeHeaders(&g, block.as_ptr(), meta.as_ptr(), field_count) - }); + // The returned tuple is rooted by the conservative stack scan until dispatched + // below. + let v = bun_jsc::h2_headers::materialize(&g, block, meta); block.clear(); meta.clear(); v @@ -4089,9 +3941,8 @@ impl crate::api::h2::connection::Sink for H2FrameParser { // state: remote-closed (6) on a stream whose local half is closed (5/7) is fully CLOSED (7), // mirroring the legacy handle_data/headers END_STREAM logic. let mut effective = state; - if let Some(stream) = self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - let legacy_state = unsafe { (*stream).state }; + if let Some(stream) = self.streams.get().get(&stream_id) { + let legacy_state = stream.state.get(); if state == 6 && matches!( legacy_state, @@ -4100,15 +3951,12 @@ impl crate::api::h2::connection::Sink for H2FrameParser { { effective = 7; } - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - unsafe { - (*stream).state = match effective { - 5 => StreamState::HALF_CLOSED_LOCAL, - 6 => StreamState::HALF_CLOSED_REMOTE, - 7 => StreamState::CLOSED, - _ => legacy_state, - }; - } + stream.state.set(match effective { + 5 => StreamState::HALF_CLOSED_LOCAL, + 6 => StreamState::HALF_CLOSED_REMOTE, + 7 => StreamState::CLOSED, + _ => legacy_state, + }); } let stream_ctx = self.rewrite_stream_ctx(stream_id); self.dispatch_with_extra( @@ -4121,9 +3969,8 @@ impl crate::api::h2::connection::Sink for H2FrameParser { // local side during the HEADERS dispatch, so the legacy send-close branch saw // OPEN and skipped its teardown — free the legacy context here or it (its Strong // JS stream root, and the engine's map entry) leaks per request. - if let Some(stream) = self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - unsafe { (*stream).free_resources::(self) }; + if let Some(stream) = self.stream(stream_id) { + stream.free_resources::(self); } // Release the per-stream JS context root so it can be collected (also done by // free_resources, but a stream may have no legacy entry). @@ -4161,13 +4008,9 @@ impl crate::api::h2::connection::Sink for H2FrameParser { // Bridge: mark the legacy stream closed with the rst code (capturing the prior state for // the aborted dispatch below). let mut old_state: u8 = StreamState::OPEN as u8; - if let Some(stream) = self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - unsafe { - old_state = (*stream).state as u8; - (*stream).state = StreamState::CLOSED; - (*stream).rst_code = code; - } + if let Some(stream) = self.streams.get().get(&stream_id) { + old_state = stream.state.replace(StreamState::CLOSED) as u8; + stream.rst_code.set(code); } let stream_ctx = self.rewrite_stream_ctx(stream_id); if code == crate::api::h2::wire::ErrorCode::Cancel.as_u32() { @@ -4186,11 +4029,10 @@ impl crate::api::h2::connection::Sink for H2FrameParser { JSValue::js_number(code as f64), ); } - // The reset closes the stream; free the legacy slot (queueing the engine eviction) + // The reset closes the stream; free the legacy entry (queueing the engine eviction) // and release its JS context root, mirroring the on_stream_end full-close path. - if let Some(stream) = self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - unsafe { (*stream).free_resources::(self) }; + if let Some(stream) = self.stream(stream_id) { + stream.free_resources::(self); } self.sctx.with_mut(|m| { m.remove(&stream_id); @@ -4205,7 +4047,7 @@ impl crate::api::h2::connection::Sink for H2FrameParser { // and the `BackRef` invariant (parser outlives this struct) holds by // construction. struct DirectWriterStruct { - writer: bun_ptr::BackRef, + writer: BackRef, } impl bun_io::Write for DirectWriterStruct { fn write_all(&mut self, data: &[u8]) -> bun_io::Result<()> { @@ -4569,13 +4411,11 @@ impl H2FrameParser { } } } - for (_, item) in this.streams.get().iter() { - // SAFETY: item is &*mut Stream from streams.iter(); the boxed Stream outlives the iteration - let stream = unsafe { &mut **item }; - if stream.used_window_size > window_size_value as u64 { + for stream in this.streams.get().values() { + if stream.used_window_size.get() > window_size_value as u64 { continue; } - stream.window_size = window_size_value as u64; + stream.window_size.set(window_size_value as u64); } Ok(JSValue::UNDEFINED) } @@ -4914,12 +4754,11 @@ impl H2FrameParser { return Err(global_object.throw(format_args!("Invalid stream id"))); } - let Some(stream) = this.streams.get().get(&stream_id).copied() else { + let Some(stream) = this.streams.get().get(&stream_id) else { return Err(global_object.throw(format_args!("Invalid stream id"))); }; - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - Ok(JSValue::from(unsafe { (*stream).end_after_headers })) + Ok(JSValue::from(stream.end_after_headers.get())) } #[bun_jsc::host_fn(method)] @@ -4942,18 +4781,20 @@ impl H2FrameParser { return Err(global_object.throw(format_args!("Invalid stream id"))); } - let Some(stream) = this.streams.get().get(&stream_id).copied() else { + let streams = this.streams.get(); + let Some(stream) = streams.get(&stream_id) else { return Err(global_object.throw(format_args!("Invalid stream id"))); }; - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the map entry exists - let stream = unsafe { &*stream }; - if let Some(signal_ref) = &stream.signal { - return Ok(JSValue::from(signal_ref.is_aborted())); + if let Some(listener) = stream.signal.get().as_ref() + && let Some(registration) = listener.registration.get().as_ref() + { + return Ok(JSValue::from(registration.signal().aborted())); } // closed with cancel = aborted Ok(JSValue::from( - stream.state == StreamState::CLOSED && stream.rst_code == ErrorCode::CANCEL.0, + stream.state.get() == StreamState::CLOSED + && stream.rst_code.get() == ErrorCode::CANCEL.0, )) } @@ -4977,22 +4818,20 @@ impl H2FrameParser { return Err(global_object.throw(format_args!("Invalid stream id"))); } - let Some(stream) = this.streams.get().get(&stream_id).copied() else { + let Some(stream) = this.stream(stream_id) else { return Err(global_object.throw(format_args!("Invalid stream id"))); }; - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the map entry exists - let stream = unsafe { &mut *stream }; let state = JSValue::create_empty_object(global_object, 6); state.put( global_object, b"localWindowSize", - JSValue::js_number(stream.window_size as f64), + JSValue::js_number(stream.window_size.get() as f64), ); state.put( global_object, b"state", - JSValue::js_number(stream.state as u8 as f64), + JSValue::js_number(stream.state.get() as u8 as f64), ); state.put( global_object, @@ -5013,7 +4852,7 @@ impl H2FrameParser { state.put( global_object, b"weight", - JSValue::js_number(stream.weight as f64), + JSValue::js_number(stream.weight.get() as f64), ); Ok(state) @@ -5039,11 +4878,11 @@ impl H2FrameParser { return Err(global_object.throw(format_args!("Invalid stream id"))); } - let Some(stream_ptr) = this.streams.get().get(&stream_id).copied() else { + let Some(stream_ptr) = this.stream(stream_id) else { return Err(global_object.throw(format_args!("Invalid stream id"))); }; - // The `options` getters below can run user JS while `stream` is borrowed. - let mut stream = this.enter_stream_dispatch(stream_ptr); + // The `options` getters below can run user JS while `stream` is held. + let stream = this.enter_stream_dispatch(stream_ptr); if !stream.can_send_data() && !stream.can_receive_data() { return Ok(JSValue::FALSE); @@ -5053,9 +4892,9 @@ impl H2FrameParser { return Err(global_object.throw(format_args!("Invalid priority"))); } - let mut weight = stream.weight; - let mut exclusive = stream.exclusive; - let mut parent_id = stream.stream_dependency; + let mut weight = stream.weight.get(); + let mut exclusive = stream.exclusive.get(); + let mut parent_id = stream.stream_dependency.get(); let mut silent = false; if let Some(js_weight) = options.get(global_object, "weight")? { if js_weight.is_number() { @@ -5103,17 +4942,16 @@ impl H2FrameParser { return Ok(JSValue::FALSE); } - stream.stream_dependency = parent_id; - stream.exclusive = exclusive; - stream.weight = weight; + stream.stream_dependency.set(parent_id); + stream.exclusive.set(exclusive); + stream.weight.set(weight); if !silent { - let stream_identifier = - UInt31WithReserved::init(stream.stream_dependency, stream.exclusive); + let stream_identifier = UInt31WithReserved::init(parent_id, exclusive); let priority = StreamPriority { stream_identifier: stream_identifier.to_uint32(), - weight: stream.weight as u8, + weight: weight as u8, }; let frame = FrameHeader { type_: FrameType::HTTP_FRAME_PRIORITY as u8, @@ -5174,7 +5012,7 @@ impl H2FrameParser { } } - let Some(stream) = this.streams.get().get(&stream_id).copied() else { + let Some(stream) = this.stream(stream_id) else { // Streams the legacy bookkeeping never registered (e.g. peer-initiated pushed streams // surfaced by the rewrite engine) get the RST_STREAM written directly. The frame is // built here rather than through the engine so this stays callable from inside an @@ -5200,8 +5038,7 @@ impl H2FrameParser { return Ok(JSValue::TRUE); }; - // SAFETY: stream is a *mut Stream from self.streams; valid while the map entry exists - this.end_stream(unsafe { &mut *stream }, ErrorCode(error_code)); + this.end_stream(&stream, ErrorCode(error_code)); Ok(JSValue::TRUE) } @@ -5242,7 +5079,7 @@ impl H2FrameParser { /// HALF_CLOSED_LOCAL, 7 = CLOSED, 0 = none) and whether `callback` was left to the caller. fn send_data( &self, - stream: &mut Stream, + stream: &Stream, payload: &[u8], callback: JSValue, options: SendDataOptions, @@ -5269,7 +5106,7 @@ impl H2FrameParser { let mut enqueued = false; let _keepalive = self.keepalive(); - let can_close = close && !stream.wait_for_trailers; + let can_close = close && !stream.wait_for_trailers.get(); if payload.is_empty() { // empty payload we still need to send a frame let data_header = FrameHeader { @@ -5305,7 +5142,8 @@ impl H2FrameParser { .min( (stream .remote_window_size - .saturating_sub(stream.remote_used_window_size)) + .get() + .saturating_sub(stream.remote_used_window_size.get())) as usize, ); let mut is_flow_control_limited = false; @@ -5326,7 +5164,7 @@ impl H2FrameParser { { // Preserve wire order: anything already batched goes out before the // queued remainder is flushed later by the drain path. - self.flush_batch_buffer(); + self.flush_batch_buffer(payload); enqueued = true; // write the full frame in memory and queue the frame // the callback will only be called after the last frame is sended @@ -5356,7 +5194,9 @@ impl H2FrameParser { max_size, payload_size ); - stream.remote_used_window_size += payload_size as u64; + stream + .remote_used_window_size + .set(stream.remote_used_window_size.get() + payload_size as u64); self.remote_used_window_size .set(self.remote_used_window_size.get() + payload_size as u64); self.note_engine_send_consumed(stream_id, payload_size as u64); @@ -5433,9 +5273,9 @@ impl H2FrameParser { off: header_off as u32, len: (batch.len() - header_off) as u32, }); - segs.push(BatchSegment::Ext { - ptr: slice.as_ptr(), - len: slice.len() as u32, + segs.push(BatchSegment::Payload { + off: (offset - size) as u32, + len: size as u32, }); }); } else { @@ -5445,7 +5285,7 @@ impl H2FrameParser { } } } - self.flush_batch_buffer(); + self.flush_batch_buffer(payload); } let mut settled_state: u8 = 0; @@ -5457,25 +5297,25 @@ impl H2FrameParser { self.dispatch_write_callback(callback); } if close { - if stream.wait_for_trailers { + if stream.wait_for_trailers.get() { self.dispatch(JSH2FrameParser::Gc::onWantTrailers, stream.get_identifier()); } else { let identifier = stream.get_identifier(); identifier.ensure_still_alive(); - if stream.state == StreamState::HALF_CLOSED_REMOTE { - stream.state = StreamState::CLOSED; + if stream.state.get() == StreamState::HALF_CLOSED_REMOTE { + stream.state.set(StreamState::CLOSED); stream.free_resources::(self); } else { - stream.state = StreamState::HALF_CLOSED_LOCAL; + stream.state.set(StreamState::HALF_CLOSED_LOCAL); } - settled_state = stream.state as u8; + settled_state = stream.state.get() as u8; if !(suppress_half_closed_local_dispatch - && stream.state == StreamState::HALF_CLOSED_LOCAL) + && stream.state.get() == StreamState::HALF_CLOSED_LOCAL) { self.dispatch_with_extra( JSH2FrameParser::Gc::onStreamEnd, identifier, - JSValue::js_number(stream.state as u8 as f64), + JSValue::js_number(stream.state.get() as u8 as f64), ); } } @@ -5506,15 +5346,13 @@ impl H2FrameParser { return Err(global_object.throw(format_args!("Invalid stream id"))); } - let Some(stream) = this.streams.get().get(&stream_id).copied() else { + let Some(stream) = this.stream(stream_id) else { return Err(global_object.throw(format_args!("Invalid stream id"))); }; - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the map entry exists - let stream = unsafe { &mut *stream }; - stream.wait_for_trailers = false; + stream.wait_for_trailers.set(false); let _ = this.send_data( - stream, + &stream, b"", JSValue::UNDEFINED, SendDataOptions { @@ -5556,12 +5394,11 @@ impl H2FrameParser { } let stream_id = stream_arg.to_u32(); let reading = reading_arg.to_boolean(); - let Some(stream) = this.streams.get().get(&stream_id).copied() else { + let Some(stream) = this.streams.get().get(&stream_id) else { // The stream already finished (or never reached the wire); nothing to backpressure. return Ok(JSValue::UNDEFINED); }; - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the map entry exists - unsafe { (*stream).reading_paused = !reading }; + stream.reading_paused.set(!reading); if reading { // Resumed: send the deferred WINDOW_UPDATE now. try_borrow: a resume issued from // inside a dispatch (the engine borrow is held by rewrite_read) is covered by the @@ -5673,12 +5510,12 @@ impl H2FrameParser { return Err(global_object.throw(format_args!("Invalid stream id"))); } - let Some(stream_ptr) = this.streams.get().get(&stream_id).copied() else { + let Some(stream_ptr) = this.stream(stream_id) else { return Err(global_object.throw(format_args!("Invalid stream id"))); }; // The header/sensitive-object getters and value coercions below can run user JS - // while `stream` is borrowed. - let mut stream = this.enter_stream_dispatch(stream_ptr); + // while `stream` is held. + let stream = this.enter_stream_dispatch(stream_ptr); let Some(headers_obj) = headers_arg.get_object() else { return Err(global_object.throw(format_args!("Expected headers to be an object"))); @@ -5791,7 +5628,7 @@ impl H2FrameParser { JSValue::js_number(ErrorCode::FRAME_SIZE_ERROR.0 as f64), ); let triggering_id = stream.id; - this.end_stream(&mut stream, ErrorCode::FRAME_SIZE_ERROR); + this.end_stream(&stream, ErrorCode::FRAME_SIZE_ERROR); this.send_go_away( triggering_id, ErrorCode::NO_ERROR, @@ -5963,16 +5800,16 @@ impl H2FrameParser { } let identifier = stream.get_identifier(); identifier.ensure_still_alive(); - if stream.state == StreamState::HALF_CLOSED_REMOTE { - stream.state = StreamState::CLOSED; + if stream.state.get() == StreamState::HALF_CLOSED_REMOTE { + stream.state.set(StreamState::CLOSED); stream.free_resources::(this); } else { - stream.state = StreamState::HALF_CLOSED_LOCAL; + stream.state.set(StreamState::HALF_CLOSED_LOCAL); } this.dispatch_with_extra( JSH2FrameParser::Gc::onStreamEnd, identifier, - JSValue::js_number(stream.state as u8 as f64), + JSValue::js_number(stream.state.get() as u8 as f64), ); Ok(JSValue::UNDEFINED) } @@ -6003,12 +5840,12 @@ impl H2FrameParser { } let close = close_arg.to_boolean(); - let Some(stream_ptr) = this.streams.get().get(&stream_id).copied() else { + let Some(stream_ptr) = this.stream(stream_id) else { return Err(global_object.throw(format_args!("Invalid stream id"))); }; // Coercing `data_arg` (a String subclass's toString) can run user JS while `stream` - // is borrowed. - let mut stream = this.enter_stream_dispatch(stream_ptr); + // is held. + let stream = this.enter_stream_dispatch(stream_ptr); if !stream.can_send_data() { this.dispatch_write_callback(callback_arg); return Ok(JSValue::FALSE); @@ -6051,7 +5888,7 @@ impl H2FrameParser { let payload = this.stable_payload(buffer.slice()); let (settled_state, callback_deferred) = this.send_data( - &mut stream, + &stream, &payload, callback_arg, SendDataOptions { @@ -6413,12 +6250,11 @@ impl H2FrameParser { return Err(global_object.throw(format_args!("Expected stream_id to be a number"))); } - let Some(stream) = this.streams.get().get(&stream_id_arg.to_u32()).copied() else { + let Some(stream) = this.streams.get().get(&stream_id_arg.to_u32()) else { return Err(global_object.throw(format_args!("Invalid stream id"))); }; - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - Ok(unsafe { (*stream).js_context.get() }.unwrap_or(JSValue::UNDEFINED)) + Ok(stream.js_context.get().get().unwrap_or(JSValue::UNDEFINED)) } #[bun_jsc::host_fn(method)] @@ -6460,9 +6296,8 @@ impl H2FrameParser { }); // Legacy path: also set on the legacy stream if it still exists (best-effort). - if let Some(stream) = this.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - unsafe { (*stream).set_context(context_arg, global_object) }; + if let Some(stream) = this.streams.get().get(&stream_id) { + stream.set_context(context_arg, global_object); } Ok(JSValue::UNDEFINED) } @@ -6486,8 +6321,7 @@ impl H2FrameParser { let mut _count: u32 = 0; let mut it = StreamResumableIterator::init(this); while let Some(stream) = it.next() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - let Some(value) = (unsafe { (*stream).js_context.get() }) else { + let Some(value) = stream.js_context.get().get() else { continue; }; this.handlers.get().vm.event_loop_mut().run_callback( @@ -6510,10 +6344,7 @@ impl H2FrameParser { // R-2: StreamResumableIterator stores a `ParentRef`; `streams` is `JsCell`-backed, // so the loop body can keep using `this` (`&Self`) directly. let mut it = StreamResumableIterator::init(this); - while let Some(stream_ptr) = it.next() { - // SAFETY: stream_ptr is a *mut Stream stored in self.streams (heap::alloc); valid for - // the lifetime of the entry. Separate heap allocation from `this`, so no aliasing. - let stream = unsafe { &mut *stream_ptr }; + while let Some(stream) = it.next() { // this is the oposite logic of emitErrorToallStreams, in this case we wanna to cancel this streams if this.is_server.get() { if stream.id % 2 == 0 { @@ -6522,10 +6353,9 @@ impl H2FrameParser { } else if stream.id % 2 != 0 { continue; } - if stream.state != StreamState::CLOSED { - let old_state = stream.state; - stream.state = StreamState::CLOSED; - stream.rst_code = ErrorCode::CANCEL.0; + if stream.state.get() != StreamState::CLOSED { + let old_state = stream.state.replace(StreamState::CLOSED); + stream.rst_code.set(ErrorCode::CANCEL.0); let identifier = stream.get_identifier(); identifier.ensure_still_alive(); stream.free_resources::(this); @@ -6552,7 +6382,7 @@ impl H2FrameParser { } // Like `goaway`: only numbers reach `to_u32` (it requires one), and the code is read - // once before any `&mut Stream` exists instead of once per stream inside the loop. + // once up front instead of once per stream inside the loop. if !error_arg.is_number() { return Err(global_object.throw(format_args!("Expected errorCode to be a number"))); } @@ -6561,13 +6391,10 @@ impl H2FrameParser { // R-2: StreamResumableIterator stores a `ParentRef`; `streams` is `JsCell`-backed, // so the loop body can keep using `this` (`&Self`) directly. let mut it = StreamResumableIterator::init(this); - while let Some(stream_ptr) = it.next() { - // SAFETY: stream_ptr is a *mut Stream stored in self.streams (heap::alloc); valid for - // the lifetime of the entry. Separate heap allocation from `this`, so no aliasing. - let stream = unsafe { &mut *stream_ptr }; - if stream.state != StreamState::CLOSED { - stream.state = StreamState::CLOSED; - stream.rst_code = rst_code; + while let Some(stream) = it.next() { + if stream.state.get() != StreamState::CLOSED { + stream.state.set(StreamState::CLOSED); + stream.rst_code.set(rst_code); let identifier = stream.get_identifier(); identifier.ensure_still_alive(); stream.free_resources::(this); @@ -6753,8 +6580,6 @@ impl H2FrameParser { let Some(stream) = this.handle_received_stream_id(stream_id) else { return Ok(JSValue::js_number(-1.0)); }; - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the map entry exists - let stream = unsafe { &mut *stream }; if !stream_ctx_arg.is_empty_or_undefined_or_null() && stream_ctx_arg.is_object() { @@ -6923,8 +6748,6 @@ impl H2FrameParser { let Some(stream) = this.handle_received_stream_id(stream_id) else { return Ok(JSValue::js_number(-1.0)); }; - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the map entry exists - let stream = unsafe { &mut *stream }; if !stream_ctx_arg.is_empty_or_undefined_or_null() && stream_ctx_arg.is_object() { @@ -6993,8 +6816,6 @@ impl H2FrameParser { let Some(stream) = this.handle_received_stream_id(stream_id) else { return Ok(JSValue::js_number(-1.0)); }; - // SAFETY: stream is a *mut Stream from self.streams (heap::alloc); valid while the map entry exists - let stream = unsafe { &mut *stream }; if !stream_ctx_arg.is_empty_or_undefined_or_null() && stream_ctx_arg.is_object() { @@ -7011,8 +6832,8 @@ impl H2FrameParser { let Some(stream_ptr) = this.handle_received_stream_id(stream_id) else { return Ok(JSValue::js_number(-1.0)); }; - // The `options` getters below can run user JS while `stream` is borrowed. - let mut stream = this.enter_stream_dispatch(stream_ptr); + // The `options` getters below can run user JS while `stream` is held. + let stream = this.enter_stream_dispatch(stream_ptr); if !stream_ctx_arg.is_empty_or_undefined_or_null() && stream_ctx_arg.is_object() { stream.set_context(stream_ctx_arg, global_object); } @@ -7027,30 +6848,30 @@ impl H2FrameParser { if callframe.arguments_count() > 4 && !options_arg.is_empty_or_undefined_or_null() { let options = options_arg; if !options.is_object() { - stream.state = StreamState::CLOSED; - stream.rst_code = ErrorCode::INTERNAL_ERROR.0; + stream.state.set(StreamState::CLOSED); + stream.rst_code.set(ErrorCode::INTERNAL_ERROR.0); this.dispatch_with_extra( JSH2FrameParser::Gc::onStreamError, stream.get_identifier(), - JSValue::js_number(stream.rst_code as f64), + JSValue::js_number(stream.rst_code.get() as f64), ); return Ok(JSValue::js_number(stream_id as f64)); } if let Some(padding_js) = options.get(global_object, "paddingStrategy")? { if padding_js.is_number() { - stream.padding_strategy = match padding_js.to_u32() { + stream.padding_strategy.set(match padding_js.to_u32() { 1 => PaddingStrategy::Aligned, 2 => PaddingStrategy::Max, _ => PaddingStrategy::None, - }; + }); } } if let Some(trailes_js) = options.get(global_object, "waitForTrailers")? { if trailes_js.is_boolean() { wait_for_trailers = trailes_js.as_boolean(); - stream.wait_for_trailers = wait_for_trailers; + stream.wait_for_trailers.set(wait_for_trailers); } } @@ -7088,7 +6909,7 @@ impl H2FrameParser { if exclusive_js.is_boolean() { if exclusive_js.as_boolean() { exclusive = true; - stream.exclusive = true; + stream.exclusive.set(true); has_priority = true; } } else { @@ -7105,16 +6926,18 @@ impl H2FrameParser { has_priority = true; parent = parent_js.to_int32(); if parent <= 0 || parent as u32 > MAX_STREAM_ID { - stream.state = StreamState::CLOSED; - stream.rst_code = ErrorCode::INTERNAL_ERROR.0; + stream.state.set(StreamState::CLOSED); + stream.rst_code.set(ErrorCode::INTERNAL_ERROR.0); this.dispatch_with_extra( JSH2FrameParser::Gc::onStreamError, stream.get_identifier(), - JSValue::js_number(stream.rst_code as f64), + JSValue::js_number(stream.rst_code.get() as f64), ); return Ok(JSValue::js_number(stream.id as f64)); } - stream.stream_dependency = u32::try_from(parent).expect("int cast"); + stream + .stream_dependency + .set(u32::try_from(parent).expect("int cast")); } else { return Err(global_object.throw_invalid_argument_type_value( b"options.parent", @@ -7129,16 +6952,16 @@ impl H2FrameParser { has_priority = true; weight = weight_js.to_int32(); if weight < 1 || weight > u8::MAX as i32 { - stream.state = StreamState::CLOSED; - stream.rst_code = ErrorCode::INTERNAL_ERROR.0; + stream.state.set(StreamState::CLOSED); + stream.rst_code.set(ErrorCode::INTERNAL_ERROR.0); this.dispatch_with_extra( JSH2FrameParser::Gc::onStreamError, stream.get_identifier(), - JSValue::js_number(stream.rst_code as f64), + JSValue::js_number(stream.rst_code.get() as f64), ); return Ok(JSValue::js_number(stream_id as f64)); } - stream.weight = u16::try_from(weight).expect("int cast"); + stream.weight.set(u16::try_from(weight).expect("int cast")); } else { return Err(global_object.throw_invalid_argument_type_value( b"options.weight", @@ -7148,31 +6971,31 @@ impl H2FrameParser { } if weight < 1 || weight > u8::MAX as i32 { - stream.state = StreamState::CLOSED; - stream.rst_code = ErrorCode::INTERNAL_ERROR.0; + stream.state.set(StreamState::CLOSED); + stream.rst_code.set(ErrorCode::INTERNAL_ERROR.0); this.dispatch_with_extra( JSH2FrameParser::Gc::onStreamError, stream.get_identifier(), - JSValue::js_number(stream.rst_code as f64), + JSValue::js_number(stream.rst_code.get() as f64), ); return Ok(JSValue::js_number(stream_id as f64)); } - stream.weight = u16::try_from(weight).expect("int cast"); + stream.weight.set(u16::try_from(weight).expect("int cast")); } if let Some(signal_arg) = options.get(global_object, "signal")? { - if let Some(signal_ptr) = AbortSignal::from_js(signal_arg) { - // SAFETY: `from_js` returns a live *mut AbortSignal owned by JSC; rooted via `signal_arg` on the stack. - let signal_ = unsafe { &mut *signal_ptr }; - if signal_.aborted() { - stream.state = StreamState::IDLE; - let wrapped = - Bun__wrapAbortError(global_object, signal_.js_reason(global_object)); - this.abort_stream(&mut stream, wrapped); + if let Some(signal) = AbortSignal::ref_from_js(signal_arg) { + if signal.aborted() { + stream.state.set(StreamState::IDLE); + let wrapped = AbortSignal::wrap_abort_error( + global_object, + signal.js_reason(global_object), + ); + this.abort_stream(&stream, wrapped); return Ok(JSValue::js_number(stream_id as f64)); } - stream.attach_signal(this, signal_); + stream.attach_signal(this, &signal); } else { return Err(global_object.throw_invalid_argument_type_value( b"options.signal", @@ -7185,13 +7008,13 @@ impl H2FrameParser { // too much memory being use if this.is_over_session_memory_limit() { - stream.state = StreamState::CLOSED; - stream.rst_code = ErrorCode::ENHANCE_YOUR_CALM.0; + stream.state.set(StreamState::CLOSED); + stream.rst_code.set(ErrorCode::ENHANCE_YOUR_CALM.0); this.rejected_streams.set(this.rejected_streams.get() + 1); this.dispatch_with_extra( JSH2FrameParser::Gc::onStreamError, stream.get_identifier(), - JSValue::js_number(stream.rst_code as f64), + JSValue::js_number(stream.rst_code.get() as f64), ); if this.rejected_streams.get() >= this.max_rejected_streams.get() { let global = this.handlers.get().global(); @@ -7221,8 +7044,8 @@ impl H2FrameParser { if this.max_send_header_block_length.get() != 0 && encoded_size > this.max_send_header_block_length.get() as usize { - stream.state = StreamState::CLOSED; - stream.rst_code = ErrorCode::REFUSED_STREAM.0; + stream.state.set(StreamState::CLOSED); + stream.rst_code.set(ErrorCode::REFUSED_STREAM.0); this.dispatch_with_2_extra( JSH2FrameParser::Gc::onFrameError, @@ -7234,7 +7057,7 @@ impl H2FrameParser { this.dispatch_with_extra( JSH2FrameParser::Gc::onStreamError, stream.get_identifier(), - JSValue::js_number(stream.rst_code as f64), + JSValue::js_number(stream.rst_code.get() as f64), ); return Ok(JSValue::js_number(stream_id as f64)); } @@ -7391,10 +7214,10 @@ impl H2FrameParser { } if end_stream { - stream.end_after_headers = true; + stream.end_after_headers.set(true); if wait_for_trailers { - stream.state = StreamState::HALF_CLOSED_LOCAL; + stream.state.set(StreamState::HALF_CLOSED_LOCAL); this.dispatch(JSH2FrameParser::Gc::onWantTrailers, stream.get_identifier()); return Ok(JSValue::js_number(stream_id as f64)); } @@ -7409,19 +7232,19 @@ impl H2FrameParser { // count) until socket close. let identifier = stream.get_identifier(); identifier.ensure_still_alive(); - if stream.state == StreamState::HALF_CLOSED_REMOTE { - stream.state = StreamState::CLOSED; + if stream.state.get() == StreamState::HALF_CLOSED_REMOTE { + stream.state.set(StreamState::CLOSED); stream.free_resources::(this); } else { - stream.state = StreamState::HALF_CLOSED_LOCAL; + stream.state.set(StreamState::HALF_CLOSED_LOCAL); } this.dispatch_with_extra( JSH2FrameParser::Gc::onStreamEnd, identifier, - JSValue::js_number(stream.state as u8 as f64), + JSValue::js_number(stream.state.get() as u8 as f64), ); } else { - stream.wait_for_trailers = wait_for_trailers; + stream.wait_for_trailers.set(wait_for_trailers); } if silent { @@ -7512,25 +7335,23 @@ impl H2FrameParser { this.detach_native_socket(); if let Some(socket) = socket_js.as_class_this_ptr::() { bun_output::scoped_log!(H2FrameParser, "TLSSocket attached"); - this.native_socket.attach( + this.native_socket.set(this.attach_to_native_socket( socket, - this.ref_guard(), BunSocket::Tls, BunSocket::TlsWriteonly, WriteonlySocketRef::Tls, - ); + )); // if we started with non native and go to native we now control the backpressure internally this.has_nonnative_backpressure.set(false); let _ = this.flush(); } else if let Some(socket) = socket_js.as_class_this_ptr::() { bun_output::scoped_log!(H2FrameParser, "TCPSocket attached"); - this.native_socket.attach( + this.native_socket.set(this.attach_to_native_socket( socket, - this.ref_guard(), BunSocket::Tcp, BunSocket::TcpWriteonly, WriteonlySocketRef::Tcp, - ); + )); // if we started with non native and go to native we now control the backpressure internally this.has_nonnative_backpressure.set(false); let _ = this.flush(); @@ -7538,8 +7359,47 @@ impl H2FrameParser { Ok(JSValue::UNDEFINED) } + /// `attach_native_callback` stores a ref on this parser in the socket; the matching + /// release happens in `NewSocket::detach_native_callback`. When the socket already has a + /// native callback attached we fall back to write-only mode and hold a + /// ref on the socket itself (`writeonly_socket_ref`), released by + /// `detach_native_socket`. + fn attach_to_native_socket( + &self, + socket: ThisPtr>, + attached: fn(BackRef, bun_ptr::Root>) -> BunSocket, + writeonly: fn(BackRef, bun_ptr::Root>) -> BunSocket, + writeonly_ref: fn(RefPtr>) -> WriteonlySocketRef, + ) -> BunSocket { + // BACKREF: `socket` is the live `m_ctx` of the JS wrapper rooted by the + // caller's `socket_js`; it strictly outlives the returned `BunSocket` via the + // attach/detach refcount protocol (see `BunSocket` docs). + if socket.has_native_callback() { + let prev = self + .writeonly_socket_ref + .replace(Some(writeonly_ref(RefPtr::from_this(socket)))); + debug_assert!(prev.is_none(), "writeonly_socket_ref already held"); + writeonly(socket.into()) + } else { + let attached_now = socket.attach_native_callback(NativeCallbacks::H2(self.new_ref())); + debug_assert!(attached_now); + attached(socket.into()) + } + } + pub(crate) fn detach_native_socket(&self) { - self.native_socket.detach(); + let native_socket = self.native_socket.replace(BunSocket::None); + + match native_socket { + // BackRef invariant: socket kept alive by attach_native_callback; this is the matching detach. + BunSocket::Tcp(socket) => socket.detach_native_callback(), + BunSocket::Tls(socket) => socket.detach_native_callback(), + // Writeonly socket was ref'd on attach; this is the matching release. + BunSocket::TcpWriteonly(_) | BunSocket::TlsWriteonly(_) => { + drop(self.writeonly_socket_ref.take()); + } + BunSocket::None => {} + } } /// Saturating read of a numeric session limit into its u32 field. @@ -7572,13 +7432,15 @@ impl H2FrameParser { } let handlers = Handlers::from_js(global_object, handler_js, this_value)?; - let init = H2FrameParser { + let this = RefPtr::new_cyclic(|self_ref| H2FrameParser { + self_ref, ref_count: bun_ptr::RefCount::init(), - native_keepalives: Cell::new(0), + keepalive_refs: JsCell::new(Vec::new()), handlers: JsCell::new(handlers), global_this: GlobalRef::from(global_object), strong_this: JsCell::new(JsRef::empty()), - native_socket: NativeSocket::default(), + native_socket: Cell::new(BunSocket::None), + writeonly_socket_ref: Cell::new(None), local_settings: Cell::new(FullSettingsPayload::default()), explicit_settings: Cell::new(0), custom_settings: JsCell::new(Vec::new()), @@ -7630,6 +7492,8 @@ impl H2FrameParser { engine_frames_sent: Cell::new(0), tx_tracker: Cell::new(TxFrameTracker::default()), auto_flusher: JsCell::new(AutoFlusher::default()), + auto_flush_ref: Cell::new(None), + cork_ref: Cell::new(None), padding_strategy: Cell::new(PaddingStrategy::None), engine: core::cell::RefCell::new(None), rewrite_tail: JsCell::new(Vec::new()), @@ -7637,57 +7501,47 @@ impl H2FrameParser { sctx: JsCell::new(BunHashMap::default()), hdr_block: JsCell::new(Vec::new()), hdr_meta: JsCell::new(Vec::new()), - }; - let this: *mut H2FrameParser = if ENABLE_ALLOCATOR_POOL { - POOL.with_borrow_mut(|pool| { - let pool = pool.get_or_insert_with(|| { - // SAFETY: `new_boxed` returns a `Box::leak`ed, fully - // initialized allocation; `from_raw` reclaims that exact - // pointer back into an owning `Box`. `ManuallyDrop` is - // `repr(transparent)` over `T`, so the pointer cast is a - // layout no-op. - unsafe { - Box::from_raw( - H2FrameParserHiveAllocator::new_boxed() - .as_ptr() - .cast::>(), - ) - } - }); - pool.get_init(init).as_ptr() - }) - } else { - bun_core::heap::into_raw(Box::new(init)) - }; - // The remaining `?` sites below may throw a JS exception; the guard - // drops `this` (its `Drop` detaches any socket attached below) and - // returns the slot to the pool / frees the Box. Defused on success. - let guard = scopeguard::guard(this, Self::release); - // SAFETY: `this` was just allocated above; unique ownership, non-null. - // R-2: deref as shared — every method below takes `&self`. - let this_ref = unsafe { &*this }; + }); + // The remaining `?` sites may throw a JS exception; the parser is freed on that path. + match Self::configure(&this, global_object, options, this_value, context_obj) { + Ok(()) => Ok(this.into_raw()), + Err(err) => { + // Give back the socket / cork slot / auto-flush refs `configure` may have handed + // out so this release is the last one. + this.detach(); + debug_assert!(this.ref_count.has_one_ref()); + drop(this); + Err(err) + } + } + } + fn configure( + this_ref: &H2FrameParser, + global_object: &JSGlobalObject, + options: JSValue, + this_value: JSValue, + context_obj: JSValue, + ) -> JsResult<()> { // check if socket is provided, and if it is a valid native socket if let Some(socket_js) = options.get(global_object, "native")? { if let Some(socket) = socket_js.as_class_this_ptr::() { bun_output::scoped_log!(H2FrameParser, "TLSSocket attached"); - this_ref.native_socket.attach( + this_ref.native_socket.set(this_ref.attach_to_native_socket( socket, - this_ref.ref_guard(), BunSocket::Tls, BunSocket::TlsWriteonly, WriteonlySocketRef::Tls, - ); + )); let _ = this_ref.flush(); } else if let Some(socket) = socket_js.as_class_this_ptr::() { bun_output::scoped_log!(H2FrameParser, "TCPSocket attached"); - this_ref.native_socket.attach( + this_ref.native_socket.set(this_ref.attach_to_native_socket( socket, - this_ref.ref_guard(), BunSocket::Tcp, BunSocket::TcpWriteonly, WriteonlySocketRef::Tcp, - ); + )); let _ = this_ref.flush(); } } @@ -7816,7 +7670,7 @@ impl H2FrameParser { this_ref.has_nonnative_backpressure.set(true); this_ref.send_preface_and_settings(); } - Ok(scopeguard::ScopeGuard::into_inner(guard)) + Ok(()) } #[bun_jsc::host_fn(method)] @@ -7829,9 +7683,7 @@ impl H2FrameParser { // so the loop body can keep using `this` (`&Self`) directly. let mut it = StreamResumableIterator::init(this); while let Some(stream) = it.next() { - // SAFETY: stream is *mut Stream from self.streams; valid until freed below / map - // cleared. `stream` points into a disjoint Box. - unsafe { (*stream).free_resources::(this) }; + stream.free_resources::(this); } this.detach(); if let Some(this_value) = this.strong_this.get().try_get() { @@ -7849,8 +7701,8 @@ impl H2FrameParser { self.unregister_auto_flush(); self.detach_native_socket(); - // Free the allocation, not just the length: detach() is reachable from JS without a - // following `deinit`, so the capacity must be released here. + // Free the allocation, not just the length: detach() is reachable from JS long before + // the parser is dropped, so the capacity must be released here. self.write_buffer.with_mut(|wb| wb.clear_and_free()); self.tx_tracker.set(TxFrameTracker::default()); // Drop every per-stream JS context root; the parser is detaching. @@ -7875,15 +7727,37 @@ impl H2FrameParser { fn release_refs_stranded_by_exit(&self) { // `uncork()` would `_write()` the corked bytes, which re-enters JS on a non-native // socket; the process is exiting, so drop them and just release the slot's ref. - if Self::corked() == Some(self.as_ctx_ptr()) { + if self.is_corked() { + CORKED_H2.set(None); CORK_OFFSET.with(|c| c.set(0)); - Self::set_corked(None); + drop(self.cork_ref.take()); } - // Removes the deferred task (its ctx is `self`) and releases the ref it holds. - self.unregister_auto_flush(); - let stranded = self.native_keepalives.replace(0); - for _ in 0..stranded { - self.deref(); + // Removes the deferred task and releases the ref it holds, even with a session error + // latched for it to report: nothing will run it. + if self.auto_flusher.get().registered.get() { + AutoFlusher::unregister_deferred_microtask_with_type(self, self.global_this.bun_vm()); + drop(self.auto_flush_ref.take()); + } + // Each stranded `Keepalive` guard's ref will never be popped (its frame cannot resume); + // release them here. + drop(self.keepalive_refs.replace(Vec::new())); + } + + /// Runs before the JS wrapper's ref is dropped. + pub(crate) fn finalize(&self) { + bun_output::scoped_log!(H2FrameParser, "finalize"); + self.strong_this.set(JsRef::empty()); + if VirtualMachine::get().is_shutting_down() { + // Free the streams first: `free_resources` releases the refs their signals hold. + // The map is emptied so `Drop` won't visit them again. + let streams = self.streams.replace(BunHashMap::default()); + for stream in streams.values() { + stream.free_resources::(self); + } + drop(streams); + // Then the refs of frames/tasks that will never run again, so the wrapper's + // release can actually reach zero and drop the parser. + self.release_refs_stranded_by_exit(); } } } @@ -7891,43 +7765,22 @@ impl H2FrameParser { impl Drop for H2FrameParser { fn drop(&mut self) { bun_output::scoped_log!(H2FrameParser, "deinit"); + // An attached socket holds a ref on the parser, so it cannot be attached here; nothing + // below (`detach_native_socket` → `on_native_close` in particular) may take a new ref. + debug_assert!(!matches!( + self.native_socket.get(), + BunSocket::Tls(_) | BunSocket::Tcp(_) + )); self.detach(); + self.strong_this.set(JsRef::empty()); + // Note: take the map out first so `self` is free for // `free_resources(self)` while we walk the entries. let streams = self.streams.replace(BunHashMap::default()); - for (_, item) in streams.iter() { - let stream = *item; - // SAFETY: stream is *mut Stream from self.streams; this is final teardown, freed exactly once via heap::take - unsafe { - (*stream).free_resources::(self); - drop(bun_core::heap::take(stream)); - } + for stream in streams.values() { + stream.free_resources::(self); } drop(streams); } } - -impl H2FrameParser { - pub(crate) fn finalize(&self) { - bun_output::scoped_log!(H2FrameParser, "finalize"); - self.strong_this.set(JsRef::empty()); - if VirtualMachine::get().is_shutting_down() { - // Free the streams first: `free_resources` releases the refs their signals hold. - // The map is emptied so a later `Drop` won't double-free. - let streams = self.streams.replace(BunHashMap::default()); - for (_, item) in streams.iter() { - let stream = *item; - // SAFETY: map has been emptied; each entry is freed exactly once. - unsafe { - (*stream).free_resources::(self); - drop(bun_core::heap::take(stream)); - } - } - drop(streams); - // Then the refs of frames/tasks that will never run again, so the - // wrapper's deref can actually reach zero and run `Drop`. - self.release_refs_stranded_by_exit(); - } - } -} diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index b900377e6431..d5dc8ad9e3a1 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -568,9 +568,13 @@ impl NewSocket { + ssl_cost } + pub(crate) fn has_native_callback(&self) -> bool { + !matches!(self.native_callback.get(), NativeCallbacks::None) + } + /// On `false` the rejected `callback` (and the ref it holds) is dropped. pub(crate) fn attach_native_callback(&self, callback: NativeCallbacks) -> bool { - if !matches!(self.native_callback.get(), NativeCallbacks::None) { + if self.has_native_callback() { return false; } self.native_callback.set(callback); diff --git a/src/runtime/webcore.rs b/src/runtime/webcore.rs index 15079c660771..571941c453e9 100644 --- a/src/runtime/webcore.rs +++ b/src/runtime/webcore.rs @@ -71,7 +71,7 @@ pub struct AutoFlusher { pub(crate) registered: core::cell::Cell, } -/// Implemented below for `FileSink` and `HTTPServerWritable<_, _>`. +/// Implemented below for `FileSink`, `HTTPServerWritable<_, _>` and `H2FrameParser`. pub trait HasAutoFlusher: Sized { fn auto_flusher(&self) -> &AutoFlusher; /// `Type.onAutoFlush` — `DeferredRepeatingTask` ABI after `@ptrCast` @@ -199,6 +199,20 @@ impl HasAutoFlusher for streams::HTTPServerWritable { } } +impl HasAutoFlusher for crate::api::h2_frame_parser_body::H2FrameParser { + #[inline] + fn auto_flusher(&self) -> &AutoFlusher { + self.auto_flusher.get() + } + /// # Safety + /// See [`HasAutoFlusher::on_auto_flush`]. + unsafe fn on_auto_flush(this: *mut Self) -> bool { + // SAFETY: `this` is the `&H2FrameParser` registered by `register_auto_flush`, kept alive + // by the parser's `auto_flush_ref` until it unregisters; only `&self` is formed. + unsafe { (*this).on_auto_flush() } + } +} + #[path = "webcore/headers_ref.rs"] pub(crate) mod headers_ref; diff --git a/test/js/node/http2/node-http2.test.js b/test/js/node/http2/node-http2.test.js index 2ed62caf61f6..59bdbeeaa8d9 100644 --- a/test/js/node/http2/node-http2.test.js +++ b/test/js/node/http2/node-http2.test.js @@ -1996,6 +1996,222 @@ it("http2 session.goaway() opaqueData survives re-entrant buffer detach over a J expect(exitCode).toBe(0); }, 15_000); +it("http2 sessions over JS Duplexes whose _write re-enters other sessions keep the cork slot consistent", async () => { + // The transport's _write runs synchronously under a parser's uncork/flush, and user code + // there can serialize a frame on this or another session, re-taking the thread's cork slot + // mid-handover. Debug builds assert the slot's ref is never double-installed (stderr stays + // empty); every session's byte stream must stay whole frames; and the sessions tear down. + const fixture = ` + const http2 = require("node:http2"); + const { Duplex } = require("node:stream"); + const { heapStats } = require("bun:jsc"); + const count = () => heapStats().objectTypeCounts.H2FrameParser ?? 0; + const connected = session => + new Promise((resolve, reject) => { + session.once("connect", resolve); + session.once("error", reject); + session.once("close", () => reject(new Error("closed before connect"))); + }); + const tick = () => new Promise(r => setImmediate(r)); + // preface, then whole frames only (9-byte header + declared length, known type) + const assertWellFormed = (name, chunks, min) => { + const all = Buffer.concat(chunks); + const preface = Buffer.from("PRI * HTTP/2.0" + String.fromCharCode(13, 10, 13, 10) + "SM" + String.fromCharCode(13, 10, 13, 10)); + if (!all.subarray(0, preface.length).equals(preface)) throw new Error(name + ": bad preface"); + let off = preface.length; + let frames = 0; + while (off < all.length) { + if (off + 9 > all.length) throw new Error(name + ": truncated frame header at " + off); + const len = all.readUIntBE(off, 3); + const type = all[off + 3]; + if (type > 0x0c) throw new Error(name + ": unknown frame type " + type + " at " + off); + if (off + 9 + len > all.length) throw new Error(name + ": truncated frame payload at " + off); + off += 9 + len; + frames++; + } + if (frames < min) throw new Error(name + ": only " + frames + " frames"); + return frames; + }; + const transport = (wire, onWrite) => + new Duplex({ + read() {}, + write(chunk, enc, cb) { + wire.push(Buffer.from(chunk)); + onWrite(); + cb(); + }, + }); + Bun.gc(true); + const baseline = count(); + const N = 3; + + // (1) one session whose transport pings the session itself + for (let i = 0; i < N; i++) { + let session; + let pings = 0; + const wire = []; + const duplex = transport(wire, () => { + if (session && pings < 20) { + pings++; + session.ping(Buffer.alloc(8), () => {}); + } + }); + session = http2.connect("http://127.0.0.1:1", { createConnection: () => duplex }); + await connected(session); + session.on("error", () => {}); + session.settings({ enablePush: false }); + const req = session.request({ ":path": "/" }); + req.on("error", () => {}); + req.end(); + await tick(); + if (pings === 0) throw new Error("transport _write never re-entered the session"); + assertWellFormed("self", wire, 3); + session.destroy(); + await tick(); + duplex.destroy(); + } + + // (2) two sessions: B's transport pings A + for (let i = 0; i < N; i++) { + let a, b; + let cross = 0; + const wa = [], wb = []; + const da = transport(wa, () => {}); + const db = transport(wb, () => { + if (a && cross < 20) { + cross++; + a.ping(Buffer.alloc(8), () => {}); + } + }); + a = http2.connect("http://127.0.0.1:1", { createConnection: () => da }); + b = http2.connect("http://127.0.0.1:1", { createConnection: () => db }); + await Promise.all([connected(a), connected(b)]); + for (const s of [a, b]) s.on("error", () => {}); + for (let j = 0; j < 5; j++) { + a.settings({ enablePush: false }); + b.settings({ enablePush: false }); + b.ping(Buffer.alloc(8), () => {}); + a.ping(Buffer.alloc(8), () => {}); + await tick(); + } + if (cross === 0) throw new Error("B's transport _write never re-entered A"); + assertWellFormed("a", wa, 5); + assertWellFormed("b", wb, 5); + for (const s of [a, b]) s.destroy(); + await tick(); + for (const d of [da, db]) d.destroy(); + } + + // (3) a cycle with no cap: B's transport pings C and C's pings B, while A writes 50 frames. + // Every A frame corks A, which forces the slot away from B or C; that flush re-enters and + // hands the slot to the other one before A resumes. + for (let i = 0; i < N; i++) { + let a, b, c; + let live = false; + let bc = 0, cb = 0; + const wa = [], wb = [], wc = []; + const da = transport(wa, () => {}); + // ping(): corked without a synchronous flush, and uncapped here, so the only thing + // bounding the B<->C exchange is the cork handover itself. + const emit = s => s.ping(Buffer.alloc(8), () => {}); + const opts = d => ({ createConnection: () => d, maxOutstandingPings: 1e9 }); + const db = transport(wb, () => { + if (live) { + bc++; + emit(c); + } + }); + const dc = transport(wc, () => { + if (live) { + cb++; + emit(b); + } + }); + a = http2.connect("http://127.0.0.1:1", opts(da)); + b = http2.connect("http://127.0.0.1:1", opts(db)); + c = http2.connect("http://127.0.0.1:1", opts(dc)); + await Promise.all([a, b, c].map(connected)); + for (const s of [a, b, c]) s.on("error", () => {}); + live = true; + emit(b); + for (let j = 0; j < 50; j++) { + emit(a); + if (j % 10 === 9) await tick(); + } + await tick(); + live = false; + await tick(); + if (bc === 0 || cb === 0) throw new Error("B/C transports never re-entered each other: " + bc + "/" + cb); + const fa = assertWellFormed("a", wa, 50); + assertWellFormed("b", wb, 5); + assertWellFormed("c", wc, 5); + for (const s of [a, b, c]) s.destroy(); + await tick(); + for (const d of [da, db, dc]) d.destroy(); + } + + // (4) a displaced owner left idle: B owns the slot, A corks (flushing B), B's transport + // pings C once so C takes the slot first, and A displaces C. C then has queued bytes but no + // cork: its auto-flush is registered for them, sends them, and retires within the tick. + const { h2AutoFlushRegistered } = require("bun:internal-for-testing"); + const nativeOf = s => s[Symbol.for("::bunhttp2native::")]; + for (let i = 0; i < N; i++) { + let a, b, c; + let armed = false; + const wa = [], wb = [], wc = []; + const emit = s => s.ping(Buffer.alloc(8), () => {}); + const opts = d => ({ createConnection: () => d, maxOutstandingPings: 1e9 }); + const da = transport(wa, () => {}); + const db = transport(wb, () => { + if (armed) { + armed = false; + emit(c); + } + }); + const dc = transport(wc, () => {}); + a = http2.connect("http://127.0.0.1:1", opts(da)); + b = http2.connect("http://127.0.0.1:1", opts(db)); + c = http2.connect("http://127.0.0.1:1", opts(dc)); + await Promise.all([a, b, c].map(connected)); + for (const s of [a, b, c]) s.on("error", () => {}); + await tick(); + emit(b); + armed = true; + emit(a); + if (armed) throw new Error("B's transport never handed the slot to C"); + if (!h2AutoFlushRegistered(nativeOf(c))) throw new Error("displaced C has no auto-flush for its queued bytes"); + for (let j = 0; j < 50 && h2AutoFlushRegistered(nativeOf(c)); j++) await new Promise(r => setTimeout(r, 10)); + if (h2AutoFlushRegistered(nativeOf(c))) throw new Error("C's auto-flush still registered after its bytes went out"); + assertWellFormed("a", wa, 2); + assertWellFormed("b", wb, 2); + assertWellFormed("c", wc, 2); + for (const s of [a, b, c]) s.destroy(); + await tick(); + for (const d of [da, db, dc]) d.destroy(); + } + + let live = -1; + for (let i = 0; i < 50; i++) { + await new Promise(r => setTimeout(r, 10)); + Bun.gc(true); + live = count() - baseline; + if (live <= 0) break; + } + console.log(JSON.stringify({ live })); + // No process.exit(): nothing the sessions registered may keep the process alive. + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", fixture], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout.trim())).toEqual({ live: 0 }); + expect(exitCode).toBe(0); +}, 60_000); + it("http2 padded DATA write survives a re-entrant stream write from a JS Duplex _write", async () => { // With padding enabled, a single-frame DATA write that crosses the 16 KiB cork boundary // flushes the cork into the JS transport mid-frame (onWrite → Duplex _write). A transport