use super::*;
pub(super) const BTRFS_SUPER_MAGIC: i64 = 0x9123_683e;
pub(super) const ZFS_SUPER_MAGIC: i64 = 0x2fc1_2fc1;
pub(super) const FS_NOCOW_FL: u64 = 0x0080_0000;
pub(super) const SUPERBLOCK_SLOT_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_SLOT_SIZE;
pub(super) const SUPERBLOCK_ZONE_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_ZONE_SIZE;
type NewestSuperblock = (
reddb_file::PagedSuperblockSelection,
[u8; SUPERBLOCK_SLOT_SIZE],
);
fn superblock_slot_image(page: &[u8; PAGE_SIZE]) -> [u8; SUPERBLOCK_SLOT_SIZE] {
let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
image.copy_from_slice(&page[..SUPERBLOCK_SLOT_SIZE]);
image
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum CowFilesystemKind {
Zfs,
BtrfsDataCow,
TestOverride,
}
pub(super) fn classify_cow_filesystem(
fs_type: i64,
mount_options: Option<&str>,
inode_flags: Option<u64>,
) -> Option<CowFilesystemKind> {
match fs_type {
ZFS_SUPER_MAGIC => Some(CowFilesystemKind::Zfs),
BTRFS_SUPER_MAGIC => {
let mount_options = mount_options?;
if mount_options.split(',').any(|option| option == "nodatacow") {
return None;
}
let inode_flags = inode_flags?;
if inode_flags & FS_NOCOW_FL != 0 {
return None;
}
Some(CowFilesystemKind::BtrfsDataCow)
}
_ => None,
}
}
#[cfg(target_os = "linux")]
fn linux_fstatfs_type(file: &File) -> Option<i64> {
use std::mem::MaybeUninit;
use std::os::fd::AsRawFd;
let mut stat = MaybeUninit::<libc::statfs>::uninit();
let rc = unsafe { libc::fstatfs(file.as_raw_fd(), stat.as_mut_ptr()) };
if rc != 0 {
return None;
}
let stat = unsafe { stat.assume_init() };
#[allow(clippy::unnecessary_cast)]
Some(stat.f_type as i64)
}
#[cfg(target_os = "linux")]
fn linux_inode_flags(file: &File) -> Option<u64> {
use std::os::fd::AsRawFd;
let mut flags: libc::c_long = 0;
let rc = unsafe { libc::ioctl(file.as_raw_fd(), libc::FS_IOC_GETFLAGS, &mut flags) };
if rc != 0 {
return None;
}
Some(flags as u64)
}
#[cfg(target_os = "linux")]
fn linux_mount_options_for_path(path: &Path) -> Option<String> {
let path = path.canonicalize().ok()?;
let mountinfo = std::fs::read_to_string("/proc/self/mountinfo").ok()?;
parse_mountinfo_options_for_path(&mountinfo, &path)
}
#[cfg(target_os = "linux")]
pub(super) fn parse_mountinfo_options_for_path(mountinfo: &str, path: &Path) -> Option<String> {
let mut best: Option<(usize, String)> = None;
for line in mountinfo.lines() {
let fields: Vec<&str> = line.split(' ').collect();
if fields.len() < 10 {
continue;
}
let Some(separator) = fields.iter().position(|field| *field == "-") else {
continue;
};
if separator + 3 >= fields.len() || separator < 6 {
continue;
}
let mount_point = mountinfo_unescape_path(fields[4]);
if !path.starts_with(&mount_point) {
continue;
}
let fs_type = fields[separator + 1];
if fs_type != "btrfs" && fs_type != "zfs" {
continue;
}
let mount_options = fields[5];
let super_options = fields[separator + 3];
let options = format!("{mount_options},{super_options}");
let depth = mount_point.components().count();
if best
.as_ref()
.map(|(best_depth, _)| depth > *best_depth)
.unwrap_or(true)
{
best = Some((depth, options));
}
}
best.map(|(_, options)| options)
}
#[cfg(target_os = "linux")]
fn mountinfo_unescape_path(value: &str) -> PathBuf {
let bytes = value.as_bytes();
let mut out = Vec::with_capacity(bytes.len());
let mut i = 0;
while i < bytes.len() {
if bytes[i] == b'\\' && i + 3 < bytes.len() {
let octal = &value[i + 1..i + 4];
if let Ok(byte) = u8::from_str_radix(octal, 8) {
out.push(byte);
i += 4;
continue;
}
}
out.push(bytes[i]);
i += 1;
}
PathBuf::from(String::from_utf8_lossy(&out).into_owned())
}
impl Pager {
pub fn open<P: AsRef<Path>>(path: P, mut config: PagerConfig) -> Result<Self, PagerError> {
let path = path.as_ref().to_path_buf();
let exists = path.exists();
if !exists && !config.create {
return Err(PagerError::InvalidDatabase(
"Database does not exist".into(),
));
}
if exists {
if let Some(sidecar) = reddb_file::layout::retired::first_present_phase1_sidecar(&path)
{
return Err(PagerError::LegacySidecarStore { sidecar });
}
}
if !exists && config.read_only {
return Err(PagerError::InvalidDatabase(
"Cannot create read-only database".into(),
));
}
let file = OpenOptions::new()
.read(true)
.write(!config.read_only)
.create(config.create && !config.read_only)
.open(&path)?;
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
if let Ok(meta) = file.metadata() {
let fs_block_size = meta.blksize();
if Self::page_size_misaligned_with_block(PAGE_SIZE, fs_block_size) {
tracing::warn!(
page_size = PAGE_SIZE,
fs_block_size,
path = %path.display(),
"database page size is not a multiple of the filesystem \
block size; page writes will straddle FS blocks \
(read-modify-write amplification). Diagnostic only — \
the page size is unchanged."
);
}
}
}
let lock_file = if !config.read_only {
let lf = OpenOptions::new().read(true).write(true).open(&path)?;
lf.try_lock_exclusive().map_err(|_| PagerError::Locked)?;
Some(lf)
} else {
let lf = OpenOptions::new().read(true).open(&path)?;
match lf.try_lock_shared() {
Ok(_) => Some(lf),
Err(_) => None,
}
};
let fold_dwb = crate::physical::fold_dwb_into_wal_enabled();
if !config.double_write && !config.read_only && !fold_dwb {
let skip_dwb_on_cow =
Self::cow_filesystem_has_atomic_page_writes(&path, &file).is_some();
if !skip_dwb_on_cow {
tracing::warn!(
path = %path.display(),
"double_write=false requested, but the data file is not proven to be on \
ZFS or btrfs datacow; keeping the double-write buffer enabled"
);
config.double_write = true;
}
}
let dwb_file = if config.double_write && !config.read_only && !fold_dwb {
let f = Self::open_dwb_file(&path)?;
Some(Mutex::new(f))
} else {
if fold_dwb && !config.read_only {
let _ = std::fs::remove_file(Self::dwb_path(&path));
}
None
};
let mut pager = Self {
path,
file: Mutex::new(file),
_lock_file: lock_file,
dwb_file,
cache: PageCache::new(config.cache_size),
freelist: RwLock::new(FreeList::new()),
header: RwLock::new(DatabaseHeader::default()),
config,
header_dirty: Mutex::new(false),
wal: RwLock::new(None),
encryption: None,
};
if exists {
pager.recover_from_dwb()?;
if !pager.config.double_write && !pager.config.read_only {
let _ = std::fs::remove_file(Self::dwb_path(&pager.path));
}
pager.load_header()?;
pager.bind_encryption_for_existing()?;
} else {
pager.initialize()?;
pager.bind_encryption_for_new()?;
}
Ok(pager)
}
pub(crate) fn page_size_misaligned_with_block(page_size: usize, fs_block_size: u64) -> bool {
fs_block_size != 0 && !(page_size as u64).is_multiple_of(fs_block_size)
}
#[cfg(test)]
fn cow_filesystem_test_override() -> Option<bool> {
match COW_ATOMIC_WRITE_TEST_OVERRIDE.load(Ordering::Relaxed) {
1 => Some(true),
2 => Some(false),
_ => None,
}
}
#[cfg(not(test))]
fn cow_filesystem_test_override() -> Option<bool> {
None
}
fn cow_filesystem_has_atomic_page_writes(
path: &Path,
file: &File,
) -> Option<CowFilesystemKind> {
if let Some(allowed) = Self::cow_filesystem_test_override() {
return allowed.then_some(CowFilesystemKind::TestOverride);
}
Self::probe_cow_filesystem(path, file)
}
#[cfg(target_os = "linux")]
fn probe_cow_filesystem(path: &Path, file: &File) -> Option<CowFilesystemKind> {
let fs_type = linux_fstatfs_type(file)?;
match fs_type {
ZFS_SUPER_MAGIC => Some(CowFilesystemKind::Zfs),
BTRFS_SUPER_MAGIC => {
let mount_options = linux_mount_options_for_path(path)?;
let inode_flags = linux_inode_flags(file)?;
classify_cow_filesystem(fs_type, Some(&mount_options), Some(inode_flags))
}
_ => None,
}
}
#[cfg(not(target_os = "linux"))]
fn probe_cow_filesystem(_path: &Path, _file: &File) -> Option<CowFilesystemKind> {
None
}
fn bind_encryption_for_existing(&mut self) -> Result<(), PagerError> {
if self.page_count().unwrap_or(0) == 0 {
return self.bind_encryption_for_new();
}
let (_, image) = self.newest_superblock()?;
let data = &image[..];
let has_marker = reddb_file::paged_encryption_marker_present(data);
let key = self.config.encryption.clone();
match (has_marker, key) {
(true, Some(key)) => {
let header_bytes =
reddb_file::paged_encryption_header_bytes(data).ok_or_else(|| {
PagerError::InvalidDatabase("encryption header parse failed".to_string())
})?;
let header = crate::storage::encryption::EncryptionHeader::from_bytes(header_bytes)
.map_err(|e| {
PagerError::InvalidDatabase(format!("encryption header parse failed: {e}"))
})?;
if !header.validate(&key) {
return Err(PagerError::InvalidKey);
}
let encryptor = crate::storage::encryption::PageEncryptor::new(key);
self.encryption = Some((encryptor, header));
Ok(())
}
(true, None) => Err(PagerError::EncryptionRequired),
(false, Some(_)) => Err(PagerError::PlainDatabaseRefusesKey),
(false, None) => Ok(()),
}
}
fn bind_encryption_for_new(&mut self) -> Result<(), PagerError> {
let Some(key) = self.config.encryption.clone() else {
return Ok(());
};
let header = crate::storage::encryption::EncryptionHeader::new(&key);
let encryptor = crate::storage::encryption::PageEncryptor::new(key);
if self.page_count().unwrap_or(0) > 0 {
let header_bytes = header.to_bytes();
self.commit_superblock(|image| {
reddb_file::write_paged_encryption_marker_and_header(image, &header_bytes)
.map_err(|err| PagerError::InvalidDatabase(err.to_string()))
})?;
}
self.encryption = Some((encryptor, header));
Ok(())
}
pub fn open_default<P: AsRef<Path>>(path: P) -> Result<Self, PagerError> {
Self::open(path, PagerConfig::default())
}
fn initialize(&self) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let initial_page_count = 3;
let header_page = Page::new_header_page(initial_page_count);
self.header_write()?.page_count = initial_page_count;
self.write_page_raw(0, &header_page)?;
let mut metadata_page = Page::new(PageType::Header, 1);
metadata_page.update_checksum();
self.write_page_raw(1, &metadata_page)?;
let mut vault_page = Page::new(PageType::Vault, 2);
vault_page.update_checksum();
self.write_page_raw(2, &vault_page)?;
let mut slot = superblock_slot_image(header_page.as_bytes());
self.write_superblock_slot(&mut slot, 0, 1)?;
self.write_superblock_slot(&mut slot, 1, 2)?;
self.sync()?;
Ok(())
}
fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
self.header.write().map_err(|_| PagerError::LockPoisoned)
}
fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
self.header.read().map_err(|_| PagerError::LockPoisoned)
}
fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
self.freelist.write().map_err(|_| PagerError::LockPoisoned)
}
fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, File>, PagerError> {
self.file.lock().map_err(|_| PagerError::LockPoisoned)
}
fn header_dirty_lock(&self) -> Result<std::sync::MutexGuard<'_, bool>, PagerError> {
self.header_dirty
.lock()
.map_err(|_| PagerError::LockPoisoned)
}
fn read_superblock_zone(&self) -> Result<[u8; SUPERBLOCK_ZONE_SIZE], PagerError> {
let mut file = self.file_lock()?;
let len = file.metadata().map(|meta| meta.len()).unwrap_or(0);
let mut zone = [0u8; SUPERBLOCK_ZONE_SIZE];
let readable = len.min(SUPERBLOCK_ZONE_SIZE as u64) as usize;
if readable > 0 {
file.seek(SeekFrom::Start(0))?;
file.read_exact(&mut zone[..readable])?;
}
Ok(zone)
}
fn newest_superblock(&self) -> Result<NewestSuperblock, PagerError> {
let zone = self.read_superblock_zone()?;
let selection = reddb_file::select_paged_superblock(&zone)
.ok_or_else(|| PagerError::SuperblockZoneUnrecoverable(self.path.clone()))?;
let start = selection.copy_index * SUPERBLOCK_SLOT_SIZE;
let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
image.copy_from_slice(&zone[start..start + SUPERBLOCK_SLOT_SIZE]);
Ok((selection, image))
}
fn write_superblock_slot(
&self,
slot: &mut [u8; SUPERBLOCK_SLOT_SIZE],
copy_index: usize,
generation: u64,
) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
reddb_file::seal_paged_superblock_slot(slot, copy_index, generation)
.map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
let mut file = self.file_lock()?;
file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
copy_index,
)))?;
file.write_all(slot)?;
file.sync_all()?;
Ok(())
}
fn commit_superblock(
&self,
mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let (newest, mut image) = self.newest_superblock()?;
mutate(&mut image)?;
let next_copy = reddb_file::paged_superblock_next_copy(Some(newest));
self.write_superblock_slot(&mut image, next_copy, newest.generation.saturating_add(1))
}
fn load_header(&self) -> Result<(), PagerError> {
let (_, image) = self.newest_superblock()?;
let decoded_header = reddb_file::decode_database_header(&image)
.map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
let freelist_head = decoded_header.freelist_head;
{
let mut header = self.header_write()?;
*header = decoded_header;
}
{
let mut freelist = self.freelist_write()?;
*freelist = FreeList::from_header(freelist_head, 0);
}
Ok(())
}
fn write_header(&self) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let header = self.header_read()?.clone();
self.commit_superblock(|image| {
reddb_file::encode_database_header(image, &header)
.map_err(|err| PagerError::InvalidDatabase(err.to_string()))
})?;
*self.header_dirty_lock()? = false;
Ok(())
}
fn read_page_raw(&self, page_id: u32) -> Result<Page, PagerError> {
let mut file = self.file_lock()?;
let offset = (page_id as u64) * (PAGE_SIZE as u64);
file.seek(SeekFrom::Start(offset))?;
let mut buf = [0u8; PAGE_SIZE];
file.read_exact(&mut buf)?;
let page = Page::from_bytes(buf);
if self.config.verify_checksums && page_id != 0 {
page.verify_checksum()?;
}
Ok(page)
}
fn write_page_raw(&self, page_id: u32, page: &Page) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let mut file = self.file_lock()?;
let offset = (page_id as u64) * (PAGE_SIZE as u64);
file.seek(SeekFrom::Start(offset))?;
file.write_all(page.as_bytes())?;
Ok(())
}
pub fn read_page(&self, page_id: u32) -> Result<Page, PagerError> {
if let Some(page) = self.cache.get(page_id) {
return Ok(page);
}
let page = self.read_page_raw(page_id)?;
if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
let evicted_id = dirty_page.page_id();
self.write_page_raw(evicted_id, &dirty_page)?;
}
Ok(page)
}
pub fn read_page_no_checksum(&self, page_id: u32) -> Result<Page, PagerError> {
if let Some(page) = self.cache.get(page_id) {
return Ok(page);
}
let mut file = self.file_lock()?;
let offset = (page_id as u64) * (PAGE_SIZE as u64);
file.seek(SeekFrom::Start(offset))?;
let mut buf = [0u8; PAGE_SIZE];
file.read_exact(&mut buf)?;
drop(file);
let page = Page::from_bytes(buf);
if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
let evicted_id = dirty_page.page_id();
self.write_page_raw(evicted_id, &dirty_page)?;
}
Ok(page)
}
pub fn write_page(&self, page_id: u32, mut page: Page) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
page.update_checksum();
if let Some(dirty_page) = self.cache.insert(page_id, page) {
let evicted_id = dirty_page.page_id();
self.write_page_raw(evicted_id, &dirty_page)?;
}
self.cache.mark_dirty(page_id);
Ok(())
}
pub fn read_page_decrypted(&self, page_id: u32) -> Result<Page, PagerError> {
if page_id == 0 || self.encryption.is_none() {
return self.read_page(page_id);
}
let raw = self.read_page_no_checksum(page_id)?;
let (enc, _) = self
.encryption
.as_ref()
.expect("encryption presence checked above");
let plaintext = enc
.decrypt(page_id, raw.as_bytes())
.map_err(|e| PagerError::InvalidDatabase(format!("decrypt page {page_id}: {e}")))?;
let mut buf = [0u8; PAGE_SIZE];
let n = plaintext.len().min(PAGE_SIZE);
buf[..n].copy_from_slice(&plaintext[..n]);
Ok(Page::from_bytes(buf))
}
pub fn write_page_encrypted(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
if page_id == 0 || self.encryption.is_none() {
return self.write_page(page_id, page);
}
const OVERHEAD: usize = crate::storage::encryption::page_encryptor::OVERHEAD;
let plaintext_len = PAGE_SIZE - OVERHEAD;
let plaintext = &page.as_bytes()[..plaintext_len];
let (enc, _) = self
.encryption
.as_ref()
.expect("encryption presence checked above");
let ciphertext = enc.encrypt(page_id, plaintext);
debug_assert_eq!(ciphertext.len(), PAGE_SIZE);
let mut buf = [0u8; PAGE_SIZE];
buf.copy_from_slice(&ciphertext);
let cipher_page = Page::from_bytes(buf);
self.write_page_no_checksum(page_id, cipher_page)
}
pub fn write_page_no_checksum(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
if let Some(dirty_page) = self.cache.insert(page_id, page) {
let evicted_id = dirty_page.page_id();
self.write_page_raw(evicted_id, &dirty_page)?;
}
self.cache.mark_dirty(page_id);
Ok(())
}
pub fn allocate_page(&self, page_type: PageType) -> Result<Page, PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let page_id = {
let mut freelist = self.freelist_write()?;
if let Some(id) = freelist.allocate() {
id
} else if freelist.trunk_head() != 0 {
let trunk_id = freelist.trunk_head();
drop(freelist);
let trunk = self.read_page(trunk_id).map_err(|e| match e {
PagerError::PageNotFound(_) => {
PagerError::InvalidDatabase("Freelist trunk missing".to_string())
}
other => other,
})?;
let mut freelist = self.freelist_write()?;
freelist
.load_from_trunk(&trunk)
.map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {}", e)))?;
let id = freelist.allocate().ok_or_else(|| {
PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
})?;
let mut header = self.header_write()?;
header.freelist_head = freelist.trunk_head();
*self.header_dirty_lock()? = true;
id
} else {
let mut header = self.header_write()?;
let id = header.page_count;
header.page_count += 1;
*self.header_dirty_lock()? = true;
id
}
};
let page = Page::new(page_type, page_id);
if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
let evicted_id = dirty_page.page_id();
self.write_page_raw(evicted_id, &dirty_page)?;
}
self.cache.mark_dirty(page_id);
Ok(page)
}
pub fn reserve_contig_extent(&self, n_pages: u32) -> Result<super::ExtentId, PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
if n_pages == 0 {
return Err(PagerError::InvalidDatabase(
"contiguous extent must reserve at least one page".to_string(),
));
}
let start_page = {
let mut header = self.header_write()?;
let start = header.page_count;
header.page_count = header.page_count.checked_add(n_pages).ok_or_else(|| {
PagerError::InvalidDatabase("contiguous extent page count overflow".to_string())
})?;
*self.header_dirty_lock()? = true;
start
};
for page_id in start_page..start_page + n_pages {
let mut page = Page::new(PageType::Vector, page_id);
page.update_checksum();
if let Some(dirty_page) = self.cache.insert(page_id, page) {
let evicted_id = dirty_page.page_id();
self.write_page_raw(evicted_id, &dirty_page)?;
}
self.cache.mark_dirty(page_id);
}
Ok(super::ExtentId {
start_page,
n_pages,
})
}
pub fn free_page(&self, page_id: u32) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
self.cache.remove(page_id);
let mut freelist = self.freelist_write()?;
freelist.free(page_id);
*self.header_dirty_lock()? = true;
Ok(())
}
pub fn header(&self) -> Result<DatabaseHeader, PagerError> {
Ok(self.header_read()?.clone())
}
pub fn physical_header(&self) -> Result<PhysicalFileHeader, PagerError> {
Ok(self.header_read()?.physical)
}
pub fn update_physical_header(&self, physical: PhysicalFileHeader) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let mut header = self.header_write()?;
header.physical = physical;
*self.header_dirty_lock()? = true;
Ok(())
}
pub fn page_count(&self) -> Result<u32, PagerError> {
Ok(self.header_read()?.page_count)
}
pub fn set_wal_writer(&self, wal: Arc<Mutex<crate::storage::wal::writer::WalWriter>>) {
let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
*slot = Some(wal);
}
pub fn clear_wal_writer(&self) {
let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
*slot = None;
}
pub fn has_wal_writer(&self) -> bool {
self.wal.read().map(|s| s.is_some()).unwrap_or(false)
}
pub fn flush(&self) -> Result<(), PagerError> {
if self.config.read_only {
return Ok(());
}
let trunks = {
let mut freelist = self.freelist_write()?;
if freelist.is_dirty() {
let mut header = self.header_write()?;
let trunks = freelist.flush_to_trunks(0, || {
let id = header.page_count;
header.page_count += 1;
id
});
header.freelist_head = freelist.trunk_head();
*self.header_dirty_lock()? = true;
freelist.mark_clean();
trunks
} else {
Vec::new()
}
};
for trunk in trunks {
let page_id = trunk.page_id();
self.cache.insert(page_id, trunk);
self.cache.mark_dirty(page_id);
}
let dirty_pages = self.cache.flush_dirty();
if !dirty_pages.is_empty() {
let max_lsn = dirty_pages
.iter()
.filter_map(|(_, page)| page.header().ok().map(|h| h.lsn))
.max()
.unwrap_or(0);
if max_lsn > 0 {
if let Ok(slot) = self.wal.read() {
if let Some(wal) = slot.as_ref() {
let wal = Arc::clone(wal);
drop(slot);
let mut wal_guard = wal.lock().unwrap_or_else(|p| p.into_inner());
wal_guard.flush_until(max_lsn).map_err(PagerError::Io)?;
}
}
}
self.write_pages_through_dwb(&dirty_pages)?;
}
if *self.header_dirty_lock()? {
self.write_header()?;
}
Ok(())
}
pub fn sync(&self) -> Result<(), PagerError> {
self.flush()?;
let file = self.file_lock()?;
file.sync_all()?;
Ok(())
}
pub fn cache_stats(&self) -> crate::storage::engine::page_cache::CacheStats {
self.cache.stats()
}
pub fn cache_len(&self) -> usize {
self.cache.len()
}
pub fn cache_capacity(&self) -> usize {
self.cache.capacity()
}
pub fn dirty_page_count(&self) -> usize {
self.cache.dirty_count()
}
pub fn dirty_fraction(&self) -> f64 {
let capacity = self.cache.capacity().max(1) as f64;
self.cache.dirty_count() as f64 / capacity
}
pub fn flush_some_dirty(&self, max: usize) -> Result<usize, PagerError> {
if self.config.read_only || max == 0 {
return Ok(0);
}
let dirty_pages = self.cache.flush_some_dirty(max);
if dirty_pages.is_empty() {
return Ok(0);
}
let count = dirty_pages.len();
for (page_id, page) in dirty_pages {
self.write_page(page_id, page)?;
}
Ok(count)
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn is_read_only(&self) -> bool {
self.config.read_only
}
pub fn file_size(&self) -> Result<u64, PagerError> {
let file = self.file_lock()?;
Ok(file.metadata()?.len())
}
pub fn prefetch_hint(&self, page_id: u32) {
if let Ok(file) = self.file_lock() {
let _ = crate::storage::btree::prefetch::prefetch_page(
&file,
page_id as u64,
PAGE_SIZE as u32,
);
}
}
fn dwb_path(db_path: &Path) -> PathBuf {
reddb_file::layout::pager_dwb_shadow_path(db_path)
}
fn open_dwb_file(db_path: &Path) -> Result<File, PagerError> {
Ok(OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(Self::dwb_path(db_path))?)
}
fn clear_dwb_file(file: &mut File) -> Result<(), PagerError> {
file.set_len(0)?;
file.seek(SeekFrom::Start(0))?;
file.sync_all()?;
Ok(())
}
pub fn read_manifest_page(&self) -> Result<Page, PagerError> {
self.read_page(1)
.map_err(|_| PagerError::ManifestZoneCorrupt(self.path.clone()))
}
fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
if let Some(dwb_mutex) = &self.dwb_file {
let mut dwb = dwb_mutex.lock().map_err(|_| PagerError::LockPoisoned)?;
let buf = reddb_file::encode_paged_dwb_frame(
pages
.iter()
.map(|(page_id, page)| (*page_id, page.as_bytes())),
);
dwb.seek(SeekFrom::Start(0))?;
dwb.write_all(&buf)?;
dwb.set_len(buf.len() as u64)?;
dwb.sync_all()?;
for (page_id, page) in pages {
self.write_page_raw(*page_id, page)?;
}
Self::clear_dwb_file(&mut dwb)?;
Ok(())
} else {
for (page_id, page) in pages {
self.write_page_raw(*page_id, page)?;
}
Ok(())
}
}
fn recover_from_dwb(&self) -> Result<(), PagerError> {
let dwb_path = Self::dwb_path(&self.path);
if !dwb_path.exists() {
return Ok(());
}
if let Some(dwb_mutex) = &self.dwb_file {
let mut file = dwb_mutex.lock().map_err(|_| PagerError::LockPoisoned)?;
return self.recover_from_dwb_file(&mut file);
}
let mut file = OpenOptions::new().read(true).write(true).open(&dwb_path)?;
self.recover_from_dwb_file(&mut file)
}
fn recover_from_dwb_file(&self, file: &mut File) -> Result<(), PagerError> {
file.seek(SeekFrom::Start(0))?;
let len = file.metadata()?.len();
let mut buf = vec![0u8; len as usize];
file.read_exact(&mut buf)?;
let entries = match reddb_file::decode_paged_dwb_frame(&buf) {
Ok(entries) => entries,
Err(_) => return Self::clear_dwb_file(file),
};
for entry in entries {
let page = Page::from_bytes(entry.page);
self.write_page_raw(entry.page_id, &page)?;
}
{
let file = self.file_lock()?;
file.sync_all()?;
}
Self::clear_dwb_file(file)
}
pub fn write_header_and_sync(&self) -> Result<(), PagerError> {
self.write_header()?;
let file = self.file_lock()?;
file.sync_all()?;
Ok(())
}
pub fn set_checkpoint_in_progress(
&self,
in_progress: bool,
target_lsn: u64,
) -> Result<(), PagerError> {
let mut header = self.header_write()?;
header.checkpoint_in_progress = in_progress;
header.checkpoint_target_lsn = target_lsn;
*self.header_dirty_lock()? = true;
drop(header);
self.write_header_and_sync()
}
pub fn complete_checkpoint(&self, lsn: u64) -> Result<(), PagerError> {
let mut header = self.header_write()?;
header.checkpoint_lsn = lsn;
header.checkpoint_in_progress = false;
header.checkpoint_target_lsn = 0;
*self.header_dirty_lock()? = true;
drop(header);
self.write_header_and_sync()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn temp_db_path(name: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"reddb-pager-{}-{}-{}.rdb",
name,
std::process::id(),
crate::utils::now_unix_nanos()
))
}
fn cleanup(path: &Path) {
let _ = std::fs::remove_file(path);
let _ = std::fs::remove_file(Pager::dwb_path(path));
}
fn read_zone(path: &Path) -> Vec<u8> {
let mut zone = vec![0u8; SUPERBLOCK_ZONE_SIZE];
let mut file = File::open(path).expect("open() should succeed");
file.read_exact(&mut zone)
.expect("read_exact() should succeed");
zone
}
fn poke_slot(path: &Path, copy_index: usize, bytes: &[u8]) {
let mut file = OpenOptions::new()
.write(true)
.open(path)
.expect("open() should succeed");
file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
copy_index,
)))
.expect("operation should succeed");
file.write_all(bytes).expect("write_all() should succeed");
file.sync_all().expect("sync_all() should succeed");
}
#[test]
fn open_refuses_future_database_version() {
let path = temp_db_path("future-version");
let pager = Pager::open_default(&path).expect("open_default() should succeed");
drop(pager);
let selection = reddb_file::select_paged_superblock(&read_zone(&path))
.expect("operation should succeed");
let mut slot = vec![0u8; SUPERBLOCK_SLOT_SIZE];
reddb_file::init_database_header_page(&mut slot, 1)
.expect("init_database_header_page() should succeed");
reddb_file::set_database_header_version(&mut slot, reddb_file::PAGE_FILE_VERSION + 1)
.expect("operation should succeed");
reddb_file::seal_paged_superblock_slot(
&mut slot,
selection.copy_index,
selection.generation + 1,
)
.expect("operation should succeed");
poke_slot(&path, selection.copy_index, &slot);
let err = match Pager::open_default(&path) {
Ok(_) => panic!("future database version should be rejected"),
Err(err) => err,
};
match err {
PagerError::InvalidDatabase(msg) => {
assert!(msg.contains("newer than supported"), "{msg}");
}
other => panic!("expected InvalidDatabase, got {other:?}"),
}
cleanup(&path);
}
#[test]
fn a_fresh_store_publishes_both_superblock_copies() {
let path = temp_db_path("fresh-pair");
let pager = Pager::open_default(&path).expect("open_default() should succeed");
drop(pager);
let zone = read_zone(&path);
for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
let start = copy_index * SUPERBLOCK_SLOT_SIZE;
assert!(
reddb_file::paged_superblock_slot_generation(
&zone[start..start + SUPERBLOCK_SLOT_SIZE],
copy_index
)
.is_some(),
"copy {copy_index} must be valid on a fresh store"
);
}
cleanup(&path);
}
#[test]
fn an_update_writes_the_stale_copy_and_leaves_the_newest_one_intact() {
let path = temp_db_path("ping-pong");
let pager = Pager::open_default(&path).expect("open_default() should succeed");
let before = reddb_file::select_paged_superblock(&read_zone(&path))
.expect("operation should succeed");
pager
.allocate_page(PageType::BTreeLeaf)
.expect("allocate_page() should succeed");
pager.sync().expect("sync() should succeed");
let zone = read_zone(&path);
let after = reddb_file::select_paged_superblock(&zone)
.expect("select_paged_superblock() should succeed");
assert_ne!(
after.copy_index, before.copy_index,
"an update must target the stale copy"
);
assert!(after.generation > before.generation);
let start = before.copy_index * SUPERBLOCK_SLOT_SIZE;
assert_eq!(
reddb_file::paged_superblock_slot_generation(
&zone[start..start + SUPERBLOCK_SLOT_SIZE],
before.copy_index
),
Some(before.generation)
);
drop(pager);
cleanup(&path);
}
#[test]
fn open_recovers_from_the_older_copy_when_the_newest_is_torn() {
let path = temp_db_path("torn-newest");
{
let pager = Pager::open_default(&path).expect("open_default() should succeed");
pager
.allocate_page(PageType::BTreeLeaf)
.expect("allocate_page() should succeed");
pager.sync().expect("sync() should succeed");
}
let newest = reddb_file::select_paged_superblock(&read_zone(&path))
.expect("operation should succeed");
let mut torn = vec![0u8; SUPERBLOCK_SLOT_SIZE];
torn.copy_from_slice(
&read_zone(&path)[newest.copy_index * SUPERBLOCK_SLOT_SIZE
..newest.copy_index * SUPERBLOCK_SLOT_SIZE + SUPERBLOCK_SLOT_SIZE],
);
torn[64] ^= 0xFF;
poke_slot(&path, newest.copy_index, &torn);
let pager = Pager::open_default(&path).expect("older copy must root the store");
assert!(pager.page_count().expect("page_count() should succeed") >= 3);
drop(pager);
cleanup(&path);
}
#[test]
fn open_refuses_didactically_when_both_superblock_copies_are_invalid() {
let path = temp_db_path("both-invalid");
{
let _ = Pager::open_default(&path).expect("open_default() should succeed");
}
let zone = read_zone(&path);
for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
let start = copy_index * SUPERBLOCK_SLOT_SIZE;
let mut slot = zone[start..start + SUPERBLOCK_SLOT_SIZE].to_vec();
slot[64] ^= 0xFF;
poke_slot(&path, copy_index, &slot);
}
let err = match Pager::open_default(&path) {
Ok(_) => panic!("a rootless store must not open"),
Err(err) => err,
};
let rendered = err.to_string();
assert!(matches!(err, PagerError::SuperblockZoneUnrecoverable(_)));
assert!(rendered.contains("superblock zone"), "{rendered}");
assert!(rendered.contains("salvage"), "{rendered}");
cleanup(&path);
}
#[test]
fn open_refuses_a_legacy_sidecar_store_and_names_the_migration_tool() {
let path = temp_db_path("legacy-sidecar");
{
let _ = Pager::open_default(&path).expect("open_default() should succeed");
}
let sidecar = reddb_file::layout::retired::pager_header_path_v0(&path);
std::fs::write(&sidecar, b"legacy header shadow").expect("write() should succeed");
let err = match Pager::open_default(&path) {
Ok(_) => panic!("legacy stores need migration first"),
Err(err) => err,
};
let rendered = err.to_string();
assert!(matches!(err, PagerError::LegacySidecarStore { .. }));
assert!(rendered.contains("migration tool"), "{rendered}");
let _ = std::fs::remove_file(&sidecar);
cleanup(&path);
}
#[test]
fn no_phase_one_sidecar_survives_a_create_write_reopen_cycle() {
let path = temp_db_path("census");
{
let pager = Pager::open_default(&path).expect("open_default() should succeed");
let page = pager
.allocate_page(PageType::BTreeLeaf)
.expect("allocate_page() should succeed");
pager
.write_page(page.page_id(), page)
.expect("operation should succeed");
pager.sync().expect("sync() should succeed");
}
{
let pager = Pager::open_default(&path).expect("open_default() should succeed");
pager.sync().expect("sync() should succeed");
}
assert_eq!(
reddb_file::layout::retired::first_present_phase1_sidecar(&path),
None,
"the pager must never create a retired phase-1 sidecar"
);
cleanup(&path);
}
}