reddb_server/storage/engine/pager/

impl.rs

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;

/// The superblock zone occupies the head of page 0 (ADR 0038 §2 phase 1).
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;
pub(super) const PHASE3_DWB_ZONE_FILE_VERSION: u32 = reddb_file::PAGE_FILE_VERSION;

/// The newest valid superblock copy together with its slot image.
type NewestSuperblock = (
    reddb_file::PagedSuperblockSelection,
    [u8; SUPERBLOCK_SLOT_SIZE],
);

/// Take the first slot's worth of a freshly-built page-0 image. The database
/// header and the encryption header both live well inside this prefix.
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() };
    // `statfs.f_type` is i32 on 32-bit glibc targets (armv7) and i64 on
    // 64-bit ones — widen explicitly so both compile.
    #[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 {
    /// Open or create a database file
    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(),
            ));
        }

        // ADR 0038 §4 phase 1 is a clean break: a store still carrying the
        // retired `rdb-hdr`/`rdb-meta` sidecars is never read by the engine.
        // The offline migration tool is the only path forward.
        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(),
            ));
        }

        // Open file
        // Note: create requires write access, so disable it for read-only mode
        let file = OpenOptions::new()
            .read(true)
            .write(!config.read_only)
            .create(config.create && !config.read_only)
            .open(&path)?;

        // gh-892: diagnostic probe of the filesystem block size. If the
        // compile-time 16 KiB PAGE_SIZE is not a multiple of the FS block
        // size, page writes straddle FS blocks and incur read-modify-write
        // amplification. Pure diagnostic — emitted once per open, never
        // mutates the page size. `blksize()` is the fstat `st_blksize`.
        #[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."
                    );
                }
            }
        }

        // Acquire file lock (exclusive for writes, shared for read-only)
        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,
            }
        };

        // ADR 0038 phase 3: the DWB sidecar is retired. When DWB is active,
        // the recovery frame is written to a reserved zone inside the data
        // file and fsynced before in-place page writes. A pre-existing sidecar
        // marks a legacy store that must go through the offline migration tool.
        //
        // gh-895: an explicit `double_write = false` request is honored only
        // when the already-open data file is proven to live on a filesystem
        // with atomic CoW page writes. Unknown and non-CoW filesystems fail
        // closed by keeping the in-file DWB zone enabled.
        let fold_dwb = crate::physical::fold_dwb_into_wal_enabled();
        if exists {
            if let Some(sidecar) =
                reddb_file::layout::retired::first_present_phase3_dwb_sidecar(&path)
            {
                return Err(PagerError::LegacySidecarStore { sidecar });
            }
        }
        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 in-file double-write buffer enabled"
                );
                config.double_write = true;
            }
        }
        let dwb_enabled = config.double_write && !config.read_only && !fold_dwb;

        let mut pager = Self {
            path,
            file: Mutex::new(file),
            _lock_file: lock_file,
            dwb_enabled,
            cache: PageCache::new(config.cache_size),
            freelist: RwLock::new(FreeList::new()),
            deferred_reclamation: Mutex::new(DeferredPageReclamation::default()),
            allocation_reused: std::sync::atomic::AtomicBool::new(false),
            allocation_fence_pending: std::sync::atomic::AtomicBool::new(false),
            header: RwLock::new(DatabaseHeader::default()),
            config,
            header_dirty: Mutex::new(false),
            wal: RwLock::new(None),
            encryption: None,
        };

        if exists {
            // Recover from the in-file double-write zone before loading the
            // header or allowing WAL replay to read pages.
            pager.ensure_phase3_dwb_zone_layout()?;
            pager.recover_from_dwb()?;
            // Load existing database (with header shadow fallback)
            pager.load_header()?;
            pager.bind_encryption_for_existing()?;
        } else {
            // Initialize new database
            pager.initialize()?;
            pager.bind_encryption_for_new()?;
        }

        Ok(pager)
    }

    /// gh-892 diagnostic predicate: returns `true` when the database
    /// `page_size` is **not** an integer multiple of the filesystem's
    /// reported block size (`st_blksize`), i.e. `page_size % fs_block_size
    /// != 0`. A `fs_block_size` of `0` (probe unavailable / unknown) is
    /// treated as aligned so a missing probe never produces a warning.
    /// Pure function with no I/O so the warn decision is unit-testable.
    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
    }

    /// Inspect page 0 for the `RDBE` encryption marker, then resolve
    /// the (key, marker) matrix:
    ///
    /// | Marker | Key supplied | Result                              |
    /// |--------|--------------|-------------------------------------|
    /// | yes    | yes          | Bind encryptor; validate key        |
    /// | yes    | no           | `EncryptionRequired` (fail closed)  |
    /// | no     | yes          | `PlainDatabaseRefusesKey`           |
    /// | no     | no           | Plain pager — no binding needed     |
    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(()),
        }
    }

    /// New DB: if a key is configured, write the marker + header to
    /// page 0 so subsequent opens detect encryption.
    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);

        // Stamp the marker + header into the superblock zone if `initialize()`
        // has already published it.
        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(())
    }

    /// Open with default configuration
    pub fn open_default<P: AsRef<Path>>(path: P) -> Result<Self, PagerError> {
        Self::open(path, PagerConfig::default())
    }

    /// Initialize a new database
    fn initialize(&self) -> Result<(), PagerError> {
        if self.config.read_only {
            return Err(PagerError::ReadOnly);
        }

        // Create header page. Page ids 1 and 2 are reserved for fixed
        // metadata/vault pages. ADR 0038 phase 3 reserves the following DWB
        // zone in-file so normal B-tree allocation starts after it.
        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;

        // Reserve page 0 (the superblock zone) plus the reserved pages so any
        // scan over 0..page_count can read every allocated page in a brand-new
        // database. The zone's own bytes are published below, one slot at a
        // time; this first whole-page write only sizes the file.
        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()?;

        // Publish both superblock copies so the ping-pong invariant — one
        // valid copy always survives a crash — holds from the very first
        // update onward, not just from the second.
        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)?;

        // Sync to disk
        self.sync()?;

        Ok(())
    }

    /// Acquire a write lock on the header RwLock, mapping poison errors.
    fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
        self.header.write().map_err(|_| PagerError::LockPoisoned)
    }

    /// Acquire a read lock on the header RwLock, mapping poison errors.
    fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
        self.header.read().map_err(|_| PagerError::LockPoisoned)
    }

    /// Acquire a write lock on the freelist RwLock, mapping poison errors.
    fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
        self.freelist.write().map_err(|_| PagerError::LockPoisoned)
    }

    /// Acquire a lock on the file Mutex, mapping poison errors.
    fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, File>, PagerError> {
        self.file.lock().map_err(|_| PagerError::LockPoisoned)
    }

    /// Acquire a lock on the header_dirty Mutex, mapping poison errors.
    fn header_dirty_lock(&self) -> Result<std::sync::MutexGuard<'_, bool>, PagerError> {
        self.header_dirty
            .lock()
            .map_err(|_| PagerError::LockPoisoned)
    }

    /// Read the whole superblock zone (both ping-pong copies) from the head of
    /// the file. A file shorter than the zone yields zero-filled tail bytes,
    /// which decode as "slot never written".
    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)
    }

    /// The newest valid superblock copy, with its slot image.
    ///
    /// Both copies invalid is the one unrecoverable superblock state (ADR 0074
    /// §2): the store does not open, and the error names the zone and points
    /// at salvage rather than panicking or silently re-initialising.
    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(())
    }

    /// Seal `slot` for `copy_index`/`generation` and write exactly that slot,
    /// then fsync. The caller guarantees `copy_index` is the stale copy.
    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(())
    }

    /// Publish `mutate`'s edit of the newest slot image into the *stale* copy.
    ///
    /// Reading the newest image first means every byte the pager does not
    /// explicitly rewrite — most importantly the encryption marker and header —
    /// carries forward untouched.
    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))
    }

    /// Load the database header from the newest valid superblock copy.
    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()))?;
        // Cache eviction (or DWB recovery) can persist pages before the next
        // superblock publication. After a crash the allocation counter may
        // therefore lag behind live pages already in the file. Reserve the
        // entire physical extent before allocating; never overwrite that tail.
        // A partial last page is reserved too, without treating it as verified.
        let physical_pages = self
            .file_lock()?
            .metadata()?
            .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;
        }

        // Initialize freelist
        {
            let mut freelist = self.freelist_write()?;
            *freelist = FreeList::from_header(freelist_head, 0);
        }

        Ok(())
    }

    /// Publish the in-memory database header as a new superblock generation.
    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(())
    }

    /// Read a page from disk (bypassing cache)
    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);

        // Page 0 is the superblock zone, not a checksummed page: its two slots
        // are sealed independently, so no whole-page checksum can hold across
        // a ping-pong write. Slot CRCs are its integrity contract.
        if self.config.verify_checksums && page_id != 0 {
            page.verify_checksum()?;
        }

        Ok(page)
    }

    /// Write a page to disk (bypassing cache)
    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(())
    }

    /// Read a page (cache-aware)
    pub fn read_page(&self, page_id: u32) -> Result<Page, PagerError> {
        // Check cache first
        if let Some(page) = self.cache.get(page_id) {
            return Ok(page);
        }

        // Cache miss - read from disk
        let page = self.read_page_raw(page_id)?;

        // Add to cache
        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
            // Evicted page was dirty, need to write it back
            let evicted_id = dirty_page.page_id();
            self.write_page_raw(evicted_id, &dirty_page)?;
        }

        Ok(page)
    }

    /// Read a page without verifying checksum (for encrypted pages)
    ///
    /// Use this when the page content has its own integrity protection
    /// (e.g., AES-GCM authentication tag for encrypted pages).
    pub fn read_page_no_checksum(&self, page_id: u32) -> Result<Page, PagerError> {
        // Check cache first
        if let Some(page) = self.cache.get(page_id) {
            return Ok(page);
        }

        // Cache miss - read from disk (skip checksum verification)
        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);

        // Add to cache (no checksum verification)
        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
            // Evicted page was dirty, need to write it back
            let evicted_id = dirty_page.page_id();
            self.write_page_raw(evicted_id, &dirty_page)?;
        }

        Ok(page)
    }

    /// Write a page (cache-aware)
    pub fn write_page(&self, page_id: u32, mut page: Page) -> Result<(), PagerError> {
        if self.config.read_only {
            return Err(PagerError::ReadOnly);
        }

        // Update checksum
        page.update_checksum();

        // Add to cache and mark dirty. If the cache had to evict a
        // dirty entry, write it through immediately — the evicted
        // page will never be flushed otherwise (same bug fixed in
        // `allocate_page`).
        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(())
    }

    /// Read a page through the configured encryptor if any. Page 0
    /// is always returned plaintext (it carries the encryption marker
    /// + header). Callers that want raw cipher bytes can use
    ///   `read_page_no_checksum` directly.
    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))
    }

    /// Write a page through the configured encryptor if any. Page 0
    /// bypasses encryption and goes through the normal checksummed
    /// path. Encrypted pages skip the checksum update because
    /// AES-GCM's authentication tag is the integrity guarantee.
    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);
        }
        // Canonical per-page envelope overhead (nonce + GCM tag),
        // owned by reddb-io-crypto (#1053, ADR 0054).
        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)
    }

    /// Write a page without updating checksum (for encrypted pages)
    ///
    /// Use this when the page content has its own integrity protection
    /// (e.g., AES-GCM authentication tag for encrypted pages).
    pub fn write_page_no_checksum(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
        if self.config.read_only {
            return Err(PagerError::ReadOnly);
        }

        // Add to cache and mark dirty (no checksum update). Same
        // eviction-write-through guard as `write_page`.
        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(())
    }

    /// Allocate a new page
    pub fn allocate_page(&self, page_type: PageType) -> Result<Page, PagerError> {
        if self.config.read_only {
            return Err(PagerError::ReadOnly);
        }

        // Try to get from freelist first
        let page_id = {
            let mut freelist = self.freelist_write()?;
            // A previous fence can fail after loading the trunk into memory.
            // Retry it before the fast path can consume any of those pages.
            self.complete_allocation_fence()?;
            if let Some(id) = freelist.allocate() {
                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,
                })?;

                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);
                // The consumed trunk itself becomes reusable. Forget its old
                // allocator role durably before cache eviction can overwrite it.
                // Keep the allocator locked so another allocator cannot race
                // ahead of this fence. A failed fence leaves allocation stopped.
                self.allocation_fence_pending
                    .store(true, std::sync::atomic::Ordering::Release);
                self.complete_allocation_fence()?;
                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 {
                // No free pages, extend file
                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);

        // Write to cache. The evicted page (if any) is dirty by
        // definition — `cache.insert` only returns `Some` when it
        // had to evict a dirty entry to make room. The previous
        // version dropped that return value, which silently lost
        // writes whenever a freshly-allocated page caused an LRU
        // eviction. This shows up under heavy ingest as
        // "B-tree insert error: Pager error: I/O error: failed to
        // fill whole buffer" later, when something tries to read
        // back the never-flushed page. Mirror `read_page`'s
        // handling: write the evicted page through immediately so
        // the on-disk image stays consistent.
        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)
    }

    /// Reserve a contiguous extent of vector pages.
    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,
        })
    }

    /// Free a page (return to freelist)
    pub fn free_page(&self, page_id: u32) -> Result<(), PagerError> {
        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(),
                ));
            }
            // Keep dirty forwarding/empty nodes until publication. An
            // incidental flush must leave the previous root traversable.
            deferred.pending.insert(page_id);
            return Ok(());
        }
        drop(deferred);

        // Remove from cache
        self.cache.remove(page_id);

        // Add to freelist
        let mut freelist = self.freelist_write()?;
        freelist.free(page_id);

        *self.header_dirty_lock()? = true;

        Ok(())
    }

    pub(crate) fn enable_deferred_reclamation(&self) -> Result<(), PagerError> {
        self.deferred_reclamation
            .lock()
            .map_err(|_| PagerError::LockPoisoned)?
            .enabled = true;
        Ok(())
    }

    pub(crate) 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(crate) fn restore_deferred_reclamation(&self, pages: &[u32]) -> Result<(), PagerError> {
        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(())
    }

    /// The caller has durably published a catalog that retains `retained`
    /// and no longer names `released`, after waiting for old tree readers.
    pub(crate) fn complete_deferred_publication(
        &self,
        retained: &[u32],
        released: &[u32],
    ) -> Result<(), PagerError> {
        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(())
    }

    /// Return a fully validated, unpublished tree generation in one allocator
    /// mutation. Call only after its removal from the manifest is durable.
    pub(crate) fn free_retired_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
        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 freelist = self.freelist_write()?;
        for id in pages {
            self.cache.remove(*id);
        }
        freelist.free_batch(pages);
        *self.header_dirty_lock()? = true;
        Ok(())
    }

    /// Get database header
    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(())
    }

    /// Get page count
    pub fn page_count(&self) -> Result<u32, PagerError> {
        Ok(self.header_read()?.page_count)
    }

    /// Attach a WAL writer to enforce WAL-first flush ordering.
    ///
    /// After this call, [`Pager::flush`] computes the maximum
    /// `header.lsn` over all dirty pages and calls
    /// `WalWriter::flush_until(max_lsn)` before any page is written
    /// to the data file. This is the postgres rule: a page on disk
    /// implies its WAL record is already durable on disk.
    ///
    /// Existing call sites that construct a Pager without a WAL
    /// keep their previous behaviour (no LSN check) — wiring is
    /// strictly opt-in.
    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);
    }

    /// Detach the WAL writer (test / shutdown path).
    pub fn clear_wal_writer(&self) {
        let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
        *slot = None;
    }

    /// Has a WAL writer been attached?
    pub fn has_wal_writer(&self) -> bool {
        self.wal.read().map(|s| s.is_some()).unwrap_or(false)
    }

    /// Flush all dirty pages to disk
    pub fn flush(&self) -> Result<(), PagerError> {
        if self.config.read_only {
            return Ok(());
        }

        // Persist freelist to trunk pages when dirty
        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) {
                // Trunk publication can evict dirty data. Keep it in the same
                // WAL-first, double-write-protected batch as resident pages.
                dirty_pages.push((evicted.page_id(), evicted));
            }
            self.cache.mark_dirty(page_id);
        }

        // Flush dirty pages from cache (through DWB if enabled)
        dirty_pages.extend(self.cache.flush_dirty());
        if !dirty_pages.is_empty() {
            // WAL-FIRST: ensure every WAL record describing a dirty
            // page is durable BEFORE the page itself reaches disk.
            // Pages with `lsn == 0` are exempt (freelist trunks, header
            // shadow pages, anything not produced by a WAL append).
            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 the read lock before taking the WAL
                        // mutex so an unrelated reader cannot block
                        // the flush path.
                        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)?;
        }

        // Write header if dirty
        if *self.header_dirty_lock()? {
            self.write_header()?;
        }

        Ok(())
    }

    /// Sync file to disk (fsync)
    pub fn sync(&self) -> Result<(), PagerError> {
        self.flush()?;

        if self
            .allocation_reused
            .swap(false, std::sync::atomic::Ordering::AcqRel)
        {
            // flush published the current allocator in one slot. Mirror that
            // ownership before callers publish roots that use reclaimed pages;
            // recovery through the sibling slot must never allocate them again.
            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(())
    }

    /// Get cache statistics
    pub fn cache_stats(&self) -> crate::storage::engine::page_cache::CacheStats {
        self.cache.stats()
    }

    /// Pages currently resident in the page cache. Slots are fixed size, so
    /// this times the page size is the cache's memory footprint (ADR 0073 §2).
    pub fn cache_len(&self) -> usize {
        self.cache.len()
    }

    /// Preallocated page-cache slots — the sizing this process derived from
    /// its page-cache budget share.
    pub fn cache_capacity(&self) -> usize {
        self.cache.capacity()
    }

    /// Count dirty pages currently in the page cache.
    pub fn dirty_page_count(&self) -> usize {
        self.cache.dirty_count()
    }

    /// Estimated fraction of the page cache holding dirty pages.
    /// Returned in `[0, 1]`. Used by the background writer to
    /// decide when to kick in aggressive flushing.
    pub fn dirty_fraction(&self) -> f64 {
        let capacity = self.cache.capacity().max(1) as f64;
        self.cache.dirty_count() as f64 / capacity
    }

    /// Flush up to `max` dirty pages from the cache. Returns the
    /// number actually written. Background-writer entry point —
    /// reuses the same persistence path as `flush()` but bounded.
    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();
        // WAL-first: every cached dirty page carries an LSN that the
        // WAL must have already persisted. The full `flush()` path
        // enforces this with `wal.flush(max_lsn)`; here we simply
        // write through the pager — safe because callers only reach
        // this path via the bgwriter, which runs asynchronously
        // alongside normal commits that already respect WAL-first.
        for (page_id, page) in dirty_pages {
            self.write_page(page_id, page)?;
        }
        Ok(count)
    }

    /// Get database file path
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Check if database is read-only
    pub fn is_read_only(&self) -> bool {
        self.config.read_only
    }

    /// Get file size in bytes
    pub fn file_size(&self) -> Result<u64, PagerError> {
        let file = self.file_lock()?;
        Ok(file.metadata()?.len())
    }

    /// Issue an OS-level read-ahead hint for `page_id`.
    ///
    /// A6 prefetch wire: called from `BTreeCursor::next` when the
    /// cursor passes 50% of the current leaf, so the kernel fetches
    /// the next leaf page while CPU processes the remaining half of
    /// the current one. Failures are silent — a missed prefetch is a
    /// performance miss, never a correctness bug.
    pub fn prefetch_hint(&self, page_id: u32) {
        if let Ok(file) = self.file_lock() {
            let _ = super::super::prefetch::prefetch_page(&file, page_id as u64, PAGE_SIZE as u32);
        }
    }

    // ── Corruption defense helpers ──────────────────────────────────

    /// Path for the double-write buffer file
    fn dwb_path(db_path: &Path) -> PathBuf {
        reddb_file::layout::pager_dwb_shadow_path(db_path)
    }

    fn dwb_zone_offset() -> u64 {
        DWB_ZONE_START_PAGE as u64 * PAGE_SIZE as u64
    }

    fn write_dwb_zone_frame(&self, buf: &[u8]) -> Result<(), PagerError> {
        if buf.len() > DWB_ZONE_BYTES {
            return Err(PagerError::InvalidDatabase(
                "DWB frame exceeds the reserved in-file zone".to_string(),
            ));
        }
        let mut file = self.file_lock()?;
        file.seek(SeekFrom::Start(Self::dwb_zone_offset()))?;
        file.write_all(buf)?;
        file.sync_all()?;
        Ok(())
    }

    /// Clear the in-file DWB zone with valid reserved pages.
    fn clear_dwb_zone(&self) -> Result<(), PagerError> {
        let mut file = self.file_lock()?;
        for page_id in DWB_ZONE_START_PAGE..FIRST_ALLOCATABLE_PAGE {
            let mut page = Page::new(PageType::Header, page_id);
            page.update_checksum();
            file.seek(SeekFrom::Start(page_id as u64 * PAGE_SIZE as u64))?;
            file.write_all(page.as_bytes())?;
        }
        file.sync_all()?;
        Ok(())
    }

    fn dwb_zone_is_clear(buf: &[u8]) -> bool {
        if buf.len() != DWB_ZONE_BYTES {
            return false;
        }
        for (idx, chunk) in buf.chunks_exact(PAGE_SIZE).enumerate() {
            let Ok(page) = Page::from_slice(chunk) else {
                return false;
            };
            if page.verify_checksum().is_err() {
                return false;
            }
            if page.page_id() != DWB_ZONE_START_PAGE + idx as u32 {
                return false;
            }
            if page.page_type().ok() != Some(PageType::Header) {
                return false;
            }
        }
        true
    }

    /// Read the internal manifest page (page 1), the zone the superblock roots.
    ///
    /// ADR 0074 §2: manifest corruption fails the open didactically, naming the
    /// zone — it never falls back to a sidecar and never returns garbage.
    pub fn read_manifest_page(&self) -> Result<Page, PagerError> {
        self.read_page(1)
            .map_err(|_| PagerError::ManifestZoneCorrupt(self.path.clone()))
    }

    /// Write pages through the double-write buffer for torn page protection.
    ///
    /// 1. Write pages to the in-file DWB zone with a header (magic + count + checksum)
    /// 2. fsync the DWB zone
    /// 3. Write all pages to their final locations in the .rdb file
    /// 4. Clear the DWB zone (marks as consumed)
    fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
        if self.dwb_enabled {
            for chunk in pages.chunks(DWB_MAX_PAGES_PER_FRAME.max(1)) {
                let buf = reddb_file::encode_paged_dwb_frame(
                    chunk
                        .iter()
                        .map(|(page_id, page)| (*page_id, page.as_bytes())),
                );

                // Write DWB zone and fsync
                self.write_dwb_zone_frame(&buf)?;

                // Now write pages to their final locations
                for (page_id, page) in chunk {
                    self.write_page_raw(*page_id, page)?;
                }
                {
                    let file = self.file_lock()?;
                    file.sync_all()?;
                }

                // Reset the zone to its inactive page images.
                self.clear_dwb_zone()?;
            }

            Ok(())
        } else {
            // DWB disabled — write directly
            for (page_id, page) in pages {
                self.write_page_raw(*page_id, page)?;
            }
            Ok(())
        }
    }

    /// Recover from double-write buffer after a crash.
    ///
    /// If the DWB zone contains valid pages, they were written before the crash
    /// interrupted writing to their home locations. Re-apply them.
    fn recover_from_dwb(&self) -> Result<(), PagerError> {
        let len = {
            let file = self.file_lock()?;
            file.metadata()?.len()
        };
        let zone_end = Self::dwb_zone_offset() + DWB_ZONE_BYTES as u64;
        if len < zone_end {
            return Ok(());
        }

        let mut buf = vec![0u8; DWB_ZONE_BYTES];
        {
            let mut file = self.file_lock()?;
            file.seek(SeekFrom::Start(Self::dwb_zone_offset()))?;
            file.read_exact(&mut buf)?;
        }
        if buf.get(0..4) != Some(reddb_file::DWB_MAGIC.as_slice()) {
            if Self::dwb_zone_is_clear(&buf) {
                return Ok(());
            }
            self.clear_dwb_zone()?;
            return Ok(());
        }

        let entries = match reddb_file::decode_paged_dwb_frame(&buf) {
            Ok(entries) => entries,
            Err(_) => {
                self.clear_dwb_zone()?;
                return Ok(());
            }
        };

        // DWB is valid — re-apply pages to main file
        for entry in entries {
            let page = Page::from_bytes(entry.page);
            self.write_page_raw(entry.page_id, &page)?;
        }

        // Sync and clean up
        {
            let file = self.file_lock()?;
            file.sync_all()?;
        }

        self.clear_dwb_zone()
    }

    /// Finish a consumed trunk's durable ownership change, retaining errors
    /// across allocation attempts until both copies have been synchronized.
    /// Called with the freelist locked when consuming or retrying a trunk.
    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(())
    }

    /// Write header and sync to disk (public for checkpointer).
    pub fn write_header_and_sync(&self) -> Result<(), PagerError> {
        self.write_header()?;
        let file = self.file_lock()?;
        file.sync_all()?;
        Ok(())
    }

    /// Set the checkpoint_in_progress flag in the header.
    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()
    }

    /// Update the checkpoint LSN and clear the in-progress flag.
    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
    }

    /// Overwrite one superblock slot in place, leaving the sibling untouched.
    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");
    }

    fn stamp_pre_phase3_layout(path: &Path, page_count: u32) {
        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();
            let generation = reddb_file::paged_superblock_slot_generation(&slot, copy_index)
                .expect("fresh store should publish both slots");
            reddb_file::set_database_header_version(&mut slot, PHASE3_DWB_ZONE_FILE_VERSION - 1)
                .expect("set_database_header_version() should succeed");
            reddb_file::set_database_header_page_count(&mut slot, page_count)
                .expect("set_database_header_page_count() should succeed");
            reddb_file::seal_paged_superblock_slot(&mut slot, copy_index, generation)
                .expect("seal_paged_superblock_slot() should succeed");
            poke_slot(path, copy_index, &slot);
        }
    }

    #[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);

        // A *valid* superblock slot — correct magic, copy index and CRC — whose
        // database header names a version we cannot read. The slot is intact;
        // the header is from the future.
        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() {
        // The ping-pong invariant must hold from the very first update, which
        // means creation cannot leave one copy blank.
        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);

        // The copy that was newest before the update is still readable — this
        // is the byte-level statement of "a crash mid-write always leaves one
        // valid copy".
        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 failed_trunk_fence_is_completed_before_retrying_allocation() {
        let path = temp_db_path("retry-trunk-fence");
        let pager = Pager::open_default(&path).expect("pager");
        let mut pages = Vec::new();
        for _ in 0..8 {
            pages.push(pager.allocate_page(PageType::BTreeLeaf).unwrap().page_id());
        }
        pager.sync().unwrap();
        for id in pages {
            pager.free_page(id).unwrap();
        }
        pager.sync().unwrap();
        let trunk = pager.header().unwrap().freelist_head;
        assert_ne!(trunk, 0);

        // A real write error after the trunk has been read must not turn a
        // subsequent allocation into an unfenced fast-path reuse.
        let writable = {
            let mut file = pager.file_lock().unwrap();
            std::mem::replace(&mut *file, File::open(&path).unwrap())
        };
        assert!(pager.allocate_page(PageType::BTreeLeaf).is_err());
        *pager.file_lock().unwrap() = writable;
        pager.allocate_page(PageType::BTreeLeaf).unwrap();

        let zone = read_zone(&path);
        for copy in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
            let start = copy * SUPERBLOCK_SLOT_SIZE;
            let header =
                reddb_file::decode_database_header(&zone[start..start + SUPERBLOCK_SLOT_SIZE])
                    .unwrap();
            assert_ne!(
                header.freelist_head, trunk,
                "copy {copy} still owns consumed trunk"
            );
        }
        drop(pager);
        cleanup(&path);
    }

    #[test]
    fn sibling_superblock_recovery_does_not_reallocate_live_reused_pages() {
        let path = temp_db_path("reused-allocator-pair");
        let pager = Pager::open_default(&path).expect("pager");
        let mut pages = Vec::new();
        for _ in 0..32 {
            pages.push(
                pager
                    .allocate_page(PageType::BTreeLeaf)
                    .expect("allocate")
                    .page_id(),
            );
        }
        pager.sync().expect("seed");
        for id in pages {
            pager.free_page(id).expect("free");
        }
        pager.sync().expect("publish free list");
        let mut live = std::collections::HashSet::new();
        for _ in 0..8 {
            let page = pager.allocate_page(PageType::BTreeLeaf).expect("reuse");
            live.insert(page.page_id());
            pager
                .write_page(page.page_id(), page)
                .expect("write valid page");
        }
        pager.sync().expect("publish reused ownership");
        drop(pager);
        let selection = reddb_file::select_paged_superblock(&read_zone(&path)).expect("superblock");
        poke_slot(&path, selection.copy_index, &[0; SUPERBLOCK_SLOT_SIZE]);
        let recovered = Pager::open_default(&path).expect("sibling recovery");
        for _ in 0..64 {
            let id = recovered
                .allocate_page(PageType::BTreeLeaf)
                .expect("allocate after recovery")
                .page_id();
            assert!(
                !live.contains(&id),
                "sibling allocator still considers live page {id} free"
            );
        }
        drop(recovered);
        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");
        // Tear the newest copy: a torn write leaves its CRC broken.
        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 pre_phase3_large_store_without_sidecar_routes_to_migration() {
        let path = temp_db_path("pre-phase3-large");
        {
            let _ = Pager::open_default(&path).expect("open_default() should succeed");
        }
        stamp_pre_phase3_layout(&path, FIRST_ALLOCATABLE_PAGE);

        let err = match Pager::open_default(&path) {
            Ok(_) => panic!("pre-phase-3 stores need migration first"),
            Err(err) => err,
        };
        let rendered = err.to_string();
        assert!(matches!(err, PagerError::LegacyPagedStore { .. }));
        assert!(rendered.contains("migration tool"), "{rendered}");

        cleanup(&path);
    }

    #[test]
    fn pre_phase3_small_store_without_sidecar_routes_to_migration() {
        let path = temp_db_path("pre-phase3-small");
        {
            let _ = Pager::open_default(&path).expect("open_default() should succeed");
        }
        stamp_pre_phase3_layout(&path, 3);
        OpenOptions::new()
            .write(true)
            .open(&path)
            .expect("open() should succeed")
            .set_len(3 * PAGE_SIZE as u64)
            .expect("set_len() should succeed");

        let err = match Pager::open_default(&path) {
            Ok(_) => panic!("pre-phase-3 stores need migration first"),
            Err(err) => err,
        };
        assert!(matches!(err, PagerError::LegacyPagedStore { .. }));

        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);
    }
}