node:http2: remove unsafe from H2FrameParser

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;

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<Self>, 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<C: AbortListener>(&self, ctx: *mut C) -> &AbortSignal {
         extern "C" fn callback<C: AbortListener>(ptr: *mut c_void, reason: JSValue) {
@@ -84,6 +112,25 @@ impl AbortSignal {
         self.add_listener(ctx.cast::<c_void>(), callback::<C>)
     }
 
+    /// 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<C: NativeAbortListener>(
+        &self,
+        listener: bun_ptr::BackRef<C, bun_ptr::Root>,
+    ) -> AbortListenerRegistration {
+        extern "C" fn callback<C: NativeAbortListener>(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::<C>()) }, reason);
+        }
+        let ctx = listener.this_ptr().as_ptr().cast::<c_void>();
+        let signal = self.ref_();
+        self.pending_activity_ref();
+        self.add_listener(ctx, callback::<C>);
+        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

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<JSValue> {
+    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)) }
+    })
+}

src/jsc/bindings/FfiSlice.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include <span>
+#include <stddef.h>
+#include <stdint.h>
+
+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<typename T = uint8_t>
+struct FfiSlice {
+    const T* ptr;
+    size_t len;
+
+    std::span<const T> span() const { return { ptr, len }; }
+};
+
+} // namespace Bun

src/jsc/bindings/H2HeadersMaterializer.cpp

@@ -18,6 +18,7 @@
 #include <wtf/text/StringView.h>
 #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<uint8_t> packedSlice,
+    FfiSlice<uint32_t> 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<unsigned>(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;

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<WebCore::WebSocket::HandshakeRawHeader>;
 
 extern "C" void WebSocket__didReceiveHandshakeResponse(WebCore::WebSocket* webSocket, uint16_t statusCode, WebCore::WebSocket::FfiSlice statusMessage, FfiRawHeaderSlice headers, WebCore::WebSocket::FfiSlice body)
 {

src/jsc/bindings/webcore/WebSocket.h

@@ -39,6 +39,7 @@
 #include <wtf/Lock.h>
 #include "FetchHeaders.h"
 #include "WebSocketErrorCode.h"
+#include "FfiSlice.h"
 
 namespace WebCore {
 class JSBlob;
@@ -215,12 +216,7 @@ class WebSocket final : public RefCounted<WebSocket>, public EventTargetWithInli
 
     void didReceiveMessage(String&& message);
     void didReceiveBinaryData(const AtomString& eventName, const std::span<const uint8_t> binaryData);
-    /// `bun_core::ffi::FfiSlice` — a borrowed `&[u8]` passed by value.
-    struct FfiSlice {
-        const uint8_t* ptr;
-        size_t len;
-        std::span<const uint8_t> span() const { return { ptr, len }; }
-    };
+    using FfiSlice = Bun::FfiSlice<uint8_t>;
     struct HandshakeRawHeader {
         FfiSlice name;
         FfiSlice value;

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"]

src/ptr/lib.rs

@@ -162,7 +162,48 @@ impl<T: ?Sized> BackRef<T, Mut> {
     }
 }
 
-impl<T, P> BackRef<T, P> {
+/// 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<T>(pub(crate) core::ptr::NonNull<T>);
+
+impl<T> SelfRoot<T> {
+    /// 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<T> {
+        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<T, Root> {
+        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<T, P: DanglingOk> BackRef<T, P> {
     #[inline]
     pub const fn dangling() -> Self {
         BackRef(core::ptr::NonNull::dangling(), core::marker::PhantomData)

src/ptr/ref_count.rs

@@ -541,6 +541,21 @@ impl<T: AnyRefCounted> RefPtr<T> {
         Self(ptr)
     }
 
+    /// [`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>) -> T) -> Self {
+        let raw: NonNull<T> = bun_core::heap::into_raw_nn(Box::<T>::new_uninit()).cast::<T>();
+        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)
+    }
+
     /// Take a new ref on `*raw_ptr`.
     ///
     /// # Safety
@@ -975,4 +990,57 @@ mod tests {
         assert_eq!(type_base_name("a::b::Foo<c::Bar>"), "Foo<c::Bar>");
         assert_eq!(type_base_name("Foo"), "Foo");
     }
+
+    struct Cyclic {
+        ref_count: RefCount<Cyclic>,
+        self_ref: crate::SelfRoot<Cyclic>,
+        payload: Box<u32>,
+    }
+
+    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<Self> {
+            // 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);
+    }
 }

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;
     }
 }