diff --git a/src/btree/internal.rs b/src/btree/internal.rs index d10a1c6..de10435 100644 --- a/src/btree/internal.rs +++ b/src/btree/internal.rs @@ -23,6 +23,35 @@ pub struct Internal { pub entries: Vec, } +/// Fixed bytes an internal record costs beyond its key: the `key_len` `u16` +/// and the `right_child` `u64`. +/// +/// Named so the sizing used to decide whether a separator fits stays tied to +/// the layout [`Internal::encode`] actually writes. +pub(crate) const SEPARATOR_RECORD_OVERHEAD: usize = 2 + 8; + +/// Bytes one slot-directory entry costs. +pub(crate) const SLOT_DIRECTORY_ENTRY_SIZE: usize = 2; + +/// Encoded byte cost of one internal separator plus its slot-directory entry. +pub(crate) fn separator_entry_size(key_len: usize) -> Result { + if key_len > u16::MAX as usize { + return Err(PagedbError::PayloadTooLarge); + } + key_len + .checked_add(SEPARATOR_RECORD_OVERHEAD) + .and_then(|size| size.checked_add(SLOT_DIRECTORY_ENTRY_SIZE)) + .ok_or(PagedbError::PayloadTooLarge) +} + +#[must_use] +pub(crate) fn separator_fits(key_len: usize, page_size: usize) -> bool { + separator_entry_size(key_len) + .ok() + .and_then(|entry_size| HEADER_LEN.checked_add(entry_size)) + .is_some_and(|needed| needed <= body_capacity(page_size)) +} + impl Internal { pub fn decode(body: &[u8]) -> Result { let h: NodeHeader = validate_node_body(body)?; @@ -48,8 +77,12 @@ impl Internal { let cap = body.len(); let prefix_len = 0usize; let slot_count = self.entries.len(); - let record_bytes: usize = self.entries.iter().map(|e| 2 + e.key.len() + 8).sum(); - let slot_dir_bytes = slot_count * 2; + let record_bytes: usize = self + .entries + .iter() + .map(|e| e.key.len() + SEPARATOR_RECORD_OVERHEAD) + .sum(); + let slot_dir_bytes = slot_count * SLOT_DIRECTORY_ENTRY_SIZE; if HEADER_LEN + prefix_len + slot_dir_bytes + record_bytes > cap { return Err(PagedbError::PayloadTooLarge); } @@ -68,7 +101,7 @@ impl Internal { } let mut tail = cap; for (i, e) in self.entries.iter().enumerate() { - let rec_size = 2 + e.key.len() + 8; + let rec_size = e.key.len() + SEPARATOR_RECORD_OVERHEAD; tail -= rec_size; let off = tail; write_u16_le( diff --git a/src/btree/leaf.rs b/src/btree/leaf.rs index 8a209bb..a709064 100644 --- a/src/btree/leaf.rs +++ b/src/btree/leaf.rs @@ -17,6 +17,13 @@ pub enum LeafValue { Overflow { total_len: u64, root_page_id: u64 }, } +/// Bytes an overflow reference occupies in an encoded record body: the +/// sentinel `u16`, then `total_len` and `root_page_id` as `u64`s. +/// +/// Named so the preflight that decides whether a record can fit and the encoder +/// that lays it out cannot drift apart. +pub(crate) const OVERFLOW_REF_ENCODED_SIZE: usize = 2 + 8 + 8; + impl LeafValue { /// Number of bytes this value occupies in the encoded record body (after /// the key suffix): 2-byte `value_len` field plus the payload. @@ -24,8 +31,7 @@ impl LeafValue { pub fn encoded_size(&self) -> usize { match self { Self::Inline(v) => 2 + v.len(), - // sentinel u16 + total_len u64 + root_page_id u64 - Self::Overflow { .. } => 2 + 8 + 8, + Self::Overflow { .. } => OVERFLOW_REF_ENCODED_SIZE, } } } @@ -237,6 +243,24 @@ impl Leaf { let slot_dir_bytes = records.len() * 2; HEADER_LEN + prefix_len + slot_dir_bytes + record_bytes <= cap } + + /// Whether one key/value record fits in an otherwise-empty leaf. + #[must_use] + pub(crate) fn single_record_fits_encoded( + key_len: usize, + value_encoded_size: usize, + page_size: usize, + ) -> bool { + if key_len > u16::MAX as usize { + return false; + } + HEADER_LEN + .checked_add(key_len) + .and_then(|size| size.checked_add(2)) + .and_then(|size| size.checked_add(2)) + .and_then(|size| size.checked_add(value_encoded_size)) + .is_some_and(|needed| needed <= body_capacity(page_size)) + } } impl Default for Leaf { diff --git a/src/btree/overflow.rs b/src/btree/overflow.rs index d68f95d..33123ee 100644 --- a/src/btree/overflow.rs +++ b/src/btree/overflow.rs @@ -13,6 +13,8 @@ //! `next` pointer sits at byte 4 in the root and byte 0 in a chain page — any //! chain walk must account for that offset. +use std::collections::BTreeSet; + use crate::errors::PagedbError; use crate::pager::Pager; use crate::pager::format::data_page::ENVELOPE_OVERHEAD; @@ -272,8 +274,12 @@ pub async fn release( if info.refcount <= 1 { // Collect entire chain. let mut freed = vec![root_page_id]; + let mut seen = BTreeSet::from([root_page_id]); let mut cur = info.next; while cur != 0 { + if !seen.insert(cur) { + return Err(PagedbError::overflow_chain_cycle(root_page_id, cur)); + } let guard = pager .read_main_page(cur, realm_id, PageKind::Overflow) .await?; @@ -303,22 +309,46 @@ pub async fn read_chain( root_page_id: u64, total_len: u64, ) -> Result> { - let mut out: Vec = Vec::with_capacity(usize::try_from(total_len).unwrap_or(0)); + let total_len = usize::try_from(total_len) + .ok() + .filter(|len| isize::try_from(*len).is_ok()) + .ok_or_else(|| PagedbError::overflow_body_malformed("chain.total_length"))?; + // Durable metadata must not choose an arbitrarily large allocation before + // any chain page has been authenticated. Grow only as bytes are verified. + let mut out: Vec = Vec::with_capacity(total_len.min(pager.page_size())); let info = read_root_page(pager, realm_id, root_page_id).await?; + if info.root_data.len() > total_len { + return Err(PagedbError::overflow_body_malformed( + "chain.assembled_length", + )); + } out.extend_from_slice(&info.root_data); + let mut seen = BTreeSet::from([root_page_id]); let mut next = info.next; while next != 0 { + if !seen.insert(next) { + return Err(PagedbError::overflow_chain_cycle(root_page_id, next)); + } let guard = pager .read_main_page(next, realm_id, PageKind::Overflow) .await?; let body = guard.body(); let (n, data) = decode_overflow(&body)?; + if out + .len() + .checked_add(data.len()) + .is_none_or(|assembled| assembled > total_len) + { + return Err(PagedbError::overflow_body_malformed( + "chain.assembled_length", + )); + } out.extend_from_slice(data); next = n; } - if out.len() as u64 != total_len { + if out.len() != total_len { return Err(PagedbError::overflow_body_malformed( "chain.assembled_length", )); @@ -334,9 +364,13 @@ pub async fn collect_chain( root_page_id: u64, ) -> Result> { let mut out = vec![root_page_id]; + let mut seen = BTreeSet::from([root_page_id]); let info = read_root_page(pager, realm_id, root_page_id).await?; let mut next = info.next; while next != 0 { + if !seen.insert(next) { + return Err(PagedbError::overflow_chain_cycle(root_page_id, next)); + } let guard = pager .read_main_page(next, realm_id, PageKind::Overflow) .await?; @@ -350,8 +384,38 @@ pub async fn collect_chain( #[cfg(test)] mod tests { + use std::sync::Arc; + use std::time::Duration; + + use crate::crypto::CipherId; + use crate::crypto::kdf::derive_mk; + use crate::errors::CorruptionDetail; + use crate::pager::PagerConfig; + use crate::vfs::memory::MemVfs; + use super::*; + const TEST_PAGE_SIZE: usize = 4096; + const TEST_REALM: RealmId = RealmId::new([0xA4; 16]); + + async fn test_pager() -> Arc> { + let mk = derive_mk(&[0xA5; 32], &[0u8; 16], 0).unwrap(); + let cfg = PagerConfig { + page_size: TEST_PAGE_SIZE, + buffer_pool_pages: 16, + segment_cache_pages: 16, + cipher_id: CipherId::Aes256Gcm, + mk_epoch: 0, + main_db_file_id: [0xB4; 16], + main_db_path: "/main.db".into(), + anchor_budget: 1_000_000, + dek_lru_capacity: 16, + observer_retry_count: 0, + metrics_enabled: true, + }; + Arc::new(Pager::open(MemVfs::new(), mk, cfg).await.unwrap()) + } + #[test] fn round_trip_chain_page() { let mut body = vec![0u8; 4096 - ENVELOPE_OVERHEAD]; @@ -378,4 +442,111 @@ mod tests { // root: 4096 - 40 - 16 = 4040 assert_eq!(overflow_root_capacity(4096), 4040); } + + async fn cyclic_chain(pager: &Pager, root_page_id: u64, chain_page_id: u64) { + let mut root_body = vec![0u8; TEST_PAGE_SIZE - ENVELOPE_OVERHEAD]; + encode_overflow_root(&mut root_body, 1, chain_page_id, b"").unwrap(); + pager + .write_main_page(root_page_id, TEST_REALM, PageKind::OverflowRoot, &root_body) + .await + .unwrap(); + + let mut chain_body = vec![0u8; TEST_PAGE_SIZE - ENVELOPE_OVERHEAD]; + encode_overflow(&mut chain_body, chain_page_id, b"").unwrap(); + pager + .write_main_page(chain_page_id, TEST_REALM, PageKind::Overflow, &chain_body) + .await + .unwrap(); + } + + #[tokio::test(flavor = "current_thread")] + async fn read_chain_rejects_absurd_total_len_without_allocation_panic() { + let pager = test_pager().await; + let root_page_id = 42; + let mut body = vec![0u8; TEST_PAGE_SIZE - ENVELOPE_OVERHEAD]; + encode_overflow_root(&mut body, 1, 0, b"small").unwrap(); + pager + .write_main_page(root_page_id, TEST_REALM, PageKind::OverflowRoot, &body) + .await + .unwrap(); + + let error = read_chain(&pager, TEST_REALM, root_page_id, u64::MAX) + .await + .unwrap_err(); + assert!(matches!( + error, + PagedbError::Corruption(CorruptionDetail::OverflowBodyMalformed { .. }) + )); + } + + #[tokio::test(flavor = "current_thread")] + async fn read_chain_rejects_more_data_than_declared() { + let pager = test_pager().await; + let root_page_id = 43; + let mut body = vec![0u8; TEST_PAGE_SIZE - ENVELOPE_OVERHEAD]; + encode_overflow_root(&mut body, 1, 0, b"too-long").unwrap(); + pager + .write_main_page(root_page_id, TEST_REALM, PageKind::OverflowRoot, &body) + .await + .unwrap(); + + let error = read_chain(&pager, TEST_REALM, root_page_id, 1) + .await + .unwrap_err(); + assert!(matches!( + error, + PagedbError::Corruption(CorruptionDetail::OverflowBodyMalformed { .. }) + )); + } + + #[tokio::test(flavor = "current_thread")] + async fn read_chain_rejects_cycle_without_hanging() { + let pager = test_pager().await; + cyclic_chain(&pager, 44, 45).await; + let error = tokio::time::timeout( + Duration::from_secs(1), + read_chain(&pager, TEST_REALM, 44, 0), + ) + .await + .expect("cycle detection should return before the timeout") + .unwrap_err(); + assert!(matches!( + error, + PagedbError::Corruption(CorruptionDetail::OverflowChainCycle { .. }) + )); + } + + #[tokio::test(flavor = "current_thread")] + async fn release_rejects_cycle_without_hanging() { + let pager = test_pager().await; + cyclic_chain(&pager, 46, 47).await; + let result = + tokio::time::timeout(Duration::from_secs(1), release(&pager, TEST_REALM, 46, 48)) + .await + .expect("cycle detection should return before the timeout"); + let Err(error) = result else { + panic!("overflow release cycles must not be accepted"); + }; + assert!(matches!( + error, + PagedbError::Corruption(CorruptionDetail::OverflowChainCycle { .. }) + )); + } + + #[tokio::test(flavor = "current_thread")] + async fn collect_chain_rejects_cycle_without_hanging() { + let pager = test_pager().await; + cyclic_chain(&pager, 49, 50).await; + let error = tokio::time::timeout( + Duration::from_secs(1), + collect_chain(&pager, TEST_REALM, 49), + ) + .await + .expect("cycle detection should return before the timeout") + .expect_err("overflow collect cycles must not be accepted"); + assert!(matches!( + error, + PagedbError::Corruption(CorruptionDetail::OverflowChainCycle { .. }) + )); + } } diff --git a/src/btree/tree/bulk.rs b/src/btree/tree/bulk.rs index 3748573..b067211 100644 --- a/src/btree/tree/bulk.rs +++ b/src/btree/tree/bulk.rs @@ -6,6 +6,7 @@ use crate::Result; use crate::errors::PagedbError; use crate::vfs::Vfs; +use crate::btree::internal; use crate::btree::leaf::{Leaf, LeafValue}; use crate::btree::node; use crate::btree::overflow; @@ -34,6 +35,15 @@ impl BTree { return Ok(()); } + for pair in pairs.windows(2) { + if pair[0].0 >= pair[1].0 { + return Err(PagedbError::BulkLoadNotMonotonic); + } + } + for (key, value) in &pairs { + self.validate_insert_record_fits(key, value)?; + } + // Resolve each value to its stored form, spilling values past the inline // threshold to overflow chains (same threshold as `put`). Inlining an // oversized value would exceed leaf capacity and fail the encode. @@ -130,13 +140,19 @@ impl BTree { let mut used = node::HEADER_LEN; for item in remaining.iter().skip(1) { let sep_key = &item.1; - let entry_bytes = 2 + sep_key.len() + 8 + 2; // record + slot - if used + entry_bytes > body_cap { + let entry_bytes = internal::separator_entry_size(sep_key.len())?; + if used + .checked_add(entry_bytes) + .is_none_or(|needed| needed > body_cap) + { break; } used += entry_bytes; count += 1; } + if count == 1 && remaining.len() > 1 { + return Err(PagedbError::PayloadTooLarge); + } let chunk = &remaining[..count]; remaining = &remaining[count..]; diff --git a/src/btree/tree/core.rs b/src/btree/tree/core.rs index 16472f8..2b1ebb2 100644 --- a/src/btree/tree/core.rs +++ b/src/btree/tree/core.rs @@ -1,6 +1,6 @@ //! `BTree` — the `CoW` shadow-paging B+ tree. -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap}; use std::sync::Arc; use crate::errors::PagedbError; @@ -9,9 +9,94 @@ use crate::pager::{PageGuard, Pager}; use crate::vfs::Vfs; use crate::{RealmId, Result}; -use crate::btree::internal::Internal; +use crate::btree::internal::{self, Internal}; use crate::btree::leaf::Leaf; use crate::btree::node::{NodeKind, body_capacity, read_header}; +use crate::btree::overflow; + +/// Encoded size a value of `value_len` bytes will occupy in a leaf record, +/// computed before the value is built. +/// +/// The preflight counterpart to [`LeafValue::encoded_size`], and it must agree +/// with it — both sides read the same constant rather than restating the +/// layout. +fn stored_leaf_value_encoded_size(value_len: usize, page_size: usize) -> Result { + if value_len > overflow::inline_value_threshold(page_size) { + Ok(crate::btree::leaf::OVERFLOW_REF_ENCODED_SIZE) + } else { + 2usize + .checked_add(value_len) + .ok_or(PagedbError::PayloadTooLarge) + } +} + +/// A descent reached a page twice, or followed a zero child pointer. +/// +/// Both mean the pointer graph has no terminator, which no honest writer can +/// produce. Named for the walk rather than reported as a generic header +/// failure, so `fsck` can say which page repeated and in which structure. +pub(super) fn malformed_btree_topology(structure: &'static str, page_id: u64) -> PagedbError { + PagedbError::page_chain_cycle(structure, page_id) +} + +// Keep common shallow descents stack-only and small; deeper trees spill into +// the exact set instead of weakening cycle detection. +const INLINE_SEEN_PAGE_IDS: usize = 4; + +/// Duplicate-page detector for B-tree descents. Shallow trees stay on the +/// inline array; deeper trees preserve exact detection through the overflow set. +pub(super) struct SeenPageIds { + /// Which walk this is, so a rejection can name the structure it was in. + structure: &'static str, + inline: [u64; INLINE_SEEN_PAGE_IDS], + len: usize, + overflow: Option>, +} + +impl SeenPageIds { + pub(super) fn new(structure: &'static str) -> Self { + Self { + structure, + inline: [0; INLINE_SEEN_PAGE_IDS], + len: 0, + overflow: None, + } + } + + pub(super) fn from_existing(structure: &'static str, path: &[u64]) -> Result { + let mut seen = Self::new(structure); + for &page_id in path { + seen.insert(page_id)?; + } + Ok(seen) + } + + pub(super) fn insert(&mut self, page_id: u64) -> Result<()> { + if page_id == 0 || !self.insert_inner(page_id) { + return Err(malformed_btree_topology(self.structure, page_id)); + } + Ok(()) + } + + fn insert_inner(&mut self, page_id: u64) -> bool { + if let Some(overflow) = &mut self.overflow { + return overflow.insert(page_id); + } + if self.inline[..self.len].contains(&page_id) { + return false; + } + if self.len < INLINE_SEEN_PAGE_IDS { + self.inline[self.len] = page_id; + self.len += 1; + return true; + } + let mut overflow = BTreeSet::new(); + overflow.extend(self.inline); + let inserted = overflow.insert(page_id); + self.overflow = Some(overflow); + inserted + } +} /// `CoW` B+ tree backed by the Pager. Single writer per instance; concurrent /// reads through `&self`. @@ -227,9 +312,14 @@ impl BTree { self.freed.push(page_id); } - pub(super) async fn read_node_kind(&self, page_id: u64) -> Result { - let (_g, kind) = self.read_node_guard(page_id).await?; - Ok(kind) + pub(super) fn validate_insert_record_fits(&self, key: &[u8], value: &[u8]) -> Result<()> { + let value_encoded_size = stored_leaf_value_encoded_size(value.len(), self.page_size)?; + if !Leaf::single_record_fits_encoded(key.len(), value_encoded_size, self.page_size) + || !internal::separator_fits(key.len(), self.page_size) + { + return Err(PagedbError::PayloadTooLarge); + } + Ok(()) } /// Read a B+ tree node page without knowing its kind in advance. The pager diff --git a/src/btree/tree/navigate.rs b/src/btree/tree/navigate.rs index 9d48060..4d2210d 100644 --- a/src/btree/tree/navigate.rs +++ b/src/btree/tree/navigate.rs @@ -7,7 +7,7 @@ use crate::btree::internal::{Internal, InternalAccessor, InternalEntry}; use crate::btree::node::NodeKind; use crate::btree::split::split_internal; -use super::core::BTree; +use super::core::{BTree, SeenPageIds}; impl BTree { /// Given an internal node and a child `page_id`, return the `page_id` of the @@ -195,7 +195,9 @@ impl BTree { pub(super) async fn path_to_leaf_for_key(&self, key: &[u8]) -> Result> { let mut path = Vec::new(); let mut page_id = self.root_page_id; + let mut seen = SeenPageIds::new("btree_descent"); loop { + seen.insert(page_id)?; path.push(page_id); // Fresh leaves (from in-txn splits) live only in `fresh_leaves` // until flush — the pager has no copy yet. Short-circuit the @@ -224,6 +226,7 @@ impl BTree { // Root is a leaf; no next leaf. return Ok(None); } + let mut seen = SeenPageIds::from_existing("btree_descent", path)?; let mut child = path[path.len() - 1]; for i in (0..path.len() - 1).rev() { let (guard, _kind) = self.read_node_guard(path[i]).await?; @@ -234,6 +237,7 @@ impl BTree { let mut new_path: Vec = path[..=i].to_vec(); let mut cur = next_child; loop { + seen.insert(cur)?; new_path.push(cur); // A fresh-from-split leaf has no pager presence yet. if self.fresh_leaves.contains_key(&cur) { @@ -260,7 +264,9 @@ impl BTree { pub(super) async fn path_to_rightmost_leaf(&self) -> Result> { let mut path = Vec::new(); let mut page_id = self.root_page_id; + let mut seen = SeenPageIds::new("btree_descent"); loop { + seen.insert(page_id)?; path.push(page_id); // Fresh-from-split leaves only live in `fresh_leaves`. if self.fresh_leaves.contains_key(&page_id) { diff --git a/src/btree/tree/read.rs b/src/btree/tree/read.rs index c2e099b..646abd9 100644 --- a/src/btree/tree/read.rs +++ b/src/btree/tree/read.rs @@ -11,7 +11,7 @@ use crate::btree::leaf::{Leaf, LeafAccessor, LeafValue, LeafValueRef}; use crate::btree::node::NodeKind; use crate::btree::overflow; -use super::core::BTree; +use super::core::{BTree, SeenPageIds}; impl BTree { /// Get a value by key. Returns `None` if not found. @@ -93,8 +93,11 @@ impl BTree { NodeKind::Internal => InternalAccessor::new(start_guard.body_ref())?.child_for(key), }; // Descend from the first child onward. Subsequent guards are owned. + let mut seen = SeenPageIds::new("btree_descent"); + seen.insert(start_page_id)?; let mut page_id = next_page_id; loop { + seen.insert(page_id)?; // Fresh-from-split leaves only live in `fresh_leaves` until // flush. If the next child is one, resolve from cache. if let Some(leaf) = self.fresh_leaves.get(&page_id) { diff --git a/src/btree/tree/write.rs b/src/btree/tree/write.rs index bd7efab..a474c76 100644 --- a/src/btree/tree/write.rs +++ b/src/btree/tree/write.rs @@ -7,30 +7,112 @@ use crate::errors::PagedbError; use crate::vfs::Vfs; use crate::btree::leaf::{Leaf, LeafValue}; -use crate::btree::node::NodeKind; use crate::btree::overflow; use crate::btree::split::split_leaf; -use super::core::BTree; +use super::core::{BTree, SeenPageIds}; impl BTree { - /// Insert or overwrite a key-value pair. - pub async fn put(&mut self, key: &[u8], value: &[u8]) -> Result<()> { - // Build the leaf value — inline if small enough, overflow chain otherwise. - let leaf_value = if value.len() > overflow::inline_value_threshold(self.page_size) { + fn invalidate_append_state(&mut self) { + self.append_cached_path = None; + self.append_last_key = None; + } + + fn cached_leaf_mut(&mut self, page_id: u64, leaf_is_fresh: bool) -> &mut Leaf { + if leaf_is_fresh { + self.fresh_leaves + .get_mut(&page_id) + .expect("fresh_leaves contains key") + } else { + self.dirty_leaves + .get_mut(&page_id) + .expect("ensure_leaf_dirty populated") + } + } + + fn apply_release_result( + &mut self, + old_root: u64, + unused_cow_page: u64, + release_result: overflow::ReleaseResult, + ) { + match release_result { + overflow::ReleaseResult::Freed { freed_pages } => { + for page_id in freed_pages { + self.free_page(page_id); + } + self.free_page(unused_cow_page); + } + overflow::ReleaseResult::Decremented { + new_root_page_id: _, + } => self.free_page(old_root), + } + } + + /// Return the pages of a chain that was written for a value the caller + /// never got to commit. + /// + /// Best-effort by design: this runs on an error path, and a failure here + /// must not replace the error the caller actually needs to see. It is + /// logged rather than swallowed, because the cost of giving up is a page + /// leak that nothing else will notice. + async fn discard_uncommitted_value(&mut self, value: Option) { + let Some(LeafValue::Overflow { root_page_id, .. }) = value else { + return; + }; + match overflow::collect_chain(&self.pager, self.realm_id, root_page_id).await { + Ok(page_ids) => { + for page_id in page_ids { + self.free_page(page_id); + } + } + Err(error) => tracing::warn!( + root_page_id, + error = %error, + "could not reclaim the overflow chain of an uncommitted value; its pages are leaked" + ), + } + } + + async fn max_key_at_path(&self, path: &[u64]) -> Result>> { + let leaf_page_id = *path.last().expect("non-empty path"); + if let Some(leaf) = self.fresh_leaves.get(&leaf_page_id) { + return Ok(leaf.records.last().map(|(key, _)| key.clone())); + } + if let Some(leaf) = self.dirty_leaves.get(&leaf_page_id) { + return Ok(leaf.records.last().map(|(key, _)| key.clone())); + } + Ok(self + .decode_leaf_from_pager(leaf_page_id) + .await? + .records + .last() + .map(|(key, _)| key.clone())) + } + + async fn leaf_value_for(&mut self, value: &[u8]) -> Result { + if value.len() > overflow::inline_value_threshold(self.page_size) { let realm = self.realm_id; - let ps = self.page_size; + let page_size = self.page_size; let pager = Arc::clone(&self.pager); - let root_id = - overflow::write_chain(&pager, realm, value, ps, &mut || self.allocate_page()) - .await?; - LeafValue::Overflow { + let root_page_id = overflow::write_chain(&pager, realm, value, page_size, &mut || { + self.allocate_page() + }) + .await?; + Ok(LeafValue::Overflow { total_len: value.len() as u64, - root_page_id: root_id, - } + root_page_id, + }) } else { - LeafValue::Inline(value.to_vec()) - }; + Ok(LeafValue::Inline(value.to_vec())) + } + } + + /// Insert or overwrite a key-value pair. + pub async fn put(&mut self, key: &[u8], value: &[u8]) -> Result<()> { + self.invalidate_append_state(); + self.validate_insert_record_fits(key, value)?; + let leaf_value = self.leaf_value_for(value).await?; if self.root_page_id == 0 { let new_root = self.allocate_page(); @@ -49,10 +131,6 @@ impl BTree { // Descend to find the leaf page_id. Returns the path; we'll fetch the // leaf via the dirty-cache-aware helper below. let path = self.path_to_leaf_for_key(key).await?; - // Regular `put` invalidates the append cache — caller may have - // mutated keys that aren't in monotonic order vs the cache. - self.append_cached_path = None; - self.append_last_key = None; let _split = self.put_at_path(path, key, leaf_value).await?; Ok(()) } @@ -86,40 +164,36 @@ impl BTree { self.ensure_leaf_dirty(leaf_page_id, &path).await?; } - let leaf: &mut Leaf = if leaf_is_fresh { - self.fresh_leaves - .get_mut(&leaf_page_id) - .expect("fresh_leaves contains key") - } else { - self.dirty_leaves - .get_mut(&leaf_page_id) - .expect("ensure_leaf_dirty populated") - }; + let page_size = self.page_size; + let leaf = self.cached_leaf_mut(leaf_page_id, leaf_is_fresh); let monotonic = leaf.records.last().is_some_and(|(k, _)| key > k.as_slice()) && !leaf.records.iter().any(|(k, _)| k.as_slice() == key); let (_is_new, old_value) = leaf.upsert(key, leaf_value); - let fits = leaf.fits(self.page_size); + let fits = leaf.fits(page_size); // Free old overflow chain if a record was replaced (refcount-aware). - if let Some(LeafValue::Overflow { - root_page_id: old_root, - .. - }) = old_value + if let Some( + old_value @ LeafValue::Overflow { + root_page_id: old_root, + .. + }, + ) = old_value { let new_cow_page = self.allocate_page(); - match overflow::release(&self.pager, self.realm_id, old_root, new_cow_page).await? { - overflow::ReleaseResult::Freed { freed_pages } => { - for pid in freed_pages { - self.free_page(pid); - } - self.free_page(new_cow_page); + match overflow::release(&self.pager, self.realm_id, old_root, new_cow_page).await { + Ok(release_result) => { + self.apply_release_result(old_root, new_cow_page, release_result); } - overflow::ReleaseResult::Decremented { - new_root_page_id: _, - } => { - self.free_page(old_root); + Err(error) => { + self.free_page(new_cow_page); + let attempted_value = self + .cached_leaf_mut(leaf_page_id, leaf_is_fresh) + .upsert(key, old_value) + .1; + self.discard_uncommitted_value(attempted_value).await; + return Err(error); } } } @@ -175,7 +249,11 @@ impl BTree { /// # Errors /// /// Returns [`PagedbError::AppendNotMonotonic`] if `key` is not strictly - /// greater than the previously-appended key in this `BTree` session. + /// greater than the previously-appended key in this `BTree` session, or — + /// on the first append of a session, when there is no such key yet — not + /// strictly greater than the largest key already in the tree. Both are the + /// same invariant: the cached rightmost path is only meaningful while every + /// append lands to the right of everything before it. /// Intended for op-logs, time-series indexes, FTS posting-list builds — /// any workload where the embedder can guarantee monotonic-key insert. pub async fn put_append(&mut self, key: &[u8], value: &[u8]) -> Result<()> { @@ -185,39 +263,38 @@ impl BTree { return Err(PagedbError::AppendNotMonotonic); } } + self.validate_insert_record_fits(key, value)?; - // Empty tree: delegate to the regular `put` path, which initializes - // the root and writes the first leaf. After it returns, seed the - // append cache with the new key. + // Initialize directly: regular put invalidates append state by design. if self.root_page_id == 0 { - self.put(key, value).await?; + let leaf_value = self.leaf_value_for(value).await?; + let new_root = self.allocate_page(); + let mut leaf = Leaf::new(); + leaf.upsert(key, leaf_value); + self.write_leaf(new_root, &leaf).await?; + self.root_page_id = new_root; self.append_last_key = Some(key.to_vec()); - // The cached path needs a real descent on the next call. + self.append_cached_path = Some(vec![new_root]); return Ok(()); } - // Build the leaf value (inline or overflow). - let leaf_value = if value.len() > self.page_size / 4 { - let realm = self.realm_id; - let ps = self.page_size; - let pager = Arc::clone(&self.pager); - let root_id = - overflow::write_chain(&pager, realm, value, ps, &mut || self.allocate_page()) - .await?; - LeafValue::Overflow { - total_len: value.len() as u64, - root_page_id: root_id, - } - } else { - LeafValue::Inline(value.to_vec()) - }; - // Fast path: cached rightmost path is valid → skip descent. - let path = if let Some(cached) = self.append_cached_path.take() { - cached - } else { - self.path_to_rightmost_leaf().await? + let path = match &self.append_cached_path { + Some(cached) => cached.clone(), + None => self.path_to_rightmost_leaf().await?, }; + // Checked before the cached path is consumed: a rejected append must + // leave the session exactly as it found it, cache included. + if self.append_last_key.is_none() + && self + .max_key_at_path(&path) + .await? + .is_some_and(|max_key| key <= max_key.as_slice()) + { + return Err(PagedbError::AppendNotMonotonic); + } + self.append_cached_path = None; + let leaf_value = self.leaf_value_for(value).await?; let path_for_retry = path.clone(); let split = self.put_at_path(path, key, leaf_value).await?; self.append_last_key = Some(key.to_vec()); @@ -258,8 +335,7 @@ impl BTree { // `delete` may have removed the previously-appended max, so the // monotonic invariant on `put_append` can no longer be enforced // against `append_last_key`. Reset the append state. - self.append_cached_path = None; - self.append_last_key = None; + self.invalidate_append_state(); if self.root_page_id == 0 { return Ok(false); } @@ -269,35 +345,27 @@ impl BTree { if !leaf_is_fresh { self.ensure_leaf_dirty(leaf_page_id, &path).await?; } - let removed = if leaf_is_fresh { - self.fresh_leaves - .get_mut(&leaf_page_id) - .expect("fresh contains key") - .remove(key) - } else { - self.dirty_leaves - .get_mut(&leaf_page_id) - .expect("dirty after ensure") - .remove(key) - }; + let removed = self + .cached_leaf_mut(leaf_page_id, leaf_is_fresh) + .remove(key); match removed { None => return Ok(false), - Some(LeafValue::Overflow { - root_page_id: old_root, - .. - }) => { + Some( + old_value @ LeafValue::Overflow { + root_page_id: old_root, + .. + }, + ) => { let new_cow_page = self.allocate_page(); - match overflow::release(&self.pager, self.realm_id, old_root, new_cow_page).await? { - overflow::ReleaseResult::Freed { freed_pages } => { - for pid in freed_pages { - self.free_page(pid); - } - self.free_page(new_cow_page); + match overflow::release(&self.pager, self.realm_id, old_root, new_cow_page).await { + Ok(release_result) => { + self.apply_release_result(old_root, new_cow_page, release_result); } - overflow::ReleaseResult::Decremented { - new_root_page_id: _, - } => { - self.free_page(old_root); + Err(error) => { + self.free_page(new_cow_page); + self.cached_leaf_mut(leaf_page_id, leaf_is_fresh) + .upsert(key, old_value); + return Err(error); } } } @@ -341,23 +409,15 @@ impl BTree { if self.root_page_id == 0 { return Ok(0); } - // Collect all keys in range (keys only, no values needed). - let mut page_id = self.root_page_id; - loop { - if self.fresh_leaves.contains_key(&page_id) { - break; - } - let kind = self.read_node_kind(page_id).await?; - if kind == NodeKind::Leaf { - break; - } - let internal = self.read_internal(page_id).await?; - page_id = internal.child_for(start); - } + // Parent paths are authoritative for fresh same-session split leaves; + // their sibling links remain zero until flush. + let mut path = self.path_to_leaf_for_key(start).await?; let mut keys_to_delete: Vec> = Vec::new(); - let mut next = page_id; - 'outer: while next != 0 { - let leaf = self.read_leaf(next).await?; + let mut seen_leaves = SeenPageIds::new("leaf_siblings"); + 'outer: loop { + let leaf_page_id = *path.last().expect("non-empty path"); + seen_leaves.insert(leaf_page_id)?; + let leaf = self.read_leaf(leaf_page_id).await?; for (k, _) in &leaf.records { if k.as_slice() >= end { break 'outer; @@ -366,7 +426,23 @@ impl BTree { keys_to_delete.push(k.clone()); } } - next = leaf.right_sibling; + let next_path = self.next_leaf_after(&path).await?; + match (leaf.right_sibling, next_path) { + (0, Some(parent_next)) => path = parent_next, + (0, None) => break, + (right_sibling, Some(parent_next)) + if parent_next.last().copied() == Some(right_sibling) => + { + path = parent_next; + } + (right_sibling, parent_next) => { + return Err(PagedbError::leaf_sibling_mismatch( + leaf_page_id, + right_sibling, + parent_next.and_then(|path| path.last().copied()), + )); + } + } } let count = keys_to_delete.len() as u64; for key in keys_to_delete { diff --git a/src/errors.rs b/src/errors.rs index d9dd3b4..7c99bab 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -127,6 +127,16 @@ pub enum PagedbError { #[error("put_append called with non-monotonic key")] AppendNotMonotonic, + /// `BTree::bulk_load` was handed input whose keys are not strictly + /// increasing — descending, or repeating the same key. + /// + /// The sibling of [`Self::AppendNotMonotonic`]: same invariant, different + /// entry point. Bulk load builds the tree bottom-up from the input order + /// itself, so unsorted input does not merely misplace a record — it would + /// produce a tree whose separators do not describe its leaves. + #[error("bulk_load keys must be strictly increasing")] + BulkLoadNotMonotonic, + /// The deferred-free backlog exceeds the configured threshold and /// active reader pins prevent draining it. #[non_exhaustive] @@ -306,6 +316,33 @@ pub enum CorruptionDetail { /// has no terminator. Distinct from a truncated chain: the links /// authenticate, they just form a loop. OverflowChainCycle { root_page_id: u64, page_id: u64 }, + /// A linked page structure revisited a page it had already walked, so the + /// walk has no terminator. + /// + /// The overflow-chain form is [`Self::OverflowChainCycle`], which can also + /// name the chain's root. This is the general case — a B+ tree root-to-leaf + /// descent, a leaf sibling walk, or the durable free-list chain — where + /// `structure` says which walk found the loop and `page_id` is the page it + /// reached twice. Every link authenticates; they simply form a loop, which + /// no honest writer can produce. + PageChainCycle { + structure: &'static str, + page_id: u64, + }, + /// A leaf's persisted right-sibling link disagrees with the leaf its parent + /// path says comes next. + /// + /// Both encode "the next leaf", so both must name it. A disagreement means + /// one of them outlived the page it points at — a sibling link left behind + /// by a page that was freed and handed to another node, or a parent whose + /// child pointer was rewritten without its leaves. `parent_next` is `None` + /// when the parent path has no successor at all yet the leaf still claims + /// one. + LeafSiblingMismatch { + leaf_page_id: u64, + right_sibling: u64, + parent_next: Option, + }, /// One physical page was reached twice in a single traversal under two /// incompatible page kinds. /// @@ -426,6 +463,29 @@ impl PagedbError { }) } + /// Canonical constructor for a cyclic linked page structure. `structure` + /// names the walk that found the loop — `"btree_descent"`, + /// `"leaf_siblings"`, `"free_list"`. + #[must_use] + pub const fn page_chain_cycle(structure: &'static str, page_id: u64) -> Self { + Self::Corruption(CorruptionDetail::PageChainCycle { structure, page_id }) + } + + /// Canonical constructor for a leaf whose sibling link and parent path + /// disagree about which leaf comes next. + #[must_use] + pub const fn leaf_sibling_mismatch( + leaf_page_id: u64, + right_sibling: u64, + parent_next: Option, + ) -> Self { + Self::Corruption(CorruptionDetail::LeafSiblingMismatch { + leaf_page_id, + right_sibling, + parent_next, + }) + } + /// Canonical constructor for a cyclic overflow chain. #[must_use] pub const fn overflow_chain_cycle(root_page_id: u64, page_id: u64) -> Self { diff --git a/src/pager/freelist.rs b/src/pager/freelist.rs index e9c120a..e38349d 100644 --- a/src/pager/freelist.rs +++ b/src/pager/freelist.rs @@ -46,8 +46,12 @@ pub async fn read_chain( ) -> Result<(Vec<(u64, u64)>, Vec)> { let mut entries = Vec::new(); let mut chain_pages = Vec::new(); + let mut seen = HashSet::new(); let mut page = head; while page != 0 { + if !seen.insert(page) { + return Err(PagedbError::page_chain_cycle("free_list", page)); + } let guard = pager.read_main_page(page, realm_id, PageKind::Free).await?; let body = guard.body_ref(); let mut next_b = [0u8; 8]; @@ -182,3 +186,63 @@ pub async fn write_chain( } Ok(chain_pages[0]) } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::Duration; + + use crate::crypto::CipherId; + use crate::crypto::kdf::derive_mk; + use crate::pager::PagerConfig; + use crate::vfs::memory::MemVfs; + + use super::*; + + const PAGE: usize = 4096; + const REALM: RealmId = RealmId::new([0xF1; 16]); + + async fn test_pager() -> Arc> { + let mk = derive_mk(&[0xF2; 32], &[0u8; 16], 0).unwrap(); + let config = PagerConfig { + page_size: PAGE, + buffer_pool_pages: 16, + segment_cache_pages: 16, + cipher_id: CipherId::Aes256Gcm, + mk_epoch: 0, + main_db_file_id: [0xF3; 16], + main_db_path: "/main.db".into(), + anchor_budget: 1_000_000, + dek_lru_capacity: 16, + observer_retry_count: 0, + metrics_enabled: true, + }; + Arc::new(Pager::open(MemVfs::new(), mk, config).await.unwrap()) + } + + #[tokio::test(flavor = "current_thread")] + async fn read_chain_rejects_cycle_without_hanging() { + let pager = test_pager().await; + let mut body = vec![0u8; body_capacity(PAGE)]; + body[0..8].copy_from_slice(&10u64.to_le_bytes()); + pager + .write_main_page(10, REALM, PageKind::Free, &body) + .await + .unwrap(); + + let error = tokio::time::timeout(Duration::from_secs(1), read_chain(&pager, REALM, 10)) + .await + .expect("cycle detection should return before timeout") + .expect_err("free-list cycles must be corruption"); + assert!( + matches!( + error, + PagedbError::Corruption(crate::errors::CorruptionDetail::PageChainCycle { + structure: "free_list", + page_id: 10, + }) + ), + "expected a free-list PageChainCycle naming page 10, got {error:?}" + ); + } +} diff --git a/tests/btree_basic.rs b/tests/btree_basic.rs index 0c0be31..e5fca6f 100644 --- a/tests/btree_basic.rs +++ b/tests/btree_basic.rs @@ -1,9 +1,12 @@ use std::sync::Arc; +use std::sync::mpsc; +use std::time::Duration; use pagedb::btree::BTree; use pagedb::btree::internal::Internal; use pagedb::btree::leaf::{Leaf, LeafValue}; use pagedb::btree::node::body_capacity; +use pagedb::btree::overflow::encode_overflow; use pagedb::crypto::CipherId; use pagedb::crypto::kdf::derive_mk; use pagedb::errors::CorruptionDetail; @@ -35,6 +38,45 @@ fn fresh_tree(pager: Arc>) -> BTree { BTree::open(pager, RealmId::new([1; 16]), 0, 4, PAGE) } +async fn tree_with_unreadable_overflow_root( + pager: Arc>, + key: &[u8], + leaf_page_id: u64, + overflow_root_page_id: u64, +) -> BTree { + let realm = RealmId::new([1; 16]); + let mut leaf = Leaf::new(); + leaf.upsert( + key, + LeafValue::Overflow { + total_len: (PAGE * 2) as u64, + root_page_id: overflow_root_page_id, + }, + ); + let mut leaf_body = vec![0u8; body_capacity(PAGE)]; + leaf.encode(&mut leaf_body).unwrap(); + pager + .write_main_page(leaf_page_id, realm, PageKind::BTreeLeaf, &leaf_body) + .await + .unwrap(); + + // Deliberately seal the referenced root under the chain-page kind. The + // leaf remains readable, but releasing its old overflow value must fail. + let mut overflow_body = vec![0u8; body_capacity(PAGE)]; + encode_overflow(&mut overflow_body, 0, b"old").unwrap(); + pager + .write_main_page( + overflow_root_page_id, + realm, + PageKind::Overflow, + &overflow_body, + ) + .await + .unwrap(); + + BTree::open(pager, realm, leaf_page_id, overflow_root_page_id + 1, PAGE) +} + #[tokio::test(flavor = "current_thread")] async fn empty_tree_get_returns_none() { let pager = fresh_pager().await; @@ -270,7 +312,10 @@ async fn put_append_after_regular_put_re_descends() { // Regular put may target any leaf — invalidates the append cache and // resets the monotonic tracker. tree.put(b"middle", b"v").await.unwrap(); - // After invalidation, put_append accepts any key (cache reset). + // Cache invalidation forces a fresh rightmost descent; append must still + // compare against the tree's actual maximum key. + let error = tree.put_append(b"b00", b"v").await.unwrap_err(); + assert!(matches!(error, PagedbError::AppendNotMonotonic)); tree.put_append(b"z99", b"v").await.unwrap(); // Now further appends must be > "z99". assert!(tree.put_append(b"z00", b"v").await.is_err()); @@ -406,3 +451,326 @@ async fn malformed_node_body_reports_corruption_instead_of_panicking() { ); } } + +#[test] +fn delete_range_rejects_leaf_sibling_cycle_without_hanging() { + let (tx, rx) = mpsc::channel(); + + std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let result = runtime.block_on(async { + let pager = fresh_pager().await; + let realm = RealmId::new([1; 16]); + let leaf_page_id = 31; + + let mut leaf = Leaf::new(); + leaf.right_sibling = leaf_page_id; + leaf.upsert(b"k", LeafValue::Inline(b"v".to_vec())); + let mut body = vec![0u8; body_capacity(PAGE)]; + leaf.encode(&mut body).unwrap(); + pager + .write_main_page(leaf_page_id, realm, PageKind::BTreeLeaf, &body) + .await + .unwrap(); + + let mut tree = BTree::open(pager, realm, leaf_page_id, 32, PAGE); + tree.delete_range(b"a", b"z").await.map(|_| ()) + }); + let _ = tx.send(result); + }); + + let result = rx + .recv_timeout(Duration::from_secs(5)) + .expect("delete_range sibling-cycle detection should return before the timeout"); + let error = result.expect_err("leaf sibling cycles must not be accepted"); + assert!( + matches!( + error, + PagedbError::Corruption(CorruptionDetail::LeafSiblingMismatch { .. }) + ), + "expected LeafSiblingMismatch, got {error:?}" + ); +} + +#[test] +fn get_rejects_internal_child_cycle_without_hanging() { + let (tx, rx) = mpsc::channel(); + + std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let result = runtime.block_on(async { + let pager = fresh_pager().await; + let realm = RealmId::new([1; 16]); + let root_page_id = 41; + let internal = Internal { + leftmost_child: root_page_id, + entries: Vec::new(), + }; + let mut body = vec![0u8; body_capacity(PAGE)]; + internal.encode(&mut body).unwrap(); + pager + .write_main_page(root_page_id, realm, PageKind::BTreeInternal, &body) + .await + .unwrap(); + let tree = BTree::open(pager, realm, root_page_id, 42, PAGE); + tree.get(b"k").await.map(|_| ()) + }); + let _ = tx.send(result); + }); + + let result = rx + .recv_timeout(Duration::from_secs(5)) + .expect("get internal-cycle detection should return before the timeout"); + let error = result.expect_err("internal child cycles must not be accepted"); + assert!( + matches!( + error, + PagedbError::Corruption(CorruptionDetail::PageChainCycle { + structure: "btree_descent", + .. + }) + ), + "expected a descent PageChainCycle, got {error:?}" + ); +} + +#[test] +fn put_rejects_internal_child_cycle_without_hanging() { + let (tx, rx) = mpsc::channel(); + + std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let result = runtime.block_on(async { + let pager = fresh_pager().await; + let realm = RealmId::new([1; 16]); + let root_page_id = 43; + let internal = Internal { + leftmost_child: root_page_id, + entries: Vec::new(), + }; + let mut body = vec![0u8; body_capacity(PAGE)]; + internal.encode(&mut body).unwrap(); + pager + .write_main_page(root_page_id, realm, PageKind::BTreeInternal, &body) + .await + .unwrap(); + let mut tree = BTree::open(pager, realm, root_page_id, 44, PAGE); + tree.put(b"k", b"v").await + }); + let _ = tx.send(result); + }); + + let result = rx + .recv_timeout(Duration::from_secs(5)) + .expect("put internal-cycle detection should return before the timeout"); + let error = result.expect_err("internal child cycles must not be accepted"); + assert!( + matches!( + error, + PagedbError::Corruption(CorruptionDetail::PageChainCycle { + structure: "btree_descent", + .. + }) + ), + "expected a descent PageChainCycle, got {error:?}" + ); +} + +#[test] +fn bulk_load_rejects_separator_that_cannot_fit_without_hanging() { + let (tx, rx) = mpsc::channel(); + + std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let result = runtime.block_on(async { + let pager = fresh_pager().await; + let mut tree = fresh_tree(pager); + let key_len = body_capacity(PAGE) - 32; + tree.bulk_load(vec![ + (vec![b'a'; key_len], Vec::new()), + (vec![b'b'; key_len], Vec::new()), + ]) + .await + }); + let _ = tx.send(result); + }); + + let result = rx + .recv_timeout(Duration::from_secs(5)) + .expect("bulk_load separator validation should return before the timeout"); + let error = result.expect_err("oversize internal separators must be rejected"); + assert!(matches!(error, PagedbError::PayloadTooLarge)); +} + +#[tokio::test(flavor = "current_thread")] +async fn bulk_load_rejects_non_strict_key_order_without_poisoning_tree() { + let cases = [ + vec![ + (b"b".to_vec(), b"two".to_vec()), + (b"a".to_vec(), b"one".to_vec()), + ], + vec![ + (b"a".to_vec(), b"one".to_vec()), + (b"a".to_vec(), b"two".to_vec()), + ], + ]; + + for pairs in cases { + let pager = fresh_pager().await; + let mut tree = fresh_tree(pager); + let error = tree + .bulk_load(pairs) + .await + .expect_err("bulk_load must reject unsorted or duplicate keys"); + assert!( + matches!(error, PagedbError::BulkLoadNotMonotonic), + "expected BulkLoadNotMonotonic, got {error:?}" + ); + assert_eq!(tree.root_page_id(), 0); + assert_eq!(tree.next_page_id(), 4); + + tree.bulk_load(vec![ + (b"a".to_vec(), b"one".to_vec()), + (b"b".to_vec(), b"two".to_vec()), + ]) + .await + .unwrap(); + assert_eq!( + tree.get(b"a").await.unwrap().as_deref(), + Some(b"one".as_ref()) + ); + assert_eq!( + tree.get(b"b").await.unwrap().as_deref(), + Some(b"two".as_ref()) + ); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn put_rejects_oversized_key_without_poisoning_tree() { + let pager = fresh_pager().await; + let mut tree = fresh_tree(pager); + tree.put(b"small", b"value").await.unwrap(); + + let oversized_key = vec![b'z'; body_capacity(PAGE)]; + let error = tree + .put(&oversized_key, b"bad") + .await + .expect_err("oversized keys must be rejected at put time"); + assert!(matches!(error, PagedbError::PayloadTooLarge)); + assert_eq!( + tree.get(b"small").await.unwrap().as_deref(), + Some(b"value".as_ref()) + ); + assert!(tree.get(&oversized_key).await.unwrap().is_none()); + tree.put(b"valid", b"good").await.unwrap(); + tree.flush().await.unwrap(); + assert_eq!( + tree.get(b"valid").await.unwrap().as_deref(), + Some(b"good".as_ref()) + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn put_append_rejects_oversized_key_without_poisoning_tree() { + let pager = fresh_pager().await; + let mut tree = fresh_tree(pager); + tree.put_append(b"small", b"value").await.unwrap(); + + let oversized_key = vec![b'z'; body_capacity(PAGE)]; + let error = tree + .put_append(&oversized_key, b"bad") + .await + .expect_err("append mode must reject oversized keys at put time"); + assert!(matches!(error, PagedbError::PayloadTooLarge)); + assert_eq!( + tree.get(b"small").await.unwrap().as_deref(), + Some(b"value".as_ref()) + ); + assert!(tree.get(&oversized_key).await.unwrap().is_none()); + tree.put_append(b"valid", b"good").await.unwrap(); + tree.flush().await.unwrap(); + assert_eq!( + tree.get(b"valid").await.unwrap().as_deref(), + Some(b"good".as_ref()) + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn put_release_failure_does_not_publish_replacement() { + let pager = fresh_pager().await; + let mut tree = tree_with_unreadable_overflow_root(pager, b"k", 80, 81).await; + + let error = tree + .put(b"k", b"new") + .await + .expect_err("replacing an unreadable overflow value must fail"); + assert!(matches!( + error, + PagedbError::ChecksumFailure | PagedbError::Corruption(_) | PagedbError::Io(_) + )); + match tree.get(b"k").await { + Err(PagedbError::ChecksumFailure | PagedbError::Corruption(_) | PagedbError::Io(_)) => {} + Ok(Some(value)) => panic!("failed replacement published value {value:?}"), + Ok(None) => panic!("failed replacement removed the original key"), + Err(error) => panic!("unexpected post-failure read error: {error:?}"), + } +} + +#[tokio::test(flavor = "current_thread")] +async fn delete_release_failure_does_not_remove_key() { + let pager = fresh_pager().await; + let mut tree = tree_with_unreadable_overflow_root(pager, b"k", 90, 91).await; + + let error = tree + .delete(b"k") + .await + .expect_err("deleting an unreadable overflow value must fail"); + assert!(matches!( + error, + PagedbError::ChecksumFailure | PagedbError::Corruption(_) | PagedbError::Io(_) + )); + match tree.get(b"k").await { + Err(PagedbError::ChecksumFailure | PagedbError::Corruption(_) | PagedbError::Io(_)) + | Ok(Some(_)) => {} + Ok(None) => panic!("failed delete removed the original key"), + Err(error) => panic!("unexpected post-failure read error: {error:?}"), + } +} + +#[tokio::test(flavor = "current_thread")] +async fn delete_range_visits_fresh_split_leaves_before_flush() { + let pager = fresh_pager().await; + let mut tree = fresh_tree(pager); + + for i in 0..2_000u32 { + let key = format!("k{i:08}"); + let value = format!("v-{i:08}-").repeat(8); + tree.put(key.as_bytes(), value.as_bytes()).await.unwrap(); + } + + let deleted = tree.delete_range(b"k00000500", b"k00001500").await.unwrap(); + assert_eq!(deleted, 1_000); + + for i in 0..2_000u32 { + let key = format!("k{i:08}"); + let got = tree.get(key.as_bytes()).await.unwrap(); + if (500..1_500).contains(&i) { + assert!(got.is_none(), "key {key} should have been deleted"); + } else { + assert!(got.is_some(), "key {key} should remain visible"); + } + } +} diff --git a/tests/smoke.rs b/tests/smoke.rs index 28a920f..54ef6a9 100644 --- a/tests/smoke.rs +++ b/tests/smoke.rs @@ -42,6 +42,9 @@ fn every_variant_displays() { PagedbError::ChecksumFailure, PagedbError::corruption(CorruptionDetail::HeaderUnverifiable), PagedbError::structural_header_invalid("main.db", "magic"), + PagedbError::page_chain_cycle("btree_descent", 41), + PagedbError::leaf_sibling_mismatch(31, 31, None), + PagedbError::BulkLoadNotMonotonic, PagedbError::vfs_contract_violated( "read_at", "reported more bytes than the caller requested",