obj_core/btree/

node.rs

//! B+tree node encode / decode.
//!
//! See `docs/format.md` § B+tree for the on-disk layout. This module
//! is the reference implementation; the tests at the bottom validate
//! round-trip equality and rejection of every documented invariant
//! violation.
//!
//! # Power-of-ten posture
//!
//! - **Rule 2.** Every loop iterates over `key_count`, the slot
//!   directory length, or the page-byte range — all bounded by
//!   `PAGE_SIZE`.
//! - **Rule 3.** The in-memory [`DecodedNode`] uses `Vec`s, not
//!   `heapless::Vec`s of the worst-case slot count: the worst-case
//!   slot count times the worst-case per-key buffer (~1 MiB) would
//!   overflow a 2 MiB stack. The B+tree's *traversal stack* uses
//!   `heapless::Vec` to satisfy Rule 3 on the hot read path; the
//!   *decoded representation* of a single node is a transient
//!   per-page artifact whose heap allocation is unavoidable and is
//!   not on a real-time path.
//! - **Rule 5.** `validate_node` is the central invariant check; it
//!   is `debug_assert!`-ed at every mutation site and surfaces as
//!   `Error::Corruption` on decode of a malformed page.
//! - **Rule 7.** No `unwrap` / `expect` on any error-bearing path;
//!   length-prefixed varint reads return `Error::Corruption` on
//!   underflow.

#![forbid(unsafe_code)]

use crate::error::{Error, Result};
use crate::pager::page::{Page, PAGE_SIZE, PAGE_TRAILER_SIZE};

/// Page-type tag for an internal B+tree node. See `docs/format.md`
/// § Page types.
pub const PAGE_TYPE_BTREE_INTERNAL: u8 = 0x02;
/// Page-type tag for a leaf B+tree node.
pub const PAGE_TYPE_BTREE_LEAF: u8 = 0x03;

/// Fixed-size B+tree node header. See `docs/format.md` § Node header.
pub const NODE_HEADER_SIZE: usize = 24;

// Field offsets within the node header.
const OFF_PAGE_TYPE: usize = 0;
const OFF_LEVEL: usize = 1;
const OFF_KEY_COUNT: usize = 2;
const OFF_FREE_SPACE_OFF: usize = 4;
const OFF_NEXT_SIBLING: usize = 8;
const OFF_RESERVED: usize = 16;

/// Bytes per slot in a leaf slot directory. The entry is a single
/// u32 LE giving the byte offset of the corresponding heap entry.
pub const LEAF_SLOT_BYTES: usize = 4;

/// Bytes per slot in an internal slot directory. The entry is a u32
/// LE offset followed by a u64 LE right-child page-id.
pub const INTERNAL_SLOT_BYTES: usize = 12;

/// Bytes the leftmost-child page-id occupies in an internal node,
/// immediately after the node header.
pub const INTERNAL_LEFTMOST_CHILD_BYTES: usize = 8;

/// Byte offset of the first byte after the node header (leaf slot
/// directory start, OR internal leftmost-child).
const PAYLOAD_OFFSET: usize = NODE_HEADER_SIZE;

/// Last byte index (exclusive) of the in-node payload. Beyond this
/// lies the page trailer.
const PAYLOAD_END: usize = PAGE_SIZE - PAGE_TRAILER_SIZE;

/// Maximum number of slots a leaf can hold at the worst-case empty
/// key/value sizes. Used to size the in-memory slot vector.
pub const LEAF_SLOT_CAP: usize = (PAYLOAD_END - PAYLOAD_OFFSET) / LEAF_SLOT_BYTES;

/// Maximum number of slots an internal node can hold at the worst
/// case. Used to size the in-memory slot vector.
pub const INTERNAL_SLOT_CAP: usize =
    (PAYLOAD_END - PAYLOAD_OFFSET - INTERNAL_LEFTMOST_CHILD_BYTES) / INTERNAL_SLOT_BYTES;

/// Maximum key length the format permits. See `docs/format.md`
/// § Key and value encoding.
#[must_use]
pub const fn max_key_len() -> usize {
    PAGE_SIZE / 4
}

/// Maximum value length that still fits inline in a leaf alongside
/// at least one slot. See `docs/format.md` § Key and value encoding.
#[must_use]
pub const fn max_inline_value(key_len: usize) -> usize {
    // Available payload: PAGE_SIZE - trailer - node header - one slot
    // dir entry. Minus the heap entry's two varints.
    let base = PAYLOAD_END - PAYLOAD_OFFSET - LEAF_SLOT_BYTES;
    // 9 bytes max per LEB128 varint at 64-bit; we use two of them.
    let varints = 9 + 9;
    if key_len + varints >= base {
        0
    } else {
        base - key_len - varints
    }
}

/// Whether a node is an internal node or a leaf.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NodeKind {
    /// Internal node — pivots + child pointers, no user values.
    Internal,
    /// Leaf node — `(key, value)` slots.
    Leaf,
}

/// One leaf slot in the decoded in-memory representation.
#[derive(Debug, Clone)]
pub struct LeafEntry {
    /// The key bytes.
    pub key: Vec<u8>,
    /// The value bytes.
    pub value: Vec<u8>,
}

/// One internal-node pivot entry in the decoded in-memory
/// representation. The corresponding right child is stored in
/// [`DecodedNode::children`] at index `i + 1`.
#[derive(Debug, Clone)]
pub struct InternalEntry {
    /// The pivot key.
    pub key: Vec<u8>,
}

/// Alias for "one of the right-child page-ids in an internal node".
/// We store children as raw `u64`s; the on-disk format permits `0`
/// only as the leaf `next_sibling` sentinel, never as an internal
/// child pointer.
pub type ChildEntry = u64;

/// Decoded in-memory representation of a B+tree node.
///
/// One [`DecodedNode`] is built per `Pager::read_page` of a B+tree
/// page; lifetime is "until the caller is done inspecting the
/// node." See the module-level Rule 3 note for why we use `Vec`
/// rather than `heapless::Vec` for the slot collections.
#[derive(Debug, Clone)]
pub struct DecodedNode {
    /// Node kind (internal or leaf).
    pub kind: NodeKind,
    /// Tree level. Leaves are level 0; internals are level ≥ 1.
    pub level: u8,
    /// Right-sibling leaf page-id (leaves only; `0` on the last leaf
    /// and on internal nodes).
    pub next_sibling: u64,
    /// Child page-ids (internal nodes only). Length is
    /// `internals.len() + 1` when the node is internal, `0` for
    /// leaves.
    pub children: Vec<ChildEntry>,
    /// Leaf slots (leaf nodes only).
    pub leaves: Vec<LeafEntry>,
    /// Internal-node pivots (internal nodes only).
    pub internals: Vec<InternalEntry>,
}

impl DecodedNode {
    /// Number of bytes the slot directory + key/value heap currently
    /// occupy in a freshly-encoded page. Used by the insert path to
    /// decide whether a node has room for one more entry.
    #[must_use]
    pub fn occupied_bytes(&self) -> usize {
        match self.kind {
            NodeKind::Leaf => {
                let slot_bytes = self.leaves.len() * LEAF_SLOT_BYTES;
                let heap_bytes: usize = self
                    .leaves
                    .iter()
                    .map(|e| {
                        varint_len(u64_from_usize(e.key.len()))
                            + e.key.len()
                            + varint_len(u64_from_usize(e.value.len()))
                            + e.value.len()
                    })
                    .sum();
                slot_bytes + heap_bytes
            }
            NodeKind::Internal => {
                let slot_bytes = self.internals.len() * INTERNAL_SLOT_BYTES;
                let heap_bytes: usize = self
                    .internals
                    .iter()
                    .map(|e| varint_len(u64_from_usize(e.key.len())) + e.key.len())
                    .sum();
                INTERNAL_LEFTMOST_CHILD_BYTES + slot_bytes + heap_bytes
            }
        }
    }

    /// Available payload bytes a fresh encode would still have.
    #[must_use]
    pub fn free_bytes(&self) -> usize {
        let cap = PAYLOAD_END - PAYLOAD_OFFSET;
        cap.saturating_sub(self.occupied_bytes())
    }
}

/// Encode `node` into `page`. Stamps the page trailer.
///
/// # Errors
///
/// Returns [`Error::BTreeInvariantViolated`] if the node's slot count
/// or key/value sizes do not fit in a single page.
pub fn encode_node(node: &DecodedNode, page: &mut Page) -> Result<()> {
    validate_node_release(node)?;
    let buf = page.as_bytes_mut();
    buf.fill(0);
    write_node_header(buf, node);
    match node.kind {
        NodeKind::Leaf => encode_leaf_body(buf, node)?,
        NodeKind::Internal => encode_internal_body(buf, node)?,
    }
    // Stamp the trailer using the pager helper so the page is
    // immediately readable through `Pager::write_page` (which leaves
    // the trailer alone — it expects a stamped page).
    crate::pager::checksum::write_page_trailer(page);
    Ok(())
}

fn write_node_header(buf: &mut [u8; PAGE_SIZE], node: &DecodedNode) {
    let page_type = match node.kind {
        NodeKind::Leaf => PAGE_TYPE_BTREE_LEAF,
        NodeKind::Internal => PAGE_TYPE_BTREE_INTERNAL,
    };
    buf[OFF_PAGE_TYPE] = page_type;
    buf[OFF_LEVEL] = node.level;
    let key_count = match node.kind {
        NodeKind::Leaf => node.leaves.len(),
        NodeKind::Internal => node.internals.len(),
    };
    // Bounded by LEAF_SLOT_CAP / INTERNAL_SLOT_CAP at encode time; the
    // narrowing cast is safe because both caps fit in u16 (LEAF_SLOT_CAP
    // = (4092 - 24) / 4 = 1017 < 65535).
    let kc16 = u16_from_usize(key_count);
    buf[OFF_KEY_COUNT..OFF_KEY_COUNT + 2].copy_from_slice(&kc16.to_le_bytes());
    // `free_space_off` is the heap watermark — the byte offset of
    // the first heap entry currently in use. After a compacting
    // encode this is `PAYLOAD_END - heap_bytes`.
    let heap_bytes: usize = match node.kind {
        NodeKind::Leaf => node
            .leaves
            .iter()
            .map(|e| {
                varint_len(u64_from_usize(e.key.len()))
                    + e.key.len()
                    + varint_len(u64_from_usize(e.value.len()))
                    + e.value.len()
            })
            .sum(),
        NodeKind::Internal => node
            .internals
            .iter()
            .map(|e| varint_len(u64_from_usize(e.key.len())) + e.key.len())
            .sum(),
    };
    let fso = u32_from_usize(PAYLOAD_END.saturating_sub(heap_bytes));
    buf[OFF_FREE_SPACE_OFF..OFF_FREE_SPACE_OFF + 4].copy_from_slice(&fso.to_le_bytes());
    buf[OFF_NEXT_SIBLING..OFF_NEXT_SIBLING + 8].copy_from_slice(&node.next_sibling.to_le_bytes());
    // OFF_RESERVED (16..24) stays zero per the spec.
    let _ = OFF_RESERVED;
}

fn encode_leaf_body(buf: &mut [u8; PAGE_SIZE], node: &DecodedNode) -> Result<()> {
    let mut heap_cursor = PAYLOAD_END;
    let mut slot_off = PAYLOAD_OFFSET;
    for entry in &node.leaves {
        let entry_len = varint_len(u64_from_usize(entry.key.len()))
            + entry.key.len()
            + varint_len(u64_from_usize(entry.value.len()))
            + entry.value.len();
        if heap_cursor < entry_len + slot_off + LEAF_SLOT_BYTES {
            return Err(Error::BTreeInvariantViolated {
                reason: "leaf encode: slot dir and heap collide",
            });
        }
        heap_cursor -= entry_len;
        let mut cur = heap_cursor;
        let kl = u64_from_usize(entry.key.len());
        cur += write_varint(&mut buf[cur..], kl);
        buf[cur..cur + entry.key.len()].copy_from_slice(&entry.key);
        cur += entry.key.len();
        let vl = u64_from_usize(entry.value.len());
        cur += write_varint(&mut buf[cur..], vl);
        buf[cur..cur + entry.value.len()].copy_from_slice(&entry.value);
        let off32 = u32_from_usize(heap_cursor);
        buf[slot_off..slot_off + LEAF_SLOT_BYTES].copy_from_slice(&off32.to_le_bytes());
        slot_off += LEAF_SLOT_BYTES;
    }
    Ok(())
}

fn encode_internal_body(buf: &mut [u8; PAGE_SIZE], node: &DecodedNode) -> Result<()> {
    if node.children.len() != node.internals.len() + 1 {
        return Err(Error::BTreeInvariantViolated {
            reason: "internal node has children.len() != pivots+1",
        });
    }
    let leftmost = node.children[0];
    buf[PAYLOAD_OFFSET..PAYLOAD_OFFSET + INTERNAL_LEFTMOST_CHILD_BYTES]
        .copy_from_slice(&leftmost.to_le_bytes());
    let mut heap_cursor = PAYLOAD_END;
    let mut slot_off = PAYLOAD_OFFSET + INTERNAL_LEFTMOST_CHILD_BYTES;
    for (i, pivot) in node.internals.iter().enumerate() {
        let right_child = node.children[i + 1];
        let entry_len = varint_len(u64_from_usize(pivot.key.len())) + pivot.key.len();
        if heap_cursor < entry_len + slot_off + INTERNAL_SLOT_BYTES {
            return Err(Error::BTreeInvariantViolated {
                reason: "internal encode: slot dir and heap collide",
            });
        }
        heap_cursor -= entry_len;
        let mut cur = heap_cursor;
        cur += write_varint(&mut buf[cur..], u64_from_usize(pivot.key.len()));
        buf[cur..cur + pivot.key.len()].copy_from_slice(&pivot.key);
        let off32 = u32_from_usize(heap_cursor);
        buf[slot_off..slot_off + 4].copy_from_slice(&off32.to_le_bytes());
        buf[slot_off + 4..slot_off + INTERNAL_SLOT_BYTES]
            .copy_from_slice(&right_child.to_le_bytes());
        slot_off += INTERNAL_SLOT_BYTES;
    }
    Ok(())
}

/// Decode a B+tree page into a [`DecodedNode`].
///
/// # Errors
///
/// Returns [`Error::Corruption`] if any field is malformed: bad
/// page-type tag, key-count exceeding the slot cap, slot offset
/// outside the page, varint length running past the page end, or
/// keys not in ascending order.
pub fn decode_node(buf: &[u8; PAGE_SIZE]) -> Result<DecodedNode> {
    let page_type = buf[OFF_PAGE_TYPE];
    let kind = match page_type {
        PAGE_TYPE_BTREE_LEAF => NodeKind::Leaf,
        PAGE_TYPE_BTREE_INTERNAL => NodeKind::Internal,
        _ => return Err(Error::Corruption { page_id: 0 }),
    };
    let level = buf[OFF_LEVEL];
    let key_count = u16::from_le_bytes([buf[OFF_KEY_COUNT], buf[OFF_KEY_COUNT + 1]]) as usize;
    let next_sibling = u64::from_le_bytes([
        buf[OFF_NEXT_SIBLING],
        buf[OFF_NEXT_SIBLING + 1],
        buf[OFF_NEXT_SIBLING + 2],
        buf[OFF_NEXT_SIBLING + 3],
        buf[OFF_NEXT_SIBLING + 4],
        buf[OFF_NEXT_SIBLING + 5],
        buf[OFF_NEXT_SIBLING + 6],
        buf[OFF_NEXT_SIBLING + 7],
    ]);
    let mut node = DecodedNode {
        kind,
        level,
        next_sibling,
        children: Vec::new(),
        leaves: Vec::new(),
        internals: Vec::new(),
    };
    match kind {
        NodeKind::Leaf => decode_leaf_body(buf, &mut node, key_count)?,
        NodeKind::Internal => decode_internal_body(buf, &mut node, key_count)?,
    }
    validate_node_release(&node)?;
    Ok(node)
}

fn decode_leaf_body(buf: &[u8; PAGE_SIZE], node: &mut DecodedNode, key_count: usize) -> Result<()> {
    if key_count > LEAF_SLOT_CAP {
        return Err(Error::Corruption { page_id: 0 });
    }
    node.leaves.reserve(key_count);
    let mut prev_key: Option<Vec<u8>> = None;
    for i in 0..key_count {
        let slot_off = PAYLOAD_OFFSET + i * LEAF_SLOT_BYTES;
        if slot_off + LEAF_SLOT_BYTES > PAYLOAD_END {
            return Err(Error::Corruption { page_id: 0 });
        }
        let entry_off = u32::from_le_bytes([
            buf[slot_off],
            buf[slot_off + 1],
            buf[slot_off + 2],
            buf[slot_off + 3],
        ]) as usize;
        if !(PAYLOAD_OFFSET..PAYLOAD_END).contains(&entry_off) {
            return Err(Error::Corruption { page_id: 0 });
        }
        let (key, after_key) = read_length_prefixed(buf, entry_off)?;
        let (value, _) = read_length_prefixed(buf, after_key)?;
        if key.len() > max_key_len() {
            return Err(Error::Corruption { page_id: 0 });
        }
        let key_vec = key.to_vec();
        if let Some(prev) = &prev_key {
            if key_vec.as_slice() <= prev.as_slice() {
                return Err(Error::Corruption { page_id: 0 });
            }
        }
        prev_key = Some(key_vec.clone());
        node.leaves.push(LeafEntry {
            key: key_vec,
            value: value.to_vec(),
        });
    }
    Ok(())
}

fn decode_internal_body(
    buf: &[u8; PAGE_SIZE],
    node: &mut DecodedNode,
    key_count: usize,
) -> Result<()> {
    if key_count > INTERNAL_SLOT_CAP {
        return Err(Error::Corruption { page_id: 0 });
    }
    let leftmost = read_u64(buf, PAYLOAD_OFFSET);
    if leftmost == 0 {
        return Err(Error::Corruption { page_id: 0 });
    }
    node.children.reserve(key_count + 1);
    node.internals.reserve(key_count);
    node.children.push(leftmost);
    let mut prev_key: Option<Vec<u8>> = None;
    for i in 0..key_count {
        let (key_vec, right_child) = decode_internal_slot(buf, i)?;
        if let Some(prev) = &prev_key {
            if key_vec.as_slice() <= prev.as_slice() {
                return Err(Error::Corruption { page_id: 0 });
            }
        }
        prev_key = Some(key_vec.clone());
        node.internals.push(InternalEntry { key: key_vec });
        node.children.push(right_child);
    }
    Ok(())
}

/// Decode the i-th slot of an internal-node body. Returns the pivot
/// key and the right-child page-id, or `Error::Corruption` if the
/// slot is malformed.
fn decode_internal_slot(buf: &[u8; PAGE_SIZE], i: usize) -> Result<(Vec<u8>, u64)> {
    let slot_off = PAYLOAD_OFFSET + INTERNAL_LEFTMOST_CHILD_BYTES + i * INTERNAL_SLOT_BYTES;
    if slot_off + INTERNAL_SLOT_BYTES > PAYLOAD_END {
        return Err(Error::Corruption { page_id: 0 });
    }
    let entry_off = u32::from_le_bytes([
        buf[slot_off],
        buf[slot_off + 1],
        buf[slot_off + 2],
        buf[slot_off + 3],
    ]) as usize;
    let right_child = read_u64(buf, slot_off + 4);
    if right_child == 0 || !(PAYLOAD_OFFSET..PAYLOAD_END).contains(&entry_off) {
        return Err(Error::Corruption { page_id: 0 });
    }
    let (key, _) = read_length_prefixed(buf, entry_off)?;
    if key.len() > max_key_len() {
        return Err(Error::Corruption { page_id: 0 });
    }
    Ok((key.to_vec(), right_child))
}

/// Read a u64 LE from `buf` at `offset`. Panics if `offset + 8` is
/// out of bounds; callers must validate the offset.
fn read_u64(buf: &[u8; PAGE_SIZE], offset: usize) -> u64 {
    u64::from_le_bytes([
        buf[offset],
        buf[offset + 1],
        buf[offset + 2],
        buf[offset + 3],
        buf[offset + 4],
        buf[offset + 5],
        buf[offset + 6],
        buf[offset + 7],
    ])
}

/// Read a `(length-prefixed bytes, next_offset)` pair from `buf`
/// starting at `offset`. Length is an unsigned LEB128 varint; the
/// payload is `length` raw bytes.
///
/// # Power-of-ten Rule 7 posture
///
/// `len` is caller-controlled (it lives on disk in a page byte that
/// can be tampered with or random in a fuzz harness). The
/// `after + len_usize` end-of-slice computation MUST use
/// `checked_add`: with `overflow-checks = true` in release (per
/// Rule 10) a raw `+` panics on a wrap; without overflow-checks the
/// bounds check is silently bypassed. Either failure mode is a
/// denial-of-service hazard reachable from
/// `Pager::read_page → BTree::open → decode_node` on any tampered
/// B+tree page (M13 #115).
fn read_length_prefixed(buf: &[u8; PAGE_SIZE], offset: usize) -> Result<(&[u8], usize)> {
    if offset >= PAYLOAD_END {
        return Err(Error::Corruption { page_id: 0 });
    }
    let (len, after) = read_varint(buf, offset)?;
    let len_usize = usize_from_u64(len)?;
    // Rule 7: `len_usize` is unbounded (varint domain is u64); raw
    // `after + len_usize` wraps on tampered input.
    let end = after
        .checked_add(len_usize)
        .ok_or(Error::Corruption { page_id: 0 })?;
    if end > PAYLOAD_END {
        return Err(Error::Corruption { page_id: 0 });
    }
    Ok((&buf[after..end], end))
}

/// Validate `node` against the format-version-0 invariants.
///
/// # Errors
///
/// Returns [`Error::BTreeInvariantViolated`] if any invariant fails.
/// Callers can `debug_assert!` on a wrapping helper for development
/// builds; release builds get the error surfaced cleanly.
pub fn validate_node(node: &DecodedNode) -> Result<()> {
    validate_node_release(node)
}

fn validate_node_release(node: &DecodedNode) -> Result<()> {
    match node.kind {
        NodeKind::Leaf => {
            if node.level != 0 {
                return Err(Error::BTreeInvariantViolated {
                    reason: "leaf has non-zero level",
                });
            }
            if node.leaves.len() > LEAF_SLOT_CAP {
                return Err(Error::BTreeInvariantViolated {
                    reason: "leaf key count exceeds slot cap",
                });
            }
            for w in node.leaves.windows(2) {
                if w[0].key.as_slice() >= w[1].key.as_slice() {
                    return Err(Error::BTreeInvariantViolated {
                        reason: "leaf keys not strictly sorted",
                    });
                }
            }
        }
        NodeKind::Internal => {
            if node.level == 0 {
                return Err(Error::BTreeInvariantViolated {
                    reason: "internal node at level 0",
                });
            }
            if node.internals.len() > INTERNAL_SLOT_CAP {
                return Err(Error::BTreeInvariantViolated {
                    reason: "internal key count exceeds slot cap",
                });
            }
            if node.children.len() != node.internals.len() + 1 {
                return Err(Error::BTreeInvariantViolated {
                    reason: "internal children.len() != pivots+1",
                });
            }
            for c in &node.children {
                if *c == 0 {
                    return Err(Error::BTreeInvariantViolated {
                        reason: "internal node has zero child page-id",
                    });
                }
            }
            for w in node.internals.windows(2) {
                if w[0].key.as_slice() >= w[1].key.as_slice() {
                    return Err(Error::BTreeInvariantViolated {
                        reason: "internal keys not strictly sorted",
                    });
                }
            }
            if node.next_sibling != 0 {
                return Err(Error::BTreeInvariantViolated {
                    reason: "internal node has non-zero next_sibling",
                });
            }
        }
    }
    Ok(())
}

// --------- varint helpers ---------

/// Number of bytes an unsigned LEB128 varint would use for `v`.
#[must_use]
pub fn varint_len(v: u64) -> usize {
    let mut n: usize = 1;
    let mut x = v >> 7;
    while x != 0 {
        n += 1;
        x >>= 7;
    }
    n
}

/// Write an unsigned LEB128 varint into `dst` starting at byte 0.
/// Returns the number of bytes written.
fn write_varint(dst: &mut [u8], mut v: u64) -> usize {
    let mut i = 0;
    loop {
        let mut byte = (v & 0x7F) as u8;
        v >>= 7;
        if v != 0 {
            byte |= 0x80;
            dst[i] = byte;
            i += 1;
        } else {
            dst[i] = byte;
            i += 1;
            return i;
        }
    }
}

/// Read an unsigned LEB128 varint from `buf` starting at `offset`.
/// Returns `(value, next_offset)`. Power-of-ten Rule 2: the loop is
/// bounded by 10 bytes (`ceil(64 / 7)`), which is the maximum a
/// 64-bit varint can occupy.
fn read_varint(buf: &[u8; PAGE_SIZE], offset: usize) -> Result<(u64, usize)> {
    let mut value: u64 = 0;
    let mut shift: u32 = 0;
    let mut i = offset;
    for _ in 0..10 {
        if i >= PAYLOAD_END {
            return Err(Error::Corruption { page_id: 0 });
        }
        let byte = buf[i];
        i += 1;
        let chunk = u64::from(byte & 0x7F);
        // `shift` is in range [0..63]; the cast cannot truncate.
        value |= chunk << shift;
        if (byte & 0x80) == 0 {
            return Ok((value, i));
        }
        shift += 7;
        if shift >= 64 {
            return Err(Error::Corruption { page_id: 0 });
        }
    }
    Err(Error::Corruption { page_id: 0 })
}

// --------- narrow casts ---------

fn u32_from_usize(v: usize) -> u32 {
    // PAYLOAD_END < 2^32; every caller passes such a value.
    debug_assert!(u32::try_from(v).is_ok());
    // Saturating downcast: the asserted bound is the binding contract;
    // a release-mode violation would still return a too-large u32, which
    // an encoder cross-check (decode_node) would catch on read.
    u32::try_from(v).unwrap_or(u32::MAX)
}

fn u16_from_usize(v: usize) -> u16 {
    debug_assert!(u16::try_from(v).is_ok());
    u16::try_from(v).unwrap_or(u16::MAX)
}

fn u64_from_usize(v: usize) -> u64 {
    v as u64
}

fn usize_from_u64(v: u64) -> Result<usize> {
    usize::try_from(v).map_err(|_| Error::Corruption { page_id: 0 })
}

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

    fn empty_leaf() -> DecodedNode {
        DecodedNode {
            kind: NodeKind::Leaf,
            level: 0,
            next_sibling: 0,
            children: Vec::new(),
            leaves: Vec::new(),
            internals: Vec::new(),
        }
    }

    fn leaf_with(entries: &[(&[u8], &[u8])]) -> DecodedNode {
        let mut leaf = empty_leaf();
        for (k, v) in entries {
            leaf.leaves.push(LeafEntry {
                key: k.to_vec(),
                value: v.to_vec(),
            });
        }
        leaf
    }

    #[test]
    fn round_trip_empty_leaf() {
        let leaf = empty_leaf();
        let mut page = Page::zeroed();
        encode_node(&leaf, &mut page).expect("encode");
        let decoded = decode_node(page.as_bytes()).expect("decode");
        assert_eq!(decoded.kind, NodeKind::Leaf);
        assert_eq!(decoded.level, 0);
        assert_eq!(decoded.next_sibling, 0);
        assert!(decoded.leaves.is_empty());
    }

    #[test]
    fn round_trip_populated_leaf() {
        let leaf = leaf_with(&[(b"alpha", b"AAA"), (b"bravo", b"BBB"), (b"charlie", b"CCC")]);
        let mut page = Page::zeroed();
        encode_node(&leaf, &mut page).expect("encode");
        let decoded = decode_node(page.as_bytes()).expect("decode");
        assert_eq!(decoded.leaves.len(), 3);
        assert_eq!(decoded.leaves[0].key.as_slice(), b"alpha");
        assert_eq!(decoded.leaves[0].value.as_slice(), b"AAA");
        assert_eq!(decoded.leaves[2].key.as_slice(), b"charlie");
    }

    #[test]
    fn round_trip_internal() {
        let mut internal = DecodedNode {
            kind: NodeKind::Internal,
            level: 1,
            next_sibling: 0,
            children: Vec::new(),
            leaves: Vec::new(),
            internals: Vec::new(),
        };
        // 3 pivots → 4 children.
        for raw in [10u64, 20, 30, 40] {
            internal.children.push(raw);
        }
        for k in [b"d".as_slice(), b"h", b"m"] {
            internal.internals.push(InternalEntry { key: k.to_vec() });
        }
        let mut page = Page::zeroed();
        encode_node(&internal, &mut page).expect("encode");
        let decoded = decode_node(page.as_bytes()).expect("decode");
        assert_eq!(decoded.kind, NodeKind::Internal);
        assert_eq!(decoded.level, 1);
        assert_eq!(decoded.internals.len(), 3);
        assert_eq!(decoded.children.as_slice(), &[10, 20, 30, 40]);
        assert_eq!(decoded.internals[0].key.as_slice(), b"d");
        assert_eq!(decoded.internals[2].key.as_slice(), b"m");
    }

    #[test]
    fn decode_rejects_unsorted_leaf() {
        // Encode a deliberately-malformed leaf by hand: write two
        // out-of-order entries and check decode errors.
        let mut page = Page::zeroed();
        let buf = page.as_bytes_mut();
        buf[OFF_PAGE_TYPE] = PAGE_TYPE_BTREE_LEAF;
        buf[OFF_LEVEL] = 0;
        buf[OFF_KEY_COUNT..OFF_KEY_COUNT + 2].copy_from_slice(&2u16.to_le_bytes());
        // Slot 0 at PAYLOAD_END - 5  ("c" + "x").
        // Slot 1 at PAYLOAD_END - 10 ("a" + "y").
        let slot0 = u32_from_usize(PAYLOAD_END - 4);
        let slot1 = u32_from_usize(PAYLOAD_END - 8);
        buf[PAYLOAD_OFFSET..PAYLOAD_OFFSET + 4].copy_from_slice(&slot0.to_le_bytes());
        buf[PAYLOAD_OFFSET + 4..PAYLOAD_OFFSET + 8].copy_from_slice(&slot1.to_le_bytes());
        // Heap entry for slot 0: key="c", value="x"  -> 1,'c',1,'x'.
        buf[PAYLOAD_END - 4] = 1;
        buf[PAYLOAD_END - 3] = b'c';
        buf[PAYLOAD_END - 2] = 1;
        buf[PAYLOAD_END - 1] = b'x';
        // Heap entry for slot 1: key="a", value="y"  -> 1,'a',1,'y'.
        buf[PAYLOAD_END - 8] = 1;
        buf[PAYLOAD_END - 7] = b'a';
        buf[PAYLOAD_END - 6] = 1;
        buf[PAYLOAD_END - 5] = b'y';
        crate::pager::checksum::write_page_trailer(&mut page);
        let err = decode_node(page.as_bytes()).expect_err("decode must reject");
        assert!(matches!(err, Error::Corruption { .. }));
    }

    #[test]
    fn varint_round_trip() {
        for v in [0u64, 1, 127, 128, 0xFFFF, 0xDEAD_BEEF, u64::MAX] {
            let mut buf = [0u8; 16];
            let n = write_varint(&mut buf, v);
            assert_eq!(n, varint_len(v));
            // Read back using a faux PAGE_SIZE buffer.
            let mut page = [0u8; PAGE_SIZE];
            page[..n].copy_from_slice(&buf[..n]);
            let (decoded, after) = read_varint(&page, 0).expect("decode varint");
            assert_eq!(decoded, v);
            assert_eq!(after, n);
        }
    }

    /// M13 #115 regression: a tampered B+tree leaf carrying a
    /// length-prefix varint that encodes `u64::MAX` must return
    /// `Error::Corruption`, NOT panic in `read_length_prefixed` on
    /// the `after + len_usize` overflow.
    ///
    /// Critical: this test must also pass under `cargo test --release`
    /// — the workspace profile enables `overflow-checks = true` per
    /// power-of-ten Rule 10, which is the mechanism that turned the
    /// arithmetic wrap into a panic in the first place. A `checked_add`
    /// fix is the only thing that makes both debug and release happy.
    #[test]
    fn decode_node_varint_max_returns_corruption_not_panic() {
        let mut page = Page::zeroed();
        {
            let buf = page.as_bytes_mut();
            buf[OFF_PAGE_TYPE] = PAGE_TYPE_BTREE_LEAF;
            buf[OFF_LEVEL] = 0;
            // One slot. The varint at the slot's entry-offset will
            // encode `u64::MAX`, which `read_length_prefixed` must
            // reject rather than wrap.
            buf[OFF_KEY_COUNT..OFF_KEY_COUNT + 2].copy_from_slice(&1u16.to_le_bytes());
            // Place the heap entry near the end of the payload. The
            // bug triggers BEFORE the bounds check, so the absolute
            // offset value only matters to the extent that it points
            // somewhere parseable.
            let entry_off = PAYLOAD_END - 16;
            let slot_off_bytes = u32_from_usize(entry_off).to_le_bytes();
            buf[PAYLOAD_OFFSET..PAYLOAD_OFFSET + 4].copy_from_slice(&slot_off_bytes);
            // 10-byte LEB128 encoding of `u64::MAX`. The 10th byte's
            // top bit is clear so `read_varint` accepts it; the
            // decoded length is `u64::MAX`, which `as usize` becomes
            // `usize::MAX` on 64-bit and triggers the `after + len`
            // overflow in `read_length_prefixed`.
            let varint = [0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x01];
            buf[entry_off..entry_off + varint.len()].copy_from_slice(&varint);
        }
        crate::pager::checksum::write_page_trailer(&mut page);
        let result = decode_node(page.as_bytes());
        match result {
            Err(Error::Corruption { .. }) => {}
            other => panic!(
                "expected Err(Error::Corruption), got {:?} \
                 (a panic here would indicate the M13 #115 bug \
                  is still present in release builds)",
                other.map(|_| "Ok(_)").unwrap_or("non-corruption err")
            ),
        }
    }

    /// M13 #115 regression for the internal-node variant: a tampered
    /// internal-node entry pointing at a varint that encodes
    /// `u64::MAX` must surface as `Error::Corruption`, not panic via
    /// `decode_internal_slot → read_length_prefixed`.
    #[test]
    fn decode_internal_node_varint_max_returns_corruption() {
        let mut page = Page::zeroed();
        {
            let buf = page.as_bytes_mut();
            buf[OFF_PAGE_TYPE] = PAGE_TYPE_BTREE_INTERNAL;
            buf[OFF_LEVEL] = 1;
            buf[OFF_KEY_COUNT..OFF_KEY_COUNT + 2].copy_from_slice(&1u16.to_le_bytes());
            // Leftmost child page-id — must be non-zero.
            buf[PAYLOAD_OFFSET..PAYLOAD_OFFSET + 8].copy_from_slice(&42u64.to_le_bytes());
            // One internal slot: u32 entry_off + u64 right_child.
            let slot_base = PAYLOAD_OFFSET + INTERNAL_LEFTMOST_CHILD_BYTES;
            let entry_off = PAYLOAD_END - 16;
            buf[slot_base..slot_base + 4].copy_from_slice(&u32_from_usize(entry_off).to_le_bytes());
            buf[slot_base + 4..slot_base + INTERNAL_SLOT_BYTES]
                .copy_from_slice(&7u64.to_le_bytes());
            // Same `u64::MAX` varint as the leaf test above.
            let varint = [0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x01];
            buf[entry_off..entry_off + varint.len()].copy_from_slice(&varint);
        }
        crate::pager::checksum::write_page_trailer(&mut page);
        let result = decode_node(page.as_bytes());
        assert!(
            matches!(result, Err(Error::Corruption { .. })),
            "expected Err(Error::Corruption) on u64::MAX varint in internal slot"
        );
    }

    #[test]
    fn max_inline_value_is_below_payload() {
        let k = 8;
        let v = max_inline_value(k);
        // Compute slot+heap bytes for one entry at this max value
        // length and assert it fits.
        let entry_len = varint_len(u64_from_usize(k)) + k + varint_len(u64_from_usize(v)) + v;
        let payload = entry_len + LEAF_SLOT_BYTES;
        assert!(payload <= PAYLOAD_END - PAYLOAD_OFFSET);
    }
}