cove_runtime/

heap.rs

//! The per-task mark-and-sweep collector.
//!
//! The Language Card says memory is managed by a precise, non-moving
//! mark-and-sweep collector, and ADR 0011 narrows that to a heap per task over
//! the values a task owns. This module is that heap.
//!
//! ADR 0008 gives each spawned task a thread and an [`crate::interp::Interpreter`]
//! of its own, so a heap belongs to one interpreter and is reached only from
//! the thread running it. That is what makes "per task" more than a
//! convention: a task's objects are unreachable from any other thread, so a
//! collection needs no safepoint from anyone else and takes no lock.
//!
//! # What the heap owns
//!
//! `Rc` reclaims a value the moment nothing points at it, which is correct for
//! every Cove value that is built once and never altered — a string, an array,
//! a map, a set, a closure, a struct, an enum case. None of those can be made
//! to point back at something that points at them, because each is built from
//! values that already exist. The one exception is
//! [`crate::value::VectorStorage`]: a vector's elements are behind a
//! `RefCell`, so `v.push(v)` is a cycle, and `Rc` alone will never free it.
//!
//! The heap therefore tracks exactly the objects that can form a cycle. `Rc`
//! remains the allocation handle and still reclaims everything acyclic on its
//! own; the collector exists for what `Rc` cannot do.
//!
//! The heap holds a [`Weak`] handle to each object rather than a strong one.
//! That is not an optimisation: `freeze()` consumes *uniquely owned* vector
//! storage and asks `Rc::strong_count` whether the caller holds the only
//! handle, so a heap holding a strong reference would make `freeze()` fail on
//! every vector. A `Weak` keeps the collector out of the language's own
//! uniqueness rule, and it costs nothing — a cycle keeps itself alive, so a
//! `Weak` to a member of one always upgrades.
//!
//! # Roots, and why reference counts are part of them
//!
//! ADR 0011 is explicit that the roots are the interpreter's own structures
//! rather than a machine stack, so there are no stack maps here. `Roots` is a
//! trait naming something that can *drive* a walk of one task's roots, and
//! `SlotRoots` is its one implementor now: every binding the interpreter
//! creates is a [`crate::interp`] `Place`, whose slot is an
//! `Rc<RefCell<Value>>` registered in a `SlotRoots` list with the same
//! push-and-truncate discipline the environment chain already has; the
//! collector borrows each cell as it walks, so what it sees is what the slot
//! holds now rather than what it held when the binding was made.
//!
//! Before ADR 0034 there was a second implementor: the predecessor VM had no
//! such cells, and put every binding into one contiguous `Vec<Value>`
//! instead, precisely so a call on that backend would allocate nothing. A
//! collector that demanded the interpreter's shape would have made that VM
//! build a cell per binding and give back what the arrangement bought; one
//! that demanded the VM's would have made the interpreter snapshot its
//! bindings into a vector, which is both a copy and a lie, since the
//! snapshot would be of the values as they were rather than as they are. A
//! walk rather than a structure is what let the collector work over both
//! shapes without asking either one to give back what it was for.
//!
//! The linear-memory backend does not implement this trait at all. Its heap
//! belongs to the run rather than to a task, and it finds its roots from a
//! frame's static reference map instead — `crate::vm::mem`'s own `Roots`
//! describes that on its own terms. Adding that backend cost this module
//! nothing, which is the walk-not-a-list design paying for itself a second
//! time: the point was never only the two shapes that existed when it was
//! written.
//!
//! A walk is asked for twice, once to count and once to mark, so an
//! implementation must be re-walkable and must yield the same values both
//! times. Nothing here consumes it.
//!
//! The walk covers what the program has named. It does not cover a value the
//! interpreter is holding in a Rust local while it is mid-evaluation — the
//! left operand of a `+` whose right operand is still being evaluated. Those
//! are the values ADR 0011 calls "values being evaluated," and a tree walker
//! has no list of them by construction. The linear-memory backend has no
//! such gap to close: a value it is computing already lives in a frame slot
//! the static reference map already covers, and an object allocated before
//! its fields are written is zeroed rather than left holding whatever
//! preceded it, so there is nothing mid-evaluation for its collector to
//! miss the way this one can.
//!
//! The collector finds them exactly, without scanning anything: it counts the
//! references it *can* see. For every shared allocation it walks — a vector, an
//! array, a map, a struct, a closure, a trait object, a task, a task scope —
//! it sums the references reachable from the walked roots and from the
//! objects it manages, and compares that with `Rc::strong_count`. A shortfall
//! is a reference held somewhere the collector cannot read — a Rust local it
//! has not walked — so that allocation, and everything it holds, is a root.
//!
//! This is the rule that makes a safepoint safe rather than merely
//! well-chosen. A value in a Rust local is itself a reference, so a value the
//! collector cannot see is a value whose count does not add up; there is no
//! arrangement of locals the interpreter can reach a safepoint in that hides
//! one. What the interpreter has to get right is therefore narrower than
//! "have everything on a stack": it has to not walk anything twice, because a
//! reference counted twice is a shortfall concealed. The linear-memory
//! backend answers the same demand — a safepoint must not reach a value its
//! collector cannot find — without this counting trick: a static reference
//! map leaves nothing held off to the side to count.
//!
//! Counting the containers as well as the objects is what makes this sound
//! rather than merely plausible. An array can hold the only reference to a
//! vector while being held itself by a garbage cycle *and* by a temporary; if
//! only the vector were counted, every reference to it would look accounted
//! for — by the garbage — and the sweep would empty something the program can
//! still reach.
//!
//! This is precise in the sense ADR 0001 asks for: no word is guessed to be a
//! pointer, and no integer is ever mistaken for one. It is also the invariant
//! that makes every other awkward case safe. A slot the interpreter has
//! mutably borrowed cannot be read, so its references go unseen, so whatever
//! it holds is treated as a root.
//!
//! # What it does not do
//!
//! No finalizers, no compaction, no generations, no concurrent or incremental
//! collection, no weak references in the language. Each is out of scope in ADR
//! 0001 and remains so in ADR 0011.
//!
//! # `Shared`
//!
//! ADR 0011 says a `Shared<T>` cell owns its contents rather than any task's
//! heap, and collects them with the cell. That is what happens, and it needs
//! no collector at all.
//!
//! A [`crate::shared::SharedCell`] holds a [`crate::task::Transfer`], not a
//! [`Value`], and `Transfer::of` refuses a `Vector`. A cell therefore cannot
//! hold a collectable object, so there is nothing in one for a heap to own and
//! no way for a cycle among a task's objects to run through one. The `Arc`
//! frees the contents with the cell, which is exactly what the ADR asks for. Each `lock` materialises
//! a fresh `Value` for the locking task, and *that* copy is an ordinary value
//! in that task's heap, collected there like any other.
//!
//! So the collector treats a `Shared` as a leaf: it never takes the cell's
//! lock. That is not only unnecessary, it is required. `lock` holds the mutex
//! for the whole of the closure it is given, that closure runs Cove code, and
//! Cove code reaches safepoints — so a collector that locked a cell would
//! sooner or later wait for a lock the collecting thread already holds.
//!
//! One thing this does not reach: a cell may hold *another* cell, including
//! itself, and that is an `Arc` cycle no heap here can see. Cells are
//! reachable from every task that was given one and outlive all of them, so
//! collecting cycles among them would need a collector that stops every
//! thread — which ADR 0011 rules out under "no concurrent collection". It is
//! a real leak, and the ADR now says so under "What this leaves uncollected".

use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use std::mem::size_of;
use std::rc::{Rc, Weak};
use std::time::Duration;

use crate::task::{Task, TaskScope};
use crate::value::{MapKey, Repr, Value, VectorStorage};
use crate::wallclock::Instant;

/// The fewest objects a task may allocate between two collections.
///
/// A collection costs a walk of the live set, so collecting after every
/// allocation would make the collector the program. This floor is what a small
/// program pays: it allocates fewer than this many vectors and is never
/// collected at all.
const MIN_ALLOCATIONS_BETWEEN_COLLECTIONS: u64 = 64;

/// How much the object count may grow past the live set before the next
/// collection.
///
/// Doubling makes the total collection work over a run proportional to total
/// allocation rather than to allocation times live size, which is the standard
/// reason to size the next collection from the last one's survivors.
const GROWTH_FACTOR: u64 = 2;

/// One task's roots, as something the collector can walk.
///
/// # Why a walk and not a list
///
/// `Interpreter` is this trait's one implementor now, giving every binding
/// an `Rc<RefCell<Value>>` of its own and handing the collector the cells,
/// so a collection reads what a binding holds *now* rather than a snapshot
/// of what it held when the binding was made. Before ADR 0034 there was a
/// second: the predecessor VM gave every binding a slot of one contiguous
/// `Vec<Value>` instead, which was the whole reason a call on that backend
/// allocated nothing. A collector that demanded the interpreter's shape
/// would have made that VM build a cell per binding and give back what the
/// arrangement bought; one that demanded the VM's would have made the
/// interpreter snapshot its bindings into a vector, which is both a copy and
/// a lie, since the snapshot would be of the values as they were rather than
/// as they are.
///
/// So neither shape was asked for; what is asked for is a walk: an
/// implementation calls `visit` once per reference it holds, and how it
/// finds them is its own business. An enum of the shapes was the alternative
/// and it was not taken, because it would put every backend's root
/// representation in this module and make adding another an edit here
/// rather than there — which is exactly what let the linear-memory backend
/// add a third kind of root, in `crate::vm::mem`'s own `Roots`, without
/// touching this file at all.
///
/// # What an implementation owes
///
/// [`Heap::collect`] walks the roots **twice** — once to count the references
/// it can see, once to mark from them — so a walk must be repeatable and must
/// yield the same references both times. Nothing consumes it.
///
/// It must also yield each reference **exactly once**. Yielding one twice is
/// not conservative: the collector's soundness rests on comparing the
/// references it can see against `Rc::strong_count`, so a reference counted
/// twice makes a live allocation's count add up when it does not, and the
/// sweep then empties something a backend temporary can still reach. Both
/// implementations here therefore de-duplicate what they know can alias:
/// [`SlotRoots`] by slot address, because a `var` parameter binds the
/// caller's slot and one cell can be registered by two environments.
///
/// Yielding *too few* is safe in the same accounting, and is the reason the
/// VM need not walk its constant pool: an unseen reference is a shortfall,
/// and a shortfall is a root.
pub(crate) trait Roots {
    /// Calls `visit` once for every value this task holds directly.
    fn walk(&self, visit: &mut dyn FnMut(&Value));
}

/// A slot the collector starts from: one binding's storage.
///
/// This is [`crate::interp`]'s `Place` slot, registered here so a collection
/// can read it. The interpreter is the only writer, and it keeps the list in
/// step with the environment chain: a binding pushes, and leaving a block or a
/// call truncates back to the length it recorded on entry.
type Slot = Rc<RefCell<Value>>;

/// Every binding one interpreter currently holds, innermost last.
///
/// The list is shared by every environment on one thread, and its
/// push-and-truncate discipline mirrors that thread's environment chain
/// exactly. There is one list per interpreter and one interpreter per task, so
/// this *is* a task's roots — nothing has to be sliced out of a larger set.
#[derive(Default)]
pub(crate) struct SlotRoots {
    slots: Vec<Slot>,
}

impl SlotRoots {
    /// An empty root set.
    pub(crate) fn new() -> SlotRoots {
        SlotRoots::default()
    }

    /// How many slots are registered. A caller records this before entering a
    /// scope and hands it back to [`SlotRoots::truncate`] on the way out.
    pub(crate) fn len(&self) -> usize {
        self.slots.len()
    }

    /// Registers one binding's slot.
    pub(crate) fn push(&mut self, slot: Slot) {
        self.slots.push(slot);
    }

    /// Drops every slot registered after `len`, which is what leaving a block
    /// or a call does.
    pub(crate) fn truncate(&mut self, len: usize) {
        self.slots.truncate(len);
    }
}

impl Roots for SlotRoots {
    /// Borrows each registered slot and yields what it holds.
    ///
    /// One cell may be registered twice — a `var` parameter binds the caller's
    /// slot, so the caller's environment and the callee's both name it — and
    /// yielding it twice would count one reference as two. The addresses
    /// already seen are what stops that.
    ///
    /// A slot that cannot be borrowed is one the interpreter is writing
    /// through. It is skipped rather than waited for, which is not a hole:
    /// its references go uncounted, so whatever it holds is short by one and
    /// is a root under the rule this module's documentation gives.
    fn walk(&self, visit: &mut dyn FnMut(&Value)) {
        let mut walked: HashSet<usize> = HashSet::new();
        for slot in &self.slots {
            if !walked.insert(Rc::as_ptr(slot) as usize) {
                continue;
            }
            if let Ok(value) = slot.try_borrow() {
                visit(&value);
            }
        }
    }
}

/// What one collection did.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Collection {
    /// Objects allocated since the previous collection.
    pub allocated: u64,
    /// Objects the sweep reclaimed.
    pub freed_objects: u64,
    /// Bytes held by what the sweep reclaimed, counting each shared
    /// allocation once and not following an edge into an object that
    /// survived. A value the reclaimed objects shared with a survivor is
    /// counted here even though it was not released, so this is an upper
    /// bound; [`Collection::live_bytes`] is the measured figure.
    pub freed_bytes: u64,
    /// Objects still live after the sweep.
    pub live_objects: u64,
    /// Bytes the live set holds: every string, array, map, set, struct, enum
    /// case and closure the live objects and the roots reach, with each
    /// shared allocation counted once.
    pub live_bytes: u64,
    /// How long the program was stopped for this collection.
    pub pause: Duration,
}

/// What a run's heaps did in total.
///
/// `cove run --stats` prints this, and the trace's `heap_summary` event
/// carries it.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct HeapStats {
    /// Collectable objects allocated over the whole run.
    pub allocated_objects: u64,
    /// Bytes those allocations asked for, at the size each object was born.
    pub allocated_bytes: u64,
    /// How many collections ran.
    pub collections: u64,
    /// Objects those collections reclaimed.
    pub freed_objects: u64,
    /// Bytes live at the most recent collection.
    pub live_bytes: u64,
    /// Objects live at the most recent collection.
    pub live_objects: u64,
    /// The largest live set any collection measured.
    pub peak_bytes: u64,
    /// Total time the program was stopped for collection.
    pub pause: Duration,
}

impl HeapStats {
    /// Folds one heap's totals into these, which is what happens when a task's
    /// thread ends and its heap is retired into the run's.
    ///
    /// The live figures are not merged: a live set is a present fact about one
    /// heap rather than a total over a run. At the end of a run every task's
    /// heap has been swept and gone with its thread, so what is live is the
    /// entry's own and [`crate::interp::Interpreter::heap_stats`] reads it
    /// from there.
    pub fn merge(&mut self, other: &HeapStats) {
        self.allocated_objects += other.allocated_objects;
        self.allocated_bytes += other.allocated_bytes;
        self.collections += other.collections;
        self.freed_objects += other.freed_objects;
        self.peak_bytes = self.peak_bytes.max(other.peak_bytes);
        self.pause += other.pause;
    }
}

/// One task's heap.
///
/// A task owns the objects it allocates. The Language Card's task-safety rule
/// is what makes that a language rule rather than an approximation: a vector
/// may not cross a task boundary, so no two tasks ever hold the same one, and
/// a task can collect without waiting for any other task to reach a safepoint.
pub(crate) struct Heap {
    /// Every object this task has allocated and not yet lost, keyed by the
    /// address the object was allocated at. The address is stable while the
    /// `Weak` lives, so it identifies the object even after `Rc` has already
    /// reclaimed it.
    objects: HashMap<usize, Weak<VectorStorage>>,
    allocations_since_collection: u64,
    next_collection_at: u64,
    stats: HeapStats,
}

impl Heap {
    /// An empty heap.
    pub(crate) fn new() -> Heap {
        Heap {
            objects: HashMap::new(),
            allocations_since_collection: 0,
            next_collection_at: MIN_ALLOCATIONS_BETWEEN_COLLECTIONS,
            stats: HeapStats::default(),
        }
    }

    /// This heap's totals so far.
    pub(crate) fn stats(&self) -> HeapStats {
        self.stats
    }

    /// Whether the heap is tracking no object at all, which is what a program
    /// that never made a vector looks like.
    pub(crate) fn is_empty(&self) -> bool {
        self.objects.is_empty()
    }

    /// This heap's totals so far, resetting the cumulative ones.
    ///
    /// A task's heap ends with its thread, and what it did is folded into the
    /// run's totals by [`crate::runtime::Runtime::retire_heap`]. Taking the
    /// counters rather than reading them keeps a second fold from counting the
    /// same allocation twice. The live figures describe the heap right now and
    /// are not counters, so they are left alone.
    pub(crate) fn take_stats(&mut self) -> HeapStats {
        let taken = self.stats;
        self.stats = HeapStats {
            live_bytes: self.stats.live_bytes,
            live_objects: self.stats.live_objects,
            peak_bytes: self.stats.peak_bytes,
            ..HeapStats::default()
        };
        taken
    }

    /// Allocates growable vector storage owned by this heap.
    ///
    /// The returned `Rc` is the program's handle; the heap keeps only a
    /// `Weak`, so the value's lifetime is still `Rc`'s to decide until a cycle
    /// takes that decision away from it.
    pub(crate) fn allocate(&mut self, elements: Vec<Value>) -> Rc<VectorStorage> {
        let storage = VectorStorage::new(elements);
        self.stats.allocated_objects += 1;
        self.stats.allocated_bytes += object_bytes(&storage);
        self.allocations_since_collection += 1;
        self.objects
            .insert(Rc::as_ptr(&storage) as usize, Rc::downgrade(&storage));
        storage
    }

    /// Whether enough has been allocated since the last collection to be worth
    /// another one.
    pub(crate) fn should_collect(&self) -> bool {
        self.allocations_since_collection >= self.next_collection_at
    }

    /// Marks from the roots and sweeps what is not marked.
    ///
    /// `roots` is walked twice and never consumed: see [`Roots`] for what a
    /// walk owes, and for why the interpreter's own implementor of it is
    /// shaped the way it is.
    pub(crate) fn collect(&mut self, roots: &dyn Roots) -> Collection {
        let started = Instant::now();

        // An object `Rc` already reclaimed leaves a dead `Weak` behind. Drop
        // those first so every count below is over objects that still exist.
        self.objects.retain(|_, weak| weak.strong_count() > 0);

        // Every strong count is read before anything here upgrades a handle,
        // so the numbers describe the program's references and not the
        // collector's.
        let strong: HashMap<usize, usize> = self
            .objects
            .iter()
            .map(|(&at, weak)| (at, weak.strong_count()))
            .collect();

        let scan = self.count_visible_references(roots);
        let live = self.mark(roots, &scan, &strong);
        // The scan holds a handle to everything it saw. Releasing them before
        // the sweep keeps the collector out of the reference counts it is
        // about to act on.
        drop(scan);
        let (freed_objects, freed_bytes) = self.sweep(&live.marked);

        let collection = Collection {
            allocated: self.allocations_since_collection,
            freed_objects,
            freed_bytes,
            live_objects: self.objects.len() as u64,
            live_bytes: live.bytes,
            pause: started.elapsed(),
        };

        self.allocations_since_collection = 0;
        self.next_collection_at =
            (collection.live_objects * GROWTH_FACTOR).max(MIN_ALLOCATIONS_BETWEEN_COLLECTIONS);
        self.stats.collections += 1;
        self.stats.freed_objects += freed_objects;
        self.stats.live_bytes = collection.live_bytes;
        self.stats.live_objects = collection.live_objects;
        self.stats.peak_bytes = self.stats.peak_bytes.max(collection.live_bytes);
        self.stats.pause += collection.pause;
        collection
    }

    /// Counts, for every shared allocation it can reach, how many references
    /// to it the collector can see.
    ///
    /// It is not enough to do this for the objects the heap manages. A `Rc`
    /// container the collector does not manage — an array, a map, a struct, a
    /// closure, a trait object, a task — can hold the only reference to an
    /// object while itself being held by nothing but an evaluator temporary.
    /// Counting the container too is what catches that: its own references do
    /// not add up either, so everything it holds is a root.
    ///
    /// Each allocation's contents are walked exactly once, so each physical
    /// reference is counted exactly once and a shortfall is exactly the set of
    /// references the collector cannot read.
    fn count_visible_references(&self, roots: &dyn Roots) -> Scan {
        let mut scan = Scan {
            seen: HashMap::new(),
        };
        roots.walk(&mut |value| scan.count(value));
        for weak in self.objects.values() {
            let Some(object) = weak.upgrade() else {
                continue;
            };
            // The heap's own table is not a reference, so this registers the
            // object without sighting one.
            scan.observe(&object);
            if let Ok(elements) = object.elements.try_borrow() {
                for element in elements.iter() {
                    scan.count(element);
                }
            };
        }
        scan
    }

    /// Marks everything reachable from this task's roots and from every
    /// object some backend temporary still holds.
    fn mark(&self, roots: &dyn Roots, scan: &Scan, strong: &HashMap<usize, usize>) -> LiveSet {
        let mut marker = Marker {
            managed: &self.objects,
            excluded: None,
            marked: HashSet::new(),
            walked: HashSet::new(),
            bytes: 0,
            work: Vec::new(),
        };
        roots.walk(&mut |value| marker.visit(value));
        // Anything whose references do not add up is held from somewhere the
        // collector cannot read — an evaluator temporary — so it is a root.
        for (at, sighting) in &scan.seen {
            // A managed object's count is the one snapshotted before this
            // collection upgraded any handle; every other allocation's was
            // read the first time it was seen, before the scan took a handle
            // of its own.
            let held = strong.get(at).copied().unwrap_or(sighting.strong);
            if held > sighting.sighted {
                marker.visit(&sighting.held);
            }
        }
        // An object whose elements are borrowed right now cannot be read, and
        // so cannot be swept either: clearing it is exactly what the borrow
        // would forbid.
        for (at, weak) in &self.objects {
            if marker.marked.contains(at) {
                continue;
            }
            if let Some(object) = weak.upgrade() {
                if object.elements.try_borrow().is_err() {
                    marker.enqueue(object);
                }
            };
        }
        marker.drain();
        LiveSet {
            marked: marker.marked,
            bytes: marker.bytes,
        }
    }

    /// Reclaims every object the mark phase did not reach.
    ///
    /// The two phases matter. Clearing every doomed object's elements while
    /// the heap still holds a strong handle to all of them breaks the cycles
    /// without any object's `Drop` running inside another's, so a long chain
    /// is torn down iteratively rather than by a recursive drop that would
    /// exhaust the native stack. Dropping the handles afterwards is what
    /// actually frees the storage.
    fn sweep(&mut self, marked: &HashSet<usize>) -> (u64, u64) {
        let mut doomed: Vec<Rc<VectorStorage>> = Vec::new();
        for (at, weak) in &self.objects {
            if marked.contains(at) {
                continue;
            }
            if let Some(object) = weak.upgrade() {
                doomed.push(object);
            }
        }
        if doomed.is_empty() {
            return (0, 0);
        }

        let freed_bytes = {
            let mut accounting = Marker {
                managed: &self.objects,
                excluded: Some(marked),
                marked: HashSet::new(),
                walked: HashSet::new(),
                bytes: 0,
                work: Vec::new(),
            };
            for object in &doomed {
                accounting.enqueue(object.clone());
            }
            accounting.drain();
            accounting.bytes
        };

        self.objects.retain(|at, _| marked.contains(at));
        for object in &doomed {
            if let Ok(mut elements) = object.elements.try_borrow_mut() {
                elements.clear();
            }
        }
        let freed_objects = doomed.len() as u64;
        drop(doomed);
        (freed_objects, freed_bytes)
    }
}

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

/// What one mark phase found: which objects are live, and how much storage the
/// values it reached hold.
struct LiveSet {
    marked: HashSet<usize>,
    bytes: u64,
}

/// One shared allocation the collector saw, and how completely it saw it.
struct Sighting {
    /// A handle to walk it again in the mark phase, taken the first time it
    /// was seen.
    held: Value,
    /// References to it the collector could read.
    sighted: usize,
    /// References to it that exist, read before the handle above was taken.
    strong: usize,
}

/// Counts the references to every shared allocation the collector can reach.
struct Scan {
    seen: HashMap<usize, Sighting>,
}

impl Scan {
    /// Records one reference to the allocation at `at`, and reports whether
    /// this was the first time the collector saw it — which is when its
    /// contents still need walking.
    fn sight(&mut self, at: usize, strong: usize, held: impl FnOnce() -> Value) -> bool {
        match self.seen.get_mut(&at) {
            Some(sighting) => {
                sighting.sighted += 1;
                false
            }
            None => {
                self.seen.insert(
                    at,
                    Sighting {
                        held: held(),
                        sighted: 1,
                        strong,
                    },
                );
                true
            }
        }
    }

    /// Registers a managed object without counting a reference to it: the
    /// heap's own table is a `Weak`, not a reference the program holds.
    fn observe(&mut self, object: &Rc<VectorStorage>) {
        let at = Rc::as_ptr(object) as usize;
        if let std::collections::hash_map::Entry::Vacant(slot) = self.seen.entry(at) {
            slot.insert(Sighting {
                held: Value(Repr::Vector(object.clone())),
                sighted: 0,
                // A managed object's real count is snapshotted by
                // `Heap::collect` before anything upgrades a handle; this
                // one is never used for a managed object.
                strong: 0,
            });
        }
    }

    /// Counts one reference for every shared allocation `value` names, and
    /// walks the contents of each the first time it is seen.
    ///
    /// A managed object's contents are not walked here: the heap's table
    /// enumerates every object, so walking from a reference as well would
    /// count its outgoing references twice.
    fn count(&mut self, value: &Value) {
        match value {
            Value(Repr::Vector(storage)) => {
                self.sight(
                    Rc::as_ptr(storage) as usize,
                    Rc::strong_count(storage),
                    || value.clone(),
                );
            }
            Value(Repr::Array(items)) => {
                if self.sight(array_addr(items), Rc::strong_count(items), || value.clone()) {
                    for item in items.iter() {
                        self.count(item);
                    }
                }
            }
            Value(Repr::Map(entries)) => {
                if self.sight(
                    Rc::as_ptr(entries) as usize,
                    Rc::strong_count(entries),
                    || value.clone(),
                ) {
                    for entry in entries.values() {
                        self.count(entry);
                    }
                }
            }
            Value(Repr::Closure(closure)) => {
                if self.sight(
                    Rc::as_ptr(closure) as usize,
                    Rc::strong_count(closure),
                    || value.clone(),
                ) {
                    for (_, captured) in &closure.captures {
                        self.count(captured);
                    }
                }
            }
            Value(Repr::Dyn(wrapped)) => {
                if self.sight(
                    Rc::as_ptr(wrapped) as usize,
                    Rc::strong_count(wrapped),
                    || value.clone(),
                ) {
                    self.count(&wrapped.value);
                }
            }
            Value(Repr::Task(task)) => self.count_task(task),
            Value(Repr::TaskScope(scope)) => {
                if self.sight(Rc::as_ptr(scope) as usize, Rc::strong_count(scope), || {
                    value.clone()
                }) {
                    if let Ok(tasks) = scope.tasks.try_borrow() {
                        for task in tasks.iter() {
                            self.count_task(task);
                        }
                    };
                }
            }
            // A `Struct` is an `Rc` — it has been since issue #104, so that a
            // non-mutating method call would stop copying its receiver — so
            // two paths can reach one, and its fields are walked on the first
            // sighting only. Walking them once per path would count every
            // reference inside it twice, and a reference counted twice is a
            // shortfall concealed.
            Value(Repr::Struct(structure)) => {
                if self.sight(
                    Rc::as_ptr(structure) as usize,
                    Rc::strong_count(structure),
                    || value.clone(),
                ) {
                    for (_, field) in &structure.fields {
                        self.count(field);
                    }
                }
            }
            // An `Enum` is `Box`ed, so it is owned by exactly one value and
            // no two paths reach the same one. That is still true of the
            // payload issue #183 moved inside the box: a `value::Payload`
            // owns its values outright, whether they sit in the `EnumValue`
            // or in the slice a longer one points at, so walking it yields
            // each reference exactly once — which is the property this
            // module's documentation says the whole accounting rests on.
            Value(Repr::Enum(enumeration)) => {
                for item in &enumeration.payload {
                    self.count(item);
                }
            }
            // A `Shared` is a leaf. Its cell holds a `Transfer`, which no
            // `Vector` can be part of, so no reference to a managed object
            // hides in one — and reading it would mean taking a lock the
            // collecting thread may already hold.
            Value(Repr::Shared(_)) => {}
            // A `Set`'s elements are `MapKey`s, which no mutable handle can
            // be, so no reference hides in one. A resource handle is a name
            // the host resolves, so it owns no Cove object either. A bound
            // host operation is an `Rc` — it has been since issue #114 — but
            // of two names rather than of any `Value`, so nothing hides
            // behind that pointer either. Every remaining case is a scalar, a
            // string, or a range.
            _ => {}
        }
    }

    fn count_task(&mut self, task: &Rc<Task>) {
        let first = self.sight(Rc::as_ptr(task) as usize, Rc::strong_count(task), || {
            Value(Repr::Task(Rc::clone(task)))
        });
        if !first {
            return;
        }
        // A task's body went to its own thread as a `Transfer` and is not
        // reachable from the handle, so the value it settled with is all a
        // handle holds.
        if let Ok(state) = task.state.try_borrow() {
            if let crate::task::TaskState::Settled(value) = &*state {
                self.count(value);
            }
        };
    }
}

/// Marks a live set and measures it.
struct Marker<'h> {
    /// The objects this heap manages. A vector allocated elsewhere — by a
    /// host, or by a test building a value directly — is not this heap's to
    /// mark or to free.
    managed: &'h HashMap<usize, Weak<VectorStorage>>,
    /// Objects to stop at, used when measuring what a sweep released: the
    /// survivors are not part of what was freed.
    excluded: Option<&'h HashSet<usize>>,
    marked: HashSet<usize>,
    walked: HashSet<usize>,
    bytes: u64,
    work: Vec<Rc<VectorStorage>>,
}

impl Marker<'_> {
    /// Marks `object` and queues its contents.
    fn enqueue(&mut self, object: Rc<VectorStorage>) {
        let at = Rc::as_ptr(&object) as usize;
        if self.excluded.is_some_and(|set| set.contains(&at)) {
            return;
        }
        if self.marked.insert(at) {
            self.work.push(object);
        }
    }

    /// Walks the queue until nothing is left.
    ///
    /// Managed objects go through this queue rather than through recursion,
    /// because a chain of vectors is as long as the program made it and
    /// recursion over one would be bounded by the native stack.
    fn drain(&mut self) {
        while let Some(object) = self.work.pop() {
            self.bytes += object_bytes(&object);
            // An object whose elements are borrowed right now cannot be read,
            // and does not need to be: its references went uncounted, so
            // everything it holds is already a root.
            if let Ok(elements) = object.elements.try_borrow() {
                for element in elements.iter() {
                    self.visit(element);
                }
            }
        }
    }

    /// Marks every managed object `value` reaches, and adds the storage
    /// `value` itself holds to the live total.
    fn visit(&mut self, value: &Value) {
        match value {
            // A vector this heap does not manage — another task's, or one
            // built outside the interpreter — is neither its to mark nor its
            // to measure.
            Value(Repr::Vector(storage))
                if self.managed.contains_key(&(Rc::as_ptr(storage) as usize)) =>
            {
                self.enqueue(storage.clone());
            }
            Value(Repr::Str(text)) => {
                if self.walked.insert(text.as_ptr() as usize) {
                    self.bytes += text.len() as u64;
                }
            }
            Value(Repr::Array(items)) => {
                if self.walked.insert(array_addr(items)) {
                    self.bytes += (items.len() * size_of::<Value>()) as u64;
                    for item in items.iter() {
                        self.visit(item);
                    }
                }
            }
            Value(Repr::Map(entries)) => {
                if self.walked.insert(Rc::as_ptr(entries) as usize) {
                    for (key, entry) in entries.iter() {
                        self.bytes += key_bytes(key) + size_of::<Value>() as u64;
                        self.visit(entry);
                    }
                }
            }
            Value(Repr::Set(items)) => {
                if self.walked.insert(Rc::as_ptr(items) as usize) {
                    for item in items.iter() {
                        self.bytes += key_bytes(item);
                    }
                }
            }
            // A `Struct` is an `Rc`, so two paths can reach the same one;
            // walk its fields on the first sighting only, exactly as every
            // other `Rc` container above does. Without the guard, a struct
            // reached twice has its header and its field names added to
            // `live_bytes` twice, and a chain of structs each holding two
            // references to the one below is walked in time exponential in
            // the chain's length.
            Value(Repr::Struct(structure)) => {
                if self.walked.insert(Rc::as_ptr(structure) as usize) {
                    self.bytes += size_of::<crate::value::StructValue>() as u64;
                    for (name, field) in &structure.fields {
                        self.bytes += (name.len() + size_of::<Value>()) as u64;
                        self.visit(field);
                    }
                }
            }
            // An `Enum` is `Box`ed, so it is owned by exactly one value and
            // no two paths reach the same one.
            //
            // Its payload is a `value::Payload`, which since issue #183
            // holds the arities that occur — none and one — inside the
            // `EnumValue` itself and allocates only from two upwards. So the
            // slots are charged by asking the payload where they live rather
            // than by counting them: an inline slot is already inside the
            // `size_of::<EnumValue>()` charged above, and charging it again
            // would say a `Some(x)` costs a `Value` more than it does.
            Value(Repr::Enum(enumeration)) => {
                self.bytes += size_of::<crate::value::EnumValue>() as u64;
                if let crate::value::Payload::Many(items) = &enumeration.payload {
                    self.bytes += (items.len() * size_of::<Value>()) as u64;
                }
                for item in &enumeration.payload {
                    self.visit(item);
                }
            }
            Value(Repr::Closure(closure)) => {
                if self.walked.insert(Rc::as_ptr(closure) as usize) {
                    self.bytes += size_of::<crate::value::Closure>() as u64;
                    for (name, captured) in &closure.captures {
                        self.bytes += (name.len() + size_of::<Value>()) as u64;
                        self.visit(captured);
                    }
                }
            }
            Value(Repr::Dyn(wrapped)) => {
                if self.walked.insert(Rc::as_ptr(wrapped) as usize) {
                    self.bytes += size_of::<crate::value::DynValue>() as u64;
                    self.visit(&wrapped.value);
                }
            }
            Value(Repr::Task(task)) => self.visit_task(task),
            // A `Shared`'s contents belong to the cell, not to this task, so
            // they are neither marked nor measured here; see this module's
            // documentation for why the lock is never taken.
            Value(Repr::Shared(_)) => {}
            Value(Repr::TaskScope(scope)) if self.walked.insert(Rc::as_ptr(scope) as usize) => {
                self.bytes += size_of::<TaskScope>() as u64;
                // A scope this thread is mid-way through mutating cannot be
                // read, so its tasks go unsighted and the shortfall rule
                // roots them.
                if let Ok(tasks) = scope.tasks.try_borrow() {
                    for task in tasks.iter() {
                        self.visit_task(task);
                    }
                }
            }
            _ => {}
        }
    }

    fn visit_task(&mut self, task: &Rc<Task>) {
        if !self.walked.insert(Rc::as_ptr(task) as usize) {
            return;
        }
        self.bytes += size_of::<Task>() as u64;
        if let Ok(state) = task.state.try_borrow() {
            if let crate::task::TaskState::Settled(value) = &*state {
                self.visit(value);
            }
        }
    }
}

/// The address of an array's shared storage.
///
/// `Rc<[Value]>` is a fat pointer, so it is narrowed to the address of its
/// first element, which identifies the allocation just as well.
fn array_addr(items: &Rc<[Value]>) -> usize {
    Rc::as_ptr(items) as *const Value as usize
}

/// The storage one object holds for itself: its header and its element slots.
///
/// What each element points at is counted separately, once, however many
/// elements point at it.
fn object_bytes(storage: &VectorStorage) -> u64 {
    let elements = storage
        .elements
        .try_borrow()
        .map(|elements| elements.len())
        .unwrap_or(0);
    (size_of::<VectorStorage>() + elements * size_of::<Value>()) as u64
}

/// The storage a map key or set element holds.
fn key_bytes(key: &MapKey) -> u64 {
    let own = size_of::<MapKey>() as u64;
    own + match key {
        MapKey::Str(text) => text.len() as u64,
        MapKey::EnumCase(type_name, case, payload) => {
            (type_name.len() + case.len()) as u64 + payload.iter().map(key_bytes).sum::<u64>()
        }
        MapKey::Struct(type_name, fields, _) => {
            type_name.len() as u64
                + fields
                    .iter()
                    .map(|(name, field)| name.len() as u64 + key_bytes(field))
                    .sum::<u64>()
        }
        MapKey::Array(items) => items.iter().map(key_bytes).sum(),
        MapKey::Set(items) => items.iter().map(key_bytes).sum(),
        MapKey::Map(entries) => entries
            .iter()
            .map(|(key, value)| key_bytes(key) + key_bytes(value))
            .sum(),
        _ => 0,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::value::StructValue;

    /// Registers `value` as a binding and returns the slot, so a test can drop
    /// the root later.
    fn root(roots: &mut SlotRoots, value: Value) -> Slot {
        let slot = Rc::new(RefCell::new(value));
        roots.push(slot.clone());
        slot
    }

    #[test]
    fn an_unreachable_object_is_reclaimed() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        let storage = heap.allocate(vec![Value(Repr::Int(1))]);
        drop(storage);
        let collected = heap.collect(&roots);
        // `Rc` already freed it, so there was nothing left for the sweep to
        // free; the heap still stops tracking it.
        assert_eq!(collected.live_objects, 0);
    }

    #[test]
    fn a_reachable_object_survives() {
        let mut roots = SlotRoots::new();
        let mut heap = Heap::new();
        let storage = heap.allocate(vec![Value(Repr::Int(1))]);
        let _slot = root(&mut roots, Value(Repr::Vector(storage)));
        let collected = heap.collect(&roots);
        assert_eq!(collected.live_objects, 1);
        assert_eq!(collected.freed_objects, 0);
    }

    /// The whole reason for the collector: two objects that point at each
    /// other keep each other's reference count above zero forever.
    #[test]
    fn a_cycle_is_reclaimed() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        let a = heap.allocate(Vec::new());
        let b = heap.allocate(Vec::new());
        a.elements.borrow_mut().push(Value(Repr::Vector(b.clone())));
        b.elements.borrow_mut().push(Value(Repr::Vector(a.clone())));
        let weak = Rc::downgrade(&a);
        drop(a);
        drop(b);

        assert!(weak.upgrade().is_some(), "`Rc` cannot free a cycle");
        let collected = heap.collect(&roots);
        assert_eq!(collected.freed_objects, 2);
        assert_eq!(collected.live_objects, 0);
        assert!(weak.upgrade().is_none(), "the cycle was not freed");
    }

    #[test]
    fn a_reachable_cycle_survives() {
        let mut roots = SlotRoots::new();
        let mut heap = Heap::new();
        let a = heap.allocate(Vec::new());
        a.elements.borrow_mut().push(Value(Repr::Vector(a.clone())));
        let _slot = root(&mut roots, Value(Repr::Vector(a.clone())));
        drop(a);
        let collected = heap.collect(&roots);
        assert_eq!(collected.freed_objects, 0);
        assert_eq!(collected.live_objects, 1);
    }

    /// A cycle whose back edge runs through a struct field is still a cycle.
    #[test]
    fn a_cycle_through_a_struct_field_is_reclaimed() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        let object = heap.allocate(Vec::new());
        object
            .elements
            .borrow_mut()
            .push(Value(Repr::Struct(Rc::new(StructValue {
                type_name: "test.Node".into(),
                fields: vec![("next".into(), Value(Repr::Vector(object.clone())))],
                opaque: false,
            }))));
        let weak = Rc::downgrade(&object);
        drop(object);
        assert!(weak.upgrade().is_some());
        assert_eq!(heap.collect(&roots).freed_objects, 1);
        assert!(weak.upgrade().is_none());
    }

    /// An object held only by a value the collector cannot read — an
    /// evaluator temporary, here modelled by a plain Rust local — is a root,
    /// found by comparing the references the collector can see with the
    /// reference count.
    #[test]
    fn an_object_held_only_by_a_temporary_is_a_root() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        let held = heap.allocate(vec![Value(Repr::Int(1))]);
        let collected = heap.collect(&roots);
        assert_eq!(collected.freed_objects, 0);
        assert_eq!(collected.live_objects, 1);
        assert_eq!(held.elements.borrow().len(), 1, "its contents survived");
    }

    /// The same rule, one level deeper: the temporary holds a container, and
    /// the object is inside it.
    #[test]
    fn an_object_inside_a_temporary_container_is_a_root() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        let inner = heap.allocate(vec![Value(Repr::Int(7))]);
        let weak = Rc::downgrade(&inner);
        let temporary = Value(Repr::Array(vec![Value(Repr::Vector(inner))].into()));
        let collected = heap.collect(&roots);
        assert_eq!(collected.freed_objects, 0);
        assert!(weak.upgrade().is_some());
        drop(temporary);
    }

    /// The subtle case, and the reason the scan counts references to shared
    /// containers and not only to managed objects. `shared` is held by a
    /// garbage cycle *and* by a temporary the collector cannot read. Counting
    /// only the object would find every reference to it accounted for — by
    /// the garbage — and free the one thing the temporary can still reach.
    #[test]
    fn an_object_reached_through_a_container_a_temporary_shares_with_garbage_survives() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        let inner = heap.allocate(vec![Value(Repr::Int(7))]);
        let alive = Rc::downgrade(&inner);
        let shared: Rc<[Value]> = vec![Value(Repr::Vector(inner))].into();

        let a = heap.allocate(Vec::new());
        let b = heap.allocate(Vec::new());
        a.elements.borrow_mut().push(Value(Repr::Vector(b.clone())));
        a.elements
            .borrow_mut()
            .push(Value(Repr::Array(Rc::clone(&shared))));
        b.elements.borrow_mut().push(Value(Repr::Vector(a.clone())));
        let cycle = Rc::downgrade(&a);
        drop(a);
        drop(b);

        let collected = heap.collect(&roots);
        assert!(
            cycle.upgrade().is_none(),
            "the garbage cycle should have gone"
        );
        let survivor = alive
            .upgrade()
            .expect("the temporary's array still holds it");
        // Sweeping clears an object's elements, so this is what a wrong answer
        // looks like: the handle is still there and its contents are gone.
        assert!(
            survivor
                .elements
                .borrow()
                .first()
                .is_some_and(|element| element.eq_value(&Value(Repr::Int(7)))),
            "the sweep emptied a vector something still holds: {collected:?}"
        );
        assert_eq!(shared.len(), 1);
    }

    /// The same rule again, with the container shared *twice* by the garbage
    /// rather than once. A `Value::Struct` is an `Rc`, so one struct value can
    /// be reached from two places; if the scan walked its fields once per
    /// path, the vector inside it would be sighted twice, its two sightings
    /// would account for the two references that exist — the struct's and the
    /// temporary's — and the shortfall that makes the temporary a root would
    /// not fire. The sweep would then empty a vector the temporary still
    /// reaches, which is the failure this whole mechanism exists to prevent.
    #[test]
    fn an_object_inside_a_struct_two_garbage_paths_share_survives() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        // The only two references to this are the struct's field and the
        // local, and the local is what stands for a backend temporary.
        let held = heap.allocate(vec![Value(Repr::Int(7))]);
        let structure = Rc::new(StructValue {
            type_name: "test.Holder".into(),
            fields: vec![("held".into(), Value(Repr::Vector(held.clone())))],
            opaque: false,
        });

        // A garbage cycle whose two members both name that one struct, so the
        // scan reaches the struct twice without the program naming it at all.
        let a = heap.allocate(Vec::new());
        let b = heap.allocate(Vec::new());
        a.elements.borrow_mut().push(Value(Repr::Vector(b.clone())));
        b.elements.borrow_mut().push(Value(Repr::Vector(a.clone())));
        a.elements
            .borrow_mut()
            .push(Value(Repr::Struct(Rc::clone(&structure))));
        b.elements
            .borrow_mut()
            .push(Value(Repr::Struct(Rc::clone(&structure))));
        let cycle = Rc::downgrade(&a);
        drop(structure);
        drop(a);
        drop(b);

        let collected = heap.collect(&roots);
        assert!(
            cycle.upgrade().is_none(),
            "the garbage cycle should have gone: {collected:?}"
        );
        assert!(
            held.elements
                .borrow()
                .first()
                .is_some_and(|element| element.eq_value(&Value(Repr::Int(7)))),
            "the sweep emptied a vector a temporary still holds: {collected:?}"
        );
    }

    #[test]
    fn a_binding_dropped_from_the_roots_is_reclaimed() {
        let mut roots = SlotRoots::new();
        let mut heap = Heap::new();
        let object = heap.allocate(Vec::new());
        let cycle = Value(Repr::Vector(object.clone()));
        object.elements.borrow_mut().push(cycle);
        drop(object);
        let base = roots.len();
        let slot = Rc::new(RefCell::new(Value(Repr::Unit)));
        roots.push(slot.clone());
        drop(slot);
        roots.truncate(base);
        assert_eq!(heap.collect(&roots).freed_objects, 1);
    }

    #[test]
    fn live_bytes_falls_when_a_cycle_is_reclaimed() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        let a = heap.allocate(vec![Value(Repr::Str("a fairly long string".into()))]);
        let b = heap.allocate(Vec::new());
        a.elements.borrow_mut().push(Value(Repr::Vector(b.clone())));
        b.elements.borrow_mut().push(Value(Repr::Vector(a.clone())));

        let before = {
            // With both handles held, the cycle is rooted by the temporaries.
            heap.collect(&roots).live_bytes
        };
        drop(a);
        drop(b);
        let after = heap.collect(&roots).live_bytes;
        assert!(before > 0);
        assert_eq!(after, 0, "live bytes should fall to nothing: {before}");
    }

    /// An object an enum case carries is reachable through the case, on both
    /// arms of `value::Payload`.
    ///
    /// There was no test here that built a `Value::Enum` at all, which is
    /// how a payload's representation could have changed under
    /// `Scan::count` and `Marker::visit` with nothing failing. Issue #183
    /// changed it — a payload of one now lives inside the `EnumValue`
    /// instead of in a vector beside it, and a payload of two or more in a
    /// boxed slice — so both arms are built here and the sweep is asked
    /// about each. Sweeping empties a vector's elements, so a wrong answer
    /// in either walker is a handle that is still there with its contents
    /// gone.
    #[test]
    fn an_object_an_enum_case_carries_is_reachable_through_the_case() {
        for arity in [1usize, 3] {
            let mut roots = SlotRoots::new();
            let mut heap = Heap::new();
            let held = heap.allocate(vec![Value(Repr::Int(7))]);
            let mut payload: Vec<Value> = vec![Value(Repr::Vector(held.clone()))];
            payload.resize(arity, Value(Repr::Unit));
            let _slot = root(
                &mut roots,
                Value(Repr::Enum(Box::new(crate::value::EnumValue {
                    type_name: "test.Carrier".into(),
                    case: "Holds".into(),
                    payload: payload.into(),
                }))),
            );
            let weak = Rc::downgrade(&held);
            drop(held);

            let collected = heap.collect(&roots);
            let survivor = weak
                .upgrade()
                .expect("the enum case still holds the vector");
            assert!(
                survivor
                    .elements
                    .borrow()
                    .first()
                    .is_some_and(|element| element.eq_value(&Value(Repr::Int(7)))),
                "the sweep emptied a vector an enum case of arity {arity} \
                 still reaches: {collected:?}"
            );
            assert_eq!(collected.freed_objects, 0);
        }
    }

    /// A payload of one is charged as the storage it is, and a payload of
    /// three as the storage *it* is.
    ///
    /// `Marker::visit` charges `size_of::<EnumValue>()` for the case and
    /// then, since issue #183, the payload slots only when they are a
    /// `Payload::Many` — because a `Payload::One`'s slot is already inside
    /// the `EnumValue` it just charged, and charging it again would say a
    /// `Some(x)` costs a `Value` more than it does. Nothing else in this
    /// suite reads an enum's byte contribution, and the differential suite
    /// strips `live_bytes` from what it compares, so this is the only place
    /// the arithmetic is stated.
    #[test]
    fn an_enum_case_charges_the_payload_it_actually_allocated() {
        let bytes_for = |arity: usize| {
            let mut roots = SlotRoots::new();
            let mut heap = Heap::new();
            let _slot = root(
                &mut roots,
                Value(Repr::Enum(Box::new(crate::value::EnumValue {
                    type_name: "test.Carrier".into(),
                    case: "Holds".into(),
                    payload: vec![Value(Repr::Int(1)); arity].into(),
                }))),
            );
            heap.collect(&roots).live_bytes
        };
        let case = size_of::<crate::value::EnumValue>() as u64;
        assert_eq!(bytes_for(0), case);
        assert_eq!(bytes_for(1), case);
        assert_eq!(bytes_for(3), case + 3 * size_of::<Value>() as u64);
    }

    /// A closure is charged for its own header, its capture names, and one
    /// `Value` per capture — once, however many live paths reach it.
    ///
    /// Nothing in this suite built a `Value::Closure` at all, which is how
    /// its byte arithmetic could have gone wrong in either walker with
    /// nothing failing. Before ADR 0034, `tests/differential`'s `same_heap`
    /// did not close that gap either: it compared the interpreter against
    /// the predecessor VM rather than against an absolute figure, so an
    /// error made identically by both would have passed unseen there too.
    /// There is no such comparison to lean on today — the linear-memory
    /// backend counts bytes of a heap this one's arithmetic has nothing to
    /// do with — so the figure is stated here, absolutely, the way
    /// `an_enum_case_charges_the_payload_it_actually_allocated` states the
    /// enum's.
    ///
    /// Two roots rather than one because the closure arm's `walked` guard is
    /// what keeps a shared closure from being charged twice, and a guard
    /// nothing exercises is a guard that can be lost by accident.
    #[test]
    fn a_closure_charges_its_captures_once_however_many_paths_reach_it() {
        let mut roots = SlotRoots::new();
        let mut heap = Heap::new();
        let closure = Rc::new(crate::value::Closure {
            is_async: false,
            arity: 1,
            body: crate::value::ClosureBody::Tree {
                params: Vec::new(),
                block: std::sync::Arc::new(cove_syntax::ast::Block {
                    statements: Vec::new(),
                    tail: None,
                    span: cove_diag::Span::new(cove_diag::FileId(0), 0, 0),
                }),
                decl: None,
            },
            module: "test".into(),
            captures: vec![
                ("species".into(), Value(Repr::Int(7))),
                ("world".into(), Value(Repr::Unit)),
            ],
        });
        let _a = root(&mut roots, Value(Repr::Closure(Rc::clone(&closure))));
        let _b = root(&mut roots, Value(Repr::Closure(Rc::clone(&closure))));
        // The same reason the struct test below drops its local: an unrooted
        // binding is a temporary, and the shortfall rule would rightly treat
        // it as a third live path.
        drop(closure);

        let expected = size_of::<crate::value::Closure>() as u64
            + ("species".len() + size_of::<Value>()) as u64
            + ("world".len() + size_of::<Value>()) as u64;
        assert_eq!(
            heap.collect(&roots).live_bytes,
            expected,
            "a closure reached from two roots should be charged once, for its \
             header and its captures"
        );
    }

    /// An object a closure captured is reachable through the capture.
    ///
    /// `Scan::count`'s closure arm is what says so, and the sweep is what
    /// asks: a wrong answer there is a `Vector` handle the program still
    /// holds with its elements emptied out from under it. This is the shape
    /// `examples/life`'s `population()` has — a closure over one captured
    /// value, handed to `filter` — with the capture made collectable so that
    /// there is something for the collector to get wrong.
    #[test]
    fn an_object_a_closure_captured_is_reachable_through_the_capture() {
        let mut roots = SlotRoots::new();
        let mut heap = Heap::new();
        let held = heap.allocate(vec![Value(Repr::Int(7))]);
        let _slot = root(
            &mut roots,
            Value(Repr::Closure(Rc::new(crate::value::Closure {
                is_async: false,
                arity: 1,
                body: crate::value::ClosureBody::Tree {
                    params: Vec::new(),
                    block: std::sync::Arc::new(cove_syntax::ast::Block {
                        statements: Vec::new(),
                        tail: None,
                        span: cove_diag::Span::new(cove_diag::FileId(0), 0, 0),
                    }),
                    decl: None,
                },
                module: "test".into(),
                captures: vec![("held".into(), Value(Repr::Vector(held.clone())))],
            }))),
        );
        let weak = Rc::downgrade(&held);
        drop(held);

        let collected = heap.collect(&roots);
        let survivor = weak.upgrade().expect("the closure still holds the vector");
        assert!(
            survivor
                .elements
                .borrow()
                .first()
                .is_some_and(|element| element.eq_value(&Value(Repr::Int(7)))),
            "the sweep emptied a vector a closure capture still reaches: {collected:?}"
        );
        assert_eq!(collected.freed_objects, 0);
    }

    /// `Marker::visit`'s struct arm has no `walked` guard, unlike every other
    /// container it walks, so a struct reached from two live roots has its
    /// header and its field names added to `live_bytes` twice. That
    /// contradicts `Heap::live_bytes`'s own documented accounting: "each
    /// shared allocation counted once."
    #[test]
    fn a_struct_shared_by_two_live_paths_reports_its_bytes_once() {
        let mut roots = SlotRoots::new();
        let mut heap = Heap::new();
        let structure = Rc::new(StructValue {
            type_name: "test.Shared".into(),
            fields: vec![("value".into(), Value(Repr::Int(7)))],
            opaque: false,
        });
        let _a = root(&mut roots, Value(Repr::Struct(Rc::clone(&structure))));
        let _b = root(&mut roots, Value(Repr::Struct(Rc::clone(&structure))));
        // Drop the local so the only two references left are the roots' —
        // otherwise this binding is itself an unrooted temporary, and the
        // shortfall rule would (correctly) treat it as a third live path.
        drop(structure);

        let collected = heap.collect(&roots);
        let expected =
            size_of::<StructValue>() as u64 + ("value".len() + size_of::<Value>()) as u64;
        assert_eq!(
            collected.live_bytes, expected,
            "a struct reached from two roots should be counted once, not once per root"
        );
    }

    /// The same missing guard makes the walk exponential rather than merely
    /// wrong. A chain of struct values, each holding two references to the
    /// one below, is walked once per *path* to a struct rather than once per
    /// struct when nothing deduplicates it — so reaching depth `N` costs
    /// `2^N` visits to the bottom of the chain.
    ///
    /// Depth 30 is chosen so the two shapes are unmistakable rather than
    /// merely different: a walk that visits each of the 31 structs once, as
    /// the fix does, finishes in microseconds regardless of depth, while one
    /// that revisits every shared struct along every path needs on the order
    /// of `2^30` (roughly a billion) visits to the deepest struct alone. That
    /// is far past anything a slow-but-linear walk could rack up by
    /// accident, so a timeout firing here can only mean the exponential
    /// shape came back, not that the machine running the suite is loaded.
    ///
    /// The whole heap, and every `Rc` it touches, is built on a spawned
    /// thread and never leaves it — only the resulting byte count, a `u64`,
    /// crosses back over the channel — because `Rc` is not `Send` and cannot
    /// be moved across the boundary a wall-clock bound needs. Not joining
    /// that thread is deliberate: if the walk is exponential again, the
    /// `recv_timeout` below still reports the failure promptly, and the
    /// still-running thread is simply killed when the test process exits,
    /// rather than hanging the suite.
    #[test]
    fn a_chain_of_shared_structs_is_walked_without_exponential_blowup() {
        const DEPTH: usize = 30;
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let mut roots = SlotRoots::new();
            let mut heap = Heap::new();

            let mut current = Value(Repr::Struct(Rc::new(StructValue {
                type_name: "test.Level0".into(),
                fields: Vec::new(),
                opaque: false,
            })));
            for level in 1..=DEPTH {
                current = Value(Repr::Struct(Rc::new(StructValue {
                    type_name: format!("test.Level{level}").into(),
                    fields: vec![("a".into(), current.clone()), ("b".into(), current)],
                    opaque: false,
                })));
            }
            let _slot = root(&mut roots, current);

            let collected = heap.collect(&roots);
            // The sender is dropped along with this thread whether or not the
            // receiver is still listening; a failed send just means the main
            // thread already gave up and timed out.
            let _ = tx.send(collected.live_bytes);
        });

        let live_bytes = rx.recv_timeout(std::time::Duration::from_secs(10)).expect(
            "collection did not finish in time — the struct arm is walking \
                 the chain exponentially again",
        );

        let mut expected = size_of::<StructValue>() as u64;
        for _ in 1..=DEPTH {
            expected += size_of::<StructValue>() as u64 + 2 * (1 + size_of::<Value>()) as u64;
        }
        assert_eq!(
            live_bytes, expected,
            "each struct in the chain should be counted once, not once per path to it"
        );
    }

    /// A closure's own bytes, its captures' names and slots, and what the
    /// captures reach — counted once however many live paths reach the
    /// closure.
    ///
    /// `Marker::visit`'s closure arm has the `walked` guard the struct arm
    /// was missing, so this passes today. It is here because nothing checked
    /// it: `Value::Closure` was reached by exactly one test in this module
    /// and that test asks about survival rather than about bytes, and before
    /// ADR 0034 the predecessor's own `vm::tests::heap`'s `same_heap` compared
    /// the interpreter against it rather than against an absolute, so an
    /// error made identically on both would have gone unseen there too.
    #[test]
    fn a_closure_shared_by_two_live_paths_reports_its_bytes_once() {
        let mut roots = SlotRoots::new();
        let mut heap = Heap::new();
        let closure = Rc::new(crate::value::Closure {
            is_async: false,
            arity: 1,
            body: crate::value::ClosureBody::Tree {
                params: Vec::new(),
                block: std::sync::Arc::new(cove_syntax::ast::Block {
                    statements: Vec::new(),
                    tail: None,
                    span: cove_diag::Span::new(cove_diag::FileId(0), 0, 0),
                }),
                decl: None,
            },
            module: "test".into(),
            captures: vec![("held".into(), Value::string("seven"))],
        });
        let _a = root(&mut roots, Value(Repr::Closure(Rc::clone(&closure))));
        let _b = root(&mut roots, Value(Repr::Closure(Rc::clone(&closure))));
        // As in the struct case above: the local would otherwise be an
        // unrooted temporary, and the shortfall rule would correctly make it
        // a third live path.
        drop(closure);

        let collected = heap.collect(&roots);
        let expected = size_of::<crate::value::Closure>() as u64
            + ("held".len() + size_of::<Value>()) as u64
            + "seven".len() as u64;
        assert_eq!(
            collected.live_bytes, expected,
            "a closure reached from two roots should be counted once, not once per root"
        );
    }

    /// An enum case is billed for the box it is, plus the payload slots it
    /// actually allocates — and a payload of one allocates none.
    ///
    /// Issue #183 put the arities an ordinary program builds inside the
    /// `EnumValue` itself, so a `Some(x)` is one allocation rather than two,
    /// and `Marker::visit` charges only a `Payload::Many` for its slice.
    /// ADR 0028 decision 2 says the prototype records whether a heap
    /// representation gives that win back before it becomes the default,
    /// which needs the present figure written down rather than assumed.
    #[test]
    fn an_enum_case_is_billed_for_the_payload_it_actually_allocates() {
        let one = {
            let mut roots = SlotRoots::new();
            let mut heap = Heap::new();
            let _slot = root(
                &mut roots,
                Value(Repr::Enum(Box::new(crate::value::EnumValue {
                    type_name: "Option".into(),
                    case: "Some".into(),
                    payload: crate::value::Payload::One(Value::string("seven")),
                }))),
            );
            heap.collect(&roots).live_bytes
        };
        assert_eq!(
            one,
            size_of::<crate::value::EnumValue>() as u64 + "seven".len() as u64,
            "a payload of one lives in the box that already exists"
        );

        let many = {
            let mut roots = SlotRoots::new();
            let mut heap = Heap::new();
            let _slot = root(
                &mut roots,
                Value(Repr::Enum(Box::new(crate::value::EnumValue {
                    type_name: "test.Pair".into(),
                    case: "Both".into(),
                    payload: crate::value::Payload::Many(
                        vec![Value::string("seven"), Value::string("eight")].into(),
                    ),
                }))),
            );
            heap.collect(&roots).live_bytes
        };
        assert_eq!(
            many,
            size_of::<crate::value::EnumValue>() as u64
                + 2 * size_of::<Value>() as u64
                + ("seven".len() + "eight".len()) as u64,
            "a payload of two or more is a slice beside the box"
        );
    }

    #[test]
    fn collections_are_spaced_by_allocation() {
        let mut heap = Heap::new();
        assert!(!heap.should_collect());
        for _ in 0..MIN_ALLOCATIONS_BETWEEN_COLLECTIONS {
            let _ = heap.allocate(Vec::new());
        }
        assert!(heap.should_collect());
    }

    #[test]
    fn stats_accumulate_over_collections() {
        let roots = SlotRoots::new();
        let mut heap = Heap::new();
        let object = heap.allocate(Vec::new());
        let cycle = Value(Repr::Vector(object.clone()));
        object.elements.borrow_mut().push(cycle);
        drop(object);
        heap.collect(&roots);
        heap.collect(&roots);
        let stats = heap.stats();
        assert_eq!(stats.allocated_objects, 1);
        assert_eq!(stats.collections, 2);
        assert_eq!(stats.freed_objects, 1);
    }
}