rayforce-sys 1.3.2

/*
 *   Copyright (c) 2025-2026 Anton Kundenko <singaraiona@gmail.com>
 *   All rights reserved.

 *   Permission is hereby granted, free of charge, to any person obtaining a copy
 *   of this software and associated documentation files (the "Software"), to deal
 *   in the Software without restriction, including without limitation the rights
 *   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *   copies of the Software, and to permit persons to whom the Software is
 *   furnished to do so, subject to the following conditions:

 *   The above copyright notice and this permission notice shall be included in all
 *   copies or substantial portions of the Software.

 *   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 *   SOFTWARE.
 */

#if defined(__APPLE__)
#  define _DARWIN_C_SOURCE
#elif !defined(_WIN32)
#  define _GNU_SOURCE       /* ftruncate, MAP_SHARED, etc. */
#endif

#include "heap.h"
#include "cow.h"
#include "sys.h"
#include "core/platform.h"
#include "core/timer.h"   /* ray_time_now_ms — idle-decay clock */
#include "core/pool.h"    /* ray_pool_each_worker_heap — maintenance targets */
#include "core/runtime.h" /* ray_sys_mem_flush — a cgroup's statistics flushed */
#include "table/sym.h"
#include "table/domain.h"
#include "lang/eval.h"
#include "store/hnsw.h"
#include "store/csr.h"
#include "store/part.h"
#include "ops/idxop.h"
#include <string.h>
#include <stdlib.h>     /* getenv */
#include <stdio.h>      /* snprintf */
#include <unistd.h>     /* getpid, close, ftruncate, unlink */
#include <fcntl.h>      /* open, fcntl, F_PREALLOCATE on macOS */
#include <errno.h>
#include <sys/stat.h>   /* O_*  modes */
#include <sys/types.h>
#include <stdatomic.h>

/* File-backed spill (a pool or direct block mapped over a preallocated temp
 * file when the anonymous mapping is refused) is POSIX-only.  Windows pools
 * take only the anonymous VirtualAlloc path (docs/architecture/memory.md):
 * the spill helpers are compiled out and swap_fd stays -1 there. */
#if defined(RAY_OS_WINDOWS)
#  define RAY_HEAP_FILE_SPILL 0
#else
#  define RAY_HEAP_FILE_SPILL 1
#  include <sys/mman.h>   /* mmap, munmap */
#endif

#ifdef DEBUG
/* =====================================================================
 * Debug-only stale-pointer detector (issue #240 investigation).
 *
 * Shadow set of block addresses that are currently FREE (sitting on a
 * freelist, in a slab cache, or queued on a foreign list).  Any
 * ray_free / ray_release / ray_retain that touches an address in this
 * set is a use-after-free; report with a backtrace and abort before
 * the allocator state is corrupted.
 *
 * Opt-in: set RAY_DFD=1 (debug builds only).  ASan cannot see
 * use-after-free inside the pool allocator, so this is the tool of
 * choice for chasing double releases (found while debugging #240).
 * RAY_DFD_NO_ABORT=1 reports without aborting.
 * ===================================================================== */
#if !defined(RAY_OS_WINDOWS)
#include <execinfo.h>
#endif

#define DFD_CAP_BITS 20
#define DFD_CAP      (1u << DFD_CAP_BITS)
#define DFD_TOMB     ((const void*)(uintptr_t)1)

static const void*  dfd_slots[DFD_CAP];
static atomic_flag  dfd_lock_flag = ATOMIC_FLAG_INIT;

static bool dfd_enabled(void) {
    static int on = -1;
    if (on < 0) {
        const char* e = getenv("RAY_DFD");
        on = (e && *e && *e != '0') ? 1 : 0;
    }
    return on == 1;
}

static void dfd_lock(void)   { while (atomic_flag_test_and_set_explicit(&dfd_lock_flag, memory_order_acquire)) {} }
static void dfd_unlock(void) { atomic_flag_clear_explicit(&dfd_lock_flag, memory_order_release); }

static inline uint32_t dfd_hash(const void* p) {
    uint64_t x = (uint64_t)(uintptr_t)p >> 5;
    x *= 0x9E3779B97F4A7C15ull;
    return (uint32_t)(x >> (64 - DFD_CAP_BITS));
}

static void dfd_add(const void* p) {
    if (!dfd_enabled()) return;
    dfd_lock();
    uint32_t i = dfd_hash(p);
    uint32_t first_tomb = UINT32_MAX;
    for (uint32_t n = 0; n < DFD_CAP; n++, i = (i + 1) & (DFD_CAP - 1)) {
        if (dfd_slots[i] == p) { dfd_unlock(); return; }       /* already present */
        if (dfd_slots[i] == DFD_TOMB && first_tomb == UINT32_MAX) first_tomb = i;
        if (dfd_slots[i] == NULL) {
            dfd_slots[first_tomb != UINT32_MAX ? first_tomb : i] = p;
            dfd_unlock();
            return;
        }
    }
    dfd_unlock();  /* table full — detector degrades, never blocks */
}

static void dfd_remove(const void* p) {
    if (!dfd_enabled()) return;
    dfd_lock();
    uint32_t i = dfd_hash(p);
    for (uint32_t n = 0; n < DFD_CAP; n++, i = (i + 1) & (DFD_CAP - 1)) {
        if (dfd_slots[i] == p) { dfd_slots[i] = DFD_TOMB; break; }
        if (dfd_slots[i] == NULL) break;
    }
    dfd_unlock();
}

static bool dfd_contains(const void* p) {
    if (!dfd_enabled()) return false;
    dfd_lock();
    uint32_t i = dfd_hash(p);
    bool found = false;
    for (uint32_t n = 0; n < DFD_CAP; n++, i = (i + 1) & (DFD_CAP - 1)) {
        if (dfd_slots[i] == p) { found = true; break; }
        if (dfd_slots[i] == NULL) break;
    }
    dfd_unlock();
    return found;
}

/* Forget every tracked address inside [lo, hi).  Must run whenever a
 * pool is munmapped (heap destroy, oversized-pool reclaim): tests
 * destroy and recreate heaps, and a new pool mapped at a recycled
 * address would otherwise alias stale "free" entries and report
 * phantom stale-retains on legitimately live blocks. */
static void dfd_purge_range(uintptr_t lo, uintptr_t hi) {
    if (!dfd_enabled()) return;
    dfd_lock();
    for (uint32_t i = 0; i < DFD_CAP; i++) {
        uintptr_t p = (uintptr_t)dfd_slots[i];
        if (p > (uintptr_t)DFD_TOMB && p >= lo && p < hi)
            dfd_slots[i] = DFD_TOMB;
    }
    dfd_unlock();
}

static void dfd_report(const char* who, const void* p) {
    fprintf(stderr, "\n=== DFD: %s on FREED block %p ===\n", who, p);
#if !defined(RAY_OS_WINDOWS)
    void* frames[64];
    int n = backtrace(frames, 64);
    backtrace_symbols_fd(frames, n, 2);
#endif
    fflush(stderr);
    if (!getenv("RAY_DFD_NO_ABORT")) abort();
}

/* Called from cow.c (ray_release / ray_retain). */
void ray_dfd_check_live(const void* p, const char* who);
void ray_dfd_check_live(const void* p, const char* who) {
    if (p && dfd_contains(p)) dfd_report(who, p);
}

/* Walk every freelist of every registered heap; abort on NULL links or
 * cycles.  Called at GC entry so corruption is caught at the same point
 * the release build crashes (issue #240), but deterministically. */
static void dfd_validate_freelists(void);
#else
#define dfd_add(p)              ((void)0)
#define dfd_remove(p)           ((void)0)
#define dfd_purge_range(lo, hi) ((void)0)
#endif

/* Portable disk-block preallocation.  Returns 0 on success, errno-style
 * code on failure (matching posix_fallocate's contract).  Linux has
 * posix_fallocate natively.  macOS uses fcntl(F_PREALLOCATE) — try
 * contiguous first, fall back to non-contiguous, then ftruncate to
 * extend the file size if needed (F_PREALLOCATE doesn't grow the file
 * beyond its current size). */
#if RAY_HEAP_FILE_SPILL
static int heap_preallocate(int fd, off_t offset, off_t len) {
#if defined(__APPLE__)
    fstore_t fs = {
        .fst_flags    = F_ALLOCATECONTIG | F_ALLOCATEALL,
        .fst_posmode  = F_PEOFPOSMODE,
        .fst_offset   = 0,
        .fst_length   = offset + len,
        .fst_bytesalloc = 0,
    };
    if (fcntl(fd, F_PREALLOCATE, &fs) == -1) {
        /* Retry without contiguous-only constraint. */
        fs.fst_flags = F_ALLOCATEALL;
        if (fcntl(fd, F_PREALLOCATE, &fs) == -1) return errno ? errno : -1;
    }
    /* F_PREALLOCATE reserves blocks but doesn't grow the logical file
     * size — extend with ftruncate so mmap'd pages past the old size
     * can actually be written without SIGBUS. */
    if (ftruncate(fd, offset + len) != 0) return errno ? errno : -1;
    return 0;
#else
    return posix_fallocate(fd, offset, len);
#endif
}
#endif /* RAY_HEAP_FILE_SPILL */

/* --------------------------------------------------------------------------
 * Static asserts
 * -------------------------------------------------------------------------- */
_Static_assert(sizeof(ray_pool_hdr_t) <= 16,
               "ray_pool_hdr_t must fit in aux (16 bytes)");

/* --------------------------------------------------------------------------
 * Thread-local state
 * -------------------------------------------------------------------------- */
RAY_TLS ray_heap_t*     ray_tl_heap  = NULL;

/* Stats tracking — gated by RAY_MEM_STATS (1 in DEBUG builds, 0 in release).
 * All stats go through the per-heap struct (ray_tl_heap->stats) so that
 * heap merges keep bytes_allocated accurate.
 *
 * bytes_allocated is only modified by the owning thread (alloc/local-free)
 * or by the main thread during GC flush (return_to_owner=true, workers idle).
 * No atomics needed. */
#if RAY_MEM_STATS
#  define RAY_STAT(x) (x)
#else
#  define RAY_STAT(x) ((void)0)
#endif

/* --------------------------------------------------------------------------
 * Bitmap-based heap ID allocator (atomic CAS, reusable IDs)
 *
 * Each bit in the bitmap represents one heap ID. Acquiring sets a bit,
 * releasing clears it. IDs are reused after release (unlike a monotonic
 * counter). Cursor rotates to spread contention across words.
 * -------------------------------------------------------------------------- */
static _Atomic(uint64_t) g_heap_id_bitmap[RAY_HEAP_ID_WORDS] = { [0] = 1ULL };
static _Atomic(uint64_t) g_heap_id_cursor = 0;

/* Direct large-allocation counters (process-wide, exact).  Direct blocks are
 * standalone mmaps not owned by any per-thread heap, so their bytes live here
 * rather than in a per-thread stats slot; ray_mem_stats surfaces them. */
static _Atomic(int64_t) g_direct_bytes = 0;
static _Atomic(int64_t) g_direct_count = 0;

/* One process-wide, explicitly-scoped allocation trace for `.mem.ts`.
 * The inactive hot-path cost is one relaxed load plus a predicted-not-taken
 * branch in ray_alloc/ray_free.  While active, atomics make allocations from
 * the main and every worker heap contribute to the same query measurement. */
static _Atomic(uint32_t) g_mem_trace_active = 0;
static _Atomic(uint64_t) g_mem_trace_alloc_count = 0;
static _Atomic(uint64_t) g_mem_trace_free_count = 0;
static _Atomic(uint64_t) g_mem_trace_allocated = 0;
static _Atomic(uint64_t) g_mem_trace_freed = 0;
static _Atomic(int64_t)  g_mem_trace_live = 0;
static _Atomic(uint64_t) g_mem_trace_peak = 0;

static inline void mem_trace_note_alloc(size_t bytes) {
    if (RAY_LIKELY(atomic_load_explicit(&g_mem_trace_active,
                                        memory_order_relaxed) != 1)) return;
    atomic_fetch_add_explicit(&g_mem_trace_alloc_count, 1, memory_order_relaxed);
    atomic_fetch_add_explicit(&g_mem_trace_allocated, (uint64_t)bytes,
                              memory_order_relaxed);
    int64_t live = atomic_fetch_add_explicit(&g_mem_trace_live, (int64_t)bytes,
                                             memory_order_relaxed) + (int64_t)bytes;
    uint64_t candidate = live > 0 ? (uint64_t)live : 0;
    uint64_t peak = atomic_load_explicit(&g_mem_trace_peak, memory_order_relaxed);
    while (candidate > peak &&
           !atomic_compare_exchange_weak_explicit(&g_mem_trace_peak, &peak,
                                                   candidate,
                                                   memory_order_relaxed,
                                                   memory_order_relaxed)) {}
}

static inline void mem_trace_note_free(size_t bytes) {
    if (RAY_LIKELY(atomic_load_explicit(&g_mem_trace_active,
                                        memory_order_relaxed) != 1)) return;
    atomic_fetch_add_explicit(&g_mem_trace_free_count, 1, memory_order_relaxed);
    atomic_fetch_add_explicit(&g_mem_trace_freed, (uint64_t)bytes,
                              memory_order_relaxed);
    atomic_fetch_sub_explicit(&g_mem_trace_live, (int64_t)bytes,
                              memory_order_relaxed);
}

bool ray_mem_trace_begin(void) {
    uint32_t expected = 0;
    if (!atomic_compare_exchange_strong_explicit(&g_mem_trace_active, &expected, 2,
                                                  memory_order_acq_rel,
                                                  memory_order_relaxed))
        return false;
    atomic_store_explicit(&g_mem_trace_alloc_count, 0, memory_order_relaxed);
    atomic_store_explicit(&g_mem_trace_free_count, 0, memory_order_relaxed);
    atomic_store_explicit(&g_mem_trace_allocated, 0, memory_order_relaxed);
    atomic_store_explicit(&g_mem_trace_freed, 0, memory_order_relaxed);
    atomic_store_explicit(&g_mem_trace_live, 0, memory_order_relaxed);
    atomic_store_explicit(&g_mem_trace_peak, 0, memory_order_relaxed);
    atomic_store_explicit(&g_mem_trace_active, 1, memory_order_release);
    return true;
}

void ray_mem_trace_end(ray_mem_trace_t* out) {
    /* Freeze first: state 2 makes allocation hooks ignore the scope while
     * preventing another caller from resetting the counters mid-snapshot. */
    atomic_store_explicit(&g_mem_trace_active, 2, memory_order_release);
    if (out) {
        out->alloc_count = atomic_load_explicit(&g_mem_trace_alloc_count,
                                                memory_order_relaxed);
        out->free_count = atomic_load_explicit(&g_mem_trace_free_count,
                                               memory_order_relaxed);
        out->allocated_bytes = atomic_load_explicit(&g_mem_trace_allocated,
                                                    memory_order_relaxed);
        out->freed_bytes = atomic_load_explicit(&g_mem_trace_freed,
                                                memory_order_relaxed);
        out->net_bytes = atomic_load_explicit(&g_mem_trace_live,
                                              memory_order_relaxed);
        out->peak_live_bytes = atomic_load_explicit(&g_mem_trace_peak,
                                                    memory_order_relaxed);
    }
    atomic_store_explicit(&g_mem_trace_active, 0, memory_order_release);
}

/* Anonymous (RAM-resident) pool, direct and sys bytes we have committed.
 * File-backed spill mappings are NOT counted (they can be evicted to disk,
 * never OOM-kill).  When a new anon mapping would push this past the
 * watermark, it is routed to a spill file instead — or, when no spill file can
 * be made, to RAM anyway: this never rejects work, it just picks disk over
 * RAM.  A spill mapping is MAP_SHARED: after fork() parent and child share
 * it instead of each having a copy, so a child that keeps running rayforce
 * code must not write blocks the parent spilled. */
static _Atomic(int64_t) g_anon_committed = 0;
static _Atomic(int64_t) g_anon_watermark = 0;   /* 0 = the default (heap_anon_watermark) */
static _Atomic(int64_t) g_anon_peak      = 0;   /* high-water of g_anon_committed */

/* What has been mapped, for ray_heap_relieve to learn what a statement did:
 * the number of anonymous mappings and the bytes of the direct and sys blocks
 * among them — sys blocks too small to spill (RAY_SYS_SPILL_MIN) left out —
 * and the number of mappings that went past RAM — a pool, direct or sys
 * block mapped over a spill file (the watermark's call, or a refused
 * anonymous mapping), or taken from RAM past the watermark because no spill
 * file could be made. */
static _Atomic(int64_t) g_anon_maps   = 0;
static _Atomic(int64_t) g_anon_large  = 0;
static _Atomic(int64_t) g_anon_spills = 0;
static _Atomic(int64_t) g_anon_stmt_peak = 0;   /* high-water since the last boundary */
static _Atomic(int64_t) g_spill_pools = 0;      /* spill pools mapped, all heaps */

static void heap_anon_spilled(void) {
    atomic_fetch_add_explicit(&g_anon_spills, 1, memory_order_relaxed);
}

/* Commit `bytes` of anonymous (RAM-resident) footprint and advance the peak
 * high-water.  Called from the anon-commit sites (pools, direct blocks, sys
 * blocks); file-backed spill mappings never call this (they must not count
 * toward the RAM watermark).  `mapped`: a mapping ray_heap_relieve counts —
 * a pool, a direct block, a sys block that could spill; `large`: one of the
 * last two. */
static void heap_anon_commit(int64_t bytes, bool mapped, bool large) {
    if (mapped) atomic_fetch_add_explicit(&g_anon_maps, 1, memory_order_relaxed);
    if (large) atomic_fetch_add_explicit(&g_anon_large, bytes, memory_order_relaxed);
    int64_t nv = atomic_fetch_add_explicit(&g_anon_committed, bytes,
                                            memory_order_relaxed) + bytes;
    int64_t pk = atomic_load_explicit(&g_anon_peak, memory_order_relaxed);
    while (nv > pk &&
           !atomic_compare_exchange_weak_explicit(&g_anon_peak, &pk, nv,
                                                   memory_order_relaxed,
                                                   memory_order_relaxed)) {
        /* pk reloaded with the current value on failure; retry. */
    }
    pk = atomic_load_explicit(&g_anon_stmt_peak, memory_order_relaxed);
    while (nv > pk &&
           !atomic_compare_exchange_weak_explicit(&g_anon_stmt_peak, &pk, nv,
                                                   memory_order_relaxed,
                                                   memory_order_relaxed)) {}
}

/* Threshold above which anon allocations spill to disk.  Default keeps our
 * anon footprint below the RAM the process may actually use — physical RAM,
 * or the container's cgroup limit when that is smaller (#688) — by a quarter
 * of it, at least 3 GiB and at most half.  That room is for the page cache:
 * the mapped inputs, the dirty pages of the files being written and those of
 * the spill files themselves.  Without it spilling would start only once
 * nothing was left, and under a cgroup the page cache is charged to the same
 * limit (#712). */
static int64_t heap_anon_watermark(void) {
    int64_t wm = atomic_load_explicit(&g_anon_watermark, memory_order_relaxed);
    if (wm > 0) return wm;
    int64_t lim = ray_sys_ram_limit();
    if (lim <= 0) return lim;
    int64_t room = lim / 4;
    if (room < (INT64_C(3) << 30)) room = INT64_C(3) << 30;
    if (room > lim / 2) room = lim / 2;
    return lim - room;
}
/* True if committing `bytes` more anonymous RAM would cross the watermark. */
static bool heap_anon_would_exceed(size_t bytes) {
    int64_t wm = heap_anon_watermark();
    if (wm <= 0) return false;   /* unknown RAM → keep the fast anon path */
    int64_t cur = atomic_load_explicit(&g_anon_committed, memory_order_relaxed);
    return cur + (int64_t)bytes > wm;
}

/* A boundary far from the watermark visits the heaps for the spill pools
 * alone (ray_heap_relieve) — to park the empty ones, discard their pages and
 * age them — so at most this often: a stream of small statements (a
 * server's) would otherwise walk every worker heap each time, 13-14 µs a
 * statement at 16 workers.  A new watermark visits at the next boundary. */
#define RAY_FAR_VISIT_MS 100
static _Atomic(int64_t) g_far_visit_ms = INT64_MIN / 2;

int64_t ray_heap_anon_committed(void) {
    return atomic_load_explicit(&g_anon_committed, memory_order_relaxed);
}
int64_t ray_heap_anon_peak(void) {
    return atomic_load_explicit(&g_anon_peak, memory_order_relaxed);
}
int64_t ray_heap_anon_watermark(void) {
    return heap_anon_watermark();
}
int64_t ray_heap_mem_budget(void) {
    int64_t wm = atomic_load_explicit(&g_anon_watermark, memory_order_relaxed);
    return wm > 0 ? wm : ray_sys_ram_limit();
}
/* A new budget: the next boundary visits the heaps. */
void ray_heap_set_anon_watermark(int64_t bytes) {
    atomic_store_explicit(&g_anon_watermark, bytes < 0 ? 0 : bytes,
                          memory_order_relaxed);
    atomic_store_explicit(&g_far_visit_ms, INT64_MIN / 2, memory_order_relaxed);
}

ray_heap_t* ray_heap_registry[RAY_HEAP_REGISTRY_SIZE];

/* Serializes registry slot writes (register / unregister) and the
 * abandoned-heap LIFO.  Workers wake on the shutdown signal together and
 * leave their heaps CONCURRENTLY, so those two structures need one owner at
 * a time.  Spinlock matches the dfd_lock pattern above; held only across the
 * short critical sections, never across munmap or syscalls.
 *
 * The free path deliberately does NOT take this lock: it only READS a
 * registry slot, and a heap in the registry is never freed while a thread
 * could still be freeing into it (see ray_heap_abandon). */
static atomic_flag ray_registry_lock_flag = ATOMIC_FLAG_INIT;
static void ray_registry_lock(void)   { while (atomic_flag_test_and_set_explicit(&ray_registry_lock_flag, memory_order_acquire)) {} }
static void ray_registry_unlock(void) { atomic_flag_clear_explicit(&ray_registry_lock_flag, memory_order_release); }

/* Pending-merge queue head (lock-free LIFO) */
_Atomic(ray_heap_t*) ray_heap_pending_merge = NULL;

static int heap_id_acquire(void) {
    uint64_t start = atomic_fetch_add_explicit(&g_heap_id_cursor, 1,
                                                memory_order_relaxed);
    for (uint64_t off = 0; off < RAY_HEAP_ID_WORDS; off++) {
        uint64_t idx = (start + off) % RAY_HEAP_ID_WORDS;
        uint64_t word = atomic_load_explicit(&g_heap_id_bitmap[idx],
                                              memory_order_relaxed);
        while (~word != 0ULL) {
            uint64_t free_bits = ~word;
            uint64_t bit = (uint64_t)__builtin_ctzll(free_bits);
            uint64_t mask = 1ULL << bit;
            uint64_t new_word = word | mask;
            if (atomic_compare_exchange_weak_explicit(
                    &g_heap_id_bitmap[idx], &word, new_word,
                    memory_order_acq_rel, memory_order_relaxed)) {
                return (int)(idx * 64 + bit);
            }
            /* CAS failed — word updated, retry with new value */
        }
    }
    return -1;  /* pool exhausted */
}

static void heap_id_release(int id) {
    if (id < 0 || id >= (int)RAY_HEAP_ID_BITS) return;
    uint64_t idx = (uint64_t)id >> 6;
    uint64_t bit = (uint64_t)id & 63ULL;
    uint64_t mask = ~(1ULL << bit);
    atomic_fetch_and_explicit(&g_heap_id_bitmap[idx], mask,
                               memory_order_release);
}

/* --------------------------------------------------------------------------
 * Parallel flag
 * -------------------------------------------------------------------------- */
_Atomic(uint32_t) ray_parallel_flag = 0;

/* --------------------------------------------------------------------------
 * Helpers
 * -------------------------------------------------------------------------- */

static uint8_t ceil_log2(size_t n) {
    if (n <= 1) return 0;
    return (uint8_t)(64 - __builtin_clzll(n - 1));
}

uint8_t ray_order_for_size(size_t data_size) {
    if (data_size > SIZE_MAX - 32) return RAY_HEAP_MAX_ORDER + 1;
    size_t total = data_size + 32;  /* 32B ray_t header (no prefix) */
    uint8_t k = ceil_log2(total);
    if (k < RAY_ORDER_MIN) k = RAY_ORDER_MIN;
    return k;
}

/* --------------------------------------------------------------------------
 * Pool management
 *
 * Self-aligned pools: pool base = ptr & ~(pool_size - 1).
 * First min-block (64B at offset 0) reserved for pool header.
 * Remaining space split via cascading buddy split.
 *
 * For oversized blocks (order > POOL_ORDER), pool_order = order + 1
 * so the cascading split produces a right-half block of the needed order.
 * -------------------------------------------------------------------------- */

static bool heap_add_pool(ray_heap_t* h, uint8_t order);
static bool heap_unpark(ray_heap_t* h, bool spill);
static void heap_unpark_all(ray_heap_t* h);
static bool heap_make_room(size_t bytes);
static int64_t heap_decay_threshold_ms(void);
static void heap_abandon_spill(ray_heap_t* h);

/* --------------------------------------------------------------------------
 * Freelist operations (circular sentinel via fl_prev/fl_next)
 *
 * Each freelist[order] is a ray_fl_head_t sentinel. fl_remove() unlinks a
 * block from ANY circular list without needing the head pointer — enabling
 * safe cross-heap buddy coalescing.
 * -------------------------------------------------------------------------- */

RAY_INLINE void heap_insert_block(ray_heap_t* h, ray_t* blk, uint8_t order) {
    dfd_add(blk);
    ray_fl_head_t* head = &h->freelist[order];
    ray_t* first = head->fl_next;
    blk->fl_prev = (ray_t*)head;
    blk->fl_next = first;
    first->fl_prev = blk;
    head->fl_next = blk;
    ray_atomic_store(&blk->rc, 0);  /* free marker */
    blk->order = order;
    /* Free-block AGE (page-release aging, see GC pass 5).  `attrs` is dead
     * state while a block is free — splits already leave garbage here and
     * every allocation rewrites the header — so it carries the number of
     * GC passes this block has survived on a freelist.  Reset on every
     * insert (fresh free, coalesce, split). */
    blk->attrs = 0;
    h->avail |= (1ULL << order);
}

RAY_INLINE void heap_split_block(ray_heap_t* h, ray_t* blk,
                                uint8_t target_order, uint8_t block_order) {
    while (block_order > target_order) {
        block_order--;
        ray_t* buddy = (ray_t*)((char*)blk + BSIZEOF(block_order));
        buddy->mmod  = 0;
        buddy->order = block_order;
        heap_insert_block(h, buddy, block_order);
    }
}

/* --------------------------------------------------------------------------
 * Coalescing: merge block with buddies up to pool_order
 *
 * Pool header at offset 0 has rc=1 and order=RAY_ORDER_MIN, so buddy
 * checks always fail before reaching the header. Safe sentinel.
 * -------------------------------------------------------------------------- */

static void heap_coalesce(ray_heap_t* h, ray_t* blk,
                          uintptr_t pool_base, uint8_t pool_order) {
    dfd_remove(blk);
    uint8_t order = blk->order;

    /* During parallel execution, coalesce only same-pool buddies owned by
     * THIS heap.  blk is local (this path runs only for local frees), and
     * buddies stay in-pool (ray_buddy_of XORs within the pool), so a free
     * (rc==0) buddy lives only on our own freelist — which only we touch
     * during parallel.  rc!=0 excludes in-flight cross-thread frees.  The
     * heap_id guard is belt-and-suspenders against unexpected aliasing. */
    if (atomic_load_explicit(&ray_parallel_flag, memory_order_relaxed) != 0) {
        for (;; order++) {
            if (order >= pool_order) break;
            ray_t* buddy = ray_buddy_of(blk, order, pool_base);
            if (ray_atomic_load(&buddy->rc) != 0 || buddy->order != order) break;
            ray_pool_hdr_t* bphdr = ray_pool_of(buddy);
            if (!bphdr || bphdr->heap_id != h->id) break;
            fl_remove(buddy);
            dfd_remove(buddy);
            if (fl_empty(&h->freelist[order]))
                h->avail &= ~(1ULL << order);
            blk = (buddy < blk) ? buddy : blk;
        }
        heap_insert_block(h, blk, order);
        return;
    }

    for (;; order++) {
        if (order >= pool_order) break;

        ray_t* buddy = ray_buddy_of(blk, order, pool_base);
        __builtin_prefetch(buddy, 0, 1);

        uint32_t buddy_rc = ray_atomic_load(&buddy->rc);
        if (buddy_rc != 0 || buddy->order != order) break;

        fl_remove(buddy);
        if (fl_empty(&h->freelist[order]))
            h->avail &= ~(1ULL << order);

        blk = (buddy < blk) ? buddy : blk;
    }

    heap_insert_block(h, blk, order);
}

/* --------------------------------------------------------------------------
 * heap_add_pool implementation
 * -------------------------------------------------------------------------- */

static bool heap_add_pool(ray_heap_t* h, uint8_t order) {
    if (h->pool_count >= RAY_MAX_POOLS) return false;

    uint8_t pool_order;
    if (order >= RAY_HEAP_POOL_ORDER)
        pool_order = order + 1;  /* need one order larger for header + block */
    else
        pool_order = RAY_HEAP_POOL_ORDER;

    if (pool_order > RAY_HEAP_MAX_ORDER) return false;
    size_t pool_size = BSIZEOF(pool_order);

    void* mem = NULL;
    int   swap_fd  = -1;
    char* swap_path = NULL;

    /* Proactively back this pool with a disk spill file if an anon mapping
     * would push our RAM footprint past the watermark — a file-backed pool
     * can't be OOM-killed.  Otherwise take the fast anon path (and still fall
     * back to a spill file below if the kernel refuses the mapping). */
    bool over = heap_anon_would_exceed(pool_size);
    if (!over)
        mem = ray_vm_alloc_aligned(pool_size, pool_size);

#if !RAY_HEAP_FILE_SPILL
    /* No spill files here: past the watermark the pool comes from RAM. */
    if (!mem && over) mem = ray_vm_alloc_aligned(pool_size, pool_size);
    if (!mem) return false;
#else
    if (!mem) do {
        /* Anonymous mmap refused — usually means RAM+swap can't satisfy
         * pool_size right now.  Fall back to file-backed mmap: create a
         * tempfile in h->swap_path, reserve `pool_size` bytes of disk
         * blocks (so writes won't SIGBUS later on disk-full), then map
         * the file at a self-aligned address using the anonymous-VM
         * reservation trick — no over-allocation of file or disk. */
        static _Atomic uint64_t swap_counter = 0;
        uint64_t cnt = atomic_fetch_add_explicit(&swap_counter, 1, memory_order_relaxed);

        size_t plen = strlen(h->swap_path);
        size_t need = plen + 64;  /* room for "rayheap_<pid>_<heap>_<cnt>.dat" */
        swap_path = (char*)ray_sys_alloc(need);
        if (!swap_path) break;
        snprintf(swap_path, need, "%srayheap_%d_%u_%llu.dat",
                 h->swap_path, (int)getpid(), (unsigned)h->id,
                 (unsigned long long)cnt);

        /* Close-on-exec, as every spill file: a child of .sys.exec would
         * otherwise hold the unlinked file — its pages, dirty ones written
         * back — for as long as it runs, past the pool's unmapping. */
        swap_fd = open(swap_path, O_RDWR | O_CREAT | O_EXCL | O_CLOEXEC, 0600);
        if (swap_fd < 0) {
            ray_sys_free(swap_path);
            swap_path = NULL;
            break;
        }

        /* Reserve EXACTLY pool_size bytes of disk blocks AND grow the
         * file to pool_size.  Crucial that the file is empty (EOF=0)
         * when this runs: macOS F_PREALLOCATE with F_PEOFPOSMODE
         * extends past the current EOF, so doing this before any other
         * ftruncate keeps the reservation == pool_size, not 2x.  ENOSPC
         * here surfaces as a clean false return -> ray_alloc NULL ->
         * ray_error("oom") at the wrapper layer. */
        if (heap_preallocate(swap_fd, 0, (off_t)pool_size) != 0) {
            close(swap_fd);
            unlink(swap_path);
            ray_sys_free(swap_path);
            swap_fd = -1;
            swap_path = NULL;
            break;
        }

        /* Reserve 2*pool_size of address space anonymously to guarantee
         * a self-aligned subrange exists.  PROT_NONE is enough — we
         * never read/write the anon mapping; it just holds the address
         * range so the kernel won't hand it out to a concurrent mmap.
         * After computing the aligned subrange, free the slack and
         * MAP_FIXED the file-backed mapping over the kept region. */
        size_t reserve_size = pool_size + pool_size;
        void* anon = mmap(NULL, reserve_size, PROT_NONE,
                          MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
        if (anon == MAP_FAILED) {
            close(swap_fd);
            unlink(swap_path);
            ray_sys_free(swap_path);
            swap_fd = -1;
            swap_path = NULL;
            break;
        }

        uintptr_t addr    = (uintptr_t)anon;
        uintptr_t aligned = (addr + pool_size - 1) & ~(pool_size - 1);
        if (aligned > addr)
            munmap(anon, aligned - addr);
        uintptr_t end         = addr + reserve_size;
        uintptr_t aligned_end = aligned + pool_size;
        if (end > aligned_end)
            munmap((void*)aligned_end, end - aligned_end);

        /* MAP_FIXED replaces the kept anon mapping atomically with the
         * file-backed one.  No address-space race since the kept range
         * is still anon-reserved at this point. */
        void* mapped = mmap((void*)aligned, pool_size,
                            PROT_READ | PROT_WRITE,
                            MAP_SHARED | MAP_FIXED, swap_fd, 0);
        if (mapped == MAP_FAILED) {
            munmap((void*)aligned, pool_size);
            close(swap_fd);
            unlink(swap_path);
            ray_sys_free(swap_path);
            swap_fd = -1;
            swap_path = NULL;
            break;
        }

        /* Count the swap-backed pool as committed working set — it is a RAM
         * substitute with preallocated blocks, not an evictable file cache.
         * The PROT_NONE reserve + slack trims above are transient and stay
         * uncounted.  Freed via ray_vm_free (which subtracts) at pool teardown. */
        ray_sys_track_add((int64_t)pool_size);
        mem = (void*)aligned;

        /* Unlink now that the mapping holds the inode: the file keeps its
         * blocks until the last mapping goes away, and the kernel reclaims
         * them on munmap or on process exit, whichever comes first.  Doing
         * it here rather than at teardown means no path out of this heap
         * can strand the file — a worker that abandons its heap for reuse
         * never runs a teardown at all. */
        unlink(swap_path);
        ray_sys_free(swap_path);
        swap_path = NULL;
    } while (0);
    /* No spill file (no writable spill directory, disk full): RAM past the
     * watermark rather than failing the allocation. */
    if (!mem && over) mem = ray_vm_alloc_aligned(pool_size, pool_size);
    if (!mem) return false;
#endif /* RAY_HEAP_FILE_SPILL */

    /* Enable transparent huge pages on anon pools (Linux).  Self-aligned
     * 32MB pools are 2MB-aligned, hence THP-eligible.  Never on file-backed
     * swap mappings.  Env RAY_HEAP_HUGEPAGE=0 disables (THP-jitter-averse
     * deployments). */
    uint8_t pool_hugepage = 0;
    if (swap_fd < 0) {  /* anon pool only */
        const char* hp = getenv("RAY_HEAP_HUGEPAGE");
        if (!(hp && hp[0] == '0'))
            pool_hugepage = ray_vm_hugepage(mem, pool_size) ? 1 : 0;
    }

    /* --- Write pool header at offset 0 --- */
    ray_t* hdr_block = (ray_t*)mem;
    memset(hdr_block, 0, BSIZEOF(RAY_ORDER_MIN));
    hdr_block->mmod  = 0;
    hdr_block->order = RAY_ORDER_MIN;
    ray_atomic_store(&hdr_block->rc, 1);  /* sentinel: never free */

    ray_pool_hdr_t* hdr = (ray_pool_hdr_t*)hdr_block;  /* overlay on aux */
    hdr->heap_id    = h->id;
    hdr->pool_order = pool_order;
    hdr->vm_base    = mem;  /* on POSIX, same as aligned base */

    /* --- Cascading split: split from pool_order down to RAY_ORDER_MIN.
     *     Right half of each split → freelist.
     *     Leftmost min-block = pool header (already set, rc=1). --- */
    for (uint8_t o = pool_order; o > RAY_ORDER_MIN; o--) {
        ray_t* right = (ray_t*)((char*)mem + BSIZEOF(o - 1));
        right->mmod  = 0;
        right->order = (uint8_t)(o - 1);
        heap_insert_block(h, right, (uint8_t)(o - 1));
    }

    /* --- Track pool --- */
    h->pools[h->pool_count] = (ray_pool_entry_t){
        .base       = mem,
        .pool_order = pool_order,
        .backed     = (swap_fd >= 0) ? 1 : 0,
        .hugepage   = pool_hugepage,
        .swap_fd    = swap_fd,
        .swap_path  = swap_path,  /* NULL when not backed */
    };
    h->pool_count++;

    if (swap_fd < 0)   /* anonymous pool: counts toward the RAM watermark */
        heap_anon_commit((int64_t)pool_size, true, false);
    else
        atomic_fetch_add_explicit(&g_spill_pools, 1, memory_order_relaxed);
    if (over || swap_fd >= 0) {
        heap_anon_spilled();
        h->spilled = 1;
    }

    return true;
}

/* --------------------------------------------------------------------------
 * Slab cache flush (with coalescing for GC effectiveness)
 * -------------------------------------------------------------------------- */

static void heap_flush_slabs(ray_heap_t* h) {
    for (int i = 0; i < RAY_SLAB_ORDERS; i++) {
        while (h->slabs[i].count > 0) {
            ray_t* blk = h->slabs[i].stack[--h->slabs[i].count];
            dfd_remove(blk);
            int pidx = heap_find_pool(h, blk);
            uintptr_t pb;
            uint8_t po;
            if (pidx >= 0) {
                pb = (uintptr_t)h->pools[pidx].base;
                po = h->pools[pidx].pool_order;
            } else {
                ray_pool_hdr_t* phdr = ray_pool_of(blk);
                if (!phdr) continue;
                pb = (uintptr_t)phdr;
                po = phdr->pool_order;
            }
            heap_coalesce(h, blk, pb, po);
        }
    }
}

/* --------------------------------------------------------------------------
 * Foreign-list drain (owner side)
 *
 * Every block on h->foreign is OWNED BY h: a cross-thread free routes the
 * block to its owner rather than parking it on the freeing heap (see the
 * foreign path in ray_free).  So the drain needs no pool-header lookup for
 * ownership, no registry access, and no "are the workers idle?" gate — it
 * takes the whole list with one exchange and coalesces into its own
 * freelists, which no other thread touches.
 *
 * The acquire on the exchange pairs with the release on the pusher's CAS,
 * so the block contents written before the push are visible here.
 * -------------------------------------------------------------------------- */

static void heap_drain_foreign(ray_heap_t* h) {
    ray_t* blk = atomic_exchange_explicit(&h->foreign, NULL,
                                          memory_order_acquire);
    while (blk) {
        ray_t* next = blk->fl_next;
        int pidx = heap_find_pool(h, blk);
        uintptr_t pb;
        uint8_t po;
        if (pidx >= 0) {
            pb = (uintptr_t)h->pools[pidx].base;
            po = h->pools[pidx].pool_order;
        } else {
            /* h owns the block but does not track its pool — only reachable
             * through ray_heap_merge's RAY_MAX_POOLS overflow, which rewrites
             * the header's heap_id without room for the pool entry. */
            ray_pool_hdr_t* phdr = ray_pool_of(blk);
            if (!phdr) { blk = next; continue; }
            pb = (uintptr_t)phdr;
            po = phdr->pool_order;
            /* Adopted from a heap that left the registry while its pools
             * were still mapped (ray_heap_push_pending).  We never charged
             * the allocation, so there is nothing to take off our books —
             * unlike the merge-overflow case above, where the header was
             * rewritten to name us. */
            if (phdr->heap_id != h->id) {
                heap_coalesce(h, blk, pb, po);
                blk = next;
                continue;
            }
        }
        RAY_STAT(h->stats.bytes_allocated -= BSIZEOF(blk->order));
        heap_coalesce(h, blk, pb, po);
        blk = next;
    }
}

/* --------------------------------------------------------------------------
 * Owned-reference helpers
 * -------------------------------------------------------------------------- */

static bool ray_atom_str_is_sso(const ray_t* s) {
    if (s->slen >= 1 && s->slen <= 7) return true;
    if (s->slen == 0 && s->obj == NULL) return true;
    return false;
}

static bool ray_atom_owns_obj(const ray_t* v) {
    if (v->type == -RAY_GUID) return v->obj != NULL;
    if (v->type == -RAY_STR) return !ray_atom_str_is_sso(v);
    return false;
}

static void ray_release_owned_refs(ray_t* v) {
    if (!v || RAY_IS_ERR(v)) return;

    if (ray_is_atom(v)) {
        if (v->type == RAY_LAMBDA) {
            /* Lambda stores params/body/bytecode/constants plus optional
             * nfo/debug/closure owned references in ray_data. */
            ray_t** slots = (ray_t**)ray_data(v);
            for (int i = 0; i < 4; i++) {
                if (slots[i] && !RAY_IS_ERR(slots[i]))
                    ray_release(slots[i]);
            }
            /* Release optional debug info slots */
            if (LAMBDA_NFO(v)) ray_release(LAMBDA_NFO(v));
            if (LAMBDA_DBG(v)) ray_release(LAMBDA_DBG(v));
            if (LAMBDA_CLOSURE(v)) ray_release(LAMBDA_CLOSURE(v));
            if (LAMBDA_CAPROWS(v)) ray_release(LAMBDA_CAPROWS(v));
            return;
        }
        if (v->type == RAY_LAZY) {
            ray_graph_t* g = RAY_LAZY_GRAPH(v);
            if (g) {
                ray_graph_free(g);
                RAY_LAZY_GRAPH(v) = NULL;
            }
            return;
        }
        /* I64 atom tagged as an HNSW handle owns a ray_hnsw_t — free it
         * when the atom's rc drops to zero so rebindings and scope-exit
         * don't leak the (potentially large) index graph. */
        if (v->type == -RAY_I64 && (v->attrs & RAY_ATTR_HNSW)) {
            ray_hnsw_t* idx = (ray_hnsw_t*)(uintptr_t)v->i64;
            if (idx) ray_hnsw_free(idx);
            v->i64 = 0;
            v->attrs &= (uint8_t)~RAY_ATTR_HNSW;
            return;
        }
        /* I64 atom tagged as a graph handle owns a ray_rel_t (CSR) — free
         * it on rc→0 so users don't leak when rebinding or going out of
         * scope without an explicit (.graph.free h). */
        if (v->type == -RAY_I64 && (v->attrs & RAY_ATTR_GRAPH)) {
            ray_rel_t* rel = (ray_rel_t*)(uintptr_t)v->i64;
            if (rel) ray_rel_free(rel);
            v->i64 = 0;
            v->attrs &= (uint8_t)~RAY_ATTR_GRAPH;
            return;
        }
        if (ray_atom_owns_obj(v) && v->obj && !RAY_IS_ERR(v->obj))
            ray_release(v->obj);
        return;
    }

    if (v->attrs & RAY_ATTR_SLICE) {
        if (v->slice_parent && !RAY_IS_ERR(v->slice_parent))
            ray_release(v->slice_parent);
        return;
    }

    /* RAY_INDEX block: release per-kind payload children + saved-aux
     * pointers.  Must run before the LIST/TABLE compound checks below
     * (which would mistreat the data[] payload as child pointers). */
    if (v->type == RAY_INDEX) {
        ray_index_t* ix = ray_index_payload(v);
        ray_index_release_payload(ix);
        ray_index_release_saved(ix);
        return;
    }

    /* Vector with attached index: aux[0..7] holds an owning ref to the
     * index ray_t.  Bytes 8-15 are NOT displaced into the index for a
     * STR or SYM parent — attach_finalize leaves the pool or domain
     * pointer in place there (ops/idxop.c, and the layout note in
     * rayforce.h), and ray_index_release_saved is a no-op.  So the ref
     * taken when that pool or domain was attached is still this
     * vector's to drop, exactly as for an unindexed one. */
    if (v->attrs & RAY_ATTR_HAS_INDEX) {
        /* A mmap-resident index (mmod==1) is a PASSENGER in this column's file
         * mapping — the column's single munmap frees it.  Releasing it here
         * would ray_free it and munmap a sub-region of the mapping.  Only a
         * heap-built index (mmod==0) is released by pointer. */
        if (v->index && !RAY_IS_ERR(v->index) && v->index->mmod != 1)
            ray_release(v->index);
        /* Fall through to the STR / SYM arms below rather than repeating
         * them: the pool or domain is still at bytes 8-15 either way, and a
         * second copy of this dispatch is the drift this whole change is
         * about. */
    }

    if (v->type == RAY_STR && v->str_pool && !RAY_IS_ERR(v->str_pool)) {
        /* Return rather than fall through, the way the SYM arm below does:
         * a mapped column's pool owns the region they share, so this
         * release can be the one that unmaps the page `v` sits on.  None
         * of the checks after this one apply to a STR vec anyway. */
        ray_release(v->str_pool);
        return;
    }

    /* RAY_SYM vec: drop the resolution-domain ref (aux bytes 8-15).
     * No-op for the immortal runtime singleton — the common case. */
    if (v->type == RAY_SYM && v->sym_domain) {
        ray_sym_domain_release(v->sym_domain);
        return;
    }

    if (RAY_IS_PARTED(v->type)) {
        int64_t n_segs = v->len;
        ray_t** segs = (ray_t**)ray_data(v);
        for (int64_t i = 0; i < n_segs; i++) {
            if (segs[i] && !RAY_IS_ERR(segs[i]))
                ray_release(segs[i]);
        }
        return;
    }

    if (v->type == RAY_MAPCOMMON) {
        ray_t** ptrs = (ray_t**)ray_data(v);
        if (ptrs[0] && !RAY_IS_ERR(ptrs[0])) ray_release(ptrs[0]);
        if (ptrs[1] && !RAY_IS_ERR(ptrs[1])) ray_release(ptrs[1]);
        return;
    }

    if (v->type == RAY_TABLE || v->type == RAY_DICT) {
        ray_t** slots = (ray_t**)ray_data(v);
        if (slots[0] && !RAY_IS_ERR(slots[0])) ray_release(slots[0]);
        if (slots[1] && !RAY_IS_ERR(slots[1])) ray_release(slots[1]);
        return;
    }

    if (v->type == RAY_LIST) {
        ray_t** ptrs = (ray_t**)ray_data(v);
        for (int64_t i = 0; i < v->len; i++) {
            ray_t* child = ptrs[i];
            if (child && !RAY_IS_ERR(child)) ray_release(child);
        }
    }
}

bool ray_retain_owned_refs(ray_t* v) {
    if (!v || RAY_IS_ERR(v)) return true;

    if (ray_is_atom(v)) {
        if (v->type == RAY_LAMBDA) {
            ray_t** slots = (ray_t**)ray_data(v);
            for (int i = 0; i < 4; i++) {
                if (slots[i] && !RAY_IS_ERR(slots[i]))
                    ray_retain(slots[i]);
            }
            if (LAMBDA_NFO(v)) ray_retain(LAMBDA_NFO(v));
            if (LAMBDA_DBG(v)) ray_retain(LAMBDA_DBG(v));
            if (LAMBDA_CLOSURE(v)) ray_retain(LAMBDA_CLOSURE(v));
            if (LAMBDA_CAPROWS(v)) ray_retain(LAMBDA_CAPROWS(v));
            return true;
        }
        /* Lazy handles own their graph uniquely — no retain on copy */
        if (v->type == RAY_LAZY) return true;
        /* HNSW handle owns its ray_hnsw_t uniquely.  Deep-clone the index
         * so the copy is an independent owner with the same semantics as
         * the source.  On clone-OOM, detach the copy (so caller can free
         * it cleanly) and signal failure — the caller must not treat the
         * copy as a valid handle. */
        if (v->type == -RAY_I64 && (v->attrs & RAY_ATTR_HNSW)) {
            ray_hnsw_t* src = (ray_hnsw_t*)(uintptr_t)v->i64;
            if (src) {
                ray_hnsw_t* dup = ray_hnsw_clone(src);
                if (!dup) {
                    v->i64 = 0;
                    v->attrs &= (uint8_t)~RAY_ATTR_HNSW;
                    return false;
                }
                v->i64 = (int64_t)(uintptr_t)dup;
            }
            return true;
        }
        /* Graph handle (ray_rel_t*).  No deep-clone API exists for CSR
         * (could be GBs), and the user's mental model is "the handle is
         * the graph" — like a file descriptor.  On block-copy we detach
         * the COPY (the source keeps ownership) so a later free of the
         * duplicate doesn't touch the source's CSR.  This matches the
         * lazy-graph (RAY_LAZY) policy above. */
        if (v->type == -RAY_I64 && (v->attrs & RAY_ATTR_GRAPH)) {
            v->i64 = 0;
            v->attrs &= (uint8_t)~RAY_ATTR_GRAPH;
            return true;
        }
        if (ray_atom_owns_obj(v) && v->obj && !RAY_IS_ERR(v->obj))
            ray_retain(v->obj);
        return true;
    }

    if (v->attrs & RAY_ATTR_SLICE) {
        if (v->slice_parent && !RAY_IS_ERR(v->slice_parent))
            ray_retain(v->slice_parent);
        return true;
    }

    if (v->type == RAY_INDEX) {
        ray_index_t* ix = ray_index_payload(v);
        ray_index_retain_payload(ix);
        ray_index_retain_saved(ix);
        return true;
    }

    if (v->attrs & RAY_ATTR_HAS_INDEX) {
        /* Mirror ray_release_owned_refs: a mmap-resident passenger index
         * (mmod==1) is owned by the column's mapping, not refcounted here,
         * and the pool or domain at bytes 8-15 is this copy's to take a
         * ref on — skipping it is what let a copy that later sheds its
         * index release a reference it never held. */
        if (v->index && !RAY_IS_ERR(v->index) && v->index->mmod != 1)
            ray_retain(v->index);
        /* Falls through to the STR / SYM arms below — see the mirror of
         * this comment in ray_release_owned_refs. */
    }

    if (v->type == RAY_STR && v->str_pool && !RAY_IS_ERR(v->str_pool))
        ray_retain(v->str_pool);

    /* RAY_SYM vec: the header copy carried the domain pointer — take a
     * ref.  No-op for the immortal runtime singleton. */
    if (v->type == RAY_SYM && v->sym_domain) {
        ray_sym_domain_retain(v->sym_domain);
        return true;
    }

    if (RAY_IS_PARTED(v->type)) {
        int64_t n_segs = v->len;
        ray_t** segs = (ray_t**)ray_data(v);
        for (int64_t i = 0; i < n_segs; i++) {
            if (segs[i] && !RAY_IS_ERR(segs[i]))
                ray_retain(segs[i]);
        }
        return true;
    }

    if (v->type == RAY_MAPCOMMON) {
        ray_t** ptrs = (ray_t**)ray_data(v);
        if (ptrs[0] && !RAY_IS_ERR(ptrs[0])) ray_retain(ptrs[0]);
        if (ptrs[1] && !RAY_IS_ERR(ptrs[1])) ray_retain(ptrs[1]);
        return true;
    }

    if (v->type == RAY_TABLE || v->type == RAY_DICT) {
        ray_t** slots = (ray_t**)ray_data(v);
        if (slots[0] && !RAY_IS_ERR(slots[0])) ray_retain(slots[0]);
        if (slots[1] && !RAY_IS_ERR(slots[1])) ray_retain(slots[1]);
        return true;
    }

    if (v->type == RAY_LIST) {
        ray_t** ptrs = (ray_t**)ray_data(v);
        for (int64_t i = 0; i < v->len; i++) {
            ray_t* child = ptrs[i];
            if (child && !RAY_IS_ERR(child)) ray_retain(child);
        }
    }
    return true;
}

static void ray_detach_owned_refs(ray_t* v) {
    if (!v || RAY_IS_ERR(v)) return;

    if (ray_is_atom(v)) {
        if (v->type == RAY_LAMBDA) {
            ray_t** slots = (ray_t**)ray_data(v);
            for (int i = 0; i < 4; i++) slots[i] = NULL;
            LAMBDA_NFO(v) = NULL;
            LAMBDA_DBG(v) = NULL;
            LAMBDA_CLOSURE(v) = NULL;
            LAMBDA_CAPROWS(v) = NULL;
            LAMBDA_CAPQ(v) = 0;
            return;
        }
        if (v->type == RAY_LAZY) {
            RAY_LAZY_GRAPH(v) = NULL;
            RAY_LAZY_OP(v)    = NULL;
            return;
        }
        /* HNSW handle: ownership has been transferred elsewhere; stop the
         * rc→0 cleanup hook from freeing the (now-foreign) index. */
        if (v->type == -RAY_I64 && (v->attrs & RAY_ATTR_HNSW)) {
            v->i64 = 0;
            v->attrs &= (uint8_t)~RAY_ATTR_HNSW;
            return;
        }
        /* Graph handle: same semantics as HNSW — ownership has moved,
         * clear the bit so rc→0 doesn't ray_rel_free a foreign pointer. */
        if (v->type == -RAY_I64 && (v->attrs & RAY_ATTR_GRAPH)) {
            v->i64 = 0;
            v->attrs &= (uint8_t)~RAY_ATTR_GRAPH;
            return;
        }
        if (ray_atom_owns_obj(v)) v->obj = NULL;
        return;
    }

    if (v->attrs & RAY_ATTR_SLICE) {
        v->slice_parent = NULL;
        v->slice_offset = 0;
        v->attrs &= (uint8_t)~RAY_ATTR_SLICE;
        return;
    }

    if (v->type == RAY_INDEX) {
        ray_index_t* ix = ray_index_payload(v);
        switch ((ray_idx_kind_t)ix->kind) {
        case RAY_IDX_HASH:
            ix->u.hash.table = ix->u.hash.gkeys = NULL;
            ix->u.hash.offs  = ix->u.hash.rows  = NULL;
            break;
        case RAY_IDX_SORT:  ix->u.sort.perm = NULL; break;
        case RAY_IDX_BLOOM: ix->u.bloom.bits = NULL; break;
        default: break;
        }
        memset(ix->saved_aux, 0, 16);
        ix->saved_attrs = 0;
        return;
    }

    if (v->attrs & RAY_ATTR_HAS_INDEX) {
        v->index    = NULL;
        v->_idx_pad = NULL;
        v->attrs   &= (uint8_t)~RAY_ATTR_HAS_INDEX;
        return;
    }

    if (v->type == RAY_STR) {
        v->str_pool = NULL;
    }

    /* RAY_SYM vec: domain ref transferred elsewhere (e.g. the header
     * memcpy in ray_scratch_realloc) — clear so ray_free won't drop it.
     * The defensive ray_sym_vec_domain fallback covers any reader that
     * sees the detached header. */
    if (v->type == RAY_SYM) {
        v->sym_domain = NULL;
        return;
    }

    if (RAY_IS_PARTED(v->type)) {
        int64_t n_segs = v->len;
        ray_t** segs = (ray_t**)ray_data(v);
        for (int64_t i = 0; i < n_segs; i++)
            segs[i] = NULL;
        return;
    }

    if (v->type == RAY_MAPCOMMON) {
        ray_t** ptrs = (ray_t**)ray_data(v);
        ptrs[0] = NULL;
        ptrs[1] = NULL;
        return;
    }

    if (v->type == RAY_TABLE || v->type == RAY_DICT) {
        ray_t** slots = (ray_t**)ray_data(v);
        slots[0] = NULL;
        slots[1] = NULL;
        v->len = 0;
        return;
    }

    if (v->type == RAY_LIST) {
        v->len = 0;
    }
}

/* --------------------------------------------------------------------------
 * ray_alloc
 * -------------------------------------------------------------------------- */

/* File-backed direct mapping: a disk-spilled standalone allocation, used when
 * an allocation cannot be RAM-resident (larger than physical RAM) or when the
 * anonymous mmap was refused.  Mirrors heap_add_pool's swap fallback but at the
 * exact size — a direct block needs no pool alignment (it is located by its
 * stored map_size, not by pool-base masking), so a plain mmap suffices.
 * Returns the mapped base and the fd to close at free (*out_path is NULL: the
 * file is already unlinked); NULL on failure. */
static void* heap_direct_map_file(ray_heap_t* h, size_t map_size,
                                  int* out_fd, char** out_path) {
#if !RAY_HEAP_FILE_SPILL
    (void)h; (void)map_size; (void)out_fd; (void)out_path;
    return NULL;
#else
    static _Atomic uint64_t direct_swap_counter = 0;
    uint64_t cnt = atomic_fetch_add_explicit(&direct_swap_counter, 1,
                                             memory_order_relaxed);
    size_t plen = strlen(h->swap_path);
    size_t need = plen + 64;  /* room for "raydirect_<pid>_<heap>_<cnt>.dat" */
    char*  path = (char*)ray_sys_alloc(need);
    if (!path) return NULL;
    snprintf(path, need, "%sraydirect_%d_%u_%llu.dat", h->swap_path,
             (int)getpid(), (unsigned)h->id, (unsigned long long)cnt);

    int fd = open(path, O_RDWR | O_CREAT | O_EXCL | O_CLOEXEC, 0600);
    if (fd < 0) { ray_sys_free(path); return NULL; }

    /* Reserve real disk blocks up front so later stores can't SIGBUS on a
     * disk-full condition.  ENOSPC here surfaces as a clean NULL -> "oom". */
    if (heap_preallocate(fd, 0, (off_t)map_size) != 0) {
        close(fd); unlink(path); ray_sys_free(path); return NULL;
    }
    void* mapped = mmap(NULL, map_size, PROT_READ | PROT_WRITE,
                        MAP_SHARED, fd, 0);
    if (mapped == MAP_FAILED) {
        close(fd); unlink(path); ray_sys_free(path); return NULL;
    }
    /* Count as committed working set — a RAM substitute with preallocated
     * blocks (matches heap_add_pool's swap-pool accounting). */
    ray_sys_track_add((int64_t)map_size);
    /* Unlink now that the mapping holds the inode, as heap_add_pool does:
     * the kernel reclaims the blocks on munmap or at process exit.  Left
     * until ray_free, a process that exited or was killed holding the block
     * left the whole file on disk. */
    unlink(path);
    ray_sys_free(path);
    *out_fd   = fd;
    *out_path = NULL;
    return mapped;
#endif /* RAY_HEAP_FILE_SPILL */
}

/* --------------------------------------------------------------------------
 * Direct-block reuse cache
 *
 * Analytical queries allocate the same few large scratch/result blocks
 * every run (radix payload maps, order maps, result columns) and freed
 * them straight back to the kernel: ~330MB of anon mmap+munmap round-trip
 * per 10M-row group query, with the kernel re-zeroing every page on the
 * next fault (kernel_init_pages was ~8% of ClickBench q17).  Keep a small
 * global stash of recently freed ANON direct blocks and serve size-matched
 * requests from it.
 *
 * Accounting: a cached block's pages stay resident, so it remains counted
 * in both the committed-RAM tracker and the g_anon_committed watermark;
 * only the live-block stats (g_direct_bytes/count) drop on stash and rise
 * again on reuse.  Under watermark pressure the cache is drained before
 * new memory is committed.  File-backed spill blocks are never cached.
 *
 * Fit: best block with map_size in [need, need + max(need/4, 2MB)] — the
 * repeat-query case hits exactly; the bound keeps waste at <= 25%.
 * Concurrency: any thread may alloc or free a direct block. A global lock
 * guards an intrusive FIFO of recently freed blocks. Its links reuse the dead
 * allocation's prefix, so entry count needs no allocation or arbitrary cap.
 * Replacement evicts the oldest cached blocks to admit newly freed scratch;
 * obsolete CSV buffers cannot monopolize the byte budget indefinitely.
 * Detached mappings are returned to the OS outside the cache lock. */
typedef struct ray_direct_cache_node {
    struct ray_direct_cache_node* next;
    struct ray_direct_cache_node* prev;
    size_t map_size;
} ray_direct_cache_node_t;
_Static_assert(sizeof(ray_direct_cache_node_t) <= RAY_DIRECT_HDR,
               "cached links must fit in the dead direct-allocation prefix");
static ray_direct_cache_node_t* g_direct_cache_head;
static ray_direct_cache_node_t* g_direct_cache_tail;
static size_t g_direct_cache_bytes;
static _Atomic(int)    g_direct_cache_spin = 0;

static inline void direct_cache_lock(void) {
    while (atomic_exchange_explicit(&g_direct_cache_spin, 1,
                                    memory_order_acquire)) { /* spin */ }
}
static inline void direct_cache_unlock(void) {
    atomic_store_explicit(&g_direct_cache_spin, 0, memory_order_release);
}

static size_t direct_cache_budget(void) {
    int64_t wm = ray_heap_mem_budget();
    size_t b = (wm > 0) ? (size_t)wm / 16 : 0;
    /* 1/16 of the memory budget is the sole bound (1GB on a 16GB box, ~8GB
     * on 128GB).  Earlier absolute caps (512MB, then 4GB) each turned
     * out to exclude exactly the blocks whose kernel re-zeroing cost the
     * most at the next data scale: a 100M-row group query cycles several
     * GB-scale scratch blocks (order map ~800MB, per-partition gather
     * arrays ~1.6GB, result columns ~500MB) per execution, and every
     * cache miss is an mmap + munmap + full page-zeroing round trip
     * (measured 14-17% of q17/q18 wall as kernel_init_pages). */
    return b;
}

/* Caller holds the lock. Ownership transfers from the cache to the caller. */
static void direct_cache_unlink(ray_direct_cache_node_t* node) {
    if (node->prev) node->prev->next = node->next;
    else g_direct_cache_head = node->next;
    if (node->next) node->next->prev = node->prev;
    else g_direct_cache_tail = node->prev;
    g_direct_cache_bytes -= node->map_size;
}

/* Only detached nodes are passed here. No cache lock is held across munmap. */
static void direct_cache_release(ray_direct_cache_node_t* node) {
    while (node) {
        ray_direct_cache_node_t* next = node->next;
        size_t bytes = node->map_size;
        atomic_fetch_sub_explicit(&g_anon_committed, (int64_t)bytes, memory_order_relaxed);
        ray_vm_free(node, bytes);
        node = next;
    }
}

/* Best fitting cached block within the existing waste bound. Exact matches
 * stop the search; similar-sized analytical scratch is the common case. */
static void* direct_cache_take(size_t need, size_t* out_size) {
    direct_cache_lock();
    size_t slack = need / 4;
    if (slack < (2u << 20)) slack = (2u << 20);
    ray_direct_cache_node_t* best = NULL;
    for (ray_direct_cache_node_t* node = g_direct_cache_head; node; node = node->next) {
        size_t bytes = node->map_size;
        if (bytes >= need && bytes - need <= slack && (!best || bytes < best->map_size)) {
            best = node;
            if (bytes == need) break;
        }
    }
    if (best) {
        *out_size = best->map_size;
        direct_cache_unlink(best);
    }
    direct_cache_unlock();
    return best;
}

/* Recently freed blocks replace the oldest cached blocks within the unchanged
 * byte budget. The caller has already removed this block from live statistics. */
static bool direct_cache_put(void* base, size_t map_size) {
    size_t budget = direct_cache_budget();
    if (map_size > budget) return false;
    ray_direct_cache_node_t* evicted = NULL;
    direct_cache_lock();
    while (g_direct_cache_bytes > budget - map_size) {
        ray_direct_cache_node_t* old = g_direct_cache_head;
        direct_cache_unlink(old);
        old->next = evicted;
        evicted = old;
    }
    ray_direct_cache_node_t* node = base;
    node->map_size = map_size;
    node->next = NULL;
    node->prev = g_direct_cache_tail;
    if (g_direct_cache_tail) g_direct_cache_tail->next = node;
    else g_direct_cache_head = node;
    g_direct_cache_tail = node;
    g_direct_cache_bytes += map_size;
    direct_cache_unlock();
    direct_cache_release(evicted);
    return true;
}

/* Detach under the lock; return mappings to the OS without blocking alloc/free
 * callers on kernel unmapping and TLB invalidation. */
static void direct_cache_drain(void) {
    direct_cache_lock();
    ray_direct_cache_node_t* nodes = g_direct_cache_head;
    g_direct_cache_head = g_direct_cache_tail = NULL;
    g_direct_cache_bytes = 0;
    direct_cache_unlock();
    direct_cache_release(nodes);
}

void ray_heap_direct_cache_drain(void) {
    direct_cache_drain();
}

/* Bytes a drain would hand back now. */
static int64_t direct_cache_bytes(void) {
    direct_cache_lock();
    size_t b = g_direct_cache_bytes;
    direct_cache_unlock();
    return (int64_t)b;
}

/* Direct large allocation: mmap the exact page-rounded size instead of a
 * power-of-2 oversized buddy pool.  Returns a marked ray_t or NULL on failure. */
static ray_t* heap_alloc_direct(ray_heap_t* h, size_t data_size) {
    if (data_size > SIZE_MAX - RAY_DIRECT_HDR - 32 - 4095) return NULL;
    size_t map_size = (RAY_DIRECT_HDR + 32 + data_size + 4095) & ~(size_t)4095;

    void* base      = NULL;
    int   swap_fd   = -1;
    char* swap_path = NULL;

    /* Reuse cache first: a hit hands back resident, already-committed
     * pages — no mmap, no watermark commit (the block never left the
     * committed totals while cached), no kernel page-zeroing on fault. */
    size_t cached_size = 0;
    base = direct_cache_take(map_size, &cached_size);
    if (base) {
        map_size = cached_size;
    } else {
        /* If keeping this in anonymous RAM would push our footprint past the
         * watermark (default: physical RAM), back it with a disk file from the
         * start so it spills instead of being OOM-killed as its pages fault in.
         * (Under lenient overcommit an anon mmap the kernel can't actually back
         * SUCCEEDS, then kills us at fault time — so a fallback-on-failure alone
         * would not catch it.)  Otherwise use anonymous RAM, and fall back to a
         * spill file only if the kernel refuses the mapping.
         * Under pressure, give back what is free before deciding
         * (heap_make_room). */
        bool force_file = heap_anon_would_exceed(map_size) && heap_make_room(map_size);
        if (!force_file)
            base = ray_vm_alloc(map_size);   /* anon RW, page-aligned, counted */
        if (!base) {
            base = heap_direct_map_file(h, map_size, &swap_fd, &swap_path);
            /* No spill file (no writable spill directory, disk full): RAM
             * past the watermark rather than failing the allocation. */
            if (!base && force_file) base = ray_vm_alloc(map_size);
            if (!base) return NULL;
        }
        if (swap_fd < 0)   /* anonymous: counts toward the RAM watermark */
            heap_anon_commit((int64_t)map_size, true, true);
        if (force_file || swap_fd >= 0)
            heap_anon_spilled();
    }

    ray_direct_hdr_t* hdr = (ray_direct_hdr_t*)base;
    hdr->map_size  = map_size;
    hdr->swap_fd   = swap_fd;
    hdr->swap_path = swap_path;

    ray_t* v = (ray_t*)((char*)base + RAY_DIRECT_HDR);
    memset(v, 0, 32);
    v->mmod  = 0;
    v->order = RAY_ORDER_DIRECT;
    if (RAY_UNLIKELY(ray_rc_sync))
        ray_atomic_store(&v->rc, 1);
    else
        v->rc = 1;

    atomic_fetch_add_explicit(&g_direct_bytes, (int64_t)map_size, memory_order_relaxed);
    atomic_fetch_add_explicit(&g_direct_count, 1, memory_order_relaxed);
    RAY_STAT(h->stats.alloc_count++);
    return v;
}

ray_t* ray_alloc(size_t data_size) {
    ray_heap_t* h = ray_tl_heap;
    if (RAY_UNLIKELY(!h)) {
        ray_heap_init();
        h = ray_tl_heap;
        if (!h) return NULL;
    }

    uint8_t order = ray_order_for_size(data_size);

    /* Anything that would need a dedicated oversized pool goes direct — exact
     * size, no power-of-2 rounding, no order+1 pool.  This includes sizes
     * above the largest buddy order: direct blocks carry their exact map_size
     * in a prefix header, so they are not bounded by RAY_HEAP_MAX_ORDER.  On
     * mmap failure, fall through to the buddy pool only when the buddy model
     * can still represent the request. */
    if (RAY_UNLIKELY(order >= RAY_HEAP_POOL_ORDER)) {
        ray_t* v = heap_alloc_direct(h, data_size);
        if (v) {
            ray_direct_hdr_t* hdr = (ray_direct_hdr_t*)((char*)v - RAY_DIRECT_HDR);
            mem_trace_note_alloc(hdr->map_size);
            return v;
        }
        if (order > RAY_HEAP_MAX_ORDER) return NULL;
    }

    if (order > RAY_HEAP_MAX_ORDER) return NULL;

    /* Slab fast path */
    if (RAY_LIKELY(IS_SLAB_ORDER(order))) {
        int idx = SLAB_INDEX(order);
        if (RAY_LIKELY(h->slabs[idx].count > 0)) {
            ray_t* v = h->slabs[idx].stack[--h->slabs[idx].count];
            dfd_remove(v);

            /* Zero full 32-byte header (hot path).
             * Null bitmap (bytes 0-15) must be cleared for null-bit correctness. */
            memset(v, 0, 32);
            v->order = order;
            if (RAY_UNLIKELY(ray_rc_sync))
                ray_atomic_store(&v->rc, 1);
            else
                v->rc = 1;

            RAY_STAT(h->stats.alloc_count++);
            RAY_STAT(h->stats.slab_hits++);
            RAY_STAT(h->stats.bytes_allocated += BSIZEOF(order));
            RAY_STAT(h->stats.peak_bytes = h->stats.bytes_allocated > h->stats.peak_bytes
                ? h->stats.bytes_allocated : h->stats.peak_bytes);
            mem_trace_note_alloc(BSIZEOF(order));
            return v;
        }
    }

    /* Find free block via avail bitmask.
     * Avail bits can be stale from cross-heap fl_remove, so we loop
     * to find a genuinely non-empty freelist. */
    uint64_t candidates = h->avail & (UINT64_MAX << order);

    /* Scan past stale avail bits (cross-heap fl_remove may have emptied lists) */
    uint8_t found_order;
    for (;;) {
        if (RAY_UNLIKELY(candidates == 0)) {
            /* Out of blocks at this order.  Take back everything other
             * threads have freed to us BEFORE asking the OS for another
             * pool: those blocks are already ours, and mapping fresh memory
             * while our own sits on the list is how a repeated query grew a
             * server by a pool per execution (issue #439).  This ordering is
             * the invariant — no pool is ever added without a drain first. */
            heap_drain_foreign(h);
            candidates = h->avail & (UINT64_MAX << order);
            /* Then a pool the heap parked (heap_park_empty_pools), an
             * anonymous one first. */
            if (candidates == 0 && h->parked
                && (heap_unpark(h, false) || heap_unpark(h, true)))
                candidates = h->avail & (UINT64_MAX << order);
            if (candidates == 0) {
                if (!heap_add_pool(h, order)) return NULL;
                candidates = h->avail & (UINT64_MAX << order);
                if (candidates == 0) return NULL;
            }
        }
        found_order = (uint8_t)__builtin_ctzll(candidates);
        if (!fl_empty(&h->freelist[found_order])) break;
        /* Clear stale bit and try next */
        h->avail &= ~(1ULL << found_order);
        candidates &= ~(1ULL << found_order);
    }

    /* Pop from circular sentinel freelist */
    ray_fl_head_t* head = &h->freelist[found_order];
    ray_t* blk = head->fl_next;
    fl_remove(blk);
    dfd_remove(blk);
    if (fl_empty(head))
        h->avail &= ~(1ULL << found_order);

    /* Split down to requested order */
    heap_split_block(h, blk, order, found_order);

    /* Zero ray_t header and set metadata */
    memset(blk, 0, 32);
    blk->mmod  = 0;
    blk->order = order;
    if (RAY_UNLIKELY(ray_rc_sync))
        ray_atomic_store(&blk->rc, 1);
    else
        blk->rc = 1;

    RAY_STAT(h->stats.alloc_count++);
    RAY_STAT(h->stats.bytes_allocated += BSIZEOF(order));
    RAY_STAT(h->stats.peak_bytes = h->stats.bytes_allocated > h->stats.peak_bytes
        ? h->stats.bytes_allocated : h->stats.peak_bytes);

    mem_trace_note_alloc(BSIZEOF(order));

    return blk;
}

/* --------------------------------------------------------------------------
 * ray_free
 * -------------------------------------------------------------------------- */

/* Drop one reference to a shared file mapping; the last one unmaps it.
 *
 * Not atomic: every reference is taken and dropped while building or
 * freeing a block, both of which already run under the owning heap's
 * single-threaded discipline.  If a mapped column ever becomes shareable
 * across heaps this needs the same treatment as ray_t's own rc. */
/* Live shared mappings, keyed by base address.
 *
 * A block inside the region has to reach its own reference without going
 * through a pointer somebody may swap: the pool keeps the descriptor in its
 * header, but the column's only link to it would be str_pool, and that is
 * exactly what str_pool_cow replaces when a mapped column is mutated.  What
 * never changes is the column's address, and the column header IS the start
 * of the mapping — so the column looks its reference up by that.
 *
 * Touched once when a column is mapped and once when it is freed, never per
 * query, so a spin lock over a small bucket array is enough.  The
 * descriptors are their own chain nodes. */
#define RAY_FMAP_BUCKETS 256
static ray_file_map_t*  g_fmap[RAY_FMAP_BUCKETS];
static _Atomic(int)     g_fmap_lock = 0;

static inline void fmap_lock(void) {
    while (atomic_exchange_explicit(&g_fmap_lock, 1, memory_order_acquire))
        RAY_CPU_RELAX();
}
static inline void fmap_unlock(void) {
    atomic_store_explicit(&g_fmap_lock, 0, memory_order_release);
}
/* Bucket by the LOW bits of the page number, deliberately.  The kernel
 * hands out consecutive file mappings a few pages apart, so those bits are
 * what actually varies between columns — measured over forty mapped string
 * columns, forty distinct buckets and no chain longer than one.  Hashing
 * the high bits instead would put every column of a table in one chain. */
static inline size_t fmap_bucket(const void* base) {
    return ((uintptr_t)base >> 12) & (RAY_FMAP_BUCKETS - 1);
}

void ray_file_map_register(const void* base, ray_file_map_t* m) {
    if (!m) return;
    size_t b = fmap_bucket(base);
    fmap_lock();
    m->next = g_fmap[b];
    g_fmap[b] = m;
    fmap_unlock();
}

ray_file_map_t* ray_file_map_lookup(const void* base) {
    size_t b = fmap_bucket(base);
    fmap_lock();
    ray_file_map_t* m = g_fmap[b];
    while (m && m->base != base) m = m->next;
    fmap_unlock();
    return m;
}

void ray_file_map_release(ray_file_map_t* m) {
    if (!m) return;
    if (ray_atomic_dec(&m->rc) > 1) return;   /* returns the value BEFORE */
    /* Unlink if it was ever registered; a mapping nobody mutated never
     * entered the table and the walk simply finds nothing. */
    size_t b = fmap_bucket(m->base);
    fmap_lock();
    ray_file_map_t** pp = &g_fmap[b];
    while (*pp && *pp != m) pp = &(*pp)->next;
    if (*pp) *pp = m->next;
    fmap_unlock();
    ray_vm_unmap_file(m->base, m->len);
    ray_sys_free(m);
}

void ray_free(ray_t* v) {
    if (!v || RAY_IS_ERR(v)) return;
    if (v->attrs & RAY_ATTR_ARENA) return;  /* arena-owned, bulk-freed */
#ifdef DEBUG
    if (dfd_contains(v)) dfd_report("ray_free (double free)", v);
#endif

    /* Guard: keep rc=1 while releasing children so buddy coalescing
     * won't merge this block prematurely (it checks buddy_rc==0). */
    ray_atomic_store(&v->rc, 1);

    /* A mapped block owning a shared region takes its descriptor now: aux
     * lives inside the region it describes. */
    ray_file_map_t* owned_map = (v->mmod == 3) ? v->file_map : NULL;

    /* A mapped string column reaches its own reference through its pool,
     * which is the cheap path and needs no lock — but only while that
     * pointer still leads to the mapped pool.  Read it before the children
     * go, since releasing the pool may be what drops it. */
    ray_file_map_t* col_map = NULL;
    if (v->mmod == 1 && v->type == RAY_STR &&
        v->str_pool && !RAY_IS_ERR(v->str_pool) && v->str_pool->mmod == 3)
        col_map = v->str_pool->file_map;

    ray_release_owned_refs(v);

    ray_heap_t* h = ray_tl_heap;

    if (owned_map) {
        ray_file_map_release(owned_map);
        if (h) RAY_STAT(h->stats.free_count++);
        return;
    }

    /* File-mapped: munmap */
    if (v->mmod == 1) {
        if (v->type == RAY_TABLE || v->type == RAY_DICT || v->type == RAY_LIST) return;
        /* A column mapped by the store registered its region and holds one
         * of its references.  Looking that up by address rather than through
         * str_pool is the point: a mutated column's pool pointer no longer
         * leads anywhere useful, while its address is what the region was
         * registered under.  The descriptor also carries the true mapped
         * length, so nothing has to re-derive it from a header that the
         * children release may already have taken away. */
        /* col_map is the answer whenever the pool pointer still led to the
         * region, which is every column that was not mutated.  Only one
         * that had its pool swapped — str_pool_cow deep-copies a mapped
         * pool before appending — has to ask the registry, and only a
         * string column was ever registered.  So the lock stays off the
         * ordinary free entirely, and a mutated column pays it once. */
        /* A column loaded with an inline index registered its region too
         * (col.c): the index may have been detached since, and only the
         * descriptor still knows the mapped length. */
        ray_file_map_t* m = col_map;
        if (!m) m = ray_file_map_lookup(v);
        if (m) {
            ray_file_map_release(m);
            if (h) RAY_STAT(h->stats.free_count++);
            return;
        }
        if (v->type > 0 && v->type < RAY_TYPE_COUNT) {
            uint8_t esz = ray_sym_elem_size(v->type, v->attrs);
            size_t data_size = 32 + (size_t)v->len * esz;
            if (v->type == RAY_STR) {
                size_t pool_len = 0;
                if (v->str_pool && !RAY_IS_ERR(v->str_pool) && v->str_pool->len > 0)
                    pool_len = (size_t)v->str_pool->len;
                data_size += 32 + pool_len;
            }
            /* Inline index region: a mmap-resident (passenger) index extends the
             * mapping past the payload by its 32-aligned inline region.  Derive
             * that size from the index itself (ray_index_inline_size) rather than
             * stashing it in aux — str_pool occupies _idx_pad on STR columns, and
             * the payload size above already accounts for descriptors + pool.
             * Heap-resident indexes (RAY_MARK_MMAP clear) keep the payload-only
             * formula. */
            if ((v->attrs & RAY_ATTR_HAS_INDEX) && v->index && !RAY_IS_ERR(v->index)) {
                ray_index_t* ix = ray_index_payload(v->index);
                if (ix->markers & RAY_MARK_MMAP) {
                    int64_t region_off = ((int64_t)data_size + 31) & ~(int64_t)31;
                    data_size = (size_t)(region_off + ray_index_inline_size(ix));
                }
            }
            size_t mapped_size = (data_size + 4095) & ~(size_t)4095;
            ray_vm_unmap_file(v, mapped_size);
        } else {
            ray_vm_unmap_file(v, 4096);
        }
        if (h) RAY_STAT(h->stats.free_count++);
        return;
    }

    /* Legacy mmod==2 guard */
    if (v->mmod == 2) return;

    /* Direct large allocation: standalone mmap, freed by munmap.  Checked
     * BEFORE any pool masking (a direct block is not a pool member).  Any
     * thread may free it; the counter is a global atomic. */
    if (RAY_UNLIKELY(v->order > RAY_HEAP_MAX_ORDER)) {
        char*             base = (char*)v - RAY_DIRECT_HDR;
        ray_direct_hdr_t* hdr  = (ray_direct_hdr_t*)base;
        size_t map_size  = hdr->map_size;
        int    swap_fd   = hdr->swap_fd;
        char*  swap_path = hdr->swap_path;
        mem_trace_note_free(map_size);
        if (h) RAY_STAT(h->stats.free_count++);
        atomic_fetch_sub_explicit(&g_direct_bytes, (int64_t)map_size, memory_order_relaxed);
        atomic_fetch_sub_explicit(&g_direct_count, 1, memory_order_relaxed);
#if RAY_HEAP_FILE_SPILL
        if (swap_fd >= 0) {
            /* File-backed spill: mapped directly (not via ray_vm_alloc), so
             * unmap + uncount by hand, then close and unlink the spill file. */
            munmap(base, map_size);
            ray_sys_track_sub((int64_t)map_size);
            close(swap_fd);
            if (swap_path) { unlink(swap_path); ray_sys_free(swap_path); }
        } else
#else
        (void)swap_fd; (void)swap_path;
#endif
        if (direct_cache_put(base, map_size)) {
            /* Stashed for reuse: pages stay resident, so the block keeps
             * its committed-RAM and watermark accounting; only the live
             * stats above dropped.  Eviction (direct_cache_drain) performs
             * the deferred un-commit + munmap. */
        } else {
            atomic_fetch_sub_explicit(&g_anon_committed, (int64_t)map_size,
                                      memory_order_relaxed);
            ray_vm_free(base, map_size);   /* munmap + committed-RAM counter -= */
        }
        return;
    }

    if (!h) return;

    uint8_t order = v->order;

    if (order < RAY_ORDER_MIN || order > RAY_HEAP_MAX_ORDER) return;

    mem_trace_note_free(BSIZEOF(order));

#if RAY_MEM_STATS
    size_t block_size = BSIZEOF(order);
#endif

    /* Warm-first ownership check: probe the MRU pool, then scan the warm
     * h->pools[] array.  A hit proves the block is local and yields
     * pool base+order WITHOUT reading the cold pool header 32MB away.
     * Only a full miss falls through to the foreign path.  heap_find_pool
     * returning >=0 is equivalent to the old phdr->heap_id == h->id check. */
    int pidx = -1;
    if (h->last_pool_idx < h->pool_count) {
        uintptr_t pb = (uintptr_t)h->pools[h->last_pool_idx].base;
        uintptr_t pe = pb + BSIZEOF(h->pools[h->last_pool_idx].pool_order);
        if ((uintptr_t)v >= pb && (uintptr_t)v < pe)
            pidx = (int)h->last_pool_idx;
    }
    if (pidx < 0) pidx = heap_find_pool(h, v);

    if (pidx >= 0) {
        /* Local block — base+order from the warm pools array. */
        h->last_pool_idx = (uint32_t)pidx;
        uintptr_t pool_base  = (uintptr_t)h->pools[pidx].base;
        uint8_t   pool_order = h->pools[pidx].pool_order;

        /* Slab fast path */
        if (IS_SLAB_ORDER(order)) {
            int idx = SLAB_INDEX(order);
            if (h->slabs[idx].count < h->slab_cap[idx]) {
                /* rc=1 so buddy coalescing skips slab-cached blocks (a buddy
                 * freed concurrently reads rc; rc==0 would wrongly merge). */
                ray_atomic_store(&v->rc, 1);
                dfd_add(v);
                h->slabs[idx].stack[h->slabs[idx].count++] = v;
                RAY_STAT(h->stats.free_count++);
                RAY_STAT(h->stats.bytes_allocated -= block_size);
                return;
            }
        }

        heap_coalesce(h, v, pool_base, pool_order);
        RAY_STAT(h->stats.free_count++);
        RAY_STAT(h->stats.bytes_allocated -= block_size);
        return;
    }

    /* Foreign: not in any of our pools.  Push it onto the OWNER's list, not
     * ours.  A block parked here is memory only the owner can reuse, and
     * until this changed nothing but an explicit ray_heap_gc() ever moved it
     * — so a server that never called one had every worker start each query
     * short by the volume the consuming thread was sitting on, and map a
     * fresh pool to make up the difference, forever (issue #439).  Handing
     * the block back at free time makes the growth impossible instead of
     * merely collectable.
     *
     * The registry lookup is unlocked, which is sound because a registered
     * heap outlives every block it owns: threads abandon their heaps rather
     * than destroying them (ray_heap_abandon).
     *
     * Do NOT adjust bytes_allocated: the block stays counted on the owning
     * heap until IT drains and coalesces. */
    ray_pool_hdr_t* phdr = ray_pool_of(v);
    if (RAY_UNLIKELY(!phdr)) return;      /* not inside any live pool */
    uint16_t owner_id = phdr->heap_id;
    ray_heap_t* owner = ray_heap_registry[owner_id % RAY_HEAP_REGISTRY_SIZE];
    /* Not in the registry.  Either the owner was destroyed — its pools went
     * with it and the mapping is gone — or it is merely waiting to be merged
     * (ray_heap_push_pending unregisters first, while the pools stay mapped).
     * The two are indistinguishable from here, so keep the block on our own
     * list: the drain coalesces it against its own pool header and adopts it.
     * Dropping it instead would strand the block and pin its pool against
     * reclamation for the life of the process. */
    if (RAY_UNLIKELY(!owner || owner->id != owner_id)) owner = h;

    dfd_add(v);
    ray_t* head = atomic_load_explicit(&owner->foreign, memory_order_relaxed);
    do {
        v->fl_next = head;
    } while (!atomic_compare_exchange_weak_explicit(
                 &owner->foreign, &head, v,
                 memory_order_release, memory_order_relaxed));
    RAY_STAT(h->stats.free_count++);
}

/* --------------------------------------------------------------------------
 * Raw buffer allocator: malloc/calloc/realloc/free for plain byte buffers that
 * are NOT ray_t values, backed by the buddy heap (no libc malloc).  The block
 * is a real ray_t header + data; we stamp type=RAY_U8 (no owned children) so
 * ray_free reclaims it without walking elements, and hand back ray_data().  The
 * header sits 32 bytes before the returned pointer; ray_free_raw recovers it.
 * -------------------------------------------------------------------------- */

void* ray_alloc_raw(size_t n) {
    ray_t* v = ray_alloc(n);
    if (!v || RAY_IS_ERR(v)) return NULL;
    v->type = RAY_U8;          /* plain byte vec: ray_free reclaims, no child walk */
    v->len  = (int64_t)n;
    return ray_data(v);
}

void* ray_calloc_raw(size_t n) {
    void* p = ray_alloc_raw(n);
    if (p && n) memset(p, 0, n);
    return p;
}

void ray_free_raw(void* p) {
    if (!p) return;
    ray_free((ray_t*)((char*)p - 32));   /* 32 = ray_t header before the data */
}

bool ray_raw_discard(void* p, size_t from, size_t upto) {
#if RAY_HEAP_FILE_SPILL && defined(MADV_REMOVE)
    if (!p || upto <= from) return false;
    ray_t* v = (ray_t*)((char*)p - 32);
    /* An anonymous block is below the watermark, and the direct-block
     * cache keeps it resident for its next use: left alone. */
    if (!ray_is_direct(v) || !ray_direct_file_backed(v)) return false;
    size_t room = ray_block_data_bytes(v);
    if (upto > room) upto = room;
    long ps = sysconf(_SC_PAGESIZE);
    uintptr_t pg = ps > 0 ? (uintptr_t)ps : 4096;
    /* whole pages only, and never the first: it holds the headers */
    uintptr_t first = ((uintptr_t)p + pg - 1) & ~(pg - 1);
    uintptr_t lo = ((uintptr_t)p + from) & ~(pg - 1);
    uintptr_t hi = ((uintptr_t)p + upto) & ~(pg - 1);
    if (lo < first) lo = first;
    if (hi <= lo) return false;
    /* punched out of the spill file: a dirty page is dropped, not written */
    return madvise((void*)lo, hi - lo, MADV_REMOVE) == 0;
#else
    /* No way to punch a mapped range here (macOS has no MADV_REMOVE, and
     * F_PUNCHHOLE does not say what it does to pages mapped shared): the
     * pages stay until the block is freed, as before. */
    (void)p; (void)from; (void)upto;
    return false;
#endif
}

void* ray_realloc_raw(void* p, size_t n) {
    if (!p) return ray_alloc_raw(n);
    ray_t* v = (ray_t*)((char*)p - 32);
    size_t cur = ray_block_data_bytes(v);   /* data capacity (buddy or direct) */
    if (n <= cur) { v->len = (int64_t)n; return p; }
    void* np = ray_alloc_raw(n);
    if (!np) return NULL;
    memcpy(np, p, cur);                     /* copy the whole old block (>= valid bytes) */
    ray_free_raw(p);
    return np;
}

/* --------------------------------------------------------------------------
 * ray_alloc_copy
 * -------------------------------------------------------------------------- */

ray_t* ray_alloc_copy(ray_t* v) {
    if (!v || RAY_IS_ERR(v)) return NULL;
    size_t data_size;
    if (v->attrs & RAY_ATTR_SLICE) {
        /* Slice blocks are header-only views; their .len reflects the
         * slice extent on the parent, not their own storage.  A naive
         * len*esz would overrun the 32-byte header.  ray_retain_owned_refs
         * then bumps slice_parent's rc, keeping the view valid. */
        data_size = 0;
    } else if (ray_is_atom(v)) {
        data_size = 0;
    } else if (v->type == RAY_TABLE || v->type == RAY_DICT) {
        data_size = 2 * sizeof(ray_t*);
    } else if (RAY_IS_PARTED(v->type) || v->type == RAY_MAPCOMMON) {
        int64_t n_ptrs = v->len;
        if (v->type == RAY_MAPCOMMON) n_ptrs = 2;
        if (n_ptrs < 0) return ray_error("oom", NULL);
        data_size = (size_t)n_ptrs * sizeof(ray_t*);
    } else if (v->type == RAY_LIST) {
        /* RAY_LIST has type==0, which the generic branch below (t <= 0)
         * would route to data_size=0, silently producing a header-only copy
         * whose item-pointer area is uninitialised — a shallow COW of a
         * shared list would then lose every element.  Handle explicitly. */
        if (v->len < 0 || (uint64_t)v->len > SIZE_MAX / sizeof(ray_t*))
            return ray_error("oom", NULL);
        data_size = (size_t)ray_len(v) * sizeof(ray_t*);
    } else {
        int8_t t = ray_type(v);
        if (t <= 0 || t >= RAY_TYPE_COUNT)
            data_size = 0;
        else {
            uint8_t esz = ray_sym_elem_size(t, v->attrs);
            if (v->len < 0 || (esz > 0 && (uint64_t)v->len > SIZE_MAX / esz))
                return ray_error("oom", NULL);
            data_size = (size_t)ray_len(v) * esz;
        }
    }
    ray_t* copy = ray_alloc(data_size);
    if (!copy) return NULL;

    uint8_t new_order = copy->order;
    uint8_t new_mmod  = copy->mmod;
    memcpy(copy, v, 32 + data_size);
    copy->mmod  = new_mmod;
    copy->order = new_order;
    if (RAY_UNLIKELY(ray_rc_sync))
        ray_atomic_store(&copy->rc, 1);
    else
        copy->rc = 1;
    /* A table's key map (RAY_IDX_UKEY, ops/idxop.h) describes the rows of the
     * table it was built for and upsert mutates it in place, recording the
     * row count it now covers.  Two tables sharing one would each be writing
     * their own count into it, so a copy starts without it and builds its own
     * on demand.  Nothing else attaches an index to a RAY_TABLE, so clearing
     * the bit here cannot discard anything else.  Before retain, so the copy
     * never takes a reference it does not keep. */
    if (copy->type == RAY_TABLE && (copy->attrs & RAY_ATTR_HAS_INDEX)) {
        copy->index    = NULL;
        copy->_idx_pad = NULL;
        copy->attrs   &= (uint8_t)~RAY_ATTR_HAS_INDEX;
    }

    if (!ray_retain_owned_refs(copy)) {
        /* Deep-clone of an owned resource failed (e.g. HNSW index OOM).
         * The copy's owned state has already been neutralized, so a plain
         * ray_free won't touch the source's resources. */
        ray_free(copy);
        return ray_error("oom", NULL);
    }
    return copy;
}

/* --------------------------------------------------------------------------
 * ray_scratch_alloc / ray_scratch_realloc
 * -------------------------------------------------------------------------- */

ray_t* ray_scratch_alloc(size_t data_size) {
    return ray_alloc(data_size);
}

ray_t* ray_scratch_realloc(ray_t* v, size_t new_data_size) {
    ray_t* new_v = ray_alloc(new_data_size);
    if (!new_v) return NULL;
    if (v && !RAY_IS_ERR(v)) {
        size_t old_data;
        if (v->attrs & RAY_ATTR_SLICE)
            old_data = 0;  /* slice blocks are header-only — no own storage */
        else if (ray_is_atom(v))
            old_data = 0;
        else if (v->type == RAY_LIST) {
            if (v->len < 0) { old_data = 0; }
            else old_data = (size_t)ray_len(v) * sizeof(ray_t*);
        } else if (v->type == RAY_TABLE || v->type == RAY_DICT) {
            old_data = 2 * sizeof(ray_t*);
        } else if (RAY_IS_PARTED(v->type) || v->type == RAY_MAPCOMMON) {
            int64_t n_ptrs = v->len;
            if (v->type == RAY_MAPCOMMON) n_ptrs = 2;
            if (n_ptrs < 0) n_ptrs = 0;
            old_data = (size_t)n_ptrs * sizeof(ray_t*);
        } else {
            int8_t t = ray_type(v);
            old_data = (t > 0 && t < RAY_TYPE_COUNT && v->len >= 0) ?
                       (size_t)ray_len(v) * ray_sym_elem_size(t, v->attrs) : 0;
        }
        /* Clamp old_data to actual allocation size */
        if (v->mmod == 0 && v->order >= RAY_ORDER_MIN) {
            size_t alloc_data = ray_block_data_bytes(v);  /* buddy or direct */
            if (old_data > alloc_data) old_data = alloc_data;
        }
        size_t copy_data = old_data < new_data_size ? old_data : new_data_size;
        uint8_t new_mmod = new_v->mmod;
        uint8_t new_order = new_v->order;
        memcpy(new_v, v, 32 + copy_data);
        new_v->mmod = new_mmod;
        new_v->order = new_order;
        if (RAY_UNLIKELY(ray_rc_sync))
            ray_atomic_store(&new_v->rc, 1);
        else
            new_v->rc = 1;
        /* Ownership transfers via memcpy — no retain needed on new_v.
         * Detach nulls old pointers so ray_free won't double-release. */
        if (!(v->attrs & RAY_ATTR_ARENA)) {
            ray_detach_owned_refs(v);
            ray_free(v);
        }
    }
    return new_v;
}

/* --------------------------------------------------------------------------
 * ray_mem_stats
 * -------------------------------------------------------------------------- */

void ray_mem_stats(ray_mem_stats_t* out) {
    if (ray_tl_heap)
        *out = ray_tl_heap->stats;
    else
        memset(out, 0, sizeof(*out));
    int64_t sc = 0, sp = 0;
    ray_sys_get_stat(&sc, &sp);
    out->sys_current = (size_t)sc;
    out->sys_peak    = (size_t)sp;
    int64_t mc = 0, mp = 0;
    ray_sys_get_mapped(&mc, &mp);
    out->sys_mapped      = (size_t)mc;
    out->sys_mapped_peak = (size_t)mp;
    /* Direct large allocations are process-wide (not per-thread) — overlay the
     * global counters so bytes_allocated + direct_bytes is the true live
     * object footprint (buddy blocks + exact direct mmaps). */
    out->direct_bytes = (size_t)atomic_load_explicit(&g_direct_bytes, memory_order_relaxed);
    out->direct_count = (size_t)atomic_load_explicit(&g_direct_count, memory_order_relaxed);
}

/* --------------------------------------------------------------------------
 * Heap lifecycle
 * -------------------------------------------------------------------------- */


/* Heaps whose thread exited: still registered, pools intact, waiting for the
 * next thread to adopt them.  Guarded by the registry lock — pushed and
 * popped once per thread lifetime, never on an allocation path. */
static ray_heap_t* g_heap_idle = NULL;

/* The spill directory, as ray_heap_init resolves it (see there). */
static void heap_swap_dir(char* out, size_t cap) {
    const char* env = getenv("RAY_HEAP_SWAP");
    if (!(env && *env)) env = getenv("TMPDIR");
    const char* sp = (env && *env && strlen(env) < cap - 16) ? env : "/tmp";
    size_t sp_len = strlen(sp);
    memcpy(out, sp, sp_len);
    out[sp_len] = '\0';
    if (sp_len > 0 && out[sp_len - 1] != '/' && sp_len < cap - 1) {
        out[sp_len] = '/';
        out[sp_len + 1] = '\0';
    }
}

/* --------------------------------------------------------------------------
 * Off-heap system blocks
 *
 * ray_sys_alloc serves storage that lives outside the per-thread heaps: the
 * string arenas of the symbol table and the column domains, their hash
 * tables, index-build scratch.  It counts toward the anonymous watermark like
 * pools and direct blocks, so -m (or the default) bounds it too, and a block
 * of RAY_SYS_SPILL_MIN bytes or more that would cross the watermark is mapped
 * over a spill file instead (#712), or taken from RAM when no spill file can be
 * made.  The file is unlinked and its descriptor closed at once — the mapping
 * keeps it — so spilled blocks hold no fds.  Either way the block reads zero
 * (fresh anonymous memory, or a new file created O_EXCL and preallocated),
 * which ray_sys_alloc promises its callers (mem/sys.h): it never comes from
 * the direct-block cache.
 * -------------------------------------------------------------------------- */
#define RAY_SYS_SPILL_MIN ((size_t)1 << 20)

#if RAY_HEAP_FILE_SPILL
static void* heap_sys_map_file(size_t size) {
    static _Atomic uint64_t sys_swap_counter = 0;
    char path[320];
    heap_swap_dir(path, 256);
    size_t plen = strlen(path);
    snprintf(path + plen, sizeof(path) - plen, "raysys_%d_%llu.dat", (int)getpid(),
             (unsigned long long)atomic_fetch_add_explicit(&sys_swap_counter, 1,
                                                           memory_order_relaxed));
    int fd = open(path, O_RDWR | O_CREAT | O_EXCL | O_CLOEXEC, 0600);
    if (fd < 0) return NULL;
    void* p = NULL;
    if (heap_preallocate(fd, 0, (off_t)size) == 0) {
        p = mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
        if (p == MAP_FAILED) p = NULL;
    }
    unlink(path);
    close(fd);
    if (p) ray_sys_track_add((int64_t)size);   /* committed working set, as a spilled pool */
    return p;
}
#endif

void* ray_heap_sys_map(size_t size, bool* spilled) {
    *spilled = false;
    void* p = NULL;
    bool big = size >= RAY_SYS_SPILL_MIN;   /* could spill, so ray_heap_relieve counts it */
#if RAY_HEAP_FILE_SPILL
    bool to_file = big && heap_anon_would_exceed(size) && heap_make_room(size);
    if (!to_file) p = ray_vm_alloc(size);
    if (!p && big) {
        p = heap_sys_map_file(size);
        if (p) { *spilled = true; heap_anon_spilled(); return p; }
    }
    /* No spill file (no writable spill directory, disk full): RAM rather
     * than failing — callers of ray_sys_alloc grow core tables with it. */
    if (!p && to_file) {
        p = ray_vm_alloc(size);
        if (p) heap_anon_spilled();
    }
#else
    p = ray_vm_alloc(size);
#endif
    if (p) heap_anon_commit((int64_t)size, big, big);
    return p;
}

void ray_heap_sys_unmap(void* p, size_t size, bool spilled) {
    if (!p) return;
#if RAY_HEAP_FILE_SPILL
    if (spilled) {
        munmap(p, size);
        ray_sys_track_sub((int64_t)size);
        return;
    }
#else
    (void)spilled;
#endif
    atomic_fetch_sub_explicit(&g_anon_committed, (int64_t)size, memory_order_relaxed);
    ray_vm_free(p, size);
}

void ray_heap_init(void) {
    if (ray_tl_heap) return;

    /* Adopt an abandoned heap before mapping a new one.  Its pools are
     * already faulted in and its slab caches are warm, and — the reason this
     * exists at all — a heap that is never destroyed can be looked up on the
     * lock-free cross-thread free path without racing a munmap. */
    ray_registry_lock();
    ray_heap_t* reuse = g_heap_idle;
    if (reuse) g_heap_idle = reuse->idle_next;
    ray_registry_unlock();
    if (reuse) {
        reuse->idle_next = NULL;
        ray_tl_heap = reuse;
        return;
    }

    size_t heap_sz = (sizeof(ray_heap_t) + 4095) & ~(size_t)4095;
    ray_heap_t* h = (ray_heap_t*)ray_vm_alloc(heap_sz);
    /* A silent return here leaves ray_tl_heap NULL, so every later
     * ray_alloc on this thread fails and the caller sees a baffling
     * downstream error (e.g. a save returning IO) with no hint that the
     * heap never came up.  Make both init failure modes loud. */
    if (!h) {
        fprintf(stderr, "ray_heap_init: ray_vm_alloc(%zu) failed (errno=%d %s) "
                        "— heap not initialized\n",
                heap_sz, errno, strerror(errno));
        return;
    }
    memset(h, 0, heap_sz);

    /* Bitmap-based ID: acquire reusable ID via atomic CAS */
    int id = heap_id_acquire();
    if (id < 0) {
        fprintf(stderr, "ray_heap_init: heap-ID pool exhausted "
                        "— heap not initialized\n");
        ray_vm_free(h, heap_sz);
        return;  /* ID pool exhausted */
    }
    h->id = (uint16_t)id;

    /* Register in global heap registry (lock: serialize the slot write
     * against a concurrent register / unregister). */
    ray_registry_lock();
    ray_heap_registry[h->id % RAY_HEAP_REGISTRY_SIZE] = h;
    ray_registry_unlock();

    /* Initialize circular sentinel freelists */
    for (int i = 0; i < RAY_HEAP_FL_SIZE; i++)
        fl_init(&h->freelist[i]);

    /* Per-order slab push caps from the byte budget (env-overridable).
     * cap[o] = clamp(budget / 2^o, 1, RAY_SLAB_CACHE_SIZE): large orders
     * keep fewer entries, giving a predictable per-thread RSS ceiling. */
    {
        size_t budget = RAY_SLAB_BUDGET;
        const char* b = getenv("RAY_SLAB_BUDGET");
        if (b && *b) {
            char* end = NULL;
            unsigned long long bv = strtoull(b, &end, 10);
            if (end && *end == '\0' && bv > 0) budget = (size_t)bv;
        }
        for (int i = 0; i < RAY_SLAB_ORDERS; i++) {
            size_t bsize = BSIZEOF(RAY_SLAB_MIN + i);
            size_t cap = budget / bsize;
            if (cap < 1) cap = 1;
            if (cap > RAY_SLAB_CACHE_SIZE) cap = RAY_SLAB_CACHE_SIZE;
            h->slab_cap[i] = (uint32_t)cap;
        }
    }

    /* Resolve swap directory for the file-backed pool fallback.  Order:
     * RAY_HEAP_SWAP override → TMPDIR → /tmp.  /tmp is the default rather
     * than the working directory, which may be read-only or shared (the old
     * "./" default littered CWD with large sparse files or failed outright);
     * swap files carry the pid so they never collide.  A trailing slash is
     * always ensured so heap_add_pool can concatenate `<swap_path><filename>`
     * unconditionally.  An empty / over-long value is rejected in favour of
     * the next option. */
    heap_swap_dir(h->swap_path, sizeof(h->swap_path));

    ray_tl_heap = h;
}

/* Forward decl — released before this heap's pools are munmap'd so a
 * stale thread-local trace doesn't survive the destroy and crash a
 * subsequent ray_eval_str → ray_clear_error_trace call when the next
 * heap takes over.  Lives in src/lang/eval.c. */
extern void ray_clear_error_trace(void);

void ray_heap_abandon(void) {
    ray_heap_t* h = ray_tl_heap;
    if (!h) return;

    /* Thread-local refs into this heap must go before another thread owns
     * it, or the next ray_eval_str on the adopting thread would release a
     * trace this thread already handed on. */
    ray_clear_error_trace();
    ray_parted_cache_clear();

    /* Take back whatever was freed to us while we ran, so the heap is
     * handed on with its memory on its freelists rather than on a list the
     * adopter has to notice. */
    heap_drain_foreign(h);
    heap_abandon_spill(h);

    ray_registry_lock();
    h->idle_next = g_heap_idle;
    g_heap_idle = h;
    ray_registry_unlock();

    ray_tl_heap = NULL;
}

/* Real teardown: unregisters the heap and munmaps its pools.
 *
 * ONLY valid when no other thread can still be freeing blocks that belong to
 * this heap — process shutdown after the worker pool is joined, or a
 * single-threaded test.  A thread that merely exits must call
 * ray_heap_abandon instead; see the comment on that function. */
void ray_heap_destroy(void) {
    ray_heap_t* h = ray_tl_heap;
    if (!h) return;

    /* Drop any thread-local refs into THIS heap's memory before we
     * unmap its pools.  Ordering matters: release while the backing
     * memory is still mapped; once the for-loop below munmap's the
     * pools, dereferencing g_error_trace becomes UB. */
    ray_clear_error_trace();
    ray_parted_cache_clear();

    /* FILE sym-domain string atoms used to live on the per-thread buddy
     * heap, requiring a drop here so the process-global cache never handed
     * out a domain backed by this heap's freed memory.  They now live in a
     * per-domain malloc arena (RAY_ATTR_ARENA, thread-independent), so the
     * cache safely outlives any heap teardown — no drop needed. */

    uint16_t saved_id = h->id;

    /* Unregister first so a free that is still to resolve an owner cannot
     * pick this heap up.  The lock serializes the slot write against
     * ray_heap_init / ray_heap_abandon.
     *
     * The cross-heap foreign-list purge that used to live here is gone with
     * the design it belonged to: a block of h's can no longer be sitting on
     * some OTHER heap's foreign list, because a cross-thread free routes the
     * block to its owner.  With it goes the tearing hazard of walking a
     * neighbour's singly-linked list while its owner mutates it. */
    ray_registry_lock();
    ray_heap_registry[h->id % RAY_HEAP_REGISTRY_SIZE] = NULL;
    ray_registry_unlock();

    /* Anything queued for us dies with the pools it lives in — the blocks are
     * free memory inside mappings we are about to drop. */
    atomic_store_explicit(&h->foreign, NULL, memory_order_relaxed);

    /* Skip flush_slabs — all pools are about to be munmap'd. */

    /* Munmap all tracked pools.  File-backed pools also need their fd
     * closed and their tempfile unlinked so the swap directory doesn't
     * accumulate orphans. */
    for (uint32_t i = 0; i < h->pool_count; i++) {
        ray_pool_hdr_t* hdr = (ray_pool_hdr_t*)h->pools[i].base;
        dfd_purge_range((uintptr_t)h->pools[i].base,
                        (uintptr_t)h->pools[i].base + BSIZEOF(h->pools[i].pool_order));
        ray_vm_free(hdr->vm_base, BSIZEOF(h->pools[i].pool_order));
        if (h->pools[i].backed) {
            atomic_fetch_sub_explicit(&g_spill_pools, 1, memory_order_relaxed);
            if (h->pools[i].swap_fd >= 0) close(h->pools[i].swap_fd);
            if (h->pools[i].swap_path) {
                unlink(h->pools[i].swap_path);
                ray_sys_free(h->pools[i].swap_path);
            }
        } else {
            atomic_fetch_sub_explicit(&g_anon_committed,
                (int64_t)BSIZEOF(h->pools[i].pool_order), memory_order_relaxed);
        }
    }

    size_t heap_sz = (sizeof(ray_heap_t) + 4095) & ~(size_t)4095;
    ray_vm_free(h, heap_sz);
    ray_tl_heap = NULL;

    /* Release bitmap ID after all memory is freed */
    heap_id_release(saved_id);
}

/* heap_return_foreign_freelist used to walk this heap's freelists looking for
 * blocks that belong to somebody else and hand them back.  Nothing can put
 * such a block there any more — a cross-thread free goes straight to the
 * owner's list, and a drain only ever coalesces the draining heap's own
 * blocks — so the whole pass is gone. */

#ifdef DEBUG
static void dfd_validate_freelists(void) {
    if (!dfd_enabled()) return;
    for (int hid = 0; hid < RAY_HEAP_REGISTRY_SIZE; hid++) {
        ray_heap_t* gh = ray_heap_registry[hid];
        if (!gh) continue;
        for (int ord = RAY_ORDER_MIN; ord < RAY_HEAP_FL_SIZE; ord++) {
            ray_fl_head_t* fh = &gh->freelist[ord];
            ray_t* blk = fh->fl_next;
            int64_t steps = 0;
            while (blk != (ray_t*)fh) {
                if (!blk) {
                    fprintf(stderr, "\n=== DFD: NULL link in heap %u freelist order %d ===\n",
                            gh->id, ord);
                    dfd_report("freelist NULL link", NULL);
                }
                if (++steps > (int64_t)1e8) {
                    fprintf(stderr, "\n=== DFD: cycle in heap %u freelist order %d ===\n",
                            gh->id, ord);
                    dfd_report("freelist cycle", blk);
                }
                blk = blk->fl_next;
            }
        }
    }
}
#endif

/* Number of maintenance visits a free block must survive on a freelist
 * before pass 5 hands its pages back; a block whose attrs exceed it has
 * already been released and must not be released again.  File scope
 * because the idle decay below shares the marker. */
#define RAY_FREE_AGE_RELEASE   3
/* Below this order a block has no whole page to give back once its first
 * page — which carries the freelist links — is kept, so every release walk
 * starts here. */
#define RAY_RELEASE_MIN_ORDER  13

/* Which pools a reclaim may unmap: oversized ones (gc pass 4), spill ones
 * (the idle decay, an abandoned heap, a relief that acts — releasing a
 * spill block's pages frees nothing, see heap_release_free_pages), or
 * anonymous ones (heap_make_room, ray_heap_relieve). */
#define RAY_RECLAIM_OVERSIZED  1u
#define RAY_RECLAIM_SPILL      2u
#define RAY_RECLAIM_ANON       4u

static bool heap_reclaim_admits(const ray_pool_entry_t* pe, unsigned mode) {
    return ((mode & RAY_RECLAIM_OVERSIZED) && pe->pool_order > RAY_HEAP_POOL_ORDER)
        || ((mode & RAY_RECLAIM_SPILL) && pe->backed)
        || ((mode & RAY_RECLAIM_ANON) && !pe->backed);
}

/* The pools of h that `mode` admits and that may be entirely free, into
 * out[]; returns how many.  h's foreign list must have been drained and its
 * slab cache flushed: every free block is then coalesced as far as it goes,
 * so an empty pool holds exactly the cascade heap_add_pool cut it into — one
 * free block of each order k from RAY_ORDER_MIN to pool_order-1, at offset
 * 2^k.  No free block at order RAY_HEAP_POOL_ORDER-1 or above therefore rules
 * out every pool at once (a clear avail bit is exact; only a set one can be
 * stale), and otherwise one header per cascade offset settles a pool — the
 * offset is always the start of a block, never inside one, since only a
 * block starting at the pool header could span it.  A filter, not the test:
 * a block another thread is freeing into h right now reads rc 0 too, before
 * it is on any list of h's.  heap_find_empty_pools confirms.  A parked
 * pool is not a candidate: it is empty by construction, and its blocks are
 * on no list. */
static uint32_t heap_empty_pool_candidates(ray_heap_t* h, unsigned mode,
                                           uint32_t* out) {
    if (h->pool_count <= 1) return 0;
    if (!(h->avail & (UINT64_MAX << (RAY_HEAP_POOL_ORDER - 1)))) return 0;
    uint32_t n = 0;
    for (uint32_t p = 0; p < h->pool_count; p++) {
        if (h->pools[p].parked || !heap_reclaim_admits(&h->pools[p], mode)) continue;
        uintptr_t pb = (uintptr_t)h->pools[p].base;
        int k = h->pools[p].pool_order - 1;
        for (; k >= RAY_ORDER_MIN; k--) {
            ray_t* b = (ray_t*)(pb + BSIZEOF(k));
            if (ray_atomic_load(&b->rc) != 0 || b->order != k) break;
        }
        if (k < RAY_ORDER_MIN) out[n++] = p;
    }
    return n;
}

/* Set empty[p] for each unparked pool p of h that `mode` admits and whose
 * bytes are all free; returns how many.  empty[] has h->pool_count entries,
 * zeroed by the caller.  The caller vouches that h is quiescent — its own
 * heap, or a parked worker's with ray_parallel_flag clear — and has drained
 * h's foreign list and flushed its slab cache; a block in either is not on a
 * freelist, so its pool is not empty as far as this goes.
 *
 * Emptiness is "every byte of the pool but its header is on h's freelists".
 * A block of h's that is live, or queued anywhere, is by construction not
 * there, so it shows up as a shortfall — no cross-heap scan is needed, since
 * a free always routes a block to its owner.  One walk of h's freelists
 * sums the free bytes of every candidate at once: walking them once per
 * pool made a boundary O(pools x free blocks), a quarter of a second on a
 * fragmented heap of a few hundred MB.  The walk only runs when the
 * candidate filter above found something, and an empty pool's blocks are
 * the cascade, so they come off the lists by offset rather than by a second
 * walk (heap_pool_unlist). */
static uint32_t heap_find_empty_pools(ray_heap_t* h, unsigned mode, uint8_t* empty) {
    uint32_t cand[RAY_MAX_POOLS];
    uint32_t n = heap_empty_pool_candidates(h, mode, cand);
    if (n == 0) return 0;

    /* Candidates by base, so a block finds its pool by binary search. */
    uint8_t top = 0;
    for (uint32_t i = 0; i < n; i++) {
        uint32_t c = cand[i];
        uint32_t j = i;
        for (; j > 0 && (uintptr_t)h->pools[cand[j - 1]].base
                        > (uintptr_t)h->pools[c].base; j--)
            cand[j] = cand[j - 1];
        cand[j] = c;
        if (h->pools[c].pool_order > top) top = h->pools[c].pool_order;
    }
    size_t free_bytes[RAY_MAX_POOLS];
    memset(free_bytes, 0, n * sizeof(free_bytes[0]));
    for (int ord = RAY_ORDER_MIN; ord < top; ord++) {
        if (!(h->avail & (1ULL << ord))) continue;
        ray_fl_head_t* fh = &h->freelist[ord];
        for (ray_t* blk = fh->fl_next; blk != (ray_t*)fh; blk = blk->fl_next) {
            uintptr_t a = (uintptr_t)blk;
            uint32_t lo = 0, hi = n;   /* first candidate based above a */
            while (lo < hi) {
                uint32_t mid = (lo + hi) / 2;
                if ((uintptr_t)h->pools[cand[mid]].base <= a) lo = mid + 1;
                else hi = mid;
            }
            if (lo == 0) continue;
            const ray_pool_entry_t* pe = &h->pools[cand[lo - 1]];
            if (a < (uintptr_t)pe->base + BSIZEOF(pe->pool_order))
                free_bytes[lo - 1] += BSIZEOF(ord);
        }
    }

    uint32_t found = 0;
    for (uint32_t i = 0; i < n; i++) {
        uint8_t po = h->pools[cand[i]].pool_order;
        if (free_bytes[i] == BSIZEOF(po) - BSIZEOF(RAY_ORDER_MIN)) {
            empty[cand[i]] = 1;
            found++;
        }
    }
    return found;
}

/* Take empty pool p's blocks — the cascade heap_add_pool cut it into — off
 * h's freelists. */
static void heap_pool_unlist(ray_heap_t* h, uint32_t p) {
    uintptr_t pb = (uintptr_t)h->pools[p].base;
    for (uint8_t k = RAY_ORDER_MIN; k < h->pools[p].pool_order; k++) {
        fl_remove((ray_t*)(pb + BSIZEOF(k)));
        if (fl_empty(&h->freelist[k]))
            h->avail &= ~(1ULL << k);
    }
}

/* Unmap pool p of h, empty and off the freelists — unlisted, or parked — and
 * drop its entry, the last one moving into p; returns the anonymous bytes
 * unmapped. */
static int64_t heap_pool_unmap(ray_heap_t* h, uint32_t p) {
    ray_pool_entry_t* pe = &h->pools[p];
    uint8_t po = pe->pool_order;
    uintptr_t pb = (uintptr_t)pe->base;
    int64_t anon = 0;
    if (pe->parked) h->parked--;

    dfd_purge_range(pb, pb + BSIZEOF(po));
    ray_vm_free(((ray_pool_hdr_t*)pb)->vm_base, BSIZEOF(po));
    /* File-backed pools also need their fd closed and tempfile
     * unlinked, mirroring the heap_destroy path.  Unmapping drops the
     * unlinked file's pages without writing them back. */
    if (pe->backed) {
        atomic_fetch_sub_explicit(&g_spill_pools, 1, memory_order_relaxed);
        if (pe->swap_fd >= 0) close(pe->swap_fd);
        if (pe->swap_path) {
            unlink(pe->swap_path);
            ray_sys_free(pe->swap_path);
        }
    } else {
        atomic_fetch_sub_explicit(&g_anon_committed,
            (int64_t)BSIZEOF(po), memory_order_relaxed);
        anon = (int64_t)BSIZEOF(po);
    }
    h->pools[p] = h->pools[--h->pool_count];
    return anon;
}

/* Unmap the pools of h that `mode` admits and whose bytes are all free,
 * parked ones included, keeping h's last pool; returns the anonymous bytes
 * unmapped.  Same conditions as heap_find_empty_pools. */
static int64_t heap_reclaim_empty_pools(ray_heap_t* h, unsigned mode) {
    uint8_t empty[RAY_MAX_POOLS];
    memset(empty, 0, h->pool_count);
    uint32_t n = heap_find_empty_pools(h, mode, empty);
    for (uint32_t p = 0; h->parked && p < h->pool_count; p++)
        if (h->pools[p].parked && heap_reclaim_admits(&h->pools[p], mode)) {
            empty[p] = 1;
            n++;
        }
    if (n == 0) return 0;

    /* Highest index first: unmapping pool p moves the last entry into p, and
     * every entry above p has been dealt with by then. */
    int64_t anon = 0;
    for (uint32_t p = h->pool_count; p-- > 0; ) {
        if (!empty[p]) continue;
        if (h->pool_count <= 1) break;   /* a heap keeps its last pool */
        if (!h->pools[p].parked) heap_pool_unlist(h, p);
        anon += heap_pool_unmap(h, p);
    }
    return anon;
}

/* Parking.  An empty pool kept for the next statement does not stay empty
 * through it: every pool holds one free block of each small order (the
 * cascade), so the first small allocations — the statement's own parsing
 * and environment — take a block from each in turn, and by the time the
 * statement wants a large block that the empty pools could make room for,
 * none of them is empty.  A parked pool keeps its mapping and its pages but
 * its blocks are off the freelists: nothing is cut from it until the heap
 * has no free block left of the order it needs (heap_unpark, on the
 * allocation slow path before a new pool is mapped), and heap_make_room can
 * still unmap it for a large block, an anonymous one.  A spill one has its
 * pages discarded at the boundary that parks it (heap_spill_pools_discard).
 * The heap's last unparked pool stays on the lists.  Parks the empty pools
 * `mode` admits; the pool header keeps when (`now`, ms) the pool was parked
 * (heap_spill_pools_age). */
static void heap_park_empty_pools(ray_heap_t* h, unsigned mode, uint32_t now) {
    uint8_t empty[RAY_MAX_POOLS];
    memset(empty, 0, h->pool_count);
    if (heap_find_empty_pools(h, mode, empty) == 0)
        return;
    for (uint32_t p = h->pool_count; p-- > 0; ) {
        if (!empty[p]) continue;
        if (h->pool_count - h->parked <= 1) break;
        heap_pool_unlist(h, p);
        h->pools[p].parked = 1;
        ((ray_pool_hdr_t*)h->pools[p].base)->parked_ms = now;
        h->parked++;
    }
}

/* Put one parked pool's blocks back on the freelists — a spill one if
 * `spill`, else an anonymous one; false if h has none. */
static bool heap_unpark(ray_heap_t* h, bool spill) {
    for (uint32_t p = h->pool_count; p-- > 0; ) {
        ray_pool_entry_t* pe = &h->pools[p];
        if (!pe->parked || pe->backed != spill) continue;
        uintptr_t pb = (uintptr_t)pe->base;
        for (uint8_t o = pe->pool_order; o > RAY_ORDER_MIN; o--)
            heap_insert_block(h, (ray_t*)(pb + BSIZEOF(o - 1)), (uint8_t)(o - 1));
        pe->parked = 0;
        pe->discarded = 0;   /* its pages are in use again */
        h->parked--;
        return true;
    }
    return false;
}

static void heap_unpark_all(ray_heap_t* h) {
    while (h->parked && (heap_unpark(h, false) || heap_unpark(h, true))) {}
}

/* An abandoned heap is visited by no boundary (heap_each_quiescent), so its
 * empty spill pools would never age (heap_spill_pools_age): they go as its
 * thread leaves it, which still owns it then. */
static void heap_abandon_spill(ray_heap_t* h) {
    for (uint32_t p = 0; p < h->pool_count; p++)
        if (h->pools[p].backed) {
            heap_flush_slabs(h);
            (void)heap_reclaim_empty_pools(h, RAY_RECLAIM_SPILL);
            return;
        }
}

/* The heaps a maintenance point may work on from this thread: its own, and
 * the parked workers' of the live pool.  Any other registered heap belongs
 * to a thread that may be allocating right now — an embedding's own threads,
 * a host-driven poll loop — or to none yet, and then may be adopted at any
 * moment; only its owner may touch its freelists.  Caller ensures
 * ray_parallel_flag is clear. */
static void heap_each_quiescent(void (*fn)(ray_heap_t* h, void* ctx), void* ctx) {
    if (ray_tl_heap) fn(ray_tl_heap, ctx);
    ray_pool_each_worker_heap(fn, ctx);
}

static void heap_reclaim_oversized(ray_heap_t* h, void* ctx) {
    (void)ctx;
    (void)heap_reclaim_empty_pools(h, RAY_RECLAIM_OVERSIZED);
}

void ray_heap_gc(void) {
    ray_parted_cache_prune();
    ray_heap_t* h = ray_tl_heap;
    if (!h) return;
#ifdef DEBUG
    dfd_validate_freelists();
#endif

    bool safe = (atomic_load_explicit(&ray_parallel_flag, memory_order_relaxed) == 0);

    /* Pass 1: take back whatever other threads freed to us, and fold the
     * slab caches back into the freelists.  The drain needs no idle-worker
     * gate: the list holds only our own blocks. */
    heap_drain_foreign(h);
    heap_flush_slabs(h);

    if (safe) {
        /* Pass 2 (return foreign blocks absorbed onto our freelists) is gone
         * — nothing absorbs a foreign block locally any more.
         *
         * Pass 3: Skip worker heaps.  Not for safety — ray_parallel_end
         * drains `pending` with acquire before clearing the flag, and a
         * worker allocates nothing between its last pending-- and sem_wait,
         * which is what lets the decay sweep below walk those very heaps.
         * It is a policy choice: folding a worker's slab cache back costs
         * it a freelist pop and a split on its next allocation of that
         * order, and the collection runs far more often than the caches
         * are worth reclaiming. */

        /* Pass 4: Reclaim OVERSIZED empty pools.
         * Standard pools (pool_order == RAY_HEAP_POOL_ORDER) are never
         * munmapped — physical pages released via madvise (phase 5)
         * re-fault cheaply on next query.  (On THP-backed pools, phase 5
         * only releases the 2MB-aligned interior of a free block, so
         * sub-2MB free blocks keep their pages resident — the deliberate
         * no-shatter tradeoff; disable per-deployment with RAY_HEAP_HUGEPAGE=0.)
         * Only oversized pools (pool_order > RAY_HEAP_POOL_ORDER) are
         * candidates — these are one-off large allocations.
         *
         * Emptiness is computed from the owning heap's freelists (see
         * heap_reclaim_empty_pools).  This avoids atomic live_count
         * operations on the alloc/free hot path.  Only our own heap and the
         * parked workers' are visited (heap_each_quiescent); a worker's
         * slab cache is not flushed here (pass 3), so a pool with a block
         * still in it waits for a later pass. */
        heap_each_quiescent(heap_reclaim_oversized, NULL);

        /* Pass 5: Release physical pages from free blocks in every
         * idle heap.  Pass 2 may have returned blocks to worker-owned
         * freelists; releasing only the caller heap leaves those worker
         * pages resident across large query repetitions.
         *
         * Use each heap's avail bitmap (set on insert, cleared on
         * remove) to skip the entire walk when no order >= 13 has any
         * free block.  Tiny-query workloads — where the per-statement
         * GC fires before any large allocation has been freed —
         * complete pass 5 without entering the body. */
        /* Aging: pages are released only after a block SURVIVES several
         * pass-5 VISITS on a freelist.  This routine runs per statement
         * and per parallel_end, so the per-query temporaries of a
         * repeated query used to be MADV_DONTNEEDed between every two
         * executions and refaulted from scratch — kernel page-fault
         * frames dominated repeated-query profiles while the freed block
         * was reused within milliseconds.  A block re-allocated before
         * reaching RAY_FREE_AGE_RELEASE keeps its pages; genuinely idle
         * memory still returns to the OS after a few maintenance points.
         * The released marker also stops re-madvising the same idle
         * block on every subsequent pass.
         *
         * INCREMENTAL CURSOR — bounded work at every maintenance point
         * (the design contract: heap maintenance must never turn into an
         * unpredictable pause).  The walk resumes at the (heap, order)
         * slot where the previous pass stopped and is capped by BOTH a
         * block-visit budget and a release budget: each madvise batch
         * costs cross-core TLB-shootdown IPIs, so releases are the
         * expensive unit, not visits.  The cursor advances BEFORE a slot
         * is processed, so a list interrupted mid-walk simply waits one
         * full cycle instead of being double-aged; the cursor never
         * stores a block pointer — blocks may be reallocated, split or
         * coalesced between passes, only (heap, order) is stable. */
        #define RAY_P5_VISIT_BUDGET    2048
        #define RAY_P5_RELEASE_BUDGET  64
        static int s_p5_hid = 0;
        static int s_p5_ord = RAY_RELEASE_MIN_ORDER;
        uint64_t large_orders_mask = ~((1ULL << 13) - 1);
        int64_t p5_visited = 0, p5_released = 0;
        int p5_slots = RAY_HEAP_REGISTRY_SIZE * (RAY_HEAP_FL_SIZE - 13);
        for (int step = 0; step < p5_slots; step++) {
            if (p5_visited >= RAY_P5_VISIT_BUDGET ||
                p5_released >= RAY_P5_RELEASE_BUDGET) break;
            int hid = s_p5_hid;
            int ord = s_p5_ord;
            if (++s_p5_ord >= RAY_HEAP_FL_SIZE) {
                s_p5_ord = RAY_RELEASE_MIN_ORDER;
                s_p5_hid = (s_p5_hid + 1) % RAY_HEAP_REGISTRY_SIZE;
            }
            ray_heap_t* gh = ray_heap_registry[hid];
            if (!gh) continue;
            if (!((gh->avail & large_orders_mask) & (1ULL << ord))) continue;
            ray_fl_head_t* head = &gh->freelist[ord];
            for (ray_t* blk = head->fl_next; blk != (ray_t*)head;
                 blk = blk->fl_next) {
                if (p5_visited++ >= RAY_P5_VISIT_BUDGET) break;
                if (blk->attrs < RAY_FREE_AGE_RELEASE) {
                    blk->attrs++;
                } else if (blk->attrs == RAY_FREE_AGE_RELEASE) {
                    size_t bsize = BSIZEOF(ord);
                    int rpidx = heap_find_pool(gh, blk);
                    bool hp = (rpidx >= 0) ? (gh->pools[rpidx].hugepage != 0)
                                           : false;
                    ray_vm_release_block(blk, bsize, hp);
                    blk->attrs = RAY_FREE_AGE_RELEASE + 1;  /* released */
                    if (++p5_released >= RAY_P5_RELEASE_BUDGET) break;
                }
            }
        }
    }

}

/* Hand back the pages of every free block of `h` at or above the release
 * floor that has not already been released, and report how many.  The three
 * callers differ only in which heaps they visit and what they do first;
 * the walk itself is this.  A block of a spill pool is only marked: its
 * pages are the spill file's page cache, which madvise unmaps from us
 * without freeing — such memory goes back when its pool is unmapped. */
static int64_t heap_release_free_pages(ray_heap_t* h) {
    int64_t released = 0;
    for (int ord = RAY_RELEASE_MIN_ORDER; ord < RAY_HEAP_FL_SIZE; ord++) {
        if (!(h->avail & (1ULL << ord))) continue;
        ray_fl_head_t* head = &h->freelist[ord];
        for (ray_t* blk = head->fl_next; blk != (ray_t*)head;
             blk = blk->fl_next) {
            if (blk->attrs > RAY_FREE_AGE_RELEASE) continue;  /* released */
            int pidx = heap_find_pool(h, blk);
            blk->attrs = RAY_FREE_AGE_RELEASE + 1;
            if (pidx >= 0 && h->pools[pidx].backed) continue;
            bool hp = (pidx >= 0) ? (h->pools[pidx].hugepage != 0) : false;
            ray_vm_release_block(blk, BSIZEOF(ord), hp);
            released++;
        }
    }
    return released;
}

/* Whether a parked spill pool's pages are discarded (heap_spill_discard) —
 * or the pool unmapped, as where the platform cannot discard them.  Policy:
 * the setter (ray_heap_set_spill_discard) is for tests. */
static _Atomic(bool) g_spill_discard = true;
static _Atomic(int64_t) g_spill_discard_calls = 0;   /* ray_heap_spill_discard_count */

bool ray_heap_set_spill_discard(bool on) {
    return atomic_exchange_explicit(&g_spill_discard, on, memory_order_relaxed);
}

/* Discard the pages of parked spill pool pe — dead data — and keep its file
 * and its mapping for the next statement that spills: from the pool's
 * second page to the end of its last page in RAM (ray_vm_resident_end: as
 * far as its use reached), the range is punched out of the file, each dirty
 * page with it, unwritten, and its memory free; then reserved again
 * (heap_preallocate), so a later store still finds its disk block.  The
 * first page — the pool header and the free headers of the cascade's
 * smallest blocks — stays as it is; the other cascade blocks' headers are
 * written anew when the pool is taken back (heap_unpark), and nothing reads
 * a parked pool's blocks before that.  Returns 1 if pages were punched, 0
 * if none past the first page were in RAM, and -1 where it cannot be done —
 * no way to punch a mapped range on this platform (macOS: F_PUNCHHOLE does
 * not say what it does to pages mapped shared), or the file system or the
 * disk refused: the pool is to be unmapped then.  With pe NULL, 0 where
 * pages would be discarded, -1 where not (ray_heap_spill_discards). */
static int heap_spill_discard(const ray_pool_entry_t* pe) {
#if RAY_HEAP_FILE_SPILL && defined(__linux__) && defined(FALLOC_FL_PUNCH_HOLE)
    if (!atomic_load_explicit(&g_spill_discard, memory_order_relaxed)) return -1;
    if (!pe) return 0;
    long ps = sysconf(_SC_PAGESIZE);
    size_t pg = ps > 0 ? (size_t)ps : 4096;
    size_t end = ray_vm_resident_end(pe->base, BSIZEOF(pe->pool_order));
    if (end <= pg) return 0;
    off_t off = (off_t)pg, len = (off_t)(end - pg);
    if (fallocate(pe->swap_fd, FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE, off, len) != 0
        || heap_preallocate(pe->swap_fd, off, len) != 0)
        return -1;
    return 1;
#else
    (void)pe;
    return -1;
#endif
}

int64_t ray_heap_spill_discard_count(void) {
    return atomic_load_explicit(&g_spill_discard_calls, memory_order_relaxed);
}

bool ray_heap_spill_discards(void) {
    return heap_spill_discard(NULL) == 0;
}

/* Discard the pages of h's parked spill pools not discarded since they were
 * last in use — those parked at this boundary — or unmap a pool whose pages
 * cannot be (the boundary that parked it left h another pool, unparked).
 * Returns whether pages were dropped: the boundary then flushes the
 * cgroup's statistics (ray_sys_mem_flush). */
static bool heap_spill_pools_discard(ray_heap_t* h) {
    bool dropped = false;
    for (uint32_t p = h->pool_count; h->parked && p-- > 0; ) {
        ray_pool_entry_t* pe = &h->pools[p];
        if (!pe->parked || !pe->backed || pe->discarded) continue;
        atomic_fetch_add_explicit(&g_spill_discard_calls, 1, memory_order_relaxed);
        int got = heap_spill_discard(pe);
        if (got < 0) {
            (void)heap_pool_unmap(h, p);
            dropped = true;
            continue;
        }
        pe->discarded = 1;
        dropped |= got > 0;
    }
    return dropped;
}

/* A parked spill pool that no statement has taken back for the idle decay's
 * threshold is unmapped, its file closed — but not h's last pool: the decay
 * pool by pool, for a process never idle that long — a server — which would
 * otherwise keep the files, their disk blocks reserved, for good.  A negative
 * threshold — the decay disabled — is never reached, as unsigned.  `now` as
 * for heap_park_empty_pools. */
static void heap_spill_pools_age(ray_heap_t* h, uint32_t now) {
    int64_t thr = heap_decay_threshold_ms();
    for (uint32_t p = h->pool_count; h->parked && h->pool_count > 1 && p-- > 0; ) {
        const ray_pool_entry_t* pe = &h->pools[p];
        if (pe->parked && pe->backed
            && now - ((const ray_pool_hdr_t*)pe->base)->parked_ms >= (uint64_t)thr)
            (void)heap_pool_unmap(h, p);
    }
}

/* A direct or sys block of `bytes` is about to cross the watermark.  Before
 * it goes to a spill file, give back what the calling thread holds free: the
 * direct-block cache, then its own heap's empty anonymous pools — parked
 * ones above all (heap_park_empty_pools) — which no block of that size is
 * ever cut from.  A statement that freed its pool temporaries and goes on to
 * large blocks would otherwise spill those while the pools sit empty and
 * counted, and no rule at the boundary can tell that the next statement
 * wants large blocks rather than its pools back.  Only the caller's heap:
 * any other, a worker's included, may be allocating right now.  Returns
 * whether `bytes` still crosses the watermark. */
static bool heap_make_room(size_t bytes) {
    direct_cache_drain();
    if (!heap_anon_would_exceed(bytes)) return false;
#if RAY_HEAP_FILE_SPILL
    ray_heap_t* h = ray_tl_heap;
    if (!h) return true;
    heap_drain_foreign(h);
    heap_flush_slabs(h);
    int64_t got = heap_reclaim_empty_pools(h, RAY_RECLAIM_ANON);
    return got == 0 || heap_anon_would_exceed(bytes);
#else
    return true;   /* past the watermark the block comes from RAM all the same */
#endif
}

void ray_heap_release_pages(void) {
    /* Explicit release: callers asking for pages back get them NOW —
     * no aging — but still skip blocks already released. */
    ray_heap_t* h = ray_tl_heap;
    if (!h) return;
    heap_unpark_all(h);
    (void)heap_release_free_pages(h);
}

/* --------------------------------------------------------------------------
 * Idle decay
 *
 * A free block keeps its physical pages so the next allocation of that size
 * reuses them without faulting; that is why pass 5 ages a block before
 * releasing it, and why a repeated query never pays to refault its own
 * temporaries.  The price of the policy is that a process which ran one
 * heavy query holds its peak footprint for the rest of its life even while
 * it does nothing.  The memory is free and reusable — it is simply not
 * available to anything else on the machine.
 *
 * The decay closes that gap with no background thread, no queue and no task
 * object.  Work stamps a coarse timestamp when it STARTS, and the two points
 * that already exist at the end of a unit of work — the statement boundary
 * and the event loop between wakeups — compare that stamp against a
 * threshold.  If the process has been quiet longer than the threshold, one
 * un-aged sweep releases the pages of every free block and unmaps every
 * empty spill pool (madvise frees nothing of a file mapping; see
 * heap_release_free_pages); otherwise the check is a relaxed load and a
 * subtraction.
 *
 * The threshold is the whole point: it is what separates "the workload has
 * stopped" from "we are between two iterations of a hot loop".  Releasing in
 * the second case drops exactly the pages the next iteration refaults, which
 * is why pass 5 ages blocks in the first place.
 *
 * The stamp is taken at the END of a unit of work, and the boundary check
 * reads the clock BEFORE the new stamp.  What it measures is therefore the
 * gap between one unit finishing and the next — the time the process
 * actually sat idle.  Stamping at the start would measure the unit's own
 * duration instead, so a loop of units longer than the threshold would
 * sweep after every one of them, discarding the working set the next is
 * about to fault back in.
 *
 * The sweep works on the pool workers' heaps as well as the caller's, and it
 * drains each one's foreign list and slab cache before walking its
 * freelists.  That is not incidental: in a server the main thread frees the
 * results of a parallel query, so the blocks land on the owning WORKER's
 * foreign list, and a worker only drains its own list when it next needs
 * memory — which a warm heap never does.  Freelist-only, the sweep sees a
 * small fraction of the free bytes; with the drain it sees nearly all of
 * them.  Draining also coalesces, so the pages come back in large runs
 * instead of scattered blocks below the order-13 floor.  No other heap is
 * visited: the registry also holds the heaps of other live threads, whose
 * freelists only their owners may touch (heap_each_quiescent).
 *
 * That is sound only because the sweep runs when ray_parallel_flag is 0.
 * The dispatcher clears the flag only after every worker's `pending--` is
 * visible, and a worker between `pending--` and sem_wait claims tasks and
 * waits — it neither allocates nor frees — so no other thread is touching
 * a worker heap's freelists or slab stack.  A concurrent cross-thread free
 * can still PUSH onto a foreign list at any moment, and that is safe on its
 * own terms: the drain takes the whole list with one atomic exchange, and
 * whatever arrives afterwards is simply left for the next sweep.
 *
 * Unlike ray_heap_gc's pass 5 this walk has no visit or release budget.
 * The budget exists there because gc runs inside a workload and must never
 * become an unpredictable pause; the decay by construction only runs when
 * there is no workload to pause.
 *
 * The armed flag makes this once per idle period.  After a sweep there is
 * nothing further to give back until new work dirties memory again, so a
 * long-idle process stops checking altogether and its event loop returns to
 * blocking indefinitely instead of waking on a timer.
 * -------------------------------------------------------------------------- */

#define RAY_HEAP_DECAY_MS_DEFAULT   10000
#define RAY_HEAP_DECAY_MS_MAX       ((int64_t)7 * 24 * 3600 * 1000)

static _Atomic int64_t g_heap_activity_ms;
static _Atomic bool    g_heap_decay_armed;

/* Read per check rather than cached at first use.  This is policy, not a
 * hot-path constant — it is consulted once per statement and once per event
 * loop wakeup — and caching it would make the value depend on which code
 * path happened to run first in the process. */
/* The policy, not a knob: there is no environment override.  Negative
 * disables the decay; the clamp keeps `activity + threshold` from wrapping,
 * and a week is already "never".  The setter exists so a test can drive the
 * policy — an operator-facing override, if one is ever wanted, belongs on an
 * existing surface rather than here. */
static int64_t g_heap_decay_ms = RAY_HEAP_DECAY_MS_DEFAULT;

void ray_heap_set_decay_ms(int64_t ms) {
    if (ms > RAY_HEAP_DECAY_MS_MAX) ms = RAY_HEAP_DECAY_MS_MAX;
    g_heap_decay_ms = (ms < 0) ? -1 : ms;
}

static int64_t heap_decay_threshold_ms(void) {
    return g_heap_decay_ms;
}

void ray_heap_note_activity(void) {
    atomic_store_explicit(&g_heap_activity_ms, ray_time_now_ms(),
                          memory_order_relaxed);
    atomic_store_explicit(&g_heap_decay_armed, true, memory_order_relaxed);
}

int64_t ray_heap_decay_due_ms(void) {
    if (!atomic_load_explicit(&g_heap_decay_armed, memory_order_relaxed))
        return -1;
    int64_t threshold = heap_decay_threshold_ms();
    if (threshold < 0) return -1;   /* decay disabled */
    int64_t due = atomic_load_explicit(&g_heap_activity_ms,
                                       memory_order_relaxed)
                + threshold - ray_time_now_ms();
    return due > 0 ? due : 0;
}

static void heap_decay_sweep(ray_heap_t* h, void* ctx) {
    /* Blocks freed to h by another thread, and h's own slab cache, are not
     * on a freelist and would be invisible to the walk below — in a server
     * that is where most of the free bytes are.  Nor are the blocks of a
     * parked pool. */
    heap_drain_foreign(h);
    heap_flush_slabs(h);
    heap_unpark_all(h);
    (void)heap_reclaim_empty_pools(h, RAY_RECLAIM_SPILL);
    *(int64_t*)ctx += heap_release_free_pages(h);
}

int64_t ray_heap_decay(void) {
    if (ray_heap_decay_due_ms() != 0) return -1;
    if (atomic_load_explicit(&ray_parallel_flag, memory_order_relaxed) != 0)
        return -1;

    /* Disarm before sweeping: what follows is everything this idle period
     * can return, so repeating it before work resumes finds nothing. */
    atomic_store_explicit(&g_heap_decay_armed, false, memory_order_relaxed);

    int64_t released = 0;
    heap_each_quiescent(heap_decay_sweep, &released);
    return released;
}

/* What a relief would hand back, as far as the candidate filter can tell:
 * the anonymous bytes of the heaps that did not spill a pool (room; of other
 * threads' heaps, room_other). */
typedef struct { int64_t room, room_other; } relieve_measure_t;

static void heap_relieve_measure(ray_heap_t* h, void* ctx) {
    relieve_measure_t* m = ctx;
    heap_drain_foreign(h);
    heap_flush_slabs(h);
    if (h->spilled) return;
    uint32_t cand[RAY_MAX_POOLS];
    uint32_t n = heap_empty_pool_candidates(h, RAY_RECLAIM_ANON, cand);
    uint8_t empty[RAY_MAX_POOLS];
    memset(empty, 0, h->pool_count);
    for (uint32_t i = 0; i < n; i++) empty[cand[i]] = 1;
    int64_t room = 0;
    uint32_t gone = 0;   /* pools the reclaim would take */
    for (uint32_t p = 0; p < h->pool_count; p++) {
        const ray_pool_entry_t* pe = &h->pools[p];
        if ((empty[p] || pe->parked) && !pe->backed) {
            room += (int64_t)BSIZEOF(pe->pool_order);
            gone++;
        }
    }
    /* With every pool empty, the reclaim keeps the first. */
    if (gone && gone == h->pool_count)
        room -= (int64_t)BSIZEOF(h->pools[0].pool_order);
    m->room += room;
    if (h != ray_tl_heap) m->room_other += room;
}

typedef struct { int64_t released; uint32_t now; bool near, act, dropped; } relieve_visit_t;

/* If the relief acts (act), a heap gives back its empty spill pools, and a
 * heap that did not spill a pool its empty anonymous ones too and the pages
 * of its other free blocks — one that did keeps those: it ran out of its
 * own.  What stays empty is parked, a parked spill pool's pages are
 * discarded, and a spill pool long unused is unmapped.  The heap has been
 * looked at: what it spilled is known now.  Far from the watermark (!near)
 * the visit is only for the spill pools: it parks the empty ones it finds on
 * the freelists as they are, without a drain or a slab flush — a pool with
 * a block still in either waits for a later visit, and the next statement
 * keeps its slab cache. */
static void heap_relieve_visit(ray_heap_t* h, void* ctx) {
    relieve_visit_t* v = ctx;
    if (v->near) {
        heap_drain_foreign(h);
        heap_flush_slabs(h);
    }
    int64_t spills = atomic_load_explicit(&g_spill_pools, memory_order_relaxed);
    heap_spill_pools_age(h, v->now);
    if (v->act) {
        (void)heap_reclaim_empty_pools(h, h->spilled ? RAY_RECLAIM_SPILL
                                                     : RAY_RECLAIM_SPILL | RAY_RECLAIM_ANON);
        if (!h->spilled) v->released += heap_release_free_pages(h);
    }
    heap_park_empty_pools(h, v->near ? RAY_RECLAIM_ANON | RAY_RECLAIM_SPILL
                                     : RAY_RECLAIM_SPILL, v->now);
    v->dropped |= heap_spill_pools_discard(h);
    v->dropped |= atomic_load_explicit(&g_spill_pools, memory_order_relaxed) != spills;
    h->spilled = 0;
}

/* What the boundaries since the last relief have shown (ray_heap_relieve). */
enum {
    RELIEF_NONE,      /* nothing against relieving */
    RELIEF_JUST,      /* the last relief has not been put to the test yet */
    RELIEF_FUTILE,    /* the statement after it spilled all the same */
};
static _Atomic(int)     g_relieve_state  = RELIEF_NONE;
static _Atomic(int64_t) g_relieve_wm     = 0;   /* the watermark of the last relief */
static _Atomic(int64_t) g_relieve_maps   = 0;   /* g_anon_maps at the last boundary */
static _Atomic(int64_t) g_relieve_large  = 0;   /* g_anon_large at the last boundary */
static _Atomic(int64_t) g_relieve_spills = 0;   /* g_anon_spills at the last boundary */
static _Atomic(bool)    g_relieve_other  = false;   /* the last relief took others' pools */

/* Statement boundary.  A free block keeps its pages and a standard pool
 * stays mapped, so the next query reuses them without faulting (see the idle
 * decay above).  Where memory is short that can turn against the next query:
 * a kept pool still counts toward the watermark, so memory the next query
 * wants in another form — a large block, or a pool of another heap — goes to
 * spill files instead.  A repeated heavy query then ran far slower than its
 * first run.  Handing back at every boundary near the watermark was no
 * answer either: the next statement maps and faults it straight back
 * (#750).  So the boundary acts on evidence, not on the footprint alone.
 *
 * The evidence is a spill: a mapping of the statement that just ended went
 * past RAM (g_anon_spills).  Then a heap that spilled a pool ran out of its
 * own, so its empty pools are its working set and stay, parked; the empty
 * anonymous pools of the heaps that did not are memory the spilling one
 * could not reach.  Those are unmapped — the relief — when that hands back
 * at least a pool and brings the next pool under the watermark; the
 * direct-block cache goes with them, and the pages of those heaps' other
 * free blocks.
 *
 * A relief can be futile: when the next statement that maps anything spills
 * again although it took the room for pools only, the room went back to the
 * same kind of memory it came from — more relief only refaults it.  No relief
 * then until a statement maps without spilling, or the watermark moves.  A
 * statement that maps nothing says nothing either way.  The other way round,
 * when the statement after a relief that took other threads' pools does not
 * spill but puts large blocks in that room and runs up to the watermark, it
 * needed the relief, and a repeat of it will again (a heavy query run again
 * and again): its boundary relieves as a spill's would.
 *
 * Without the evidence, near the watermark (the next pool would not fit),
 * the empty pools are parked (heap_park_empty_pools): the next statement
 * takes them back without a fault, and its large blocks can still have them
 * (heap_make_room).
 *
 * An empty spill pool is parked like the rest, its file and its mapping
 * kept for the next statement that spills, but its pages — dead data — are
 * discarded at the boundary that parks it (heap_spill_pools_discard): a
 * dirty one is dropped unwritten, which the kernel would otherwise write
 * back for nothing, and its memory is free; the next statement that spills
 * takes the pool back, its pages zero-filled anew.  Where pages cannot be
 * discarded the pool is unmapped instead.
 * When the relief acts, the empty spill pools go with the anonymous ones,
 * from every heap the relief visits, the spilling one's too, and a parked
 * spill pool unused for the idle decay's threshold goes too
 * (heap_spill_pools_age), so a server that is never idle does not keep its
 * file for good.  After spill pages are dropped, the limiting cgroup's
 * statistics are flushed (ray_sys_mem_flush): the kernel updates them
 * lazily and holds writers to them, and the dirty pages just dropped would
 * otherwise throttle the next writes as if they were still there (measured:
 * a 64 MB spill right after dropping 137 MB of dirty pages, 1.2 s instead of
 * 0.1 s, 900 MB cgroup).  The boundary
 * visits the heaps while any spill pool is mapped — far from the watermark
 * at most every RAY_FAR_VISIT_MS, and only for the spill pools.
 *
 * Every check is a relaxed load or a counter comparison; the walks — a
 * drain and a slab flush per heap, a few header reads per possibly-empty
 * pool, one freelist walk where a pool may be empty — run only on a spill or
 * near the watermark (far from it, while a spill pool is mapped: the header
 * reads and the walk).  A spill pool parked at this boundary costs a
 * mincore, a hole punched and the range reserved again; one parked before,
 * nothing.
 *
 * Only this thread's heap and the parked workers' are visited
 * (heap_each_quiescent).  Without file spill (Windows) the watermark does not
 * decide where memory goes — past it a pool comes from RAM all the same — so
 * there is nothing to win, and the memory is left to the idle decay.
 * Returns the number of blocks whose pages were released. */
int64_t ray_heap_relieve(void) {
    if (!RAY_HEAP_FILE_SPILL) return 0;
    if (atomic_load_explicit(&ray_parallel_flag, memory_order_relaxed) != 0)
        return 0;
    const int64_t pool = (int64_t)BSIZEOF(RAY_HEAP_POOL_ORDER);
    int64_t wm = heap_anon_watermark();
    int64_t committed = atomic_load_explicit(&g_anon_committed, memory_order_relaxed);

    /* What the statement that just ended did, and what that says of the
     * last relief. */
    int64_t maps   = atomic_load_explicit(&g_anon_maps, memory_order_relaxed);
    int64_t large  = atomic_load_explicit(&g_anon_large, memory_order_relaxed);
    int64_t spills = atomic_load_explicit(&g_anon_spills, memory_order_relaxed);
    int64_t peak   = atomic_exchange_explicit(&g_anon_stmt_peak, committed,
                                              memory_order_relaxed);
    bool spilled = atomic_exchange_explicit(&g_relieve_spills, spills,
                                            memory_order_relaxed) != spills;
    bool mapped = atomic_exchange_explicit(&g_relieve_maps, maps,
                                           memory_order_relaxed) != maps || spilled;
    large -= atomic_exchange_explicit(&g_relieve_large, large, memory_order_relaxed);
    bool needed = false;
    int state = atomic_load_explicit(&g_relieve_state, memory_order_relaxed);
    if (wm != atomic_load_explicit(&g_relieve_wm, memory_order_relaxed))
        state = RELIEF_NONE;
    else if (mapped && state == RELIEF_JUST) {
        needed = large >= pool && peak + pool > wm
              && atomic_load_explicit(&g_relieve_other, memory_order_relaxed);
        state = (spilled && large < pool) ? RELIEF_FUTILE : RELIEF_NONE;
    } else if (mapped && state == RELIEF_FUTILE && !spilled)
        state = RELIEF_NONE;
    atomic_store_explicit(&g_relieve_state, state, memory_order_relaxed);
    if (wm <= 0) return 0;

    bool relieve = (spilled || needed) && state == RELIEF_NONE;
    bool near = relieve || spilled || committed + pool > wm;
    if (!near && atomic_load_explicit(&g_spill_pools, memory_order_relaxed) == 0)
        return 0;
    int64_t now = ray_time_now_ms();
    if (!near) {
        if (now - atomic_load_explicit(&g_far_visit_ms, memory_order_relaxed)
            < RAY_FAR_VISIT_MS)
            return 0;
        atomic_store_explicit(&g_far_visit_ms, now, memory_order_relaxed);
    }
    relieve_visit_t v = { 0, (uint32_t)now, near, false, false };
    if (relieve) {
        relieve_measure_t m = { direct_cache_bytes(), 0 };
        heap_each_quiescent(heap_relieve_measure, &m);
        v.act = m.room >= pool && committed - m.room + pool <= wm;
        if (v.act) {
            direct_cache_drain();
            atomic_store_explicit(&g_relieve_wm, wm, memory_order_relaxed);
            atomic_store_explicit(&g_relieve_state, RELIEF_JUST, memory_order_relaxed);
            atomic_store_explicit(&g_relieve_other, m.room_other >= pool,
                                  memory_order_relaxed);
        }
    }
    heap_each_quiescent(heap_relieve_visit, &v);
    /* The next statement's peak starts from the footprint this boundary
     * leaves, not the one it found: what the relief handed back is no part
     * of it (its peak decides whether it needed a relief). */
    atomic_store_explicit(&g_anon_stmt_peak,
                          atomic_load_explicit(&g_anon_committed, memory_order_relaxed),
                          memory_order_relaxed);
    if (v.dropped) ray_sys_mem_flush();
    return v.released;
}

void ray_heap_merge(ray_heap_t* src) {
    ray_heap_t* dst = ray_tl_heap;
    if (!dst || !src) return;
    /* A parked pool's blocks are on no list; the freelist splice below
     * carries only listed ones. */
    heap_unpark_all(src);

    /* Merge stats: dst inherits src's outstanding allocations so that
     * future local frees of those blocks correctly decrement dst. */
    dst->stats.alloc_count     += src->stats.alloc_count;
    dst->stats.free_count      += src->stats.free_count;
    dst->stats.bytes_allocated += src->stats.bytes_allocated;
    dst->stats.slab_hits       += src->stats.slab_hits;
    dst->stats.direct_count    += src->stats.direct_count;
    dst->stats.direct_bytes    += src->stats.direct_bytes;
    if (src->stats.peak_bytes > dst->stats.peak_bytes)
        dst->stats.peak_bytes = src->stats.peak_bytes;

    /* Transfer slabs: fit into dst cache, coalesce overflow */
    for (int i = 0; i < RAY_SLAB_ORDERS; i++) {
        while (src->slabs[i].count > 0 && dst->slabs[i].count < dst->slab_cap[i])
            dst->slabs[i].stack[dst->slabs[i].count++] =
                src->slabs[i].stack[--src->slabs[i].count];
        while (src->slabs[i].count > 0) {
            ray_t* blk = src->slabs[i].stack[--src->slabs[i].count];
            int pidx = heap_find_pool(dst, blk);
            uintptr_t pb;
            uint8_t po;
            if (pidx >= 0) {
                pb = (uintptr_t)dst->pools[pidx].base;
                po = dst->pools[pidx].pool_order;
            } else {
                ray_pool_hdr_t* phdr = ray_pool_of(blk);
                if (!phdr) continue;
                pb = (uintptr_t)phdr;
                po = phdr->pool_order;
            }
            heap_coalesce(dst, blk, pb, po);
        }
    }

    /* Blocks other threads freed back to src, taken over by dst along with
     * src's pools. */
    ray_t* fblk = atomic_exchange_explicit(&src->foreign, NULL,
                                           memory_order_acquire);
    while (fblk) {
        ray_t* next = fblk->fl_next;
        int pidx = heap_find_pool(dst, fblk);
        uintptr_t pb;
        uint8_t po;
        if (pidx >= 0) {
            pb = (uintptr_t)dst->pools[pidx].base;
            po = dst->pools[pidx].pool_order;
        } else {
            ray_pool_hdr_t* phdr = ray_pool_of(fblk);
            if (!phdr) { fblk = next; continue; }
            pb = (uintptr_t)phdr;
            po = phdr->pool_order;
        }
        /* dst inherited src's bytes_allocated above, and these blocks are
         * still charged in it — they were freed to their owner but never
         * drained.  Coalescing them without this leaves the charge standing
         * for good, so .sys.mem drifts up by every merged pending block. */
        RAY_STAT(dst->stats.bytes_allocated -= BSIZEOF(fblk->order));
        heap_coalesce(dst, fblk, pb, po);
        fblk = next;
    }

    /* Merge freelists: circular list splice (src chain into dst chain) */
    for (int i = RAY_ORDER_MIN; i < RAY_HEAP_FL_SIZE; i++) {
        if (fl_empty(&src->freelist[i])) continue;

        ray_fl_head_t* src_head = &src->freelist[i];
        ray_fl_head_t* dst_head = &dst->freelist[i];

        /* Splice: src's chain [src_first...src_last] into dst after sentinel */
        ray_t* src_first = src_head->fl_next;
        ray_t* src_last  = src_head->fl_prev;
        ray_t* dst_first = dst_head->fl_next;

        /* src_first goes after dst sentinel */
        dst_head->fl_next = src_first;
        src_first->fl_prev = (ray_t*)dst_head;

        /* src_last connects to old dst_first */
        src_last->fl_next = dst_first;
        dst_first->fl_prev = src_last;

        dst->avail |= (1ULL << i);

        /* Reset src sentinel to empty */
        fl_init(src_head);
    }

    src->avail = 0;

    /* Update pool headers: set heap_id to dst, transfer pool entries.
     * Do NOT rewrite heap_id for pools that can't be tracked — that would
     * make coalescing reference a pool not in dst's pool table. */
    for (uint32_t i = 0; i < src->pool_count; i++) {
        if (dst->pool_count < RAY_MAX_POOLS) {
            ray_pool_hdr_t* hdr = (ray_pool_hdr_t*)src->pools[i].base;
            hdr->heap_id = dst->id;
            dst->pools[dst->pool_count++] = src->pools[i];
        } else {
            /* Pool overflow: only triggers at RAY_MAX_POOLS (512 pools = 16GB+).
             * dst cannot track the pool, so it could never be released —
             * fatal in every build.  Explicit rather than assert(0): under
             * -DNDEBUG that compiled away and the pool leaked silently (#652). */
            fprintf(stderr, "rayforce: ray_heap_merge: pool overflow at "
                            "RAY_MAX_POOLS (%d)\n", (int)RAY_MAX_POOLS);
            abort();
        }
    }
    src->pool_count = 0;
}

/* --------------------------------------------------------------------------
 * Public foreign-blocks flush
 * -------------------------------------------------------------------------- */

void ray_heap_flush_foreign(void) {
    ray_heap_t* h = ray_tl_heap;
    if (!h) return;
    /* No ray_parallel_flag gate: the list holds only this heap's own blocks,
     * and only this thread coalesces into its own freelists. */
    heap_drain_foreign(h);
}

/* --------------------------------------------------------------------------
 * Post-dispatch reclaim (dispatcher side)
 *
 * A parallel operator's workers allocate their per-task buffers from their
 * own heaps and the main thread frees them once the dispatch has completed,
 * so every such block ends up on the owning worker's foreign list.  The
 * owner takes that list back only when an allocation finds its freelists
 * empty at the order it needs — and a warm worker with a partly cut pool
 * rarely does: it keeps splitting fresh pool space instead, touching new
 * pages every round while its own freed blocks wait on the list.  Under a
 * steady stream of such operators the process grows by one full pool per
 * worker before any block is reused, and nothing short of the idle decay
 * (which needs the process to sit quiet) drains it earlier.
 *
 * So the dispatcher drains each worker's list at the end of each parallel
 * region.  The conditions are the ones the decay sweep relies on: the flag
 * is clear, so every worker has done its last pending-- and is claiming
 * nothing, allocating nothing and freeing nothing until the next dispatch;
 * a concurrent push onto a foreign list from some other thread is safe
 * because the drain takes the whole list in one exchange and leaves later
 * arrivals for the next round.  Only pool workers qualify: the registry
 * also holds the heaps of other live threads (a server's poll thread, an
 * embedding's own threads), whose freelists only their owner may touch, so
 * the pool names the heaps rather than this walking the registry.  No
 * pages are released here — the blocks only go back to freelists, so the
 * next round reuses them without faulting.  Cost: one load per worker and
 * the coalescing of whatever was freed cross-thread since the last
 * dispatch, work the owner would otherwise do on its next dry allocation.
 * -------------------------------------------------------------------------- */

void ray_heap_reclaim_worker(ray_heap_t* h) {
    if (!h) return;
    if (atomic_load_explicit(&ray_parallel_flag, memory_order_acquire) != 0)
        return;
    if (!atomic_load_explicit(&h->foreign, memory_order_relaxed)) return;
    heap_drain_foreign(h);
}

/* --------------------------------------------------------------------------
 * Pending-merge queue (lock-free LIFO)
 *
 * Workers that are torn down push their heap onto this queue instead of
 * destroying it immediately. The main thread drains the queue, merging
 * each pending heap into its own and then destroying it.
 * -------------------------------------------------------------------------- */

void ray_heap_push_pending(ray_heap_t* heap) {
    if (!heap) return;
    /* Unregister so no new foreign blocks target this heap (lock: serialize
     * against concurrent destroyers walking the registry). */
    ray_registry_lock();
    ray_heap_registry[heap->id % RAY_HEAP_REGISTRY_SIZE] = NULL;
    ray_registry_unlock();
    /* Lock-free push: CAS loop on global LIFO head */
    heap->pending_next = atomic_load_explicit(&ray_heap_pending_merge, memory_order_relaxed);
    while (!atomic_compare_exchange_weak_explicit(
            &ray_heap_pending_merge,
            &heap->pending_next, heap,
            memory_order_release, memory_order_relaxed))
        ;
}

void ray_heap_drain_pending(void) {
    /* Atomically steal the entire pending list */
    ray_heap_t* pending = atomic_exchange_explicit(
        &ray_heap_pending_merge, NULL,
        memory_order_acquire);
    while (pending) {
        ray_heap_t* next = pending->pending_next;
        ray_heap_merge(pending);
        /* Free the heap struct (pools already transferred by merge) */
        uint16_t saved_id = pending->id;
        size_t heap_sz = (sizeof(ray_heap_t) + 4095) & ~(size_t)4095;
        ray_vm_free(pending, heap_sz);
        heap_id_release(saved_id);
        pending = next;
    }
}

/* --------------------------------------------------------------------------
 * Scratch arena: bump allocator backed by buddy-allocated 64KB blocks
 * -------------------------------------------------------------------------- */

void* ray_scratch_arena_push(ray_scratch_arena_t* a, size_t nbytes) {
    /* 16-byte alignment */
    nbytes = (nbytes + 15) & ~(size_t)15;

    if (RAY_LIKELY(a->ptr != NULL && a->ptr + nbytes <= a->end))
        goto bump;

    /* Need a new backing block */
    if (a->n_backing >= RAY_ARENA_MAX_BACKING) return NULL;

    size_t block_data = BSIZEOF(RAY_ARENA_BLOCK_ORDER) - 32;
    /* If request exceeds standard block, allocate exact-fit */
    size_t alloc_size = nbytes > block_data ? nbytes : block_data;
    ray_t* blk = ray_alloc(alloc_size);
    if (!blk) return NULL;
    a->backing[a->n_backing++] = blk;
    a->ptr = (char*)ray_data(blk);
    /* The window ends at the block's real capacity.  `order` describes the
     * block size only for buddy blocks: a request at or above
     * RAY_HEAP_POOL_ORDER comes back as a DIRECT block, mapped at its exact
     * page-rounded size and tagged with the RAY_ORDER_DIRECT sentinel.
     * Reading that sentinel as an order claims a 2^39-byte block, and every
     * push after the one that triggered it is then bump-allocated past the
     * end of the mapping — a write straight into unmapped address space.
     * ray_block_data_bytes covers both block kinds. */
    a->end = a->ptr + ray_block_data_bytes(blk);

bump:;
    void* ret = a->ptr;
    a->ptr += nbytes;
    return ret;
}

void ray_scratch_arena_reset(ray_scratch_arena_t* a) {
    for (int i = 0; i < a->n_backing; i++)
        ray_free(a->backing[i]);
    a->n_backing = 0;
    a->ptr = NULL;
    a->end = NULL;
}

/* --------------------------------------------------------------------------
 * Parallel begin / end
 * -------------------------------------------------------------------------- */

void ray_parallel_begin(void) { atomic_store(&ray_parallel_flag, 1); }
void ray_parallel_end(void) {
    atomic_store(&ray_parallel_flag, 0);
    ray_heap_gc();
}

/* --------------------------------------------------------------------------
 * Native footprint — ray_shallow_bytes / ray_retained_bytes (rayforce.h)
 *
 * Read-only mirrors of the ownership rules in ray_release_owned_refs and
 * of the mapping arithmetic in ray_free.  Keep the three in step: a new
 * owned child kind belongs in retained_walk, a new private-storage kind
 * (one only its holder releases and no API hands out) in shallow too.
 * -------------------------------------------------------------------------- */

/* Bytes of a file mapping owned by block v (mmod 1) — the size ray_free
 * hands to the unmap.  A string column reaches its region descriptor
 * through its pool (or the registry once the pool was swapped); anything
 * else is the page-rounded payload plus an inline passenger index. */
static size_t mapped_block_bytes(const ray_t* v) {
    if (v->type == RAY_TABLE || v->type == RAY_DICT || v->type == RAY_LIST) return 0;
    {
        /* Same route as ray_free: the pool's descriptor for a string
         * column, else the registry — which also holds every indexed
         * mapped column (col.c), whether or not its index is still
         * attached. */
        ray_file_map_t* m = NULL;
        if (v->type == RAY_STR && v->str_pool && !RAY_IS_ERR(v->str_pool) &&
            v->str_pool->mmod == 3)
            m = v->str_pool->file_map;
        if (!m) m = ray_file_map_lookup(v);
        if (m) return m->len;
    }
    if (v->type <= 0 || v->type >= RAY_TYPE_COUNT) return 0;
    uint8_t esz = ray_sym_elem_size(v->type, v->attrs);
    size_t data_size = 32 + (size_t)v->len * esz;
    if (v->type == RAY_STR) {
        size_t pool_len = 0;
        if (v->str_pool && !RAY_IS_ERR(v->str_pool) && v->str_pool->len > 0)
            pool_len = (size_t)v->str_pool->len;
        data_size += 32 + pool_len;
    }
    if ((v->attrs & RAY_ATTR_HAS_INDEX) && v->index && !RAY_IS_ERR(v->index)) {
        ray_index_t* ix = ray_index_payload(v->index);
        if (ix->markers & RAY_MARK_MMAP) {
            int64_t region_off = ((int64_t)data_size + 31) & ~(int64_t)31;
            data_size = (size_t)(region_off + ray_index_inline_size(ix));
        }
    }
    return (data_size + 4095) & ~(size_t)4095;
}

/* The allocation behind one block, and nothing it points at. */
static size_t block_own_bytes(const ray_t* v) {
    if (!v) return 0;
    if (v->attrs & RAY_ATTR_ARENA) return 0;          /* singletons, arena values */
    switch (v->mmod) {
    case 1:  return mapped_block_bytes(v);            /* owns its mapping */
    case 2:  return 0;                                /* borrowed */
    case 3:  return 0;                                /* pool inside a column's region */
    default: break;
    }
    if (ray_is_direct(v)) return ray_direct_map_size(v);
    if (v->order < RAY_ORDER_MIN || v->order > RAY_HEAP_MAX_ORDER) return 0;
    return BSIZEOF(v->order);
}

/* Heap-resident index: its block plus its tables.  A mapping passenger is
 * charged through the column's mapping. */
static size_t index_own_bytes(const ray_t* idx) {
    if (!idx || RAY_IS_ERR(idx) || idx->mmod == 1) return 0;
    ray_index_t* ix = ray_index_payload((ray_t*)idx);
    if (ix->markers & RAY_MARK_MMAP) return 0;
    size_t n = block_own_bytes(idx);
    ray_t* kids[4];
    int k = ray_index_child_blocks(ix, kids, 4);
    for (int i = 0; i < k; i++) n += block_own_bytes(kids[i]);
    return n;
}

size_t ray_shallow_bytes(const ray_t* v) {
    if (!v) return 0;
    size_t n = block_own_bytes(v);
    if (RAY_IS_ERR(v)) return n;
    if (ray_is_atom(v)) {
        if (ray_atom_owns_obj(v) && v->obj && !RAY_IS_ERR(v->obj))
            n += block_own_bytes(v->obj);
        return n;
    }
    if (v->attrs & RAY_ATTR_SLICE) return n;
    if (v->type == RAY_INDEX) return index_own_bytes(v);
    if (v->attrs & RAY_ATTR_HAS_INDEX) n += index_own_bytes(v->index);
    if (v->type == RAY_STR && v->str_pool && !RAY_IS_ERR(v->str_pool))
        n += block_own_bytes(v->str_pool);
    return n;
}

/* Pointer set for the retained walk: open addressing over ray_sys_alloc,
 * so the walk never touches the heap it is measuring.  On OOM the set stops
 * growing and add() answers "new" for everything — the walk then still
 * terminates (the value graph is acyclic) and can only over-count. */
typedef struct { const ray_t** slots; size_t cap; size_t n; } ptrset_t;

static bool ptrset_add(ptrset_t* s, const ray_t* p) {
    if (s->n * 2 >= s->cap) {
        size_t ncap = s->cap ? s->cap * 2 : 64;
        const ray_t** ns = (const ray_t**)ray_sys_alloc(ncap * sizeof(*ns));
        if (!ns) return true;
        memset(ns, 0, ncap * sizeof(*ns));
        for (size_t i = 0; i < s->cap; i++) {
            const ray_t* q = s->slots[i];
            if (!q) continue;
            size_t h = ((uintptr_t)q >> 5) & (ncap - 1);
            while (ns[h]) h = (h + 1) & (ncap - 1);
            ns[h] = q;
        }
        if (s->slots) ray_sys_free((void*)s->slots);
        s->slots = ns; s->cap = ncap;
    }
    size_t h = ((uintptr_t)p >> 5) & (s->cap - 1);
    while (s->slots[h]) {
        if (s->slots[h] == p) return false;
        h = (h + 1) & (s->cap - 1);
    }
    s->slots[h] = p; s->n++;
    return true;
}

static size_t retained_walk(const ray_t* v, ptrset_t* seen) {
    if (!v || RAY_IS_ERR(v)) return 0;
    if (!ptrset_add(seen, v)) return 0;
    size_t n = block_own_bytes(v);

    if (ray_is_atom(v)) {
        if (v->type == RAY_LAMBDA) {
            ray_t*  lam   = (ray_t*)v;
            ray_t** slots = (ray_t**)ray_data(lam);
            for (int i = 0; i < 4; i++) n += retained_walk(slots[i], seen);
            n += retained_walk(LAMBDA_NFO(lam), seen);
            n += retained_walk(LAMBDA_DBG(lam), seen);
            n += retained_walk(LAMBDA_CLOSURE(lam), seen);
            n += retained_walk(LAMBDA_CAPROWS(lam), seen);
            return n;
        }
        if (ray_atom_owns_obj(v)) n += retained_walk(v->obj, seen);
        return n;
    }
    if (v->attrs & RAY_ATTR_SLICE) return n + retained_walk(v->slice_parent, seen);
    if (v->type == RAY_INDEX) {
        ray_index_t* ix = ray_index_payload((ray_t*)v);
        if (v->mmod == 1 || (ix->markers & RAY_MARK_MMAP)) return 0;
        ray_t* kids[4];
        int k = ray_index_child_blocks(ix, kids, 4);
        for (int i = 0; i < k; i++) n += retained_walk(kids[i], seen);
        return n;
    }
    if ((v->attrs & RAY_ATTR_HAS_INDEX) && v->index && !RAY_IS_ERR(v->index) &&
        v->index->mmod != 1)
        n += retained_walk(v->index, seen);
    if (v->type == RAY_STR) {
        if (v->str_pool && !RAY_IS_ERR(v->str_pool)) n += retained_walk(v->str_pool, seen);
        return n;
    }
    if (v->type == RAY_SYM) return n;   /* the domain is not a ray_t */
    if (RAY_IS_PARTED(v->type) || v->type == RAY_LIST) {
        ray_t** ptrs = (ray_t**)ray_data((ray_t*)v);
        for (int64_t i = 0; i < v->len; i++) n += retained_walk(ptrs[i], seen);
        return n;
    }
    if (v->type == RAY_MAPCOMMON || v->type == RAY_TABLE || v->type == RAY_DICT) {
        ray_t** slots = (ray_t**)ray_data((ray_t*)v);
        n += retained_walk(slots[0], seen);
        n += retained_walk(slots[1], seen);
    }
    return n;
}

size_t ray_retained_bytes(const ray_t* v) {
    ptrset_t seen = { NULL, 0, 0 };
    size_t n = retained_walk(v, &seen);
    if (seen.slots) ray_sys_free((void*)seen.slots);
    return n;
}