minerva 0.2.0

//! The paged occupancy plane (S205): the visible plane's possession-diff
//! shell.
//!
//! The collation's possession scans cost their representation, not the
//! change (ruling R-30's correction; alma COLLAB-9's ~135 us per churned
//! keystroke): `DotSet::difference` pays a membership probe per retained
//! exception, and a churned document's visible plane is nearly all
//! exceptions above a punched floor. This plane re-realizes the same set
//! for diff cost, in the substructure program's one factoring: per
//! `(station, page-of-64)`, a `u64` occupancy mask, so comparing two
//! states of the plane is one sorted walk of `O(pages)` word compares and
//! the changed dots fall out of the XOR as bits. Content-exact for ANY
//! pair by construction (masks are compared, never counters or journal
//! claims), so the collation's totality contract needs no narrowing here.
//!
//! A derived coordinate under the shell discipline: the [`DotSet`] stays
//! the visible plane's carried form and the survivor law's store, this
//! plane is maintained beside it on the `&mut` write paths (C8, no
//! interior mutability), excluded from the store's equality and hash, and
//! never on the wire. Canonical form: no zero masks retained and keys
//! strictly ascending, so equal sets spell equal planes.
//!
//! [`DotSet`]: crate::metis::DotSet

extern crate alloc;

use alloc::collections::BTreeMap;
use alloc::vec::Vec;
use core::num::NonZeroU64;

use super::placement::Dot;

/// One page key: a station and its 64-dot page (`dot / 64`).
pub(super) type PageKey = (u32, u64);

/// Dots per occupancy page: the crate's one factoring (the identity
/// plane's page width, the thread's fragment cap, the wire's run economy).
const PAGE: u64 = 64;

/// The paged occupancy plane: per station, its pages ascending (the
/// fiber idiom the identity plane already uses), an absent page meaning
/// the zero mask. Station partitioning is load-bearing (the gate
/// review's cross-station finding): a page birth shifts rows only
/// within its own station's fiber, and a station's own pages are born
/// in mint order, so the common birth is a tail append; a low-keyed
/// station writing into a document populated by high-keyed stations
/// never moves the other fibers.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct OccupancyPlane {
    /// Strictly ascending pages per station; no zero masks and no empty
    /// stations retained.
    stations: BTreeMap<u32, Vec<(u64, u64)>>,
}

impl OccupancyPlane {
    /// The empty plane.
    pub(super) const fn new() -> Self {
        Self {
            stations: BTreeMap::new(),
        }
    }

    /// Nonzero page rows across all station fibers.
    #[cfg(feature = "instrumentation")]
    pub(super) fn page_count(&self) -> usize {
        self.stations.values().map(Vec::len).sum()
    }

    /// Station fibers retaining at least one nonzero page.
    #[cfg(feature = "instrumentation")]
    pub(super) fn station_count(&self) -> usize {
        self.stations.len()
    }

    /// The page and bit of one dot. Dot `0` is the non-dot and never
    /// enters (its bit in page zero stays permanently clear).
    const fn page_bit(index: u64) -> (u64, u64) {
        (index / PAGE, 1u64 << (index % PAGE))
    }

    /// Sets or clears one dot's bit, keeping the canonical form (a page
    /// is born on its first set bit and retires with its last; a station
    /// retires with its last page). `O(log)` search plus a row shift
    /// bounded by the ONE station's fiber.
    pub(super) fn set(&mut self, station: u32, index: u64, present: bool) {
        let (page, bit) = Self::page_bit(index);
        if present {
            let rows = self.stations.entry(station).or_default();
            match rows.binary_search_by_key(&page, |&(held, _)| held) {
                Ok(at) => rows[at].1 |= bit,
                Err(at) => rows.insert(at, (page, bit)),
            }
        } else if let Some(rows) = self.stations.get_mut(&station)
            && let Ok(at) = rows.binary_search_by_key(&page, |&(held, _)| held)
        {
            rows[at].1 &= !bit;
            if rows[at].1 == 0 {
                let _ = rows.remove(at);
                if rows.is_empty() {
                    let _ = self.stations.remove(&station);
                }
            }
        }
    }

    /// Sets a contiguous dot run `first .. first + len` of one station in
    /// `O(pages)` staging (full interior pages stage one all-ones mask
    /// each, the two partial edges OR into what the plane already holds)
    /// plus one [`apply_pages`](Self::apply_pages) application: the
    /// bulk-ingest presence form of [`set`](Self::set), agreeing with it
    /// exactly. A run truncates at the `u64::MAX` dot ceiling (the
    /// per-dot form can name nothing past it), and the non-dot `0` never
    /// enters.
    pub(super) fn set_run(&mut self, station: u32, first: u64, len: u32) {
        if len == 0 || first.max(1) > first.saturating_add(u64::from(len) - 1) {
            return;
        }
        let start = first.max(1);
        let last = first.saturating_add(u64::from(len) - 1);
        let (first_page, last_page) = (start / PAGE, last / PAGE);
        let mut staged: Vec<(PageKey, u64)> = Vec::new();
        let mut page = first_page;
        loop {
            // The run's bits within this page: all 64, unless an edge trims.
            let low = if page == first_page { start % PAGE } else { 0 };
            let high = if page == last_page {
                last % PAGE
            } else {
                PAGE - 1
            };
            let mask = if (high - low) == PAGE - 1 {
                u64::MAX
            } else {
                ((1u64 << (high - low + 1)) - 1) << low
            };
            let current = self
                .stations
                .get(&station)
                .and_then(|rows| {
                    rows.binary_search_by_key(&page, |&(held, _)| held)
                        .ok()
                        .map(|slot| rows[slot].1)
                })
                .unwrap_or(0);
            staged.push(((station, page), current | mask));
            if page == last_page {
                break;
            }
            page += 1;
        }
        self.apply_pages(&staged);
    }

    /// Builds the plane from an ascending page walk (the
    /// [`DotSet::occupancy_pages`](crate::metis::DotSet) read, computed
    /// off the compact form in `O(pages + exceptions)`, never a per-dot
    /// walk of the prefix): the whole-store construction path. Adjacent
    /// rows sharing a key coalesce (the walk's floor/exception boundary).
    pub(super) fn from_pages(pages: impl Iterator<Item = (PageKey, u64)>) -> Self {
        let mut stations: BTreeMap<u32, Vec<(u64, u64)>> = BTreeMap::new();
        for ((station, page), mask) in pages {
            debug_assert!(mask != 0, "the page walk carries no zero masks");
            let rows = stations.entry(station).or_default();
            match rows.last_mut() {
                Some((held, bits)) if *held == page => *bits |= mask,
                _ => {
                    debug_assert!(
                        rows.last().is_none_or(|&(held, _)| held < page),
                        "the page walk ascends"
                    );
                    rows.push((page, mask));
                }
            }
        }
        Self { stations }
    }

    /// Applies a sorted batch of whole page rows (the collation's mark
    /// refresh and the performed store's flip settle). Per station, two
    /// regimes keep the cost linear in every shape: while every change
    /// swaps the mask of a row that exists and stays alive, each is one
    /// binary-search point write; the first page birth or death in that
    /// station switches its group to ONE linear merge of the fiber, so
    /// retiring `P` pages costs `O(fiber rows + changes)`, never
    /// `O(P x rows)` (the gate review's quadratic finding), and no
    /// station's changes ever shift another station's fiber (its
    /// cross-station sequel).
    pub(super) fn apply_pages(&mut self, changes: &[(PageKey, u64)]) {
        debug_assert!(
            changes.windows(2).all(|pair| pair[0].0 < pair[1].0),
            "the change batch ascends strictly"
        );
        let mut at = 0;
        while at < changes.len() {
            let station = changes[at].0.0;
            let end = at
                + changes[at..]
                    .iter()
                    .take_while(|&&((held, _), _)| held == station)
                    .count();
            self.apply_station_pages(station, &changes[at..end]);
            at = end;
        }
    }

    /// One station's batch, the two-regime rule over its own fiber.
    fn apply_station_pages(&mut self, station: u32, changes: &[(PageKey, u64)]) {
        let rows = self.stations.entry(station).or_default();
        // Incremental shapes stay in place: an alive row's mask swaps by
        // point write, and a birth BEYOND the fiber's tail is a push
        // (the forward-minted delivery, one page birth per 64 dots,
        // which a whole-fiber rebuild would turn into cumulative
        // quadratic copying: the gate review's singleton-delta finding).
        // Beyond-tail births ascend with the batch, so each classifies
        // against the unmodified fiber's end and the application pushes
        // in order. Only interior restructuring (an interior birth, any
        // death) takes the linear merge, whose cost then amortizes
        // against the change that forced it.
        let incremental = changes.iter().all(|&((_, page), mask)| {
            mask != 0
                && match rows.binary_search_by_key(&page, |&(held, _)| held) {
                    Ok(_) => true,
                    Err(at) => at == rows.len(),
                }
        });
        if incremental {
            for &((_, page), mask) in changes {
                match rows.binary_search_by_key(&page, |&(held, _)| held) {
                    Ok(slot) => rows[slot].1 = mask,
                    Err(at) => {
                        debug_assert_eq!(at, rows.len(), "a birth lands past the tail");
                        rows.push((page, mask));
                    }
                }
            }
            return;
        }
        let mut merged: Vec<(u64, u64)> = Vec::with_capacity(rows.len() + changes.len());
        let mut held = rows.iter().copied().peekable();
        let mut edits = changes
            .iter()
            .map(|&((_, page), mask)| (page, mask))
            .peekable();
        loop {
            let take_edit = match (held.peek(), edits.peek()) {
                (None, None) => break,
                (Some(_), None) => false,
                (None, Some(_)) => true,
                (Some(&(row_page, _)), Some(&(edit_page, _))) => {
                    if row_page == edit_page {
                        let _ = held.next();
                    }
                    row_page >= edit_page
                }
            };
            let (page, mask) = if take_edit {
                edits.next().expect("peeked")
            } else {
                held.next().expect("peeked")
            };
            if mask != 0 {
                merged.push((page, mask));
            }
        }
        if merged.is_empty() {
            let _ = self.stations.remove(&station);
        } else {
            *rows = merged;
        }
    }

    /// Stages a sorted per-dot flip stream into whole page rows and
    /// applies them through [`apply_pages`](Self::apply_pages): the
    /// batched form of [`set`](Self::set) for the delta-sized settles
    /// (the in-place merge's residual arms), linear in the flips and the
    /// touched fibers.
    pub(super) fn apply_flips(&mut self, flips: impl Iterator<Item = (Dot, bool)>) {
        let mut staged: Vec<(PageKey, u64)> = Vec::new();
        for (dot, present) in flips {
            let (page, bit) = Self::page_bit(dot.counter());
            let key = (dot.station(), page);
            if staged.last().is_none_or(|&(held, _)| held != key) {
                debug_assert!(
                    staged.last().is_none_or(|&(held, _)| held < key),
                    "the flip stream ascends"
                );
                let current = self
                    .stations
                    .get(&dot.station())
                    .and_then(|rows| {
                        rows.binary_search_by_key(&page, |&(held, _)| held)
                            .ok()
                            .map(|slot| rows[slot].1)
                    })
                    .unwrap_or(0);
                staged.push((key, current));
            }
            let mask = &mut staged.last_mut().expect("just staged").1;
            if present {
                *mask |= bit;
            } else {
                *mask &= !bit;
            }
        }
        self.apply_pages(&staged);
    }

    /// Adopts another plane's rows wholesale (test machinery; the
    /// collation's mark refresh applies exactly the changed rows through
    /// [`apply_pages`](Self::apply_pages) instead).
    #[cfg(test)]
    pub(super) fn adopt(&mut self, other: &Self) {
        self.stations.clone_from(&other.stations);
    }

    /// The sorted difference walk: every page whose mask differs, with
    /// both masks (an absent page as zero). `O(pages of either side)`
    /// word compares over the station fibers; the changed dots are the
    /// XOR's set bits, extracted by the caller. Content-exact for any
    /// pair.
    pub(super) fn changed_pages<'a>(
        &'a self,
        other: &'a Self,
    ) -> impl Iterator<Item = (PageKey, u64, u64)> + 'a {
        let mut held = PageCursor::new(&self.stations);
        let mut live = PageCursor::new(&other.stations);
        core::iter::from_fn(move || {
            loop {
                let (key, held_mask, live_mask) = match (held.peek(), live.peek()) {
                    (None, None) => return None,
                    (Some(h), None) => {
                        held.bump();
                        (h.0, h.1, 0)
                    }
                    (None, Some(l)) => {
                        live.bump();
                        (l.0, 0, l.1)
                    }
                    (Some(h), Some(l)) => match h.0.cmp(&l.0) {
                        core::cmp::Ordering::Less => {
                            held.bump();
                            (h.0, h.1, 0)
                        }
                        core::cmp::Ordering::Greater => {
                            live.bump();
                            (l.0, 0, l.1)
                        }
                        core::cmp::Ordering::Equal => {
                            held.bump();
                            live.bump();
                            (h.0, h.1, l.1)
                        }
                    },
                };
                if held_mask != live_mask {
                    return Some((key, held_mask, live_mask));
                }
            }
        })
    }

    /// The dots of one page's mask, ascending (the XOR extraction).
    pub(super) fn dots_of(key: PageKey, mut mask: u64) -> impl Iterator<Item = Dot> {
        core::iter::from_fn(move || {
            loop {
                if mask == 0 {
                    return None;
                }
                let bit = mask.trailing_zeros();
                mask &= mask - 1;
                // The non-dot zero never enters an occupancy plane (page
                // zero keeps bit zero clear), so the skip is unreachable;
                // it keeps the extraction total rather than asserting the
                // law a second time (ruling R-91).
                if let Some(counter) = NonZeroU64::new(key.1 * PAGE + u64::from(bit)) {
                    return Some(Dot::new(key.0, counter));
                }
            }
        })
    }

    /// Whether one dot's bit is set (the debug belt's read: the choke
    /// point asserts the plane never LEADS its carried set, which is
    /// what makes the collation's born-bit hygiene observable; the
    /// release build type-checks the assert's expression, so the read
    /// stays unconditional and branch-eliminated).
    pub(super) fn contains(&self, station: u32, index: u64) -> bool {
        let (page, bit) = Self::page_bit(index);
        self.stations.get(&station).is_some_and(|rows| {
            rows.binary_search_by_key(&page, |&(held, _)| held)
                .is_ok_and(|slot| rows[slot].1 & bit != 0)
        })
    }

    /// A point-built plane from ascending dots (test machinery: the
    /// contract-form build the checks and pins compare against).
    #[cfg(test)]
    fn from_dots(dots: impl Iterator<Item = Dot>) -> Self {
        let mut plane = Self::new();
        for dot in dots {
            plane.set(dot.station(), dot.counter(), true);
        }
        plane
    }

    /// Exhaustive structural check for the test suites: canonical form
    /// and exact agreement with a fresh build from the carried set.
    #[cfg(test)]
    pub(super) fn check_against(&self, dots: impl Iterator<Item = Dot>) {
        for rows in self.stations.values() {
            assert!(!rows.is_empty(), "no empty stations retained");
            assert!(
                rows.windows(2).all(|pair| pair[0].0 < pair[1].0),
                "pages ascend strictly within a fiber"
            );
            assert!(
                rows.iter().all(|&(_, mask)| mask != 0),
                "no zero masks retained"
            );
        }
        assert_eq!(
            self,
            &Self::from_dots(dots),
            "the occupancy plane agrees with its carried set"
        );
    }
}

/// A flattening cursor over one plane's station fibers, for the
/// difference walk: peeks and bumps `((station, page), mask)` rows in
/// ascending key order without allocation.
struct PageCursor<'a> {
    stations: alloc::collections::btree_map::Iter<'a, u32, Vec<(u64, u64)>>,
    current: Option<(u32, &'a [(u64, u64)])>,
}

impl<'a> PageCursor<'a> {
    fn new(stations: &'a BTreeMap<u32, Vec<(u64, u64)>>) -> Self {
        let mut cursor = Self {
            stations: stations.iter(),
            current: None,
        };
        cursor.refill();
        cursor
    }

    fn refill(&mut self) {
        while self.current.is_none_or(|(_, rows)| rows.is_empty()) {
            if let Some((&station, rows)) = self.stations.next() {
                self.current = Some((station, rows.as_slice()));
            } else {
                self.current = None;
                return;
            }
        }
    }

    fn peek(&self) -> Option<(PageKey, u64)> {
        self.current.map(|(station, rows)| {
            let (page, mask) = rows[0];
            ((station, page), mask)
        })
    }

    fn bump(&mut self) {
        if let Some((station, rows)) = self.current {
            self.current = Some((station, &rows[1..]));
            self.refill();
        }
    }
}

#[cfg(test)]
mod tests;