graphitesql 0.1.2

//! The write-side pager: buffered page mutations with atomic, journaled commit.
//!
//! All mutations during a transaction are buffered in an in-memory *overlay*
//! (page number → full page image). Nothing touches the database file until
//! [`commit`](WritePager::commit):
//!
//! * **ROLLBACK** is therefore trivial and always correct — drop the overlay.
//! * **COMMIT** is made crash-safe with a rollback journal: the original
//!   contents of every page about to be overwritten are written to a journal
//!   file and synced *before* the database file is modified; the journal is
//!   cleared only after the database is synced. A crash mid-commit leaves a
//!   journal that the next [`open`](WritePager::open) replays via recovery.
//!
//! Reads consult the overlay first, so within a transaction the pager is
//! read-your-writes consistent. It implements [`PageSource`], so the existing
//! b-tree cursors and schema reader work over it unchanged.

use super::pcache::{self, PageCache};
use super::wal::{SharedWalIndex, WalSnapshot};
use super::{Page, PageSource};
use crate::btree::page::{BtreePage, PageType};
use crate::btree::ptrmap::{self, PtrmapType};
use crate::error::{Error, Result};
use crate::format::header::HEADER_LEN;
use crate::format::{DatabaseHeader, TextEncoding};
use crate::vfs::File;
use alloc::boxed::Box;
use alloc::collections::BTreeMap;
use alloc::format;
use alloc::rc::Rc;
use alloc::string::String;
use alloc::vec;
use alloc::vec::Vec;
use core::cell::{Cell, RefCell};

/// The 8-byte magic that opens every SQLite-format rollback journal.
///
/// SQLite writes these exact bytes at offset 0; a journal is only "well-formed"
/// (and therefore a candidate for hot-journal recovery) if they are present.
const JOURNAL_MAGIC: [u8; 8] = [0xd9, 0xd5, 0x05, 0xf9, 0x20, 0xa1, 0x63, 0xd7];

/// Bytes occupied by the fixed fields of the journal header (magic, record
/// count, checksum nonce, initial page count, sector size, page size). The
/// header is then zero-padded out to [`JOURNAL_SECTOR`].
const JOURNAL_HDR_FIELDS: usize = 28;

/// Sector size assumed for the rollback journal. SQLite hard-codes 512 bytes
/// (there is no portable way to discover the true sector size), so we record the
/// same value at header offset 20 and pad the header out to this boundary. Each
/// page record begins on this boundary so a torn write of the header sector
/// cannot damage the page records that follow.
const JOURNAL_SECTOR: u64 = 512;

/// Compute the 4-byte page checksum exactly as SQLite does: seed with the
/// per-transaction `nonce`, then add the unsigned byte value at offsets
/// `len-200, len-400, …` down to the first non-negative index. The sum wraps in
/// `u32`. A sparse sample (every 200th byte) is intentional — it is what SQLite
/// uses, so the value must match byte-for-byte for cross-recovery.
fn journal_page_checksum(nonce: u32, page: &[u8]) -> u32 {
    let mut cksum = nonce;
    // X starts at len-200 and steps down by 200 while >= 0.
    let mut x = page.len() as isize - 200;
    while x >= 0 {
        cksum = cksum.wrapping_add(page[x as usize] as u32);
        x -= 200;
    }
    cksum
}

/// Fixed file offset of the SQLite "lock byte". The page that contains this byte
/// is never used to store data; on databases larger than 1 GiB it falls on a
/// page after page 1 that the allocator skips.
const PENDING_BYTE: u64 = 0x4000_0000;

/// A pointer-map entry: the page's type and its parent page number.
type PtrmapEntry = (PtrmapType, u32);

/// Auto-vacuum mode of a database, as recorded in the file header.
///
/// SQLite encodes the mode in two header fields: `largest_root_page` (offset 52)
/// is non-zero iff auto-vacuum is enabled, and `incremental_vacuum` (offset 64)
/// distinguishes FULL (0) from INCREMENTAL (non-zero). See the file-format spec,
/// "The Database Header".
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AutoVacuum {
    /// Auto-vacuum disabled (the default). `PRAGMA auto_vacuum` reports `0`.
    None,
    /// Full auto-vacuum: free pages are reclaimed automatically on each commit.
    /// `PRAGMA auto_vacuum` reports `1`.
    Full,
    /// Incremental auto-vacuum: free pages are tracked but only reclaimed on
    /// `PRAGMA incremental_vacuum`. `PRAGMA auto_vacuum` reports `2`.
    Incremental,
}

/// A writable pager over a database file, with an optional journal file.
pub struct WritePager {
    file: Box<dyn File>,
    journal: Option<Box<dyn File>>,
    header: DatabaseHeader,
    /// The header as it stands in the **last committed** database. `header` is
    /// mutated in place throughout a transaction (freelist trunk/count,
    /// `largest_root_page`, `user_version`, …), so on ROLLBACK it must be
    /// restored to the committed state — otherwise a stale freelist pointer left
    /// by the rolled-back transaction sends the next allocation reading a page
    /// that only ever existed in the discarded, grown file. Refreshed on every
    /// successful commit (and on the VACUUM image replace), restored by
    /// [`rollback`](Self::rollback).
    committed_header: DatabaseHeader,
    page_size: usize,
    /// Pages currently on disk (the durable page count).
    disk_pages: u32,
    /// Logical page count including not-yet-flushed allocations.
    page_count: u32,
    overlay: BTreeMap<u32, Vec<u8>>,
    /// The `-wal` file handle, when one was supplied (file-backed databases).
    wal_file: Option<Box<dyn File>>,
    /// WAL runtime state; `Some` when the database is in WAL mode.
    wal: Option<WalRuntime>,
    /// The write lock currently held on the main file. A write transaction takes
    /// `Reserved` on its first staged page and upgrades to `Exclusive` while
    /// flushing at commit, so a second writer to the same file gets
    /// [`Error::Busy`](crate::Error::Busy). Released on commit/rollback.
    ///
    /// A `Cell` because the persistent read lock is taken lazily on the `&self`
    /// read path ([`begin_read_txn`](Self::begin_read_txn)) — the first `SELECT`
    /// inside an explicit transaction acquires `Shared` through a shared borrow.
    held: Cell<crate::vfs::LockLevel>,
    /// Whether an explicit read transaction is open on this connection. When set,
    /// the pager holds a persistent `Shared` lock for the whole read transaction
    /// (taken by [`begin_read_txn`](Self::begin_read_txn)) rather than only
    /// transiently per page. This makes the reader visible to a concurrent writer
    /// on the same file: the writer's commit-time upgrade to `Exclusive` BUSYs
    /// until the reader ends its transaction, matching SQLite's locking model
    /// (`pager.c` `PAGER_SHARED` held across a read txn). Multiple such readers
    /// still coexist (`Shared` is a counted lock). Cleared and the lock released
    /// by [`end_read_txn`](Self::end_read_txn). A `Cell` for the same `&self`
    /// reason as [`held`](Self::held).
    read_txn: Cell<bool>,
    /// Whether this connection is holding a *transient* autocommit read lock
    /// (ROADMAP C9b-3). A bare autocommit `SELECT` (no open transaction) takes a
    /// `Shared` lock for the duration of the statement so a foreign *process* mid-
    /// write (holding the OS-exclusive lock under the pessimistic whole-file model)
    /// cannot be read torn; it is released the moment the statement finishes. Unlike
    /// [`read_txn`](Self::read_txn) this does not pin a repeatable-read snapshot —
    /// it exists only to gate the read against a concurrent foreign writer. Set by
    /// [`begin_autocommit_read`](Self::begin_autocommit_read), cleared by
    /// [`end_autocommit_read`](Self::end_autocommit_read). A `Cell` for the same
    /// `&self` read-path reason as [`held`](Self::held).
    transient_read: Cell<bool>,
    /// Open savepoints (innermost last); each snapshots the staged state so
    /// `ROLLBACK TO` can restore it.
    savepoints: Vec<Savepoint>,
    /// `PRAGMA secure_delete`: when set, the content of a page handed to the
    /// freelist is overwritten with zeros (so deleted data does not linger on
    /// disk). A per-connection runtime setting, not persisted in the file.
    secure_delete: bool,
    /// Bounded LRU cache of **clean** pages read from the main file (ROADMAP
    /// C8c). It only ever holds pages served straight from disk — never an
    /// overlay (dirty) or WAL page, both of which `read_page` consults first — so
    /// no dirty page can be evicted from here. A long read-heavy scan evicts
    /// least-recently-used clean pages instead of growing without bound; an
    /// evicted page is simply re-read from disk. Wrapped in a `RefCell` because
    /// `read_page` takes `&self`.
    read_cache: RefCell<PageCache>,
    /// Coherency token for the clean read cache **on the read-only path** (ROADMAP
    /// C8c-2): the database change counter (page 1, bytes 24–27) that the cached
    /// clean pages were read under.
    ///
    /// A pure read-only connection holds no write lock, so — unlike a writer — it
    /// cannot assume the on-disk pages are stable between statements: another
    /// in-process `Connection` may commit and bump the change counter. Caching is
    /// therefore only sound while that counter is unchanged. The token is
    /// established/re-checked at a statement boundary by
    /// [`revalidate_read_cache`](Self::revalidate_read_cache): if page 1's on-disk
    /// counter differs from this token, the cache is stale and is dropped before
    /// the statement reads anything.
    ///
    /// `None` means "not yet validated this statement" — the read cache is not
    /// trusted until a revalidation stamps a token. The write path never consults
    /// this field (it clears the cache on lock transitions instead). A `Cell`
    /// because the read path runs through `&self`.
    read_cache_token: Cell<Option<u32>>,
}

/// A snapshot of the pager's staged state captured by `SAVEPOINT`.
struct Savepoint {
    name: String,
    overlay: BTreeMap<u32, Vec<u8>>,
    header: DatabaseHeader,
    page_count: u32,
}

/// Per-connection live WAL state. Committed frames now live in a **shared**
/// [`SharedWalIndex`] (keyed by path in the VFS) so multiple in-process
/// `Connection`s over the same file read WAL frames coherently (ROADMAP C9c):
///
/// * A reader pins [`snapshot`](WalRuntime::snapshot) (the shared index's
///   high-water mark at read-transaction start) and resolves every page against
///   frames ≤ that snapshot — so a concurrent writer's later commits are invisible
///   until the reader ends its transaction (repeatable read, `wal.c`'s `mxFrame`).
/// * A writer appends its committed frames to both the `-wal` file and the shared
///   index under the write-intent lock, so the next reader of any other connection
///   sees them.
///
/// [`salt`](WalRuntime::salt) and the append cursor for *this* connection's writes
/// are seeded from the shared index at commit time, so two connections writing in
/// turn produce one continuous, checksum-consistent WAL.
struct WalRuntime {
    /// The process-shared wal-index for this file (all connections share it).
    index: SharedWalIndex,
    /// This connection's pinned read snapshot: the frame high-water mark and the
    /// generation it was captured under. Refreshed to the latest commit at each
    /// autocommit statement boundary, but held fixed (pinned) while a read
    /// transaction is open. A `Cell` because the read path and the read-txn
    /// begin/end hooks run through `&self`.
    snapshot: Cell<WalSnapshot>,
    /// The exact `mx_frame` this connection registered as a pinned reader in the
    /// shared index, or `None` if it has no open read transaction. Tracked
    /// separately from [`snapshot`](WalRuntime::snapshot) so that a read-then-write
    /// transaction (whose snapshot advances at commit) still unregisters the mark
    /// it originally registered. A `Cell` for the same `&self` reason as
    /// [`snapshot`](WalRuntime::snapshot).
    registered_mark: Cell<Option<u32>>,
    /// Database size in pages at [`snapshot`](WalRuntime::snapshot) (cached from
    /// the newest visible commit frame, or the main file if none). A `Cell` so the
    /// autocommit snapshot refresh through `&self` can update it.
    db_size: Cell<u32>,
    /// This connection's WAL salts (used when it is the one appending frames).
    salt: [u8; 8],
}

const WAL_MAGIC_LE: u32 = 0x377f_0682; // little-endian checksum variant
const WAL_HDR_LEN: usize = 32;
const WAL_FRAME_HDR_LEN: usize = 24;

/// One frame written during a WAL commit, staged for publication into the shared
/// wal-index after the `-wal` file is synced: `(page_no, commit_db_size, bytes,
/// next append offset, running checksum)`.
type AppendedFrame = (u32, u32, Vec<u8>, u64, (u32, u32));

impl WritePager {
    /// Open an existing database file for writing. Replays the journal first if a
    /// previous commit was interrupted.
    pub fn open(file: Box<dyn File>, journal: Option<Box<dyn File>>) -> Result<WritePager> {
        Self::open_wal(file, journal, None)
    }

    /// Like [`open`](Self::open), but also given the `-wal` companion file so the
    /// database can be opened (and reopened) in WAL mode.
    pub fn open_wal(
        mut file: Box<dyn File>,
        mut journal: Option<Box<dyn File>>,
        mut wal_file: Option<Box<dyn File>>,
    ) -> Result<WritePager> {
        if let Some(j) = journal.as_mut() {
            Self::recover(file.as_mut(), j.as_mut())?;
        }
        let file_size = file.size()?;
        if file_size < HEADER_LEN as u64 {
            return Err(Error::Corrupt("file too small to be a database".into()));
        }
        let mut head = [0u8; HEADER_LEN];
        file.read_exact_at(&mut head, 0)?;
        let header = DatabaseHeader::parse(&head)?;
        let page_size = header.page_size as usize;
        if file_size % page_size as u64 != 0 {
            return Err(Error::Corrupt(
                "file size not a multiple of page size".into(),
            ));
        }
        let pages = (file_size / page_size as u64) as u32;
        // If the database is in WAL mode, attach to the shared wal-index for this
        // path. Seed it from the `-wal` file if it is still empty (first open of
        // this file in the process); otherwise adopt whatever a sibling connection
        // has already published, so reads are coherent across connections.
        let wal = if header.read_version == 2 {
            match wal_file.as_mut() {
                Some(w) => Self::attach_wal(w.as_mut(), page_size)?,
                None => None,
            }
        } else {
            None
        };
        let page_count = wal.as_ref().map(|w| w.db_size.get()).unwrap_or(pages);
        Ok(WritePager {
            file,
            journal,
            committed_header: header.clone(),
            header,
            page_size,
            disk_pages: pages,
            page_count,
            overlay: BTreeMap::new(),
            wal_file,
            wal,
            held: Cell::new(crate::vfs::LockLevel::Unlocked),
            read_txn: Cell::new(false),
            transient_read: Cell::new(false),
            savepoints: Vec::new(),
            secure_delete: false,
            read_cache: RefCell::new(PageCache::new(pcache::DEFAULT_CACHE_SIZE, page_size)),
            read_cache_token: Cell::new(None),
        })
    }

    /// Create a brand-new, empty database (a single `sqlite_schema` leaf page).
    pub fn create(
        file: Box<dyn File>,
        journal: Option<Box<dyn File>>,
        page_size: u32,
    ) -> Result<WritePager> {
        Self::create_wal(file, journal, None, page_size)
    }

    /// Like [`create`](Self::create), with the `-wal` companion file available.
    pub fn create_wal(
        file: Box<dyn File>,
        journal: Option<Box<dyn File>>,
        wal_file: Option<Box<dyn File>>,
        page_size: u32,
    ) -> Result<WritePager> {
        Self::create_auto_vacuum(file, journal, wal_file, page_size, AutoVacuum::None)
    }

    /// Create a brand-new, empty database in the given auto-vacuum `mode`.
    ///
    /// For an *empty* database (just page 1, no user tables) SQLite records the
    /// mode purely in the header: `largest_root_page` (offset 52) is set to `1`
    /// when auto-vacuum is enabled (FULL or INCREMENTAL) and `0` when it is off,
    /// and `incremental_vacuum` (offset 64) is `1` for INCREMENTAL, `0`
    /// otherwise. The file is still a single page — no pointer-map page exists
    /// yet, so there is no ptrmap maintenance to do here. (Those values were
    /// confirmed empirically against `sqlite3 3.50.4`.)
    ///
    /// This is the storage foundation for auto-vacuum; the ptrmap *write*
    /// maintenance that keeps the map current as pages are allocated/freed is a
    /// separate concern layered on top.
    pub fn create_auto_vacuum(
        file: Box<dyn File>,
        journal: Option<Box<dyn File>>,
        wal_file: Option<Box<dyn File>>,
        page_size: u32,
        mode: AutoVacuum,
    ) -> Result<WritePager> {
        if page_size < 512 || !page_size.is_power_of_two() {
            return Err(Error::Error(format!("invalid page size {page_size}")));
        }
        // Empty auto-vacuum db: page 1 is its own largest (and only) root.
        let (largest_root_page, incremental_vacuum) = match mode {
            AutoVacuum::None => (0, 0),
            AutoVacuum::Full => (1, 0),
            AutoVacuum::Incremental => (1, 1),
        };
        let header = DatabaseHeader {
            page_size,
            write_version: 1,
            read_version: 1,
            reserved_space: 0,
            change_counter: 1,
            size_in_pages: 1,
            freelist_trunk: 0,
            freelist_count: 0,
            schema_cookie: 0,
            schema_format: 4,
            default_cache_size: 0,
            largest_root_page,
            text_encoding: TextEncoding::Utf8,
            user_version: 0,
            incremental_vacuum,
            application_id: 0,
            version_valid_for: 1,
            sqlite_version_number: 3_053_002,
        };
        let mut wp = WritePager {
            file,
            journal,
            committed_header: header.clone(),
            header,
            page_size: page_size as usize,
            disk_pages: 0,
            page_count: 1,
            overlay: BTreeMap::new(),
            wal_file,
            wal: None,
            held: Cell::new(crate::vfs::LockLevel::Unlocked),
            read_txn: Cell::new(false),
            transient_read: Cell::new(false),
            savepoints: Vec::new(),
            secure_delete: false,
            read_cache: RefCell::new(PageCache::new(
                pcache::DEFAULT_CACHE_SIZE,
                page_size as usize,
            )),
            read_cache_token: Cell::new(None),
        };
        // Page 1: db header (0..100) + an empty table-leaf b-tree at offset 100.
        let mut page1 = vec![0u8; page_size as usize];
        wp.header.write_to(&mut page1)?;
        write_empty_leaf_header(&mut page1, HEADER_LEN, page_size);
        wp.overlay.insert(1, page1);
        Ok(wp)
    }

    /// The database header (reflects in-transaction changes once committed).
    /// Touching it marks page 1 dirty so a header-only change (e.g.
    /// `PRAGMA user_version=…`) is still flushed by `commit`, which otherwise
    /// short-circuits when no pages changed.
    pub fn header_mut(&mut self) -> &mut DatabaseHeader {
        if !self.overlay.contains_key(&1)
            && let Ok(p) = self.read_page(1)
        {
            self.overlay.insert(1, p);
        }
        &mut self.header
    }

    /// Read the full bytes of page `number` (overlay first, then WAL, then disk).
    ///
    /// Clean pages read from the main file are served through a bounded LRU cache
    /// (the `read_cache` field) so a long scan does not grow the resident set
    /// without bound. The overlay (dirty pages) and the WAL are consulted *before*
    /// the cache, so the cache only ever holds — and only ever returns — clean
    /// on-disk pages; a dirty page is therefore never evictable from it.
    ///
    /// # Coherency
    ///
    /// The clean read cache is trusted under exactly two regimes, both of which
    /// guarantee the cached bytes still match the file:
    ///
    /// * **Write transaction** — this connection holds a write lock (`Reserved`+),
    ///   so it is the only writer and the on-disk clean pages cannot change beneath
    ///   the cache. The cache is cleared on every lock transition (see
    ///   `acquire_write_intent` and `release_locks`).
    /// * **Read-only, validated** — this connection holds no write lock, but a
    ///   statement boundary has just confirmed (via
    ///   [`revalidate_read_cache`](Self::revalidate_read_cache)) that page 1's
    ///   on-disk change counter still equals the token the cache was populated
    ///   under. A foreign `Connection` bumps that counter on every commit, so an
    ///   unchanged counter means no foreign write has landed since the cache was
    ///   filled. The token is `Some` only while that invariant holds; the next
    ///   statement re-checks and drops the cache if it changed.
    ///
    /// With neither regime active (`read_cache_token` is `None` and no write lock),
    /// the cache is bypassed and the page is read straight from disk, exactly as
    /// before — the read-only path stays correct even if the exec layer never calls
    /// the revalidation hook.
    pub fn read_page(&self, number: u32) -> Result<Vec<u8>> {
        if let Some(bytes) = self.overlay.get(&number) {
            return Ok(bytes.clone());
        }
        // In WAL mode, the newest committed version of a page (up to this
        // connection's pinned snapshot) may live in the shared wal-index; any page
        // ≤ the snapshot db size that is *not* in the WAL comes from the main file
        // (checkpoint guarantees it is there). A sibling connection may have grown
        // or checkpointed the main file since this connection last synced its
        // `disk_pages`, so bound the read by the snapshot db size and read the
        // bytes straight from the file (bypassing the possibly-stale `disk_pages`
        // and the clean read cache, which is not the read authority in WAL mode).
        if let Some(w) = &self.wal {
            if let Some(bytes) = w
                .index
                .with(|ix| ix.find_frame(number, w.snapshot.get().mx_frame))
            {
                return Ok(bytes);
            }
            let db_size = w.db_size.get();
            if number == 0 || number > db_size {
                return Err(Error::Corrupt(format!("page {number} out of range")));
            }
            let mut buf = vec![0u8; self.page_size];
            self.file
                .read_exact_at(&mut buf, (number as u64 - 1) * self.page_size as u64)?;
            return Ok(buf);
        }
        if number == 0 || number > self.disk_pages {
            return Err(Error::Corrupt(format!("page {number} out of range")));
        }
        // Trust the cache under a write lock (we are the only writer) or on the
        // read-only path once a statement boundary has stamped a still-valid
        // change-counter token (see method docs).
        let cacheable = self.held.get() >= crate::vfs::LockLevel::Reserved
            || self.read_cache_token.get().is_some();
        if cacheable && let Some(bytes) = self.read_cache.borrow_mut().get(number) {
            return Ok(bytes.as_ref().clone());
        }
        let mut buf = vec![0u8; self.page_size];
        self.file
            .read_exact_at(&mut buf, (number as u64 - 1) * self.page_size as u64)?;
        if cacheable {
            let data = Rc::new(buf);
            self.read_cache
                .borrow_mut()
                .insert(number, Rc::clone(&data));
            return Ok(data.as_ref().clone());
        }
        Ok(buf)
    }

    /// Revalidate the clean read cache against the database's on-disk **change
    /// counter** (page 1, bytes 24–27) at a statement boundary, for the read-only
    /// path (ROADMAP C8c-2).
    ///
    /// A pure read-only connection holds no write lock, so between statements a
    /// foreign in-process `Connection` may commit and rewrite pages under it. The
    /// change counter is SQLite's monotonic "the file changed" token: it is bumped
    /// on every commit. This method reads that counter directly from disk and:
    ///
    /// * if it differs from the token the cache was populated under (or the cache
    ///   was never validated), **drops the entire cache** and re-stamps the token,
    ///   so the upcoming statement re-reads fresh pages from disk;
    /// * if it is unchanged, leaves the cache intact — the cached clean pages are
    ///   still byte-identical to the file, so the statement may reuse them.
    ///
    /// Either way, after this call `read_cache_token` is `Some(current counter)`,
    /// which is what enables the cache for the duration of the statement (see
    /// [`read_page`](Self::read_page)). Call it once at the start of each read
    /// statement, before any page is read.
    ///
    /// A no-op while a write lock is held: a writer owns the file and manages the
    /// cache through its lock transitions, so the read-only token must not
    /// interfere. In WAL mode the change counter is not the authority for reads
    /// (the WAL is), so this also leaves the cache untouched there.
    ///
    /// Reading page 1 goes through [`read_page`](Self::read_page); costing one page
    /// read per statement (itself cached once the token is set), which is far
    /// cheaper than re-reading every page of every statement from disk.
    pub fn revalidate_read_cache(&self) {
        // In WAL mode the shared wal-index — not the change counter — is the read
        // authority. Advance this connection's read snapshot to the latest commit
        // so an autocommit statement sees a sibling connection's committed WAL
        // frames (ROADMAP C9c). A pinned reader (open read transaction) keeps its
        // fixed snapshot for repeatable read; a writer owns coherency through its
        // lock. Then leave the clean read cache alone (it is bypassed in WAL mode).
        if let Some(w) = &self.wal {
            if self.held.get() < crate::vfs::LockLevel::Reserved
                && w.registered_mark.get().is_none()
            {
                let snap = w.index.with(|ix| ix.snapshot());
                w.snapshot.set(snap);
                // The snapshot db size comes from the newest visible WAL commit;
                // if the WAL was reset (checkpointed) it has none, so fall back to
                // the *current* main-file page count (a sibling may have grown or
                // checkpointed it since this connection last synced `disk_pages`).
                let db = w
                    .index
                    .with(|ix| ix.snapshot_db_size(snap.mx_frame))
                    .or_else(|| self.main_file_page_count())
                    .unwrap_or(self.disk_pages);
                w.db_size.set(db);
            }
            return;
        }
        // Under a write lock the writer already owns coherency.
        if self.held.get() >= crate::vfs::LockLevel::Reserved {
            return;
        }
        let current = match self.disk_change_counter() {
            Some(c) => c,
            // Could not read the counter (e.g. an empty/absent page 1): fall back
            // to the always-correct uncached path by clearing the token.
            None => {
                self.read_cache_token.set(None);
                self.read_cache.borrow_mut().clear();
                return;
            }
        };
        if self.read_cache_token.get() != Some(current) {
            // The file changed under us (or this is the first statement): the
            // cached clean pages may be stale, so drop them and re-stamp.
            self.read_cache.borrow_mut().clear();
            self.read_cache_token.set(Some(current));
        }
    }

    /// The current number of pages in the **main** database file, read fresh from
    /// its size. Used in WAL mode to bound reads that fall through to the main
    /// file after a sibling connection has grown or checkpointed it (so a stale
    /// per-connection `disk_pages` does not reject a page that now exists).
    /// Returns `None` if the size cannot be read or is not a page multiple.
    fn main_file_page_count(&self) -> Option<u32> {
        let size = self.file.size().ok()?;
        if self.page_size == 0 || size % self.page_size as u64 != 0 {
            return None;
        }
        Some((size / self.page_size as u64) as u32)
    }

    /// Read the database change counter (page 1, bytes 24–27, big-endian) straight
    /// from disk, bypassing the cache. Returns `None` if page 1 cannot be read.
    fn disk_change_counter(&self) -> Option<u32> {
        if self.disk_pages == 0 {
            return None;
        }
        let mut hdr = [0u8; 28];
        self.file.read_exact_at(&mut hdr, 0).ok()?;
        Some(be32(&hdr, 24))
    }

    /// Reconfigure the bounded clean-page read cache from a `cache_size` value
    /// (the `cache_size` PRAGMA convention: a positive value is a page count, a
    /// negative value is KiB of memory). Lowering it evicts the
    /// least-recently-used clean pages immediately. Dirty pages live in the
    /// overlay and are unaffected.
    pub fn set_cache_size(&self, cache_size: i64) {
        self.read_cache
            .borrow_mut()
            .set_cache_size(cache_size, self.page_size);
    }

    /// The number of **clean** pages currently resident in the read cache.
    /// Read-only accessor used to assert the LRU bound holds; not part of the
    /// stable API. Dirty (overlay) pages are not counted here — they are tracked
    /// separately and never evicted.
    #[doc(hidden)]
    pub fn resident_clean_pages(&self) -> usize {
        self.read_cache.borrow().len()
    }

    /// Cumulative `(hits, misses)` of the clean read cache since this pager was
    /// opened. A hit is a page served from cache; a miss is one that had to be
    /// read from disk. Read-only accessor used by tests to prove repeated
    /// same-page reads within a statement avoid disk; not part of the stable API.
    #[doc(hidden)]
    pub fn read_cache_stats(&self) -> (u64, u64) {
        self.read_cache.borrow().stats()
    }

    /// The number of dirty (staged, not-yet-committed) pages held in the overlay.
    /// These are never evictable. Read-only accessor for tests; not stable API.
    #[doc(hidden)]
    pub fn resident_dirty_pages(&self) -> usize {
        self.overlay.len()
    }

    /// Stage a full page image into the overlay.
    pub fn write_page(&mut self, number: u32, bytes: Vec<u8>) -> Result<()> {
        if bytes.len() != self.page_size {
            return Err(Error::Error("page image has wrong size".into()));
        }
        self.acquire_write_intent()?;
        self.overlay.insert(number, bytes);
        Ok(())
    }

    /// Begin an explicit read transaction: acquire a **persistent** `Shared`
    /// (read) lock and hold it until [`end_read_txn`](Self::end_read_txn).
    ///
    /// The pager otherwise only touches `Shared` on the path to a write lock;
    /// pure reads are served without holding one. That is fine for autocommit
    /// reads, but an *open* read transaction (`BEGIN; SELECT …`) must keep the
    /// database's committed snapshot stable and — like SQLite — make itself
    /// visible to a concurrent writer, so the writer's commit-time upgrade to
    /// `Exclusive` BUSYs until this reader finishes. This method installs that
    /// persistent lock.
    ///
    /// - Idempotent: calling it again inside the same read transaction is a no-op.
    /// - If a *write* lock is already held (a read followed by a write in the same
    ///   transaction, or a write transaction that also reads), the existing
    ///   `Reserved`/`Exclusive` lock already excludes other writers, so this just
    ///   records that a read transaction is open without weakening the lock.
    /// - Multiple connections may hold the persistent `Shared` lock at once
    ///   (`Shared` is a counted lock); readers never block each other.
    ///
    /// Returns [`Error::Busy`] if a foreign writer already holds `Pending`/
    /// `Exclusive` on the file (a reader cannot start while a writer is draining
    /// readers, matching SQLite).
    ///
    /// Takes `&self`: the SQL read path (`SELECT`) runs through `&self`, and the
    /// C9a rule is to acquire the persistent lock lazily at the *first* read
    /// inside an explicit transaction. The lock state lives in `Cell`s and the
    /// `File::lock` trait method takes `&self`, so this needs no mutable borrow.
    pub fn begin_read_txn(&self) -> Result<()> {
        use crate::vfs::LockLevel;
        if self.held.get() < LockLevel::Shared {
            self.file.lock(LockLevel::Shared)?;
            self.held.set(LockLevel::Shared);
        }
        // In WAL mode, pin a stable snapshot for the whole read transaction and
        // register it with the shared index so a concurrent checkpoint cannot drop
        // the frames this reader still needs (`wal.c`'s per-reader mxFrame). Only
        // the *first* read inside the transaction pins it (idempotent thereafter).
        if let Some(w) = &self.wal
            && w.registered_mark.get().is_none()
        {
            let snap = w.index.with(|ix| {
                let s = ix.snapshot();
                ix.register_reader(s.mx_frame);
                s
            });
            w.snapshot.set(snap);
            w.db_size.set(
                w.index
                    .with(|ix| ix.snapshot_db_size(snap.mx_frame))
                    .or_else(|| self.main_file_page_count())
                    .unwrap_or(self.disk_pages),
            );
            w.registered_mark.set(Some(snap.mx_frame));
        }
        self.read_txn.set(true);
        Ok(())
    }

    /// End an explicit read transaction opened by
    /// [`begin_read_txn`](Self::begin_read_txn), releasing the persistent
    /// `Shared` lock so a waiting writer can now upgrade to `Exclusive`.
    ///
    /// Only the read-transaction bookkeeping and a *pure* `Shared` lock are
    /// dropped here. If the transaction turned into a write (the pager now holds
    /// `Reserved`/`Exclusive`), the lock is left to the write path's
    /// commit/rollback (`release_locks`) — ending the read side must not strand a
    /// half-committed writer. Idempotent when no read transaction is open.
    ///
    /// Takes `&self` (the lock state is interior-mutable) so the exec layer can
    /// release a read-only explicit transaction's lock on COMMIT/ROLLBACK without
    /// a mutable pager borrow.
    pub fn end_read_txn(&self) {
        use crate::vfs::LockLevel;
        self.read_txn.set(false);
        // Release this connection's pinned WAL reader mark so a checkpoint may now
        // reset the WAL. The next autocommit statement re-snapshots at the latest
        // commit.
        if let Some(w) = &self.wal
            && let Some(mark) = w.registered_mark.get()
        {
            w.index.with(|ix| ix.unregister_reader(mark));
            w.registered_mark.set(None);
        }
        if self.held.get() == LockLevel::Shared {
            let _ = self.file.unlock(LockLevel::Unlocked);
            self.held.set(LockLevel::Unlocked);
            // A foreign writer may now change the file, so anything cached under
            // the read lock must not be served to a later read. Force the next
            // read-only statement to re-validate the change counter before it
            // trusts the cache again.
            self.read_cache.borrow_mut().clear();
            self.read_cache_token.set(None);
        }
    }

    /// Whether an explicit read transaction (persistent `Shared` lock) is open.
    /// Read-only accessor for tests and the exec layer's txn bookkeeping.
    pub fn in_read_txn(&self) -> bool {
        self.read_txn.get()
    }

    /// Take a **transient** `Shared` lock around a bare autocommit read (ROADMAP
    /// C9b-3), returning `true` if it was taken (so the caller knows to release it
    /// with [`end_autocommit_read`](Self::end_autocommit_read)).
    ///
    /// A bare autocommit `SELECT` otherwise reads without holding any lock, which is
    /// safe in-process (the cache + write locks coordinate) but lets a *foreign
    /// process* mid-write be read torn: under the pessimistic whole-file model a
    /// cross-process writer holds the OS-exclusive lock for its whole write txn, and
    /// without a `Shared` lock this reader never contends with it. Acquiring `Shared`
    /// makes the read wait for (or [`Error::Busy`] against) that foreign exclusive
    /// lock, exactly as an explicit read transaction does — but released immediately
    /// at statement end rather than pinned for a repeatable-read snapshot.
    ///
    /// A no-op (returns `false`) when a lock is already held — an explicit read/write
    /// transaction owns coordination — so it only ever fires for a true autocommit
    /// read. In-process this never adds contention: the OS lock is process-wide, so a
    /// sibling connection's write already holds it and `reconcile` is a no-op for a
    /// same-process `Shared` acquire; only a cross-process exclusive holder BUSYs.
    pub fn begin_autocommit_read(&self) -> Result<bool> {
        use crate::vfs::LockLevel;
        if self.read_txn.get() || self.transient_read.get() || self.held.get() >= LockLevel::Shared
        {
            return Ok(false);
        }
        self.file.lock(LockLevel::Shared)?;
        self.held.set(LockLevel::Shared);
        self.transient_read.set(true);
        Ok(true)
    }

    /// Release the transient autocommit read lock taken by
    /// [`begin_autocommit_read`](Self::begin_autocommit_read). Idempotent; a no-op
    /// if no transient lock is held (e.g. the statement upgraded to a write, whose
    /// own commit/rollback path releases the lock instead). The clean read cache is
    /// left intact — the next statement's `revalidate_read_cache` drops it via the
    /// change-counter token if a foreign writer committed meanwhile.
    pub fn end_autocommit_read(&self) {
        use crate::vfs::LockLevel;
        if !self.transient_read.get() {
            return;
        }
        self.transient_read.set(false);
        // Only release if the statement did not escalate to a write lock (which the
        // write path owns and releases at commit/rollback).
        if self.held.get() == LockLevel::Shared {
            let _ = self.file.unlock(LockLevel::Unlocked);
            self.held.set(LockLevel::Unlocked);
        }
    }

    /// Take the write-intent (`RESERVED`) lock on the main file before staging
    /// changes, so a concurrent writer to the same file is rejected with
    /// [`Error::Busy`] rather than corrupting it. Idempotent within a transaction.
    fn acquire_write_intent(&mut self) -> Result<()> {
        use crate::vfs::LockLevel;
        let entry = self.held.get();
        if self.held.get() < LockLevel::Shared {
            self.file.lock(LockLevel::Shared)?;
            self.held.set(LockLevel::Shared);
        }
        if self.held.get() < LockLevel::Reserved {
            if let Err(e) = self.file.lock(LockLevel::Reserved) {
                // A rejected first write must not keep the SHARED lock it just
                // took, or it would block the current writer from committing.
                if entry == LockLevel::Unlocked {
                    self.release_locks();
                }
                return Err(e);
            }
            self.held.set(LockLevel::Reserved);
            // Entering a write transaction: a foreign connection over the same
            // VFS may have committed since we last held a lock, so anything in
            // the clean read cache could be stale. Drop it; pages are re-read
            // from disk and re-cached fresh under the write lock. Also drop the
            // read-only coherency token so the read-only cache regime does not
            // shadow the write-lock regime while the lock is held.
            self.read_cache.borrow_mut().clear();
            self.read_cache_token.set(None);
        }
        Ok(())
    }

    /// Upgrade to the `EXCLUSIVE` lock for the flush phase of a commit.
    fn acquire_exclusive(&mut self) -> Result<()> {
        use crate::vfs::LockLevel;
        self.acquire_write_intent()?;
        if self.held.get() < LockLevel::Exclusive {
            self.file.lock(LockLevel::Exclusive)?;
            self.held.set(LockLevel::Exclusive);
        }
        Ok(())
    }

    /// Drop all locks at the end of a transaction — the persistent read
    /// (`Shared`) lock as well as any write (`Reserved`/`Exclusive`) lock.
    fn release_locks(&mut self) {
        use crate::vfs::LockLevel;
        self.read_txn.set(false);
        // Drop this connection's pinned WAL reader mark (if any) so a checkpoint
        // may now reset the WAL, and so the next autocommit statement re-snapshots
        // at the latest commit. Mirrors the unpin in `end_read_txn`.
        if let Some(w) = &self.wal
            && let Some(mark) = w.registered_mark.get()
        {
            w.index.with(|ix| ix.unregister_reader(mark));
            w.registered_mark.set(None);
        }
        if self.held.get() != LockLevel::Unlocked {
            let _ = self.file.unlock(LockLevel::Unlocked);
            self.held.set(LockLevel::Unlocked);
            // Once unlocked, a foreign writer may change the file, so anything
            // we cached under the lock must not be served to a later read. Force
            // the next read-only statement to re-validate before trusting it.
            self.read_cache.borrow_mut().clear();
            self.read_cache_token.set(None);
        }
    }

    /// Allocate a page, reusing one from the freelist if available, otherwise
    /// extending the file. The returned page is staged zeroed in the overlay.
    ///
    /// In auto-vacuum mode the file is interleaved with pointer-map pages; when
    /// extending the file we must never hand out a ptrmap page number. If the
    /// next page would be a ptrmap page (or the lock-byte page), we allocate and
    /// zero it in place and advance to the following page.
    pub fn allocate_page(&mut self) -> Result<u32> {
        if self.header.freelist_count > 0 && self.header.freelist_trunk != 0 {
            return self.alloc_from_freelist();
        }
        let auto_vacuum = self.auto_vacuum_on();
        let usable = self.usable_size() as u32;
        loop {
            self.page_count += 1;
            let n = self.page_count;
            if auto_vacuum && self.is_reserved_layout_page(n, usable) {
                // Materialize the ptrmap (or lock-byte) page zeroed and keep
                // going; its contents are filled in by `rebuild_ptrmap` at commit.
                self.overlay.insert(n, vec![0u8; self.page_size]);
                continue;
            }
            self.overlay.insert(n, vec![0u8; self.page_size]);
            return Ok(n);
        }
    }

    /// Whether the database is in any auto-vacuum mode (FULL or INCREMENTAL).
    fn auto_vacuum_on(&self) -> bool {
        self.header.largest_root_page != 0
    }

    /// Whether `pgno` is a structural page that must not be handed out as data:
    /// a pointer-map page, or the page that contains the lock byte (only when the
    /// page size makes that byte fall on a page other than page 1).
    fn is_reserved_layout_page(&self, pgno: u32, usable: u32) -> bool {
        if ptrmap::is_ptrmap_page(usable, pgno) {
            return true;
        }
        // The lock byte lives at fixed file offset 2^30. The page holding it is
        // skipped by the allocator. For page sizes ≤ 1 GiB on small databases
        // this never triggers, but we stay correct for large files.
        let lock_page = (PENDING_BYTE / self.page_size as u64) as u32 + 1;
        lock_page > 1 && pgno == lock_page
    }

    /// Pop a page off the freelist (file-format spec, "The Freelist"). The
    /// freelist is trunk pages, each holding a next-trunk pointer, a leaf count,
    /// and that many free-page numbers.
    fn alloc_from_freelist(&mut self) -> Result<u32> {
        let trunk = self.header.freelist_trunk;
        let mut tbytes = self.read_page(trunk)?;
        let leaf_count = be32(&tbytes, 4);
        if leaf_count > 0 {
            // Reuse the last leaf; the trunk stays on the list.
            let idx = 8 + 4 * (leaf_count as usize - 1);
            let leaf = be32(&tbytes, idx);
            put32(&mut tbytes, 4, leaf_count - 1);
            self.write_page(trunk, tbytes)?;
            self.header.freelist_count -= 1;
            self.overlay.insert(leaf, vec![0u8; self.page_size]);
            Ok(leaf)
        } else {
            // No leaves: consume the trunk page itself; its successor heads the list.
            let next = be32(&tbytes, 0);
            self.header.freelist_trunk = next;
            self.header.freelist_count -= 1;
            self.overlay.insert(trunk, vec![0u8; self.page_size]);
            Ok(trunk)
        }
    }

    /// Set `PRAGMA secure_delete`: when `on`, the content of a freed page is
    /// zeroed before it joins the freelist.
    pub fn set_secure_delete(&mut self, on: bool) {
        self.secure_delete = on;
    }

    /// Return `page` to the freelist (appending it as a leaf of the first trunk
    /// if there is room, else making it a new trunk page).
    pub fn free_page(&mut self, page: u32) -> Result<()> {
        let trunk = self.header.freelist_trunk;
        // SQLite caps trunk leaves at usable/4 - 2 (integrity_check enforces it).
        let max_leaves = (self.usable_size() / 4).saturating_sub(2) as u32;
        if trunk != 0 {
            let mut tbytes = self.read_page(trunk)?;
            let leaf_count = be32(&tbytes, 4);
            if leaf_count < max_leaves {
                let idx = 8 + 4 * leaf_count as usize;
                put32(&mut tbytes, idx, page);
                put32(&mut tbytes, 4, leaf_count + 1);
                self.write_page(trunk, tbytes)?;
                // With secure_delete, overwrite the freed page's old content. As a
                // freelist *leaf* it carries no required bytes, so a zero page is
                // valid (integrity_check ignores freelist-leaf content).
                if self.secure_delete {
                    self.write_page(page, vec![0u8; self.page_size])?;
                }
                self.header.freelist_count += 1;
                return Ok(());
            }
        }
        // Make `page` a new trunk pointing at the previous head.
        let mut nb = vec![0u8; self.page_size];
        put32(&mut nb, 0, trunk); // next trunk
        put32(&mut nb, 4, 0); // leaf count
        self.write_page(page, nb)?;
        self.header.freelist_trunk = page;
        self.header.freelist_count += 1;
        Ok(())
    }

    /// Rebuild every pointer-map page from the current logical page contents.
    ///
    /// Only called in auto-vacuum mode, just before a commit flushes pages. It
    /// derives the (type, parent) of every tracked page by walking the b-tree
    /// forest (roots from `sqlite_schema`, plus page 1's own schema tree) and the
    /// freelist, then re-stages the affected ptrmap pages into the overlay so the
    /// flush writes them. This whole-map rebuild keeps the ptrmap correct across
    /// arbitrary structural changes (splits, merges, overflow chains, frees)
    /// without threading parent bookkeeping through every b-tree writer.
    ///
    /// It also stamps `largest_root_page` = max root page number, which
    /// `integrity_check` requires to match exactly.
    /// Derive the desired pointer-map entries `(type, parent)` for every tracked
    /// page from the current logical page contents, plus the largest root page
    /// number. Shared by [`rebuild_ptrmap`](Self::rebuild_ptrmap) and the
    /// FULL-mode commit-time relocator, which needs the same parent bookkeeping
    /// to fix references to moved pages.
    fn compute_want(&self) -> Result<(BTreeMap<u32, PtrmapEntry>, u32)> {
        let mut want: BTreeMap<u32, PtrmapEntry> = BTreeMap::new();
        // 1. Discover all b-tree roots: the schema tree (page 1) and every
        //    `rootpage` recorded in sqlite_schema. Roots get a RootPage entry
        //    (page 1 is never tracked). Then walk each tree for Btree/overflow.
        let mut roots: Vec<u32> = Vec::new();
        roots.push(1);
        let mut max_root = 1u32;
        for root in self.schema_roots()? {
            if root >= 2 {
                want.insert(root, (PtrmapType::RootPage, 0));
            }
            if root > max_root {
                max_root = root;
            }
            roots.push(root);
        }
        for root in roots {
            self.walk_btree_ptrmap(root, &mut want)?;
        }
        // 2. Freelist pages (trunk + leaves) are FreePage with parent 0.
        self.walk_freelist_ptrmap(&mut want)?;
        Ok((want, max_root))
    }

    fn rebuild_ptrmap(&mut self) -> Result<()> {
        let usable = self.usable_size() as u32;
        // Desired (type, parent) for every tracked page, plus the max root page.
        let (want, max_root) = self.compute_want()?;

        // Keep the header's largest_root_page in sync.
        self.header.largest_root_page = max_root;

        // Materialize the ptrmap pages: gather the entries that fall on each,
        //    and write them. Pages with no live entry keep their slot's previous
        //    bytes (integrity_check never reads an unreferenced slot).
        let mut by_map: BTreeMap<u32, Vec<(u32, PtrmapEntry)>> = BTreeMap::new();
        for (&pgno, &ent) in &want {
            let map = ptrmap::ptrmap_pageno(usable, pgno);
            by_map.entry(map).or_default().push((pgno, ent));
        }
        for (map_pg, entries) in by_map {
            let mut page = if self.has_page(map_pg) {
                self.read_page(map_pg)?
            } else {
                vec![0u8; self.page_size]
            };
            for (pgno, (kind, parent)) in entries {
                let off = ptrmap::ptrmap_entry_offset(usable, pgno);
                let enc = ptrmap::encode_entry(kind, parent);
                page[off..off + ptrmap::ENTRY_SIZE].copy_from_slice(&enc);
            }
            self.overlay.insert(map_pg, page);
        }
        Ok(())
    }

    /// Run FULL-mode commit-time truncation, best-effort: only in FULL auto-vacuum
    /// mode, and any error rolls back to the pre-relocation staged state so the
    /// commit proceeds via the sound "leave freed pages in place" path. A
    /// sound-but-not-maximally-compact file is always acceptable; a corrupt one
    /// never is.
    fn maybe_autovacuum_truncate(&mut self) {
        if self.auto_vacuum() != AutoVacuum::Full {
            return;
        }
        let saved_overlay = self.overlay.clone();
        let saved_header = self.header.clone();
        let saved_count = self.page_count;
        // `limit == 0` ⇒ reclaim as much as possible (full compaction).
        if self.autovacuum_truncate(0).is_err() {
            self.overlay = saved_overlay;
            self.header = saved_header;
            self.page_count = saved_count;
        }
    }

    /// `PRAGMA incremental_vacuum(n)`: reclaim up to `n` free pages off the end of
    /// the file for an `auto_vacuum=INCREMENTAL` database, returning the number of
    /// pages actually removed. When `n <= 0` it reclaims as many as possible (full
    /// compaction). The reclaimed pages are dropped from the staged image; a
    /// subsequent [`commit`](Self::commit) makes the smaller file durable.
    ///
    /// Outside INCREMENTAL mode this is a no-op that reclaims nothing, matching
    /// SQLite (which does nothing for `auto_vacuum` NONE/FULL). Like the FULL-mode
    /// commit-time truncation it reuses, it is best-effort: any internal relocation
    /// error rolls the staged state back to its pre-reclamation form, so the worst
    /// case is a sound-but-less-compact file, never a corrupt one.
    pub fn incremental_vacuum(&mut self, n: i64) -> Result<u32> {
        if self.auto_vacuum() != AutoVacuum::Incremental {
            return Ok(0);
        }
        // `n <= 0` ⇒ unbounded (`limit == 0`); otherwise cap the pages truncated.
        let limit = if n <= 0 {
            0
        } else {
            n.min(u32::MAX as i64) as u32
        };
        let before = self.page_count;
        let saved_overlay = self.overlay.clone();
        let saved_header = self.header.clone();
        let saved_count = self.page_count;
        if self.autovacuum_truncate(limit).is_err() {
            self.overlay = saved_overlay;
            self.header = saved_header;
            self.page_count = saved_count;
            return Ok(0);
        }
        Ok(before - self.page_count)
    }

    /// FULL `auto_vacuum` commit-time truncation (SQLite's `autoVacuumCommit`).
    ///
    /// Relocates pages from the end of the file into lower-numbered free slots and
    /// shrinks `page_count`, so the file stays compact after deletes. Runs only in
    /// FULL mode, just before [`rebuild_ptrmap`](Self::rebuild_ptrmap) regenerates
    /// the pointer map and `largest_root_page` from the relocated structure.
    ///
    /// The relocation moves each trailing data page into the lowest free page,
    /// then fixes the single pointer that referenced it (a parent's child pointer,
    /// an overflow holder's link, or the previous overflow page's next-pointer),
    /// looked up from the freshly-derived `want` map. B-tree **root** pages are
    /// never relocated: their page number is cached in the executor's in-memory
    /// schema, which this layer must not invalidate. When the lowest movable page
    /// is a root (or no free slot remains, or a structural page blocks the way),
    /// the loop stops and leaves the remaining pages in place — still sound, just
    /// less compact. Freed pages that survive below the new size are rebuilt onto
    /// the freelist; everything above it is simply truncated away.
    ///
    /// Any error leaves the staged state untouched-enough that the caller can fall
    /// back to the in-place path; callers treat a relocation error as "skip
    /// truncation" rather than failing the commit.
    ///
    /// `limit` bounds how many pages are truncated off the end: `0` means
    /// unbounded (full compaction, the FULL-mode commit behavior), while a
    /// non-zero `limit` stops once that many trailing pages have been removed —
    /// the bound `PRAGMA incremental_vacuum(n)` relies on. The relocation,
    /// reference-fixing, freelist-rebuild and root/lock-byte safety rules are
    /// identical in both cases.
    fn autovacuum_truncate(&mut self, limit: u32) -> Result<()> {
        let usable = self.usable_size() as u32;
        let (want, _max_root) = self.compute_want()?;

        // The set of currently-free data pages (freelist trunk + leaves). We empty
        // the freelist and rebuild it at the end from whatever stays free below the
        // truncation point.
        let mut free: alloc::collections::BTreeSet<u32> = alloc::collections::BTreeSet::new();
        {
            let mut probe: BTreeMap<u32, PtrmapEntry> = BTreeMap::new();
            self.walk_freelist_ptrmap(&mut probe)?;
            for (pg, (kind, _)) in probe {
                if kind == PtrmapType::FreePage {
                    free.insert(pg);
                }
            }
        }
        if free.is_empty() {
            return Ok(()); // nothing to reclaim
        }
        // Detach the existing freelist; we rebuild it from `free` at the end.
        self.header.freelist_trunk = 0;
        self.header.freelist_count = 0;

        // Translation from a moved page's old number to its new (lower) number,
        // so a later fixup that names a since-relocated parent/holder/prev follows
        // it to where its bytes now live.
        let mut moved: BTreeMap<u32, u32> = BTreeMap::new();
        let resolve = |moved: &BTreeMap<u32, u32>, p: u32| -> u32 { *moved.get(&p).unwrap_or(&p) };

        let lock_page = (PENDING_BYTE / self.page_size as u64) as u32 + 1;
        let is_lock = |p: u32| lock_page > 1 && p == lock_page;

        let start = self.page_count;
        let mut last = self.page_count;
        while last > 1 {
            // Stop once `limit` trailing pages have been truncated (0 = unbounded).
            // Counting pages removed from the *end* of the file, not relocations,
            // matches `PRAGMA incremental_vacuum(n)`: each pass shrinks the file by
            // one page, whether that tail page was free/structural (dropped) or a
            // live page displaced by relocating it into a lower free slot.
            if limit != 0 && start - last >= limit {
                break;
            }
            // Trailing structural / free pages can be dropped without moving.
            if ptrmap::is_ptrmap_page(usable, last) {
                // A ptrmap page at the very end tracks only now-gone pages.
                self.overlay.remove(&last);
                last -= 1;
                continue;
            }
            if is_lock(last) {
                break; // never relocate across the lock-byte page
            }
            if free.remove(&last) {
                // A free page at the tail: just drop it.
                self.overlay.remove(&last);
                last -= 1;
                continue;
            }
            // `last` is a live data page. Find the lowest free slot below it that
            // is itself usable (not a ptrmap/lock page).
            let dest = free
                .iter()
                .copied()
                .find(|&f| f < last && !ptrmap::is_ptrmap_page(usable, f) && !is_lock(f));
            let Some(dest) = dest else {
                break; // no free slot below: cannot compact further
            };

            let (kind, parent) = match want.get(&last) {
                Some(&e) => e,
                // A live page with no ptrmap entry would be a bug; bail to the
                // safe in-place path rather than risk corruption.
                None => return Err(Error::Corrupt("relocate: page not in ptrmap".into())),
            };
            if kind == PtrmapType::RootPage {
                break; // do not relocate roots (in-memory schema caches them)
            }

            // Move the page contents into `dest`.
            let contents = self.read_page(last)?;
            self.overlay.insert(dest, contents);
            self.overlay.remove(&last);
            free.remove(&dest);
            moved.insert(last, dest);

            // Fix the one reference that pointed at `last`.
            self.fix_reference(kind, resolve(&moved, parent), last, dest, usable)?;

            last -= 1;
        }

        // Rebuild the freelist from the pages that remain free at or below the new
        // size; free pages above it were truncated away.
        let survivors: Vec<u32> = free.iter().copied().filter(|&pg| pg <= last).collect();
        for pg in survivors {
            self.free_page(pg)?;
        }

        // Apply the new page count and drop any overlay pages past the end.
        self.page_count = last;
        let stale: Vec<u32> = self.overlay.keys().copied().filter(|&p| p > last).collect();
        for p in stale {
            self.overlay.remove(&p);
        }
        Ok(())
    }

    /// Rewrite the single on-page pointer that referenced page `old` so it points
    /// at `new`, given the moved page's ptrmap `kind` and its (already
    /// move-resolved) `holder` page. Used by [`autovacuum_truncate`].
    fn fix_reference(
        &mut self,
        kind: PtrmapType,
        holder: u32,
        old: u32,
        new: u32,
        usable: u32,
    ) -> Result<()> {
        match kind {
            PtrmapType::Btree => {
                // `holder` is the parent b-tree page; find the child pointer == old.
                let mut bytes = self.read_page(holder)?;
                let bt = BtreePage::parse(Page::from_bytes(holder, bytes.clone()))?;
                let mut fixed = false;
                for i in 0..=bt.num_cells() {
                    if bt.child_pointer(i)? == old {
                        let off = bt.child_pointer_offset(i)?;
                        bytes[off..off + 4].copy_from_slice(&new.to_be_bytes());
                        fixed = true;
                        break;
                    }
                }
                if !fixed {
                    return Err(Error::Corrupt("relocate: child pointer not found".into()));
                }
                self.overlay.insert(holder, bytes);
            }
            PtrmapType::Overflow1 => {
                // `holder` is a b-tree page; one of its cells owns the chain whose
                // first page is `old`. Rewrite that cell's first-overflow pointer.
                let mut bytes = self.read_page(holder)?;
                let bt = BtreePage::parse(Page::from_bytes(holder, bytes.clone()))?;
                let mut fixed = false;
                for i in 0..bt.num_cells() {
                    if let Some(off) = bt.cell_overflow_offset(i, usable as usize)?
                        && u32::from_be_bytes([
                            bytes[off],
                            bytes[off + 1],
                            bytes[off + 2],
                            bytes[off + 3],
                        ]) == old
                    {
                        bytes[off..off + 4].copy_from_slice(&new.to_be_bytes());
                        fixed = true;
                        break;
                    }
                }
                if !fixed {
                    return Err(Error::Corrupt("relocate: overflow holder not found".into()));
                }
                self.overlay.insert(holder, bytes);
            }
            PtrmapType::Overflow2 => {
                // `holder` is the previous overflow page; its first 4 bytes are the
                // next-pointer that named `old`.
                let mut bytes = self.read_page(holder)?;
                if u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) != old {
                    return Err(Error::Corrupt("relocate: overflow link mismatch".into()));
                }
                bytes[0..4].copy_from_slice(&new.to_be_bytes());
                self.overlay.insert(holder, bytes);
            }
            PtrmapType::RootPage | PtrmapType::FreePage => {
                return Err(Error::Corrupt("relocate: unexpected page kind".into()));
            }
        }
        Ok(())
    }

    /// Whether page `n` currently exists (in the overlay, WAL, or on disk).
    fn has_page(&self, n: u32) -> bool {
        if self.overlay.contains_key(&n) {
            return true;
        }
        if let Some(w) = &self.wal
            && w.index
                .with(|ix| ix.find_frame(n, w.snapshot.get().mx_frame))
                .is_some()
        {
            return true;
        }
        n >= 1 && n <= self.disk_pages
    }

    /// Read the `rootpage` of every object in `sqlite_schema` (the table b-tree
    /// rooted at page 1), returning `(rootpage, _)` pairs. Rows with a null/zero
    /// root (views, triggers) are skipped.
    fn schema_roots(&self) -> Result<Vec<u32>> {
        let enc = self.header.text_encoding;
        let mut out = Vec::new();
        let mut stack = alloc::vec![1u32];
        while let Some(pg) = stack.pop() {
            let bt = BtreePage::parse(Page::from_bytes(pg, self.read_page(pg)?))?;
            match bt.page_type() {
                PageType::LeafTable => {
                    let usable = self.usable_size();
                    for i in 0..bt.num_cells() {
                        let cell = bt.table_leaf_cell(i, usable)?;
                        let full =
                            crate::btree::cursor::read_payload(self, bt.data(), &cell.payload)?;
                        let cols = crate::format::record::decode_record(&full, enc)?;
                        // sqlite_schema column 3 is `rootpage`.
                        if let Some(crate::value::Value::Integer(r)) = cols.get(3)
                            && *r > 0
                        {
                            out.push(*r as u32);
                        }
                    }
                }
                PageType::InteriorTable => {
                    for i in 0..bt.num_cells() {
                        stack.push(bt.child_pointer(i)?);
                    }
                    stack.push(bt.right_pointer());
                }
                _ => return Err(Error::Corrupt("sqlite_schema is not a table b-tree".into())),
            }
        }
        Ok(out)
    }

    /// Walk the b-tree rooted at `root`, recording the ptrmap entry of every
    /// non-root page below it: child pages get `Btree(parent)`, and each cell's
    /// overflow chain gets `Overflow1(holding page)` then `Overflow2(prev)`.
    fn walk_btree_ptrmap(&self, root: u32, want: &mut BTreeMap<u32, PtrmapEntry>) -> Result<()> {
        let usable = self.usable_size();
        // Iterative DFS over (page, is_root) so we don't record the root as Btree.
        let mut stack = alloc::vec![root];
        while let Some(pg) = stack.pop() {
            let bt = BtreePage::parse(Page::from_bytes(pg, self.read_page(pg)?))?;
            match bt.page_type() {
                PageType::LeafTable => {
                    for i in 0..bt.num_cells() {
                        let ov = bt.table_leaf_cell(i, usable)?.payload.overflow;
                        self.record_overflow_chain(ov, pg, want)?;
                    }
                }
                PageType::LeafIndex => {
                    for i in 0..bt.num_cells() {
                        let ov = bt.index_cell(i, usable)?.payload.overflow;
                        self.record_overflow_chain(ov, pg, want)?;
                    }
                }
                PageType::InteriorTable => {
                    for i in 0..bt.num_cells() {
                        let child = bt.child_pointer(i)?;
                        want.insert(child, (PtrmapType::Btree, pg));
                        stack.push(child);
                    }
                    let r = bt.right_pointer();
                    want.insert(r, (PtrmapType::Btree, pg));
                    stack.push(r);
                }
                PageType::InteriorIndex => {
                    for i in 0..bt.num_cells() {
                        let ov = bt.index_cell(i, usable)?.payload.overflow;
                        self.record_overflow_chain(ov, pg, want)?;
                        let child = bt.child_pointer(i)?;
                        want.insert(child, (PtrmapType::Btree, pg));
                        stack.push(child);
                    }
                    let r = bt.right_pointer();
                    want.insert(r, (PtrmapType::Btree, pg));
                    stack.push(r);
                }
            }
        }
        Ok(())
    }

    /// Record the ptrmap entries for an overflow chain whose owning cell lives on
    /// `holder`: first page `Overflow1(holder)`, the rest `Overflow2(prev)`.
    fn record_overflow_chain(
        &self,
        first: u32,
        holder: u32,
        want: &mut BTreeMap<u32, PtrmapEntry>,
    ) -> Result<()> {
        if first == 0 {
            return Ok(());
        }
        want.insert(first, (PtrmapType::Overflow1, holder));
        let mut prev = first;
        let mut cur = {
            let pg = self.read_page(first)?;
            u32::from_be_bytes([pg[0], pg[1], pg[2], pg[3]])
        };
        while cur != 0 {
            want.insert(cur, (PtrmapType::Overflow2, prev));
            let pg = self.read_page(cur)?;
            prev = cur;
            cur = u32::from_be_bytes([pg[0], pg[1], pg[2], pg[3]]);
        }
        Ok(())
    }

    /// Record `FreePage(0)` for every page on the freelist (trunk pages and the
    /// leaf pages they list).
    fn walk_freelist_ptrmap(&self, want: &mut BTreeMap<u32, PtrmapEntry>) -> Result<()> {
        let mut trunk = self.header.freelist_trunk;
        let mut guard = 0u32;
        let cap = self.header.freelist_count + 8;
        while trunk != 0 {
            guard += 1;
            if guard > cap {
                return Err(Error::Corrupt("freelist trunk cycle".into()));
            }
            want.insert(trunk, (PtrmapType::FreePage, 0));
            let tb = self.read_page(trunk)?;
            let next = u32::from_be_bytes([tb[0], tb[1], tb[2], tb[3]]);
            let leaf_count = u32::from_be_bytes([tb[4], tb[5], tb[6], tb[7]]);
            for i in 0..leaf_count as usize {
                let idx = 8 + 4 * i;
                let leaf = u32::from_be_bytes([tb[idx], tb[idx + 1], tb[idx + 2], tb[idx + 3]]);
                want.insert(leaf, (PtrmapType::FreePage, 0));
            }
            trunk = next;
        }
        Ok(())
    }

    /// Discard all staged changes (ROLLBACK).
    pub fn rollback(&mut self) {
        self.overlay.clear();
        // `header` was mutated in place during the transaction (freelist
        // trunk/count, `largest_root_page`, `user_version`, …); restore it to the
        // last committed state. Skipping this leaves a stale freelist pointer that
        // sends the next allocation reading a page that only existed in the
        // discarded, grown file (a spurious "page N out of range").
        self.header = self.committed_header.clone();
        // The durable page count is the last WAL commit's in WAL mode, else the
        // main file's.
        self.page_count = match &self.wal {
            Some(w) => w.db_size.get(),
            None => self.disk_pages,
        };
        self.savepoints.clear();
        self.release_locks();
    }

    /// Open a savepoint, snapshotting the current staged state.
    pub fn savepoint(&mut self, name: &str) {
        self.savepoints.push(Savepoint {
            name: String::from(name),
            overlay: self.overlay.clone(),
            header: self.header.clone(),
            page_count: self.page_count,
        });
    }

    /// Number of open savepoints.
    pub fn savepoint_depth(&self) -> usize {
        self.savepoints.len()
    }

    /// `RELEASE name`: drop the named savepoint and any nested inside it, keeping
    /// the staged changes. Errors if there is no such savepoint.
    pub fn release_savepoint(&mut self, name: &str) -> Result<()> {
        match self
            .savepoints
            .iter()
            .rposition(|s| s.name.eq_ignore_ascii_case(name))
        {
            Some(idx) => {
                self.savepoints.truncate(idx);
                Ok(())
            }
            None => Err(Error::Error(format!("no such savepoint: {name}"))),
        }
    }

    /// `ROLLBACK TO name`: restore the staged state to the named savepoint and
    /// discard any nested inside it, but keep the savepoint open.
    pub fn rollback_to_savepoint(&mut self, name: &str) -> Result<()> {
        match self
            .savepoints
            .iter()
            .rposition(|s| s.name.eq_ignore_ascii_case(name))
        {
            Some(idx) => {
                let snap = &self.savepoints[idx];
                self.overlay = snap.overlay.clone();
                self.header = snap.header.clone();
                self.page_count = snap.page_count;
                self.savepoints.truncate(idx + 1);
                Ok(())
            }
            None => Err(Error::Error(format!("no such savepoint: {name}"))),
        }
    }

    /// Atomically flush all staged changes to the database file.
    pub fn commit(&mut self) -> Result<()> {
        if self.overlay.is_empty() {
            self.release_locks();
            return Ok(());
        }
        // In WAL mode, append the dirty pages as WAL frames instead.
        if self.wal.is_some() {
            return self.commit_wal();
        }
        // Take the EXCLUSIVE lock for the flush (no other writer or reader).
        self.acquire_exclusive()?;
        // In auto-vacuum mode, first relocate-and-truncate (FULL only), then bring
        // the pointer-map pages up to date before flushing.
        if self.auto_vacuum_on() {
            self.maybe_autovacuum_truncate();
            self.rebuild_ptrmap()?;
        }
        // Refresh header bookkeeping and re-stamp page 1.
        self.header.change_counter = self.header.change_counter.wrapping_add(1);
        self.header.size_in_pages = self.page_count;
        self.header.version_valid_for = self.header.change_counter;
        let mut page1 = self.overlay.get(&1).cloned().unwrap_or(self.read_page(1)?);
        self.header.write_to(&mut page1)?;
        self.overlay.insert(1, page1);

        // 1. Journal the originals of pages that already exist on disk.
        if self.journal.is_some() {
            self.write_journal()?;
        }

        // 2. Write all dirty pages, then make sure the file is exactly sized.
        let page_size = self.page_size as u64;
        let pages: Vec<(u32, Vec<u8>)> =
            self.overlay.iter().map(|(k, v)| (*k, v.clone())).collect();
        for (n, bytes) in &pages {
            self.file.write_all_at(bytes, (*n as u64 - 1) * page_size)?;
        }
        self.file.truncate(self.page_count as u64 * page_size)?;
        self.file.sync()?;

        // 3. Clear the journal — the commit is now durable.
        if let Some(j) = self.journal.as_mut() {
            j.truncate(0)?;
            j.sync()?;
        }

        self.disk_pages = self.page_count;
        // The header just became durable: it is now the committed state a later
        // ROLLBACK must restore to.
        self.committed_header = self.header.clone();
        self.overlay.clear();
        // The main file's contents just changed under the clean read cache; drop
        // it so subsequent reads re-fetch the committed bytes.
        self.read_cache.borrow_mut().clear();
        self.savepoints.clear();
        self.release_locks();
        Ok(())
    }

    /// Whether the database is in WAL mode.
    pub fn wal_mode(&self) -> bool {
        self.wal.is_some()
    }

    /// The auto-vacuum mode recorded in the file header.
    ///
    /// Derived from `largest_root_page` (non-zero ⇒ auto-vacuum on) and
    /// `incremental_vacuum` (non-zero ⇒ INCREMENTAL), matching how SQLite
    /// reports `PRAGMA auto_vacuum`.
    pub fn auto_vacuum(&self) -> AutoVacuum {
        if self.header.largest_root_page == 0 {
            AutoVacuum::None
        } else if self.header.incremental_vacuum != 0 {
            AutoVacuum::Incremental
        } else {
            AutoVacuum::Full
        }
    }

    /// Switch an *empty* database (page 1 only, no user tables) into the given
    /// auto-vacuum `mode` by stamping the header fields, mirroring how SQLite
    /// honours `PRAGMA auto_vacuum=FULL|INCREMENTAL` only before any table is
    /// created. Returns `Ok(true)` if the mode was applied, `Ok(false)` if the
    /// database already contains data (in which case SQLite makes the pragma a
    /// no-op until the next `VACUUM`, which we mirror).
    pub fn set_auto_vacuum_if_empty(&mut self, mode: AutoVacuum) -> Result<bool> {
        // "Empty" means a single page whose schema b-tree has no rows.
        if self.page_count != 1 {
            return Ok(false);
        }
        let p1 = self.read_page(1)?;
        let bt = BtreePage::parse(Page::from_bytes(1, p1))?;
        let non_empty = match bt.page_type() {
            PageType::LeafTable => bt.num_cells() != 0,
            _ => true,
        };
        if non_empty {
            return Ok(false);
        }
        let (largest_root_page, incremental_vacuum) = match mode {
            AutoVacuum::None => (0, 0),
            AutoVacuum::Full => (1, 0),
            AutoVacuum::Incremental => (1, 1),
        };
        let h = self.header_mut();
        h.largest_root_page = largest_root_page;
        h.incremental_vacuum = incremental_vacuum;
        Ok(true)
    }

    /// Switch the database into WAL mode (`PRAGMA journal_mode = WAL`). Stamps the
    /// file header's read/write version = 2 via a normal journaled commit, then
    /// initializes empty WAL state. A no-op if already in WAL mode or if no
    /// `-wal` file was supplied (e.g. a bare in-memory database).
    pub fn set_wal_mode(&mut self) -> Result<bool> {
        if self.wal.is_some() {
            return Ok(true);
        }
        if self.wal_file.is_none() {
            return Ok(false);
        }
        // Persist the WAL version bytes in the main header first.
        if self.header.read_version != 2 || self.header.write_version != 2 {
            self.header.read_version = 2;
            self.header.write_version = 2;
            let mut page1 = self.overlay.get(&1).cloned().unwrap_or(self.read_page(1)?);
            self.header.write_to(&mut page1)?;
            self.overlay.insert(1, page1);
            self.commit()?; // journaled commit of the header change
        }
        // Fresh WAL: truncate any stale -wal and start with new salts.
        if let Some(w) = self.wal_file.as_mut() {
            w.truncate(0)?;
            w.sync()?;
        }
        let salt = (self.header.change_counter as u64)
            .wrapping_mul(0x9E37_79B9)
            .to_be_bytes();
        // Attach to (and reset) the shared wal-index for this path so sibling
        // connections observe the fresh, empty WAL.
        let index = self
            .wal_file
            .as_ref()
            .and_then(|w| w.wal_index())
            .unwrap_or_default();
        let snapshot = index.with(|ix| {
            ix.reset(salt);
            ix.snapshot()
        });
        self.wal = Some(WalRuntime {
            index,
            snapshot: Cell::new(snapshot),
            registered_mark: Cell::new(None),
            db_size: Cell::new(self.page_count),
            salt,
        });
        Ok(true)
    }

    /// Commit the overlay as WAL frames appended to the `-wal` file.
    fn commit_wal(&mut self) -> Result<()> {
        if self.auto_vacuum_on() {
            self.maybe_autovacuum_truncate();
            self.rebuild_ptrmap()?;
        }
        self.header.change_counter = self.header.change_counter.wrapping_add(1);
        self.header.size_in_pages = self.page_count;
        self.header.version_valid_for = self.header.change_counter;
        let mut page1 = self.overlay.get(&1).cloned().unwrap_or(self.read_page(1)?);
        self.header.write_to(&mut page1)?;
        self.overlay.insert(1, page1);

        let page_size = self.page_size;
        let pages: Vec<(u32, Vec<u8>)> =
            self.overlay.iter().map(|(k, v)| (*k, v.clone())).collect();
        let new_count = self.page_count;
        let wal = self.wal.as_mut().expect("wal mode");
        let index = wal.index.clone();
        let file = self.wal_file.as_mut().expect("wal file");

        // Continue whatever the shared index already published (a sibling
        // connection may have written the header and earlier frames); if the WAL
        // is empty, we write the header and start fresh under *our* salts. Reading
        // the shared writer state under the write-intent lock is safe: WAL writers
        // serialize on that lock.
        let (mut offset, mut cksum, salt) = match index.with(|ix| ix.writer_state()) {
            Some((off, ck, salt, _ps)) => (off, ck, salt),
            None => {
                let salt = wal.salt;
                let mut hdr = [0u8; WAL_HDR_LEN];
                hdr[0..4].copy_from_slice(&WAL_MAGIC_LE.to_be_bytes());
                hdr[4..8].copy_from_slice(&3_007_000u32.to_be_bytes()); // format version
                hdr[8..12].copy_from_slice(&(page_size as u32).to_be_bytes());
                hdr[12..16].copy_from_slice(&0u32.to_be_bytes()); // checkpoint sequence
                hdr[16..24].copy_from_slice(&salt);
                let (h0, h1) = super::wal::checksum(false, 0, 0, &hdr[0..24]);
                hdr[24..28].copy_from_slice(&h0.to_be_bytes());
                hdr[28..32].copy_from_slice(&h1.to_be_bytes());
                file.write_all_at(&hdr, 0)?;
                (WAL_HDR_LEN as u64, (h0, h1), salt)
            }
        };

        let n = pages.len();
        let frame_len = WAL_FRAME_HDR_LEN + page_size;
        // Frames written this commit, to publish into the shared index after sync:
        // (page_no, commit_db_size, bytes, next append offset, running checksum).
        let mut appended: Vec<AppendedFrame> = Vec::with_capacity(n);
        for (i, (page_no, data)) in pages.iter().enumerate() {
            // The last frame of the commit carries the post-commit db size.
            let db_size = if i + 1 == n { new_count } else { 0 };
            let mut fhdr = [0u8; WAL_FRAME_HDR_LEN];
            fhdr[0..4].copy_from_slice(&page_no.to_be_bytes());
            fhdr[4..8].copy_from_slice(&db_size.to_be_bytes());
            fhdr[8..16].copy_from_slice(&salt);
            let (c0, c1) = super::wal::checksum(false, cksum.0, cksum.1, &fhdr[0..8]);
            let (c0, c1) = super::wal::checksum(false, c0, c1, data);
            fhdr[16..20].copy_from_slice(&c0.to_be_bytes());
            fhdr[20..24].copy_from_slice(&c1.to_be_bytes());
            let mut frame = Vec::with_capacity(frame_len);
            frame.extend_from_slice(&fhdr);
            frame.extend_from_slice(data);
            file.write_all_at(&frame, offset)?;
            offset += frame_len as u64;
            cksum = (c0, c1);
            appended.push((*page_no, db_size, data.clone(), offset, cksum));
        }
        file.sync()?;

        // Publish the durable frames into the shared wal-index and refresh this
        // connection's snapshot to include its own writes.
        wal.salt = salt;
        let snapshot = index.with(|ix| {
            for (page_no, db_size, data, next_off, ck) in appended {
                ix.append(page_no, db_size, salt, data, next_off, ck, page_size as u32);
            }
            ix.snapshot()
        });
        wal.snapshot.set(snapshot);
        wal.db_size.set(new_count);
        // The header just became durable in the WAL: it is now the committed
        // state a later ROLLBACK must restore to.
        self.committed_header = self.header.clone();
        self.overlay.clear();
        self.savepoints.clear();
        // WAL writers serialize via the write-intent lock taken in write_page;
        // readers stay concurrent. Release it now that the frames are durable.
        self.release_locks();
        Ok(())
    }

    /// Checkpoint: fold all committed WAL frames into the main database file, and
    /// — if no other in-process reader is still pinned to an older WAL snapshot —
    /// reset the `-wal` file and shared index with fresh salts. If a sibling reader
    /// *is* pinned, the frames are still backfilled into the main file (safe: the
    /// main file catches up) but the WAL log is left intact so that reader keeps
    /// resolving the frames its snapshot needs, matching `wal.c`'s "checkpoint
    /// cannot pass an active reader" rule.
    pub fn checkpoint(&mut self) -> Result<()> {
        let Some(wal) = self.wal.as_mut() else {
            return Ok(());
        };
        let page_size = self.page_size as u64;
        let index = wal.index.clone();
        let salt = wal.salt;
        // Snapshot the full log and decide whether it is safe to reset.
        let (frames, full_db_size, can_reset) = index.with(|ix| {
            let mx = ix.mx_frame();
            // Safe to reset only when every pinned reader is already at the full
            // high-water mark (they need nothing that a reset would drop). This
            // connection's own snapshot does not block a reset — it is refreshed
            // below to the post-checkpoint state.
            let others_ok = ix.min_reader_mark().map(|m| m >= mx).unwrap_or(true);
            (ix.checkpoint_pages(), ix.full_db_size(), others_ok)
        });
        if frames.is_empty() {
            return Ok(());
        }
        for (page_no, data) in &frames {
            self.file
                .write_all_at(data, (*page_no as u64 - 1) * page_size)?;
        }
        self.file.truncate(full_db_size as u64 * page_size)?;
        self.file.sync()?;
        self.disk_pages = full_db_size;
        // The checkpoint rewrote the main file; drop the clean read cache.
        self.read_cache.borrow_mut().clear();

        if can_reset {
            // No pinned sibling reader needs the WAL: empty it and restart with a
            // fresh salt so subsequent writes begin a new WAL generation.
            if let Some(w) = self.wal_file.as_mut() {
                w.truncate(0)?;
                w.sync()?;
            }
            let new_salt = {
                let mut s = salt;
                let v = u32::from_be_bytes([s[0], s[1], s[2], s[3]]).wrapping_add(1);
                s[0..4].copy_from_slice(&v.to_be_bytes());
                s
            };
            let snapshot = index.with(|ix| {
                ix.reset(new_salt);
                ix.snapshot()
            });
            self.wal = Some(WalRuntime {
                index,
                snapshot: Cell::new(snapshot),
                registered_mark: Cell::new(None),
                db_size: Cell::new(full_db_size),
                salt: new_salt,
            });
        } else {
            // A sibling reader is still pinned; keep the WAL log intact. This
            // connection now reads from the freshly-checkpointed main file, so pin
            // its snapshot at the current high-water mark (its own reads see the
            // WAL frames it just backfilled, which equal the main file).
            let wal = self.wal.as_mut().expect("wal mode");
            wal.db_size.set(full_db_size);
            wal.snapshot.set(index.with(|ix| ix.snapshot()));
        }
        Ok(())
    }

    /// Replace the entire database file with a freshly-built, compact `image`
    /// (page 1 first, carrying the new header). Used by `VACUUM`. Bypasses the
    /// rollback journal — the rebuild is all-or-nothing on success. In WAL mode
    /// the image is written to the main file and the WAL is reset empty.
    pub fn replace_image(&mut self, image: Vec<Vec<u8>>) -> Result<()> {
        if image.is_empty() || image[0].len() != self.page_size {
            return Err(Error::Error("invalid VACUUM image".into()));
        }
        let ps = self.page_size as u64;
        for (i, bytes) in image.iter().enumerate() {
            self.file.write_all_at(bytes, i as u64 * ps)?;
        }
        self.file.truncate(image.len() as u64 * ps)?;
        self.file.sync()?;
        let count = image.len() as u32;
        self.header = DatabaseHeader::parse(&image[0])?;
        self.disk_pages = count;
        self.page_count = count;
        self.overlay.clear();
        // The whole file was rewritten; drop the clean read cache.
        self.read_cache.borrow_mut().clear();
        // Reset any WAL: its frames now refer to the pre-VACUUM image.
        if let Some(w) = self.wal.as_ref() {
            let index = w.index.clone();
            if let Some(f) = self.wal_file.as_mut() {
                f.truncate(0)?;
                f.sync()?;
            }
            self.header.read_version = 2;
            self.header.write_version = 2;
            let salt = (count as u64).wrapping_mul(0x9E37_79B9).to_be_bytes();
            let snapshot = index.with(|ix| {
                ix.reset(salt);
                ix.snapshot()
            });
            self.wal = Some(WalRuntime {
                index,
                snapshot: Cell::new(snapshot),
                registered_mark: Cell::new(None),
                db_size: Cell::new(count),
                salt,
            });
        }
        // The rebuilt image is now the committed state a later ROLLBACK restores.
        self.committed_header = self.header.clone();
        Ok(())
    }

    /// Attach to the shared wal-index for this database (used on open, ROADMAP
    /// C9c). The **on-disk `-wal` is authoritative**: it is always parsed to
    /// establish the committed state (so a reopen after a crash trusts the disk,
    /// not a stale in-memory index). The shared index is then reconciled — adopted
    /// as-is when it already matches the disk (a live sibling published it in
    /// lockstep), or reseeded from the disk otherwise. Returns the per-connection
    /// [`WalRuntime`] with a fresh read snapshot, or `None` if there is no
    /// committed WAL data.
    fn attach_wal(wal: &mut dyn File, page_size: usize) -> Result<Option<WalRuntime>> {
        let index = wal.wal_index().unwrap_or_default();

        // Parse the on-disk `-wal` (authoritative committed state).
        let size = wal.size()?;
        if size < WAL_HDR_LEN as u64 {
            // No on-disk WAL. If a sibling has a live in-memory index (e.g. an
            // in-memory VFS whose truncate keeps the bytes but the index holds the
            // committed frames), adopt it; otherwise there is nothing to attach.
            return Self::attach_from_index_only(index);
        }
        let mut hdr = [0u8; WAL_HDR_LEN];
        wal.read_exact_at(&mut hdr, 0)?;
        let magic = u32::from_be_bytes([hdr[0], hdr[1], hdr[2], hdr[3]]);
        if magic & 0xFFFF_FFFE != WAL_MAGIC_LE {
            return Self::attach_from_index_only(index);
        }
        let big_endian = (magic & 1) == 1;
        let wal_ps = u32::from_be_bytes([hdr[8], hdr[9], hdr[10], hdr[11]]) as usize;
        if wal_ps != page_size {
            return Self::attach_from_index_only(index);
        }
        let mut salt = [0u8; 8];
        salt.copy_from_slice(&hdr[16..24]);
        let (h0, h1) = super::wal::checksum(big_endian, 0, 0, &hdr[0..24]);
        if h0 != u32::from_be_bytes([hdr[24], hdr[25], hdr[26], hdr[27]])
            || h1 != u32::from_be_bytes([hdr[28], hdr[29], hdr[30], hdr[31]])
        {
            return Self::attach_from_index_only(index);
        }
        let frame_len = WAL_FRAME_HDR_LEN + page_size;
        let (mut s0, mut s1) = (h0, h1);
        let mut off = WAL_HDR_LEN as u64;
        // Every valid frame in commit order: (page_no, commit_db_size, data).
        let mut log: Vec<(u32, u32, Vec<u8>)> = Vec::new();
        // The count/offset/cksum as of the last *committed* frame.
        let mut committed_len = 0usize;
        let mut committed_off = WAL_HDR_LEN as u64;
        let mut committed_cksum = (h0, h1);
        let mut db_size = 0u32;
        while off + frame_len as u64 <= size {
            let mut fhdr = [0u8; WAL_FRAME_HDR_LEN];
            wal.read_exact_at(&mut fhdr, off)?;
            let mut page = vec![0u8; page_size];
            wal.read_exact_at(&mut page, off + WAL_FRAME_HDR_LEN as u64)?;
            if fhdr[8..16] != salt {
                break;
            }
            let (c0, c1) = super::wal::checksum(big_endian, s0, s1, &fhdr[0..8]);
            let (c0, c1) = super::wal::checksum(big_endian, c0, c1, &page);
            if c0 != u32::from_be_bytes([fhdr[16], fhdr[17], fhdr[18], fhdr[19]])
                || c1 != u32::from_be_bytes([fhdr[20], fhdr[21], fhdr[22], fhdr[23]])
            {
                break;
            }
            s0 = c0;
            s1 = c1;
            let page_no = u32::from_be_bytes([fhdr[0], fhdr[1], fhdr[2], fhdr[3]]);
            let commit = u32::from_be_bytes([fhdr[4], fhdr[5], fhdr[6], fhdr[7]]);
            log.push((page_no, commit, page));
            off += frame_len as u64;
            if commit != 0 {
                committed_len = log.len();
                committed_off = off;
                committed_cksum = (s0, s1);
                db_size = commit;
            }
        }
        if committed_len == 0 {
            // The on-disk WAL has no complete commit. Do not adopt a stale index —
            // the disk is authoritative and says nothing is committed.
            return Ok(None);
        }
        // Drop any trailing frames past the last commit (torn/uncommitted tail).
        log.truncate(committed_len);
        let snapshot = index.with(|ix| {
            // Reconcile: keep the shared index only if it already matches the disk
            // exactly (same salt and same committed extent — a live sibling
            // published it). Otherwise the disk is authoritative: reseed.
            let matches_disk = ix.writer_state().is_some_and(|(off, _, s, ps)| {
                off == committed_off && s == salt && ps == page_size as u32
            });
            if !matches_disk {
                ix.seed(log, salt, committed_off, committed_cksum, page_size as u32);
            }
            ix.snapshot()
        });
        Ok(Some(WalRuntime {
            index,
            snapshot: Cell::new(snapshot),
            registered_mark: Cell::new(None),
            db_size: Cell::new(db_size),
            salt,
        }))
    }

    /// Attach purely from a live in-memory shared index when the on-disk `-wal`
    /// carries no parseable committed state (e.g. a `MemoryVfs` whose `-wal` bytes
    /// were reset but whose shared index still holds the sibling's frames).
    /// Returns `None` if the index is also empty.
    fn attach_from_index_only(index: SharedWalIndex) -> Result<Option<WalRuntime>> {
        let (snapshot, db_size, salt) = index.with(|ix| {
            let snap = ix.snapshot();
            let db = ix.snapshot_db_size(snap.mx_frame).unwrap_or(0);
            let salt = ix.writer_state().map(|(_, _, s, _)| s).unwrap_or([0; 8]);
            (snap, db, salt)
        });
        if db_size == 0 {
            return Ok(None);
        }
        Ok(Some(WalRuntime {
            index,
            snapshot: Cell::new(snapshot),
            registered_mark: Cell::new(None),
            db_size: Cell::new(db_size),
            salt,
        }))
    }

    /// Write the SQLite-format rollback journal: the originals of every page
    /// this commit is about to overwrite, in the on-disk byte layout that the
    /// real `sqlite3` (and our own [`recover`](Self::recover)) plays back to undo
    /// an interrupted transaction.
    ///
    /// Layout (all integers big-endian), matching the file-format spec:
    /// * **Header** (zero-padded out to one [`JOURNAL_SECTOR`]): the 8-byte
    ///   [`JOURNAL_MAGIC`]; record count (offset 8); a per-transaction checksum
    ///   nonce (offset 12); the initial database size in pages (offset 16); the
    ///   sector size (offset 20); and the page size (offset 24).
    /// * **Page records** (packed contiguously after the header sector): for each
    ///   saved page, its 4-byte page number, its full original content, then the
    ///   4-byte [`journal_page_checksum`].
    ///
    /// The record count at offset 8 is written *after* the page records, mirroring
    /// SQLite's two-flush commit: the page bodies are durable before the header
    /// advertises how many records to trust, so a crash before the final sync
    /// leaves a count of 0 (an empty, ignorable journal) rather than a header that
    /// promises records that were never written.
    fn write_journal(&mut self) -> Result<()> {
        // Collect originals of pages being overwritten (those already on disk).
        let mut originals: Vec<(u32, Vec<u8>)> = Vec::new();
        for &n in self.overlay.keys() {
            if n <= self.disk_pages {
                let mut buf = vec![0u8; self.page_size];
                self.file
                    .read_exact_at(&mut buf, (n as u64 - 1) * self.page_size as u64)?;
                originals.push((n, buf));
            }
        }
        // A per-transaction nonce: deterministic (no RNG dependency) but varying
        // commit-to-commit via the change counter, so a stale page left over from
        // a previous journal fails this transaction's checksum.
        let nonce = (self.header.change_counter)
            .wrapping_mul(0x9E37_79B1)
            .wrapping_add(self.disk_pages.wrapping_mul(2_654_435_761));
        let page_size = self.page_size as u64;
        let j = self.journal.as_mut().unwrap();
        j.truncate(0)?;

        // Header, padded with zeros out to a full sector.
        let mut hdr = vec![0u8; JOURNAL_SECTOR as usize];
        hdr[0..8].copy_from_slice(&JOURNAL_MAGIC);
        // Record count is filled in after the records are synced (see below).
        hdr[12..16].copy_from_slice(&nonce.to_be_bytes());
        hdr[16..20].copy_from_slice(&self.disk_pages.to_be_bytes());
        hdr[20..24].copy_from_slice(&(JOURNAL_SECTOR as u32).to_be_bytes());
        hdr[24..28].copy_from_slice(&(page_size as u32).to_be_bytes());
        debug_assert_eq!(JOURNAL_HDR_FIELDS, 28);
        j.write_all_at(&hdr, 0)?;

        // Page records begin on the sector boundary.
        let mut off = JOURNAL_SECTOR;
        for (n, bytes) in &originals {
            j.write_all_at(&n.to_be_bytes(), off)?;
            off += 4;
            j.write_all_at(bytes, off)?;
            off += page_size;
            let cksum = journal_page_checksum(nonce, bytes);
            j.write_all_at(&cksum.to_be_bytes(), off)?;
            off += 4;
        }
        // Sync the records, then publish the count and sync the header sector.
        j.sync()?;
        let nrec = originals.len() as u32;
        j.write_all_at(&nrec.to_be_bytes(), 8)?;
        j.sync()?;
        Ok(())
    }

    /// Replay a hot SQLite-format journal onto `file`, restoring the pre-commit
    /// state, then clear it.
    ///
    /// Mirrors SQLite's hot-journal recovery: validate the header magic, read the
    /// initial page count / sector size / page size, then play back each page
    /// record (page number, original content, checksum) into the database file.
    /// Records whose checksum does not match the header nonce are treated as the
    /// torn tail of an interrupted write and stop the replay — everything up to
    /// that point is still rolled back. Finally the database is truncated to its
    /// recorded initial size and the journal is truncated away.
    ///
    /// A journal whose record count is `0` (or whose magic is absent) is not hot
    /// and is ignored. When a segment's count is `-1`/`0xFFFFFFFF` (SQLite's
    /// cache-spill sentinel) or otherwise larger than the remaining file, recovery
    /// reads records until the next segment header or end of file.
    ///
    /// SQLite may append **multiple journal segments** (each a sector-aligned
    /// header followed by its records) when the page cache spills mid-transaction;
    /// every segment shares the same page size and sector size, but carries its
    /// own record count and checksum nonce. Recovery walks each segment in turn,
    /// playing every valid record back, so a large interrupted transaction is
    /// fully rolled back.
    fn recover(file: &mut dyn File, journal: &mut dyn File) -> Result<()> {
        let jsize = journal.size()?;
        if jsize < JOURNAL_SECTOR {
            return Ok(()); // too small to hold a header: nothing to do
        }
        // Read the first header to learn the geometry and the original db size.
        let mut hdr = [0u8; JOURNAL_HDR_FIELDS];
        journal.read_exact_at(&mut hdr, 0)?;
        if hdr[0..8] != JOURNAL_MAGIC {
            return Ok(()); // not a SQLite journal (or zeroed/persist-invalidated)
        }
        let orig_pages = be32(&hdr, 16);
        let sector = be32(&hdr, 20) as u64;
        let page_size = be32(&hdr, 24) as u64;
        if page_size < 512 || !page_size.is_power_of_two() || sector == 0 {
            return Ok(()); // bogus geometry: not a journal we can trust
        }
        // The first segment must carry at least one record to be hot.
        if be32(&hdr, 8) == 0 {
            return Ok(());
        }

        let rec_len = 4 + page_size + 4;
        let mut seg_off = 0u64;
        let mut played_any = false;
        // Walk each journal segment (header + records), sector-aligned.
        while seg_off + JOURNAL_HDR_FIELDS as u64 <= jsize {
            let mut sh = [0u8; JOURNAL_HDR_FIELDS];
            journal.read_exact_at(&mut sh, seg_off)?;
            if sh[0..8] != JOURNAL_MAGIC {
                break; // no further segment
            }
            let nrec = be32(&sh, 8);
            let nonce = be32(&sh, 12);
            // Records begin on the sector boundary after this segment's header.
            let recs_start = seg_off + sector;
            if recs_start > jsize {
                break;
            }
            let avail = (jsize - recs_start) / rec_len;
            // -1/over-large count ⇒ read all that fit before the next thing.
            let want = if nrec as u64 > avail {
                avail
            } else {
                nrec as u64
            };
            let mut off = recs_start;
            let mut played = 0u64;
            for _ in 0..want {
                let mut nb = [0u8; 4];
                journal.read_exact_at(&mut nb, off)?;
                let n = u32::from_be_bytes(nb);
                let mut buf = vec![0u8; page_size as usize];
                journal.read_exact_at(&mut buf, off + 4)?;
                let mut cb = [0u8; 4];
                journal.read_exact_at(&mut cb, off + 4 + page_size)?;
                // A mismatched checksum (or a zero page number) marks the torn
                // tail of an interrupted write: stop here, having rolled back the
                // valid prefix, exactly as SQLite does.
                if n == 0 || u32::from_be_bytes(cb) != journal_page_checksum(nonce, &buf) {
                    break;
                }
                file.write_all_at(&buf, (n as u64 - 1) * page_size)?;
                played_any = true;
                played += 1;
                off += rec_len;
            }
            // Advance to the next segment: past the records, padded up to the
            // next sector boundary. If this segment was truncated early (torn
            // tail) there can be no further valid segment.
            if played < want {
                break;
            }
            let consumed = recs_start + played * rec_len;
            let next = consumed.div_ceil(sector) * sector;
            if next <= seg_off {
                break; // no forward progress: avoid looping
            }
            seg_off = next;
        }

        if !played_any {
            return Ok(());
        }
        // Restore the original length, discarding pages appended by the aborted tx.
        file.truncate(orig_pages as u64 * page_size)?;
        file.sync()?;
        journal.truncate(0)?;
        journal.sync()?;
        Ok(())
    }
}

impl PageSource for WritePager {
    fn page(&self, number: u32) -> Result<Page> {
        Ok(Page::from_bytes(number, self.read_page(number)?))
    }
    fn header(&self) -> &DatabaseHeader {
        &self.header
    }
    fn usable_size(&self) -> usize {
        self.header.usable_size() as usize
    }
    fn page_count(&self) -> u32 {
        // In WAL mode, reads see the snapshot's page count (which a sibling's
        // commit may have advanced beyond this connection's own logical count).
        // While a write transaction is staged (overlay non-empty) the logical
        // count is authoritative — the writer is mid-transaction over its own
        // pages. Outside WAL, the logical count as before.
        match &self.wal {
            Some(w) if self.overlay.is_empty() => w.db_size.get(),
            _ => self.page_count,
        }
    }
    fn revalidate_cache(&self) {
        self.revalidate_read_cache();
    }
}

#[inline]
fn be32(b: &[u8], at: usize) -> u32 {
    u32::from_be_bytes([b[at], b[at + 1], b[at + 2], b[at + 3]])
}

#[inline]
fn put32(b: &mut [u8], at: usize, v: u32) {
    b[at..at + 4].copy_from_slice(&v.to_be_bytes());
}

/// Write an empty table-leaf b-tree header at `offset` within `page`.
fn write_empty_leaf_header(page: &mut [u8], offset: usize, page_size: u32) {
    page[offset] = 0x0d; // leaf table b-tree
    page[offset + 1] = 0; // first freeblock
    page[offset + 2] = 0;
    page[offset + 3] = 0; // num cells = 0
    page[offset + 4] = 0;
    // Cell content area start: top of page (page_size, or 0 if 65536).
    let ccs: u16 = if page_size == 65536 {
        0
    } else {
        page_size as u16
    };
    page[offset + 5] = (ccs >> 8) as u8;
    page[offset + 6] = ccs as u8;
    page[offset + 7] = 0; // fragmented free bytes
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::vfs::{OpenFlags, Vfs, memory::MemoryVfs};

    fn mem_wp() -> WritePager {
        let vfs = MemoryVfs::new();
        let file = vfs.open("db", OpenFlags::READ_WRITE_CREATE).unwrap();
        WritePager::create(file, None, 4096).unwrap()
    }

    #[test]
    fn create_yields_valid_empty_db() {
        let mut wp = mem_wp();
        wp.commit().unwrap();
        // Page 1 parses as a header with one page.
        let p1 = wp.read_page(1).unwrap();
        let h = DatabaseHeader::parse(&p1).unwrap();
        assert_eq!(h.page_size, 4096);
        assert_eq!(h.size_in_pages, 1);
        assert_eq!(p1[100], 0x0d); // empty leaf
    }

    #[test]
    fn auto_vacuum_header_bytes_match_sqlite() {
        // Empirically confirmed against sqlite3 3.50.4: an empty auto-vacuum db
        // sets largest_root_page (offset 52) = 1, with incremental_vacuum
        // (offset 64) = 0 for FULL and = 1 for INCREMENTAL. NONE leaves both 0.
        for (mode, lrp, inc, reported) in [
            (AutoVacuum::None, 0u32, 0u32, AutoVacuum::None),
            (AutoVacuum::Full, 1, 0, AutoVacuum::Full),
            (AutoVacuum::Incremental, 1, 1, AutoVacuum::Incremental),
        ] {
            let vfs = MemoryVfs::new();
            let file = vfs.open("db", OpenFlags::READ_WRITE_CREATE).unwrap();
            let mut wp = WritePager::create_auto_vacuum(file, None, None, 4096, mode).unwrap();
            wp.commit().unwrap();
            let p1 = wp.read_page(1).unwrap();
            assert_eq!(be32(&p1, 52), lrp, "largest_root_page for {mode:?}");
            assert_eq!(be32(&p1, 64), inc, "incremental_vacuum for {mode:?}");
            // The header round-trips and the mode is reported back.
            let h = DatabaseHeader::parse(&p1).unwrap();
            assert_eq!(h.largest_root_page, lrp);
            assert_eq!(h.incremental_vacuum, inc);
            assert_eq!(wp.auto_vacuum(), reported);
            // And a reopen still reports the mode.
            let file = vfs.open("db", OpenFlags::READ_WRITE).unwrap();
            let wp2 = WritePager::open(file, None).unwrap();
            assert_eq!(wp2.auto_vacuum(), reported);
        }
    }

    #[test]
    fn allocate_and_readback() {
        let mut wp = mem_wp();
        let n = wp.allocate_page().unwrap();
        assert_eq!(n, 2);
        let mut img = vec![0u8; 4096];
        img[0] = 0x0d;
        wp.write_page(n, img).unwrap();
        wp.commit().unwrap();
        assert_eq!(wp.page_count(), 2);
        assert_eq!(wp.read_page(2).unwrap()[0], 0x0d);
    }

    #[test]
    fn rollback_discards_overlay() {
        let mut wp = mem_wp();
        wp.commit().unwrap(); // establish baseline (1 page)
        wp.allocate_page().unwrap();
        wp.rollback();
        assert_eq!(wp.page_count(), 1);
    }

    #[test]
    fn journal_recovery_restores_originals() {
        // Simulate a crash: write a journal, corrupt the file, then recover.
        let vfs = MemoryVfs::new();
        {
            let file = vfs.open("db", OpenFlags::READ_WRITE_CREATE).unwrap();
            let jf = vfs
                .open("db-journal", OpenFlags::READ_WRITE_CREATE)
                .unwrap();
            let mut wp = WritePager::create(file, Some(jf), 4096).unwrap();
            wp.commit().unwrap(); // 1-page db on disk, journal cleared
        }
        // Manually craft a SQLite-format journal as if a commit had begun: save
        // page 1's original content with a valid header + checksum.
        let orig_p1 = {
            let f = vfs.open("db", OpenFlags::READ_ONLY).unwrap();
            let mut b = vec![0u8; 4096];
            f.read_exact_at(&mut b, 0).unwrap();
            b
        };
        {
            let mut j = vfs.open("db-journal", OpenFlags::READ_WRITE).unwrap();
            let nonce = 0x1234_5678u32;
            let mut hdr = vec![0u8; JOURNAL_SECTOR as usize];
            hdr[0..8].copy_from_slice(&JOURNAL_MAGIC);
            hdr[8..12].copy_from_slice(&1u32.to_be_bytes()); // nrec
            hdr[12..16].copy_from_slice(&nonce.to_be_bytes());
            hdr[16..20].copy_from_slice(&1u32.to_be_bytes()); // orig pages
            hdr[20..24].copy_from_slice(&(JOURNAL_SECTOR as u32).to_be_bytes());
            hdr[24..28].copy_from_slice(&4096u32.to_be_bytes());
            j.write_all_at(&hdr, 0).unwrap();
            let mut off = JOURNAL_SECTOR;
            j.write_all_at(&1u32.to_be_bytes(), off).unwrap();
            off += 4;
            j.write_all_at(&orig_p1, off).unwrap();
            off += 4096;
            let cksum = journal_page_checksum(nonce, &orig_p1);
            j.write_all_at(&cksum.to_be_bytes(), off).unwrap();
            j.sync().unwrap();
        }
        // Corrupt the live file.
        {
            let mut f = vfs.open("db", OpenFlags::READ_WRITE).unwrap();
            f.write_all_at(&[0xFFu8; 16], 0).unwrap();
        }
        // Reopen: recovery should restore page 1 from the journal.
        let file = vfs.open("db", OpenFlags::READ_WRITE).unwrap();
        let jf = vfs.open("db-journal", OpenFlags::READ_WRITE).unwrap();
        let wp = WritePager::open(file, Some(jf)).unwrap();
        assert_eq!(wp.read_page(1).unwrap(), orig_p1);
    }

    /// A multi-segment journal (two sector-aligned headers, each with its own
    /// records and nonce) is fully played back — covering SQLite's cache-spill
    /// layout.
    #[test]
    fn multi_segment_journal_recovery() {
        let vfs = MemoryVfs::new();
        // Baseline 3-page db.
        {
            let file = vfs.open("db", OpenFlags::READ_WRITE_CREATE).unwrap();
            let jf = vfs
                .open("db-journal", OpenFlags::READ_WRITE_CREATE)
                .unwrap();
            let mut wp = WritePager::create(file, Some(jf), 4096).unwrap();
            for _ in 0..2 {
                let p = wp.allocate_page().unwrap();
                let mut b = vec![0u8; 4096];
                b[0] = 0x0d;
                wp.write_page(p, b).unwrap();
            }
            wp.commit().unwrap();
        }
        // Originals to restore.
        let originals: Vec<(u32, Vec<u8>)> = (1u32..=3)
            .map(|n| {
                let f = vfs.open("db", OpenFlags::READ_ONLY).unwrap();
                let mut b = vec![0u8; 4096];
                f.read_exact_at(&mut b, (n as u64 - 1) * 4096).unwrap();
                (n, b)
            })
            .collect();
        // Mutate the live file so recovery has something to undo.
        {
            let mut f = vfs.open("db", OpenFlags::READ_WRITE).unwrap();
            for n in 1u32..=3 {
                f.write_all_at(&[0xAA; 64], (n as u64 - 1) * 4096).unwrap();
            }
        }
        // Build a two-segment journal: segment A saves pages 1,2; segment B
        // (its own header on the next sector boundary) saves page 3.
        let write_seg =
            |buf: &mut Vec<u8>, base: u64, nonce: u32, recs: &[(u32, &Vec<u8>)], orig: u32| {
                let mut hdr = vec![0u8; JOURNAL_SECTOR as usize];
                hdr[0..8].copy_from_slice(&JOURNAL_MAGIC);
                hdr[8..12].copy_from_slice(&(recs.len() as u32).to_be_bytes());
                hdr[12..16].copy_from_slice(&nonce.to_be_bytes());
                hdr[16..20].copy_from_slice(&orig.to_be_bytes());
                hdr[20..24].copy_from_slice(&(JOURNAL_SECTOR as u32).to_be_bytes());
                hdr[24..28].copy_from_slice(&4096u32.to_be_bytes());
                let need = (base + JOURNAL_SECTOR) as usize;
                if buf.len() < need {
                    buf.resize(need, 0);
                }
                buf[base as usize..need].copy_from_slice(&hdr);
                for (n, data) in recs {
                    buf.extend_from_slice(&n.to_be_bytes());
                    buf.extend_from_slice(data);
                    buf.extend_from_slice(&journal_page_checksum(nonce, data).to_be_bytes());
                }
                // Pad up to the next sector boundary so the next header is aligned.
                while !(buf.len() as u64).is_multiple_of(JOURNAL_SECTOR) {
                    buf.push(0);
                }
            };
        let mut jbytes = Vec::new();
        write_seg(
            &mut jbytes,
            0,
            0x1111_1111,
            &[(1, &originals[0].1), (2, &originals[1].1)],
            3,
        );
        let seg_b_base = jbytes.len() as u64;
        write_seg(
            &mut jbytes,
            seg_b_base,
            0x2222_2222,
            &[(3, &originals[2].1)],
            3,
        );
        {
            let mut j = vfs.open("db-journal", OpenFlags::READ_WRITE).unwrap();
            j.truncate(0).unwrap();
            j.write_all_at(&jbytes, 0).unwrap();
            j.sync().unwrap();
        }
        // Recover and assert all three pages were restored.
        let file = vfs.open("db", OpenFlags::READ_WRITE).unwrap();
        let jf = vfs.open("db-journal", OpenFlags::READ_WRITE).unwrap();
        let wp = WritePager::open(file, Some(jf)).unwrap();
        for (n, data) in &originals {
            assert_eq!(&wp.read_page(*n).unwrap(), data, "page {n} restored");
        }
    }
}