ext/node/ops/http2/session.rs - main - githubmirror/deno

ext/node/ops/http2/session.rs
3 756 строк130 KB
Nathan Whitaker
fix(ext/node): apply backpressure to http2 stream writes (#36044)
20 июл 2026, 21:45
20 июл 2026, 21:4519c7442
// Copyright 2018-2026 the Deno authors. MIT license.

// On Windows MSVC, bindgen generates nghttp2 enum constants as i32 (C `int`),
// while on Unix they are u32. Explicit `as` casts are needed for cross-platform
// compatibility but trigger unnecessary_cast on the platform where the type
// already matches.
#![allow(clippy::unnecessary_cast, reason = "platform specific")]

use std::cell::Cell;
use std::cell::RefCell;
use std::cell::UnsafeCell;
use std::collections::HashMap;
use std::collections::VecDeque;
use std::ffi::c_void;
use std::rc::Rc;

use deno_core::OpState;
use deno_core::ToV8;
use deno_core::cppgc;
use deno_core::op2;
use deno_core::v8;
use libnghttp2 as ffi;

/// Match the exact callback return type generated by libnghttp2 bindings on
/// the current platform instead of guessing the Windows `ssize_t` mapping.
type CSsizeT = ffi::nghttp2_ssize;

use super::stream::Http2Headers;
use super::stream::Http2Priority;
use super::stream::Http2Stream;
use super::types::*;

// Thread-local state buffers

const SESSION_STATE_LEN: usize = SessionStateIndex::Count as usize;
const STREAM_STATE_LEN: usize = StreamStateIndex::Count as usize;
const OPTIONS_LEN: usize = OptionsIndex::Flags as usize + 1;
const SETTINGS_LEN: usize =
  SettingsIndex::Count as usize + 1 + 1 + (2 * MAX_ADDITIONAL_SETTINGS);

thread_local! {
  static SESSION_STATE: UnsafeCell<[f32; SESSION_STATE_LEN]> =
    const { UnsafeCell::new([0.0; SESSION_STATE_LEN]) };

  static STREAM_STATE: UnsafeCell<[f32; STREAM_STATE_LEN]> =
    const { UnsafeCell::new([0.0; STREAM_STATE_LEN]) };

  static OPTIONS: UnsafeCell<[u32; OPTIONS_LEN]> =
    const { UnsafeCell::new([0; OPTIONS_LEN]) };

  static SETTINGS: UnsafeCell<[u32; SETTINGS_LEN]> =
    const { UnsafeCell::new([0; SETTINGS_LEN]) };
}

#[derive(ToV8)]
pub struct JSHttp2State<'a> {
  session_state: v8::Local<'a, v8::Value>,
  stream_state: v8::Local<'a, v8::Value>,
  options_buffer: v8::Local<'a, v8::Value>,
  settings_buffer: v8::Local<'a, v8::Value>,
}

impl<'a> JSHttp2State<'a> {
  pub fn create(scope: &mut v8::PinScope<'a, 'a>) -> Self {
    let session_state = SESSION_STATE.with(|cell| {
      // SAFETY: thread-local UnsafeCell, no concurrent access possible
      let ptr = unsafe { (*cell.get()).as_mut_ptr() };
      create_f32_array(scope, ptr, SESSION_STATE_LEN)
    });

    let stream_state = STREAM_STATE.with(|cell| {
      // SAFETY: thread-local UnsafeCell, no concurrent access possible
      let ptr = unsafe { (*cell.get()).as_mut_ptr() };
      create_f32_array(scope, ptr, STREAM_STATE_LEN)
    });

    let options_buffer = OPTIONS.with(|cell| {
      // SAFETY: thread-local UnsafeCell, no concurrent access possible
      let ptr = unsafe { (*cell.get()).as_mut_ptr() };
      create_u32_array(scope, ptr, OPTIONS_LEN)
    });

    let settings_buffer = SETTINGS.with(|cell| {
      // SAFETY: thread-local UnsafeCell, no concurrent access possible
      let ptr = unsafe { (*cell.get()).as_mut_ptr() };
      create_u32_array(scope, ptr, SETTINGS_LEN)
    });

    Self {
      session_state,
      stream_state,
      options_buffer,
      settings_buffer,
    }
  }
}

fn create_f32_array<'a>(
  scope: &mut v8::PinScope<'a, 'a>,
  buffer: *mut f32,
  len: usize,
) -> v8::Local<'a, v8::Value> {
  // SAFETY: buffer points to valid thread-local static with matching len
  unsafe {
    let bs = v8::ArrayBuffer::new_backing_store_from_ptr(
      buffer as *mut c_void,
      len * std::mem::size_of::<f32>(),
      nop_deleter,
      std::ptr::null_mut(),
    );
    let ab = v8::ArrayBuffer::with_backing_store(scope, &bs.make_shared());
    v8::Float32Array::new(scope, ab, 0, len).unwrap().into()
  }
}

fn create_u32_array<'a>(
  scope: &mut v8::PinScope<'a, 'a>,
  buffer: *mut u32,
  len: usize,
) -> v8::Local<'a, v8::Value> {
  // SAFETY: buffer points to valid thread-local static with matching len
  unsafe {
    let bs = v8::ArrayBuffer::new_backing_store_from_ptr(
      buffer as *mut c_void,
      len * std::mem::size_of::<u32>(),
      nop_deleter,
      std::ptr::null_mut(),
    );
    let ab = v8::ArrayBuffer::with_backing_store(scope, &bs.make_shared());
    v8::Uint32Array::new(scope, ab, 0, len).unwrap().into()
  }
}

unsafe extern "C" fn nop_deleter(
  _data: *mut c_void,
  _byte_length: usize,
  _deleter_data: *mut c_void,
) {
}

fn with_settings<F, R>(f: F) -> R
where
  F: FnOnce(&mut [u32; SETTINGS_LEN]) -> R,
{
  SETTINGS.with(|cell| {
    // SAFETY: thread-local UnsafeCell, no concurrent access possible
    let buffer = unsafe { &mut *cell.get() };
    f(buffer)
  })
}

/// Serialize the session's tracked custom settings into the shared settings
/// buffer so JS's `addCustomSettingsToObj` can read them. Mirrors the custom
/// settings tail of Node's `Http2Settings::Update`. Entries whose high bits
/// are set (registered but never received from the peer) are skipped so the
/// JS object only exposes settings that actually have a value.
fn write_custom_settings_to_buffer(
  buffer: &mut [u32; SETTINGS_LEN],
  list: &[(i32, u32)],
) {
  let offset = SettingsIndex::Count as usize + 2;
  let mut count: usize = 0;
  for (key, val) in list {
    let id = *key as u32;
    if id & !0xffff_u32 != 0 {
      continue;
    }
    let lo = id & 0xffff;
    let mut updated = false;
    for j in 0..count {
      if buffer[offset + j * 2] & 0xffff == lo {
        buffer[offset + j * 2] = lo;
        buffer[offset + j * 2 + 1] = *val;
        updated = true;
        break;
      }
    }
    if !updated && count < MAX_ADDITIONAL_SETTINGS {
      buffer[offset + count * 2] = lo;
      buffer[offset + count * 2 + 1] = *val;
      count += 1;
    }
  }
  buffer[SettingsIndex::Count as usize + 1] = count as u32;
}

fn with_options<F, R>(f: F) -> R
where
  F: FnOnce(&[u32; OPTIONS_LEN]) -> R,
{
  OPTIONS.with(|cell| {
    // SAFETY: thread-local UnsafeCell, no concurrent access possible
    let buffer = unsafe { &*cell.get() };
    f(buffer)
  })
}

#[repr(C)]
struct H2WriteReq {
  uv_req: deno_core::uv_compat::UvWrite,
  data: Vec<u8>,
}

unsafe extern "C" fn h2_write_cb(
  req: *mut deno_core::uv_compat::UvWrite,
  _status: i32,
) {
  // Get the session pointer from the stream handle's data field.
  // SAFETY: req is valid per libuv write callback contract; handle was set by uv_write
  let stream_handle = unsafe { (*req).handle };
  if !stream_handle.is_null() {
    // SAFETY: stream_handle is valid per libuv; casting to UvHandle to read data field
    let session_ptr =
      unsafe { (*(stream_handle as *mut deno_core::uv_compat::UvHandle)).data };
    if !session_ptr.is_null() {
      // SAFETY: session_ptr was set in consume_stream and points to a valid Session
      let session = unsafe { &mut *(session_ptr as *mut Session) };
      session.maybe_notify_graceful_close_complete();
    }
  }

  // Free the write request
  // SAFETY: req was created by Box::into_raw in send_pending_data
  let _ = unsafe { Box::from_raw(req as *mut H2WriteReq) };
}

unsafe extern "C" fn h2_stream_close_cb(
  handle: *mut deno_core::uv_compat::UvHandle,
) {
  // The stream handle was a UvTcp allocated by the TCP cppgc object.
  // `consume_stream` transferred ownership to the session via
  // `TCPWrap::detach`, so the session is now responsible for freeing it.
  // SAFETY: handle was allocated by Box::into_raw and ownership transferred via detach
  let _ = unsafe { Box::from_raw(handle as *mut deno_core::uv_compat::UvTcp) };
}

unsafe extern "C" fn h2_shutdown_cb(
  req: *mut deno_core::uv_compat::UvShutdown,
  _status: i32,
) {
  // After graceful shutdown (FIN sent), close the handle to free it.
  // SAFETY: req.handle was set by uv_shutdown and is a valid stream handle
  let stream_handle = unsafe { (*req).handle };
  if !stream_handle.is_null() {
    // SAFETY: stream_handle is valid per uv_shutdown callback contract
    unsafe {
      deno_core::uv_compat::uv_close(
        stream_handle as *mut deno_core::uv_compat::UvHandle,
        Some(h2_stream_close_cb),
      );
    }
  }
  // SAFETY: req was allocated by Box::into_raw in destroy()
  let _ = unsafe { Box::from_raw(req) };
}

unsafe extern "C" fn h2_alloc_cb(
  _handle: *mut deno_core::uv_compat::UvHandle,
  suggested_size: usize,
  buf: *mut deno_core::uv_compat::UvBuf,
) {
  let data = vec![0u8; suggested_size];
  let leaked = Box::into_raw(data.into_boxed_slice());
  // SAFETY: buf is valid per libuv alloc callback contract; leaked is freshly allocated
  unsafe {
    (*buf).base = (*leaked).as_mut_ptr() as *mut _;
    (*buf).len = suggested_size;
  }
}

unsafe extern "C" fn h2_read_cb(
  handle: *mut deno_core::uv_compat::UvStream,
  nread: isize,
  buf: *const deno_core::uv_compat::UvBuf,
) {
  // Get the session from the handle's data pointer
  // SAFETY: handle is valid per libuv read callback contract
  let session_ptr = unsafe { (*handle).data as *mut Session };
  if session_ptr.is_null() {
    // Free the buffer
    // SAFETY: buf is valid per libuv read callback contract
    if !buf.is_null() && !unsafe { (*buf).base.is_null() } {
      // SAFETY: buf.base was allocated by Box::into_raw in h2_alloc_cb
      let _ = unsafe {
        Box::from_raw(std::ptr::slice_from_raw_parts_mut(
          (*buf).base as *mut u8,
          (*buf).len,
        ))
      };
    }
    return;
  }

  // SAFETY: session_ptr was set in consume_stream and is non-null (checked above)
  let session = unsafe { &mut *session_ptr };

  if nread < 0 {
    // EOF or error - stop reading and notify the session so that
    // streams are properly closed (emitting 'aborted'/'close' events).
    // SAFETY: handle is valid per libuv read callback contract
    unsafe {
      deno_core::uv_compat::uv_read_stop(handle);
    }
    // Notify nghttp2 about the EOF by terminating the session.
    // SAFETY: session.session is a valid nghttp2_session pointer
    unsafe {
      ffi::nghttp2_session_terminate_session(
        session.session,
        ffi::NGHTTP2_CONNECT_ERROR as u32,
      );
    }
    session.send_pending_data();

    // Notify JS that the underlying transport has closed. In Node.js,
    // EOF flows through PassReadErrorToPreviousListener → socket 'close'
    // event → socketOnClose, which closes all streams and destroys the
    // session. Since consume_stream bypasses the socket layer, we call
    // the handle's onstreamclose callback directly.
    {
      // SAFETY: isolate pointer is valid for the session's lifetime
      let mut isolate =
        unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
      v8::scope!(let scope, &mut isolate);
      let context = v8::Local::new(scope, session.context.clone());
      let scope = &mut v8::ContextScope::new(scope, context);
      if let Some(this) = session.this.as_ref() {
        let this_local = v8::Local::new(scope, this);
        let key = v8::String::new(scope, "onstreamclose").unwrap();
        if let Some(Ok(cb)) = this_local
          .get(scope, key.into())
          .map(v8::Local::<v8::Function>::try_from)
        {
          cb.call(scope, this_local.into(), &[]);
        }
      }
    }

    // Free the buffer
    // SAFETY: buf is valid per libuv read callback contract
    if !buf.is_null() && !unsafe { (*buf).base.is_null() } {
      // SAFETY: buf.base was allocated by Box::into_raw in h2_alloc_cb
      let _ = unsafe {
        Box::from_raw(std::ptr::slice_from_raw_parts_mut(
          (*buf).base as *mut u8,
          (*buf).len,
        ))
      };
    }
    return;
  }

  if nread > 0 {
    // SAFETY: buf.base is valid for nread bytes per libuv contract
    let data = unsafe {
      std::slice::from_raw_parts((*buf).base as *const u8, nread as usize)
    };
    session.receive_data(data);
  }

  // Free the buffer
  // SAFETY: buf is valid per libuv read callback contract
  if !buf.is_null() && !unsafe { (*buf).base.is_null() } {
    // SAFETY: buf.base was allocated by Box::into_raw in h2_alloc_cb
    let _ = unsafe {
      Box::from_raw(std::ptr::slice_from_raw_parts_mut(
        (*buf).base as *mut u8,
        (*buf).len,
      ))
    };
  }
}

// Http2Options

const DEFAULT_MAX_HEADER_LIST_PAIRS: u32 = 128;

fn option_present(flags: u32, index: OptionsIndex) -> bool {
  flags & (1 << index as u32) != 0
}

struct Http2Options {
  options: *mut ffi::nghttp2_option,
  padding_strategy: PaddingStrategy,
  max_header_pairs: u32,
}

impl Http2Options {
  fn new(
    session_type: SessionType,
    no_strict_field_ws_validation: bool,
  ) -> Self {
    let mut options: *mut ffi::nghttp2_option = std::ptr::null_mut();
    // SAFETY: passing valid pointer to be initialized by nghttp2
    unsafe { ffi::nghttp2_option_new(&mut options) };

    let mut max_header_pairs: u32 = DEFAULT_MAX_HEADER_LIST_PAIRS;
    let padding_strategy = with_options(|buffer| {
      let flags = buffer[OptionsIndex::Flags as usize];

      if option_present(flags, OptionsIndex::MaxHeaderListPairs) {
        max_header_pairs = buffer[OptionsIndex::MaxHeaderListPairs as usize];
      }

      // SAFETY: options was successfully initialized by nghttp2_option_new
      unsafe {
        ffi::nghttp2_option_set_no_closed_streams(options, 1);

        // Always disable nghttp2's automatic WINDOW_UPDATE replenishment.
        // Mirrors Node's `Http2Session::Http2Session` (`src/node_http2.cc`),
        // which unconditionally sets this so flow control is gated on
        // application-level consumption (`consume_stream`/`consume_connection`).
        // Without this, nghttp2 auto-replenishes the local stream window and a
        // misbehaving peer never trips NGHTTP2_FLOW_CONTROL_ERROR even when its
        // sends exceed the advertised initial window.
        ffi::nghttp2_option_set_no_auto_window_update(options, 1);

        let max_deflate =
          buffer[OptionsIndex::MaxDeflateDynamicTableSize as usize];
        if option_present(flags, OptionsIndex::MaxDeflateDynamicTableSize)
          && max_deflate > 0
        {
          ffi::nghttp2_option_set_max_deflate_dynamic_table_size(
            options,
            max_deflate as usize,
          );
        }

        let max_reserved =
          buffer[OptionsIndex::MaxReservedRemoteStreams as usize];
        if option_present(flags, OptionsIndex::MaxReservedRemoteStreams)
          && max_reserved > 0
        {
          ffi::nghttp2_option_set_max_reserved_remote_streams(
            options,
            max_reserved,
          );
        }

        let max_send_header =
          buffer[OptionsIndex::MaxSendHeaderBlockLength as usize];
        if option_present(flags, OptionsIndex::MaxSendHeaderBlockLength)
          && max_send_header > 0
        {
          ffi::nghttp2_option_set_max_send_header_block_length(
            options,
            max_send_header as usize,
          );
        }

        let peer_max_concurrent =
          buffer[OptionsIndex::PeerMaxConcurrentStreams as usize];
        if option_present(flags, OptionsIndex::PeerMaxConcurrentStreams)
          && peer_max_concurrent > 0
        {
          ffi::nghttp2_option_set_peer_max_concurrent_streams(
            options,
            peer_max_concurrent,
          );
        } else {
          ffi::nghttp2_option_set_peer_max_concurrent_streams(options, 100);
        }

        // Gate maxOutstandingPings on its flag bit. The optionsBuffer is a
        // process-global Uint32Array reused across sessions, so a prior
        // session's value would otherwise leak into a later session that
        // didn't pass the option and could lower nghttp2's max_outbound_ack
        // (e.g. to 1), making a legitimate second concurrent ping trip
        // NGHTTP2_ERR_FLOODED.
        if option_present(flags, OptionsIndex::MaxOutstandingPings) {
          let max_outstanding_pings =
            buffer[OptionsIndex::MaxOutstandingPings as usize];
          if max_outstanding_pings > 0 {
            ffi::nghttp2_option_set_max_outbound_ack(
              options,
              max_outstanding_pings as usize,
            );
          }
        }

        // Note: maxOutstandingSettings is a Node.js-internal queue limit for
        // outbound SETTINGS frames awaiting peer ACK. nghttp2 has no equivalent
        // option, so we accept the value for API compatibility but do not map
        // it to nghttp2_option_set_max_settings (which is the maxSettings limit
        // on inbound SETTINGS *entries* per frame — handled below).

        let max_settings = buffer[OptionsIndex::MaxSettings as usize];
        if option_present(flags, OptionsIndex::MaxSettings) && max_settings > 0
        {
          ffi::nghttp2_option_set_max_settings(options, max_settings as usize);
        }

        if matches!(session_type, SessionType::Client) {
          ffi::nghttp2_option_set_builtin_recv_extension_type(
            options,
            ffi::NGHTTP2_ALTSVC as u8,
          );
          ffi::nghttp2_option_set_builtin_recv_extension_type(
            options,
            ffi::NGHTTP2_ORIGIN as u8,
          );
        }

        // strictFieldWhitespaceValidation: when disabled, tell nghttp2 to skip
        // RFC 9113 leading/trailing whitespace validation so headers with
        // surrounding whitespace are delivered to on_header_callback instead
        // of being routed to on_invalid_header_callback (and dropped).
        if no_strict_field_ws_validation {
          ffi::nghttp2_option_set_no_rfc9113_leading_and_trailing_ws_validation(
            options, 1,
          );
        }
      }

      let padding = if option_present(flags, OptionsIndex::PaddingStrategy) {
        buffer[OptionsIndex::PaddingStrategy as usize]
      } else {
        0
      };
      match padding {
        1 => PaddingStrategy::Aligned,
        2 => PaddingStrategy::Max,
        3 => PaddingStrategy::Callback,
        _ => PaddingStrategy::None,
      }
    });

    // Apply Node.js semantics: server min 4, client min 1.
    // See node_http_common-inl.h: GetServerMaxHeaderPairs / GetClientMaxHeaderPairs.
    let min_pairs = match session_type {
      SessionType::Server => 4,
      SessionType::Client => 1,
    };
    let max_header_pairs = std::cmp::max(max_header_pairs, min_pairs);

    Self {
      options,
      padding_strategy,
      max_header_pairs,
    }
  }

  fn ptr(&self) -> *mut ffi::nghttp2_option {
    self.options
  }

  fn padding_strategy(&self) -> PaddingStrategy {
    self.padding_strategy
  }

  fn max_header_pairs(&self) -> u32 {
    self.max_header_pairs
  }
}

impl Drop for Http2Options {
  fn drop(&mut self) {
    if !self.options.is_null() {
      // SAFETY: options was allocated by nghttp2_option_new and is non-null
      unsafe { ffi::nghttp2_option_del(self.options) };
    }
  }
}

// Http2Settings

const SETTINGS_ENTRY_COUNT: usize =
  SettingsIndex::Count as usize + MAX_ADDITIONAL_SETTINGS;

struct Http2Settings {
  entries: [ffi::nghttp2_settings_entry; SETTINGS_ENTRY_COUNT],
  count: usize,
  session: *mut Session,
}

impl Http2Settings {
  fn init(session: *mut Session) -> Self {
    with_settings(|buffer| {
      let flags = buffer[SettingsIndex::Count as usize];
      let mut count: usize = 0;
      let mut entries = [ffi::nghttp2_settings_entry {
        settings_id: 0,
        value: 0,
      }; SETTINGS_ENTRY_COUNT];

      macro_rules! grab_setting {
        ($index:expr, $nghttp2_id:expr) => {
          if flags & (1 << $index as u8) != 0 {
            let val = buffer[$index as usize];
            if count < entries.len() {
              entries[count] = ffi::nghttp2_settings_entry {
                settings_id: $nghttp2_id as _,
                value: val,
              };
              count += 1;
            }
          }
        };
      }

      grab_setting!(
        SettingsIndex::HeaderTableSize,
        ffi::NGHTTP2_SETTINGS_HEADER_TABLE_SIZE
      );
      grab_setting!(
        SettingsIndex::EnablePush,
        ffi::NGHTTP2_SETTINGS_ENABLE_PUSH
      );
      grab_setting!(
        SettingsIndex::InitialWindowSize,
        ffi::NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE
      );
      grab_setting!(
        SettingsIndex::MaxFrameSize,
        ffi::NGHTTP2_SETTINGS_MAX_FRAME_SIZE
      );
      grab_setting!(
        SettingsIndex::MaxConcurrentStreams,
        ffi::NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS
      );
      grab_setting!(
        SettingsIndex::MaxHeaderListSize,
        ffi::NGHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE
      );
      grab_setting!(
        SettingsIndex::EnableConnectProtocol,
        ffi::NGHTTP2_SETTINGS_ENABLE_CONNECT_PROTOCOL
      );

      let num_add_settings = buffer[SettingsIndex::Count as usize + 1] as usize;

      if num_add_settings > 0 {
        let offset = SettingsIndex::Count as usize + 2;
        for i in 0..num_add_settings {
          let key = buffer[offset + i * 2];
          let val = buffer[offset + i * 2 + 1];
          if count < entries.len() {
            entries[count] = ffi::nghttp2_settings_entry {
              settings_id: key as i32,
              value: val,
            };
            count += 1;
          }
        }
      }

      Self {
        session,
        entries,
        count,
      }
    })
  }

  fn send(&self) {
    // SAFETY: self.session points to a valid Session allocated in Http2Session::create
    unsafe {
      let session = &mut *self.session;
      session.update_local_custom_settings(&self.entries[..self.count]);
      ffi::nghttp2_submit_settings(
        session.session,
        ffi::NGHTTP2_FLAG_NONE as _,
        self.entries.as_ptr(),
        self.count,
      );
    }
  }
}

// Callbacks

fn frame_id(frame: *const ffi::nghttp2_frame) -> i32 {
  // SAFETY: frame is valid per nghttp2 callback contract
  unsafe {
    let frame = &*frame;
    if frame.hd.type_ as u32 == ffi::NGHTTP2_PUSH_PROMISE as u32 {
      frame.push_promise.promised_stream_id
    } else {
      frame.hd.stream_id
    }
  }
}

fn frame_type(frame: *const ffi::nghttp2_frame) -> u8 {
  // SAFETY: frame is valid per nghttp2 callback contract
  unsafe { (*frame).hd.type_ }
}

fn frame_flags(frame: *const ffi::nghttp2_frame) -> u8 {
  // SAFETY: frame is valid per nghttp2 callback contract
  unsafe { (*frame).hd.flags }
}

fn frame_headers_category(
  frame: *const ffi::nghttp2_frame,
) -> ffi::nghttp2_headers_category {
  // SAFETY: frame is valid per nghttp2 callback contract
  unsafe { (*frame).headers.cat }
}

fn rcbuf_to_slice(rcbuf: *mut ffi::nghttp2_rcbuf) -> &'static [u8] {
  // SAFETY: rcbuf is valid per nghttp2 callback contract
  unsafe {
    let buf = ffi::nghttp2_rcbuf_get_buf(rcbuf);
    std::slice::from_raw_parts(buf.base, buf.len)
  }
}

fn frame_header_length(frame: *const ffi::nghttp2_frame) -> usize {
  // SAFETY: frame is valid per nghttp2 callback contract
  unsafe { (*frame).hd.length }
}

unsafe extern "C" fn on_begin_headers_callbacks(
  ng_session: *mut ffi::nghttp2_session,
  frame: *const ffi::nghttp2_frame,
  data: *mut c_void,
) -> i32 {
  // SAFETY: data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(data) };
  let id = frame_id(frame);
  let cat = frame_headers_category(frame);

  match session.find_stream(id) {
    None => {
      if session.is_graceful_closing() {
        // SAFETY: ng_session is valid per nghttp2 callback contract
        unsafe {
          ffi::nghttp2_submit_rst_stream(
            ng_session,
            ffi::NGHTTP2_FLAG_NONE as u8,
            id,
            ffi::NGHTTP2_REFUSED_STREAM as u32,
          );
        }
        return ffi::NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE as i32;
      }
      let (obj, stream) = Http2Stream::new(session, id, cat);
      stream.start_headers(cat);
      session.streams.insert(id, (obj, stream));
    }
    Some(s) => {
      s.start_headers(cat);
    }
  }

  0
}

unsafe extern "C" fn on_header_callback(
  _session: *mut ffi::nghttp2_session,
  frame: *const ffi::nghttp2_frame,
  name: *mut ffi::nghttp2_rcbuf,
  value: *mut ffi::nghttp2_rcbuf,
  flags: u8,
  data: *mut c_void,
) -> i32 {
  // SAFETY: data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(data) };
  let id = frame_id(frame);

  if let Some(stream) = session.find_stream(id) {
    let name_slice = rcbuf_to_slice(name);
    let value_slice = rcbuf_to_slice(value);

    if !stream.add_header(name_slice, value_slice, flags) {
      // SAFETY: session.session is a valid nghttp2 session pointer
      unsafe {
        ffi::nghttp2_submit_rst_stream(
          session.session,
          ffi::NGHTTP2_FLAG_NONE as u8,
          id,
          ffi::NGHTTP2_ENHANCE_YOUR_CALM as u32,
        );
      }
      return ffi::NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE as i32;
    }
  }

  0
}

unsafe extern "C" fn on_frame_recv_callback(
  _session: *mut ffi::nghttp2_session,
  frame: *const ffi::nghttp2_frame,
  data: *mut c_void,
) -> i32 {
  // SAFETY: data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(data) };

  // Count every received frame so the JS layer can surface it via the
  // `Http2Session` PerformanceObserver entry's `framesReceived` field.
  // Mirrors Node's `Http2Session::OnFrameReceive` updating `statistics_`.
  session.frames_received = session.frames_received.saturating_add(1);

  let ft = frame_type(frame) as u32;
  let ff = frame_flags(frame);
  #[allow(clippy::unnecessary_cast, reason = "cast needed for type alignment")]
  if ft == ffi::NGHTTP2_DATA as u32 {
    let has_end_stream = ff & ffi::NGHTTP2_FLAG_END_STREAM as u8 != 0;
    if has_end_stream {
      handle_data_end_stream(session, frame);
    } else if frame_header_length(frame) == 0 {
      // Empty DATA frame without END_STREAM. nghttp2 doesn't treat this as
      // a protocol violation, but Node.js counts it against
      // `maxSessionInvalidFrames` (see HandleDataFrame in node_http2.cc).
      // Post-increment comparison: with max=0 the second such frame trips.
      let count = session.invalid_frame_count;
      session.invalid_frame_count = count.saturating_add(1);
      if count > session.max_invalid_frames {
        session.custom_recv_error_code =
          Some("ERR_HTTP2_TOO_MANY_INVALID_FRAMES");
        return 1;
      }
    }
  } else if ft == ffi::NGHTTP2_PUSH_PROMISE as u32
    || ft == ffi::NGHTTP2_HEADERS as u32
  {
    handle_headers_frame(session, frame);
    if ff & ffi::NGHTTP2_FLAG_END_STREAM as u8 != 0 {
      handle_data_end_stream(session, frame);
    }
  } else if ft == ffi::NGHTTP2_SETTINGS as u32 {
    if ff & ffi::NGHTTP2_FLAG_ACK as u8 == 0 {
      // Peer's actual settings. Update our tracked remote custom settings
      // before firing the JS callback so `session.remoteSettings` returns
      // the latest values.
      // SAFETY: frame is a valid SETTINGS frame; the union access matches
      // the frame type checked above. niv/iv come straight from nghttp2.
      unsafe {
        let settings = &(*frame).settings;
        if settings.niv > 0 && !settings.iv.is_null() {
          let iv =
            std::slice::from_raw_parts(settings.iv, settings.niv as usize);
          session.update_remote_custom_settings_from_iv(iv);
        }
      }
      // Mark handshake-time as over so on_frame_send_callback's protocol
      // error hook only fires on early (pre-SETTINGS) GOAWAYs.
      session.remote_settings_received = true;
      handle_settings_frame(session);
    } else {
      // ACK for one of our outgoing SETTINGS frames. Pop the FIFO of pending
      // callbacks and invoke it so JS can fire the `localSettings` event.
      handle_settings_ack(session);
    }
  } else if ft == ffi::NGHTTP2_PRIORITY as u32 {
    handle_priority_frame(session, frame);
  } else if ft == ffi::NGHTTP2_GOAWAY as u32 {
    handle_goaway_frame(session, frame);
  } else if ft == ffi::NGHTTP2_PING as u32 {
    if ff & ffi::NGHTTP2_FLAG_ACK as u8 != 0 {
      // PING ACK from the peer. If we have an outstanding PING, consume it
      // and let JS handle the ack. Otherwise the peer sent us an
      // unsolicited PING ACK — Node treats this as a connection-level
      // protocol error (see HandlePingFrame in node_http2.cc) and surfaces
      // it via the session's internal-error callback so the JS layer can
      // destroy the session with NghttpError(NGHTTP2_ERR_PROTO).
      if session.pending_pings > 0 {
        session.pending_pings -= 1;
        handle_ping_frame(session, frame, true);
      } else {
        handle_unsolicited_ping_ack(session);
      }
    } else {
      handle_ping_frame(session, frame, false);
    }
  } else if ft == ffi::NGHTTP2_ALTSVC as u32 {
    handle_alt_svc_frame(session, frame);
  } else if ft == ffi::NGHTTP2_ORIGIN as u32 {
    handle_origin_frame(session, frame);
  }

  0
}

fn handle_data_end_stream(session: &Session, frame: *const ffi::nghttp2_frame) {
  let id = frame_id(frame);
  let Some(stream_obj) = session.find_stream_obj(id) else {
    return;
  };

  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let handle = v8::Local::new(scope, stream_obj);

  let onread_key = v8::String::new(scope, "onread").unwrap();
  let Some(onread_val) = handle.get(scope, onread_key.into()) else {
    return;
  };
  let Ok(onread_fn) = v8::Local::<v8::Function>::try_from(onread_val) else {
    return;
  };

  // Signal EOF: onStreamRead(undefined, UV_EOF)
  let eof = v8::Number::new(scope, -4095.0);
  let undef = v8::undefined(scope);
  onread_fn.call(scope, handle.into(), &[undef.into(), eof.into()]);
}

fn handle_headers_frame(session: &Session, frame: *const ffi::nghttp2_frame) {
  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let id = frame_id(frame);
  let Some(stream_ref) = session.find_stream(id) else {
    return;
  };

  let headers = stream_ref.current_headers.borrow();
  if headers.is_empty() {
    return;
  }

  let headers_array = v8::Array::new(scope, (headers.len() * 2) as i32);
  // Mirrors Node's HandleHeadersFrame: collect names of headers that arrived
  // with NGHTTP2_NV_FLAG_NO_INDEX so JS can expose them via
  // headers[http2.sensitiveHeaders].
  let sensitive_array = v8::Array::new(scope, 0);
  let mut sensitive_count: u32 = 0;
  for (i, (name, value, flags)) in headers.iter().enumerate() {
    // Header bytes are arbitrary; decode as Latin-1 (one byte per code unit)
    // to match Node's LATIN1 path and preserve non-UTF-8 byte values.
    let name_str =
      v8::String::new_from_one_byte(scope, name, v8::NewStringType::Normal)
        .unwrap();
    let value_str =
      v8::String::new_from_one_byte(scope, value, v8::NewStringType::Normal)
        .unwrap();
    headers_array.set_index(scope, (i * 2) as u32, name_str.into());
    headers_array.set_index(scope, (i * 2 + 1) as u32, value_str.into());
    if flags & (ffi::NGHTTP2_NV_FLAG_NO_INDEX as u8) != 0 {
      sensitive_array.set_index(scope, sensitive_count, name_str.into());
      sensitive_count += 1;
    }
  }

  drop(headers);
  stream_ref.clear_headers();

  let stream_obj = session.find_stream_obj(id).unwrap();
  let Some(this) = session.this.as_ref() else {
    return;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.headers_frame_cb);
  drop(state);

  let handle = v8::Local::new(scope, stream_obj);
  let id_num = v8::Number::new(scope, id.into());
  // For HEADERS frames, expose the nghttp2 category so the JS polyfill can
  // distinguish 'response', 'push', 'trailers', 'headers'. PUSH_PROMISE
  // frames don't carry a category — leave it null and let the polyfill
  // treat the new-stream branch (the stream is still being created).
  let cat: v8::Local<v8::Value> =
    if frame_type(frame) as u32 == ffi::NGHTTP2_HEADERS as u32 {
      // SAFETY: frame is a HEADERS frame per the type check above
      let c = unsafe { (*frame).headers.cat } as i32;
      v8::Integer::new(scope, c).into()
    } else {
      v8::null(scope).into()
    };
  let flags = v8::Number::new(scope, frame_flags(frame).into());

  callback.call(
    scope,
    recv.into(),
    &[
      handle.into(),
      id_num.into(),
      cat,
      flags.into(),
      headers_array.into(),
      sensitive_array.into(),
    ],
  );
}

fn handle_settings_ack(session: &mut Session) {
  // FIFO pop. Mirrors Node's BaseObjectPtr<Http2Settings> PopSettings().
  let Some(cb) = session.pending_settings_acks.pop_front() else {
    return;
  };

  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let recv = v8::undefined(scope);
  let ack = v8::Boolean::new(scope, true);
  let duration = v8::Number::new(scope, 0.0);
  let cb_local = v8::Local::new(scope, &cb);
  cb_local.call(scope, recv.into(), &[ack.into(), duration.into()]);
}

fn handle_settings_frame(session: &Session) {
  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let Some(this) = session.this.as_ref() else {
    return;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.settings_frame_cb);
  drop(state);

  callback.call(scope, recv.into(), &[]);
}

fn handle_ping_frame(
  session: &Session,
  frame: *const ffi::nghttp2_frame,
  is_ack: bool,
) {
  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  // SAFETY: frame is a valid PING frame per nghttp2 callback contract
  let opaque = unsafe { (*frame).ping.opaque_data };
  let array_buffer = v8::ArrayBuffer::new(scope, opaque.len());
  let backing_store = array_buffer.get_backing_store();
  if let Some(backing_data) = backing_store.data() {
    // SAFETY: src and dst are valid, non-overlapping, dst has room for 8 bytes
    unsafe {
      std::ptr::copy_nonoverlapping(
        opaque.as_ptr(),
        backing_data.as_ptr() as *mut u8,
        opaque.len(),
      );
    }
  }
  let payload =
    v8::Uint8Array::new(scope, array_buffer, 0, opaque.len()).unwrap();

  let Some(this) = session.this.as_ref() else {
    return;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.ping_frame_cb);
  drop(state);

  let is_ack_v = v8::Boolean::new(scope, is_ack);
  callback.call(scope, recv.into(), &[payload.into(), is_ack_v.into()]);
}

/// Inbound PING ACK with no matching outstanding PING. Mirrors Node's
/// HandlePingFrame "unsolicited ack" branch in src/node_http2.cc: surface
/// `NGHTTP2_ERR_PROTO` to the JS internal-error callback so the session is
/// destroyed with `NghttpError(-505)` (code `ERR_HTTP2_ERROR`,
/// message "Protocol error").
fn handle_unsolicited_ping_ack(session: &Session) {
  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let Some(this) = session.this.as_ref() else {
    return;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.session_internal_error_cb);
  drop(state);

  let code = v8::Integer::new(scope, ffi::NGHTTP2_ERR_PROTO);
  callback.call(scope, recv.into(), &[code.into()]);
}

fn handle_goaway_frame(session: &Session, frame: *const ffi::nghttp2_frame) {
  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  // SAFETY: frame is valid per nghttp2 callback contract
  let goaway_frame = unsafe { (*frame).goaway };

  let error_code = v8::Number::new(scope, goaway_frame.error_code.into());
  let last_stream_id =
    v8::Number::new(scope, goaway_frame.last_stream_id.into());

  let opaque_data: v8::Local<v8::Value> = if goaway_frame.opaque_data_len > 0 {
    // SAFETY: opaque_data pointer and length are set by nghttp2
    let data_slice = unsafe {
      std::slice::from_raw_parts(
        goaway_frame.opaque_data,
        goaway_frame.opaque_data_len,
      )
    };
    let array_buffer = v8::ArrayBuffer::new(scope, data_slice.len());
    let backing_store = array_buffer.get_backing_store();
    if let Some(backing_data) = backing_store.data() {
      // SAFETY: src and dst are valid, non-overlapping, and dst has sufficient capacity
      unsafe {
        std::ptr::copy_nonoverlapping(
          data_slice.as_ptr(),
          backing_data.as_ptr() as *mut u8,
          data_slice.len(),
        );
      }
    }
    v8::Uint8Array::new(scope, array_buffer, 0, data_slice.len())
      .unwrap()
      .into()
  } else {
    v8::undefined(scope).into()
  };

  let Some(this) = session.this.as_ref() else {
    return;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.goaway_data_cb);
  drop(state);

  callback.call(
    scope,
    recv.into(),
    &[error_code.into(), last_stream_id.into(), opaque_data],
  );
}

fn handle_priority_frame(session: &Session, frame: *const ffi::nghttp2_frame) {
  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  // SAFETY: frame is valid per nghttp2 callback contract
  let priority_frame = unsafe { (*frame).priority };
  let id = frame_id(frame);
  let spec = priority_frame.pri_spec;

  let stream_id = v8::Number::new(scope, id.into());
  let parent_stream_id = v8::Number::new(scope, spec.stream_id.into());
  let weight = v8::Number::new(scope, spec.weight.into());
  let exclusive = v8::Boolean::new(scope, spec.exclusive != 0);

  let Some(this) = session.this.as_ref() else {
    return;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.priority_frame_cb);
  drop(state);

  callback.call(
    scope,
    recv.into(),
    &[
      stream_id.into(),
      parent_stream_id.into(),
      weight.into(),
      exclusive.into(),
    ],
  );
}

fn handle_alt_svc_frame(session: &Session, frame: *const ffi::nghttp2_frame) {
  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let id = frame_id(frame);

  // SAFETY: frame is valid per nghttp2 callback contract
  let ext = unsafe { (*frame).ext };
  let altsvc = ext.payload as *const ffi::nghttp2_ext_altsvc;

  // SAFETY: altsvc origin pointer and length are set by nghttp2
  let origin_slice = unsafe {
    std::slice::from_raw_parts((*altsvc).origin, (*altsvc).origin_len)
  };
  // SAFETY: altsvc field_value pointer and length are set by nghttp2
  let field_value_slice = unsafe {
    std::slice::from_raw_parts((*altsvc).field_value, (*altsvc).field_value_len)
  };

  let origin_str = std::str::from_utf8(origin_slice)
    .map(|s| v8::String::new(scope, s).unwrap())
    .unwrap_or_else(|_| v8::String::new(scope, "").unwrap());
  let field_value_str = std::str::from_utf8(field_value_slice)
    .map(|s| v8::String::new(scope, s).unwrap())
    .unwrap_or_else(|_| v8::String::new(scope, "").unwrap());

  let Some(this) = session.this.as_ref() else {
    return;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.alt_svc_cb);
  drop(state);

  let stream_id = v8::Number::new(scope, id.into());
  callback.call(
    scope,
    recv.into(),
    &[stream_id.into(), origin_str.into(), field_value_str.into()],
  );
}

fn handle_origin_frame(session: &Session, frame: *const ffi::nghttp2_frame) {
  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  // SAFETY: frame is valid per nghttp2 callback contract
  let ext = unsafe { (*frame).ext };
  let origin = ext.payload as *const ffi::nghttp2_ext_origin;

  // SAFETY: origin payload is valid and initialized by nghttp2
  let nov = unsafe { (*origin).nov };
  // SAFETY: origin payload is valid and initialized by nghttp2
  let origins_ptr = unsafe { (*origin).ov };

  if nov == 0 {
    return;
  }

  let origins_array = v8::Array::new(scope, nov as i32);
  for i in 0..nov {
    // SAFETY: origins_ptr points to nov valid entries per nghttp2 contract
    let entry = unsafe { *origins_ptr.add(i) };
    let origin_slice =
      // SAFETY: entry origin pointer and length are set by nghttp2
      unsafe { std::slice::from_raw_parts(entry.origin, entry.origin_len) };
    if let Ok(origin_str) = std::str::from_utf8(origin_slice) {
      let js_string = v8::String::new(scope, origin_str).unwrap();
      origins_array.set_index(scope, i as u32, js_string.into());
    }
  }

  let Some(this) = session.this.as_ref() else {
    return;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.origin_frame_cb);
  drop(state);

  callback.call(scope, recv.into(), &[origins_array.into()]);
}

unsafe extern "C" fn on_stream_close_callback(
  _session: *mut ffi::nghttp2_session,
  stream_id: i32,
  error_code: u32,
  data: *mut c_void,
) -> i32 {
  // SAFETY: data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(data) };
  let Some(stream_obj) = session.find_stream_obj(stream_id).cloned() else {
    return 0;
  };

  // Mark the stream as being closed by nghttp2 BEFORE calling JS.
  // This prevents shutdown() from calling resume_data(), which would
  // re-activate the data provider for a stream that close_stream is
  // about to destroy (double-free with no_closed_streams=1).
  if let Some(stream) = session.find_stream(stream_id) {
    *stream.closed_by_nghttp2.borrow_mut() = true;
  }

  // SAFETY: isolate pointer is valid for the session's lifetime
  let mut isolate =
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let callback = v8::Local::new(scope, &callbacks.stream_close_cb);
  drop(state);

  let recv = v8::Local::new(scope, stream_obj);
  let code = v8::Integer::new_from_unsigned(scope, error_code);

  let result = callback.call(scope, recv.into(), &[code.into()]);

  if result.is_none() || result.map(|v| v.is_false()).unwrap_or(false) {
    session.streams.remove(&stream_id);
  }

  0
}

unsafe extern "C" fn on_data_chunk_recv_callback(
  ng_session: *mut ffi::nghttp2_session,
  _flags: u8,
  stream_id: i32,
  data_ptr: *const u8,
  len: usize,
  user_data: *mut c_void,
) -> i32 {
  if len == 0 {
    return 0;
  }
  // SAFETY: user_data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(user_data) };

  // Always replenish the connection-level flow-control window. Stream-level
  // consumption is gated on whether the JS Readable side is actively
  // reading: when paused, defer it so a peer that ignores stream-level
  // flow control gets a NGHTTP2_FLOW_CONTROL_ERROR. Mirrors Node's
  // `Http2Session::OnDataChunkReceived` (`src/node_http2.cc`).
  // SAFETY: ng_session is valid per nghttp2 callback contract
  unsafe {
    ffi::nghttp2_session_consume_connection(ng_session, len);
  };
  if let Some(stream) = session.find_stream(stream_id) {
    if *stream.reading.borrow() {
      // SAFETY: ng_session is valid per nghttp2 callback contract
      unsafe {
        ffi::nghttp2_session_consume_stream(ng_session, stream_id, len);
      };
    } else {
      *stream.inbound_consumed_data_while_paused.borrow_mut() += len;
    }
  } else {
    // SAFETY: ng_session is valid per nghttp2 callback contract
    unsafe {
      ffi::nghttp2_session_consume_stream(ng_session, stream_id, len);
    };
  }

  // Deliver data to the JS stream via its onread callback
  let Some(stream_obj) = session.find_stream_obj(stream_id) else {
    return 0;
  };

  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let handle = v8::Local::new(scope, stream_obj);

  // Get the onread property
  let onread_key = v8::String::new(scope, "onread").unwrap();
  let Some(onread_val) = handle.get(scope, onread_key.into()) else {
    return 0;
  };
  let Ok(onread_fn) = v8::Local::<v8::Function>::try_from(onread_val) else {
    return 0;
  };

  // Create a Buffer with the received data
  // SAFETY: data_ptr is valid for len bytes per nghttp2 callback contract
  let data_slice = unsafe { std::slice::from_raw_parts(data_ptr, len) };
  let ab = v8::ArrayBuffer::new(scope, len);
  let backing_store = ab.get_backing_store();
  let dst = backing_store.data().unwrap().as_ptr() as *mut u8;
  // SAFETY: src and dst are valid, non-overlapping, and dst has len bytes capacity
  unsafe { std::ptr::copy_nonoverlapping(data_slice.as_ptr(), dst, len) };
  let uint8_array = v8::Uint8Array::new(scope, ab, 0, len).unwrap();

  let nread = v8::Number::new(scope, len as f64);
  // onStreamRead(arrayBuffer, nread) with `this` = handle
  onread_fn.call(scope, handle.into(), &[uint8_array.into(), nread.into()]);

  0
}

pub unsafe extern "C" fn on_stream_read_callback(
  _session: *mut ffi::nghttp2_session,
  stream_id: i32,
  buf: *mut u8,
  length: usize,
  data_flags: *mut u32,
  _source: *mut ffi::nghttp2_data_source,
  user_data: *mut c_void,
) -> CSsizeT {
  // SAFETY: user_data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(user_data) };

  // Gather data and determine flags while holding borrows, then call
  // into JS (on_trailers) only after dropping all borrows. This prevents
  // holding a RefMut<pending_data> or &Ref<Http2Stream> (borrowed from
  // session.streams HashMap) while JS callbacks fire, which could
  // invalidate the references.
  //
  // For aligned padding, switch to NGHTTP2_DATA_FLAG_NO_COPY so the
  // actual payload is emitted by `on_send_data_callback` as separate
  // chunks (header / pad_length / data / padding). For the default
  // strategy (no padding) we keep the in-place copy: nghttp2 packs the
  // data into the framebuf and OB_SEND_DATA returns it as one chunk,
  // matching how Deno previously framed unpadded responses (and what
  // tests like test-http2-res-corked rely on for chunk count).
  let no_copy = matches!(session.padding_strategy, PaddingStrategy::Aligned);

  let mut amount: usize = 0;
  let mut need_eof = false;
  let mut need_trailers = false;
  let mut is_deferred = false;
  let mut have_data = false;
  let mut completed = Vec::new();

  if let Some(stream) = session.find_stream(stream_id) {
    if no_copy {
      let pending_data = stream.pending_data.borrow();
      if !pending_data.is_empty() {
        let amt = std::cmp::min(pending_data.len(), length);
        if amt > 0 {
          amount = amt;
          have_data = true;
          // pending_data is consumed in on_send_data_callback so the
          // bytes survive across the read -> send transition.
          if pending_data.len() == amt && *stream.writable_ended.borrow() {
            need_eof = true;
            need_trailers = stream.has_trailers();
          }
        }
      } else if *stream.writable_ended.borrow() {
        need_eof = true;
        need_trailers = stream.has_trailers();
      } else {
        is_deferred = true;
      }
    } else {
      let mut pending_data = stream.pending_data.borrow_mut();
      if !pending_data.is_empty() {
        let amt = std::cmp::min(pending_data.len(), length);
        if amt > 0 {
          let data_slice = pending_data.split_to(amt);
          // SAFETY: buf has capacity for `length` bytes per nghttp2 contract
          unsafe {
            std::ptr::copy_nonoverlapping(data_slice.as_ptr(), buf, amt)
          };
          amount = amt;

          if pending_data.is_empty() && *stream.writable_ended.borrow() {
            need_eof = true;
            need_trailers = stream.has_trailers();
          }
          drop(pending_data);
          stream.consume_outbound(amt, &mut completed);
        }
      } else if *stream.writable_ended.borrow() {
        need_eof = true;
        need_trailers = stream.has_trailers();
      } else {
        is_deferred = true;
      }
    }
    // pending_data borrow and stream reference are dropped here
  } else {
    return ffi::NGHTTP2_ERR_DEFERRED as _;
  }

  // The stream borrow is released, so the session can be mutated again. The
  // completions are only run by send_pending_data, once this mem_send pass
  // is over.
  session.write_completions.append(&mut completed);

  if is_deferred {
    return ffi::NGHTTP2_ERR_DEFERRED as _;
  }

  if no_copy && have_data {
    // SAFETY: data_flags is a valid out-pointer per nghttp2 contract
    unsafe { *data_flags |= ffi::NGHTTP2_DATA_FLAG_NO_COPY as u32 };
  }

  if need_eof {
    // SAFETY: data_flags is a valid out-pointer per nghttp2 contract
    unsafe { *data_flags |= ffi::NGHTTP2_DATA_FLAG_EOF as u32 };
    if need_trailers {
      // SAFETY: data_flags is a valid out-pointer per nghttp2 contract
      unsafe { *data_flags |= ffi::NGHTTP2_DATA_FLAG_NO_END_STREAM as u32 };
      // Re-lookup stream after dropping previous borrows; the JS
      // callback in on_trailers() could modify session.streams.
      if let Some(stream) = session.find_stream(stream_id) {
        stream.on_trailers();
      }
    }
    // Mark EOF as emitted on this stream so the subsequent shutdown()
    // op (from Writable._final / shutdownWritable) skips its
    // resume_data and nghttp2 doesn't pack a redundant empty trailing
    // DATA frame just to carry END_STREAM.
    if let Some(stream) = session.find_stream(stream_id) {
      *stream.eof_sent.borrow_mut() = true;
    }
    // Complete shutdown now. All borrows have been dropped above,
    // so it's safe to call into JS. This must happen before
    // on_stream_close_callback fires (also during mem_send) so that
    // writableFinished is true when the close event is emitted.
    if let Some(stream) = session.find_stream(stream_id) {
      stream.complete_shutdown();
    }
  }

  amount as CSsizeT
}

unsafe extern "C" fn on_select_padding(
  _session: *mut ffi::nghttp2_session,
  frame: *const ffi::nghttp2_frame,
  max_payload_len: usize,
  user_data: *mut c_void,
) -> CSsizeT {
  // SAFETY: user_data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(user_data) };
  let padding = frame_header_length(frame);

  let result = match session.padding_strategy {
    PaddingStrategy::None => padding,
    PaddingStrategy::Max => {
      session.on_max_frame_size_padding(padding, max_payload_len)
    }
    PaddingStrategy::Aligned | PaddingStrategy::Callback => {
      session.on_dword_aligned_padding(padding, max_payload_len)
    }
  };

  result as CSsizeT
}

unsafe extern "C" fn on_frame_not_send_callback(
  _session: *mut ffi::nghttp2_session,
  frame: *const ffi::nghttp2_frame,
  lib_error_code: i32,
  data: *mut c_void,
) -> i32 {
  // Per Node.js parity, swallow events for frames that fail because the
  // session/stream is already closing — the close path will surface the
  // error through other channels.
  if lib_error_code == ffi::NGHTTP2_ERR_SESSION_CLOSING
    || lib_error_code == ffi::NGHTTP2_ERR_STREAM_CLOSED
    || lib_error_code == ffi::NGHTTP2_ERR_STREAM_CLOSING
  {
    return 0;
  }

  // SAFETY: data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(data) };

  let id = frame_id(frame);
  let ftype = frame_type(frame);
  let translated = translate_nghttp2_error_code(lib_error_code);

  let mut isolate =
    // SAFETY: isolate pointer is valid for the session's lifetime
    unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
  v8::scope!(let scope, &mut isolate);
  let context = v8::Local::new(scope, session.context.clone());
  let scope = &mut v8::ContextScope::new(scope, context);

  let stream_id = v8::Integer::new(scope, id);
  let frame_type_v = v8::Integer::new_from_unsigned(scope, ftype as u32);
  let code = v8::Integer::new_from_unsigned(scope, translated);

  let Some(this) = session.this.as_ref() else {
    return 0;
  };
  let state = session.op_state.borrow();
  let callbacks = state.borrow::<SessionCallbacks>();
  let recv = v8::Local::new(scope, this);
  let callback = v8::Local::new(scope, &callbacks.frame_error_cb);
  drop(state);

  callback.call(
    scope,
    recv.into(),
    &[stream_id.into(), frame_type_v.into(), code.into()],
  );
  0
}

// Maps an nghttp2 library error (negative) to the HTTP/2 protocol error
// code (RFC 7540 §7) surfaced to JavaScript, matching Node's
// TranslateNghttp2ErrorCode in src/node_http2.cc.
fn translate_nghttp2_error_code(lib_err: i32) -> u32 {
  match lib_err {
    ffi::NGHTTP2_ERR_STREAM_CLOSED => 5, // NGHTTP2_STREAM_CLOSED
    ffi::NGHTTP2_ERR_HEADER_COMP => 9,   // NGHTTP2_COMPRESSION_ERROR
    ffi::NGHTTP2_ERR_FRAME_SIZE_ERROR => 6, // NGHTTP2_FRAME_SIZE_ERROR
    ffi::NGHTTP2_ERR_FLOW_CONTROL => 3,  // NGHTTP2_FLOW_CONTROL_ERROR
    ffi::NGHTTP2_ERR_REFUSED_STREAM => 7, // NGHTTP2_REFUSED_STREAM
    ffi::NGHTTP2_ERR_PROTO
    | ffi::NGHTTP2_ERR_HTTP_HEADER
    | ffi::NGHTTP2_ERR_HTTP_MESSAGING => 1, // NGHTTP2_PROTOCOL_ERROR
    _ => 2,                              // NGHTTP2_INTERNAL_ERROR
  }
}

unsafe extern "C" fn on_invalid_header_callback(
  _session: *mut ffi::nghttp2_session,
  _frame: *const ffi::nghttp2_frame,
  _name: *mut ffi::nghttp2_rcbuf,
  _value: *mut ffi::nghttp2_rcbuf,
  _flags: u8,
  _data: *mut c_void,
) -> i32 {
  0
}

unsafe extern "C" fn on_nghttp_error_callback(
  _session: *mut ffi::nghttp2_session,
  _lib_error_code: i32,
  _msg: *const std::ffi::c_char,
  _len: usize,
  _data: *mut c_void,
) -> i32 {
  0
}

unsafe extern "C" fn on_send_data_callback(
  _session: *mut ffi::nghttp2_session,
  frame: *mut ffi::nghttp2_frame,
  framehd: *const u8,
  length: usize,
  _source: *mut ffi::nghttp2_data_source,
  user_data: *mut c_void,
) -> i32 {
  // SAFETY: user_data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(user_data) };

  // SAFETY: frame is valid for the duration of the callback. `data`
  // and `hd` are union fields in `nghttp2_frame`; accessing them is
  // unsafe but valid because pack_data set the active variant before
  // dispatching this callback.
  let (stream_id, padlen) = unsafe {
    let frame = &*frame;
    // padlen is total trailing padding INCLUDING the pad_length byte,
    // so pad_length value (the byte itself) = padlen - 1, and the
    // zero-padding bytes count = padlen - 1.
    (frame.hd.stream_id, frame.data.padlen)
  };

  // Header (NGHTTP2_FRAME_HDLEN = 9 bytes per RFC 7540 §4.1) packed
  // into the framebuf by pack_data; emit it as one chunk on the wire.
  let header_len = 9usize;
  // SAFETY: framehd points to the header packed by nghttp2; we copy
  // it before returning so it doesn't outlive the callback.
  let header = unsafe { std::slice::from_raw_parts(framehd, header_len) };
  session.outgoing_chunks.push_back(header.to_vec());

  if padlen > 0 {
    // Pad length octet: value is the trailing-zero byte count, i.e.
    // padlen - 1. nghttp2 reserved this byte slot in the framebuf via
    // frame_set_pad's `framehd_only` shift but didn't fill it.
    session.outgoing_chunks.push_back(vec![(padlen - 1) as u8]);
  }

  if length > 0 {
    let mut completed = Vec::new();
    let chunk_opt = session.find_stream(stream_id).and_then(|stream| {
      let mut pending = stream.pending_data.borrow_mut();
      let amt = std::cmp::min(pending.len(), length);
      if amt > 0 {
        let chunk = pending.split_to(amt);
        drop(pending);
        stream.consume_outbound(amt, &mut completed);
        Some(chunk.to_vec())
      } else {
        None
      }
    });
    // Borrow released; safe to mutate session now.
    if let Some(chunk) = chunk_opt {
      session.outgoing_chunks.push_back(chunk);
    }
    session.write_completions.append(&mut completed);
    // If pending shrank short of `length` (shouldn't happen — read
    // callback already promised this many bytes), the wire will be
    // short by the missing amount; nghttp2 will fail the session via
    // its error callback rather than silently corrupt the stream.
  }

  if padlen > 1 {
    // Trailing padding bytes (zeros). padlen - 1 of them.
    session
      .outgoing_chunks
      .push_back(vec![0u8; (padlen - 1) as usize]);
  }

  0
}

unsafe extern "C" fn on_invalid_frame_recv_callback(
  _session: *mut ffi::nghttp2_session,
  _frame: *const ffi::nghttp2_frame,
  lib_error_code: i32,
  data: *mut c_void,
) -> i32 {
  // SAFETY: data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(data) };
  let count = session.invalid_frame_count;
  session.invalid_frame_count = count.saturating_add(1);
  // Node.js compares post-increment against the limit (see node_http2.cc
  // OnInvalidFrame: `if (invalid_frame_count_++ > max_invalid_frames)`).
  // Returning a non-zero value tells nghttp2 to terminate the session
  // immediately; receive_data picks up the custom error code below and
  // surfaces it to JS as ERR_HTTP2_TOO_MANY_INVALID_FRAMES.
  if count > session.max_invalid_frames {
    session.custom_recv_error_code = Some("ERR_HTTP2_TOO_MANY_INVALID_FRAMES");
    return 1;
  }
  // Surface protocol-level violations (e.g. server attempting to disable
  // SETTINGS_ENABLE_CONNECT_PROTOCOL after enabling it) to JS as a session
  // 'error' event. Mirrors Node.js OnInvalidFrame which forwards specific
  // lib error codes to http2session_on_error_function.
  // SAFETY: nghttp2_is_fatal is a pure function on the lib_error_code.
  let is_fatal = unsafe { ffi::nghttp2_is_fatal(lib_error_code) } != 0;
  if is_fatal
    || lib_error_code == ffi::NGHTTP2_ERR_PROTO as i32
    || lib_error_code == ffi::NGHTTP2_ERR_STREAM_CLOSED as i32
    || lib_error_code == ffi::NGHTTP2_ERR_FLOW_CONTROL as i32
  {
    // Do not enter JS while nghttp2_session_mem_recv is still active. The JS
    // callback destroys the session and submits a GOAWAY, both of which may
    // mutate nghttp2 state. send_pending_data emits this immediately after
    // mem_recv unwinds.
    session.pending_internal_error = Some(lib_error_code);
  }
  0
}

unsafe extern "C" fn on_frame_send_callback(
  _session: *mut ffi::nghttp2_session,
  frame: *const ffi::nghttp2_frame,
  data: *mut c_void,
) -> i32 {
  // SAFETY: data is the user_data pointer set during session creation
  let session = unsafe { Session::from_user_data(data) };
  session.frames_sent = session.frames_sent.saturating_add(1);
  // SAFETY: frame is valid per nghttp2 callback contract
  let f = unsafe { &*frame };
  // SAFETY: union access of `hd` is always valid (every nghttp2 frame has a
  // header). Reading `type_` to discriminate which other variant is active.
  let frame_type = unsafe { f.hd.type_ };
  if frame_type == ffi::NGHTTP2_GOAWAY as u8 {
    // SAFETY: union access valid because we verified type_ == NGHTTP2_GOAWAY
    let goaway = unsafe { &f.goaway };
    session.sent_goaway_code = Some(goaway.error_code);
    // When nghttp2 detects a connection-level violation while parsing inbound
    // bytes (e.g. an unknown extension frame whose declared length exceeds
    // SETTINGS_MAX_FRAME_SIZE, or any other terminate_session_with_reason
    // path), it queues a GOAWAY whose error_code carries the protocol-level
    // reason. mem_recv itself returns success in this case, so without
    // observing the outgoing GOAWAY we'd lose the error and the JS layer
    // would just see a graceful close. Mirror Node's
    // `Http2Session::OnFrameSent` GOAWAY branch: if the GOAWAY we're sending
    // isn't NGHTTP2_NO_ERROR, surface it to JS through
    // `onSessionInternalError(NGHTTP2_ERR_PROTO)` so the session is destroyed
    // with the same `NghttpError("Protocol error")` Node produces.
    if session.pending_user_goaway > 0 {
      // User-initiated GOAWAY (from the `goaway()` op): consume one pending
      // marker and stay quiet. nghttp2 only ever sends one GOAWAY per
      // submit, so the user counter mirrors what we put on the wire.
      //
      // Known race: the marker is associated with a *count*, not with a
      // specific outgoing frame. If a user `goaway()` and an internal
      // `terminate_session_with_reason` GOAWAY are queued simultaneously and
      // ship in the opposite order from how they were submitted, we'll
      // attribute the internal GOAWAY to the user (no protocol error fired)
      // and the user GOAWAY to nghttp2 (may fire if the user passed a
      // non-NO_ERROR code and SETTINGS hasn't arrived yet). This is
      // acceptable since racing `destroy()` against parser-induced internal
      // GOAWAYs is not a real-world program shape, and the worst-case
      // outcome is just a misclassified error -- not a crash or hang.
      session.pending_user_goaway =
        session.pending_user_goaway.saturating_sub(1);
    } else if goaway.error_code != ffi::NGHTTP2_NO_ERROR as u32
      && !session.protocol_error_emitted
      && ((session.is_client && !session.remote_settings_received)
        || (!session.is_client
          && goaway.error_code == ffi::NGHTTP2_FRAME_SIZE_ERROR as u32))
    {
      // Surface client handshake failures and server-side frame-size
      // violations as `NghttpError("Protocol error")`, matching Node's
      // `test-http2-client-http1-server` and
      // `test-http2-session-cleanup-on-nghttp2-goaway` behavior.
      //
      // Other late server-side GOAWAYs stay on the normal close path; treating
      // all of them as protocol errors regresses legitimate shutdowns. The
      // deferred emission below also ensures the GOAWAY bytes are handed to
      // the socket before JS destroys the session.
      session.protocol_error_emitted = true;
      // Do not call into JS from inside nghttp2_session_mem_send. The JS
      // callback destroys the session, which would otherwise release the
      // nghttp2 allocation while mem_send still has it on the stack. The
      // native send path emits this after queueing the frame; the JS socket
      // path emits it after getOutgoingChunk has returned the frame to JS.
      session.pending_internal_error = Some(ffi::NGHTTP2_ERR_PROTO);
    }
  }
  0
}

/// Detect stream-level flow-control violations on inbound DATA frames before
/// nghttp2 processes the payload. Upstream nghttp2 (>= 1.65) reacts to a
/// stream-level overflow by tearing down the *whole session* with GOAWAY
/// (NGHTTP2_FLOW_CONTROL_ERROR), which closes every active stream with code
/// `NGHTTP2_REFUSED_STREAM`. Node ships an older vendored nghttp2 whose
/// `nghttp2_session_update_recv_stream_window_size` calls
/// `nghttp2_session_add_rst_stream(stream_id, NGHTTP2_FLOW_CONTROL_ERROR)`
/// instead, so the offending stream alone closes with
/// `NGHTTP2_FLOW_CONTROL_ERROR`. Tests like
/// `parallel/test-http2-misbehaving-flow-control-paused` assert the
/// stream-level form, so we replicate it here:
///
///   1. peek the DATA frame header
///   2. if `effective_recv_data_length + length > effective_local_window_size`,
///      bump the stream's local window to NGHTTP2_MAX_WINDOW_SIZE so nghttp2's
///      own check won't fire (and won't tear down the whole session), and
///   3. submit `RST_STREAM(NGHTTP2_FLOW_CONTROL_ERROR)` for the offending
///      stream, so when nghttp2 closes it the registered close callback emits
///      `ERR_HTTP2_STREAM_ERROR("Stream closed with error code NGHTTP2_FLOW_CONTROL_ERROR")`.
unsafe extern "C" fn on_begin_frame_callback(
  ng_session: *mut ffi::nghttp2_session,
  hd: *const ffi::nghttp2_frame_hd,
  _user_data: *mut c_void,
) -> i32 {
  // SAFETY: hd is valid per nghttp2 callback contract
  let hd = unsafe { &*hd };
  if hd.type_ != ffi::NGHTTP2_DATA as u8 || hd.stream_id == 0 {
    return 0;
  }

  // SAFETY: ng_session is valid per nghttp2 callback contract
  let eff_local_window = unsafe {
    ffi::nghttp2_session_get_stream_effective_local_window_size(
      ng_session,
      hd.stream_id,
    )
  };
  if eff_local_window < 0 {
    return 0;
  }
  // SAFETY: ng_session is valid per nghttp2 callback contract
  let eff_recv = unsafe {
    ffi::nghttp2_session_get_stream_effective_recv_data_length(
      ng_session,
      hd.stream_id,
    )
  };
  if eff_recv < 0 {
    return 0;
  }
  let projected = (eff_recv as i64) + (hd.length as i64);
  if projected <= eff_local_window as i64 {
    return 0;
  }

  // Disable nghttp2's own flow-control check for this stream by raising
  // its local window to the protocol maximum. We've already decided to
  // RST_STREAM the offender; we don't want nghttp2 to also terminate the
  // entire session before our RST_STREAM frame ships out.
  // SAFETY: ng_session is valid per nghttp2 callback contract
  unsafe {
    // NGHTTP2_MAX_WINDOW_SIZE = (1 << 31) - 1 (RFC 7540).
    ffi::nghttp2_session_set_local_window_size(
      ng_session,
      ffi::NGHTTP2_FLAG_NONE as u8,
      hd.stream_id,
      i32::MAX,
    );
    ffi::nghttp2_submit_rst_stream(
      ng_session,
      ffi::NGHTTP2_FLAG_NONE as u8,
      hd.stream_id,
      ffi::NGHTTP2_FLOW_CONTROL_ERROR as u32,
    );
  }
  0
}

fn create_callbacks() -> *mut ffi::nghttp2_session_callbacks {
  let mut callbacks: *mut ffi::nghttp2_session_callbacks = std::ptr::null_mut();

  // SAFETY: passing valid pointer to be initialized by nghttp2
  unsafe {
    assert_eq!(ffi::nghttp2_session_callbacks_new(&mut callbacks), 0);

    ffi::nghttp2_session_callbacks_set_on_begin_headers_callback(
      callbacks,
      Some(on_begin_headers_callbacks),
    );
    ffi::nghttp2_session_callbacks_set_on_header_callback2(
      callbacks,
      Some(on_header_callback),
    );
    ffi::nghttp2_session_callbacks_set_on_frame_recv_callback(
      callbacks,
      Some(on_frame_recv_callback),
    );
    ffi::nghttp2_session_callbacks_set_on_stream_close_callback(
      callbacks,
      Some(on_stream_close_callback),
    );
    ffi::nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
      callbacks,
      Some(on_data_chunk_recv_callback),
    );
    ffi::nghttp2_session_callbacks_set_on_frame_not_send_callback(
      callbacks,
      Some(on_frame_not_send_callback),
    );
    ffi::nghttp2_session_callbacks_set_on_invalid_header_callback2(
      callbacks,
      Some(on_invalid_header_callback),
    );
    ffi::nghttp2_session_callbacks_set_error_callback2(
      callbacks,
      Some(on_nghttp_error_callback),
    );
    ffi::nghttp2_session_callbacks_set_send_data_callback(
      callbacks,
      Some(on_send_data_callback),
    );
    ffi::nghttp2_session_callbacks_set_on_invalid_frame_recv_callback(
      callbacks,
      Some(on_invalid_frame_recv_callback),
    );
    ffi::nghttp2_session_callbacks_set_on_frame_send_callback(
      callbacks,
      Some(on_frame_send_callback),
    );
    ffi::nghttp2_session_callbacks_set_select_padding_callback2(
      callbacks,
      Some(on_select_padding),
    );
    ffi::nghttp2_session_callbacks_set_on_begin_frame_callback(
      callbacks,
      Some(on_begin_frame_callback),
    );
  }

  callbacks
}

// Session

#[allow(dead_code, reason = "fields are stored for prevent GC of v8 handles")]
pub struct SessionCallbacks {
  pub session_internal_error_cb: v8::Global<v8::Function>,
  pub priority_frame_cb: v8::Global<v8::Function>,
  pub settings_frame_cb: v8::Global<v8::Function>,
  pub ping_frame_cb: v8::Global<v8::Function>,
  pub headers_frame_cb: v8::Global<v8::Function>,
  pub frame_error_cb: v8::Global<v8::Function>,
  pub goaway_data_cb: v8::Global<v8::Function>,
  pub alt_svc_cb: v8::Global<v8::Function>,
  pub stream_trailers_cb: v8::Global<v8::Function>,
  pub stream_close_cb: v8::Global<v8::Function>,
  pub origin_frame_cb: v8::Global<v8::Function>,
}

#[derive(Debug)]
pub struct NgHttp2StreamWrite {
  pub data: bytes::Bytes,
  #[allow(dead_code, reason = "stored for debugging")]
  pub stream_id: i32,
}

impl NgHttp2StreamWrite {
  pub fn new(data: bytes::Bytes, stream_id: i32) -> Self {
    Self { data, stream_id }
  }

  pub fn len(&self) -> usize {
    self.data.len()
  }
}

/// A finished stream write waiting for its JS `oncomplete` to run.
pub struct WriteCompletion {
  /// The JS `WriteWrap` handed to `writeBuffer` / `writeUtf8String`.
  pub req: v8::Global<v8::Object>,
  /// 0 once nghttp2 framed the bytes, or a negative libuv errno if the write
  /// was cancelled before that could happen.
  pub status: i32,
}

pub struct Session {
  pub session: *mut ffi::nghttp2_session,
  native: Rc<NativeResources>,
  pub streams: HashMap<i32, (v8::Global<v8::Object>, cppgc::Ref<Http2Stream>)>,
  /// Stream writes that have completed since the last flush. The nghttp2
  /// callbacks that drain `pending_data` must not re-enter JS, so they park
  /// completions here and `send_pending_data` runs them once `mem_send` is
  /// done. Mirrors how Node defers write completion to
  /// `Http2Session::ClearOutgoing` (`src/node_http2.cc`).
  pub write_completions: Vec<WriteCompletion>,
  pub outgoing_buffers: Vec<NgHttp2StreamWrite>,
  pub outgoing_length: usize,
  pub isolate: v8::UnsafeRawIsolatePtr,
  pub context: v8::Global<v8::Context>,
  pub op_state: Rc<RefCell<OpState>>,
  pub this: Option<v8::Global<v8::Object>>,
  pub padding_strategy: PaddingStrategy,
  pub graceful_close_initiated: bool,
  pub stream: Option<*mut deno_core::uv_compat::UvStream>,
  /// Prevents recursive send_pending_data calls. When mem_recv fires
  /// JS callbacks that call back into send_pending_data, the nested
  /// mem_send can close/free streams that mem_recv still references
  /// (double-free). Like Node.js's is_sending() guard.
  pub is_sending: bool,
  /// Set when destroy() is called while is_sending is true. The TCP
  /// handle close is deferred until send_pending_data can run, allowing
  /// GOAWAY to be sent before the connection closes.
  pub pending_destroy: bool,
  /// True while `get_outgoing_chunk` is inside `nghttp2_session_mem_send`
  /// on the JS-socket transport. It complements `is_sending`: native
  /// teardown uses `is_sending` to defer resource release, while write
  /// completions use this flag to remain parked until the outer drain's
  /// trailing send pass. This prevents producer JS from nesting a second
  /// `mem_send` while the outer one is still on the stack (double-free with
  /// `no_closed_streams=1`).
  pub draining_outgoing: bool,
  /// RST_STREAM submissions deferred because is_sending was true.
  /// Matches Node.js's pending_rst_streams_ mechanism: submitting
  /// RST_STREAM during mem_recv/mem_send can cause nghttp2 to
  /// double-free a stream (with no_closed_streams=1). Instead, we
  /// defer and flush them after send_pending_data completes.
  pub pending_rst_streams: Vec<(i32, u32)>,
  /// Maximum number of header pairs allowed per stream. Mirrors Node.js's
  /// per-session limit derived from the `maxHeaderListPairs` option.
  /// Streams that receive more headers are reset with NGHTTP2_ENHANCE_YOUR_CALM.
  pub max_header_pairs: u32,
  /// Pre-formatted chunks queued by `on_send_data_callback` (NO_COPY DATA
  /// frames). `get_outgoing_chunk` drains this before falling through to
  /// `nghttp2_session_mem_send`, so each part of a padded DATA frame
  /// (header / pad_length / payload / padding) becomes its own
  /// socket.write, matching the per-uv_buf_t output of Node's libuv send
  /// path that test-http2-padding-aligned asserts.
  pub outgoing_chunks: VecDeque<Vec<u8>>,
  /// Maximum number of invalid HTTP/2 frames the peer may send before the
  /// session is terminated. Mirrors Node.js's `maxSessionInvalidFrames`
  /// option (default 1000). The check uses post-increment comparison
  /// (`count++ > max`), so `0` means the second invalid frame triggers.
  pub max_invalid_frames: u32,
  /// Running count of invalid frames received on this session.
  pub invalid_frame_count: u32,
  /// Total HTTP/2 frames received on this session, used for the
  /// `framesReceived` field of perf_hooks `Http2Session` entries.
  pub frames_received: u32,
  /// Total HTTP/2 frames sent on this session, used for the
  /// `framesSent` field of perf_hooks `Http2Session` entries.
  pub frames_sent: u32,
  /// Custom error code set by an nghttp2 callback when it returns a fatal
  /// error so that `receive_data` can surface it to JS via
  /// `session_internal_error_cb`. Mirrors Node's
  /// `Http2Session::custom_recv_error_code_`.
  pub custom_recv_error_code: Option<&'static str>,
  /// Error code from the last GOAWAY frame this side sent. Set in
  /// `on_frame_send_callback`. The JS layer reads it via
  /// `handle.lastSentGoawayCode()` so that streams torn down because
  /// nghttp2 has already terminated the session can carry the actual
  /// connection-level error (e.g. NGHTTP2_FLOW_CONTROL_ERROR) instead of
  /// being papered over with NGHTTP2_CANCEL.
  pub sent_goaway_code: Option<u32>,
  /// Custom settings the local peer has sent. Surfaced via
  /// `session.localSettings.customSettings`. Mirrors Node's
  /// `Http2Session::local_custom_settings_`.
  pub local_custom_settings: Vec<(i32, u32)>,
  /// Custom settings the remote peer is allowed to send (`remoteCustomSettings`
  /// option) plus their last-received values. The high bit (1 << 16) of the
  /// stored ID flags an entry as "registered but no value received yet".
  /// Mirrors Node's `Http2Session::remote_custom_settings_`.
  pub remote_custom_settings: Vec<(i32, u32)>,
  /// FIFO of JS callbacks waiting for SETTINGS ACK from the peer. Each
  /// `session.settings()` invocation enqueues one entry; an inbound SETTINGS
  /// frame with the ACK flag pops the front and invokes it with
  /// `(ack=true, duration=0)`. Mirrors Node's `outstanding_settings_` queue.
  ///
  /// Invariant: every SETTINGS frame submitted via `Http2Settings::send`
  /// (including the constructor's initial settings, which the JS layer
  /// submits via `Http2Session.prototype.settings` with a bound
  /// `settingsCallback`) pushes exactly one entry; every inbound ACK pops
  /// exactly one. Because the only call site is `fn settings(cb)`, push and
  /// submit are paired atomically.
  pub pending_settings_acks: VecDeque<v8::Global<v8::Function>>,
  /// Count of outstanding outbound PINGs awaiting an ACK from the peer.
  /// Mirrors Node's `outstanding_pings_` queue depth. Receiving a PING ACK
  /// when this is zero is treated as a connection-level protocol error
  /// (see `HandlePingFrame` in node_http2.cc): there is no legitimate
  /// reason for a peer to send an unsolicited PING ACK.
  pub pending_pings: u32,
  /// True after we've surfaced an internal protocol error to JS. Used by
  /// `on_frame_send_callback`'s GOAWAY hook so we don't re-fire the
  /// `onSessionInternalError` JS callback for every GOAWAY in a teardown
  /// sequence (e.g. a peer GOAWAY echo).
  pub protocol_error_emitted: bool,
  /// Error discovered while serializing an outgoing frame. Emitted only
  /// after nghttp2_session_mem_send has returned so JS teardown cannot free
  /// the active nghttp2 session reentrantly.
  pub pending_internal_error: Option<i32>,
  /// Number of GOAWAYs the user has submitted via the `goaway()` op that
  /// have not yet drained through `on_frame_send_callback`. Used to skip
  /// the protocol-error hook for user-initiated GOAWAYs (e.g. a normal
  /// `session.destroy(code)`); only nghttp2-internal GOAWAYs (queued from
  /// `terminate_session_with_reason` after a protocol/frame-size violation)
  /// should surface as `NghttpError("Protocol error")` to JS.
  pub pending_user_goaway: u32,
  /// Set true once we receive any SETTINGS frame from the peer. We use this
  /// to gate the `on_frame_send_callback` protocol-error hook so it only
  /// fires for handshake-time failures (where nghttp2 internally tears the
  /// session down before SETTINGS exchange completes). Late connection-level
  /// GOAWAYs originating from server-side state (e.g. timeout-driven
  /// tear-downs that don't go through the user `goaway()` op) flow through
  /// the normal close path instead.
  pub remote_settings_received: bool,
  /// True if this is a client session, false for server. Used to distinguish
  /// pre-SETTINGS client handshake failures from the narrow server-side
  /// frame-size error path in `on_frame_send_callback`.
  pub is_client: bool,
}

impl Session {
  /// Read the list of remote custom settings IDs the user wants to track.
  /// JS writes them into the shared settings buffer via
  /// `remoteCustomSettingsToBuffer` immediately before constructing the
  /// session. Mirrors Node's `Http2Session::FetchAllowedRemoteCustomSettings`.
  pub fn fetch_allowed_remote_custom_settings(&mut self) {
    with_settings(|buffer| {
      let count = buffer[SettingsIndex::Count as usize + 1] as usize;
      if count == 0 {
        return;
      }
      let imax = count.min(MAX_ADDITIONAL_SETTINGS);
      let offset = SettingsIndex::Count as usize + 2;
      for i in 0..imax {
        let key = buffer[offset + i * 2] & 0xffff;
        // bit 16 marks "registered but not yet received".
        let marked = (key | (1 << 16)) as i32;
        self.remote_custom_settings.push((marked, 0));
      }
      // Reset the count so a later read of localSettings/remoteSettings on
      // a session without any local custom settings doesn't see leftover.
      buffer[SettingsIndex::Count as usize + 1] = 0;
    });
  }

  /// Record the custom settings we just sent so they can be surfaced via
  /// `session.localSettings.customSettings`. Mirrors Node's
  /// `Http2Session::UpdateLocalCustomSettings`.
  pub fn update_local_custom_settings(
    &mut self,
    entries: &[ffi::nghttp2_settings_entry],
  ) {
    for entry in entries {
      // Standard nghttp2 setting IDs are 1..=6 and 8. Node treats anything
      // >= IDX_SETTINGS_COUNT (7) as custom; we follow the same rule.
      if entry.settings_id < SettingsIndex::Count as i32 {
        continue;
      }
      let id = entry.settings_id;
      let mut updated = false;
      for slot in self.local_custom_settings.iter_mut() {
        if slot.0 == id {
          slot.1 = entry.value;
          updated = true;
          break;
        }
      }
      if !updated && self.local_custom_settings.len() < MAX_ADDITIONAL_SETTINGS
      {
        self.local_custom_settings.push((id, entry.value));
      }
    }
  }

  /// Walk a received SETTINGS frame's iv array and update the registered
  /// remote custom settings with the new values. Settings whose IDs were
  /// not registered via `remoteCustomSettings` are dropped, matching Node's
  /// behaviour in `Http2Session::HandleSettingsFrame`.
  pub fn update_remote_custom_settings_from_iv(
    &mut self,
    iv: &[ffi::nghttp2_settings_entry],
  ) {
    if self.remote_custom_settings.is_empty() {
      return;
    }
    for entry in iv {
      if entry.settings_id < SettingsIndex::Count as i32 {
        continue;
      }
      let id_lo = (entry.settings_id as u32) & 0xffff;
      for slot in self.remote_custom_settings.iter_mut() {
        if (slot.0 as u32) & 0xffff == id_lo {
          slot.0 = id_lo as i32; // clear bit 16 to mark "received".
          slot.1 = entry.value;
          break;
        }
      }
    }
  }

  pub fn find_stream(&self, id: i32) -> Option<&cppgc::Ref<Http2Stream>> {
    self.streams.get(&id).map(|v| &v.1)
  }

  pub fn find_stream_obj(&self, id: i32) -> Option<&v8::Global<v8::Object>> {
    self.streams.get(&id).map(|v| &v.0)
  }

  pub fn push_outgoing_buffer(&mut self, write: NgHttp2StreamWrite) {
    self.outgoing_length += write.len();
    self.outgoing_buffers.push(write);
  }

  pub fn clear_outgoing(&mut self) {
    self.outgoing_buffers.clear();
    self.outgoing_length = 0;
  }

  /// Run `oncomplete` for every write that finished since the last flush,
  /// releasing the producer that was waiting on it. Mirrors Node's
  /// `Http2Session::ClearOutgoing` (`src/node_http2.cc`).
  ///
  /// Callers must be outside any nghttp2 `mem_send` / `mem_recv` pass, since
  /// each `oncomplete` runs arbitrary JS.
  pub fn flush_write_completions(&mut self) {
    if self.write_completions.is_empty() {
      return;
    }

    // An `oncomplete` runs arbitrary JS that can call `stream.write()` and
    // re-enter the send path. On the libuv transport `send_pending_data`
    // clears `is_sending` before flushing, so any nested send is blocked by
    // its own guard. The JS-socket transport instead drains via
    // `get_outgoing_chunk`, which sets `draining_outgoing` around its
    // `mem_send`; flushing there would let the nested write start a second
    // `mem_send` inside the outer one (double-free, `no_closed_streams=1`).
    // Leave the completions queued — the drain's trailing send pass
    // (`origSendPending` -> `send_pending_data`) or the session teardown
    // flush runs them once the outer `mem_send` has unwound.
    if self.draining_outgoing {
      return;
    }

    // Take the queue up front: an `oncomplete` can write again and re-enter
    // send_pending_data, which must not see entries already being completed.
    let completions = std::mem::take(&mut self.write_completions);

    // SAFETY: isolate pointer is valid during session lifetime
    let mut isolate =
      unsafe { v8::Isolate::from_raw_isolate_ptr(self.isolate) };
    v8::scope!(let scope, &mut isolate);
    let context = v8::Local::new(scope, self.context.clone());
    let scope = &mut v8::ContextScope::new(scope, context);
    let key =
      v8::String::new_external_onebyte_static(scope, b"oncomplete").unwrap();

    for completion in completions {
      let req = v8::Local::new(scope, completion.req);
      if let Some(oncomplete) = req.get(scope, key.into())
        && let Ok(oncomplete) = v8::Local::<v8::Function>::try_from(oncomplete)
      {
        let status = v8::Integer::new(scope, completion.status);
        oncomplete.call(scope, req.into(), &[status.into()]);
      }
    }
  }

  /// Pull the next outgoing chunk from nghttp2's send queue. The body of the
  /// `get_outgoing_chunk` op, split out so the op can bracket it with the
  /// `draining_outgoing` re-entrancy guard on every return path.
  fn drain_outgoing_chunk(&mut self) -> Box<[u8]> {
    loop {
      if let Some(chunk) = self.outgoing_chunks.pop_front() {
        return chunk.into_boxed_slice();
      }
      let mut src = std::ptr::null();
      let src_len =
        // SAFETY: self.session is a valid nghttp2 session pointer
        unsafe { ffi::nghttp2_session_mem_send(self.session, &mut src) };
      if src_len > 0 {
        // SAFETY: src and src_len are valid per nghttp2_session_mem_send
        let data = unsafe { std::slice::from_raw_parts(src, src_len as usize) };
        return data.to_vec().into_boxed_slice();
      }
      if src_len < 0 {
        // SAFETY: nghttp2_strerror returns a static C string for any input
        let msg = unsafe {
          let p = ffi::nghttp2_strerror(src_len as i32);
          std::ffi::CStr::from_ptr(p).to_string_lossy()
        };
        log::debug!("nghttp2_session_mem_send failed: {} ({})", msg, src_len);
        return Box::new([]);
      }
      // src_len == 0: nghttp2 has nothing more directly, but
      // on_send_data_callback may have just pushed chunks (NO_COPY
      // DATA) — loop and pop them. If the queue is also empty, the
      // next iteration's pop returns None and we hit mem_send again
      // which returns 0, exiting via the early return below.
      if self.outgoing_chunks.is_empty() {
        return Box::new([]);
      }
    }
  }

  /// Submit RST_STREAM, or defer it if we're inside mem_recv/mem_send.
  /// Matches Node.js's Http2Stream::SubmitRstStream: submitting
  /// RST_STREAM while nghttp2 is processing frames can cause
  /// double-free with no_closed_streams=1.
  pub fn submit_rst_stream(&mut self, stream_id: i32, code: u32) {
    if self.session.is_null() {
      return;
    }
    if self.is_sending {
      self.pending_rst_streams.push((stream_id, code));
      return;
    }

    // Submit RST_STREAM before flushing so nghttp2 can prioritise it over
    // any queued DATA frames (e.g. an END_STREAM data frame queued by a
    // prior stream.end()). Flushing first would push the END_STREAM frame
    // and let nghttp2 free the stream (no_closed_streams=1), after which
    // the RST_STREAM would be silently dropped and the peer would never
    // see an error. This matches Node.js's SubmitRstStream which calls
    // nghttp2_submit_rst_stream directly and then schedules a write.
    // SAFETY: self.session is a valid nghttp2 session pointer
    let stream_ptr =
      unsafe { ffi::nghttp2_session_find_stream(self.session, stream_id) };
    if stream_ptr.is_null() {
      return;
    }

    // SAFETY: self.session is a valid nghttp2 session pointer
    unsafe {
      ffi::nghttp2_submit_rst_stream(
        self.session,
        ffi::NGHTTP2_FLAG_NONE as u8,
        stream_id,
        code,
      );
    }

    self.send_pending_data();
  }

  /// Flush deferred RST_STREAM submissions. Called after
  /// send_pending_data completes.
  fn flush_pending_rst_streams(&mut self) {
    if self.pending_rst_streams.is_empty() {
      return;
    }
    if self.session.is_null() {
      self.pending_rst_streams.clear();
      return;
    }
    let pending: Vec<_> = std::mem::take(&mut self.pending_rst_streams);
    for (stream_id, code) in pending {
      // Check if the stream still exists.
      // SAFETY: self.session is a valid nghttp2 session pointer
      let stream_ptr =
        unsafe { ffi::nghttp2_session_find_stream(self.session, stream_id) };
      if stream_ptr.is_null() {
        continue;
      }
      // Submit RST_STREAM before flushing so nghttp2 can prioritise it
      // over any queued DATA frames (e.g. an END_STREAM data frame).
      // SAFETY: self.session is a valid nghttp2 session pointer
      unsafe {
        ffi::nghttp2_submit_rst_stream(
          self.session,
          ffi::NGHTTP2_FLAG_NONE as u8,
          stream_id,
          code,
        );
      }
    }
    self.send_pending_data();
  }

  pub fn is_graceful_closing(&self) -> bool {
    self.graceful_close_initiated
  }

  pub fn start_graceful_close(&mut self) {
    self.graceful_close_initiated = true;
  }

  pub fn active_stream_count(&self) -> usize {
    self.streams.len()
  }

  fn detach_js_refs(&mut self) {
    self.this.take();
    self.streams.clear();
    self.pending_settings_acks.clear();
  }

  fn close_native_resources(&mut self) {
    self.detach_js_refs();
    self.native.close();
    self.session = std::ptr::null_mut();
  }

  fn take_stream_for_close(
    &mut self,
  ) -> Option<*mut deno_core::uv_compat::UvStream> {
    let stream = self.stream.take()?;
    // Stop pending libuv callbacks from observing the Session after native
    // teardown or cppgc finalization. TCPWrap::detach already detached its
    // StreamHandleData, so there is no previous owner to restore here.
    // SAFETY: stream is a valid libuv handle owned by this session.
    unsafe {
      (*stream).data = std::ptr::null_mut();
    }
    Some(stream)
  }

  fn close_stream_gracefully(&mut self) {
    let Some(stream) = self.take_stream_for_close() else {
      return;
    };
    // SAFETY: stream is a valid libuv handle owned by this session.
    unsafe {
      deno_core::uv_compat::uv_read_stop(stream);
      let req = Box::into_raw(Box::new(deno_core::uv_compat::new_shutdown()));
      let ret =
        deno_core::uv_compat::uv_shutdown(req, stream, Some(h2_shutdown_cb));
      if ret != 0 {
        let _ = Box::from_raw(req);
        deno_core::uv_compat::uv_close(
          stream as *mut deno_core::uv_compat::UvHandle,
          Some(h2_stream_close_cb),
        );
      }
    }
  }

  fn finish_pending_destroy(&mut self) {
    self.pending_destroy = false;
    self.close_stream_gracefully();
    self.close_native_resources();
  }

  fn emit_pending_internal_error(&mut self) {
    if let Some(errno) = self.pending_internal_error.take() {
      self.emit_session_internal_error(errno);
    }
  }

  fn close_stream_finalizer_safe(&mut self) {
    let Some(stream) = self.take_stream_for_close() else {
      return;
    };
    // SAFETY: stream is a valid libuv handle owned by this session. This path
    // is used from cppgc Drop and must not call back into JS.
    unsafe {
      deno_core::uv_compat::uv_read_stop(stream);
      deno_core::uv_compat::uv_close(
        stream as *mut deno_core::uv_compat::UvHandle,
        Some(h2_stream_close_cb),
      );
    }
  }

  /// Check if graceful close is complete and notify JS if so.
  /// Mirrors Node.js's MaybeNotifyGracefulCloseComplete in node_http2.cc.
  /// Called after writes complete to detect when the session can be destroyed.
  pub fn maybe_notify_graceful_close_complete(&mut self) {
    if !self.graceful_close_initiated {
      return;
    }
    if self.session.is_null() {
      return;
    }

    let want_write =
      // SAFETY: self.session is a valid nghttp2 session pointer
      unsafe { ffi::nghttp2_session_want_write(self.session) };
    let want_read =
      // SAFETY: self.session is a valid nghttp2 session pointer
      unsafe { ffi::nghttp2_session_want_read(self.session) };

    if want_write != 0 || want_read != 0 {
      return;
    }

    // SAFETY: isolate pointer is valid for the session's lifetime
    let mut isolate =
      unsafe { v8::Isolate::from_raw_isolate_ptr(self.isolate) };
    v8::scope!(let scope, &mut isolate);
    let context = v8::Local::new(scope, self.context.clone());
    let scope = &mut v8::ContextScope::new(scope, context);

    let Some(this) = self.this.as_ref() else {
      return;
    };
    let this_local = v8::Local::new(scope, this);
    let key = v8::String::new(scope, "ongracefulclosecomplete").unwrap();
    if let Some(Ok(cb)) = this_local
      .get(scope, key.into())
      .map(v8::Local::<v8::Function>::try_from)
    {
      cb.call(scope, this_local.into(), &[]);
    }
  }

  pub unsafe fn from_user_data<'a>(user_data: *mut c_void) -> &'a mut Self {
    // SAFETY: caller guarantees user_data points to a valid Session
    unsafe { &mut *(user_data as *mut Session) }
  }

  pub fn send_pending_data(&mut self) {
    if self.session.is_null() {
      return;
    }
    // Prevent recursive calls. JS callbacks from nghttp2_session_mem_recv
    // can call back into send_pending_data via scheduleSendPending. Nested
    // nghttp2_session_mem_send can close/free streams that mem_recv still
    // references (double-free with no_closed_streams=1). Matches Node.js
    // is_sending() guard in SendPendingData.
    if self.is_sending {
      return;
    }
    let stream = match self.stream {
      Some(stream) => stream,
      None => {
        // JS write path: no consumed stream, but still check for graceful
        // close completion. The JS side handles data serialization via
        // getOutgoingChunk/sendPending, but the graceful close notification
        // must still fire when nghttp2 reports no more pending I/O.
        //
        // getOutgoingChunk drives mem_send on this path, so by the time the
        // JS sendPending override calls through to here the writes it framed
        // are ready to complete.
        self.flush_write_completions();
        self.emit_pending_internal_error();
        self.maybe_notify_graceful_close_complete();
        if self.pending_destroy {
          self.finish_pending_destroy();
        }
        return;
      }
    };

    // Safety check: ensure the stream handle is still a valid TCP handle
    let handle_type =
      // SAFETY: stream pointer is valid, stored in self.stream by consume_stream
      unsafe { (*(stream as *mut deno_core::uv_compat::UvHandle)).r#type };
    if handle_type != deno_core::uv_compat::uv_handle_type::UV_TCP {
      self.stream = None;
      self.emit_pending_internal_error();
      if self.pending_destroy {
        self.finish_pending_destroy();
      }
      return;
    }

    self.is_sending = true;
    loop {
      // Drain any chunks queued by on_send_data_callback (NO_COPY DATA
      // frames) before pulling the next frame from nghttp2.
      while let Some(chunk) = self.outgoing_chunks.pop_front() {
        let write_req = Box::new(H2WriteReq {
          uv_req: deno_core::uv_compat::new_write(),
          data: chunk,
        });
        let write_ptr = Box::into_raw(write_req);
        // SAFETY: write_ptr is freshly allocated; stream is a valid libuv handle
        unsafe {
          let buf = deno_core::uv_compat::UvBuf {
            base: (*write_ptr).data.as_ptr() as *mut _,
            len: (*write_ptr).data.len(),
          };
          let ret = deno_core::uv_compat::uv_write(
            &mut (*write_ptr).uv_req,
            stream,
            &buf,
            1,
            Some(h2_write_cb),
          );
          if ret != 0 {
            let _ = Box::from_raw(write_ptr);
          }
        }
      }
      let mut src = std::ptr::null();
      let src_len =
        // SAFETY: self.session is a valid nghttp2 session pointer
        unsafe { ffi::nghttp2_session_mem_send(self.session, &mut src) };

      if src_len > 0 {
        // SAFETY: src and src_len are valid per nghttp2_session_mem_send contract
        let data = unsafe { std::slice::from_raw_parts(src, src_len as usize) };
        // Write to libuv stream
        let data_copy = data.to_vec();
        let write_req = Box::new(H2WriteReq {
          uv_req: deno_core::uv_compat::new_write(),
          data: data_copy,
        });
        let write_ptr = Box::into_raw(write_req);
        // SAFETY: write_ptr is freshly allocated; stream is a valid libuv handle
        unsafe {
          let buf = deno_core::uv_compat::UvBuf {
            base: (*write_ptr).data.as_ptr() as *mut _,
            len: (*write_ptr).data.len(),
          };
          let ret = deno_core::uv_compat::uv_write(
            &mut (*write_ptr).uv_req,
            stream,
            &buf,
            1,
            Some(h2_write_cb),
          );
          if ret != 0 {
            let _ = Box::from_raw(write_ptr);
          }
        }
      } else if self.outgoing_chunks.is_empty() {
        break;
      }
    }
    self.is_sending = false;

    if !self.outgoing_buffers.is_empty() {
      for buffer in &self.outgoing_buffers {
        let data = buffer.data.as_ref();
        let data_copy = data.to_vec();
        let write_req = Box::new(H2WriteReq {
          uv_req: deno_core::uv_compat::new_write(),
          data: data_copy,
        });
        let write_ptr = Box::into_raw(write_req);
        // SAFETY: write_ptr is freshly allocated; stream is a valid libuv handle
        unsafe {
          let buf = deno_core::uv_compat::UvBuf {
            base: (*write_ptr).data.as_ptr() as *mut _,
            len: (*write_ptr).data.len(),
          };
          let ret = deno_core::uv_compat::uv_write(
            &mut (*write_ptr).uv_req,
            stream,
            &buf,
            1,
            Some(h2_write_cb),
          );
          if ret != 0 {
            let _ = Box::from_raw(write_ptr);
          }
        }
      }
      self.clear_outgoing();
    }

    // mem_send is done and is_sending is clear, so it's safe to hand the
    // writes it framed back to JS.
    self.flush_write_completions();

    self.emit_pending_internal_error();
    self.maybe_notify_graceful_close_complete();

    // Flush any RST_STREAM submissions that were deferred during
    // mem_recv/mem_send (is_sending was true). Matches Node.js's
    // pending_rst_streams_ flush at the end of SendPendingData.
    self.flush_pending_rst_streams();

    // destroy() may be called reentrantly by an nghttp2 callback above. The
    // session and its callback table must remain alive until mem_send has
    // completely unwound, then can be released before returning to JS.
    if self.pending_destroy {
      self.finish_pending_destroy();
    }
  }

  pub fn receive_data(&mut self, data: &[u8]) {
    if data.is_empty() {
      return;
    }
    if self.session.is_null() {
      return;
    }
    // Block re-entrant send_pending_data calls from JS callbacks during
    // mem_recv. A single mem_recv can process multiple frames (e.g.
    // END_STREAM + RST_STREAM). If a callback from frame N triggers
    // mem_send which closes/frees a stream, frame N+1 (RST_STREAM for
    // the same stream) would crash with a double-free. Matches Node.js
    // behavior where sends are deferred until after mem_recv completes.
    self.is_sending = true;
    // SAFETY: self.session is valid; data slice pointer and length are valid
    let ret = unsafe {
      ffi::nghttp2_session_mem_recv(
        self.session,
        data.as_ptr() as _,
        data.len(),
      )
    };
    self.is_sending = false;
    if (ret as i64) < 0 {
      // nghttp2 reported a fatal error processing the inbound frames.
      // Mirrors Node.js HTTP2Session::OnStreamRead: hand the error to JS
      // via onSessionInternalError so it can destroy the session.
      self.emit_session_internal_error(ret as i32);
    }
    self.send_pending_data();
  }

  /// Invoke `onSessionInternalError(integerCode, customErrorCode)` on the
  /// JS handle. Used when nghttp2 returns a fatal error from mem_recv —
  /// the JS side maps `customErrorCode` (e.g.
  /// `ERR_HTTP2_TOO_MANY_INVALID_FRAMES`) to the proper error and destroys
  /// the session, which in turn propagates the error to streams.
  fn emit_session_internal_error(&mut self, errno: i32) {
    let custom_code = self.custom_recv_error_code.take();

    let mut isolate =
      // SAFETY: isolate pointer is valid for the session's lifetime
      unsafe { v8::Isolate::from_raw_isolate_ptr(self.isolate) };
    v8::scope!(let scope, &mut isolate);
    let context = v8::Local::new(scope, self.context.clone());
    let scope = &mut v8::ContextScope::new(scope, context);

    let state = self.op_state.borrow();
    let callbacks = state.borrow::<SessionCallbacks>();
    let callback = v8::Local::new(scope, &callbacks.session_internal_error_cb);
    let Some(this) = self.this.as_ref() else {
      return;
    };
    let recv = v8::Local::new(scope, this);
    drop(state);

    let errno_v = v8::Integer::new(scope, errno);
    let custom_code_v: v8::Local<v8::Value> = match custom_code {
      Some(code) => v8::String::new(scope, code).unwrap().into(),
      None => v8::undefined(scope).into(),
    };

    callback.call(scope, recv.into(), &[errno_v.into(), custom_code_v]);
  }

  pub fn on_dword_aligned_padding(
    &self,
    frame_len: usize,
    max_payload_len: usize,
  ) -> usize {
    let r = (frame_len + 9) % 8;
    if r == 0 {
      return frame_len;
    }
    let pad = frame_len + (8 - r);
    std::cmp::min(max_payload_len, pad)
  }

  pub fn on_max_frame_size_padding(
    &self,
    _frame_len: usize,
    max_payload_len: usize,
  ) -> usize {
    max_payload_len
  }
}

#[derive(ToV8)]
pub struct Http2SessionState {
  pub effective_local_window_size: f64,
  pub effective_recv_data_length: f64,
  pub next_stream_id: f64,
  pub local_window_size: f64,
  pub last_proc_stream_id: f64,
  pub remote_window_size: f64,
  pub outbound_queue_size: f64,
  pub hd_deflate_dynamic_table_size: f64,
  pub hd_inflate_dynamic_table_size: f64,
}

struct NativeResources {
  session: Cell<*mut ffi::nghttp2_session>,
  callbacks: Cell<*mut ffi::nghttp2_session_callbacks>,
}

impl NativeResources {
  fn new() -> Self {
    Self {
      session: Cell::new(std::ptr::null_mut()),
      callbacks: Cell::new(std::ptr::null_mut()),
    }
  }

  fn close(&self) {
    let session = self.session.replace(std::ptr::null_mut());
    if !session.is_null() {
      // SAFETY: session was created by nghttp2_session_*_new3 and is owned by
      // these native resources.
      unsafe {
        ffi::nghttp2_session_del(session);
      }
    }

    let callbacks = self.callbacks.replace(std::ptr::null_mut());
    if !callbacks.is_null() {
      // SAFETY: callbacks was created by nghttp2_session_callbacks_new and is
      // owned by these native resources.
      unsafe {
        ffi::nghttp2_session_callbacks_del(callbacks);
      }
    }
  }
}

impl Drop for NativeResources {
  fn drop(&mut self) {
    self.close();
  }
}

pub struct Http2Session {
  #[allow(dead_code, reason = "stored for future use")]
  type_: SessionType,
  native: Rc<NativeResources>,
  pub(crate) inner: Cell<*mut Session>,
}

// SAFETY: Http2Session pointers are traced by cppgc
unsafe impl deno_core::GarbageCollected for Http2Session {
  fn trace(&self, _: &mut v8::cppgc::Visitor) {}

  fn get_name(&self) -> &'static std::ffi::CStr {
    c"Http2Session"
  }
}

impl Drop for Http2Session {
  fn drop(&mut self) {
    self.teardown();
  }
}

impl Http2Session {
  fn teardown(&self) {
    let inner_ptr = self.inner.replace(std::ptr::null_mut());
    if !inner_ptr.is_null() {
      // SAFETY: inner was allocated by Box::into_raw in Http2Session::create
      // and is owned by this Http2Session.
      let mut inner = unsafe { Box::from_raw(inner_ptr) };
      inner.close_native_resources();
      inner.close_stream_finalizer_safe();
    }
  }

  fn create(
    this: v8::Global<v8::Object>,
    isolate: &v8::Isolate,
    scope: &mut v8::PinScope<'_, '_>,
    op_state: Rc<RefCell<OpState>>,
    session_type: SessionType,
    no_strict_field_ws_validation: bool,
  ) -> Self {
    let mut session: *mut ffi::nghttp2_session = std::ptr::null_mut();
    let options =
      Http2Options::new(session_type, no_strict_field_ws_validation);
    let native = Rc::new(NativeResources::new());

    let context = scope.get_current_context();
    let context = v8::Global::new(scope, context);
    // SAFETY: isolate reference is valid; raw pointer stored for later use
    let isolate_ptr = unsafe { isolate.as_raw_isolate_ptr() };

    let inner = Box::into_raw(Box::new(Session {
      session,
      native: native.clone(),
      streams: HashMap::new(),
      write_completions: Vec::new(),
      op_state,
      context,
      isolate: isolate_ptr,
      this: Some(this),
      outgoing_buffers: Vec::with_capacity(32),
      outgoing_length: 0,
      padding_strategy: options.padding_strategy(),
      graceful_close_initiated: false,
      stream: None,
      is_sending: false,
      pending_destroy: false,
      draining_outgoing: false,
      pending_rst_streams: Vec::new(),
      max_header_pairs: options.max_header_pairs(),
      outgoing_chunks: VecDeque::new(),
      max_invalid_frames: 1000,
      invalid_frame_count: 0,
      frames_received: 0,
      frames_sent: 0,
      custom_recv_error_code: None,
      sent_goaway_code: None,
      local_custom_settings: Vec::new(),
      remote_custom_settings: Vec::new(),
      pending_settings_acks: VecDeque::new(),
      pending_pings: 0,
      protocol_error_emitted: false,
      pending_internal_error: None,
      pending_user_goaway: 0,
      remote_settings_received: false,
      is_client: matches!(session_type, SessionType::Client),
    }));

    let callbacks;
    // SAFETY: inner is valid (just allocated); callbacks and options are valid
    unsafe {
      callbacks = create_callbacks();
      match session_type {
        SessionType::Server => ffi::nghttp2_session_server_new3(
          &mut session,
          callbacks,
          inner as *mut _,
          options.ptr(),
          std::ptr::null_mut(),
        ),
        SessionType::Client => ffi::nghttp2_session_client_new3(
          &mut session,
          callbacks,
          inner as *mut _,
          options.ptr(),
          std::ptr::null_mut(),
        ),
      };
      (*inner).session = session;
      native.session.set(session);
      native.callbacks.set(callbacks);
      // Import the user's remoteCustomSettings list (JS writes the IDs to
      // the shared settings buffer immediately before constructing us).
      (*inner).fetch_allowed_remote_custom_settings();
    }

    Self {
      type_: session_type,
      native,
      inner: Cell::new(inner),
    }
  }

  fn submit_request(
    &self,
    priority: Http2Priority,
    headers: Http2Headers,
    options: i32,
  ) -> i32 {
    if self.native.session.get().is_null() {
      return -1;
    }
    let has_data = (options & STREAM_OPTION_EMPTY_PAYLOAD) == 0;
    let mut data_provider = ffi::nghttp2_data_provider2 {
      source: ffi::nghttp2_data_source {
        ptr: std::ptr::null_mut(),
      },
      read_callback: Some(on_stream_read_callback),
    };

    let dp_ptr = if has_data {
      &mut data_provider as *mut _
    } else {
      std::ptr::null_mut()
    };

    // SAFETY: self.session, priority, headers, and data_provider are valid
    let ret = unsafe {
      ffi::nghttp2_submit_request2(
        self.native.session.get(),
        &priority.spec,
        headers.data(),
        headers.len(),
        dp_ptr,
        std::ptr::null_mut(),
      )
    };

    const NGHTTP2_ERR_NOMEM: i32 = -901;
    assert_ne!(ret, NGHTTP2_ERR_NOMEM);

    if ret > 0 {
      // SAFETY: self.inner was allocated by Box::into_raw and is valid
      let session = unsafe { &mut *self.inner.get() };
      let (obj, stream) =
        Http2Stream::new(session, ret, ffi::NGHTTP2_HCAT_HEADERS);
      stream.start_headers(ffi::NGHTTP2_HCAT_HEADERS);
      if (options & STREAM_OPTION_GET_TRAILERS) != 0 {
        stream.set_has_trailers(true);
      }
      session.streams.insert(ret, (obj, stream));
      session.send_pending_data();
    }

    ret
  }
}

#[op2]
impl Http2Session {
  #[constructor]
  #[cppgc]
  fn new(
    #[this] this: v8::Global<v8::Object>,
    isolate: &v8::Isolate,
    scope: &mut v8::PinScope<'_, '_>,
    op_state: Rc<RefCell<OpState>>,
    #[smi] type_: i32,
    no_strict_field_ws_validation: bool,
  ) -> Http2Session {
    Http2Session::create(
      this,
      isolate,
      scope,
      op_state,
      match type_ {
        0 => SessionType::Server,
        1 => SessionType::Client,
        _ => unreachable!(),
      },
      no_strict_field_ws_validation,
    )
  }

  #[fast]
  fn consume_stream(&self, #[cppgc] tcp: &crate::ops::tcp_wrap::TCPWrap) {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    if session.session.is_null() {
      return;
    }
    // Take ownership of the underlying TCP handle away from the `TCPWrap`. The
    // session now owns the `UvTcp` allocation and is solely responsible for
    // freeing it (via `h2_stream_close_cb`). Without this transfer both the
    // `TCPWrap` and the session would free the same allocation, causing a
    // double free / use-after-free.
    let stream = tcp.detach() as *mut deno_core::uv_compat::UvStream;
    if stream.is_null() {
      // The TCP handle was never allocated or was already detached; nothing to
      // consume.
      return;
    }

    // Stop the existing read on the TCP handle
    // SAFETY: stream is a valid libuv stream handle from TCP object
    unsafe {
      deno_core::uv_compat::uv_read_stop(stream);
    }

    // Store the session pointer in the stream's data field so
    // the read callback can access it
    // SAFETY: stream is a valid libuv handle; self.inner is a valid pointer
    unsafe {
      (*stream).data = self.inner.get() as *mut std::ffi::c_void;
    }

    session.stream = Some(stream);

    // Start reading from the stream
    // SAFETY: stream is a valid libuv stream handle with valid callbacks
    let ret = unsafe {
      deno_core::uv_compat::uv_read_start(
        stream,
        Some(h2_alloc_cb),
        Some(h2_read_cb),
      )
    };
    let _ = ret;
  }

  #[fast]
  #[reentrant]
  fn receive(&self, #[buffer] data: &[u8]) {
    if self.inner.get().is_null() {
      return;
    }
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    if session.session.is_null() {
      return;
    }
    session.receive_data(data);
  }

  #[fast]
  #[reentrant]
  fn send_pending(&self) {
    if self.inner.get().is_null() {
      return;
    }
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    session.send_pending_data();
  }

  /// Drain a single outgoing h2 chunk from nghttp2's send queue.
  ///
  /// Returns one buffer per logical frame slice so the JS write path
  /// can issue one socket.write per chunk, matching the libuv
  /// consume_stream path that queues a separate uv_write per chunk in
  /// `send_pending_data`. For NO_COPY DATA frames, `on_send_data_callback`
  /// pushes header / pad_length / payload / padding into
  /// `session.outgoing_chunks`, which we drain first; non-DATA frames
  /// come straight from `nghttp2_session_mem_send`. An empty buffer
  /// signals "no more pending data"; fatal nghttp2 errors are logged
  /// here via `nghttp2_strerror` so they aren't silently swallowed.
  #[buffer]
  #[reentrant]
  fn get_outgoing_chunk(&self) -> Box<[u8]> {
    if self.inner.get().is_null() {
      return Box::new([]);
    }
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    if session.session.is_null() {
      return Box::new([]);
    }
    // A native callback may synchronously destroy the JS session while
    // nghttp2_session_mem_recv is still on the stack. closeSession then tries
    // to drain the queued GOAWAY through this method. Entering mem_send before
    // mem_recv has unwound corrupts nghttp2's session state, so leave the
    // frame queued for the outer receive_data -> send_pending_data path.
    if session.is_sending || session.draining_outgoing {
      return Box::new([]);
    }

    // nghttp2 may invoke reentrant JS callbacks while serializing a frame.
    // Keep native teardown and write completions deferred until mem_send has
    // unwound and the JS send loop calls send_pending().
    session.draining_outgoing = true;
    session.is_sending = true;
    let chunk = session.drain_outgoing_chunk();
    session.is_sending = false;
    session.draining_outgoing = false;
    chunk
  }

  #[fast]
  #[reentrant]
  fn destroy(
    &self,
    #[this] this: v8::Global<v8::Object>,
    scope: &mut v8::PinScope<'_, '_>,
  ) {
    if self.inner.get().is_null() {
      return;
    }
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };

    // Flush any completions left parked on the session queue. `closeSession`
    // frames a trailing GOAWAY through `get_outgoing_chunk` with no following
    // send pass, so a write completed by that drain would otherwise only be
    // run by a subsequent per-stream `destroy()`; if every stream is already
    // gone the session queue would never drain. Deferred safely if this
    // teardown itself runs inside an outgoing drain.
    session.flush_write_completions();

    let destroy_deferred = session.is_sending;
    if destroy_deferred {
      // We're inside receive_data's mem_recv. Defer the TCP handle
      // and native-resource close so nghttp2 and the Session remain valid
      // until the active mem_recv/mem_send call has fully unwound.
      session.pending_destroy = true;
    } else {
      // Close the stream handle we took ownership of via consume_stream.
      // Use uv_shutdown first to send TCP FIN (graceful close) so the
      // peer can read any remaining buffered data. On Windows, calling
      // uv_close directly sends TCP RST which discards buffered data.
      session.close_stream_gracefully();
    }

    // Call ondone callback if set
    let this_local = v8::Local::new(scope, &this);
    let ondone_key = v8::String::new(scope, "ondone").unwrap();
    if let Some(Ok(ondone_fn)) = this_local
      .get(scope, ondone_key.into())
      .map(v8::Local::<v8::Function>::try_from)
    {
      ondone_fn.call(scope, this_local.into(), &[]);
    }
    if !destroy_deferred {
      // Release nghttp2 and its callback table immediately, but retain the
      // small Rust Session allocation until cppgc finalizes this handle. That
      // keeps late libuv callbacks and retained internal handles harmless.
      session.close_native_resources();
    }
  }

  #[fast]
  fn settings(&self, cb: v8::Local<v8::Function>) -> bool {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    if session.session.is_null() {
      return false;
    }
    // SAFETY: session.isolate is valid for this session's lifetime
    let isolate = unsafe { v8::Isolate::from_raw_isolate_ptr(session.isolate) };
    // Enqueue BEFORE submit so the FIFO entry is in place no matter what.
    // The JS layer guarantees this op is the sole entry point that submits
    // SETTINGS — including the constructor's initial settings, which flows
    // through `Http2Session.prototype.settings` with a bound
    // `settingsCallback`. That keeps `pending_settings_acks` and outbound
    // SETTINGS frames 1:1.
    session
      .pending_settings_acks
      .push_back(v8::Global::new(&isolate, cb));
    let settings = Http2Settings::init(self.inner.get());
    settings.send();
    session.send_pending_data();
    true
  }

  #[reentrant]
  fn goaway(
    &self,
    code: u32,
    last_stream_id: i32,
    #[anybuffer] maybe_data: Option<&[u8]>,
  ) {
    if self.native.session.get().is_null() {
      return;
    }
    let (data_ptr, data_len) = maybe_data
      .map(|d| (d.as_ptr(), d.len()))
      .unwrap_or((std::ptr::null(), 0));

    // When lastStreamID <= 0, use the last processed stream ID
    // so that in-progress streams are not refused.
    let effective_last_stream_id = if last_stream_id <= 0 {
      // SAFETY: self.session is a valid nghttp2 session pointer
      unsafe {
        ffi::nghttp2_session_get_last_proc_stream_id(self.native.session.get())
      }
    } else {
      last_stream_id
    };

    // SAFETY: self.session is valid; data_ptr and data_len are valid
    unsafe {
      ffi::nghttp2_submit_goaway(
        self.native.session.get(),
        ffi::NGHTTP2_FLAG_NONE as _,
        effective_last_stream_id,
        code,
        data_ptr,
        data_len,
      );
    }
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    // Mark this GOAWAY as user-initiated so on_frame_send_callback skips the
    // internal-protocol-error hook when nghttp2 ships it.
    session.pending_user_goaway = session.pending_user_goaway.saturating_add(1);
    session.send_pending_data();
  }

  #[fast]
  fn set_graceful_close(&self) {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    session.graceful_close_initiated = true;
  }

  #[fast]
  fn is_graceful_closing(&self) -> bool {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    session.is_graceful_closing()
  }

  #[fast]
  fn submit_shutdown_notice(&self) {
    if self.native.session.get().is_null() {
      return;
    }
    // SAFETY: self.session is a valid nghttp2 session pointer
    unsafe { ffi::nghttp2_submit_shutdown_notice(self.native.session.get()) };
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    session.start_graceful_close();
    session.send_pending_data();
  }

  #[fast]
  #[smi]
  fn active_stream_count(&self) -> u32 {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    session.active_stream_count() as u32
  }

  #[fast]
  #[smi]
  fn frames_received(&self) -> u32 {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    session.frames_received
  }

  #[fast]
  #[smi]
  fn frames_sent(&self) -> u32 {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    session.frames_sent
  }

  #[fast]
  fn has_pending_data(&self) -> bool {
    if self.native.session.get().is_null() {
      return false;
    }
    // SAFETY: self.session is a valid nghttp2 session pointer
    unsafe {
      let want_write =
        ffi::nghttp2_session_want_write(self.native.session.get());
      let want_read = ffi::nghttp2_session_want_read(self.native.session.get());
      want_write != 0 || want_read != 0
    }
  }

  /// Returns the error code from the most recent GOAWAY frame this session
  /// has sent, or -1 if no GOAWAY has been sent yet.
  #[fast]
  #[smi]
  fn last_sent_goaway_code(&self) -> i32 {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    match session.sent_goaway_code {
      Some(code) => code as i32,
      None => -1,
    }
  }

  #[fast]
  fn local_settings(&self) {
    if self.native.session.get().is_null() {
      return;
    }
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    // SAFETY: self.session is a valid nghttp2 session pointer
    with_settings(|buffer| unsafe {
      buffer[SettingsIndex::HeaderTableSize as usize] =
        ffi::nghttp2_session_get_local_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_HEADER_TABLE_SIZE,
        ) as u32;
      buffer[SettingsIndex::EnablePush as usize] =
        ffi::nghttp2_session_get_local_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_ENABLE_PUSH,
        ) as u32;
      buffer[SettingsIndex::MaxConcurrentStreams as usize] =
        ffi::nghttp2_session_get_local_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS,
        ) as u32;
      buffer[SettingsIndex::InitialWindowSize as usize] =
        ffi::nghttp2_session_get_local_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE,
        ) as u32;
      buffer[SettingsIndex::MaxFrameSize as usize] =
        ffi::nghttp2_session_get_local_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_MAX_FRAME_SIZE,
        ) as u32;
      buffer[SettingsIndex::MaxHeaderListSize as usize] =
        ffi::nghttp2_session_get_local_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE,
        ) as u32;
      buffer[SettingsIndex::EnableConnectProtocol as usize] =
        ffi::nghttp2_session_get_local_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_ENABLE_CONNECT_PROTOCOL,
        ) as u32;
      write_custom_settings_to_buffer(buffer, &session.local_custom_settings);
    });
  }

  #[fast]
  fn remote_settings(&self) {
    if self.native.session.get().is_null() {
      return;
    }
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    // SAFETY: self.session is a valid nghttp2 session pointer
    with_settings(|buffer| unsafe {
      buffer[SettingsIndex::HeaderTableSize as usize] =
        ffi::nghttp2_session_get_remote_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_HEADER_TABLE_SIZE,
        ) as u32;
      buffer[SettingsIndex::EnablePush as usize] =
        ffi::nghttp2_session_get_remote_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_ENABLE_PUSH,
        ) as u32;
      buffer[SettingsIndex::MaxConcurrentStreams as usize] =
        ffi::nghttp2_session_get_remote_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS,
        ) as u32;
      buffer[SettingsIndex::InitialWindowSize as usize] =
        ffi::nghttp2_session_get_remote_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE,
        ) as u32;
      buffer[SettingsIndex::MaxFrameSize as usize] =
        ffi::nghttp2_session_get_remote_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_MAX_FRAME_SIZE,
        ) as u32;
      buffer[SettingsIndex::MaxHeaderListSize as usize] =
        ffi::nghttp2_session_get_remote_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE,
        ) as u32;
      buffer[SettingsIndex::EnableConnectProtocol as usize] =
        ffi::nghttp2_session_get_remote_settings(
          self.native.session.get(),
          ffi::NGHTTP2_SETTINGS_ENABLE_CONNECT_PROTOCOL,
        ) as u32;
      write_custom_settings_to_buffer(buffer, &session.remote_custom_settings);
    });
  }

  fn get_state(&self) -> Http2SessionState {
    if self.native.session.get().is_null() {
      return Http2SessionState {
        effective_local_window_size: 0.0,
        effective_recv_data_length: 0.0,
        next_stream_id: 0.0,
        local_window_size: 0.0,
        last_proc_stream_id: 0.0,
        remote_window_size: 0.0,
        outbound_queue_size: 0.0,
        hd_deflate_dynamic_table_size: 0.0,
        hd_inflate_dynamic_table_size: 0.0,
      };
    }
    // SAFETY: self.session is a valid nghttp2 session pointer
    unsafe {
      Http2SessionState {
        effective_local_window_size:
          ffi::nghttp2_session_get_effective_local_window_size(
            self.native.session.get(),
          ) as f64,
        effective_recv_data_length:
          ffi::nghttp2_session_get_effective_recv_data_length(
            self.native.session.get(),
          ) as f64,
        next_stream_id: ffi::nghttp2_session_get_next_stream_id(
          self.native.session.get(),
        ) as f64,
        local_window_size: ffi::nghttp2_session_get_local_window_size(
          self.native.session.get(),
        ) as f64,
        last_proc_stream_id: ffi::nghttp2_session_get_last_proc_stream_id(
          self.native.session.get(),
        ) as f64,
        remote_window_size: ffi::nghttp2_session_get_remote_window_size(
          self.native.session.get(),
        ) as f64,
        outbound_queue_size: ffi::nghttp2_session_get_outbound_queue_size(
          self.native.session.get(),
        ) as f64,
        hd_deflate_dynamic_table_size:
          ffi::nghttp2_session_get_hd_deflate_dynamic_table_size(
            self.native.session.get(),
          ) as f64,
        hd_inflate_dynamic_table_size:
          ffi::nghttp2_session_get_hd_inflate_dynamic_table_size(
            self.native.session.get(),
          ) as f64,
      }
    }
  }

  #[fast]
  #[rename("setNextStreamID")]
  fn set_next_stream_id(&self, id: i32) -> bool {
    if self.native.session.get().is_null() {
      return false;
    }
    let ret =
      // SAFETY: self.session is a valid nghttp2 session pointer
      unsafe {
        ffi::nghttp2_session_set_next_stream_id(self.native.session.get(), id)
      };
    if ret < 0 {
      log::debug!("failed to set next stream id to {}", id);
      return false;
    }
    log::debug!("set next stream id to {}", id);
    true
  }

  #[fast]
  fn set_max_invalid_frames(&self, value: u32) {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    session.max_invalid_frames = value;
  }

  #[fast]
  fn set_local_window_size(&self, window_size: i32) -> i32 {
    if self.native.session.get().is_null() {
      return -1;
    }
    // SAFETY: self.session is a valid nghttp2 session pointer
    unsafe {
      ffi::nghttp2_session_set_local_window_size(
        self.native.session.get(),
        ffi::NGHTTP2_FLAG_NONE as u8,
        0,
        window_size,
      )
    }
  }

  #[fast]
  fn update_chunks_sent(&self) -> u32 {
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    session.outgoing_buffers.len() as u32
  }

  #[fast]
  fn origin(&self, #[string] origins: &str, count: i32) -> i32 {
    if self.native.session.get().is_null() {
      return -1;
    }
    // Origins are concatenated and separated by NUL bytes (Node-compatible
    // serialization from JS: `arr += `${origin}\0``).
    let mut ov: Vec<ffi::nghttp2_origin_entry> =
      Vec::with_capacity(count as usize);
    let origins_bytes = origins.as_bytes();
    let mut start = 0;

    while ov.len() < count as usize && start < origins_bytes.len() {
      let end = origins_bytes[start..]
        .iter()
        .position(|&b| b == 0)
        .map(|p| start + p)
        .unwrap_or(origins_bytes.len());
      ov.push(ffi::nghttp2_origin_entry {
        origin: origins_bytes[start..end].as_ptr() as *mut u8,
        origin_len: end - start,
      });
      start = end + 1;
    }

    // SAFETY: self.session is valid; ov slice pointer and length are valid
    let ret = unsafe {
      ffi::nghttp2_submit_origin(
        self.native.session.get(),
        ffi::NGHTTP2_FLAG_NONE as u8,
        ov.as_ptr(),
        ov.len(),
      )
    };
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    session.send_pending_data();
    ret
  }

  #[fast]
  fn altsvc(
    &self,
    stream_id: i32,
    #[string] origin: &str,
    #[string] value: &str,
  ) -> i32 {
    if self.native.session.get().is_null() {
      return -1;
    }
    let origin_bytes = origin.as_bytes();
    let value_bytes = value.as_bytes();

    if origin_bytes.len() + value_bytes.len() > 16382 {
      return -1;
    }

    if (origin_bytes.is_empty() && stream_id == 0)
      || (!origin_bytes.is_empty() && stream_id != 0)
    {
      return -1;
    }

    // SAFETY: self.session is valid; origin and value byte slices are valid
    let ret = unsafe {
      ffi::nghttp2_submit_altsvc(
        self.native.session.get(),
        ffi::NGHTTP2_FLAG_NONE as u8,
        stream_id,
        origin_bytes.as_ptr(),
        origin_bytes.len(),
        value_bytes.as_ptr(),
        value_bytes.len(),
      )
    };
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    session.send_pending_data();
    ret
  }

  #[fast]
  fn ping(&self, #[buffer] payload: &[u8]) -> i32 {
    if self.native.session.get().is_null() {
      return -1;
    }
    if payload.len() != 8 {
      return -1;
    }

    // SAFETY: self.session is valid; payload is exactly 8 bytes (checked above)
    let ret = unsafe {
      ffi::nghttp2_submit_ping(
        self.native.session.get(),
        ffi::NGHTTP2_FLAG_NONE as u8,
        payload.as_ptr(),
      )
    };
    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &mut *self.inner.get() };
    if ret == 0 {
      // Track the outstanding ping so an inbound ACK can be matched. If
      // an ACK arrives when this counter is zero, the peer sent an
      // unsolicited PING ACK and we treat it as a protocol error.
      session.pending_pings = session.pending_pings.saturating_add(1);
    }
    session.send_pending_data();
    ret
  }

  fn request<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    headers: v8::Local<v8::String>,
    count: u32,
    options: i32,
    stream_id: i32,
    weight: i32,
    exclusive: bool,
  ) -> v8::Local<'s, v8::Value> {
    let priority = Http2Priority::new(stream_id, weight, exclusive);
    let headers = Http2Headers::from_v8_string(scope, headers, count as usize);

    let ret = self.submit_request(priority, headers, options);
    if ret <= 0 {
      return v8::Integer::new(scope, ret).into();
    }

    // SAFETY: self.inner was allocated by Box::into_raw and is valid
    let session = unsafe { &*self.inner.get() };
    if let Some(stream_obj) = session.find_stream_obj(ret) {
      return v8::Local::new(scope, stream_obj).into();
    }
    v8::Integer::new(scope, -1).into()
  }
}

#[op2]
pub fn op_http2_callbacks(
  state: &mut OpState,
  #[scoped] session_internal_error_cb: v8::Global<v8::Function>,
  #[scoped] priority_frame_cb: v8::Global<v8::Function>,
  #[scoped] settings_frame_cb: v8::Global<v8::Function>,
  #[scoped] ping_frame_cb: v8::Global<v8::Function>,
  #[scoped] headers_frame_cb: v8::Global<v8::Function>,
  #[scoped] frame_error_cb: v8::Global<v8::Function>,
  #[scoped] goaway_data_cb: v8::Global<v8::Function>,
  #[scoped] alt_svc_cb: v8::Global<v8::Function>,
  #[scoped] origin_frame_cb: v8::Global<v8::Function>,
  #[scoped] stream_trailers_cb: v8::Global<v8::Function>,
  #[scoped] stream_close_cb: v8::Global<v8::Function>,
) {
  state.put(SessionCallbacks {
    session_internal_error_cb,
    priority_frame_cb,
    settings_frame_cb,
    ping_frame_cb,
    headers_frame_cb,
    frame_error_cb,
    goaway_data_cb,
    alt_svc_cb,
    origin_frame_cb,
    stream_trailers_cb,
    stream_close_cb,
  });
}

#[op2]
pub fn op_http2_http_state<'a>(
  scope: &mut v8::PinScope<'a, 'a>,
) -> JSHttp2State<'a> {
  JSHttp2State::create(scope)
}

/// Look up the human-readable string for an nghttp2 integer error code via
/// `nghttp2_strerror`. Used by the JS `NghttpError` class so an
/// `ERR_HTTP2_ERROR` raised from the binding carries the same `.message`
/// (e.g. "Protocol error" for `NGHTTP2_ERR_PROTO`) as Node.js produces from
/// its own binding.
#[op2]
#[string]
pub fn op_http2_error_string(code: i32) -> String {
  // Per https://nghttp2.org/documentation/nghttp2_strerror.html the input
  // must be one of `nghttp2_error`; for unknown codes the function may
  // return NULL, so we guard before constructing a `CStr`.
  // SAFETY: nghttp2_strerror returns either NULL or a static C string.
  let p = unsafe { ffi::nghttp2_strerror(code) };
  if p.is_null() {
    return String::new();
  }
  // SAFETY: p is a non-null, NUL-terminated static C string from nghttp2.
  unsafe { std::ffi::CStr::from_ptr(p) }
    .to_string_lossy()
    .into_owned()
}