cljrs-gc 0.1.248

//! Region (arena/bump) allocator for short-lived GC objects.
//!
//! A [`Region`] allocates objects via fast bump-pointer allocation without
//! touching the global GC heap's mutex or linked list.  Objects allocated in a
//! region are dropped in bulk when the region is reset or dropped — there is no
//! per-object deallocation.
//!
//! # Safety
//!
//! The caller **must** ensure that no [`GcPtr`] to a region-allocated object
//! outlives the [`Region`].  The GC will never collect region objects (they are
//! not in the heap list).  Accessing a region-allocated pointer after the region
//! is dropped is undefined behaviour.
//!
//! A [`RegionGuard`] provides RAII-based activation of a thread-local region
//! that can be queried by allocation-site code.

use std::alloc::{self, Layout};
use std::cell::RefCell;
use std::ptr::{self, NonNull};

#[cfg(not(feature = "no-gc"))]
use crate::gc_header::GcBoxHeader;
use crate::{GcBox, GcPtr, Trace};

// ── Constants ───────────────────────────────────────────────────────────────

/// Default chunk size (4 KiB).  Chunks grow if a single allocation is larger.
const DEFAULT_CHUNK_SIZE: usize = 4096;

/// Raised when a bounded region cannot satisfy an allocation.
///
/// `GcPtr::new` is intentionally infallible throughout the runtime, so a
/// bounded invocation reports exhaustion by unwinding with this typed payload.
/// Execution boundaries can catch it with `catch_unwind` and turn it into a
/// normal transaction error without confusing it with a user panic.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RegionLimitExceeded {
    pub limit: usize,
    pub used: usize,
    pub requested: usize,
}

// ── Internal: chunk of raw memory ───────────────────────────────────────────

struct Chunk {
    data: NonNull<u8>,
    layout: Layout,
}

impl Chunk {
    fn new(size: usize, align: usize) -> Self {
        let layout =
            Layout::from_size_align(size, align.max(16)).expect("Region: invalid chunk layout");
        // SAFETY: layout has non-zero size.
        let data =
            unsafe { NonNull::new(alloc::alloc(layout)).expect("Region: chunk allocation failed") };
        Self { data, layout }
    }
}

// ── Internal: drop entry ────────────────────────────────────────────────────

/// Type-erased destructor entry.  Runs `drop_in_place` on the GcBox without
/// freeing memory (the region owns the backing storage).
struct DropEntry {
    ptr: *mut u8,
    drop_fn: unsafe fn(*mut u8),
}

/// Drop the *value* inside a `GcBox<T>` in place, without freeing memory.
///
/// # Safety
/// `ptr` must point to a valid, initialised `GcBox<T>`.
unsafe fn drop_gcbox_in_place<T: Trace + 'static>(ptr: *mut u8) {
    unsafe { ptr::drop_in_place(ptr as *mut GcBox<T>) };
}

// ── Region ──────────────────────────────────────────────────────────────────

/// A bump allocator that produces [`GcPtr`]-compatible objects.
///
/// All memory is freed in bulk on [`reset`](Region::reset) or [`drop`].
pub struct Region {
    /// Allocated chunks, oldest first.
    chunks: Vec<Chunk>,
    /// Current bump pointer (byte offset into the active chunk).
    ptr: usize,
    /// End of the active chunk.
    end: usize,
    /// Drop entries, in allocation order.
    drops: Vec<DropEntry>,
    /// Cumulative bytes consumed by objects (excludes alignment padding).
    bytes_used: usize,
    /// Managed bytes charged to this region, including out-of-line storage
    /// reported by `Trace::gc_size_extra`.
    accounted_bytes: usize,
    /// Optional hard ceiling for managed bytes. A bounded region is backed by
    /// one fixed-size chunk and never grows.
    byte_limit: Option<usize>,
    /// Number of objects allocated.
    object_count: usize,
}

impl Region {
    /// Create a new region with the default chunk size (4 KiB).
    pub fn new() -> Self {
        Self::with_capacity(DEFAULT_CHUNK_SIZE)
    }

    /// Create a new region whose first chunk is at least `cap` bytes.
    pub fn with_capacity(cap: usize) -> Self {
        let cap = cap.max(64); // minimum sensible size
        let chunk = Chunk::new(cap, 16);
        let base = chunk.data.as_ptr() as usize;
        Self {
            chunks: vec![chunk],
            ptr: base,
            end: base + cap,
            drops: Vec::new(),
            bytes_used: 0,
            accounted_bytes: 0,
            byte_limit: None,
            object_count: 0,
        }
    }

    /// Create a region with a fixed managed-memory budget.
    ///
    /// The region allocates one `limit`-byte chunk and will never grow it.
    /// Object boxes, alignment, and out-of-line bytes reported through
    /// [`Trace::gc_size_extra`] are charged to the limit. Allocations made by
    /// ordinary Rust containers that are not owned by a traced value are not
    /// visible here; callers that require a process-wide hard RSS ceiling must
    /// add an outer sandbox (for example a WASM linear-memory limit).
    pub fn with_limit(limit: usize) -> Self {
        assert!(limit >= 64, "Region: byte limit must be at least 64");
        let chunk = Chunk::new(limit, 16);
        let base = chunk.data.as_ptr() as usize;
        Self {
            chunks: vec![chunk],
            ptr: base,
            end: base + limit,
            drops: Vec::new(),
            bytes_used: 0,
            accounted_bytes: 0,
            byte_limit: Some(limit),
            object_count: 0,
        }
    }

    /// Allocate a GC-compatible object in this region.
    ///
    /// The returned [`GcPtr`] is valid until this region is reset or dropped.
    /// The object is **not** registered in the global GC heap.
    pub fn alloc<T: Trace + 'static>(&mut self, value: T) -> GcPtr<T> {
        let layout = Layout::new::<GcBox<T>>();
        let requested = layout.size().saturating_add(value.gc_size_extra());
        self.charge(requested);
        let raw = self.bump_alloc(layout);

        let gc_box = raw as *mut GcBox<T>;
        // SAFETY: `raw` is properly aligned and sized for GcBox<T>.
        #[cfg(not(feature = "no-gc"))]
        unsafe {
            ptr::write(
                gc_box,
                GcBox {
                    header: GcBoxHeader::new::<T>(0),
                    value,
                },
            );
        }
        #[cfg(feature = "no-gc")]
        unsafe {
            ptr::write(gc_box, GcBox { value });
        }

        self.drops.push(DropEntry {
            ptr: raw,
            drop_fn: drop_gcbox_in_place::<T>,
        });
        self.object_count += 1;
        crate::stats::GC_STATS.record_region_alloc(layout.size());

        // SAFETY: `gc_box` is non-null (from bump_alloc) and ≥16-aligned.
        // In GC builds the pointer is tagged region-local so the collector
        // never dereferences it after this region is reset; in no-gc builds
        // there is no collector and the pointer is untagged.
        #[cfg(not(feature = "no-gc"))]
        {
            unsafe { GcPtr::from_region_raw(gc_box) }
        }
        #[cfg(feature = "no-gc")]
        {
            GcPtr(unsafe { NonNull::new_unchecked(gc_box) })
        }
    }

    /// Trace every live object in this region, marking any GC-heap objects
    /// they reference so the collector keeps those alive.
    ///
    /// Called by [`trace_active_regions`] during GC root scanning.  Each
    /// drop-entry pointer is a live `GcBox` whose header `trace_fn` walks the
    /// value's `GcPtr` children; the region is on the active stack, so its
    /// headers are valid.
    #[cfg(not(feature = "no-gc"))]
    pub(crate) fn trace_live(&self, visitor: &mut crate::MarkVisitor) {
        for entry in &self.drops {
            let header = entry.ptr as *const GcBoxHeader;
            // SAFETY: `entry.ptr` is a live GcBox in this (active) region.
            unsafe { ((*header).trace_fn)(header, visitor) };
        }
    }

    /// Drop all objects and reclaim memory, keeping the first chunk for reuse.
    pub fn reset(&mut self) {
        // Run destructors in reverse (LIFO) order.
        for entry in self.drops.drain(..).rev() {
            unsafe { (entry.drop_fn)(entry.ptr) };
        }

        // Free all chunks except the first.
        while self.chunks.len() > 1 {
            let chunk = self.chunks.pop().unwrap();
            unsafe { alloc::dealloc(chunk.data.as_ptr(), chunk.layout) };
        }

        // Reset bump pointer to the start of the first chunk.
        if let Some(first) = self.chunks.first() {
            let base = first.data.as_ptr() as usize;
            self.ptr = base;
            self.end = base + first.layout.size();
        }

        self.bytes_used = 0;
        self.accounted_bytes = 0;
        self.object_count = 0;
    }

    /// Total bytes consumed by allocated objects (excludes padding).
    pub fn bytes_used(&self) -> usize {
        self.bytes_used
    }

    /// Number of objects currently in the region.
    pub fn object_count(&self) -> usize {
        self.object_count
    }

    /// Managed bytes charged against this region's optional limit.
    pub fn accounted_bytes(&self) -> usize {
        self.accounted_bytes
    }

    /// Configured managed-memory limit, if this is a bounded region.
    pub fn byte_limit(&self) -> Option<usize> {
        self.byte_limit
    }

    // ── internal ────────────────────────────────────────────────────────────

    /// Bump-allocate `layout.size()` bytes with `layout.align()` alignment.
    fn bump_alloc(&mut self, layout: Layout) -> *mut u8 {
        let align = layout.align();
        let size = layout.size();

        // Align the current pointer up.
        let aligned = (self.ptr + align - 1) & !(align - 1);
        let new_ptr = aligned + size;

        if new_ptr <= self.end {
            self.ptr = new_ptr;
            self.bytes_used += size;
            aligned as *mut u8
        } else {
            self.grow_and_alloc(layout)
        }
    }

    fn charge(&mut self, requested: usize) {
        if let Some(limit) = self.byte_limit
            && self.accounted_bytes.saturating_add(requested) > limit
        {
            std::panic::panic_any(RegionLimitExceeded {
                limit,
                used: self.accounted_bytes,
                requested,
            });
        }
        self.accounted_bytes = self.accounted_bytes.saturating_add(requested);
    }

    /// Allocate a new chunk large enough, then bump-allocate from it.
    fn grow_and_alloc(&mut self, layout: Layout) -> *mut u8 {
        let size = layout.size();
        if let Some(limit) = self.byte_limit {
            std::panic::panic_any(RegionLimitExceeded {
                limit,
                used: self.accounted_bytes.saturating_sub(size),
                requested: size,
            });
        }
        let chunk_size = DEFAULT_CHUNK_SIZE.max(size * 2);
        let chunk = Chunk::new(chunk_size, layout.align());
        let base = chunk.data.as_ptr() as usize;

        let aligned = (base + layout.align() - 1) & !(layout.align() - 1);
        self.ptr = aligned + size;
        self.end = base + chunk_size;
        self.bytes_used += size;

        self.chunks.push(chunk);
        aligned as *mut u8
    }
}

impl Default for Region {
    fn default() -> Self {
        Self::new()
    }
}

impl Drop for Region {
    fn drop(&mut self) {
        // Run destructors in reverse order.
        for entry in self.drops.drain(..).rev() {
            unsafe { (entry.drop_fn)(entry.ptr) };
        }
        // Free all chunks.
        for chunk in self.chunks.drain(..) {
            unsafe { alloc::dealloc(chunk.data.as_ptr(), chunk.layout) };
        }
    }
}

// ── Thread-local region stack ───────────────────────────────────────────────

thread_local! {
    /// Stack of active regions for the current thread.
    ///
    /// [`RegionGuard`] pushes/pops.  Allocation-site code calls
    /// [`try_alloc_in_region`] to opportunistically use the top region.
    static REGION_STACK: RefCell<Vec<*mut Region>> = const { RefCell::new(Vec::new()) };
}

/// RAII guard that activates a [`Region`] on the thread-local stack.
///
/// When dropped, the region is popped from the stack.  The caller still owns
/// the region and is responsible for its lifetime.
pub struct RegionGuard {
    _not_send: std::marker::PhantomData<*mut ()>, // !Send
}

impl RegionGuard {
    /// Push `region` onto the thread-local region stack.
    ///
    /// # Safety
    /// The `Region` must outlive this guard.
    pub unsafe fn new(region: &mut Region) -> Self {
        let ptr = region as *mut Region;
        REGION_STACK.with(|stack| stack.borrow_mut().push(ptr));
        Self {
            _not_send: std::marker::PhantomData,
        }
    }
}

impl Drop for RegionGuard {
    fn drop(&mut self) {
        REGION_STACK.with(|stack| {
            stack.borrow_mut().pop();
        });
    }
}

// ── Poisoned / retired regions (GC build) ───────────────────────────────────
//
// The heap-promotion fallback (Phase 10.5): when a value that is *opaque* to
// the promotion scan (e.g. an unrealized lazy seq) is published to a
// program-lifetime cell while regions are active, we can no longer prove the
// active regions hold only non-escaping values.  Rather than risk a dangling
// pointer, the publisher "poisons" the active regions: each one is *retired*
// when its scope closes — its memory is kept alive forever (and traced as a
// GC root so heap children stay live) instead of being reset.  A deliberate,
// bounded leak, mirroring the JIT's pinned-epoch precedent.

#[cfg(not(feature = "no-gc"))]
thread_local! {
    /// Regions whose scopes have closed but which may still be referenced —
    /// kept alive for the rest of the process and traced as roots.  Moving
    /// the `Region` struct out of its `Box` is safe: object memory lives in
    /// separately-allocated chunks, and no raw pointer to the struct itself
    /// survives once it has left the active stack.
    static RETIRED_REGIONS: RefCell<Vec<Region>> = const { RefCell::new(Vec::new()) };
    /// Poison watermark: every region at stack depth ≤ this value must be
    /// retired (not reset) when it closes.
    static POISON_WATERMARK: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

/// Poison every region currently active on this thread: each will be retired
/// instead of reset when its scope closes.  No-op when no region is active.
#[cfg(not(feature = "no-gc"))]
pub fn poison_active_regions() {
    let depth = region_stack_depth();
    if depth == 0 {
        return;
    }
    POISON_WATERMARK.with(|w| w.set(w.get().max(depth)));
    crate::stats::GC_STATS.record_region_poison();
}

#[cfg(feature = "no-gc")]
pub fn poison_active_regions() {}

/// Close a region whose scope has ended: pop it from the thread-local stack,
/// then either drop it (running destructors, freeing memory) or — if it was
/// poisoned — retire it.
///
/// All owners of stack-registered regions (the rt_abi bridge, the IR
/// interpreter, scratch guards) must close through here so the poison
/// protocol is honoured.
#[cfg(not(feature = "no-gc"))]
pub fn close_region(region: Box<Region>) {
    let depth = region_stack_depth();
    pop_region_guard();
    let poisoned = POISON_WATERMARK.with(|w| {
        if depth != 0 && depth <= w.get() {
            w.set(depth - 1);
            true
        } else {
            false
        }
    });
    if poisoned {
        tracing::debug!(
            target: "gc",
            "retiring poisoned region ({} objects, {} bytes)",
            region.object_count(),
            region.bytes_used()
        );
        RETIRED_REGIONS.with(|r| r.borrow_mut().push(*region));
    }
    // else: `region` drops here — destructors run, chunks are freed.
}

#[cfg(feature = "no-gc")]
pub fn close_region(region: Box<Region>) {
    pop_region_guard();
    drop(region);
}

/// Trace every retired region's objects (their heap children must stay
/// alive).  Called by `GcHeap::collect` alongside [`trace_active_regions`].
#[cfg(not(feature = "no-gc"))]
pub(crate) fn trace_retired_regions(visitor: &mut crate::MarkVisitor) {
    RETIRED_REGIONS.with(|r| {
        for region in r.borrow().iter() {
            region.trace_live(visitor);
        }
    });
}

/// Trace every object in every region on this thread's active region stack,
/// marking the GC-heap objects they reference.
///
/// Invoked by `GcHeap::collect` during root scanning so that region→heap
/// references are honoured even though the mark phase treats region objects
/// themselves as opaque (see `MarkVisitor::visit`).
///
/// Scanning only the *current* thread's regions is complete by design: each
/// `GcHeap` lives on a single OS thread, regions are thread-local, and region
/// objects are only ever bump-allocated into the region at the top of *this*
/// thread's stack (`try_alloc_in_region`).  A region is freed when it leaves
/// the stack, so every live region able to hold a pointer into this thread's
/// heap is exactly one of the regions iterated here.  (clojurust uses parallel
/// per-thread heaps with independent collections rather than a cross-thread
/// stop-the-world GC, so there is no other thread's heap to consider.)
#[cfg(not(feature = "no-gc"))]
pub(crate) fn trace_active_regions(visitor: &mut crate::MarkVisitor) {
    REGION_STACK.with(|stack| {
        for &region_ptr in stack.borrow().iter() {
            // SAFETY: `RegionGuard`/`push_region_raw` guarantee every pointer
            // on the stack refers to a live `Region`.
            let region = unsafe { &*region_ptr };
            region.trace_live(visitor);
        }
    });
}

/// Allocate in the currently active thread-local region, if one exists.
///
/// Returns `Some(GcPtr<T>)` if a region is active, `None` otherwise (caller
/// should fall back to [`GcPtr::new`]).
///
/// # Safety
/// The returned `GcPtr` is only valid while the region is alive.  The caller
/// must ensure the pointer does not outlive the region.
pub unsafe fn try_alloc_in_region<T: Trace + 'static>(value: T) -> Option<GcPtr<T>> {
    REGION_STACK.with(|stack| {
        let stack = stack.borrow();
        if let Some(&region_ptr) = stack.last() {
            // SAFETY: RegionGuard guarantees the pointer is valid.
            let region = unsafe { &mut *region_ptr };
            Some(region.alloc(value))
        } else {
            None
        }
    })
}

/// Explicitly pop the top region from the thread-local stack.
///
/// This is used by the AOT runtime ABI where [`RegionGuard`]'s RAII `Drop`
/// cannot be used across `extern "C"` boundaries.
///
/// # Safety
/// The caller must ensure that the corresponding [`Region`] is cleaned up
/// after this call.  No `GcPtr` allocated in that region may be used
/// afterwards.
pub fn pop_region_guard() {
    REGION_STACK.with(|stack| {
        stack.borrow_mut().pop();
    });
}

/// Returns `true` if a region is currently active on this thread.
pub fn region_is_active() -> bool {
    REGION_STACK.with(|stack| !stack.borrow().is_empty())
}

/// Returns the current depth of the region stack (number of active regions).
///
/// Used by exception handling to save/restore region state on throw.
pub fn region_stack_depth() -> usize {
    REGION_STACK.with(|stack| stack.borrow().len())
}

/// Pop regions until the stack depth matches `target_depth`.
///
/// Used by exception handling to unwind region scopes on throw.
///
/// # Safety
/// The caller must ensure that the corresponding `Region` objects are
/// also cleaned up (dropped/reset) for each popped entry.
pub fn unwind_region_stack_to(target_depth: usize) {
    REGION_STACK.with(|stack| {
        let mut stack = stack.borrow_mut();
        while stack.len() > target_depth {
            stack.pop();
        }
    });
}

/// Push a raw region pointer onto the thread-local stack.
///
/// This is the non-RAII equivalent of [`RegionGuard::new`], used by the
/// AOT runtime ABI where the region's lifetime is managed explicitly.
///
/// # Safety
/// The `Region` must remain valid until the corresponding
/// [`pop_region_guard`] call.
pub unsafe fn push_region_raw(region: *mut Region) {
    REGION_STACK.with(|stack| stack.borrow_mut().push(region));
}

// ── Tests ───────────────────────────────────────────────────────────────────

#[cfg(all(test, not(feature = "no-gc")))]
mod tests {
    use super::*;
    use crate::MarkVisitor;
    use std::sync::{Arc, Mutex};

    // A traceable type that records when it's dropped.
    #[derive(Debug)]
    struct Tracked {
        id: i32,
        dropped: Arc<Mutex<Vec<i32>>>,
    }

    impl Drop for Tracked {
        fn drop(&mut self) {
            self.dropped.lock().unwrap().push(self.id);
        }
    }

    impl Trace for Tracked {
        fn trace(&self, _: &mut MarkVisitor) {}
    }

    // A traceable type with a child GcPtr.
    #[derive(Debug)]
    struct Parent {
        child: GcPtr<Tracked>,
    }

    impl Trace for Parent {
        fn trace(&self, visitor: &mut MarkVisitor) {
            use crate::GcVisitor as _;
            visitor.visit(&self.child);
        }
    }

    #[test]
    fn basic_alloc_and_read() {
        let mut region = Region::new();
        let p = region.alloc(42i64);
        assert_eq!(*p.get(), 42);
        assert_eq!(region.object_count(), 1);
    }

    #[test]
    fn multiple_allocs() {
        let mut region = Region::new();
        let a = region.alloc(10i64);
        let b = region.alloc(20i64);
        let c = region.alloc(30i64);
        assert_eq!(*a.get(), 10);
        assert_eq!(*b.get(), 20);
        assert_eq!(*c.get(), 30);
        assert_eq!(region.object_count(), 3);
    }

    #[test]
    fn drop_runs_on_region_drop() {
        let dropped = Arc::new(Mutex::new(Vec::new()));
        {
            let mut region = Region::new();
            region.alloc(Tracked {
                id: 1,
                dropped: dropped.clone(),
            });
            region.alloc(Tracked {
                id: 2,
                dropped: dropped.clone(),
            });
            region.alloc(Tracked {
                id: 3,
                dropped: dropped.clone(),
            });
            // Region drops here.
        }
        let order = dropped.lock().unwrap();
        // Dropped in reverse (LIFO) order.
        assert_eq!(*order, vec![3, 2, 1]);
    }

    #[test]
    fn reset_drops_and_reuses() {
        let dropped = Arc::new(Mutex::new(Vec::new()));
        let mut region = Region::new();

        region.alloc(Tracked {
            id: 1,
            dropped: dropped.clone(),
        });
        region.alloc(Tracked {
            id: 2,
            dropped: dropped.clone(),
        });

        region.reset();

        {
            let order = dropped.lock().unwrap();
            assert_eq!(*order, vec![2, 1]);
        }
        assert_eq!(region.object_count(), 0);

        // Allocate again after reset.
        let p = region.alloc(99i64);
        assert_eq!(*p.get(), 99);
        assert_eq!(region.object_count(), 1);
    }

    #[test]
    fn large_alloc_triggers_new_chunk() {
        // Allocate many objects to exceed the default chunk size.
        let mut region = Region::with_capacity(128);
        for i in 0..100 {
            let p = region.alloc(i);
            assert_eq!(*p.get(), i);
        }
        assert_eq!(region.object_count(), 100);
        assert!(region.chunks.len() > 1);
    }

    #[test]
    fn region_objects_not_in_gc_heap() {
        let heap = crate::GcHeap::new();
        let heap_before = heap.count();

        let mut region = Region::new();
        let _p = region.alloc(42i64);
        let _q = region.alloc(99i64);

        // Region allocations should NOT increase the GC heap count.
        assert_eq!(heap.count(), heap_before);
    }

    #[test]
    fn gc_skips_region_objects_from_heap_parent() {
        // A GC-heap parent holds a GcPtr to a region-allocated child.  The
        // marker must treat the region child as opaque (skip it), not trace
        // into it — region objects are not heap-managed.  Marking must
        // succeed and the parent must survive.
        let dropped = Arc::new(Mutex::new(Vec::new()));
        let mut region = Region::new();
        let child = region.alloc(Tracked {
            id: 1,
            dropped: dropped.clone(),
        });
        assert!(child.is_region_alloc(), "region alloc must be tagged");

        // The parent is on the GC heap, pointing to a region-allocated child.
        let heap = crate::GcHeap::new();
        let parent = heap.alloc(Parent {
            child: child.clone(),
        });

        heap.collect(|vis| {
            use crate::GcVisitor as _;
            vis.visit(&parent);
        });

        // Parent should survive.
        assert_eq!(heap.count(), 1);
        // Child is region-managed, not in heap — still alive.
        assert!(dropped.lock().unwrap().is_empty());
    }

    #[test]
    fn gc_does_not_follow_reset_region_pointer() {
        // The exact crash this fix targets: a GC-heap object references a
        // region object, the region is reset (its memory freed/reused), and a
        // later GC marks the heap object.  The dangling region pointer must be
        // skipped (via its provenance tag), not dereferenced.
        let dropped = Arc::new(Mutex::new(Vec::new()));
        let heap = crate::GcHeap::new();

        let mut region = Region::new();
        let region_child = region.alloc(Tracked {
            id: 1,
            dropped: dropped.clone(),
        });
        let parent = heap.alloc(Parent {
            child: region_child,
        });

        // Reset the region: `region_child`'s backing storage is now dead.
        region.reset();

        // Marking the parent must not touch the dangling region pointer.
        heap.collect(|vis| {
            use crate::GcVisitor as _;
            vis.visit(&parent);
        });
        assert_eq!(
            heap.count(),
            1,
            "parent survives; no crash on dangling region ptr"
        );
    }

    #[test]
    fn active_region_keeps_heap_child_alive() {
        // A region object holds a GcPtr to a heap object that is otherwise
        // unreachable.  While the region is active it is traced as a GC root,
        // so the heap child must survive collection.
        let dropped = Arc::new(Mutex::new(Vec::new()));
        let heap = crate::GcHeap::new();
        let heap_child = heap.alloc(Tracked {
            id: 9,
            dropped: dropped.clone(),
        });

        let mut region = Region::new();
        let _parent = region.alloc(Parent { child: heap_child });
        let _guard = unsafe { RegionGuard::new(&mut region) };

        // Two collections (one to exhaust the grace period) with no explicit
        // roots: the heap child survives only because the active region is a
        // root.
        heap.collect(|_vis| {});
        heap.collect(|_vis| {});
        assert!(
            dropped.lock().unwrap().is_empty(),
            "heap object reachable only through an active region must survive GC"
        );
    }

    #[test]
    fn thread_local_region_guard() {
        assert!(!region_is_active());

        let mut region = Region::new();
        {
            let _guard = unsafe { RegionGuard::new(&mut region) };
            assert!(region_is_active());

            // Allocate through the thread-local API.
            let p: GcPtr<i64> = unsafe { try_alloc_in_region(42i64) }.unwrap();
            assert_eq!(*p.get(), 42);
        }

        assert!(!region_is_active());
    }

    #[test]
    fn try_alloc_returns_none_without_region() {
        assert!(!region_is_active());
        let result: Option<GcPtr<i64>> = unsafe { try_alloc_in_region(42i64) };
        assert!(result.is_none());
    }

    #[test]
    fn nested_region_guards() {
        let mut r1 = Region::new();
        let mut r2 = Region::new();

        let _g1 = unsafe { RegionGuard::new(&mut r1) };
        assert!(region_is_active());

        {
            let _g2 = unsafe { RegionGuard::new(&mut r2) };
            assert!(region_is_active());

            // Allocations go to r2 (innermost).
            unsafe { try_alloc_in_region(1i64) };
            assert_eq!(r2.object_count(), 1);
            assert_eq!(r1.object_count(), 0);
        }

        // After g2 drops, allocations go to r1.
        unsafe { try_alloc_in_region(2i64) };
        assert_eq!(r1.object_count(), 1);
    }

    #[test]
    fn bytes_used_tracking() {
        let mut region = Region::new();
        let size = std::mem::size_of::<GcBox<i64>>();
        region.alloc(1i64);
        region.alloc(2i64);
        // At least 2 * size_of::<GcBox<i64>> bytes used.
        assert!(region.bytes_used() >= size * 2);
    }

    #[test]
    fn close_region_resets_when_not_poisoned() {
        let dropped = Arc::new(Mutex::new(Vec::new()));
        let mut region = Box::new(Region::new());
        region.alloc(Tracked {
            id: 1,
            dropped: dropped.clone(),
        });
        unsafe { push_region_raw(region.as_mut() as *mut Region) };
        close_region(region);
        assert_eq!(*dropped.lock().unwrap(), vec![1], "destructor must run");
        assert!(!region_is_active());
    }

    #[test]
    fn poisoned_region_is_retired_not_reset() {
        // A publish barrier hit an opaque value while this region was open:
        // the region must be kept alive (its objects remain valid) rather
        // than reset when its scope closes.
        let dropped = Arc::new(Mutex::new(Vec::new()));
        let mut region = Box::new(Region::new());
        let p = region.alloc(Tracked {
            id: 7,
            dropped: dropped.clone(),
        });
        unsafe { push_region_raw(region.as_mut() as *mut Region) };
        poison_active_regions();
        close_region(region);
        assert!(
            dropped.lock().unwrap().is_empty(),
            "poisoned region must not run destructors"
        );
        // The object is still readable — retirement means the memory lives on.
        assert_eq!(p.get().id, 7);
        assert!(!region_is_active());

        // A region opened *after* the poison closes normally.
        let dropped2 = Arc::new(Mutex::new(Vec::new()));
        let mut r2 = Box::new(Region::new());
        r2.alloc(Tracked {
            id: 9,
            dropped: dropped2.clone(),
        });
        unsafe { push_region_raw(r2.as_mut() as *mut Region) };
        close_region(r2);
        assert_eq!(*dropped2.lock().unwrap(), vec![9]);
    }

    #[test]
    fn poison_with_no_active_region_is_a_no_op() {
        poison_active_regions();
        let dropped = Arc::new(Mutex::new(Vec::new()));
        let mut region = Box::new(Region::new());
        region.alloc(Tracked {
            id: 3,
            dropped: dropped.clone(),
        });
        unsafe { push_region_raw(region.as_mut() as *mut Region) };
        close_region(region);
        assert_eq!(*dropped.lock().unwrap(), vec![3]);
    }

    #[test]
    fn alloc_throughput_region_vs_heap() {
        const N: usize = 10_000;

        // Region allocation (bump pointer, no mutex).
        let region_start = std::time::Instant::now();
        let mut region = Region::with_capacity(N * std::mem::size_of::<GcBox<i64>>() + 1024);
        for i in 0..N as i64 {
            let p = region.alloc(i);
            std::hint::black_box(p.get());
        }
        let region_dur = region_start.elapsed();
        drop(region);

        // GC heap allocation (Box + mutex lock per allocation).
        let heap = crate::GcHeap::new();
        let heap_start = std::time::Instant::now();
        for i in 0..N as i64 {
            let p = heap.alloc(i);
            std::hint::black_box(p.get());
        }
        let heap_dur = heap_start.elapsed();

        // Region should be faster — bump allocation avoids mutex contention
        // and individual Box::new calls.
        eprintln!(
            "Region: {:?} ({:.0} ns/alloc), Heap: {:?} ({:.0} ns/alloc), speedup: {:.1}x",
            region_dur,
            region_dur.as_nanos() as f64 / N as f64,
            heap_dur,
            heap_dur.as_nanos() as f64 / N as f64,
            heap_dur.as_nanos() as f64 / region_dur.as_nanos().max(1) as f64,
        );
        // No timing assertion: wall-clock comparisons are unreliable on shared
        // CI machines.  The eprintln above captures the numbers for inspection.
    }
}