rusty_alloc 2.4.0

//! Segments: 32 MiB-aligned reservations sliced into 64 KiB slices (M3 subset
//! of upstream `segment.c`). The alignment IS the addressing scheme —
//! `ptr → segment` is a mask, `ptr → page` two shifts and a table walk.
//!
//! M3 adds SPAN RECLAMATION: freed page spans return to a per-segment
//! first-fit free list with left/right coalescing, so pages of any size class
//! can reuse the space (M2 leaked slices to their first size class forever).
//!
//! Span invariants (what makes coalescing O(1)):
//! - The carved region `[HEADER_SLICES, next_free_slice)` is partitioned into
//!   spans; every span's FIRST slot has `slice_offset == 0`, and slot
//!   `block_size > 0` ⟺ the span is a live page (0 ⟺ free span).
//! - Every span's LAST slot's `slice_offset` points back to its first slot, so
//!   the left neighbor of any span is found by one follow-back.
//! - Free spans link through `next`/`prev` into `Segment::free_spans`.
//!
//! Commit: a Normal segment on Windows or Unix commits its slices as the
//! bump cursor reaches them ([`LAZY_COMMIT`], `Segment::commit_end`); Huge
//! segments, and every segment on targets whose commit is a no-op, are
//! committed when reserved. Decommit/purge of free spans is the purge policy
//! work of M7.

use core::ptr;
use core::sync::atomic::{AtomicUsize, Ordering};

use crate::os;
use crate::page::Page;
use crate::prim::PrimError;
use crate::segment_map;
use crate::types::{SEGMENT_SIZE, SEGMENT_SLICE_SIZE, SLICES_PER_SEGMENT};

/// What a segment holds.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u32)]
pub enum SegmentKind {
    /// Sliced segment holding small/medium/large pages.
    Normal = 0xA110C,
    /// Dedicated reservation for one huge block (> 16 MiB).
    Huge = 0x40BE,
}

/// Segment header — occupies slice 0 of the reservation.
pub struct Segment {
    /// Kind tag (doubles as a magic for `debug_checks`).
    pub kind: SegmentKind,
    /// Total reserved bytes (== SEGMENT_SIZE for Normal; the whole reservation
    /// for Huge — needed to free it).
    pub total_size: usize,
    /// Bump cursor: slices at/after this index have never been carved.
    pub next_free_slice: u32,
    /// Slices below this index are committed, apart from free spans flagged
    /// `purged`; slices at/after it are reserved only. Always
    /// `>= next_free_slice`, because only the bump cursor advances into new
    /// slices ([`span_alloc`] commits ahead of it) and freed spans never
    /// rejoin the frontier. `SLICES_PER_SEGMENT` for a segment committed when
    /// it was reserved.
    pub commit_end: u32,
    /// Live pages carved from this segment.
    pub used_pages: u32,
    /// Owning thread id (`prim::thread_id`); 0 = abandoned. Gates the
    /// local-vs-remote routing in `alloc::free`.
    pub thread_id: AtomicUsize,
    /// The mapping's never-carved region is known zero (fresh OS memory).
    /// False for recycled arena chunks — bump-fresh spans then are NOT zero.
    pub mem_is_zero: bool,
    /// Any span of this segment was purged (decommitted) at some point. The
    /// segment must be RE-COMMITTED in full before it goes back to an arena:
    /// the next tenant bump-allocates from it and would otherwise touch
    /// decommitted pages (Windows access violation — the M8 defect).
    pub purged_any: bool,
    /// `prim::clock_now` deadline after which this segment's purge candidates
    /// (free, unpurged spans of at least `MEDIUM_PAGE_SLICES` slices) are
    /// purged; 0 = none scheduled. Set by the first candidate freed under a
    /// positive `purge_delay`, cleared when the candidates are purged
    /// ([`purge_scheduled`]).
    pub purge_expire: u64,
    /// This segment carries a PROT_NONE guard page (guarded allocation). The
    /// protection MUST be lifted before the memory is recycled through an
    /// arena, or the next tenant faults on a page it legitimately owns —
    /// the M8 P0. (OS-released memory is unmapped, so this only bites the
    /// arena path, which is exactly why "arenas off" was the one clean arm.)
    pub guarded: bool,
    /// Next segment in the owning heap's list (reused as the global
    /// abandoned-list link while abandoned).
    pub next: *mut Segment,
    /// Head of the free-span list (slots inside `pages`).
    pub free_spans: *mut Page,
    /// For each slice, the byte offset from THIS SEGMENT'S BASE to the `Page`
    /// that owns it.
    ///
    /// It is the same fact `Page::slice_offset` already carries, in the one
    /// form the free path can use in two instructions instead of five.
    /// Resolving a pointer to its page used to mean: multiply the slice index
    /// by `size_of::<Page>()` — 88, not a power of two, so a real `imul` —
    /// add it to the slot array to reach the slot, load the slot's
    /// `slice_offset`, and subtract it. Four dependent steps to reach a page
    /// whose address is a pure function of the slice index. Indexed by a
    /// `u32` array instead, the whole thing is
    /// `mov off(%seg,%idx,4); lea (%seg,%off)`.
    ///
    /// `free` runs this on every call, so those two instructions are the
    /// second-largest item in it. The table costs 2 KiB in a 32 MiB segment
    /// header that has ~20 KiB spare (0.006% of the segment) and is written
    /// only when a span is carved.
    ///
    /// It duplicates state, which is a real hazard, so `debug_validate_segment`
    /// checks the two agree on every slice of every span it walks.
    pub page_off: [u32; SLICES_PER_SEGMENT],
    /// One metadata slot per slice; a span's slot is its first slice.
    pub pages: [Page; SLICES_PER_SEGMENT],
}

/// Byte offset from a segment's base to its slot array — the base value stored
/// in [`Segment::page_off`].
pub const PAGES_OFFSET: usize = core::mem::offset_of!(Segment, pages);

/// The byte offset a slice owned by the span starting at `start` must record.
#[inline(always)]
pub const fn page_off_for(start: usize) -> u32 {
    (PAGES_OFFSET + start * core::mem::size_of::<Page>()) as u32
}

const _: () = assert!(
    core::mem::size_of::<Segment>() <= SEGMENT_SLICE_SIZE,
    "segment header must fit in slice 0"
);

/// Normal segments commit their slices on demand instead of all 32 MiB when
/// reserved.
///
/// Measured on Windows before this (2.3.1, threads that each allocate 100
/// bytes and wait): commit charge 33 MiB with no extra thread, 65 with one,
/// 162 with four, 547 with sixteen, every one with a 4 MiB working set. Each
/// thread's first allocation claims a segment, and each segment charged its
/// whole 32 MiB. Linux charges `Committed_AS` the same way under
/// `overcommit_memory = 2` (`MAP_NORESERVE` is ignored there).
///
/// Only where commit is a real OS call: the fixed-region, wasm and miri prims
/// commit nothing, and a small-profile segment (64 KiB) is too small for the
/// syscalls to buy anything.
pub const LAZY_COMMIT: bool =
    cfg!(all(any(windows, unix), not(miri))) && !crate::FIXED_REGION && SEGMENT_SIZE >= 4 << 20;

/// The most a frontier commit extends `commit_end` past what it must: commits
/// double from the header up to this step, so a full segment costs about a
/// dozen commit calls and the committed-but-uncarved tail stays under
/// max(the carved size, 4 MiB).
const COMMIT_STEP_MAX_SLICES: usize = (4 << 20) / SEGMENT_SLICE_SIZE;

/// Slices reserved for the header.
pub const HEADER_SLICES: usize = 1;

/// Usable slices in a Normal segment.
pub const USABLE_SLICES: usize = SLICES_PER_SEGMENT - HEADER_SLICES;

// `LARGE_OBJ_SIZE_MAX` is the byte capacity of exactly this carved region —
// the routing constant and the geometry must never drift apart, or malloc
// would either send span-sized requests to dedicated huge segments (the
// segment tax, docs/plans/segment-tax.md) or ask `span_from_segments` for a
// span no segment can hold.
const _: () = assert!(crate::types::LARGE_OBJ_SIZE_MAX == USABLE_SLICES * SEGMENT_SLICE_SIZE);

/// Recover the owning segment from any pointer into it. `with_addr` keeps
/// provenance so the mask trick is miri-clean.
///
/// Two arms on one predicate. Hosted segments are `SEGMENT_SIZE`-aligned and
/// the mask recovers the base from the address. On a fixed region
/// (`crate::REGION_STRIDES`) segments stride from the region's BASE, so the
/// mask is applied to the offset from it: that is what lets a firmware's
/// region be `MAX_ALIGN_SIZE`-aligned rather than segment-aligned, and it
/// returns the gap the linker used to leave in front of the aligned static —
/// 24,148 bytes on the ESP32-S3 that measured it
/// (`docs/plans/finished/region-alignment-dissolve.md`).
#[cfg(not(all(target_arch = "wasm32", not(miri))))]
#[inline]
pub fn segment_of(p: *mut u8) -> *mut Segment {
    if crate::REGION_STRIDES {
        // COST, on the ESP32-S3 (the bench firmware's `dealloc`, which is
        // `free` inlined, objdump of the shipped image, 2026-09-09): the mask
        // is `l32r 0xffff0000; and` — two instructions — and this is
        // `l32r &REGION_BASE; l32i; sub; l32r 0xffff; and; sub`, of which the
        // page-index computation reuses the first `sub`, so the free path
        // grows from 36 to 39 instructions. The board prices that at 9-17 ns
        // per alloc/free pair (595 / 833 / 1011 / 1134 against 578 / 820 /
        // 997 / 1125 ns with `--cfg ra_aligned_region`, which is the mask),
        // for 24,144 bytes of stack returned on the firmware that asked.
        // (The esp LLVM backend materialises `0xffff` through the literal
        // pool rather than as an `extui`; that is one instruction of the
        // three and not worth a special case.)
        // `wrapping_sub`: a pointer below the base is foreign, and the answer
        // for a foreign pointer is garbage here exactly as the mask's is; the
        // `debug_checks` guard in `free` is what refuses those.
        let base = crate::prim::fixed::stride_base();
        let off = p.addr().wrapping_sub(base) & (SEGMENT_SIZE - 1);
        p.with_addr(p.addr().wrapping_sub(off)).cast()
    } else {
        // NOTE (2026-08-21): the GEP form `p.wrapping_sub(p.addr() &
        // (SEGMENT_SIZE - 1))` — proposed because `with_addr` is an integer
        // round-trip LLVM is conservative about — measured +1.00 Ir on EVERY
        // malloc and free (small 59.32 -> 60.32, batch_lifo 59.53 -> 60.53).
        // The round-trip is free here and the subtraction is not. Do not
        // retry.
        p.with_addr(p.addr() & !(SEGMENT_SIZE - 1)).cast()
    }
}

/// wasm: segments are SLICE-aligned, not SEGMENT_SIZE-aligned (F2,
/// docs/plans/segment-tax.md), so the mask cannot recover the base — the
/// slice-granular table in `segment_map` does. One load; this platform's
/// free path is plain Rust (the x86 asm fast path is cfg'd out) and
/// single-threaded, so the table lookup is the whole cost of dissolving the
/// alignment constraint that made every ragged reservation strand its tail.
#[cfg(all(target_arch = "wasm32", not(miri)))]
#[inline]
pub fn segment_of(p: *mut u8) -> *mut Segment {
    let base = crate::segment_map::base_of(p.addr());
    debug_assert!(
        base != 0,
        "segment_of: unmapped wasm address {:#x}",
        p.addr()
    );
    p.with_addr(base).cast()
}

/// Reserve backing memory for a segment of `want` bytes.
///
/// Native: an eagerly committed, SEGMENT_SIZE-aligned OS block, exactly as
/// always (on a fixed region, aligned on the region's own segment strides —
/// `crate::REGION_STRIDES`). wasm: the slice pool first — recycled memory at 64 KiB
/// granularity — then a fresh slice-aligned reservation sized to the SLICE
/// round of `want`, not the chunk round. This is where the segment tax
/// dies: a 33 MiB huge block reserves 33.06 MiB instead of 64 MiB, and on
/// free every slice of it becomes serviceable again.
///
/// `commit = false` reserves without committing (native only; wasm memory is
/// always committed).
fn reserve_backing(want: usize, commit: bool) -> Result<(*mut u8, usize, bool), PrimError> {
    #[cfg(all(target_arch = "wasm32", not(miri)))]
    {
        let _ = commit;
        let size = os::page_align_up(want); // wasm page == slice == 64 KiB
        debug_assert!(size.is_multiple_of(SEGMENT_SLICE_SIZE));
        if let Some(addr) = crate::slice_pool::alloc_run(size / SEGMENT_SLICE_SIZE) {
            // Provenance: pool addresses were exposed when their blocks were
            // freed (`expose_provenance` at the free sites).
            return Ok((core::ptr::with_exposed_provenance_mut(addr), size, false));
        }
        let b = os::alloc_aligned(size, SEGMENT_SLICE_SIZE, true, false)?;
        Ok((b.ptr, b.size, b.is_zero))
    }
    #[cfg(not(all(target_arch = "wasm32", not(miri))))]
    {
        let b = os::alloc_aligned(want, SEGMENT_SIZE, commit, false)?;
        Ok((b.ptr, b.size, b.is_zero))
    }
}

/// Resolve a block pointer to its page slot (follows `slice_offset` back to
/// the span start).
///
/// # Safety
/// `p` must point into a live segment owned by this allocator.
#[inline]
pub unsafe fn page_of(seg: *mut Segment, p: *mut u8) -> *mut Page {
    // NOTE (2026-08-21): deriving this as `p.addr() & (SEGMENT_SIZE - 1)` —
    // the same value without the subtraction, since `seg` is `p` masked —
    // measured FLAT. LLVM already sees through the mask here.
    let idx = (p.addr() - seg.addr()) / SEGMENT_SLICE_SIZE;
    debug_assert!(
        idx < SLICES_PER_SEGMENT,
        "page_of: slice index out of range"
    );
    // SAFETY: `p` lies inside this 32 MiB segment (caller contract), so
    // `idx < SLICES_PER_SEGMENT` BY CONSTRUCTION — the offset cannot exceed
    // SEGMENT_SIZE and the divisor is SEGMENT_SLICE_SIZE. Indexing the array
    // with `[idx]` instead makes LLVM emit a bounds check it cannot discharge
    // (the bound is a property of the caller's contract, not of the
    // arithmetic), and this is the hottest function in the allocator: two
    // resolutions per free. `add` on the base pointer keeps the same
    // provenance and the same address, with the check kept under
    // `debug_checks`.
    unsafe {
        // ONE scale-4 load and an add. The table records, per slice, the byte
        // offset from this segment's base to the page that owns it, so the
        // slot address, the `imul` by `size_of::<Page>()` and the
        // `slice_offset` follow-back all disappear — see `Segment::page_off`.
        let tab: *const u32 = (&raw const (*seg).page_off).cast();
        let off = *tab.add(idx) as usize;
        // The two representations must agree. This is the check that makes the
        // duplicated state safe to keep, and it is free in release.
        // (A Huge segment keeps `page_off` alone: its interior slots' offsets
        // are not written, see `huge_alloc`.)
        debug_assert!(
            (*seg).kind == SegmentKind::Huge || {
                let base: *mut Page = (&raw mut (*seg).pages).cast();
                let slot = base.add(idx);
                let back = (*slot).slice_offset as usize * slot_stride();
                back <= idx * core::mem::size_of::<Page>()
                    && slot.cast::<u8>().sub(back).cast::<Page>()
                        == seg.cast::<u8>().add(off).cast::<Page>()
            },
            "page_of: page_off table disagrees with slice_offset"
        );
        seg.cast::<u8>().add(off).cast::<Page>()
    }
}

/// Byte distance between adjacent slice slots. No longer the unit of
/// [`Page::slice_offset`], which counts slices; this scales it where a byte
/// distance is actually wanted.
#[inline]
pub const fn slot_stride() -> usize {
    core::mem::size_of::<Page>()
}

// A span can start at most SLICES_PER_SEGMENT-1 slots before an interior slot,
// so the slice distance must fit the u16 it is stored in. This had far less
// headroom when the field held bytes: 511 * 88 = 44,968, inside u16 but only
// just — a larger `Page` or segment would have overflowed it silently.
const _: () = assert!(
    SLICES_PER_SEGMENT - 1 <= u16::MAX as usize,
    "slice_offset (slices) must fit in u16"
);

/// Payload start of the page whose slot index is `idx`.
///
/// # Safety
/// `idx` must be a span-start slice of a live segment.
#[inline]
pub unsafe fn page_area(seg: *mut Segment, idx: usize) -> *mut u8 {
    // SAFETY: stays within the segment reservation by the idx contract.
    unsafe { seg.cast::<u8>().add(idx * SEGMENT_SLICE_SIZE) }
}

/// Slot index of a page within its segment.
///
/// # Safety
/// `page` must be a slot inside `seg`'s table.
#[inline]
pub unsafe fn page_index(seg: *mut Segment, page: *mut Page) -> usize {
    // From the SLOT pointer this is `(page - base) / size_of::<Page>()`, and
    // `size_of::<Page>()` is 88 — `8 * 11`, not a power of two — so LLVM emits
    // `movabs $0x2e8ba2e8ba2e8ba3; mul; shr`, a 3-instruction magic multiply
    // with a 10-byte immediate. That is `docs/opps.md` #1, and it appeared at
    // all five call sites.
    //
    // `Page::area` is `seg + idx * SEGMENT_SLICE_SIZE` (written by `span_mark`
    // for every carved span, free or allocated, and by `huge_alloc` for its
    // one slot), and SEGMENT_SLICE_SIZE **is** a power of two. Recovering the
    // index through the payload pointer is therefore a shift.
    //
    // SAFETY: both point into the same header per the contract.
    unsafe {
        let idx = ((*page).area.addr() - seg.addr()) / SEGMENT_SLICE_SIZE;
        // (A Huge segment keeps `page_off` alone: its interior slots' offsets
        // are not written, see `huge_alloc`.)
        debug_assert!(
            (*seg).kind == SegmentKind::Huge || {
                let base: *mut Page = (&raw mut (*seg).pages).cast();
                idx == (page.addr() - base.addr()) / core::mem::size_of::<Page>()
            },
            "page_index: Page::area disagrees with the slot-pointer form"
        );
        idx
    }
}

/// Spin until no page of `seg` has a remote free IN FLIGHT.
///
/// **This closes a use-after-free.** `remote_free` claims `XFLAG_FREEING`
/// BEFORE pushing a block onto the owner's delayed list and holds it until it
/// restores `XFLAG_DELAYED` afterwards. The owner, draining that very push,
/// can free the block, retire the page, empty the segment and release it to an
/// arena — where the next tenant `memset`s the whole header, including the
/// `xthread_free` atomic the remote is about to CAS. Miri caught exactly that:
/// an atomic store in `remote_free` racing `huge_alloc`'s header scrub.
///
/// The window is bounded, which is why a barrier suffices rather than an epoch
/// scheme: before the remote sets FREEING it has not pushed yet, so the owner
/// cannot have drained it, so `used > 0` and no retire is possible. Every
/// dangerous instant therefore has FREEING observably set.
///
/// Cost is a 512-slot scan, paid only when a segment is actually released.
///
/// # Safety
/// `seg` must be a live segment header.
unsafe fn wait_no_remote_in_flight(seg: *mut Segment) {
    use crate::page::{XFLAG_FREEING, XMASK};
    // SAFETY: the page table is inside the live header; atomics only.
    unsafe {
        let base: *mut Page = (&raw mut (*seg).pages).cast();
        // Only the CARVED region can hold a page with a remote free in flight.
        // Slots at/after `next_free_slice` were never carved, so their
        // `xthread_free` is still 0 (`& XMASK == XFLAG_NORMAL`, never
        // `XFLAG_FREEING`) — scanning them is guaranteed-idle work. Bounding to
        // `[HEADER_SLICES, next_free_slice)` turns a fixed 512-slot sweep into
        // one proportional to how much of the segment was ever used: a segment
        // that carved 10 slices scans 10, not 512. (A Huge segment sets
        // `next_free_slice = SLICES_PER_SEGMENT`, so it is unaffected — correct,
        // since its one page occupies the whole reservation.)
        //
        // A Huge segment's one page is slot 1, so slot 1 is the only one
        // scanned there. Its other slots are never read by anything but this
        // scan (every slice resolves to slot 1 through `page_off`), and a
        // recycled chunk no longer scrubs them (`scrub_recycled`): sweeping
        // them read stale bytes, and cost 511 atomic loads per huge free.
        let end = if (*seg).kind == SegmentKind::Huge {
            HEADER_SLICES + 1
        } else {
            (*seg).next_free_slice as usize
        };
        for i in HEADER_SLICES..end {
            let pg = base.add(i);
            while (*pg)
                .xthread_free
                .load(core::sync::atomic::Ordering::Acquire)
                & XMASK
                == XFLAG_FREEING
            {
                core::hint::spin_loop();
            }
        }
    }
}

/// Scrub what the next tenant of a RECYCLED chunk could read before it writes
/// it: the owner table (so an uncarved slice resolves exactly as on fresh
/// memory) and the first `slots` page slots. The rest of `pages` is left as
/// it was.
///
/// This used to zero the whole `Segment`, 512 slots of 88 bytes: 45 KB
/// written, and its pages first-touched, on every thread's first segment and
/// every huge allocation once the arena had warmed up. That was 47 k of the
/// 53.5 k instructions of a 64 MiB malloc/free pair and almost all of what a
/// thread's life cost over glibc (`docs/opps.md`, vein census 2026-10-07). A
/// Normal segment's slots are zeroed instead as the bump cursor first carves
/// them; a Huge segment reads only slots 0 and 1.
///
/// # Safety
/// `seg` is the base of a mapping at least `size_of::<Segment>()` long that
/// no one else is using.
///
/// `owner_table` is false for a Huge segment, which overwrites every entry of
/// the owner table but the first a few lines later: zeroing it first was a
/// 2 KB memset that bought nothing.
unsafe fn scrub_recycled(seg: *mut Segment, slots: usize, owner_table: bool) {
    // SAFETY: per contract, both ranges lie inside the header.
    unsafe {
        let tab = (&raw mut (*seg).page_off).cast::<u32>();
        core::ptr::write_bytes(tab, 0, if owner_table { SLICES_PER_SEGMENT } else { 1 });
        core::ptr::write_bytes((&raw mut (*seg).pages).cast::<Page>(), 0, slots);
    }
}

/// Reserve a Normal segment — from an arena when one qualifies (respecting a
/// heap's `arena_id` restriction), else fresh OS memory — and register it in
/// the segment map. Under [`LAZY_COMMIT`] only the header is committed here.
pub fn segment_alloc(arena_id: i32) -> Result<*mut Segment, PrimError> {
    let claim = if LAZY_COMMIT {
        crate::arena::chunk_claim(arena_id)
    } else {
        crate::arena::chunk_alloc(arena_id).map(|(p, zero)| (p, zero, true))
    };
    let (ptr_, size, mem_zero, committed) = match claim {
        Some((p, zero, committed)) => (p, SEGMENT_SIZE, zero, committed),
        None => {
            if arena_id >= 0 {
                return Err(0); // exclusive-arena heap and its arena is full
            }
            let (p, sz, zero) = reserve_backing(SEGMENT_SIZE, !LAZY_COMMIT)?;
            (p, sz, zero, !LAZY_COMMIT)
        }
    };
    if !committed {
        // The header is written next, so it is committed first. On refusal
        // the memory goes back untouched: its chunk keeps a clear
        // `committed` bit, a fresh reservation is released whole.
        // SAFETY: the header slice lies inside the reservation just obtained.
        if unsafe { os::commit(ptr_, HEADER_SLICES * SEGMENT_SLICE_SIZE) }.is_err() {
            crate::stats::commit_failed();
            if !crate::arena::chunk_free_marked(ptr_, false) {
                let block = os::OsBlock {
                    ptr: ptr_,
                    size,
                    is_large: false,
                    is_zero: false,
                };
                // SAFETY: the reservation obtained above, never published.
                let _ = unsafe { os::free(block) };
            }
            return Err(0);
        }
    }
    let seg: *mut Segment = ptr_.cast();
    // SAFETY: 32 MiB mapping (fresh or recycled arena chunk); header written
    // in full. A recycled chunk's stale bytes are all overwritten here and
    // page slots are re-initialized by span_mark on every carve.
    unsafe {
        // Recycled chunks carry stale page slots. Only the owner table and the
        // header slot are scrubbed here; each later slot is zeroed when the
        // bump cursor first carves it (`span_alloc`). Every scalar field is
        // assigned below.
        if !mem_zero {
            scrub_recycled(seg, HEADER_SLICES, true);
        }
        (*seg).kind = SegmentKind::Normal;
        (*seg).total_size = size;
        (*seg).next_free_slice = HEADER_SLICES as u32;
        (*seg).commit_end = if committed {
            SLICES_PER_SEGMENT
        } else {
            HEADER_SLICES
        } as u32;
        (*seg).used_pages = 0;
        (*seg).thread_id = AtomicUsize::new(crate::init::thread_id());
        (*seg).mem_is_zero = mem_zero;
        (*seg).purged_any = false;
        (*seg).purge_expire = 0;
        (*seg).guarded = false;
        (*seg).next = ptr::null_mut();
        (*seg).free_spans = ptr::null_mut();
    }
    segment_map::register(seg);
    Ok(seg)
}

/// Make a dead segment's memory fit for a next tenant: lift guard-page
/// protection and re-commit purged spans (see `purged_any` / `guarded`).
/// Recycled memory is handed out as-is, so skipping this re-tenants an
/// inaccessible page (the M8 P0). Returns whether the memory is usable; when
/// the OS refuses either step (Windows refuses `MEM_COMMIT` once the system's
/// commit is exhausted), the caller must not recycle the segment, and the
/// failure is counted in [`crate::stats::commit_failures`].
///
/// # Safety
/// `seg` must be a live segment with no live blocks or references into it.
unsafe fn restore_for_reuse(seg: *mut Segment) -> bool {
    // SAFETY: per contract; the range lies inside the segment's reservation.
    unsafe {
        if !((*seg).purged_any || (*seg).guarded) {
            return true;
        }
        let base = seg.cast::<u8>().add(HEADER_SLICES * SEGMENT_SLICE_SIZE);
        let bytes = (*seg).total_size - HEADER_SLICES * SEGMENT_SLICE_SIZE;
        // Commit BEFORE lifting the guard. A purged guarded segment (purging
        // on, `huge_free`) has its guard page decommitted with the rest, and
        // Windows' `VirtualProtect` refuses decommitted pages, so the old
        // order failed on every such free and retired the chunk in place
        // (`recommit_tests::a_freed_guarded_object_recycles_its_chunk_with_purging_on`).
        // Once all of it is committed the protect succeeds either way.
        if os::commit(base, bytes).is_err() || os::protect(base, bytes, false).is_err() {
            crate::stats::commit_failed();
            return false;
        }
        (*seg).purged_any = false;
        (*seg).guarded = false;
        true
    }
}

/// Release an empty Normal segment (caller has unlinked it from the heap):
/// back to its arena when it came from one, else to the OS.
///
/// # Safety
/// `seg` must be a live Normal segment with `used_pages == 0` and no live
/// blocks or references into it.
pub unsafe fn segment_free(seg: *mut Segment) -> Result<(), PrimError> {
    // SAFETY: seg is live per the contract; this only reads atomics.
    unsafe { wait_no_remote_in_flight(seg) };
    segment_map::unregister(seg);
    // Under a positive `purge_delay` the chunk goes back resident and marked
    // pending, so the arena purges it once it has stayed free for the delay
    // and a segment claimed again inside the window refaults nothing. Under
    // `0` its spans were purged as they were freed; under `< 0` nothing is.
    let pending = crate::OS_PAGES && crate::options::get(15) > 0;
    if LAZY_COMMIT {
        // Nothing to restore: guard pages live only in Huge segments, and
        // purged or never-committed slices are recorded on the chunk instead
        // of committed back, so the next tenant commits only what it carves.
        // SAFETY: seg is live per the contract; reading its header.
        let full = unsafe {
            debug_assert!(!(*seg).guarded, "guard page in a Normal segment");
            (*seg).commit_end as usize == SLICES_PER_SEGMENT && !(*seg).purged_any
        };
        if crate::arena::chunk_release(seg.cast(), 1, Some(full), pending) {
            return Ok(());
        }
    }
    // SAFETY: seg is live and empty per the contract.
    let usable = LAZY_COMMIT || unsafe { restore_for_reuse(seg) };
    if !usable {
        // The memory could not be made accessible and committed again, so it
        // must not be re-tenanted. Outside an arena, releasing it to the OS
        // (below) needs neither; inside one, it cannot be released alone, so
        // it stays marked used for the life of the process.
        if crate::arena::owns(seg.cast()) {
            return Ok(());
        }
    } else if crate::arena::chunk_release(seg.cast(), 1, None, pending) {
        return Ok(());
    }
    // wasm: the slice pool is the free list (F2) — `expose_provenance` so a
    // later `reserve_backing` may reconstruct a pointer to this range.
    #[cfg(all(target_arch = "wasm32", not(miri)))]
    // SAFETY: seg is live per the contract; reading total_size.
    unsafe {
        if usable
            && crate::slice_pool::free_range(
                seg.cast::<u8>().expose_provenance(),
                (*seg).total_size,
            )
        {
            return Ok(());
        }
    }
    // SAFETY: per contract; reconstruct the OsBlock we allocated with.
    unsafe {
        let block = os::OsBlock {
            ptr: seg.cast(),
            size: (*seg).total_size,
            is_large: false,
            is_zero: false,
        };
        os::free(block)
    }
}

/// Write the span markers for a span at `idx` of `len` slices: first slot
/// offset 0, interior+last offsets pointing back.
///
/// Interior slices below `from` are NOT rewritten: the caller knows they
/// already point back to `idx` in both representations. That holds for every
/// slice of every span in a normal segment (this is their only writer, and
/// `debug_validate_segment` checks it), so a span taken from the FRONT of a
/// free span, or a freed span that did not merge into its left neighbour,
/// starts where its interiors already point. Rewriting them was the largest
/// loop in both `span_alloc` and `span_free`, and it stored values that were
/// already there: 31 of 31 interior slices, twice, per 2 MiB allocate-and-free
/// (opscan `huge`, callgrind per instruction). `from == 1` marks everything.
///
/// # Safety
/// `[idx, idx+len)` must lie in the carved region of `seg` under the heap lock;
/// `from >= 1`, and slices `idx+1 .. idx+min(from, len)` must already point
/// back to `idx`.
unsafe fn span_mark(seg: *mut Segment, idx: usize, len: usize, from: usize) {
    debug_assert!(from >= 1, "span_mark: slot 0 is the head, not an interior");
    // SAFETY: caller contract keeps every index in bounds.
    unsafe {
        (*seg).pages[idx].slice_offset = 0;
        (*seg).pages[idx].slice_count = len as u16;
        // Cache the payload start now, while `idx` is an integer — a shift, no
        // division. The refill path then reads `(*p).area` instead of
        // recovering it with `page_index` (see `Page::area`).
        (*seg).pages[idx].area = seg.cast::<u8>().add(idx * SEGMENT_SLICE_SIZE);
        // Interior slices, walked with RUNNING values rather than recomputed
        // indices. `(*seg).pages[idx + j]` costs a multiply by `size_of::<Page>()`
        // — which is 88, not a power of two — and `j * slot_stride()` costs a
        // second; both become adds when the loop carries the slot pointer and
        // the byte offset it is about to store. Every span carved and every
        // span released walks this loop, and a workload that churns spans
        // walks it constantly.
        // Every slice of this span resolves to the span's own slot, so the
        // table entry is one constant written across the run.
        let owner = page_off_for(idx);
        let tab: *mut u32 = (&raw mut (*seg).page_off).cast();
        *tab.wrapping_add(idx) = owner;
        // NOTE (2026-09-24, REFUTED): splitting the owner-table stores into
        // their own `slice::fill` pass, away from the stride-88 `slice_offset`
        // stores, made both loops cheaper (151 -> 92 + 43 instructions for a
        // 32-slice span) and the function DEARER: each loop brings its own
        // unroll setup and scalar remainder. Opscan `huge` +28.00, perl +340,
        // lua -8,327, sqlite -862 allocator Ir — a sign flip by span length,
        // not a win. Same verdict as the fill `huge_alloc` measured flat.
        let base: *mut Page = (&raw mut (*seg).pages).cast();
        let mut slot = base.wrapping_add(idx + from);
        let mut ent = tab.wrapping_add(idx + from);
        let mut j = from;
        while j < len {
            // SLICES back to the span start, not bytes — see Page::slice_offset.
            (*slot).slice_offset = j as u16;
            *ent = owner;
            slot = slot.wrapping_add(1);
            ent = ent.wrapping_add(1);
            j += 1;
        }
    }
}

// ASSESSED AND DECLINED 2026-08-21 — `span_mark_free` writing only the span's
// FIRST and LAST slot instead of every interior one.
//
// The reasoning holds: interior `slice_offset`s are read by exactly one
// function, `page_of`, which is only ever called on a pointer to a LIVE block,
// and coalescing follows back only from a neighbour's LAST slot. Free spans
// get long once they coalesce — ~16 interior slices per call on `sh8bench` —
// so this was the largest line in both `span_alloc` and `span_free`.
//
// It was not landed. `debug_validate_segment` asserts the STRONGER invariant —
// every span, free or live, has every interior slice pointing back — and that
// walk is the net which caught the M8 parallel-gate race that `adopt_segment`
// documents. Landing this meant weakening a validated layout invariant of an
// allocator to buy ~20M Ir on a benchmark whose own noise floor is ~100M: a
// change we could not measure where it matters, paid for in the one property
// this code most needs to keep. If it is ever wanted, it needs the invariant
// re-specified deliberately — free spans validated on bounds, live spans on
// interiors — not weakened as a side effect of an optimisation.

/// Mark `[idx, idx+len)` as a FREE span and push it on the free list.
///
/// # Safety
/// As [`span_mark`]; the span must not be in the free list already.
unsafe fn span_mark_free(seg: *mut Segment, idx: usize, len: usize, from: usize) {
    // SAFETY: caller contract.
    unsafe {
        span_mark(seg, idx, len, from);
        let slot: *mut Page = &raw mut (*seg).pages[idx];
        (*slot).block_size = 0; // free marker
        (*slot).prev = ptr::null_mut();
        (*slot).next = (*seg).free_spans;
        if !(*seg).free_spans.is_null() {
            (*(*seg).free_spans).prev = slot;
        }
        (*seg).free_spans = slot;
    }
}

/// Unlink a free span from the free list.
///
/// # Safety
/// `span` must be a free-span start slot currently linked in `seg`'s list.
unsafe fn span_list_remove(seg: *mut Segment, span: *mut Page) {
    // SAFETY: caller contract.
    unsafe {
        if (*span).prev.is_null() {
            (*seg).free_spans = (*span).next;
        } else {
            (*(*span).prev).next = (*span).next;
        }
        if !(*span).next.is_null() {
            (*(*span).next).prev = (*span).prev;
        }
        (*span).next = ptr::null_mut();
        (*span).prev = ptr::null_mut();
    }
}

/// `debug_checks`: verify the carved region is exactly TILED by spans, every
/// span start is marked, and the free-span list is consistent. Layout
/// corruption shows up here instead of as a wild pointer later.
///
/// # Safety
/// `seg` must be a live segment the caller is already operating on.
pub unsafe fn debug_validate_segment(seg: *mut Segment, where_: &str) {
    #[cfg(feature = "debug_checks")]
    {
        // SAFETY: caller is operating on this live segment already.
        unsafe {
            let end = (*seg).next_free_slice as usize;
            let mut idx = HEADER_SLICES;
            let mut spans = 0usize;
            while idx < end {
                let slot: *mut Page = &raw mut (*seg).pages[idx];
                assert_eq!(
                    (*slot).slice_offset,
                    0,
                    "{where_}: slice {idx} is not a span start (layout not tiled)"
                );
                let len = (*slot).slice_count as usize;
                assert!(
                    len > 0 && idx + len <= end,
                    "{where_}: slice {idx} has bad slice_count {len} (end {end})"
                );
                // Interior slices must point back to this start.
                for j in 1..len {
                    assert_eq!(
                        (*seg).pages[idx + j].slice_offset as usize,
                        j,
                        "{where_}: slice {} lost its back-pointer",
                        idx + j
                    );
                }
                spans += 1;
                assert!(
                    spans <= SLICES_PER_SEGMENT,
                    "{where_}: span walk did not terminate"
                );
                idx += len;
            }
            assert_eq!(idx, end, "{where_}: spans do not tile the carved region");
            // Free-span list: every node is a free span start inside the region.
            let mut s = (*seg).free_spans;
            let mut n = 0usize;
            while !s.is_null() {
                let i = page_index(seg, s);
                assert!(
                    i >= HEADER_SLICES && i < end,
                    "{where_}: free span {i} out of region"
                );
                assert_eq!((*s).slice_offset, 0, "{where_}: free span {i} not a start");
                assert_eq!((*s).block_size, 0, "{where_}: free span {i} marked live");
                n += 1;
                assert!(
                    n <= SLICES_PER_SEGMENT,
                    "{where_}: free-span list is cyclic"
                );
                s = (*s).next;
            }
        }
    }
    #[cfg(not(feature = "debug_checks"))]
    {
        let _ = (seg, where_);
    }
}

/// Allocate a span of `slices` from the segment: first-fit over reclaimed
/// spans, else bump. Returns (span-start slot, span-is-fresh-zero) or null.
///
/// # Safety
/// `seg` must be a live Normal segment under the heap lock.
pub unsafe fn span_alloc(seg: *mut Segment, slices: usize) -> (*mut Page, bool) {
    // SAFETY: caller contract — seg is live under the owner.
    unsafe { debug_validate_segment(seg, "span_alloc:enter") };
    // SAFETY: heap lock held; list and indices maintained by the invariants
    // in the module docs.
    unsafe {
        // First fit over reclaimed spans (recycled memory — NOT zero).
        let mut s = (*seg).free_spans;
        while !s.is_null() {
            let len = (*s).slice_count as usize;
            if len >= slices {
                let idx = page_index(seg, s);
                // Re-commit BEFORE splitting so both halves are backed; the
                // remainder inherits the (now cleared) purged state.
                //
                // A failed re-commit is a failed allocation, never a span
                // handed out unbacked: Windows has no overcommit, so
                // `MEM_COMMIT` fails when the system's commit is exhausted,
                // and the page layer's first store would then fault on
                // reserved memory (an access violation in `page_extend`,
                // `docs/plans/recommit-failure-ignored.md`). The span stays
                // on the free list, still marked purged; null sends the
                // caller to another segment, and a fresh one fails its own
                // commit cleanly.
                if !span_recommit(seg, idx, len) {
                    return (ptr::null_mut(), false);
                }
                span_list_remove(seg, s);
                if len > slices {
                    // The remainder starts at a new slice: re-mark all of it.
                    span_mark_free(seg, idx + slices, len - slices, 1);
                    // It has been free as long as the span it came from.
                    (*seg).pages[idx + slices].capacity = (*s).capacity;
                }
                // The FRONT of a free span: its interiors already point here.
                span_mark(seg, idx, slices, slices);
                (*seg).used_pages += 1;
                return (s, false);
            }
            s = (*s).next;
        }
        // Bump (never-carved region — zero iff the mapping came fresh from
        // the OS; recycled arena chunks are dirty).
        let idx = (*seg).next_free_slice as usize;
        if idx + slices > SLICES_PER_SEGMENT {
            return (ptr::null_mut(), false);
        }
        // Commit ahead of the cursor before anything is carved. A refusal is
        // a failed span, exactly as a refused re-commit above.
        if LAZY_COMMIT
            && idx + slices > (*seg).commit_end as usize
            && !commit_frontier(seg, idx + slices)
        {
            return (ptr::null_mut(), false);
        }
        // First carve of these slots in this tenancy: on a recycled chunk
        // they still hold the previous tenant's bytes (`scrub_recycled` left
        // them), and a stale `free_is_zero` or `xthread_free` would be read
        // as this page's. On fresh OS memory they are already zero.
        if !(*seg).mem_is_zero {
            ptr::write_bytes((&raw mut (*seg).pages).cast::<Page>().add(idx), 0, slices);
        }
        (*seg).next_free_slice = (idx + slices) as u32;
        (*seg).used_pages += 1;
        let start: *mut Page = &raw mut (*seg).pages[idx];
        // Never-carved slices: mark all of them.
        span_mark(seg, idx, slices, 1);
        (start, (*seg).mem_is_zero)
    }
}

/// Return a page's span to the segment, coalescing with free neighbors.
/// Freed spans NEVER rejoin the bump frontier: `[next_free_slice, …)` is the
/// virgin-zero region and bump allocations report `fresh = true` — giving
/// recycled slices back to it would let dirty memory skip zalloc's memset
/// (caught by the spans G1c gate, 2026-08-05).
/// The page must already be unlinked from heap queues.
///
/// Returns how many spans were purged back to the OS: the freed one under
/// `purge_delay = 0`, or every due candidate when a positive delay had
/// expired ([`schedule_purge`]).
///
/// # Safety
/// `page` must be a live span-start of `seg` with no live blocks, under the
/// heap lock.
pub unsafe fn span_free(seg: *mut Segment, page: *mut Page) -> usize {
    // SAFETY: caller contract — seg is live under the owner.
    unsafe { debug_validate_segment(seg, "span_free:enter") };
    // SAFETY: heap lock held; boundary reads follow the span invariants.
    unsafe {
        let mut idx = page_index(seg, page);
        let mut len = (*page).slice_count as usize;
        // Where the freed span sat before any merge: its interiors already
        // point back to `freed_idx` (see `span_mark`).
        let (freed_idx, freed_len) = (idx, len);
        (*seg).used_pages -= 1;
        // Scrub page state so a stale slot can't masquerade as live.
        (*page).block_size = 0;
        (*page).free = ptr::null_mut();
        (*page).local_free = ptr::null_mut();
        (*page).used = 0;
        (*page).capacity = 0;
        (*page).reserved = 0;
        // `blockmap`: the map pointer is page state like the rest — leave it
        // and the next tenant of this slot inherits a map it never carved.
        #[cfg(feature = "blockmap")]
        {
            (*page).payload = ptr::null_mut();
        }
        (*page).flags.store(0, Ordering::Relaxed);
        (*page).free_is_zero = false;

        // Merge right: the slot at idx+len (if carved) is a span START.
        // Whether any free neighbour merged in below is still purged. Only a
        // span's FIRST slot carries the flag (`span_recommit` reads nothing
        // else), so a merge that is not re-purged as a whole must carry its
        // parts' flags onto the merged start, or a decommitted neighbour ends
        // up behind a clear flag and is reused unbacked: the write fault of
        // `docs/plans/recommit-failure-ignored.md`, mechanism B. A spurious
        // flag costs only a re-commit of committed memory.
        let mut merged_purged = false;
        let right = idx + len;
        if right < (*seg).next_free_slice as usize {
            let rslot: *mut Page = &raw mut (*seg).pages[right];
            debug_assert_eq!((*rslot).slice_offset, 0, "right neighbor not a span start");
            if (*rslot).block_size == 0 {
                span_list_remove(seg, rslot);
                len += (*rslot).slice_count as usize;
                merged_purged |= (*rslot).purged;
            }
        }
        // Merge left: follow the left slot back to its span start.
        if idx > HEADER_SLICES {
            let lslot_idx = idx - 1;
            // slice_offset counts SLICES, so the span start is a subtract.
            // This was `bytes / slot_stride()` — a `movabs; mul; shr` magic
            // multiply by 88, `docs/opps.md` #1, now gone.
            let lstart_idx = lslot_idx - (*seg).pages[lslot_idx].slice_offset as usize;
            let lstart: *mut Page = &raw mut (*seg).pages[lstart_idx];
            if (*lstart).block_size == 0 {
                span_list_remove(seg, lstart);
                len += idx - lstart_idx;
                idx = lstart_idx;
                merged_purged |= (*lstart).purged;
            }
        }
        // Re-mark only what does not already point to `idx`: with no left
        // merge the freed span itself starts here and only an absorbed right
        // neighbour moves; after a left merge the left span is already right
        // and everything from the freed span on moves.
        let from = if idx == freed_idx {
            freed_len
        } else {
            freed_idx - idx
        };
        span_mark_free(seg, idx, len, from);
        // PURGE the coalesced free span (the RSS lever): return its pages to
        // the OS. Only worthwhile for multi-slice spans — a syscall costs
        // more than the pages a single 64 KiB slice returns. The span is
        // marked `purged` so reuse re-commits it; skipping that recommit is
        // an access violation on Windows (MEM_DECOMMIT), which is how this
        // was caught (2026-08-05).
        // Length FIRST: it is in a register, the option is a table load, a
        // bounds test and a sentinel test — and most freed spans are short, so
        // most frees never need the option at all.
        //
        // `purge_delay = 0` purges here. A positive delay only schedules: the
        // span is purged once it has stayed free past the segment's deadline,
        // so a span reused inside the window is never purged and refaulted.
        // Purging at once turned a loop that frees and re-allocates a 1 MiB
        // buffer from 0.7 to 207 us per iteration, every page refaulted
        // (`docs/plans/free-slot-purge.md`).
        let delay = if len >= crate::types::MEDIUM_PAGE_SLICES {
            crate::options::get(15)
        } else {
            -1
        };
        if delay == 0 {
            let area = page_area(seg, idx);
            let bytes = len * SEGMENT_SLICE_SIZE;
            let decommits = crate::options::is_enabled(5); // purge_decommits
            if os::purge(area, bytes, decommits).is_ok() {
                (*seg).pages[idx].purged = true;
                (*seg).purged_any = true;
                return 1;
            }
        }
        // Not re-purged as a whole (purging off by now, the span too small,
        // the purge failed or is only scheduled): the merged span is backed
        // only if every part was. Never cleared here: the start may already
        // be flagged.
        if merged_purged {
            (*seg).pages[idx].purged = true;
        }
        if crate::OS_PAGES && delay > 0 {
            return schedule_purge(seg, idx, delay as u64);
        }
        0
    }
}

/// When a free span became free, in milliseconds of `prim::clock_now`,
/// truncated to `u32` and compared with wrapping arithmetic (a free span's
/// age is far below 49 days). Kept in the span-start slot's `capacity`, which
/// no reader looks at while the span is free (every reader runs on a live
/// page, and a span taken from the free list has it re-initialised), so
/// `Page` keeps its 88 bytes. A free span whose slot was never stamped reads
/// 0: free since long ago, the right answer for a span freed while purging
/// was off.
#[inline]
fn now_ms(now_ns: u64) -> u32 {
    (now_ns / 1_000_000) as u32
}

/// The `purge_delay` option as a millisecond count the wrapping age compare
/// can use.
fn delay_ms() -> u32 {
    crate::options::get(15).clamp(0, i64::from(u32::MAX / 2)) as u32
}

/// The purge candidate starting at slot `idx` was just freed under a positive
/// `purge_delay` of `delay_ms`: stamp when, then start the segment's deadline
/// if none is running, or, if it has passed, purge the candidates that are
/// due. Returns how many spans were purged.
///
/// Only a span that has stayed free for the whole delay is purged, so the one
/// just freed never is: a loop that frees and re-allocates a span within the
/// delay never refaults it, however long the loop runs.
///
/// # Safety
/// `seg` is a live Normal segment under its owner; `idx` a free span start.
#[cold]
unsafe fn schedule_purge(seg: *mut Segment, idx: usize, delay_ms: u64) -> usize {
    let now = crate::prim::clock_now();
    // SAFETY: caller contract.
    unsafe {
        (*seg).pages[idx].capacity = now_ms(now);
        let exp = (*seg).purge_expire;
        if exp == 0 {
            (*seg).purge_expire = now
                .saturating_add(delay_ms.saturating_mul(1_000_000))
                .max(1);
            0
        } else if now >= exp {
            purge_candidates(seg, Some(now))
        } else {
            0
        }
    }
}

/// Purge `seg`'s due candidates if its deadline has passed at `now`, or all
/// of them when `now` is `None` (a forced collect). Returns how many spans
/// were purged.
///
/// # Safety
/// `seg` is a live Normal segment under its owner.
pub unsafe fn purge_scheduled(seg: *mut Segment, now: Option<u64>) -> usize {
    // SAFETY: caller contract.
    unsafe {
        let exp = (*seg).purge_expire;
        if exp == 0 || now.is_some_and(|t| t < exp) {
            return 0;
        }
        purge_candidates(seg, now)
    }
}

/// Purge the candidates of `seg` (free, unpurged spans of at least
/// `MEDIUM_PAGE_SLICES` slices, the spans `purge_delay = 0` purges as they
/// are freed) that have been free for the whole delay at `now`, or all of
/// them when `now` is `None`; then set the deadline to when the next one left
/// falls due, or clear it. Each purged span is marked so reuse re-commits it.
/// Returns how many were purged.
///
/// # Safety
/// `seg` is a live Normal segment under its owner.
unsafe fn purge_candidates(seg: *mut Segment, now: Option<u64>) -> usize {
    // SAFETY: caller contract; free spans hold no live blocks.
    unsafe {
        let delay = delay_ms();
        let decommits = crate::options::is_enabled(5); // purge_decommits
        let mut next = u64::MAX;
        let mut n = 0;
        let mut s = (*seg).free_spans;
        while !s.is_null() {
            let len = (*s).slice_count as usize;
            if !(*s).purged && len >= crate::types::MEDIUM_PAGE_SLICES {
                // Milliseconds this span must still wait; none when forced.
                let wait = now.map_or(0, |t| {
                    delay.saturating_sub(now_ms(t).wrapping_sub((*s).capacity))
                });
                if let (Some(t), true) = (now, wait > 0) {
                    next = next.min(t.saturating_add(u64::from(wait) * 1_000_000));
                } else {
                    let area = page_area(seg, page_index(seg, s));
                    if os::purge(area, len * SEGMENT_SLICE_SIZE, decommits).is_ok() {
                        (*s).purged = true;
                        (*seg).purged_any = true;
                        n += 1;
                    }
                }
            }
            s = (*s).next;
        }
        (*seg).purge_expire = if next == u64::MAX { 0 } else { next.max(1) };
        n
    }
}

/// Purge every FREE span of `seg` — used when a dying thread abandons it.
///
/// An abandoned segment is orphaned until some other thread happens to adopt
/// it, and until then it holds every page it ever touched RESIDENT. Measured:
/// 25 orphans accumulated across 8 waves of thread churn and a full 2048-block
/// allocation burst adopted only 4 of them. At 32 MiB each, that is the RSS
/// tail FFAI saw — max 403 MB against mimalloc's 134, with a 4.4x run-to-run
/// spread because whether anything adopts is pure scheduling.
///
/// Deliberately NOT gated on `purge_delay`. That option governs a LIVE heap's
/// own free spans, where the pages are likely to be reused shortly and a
/// syscall would be wasted. An abandoned segment has no owner to reuse them —
/// holding them costs memory for an unbounded time and buys nothing. Upstream
/// draws the same distinction with `abandoned_page_purge`.
///
/// Each purged span is marked so reuse re-commits it (skipping that is an
/// access violation on Windows), exactly as `span_free` does.
///
/// # Safety
/// `seg` must be a live Normal segment whose free-span list is stable — i.e.
/// the caller still owns it, which is true right up to the abandon publish.
pub unsafe fn purge_free_spans(seg: *mut Segment) {
    // SAFETY: caller still owns the segment; free spans have no live blocks.
    unsafe {
        let decommits = crate::options::is_enabled(5); // purge_decommits
        let mut s = (*seg).free_spans;
        while !s.is_null() {
            let next = (*s).next;
            if !(*s).purged {
                let idx = page_index(seg, s);
                let len = (*s).slice_count as usize;
                if len > 0 {
                    let area = page_area(seg, idx);
                    let bytes = len * SEGMENT_SLICE_SIZE;
                    if os::purge(area, bytes, decommits).is_ok() {
                        (*s).purged = true;
                        (*seg).purged_any = true;
                    }
                }
            }
            s = next;
        }
    }
}

/// Re-commit a span that was purged while free (no-op otherwise). Returns
/// whether the span is backed. On failure the span stays marked purged, so
/// nothing forgets that its memory is not there.
///
/// # Safety
/// `[idx, idx+len)` is a span of `seg` about to be handed to a caller.
unsafe fn span_recommit(seg: *mut Segment, idx: usize, len: usize) -> bool {
    // SAFETY: caller contract; range lies inside the segment reservation.
    unsafe {
        if !(*seg).pages[idx].purged {
            return true;
        }
        let area = page_area(seg, idx);
        if os::commit(area, len * SEGMENT_SLICE_SIZE).is_err() {
            crate::stats::commit_failed();
            return false;
        }
        (*seg).pages[idx].purged = false;
        true
    }
}

/// Commit `seg` up to at least slice `need` (`need > commit_end`): doubling
/// the committed prefix, by at most [`COMMIT_STEP_MAX_SLICES`] beyond `need`,
/// and falling back to exactly `need` if the OS refuses the larger range.
/// Returns false, counted in [`crate::stats::commit_failures`], when even
/// that is refused; `commit_end` is then unchanged.
///
/// # Safety
/// `seg` is a live Normal segment under its owner; `need <= SLICES_PER_SEGMENT`.
#[cold]
unsafe fn commit_frontier(seg: *mut Segment, need: usize) -> bool {
    // SAFETY: caller contract; every range lies inside the reservation.
    unsafe {
        let have = (*seg).commit_end as usize;
        debug_assert!(need > have && need <= SLICES_PER_SEGMENT);
        let want = need
            .max(have + have.min(COMMIT_STEP_MAX_SLICES))
            .min(SLICES_PER_SEGMENT);
        let base = seg.cast::<u8>().add(have * SEGMENT_SLICE_SIZE);
        let end = if os::commit(base, (want - have) * SEGMENT_SLICE_SIZE).is_ok() {
            want
        } else if want > need && os::commit(base, (need - have) * SEGMENT_SLICE_SIZE).is_ok() {
            need
        } else {
            crate::stats::commit_failed();
            return false;
        };
        (*seg).commit_end = end as u32;
        true
    }
}

/// Allocate a dedicated huge segment for one block of `size` bytes, placed so
/// `(block + offset) % align == 0` (align ≤ SEGMENT_SIZE/2). Returns
/// (segment, block ptr).
pub fn huge_alloc(
    size: usize,
    align: usize,
    offset: usize,
    arena_id: i32,
) -> Result<(*mut Segment, *mut u8), PrimError> {
    // A garbage alignment is EINVAL, not a `debug_assert!`: this is a safe
    // `pub fn`, and the mask arithmetic below is only sound for a power of
    // two. Cold path (dedicated segment), so the two compares are free.
    if !align.is_power_of_two() || align > SEGMENT_SIZE / 2 {
        return Err(22);
    }
    let header = SEGMENT_SLICE_SIZE;
    // Worst-case room for placing the block within the reservation. Hosted,
    // the area is slice-aligned, so only larger alignments — or offsets that
    // shift the placement off the natural boundary — need slack. On a fixed
    // region it is only `REGION_ALIGN`-aligned in absolute terms, because
    // segments stride from a `MAX_ALIGN_SIZE`-aligned base
    // (`crate::REGION_STRIDES`), so anything coarser needs the slack there.
    // `is_multiple_of` on a RUNTIME `align` is a modulo, i.e. a `div`.
    // `bins::is_aligned_to` is the mask, and its conservative direction is the
    // safe one here: a `false` makes this reserve the extra slack it would have
    // reserved for a misaligned offset anyway.
    let natural = if crate::REGION_STRIDES {
        crate::prim::fixed::REGION_ALIGN
    } else {
        SEGMENT_SLICE_SIZE
    };
    let extra = if align > natural || !crate::bins::is_aligned_to(offset, align) {
        align
    } else {
        0
    };
    // An unsatisfiable reservation (`malloc(usize::MAX)`) is an error, not a
    // `header + size` debug overflow or a wrapped small reservation.
    let Some(raw) = header.checked_add(size).and_then(|s| s.checked_add(extra)) else {
        return Err(0);
    };
    let want = os::page_align_up(raw);
    if want < raw {
        return Err(0);
    }
    // Huge blocks recycle through arenas too (contiguous chunks) — without
    // this, every huge alloc/free cycle is an OS round-trip (the Tier-A
    // malloc-large gate measured 3–4× slower before this path).
    //
    // `arena_id` is the OWNING HEAP's, exactly as `segment_alloc` above uses
    // it. It used to be a hardcoded `-1`, which meant an exclusive-arena heap
    // — the whole point of which is that its memory comes from ONE region —
    // silently took its huge blocks from the default arena or straight from
    // the OS. Upstream passes `heap->arena_id` here
    // (`mi_segment_huge_page_alloc`, oracle segment.c:1671/1683); we did not.
    // See `tests/heaps.rs::exclusive_arena_confines_huge_allocations`.
    let chunks = want.div_ceil(SEGMENT_SIZE);
    let Some(chunk_bytes) = chunks.checked_mul(SEGMENT_SIZE) else {
        return Err(0);
    };
    let (bptr, total, mem_zero) = match crate::arena::chunk_alloc_n(arena_id, chunks) {
        Some((p, zero)) => (p, chunk_bytes, zero),
        None => {
            if arena_id >= 0 {
                return Err(0); // exclusive-arena heap and its arena is full
            }
            let (p, sz, zero) = reserve_backing(want, true)?;
            (p, sz, zero)
        }
    };
    let seg: *mut Segment = bptr.cast();
    // SAFETY: header scalars fully written below; a recycled chunk's owner
    // table and the two slots a huge segment uses (0, and 1 for its page) are
    // scrubbed. Slots 2.. only get `slice_offset` below and are read by
    // nothing else (every slice resolves to slot 1 through `page_off`, and
    // `wait_no_remote_in_flight` scans slot 1 alone for a Huge segment).
    unsafe {
        if !mem_zero {
            scrub_recycled(seg, HEADER_SLICES + 1, false);
        }
    }
    let b = os::OsBlock {
        ptr: bptr,
        size: total,
        is_large: false,
        is_zero: mem_zero,
    };
    segment_map::register_range(seg.addr(), b.size);
    // SAFETY: fresh zeroed reservation ≥ header + size + extra.
    unsafe {
        (*seg).kind = SegmentKind::Huge;
        (*seg).total_size = b.size;
        (*seg).next_free_slice = SLICES_PER_SEGMENT as u32;
        (*seg).commit_end = SLICES_PER_SEGMENT as u32;
        (*seg).used_pages = 1;
        (*seg).thread_id = AtomicUsize::new(crate::init::thread_id());
        (*seg).mem_is_zero = b.is_zero;
        (*seg).purged_any = false;
        (*seg).purge_expire = 0;
        (*seg).guarded = false;
        (*seg).next = ptr::null_mut();
        (*seg).free_spans = ptr::null_mut();
        let area = b.ptr.add(header);
        // (block + offset) aligned: round (area + offset) up, subtract offset.
        // `area + offset + align` can wrap for a caller-chosen offset; that is
        // an unsatisfiable request, not a block placed by wrapped arithmetic.
        let Some(addr) = crate::bins::aligned_at_from(area.addr(), offset, align) else {
            // SAFETY: the reservation is registered and unused; no block escaped.
            let _ = huge_free(seg);
            return Err(0);
        };
        if addr < area.addr() {
            // SAFETY: as the overflow arm.
            let _ = huge_free(seg);
            return Err(0);
        }
        let block = area.with_addr(addr);
        debug_assert!(block.addr() >= area.addr() && (block.addr() + offset).is_multiple_of(align));
        // The single page lives in slot 1; every reachable interior slice
        // offsets back to it (only slices 1..512 are addressable via the mask
        // trick, and aligned offsets stay < SEGMENT_SIZE/2 by the contract).
        let page: *mut Page = &raw mut (*seg).pages[1];
        // The block's usable size is what was ASKED for, rounded up to a
        // slice — upstream's `psize` (`mi_segment_huge_page_alloc`, oracle
        // segment.c:1599) — and NOT the reservation. The reservation is
        // chunk-rounded (`chunks * SEGMENT_SIZE` through an arena), so a 33 MB
        // request holds a 64 MiB chunk pair and a 64 MB one holds 96 MiB.
        // This field used to report that whole extent, and `usable_size` is
        // the copy length of every `realloc` that MOVES: growing a 33 MB block
        // copied — and first-touched, on both sides — 64 MiB, and a 64 MB
        // block 96 MiB. Measured from a consumer at **1.41x / 1.23x mimalloc,
        // 0/6 pairs** for exactly those two steps, at parity for every step
        // below the segment size (`docs/plans/youslowbro.md` §3). `zalloc` on
        // a recycled chunk zeroes `usable_size` bytes and paid the same tax.
        // The capacity is still `total_size`, and nothing reads past
        // `block_size`; `usable_size` now agrees with `mi_usable_size` here.
        let capacity = b.size - (block.addr() - seg.addr());
        (*page).block_size = crate::prim::align_up(size.max(1), SEGMENT_SLICE_SIZE).min(capacity);
        (*page).used = 1;
        (*page).capacity = 1;
        (*page).reserved = 1;
        (*page).slice_count = (SLICES_PER_SEGMENT - 1) as u16;
        (*page).slice_offset = 0;
        // `header == SEGMENT_SLICE_SIZE`, so this is exactly `page_area(seg, 1)`.
        // Huge segments build their one slot by hand instead of through
        // `span_mark`, which is the only other writer of this field — so this
        // was the single page in the allocator whose `area` stayed null.
        // `page_index` and `unalign` both read it now, and both are reachable
        // for a huge block (`SINGLE_BLOCK` is set, so `unalign` does not take
        // its early return).
        (*page).area = area;
        debug_assert_eq!(
            (*page).area,
            page_area(seg, 1),
            "huge: area != page_area(1)"
        );
        (*page).flags.store(
            crate::page::pflags::HUGE_SEGMENT | crate::page::pflags::SINGLE_BLOCK,
            Ordering::Relaxed,
        );
        (*page).free_is_zero = b.is_zero;
        // Every slice of a huge segment resolves to slot 1, so the owner table
        // is one value repeated across the run, and it is the ONLY record of
        // that: the interior slots' `slice_offset` used to be written too, 511
        // stores 88 bytes apart that touched the whole 45 KB slot array, and
        // their one reader was the `page_of` debug cross-check (which now
        // skips Huge segments). Nothing else reads a Huge segment's slots past
        // slot 1. With the strided stores gone, this is a contiguous 2 KB fill.
        let owner = page_off_for(1);
        let tab: *mut u32 = (&raw mut (*seg).page_off).cast();
        let mut j = 1;
        while j < SLICES_PER_SEGMENT {
            *tab.add(j) = owner;
            j += 1;
        }
        Ok((seg, block))
    }
}

/// Free a huge segment (the whole reservation).
///
/// # Safety
/// `seg` must be a live Huge segment with no live references into it.
pub unsafe fn huge_free(seg: *mut Segment) -> Result<(), PrimError> {
    // SAFETY: per contract; reconstruct the OsBlock we allocated with.
    unsafe {
        // SAME BARRIER AS `segment_free`, and this is the path that actually
        // raced: a huge segment recycles through `chunk_free_n` WITHOUT going
        // through `segment_free`, so guarding only there left the hole open.
        // Every route by which memory can reach an arena needs it.
        wait_no_remote_in_flight(seg);
        segment_map::unregister_range(seg.addr(), (*seg).total_size);
        // Arena-backed huge blocks recycle their contiguous chunks — but the
        // memory must be handed back in a USABLE state: lift any guard-page
        // protection and restore commitment first. Skipping this recycles an
        // inaccessible page into the next tenant (the M8 P0).
        //
        // When that restore fails, the memory must not be recycled at all:
        // released to the OS if it is a reservation of its own, retired in
        // place (left marked used) if it is chunks of an arena.
        // Only chunk-multiple segments can reach an arena and need the
        // restore; a ragged one is released whole (or pooled, on wasm).
        let chunked = (*seg).total_size.is_multiple_of(SEGMENT_SIZE);
        // With purging on, a freed huge block gives its pages back BEFORE its
        // chunks rejoin the arena, exactly as `span_free` does for a span.
        // It never did: arena memory is committed once and handed out as-is,
        // so a dead 150 MB tensor stayed resident for the life of the process
        // whatever `purge_delay` said — the unexplained ~150 MB of
        // `docs/plans/huge-free-retention.md` (Whisper small's embedding is
        // 152 MiB). Marked `purged_any`, so `restore_for_reuse` below
        // re-commits it (checked since 2.2.3) and the next tenant gets
        // demand-zero pages, not a decommitted range. A segment outside an
        // arena is released whole below and needs none of this.
        //
        // That is under `purge_delay = 0`. A positive delay returns the
        // chunks resident and pending instead, for the arena to purge once
        // they have stayed free that long (`arena::purge_due`): a loop that
        // frees and re-allocates a huge buffer then refaults nothing.
        let delay = crate::options::get(15);
        if chunked && delay == 0 && crate::arena::owns(seg.cast()) {
            let base = seg.cast::<u8>().add(HEADER_SLICES * SEGMENT_SLICE_SIZE);
            let bytes = (*seg).total_size - HEADER_SLICES * SEGMENT_SLICE_SIZE;
            if os::purge(base, bytes, crate::options::is_enabled(5)).is_ok() {
                (*seg).purged_any = true;
            }
        }
        let usable = !chunked || restore_for_reuse(seg);
        if chunked {
            if !usable {
                if crate::arena::owns(seg.cast()) {
                    return Ok(());
                }
            } else if crate::arena::chunk_release(
                seg.cast(),
                (*seg).total_size / SEGMENT_SIZE,
                None,
                crate::OS_PAGES && delay > 0,
            ) {
                return Ok(());
            }
        }
        // wasm: ragged (slice-granular) huge reservations recycle through
        // the slice pool — the F2 fix itself. Chunk-multiple ones only
        // reach here when no arena owns them, and the pool takes those too.
        #[cfg(all(target_arch = "wasm32", not(miri)))]
        if usable
            && crate::slice_pool::free_range(
                seg.cast::<u8>().expose_provenance(),
                (*seg).total_size,
            )
        {
            return Ok(());
        }
        let block = os::OsBlock {
            ptr: seg.cast(),
            size: (*seg).total_size,
            is_large: false,
            is_zero: false,
        };
        os::free(block)
    }
}

#[cfg(all(test, feature = "std"))]
mod recommit_tests {
    use super::*;
    use crate::alloc::{collect, free, malloc, stats};
    use crate::os::test_hooks::fail_next_commits;
    use crate::types::MEDIUM_PAGE_SLICES;

    /// A purged span whose re-commit FAILS must not be handed out.
    ///
    /// The MATA desktop app died with an access violation writing a heap
    /// address in `page_extend` (2026-10-04, `secure` on, `purge_delay = 0`,
    /// the machine at 98.5 % of its commit limit). `span_recommit` dropped
    /// the result of `os::commit`, and on Windows that call fails when the
    /// system's commit is exhausted, so a span whose re-commit failed was
    /// carved anyway and the page layer's first store hit reserved memory
    /// (`docs/plans/recommit-failure-ignored.md`, mechanism A).
    ///
    /// Without the fix the second allocation below lands back in the purged
    /// span: on Windows, where the purge is a real `MEM_DECOMMIT`, writing it
    /// is that access violation; everywhere, the "not in the purged span"
    /// assertion fails. The last step checks the span still knows it is
    /// purged, so a later re-commit that succeeds makes it usable again.
    #[test]
    fn a_failed_recommit_is_a_failed_span_not_an_unbacked_one() {
        // Serialised with each other (they set the PROCESS-WIDE `purge_delay`)
        // and with the arena adoption tests (they allocate segments, which
        // can take a chunk from an arena one of those tests just adopted).
        let _g = crate::arena::adopt_tests::lock();
        // A span `span_free` will purge: at least MEDIUM_PAGE_SLICES slices.
        let span = SEGMENT_SLICE_SIZE * MEDIUM_PAGE_SLICES + SEGMENT_SLICE_SIZE;
        crate::options::set(15, 0); // purge_delay: purge at once
        let pin = malloc(span);
        let p1 = malloc(span);
        assert!(!pin.is_null() && !p1.is_null());
        assert_eq!(
            segment_of(pin),
            segment_of(p1),
            "setup: the pin must keep p1's segment alive"
        );
        // SAFETY: live blocks of `span` bytes.
        unsafe {
            core::ptr::write_bytes(pin, 1, span);
            core::ptr::write_bytes(p1, 1, span);
        }
        let purges = stats().purges;
        // SAFETY: p1 is live and freed once.
        unsafe { free(p1) };
        collect(true);
        assert!(stats().purges > purges, "setup: p1's span was not purged");

        let failures = crate::stats::commit_failures();
        fail_next_commits(1);
        let p2 = malloc(span);
        fail_next_commits(0);
        assert!(
            !p2.is_null(),
            "a failed re-commit must move on, not fail outright here"
        );
        assert_eq!(
            crate::stats::commit_failures(),
            failures + 1,
            "the re-commit was attempted, failed, and was counted"
        );
        let in_p1 = (p1.addr()..p1.addr() + span).contains(&p2.addr());
        assert!(!in_p1, "a span whose re-commit failed was handed out");
        // SAFETY: p2 is a live block of `span` bytes; the write is the check.
        unsafe { core::ptr::write_bytes(p2, 2, span) };

        // The span kept its `purged` mark, so with commit available again
        // whichever allocation reaches it re-commits it before use.
        let p3 = malloc(span);
        assert!(!p3.is_null());
        // SAFETY: live block of `span` bytes.
        unsafe { core::ptr::write_bytes(p3, 3, span) };

        // SAFETY: all live, each freed once.
        unsafe {
            free(p3);
            free(p2);
            free(pin);
        }
        crate::options::set(15, -1); // restore the shipped default
    }

    /// Two adjacent purge-eligible spans in one segment, pinned live on the
    /// left so nothing merges past them. Returns `(pin, a, b, span)` with `b`
    /// immediately right of `a`.
    fn adjacent_pair() -> (*mut u8, *mut u8, *mut u8, usize) {
        let span = SEGMENT_SLICE_SIZE * MEDIUM_PAGE_SLICES + SEGMENT_SLICE_SIZE;
        let pin = malloc(span);
        let a = malloc(span);
        let b = malloc(span);
        assert!(!pin.is_null() && !a.is_null() && !b.is_null());
        let seg = segment_of(a);
        assert!(
            segment_of(pin) == seg && segment_of(b) == seg,
            "setup: one segment"
        );
        // SAFETY: live blocks of this segment.
        unsafe {
            let pa = page_of(seg, a);
            let pb = page_of(seg, b);
            assert_eq!(
                page_index(seg, pa) + (*pa).slice_count as usize,
                page_index(seg, pb),
                "setup: b must be a's right neighbour"
            );
            core::ptr::write_bytes(a, 1, span);
            core::ptr::write_bytes(b, 1, span);
        }
        (pin, a, b, span)
    }

    /// Frees `b` with purging on, so its span is purged (decommitted on
    /// Windows), and checks that it was.
    fn purge_right(b: *mut u8) {
        let seg = segment_of(b);
        // SAFETY: `b` is live and freed once; its slot is read after.
        unsafe {
            let ib = page_index(seg, page_of(seg, b));
            free(b);
            collect(true);
            assert!((*seg).pages[ib].purged, "setup: b's span was not purged");
        }
    }

    /// After `a` merged with the purged `b`: the merged span must still say
    /// purged, and reusing the whole of it must be backed. On Windows the
    /// write is the check (a decommitted page faults); the flag assertion
    /// fails first everywhere.
    fn merged_span_is_backed(a: *mut u8, a_idx: usize, span: usize) {
        let seg = segment_of(a);
        // SAFETY: `seg` is pinned live; slot reads only.
        unsafe {
            assert!(
                (*seg).pages[a_idx].purged,
                "a merge kept a clear flag over a purged neighbour: the next tenant gets unbacked memory"
            );
        }
        let whole = 2 * span;
        let c = malloc(whole);
        assert!(!c.is_null());
        assert_eq!(c, a, "setup: the merged span should be reused first-fit");
        // SAFETY: a live block of `whole` bytes.
        unsafe {
            core::ptr::write_bytes(c, 3, whole);
            free(c);
        }
    }

    /// Mechanism B (`docs/plans/recommit-failure-ignored.md`): a freed span
    /// merges with a PURGED right neighbour, and re-purging the merged span
    /// FAILS. The merged start is the freed span's slot, whose flag is clear,
    /// so without carrying the neighbour's flag the decommitted half is
    /// handed out as if it were backed.
    #[test]
    fn a_failed_repurge_of_a_merge_keeps_the_span_purged() {
        // Serialised with each other (they set the PROCESS-WIDE `purge_delay`)
        // and with the arena adoption tests (they allocate segments, which
        // can take a chunk from an arena one of those tests just adopted).
        let _g = crate::arena::adopt_tests::lock();
        crate::options::set(15, 0); // purge_delay: purge at once
        let (pin, a, b, span) = adjacent_pair();
        let seg = segment_of(a);
        // SAFETY: `a` is live.
        let ia = unsafe { page_index(seg, page_of(seg, a)) };
        purge_right(b);
        crate::os::test_hooks::fail_next_purges(1);
        // SAFETY: `a` is live and freed once.
        unsafe { free(a) };
        collect(true);
        crate::os::test_hooks::fail_next_purges(0);
        merged_span_is_backed(a, ia, span);
        // SAFETY: live, freed once.
        unsafe { free(pin) };
        crate::options::set(15, -1);
    }

    /// The same merge with purging switched OFF at runtime after the
    /// neighbour was purged: the merged span is not re-purged at all.
    #[test]
    fn a_merge_after_purging_is_turned_off_keeps_the_span_purged() {
        // Serialised with each other (they set the PROCESS-WIDE `purge_delay`)
        // and with the arena adoption tests (they allocate segments, which
        // can take a chunk from an arena one of those tests just adopted).
        let _g = crate::arena::adopt_tests::lock();
        crate::options::set(15, 0);
        let (pin, a, b, span) = adjacent_pair();
        let seg = segment_of(a);
        // SAFETY: `a` is live.
        let ia = unsafe { page_index(seg, page_of(seg, a)) };
        purge_right(b);
        crate::options::set(15, -1); // purging off from here
        // SAFETY: `a` is live and freed once; `collect` flushes the large
        // cache, so its span reaches `span_free` and merges.
        unsafe { free(a) };
        collect(true);
        merged_span_is_backed(a, ia, span);
        // SAFETY: live, freed once.
        unsafe { free(pin) };
    }

    /// The `secure` guard-page route (`docs/plans/recommit-failure-ignored.md`,
    /// "still open"): a guarded object, freed with purging on, must give its
    /// chunk back to the arena usable, not retire it.
    ///
    /// A guarded object lives in a dedicated huge segment whose page after
    /// the object is no-access. `huge_free` purges the whole segment, guard
    /// page included, then `restore_for_reuse` lifts the guard and
    /// re-commits. With the protect FIRST, Windows refuses it:
    /// `VirtualProtect` fails on pages the purge just decommitted. The
    /// restore then reports failure, the chunk is retired in place for the
    /// life of the process, and the failure is counted as a refused commit,
    /// which is the counter that tells mechanism A apart.
    #[test]
    #[cfg(not(miri))] // guard pages need real OS protection
    fn a_freed_guarded_object_recycles_its_chunk_with_purging_on() {
        let _g = crate::arena::adopt_tests::lock();
        let h = crate::init::create_heap(0, true, -1);
        // SAFETY: heap ours on this thread; each block is ours, freed once.
        unsafe {
            (*(*h).heap.get()).guarded_set_size_bound(1, 64 * 1024);
            (*(*h).heap.get()).guarded_set_sample_rate(1, 0x1234);
        }
        // Both purge forms: decommit (the default) and reset, which keeps
        // the range committed and so has to cope with the no-access page.
        for decommits in [1, 0] {
            crate::options::set(5, decommits); // purge_decommits
            crate::options::set(15, 0);
            // SAFETY: as above.
            unsafe { free_one_guarded(h, decommits) };
            crate::options::set(15, -1);
        }
        crate::options::set(5, 1);
        // SAFETY: heap ours, no live blocks.
        unsafe { crate::init::heap_destroy(h) };
    }

    /// One guarded 4 KiB object from `h`, written and freed; the free must not
    /// count a refused commit.
    ///
    /// # Safety
    /// `h` is a live heap of this thread with guarded sampling on.
    unsafe fn free_one_guarded(h: *mut crate::init::HeapBox, decommits: i64) {
        // SAFETY: caller contract; the block is ours, freed once.
        unsafe {
            let p = crate::alloc::heap_malloc(h, 4096);
            assert!(!p.is_null());
            let seg = segment_of(p);
            assert!((*seg).guarded, "setup: the object was not guarded");
            assert!(
                crate::arena::owns(seg.cast()),
                "setup: the guarded segment is not arena memory"
            );
            p.write_bytes(0xA7, 4096);
            let failures = crate::stats::commit_failures();
            free(p);
            assert_eq!(
                crate::stats::commit_failures(),
                failures,
                "purge_decommits = {decommits}: the guarded chunk could not be restored, so it was retired"
            );
        }
    }
}

#[cfg(all(test, feature = "std"))]
mod lazy_commit_tests {
    use super::*;
    use crate::os::test_hooks::fail_next_commits;

    /// One exclusive reserve-only arena of one chunk for these tests: the
    /// arena table holds 32 per process and is never emptied.
    fn arena() -> i32 {
        static ID: std::sync::OnceLock<i32> = std::sync::OnceLock::new();
        *ID.get_or_init(|| {
            crate::arena::reserve_os_memory_ex(SEGMENT_SIZE, false, false, true)
                .expect("32 MiB reserve-only arena")
        })
    }

    /// Carve `slices` slices from `seg` and write every byte of them: on
    /// Windows a reserved page faults on touch and on Linux a PROT_NONE one
    /// does, so an uncommitted slice crashes the test here.
    ///
    /// # Safety
    /// `seg` is a live Normal segment this test owns.
    unsafe fn carve_and_touch(seg: *mut Segment, slices: usize) -> *mut Page {
        // SAFETY: caller contract.
        unsafe {
            let (pg, _) = span_alloc(seg, slices);
            assert!(!pg.is_null(), "span of {slices} slices refused");
            // Live, as the page layer would mark it: `span_free` reads a
            // zero `block_size` on a neighbour as a free span to merge.
            (*pg).block_size = SEGMENT_SLICE_SIZE;
            ptr::write_bytes((*pg).area, 0xA5, slices * SEGMENT_SLICE_SIZE);
            pg
        }
    }

    /// A segment commits its header when claimed and then only what its bump
    /// cursor reaches, doubling; it hands its chunk back marked uncommitted,
    /// and a tenant that carved the whole segment hands it back committed.
    #[test]
    fn a_segment_commits_what_it_carves_and_its_chunk_remembers() {
        if !LAZY_COMMIT {
            return;
        }
        let _g = crate::arena::adopt_tests::lock();
        let id = arena();
        // SAFETY: segments from our own exclusive arena, used by this thread
        // alone, freed empty.
        unsafe {
            let seg = segment_alloc(id).expect("segment from the arena");
            assert_eq!(
                (*seg).commit_end as usize,
                HEADER_SLICES,
                "claimed with more than its header"
            );
            let pg = carve_and_touch(seg, 3);
            assert_eq!(
                (*seg).commit_end,
                4,
                "the first carve commits max(need, double)"
            );
            let pg2 = carve_and_touch(seg, 1);
            assert_eq!((*seg).commit_end, 8, "committed past need by doubling");
            span_free(seg, pg);
            span_free(seg, pg2);
            segment_free(seg).expect("free");

            let (p, _, committed) = crate::arena::chunk_claim(id).expect("the chunk came back");
            assert_eq!(p, seg.cast(), "the arena has one chunk");
            assert!(
                !committed,
                "a partly committed chunk came back marked committed"
            );
            assert!(crate::arena::chunk_free_marked(p, false));

            // The whole segment, carved: the chunk returns committed in full,
            // and the next tenant commits nothing.
            let seg = segment_alloc(id).expect("second tenant");
            let pg = carve_and_touch(seg, USABLE_SLICES);
            assert_eq!((*seg).commit_end as usize, SLICES_PER_SEGMENT);
            span_free(seg, pg);
            segment_free(seg).expect("free");
            let seg = segment_alloc(id).expect("third tenant");
            assert_eq!(
                (*seg).commit_end as usize,
                SLICES_PER_SEGMENT,
                "a fully committed chunk was committed again"
            );
            segment_free(seg).expect("free");
            let (p, _, committed) = crate::arena::chunk_claim(id).expect("the chunk came back");
            assert!(committed, "an untouched tenant lost the chunk's commit");
            // Leave it uncommitted for the next test, as a fresh arena would be.
            assert!(crate::arena::chunk_free_marked(p, false));
        }
    }

    /// A refused frontier commit is a failed span: nothing is carved, the
    /// committed prefix is unchanged, the failure is counted, and the next
    /// attempt with commit available succeeds; a refused doubled range falls
    /// back to exactly the slices needed.
    #[test]
    fn a_refused_frontier_commit_is_a_failed_span() {
        if !LAZY_COMMIT {
            return;
        }
        let _g = crate::arena::adopt_tests::lock();
        let id = arena();
        // SAFETY: as above.
        unsafe {
            let seg = segment_alloc(id).expect("segment from the arena");
            let (cursor, end) = ((*seg).next_free_slice, (*seg).commit_end);
            let failures = crate::stats::commit_failures();
            fail_next_commits(2);
            let (pg, _) = span_alloc(seg, 2);
            fail_next_commits(0);
            assert!(pg.is_null(), "a span was carved over uncommitted memory");
            assert_eq!((*seg).next_free_slice, cursor, "the cursor moved");
            assert_eq!((*seg).commit_end, end, "the committed prefix moved");
            assert_eq!((*seg).used_pages, 0);
            assert!(
                crate::stats::commit_failures() > failures,
                "the refusal was not counted"
            );

            // With commit available the span is carved. Then one refusal of
            // a carve that would commit past its need (slices 2..4 for a
            // need of 3): the doubled range fails, exactly `need` succeeds.
            let pg = carve_and_touch(seg, 1);
            assert_eq!((*seg).commit_end, 2);
            fail_next_commits(1);
            let pg2 = carve_and_touch(seg, 1);
            fail_next_commits(0);
            assert_eq!(
                (*seg).commit_end,
                3,
                "the fallback commits exactly what is needed"
            );
            span_free(seg, pg);
            span_free(seg, pg2);
            segment_free(seg).expect("free");
        }
    }
}

#[cfg(all(test, feature = "std"))]
mod purge_delay_tests {
    use super::*;
    use crate::types::MEDIUM_PAGE_SLICES;

    /// One exclusive reserve-only arena of one chunk (see `lazy_commit_tests`).
    fn arena() -> i32 {
        static ID: std::sync::OnceLock<i32> = std::sync::OnceLock::new();
        *ID.get_or_init(|| {
            crate::arena::reserve_os_memory_ex(SEGMENT_SIZE, false, false, true)
                .expect("reserve-only arena")
        })
    }

    /// A purge candidate (`MEDIUM_PAGE_SLICES + 1` slices) freed between two
    /// live spans, so it cannot merge, under `purge_delay = delay_ms`.
    /// Returns (segment, the candidate's slot index, the two pins).
    ///
    /// # Safety
    /// The caller holds the options lock and frees everything it is given.
    unsafe fn freed_candidate(delay_ms: i64) -> (*mut Segment, usize, [*mut Page; 2]) {
        let n = MEDIUM_PAGE_SLICES + 1;
        // SAFETY: a segment of our own exclusive arena, this thread only.
        unsafe {
            let seg = segment_alloc(arena()).expect("segment");
            let (a, _) = span_alloc(seg, 1);
            let (v, _) = span_alloc(seg, n);
            let (b, _) = span_alloc(seg, 1);
            assert!(!a.is_null() && !v.is_null() && !b.is_null());
            // Live, as the page layer would mark them: `span_free` reads a
            // zero `block_size` on a neighbour as a free span to merge.
            for pg in [a, v, b] {
                (*pg).block_size = SEGMENT_SLICE_SIZE;
            }
            ptr::write_bytes((*v).area, 0x5A, n * SEGMENT_SLICE_SIZE);
            let idx = page_index(seg, v);
            crate::options::set(15, delay_ms);
            assert_eq!(span_free(seg, v), 0, "a delayed purge ran at once");
            (seg, idx, [a, b])
        }
    }

    /// # Safety
    /// As [`freed_candidate`]'s results.
    unsafe fn release(seg: *mut Segment, pins: [*mut Page; 2]) {
        crate::options::set(15, -1);
        // SAFETY: the pins are live spans of `seg`, freed once; then empty.
        unsafe {
            for p in pins {
                span_free(seg, p);
            }
            segment_free(seg).expect("free");
        }
    }

    /// Freed under a delay, the span is scheduled, not purged, and a reuse
    /// inside the window takes it back unpurged: no refault.
    #[test]
    fn a_span_reused_inside_the_delay_is_not_purged() {
        let _g = crate::arena::adopt_tests::lock();
        // SAFETY: as documented on the helpers.
        unsafe {
            let (seg, idx, pins) = freed_candidate(60_000);
            assert_ne!((*seg).purge_expire, 0, "no deadline was set");
            assert!(!(*seg).pages[idx].purged);
            let (v, _) = span_alloc(seg, MEDIUM_PAGE_SLICES + 1);
            assert_eq!(page_index(seg, v), idx, "the freed span was not reused");
            assert!(!(*v).purged, "a span reused inside the window was purged");
            (*v).block_size = SEGMENT_SLICE_SIZE;
            assert_eq!(
                purge_scheduled(seg, None),
                0,
                "a reused span is not a candidate"
            );
            span_free(seg, v);
            release(seg, pins);
        }
    }

    /// Once the deadline has passed, a purge pass purges the candidate and
    /// clears the deadline; before it, an unforced pass does nothing and a
    /// forced one purges at once.
    #[test]
    #[cfg_attr(miri, ignore)] // real time: the mock prim's clock counts calls
    fn an_expired_deadline_purges_and_a_forced_pass_does_not_wait() {
        let _g = crate::arena::adopt_tests::lock();
        // SAFETY: as documented on the helpers.
        unsafe {
            let (seg, idx, pins) = freed_candidate(1);
            std::thread::sleep(std::time::Duration::from_millis(5));
            assert_eq!(purge_scheduled(seg, Some(crate::prim::clock_now())), 1);
            assert!(
                (*seg).pages[idx].purged,
                "the expired candidate was not purged"
            );
            assert_eq!((*seg).purge_expire, 0, "the deadline was not cleared");
            release(seg, pins);

            let (seg, idx, pins) = freed_candidate(60_000);
            let now = crate::prim::clock_now();
            assert_eq!(
                purge_scheduled(seg, Some(now)),
                0,
                "purged before its deadline"
            );
            assert!(!(*seg).pages[idx].purged);
            assert_eq!(purge_scheduled(seg, None), 1, "a forced pass waited");
            assert!((*seg).pages[idx].purged);
            release(seg, pins);
        }
    }

    /// The deadline passing does not purge a span freed a moment ago: only a
    /// span free for the whole delay is due. A is freed, the delay passes, B
    /// is freed in the same segment: A is purged, B is not, and the deadline
    /// is re-armed for B.
    #[test]
    #[cfg_attr(miri, ignore)] // real time: the mock prim's clock counts calls
    fn an_expired_deadline_spares_a_span_freed_just_now() {
        let _g = crate::arena::adopt_tests::lock();
        let n = MEDIUM_PAGE_SLICES + 1;
        // SAFETY: a segment of our own exclusive arena, this thread only;
        // every span is freed once and the segment freed empty.
        unsafe {
            let seg = segment_alloc(arena()).expect("segment");
            let mut spans = [ptr::null_mut(); 5]; // pin, A, pin, B, pin
            for (k, slot) in spans.iter_mut().enumerate() {
                let (pg, _) = span_alloc(seg, if k % 2 == 1 { n } else { 1 });
                assert!(!pg.is_null());
                (*pg).block_size = SEGMENT_SLICE_SIZE;
                *slot = pg;
            }
            let (ia, ib) = (page_index(seg, spans[1]), page_index(seg, spans[3]));
            crate::options::set(15, 2);
            assert_eq!(span_free(seg, spans[1]), 0);
            std::thread::sleep(std::time::Duration::from_millis(20));
            assert_eq!(span_free(seg, spans[3]), 1, "the expired A was not purged");
            assert!(
                (*seg).pages[ia].purged,
                "A, free past the delay, was not purged"
            );
            assert!(!(*seg).pages[ib].purged, "B, freed just now, was purged");
            assert_ne!(
                (*seg).purge_expire,
                0,
                "the deadline was not re-armed for B"
            );
            crate::options::set(15, -1);
            for k in [0, 2, 4] {
                span_free(seg, spans[k]);
            }
            segment_free(seg).expect("free");
        }
    }

    /// The loop the delay exists for, through the heap: a 1 MiB buffer freed
    /// and allocated again 200 times under a 10 s delay purges nothing (under
    /// `purge_delay = 0` every free purged and every reuse refaulted), and a
    /// forced collect afterwards still gives the span back.
    #[test]
    #[cfg_attr(miri, ignore)] // megabytes of writes per round: too slow under miri
    fn a_large_buffer_loop_under_a_delay_never_purges_until_collected() {
        use crate::alloc::{collect, free, malloc, stats};
        let _g = crate::arena::adopt_tests::lock();
        let size = (MEDIUM_PAGE_SLICES + 8) * SEGMENT_SLICE_SIZE;
        if size > crate::types::LARGE_OBJ_SIZE_MAX {
            return; // a huge block in this geometry: released, not purged
        }
        crate::options::set(15, 10_000);
        let pin = malloc(64);
        let before = stats().purges;
        for i in 0..200u8 {
            let p = malloc(size);
            assert!(!p.is_null());
            // SAFETY: a live block of `size` bytes, freed once.
            unsafe {
                p.write_bytes(i, size);
                free(p);
            }
        }
        assert_eq!(stats().purges, before, "the loop purged under a delay");
        collect(true);
        assert!(
            stats().purges > before,
            "a forced collect did not purge the span"
        );
        crate::options::set(15, -1);
        // SAFETY: live, freed once.
        unsafe { free(pin) };
    }

    /// The same loop for a huge buffer (a dedicated segment, released to the
    /// arena on every free): under a delay its chunks go back pending and
    /// resident, so 20 rounds purge nothing; a forced collect then purges
    /// them through the arena pass.
    #[test]
    #[cfg_attr(miri, ignore)] // megabytes of writes per round: too slow under miri
    fn a_huge_buffer_loop_under_a_delay_never_purges_until_collected() {
        use crate::alloc::{collect, free, malloc, stats};
        let _g = crate::arena::adopt_tests::lock();
        let size = crate::types::LARGE_OBJ_SIZE_MAX + SEGMENT_SIZE / 2;
        crate::options::set(15, 10_000);
        let before = stats().purges;
        for i in 0..20u8 {
            let p = malloc(size);
            assert!(!p.is_null());
            assert!(
                crate::arena::owns(p),
                "setup: the huge block is not arena memory"
            );
            // SAFETY: a live block of `size` bytes, freed once; one byte per
            // OS page is enough to make it resident.
            unsafe {
                let mut o = 0;
                while o < size {
                    p.add(o).write(i);
                    o += 4096;
                }
                free(p);
            }
        }
        assert_eq!(stats().purges, before, "the huge loop purged under a delay");
        collect(true);
        assert!(
            stats().purges > before,
            "a forced collect did not purge the chunks"
        );
        crate::options::set(15, -1);
    }
}

#[cfg(all(test, feature = "std"))]
mod free_slot_tests {
    use super::*;
    use crate::alloc::{collect, free, malloc, stats};
    use crate::types::{MEDIUM_OBJ_SIZE_MAX, MEDIUM_PAGE_SLICES};

    /// A medium page full of `bs`-byte blocks, every byte written, then all
    /// freed but the first (the pin). `None` when this geometry has no such
    /// page (the block would not be medium, or not span two OS pages).
    fn pinned_page(bs: usize) -> Option<Vec<*mut u8>> {
        let page_bytes = MEDIUM_PAGE_SLICES * SEGMENT_SLICE_SIZE;
        if bs > MEDIUM_OBJ_SIZE_MAX || bs < 2 * os::page_size() {
            return None;
        }
        let first = malloc(bs);
        assert!(!first.is_null());
        // As many as the page holds: `reserved`, not `page_bytes / bs`,
        // because `blockmap` keeps its map after the blocks.
        // SAFETY: a page lookup on a live block.
        let k = unsafe { (*page_of(segment_of(first), first)).reserved as usize };
        assert!(k <= page_bytes / bs);
        let mut blocks = vec![first];
        blocks.extend((1..k).map(|_| malloc(bs)));
        let seg = segment_of(blocks[0]);
        // SAFETY: live blocks of `bs` bytes; page lookups on live blocks.
        unsafe {
            let pg = page_of(seg, blocks[0]);
            assert!(
                blocks
                    .iter()
                    .all(|&b| !b.is_null() && page_of(segment_of(b), b) == pg),
                "setup: the blocks did not share one page"
            );
            for &b in &blocks {
                b.write_bytes(0xAB, bs);
            }
            for &b in &blocks[1..] {
                free(b);
            }
        }
        Some(blocks)
    }

    /// One live block pins a medium page; a forced collect gives back every
    /// other slot's pages past the first, once, and the slots stay usable.
    #[test]
    fn a_pinned_medium_page_gives_back_its_free_slots() {
        let _g = crate::arena::adopt_tests::lock();
        crate::options::set(15, 0);
        let bs = 32 * 1024;
        let Some(blocks) = pinned_page(bs) else {
            crate::options::set(15, -1);
            return;
        };
        let before = stats().purges;
        collect(true);
        let purged = stats().purges - before;
        assert!(
            purged as usize >= blocks.len() - 1,
            "{purged} of {} free slots purged",
            blocks.len() - 1
        );
        // Linux and macOS give back zero pages; Windows' MEM_RESET leaves
        // the contents undefined, so only the count is checked there.
        // (miri's mock prim keeps contents too.)
        #[cfg(all(unix, not(miri)))]
        if crate::options::is_enabled(5) {
            for &b in &blocks[1..] {
                // SAFETY: the slot is free but still mapped and accessible:
                // the purge keeps it so, which is what this reads.
                let tail = unsafe { b.add(bs - 1).read_volatile() };
                assert_eq!(tail, 0, "a purged slot kept its contents");
            }
        }
        let before = stats().purges;
        collect(true);
        assert_eq!(stats().purges, before, "a purged slot was purged again");
        // The slots are ordinary free blocks: allocate and fill them all.
        let again: Vec<*mut u8> = (1..blocks.len()).map(|_| malloc(bs)).collect();
        // Every one comes back from the pinned page, i.e. the free list
        // survived the purge whole. A purge that reached a block's first OS
        // page would zero its link on Linux and end the list there: the
        // slots past it would leak and these would come from a new page.
        let mut sorted = again.clone();
        sorted.sort();
        sorted.dedup();
        assert_eq!(sorted.len(), again.len(), "a slot was handed out twice");
        // SAFETY: page lookups on live blocks.
        unsafe {
            let pg = page_of(segment_of(blocks[0]), blocks[0]);
            assert!(
                again.iter().all(|&b| page_of(segment_of(b), b) == pg),
                "a purged slot was lost from the free list"
            );
        }
        // SAFETY: live blocks, each written in full and freed once.
        unsafe {
            for &b in &again {
                assert!(!b.is_null());
                b.write_bytes(0xCD, bs);
                assert_eq!(b.add(bs - 1).read(), 0xCD);
            }
            for &b in again.iter().chain(&blocks[..1]) {
                free(b);
            }
        }
        crate::options::set(15, -1);
    }

    /// Under a positive delay an unforced collect only stamps a free slot;
    /// a later one past the delay purges it; a forced one never waits.
    #[test]
    #[cfg_attr(miri, ignore)] // real time: the mock prim's clock counts calls
    fn a_free_slot_waits_out_the_delay() {
        let _g = crate::arena::adopt_tests::lock();
        let bs = 16 * 1024;
        crate::options::set(15, 60_000);
        let Some(blocks) = pinned_page(bs) else {
            crate::options::set(15, -1);
            return;
        };
        let before = stats().purges;
        collect(false);
        assert_eq!(
            stats().purges,
            before,
            "an unforced collect purged inside the delay"
        );
        crate::options::set(15, 2);
        std::thread::sleep(std::time::Duration::from_millis(20));
        collect(false);
        let purged = stats().purges - before;
        assert!(
            purged as usize >= blocks.len() - 1,
            "only {purged} slots purged after the delay"
        );
        crate::options::set(15, -1);
        // SAFETY: the pin is live, freed once.
        unsafe { free(blocks[0]) };
    }

    /// Purging off: the free slots are left alone.
    #[test]
    fn no_slot_is_purged_with_purging_off() {
        let _g = crate::arena::adopt_tests::lock();
        crate::options::set(15, -1);
        let Some(blocks) = pinned_page(32 * 1024) else {
            return;
        };
        let before = stats().purges;
        collect(true);
        assert_eq!(stats().purges, before);
        // SAFETY: the pin is live, freed once.
        unsafe { free(blocks[0]) };
    }
}