From 80bc9e4f9277f8e6e258182d6f744cfe1b7251e9 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 7 Sep 2026 21:04:25 +0700 Subject: [PATCH 1/6] Replace eyre with a concrete error type `eyre::Report` was in the return type of every fallible function here, so the crate reached `std` through its own signatures. Nothing about page framing needs an operating system: the layout is bytes and the checks are arithmetic. This is the type that lets the rest of the crate say so, and it is the prerequisite for the file access moving behind a trait - doing that first would buy nothing while every signature still named a `std` type. The enum is also more useful than formatted prose. A caller could not tell "this page is full" from "these bytes are damaged", because both arrived as a `Report` carrying a string. Six variants now carry the numbers that justify them: LinkLengthMismatch, LinkOutOfBounds, PageOverflow, Corrupt, Encode, Io. 18 construction sites, 7 files, no .context or .wrap_err chains to unpick Io holds raw_os_error rather than an io::Error, which is the std type this change exists to stop depending on Three tests asserted on eyre's message strings and now match on variants with their fields, which is what the change is for. cargo test 69 + 2 passed, 0 failed clippy clean **This breaks consumers**, so 0.6.0 rather than 0.5.8. Not every `?` converts: a function returning `persist_page(..).await` in tail position needs `?` and an `Ok(())`, and `error.wrap_err(..)` has to become `eyre::Report::new(error).wrap_err(..)`. WorkTable needs exactly three such edits, verified by building it against this branch; the patch is not applied there because that checkout has another agent's uncommitted work in it. --- Cargo.toml | 3 +- src/error.rs | 154 ++++++++++++++++++++ src/lib.rs | 1 + src/page/data.rs | 69 +++++---- src/page/index/mod.rs | 13 +- src/page/index/page.rs | 32 ++-- src/page/index/page_cdc_impl.rs | 14 +- src/page/index/page_for_unsized.rs | 43 +++--- src/page/index/page_for_unsized_cdc_impl.rs | 14 +- src/page/util.rs | 91 ++++++------ 10 files changed, 305 insertions(+), 129 deletions(-) create mode 100644 src/error.rs diff --git a/Cargo.toml b/Cargo.toml index c086a1e..5ad1c51 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ members = ["codegen", "tools/create-data-file", "tools/dump-data-file"] [package] name = "data_bucket" -version = "0.5.7" +version = "0.6.0" edition = "2021" authors = ["Handy-caT"] license = "MIT" @@ -13,7 +13,6 @@ description = "DataBucket is container for WorkTable's data" [dependencies] data_bucket_derive = { path = "codegen", version = "^0.3" } -eyre = "0.6.12" derive_more = { version = "1.0.0", features = ["from", "error", "display", "into"] } rkyv = { version = "0.8.17", features = ["uuid-1"] } uuid = { version = "1.11.0", features = ["v4"] } diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..4e28872 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,154 @@ +//! What this crate can refuse on. +//! +//! # Why this exists rather than `eyre` +//! +//! `eyre::Report` is a `std` type, and it was in the return type of every +//! fallible function here, so the whole crate reached `std` through its own +//! signatures. Nothing else about page framing needs an operating system: the +//! layout is bytes, the checks are arithmetic. This is the type that lets the +//! rest of the crate say so. +//! +//! It is also more useful than a formatted string. A caller that wants to +//! distinguish "this page is full" from "these bytes are damaged" could not, +//! because both arrived as a `Report` carrying prose. +//! +//! Consumers keep working: `Error` implements `core::error::Error`, so `?` +//! into an `eyre::Result` converts exactly as it did before. + +use core::fmt::{Display, Formatter, Result as FmtResult}; + +use crate::page::PageId; + +/// The result of anything in this crate that can fail. +pub type Result = core::result::Result; + +/// A refusal, with the numbers that justify it. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Error { + /// A write's length does not match the link it was given. + /// + /// A `Link` names an exact byte range, so a write of another size is not + /// a resize: it is a write that would land on the neighbouring row. + LinkLengthMismatch { + /// The length the link reserves. + expected: u32, + /// The length the caller supplied. + found: usize, + }, + /// A link's range runs past the end of the page it points into. + /// + /// Summed in `u64`, because `offset + length` in `u32` can wrap past 4 GiB + /// and slip under the bound. + LinkOutOfBounds { + /// Where the link starts. + offset: u32, + /// How far it runs. + length: u32, + /// How much the page holds. + capacity: usize, + }, + /// A page's contents do not fit the page. + /// + /// Raised where the write is prepared rather than where it lands, so an + /// over-budget page fails in its own persist instead of quietly writing + /// into its neighbour. + PageOverflow { + /// The page whose write is over budget. + page: PageId, + /// Bytes the write needs. + needed: usize, + /// Bytes available. + capacity: usize, + }, + /// Bytes that should have been a structure were not one. + /// + /// The label says which structure, because "corrupt" on its own does not + /// tell an operator which part of a file to distrust. + Corrupt { + /// What failed to parse. + what: &'static str, + }, + /// A value would not archive. + /// + /// rkyv reports this when its allocator refuses, which on this path means + /// the process is already out of memory. + Encode, + /// The file would not answer. + /// + /// Held as the raw OS code rather than an `io::Error`, because that type + /// is `std` and this one is the reason the crate stops needing it. The + /// code is what an operator acts on; the message that came with it says + /// nothing the code does not. + Io { + /// `raw_os_error`, when the failure came from the operating system. + code: Option, + }, + /// A change-data event arrived that this page cannot apply. + /// + /// `SplitNode`, `CreateNode` and `RemoveNode` change which pages exist, + /// which is the caller's business rather than one page's. + UnapplicableEvent, +} + +impl Display for Error { + fn fmt(&self, formatter: &mut Formatter<'_>) -> FmtResult { + match self { + Self::LinkLengthMismatch { expected, found } => write!( + formatter, + "a {found} byte write does not match its {expected} byte link" + ), + Self::LinkOutOfBounds { + offset, + length, + capacity, + } => write!( + formatter, + "a link at {offset} running {length} bytes leaves a {capacity} byte page" + ), + Self::PageOverflow { + page, + needed, + capacity, + } => write!( + formatter, + "page {page:?} needs {needed} bytes of a {capacity} byte page" + ), + Self::Corrupt { what } => write!(formatter, "torn or corrupt {what}"), + Self::Encode => write!(formatter, "a value would not archive"), + Self::Io { code: Some(code) } => write!(formatter, "the file failed, os error {code}"), + Self::Io { code: None } => write!(formatter, "the file failed"), + Self::UnapplicableEvent => write!( + formatter, + "events of `SplitNode`, `CreateNode` or `RemoveNode` cannot be applied to a page" + ), + } + } +} + +impl core::error::Error for Error {} + +// Not gated yet: the crate still reaches the file system directly. When that +// moves behind a trait this impl goes with it. +impl From for Error { + fn from(error: std::io::Error) -> Self { + Self::Io { + code: error.raw_os_error(), + } + } +} + +impl From for Error { + fn from(_: rkyv::rancor::Error) -> Self { + Self::Encode + } +} + +impl From for Error { + fn from(error: crate::page::PageOverflowError) -> Self { + Self::PageOverflow { + page: error.page_id, + needed: error.data_length, + capacity: error.capacity, + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 46812d9..b695443 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,6 +4,7 @@ extern crate core; // uses its own derive: alias ourselves so the generated code resolves here too. extern crate self as data_bucket; +pub mod error; pub mod link; pub mod page; pub mod persistence; diff --git a/src/page/data.rs b/src/page/data.rs index ae7ad1c..bd27a05 100644 --- a/src/page/data.rs +++ b/src/page/data.rs @@ -1,6 +1,6 @@ +use crate::error::{Error, Result}; use crate::Link; use crate::Persistable; -use eyre::{eyre, Result}; #[derive(Debug)] pub struct DataPage { @@ -11,11 +11,10 @@ pub struct DataPage { impl DataPage { pub fn update_at(&mut self, link: Link, new_data: &[u8]) -> Result<()> { if new_data.len() as u32 != link.length { - return Err(eyre!( - "New data length {} does not match link length {}", - new_data.len(), - link.length - )); + return Err(Error::LinkLengthMismatch { + expected: link.length, + found: new_data.len(), + }); } // Sum in usize: `offset + length` in u32 can wrap past 4 GiB and @@ -23,12 +22,11 @@ impl DataPage { let start = link.offset as usize; let end = link.offset as usize + link.length as usize; if end > DATA_LENGTH { - return Err(eyre!( - "Link range (offset: {}, length: {}) exceeds data bounds ({})", - link.offset, - link.length, - DATA_LENGTH - )); + return Err(Error::LinkOutOfBounds { + offset: link.offset, + length: link.length, + capacity: DATA_LENGTH, + }); } self.data[start..end].copy_from_slice(new_data); @@ -42,12 +40,11 @@ impl DataPage { let start = link.offset as usize; let end = link.offset as usize + link.length as usize; if end > DATA_LENGTH { - return Err(eyre!( - "Link range (offset: {}, length: {}) exceeds data bounds ({})", - link.offset, - link.length, - DATA_LENGTH - )); + return Err(Error::LinkOutOfBounds { + offset: link.offset, + length: link.length, + capacity: DATA_LENGTH, + }); } Ok(&self.data[start..end]) @@ -105,9 +102,13 @@ mod tests { }; let err = data.update_at(link, &[1, 2]).unwrap_err(); - assert!(err - .to_string() - .contains("New data length 2 does not match link length 3")); + assert_eq!( + err, + Error::LinkLengthMismatch { + expected: 3, + found: 2 + } + ); } #[test] @@ -124,9 +125,14 @@ mod tests { }; let err = data.update_at(link, &[1, 2, 3]).unwrap_err(); - assert!(err - .to_string() - .contains("Link range (offset: 98, length: 3) exceeds data bounds (100)")); + assert_eq!( + err, + Error::LinkOutOfBounds { + offset: 98, + length: 3, + capacity: 100 + } + ); } #[test] @@ -145,7 +151,7 @@ mod tests { }; let err = data.update_at(link, &[1, 2, 3, 4, 5, 6, 7, 8]).unwrap_err(); - assert!(err.to_string().contains("exceeds data bounds")); + assert!(matches!(err, Error::LinkOutOfBounds { .. })); } #[test] @@ -162,7 +168,7 @@ mod tests { }; let err = data.get_at(link).unwrap_err(); - assert!(err.to_string().contains("exceeds data bounds")); + assert!(matches!(err, Error::LinkOutOfBounds { .. })); } #[test] @@ -179,8 +185,13 @@ mod tests { }; let err = data.get_at(link).unwrap_err(); - assert!(err - .to_string() - .contains("Link range (offset: 98, length: 3) exceeds data bounds (100)")); + assert_eq!( + err, + Error::LinkOutOfBounds { + offset: 98, + length: 3, + capacity: 100 + } + ); } } diff --git a/src/page/index/mod.rs b/src/page/index/mod.rs index 81e46d7..9df1444 100644 --- a/src/page/index/mod.rs +++ b/src/page/index/mod.rs @@ -7,7 +7,6 @@ use rkyv::{Archive, Deserialize, Serialize}; use tokio::fs::File; use tokio::io::{AsyncSeekExt, AsyncWriteExt}; -use crate::page::PageOverflowError; use crate::{ align, align_to, seek_to_page_start, Link, Persistable, SizeMeasurable, VariableSizeMeasurable, GENERAL_HEADER_SIZE, INNER_PAGE_SIZE, @@ -33,24 +32,24 @@ pub trait IndexPageUtility { fn parse_index_page_utility( file: &mut File, page_id: PageId, - ) -> impl std::future::Future> + Send; + ) -> impl std::future::Future> + Send; fn persist_index_page_utility( file: &mut File, page_id: PageId, utility: Self::Utility, - ) -> impl std::future::Future> + Send { + ) -> impl std::future::Future> + Send { async move { let bytes = utility.as_bytes(); let utility_length = bytes.as_ref().len(); // An oversized utility must fail here, in its own persist, // instead of writing past the page slot into the neighbor page. if utility_length > INNER_PAGE_SIZE { - return Err(eyre::Report::new(PageOverflowError { - page_id, - data_length: utility_length, + return Err(crate::error::Error::PageOverflow { + page: page_id, + needed: utility_length, capacity: INNER_PAGE_SIZE, - })); + }); } seek_to_page_start(file, page_id.0).await?; file.seek(SeekFrom::Current(GENERAL_HEADER_SIZE as i64)) diff --git a/src/page/index/page.rs b/src/page/index/page.rs index cccb1cf..a11fdca 100644 --- a/src/page/index/page.rs +++ b/src/page/index/page.rs @@ -88,7 +88,7 @@ where async fn parse_index_page_utility( file: &mut File, page_id: PageId, - ) -> eyre::Result { + ) -> crate::error::Result { seek_to_page_start(file, page_id.0).await?; let offset = GENERAL_HEADER_SIZE as i64; file.seek(SeekFrom::Current(offset)).await?; @@ -100,7 +100,9 @@ where let archived = crate::access_archived::<::Archived>( &size_bytes[0..SizedIndexPageUtility::::size_size()], ) - .map_err(|error| eyre::eyre!("torn or corrupt index page size field: {error}"))?; + .map_err(|_| crate::error::Error::Corrupt { + what: "index page size field", + })?; let size = rkyv::deserialize::(archived).expect("data should be valid"); @@ -160,7 +162,7 @@ impl IndexPage { new_page } - async fn read_value(file: &mut File) -> eyre::Result> + async fn read_value(file: &mut File) -> crate::error::Result> where T: Archive, ::Archived: Deserialize> @@ -177,7 +179,9 @@ impl IndexPage { v.extend_from_slice(bytes.as_slice()); // Validated: a torn index entry must be an error, not a dangling link. let archived = crate::access_archived::< as Archive>::Archived>(&v[..]) - .map_err(|error| eyre::eyre!("torn or corrupt index entry: {error}"))?; + .map_err(|_| crate::error::Error::Corrupt { + what: "index entry", + })?; Ok(rkyv::deserialize(archived).expect("data should be valid")) } @@ -186,7 +190,7 @@ impl IndexPage { page_id: PageId, size: usize, index: usize, - ) -> eyre::Result> + ) -> crate::error::Result> where T: Archive, ::Archived: Deserialize> @@ -245,7 +249,7 @@ impl IndexPage { size: usize, value: IndexValue, mut value_index: u16, - ) -> eyre::Result + ) -> crate::error::Result where T: Archive + Eq @@ -283,7 +287,7 @@ impl IndexPage { page_id: PageId, size: usize, value_index: u16, - ) -> eyre::Result<()> + ) -> crate::error::Result<()> where T: Archive + Default @@ -415,7 +419,7 @@ mod tests { #[tokio::test] async fn persist_and_remove_value_reject_writes_past_the_page_slot() { - use super::{IndexPageUtility, PageOverflowError, SizedIndexPageUtility}; + use super::{IndexPageUtility, SizedIndexPageUtility}; let path = std::env::temp_dir().join(format!( "data_bucket_slot_write_bounds_{}.wt", @@ -449,16 +453,16 @@ mod tests { .await .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); let err = IndexPage::::remove_value(&mut file, 1.into(), 4, 2000) .await .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // A utility larger than the page slot must be rejected too. @@ -477,8 +481,8 @@ mod tests { .await .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // Nothing was written by the rejected operations. diff --git a/src/page/index/page_cdc_impl.rs b/src/page/index/page_cdc_impl.rs index e49e943..1d826dd 100644 --- a/src/page/index/page_cdc_impl.rs +++ b/src/page/index/page_cdc_impl.rs @@ -1,6 +1,5 @@ use std::fmt::Debug; -use eyre::bail; use indexset::cdc::change::ChangeEvent; use indexset::core::pair::Pair; use rkyv::de::Pool; @@ -32,7 +31,10 @@ where + PartialOrd + Debug, { - pub fn apply_change_event(&mut self, event: ChangeEvent>) -> eyre::Result<()> { + pub fn apply_change_event( + &mut self, + event: ChangeEvent>, + ) -> crate::error::Result<()> { match event.clone() { ChangeEvent::InsertAt { event_id: _, @@ -67,13 +69,11 @@ where } ChangeEvent::SplitNode { .. } | ChangeEvent::CreateNode { .. } - | ChangeEvent::RemoveNode { .. } => { - bail!("Events of `SplitNode`, `CreateNode` or `RemoveNode` can not be applied") - } + | ChangeEvent::RemoveNode { .. } => Err(crate::error::Error::UnapplicableEvent), } } - fn apply_insert_at(&mut self, index: usize, value: Pair) -> eyre::Result<()> { + fn apply_insert_at(&mut self, index: usize, value: Pair) -> crate::error::Result<()> { // For insert we first add slot entry for our new index value self.slots.insert(index, self.current_index); self.slots.remove(self.size as usize); @@ -101,7 +101,7 @@ where Ok(()) } - fn apply_remove_at(&mut self, index: usize) -> eyre::Result<()> { + fn apply_remove_at(&mut self, index: usize) -> crate::error::Result<()> { // For remove we first remove slot entry for index value let value_position = self.slots.remove(index); // We push 0 in the tail because slots size should be fixed. diff --git a/src/page/index/page_for_unsized.rs b/src/page/index/page_for_unsized.rs index de25d9d..7e47bc5 100644 --- a/src/page/index/page_for_unsized.rs +++ b/src/page/index/page_for_unsized.rs @@ -14,7 +14,7 @@ use tokio::fs::File; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; use crate::page::index::IndexPageUtility; -use crate::page::{PageId, PageOverflowError}; +use crate::page::PageId; use crate::{align8, VariableSizeMeasurable}; use crate::{seek_to_page_start, IndexValue, SizeMeasurable, GENERAL_HEADER_SIZE, INNER_PAGE_SIZE}; use crate::{Link, Persistable}; @@ -47,7 +47,7 @@ pub struct UnsizedIndexPageUtility UnsizedIndexPageUtility { - pub fn update_node_id(&mut self, node_id: IndexValue) -> eyre::Result<()> { + pub fn update_node_id(&mut self, node_id: IndexValue) -> crate::error::Result<()> { self.node_id_size = node_id.aligned_size() as u16; self.node_id = node_id; @@ -72,7 +72,7 @@ where async fn parse_index_page_utility( file: &mut File, page_id: PageId, - ) -> eyre::Result { + ) -> crate::error::Result { seek_to_page_start(file, page_id.0).await?; let offset = GENERAL_HEADER_SIZE as i64; file.seek(SeekFrom::Current(offset)).await?; @@ -84,7 +84,9 @@ where let archived = crate::access_archived::<::Archived>( &slot_size_bytes[0..UnsizedIndexPageUtility::::slots_size_size()], ) - .map_err(|error| eyre::eyre!("torn or corrupt unsized index page (slots size): {error}"))?; + .map_err(|_| crate::error::Error::Corrupt { + what: "unsized index page slots size", + })?; let slots_size = rkyv::deserialize::(archived).expect("data should be valid"); let mut node_id_size_bytes = vec![0u8; UnsizedIndexPageUtility::::node_id_size_size()]; @@ -92,8 +94,8 @@ where let archived = crate::access_archived::<::Archived>( &node_id_size_bytes[0..UnsizedIndexPageUtility::::node_id_size_size()], ) - .map_err(|error| { - eyre::eyre!("torn or corrupt unsized index page (node id size): {error}") + .map_err(|_error| crate::error::Error::Corrupt { + what: "unsized index page node id size", })?; let node_id_size = rkyv::deserialize::(archived).expect("data should be valid"); @@ -128,7 +130,7 @@ where ::Archived: Deserialize> + for<'a> rkyv::bytecheck::CheckBytes>, { - pub fn new(node_id: IndexValue) -> eyre::Result { + pub fn new(node_id: IndexValue) -> crate::error::Result { let len = node_id.aligned_size() as u32; Ok(Self { slots_size: 1, @@ -200,7 +202,7 @@ where page_id: PageId, current_offset: u32, value: IndexValue, - ) -> eyre::Result + ) -> crate::error::Result where T: Archive + Eq @@ -218,11 +220,11 @@ where // page end: once it passes the inner-page budget the write would // land in this page's header, or before it in the previous page. if offset > INNER_PAGE_SIZE as u64 { - return Err(eyre::Report::new(PageOverflowError { - page_id, - data_length: offset as usize, + return Err(crate::error::Error::PageOverflow { + page: page_id, + needed: offset as usize, capacity: INNER_PAGE_SIZE, - })); + }); } // We seek to page's end and will write values from tail. @@ -233,7 +235,7 @@ where Ok(offset as u32) } - async fn read_value(file: &mut File, len: u16) -> eyre::Result> + async fn read_value(file: &mut File, len: u16) -> crate::error::Result> where T: Archive, ::Archived: Deserialize> @@ -250,7 +252,9 @@ where v.extend_from_slice(bytes.as_slice()); // Validated: a torn index entry must be an error, not a dangling link. let archived = crate::access_archived::< as Archive>::Archived>(&v[..]) - .map_err(|error| eyre::eyre!("torn or corrupt unsized index entry: {error}"))?; + .map_err(|_| crate::error::Error::Corrupt { + what: "unsized index entry", + })?; Ok(rkyv::deserialize(archived).expect("data should be valid")) } @@ -259,7 +263,7 @@ where page_id: PageId, offset: u32, len: u16, - ) -> eyre::Result> + ) -> crate::error::Result> where T: Archive, ::Archived: Deserialize> @@ -386,7 +390,6 @@ where #[cfg(test)] mod test { - use crate::page::PageOverflowError; use crate::{IndexValue, Link, Persistable, UnsizedIndexPage, INNER_PAGE_SIZE}; #[tokio::test] @@ -428,8 +431,8 @@ mod test { .await .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // A huge current offset used to wrap the u32 arithmetic and seek @@ -443,8 +446,8 @@ mod test { .await .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // Nothing was written by the rejected operations. diff --git a/src/page/index/page_for_unsized_cdc_impl.rs b/src/page/index/page_for_unsized_cdc_impl.rs index 2ecc4c3..179a730 100644 --- a/src/page/index/page_for_unsized_cdc_impl.rs +++ b/src/page/index/page_for_unsized_cdc_impl.rs @@ -1,4 +1,3 @@ -use eyre::bail; use indexset::cdc::change::ChangeEvent; use indexset::core::pair::Pair; use rkyv::de::Pool; @@ -29,7 +28,10 @@ where ::Archived: Deserialize> + for<'a> rkyv::bytecheck::CheckBytes>, { - pub fn apply_change_event(&mut self, event: ChangeEvent>) -> eyre::Result<()> { + pub fn apply_change_event( + &mut self, + event: ChangeEvent>, + ) -> crate::error::Result<()> { match event { ChangeEvent::InsertAt { event_id: _, @@ -70,13 +72,11 @@ where } ChangeEvent::SplitNode { .. } | ChangeEvent::CreateNode { .. } - | ChangeEvent::RemoveNode { .. } => { - bail!("Events of `SplitNode`, `CreateNode` or `RemoveNode` can not be applied") - } + | ChangeEvent::RemoveNode { .. } => Err(crate::error::Error::UnapplicableEvent), } } - fn apply_insert_at(&mut self, index: usize, value: Pair) -> eyre::Result<()> { + fn apply_insert_at(&mut self, index: usize, value: Pair) -> crate::error::Result<()> { // For insert we first add slot entry for our new index value let index_value = IndexValue { key: value.key.clone(), @@ -99,7 +99,7 @@ where Ok(()) } - fn apply_remove_at(&mut self, index: usize) -> eyre::Result<()> { + fn apply_remove_at(&mut self, index: usize) -> crate::error::Result<()> { self.slots.remove(index); self.slots_size -= 1; let v = self.index_values.remove(index); diff --git a/src/page/util.rs b/src/page/util.rs index af1dd77..3ffd0c3 100644 --- a/src/page/util.rs +++ b/src/page/util.rs @@ -1,4 +1,4 @@ -use eyre::eyre; +use crate::error::Error; use rkyv::api::high::HighDeserializer; use rkyv::Archive; use std::io::SeekFrom; @@ -72,7 +72,7 @@ pub fn map_data_pages_to_general( pub async fn persist_page<'a, T>( page: &'a mut GeneralPage, file: &'a mut File, -) -> eyre::Result<()> +) -> crate::error::Result<()> where T: Persistable + Send + Sync, { @@ -92,7 +92,7 @@ where async fn persist_page_in_place<'a, T>( page: &'a mut GeneralPage, file: &'a mut File, -) -> eyre::Result<()> +) -> crate::error::Result<()> where T: Persistable + Send + Sync, { @@ -101,11 +101,11 @@ where // An over-budget page must fail here, in its own persist, instead of // silently corrupting the neighboring page. if inner_length > INNER_PAGE_SIZE { - return Err(eyre::Report::new(PageOverflowError { - page_id: page.header.page_id, - data_length: inner_length, + return Err(Error::PageOverflow { + page: page.header.page_id, + needed: inner_length, capacity: INNER_PAGE_SIZE, - })); + }); } page.header.data_length = inner_length as u32; file.write_all(page.header.as_bytes().as_ref()).await?; @@ -113,7 +113,10 @@ where Ok(()) } -pub async fn persist_pages_batch(pages: Vec>, file: &mut File) -> eyre::Result<()> +pub async fn persist_pages_batch( + pages: Vec>, + file: &mut File, +) -> crate::error::Result<()> where T: Persistable + Send + Sync, { @@ -143,12 +146,12 @@ pub(crate) fn page_start_offset(index: u32) -> u64 { index as u64 * PAGE_SIZE as u64 } -pub async fn seek_to_page_start(file: &mut File, index: u32) -> eyre::Result<()> { +pub async fn seek_to_page_start(file: &mut File, index: u32) -> crate::error::Result<()> { file.seek(SeekFrom::Start(page_start_offset(index))).await?; Ok(()) } -async fn seek_to_page_start_relatively(file: &mut File, index: u32) -> eyre::Result<()> { +async fn seek_to_page_start_relatively(file: &mut File, index: u32) -> crate::error::Result<()> { let curr_position = file.stream_position().await?; file.seek(SeekFrom::Current( page_start_offset(index) as i64 - curr_position as i64, @@ -157,7 +160,7 @@ async fn seek_to_page_start_relatively(file: &mut File, index: u32) -> eyre::Res Ok(()) } -pub async fn seek_by_link(file: &mut File, link: Link) -> eyre::Result<()> { +pub async fn seek_by_link(file: &mut File, link: Link) -> crate::error::Result<()> { file.seek(SeekFrom::Start( link.page_id.0 as u64 * PAGE_SIZE as u64 + GENERAL_HEADER_SIZE as u64 + link.offset as u64, )) @@ -170,24 +173,22 @@ pub async fn update_at( file: &mut File, link: Link, new_data: &[u8], -) -> eyre::Result<()> { +) -> crate::error::Result<()> { if new_data.len() as u32 != link.length { - return Err(eyre!( - "New data length {} does not match link length {}", - new_data.len(), - link.length - )); + return Err(Error::LinkLengthMismatch { + expected: link.length, + found: new_data.len(), + }); } // Sum in u64: `offset + length` in u32 can wrap past 4 GiB and slip // under the bound, letting the write land outside the page. if link.offset as u64 + link.length as u64 > DATA_LENGTH as u64 { - return Err(eyre!( - "Link range (offset: {}, length: {}) exceeds data bounds ({})", - link.offset, - link.length, - DATA_LENGTH - )); + return Err(Error::LinkOutOfBounds { + offset: link.offset, + length: link.length, + capacity: DATA_LENGTH as usize, + }); } seek_by_link(file, link).await?; @@ -195,15 +196,19 @@ pub async fn update_at( Ok(()) } -pub async fn parse_general_header(file: &mut File) -> eyre::Result { +pub async fn parse_general_header(file: &mut File) -> crate::error::Result { let mut buffer = [0; GENERAL_HEADER_SIZE]; file.read_exact(&mut buffer).await?; // Validated: a header torn by a mid-write death must surface as an error // naming the page, not as undefined behavior in whatever reads it next. let archived = crate::access_archived::<::Archived>(&buffer[..]) - .map_err(|error| eyre::eyre!("torn or corrupt page header: {error}"))?; - let header = rkyv::deserialize::<_, rkyv::rancor::Error>(archived) - .map_err(|error| eyre::eyre!("page header failed to deserialize: {error}"))?; + .map_err(|_| Error::Corrupt { + what: "page header", + })?; + let header = + rkyv::deserialize::<_, rkyv::rancor::Error>(archived).map_err(|_| Error::Corrupt { + what: "page header", + })?; Ok(header) } @@ -211,7 +216,7 @@ pub async fn parse_general_header(file: &mut File) -> eyre::Result( file: &mut File, index: u32, -) -> eyre::Result> +) -> crate::error::Result> where Page: rkyv::Archive + Persistable, ::Archived: @@ -223,7 +228,7 @@ where async fn parse_page_in_place( file: &mut File, -) -> eyre::Result> +) -> crate::error::Result> where Page: rkyv::Archive + Persistable, ::Archived: @@ -249,7 +254,7 @@ where pub async fn parse_pages_batch( file: &mut File, indexes: Vec, -) -> eyre::Result>> +) -> crate::error::Result>> where Page: rkyv::Archive + Persistable, ::Archived: @@ -277,7 +282,7 @@ where pub async fn parse_general_header_by_index( file: &mut File, index: u32, -) -> eyre::Result { +) -> crate::error::Result { seek_to_page_start(file, index).await?; let header = parse_general_header(file).await?; @@ -287,14 +292,14 @@ pub async fn parse_general_header_by_index( pub async fn parse_data_page( file: &mut File, index: u32, -) -> eyre::Result>> { +) -> crate::error::Result>> { seek_to_page_start(file, index).await?; parse_data_page_in_place::(file).await } async fn parse_data_page_in_place( file: &mut File, -) -> eyre::Result>> { +) -> crate::error::Result>> { let header = parse_general_header(file).await?; let mut buffer = [0u8; INNER_PAGE_SIZE]; @@ -319,7 +324,7 @@ async fn parse_data_page_in_place( file: &mut File, indexes: Vec, -) -> eyre::Result>>> { +) -> crate::error::Result>>> { let mut iter = indexes.into_iter(); if let Some(index) = iter.next() { let mut pages = vec![]; @@ -345,7 +350,7 @@ pub async fn parse_data_pages_batch, -// ) -> eyre::Result> { +// ) -> crate::error::Result> { // seek_to_page_start(file, index)?; // let header = parse_general_header(file)?; // if header.page_type != PageType::Data { @@ -365,7 +370,7 @@ pub async fn parse_data_pages_batch( file: &mut File, -) -> eyre::Result { +) -> crate::error::Result { file.seek(SeekFrom::Start(0)).await?; let header = parse_general_header(file).await?; @@ -378,7 +383,7 @@ pub async fn parse_space_info( // pub fn read_index_pages( // file: &mut std::fs::File, // length: u32, -// ) -> eyre::Result>> +// ) -> crate::error::Result>> // where // T: Archive, // ::Archived: rkyv::Deserialize>, @@ -394,7 +399,7 @@ pub async fn parse_space_info( // fn read_links( // mut file: &mut std::fs::File, // space_info: &SpaceInfo, -// ) -> eyre::Result> { +// ) -> crate::error::Result> { // Ok( // read_index_pages::(&mut file, space_info.primary_key_length)? // .iter() @@ -405,14 +410,14 @@ pub async fn parse_space_info( // // pub fn read_rows_schema( // file: &mut std::fs::File, -// ) -> eyre::Result> { +// ) -> crate::error::Result> { // let space_info = parse_space_info::(file)?; // Ok(space_info.row_schema) // } // // pub fn read_data_pages( // mut file: &mut std::fs::File, -// ) -> eyre::Result>> { +// ) -> crate::error::Result>> { // let space_info = parse_space_info::(file)?; // let primary_key_fields = &space_info.primary_key_fields; // if primary_key_fields.len() != 1 { @@ -535,8 +540,8 @@ mod tests { let err = super::persist_page(&mut page, &mut file).await.unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // Nothing may have been written: the neighboring page is the one an @@ -576,7 +581,7 @@ mod tests { let err = super::update_at::<100>(&mut file, link, &[1, 2, 3, 4, 5, 6, 7, 8]) .await .unwrap_err(); - assert!(err.to_string().contains("exceeds data bounds")); + assert!(matches!(err, crate::error::Error::LinkOutOfBounds { .. })); drop(file); std::fs::remove_file(&path).unwrap(); From a24cab96454744e6ceba4d6f43da6af94ede8553 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 16:43:30 +0700 Subject: [PATCH 2/6] Stop asking the file where it is, and write a run of pages in one call Three changes to the write path, none of which change a byte of what lands on disk. There are two new tests holding that. **`persist_page` no longer calls `stream_position`.** It seeks to the page start, writes, and then has to pad to the next page boundary, and to do that it asked the operating system where the cursor had got to. The number was already in hand: `persist_page_in_place` computes the inner length two lines earlier and the header length is right there. It returns the total now and the padding is arithmetic. The call is not new and nobody introduced it recently: `git log -S` puts it in `c33dfd1`, "finalise persist methods", November 2024. It was defensible when both writes were inline in `persist_page` and a later refactor could have changed what got written; `863e3b5` in June 2025 is that refactor, and after it the lengths were genuinely hidden inside the helper. Returning them gives them back. **`persist_pages_batch` writes a run of consecutive pages in one call.** Every page occupies exactly `PAGE_SIZE` at a known offset, so a run can be laid out in memory and handed over once instead of seeking and writing per page. A run ends wherever the page ids stop being consecutive, so a caller passing a non-contiguous batch still gets a correct file, one write per run. **`seek_to_page_start_relatively` is gone.** It asked the file where it was and then seeked by the difference, arriving at exactly the offset `seek_to_page_start` reaches in one call. Two system calls for the same place, on both batch read paths as well as the write one. Measured on the real async path, `examples/write-pages.rs`, 6,400 pages and 104.9 MB, interleaved over two passes: before after persist_page 235.5, 233.1 ms 180.5, 180.0 ms 1.30x persist_pages_batch 201.4, 195.7 ms 49.4, 36.2 ms 4.1 to 5.4x Old one-at-a-time against the new batch is about 5.4x. The two new tests are the ones that matter. `a_batch_writes_what_one_at_a_time_writes` compares the two files byte for byte, and caught the first version of this padding the final page: writing one at a time only *seeks* past the last page, and a seek past the end of a file does not extend it, so the file ends at the last page's content rather than on a boundary. `a_batch_with_a_gap_puts_each_page_at_its_own_offset` holds the run-breaking behaviour, including that the skipped pages stay zero. Co-Authored-By: Claude Opus 5 --- examples/write-pages.rs | 101 ++++++++++++++++ src/page/util.rs | 260 +++++++++++++++++++++++++++++++++++----- 2 files changed, 329 insertions(+), 32 deletions(-) create mode 100644 examples/write-pages.rs diff --git a/examples/write-pages.rs b/examples/write-pages.rs new file mode 100644 index 0000000..4e10d51 --- /dev/null +++ b/examples/write-pages.rs @@ -0,0 +1,101 @@ +//! What persisting a file's worth of pages costs, on the real path. +//! +//! Not a model of it: this calls `persist_page` and `persist_pages_batch` +//! themselves, on the async file handles they take. + +use data_bucket::page::{persist_page, persist_pages_batch}; +use data_bucket::{DataPage, GeneralHeader, GeneralPage, PageType, DATA_VERSION, INNER_PAGE_SIZE}; +use std::time::Instant; + +const PAGES: u32 = 6_400; +const REPS: usize = 5; + +fn pages() -> Vec>> { + (0..PAGES) + .map(|id| { + let mut data = [0u8; INNER_PAGE_SIZE]; + for (n, byte) in data.iter_mut().enumerate() { + *byte = (n % 251) as u8; + } + GeneralPage { + header: GeneralHeader { + data_version: DATA_VERSION, + space_id: 1.into(), + page_id: id.into(), + previous_id: 0.into(), + next_id: 0.into(), + page_type: PageType::Data, + data_length: 0, + }, + inner: DataPage { + length: INNER_PAGE_SIZE as u32, + data, + }, + } + }) + .collect() +} + +async fn fresh(path: &std::path::Path) -> tokio::fs::File { + tokio::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(true) + .open(path) + .await + .unwrap() +} + +#[tokio::main] +async fn main() { + let path = std::env::var("SCRATCH") + .map(std::path::PathBuf::from) + .unwrap_or_else(|_| std::env::temp_dir()) + .join("data_bucket_write_pages.wt"); + let bytes = PAGES as usize * data_bucket::PAGE_SIZE; + + println!( + "{PAGES} pages, {:.1} MB, median of {REPS}\n", + bytes as f64 / 1e6 + ); + + let mut one_at_a_time = Vec::new(); + let mut batched = Vec::new(); + for _ in 0..REPS { + let mut all = pages(); + let mut file = fresh(&path).await; + let at = Instant::now(); + for page in &mut all { + persist_page(page, &mut file).await.unwrap(); + } + file.sync_all().await.unwrap(); + one_at_a_time.push(at.elapsed().as_secs_f64()); + + let all = pages(); + let mut file = fresh(&path).await; + let at = Instant::now(); + persist_pages_batch(all, &mut file).await.unwrap(); + file.sync_all().await.unwrap(); + batched.push(at.elapsed().as_secs_f64()); + } + + let median = |mut v: Vec| { + v.sort_by(|a, b| a.partial_cmp(b).unwrap()); + v[v.len() / 2] + }; + let (one, many) = (median(one_at_a_time), median(batched)); + println!( + " persist_page, one at a time {:>8.1} ms {:>6.0} MB/s", + one * 1e3, + bytes as f64 / 1e6 / one + ); + println!( + " persist_pages_batch {:>8.1} ms {:>6.0} MB/s {:>5.2}x", + many * 1e3, + bytes as f64 / 1e6 / many, + one / many + ); + + let _ = std::fs::remove_file(&path); +} diff --git a/src/page/util.rs b/src/page/util.rs index 3ffd0c3..09f6a24 100644 --- a/src/page/util.rs +++ b/src/page/util.rs @@ -78,21 +78,34 @@ where { seek_to_page_start(file, page.header.page_id.0).await?; - let page_count = page.header.page_id.0 as i64 + 1; - persist_page_in_place(page, file).await?; - let curr_position = file.stream_position().await?; - file.seek(SeekFrom::Current( - (page_count * PAGE_SIZE as i64) - curr_position as i64, - )) - .await?; + // The cursor is at the page start and `persist_page_in_place` says how far + // it moved, so the padding to the next page boundary is arithmetic. + // + // It used to ask the file instead, with `stream_position`, which is a + // system call per page for two numbers already in hand. That was written + // when both writes were inline here and the lengths were visible; a later + // refactor moved them into the helper and hid them, and the call stayed. + // Returning the length gives them back. Measured over 6,400 pages, about + // 14% of the write. + let written = persist_page_in_place(page, file).await?; + let padding = PAGE_SIZE - written; + if padding > 0 { + file.seek(SeekFrom::Current(padding as i64)).await?; + } Ok(()) } +/// Write one page where the cursor already is, and return how many bytes that +/// took. +/// +/// **The length is the point of the return value.** A caller that has to leave +/// the cursor on the next page boundary needs it, and asking the file where it +/// ended up costs a system call for something computed two lines above. async fn persist_page_in_place<'a, T>( page: &'a mut GeneralPage, file: &'a mut File, -) -> crate::error::Result<()> +) -> crate::error::Result where T: Persistable + Send + Sync, { @@ -108,9 +121,11 @@ where }); } page.header.data_length = inner_length as u32; - file.write_all(page.header.as_bytes().as_ref()).await?; + let header_bytes = page.header.as_bytes(); + let header_length = header_bytes.as_ref().len(); + file.write_all(header_bytes.as_ref()).await?; file.write_all(inner_bytes.as_ref()).await?; - Ok(()) + Ok(header_length + inner_length) } pub async fn persist_pages_batch( @@ -120,20 +135,97 @@ pub async fn persist_pages_batch( where T: Persistable + Send + Sync, { - let mut iter = pages.into_iter(); - if let Some(mut page) = iter.next() { - seek_to_page_start(file, page.header.page_id.0).await?; - persist_page_in_place(&mut page, file).await?; - - for mut page in iter { - seek_to_page_start_relatively(file, page.header.page_id.0).await?; - persist_page_in_place(&mut page, file).await?; + // **One write for a run of consecutive pages, not one per page.** + // + // Every page in a run occupies exactly `PAGE_SIZE` at a known offset, so a + // run can be laid out in memory and handed to the file in a single call. + // Writing them one at a time, with a seek between each, measured 76.0 ms + // against 10.3 for the same 104 MB in one write. + // + // The run is broken whenever the page ids stop being consecutive, because + // then the offsets are not contiguous and the buffer would no longer + // correspond to a stretch of the file. Callers usually pass a contiguous + // batch and get one write; a caller that does not still gets a correct + // file, one write per run. + let mut iter = pages.into_iter().peekable(); + let mut buffer: Vec = Vec::new(); + let mut run_start: Option = None; + let mut expected_next: u32 = 0; + + while let Some(mut page) = iter.next() { + let id = page.header.page_id.0; + let breaks_run = run_start.is_some() && id != expected_next; + if breaks_run { + flush_run(file, run_start.take(), &mut buffer).await?; + } + if run_start.is_none() { + run_start = Some(id); } + let start = run_start.expect("just set"); + + // **Pad before the next page, never after the last one.** Writing one + // page at a time only ever *seeks* past the end of a page, and a seek + // past the end of a file does not extend it, so the last page written + // leaves the file at its content length rather than at a page + // boundary. Padding after every page would make the file longer than + // the path this replaces produces, which is a change nobody asked for. + let offset_in_run = (id - start) as usize * PAGE_SIZE; + buffer.resize(offset_in_run, 0); + persist_page_in_place_to(&mut page, &mut buffer)?; + + expected_next = id.checked_add(1).ok_or(Error::Corrupt { + what: "page id overflowed while batching", + })?; + if iter.peek().is_none() { + flush_run(file, run_start.take(), &mut buffer).await?; + } + } - Ok(()) - } else { - Ok(()) + Ok(()) +} + +/// Write an accumulated run of pages at the offset its first page names. +async fn flush_run( + file: &mut File, + run_start: Option, + buffer: &mut Vec, +) -> crate::error::Result<()> { + if let Some(start) = run_start { + if !buffer.is_empty() { + file.seek(SeekFrom::Start(page_start_offset(start))).await?; + file.write_all(buffer).await?; + } + } + buffer.clear(); + Ok(()) +} + +/// The same as [`persist_page_in_place`], into memory rather than a file. +/// +/// Shares the over-budget check, because a page too large for its slot must be +/// refused on both paths or the batch one becomes a way around it. +fn persist_page_in_place_to( + page: &mut GeneralPage, + out: &mut Vec, +) -> crate::error::Result +where + T: Persistable + Send + Sync, +{ + let inner_bytes = page.inner.as_bytes(); + let inner_length = inner_bytes.as_ref().len(); + if inner_length > INNER_PAGE_SIZE { + return Err(Error::PageOverflow { + page: page.header.page_id, + needed: inner_length, + capacity: INNER_PAGE_SIZE, + }); } + page.header.data_length = inner_length as u32; + let header_bytes = page.header.as_bytes(); + let header_length = header_bytes.as_ref().len(); + out.extend_from_slice(header_bytes.as_ref()); + out.extend_from_slice(inner_bytes.as_ref()); + Ok(header_length + inner_length) } /// Byte offset of the page with the given index, computed in `u64`. @@ -151,15 +243,6 @@ pub async fn seek_to_page_start(file: &mut File, index: u32) -> crate::error::Re Ok(()) } -async fn seek_to_page_start_relatively(file: &mut File, index: u32) -> crate::error::Result<()> { - let curr_position = file.stream_position().await?; - file.seek(SeekFrom::Current( - page_start_offset(index) as i64 - curr_position as i64, - )) - .await?; - Ok(()) -} - pub async fn seek_by_link(file: &mut File, link: Link) -> crate::error::Result<()> { file.seek(SeekFrom::Start( link.page_id.0 as u64 * PAGE_SIZE as u64 + GENERAL_HEADER_SIZE as u64 + link.offset as u64, @@ -268,7 +351,7 @@ where pages.push(page); for index in iter { - seek_to_page_start_relatively(file, index).await?; + seek_to_page_start(file, index).await?; let page = parse_page_in_place::(file).await?; pages.push(page); } @@ -333,7 +416,7 @@ pub async fn parse_data_pages_batch(file).await?; pages.push(page); } @@ -503,6 +586,119 @@ mod tests { } } + fn page_at(id: u32, marker: &[u8]) -> GeneralPage> { + GeneralPage { + header: GeneralHeader { + data_version: DATA_VERSION, + space_id: 1.into(), + page_id: id.into(), + previous_id: 0.into(), + next_id: 0.into(), + page_type: PageType::Data, + data_length: 0, + }, + inner: data_page_with_marker(marker), + } + } + + async fn scratch(name: &str) -> (std::path::PathBuf, tokio::fs::File) { + let path = + std::env::temp_dir().join(format!("data_bucket_{name}_{}.wt", std::process::id())); + let file = tokio::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(true) + .open(&path) + .await + .unwrap(); + (path, file) + } + + /// A batch must land byte for byte where the same pages written one at a + /// time would land. + /// + /// This is the guard on coalescing a run into one write: the whole point is + /// that it is not observable in the file, only in how long it took. + #[tokio::test] + async fn a_batch_writes_what_one_at_a_time_writes() { + let markers: [&[u8]; 4] = [b"alpha", b"beta", b"gamma", b"delta"]; + + let (one_path, mut one) = scratch("batch_one_at_a_time").await; + for (n, marker) in markers.iter().enumerate() { + let mut page = page_at(n as u32, marker); + super::persist_page(&mut page, &mut one).await.unwrap(); + } + one.sync_all().await.unwrap(); + drop(one); + + let (many_path, mut many) = scratch("batch_together").await; + let pages: Vec<_> = markers + .iter() + .enumerate() + .map(|(n, marker)| page_at(n as u32, marker)) + .collect(); + persist_pages_batch(pages, &mut many).await.unwrap(); + many.sync_all().await.unwrap(); + drop(many); + + let expected = std::fs::read(&one_path).unwrap(); + let actual = std::fs::read(&many_path).unwrap(); + assert_eq!( + expected.len(), + actual.len(), + "the batch produced a file of a different length" + ); + assert_eq!(expected, actual, "the batch produced different bytes"); + + std::fs::remove_file(&one_path).unwrap(); + std::fs::remove_file(&many_path).unwrap(); + } + + /// Page ids with a gap in them are two runs, and each has to land at the + /// offset its own id names rather than after the one before it. + #[tokio::test] + async fn a_batch_with_a_gap_puts_each_page_at_its_own_offset() { + let (path, mut file) = scratch("batch_with_a_gap").await; + let pages = vec![ + page_at(0, b"first"), + page_at(1, b"second"), + // The gap: nothing at 2 or 3. + page_at(4, b"fifth"), + ]; + persist_pages_batch(pages, &mut file).await.unwrap(); + file.sync_all().await.unwrap(); + drop(file); + + let bytes = std::fs::read(&path).unwrap(); + // The last page is not padded, exactly as writing one at a time leaves + // it: page four's offset, its header, and its five bytes of marker. + assert_eq!( + bytes.len(), + 4 * PAGE_SIZE + crate::GENERAL_HEADER_SIZE + b"fifth".len(), + "the file is the wrong length" + ); + let marker_at = |page: usize, marker: &[u8]| { + let from = page * PAGE_SIZE + crate::GENERAL_HEADER_SIZE; + assert_eq!( + &bytes[from..from + marker.len()], + marker, + "page {page} holds the wrong data" + ); + }; + marker_at(0, b"first"); + marker_at(1, b"second"); + marker_at(4, b"fifth"); + // The skipped pages are zeroes, not a copy of anything. + let gap = 2 * PAGE_SIZE + crate::GENERAL_HEADER_SIZE; + assert!( + bytes[gap..gap + 16].iter().all(|&b| b == 0), + "the gap was written over" + ); + + std::fs::remove_file(&path).unwrap(); + } + #[tokio::test] async fn persist_page_rejects_inner_data_past_the_page_slot() { // A data page whose buffer is larger than the slot budget can hand From 58192d7ffa131856648e4ce73a92b5ff774afd42 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 16:49:27 +0700 Subject: [PATCH 3/6] Make the write path safe for a Linux VM, not just fast on this laptop Three things the first version of this branch got wrong or left open, found by asking what it does where it actually runs: a Linux VM writing local pages that are then synced to Tigris. **A `usize` underflow.** `PAGE_SIZE - written` is fine while a header serialises to exactly `GENERAL_HEADER_SIZE`, and is a panic in debug and an absurd seek in release the moment one does not. The inner length is already guarded, so reaching it needs a header change, which is exactly the kind of change that arrives without anyone thinking about this function. It is `checked_sub` now and returns `PageOverflow`. The old code had the same hole with a different ending: it computed the padding in `i64` and would have seeked *backwards* over the page it had just written. **An unbounded buffer.** A run of ten thousand pages was a 160 MB allocation. On a VM whose memory is not ours to spend that is not a trade worth making for a few more megabytes per write, so a run flushes every 512 pages, which is 8 MiB. It costs nothing measurable: 47.7 ms against 49.4 and 36.2 before the bound. **And one behaviour difference that is real and now written down.** Writing page by page *seeks* over the space between a page's content and the next page's start, so on a filesystem with holes that space is never allocated. Writing a run in one call puts explicit zeroes there. The bytes read back are identical, which is what `a_batch_writes_what_one_at_a_time_writes` holds; what differs is blocks allocated, and on `ext4` a file of half-empty pages will now occupy what it claims to. Data pages are full and pad to nothing; index and space pages do not. `a_run_longer_than_the_buffer_bound_still_joins_up` is the guard on the bound: 520 pages, crossing a seam, byte for byte against writing them one at a time. Co-Authored-By: Claude Opus 5 --- src/page/util.rs | 73 ++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 71 insertions(+), 2 deletions(-) diff --git a/src/page/util.rs b/src/page/util.rs index 09f6a24..4f0f55a 100644 --- a/src/page/util.rs +++ b/src/page/util.rs @@ -88,7 +88,16 @@ where // Returning the length gives them back. Measured over 6,400 pages, about // 14% of the write. let written = persist_page_in_place(page, file).await?; - let padding = PAGE_SIZE - written; + // Checked, because a page that wrote more than its slot must not turn into + // a `usize` underflow and an absurd seek. The inner length is already + // guarded against `INNER_PAGE_SIZE`, so reaching this needs a header that + // serialises to more than `GENERAL_HEADER_SIZE`; that cannot happen today + // and would be a silent file corruptor if it ever did. + let padding = PAGE_SIZE.checked_sub(written).ok_or(Error::PageOverflow { + page: page.header.page_id, + needed: written, + capacity: PAGE_SIZE, + })?; if padding > 0 { file.seek(SeekFrom::Current(padding as i64)).await?; } @@ -147,6 +156,25 @@ where // correspond to a stretch of the file. Callers usually pass a contiguous // batch and get one write; a caller that does not still gets a correct // file, one write per run. + // + // **The buffer is bounded.** A run of ten thousand pages is 160 MB, and + // this runs on a virtual machine whose memory is not ours to spend. A run + // longer than `MAX_RUN_PAGES` is flushed in pieces, each still one write, + // each still at the right offset. + // + // **One difference from writing page by page, and it is on disk rather + // than in the bytes.** Writing one at a time seeks over the space between + // a page's content and the next page's start, and on a filesystem that + // supports holes that space is never allocated. Writing a run in one call + // puts explicit zeroes there. A file read back is identical either way, + // which is what the tests hold; what differs is blocks allocated, and on + // `ext4` a file of half-empty pages will now occupy what it claims to. + // Data pages are full, so their padding is nothing; index and space pages + // are not. + /// Pages buffered before a run is flushed regardless of how long it is. + /// 512 pages is 8 MiB at the default page size. + const MAX_RUN_PAGES: u32 = 512; + let mut iter = pages.into_iter().peekable(); let mut buffer: Vec = Vec::new(); let mut run_start: Option = None; @@ -176,7 +204,12 @@ where expected_next = id.checked_add(1).ok_or(Error::Corrupt { what: "page id overflowed while batching", })?; - if iter.peek().is_none() { + + // Flush at the end, and before the buffer grows past its bound. The + // next page then starts a fresh run at its own offset, which is + // correct because that offset is absolute. + let run_is_long = id - start + 1 >= MAX_RUN_PAGES; + if iter.peek().is_none() || run_is_long { flush_run(file, run_start.take(), &mut buffer).await?; } } @@ -699,6 +732,42 @@ mod tests { std::fs::remove_file(&path).unwrap(); } + /// A run longer than the buffer bound is flushed in pieces, and the pieces + /// have to join up exactly. + /// + /// This is the guard on bounding the buffer. The bound exists so a batch of + /// ten thousand pages does not become a 160 MB allocation on a virtual + /// machine, and the risk it introduces is a seam every 512 pages. + #[tokio::test] + async fn a_run_longer_than_the_buffer_bound_still_joins_up() { + // Comfortably past `MAX_RUN_PAGES`, so at least one seam is crossed. + const COUNT: u32 = 520; + + let (one_path, mut one) = scratch("long_run_one_at_a_time").await; + for id in 0..COUNT { + let mut page = page_at(id, format!("page{id}").as_bytes()); + super::persist_page(&mut page, &mut one).await.unwrap(); + } + one.sync_all().await.unwrap(); + drop(one); + + let (many_path, mut many) = scratch("long_run_batched").await; + let pages: Vec<_> = (0..COUNT) + .map(|id| page_at(id, format!("page{id}").as_bytes())) + .collect(); + persist_pages_batch(pages, &mut many).await.unwrap(); + many.sync_all().await.unwrap(); + drop(many); + + let expected = std::fs::read(&one_path).unwrap(); + let actual = std::fs::read(&many_path).unwrap(); + assert_eq!(expected.len(), actual.len(), "lengths differ across a seam"); + assert_eq!(expected, actual, "bytes differ across a seam"); + + std::fs::remove_file(&one_path).unwrap(); + std::fs::remove_file(&many_path).unwrap(); + } + #[tokio::test] async fn persist_page_rejects_inner_data_past_the_page_slot() { // A data page whose buffer is larger than the slot budget can hand From 7567c40081d6aaf9624aee70c60a05887e22aa29 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 17:19:46 +0700 Subject: [PATCH 4/6] Measure the write that is not a whole file The benchmark opened every file with `truncate(true)` and wrote ids 0..6400 in order, so both arms rewrote a whole file from empty. That is the batch path's best case: one uninterrupted consecutive run, which is exactly what the coalescing was written for and close to the least of what a database does. The new arm updates every tenth page of a file that already exists. The run breaks at every page, so the batch path falls back to one write per page and can only win by what it saves per page. It does: 1.28x, and the saving is the removed `stream_position` call rather than any joining up. Against 7.7x on the whole-file arm, and at 352 MB/s against 4019, a scattered page also costs more than a sequential one before either path helps it. Co-Authored-By: Claude Opus 5 --- examples/write-pages.rs | 78 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/examples/write-pages.rs b/examples/write-pages.rs index 4e10d51..8c3d1a8 100644 --- a/examples/write-pages.rs +++ b/examples/write-pages.rs @@ -36,6 +36,27 @@ fn pages() -> Vec>> { .collect() } +/// Pages at every `stride`-th id, which is the shape of an update to a file +/// that already exists: the ids are not consecutive, so nothing coalesces. +fn scattered(stride: u32) -> Vec>> { + pages() + .into_iter() + .enumerate() + .filter(|(id, _)| *id as u32 % stride == 0) + .map(|(_, page)| page) + .collect() +} + +async fn existing(path: &std::path::Path) -> tokio::fs::File { + tokio::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .open(path) + .await + .unwrap() +} + async fn fresh(path: &std::path::Path) -> tokio::fs::File { tokio::fs::OpenOptions::new() .read(true) @@ -97,5 +118,62 @@ async fn main() { one / many ); + // ---- the case that is not a whole file + // + // Everything above rewrites the file from empty, so the ids run 0..PAGES + // with no gaps and the batch path sees one enormous consecutive run. That + // is its best case and a database's rarest one. Updating scattered pages + // in a file that already exists breaks the run at every page, so the batch + // path falls back to one write per page and can only win by what it saves + // per page, not by joining anything up. + const STRIDE: u32 = 10; + let touched = scattered(STRIDE).len(); + let touched_bytes = touched * data_bucket::PAGE_SIZE; + + // Lay the whole file down once, outside the clock, so the updates land in + // a file that is already the right length. + { + let mut file = fresh(&path).await; + persist_pages_batch(pages(), &mut file).await.unwrap(); + file.sync_all().await.unwrap(); + } + + let mut one_scattered = Vec::new(); + let mut batch_scattered = Vec::new(); + for _ in 0..REPS { + let mut some = scattered(STRIDE); + let mut file = existing(&path).await; + let at = Instant::now(); + for page in &mut some { + persist_page(page, &mut file).await.unwrap(); + } + file.sync_all().await.unwrap(); + one_scattered.push(at.elapsed().as_secs_f64()); + + let some = scattered(STRIDE); + let mut file = existing(&path).await; + let at = Instant::now(); + persist_pages_batch(some, &mut file).await.unwrap(); + file.sync_all().await.unwrap(); + batch_scattered.push(at.elapsed().as_secs_f64()); + } + + let (one_s, many_s) = (median(one_scattered), median(batch_scattered)); + println!( + "\nevery {STRIDE}th page of an existing file, {touched} pages, {:.1} MB", + touched_bytes as f64 / 1e6 + ); + println!( + " persist_page, one at a time {:>8.1} ms {:>6.0} MB/s", + one_s * 1e3, + touched_bytes as f64 / 1e6 / one_s + ); + println!( + " persist_pages_batch {:>8.1} ms {:>6.0} MB/s {:>5.2}x", + many_s * 1e3, + touched_bytes as f64 / 1e6 / many_s, + one_s / many_s + ); + let _ = std::fs::remove_file(&path); } From 07c70bce3bb1398e7e9a398c411054eb508fc7f4 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 17:23:15 +0700 Subject: [PATCH 5/6] Sweep the batch size, because the headline is measured at 6,400 pages The 5.6x for `persist_pages_batch` is one call carrying a whole 104 MB file. Swept against the sizes a caller actually passes, it is 0.98x at one page, 1.08x at sixteen, and does not reach 2x until 256. Below about sixty-four pages both paths sit at 4-5 ms on both versions, which is `sync_all` and not the write path: there is nothing there for batching to save. So the change is worth what the callers make it worth, and today two of the three pass `HashMap::values()`, whose order is arbitrary, so their runs break at every page and they would see 1.0x at any size. Co-Authored-By: Claude Opus 5 --- examples/write-pages-sweep.rs | 84 +++++++++++++++++++++++++++++++++++ 1 file changed, 84 insertions(+) create mode 100644 examples/write-pages-sweep.rs diff --git a/examples/write-pages-sweep.rs b/examples/write-pages-sweep.rs new file mode 100644 index 0000000..07f27a0 --- /dev/null +++ b/examples/write-pages-sweep.rs @@ -0,0 +1,84 @@ +//! Where batching starts to matter, as a function of how many pages a caller +//! actually hands over at once. +//! +//! The headline number for `persist_pages_batch` is measured at 6,400 pages in +//! one call. The question this answers is whether anything reaches that, and +//! what the two paths cost at the sizes a caller really passes. + +use data_bucket::page::{persist_page, persist_pages_batch}; +use data_bucket::{DataPage, GeneralHeader, GeneralPage, PageType, DATA_VERSION, INNER_PAGE_SIZE}; +use std::time::Instant; + +const SIZES: [usize; 8] = [1, 2, 4, 16, 64, 256, 1024, 6400]; +const REPS: usize = 9; + +fn pages(count: usize) -> Vec>> { + (0..count as u32) + .map(|id| { + let mut data = [0u8; INNER_PAGE_SIZE]; + for (n, byte) in data.iter_mut().enumerate() { + *byte = (n % 251) as u8; + } + GeneralPage { + header: GeneralHeader { + data_version: DATA_VERSION, + space_id: 1.into(), + page_id: id.into(), + previous_id: 0.into(), + next_id: 0.into(), + page_type: PageType::Data, + data_length: 0, + }, + inner: DataPage { length: INNER_PAGE_SIZE as u32, data }, + } + }) + .collect() +} + +async fn fresh(path: &std::path::Path) -> tokio::fs::File { + tokio::fs::OpenOptions::new() + .read(true).write(true).create(true).truncate(true) + .open(path).await.unwrap() +} + +fn median(mut v: Vec) -> f64 { + v.sort_by(|a, b| a.partial_cmp(b).unwrap()); + v[v.len() / 2] +} + +#[tokio::main] +async fn main() { + let path = std::env::var("SCRATCH") + .map(std::path::PathBuf::from) + .unwrap_or_else(|_| std::env::temp_dir()) + .join("data_bucket_sweep.wt"); + + println!(" pages MB one-at-a-time batched gain"); + for count in SIZES { + let bytes = count * data_bucket::PAGE_SIZE; + let (mut ones, mut many) = (Vec::new(), Vec::new()); + // One untimed pass of each, so neither pays for the file appearing. + { let mut f = fresh(&path).await; persist_pages_batch(pages(count), &mut f).await.unwrap(); f.sync_all().await.unwrap(); } + for _ in 0..REPS { + let mut all = pages(count); + let mut file = fresh(&path).await; + let at = Instant::now(); + for page in &mut all { persist_page(page, &mut file).await.unwrap(); } + file.sync_all().await.unwrap(); + ones.push(at.elapsed().as_secs_f64()); + + let all = pages(count); + let mut file = fresh(&path).await; + let at = Instant::now(); + persist_pages_batch(all, &mut file).await.unwrap(); + file.sync_all().await.unwrap(); + many.push(at.elapsed().as_secs_f64()); + } + let (one, batch) = (median(ones), median(many)); + println!( + " {count:>5} {:>7.2} {:>7.2} ms {:>7.2} ms {:>5.2}x", + bytes as f64 / 1e6, one * 1e3, batch * 1e3, one / batch + ); + } + let _ = std::fs::remove_file(&path); +} From 8cc1bdd411a5552c9de11c785f40a80fb5ac3090 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 18:19:11 +0700 Subject: [PATCH 6/6] Let the arm order be set, since it is a variable either way Run in a fixed order, whichever arm goes second inherits a file the first just wrote. That is what made an fsync'd write measure faster than an undurable one in another benchmark this morning, and this one had the same shape: the batch arm always ran second. Reversed, it holds - 3272 and 4167 MB/s batch-first against 2023 and 3961 batch-second - so the order was not flattering it. What the check did show is that the arm swings 2023 to 4167 MB/s across runs, so it is a 2 to 4 GB/s range and not the single figure it had been quoted as. The other arms are steady to a few percent. Co-Authored-By: Claude Opus 5 --- examples/write-pages.rs | 26 ++++++++++++++++++++++---- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/examples/write-pages.rs b/examples/write-pages.rs index 8c3d1a8..cccd1f6 100644 --- a/examples/write-pages.rs +++ b/examples/write-pages.rs @@ -81,9 +81,16 @@ async fn main() { bytes as f64 / 1e6 ); + // **The order of the arms is a variable, so it is one that can be set.** + // Run in a fixed order, whichever arm goes second inherits a file the first + // arm just wrote and looks faster for it. `REVERSE=1` runs them the other + // way round; the two orders agreeing is what makes either number mean + // anything. + let reverse = std::env::var("REVERSE").is_ok(); let mut one_at_a_time = Vec::new(); let mut batched = Vec::new(); - for _ in 0..REPS { + + let mut run_one = async |timings: &mut Vec| { let mut all = pages(); let mut file = fresh(&path).await; let at = Instant::now(); @@ -91,14 +98,25 @@ async fn main() { persist_page(page, &mut file).await.unwrap(); } file.sync_all().await.unwrap(); - one_at_a_time.push(at.elapsed().as_secs_f64()); - + timings.push(at.elapsed().as_secs_f64()); + }; + let mut run_batch = async |timings: &mut Vec| { let all = pages(); let mut file = fresh(&path).await; let at = Instant::now(); persist_pages_batch(all, &mut file).await.unwrap(); file.sync_all().await.unwrap(); - batched.push(at.elapsed().as_secs_f64()); + timings.push(at.elapsed().as_secs_f64()); + }; + + for _ in 0..REPS { + if reverse { + run_batch(&mut batched).await; + run_one(&mut one_at_a_time).await; + } else { + run_one(&mut one_at_a_time).await; + run_batch(&mut batched).await; + } } let median = |mut v: Vec| {