use super::freelist::FreeList;
use super::page::{Page, PageError, PageType, PAGE_SIZE};
use super::page_cache::PageCache;
use super::pager_io::{self, FIRST_ALLOCATABLE_PAGE};
use reddb_file::VfsFile;
pub use reddb_file::{DatabaseHeader, PhysicalFileHeader};
use std::path::{Path, PathBuf};
use std::sync::{Mutex, RwLock};
#[derive(Debug)]
pub enum PagerError {
Io(std::io::Error),
Page(PageError),
InvalidDatabase(String),
ReadOnly,
PageNotFound(u32),
Locked,
LockPoisoned,
EncryptionRequired,
PlainDatabaseRefusesKey,
InvalidKey,
SuperblockZoneUnrecoverable(PathBuf),
ManifestZoneCorrupt(PathBuf),
DwbZoneCorrupt(PathBuf),
LegacySidecarStore { sidecar: PathBuf },
LegacyPagedStore { path: PathBuf },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ExtentId {
pub start_page: u32,
pub n_pages: u32,
}
impl std::fmt::Display for PagerError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Io(e) => write!(f, "I/O error: {}", e),
Self::Page(e) => write!(f, "Page error: {}", e),
Self::InvalidDatabase(msg) => write!(f, "Invalid database: {}", msg),
Self::ReadOnly => write!(f, "Database is read-only"),
Self::PageNotFound(id) => write!(f, "Page {} not found", id),
Self::Locked => write!(f, "Database is locked"),
Self::LockPoisoned => write!(f, "Internal lock poisoned (concurrent thread panicked)"),
Self::EncryptionRequired => write!(
f,
"Database is encrypted but no key was supplied (set PagerConfig::encryption)"
),
Self::PlainDatabaseRefusesKey => write!(
f,
"Plain (unencrypted) database opened with an encryption key — refusing"
),
Self::InvalidKey => write!(f, "Encryption key validation failed for this database"),
Self::SuperblockZoneUnrecoverable(path) => write!(
f,
"superblock zone of {} is unrecoverable: both ping-pong copies failed \
validation, so no generation can be trusted to root the store. The store \
will not be opened. Recover what survives with red salvage \
(ADR 0074 §4); it reads the damaged file without writing to it and \
reports what it could not verify.",
path.display()
),
Self::ManifestZoneCorrupt(path) => write!(
f,
"internal manifest zone of {} failed its checksum: the zone that names \
collections, indexes and the checkpoint boundary cannot be trusted, so no \
rows are returned rather than garbage ones. Run the `SCRUB` statement to classify the fault \
and `red salvage` to extract what survives (ADR 0074 §2/§4).",
path.display()
),
Self::DwbZoneCorrupt(path) => write!(
f,
"double-write zone of {} failed validation: page recovery cannot prove \
whether the protected in-place write completed. The store will not be opened \
silently; run the `SCRUB` statement to classify the fault and `red salvage` to extract trusted \
rows (ADR 0074 §2/§4).",
path.display()
),
Self::LegacySidecarStore { sidecar } => write!(
f,
"refusing to open a legacy sidecar-backed store: found {}. The superblock \
pair, internal manifest, and double-write recovery state now live inside the \
.rdb file (ADR 0038 §2/§4), \
and the engine never reads the retired sidecars silently. Convert the store \
first with the offline migration tool \
(`reddb_server::pager_zone_migration::migrate_to_zoned`); it is reversible.",
sidecar.display()
),
Self::LegacyPagedStore { path } => write!(
f,
"refusing to open a pre-phase-3 paged store: {} does not carry the layout marker \
for the in-file double-write zone. Convert the store first with the offline \
migration tool (`reddb_server::pager_zone_migration::migrate_to_zoned`); it is \
reversible.",
path.display()
),
}
}
}
impl std::error::Error for PagerError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io(e) => Some(e),
Self::Page(e) => Some(e),
_ => None,
}
}
}
impl From<std::io::Error> for PagerError {
fn from(e: std::io::Error) -> Self {
Self::Io(e)
}
}
impl From<PageError> for PagerError {
fn from(e: PageError) -> Self {
Self::Page(e)
}
}
impl From<pager_io::PagerIoError> for PagerError {
fn from(error: pager_io::PagerIoError) -> Self {
match error {
pager_io::PagerIoError::Io(error) => Self::Io(error),
pager_io::PagerIoError::Page(error) => Self::Page(error),
pager_io::PagerIoError::InvalidDatabase(message) => Self::InvalidDatabase(message),
}
}
}
#[derive(Debug, Clone)]
pub struct PagerConfig {
pub cache_size: usize,
pub read_only: bool,
pub verify_checksums: bool,
pub double_write: bool,
}
impl Default for PagerConfig {
fn default() -> Self {
Self {
cache_size: 10_000,
read_only: false,
verify_checksums: true,
double_write: true,
}
}
}
pub trait WalBarrier {
fn flush_until(&self, lsn: u64) -> std::io::Result<()>;
fn needs_page_image(&self, _page_id: u32) -> std::io::Result<bool> {
Ok(false)
}
fn capture_page_image(&self, _page: &Page) -> std::io::Result<()> {
Ok(())
}
fn reserve_pages(&self, _start: u32, _count: u32) -> std::io::Result<()> {
Ok(())
}
fn detach_free_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
Ok(())
}
fn retire_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
Ok(())
}
}
pub struct NoWal;
impl WalBarrier for NoWal {
fn flush_until(&self, _lsn: u64) -> std::io::Result<()> {
Err(std::io::Error::other(
"nonzero page LSN requires a WAL durability barrier",
))
}
}
pub struct Pager<F: VfsFile, W: WalBarrier> {
path: PathBuf,
file: Mutex<F>,
dwb_enabled: bool,
writeback_failed: std::sync::atomic::AtomicBool,
cache: PageCache,
freelist: RwLock<FreeList>,
deferred_reclamation: Mutex<DeferredPageReclamation>,
ownership_mutations: parking_lot::ReentrantMutex<()>,
allocation_reused: std::sync::atomic::AtomicBool,
allocation_fence_pending: std::sync::atomic::AtomicBool,
header: RwLock<DatabaseHeader>,
superblock_publication: Mutex<()>,
config: PagerConfig,
header_dirty: Mutex<bool>,
wal: W,
}
#[derive(Default)]
struct DeferredPageReclamation {
enabled: bool,
pending: std::collections::BTreeSet<u32>,
published: std::collections::BTreeSet<u32>,
}
const SUPERBLOCK_SLOT_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_SLOT_SIZE;
const SUPERBLOCK_ZONE_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_ZONE_SIZE;
const PHASE3_DWB_ZONE_FILE_VERSION: u32 = 0x0001_0001;
pub 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
}
impl<F: VfsFile, W: WalBarrier> Pager<F, W> {
pub fn open_owned(
file: F,
path: PathBuf,
config: PagerConfig,
wal: W,
create: bool,
) -> Result<Self, PagerError> {
if create && file.file_len()? != 0 {
return Err(PagerError::InvalidDatabase(
"new pager file is not empty".into(),
));
}
let pager = Self {
path,
file: Mutex::new(file),
dwb_enabled: config.double_write && !config.read_only,
writeback_failed: std::sync::atomic::AtomicBool::new(false),
cache: PageCache::new(config.cache_size),
freelist: RwLock::new(FreeList::new()),
deferred_reclamation: Mutex::new(DeferredPageReclamation::default()),
ownership_mutations: parking_lot::ReentrantMutex::new(()),
allocation_reused: std::sync::atomic::AtomicBool::new(false),
allocation_fence_pending: std::sync::atomic::AtomicBool::new(false),
header: RwLock::new(DatabaseHeader::default()),
superblock_publication: Mutex::new(()),
config,
header_dirty: Mutex::new(false),
wal,
};
if create {
pager.initialize()?;
} else {
pager.ensure_phase3_dwb_zone_layout()?;
pager.recover_from_dwb()?;
pager.load_header()?;
}
Ok(pager)
}
pub fn wal_barrier(&self) -> &W {
&self.wal
}
pub fn lock_ownership_mutations(&self) -> parking_lot::ReentrantMutexGuard<'_, ()> {
self.ownership_mutations.lock()
}
pub fn fence_interrupted_ownership(&self) {
self.writeback_failed
.store(true, std::sync::atomic::Ordering::Release);
}
#[doc(hidden)]
pub fn with_file<T>(&self, operation: impl FnOnce(&mut F) -> T) -> Result<T, PagerError> {
Ok(operation(&mut *self.file_lock()?))
}
fn initialize(&self) -> Result<(), PagerError> {
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let initial_page_count = FIRST_ALLOCATABLE_PAGE;
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)?;
self.clear_dwb_zone()?;
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<'_, F>, 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()?;
Ok(pager_io::read_superblock_zone(&mut *file)?)
}
#[doc(hidden)]
pub 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 ensure_phase3_dwb_zone_layout(&self) -> Result<(), PagerError> {
let (_, image) = self.newest_superblock()?;
let header = reddb_file::decode_database_header(&image)
.map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
if header.version < PHASE3_DWB_ZONE_FILE_VERSION {
return Err(PagerError::LegacyPagedStore {
path: self.path.clone(),
});
}
Ok(())
}
fn write_superblock_slot(
&self,
slot: &mut [u8; SUPERBLOCK_SLOT_SIZE],
copy_index: usize,
generation: u64,
) -> Result<(), PagerError> {
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let mut file = self.file_lock()?;
self.ensure_writeback_available()?;
Ok(pager_io::write_superblock_slot(
&mut *file, slot, copy_index, generation,
)?)
}
#[doc(hidden)]
pub fn commit_superblock(
&self,
mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
let _publication = self
.superblock_publication
.lock()
.map_err(|_| PagerError::LockPoisoned)?;
self.commit_superblock_locked(mutate)
}
fn commit_superblock_locked(
&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))
}
pub fn fence_file_format_version(&self, version: u32) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
if !(PHASE3_DWB_ZONE_FILE_VERSION..=reddb_file::PAGE_FILE_VERSION).contains(&version) {
return Err(PagerError::InvalidDatabase(
"unsupported file format fence".into(),
));
}
let _publication = self
.superblock_publication
.lock()
.map_err(|_| PagerError::LockPoisoned)?;
if version < self.header_read()?.version {
return Err(PagerError::InvalidDatabase(
"file format downgrade is not supported".into(),
));
}
let result = (|| {
for _ in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
self.commit_superblock_locked(|image| {
reddb_file::set_database_header_version(image, version)
.map_err(|error| PagerError::InvalidDatabase(error.to_string()))
})?;
}
self.header_write()?.version = version;
Ok(())
})();
if result.is_err() {
self.writeback_failed
.store(true, std::sync::atomic::Ordering::Release);
}
result
}
fn load_header(&self) -> Result<(), PagerError> {
let (_, image) = self.newest_superblock()?;
let mut decoded_header = reddb_file::decode_database_header(&image)
.map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
let physical_pages = self.file_lock()?.file_len()?.div_ceil(PAGE_SIZE as u64);
let physical_pages = u32::try_from(physical_pages).map_err(|_| {
PagerError::InvalidDatabase("data file exceeds the page address space".into())
})?;
if decoded_header.page_count < physical_pages {
decoded_header.page_count = physical_pages;
if !self.config.read_only {
*self.header_dirty_lock()? = true;
}
}
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()?;
Ok(pager_io::read_page(
&mut *file,
page_id,
self.config.verify_checksums,
)?)
}
fn ensure_writeback_available(&self) -> Result<(), PagerError> {
if self
.writeback_failed
.load(std::sync::atomic::Ordering::Acquire)
{
return Err(PagerError::Io(std::io::Error::other(
"page writeback failed; reopen the database to recover before writing",
)));
}
Ok(())
}
fn record_ownership(
&self,
operation: impl FnOnce(&W) -> std::io::Result<()>,
) -> Result<(), PagerError> {
if let Err(error) = operation(&self.wal) {
self.writeback_failed
.store(true, std::sync::atomic::Ordering::Release);
return Err(PagerError::Io(error));
}
Ok(())
}
fn write_evicted_dirty_page(&self, page: Page) -> Result<(), PagerError> {
let mut ids = self
.cache
.dirty_page_ids(pager_io::DWB_MAX_PAGES_PER_FRAME.saturating_sub(1));
ids.sort_unstable();
let mut pages = self.cache.flush_dirty_pages(&ids);
pages.push((page.page_id(), page));
pages.sort_unstable_by_key(|(id, _)| *id);
self.write_dirty_page_batch(&pages)
}
fn write_page_raw(&self, page_id: u32, page: &Page) -> Result<(), PagerError> {
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let mut file = self.file_lock()?;
self.ensure_writeback_available()?;
Ok(pager_io::write_page(&mut *file, page_id, page)?)
}
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()) {
self.write_evicted_dirty_page(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 page = pager_io::read_page(&mut *file, page_id, false)?;
drop(file);
if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
self.write_evicted_dirty_page(dirty_page)?;
}
Ok(page)
}
pub fn write_page(&self, page_id: u32, mut page: Page) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
self.capture_page_before_write(page_id)?;
page.update_checksum();
if let Some(dirty_page) = self.cache.insert(page_id, page) {
self.write_evicted_dirty_page(dirty_page)?;
}
self.cache.mark_dirty(page_id);
Ok(())
}
pub fn write_page_no_checksum(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
self.capture_page_before_write(page_id)?;
if let Some(dirty_page) = self.cache.insert(page_id, page) {
self.write_evicted_dirty_page(dirty_page)?;
}
self.cache.mark_dirty(page_id);
Ok(())
}
fn capture_page_before_write(&self, page_id: u32) -> Result<(), PagerError> {
let mut required = false;
self.record_ownership(|wal| {
required = wal.needs_page_image(page_id)?;
Ok(())
})?;
if required {
let previous = self.read_page(page_id)?;
self.record_ownership(|wal| wal.capture_page_image(&previous))?;
}
Ok(())
}
pub fn allocate_page(&self, page_type: PageType) -> Result<Page, PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let page_id = {
let mut freelist = self.freelist_write()?;
self.complete_allocation_fence()?;
if let Some(id) = freelist.next_in_memory_page() {
self.record_ownership(|wal| wal.reserve_pages(id, 1))?;
let id = freelist
.allocate()
.expect("free page checked while allocator is locked");
self.allocation_reused
.store(true, std::sync::atomic::Ordering::Release);
id
} else if freelist.trunk_head() != 0 {
let trunk_id = freelist.trunk_head();
let trunk = self.read_page(trunk_id).map_err(|e| match e {
PagerError::PageNotFound(_) => {
PagerError::InvalidDatabase("Freelist trunk missing".to_string())
}
other => other,
})?;
if trunk.page_id() != trunk_id {
return Err(PagerError::InvalidDatabase(
"misdirected freelist trunk".into(),
));
}
let mut detached = FreeList::new();
detached
.load_from_trunk(&trunk)
.map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {e}")))?;
let mut pages = Vec::new();
let mut seen = std::collections::HashSet::new();
let page_count = self.page_count()?;
while let Some(id) = detached.allocate() {
if id < FIRST_ALLOCATABLE_PAGE || id >= page_count || !seen.insert(id) {
return Err(PagerError::InvalidDatabase(
"invalid freelist trunk page".into(),
));
}
pages.push(id);
}
self.record_ownership(|wal| wal.detach_free_pages(&pages))?;
freelist
.load_from_trunk(&trunk)
.map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {}", e)))?;
let mut header = self.header_write()?;
header.freelist_head = freelist.trunk_head();
*self.header_dirty_lock()? = true;
drop(header);
self.allocation_fence_pending
.store(true, std::sync::atomic::Ordering::Release);
self.complete_allocation_fence()?;
let next = freelist.next_in_memory_page().ok_or_else(|| {
PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
})?;
self.record_ownership(|wal| wal.reserve_pages(next, 1))?;
let id = freelist.allocate().ok_or_else(|| {
PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
})?;
self.allocation_reused
.store(true, std::sync::atomic::Ordering::Release);
id
} else {
let mut header = self.header_write()?;
let id = header.page_count;
let next = id
.checked_add(1)
.ok_or_else(|| PagerError::InvalidDatabase("page count overflow".into()))?;
self.record_ownership(|wal| wal.reserve_pages(id, 1))?;
header.page_count = next;
*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()) {
self.write_evicted_dirty_page(dirty_page)?;
}
self.cache.mark_dirty(page_id);
Ok(page)
}
pub fn reserve_contig_extent(&self, n_pages: u32) -> Result<ExtentId, PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
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;
let end = header.page_count.checked_add(n_pages).ok_or_else(|| {
PagerError::InvalidDatabase("contiguous extent page count overflow".to_string())
})?;
self.record_ownership(|wal| wal.reserve_pages(start, n_pages))?;
header.page_count = end;
*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) {
self.write_evicted_dirty_page(dirty_page)?;
}
self.cache.mark_dirty(page_id);
}
Ok(ExtentId {
start_page,
n_pages,
})
}
pub fn free_page(&self, page_id: u32) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let mut deferred = self
.deferred_reclamation
.lock()
.map_err(|_| PagerError::LockPoisoned)?;
if deferred.enabled {
if page_id < FIRST_ALLOCATABLE_PAGE || page_id >= self.page_count()? {
return Err(PagerError::InvalidDatabase(
"invalid deferred free page".into(),
));
}
self.record_ownership(|wal| wal.retire_pages(&[page_id]))?;
deferred.pending.insert(page_id);
return Ok(());
}
drop(deferred);
self.record_ownership(|wal| wal.retire_pages(&[page_id]))?;
self.cache.remove(page_id);
let mut freelist = self.freelist_write()?;
freelist.free(page_id);
*self.header_dirty_lock()? = true;
Ok(())
}
pub fn enable_deferred_reclamation(&self) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.deferred_reclamation
.lock()
.map_err(|_| PagerError::LockPoisoned)?
.enabled = true;
Ok(())
}
pub fn deferred_reclamation_pages(&self) -> Result<(Vec<u32>, Vec<u32>), PagerError> {
let deferred = self
.deferred_reclamation
.lock()
.map_err(|_| PagerError::LockPoisoned)?;
Ok((
deferred.pending.iter().copied().collect(),
deferred.published.iter().copied().collect(),
))
}
pub fn restore_deferred_reclamation(&self, pages: &[u32]) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
let count = self.page_count()?;
if pages
.iter()
.any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= count)
{
return Err(PagerError::InvalidDatabase(
"invalid persisted deferred free page".into(),
));
}
let mut deferred = self
.deferred_reclamation
.lock()
.map_err(|_| PagerError::LockPoisoned)?;
deferred.pending.extend(pages);
deferred.published.extend(pages);
Ok(())
}
pub fn complete_deferred_publication(
&self,
retained: &[u32],
released: &[u32],
) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
let mut deferred = self
.deferred_reclamation
.lock()
.map_err(|_| PagerError::LockPoisoned)?;
if !released.is_empty() {
self.free_retired_pages(released)?;
}
for id in released {
deferred.pending.remove(id);
}
deferred.published = retained.iter().copied().collect();
Ok(())
}
pub fn record_retirement_intent(&self, pages: &[u32]) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let page_count = self.page_count()?;
let mut seen = std::collections::HashSet::with_capacity(pages.len());
if pages
.iter()
.any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count || !seen.insert(*id))
{
return Err(PagerError::InvalidDatabase(
"invalid retirement intent".into(),
));
}
self.record_ownership(|wal| wal.retire_pages(pages))
}
pub fn free_retired_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let page_count = self.page_count()?;
if pages
.iter()
.any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count)
{
return Err(PagerError::InvalidDatabase(
"invalid retired page range".into(),
));
}
let mut seen = std::collections::HashSet::with_capacity(pages.len());
if pages.iter().any(|id| !seen.insert(*id)) {
return Err(PagerError::InvalidDatabase("duplicate retired page".into()));
}
let mut freelist = self.freelist_write()?;
if freelist.contains_any(&seen) {
return Err(PagerError::InvalidDatabase(
"retired page is already free".into(),
));
}
let mut trunk_id = freelist.trunk_head();
let mut visited = std::collections::HashSet::new();
while trunk_id != 0 && !seen.is_empty() {
if trunk_id < FIRST_ALLOCATABLE_PAGE
|| trunk_id >= page_count
|| !visited.insert(trunk_id)
{
return Err(PagerError::InvalidDatabase(
"invalid freelist trunk chain".into(),
));
}
let trunk = self.read_page(trunk_id)?;
if trunk.page_id() != trunk_id {
return Err(PagerError::InvalidDatabase(
"misdirected freelist trunk".into(),
));
}
let mut loaded = FreeList::new();
let next = loaded
.load_from_trunk(&trunk)
.map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
while let Some(id) = loaded.allocate() {
if id < FIRST_ALLOCATABLE_PAGE || id >= page_count {
return Err(PagerError::InvalidDatabase(
"invalid free page range".into(),
));
}
if seen.contains(&id) {
return Err(PagerError::InvalidDatabase(
"retired page is already free".into(),
));
}
}
trunk_id = next.unwrap_or(0);
}
self.record_ownership(|wal| wal.retire_pages(pages))?;
for id in pages {
self.cache.remove(*id);
}
freelist.free_batch(pages);
*self.header_dirty_lock()? = true;
Ok(())
}
fn pages_not_in_freelist(
&self,
pages: &[u32],
freelist: &FreeList,
page_count: u32,
) -> Result<std::collections::HashSet<u32>, PagerError> {
let mut remaining = std::collections::HashSet::with_capacity(pages.len());
if pages
.iter()
.any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count || !remaining.insert(*id))
{
return Err(PagerError::InvalidDatabase(
"invalid recovery page set".into(),
));
}
let requested = remaining.clone();
freelist
.remove_existing_from(&requested, &mut remaining)
.map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
let mut trunk_id = freelist.trunk_head();
let mut visited = std::collections::HashSet::new();
while trunk_id != 0 {
if trunk_id < FIRST_ALLOCATABLE_PAGE
|| trunk_id >= page_count
|| !visited.insert(trunk_id)
{
return Err(PagerError::InvalidDatabase(
"invalid freelist trunk chain".into(),
));
}
let trunk = self.read_page(trunk_id)?;
if trunk.page_id() != trunk_id {
return Err(PagerError::InvalidDatabase(
"misdirected freelist trunk".into(),
));
}
let mut loaded = FreeList::new();
let next = loaded
.load_from_trunk(&trunk)
.map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
while let Some(id) = loaded.allocate() {
if id < FIRST_ALLOCATABLE_PAGE || id >= page_count {
return Err(PagerError::InvalidDatabase(
"invalid free page range".into(),
));
}
if requested.contains(&id) && !remaining.remove(&id) {
return Err(PagerError::InvalidDatabase(
"duplicate recovery page".into(),
));
}
}
trunk_id = next.unwrap_or(0);
}
Ok(remaining)
}
pub fn recover_unowned_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
if self.config.read_only {
return Err(PagerError::ReadOnly);
}
let page_count = self.page_count()?;
let mut freelist = self.freelist_write()?;
let remaining = self.pages_not_in_freelist(pages, &freelist, page_count)?;
if !remaining.is_empty() {
let mut missing: Vec<_> = remaining.into_iter().collect();
missing.sort_unstable();
for id in &missing {
self.cache.remove(*id);
}
freelist.free_batch(&missing);
*self.header_dirty_lock()? = true;
}
Ok(())
}
pub fn validate_owned_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
let page_count = self.page_count()?;
let freelist = self.freelist_write()?;
let absent = self.pages_not_in_freelist(pages, &freelist, page_count)?;
if absent.len() != pages.len() {
return Err(PagerError::InvalidDatabase(
"owned recovery page is free".into(),
));
}
Ok(())
}
pub fn sync_recovered_allocator(&self) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
let result = (|| {
self.sync()?;
let _publication = self
.superblock_publication
.lock()
.map_err(|_| PagerError::LockPoisoned)?;
let header = self.header_read()?.clone();
for _ in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
self.commit_superblock_locked(|image| {
reddb_file::encode_database_header(image, &header)
.map_err(|error| PagerError::InvalidDatabase(error.to_string()))
})?;
}
Ok(())
})();
if result.is_err() {
self.writeback_failed
.store(true, std::sync::atomic::Ordering::Release);
}
result
}
pub fn free_page_count(&self) -> Result<u32, PagerError> {
Ok(self.freelist_write()?.total_free())
}
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> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
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 flush(&self) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
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()
}
};
let mut dirty_pages = Vec::new();
for trunk in trunks {
let page_id = trunk.page_id();
if let Some(evicted) = self.cache.insert(page_id, trunk) {
dirty_pages.push((evicted.page_id(), evicted));
}
self.cache.mark_dirty(page_id);
}
const FLUSH_BATCH_PAGES: usize = 256;
for batch in dirty_pages.chunks(FLUSH_BATCH_PAGES) {
self.write_dirty_page_batch(batch)?;
}
drop(dirty_pages);
let mut page_ids = self.cache.dirty_page_ids(usize::MAX);
page_ids.sort_unstable();
for ids in page_ids.chunks(FLUSH_BATCH_PAGES) {
let pages = self.cache.flush_dirty_pages(ids);
self.write_dirty_page_batch(&pages)?;
}
if *self.header_dirty_lock()? {
self.write_header()?;
}
Ok(())
}
fn write_dirty_page_batch(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
if pages.is_empty() {
return Ok(());
}
let result = (|| {
let max_lsn = pages
.iter()
.filter_map(|(_, page)| page.header().ok().map(|header| header.lsn))
.max()
.unwrap_or(0);
if max_lsn > 0 {
self.wal.flush_until(max_lsn)?;
}
self.write_pages_through_dwb(pages)
})();
if result.is_err() {
self.writeback_failed
.store(true, std::sync::atomic::Ordering::Release);
for (page_id, _) in pages {
self.cache.mark_dirty(*page_id);
}
}
result
}
pub fn sync(&self) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
self.flush()?;
if self
.allocation_reused
.swap(false, std::sync::atomic::Ordering::AcqRel)
{
if let Err(error) = self.write_header_and_sync() {
self.allocation_reused
.store(true, std::sync::atomic::Ordering::Release);
return Err(error);
}
}
let file = self.file_lock()?;
file.sync_all()?;
Ok(())
}
pub fn cache_stats(&self) -> super::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> {
let _ownership = self.lock_ownership_mutations();
self.ensure_writeback_available()?;
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.file_len()?)
}
fn clear_dwb_zone(&self) -> Result<(), PagerError> {
let mut file = self.file_lock()?;
Ok(pager_io::clear_dwb_zone(&mut *file)?)
}
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 self.config.read_only {
return Err(PagerError::ReadOnly);
}
let mut file = self.file_lock()?;
self.ensure_writeback_available()?;
let result = pager_io::write_pages(&mut *file, pages, self.dwb_enabled);
if result.is_err() {
self.writeback_failed
.store(true, std::sync::atomic::Ordering::Release);
}
Ok(result?)
}
fn recover_from_dwb(&self) -> Result<(), PagerError> {
let mut file = self.file_lock()?;
Ok(pager_io::recover_from_dwb(&mut *file)?)
}
fn complete_allocation_fence(&self) -> Result<(), PagerError> {
if self
.allocation_fence_pending
.load(std::sync::atomic::Ordering::Acquire)
{
self.write_header_and_sync()?;
self.write_header_and_sync()?;
self.allocation_fence_pending
.store(false, std::sync::atomic::Ordering::Release);
}
Ok(())
}
pub fn write_header_and_sync(&self) -> Result<(), PagerError> {
let _ownership = self.lock_ownership_mutations();
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 _ownership = self.lock_ownership_mutations();
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 _ownership = self.lock_ownership_mutations();
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()
}
}