use crate::{Error, Result};
pub const PAGE_SIZE: usize = 4096;
pub const HEADER_LEN: usize = 40;
const MAGIC: u32 = 0x5345_4B32;
const VERSION: u16 = 1;
const SLOT_LEN: usize = 4;
pub const FORMAT_VERSION: u16 = 2;
const FORMAT_AT: usize = 18;
pub fn format_version(b: &[u8]) -> u16 { rd_u16(b, FORMAT_AT) }
pub const MAX_RECORD_LEN: usize = PAGE_SIZE - HEADER_LEN - SLOT_LEN;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u16)]
pub enum PageKind { Free = 0, Meta = 1, Leaf = 2, Interior = 3, Overflow = 4 }
impl PageKind {
fn from_u16(v: u16) -> Option<Self> {
Some(match v {
0 => PageKind::Free, 1 => PageKind::Meta, 2 => PageKind::Leaf,
3 => PageKind::Interior, 4 => PageKind::Overflow, _ => return None,
})
}
}
fn rd_u16(b: &[u8], at: usize) -> u16 { u16::from_le_bytes([b[at], b[at + 1]]) }
fn rd_u32(b: &[u8], at: usize) -> u32 { u32::from_le_bytes(b[at..at + 4].try_into().unwrap()) }
fn rd_u64(b: &[u8], at: usize) -> u64 { u64::from_le_bytes(b[at..at + 8].try_into().unwrap()) }
pub fn checksum(b: &[u8]) -> u32 {
#[cfg(feature = "write-trace")]
let trace_started = crate::write_trace::active().then(std::time::Instant::now);
let mut h = crc32c::crc32c(&b[0..36]);
h = crc32c::crc32c_append(h, &b[40..PAGE_SIZE]);
#[cfg(feature = "write-trace")]
if let Some(started) = trace_started {
crate::write_trace::add(crate::write_trace::Field::PageCrc, started.elapsed());
crate::write_trace::page_crc();
}
h
}
pub struct PageMut<'a> { b: &'a mut [u8] }
impl<'a> PageMut<'a> {
pub fn init(b: &'a mut [u8], kind: PageKind, tree_id: u16, page_no: u32) -> Self {
assert_eq!(b.len(), PAGE_SIZE);
b.fill(0);
b[0..4].copy_from_slice(&MAGIC.to_le_bytes());
b[4..6].copy_from_slice(&VERSION.to_le_bytes());
b[6..8].copy_from_slice(&(kind as u16).to_le_bytes());
b[8..10].copy_from_slice(&tree_id.to_le_bytes());
b[12..16].copy_from_slice(&page_no.to_le_bytes());
b[16..18].copy_from_slice(&(PAGE_SIZE as u16).to_le_bytes());
b[FORMAT_AT..FORMAT_AT + 2].copy_from_slice(&FORMAT_VERSION.to_le_bytes());
PageMut { b }
}
pub fn reopen(b: &'a mut [u8]) -> Self { PageMut { b } }
fn nentries(&self) -> usize { rd_u16(self.b, 10) as usize }
pub fn nentries_pub(&self) -> usize { self.nentries() }
fn free_ptr(&self) -> usize { rd_u16(self.b, 16) as usize }
pub fn free_space(&self) -> usize {
self.free_ptr().saturating_sub(HEADER_LEN + self.nentries() * SLOT_LEN)
}
pub fn set_next_leaf(&mut self, p: u32) { self.b[20..24].copy_from_slice(&p.to_le_bytes()); }
pub fn set_child0(&mut self, p: u32) { self.set_next_leaf(p) }
pub fn insert_slot(&mut self, at: usize, rec: &[u8]) -> Result<()> {
let n = self.nentries();
if at > n { return Err(Error::TooLarge); }
if rec.len() + SLOT_LEN > self.free_space() { return Err(Error::TooLarge); }
let new_ptr = self.free_ptr() - rec.len();
self.b[new_ptr..new_ptr + rec.len()].copy_from_slice(rec);
self.b[16..18].copy_from_slice(&(new_ptr as u16).to_le_bytes());
let dir = HEADER_LEN;
let from = dir + at * SLOT_LEN;
let to = dir + n * SLOT_LEN;
self.b.copy_within(from..to, from + SLOT_LEN);
self.b[from..from + 2].copy_from_slice(&(new_ptr as u16).to_le_bytes());
self.b[from + 2..from + 4].copy_from_slice(&(rec.len() as u16).to_le_bytes());
self.b[10..12].copy_from_slice(&((n + 1) as u16).to_le_bytes());
Ok(())
}
pub fn remove_slot(&mut self, at: usize) {
let n = self.nentries();
assert!(at < n);
let dir = HEADER_LEN;
let from = dir + (at + 1) * SLOT_LEN;
let to = dir + n * SLOT_LEN;
self.b.copy_within(from..to, from - SLOT_LEN);
self.b[10..12].copy_from_slice(&((n - 1) as u16).to_le_bytes());
}
pub fn compact(&mut self) {
let n = self.nentries();
let payloads: Vec<Vec<u8>> = (0..n)
.map(|i| {
let base = HEADER_LEN + i * SLOT_LEN;
let off = rd_u16(self.b, base) as usize;
let len = rd_u16(self.b, base + 2) as usize;
self.b[off..off + len].to_vec()
})
.collect();
let mut ptr = PAGE_SIZE;
for (i, payload) in payloads.iter().enumerate() {
ptr -= payload.len();
self.b[ptr..ptr + payload.len()].copy_from_slice(payload);
let base = HEADER_LEN + i * SLOT_LEN;
self.b[base..base + 2].copy_from_slice(&(ptr as u16).to_le_bytes());
}
self.b[16..18].copy_from_slice(&(ptr as u16).to_le_bytes());
}
pub fn finalise(&mut self, lsn: u64) {
self.b[24..32].copy_from_slice(&lsn.to_le_bytes());
}
}
pub fn seal(b: &mut [u8], gen: u64) {
b[24..32].copy_from_slice(&gen.to_le_bytes());
b[FORMAT_AT..FORMAT_AT + 2].copy_from_slice(&FORMAT_VERSION.to_le_bytes());
let c = checksum(b);
b[36..40].copy_from_slice(&c.to_le_bytes());
}
#[derive(Debug)]
pub struct PageRef<'a> { b: &'a [u8] }
impl<'a> PageRef<'a> {
pub fn open(b: &'a [u8], want: u32) -> Result<Self> {
Self::open_inner(b, want, true)
}
pub fn open_resident(b: &'a [u8], want: u32) -> Result<Self> {
Self::open_inner(b, want, false)
}
fn open_inner(b: &'a [u8], want: u32, verify_crc: bool) -> Result<Self> {
let bad = |why| Err(Error::Corrupt { page_no: want, why });
if b.len() != PAGE_SIZE { return bad("wrong buffer length"); }
if rd_u32(b, 0) != MAGIC { return bad("bad magic"); }
if rd_u16(b, 4) != VERSION { return bad("unknown format version"); }
if verify_crc && rd_u32(b, 36) != checksum(b) { return bad("checksum mismatch"); }
let stamp = rd_u16(b, FORMAT_AT);
if stamp != FORMAT_VERSION { return Err(Error::UnsupportedFormat { found: stamp }); }
if rd_u32(b, 12) != want { return bad("page_no mismatch"); }
if PageKind::from_u16(rd_u16(b, 6)).is_none() { return bad("unknown page kind"); }
let n = rd_u16(b, 10) as usize;
let dir_end = HEADER_LEN + n * SLOT_LEN;
if dir_end > PAGE_SIZE { return bad("nentries exceeds page"); }
let free_ptr = rd_u16(b, 16) as usize;
if free_ptr < dir_end || free_ptr > PAGE_SIZE {
return bad("free pointer is outside the page payload area");
}
for i in 0..n {
let off = rd_u16(b, HEADER_LEN + i * SLOT_LEN) as usize;
let len = rd_u16(b, HEADER_LEN + i * SLOT_LEN + 2) as usize;
if off < free_ptr || off.checked_add(len).is_none_or(|end| end > PAGE_SIZE) {
return bad("slot out of bounds");
}
}
Ok(PageRef { b })
}
pub fn open_resident_validated(b: &'a [u8], want: u32) -> Result<Self> {
let bad = |why| Err(Error::Corrupt { page_no: want, why });
if b.len() != PAGE_SIZE { return bad("wrong buffer length"); }
if rd_u32(b, 0) != MAGIC { return bad("bad magic"); }
if rd_u16(b, 4) != VERSION { return bad("unknown format version"); }
let stamp = rd_u16(b, FORMAT_AT);
if stamp != FORMAT_VERSION { return Err(Error::UnsupportedFormat { found: stamp }); }
if rd_u32(b, 12) != want { return bad("page_no mismatch"); }
if PageKind::from_u16(rd_u16(b, 6)).is_none() { return bad("unknown page kind"); }
Ok(PageRef { b })
}
pub fn kind(&self) -> PageKind { PageKind::from_u16(rd_u16(self.b, 6)).unwrap() }
pub fn tree_id(&self) -> u16 { rd_u16(self.b, 8) }
pub fn nentries(&self) -> usize { rd_u16(self.b, 10) as usize }
pub fn page_no(&self) -> u32 { rd_u32(self.b, 12) }
pub fn next_leaf(&self) -> u32 { rd_u32(self.b, 20) }
pub fn child0(&self) -> u32 { self.next_leaf() }
pub fn lsn(&self) -> u64 { rd_u64(self.b, 24) }
pub fn free_space(&self) -> usize {
let free_ptr = rd_u16(self.b, 16) as usize;
free_ptr.saturating_sub(HEADER_LEN + self.nentries() * SLOT_LEN)
}
pub fn slot_bounds(&self, i: usize) -> (u16, u16) {
let base = HEADER_LEN + i * SLOT_LEN;
(rd_u16(self.b, base), rd_u16(self.b, base + 2))
}
pub fn slot(&self, i: usize) -> &'a [u8] {
let off = rd_u16(self.b, HEADER_LEN + i * SLOT_LEN) as usize;
let len = rd_u16(self.b, HEADER_LEN + i * SLOT_LEN + 2) as usize;
&self.b[off..off + len]
}
}
#[cfg(test)]
mod tests {
use super::*;
fn buf() -> Vec<u8> { vec![0u8; PAGE_SIZE] }
#[test]
fn a_finalised_page_reads_back_its_identity() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Leaf, 7, 12345);
p.finalise(99);
seal(&mut b, 99); let r = PageRef::open(&b, 12345).expect("verifies");
assert_eq!(r.kind(), PageKind::Leaf);
assert_eq!(r.tree_id(), 7);
assert_eq!(r.page_no(), 12345);
assert_eq!(r.lsn(), 99);
assert_eq!(r.nentries(), 0);
}
#[test]
fn entries_round_trip_in_slot_order() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Leaf, 1, 4);
p.insert_slot(0, b"alpha").unwrap();
p.insert_slot(1, b"beta").unwrap();
p.insert_slot(1, b"between").unwrap(); p.finalise(1);
seal(&mut b, 7);
let r = PageRef::open(&b, 4).unwrap();
assert_eq!(r.nentries(), 3);
assert_eq!(r.slot(0), b"alpha");
assert_eq!(r.slot(1), b"between");
assert_eq!(r.slot(2), b"beta");
}
#[test]
fn a_free_pointer_outside_its_page_is_refused() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Leaf, 1, 4);
p.finalise(1);
b[16..18].copy_from_slice(&((PAGE_SIZE + 1) as u16).to_le_bytes());
seal(&mut b, 7);
assert!(matches!(
PageRef::open(&b, 4),
Err(Error::Corrupt { page_no: 4, .. })
));
}
#[test]
fn a_page_asked_for_by_the_wrong_number_is_refused() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Leaf, 1, 500);
p.finalise(1);
seal(&mut b, 7);
match PageRef::open(&b, 501) {
Err(crate::Error::Corrupt { page_no: 501, why }) => assert_eq!(why, "page_no mismatch"),
other => panic!("expected identity refusal, got {other:?}"),
}
}
#[test]
fn a_single_flipped_byte_anywhere_is_refused() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Leaf, 1, 3);
p.insert_slot(0, b"payload-bytes").unwrap();
p.finalise(1);
seal(&mut b, 7);
assert!(PageRef::open(&b, 3).is_ok());
for pos in [0usize, 9, 21, 37, 41, 100, PAGE_SIZE - 1] {
let mut damaged = b.clone();
damaged[pos] ^= 0xff;
assert!(
PageRef::open(&damaged, 3).is_err(),
"corruption at byte {pos} was not detected"
);
}
}
#[test]
fn a_corrupt_entry_count_cannot_read_past_the_page() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Leaf, 1, 3);
p.insert_slot(0, b"x").unwrap();
p.finalise(1);
seal(&mut b, 7);
b[10..12].copy_from_slice(&60_000u16.to_le_bytes());
recrc(&mut b);
match PageRef::open(&b, 3) {
Err(crate::Error::Corrupt { why, .. }) => assert_eq!(why, "nentries exceeds page"),
other => panic!("expected bound refusal, got {other:?}"),
}
}
#[test]
fn a_record_larger_than_the_page_is_refused_not_panicked() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Leaf, 1, 3);
assert!(matches!(p.insert_slot(0, &vec![0u8; PAGE_SIZE]), Err(crate::Error::TooLarge)));
}
#[test]
fn compact_reclaims_a_removed_entrys_payload() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Leaf, 1, 5);
p.insert_slot(0, b"alpha").unwrap();
p.insert_slot(1, b"between-bytes").unwrap();
p.insert_slot(2, b"beta").unwrap();
let free_before_remove = p.free_space();
p.remove_slot(1); let free_after_remove = p.free_space();
assert_eq!(free_after_remove, free_before_remove + 4);
p.compact();
let free_after_compact = p.free_space();
assert_eq!(
free_after_compact,
free_after_remove + "between-bytes".len(),
"compact must reclaim the abandoned payload, not just the slot"
);
p.finalise(1);
seal(&mut b, 7);
let r = PageRef::open(&b, 5).unwrap();
assert_eq!(r.nentries(), 2);
assert_eq!(r.slot(0), b"alpha");
assert_eq!(r.slot(1), b"beta");
}
#[test]
fn child0_round_trips_through_an_interior_page() {
let mut b = buf();
let mut p = PageMut::init(&mut b, PageKind::Interior, 1, 9);
p.set_child0(4242);
p.finalise(1);
seal(&mut b, 7);
let r = PageRef::open(&b, 9).unwrap();
assert_eq!(r.kind(), PageKind::Interior);
assert_eq!(r.child0(), 4242);
}
fn recrc(b: &mut [u8]) {
let c = crate::page::checksum(b);
b[36..40].copy_from_slice(&c.to_le_bytes());
}
}