ic-storage 0.65.3

//! Owned degree-six B-tree. The multi-entry search, split/borrow/merge design and
//! selective payload access follow the established B-tree techniques used by
//! ic-stable-structures 0.7.2 (see docs/storage/upstreams.md). This is a new checked
//! format, not an implementation of its durable layout or compatibility reader.
//! Routing records contain small inline values; larger values have separate
//! checksummed records. Mutations stage bytes and reserve before publication.
use super::{
    Memory, Storable,
    error::{CollectionError as Error, Result, ensure},
    record,
};
use serde::{Deserialize, Serialize};
use std::{
    borrow::Cow,
    cell::{Cell, RefCell},
    collections::BTreeMap,
    marker::PhantomData,
    ops::{Bound, RangeBounds},
    rc::Rc,
};
mod read;
use read::{NodeCache, ReadNode};

pub const MAGIC: [u8; 8] = *b"ICMAP002";
const NODE: [u8; 8] = *b"ICBTND01";
const VALUE: [u8; 8] = *b"ICBTVAL1";
const FREE: [u8; 8] = *b"ICFREE01";
const BINS: usize = 32;
const MAX_DEPTH: usize = 32;
const DEGREE: usize = 6;
const MAX_ENTRIES: usize = DEGREE * 2 - 1;
const MIN_ENTRIES: usize = DEGREE - 1;
const INLINE: usize = 32;

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
struct Pointer {
    offset: u64,
    order: u8,
}
impl Pointer {
    fn capacity(self) -> u64 {
        1_u64 << self.order
    }
    fn check(self, cursor: u64) -> Result<()> {
        ensure(
            (8..32).contains(&self.order),
            Error::Corrupt("tree block order"),
        )?;
        ensure(
            self.offset >= record::HEADER_BYTES
                && self
                    .offset
                    .checked_add(self.capacity())
                    .is_some_and(|end| end <= cursor),
            Error::Corrupt("tree block bounds"),
        )
    }
    fn put(self, bytes: &mut Vec<u8>) {
        bytes.extend_from_slice(&self.offset.to_le_bytes());
        bytes.push(self.order);
    }
}
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
struct Root {
    top: Option<Pointer>,
    length: u64,
    cursor: u64,
    free: Vec<Option<Pointer>>,
    key_bound: Option<(u32, bool)>,
    value_bound: Option<(u32, bool)>,
}
#[derive(Clone)]
enum Value {
    Inline(Vec<u8>),
    External(Pointer),
}
#[derive(Clone)]
struct Entry {
    key: Vec<u8>,
    value: Value,
}
#[derive(Clone, Default)]
struct Node {
    entries: Vec<Entry>,
    children: Vec<Pointer>,
}
impl Node {
    fn encode(&self) -> Result<Vec<u8>> {
        ensure(
            self.entries.len() <= MAX_ENTRIES,
            Error::Corrupt("B-tree node capacity"),
        )?;
        ensure(
            self.children.is_empty() || self.children.len() == self.entries.len() + 1,
            Error::Corrupt("B-tree child count"),
        )?;
        let capacity =
            self.entries
                .iter()
                .try_fold(2 + self.children.len() * 9, |size, entry| {
                    let value_size = match &entry.value {
                        Value::Inline(v) => v.len(),
                        Value::External(_) => 9,
                    };
                    size.checked_add(5)
                        .and_then(|n| n.checked_add(entry.key.len()))
                        .and_then(|n| n.checked_add(value_size))
                        .ok_or(Error::Budget)
                })?;
        ensure(capacity <= record::MAX_RECORD, Error::Budget)?;
        let mut bytes = Vec::with_capacity(capacity);
        bytes.extend_from_slice(&[
            u8::try_from(self.entries.len()).unwrap(),
            u8::from(!self.children.is_empty()),
        ]);
        for p in &self.children {
            p.put(&mut bytes);
        }
        for entry in &self.entries {
            bytes.extend_from_slice(
                &u32::try_from(entry.key.len())
                    .map_err(|_| Error::Budget)?
                    .to_le_bytes(),
            );
            bytes.extend_from_slice(&entry.key);
            match &entry.value {
                Value::Inline(value) => {
                    bytes.push(u8::try_from(value.len()).unwrap());
                    bytes.extend_from_slice(value);
                }
                Value::External(p) => {
                    bytes.push(255);
                    p.put(&mut bytes);
                }
            }
        }
        record::encode_payload(NODE, &bytes)
    }
}
struct Reader<'a>(&'a [u8]);
impl<'a> Reader<'a> {
    fn take(&mut self, n: usize) -> Result<&'a [u8]> {
        ensure(n <= self.0.len(), Error::Corrupt("truncated B-tree node"))?;
        let (a, b) = self.0.split_at(n);
        self.0 = b;
        Ok(a)
    }
    fn byte(&mut self) -> Result<u8> {
        Ok(self.take(1)?[0])
    }
    fn pointer(&mut self, cursor: u64) -> Result<Pointer> {
        let offset = u64::from_le_bytes(self.take(8)?.try_into().unwrap());
        let p = Pointer {
            offset,
            order: self.byte()?,
        };
        p.check(cursor)?;
        Ok(p)
    }
}

pub struct Tree<K, V, M: Memory> {
    pub memory: M,
    root: RefCell<Root>,
    poisoned: Cell<bool>,
    cache: RefCell<NodeCache>,
    types: PhantomData<(K, V)>,
}
impl<K: Storable + Ord + Clone, V: Storable, M: Memory> Tree<K, V, M> {
    pub fn reset(memory: M) -> Result<Self> {
        let root = Root {
            top: None,
            length: 0,
            cursor: record::HEADER_BYTES,
            free: vec![None; BINS],
            key_bound: record::bound::<K>(),
            value_bound: record::bound::<V>(),
        };
        let bytes = record::encode(MAGIC, &root)?;
        ensure(bytes.len() as u64 <= record::HEADER_BYTES, Error::Budget)?;
        record::reserve(&memory, record::HEADER_BYTES)?;
        memory.write(0, &bytes);
        Ok(Self {
            memory,
            root: RefCell::new(root),
            poisoned: Cell::new(false),
            cache: RefCell::default(),
            types: PhantomData,
        })
    }
    pub fn load(memory: M) -> Result<Self> {
        let root: Root = record::read(&memory, 0, record::HEADER_BYTES, MAGIC)?;
        ensure(
            root.key_bound == record::bound::<K>() && root.value_bound == record::bound::<V>(),
            Error::Corrupt("map codec bound changed"),
        )?;
        ensure(
            root.free.len() == BINS
                && root.cursor >= record::HEADER_BYTES
                && root.cursor
                    <= memory
                        .size()
                        .checked_mul(crate::WASM_PAGE_SIZE_BYTES)
                        .ok_or(Error::Budget)?,
            Error::Corrupt("map allocator bounds"),
        )?;
        ensure(
            root.top.is_none() == (root.length == 0),
            Error::Corrupt("map root/count disagree"),
        )?;
        if let Some(p) = root.top {
            p.check(root.cursor)?;
        }
        for (order, p) in root.free.iter().enumerate() {
            if let Some(p) = p {
                p.check(root.cursor)?;
                ensure(
                    usize::from(p.order) == order,
                    Error::Corrupt("free block order"),
                )?;
            }
        }
        Ok(Self {
            memory,
            root: RefCell::new(root),
            poisoned: Cell::new(false),
            cache: RefCell::default(),
            types: PhantomData,
        })
    }
    fn healthy(&self) -> Result<()> {
        ensure(!self.poisoned.get(), Error::Poisoned)?;
        // A different allocation can poison the shared governed backing. This
        // also covers iterator entries already held in a routing frame.
        let capacity = self
            .memory
            .size()
            .checked_mul(crate::WASM_PAGE_SIZE_BYTES)
            .ok_or(Error::Budget)?;
        ensure(
            self.root.borrow().cursor <= capacity,
            Error::Corrupt("map allocator bounds"),
        )
    }
    pub fn len(&self) -> u64 {
        self.healthy().expect("map owner is poisoned");
        self.root.borrow().length
    }
    #[cfg(test)]
    fn node(&self, p: Pointer) -> Result<Node> {
        Ok(self.read_node(p, MAX_DEPTH)?.owned())
    }
    pub fn set_cache_budget(&mut self, bytes: usize) {
        self.cache.get_mut().configure(bytes);
    }
    fn read_node(&self, p: Pointer, depth: usize) -> Result<Rc<ReadNode>> {
        let cursor = self.root.borrow().cursor;
        p.check(cursor)?;
        if let Some(node) = self.cache.borrow_mut().get(p, depth) {
            return Ok(node);
        }
        let node = Rc::new(ReadNode::decode::<K, V>(
            record::read_bytes(&self.memory, p.offset, p.capacity(), NODE)?,
            cursor,
        )?);
        self.cache.borrow_mut().put(p, Rc::clone(&node), depth);
        Ok(node)
    }
    fn value(&self, v: Value) -> Result<Vec<u8>> {
        let bytes = match v {
            Value::Inline(bytes) => bytes,
            Value::External(p) => {
                p.check(self.root.borrow().cursor)?;
                let bytes = record::read_bytes(&self.memory, p.offset, p.capacity(), VALUE)?;
                record::decode_payload(VALUE, &bytes)?.to_vec()
            }
        };
        V::check_bounds(&bytes)?;
        Ok(bytes)
    }
    fn find(&self, key: &K) -> Result<Option<Value>> {
        self.healthy()?;
        let mut p = self.root.borrow().top;
        for depth in 0..MAX_DEPTH {
            let Some(current) = p else { return Ok(None) };
            let node = self.read_node(current, depth)?;
            match node.search(key)? {
                Ok(i) => return Ok(Some(node.value(i))),
                Err(i) => p = node.children.get(i).copied(),
            }
        }
        Err(Error::Corrupt("map path exceeds height bound"))
    }
    pub fn get(&self, key: &K) -> Result<Option<V>> {
        self.find(key)?.map(|v| decode(self.value(v)?)).transpose()
    }
    pub fn contains_key(&self, key: &K) -> Result<bool> {
        Ok(self.find(key)?.is_some())
    }
    pub fn insert(&mut self, key: K, value: V) -> Result<Option<V>> {
        self.healthy()?;
        let key_bytes = key.to_bytes_checked()?.into_owned();
        let bytes = value.into_bytes_checked()?;
        let mut tx = Transaction::new(self);
        let (top, split, old) = tx.insert(tx.root.top, &key, key_bytes, bytes, 0)?;
        tx.root.top = Some(if let Some((entry, right)) = split {
            tx.save(
                None,
                &Node {
                    entries: vec![entry],
                    children: vec![top, right],
                },
            )?
        } else {
            top
        });
        drop(key);
        let old = old.map(decode).transpose()?;
        tx.commit()?;
        Ok(old)
    }
    pub fn remove(&mut self, key: &K) -> Result<Option<V>> {
        self.healthy()?;
        let mut tx = Transaction::new(self);
        let Some(top) = tx.root.top else {
            return Ok(None);
        };
        let (p, old, entries) = tx.remove(top, key, 0)?;
        let Some(old) = old else { return Ok(None) };
        let value = decode(old)?;
        tx.root.top = if entries == 0 {
            let root = tx.read_node(p, 0)?;
            tx.free(p)?;
            root.children.first().copied()
        } else {
            Some(p)
        };
        tx.root.length -= 1;
        tx.commit()?;
        Ok(Some(value))
    }
    pub fn clear(&mut self) -> Result<()> {
        self.healthy()?;
        let mut tx = Transaction::new(self);
        tx.root.top = None;
        tx.root.length = 0;
        tx.root.cursor = record::HEADER_BYTES;
        tx.root.free.fill(None);
        tx.commit()
    }
    fn seek_path(&self, bound: &Bound<K>, forward: bool) -> Result<Vec<Frame>> {
        self.healthy()?;
        let mut path = Vec::new();
        let mut pointer = self.root.borrow().top;
        while let Some(p) = pointer {
            ensure(
                path.len() < MAX_DEPTH,
                Error::Corrupt("map cursor exceeds height bound"),
            )?;
            let node = self.read_node(p, path.len())?;
            let index = match bound {
                Bound::Unbounded => {
                    if forward {
                        0
                    } else {
                        node.len()
                    }
                }
                Bound::Included(k) => match node.search(k)? {
                    Ok(i) => {
                        if forward {
                            i
                        } else {
                            i + 1
                        }
                    }
                    Err(i) => i,
                },
                Bound::Excluded(k) => match node.search(k)? {
                    Ok(i) => {
                        if forward {
                            i + 1
                        } else {
                            i
                        }
                    }
                    Err(i) => i,
                },
            };
            pointer = node.children.get(index).copied();
            path.push(Frame {
                node,
                index,
                descend: false,
            });
        }
        Ok(path)
    }
    fn advance_path(&self, path: &mut Vec<Frame>, forward: bool) -> Result<Option<Entry>> {
        self.healthy()?;
        while let Some(frame) = path.last_mut() {
            if frame.descend {
                frame.descend = false;
                let mut child = frame.node.children.get(frame.index).copied();
                while let Some(p) = child {
                    ensure(
                        path.len() < MAX_DEPTH,
                        Error::Corrupt("map cursor exceeds height bound"),
                    )?;
                    let node = self.read_node(p, path.len())?;
                    let index = if forward { 0 } else { node.len() };
                    child = node.children.get(index).copied();
                    path.push(Frame {
                        node,
                        index,
                        descend: false,
                    });
                }
                continue;
            }
            let index = if forward {
                frame.index
            } else {
                if frame.index == 0 {
                    path.pop();
                    continue;
                }
                frame.index - 1
            };
            if index >= frame.node.len() {
                path.pop();
                continue;
            }
            let entry = frame.node.entry(index);
            frame.index = if forward { index + 1 } else { index };
            // A caller may stop at this entry. Read the following subtree only
            // when another step in this direction is actually requested.
            frame.descend = !frame.node.children.is_empty();
            return Ok(Some(entry));
        }
        Ok(None)
    }
    pub fn range(&self, bounds: impl RangeBounds<K>) -> TreeIter<'_, K, V, M> {
        let lower = bounds.start_bound().cloned();
        let upper = bounds.end_bound().cloned();
        let invalid = match (&lower, &upper) {
            (Bound::Included(a) | Bound::Excluded(a), Bound::Included(b) | Bound::Excluded(b)) => {
                a > b
                    || (a == b
                        && matches!((&lower, &upper), (Bound::Excluded(_), Bound::Excluded(_))))
            }
            _ => false,
        };
        TreeIter {
            tree: self,
            lower,
            upper,
            front: None,
            back: None,
            exhausted: false,
            invalid,
            keys_only: false,
        }
    }
    pub fn keys_range(
        &self,
        bounds: impl RangeBounds<K>,
    ) -> impl DoubleEndedIterator<Item = Result<K>> + '_ {
        let mut iter = self.range(bounds);
        iter.keys_only = true;
        iter.map(|e| e.map(MapEntry::into_key))
    }
}
struct Frame {
    node: Rc<ReadNode>,
    index: usize,
    descend: bool,
}
fn decode<T: Storable>(bytes: Vec<u8>) -> Result<T> {
    T::check_bounds(&bytes)?;
    Ok(T::from_bytes(Cow::Owned(bytes))?)
}

type Split = Option<(Entry, Pointer)>;
struct Transaction<'a, K, V, M: Memory> {
    tree: &'a Tree<K, V, M>,
    root: Root,
    writes: BTreeMap<u64, Vec<u8>>,
}
impl<'a, K: Storable + Ord + Clone, V: Storable, M: Memory> Transaction<'a, K, V, M> {
    fn new(tree: &'a Tree<K, V, M>) -> Self {
        Self {
            tree,
            root: tree.root.borrow().clone(),
            writes: BTreeMap::new(),
        }
    }
    fn bytes(&self, p: Pointer, magic: [u8; 8]) -> Result<Vec<u8>> {
        p.check(self.root.cursor)?;
        if let Some(bytes) = self.writes.get(&p.offset) {
            Ok(bytes.clone())
        } else {
            record::read_bytes(&self.tree.memory, p.offset, p.capacity(), magic)
        }
    }
    fn node(&self, p: Pointer, depth: usize) -> Result<Node> {
        Ok(self.read_node(p, depth)?.owned())
    }
    fn read_node(&self, p: Pointer, depth: usize) -> Result<Rc<ReadNode>> {
        if let Some(bytes) = self.writes.get(&p.offset) {
            Ok(Rc::new(ReadNode::decode::<K, V>(
                bytes.clone(),
                self.root.cursor,
            )?))
        } else {
            self.tree.read_node(p, depth)
        }
    }
    fn value(&self, v: &Value) -> Result<Vec<u8>> {
        let bytes = match v {
            Value::Inline(b) => b.clone(),
            Value::External(p) => record::decode_payload(VALUE, &self.bytes(*p, VALUE)?)?.to_vec(),
        };
        V::check_bounds(&bytes)?;
        Ok(bytes)
    }
    fn free(&mut self, p: Pointer) -> Result<()> {
        let index = usize::from(p.order);
        self.writes
            .insert(p.offset, record::encode(FREE, &self.root.free[index])?);
        self.root.free[index] = Some(p);
        Ok(())
    }
    fn allocate(&mut self, len: usize) -> Result<Pointer> {
        let size = len
            .max(256)
            .checked_next_power_of_two()
            .ok_or(Error::Budget)?;
        let order = size.trailing_zeros() as usize;
        ensure(order < BINS, Error::Budget)?;
        for bin in order..BINS {
            if let Some(mut p) = self.root.free[bin] {
                let next: Option<Pointer> = record::decode(FREE, &self.bytes(p, FREE)?)?;
                if let Some(next) = next {
                    next.check(self.root.cursor)?;
                    ensure(
                        next.order == p.order && next != p,
                        Error::Corrupt("map free-list link"),
                    )?;
                }
                self.root.free[bin] = next;
                // Return only the requested capacity. Each unused half remains
                // an ordinary free-list block, staged with the same mutation.
                while usize::from(p.order) > order {
                    p.order -= 1;
                    let sibling = Pointer {
                        offset: p.offset + p.capacity(),
                        order: p.order,
                    };
                    self.free(sibling)?;
                }
                return Ok(p);
            }
        }
        let p = Pointer {
            offset: self.root.cursor,
            order: u8::try_from(order).unwrap(),
        };
        self.root.cursor = self
            .root
            .cursor
            .checked_add(p.capacity())
            .ok_or(Error::Budget)?;
        Ok(p)
    }
    fn save_bytes(&mut self, previous: Option<Pointer>, bytes: Vec<u8>) -> Result<Pointer> {
        let p = if let Some(p) = previous.filter(|p| p.capacity() >= bytes.len() as u64) {
            let needed = bytes.len().max(256).next_power_of_two() as u64;
            // Reclaim only into already reusable smaller capacity. Splitting the
            // live block itself loses its large size class and causes unbounded
            // growth under repeated large/small replacements without coalescing.
            let smaller_available = (needed.trailing_zeros() as usize..usize::from(p.order))
                .any(|order| self.root.free[order].is_some());
            if p.capacity() / needed >= 4 && smaller_available {
                let smaller = self.allocate(bytes.len())?;
                self.free(p)?;
                smaller
            } else {
                p
            }
        } else {
            if let Some(p) = previous {
                self.free(p)?;
            }
            self.allocate(bytes.len())?
        };
        self.writes.insert(p.offset, bytes);
        Ok(p)
    }
    fn save(&mut self, p: Option<Pointer>, node: &Node) -> Result<Pointer> {
        self.save_bytes(p, node.encode()?)
    }
    fn save_value(&mut self, previous: Option<&Value>, bytes: Vec<u8>) -> Result<Value> {
        let old = match previous {
            Some(Value::External(p)) => Some(*p),
            _ => None,
        };
        if bytes.len() <= INLINE {
            if let Some(p) = old {
                self.free(p)?;
            }
            Ok(Value::Inline(bytes))
        } else {
            Ok(Value::External(
                self.save_bytes(old, record::encode_payload(VALUE, &bytes)?)?,
            ))
        }
    }
    fn insert(
        &mut self,
        p: Option<Pointer>,
        key: &K,
        key_bytes: Vec<u8>,
        bytes: Vec<u8>,
        depth: usize,
    ) -> Result<(Pointer, Split, Option<Vec<u8>>)> {
        ensure(
            depth < MAX_DEPTH,
            Error::Corrupt("map path exceeds height bound"),
        )?;
        let read = p.map(|p| self.read_node(p, depth)).transpose()?;
        let search = read.as_ref().map_or(Ok(Err(0)), |node| node.search(key))?;
        let (mut node, old) = match search {
            Ok(i) => {
                let read = read.as_ref().unwrap();
                let previous = read.value(i);
                let old = self.value(&previous)?;
                let value = self.save_value(Some(&previous), bytes)?;
                let same_pointer =
                    matches!((&previous,&value),(Value::External(a),Value::External(b)) if a==b);
                if same_pointer {
                    return Ok((p.unwrap(), None, Some(old)));
                }
                let mut node = read.owned();
                node.entries[i].value = value;
                (node, Some(old))
            }
            Err(i) => {
                if let Some(child) = read.as_ref().and_then(|node| node.children.get(i)).copied() {
                    let (next, split, old) =
                        self.insert(Some(child), key, key_bytes, bytes, depth + 1)?;
                    if split.is_none() && next == child {
                        return Ok((p.unwrap(), None, old));
                    }
                    let mut node = read.as_ref().unwrap().owned();
                    node.children[i] = next;
                    if let Some((entry, right)) = split {
                        node.entries.insert(i, entry);
                        node.children.insert(i + 1, right);
                    }
                    (node, old)
                } else {
                    let value = self.save_value(None, bytes)?;
                    let mut node = read
                        .as_ref()
                        .map_or_else(Node::default, |node| node.owned());
                    node.entries.insert(
                        i,
                        Entry {
                            key: key_bytes,
                            value,
                        },
                    );
                    self.root.length = self.root.length.checked_add(1).ok_or(Error::Budget)?;
                    (node, None)
                }
            }
        };
        let split = if node.entries.len() > MAX_ENTRIES {
            let right_entries = node.entries.split_off(DEGREE + 1);
            let median = node.entries.pop().unwrap();
            let children = if node.children.is_empty() {
                vec![]
            } else {
                node.children.split_off(DEGREE + 1)
            };
            let right = self.save(
                None,
                &Node {
                    entries: right_entries,
                    children,
                },
            )?;
            Some((median, right))
        } else {
            None
        };
        Ok((self.save(p, &node)?, split, old))
    }
    fn remove(
        &mut self,
        p: Pointer,
        key: &K,
        depth: usize,
    ) -> Result<(Pointer, Option<Vec<u8>>, usize)> {
        ensure(
            depth < MAX_DEPTH,
            Error::Corrupt("map path exceeds height bound"),
        )?;
        let read = self.read_node(p, depth)?;
        let (node, old) = match read.search(key)? {
            Ok(i) => {
                let mut node = read.owned();
                let old = self.value(&node.entries[i].value)?;
                if let Value::External(p) = node.entries[i].value {
                    self.free(p)?;
                }
                if node.children.is_empty() {
                    node.entries.remove(i);
                } else {
                    let (child, entry, entries) =
                        self.extract_min(node.children[i + 1], depth + 1)?;
                    node.entries[i] = entry;
                    node.children[i + 1] = child;
                    if entries < MIN_ENTRIES {
                        self.rebalance(&mut node, i + 1, depth)?;
                    }
                }
                (node, Some(old))
            }
            Err(i) => {
                let Some(child) = read.children.get(i).copied() else {
                    return Ok((p, None, read.len()));
                };
                let (next, old, entries) = self.remove(child, key, depth + 1)?;
                if old.is_none() {
                    return Ok((p, None, read.len()));
                }
                if next == child && entries >= MIN_ENTRIES {
                    return Ok((p, old, read.len()));
                }
                let mut node = read.owned();
                node.children[i] = next;
                if entries < MIN_ENTRIES {
                    self.rebalance(&mut node, i, depth)?;
                }
                (node, old)
            }
        };
        Ok((self.save(Some(p), &node)?, old, node.entries.len()))
    }
    fn extract_min(&mut self, p: Pointer, depth: usize) -> Result<(Pointer, Entry, usize)> {
        ensure(
            depth < MAX_DEPTH,
            Error::Corrupt("map path exceeds height bound"),
        )?;
        let read = self.read_node(p, depth)?;
        ensure(read.len() != 0, Error::Corrupt("empty successor node"))?;
        let (node, entry) = if read.children.is_empty() {
            let mut node = read.owned();
            let entry = node.entries.remove(0);
            (node, entry)
        } else {
            let (child, entry, entries) = self.extract_min(read.children[0], depth + 1)?;
            if read.children[0] == child && entries >= MIN_ENTRIES {
                return Ok((p, entry, read.len()));
            }
            let mut node = read.owned();
            node.children[0] = child;
            if entries < MIN_ENTRIES {
                self.rebalance(&mut node, 0, depth)?;
            }
            (node, entry)
        };
        Ok((self.save(Some(p), &node)?, entry, node.entries.len()))
    }
    fn rebalance(&mut self, parent: &mut Node, i: usize, parent_depth: usize) -> Result<()> {
        let depth = parent_depth + 1;
        let p = parent.children[i];
        let mut child = self.node(p, depth)?;
        if child.entries.len() >= MIN_ENTRIES {
            return Ok(());
        }
        if i > 0 {
            let left_p = parent.children[i - 1];
            let mut left = self.node(left_p, depth)?;
            ensure(
                left.children.is_empty() == child.children.is_empty(),
                Error::Corrupt("B-tree sibling levels"),
            )?;
            if left.entries.len() > MIN_ENTRIES {
                child.entries.insert(
                    0,
                    std::mem::replace(&mut parent.entries[i - 1], left.entries.pop().unwrap()),
                );
                if let Some(p) = left.children.pop() {
                    child.children.insert(0, p);
                }
                parent.children[i - 1] = self.save(Some(left_p), &left)?;
                parent.children[i] = self.save(Some(p), &child)?;
                return Ok(());
            }
        }
        if i + 1 < parent.children.len() {
            let right_p = parent.children[i + 1];
            let mut right = self.node(right_p, depth)?;
            ensure(
                right.children.is_empty() == child.children.is_empty(),
                Error::Corrupt("B-tree sibling levels"),
            )?;
            if right.entries.len() > MIN_ENTRIES {
                child.entries.push(std::mem::replace(
                    &mut parent.entries[i],
                    right.entries.remove(0),
                ));
                if !right.children.is_empty() {
                    child.children.push(right.children.remove(0));
                }
                parent.children[i + 1] = self.save(Some(right_p), &right)?;
                parent.children[i] = self.save(Some(p), &child)?;
                return Ok(());
            }
        }
        let left_index = if i > 0 { i - 1 } else { i };
        ensure(
            left_index < parent.entries.len(),
            Error::Corrupt("B-tree missing separator"),
        )?;
        let left_p = parent.children[left_index];
        let right_p = parent.children[left_index + 1];
        let mut left = self.node(left_p, depth)?;
        let right = self.node(right_p, depth)?;
        ensure(
            left.children.is_empty() == right.children.is_empty(),
            Error::Corrupt("B-tree sibling levels"),
        )?;
        left.entries.push(parent.entries.remove(left_index));
        left.entries.extend(right.entries);
        left.children.extend(right.children);
        self.free(right_p)?;
        parent.children[left_index] = self.save(Some(left_p), &left)?;
        parent.children.remove(left_index + 1);
        Ok(())
    }
    fn commit(self) -> Result<()> {
        let changed = self.root != *self.tree.root.borrow();
        let header = if changed {
            Some(record::encode(MAGIC, &self.root)?)
        } else {
            None
        };
        if let Some(h) = &header {
            ensure(h.len() as u64 <= record::HEADER_BYTES, Error::Budget)?;
        }
        record::reserve(&self.tree.memory, self.root.cursor)?;
        self.tree.poisoned.set(true);
        if self.root.cursor < self.tree.root.borrow().cursor
            || self.root.top != self.tree.root.borrow().top
        {
            self.tree.cache.borrow_mut().clear();
        }
        for (offset, bytes) in self.writes {
            self.tree.cache.borrow_mut().invalidate(offset);
            self.tree.memory.write(offset, &bytes);
        }
        if let Some(h) = header {
            self.tree.memory.write(0, &h);
        }
        *self.tree.root.borrow_mut() = self.root;
        self.tree.poisoned.set(false);
        Ok(())
    }
}
///
/// `MapEntry`
///
/// An owned key and lazily decoded value from an ordered scan.
///
pub struct MapEntry<K, V> {
    key: K,
    bytes: Vec<u8>,
    value: PhantomData<V>,
}
impl<K, V: Storable> MapEntry<K, V> {
    /// Borrow the decoded key.
    pub const fn key(&self) -> &K {
        &self.key
    }
    /// Decode the value using its declared codec.
    ///
    /// # Errors
    /// Returns the application codec or declared-bound error for this value.
    pub fn value(&self) -> Result<V> {
        decode(self.bytes.clone())
    }
    /// Consume the entry and decode its value.
    ///
    /// # Errors
    /// Returns the application codec or declared-bound error for this value.
    pub fn into_pair(self) -> Result<(K, V)> {
        Ok((self.key, decode(self.bytes)?))
    }
    /// Consume the entry without decoding its value.
    pub fn into_key(self) -> K {
        self.key
    }
    /// Consume the entry and decode only its value.
    ///
    /// # Errors
    /// Returns the application codec or declared-bound error for this value.
    pub fn into_value(self) -> Result<V> {
        decode(self.bytes)
    }
}

///
/// `TreeIter`
///
/// Lazy bidirectional traversal with bounded routing-node stacks. External values
/// are read only for in-range entries and decoded when requested by the caller.
///
pub struct TreeIter<'a, K, V, M: Memory> {
    tree: &'a Tree<K, V, M>,
    lower: Bound<K>,
    upper: Bound<K>,
    front: Option<Vec<Frame>>,
    back: Option<Vec<Frame>>,
    exhausted: bool,
    invalid: bool,
    keys_only: bool,
}
impl<K: Storable + Ord + Clone, V: Storable, M: Memory> TreeIter<'_, K, V, M> {
    fn step(&mut self, forward: bool) -> Result<Option<MapEntry<K, V>>> {
        ensure(!self.invalid, Error::Bounds)?;
        self.tree.healthy()?;
        let empty = match (&self.lower, &self.upper) {
            (Bound::Included(a), Bound::Included(b)) => a > b,
            (Bound::Included(a) | Bound::Excluded(a), Bound::Included(b) | Bound::Excluded(b)) => {
                a >= b
            }
            _ => false,
        };
        if empty {
            return Ok(None);
        }
        let (path, bound) = if forward {
            (&mut self.front, &self.lower)
        } else {
            (&mut self.back, &self.upper)
        };
        if path.is_none() {
            *path = Some(self.tree.seek_path(bound, forward)?);
        }
        let entry = self.tree.advance_path(path.as_mut().unwrap(), forward)?;
        if let Some(entry) = entry {
            let key: K = decode(entry.key)?;
            let after = match &self.lower {
                Bound::Unbounded => true,
                Bound::Included(k) => key >= *k,
                Bound::Excluded(k) => key > *k,
            };
            let before = match &self.upper {
                Bound::Unbounded => true,
                Bound::Included(k) => key <= *k,
                Bound::Excluded(k) => key < *k,
            };
            ensure(
                if forward { after } else { before },
                Error::Corrupt("map cursor key order"),
            )?;
            if !(after && before) {
                return Ok(None);
            }
            if forward {
                self.lower = Bound::Excluded(key.clone());
            } else {
                self.upper = Bound::Excluded(key.clone());
            }
            return Ok(Some(MapEntry {
                key,
                bytes: if self.keys_only {
                    vec![]
                } else {
                    self.tree.value(entry.value)?
                },
                value: PhantomData,
            }));
        }
        Ok(None)
    }
    fn advance(&mut self, forward: bool) -> Option<Result<MapEntry<K, V>>> {
        if self.exhausted {
            return None;
        }
        match self.step(forward) {
            Ok(Some(entry)) => Some(Ok(entry)),
            Ok(None) => {
                self.exhausted = true;
                None
            }
            Err(error) => {
                self.exhausted = true;
                Some(Err(error))
            }
        }
    }
}
impl<K: Storable + Ord + Clone, V: Storable, M: Memory> Iterator for TreeIter<'_, K, V, M> {
    type Item = Result<MapEntry<K, V>>;
    fn next(&mut self) -> Option<Self::Item> {
        self.advance(true)
    }
}
impl<K: Storable + Ord + Clone, V: Storable, M: Memory> DoubleEndedIterator
    for TreeIter<'_, K, V, M>
{
    fn next_back(&mut self) -> Option<Self::Item> {
        self.advance(false)
    }
}
impl<K: Storable + Ord + Clone, V: Storable, M: Memory> std::iter::FusedIterator
    for TreeIter<'_, K, V, M>
{
}

#[cfg(test)]
mod tests;