sqlrite/sql/pager/

wal.rs

//! Write-Ahead Log (WAL) file format.
//!
//! Phase 4b introduces the `.sqlrite-wal` sidecar file. Writes don't go to
//! the main `.sqlrite` file anymore once the WAL is wired in (Phase 4c);
//! instead they append **frames** to this log, and a periodic checkpoint
//! (Phase 4d) applies frames back into the main file.
//!
//! This module is the format layer — header, frame, codec, reader,
//! writer. It doesn't know anything about the `Pager` yet; that wiring is
//! the next slice.
//!
//! **On-disk layout**
//!
//! ```text
//!   byte 0..32   WAL header
//!                   0..8    magic "SQLRWAL\0"
//!                   8..12   format version (u32 LE)
//!                            v1: pre-Phase-11
//!                            v2: Phase 11.2 — adds clock_high_water
//!                                              in bytes 24..32
//!                  12..16   page size     (u32 LE) = 4096
//!                  16..20   salt          (u32 LE) — random on create,
//!                                                    re-rolled per checkpoint
//!                  20..24   checkpoint seq (u32 LE) — bumps per checkpoint
//!                  24..32   v2: clock_high_water (u64 LE) — last
//!                            persisted MVCC logical clock value;
//!                            `crate::mvcc::MvccClock::observe`'d on
//!                            reopen so timestamps don't reuse values
//!                            across restarts.
//!                            v1: reserved / zero (read as 0).
//!
//!   byte 32..    sequence of frames, each `FRAME_SIZE` bytes:
//!                   0..4    page number           (u32 LE)
//!                   4..8    commit-page-count     (u32 LE)
//!                             0 = dirty frame (part of an open write)
//!                            >0 = commit frame; value = page count at commit
//!                   8..12   salt (u32 LE)         — copied from WAL header,
//!                                                    detects truncation / file swap
//!                  12..16   checksum (u32 LE)     — rolling sum over the
//!                                                    frame header bytes
//!                                                    [0..12] + the payload
//!                  16..16+PAGE_SIZE  page bytes
//! ```
//!
//! **Format version compatibility.** v1 WALs (written by pre-Phase-11
//! builds) open cleanly: their reserved bytes are zero, which we
//! interpret as `clock_high_water = 0` — exactly what a fresh-from-
//! checkpoint clock would carry. The next time the WAL is rewritten
//! (any checkpoint, including the auto-checkpoint that fires past the
//! frame threshold) it lands on disk as v2. There's no "you must
//! upgrade your files" step.
//!
//! **Checksum.** A rolling `rotate_left(1) + byte` sum over the
//! concatenation of the frame's first 12 header bytes (page_num,
//! commit-page-count, salt) and its PAGE_SIZE body. Catches bit flips
//! and most multi-byte corruption without pulling in a dep. The 13th
//! through 16th header bytes (the checksum field itself) are excluded
//! from the computation, obviously.
//!
//! **Torn-write recovery.** On open, the reader walks frames from the
//! start and verifies each checksum. The first invalid or incomplete
//! frame marks where the WAL effectively ends; anything past it is
//! treated as if it doesn't exist. Callers learn what's committed vs
//! what's speculative from `Wal::last_commit_offset` / the `is_commit`
//! flag of each scanned frame.

use std::collections::HashMap;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};

use crate::error::{Result, SQLRiteError};
use crate::sql::pager::page::PAGE_SIZE;
use crate::sql::pager::pager::{AccessMode, acquire_lock};

pub const WAL_HEADER_SIZE: usize = 32;
pub const WAL_MAGIC: &[u8; 8] = b"SQLRWAL\0";
/// The version the engine writes today. Phase 11.2 bumped 1 → 2 to
/// Bumped 2 → 3 in Phase 11.9 to mark "may contain MVCC log-record
/// frames" — frames whose `page_num` field carries
/// [`crate::mvcc::MVCC_FRAME_MARKER`] (`u32::MAX`) instead of a
/// real page number. v1 and v2 readers had no special-case for that
/// marker and a pre-Phase-11.9 checkpoint would try to flush it to
/// the main file at offset `u32::MAX * PAGE_SIZE` (way past EOF),
/// which is why the bump is needed.
pub const WAL_FORMAT_VERSION: u32 = 3;
/// Lowest format version we know how to open. v1 had the bytes that
/// now hold `clock_high_water` reserved-as-zero, which is identical
/// to "clock has never been persisted" and round-trips cleanly. v2
/// adds the clock_high_water field but no MVCC frames. v3 adds MVCC
/// frames; downgrading a v3 WAL into a v2 reader would mis-handle
/// the MVCC marker page number — but a fresh `truncate` (every
/// checkpoint) rewrites the header at the engine's current version,
/// so the cross-version exposure is bounded.
pub const WAL_FORMAT_VERSION_MIN_SUPPORTED: u32 = 1;
pub const FRAME_HEADER_SIZE: usize = 16;
pub const FRAME_SIZE: usize = FRAME_HEADER_SIZE + PAGE_SIZE;

/// Parsed WAL header. `page_size` is redundant with the engine's compile-
/// time constant; we persist it for forward-compat and reject anything
/// that doesn't match at open time.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WalHeader {
    pub salt: u32,
    pub checkpoint_seq: u32,
    /// Phase 11.2 — the last MVCC logical-clock value persisted to
    /// this WAL. Rewritten by `truncate` (= every checkpoint) so a
    /// reopen-then-tick can never reuse a timestamp the previous run
    /// already handed out. Always 0 for v1 WALs (the bytes were
    /// reserved-zero before the bump); read back as 0 on any v1 file
    /// that pre-existed this build.
    pub clock_high_water: u64,
}

/// Parsed per-frame header (everything but the page body).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FrameHeader {
    pub page_num: u32,
    pub commit_page_count: u32,
    pub salt: u32,
    pub checksum: u32,
}

impl FrameHeader {
    /// A commit frame is the "transaction barrier": every preceding dirty
    /// frame up to the previous commit frame (or the WAL header) belongs
    /// to the transaction this one seals.
    pub fn is_commit(&self) -> bool {
        self.commit_page_count != 0
    }
}

pub struct Wal {
    // File carries a Debug impl; we don't derive on Wal because we don't
    // want to dump the whole latest_frame map into the default Debug output.
    file: File,
    path: PathBuf,
    header: WalHeader,
    /// Page → byte offset of the LATEST frame carrying that page's
    /// content. Offsets point at the start of the 16-byte frame header.
    /// A reader consults this to resolve a page via the WAL before
    /// falling back to the main DB file (that's Phase 4c).
    latest_frame: HashMap<u32, u64>,
    /// Byte offset just past the last valid commit frame. Anything past
    /// this is uncommitted and should be ignored by readers. Equals
    /// `WAL_HEADER_SIZE` when there's nothing committed yet.
    last_commit_offset: u64,
    /// Post-commit page count carried in the most recent commit frame.
    last_commit_page_count: Option<u32>,
    /// Total valid frames (up to and including `last_commit_offset`).
    /// Used by the checkpointer in Phase 4d to decide whether to run.
    frame_count: usize,
    /// Phase 11.9 — MVCC commit batches recovered from the WAL on
    /// open, in the order they were committed (oldest first).
    /// Populated by `replay_frames` from frames carrying
    /// `page_num = MVCC_FRAME_MARKER`. `Pager::open` exposes the
    /// vector so the engine can replay them into `MvStore` on
    /// reopen.
    recovered_mvcc: Vec<crate::mvcc::MvccCommitBatch>,
}

impl Wal {
    /// Creates a fresh WAL file, truncating any existing one. Writes the
    /// header synchronously so a subsequent `open` sees a valid file even
    /// if the caller panics before appending any frames. Always takes an
    /// exclusive lock — create is a write operation by definition.
    pub fn create(path: &Path) -> Result<Self> {
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(true)
            .open(path)?;
        acquire_lock(&file, path, AccessMode::ReadWrite)?;

        let salt = random_salt();
        let header = WalHeader {
            salt,
            checkpoint_seq: 0,
            clock_high_water: 0,
        };
        let mut wal = Self {
            file,
            path: path.to_path_buf(),
            header,
            latest_frame: HashMap::new(),
            last_commit_offset: WAL_HEADER_SIZE as u64,
            last_commit_page_count: None,
            frame_count: 0,
            recovered_mvcc: Vec::new(),
        };
        wal.write_header()?;
        wal.file.flush()?;
        wal.file.sync_all()?;
        Ok(wal)
    }

    /// Opens an existing WAL file with an exclusive lock (read-write).
    /// Convenience wrapper around [`Wal::open_with_mode`].
    pub fn open(path: &Path) -> Result<Self> {
        Self::open_with_mode(path, AccessMode::ReadWrite)
    }

    /// Opens an existing WAL file with the given access mode. In
    /// `ReadOnly` mode the file descriptor is opened read-only and the
    /// advisory lock is shared — multiple read-only openers may coexist.
    /// Walks every frame from the start, validates checksums, and builds
    /// the in-memory `latest_frame` index. A torn or corrupted frame is
    /// treated as the end of the log — its bytes and anything after stay
    /// on disk but are ignored by reads.
    pub fn open_with_mode(path: &Path, mode: AccessMode) -> Result<Self> {
        let mut file = match mode {
            AccessMode::ReadWrite => OpenOptions::new().read(true).write(true).open(path)?,
            AccessMode::ReadOnly => OpenOptions::new().read(true).open(path)?,
        };
        acquire_lock(&file, path, mode)?;

        let header = read_header(&mut file)?;
        let mut wal = Self {
            file,
            path: path.to_path_buf(),
            header,
            latest_frame: HashMap::new(),
            last_commit_offset: WAL_HEADER_SIZE as u64,
            last_commit_page_count: None,
            frame_count: 0,
            recovered_mvcc: Vec::new(),
        };
        wal.replay_frames()?;
        Ok(wal)
    }

    pub fn header(&self) -> WalHeader {
        self.header
    }

    pub fn frame_count(&self) -> usize {
        self.frame_count
    }

    pub fn last_commit_page_count(&self) -> Option<u32> {
        self.last_commit_page_count
    }

    /// Phase 11.2 — the MVCC logical-clock high-water mark persisted
    /// in this WAL's header. Returns 0 for fresh-from-create WALs and
    /// for v1 WALs (where the bytes were reserved-zero). The Pager
    /// (Phase 11.3) seeds the in-memory `MvccClock` from this on open.
    pub fn clock_high_water(&self) -> u64 {
        self.header.clock_high_water
    }

    /// Phase 11.2 — overrides the in-memory clock high-water value.
    /// `truncate` writes whatever value the WAL is carrying when it
    /// runs, so callers update this just before checkpoint to persist
    /// the latest in-memory clock value. The setter rejects a
    /// non-monotonic update (a value below the existing high-water
    /// mark) — that would either be a bug in the caller or evidence
    /// of a corrupted in-memory clock. Same value is a no-op.
    pub fn set_clock_high_water(&mut self, value: u64) -> Result<()> {
        if value < self.header.clock_high_water {
            return Err(SQLRiteError::General(format!(
                "WAL clock_high_water cannot move backwards: \
                 attempted {value}, current {}",
                self.header.clock_high_water
            )));
        }
        self.header.clock_high_water = value;
        Ok(())
    }

    /// Bulk-loads every committed page from the WAL into `dest`. Used by
    /// `Pager::open` to warm a WAL cache so subsequent reads don't have
    /// to seek back into the WAL file. Uncommitted frames are skipped
    /// (same rule as `read_page`).
    pub fn load_committed_into(
        &mut self,
        dest: &mut HashMap<u32, Box<[u8; PAGE_SIZE]>>,
    ) -> Result<()> {
        // Snapshot the page numbers upfront so we don't hold a borrow of
        // `self` while calling the mutating `read_page`.
        let pages: Vec<u32> = self.latest_frame.keys().copied().collect();
        for page_num in pages {
            if let Some(body) = self.read_page(page_num)? {
                dest.insert(page_num, body);
            }
        }
        Ok(())
    }

    /// Appends a new frame at the current end of file. `commit_page_count`
    /// of `None` writes a dirty (in-progress) frame; `Some(n)` writes a
    /// commit frame carrying the post-commit page count. On commit the
    /// frame is fsync'd; dirty frames are not — torn writes are recovered
    /// by the checksum check on next open.
    pub fn append_frame(
        &mut self,
        page_num: u32,
        content: &[u8; PAGE_SIZE],
        commit_page_count: Option<u32>,
    ) -> Result<()> {
        // Build the header in a buffer so we can checksum + write it
        // atomically alongside the body.
        let mut header_buf = [0u8; FRAME_HEADER_SIZE];
        header_buf[0..4].copy_from_slice(&page_num.to_le_bytes());
        header_buf[4..8].copy_from_slice(&commit_page_count.unwrap_or(0).to_le_bytes());
        header_buf[8..12].copy_from_slice(&self.header.salt.to_le_bytes());
        let sum = compute_checksum(&header_buf[0..12], content);
        header_buf[12..16].copy_from_slice(&sum.to_le_bytes());

        // Frame lands at the current tail.
        let offset = self.file.seek(SeekFrom::End(0))?;
        self.file.write_all(&header_buf)?;
        self.file.write_all(content)?;

        // Commit frames sync; dirty frames are buffered.
        if commit_page_count.is_some() {
            self.file.flush()?;
            self.file.sync_all()?;
        }

        // Update in-memory state — the latest-frame map always points at the
        // newest frame, whether dirty or committed. Readers consult the
        // commit-barrier separately to decide what's visible.
        self.latest_frame.insert(page_num, offset);
        if let Some(pc) = commit_page_count {
            self.last_commit_offset = offset + FRAME_SIZE as u64;
            self.last_commit_page_count = Some(pc);
        }
        self.frame_count += 1;
        Ok(())
    }

    /// Reads the most recent committed copy of a page from the WAL, or
    /// `None` if no committed frame has been written for this page since
    /// the last checkpoint. Uncommitted (dirty) frames are skipped — a
    /// reader must only see committed state.
    pub fn read_page(&mut self, page_num: u32) -> Result<Option<Box<[u8; PAGE_SIZE]>>> {
        let Some(&offset) = self.latest_frame.get(&page_num) else {
            return Ok(None);
        };
        // If this frame sits past the last commit barrier it's
        // uncommitted — not visible.
        if offset + FRAME_SIZE as u64 > self.last_commit_offset {
            return Ok(None);
        }
        let (_hdr, body) = self.read_frame_at(offset)?;
        Ok(Some(body))
    }

    /// Truncates the WAL back to just the header and rolls the salt.
    /// Called by the checkpointer (Phase 4d) once it has applied
    /// accumulated frames to the main file.
    pub fn truncate(&mut self) -> Result<()> {
        self.header.salt = random_salt();
        self.header.checkpoint_seq = self.header.checkpoint_seq.wrapping_add(1);
        self.file.set_len(WAL_HEADER_SIZE as u64)?;
        self.write_header()?;
        self.file.flush()?;
        self.file.sync_all()?;
        self.latest_frame.clear();
        self.last_commit_offset = WAL_HEADER_SIZE as u64;
        self.last_commit_page_count = None;
        self.frame_count = 0;
        // Phase 11.9 — the recovered MVCC batches were a snapshot
        // taken at WAL replay time and represent commits the
        // current process has now seen. Clearing them on truncate
        // matches the legacy `latest_frame.clear()` policy: the
        // WAL is now empty on disk, so the in-memory mirror is
        // empty too. (The engine still holds the *applied* state
        // in `MvStore`; this vector is only the rebuildable-from-
        // WAL portion.)
        self.recovered_mvcc.clear();
        Ok(())
    }

    // ---- internal helpers ------------------------------------------------

    fn write_header(&mut self) -> Result<()> {
        let mut buf = [0u8; WAL_HEADER_SIZE];
        buf[0..8].copy_from_slice(WAL_MAGIC);
        buf[8..12].copy_from_slice(&WAL_FORMAT_VERSION.to_le_bytes());
        buf[12..16].copy_from_slice(&(PAGE_SIZE as u32).to_le_bytes());
        buf[16..20].copy_from_slice(&self.header.salt.to_le_bytes());
        buf[20..24].copy_from_slice(&self.header.checkpoint_seq.to_le_bytes());
        // Phase 11.2: bytes 24..32 carry the MVCC clock high-water
        // mark (u64 LE). v1 WALs left this as zeros, which v2 reads
        // as "no timestamps have been issued" — the same value a
        // newly-created v2 WAL starts at. That's how the v1 → v2
        // upgrade stays seamless.
        buf[24..32].copy_from_slice(&self.header.clock_high_water.to_le_bytes());
        self.file.seek(SeekFrom::Start(0))?;
        self.file.write_all(&buf)?;
        Ok(())
    }

    /// Reads and parses one frame at `offset`. Returns `(header, body)`.
    /// Errors if the frame is truncated or the checksum fails.
    fn read_frame_at(&mut self, offset: u64) -> Result<(FrameHeader, Box<[u8; PAGE_SIZE]>)> {
        self.file.seek(SeekFrom::Start(offset))?;
        let mut header_buf = [0u8; FRAME_HEADER_SIZE];
        self.file.read_exact(&mut header_buf)?;
        let mut body = Box::new([0u8; PAGE_SIZE]);
        self.file.read_exact(body.as_mut())?;

        let page_num = u32::from_le_bytes(header_buf[0..4].try_into().unwrap());
        let commit_page_count = u32::from_le_bytes(header_buf[4..8].try_into().unwrap());
        let salt = u32::from_le_bytes(header_buf[8..12].try_into().unwrap());
        let stored_checksum = u32::from_le_bytes(header_buf[12..16].try_into().unwrap());

        if salt != self.header.salt {
            return Err(SQLRiteError::General(format!(
                "WAL frame at offset {offset}: salt mismatch (expected {:x}, got {:x})",
                self.header.salt, salt
            )));
        }
        let computed = compute_checksum(&header_buf[0..12], &body);
        if computed != stored_checksum {
            return Err(SQLRiteError::General(format!(
                "WAL frame at offset {offset}: bad checksum (expected {stored_checksum:x}, got {computed:x})"
            )));
        }

        Ok((
            FrameHeader {
                page_num,
                commit_page_count,
                salt,
                checksum: stored_checksum,
            },
            body,
        ))
    }

    /// Walks every frame from `WAL_HEADER_SIZE` to end-of-file, validating
    /// each checksum and building `latest_frame`. A frame with a salt
    /// mismatch or bad checksum marks the end of the usable log (earlier
    /// frames are still valid). The last commit frame we successfully
    /// read defines `last_commit_offset`.
    ///
    /// Key invariant: `latest_frame` only holds offsets of *committed*
    /// frames. Dirty frames belonging to an in-progress (or crashed)
    /// transaction accumulate in a pending map and are promoted on the
    /// commit frame that seals them — or discarded if the log ends before
    /// a commit arrives. Without this, an orphan dirty frame for page N
    /// would shadow the previous committed frame for page N, erasing it
    /// from visibility.
    fn replay_frames(&mut self) -> Result<()> {
        use crate::mvcc::{MVCC_FRAME_MARKER, MvccCommitBatch};

        let file_len = self.file.seek(SeekFrom::End(0))?;
        let mut offset = WAL_HEADER_SIZE as u64;
        let mut pending: HashMap<u32, u64> = HashMap::new();
        // Phase 11.9 — MVCC batches waiting for the next commit
        // barrier. They share the legacy page-commit barrier's
        // fsync, so we promote them at the same `is_commit` point
        // page frames are promoted.
        let mut pending_mvcc: Vec<MvccCommitBatch> = Vec::new();
        while offset + FRAME_SIZE as u64 <= file_len {
            match self.read_frame_at(offset) {
                Ok((header, body)) => {
                    self.frame_count += 1;
                    if header.page_num == MVCC_FRAME_MARKER {
                        // MVCC log-record frame. Decode body now —
                        // if it's corrupt the whole frame falls
                        // out of the log (treated the same as a
                        // bad checksum).
                        match MvccCommitBatch::decode(body.as_ref()) {
                            Ok(batch) => pending_mvcc.push(batch),
                            Err(_) => break,
                        }
                    } else {
                        pending.insert(header.page_num, offset);
                    }
                    if header.is_commit() {
                        // Seal: promote all pending page frames
                        // (including this commit frame itself)
                        // into latest_frame, plus all pending
                        // MVCC batches into recovered_mvcc.
                        for (p, o) in pending.drain() {
                            self.latest_frame.insert(p, o);
                        }
                        self.recovered_mvcc.extend(pending_mvcc.drain(..));
                        self.last_commit_offset = offset + FRAME_SIZE as u64;
                        self.last_commit_page_count = Some(header.commit_page_count);
                    }
                    offset += FRAME_SIZE as u64;
                }
                // A bad frame is the torn-write boundary. Keep everything
                // before it.
                Err(_) => break,
            }
        }
        // Anything still in `pending` or `pending_mvcc` belongs to
        // a transaction that never committed (crash, or a writer
        // that died mid-append). Drop it — the legacy and the
        // MVCC writes both fail together, matching the atomicity
        // contract.
        Ok(())
    }

    /// Phase 11.9 — appends an MVCC commit batch as a single WAL
    /// frame whose `page_num` is set to
    /// [`MVCC_FRAME_MARKER`](crate::mvcc::MVCC_FRAME_MARKER).
    ///
    /// Writes the frame as a "dirty" frame
    /// (`commit_page_count = None`) so it doesn't seal a
    /// transaction by itself — it piggybacks on the next legacy
    /// page-commit frame's fsync. This is the durability boundary
    /// for `BEGIN CONCURRENT`: the MVCC batch and the legacy page
    /// updates of the same transaction land in the WAL together
    /// and either both survive the next fsync or both fall away
    /// at the torn-write boundary on reopen.
    ///
    /// `Connection::commit_concurrent` calls this right before
    /// `save_database`, so the same fsync covers both.
    pub fn append_mvcc_batch(&mut self, batch: &crate::mvcc::MvccCommitBatch) -> Result<()> {
        let body = batch.encode()?;
        self.append_frame(crate::mvcc::MVCC_FRAME_MARKER, body.as_ref(), None)
    }

    /// Phase 11.9 — returns the MVCC commit batches recovered from
    /// the WAL on open, in commit order. Empty if no MVCC frames
    /// were present (a brand-new WAL, a pre-11.9 v1/v2 WAL, or one
    /// where the last commit barrier dropped MVCC frames).
    pub fn recovered_mvcc_commits(&self) -> &[crate::mvcc::MvccCommitBatch] {
        &self.recovered_mvcc
    }
}

impl std::fmt::Debug for Wal {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Wal")
            .field("path", &self.path)
            .field("salt", &format_args!("{:#x}", self.header.salt))
            .field("checkpoint_seq", &self.header.checkpoint_seq)
            .field("frame_count", &self.frame_count)
            .field("last_commit_page_count", &self.last_commit_page_count)
            .finish()
    }
}

fn read_header(file: &mut File) -> Result<WalHeader> {
    let mut buf = [0u8; WAL_HEADER_SIZE];
    file.seek(SeekFrom::Start(0))?;
    // read_exact on a short file would bubble up as a generic io error —
    // surface it as "bad magic" instead so the caller gets a consistent
    // diagnosis regardless of whether the file is short-and-garbage or
    // long-and-garbage.
    if file.read_exact(&mut buf).is_err() {
        return Err(SQLRiteError::General(
            "file is not a SQLRite WAL (too short / bad magic)".to_string(),
        ));
    }
    if &buf[0..8] != WAL_MAGIC {
        return Err(SQLRiteError::General(
            "file is not a SQLRite WAL (bad magic)".to_string(),
        ));
    }
    let version = u32::from_le_bytes(buf[8..12].try_into().unwrap());
    // Phase 11.2 — accept v1 (clock bytes are reserved-zero) and v2
    // (clock bytes carry the high-water mark). Anything else is
    // either a corrupt header or a forward-version we don't
    // understand; reject with a clean error.
    if !(WAL_FORMAT_VERSION_MIN_SUPPORTED..=WAL_FORMAT_VERSION).contains(&version) {
        return Err(SQLRiteError::General(format!(
            "unsupported WAL format version {version}; this build reads \
             v{WAL_FORMAT_VERSION_MIN_SUPPORTED}..=v{WAL_FORMAT_VERSION}"
        )));
    }
    let page_size = u32::from_le_bytes(buf[12..16].try_into().unwrap()) as usize;
    if page_size != PAGE_SIZE {
        return Err(SQLRiteError::General(format!(
            "WAL page size {page_size} doesn't match engine's {PAGE_SIZE}"
        )));
    }
    let salt = u32::from_le_bytes(buf[16..20].try_into().unwrap());
    let checkpoint_seq = u32::from_le_bytes(buf[20..24].try_into().unwrap());
    // v1 wrote zeros into bytes 24..32; v2 puts the clock high-water
    // there. Either way, decoding as `u64::from_le_bytes` produces
    // the right value — 0 for v1, the persisted clock for v2.
    let clock_high_water = u64::from_le_bytes(buf[24..32].try_into().unwrap());
    Ok(WalHeader {
        salt,
        checkpoint_seq,
        clock_high_water,
    })
}

fn random_salt() -> u32 {
    // Seeded from SystemTime. Crypto-grade randomness isn't needed — the
    // salt's only job is to make a post-truncate WAL file visibly
    // different from the pre-truncate one (so stale tail bytes can't
    // collide).
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| (d.as_nanos() as u32) ^ (d.as_secs() as u32).rotate_left(13))
        .unwrap_or(0xdeadbeef)
}

/// Rolling sum over `(header_bytes ++ body)`. `rotate_left(1)` per byte
/// makes the checksum order-sensitive, so bit flips AND byte shuffles
/// are detected.
fn compute_checksum(header_bytes: &[u8], body: &[u8; PAGE_SIZE]) -> u32 {
    let mut sum: u32 = 0;
    for &b in header_bytes {
        sum = sum.rotate_left(1).wrapping_add(b as u32);
    }
    for &b in body.iter() {
        sum = sum.rotate_left(1).wrapping_add(b as u32);
    }
    sum
}

#[cfg(test)]
mod tests {
    use super::*;

    fn tmp_wal(name: &str) -> PathBuf {
        let mut p = std::env::temp_dir();
        let pid = std::process::id();
        let nanos = std::time::SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        p.push(format!("sqlrite-wal-{pid}-{nanos}-{name}.wal"));
        p
    }

    fn page(byte: u8) -> Box<[u8; PAGE_SIZE]> {
        let mut b = Box::new([0u8; PAGE_SIZE]);
        for (i, slot) in b.iter_mut().enumerate() {
            *slot = byte.wrapping_add(i as u8);
        }
        b
    }

    #[test]
    fn create_then_open_round_trips_an_empty_wal() {
        let p = tmp_wal("empty");
        let w = Wal::create(&p).unwrap();
        assert_eq!(w.frame_count(), 0);
        assert_eq!(w.last_commit_page_count(), None);
        let salt = w.header().salt;
        drop(w);

        let w2 = Wal::open(&p).unwrap();
        assert_eq!(w2.header().salt, salt);
        assert_eq!(w2.frame_count(), 0);
        assert_eq!(w2.last_commit_page_count(), None);

        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn single_commit_frame_round_trips() {
        let p = tmp_wal("one_frame");
        let mut w = Wal::create(&p).unwrap();
        let content = page(0xab);
        w.append_frame(7, &content, Some(42)).unwrap();
        assert_eq!(w.frame_count(), 1);
        assert_eq!(w.last_commit_page_count(), Some(42));
        drop(w);

        let mut w2 = Wal::open(&p).unwrap();
        assert_eq!(w2.frame_count(), 1);
        assert_eq!(w2.last_commit_page_count(), Some(42));
        let read = w2.read_page(7).unwrap().expect("frame should be visible");
        assert_eq!(read.as_ref(), content.as_ref());
        assert!(
            w2.read_page(99).unwrap().is_none(),
            "untouched page is None"
        );

        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn multi_frame_commits_and_latest_wins() {
        // Three commits to the same page; the latest one should be what
        // read_page returns.
        let p = tmp_wal("latest_wins");
        let mut w = Wal::create(&p).unwrap();
        w.append_frame(1, &page(1), Some(10)).unwrap();
        w.append_frame(1, &page(2), Some(10)).unwrap();
        w.append_frame(1, &page(3), Some(10)).unwrap();
        w.append_frame(2, &page(9), Some(10)).unwrap();
        assert_eq!(w.frame_count(), 4);
        drop(w);

        let mut w2 = Wal::open(&p).unwrap();
        assert_eq!(w2.read_page(1).unwrap().unwrap().as_ref(), page(3).as_ref());
        assert_eq!(w2.read_page(2).unwrap().unwrap().as_ref(), page(9).as_ref());
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn orphan_dirty_tail_preserves_previous_commit() {
        // A dirty frame at the tail with no commit frame following it
        // belongs to a transaction that never sealed. The reader must
        // fall back to the previous committed frame for that page rather
        // than treating the page as absent — otherwise a crash mid-write
        // would erase the page's last durable value.
        let p = tmp_wal("dirty_tail");
        let mut w = Wal::create(&p).unwrap();
        w.append_frame(5, &page(50), Some(10)).unwrap(); // committed V1
        w.append_frame(5, &page(51), None).unwrap(); // orphan dirty V2
        drop(w);

        let mut w2 = Wal::open(&p).unwrap();
        // latest_frame points at the committed offset, NOT the orphan's.
        // read_page returns V1 — the orphan is invisible.
        let got = w2
            .read_page(5)
            .unwrap()
            .expect("committed V1 should still be visible");
        assert_eq!(got.as_ref(), page(50).as_ref());
        // Both frames are still present on disk; frame_count reflects that.
        assert_eq!(w2.frame_count(), 2);
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn uncommitted_frame_for_untouched_page_returns_none() {
        // Contrast with the previous test: a dirty frame for a page that
        // was never committed has no fallback, so read_page returns None.
        let p = tmp_wal("dirty_only");
        let mut w = Wal::create(&p).unwrap();
        w.append_frame(7, &page(70), None).unwrap(); // dirty, no commit
        drop(w);

        let mut w2 = Wal::open(&p).unwrap();
        assert_eq!(w2.read_page(7).unwrap(), None);
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn truncate_resets_to_empty_and_rolls_salt() {
        let p = tmp_wal("truncate");
        let mut w = Wal::create(&p).unwrap();
        w.append_frame(1, &page(11), Some(5)).unwrap();
        w.append_frame(2, &page(22), Some(5)).unwrap();
        let seq_before = w.header().checkpoint_seq;
        let salt_before = w.header().salt;
        w.truncate().unwrap();
        assert_eq!(w.frame_count(), 0);
        assert_eq!(w.last_commit_page_count(), None);
        assert_eq!(w.header().checkpoint_seq, seq_before + 1);
        // Salt is randomly rolled; we can't assert a specific value, but
        // it should almost never match the previous one.
        let _ = salt_before; // the SystemTime-based salt can collide in a
        // theoretical tie; don't assert inequality to avoid flakes.

        // Drop w so its exclusive lock releases before we reopen the same
        // path for verification.
        drop(w);

        // After truncate, read_page returns None for pages we previously
        // wrote — the frames are gone.
        let mut w2 = Wal::open(&p).unwrap();
        assert_eq!(w2.frame_count(), 0);
        assert_eq!(w2.read_page(1).unwrap(), None);
        assert_eq!(w2.read_page(2).unwrap(), None);

        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn bad_magic_file_is_rejected() {
        let p = tmp_wal("bad_magic");
        std::fs::write(&p, b"not a WAL file").unwrap();
        let err = Wal::open(&p).unwrap_err();
        assert!(format!("{err}").contains("bad magic"));
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn corrupt_frame_body_marks_end_of_log() {
        // Write two valid commit frames, then flip a byte in the second
        // frame's body. The reader should accept the first frame and
        // treat the second as end-of-log.
        let p = tmp_wal("bit_flip");
        let mut w = Wal::create(&p).unwrap();
        w.append_frame(1, &page(0x11), Some(5)).unwrap();
        w.append_frame(2, &page(0x22), Some(5)).unwrap();
        drop(w);

        // Flip a byte in the second frame's body. Frame 2's body starts
        // at offset WAL_HEADER_SIZE + FRAME_SIZE + FRAME_HEADER_SIZE.
        let body_offset = WAL_HEADER_SIZE + FRAME_SIZE + FRAME_HEADER_SIZE;
        let mut buf = std::fs::read(&p).unwrap();
        buf[body_offset] ^= 0xff;
        std::fs::write(&p, &buf).unwrap();

        let mut w2 = Wal::open(&p).unwrap();
        // First frame survived.
        assert_eq!(
            w2.read_page(1).unwrap().unwrap().as_ref(),
            page(0x11).as_ref()
        );
        // Second frame was truncated out — its content isn't readable.
        assert_eq!(w2.read_page(2).unwrap(), None);
        assert_eq!(w2.frame_count(), 1);

        let _ = std::fs::remove_file(&p);
    }

    // -----------------------------------------------------------------
    // Phase 11.2 — clock_high_water in the WAL header
    // -----------------------------------------------------------------

    #[test]
    fn fresh_wal_starts_clock_at_zero() {
        let p = tmp_wal("clock_fresh");
        let w = Wal::create(&p).unwrap();
        assert_eq!(w.clock_high_water(), 0);
        drop(w);
        let w2 = Wal::open(&p).unwrap();
        assert_eq!(w2.clock_high_water(), 0);
        let _ = std::fs::remove_file(&p);
    }

    /// Round-trip: setting the clock and triggering `truncate`
    /// (= every checkpoint) must persist the high-water mark across a
    /// reopen. This is the property Phase 11.6 GC relies on — without
    /// it, two transactions on either side of a reopen could share a
    /// timestamp and corrupt visibility.
    #[test]
    fn clock_high_water_round_trips_through_truncate() {
        let p = tmp_wal("clock_truncate");
        let mut w = Wal::create(&p).unwrap();
        // Append a frame so truncate has something to drop. Doesn't
        // matter what the body is — we're testing the header path.
        w.append_frame(1, &page(0xaa), Some(1)).unwrap();
        w.set_clock_high_water(12_345).unwrap();
        w.truncate().unwrap();
        assert_eq!(w.clock_high_water(), 12_345);
        drop(w);

        let w2 = Wal::open(&p).unwrap();
        assert_eq!(w2.clock_high_water(), 12_345);
        let _ = std::fs::remove_file(&p);
    }

    /// `truncate` rewrites the header. Bump-then-truncate-twice must
    /// keep advancing the on-disk value as the in-memory clock moves.
    #[test]
    fn clock_high_water_is_monotonically_persisted_across_truncates() {
        let p = tmp_wal("clock_monotonic_persist");
        let mut w = Wal::create(&p).unwrap();
        w.set_clock_high_water(100).unwrap();
        w.truncate().unwrap();
        w.set_clock_high_water(200).unwrap();
        w.truncate().unwrap();
        drop(w);

        let w2 = Wal::open(&p).unwrap();
        assert_eq!(w2.clock_high_water(), 200);
        let _ = std::fs::remove_file(&p);
    }

    /// Setter rejects a value below the current high-water mark with
    /// a typed error. Same value is accepted as a no-op (test below).
    /// Same-or-greater is the contract every consumer relies on; a
    /// silent saturate-to-current would mask a real bug in the caller.
    #[test]
    fn set_clock_high_water_rejects_regressions() {
        let p = tmp_wal("clock_no_regress");
        let mut w = Wal::create(&p).unwrap();
        w.set_clock_high_water(500).unwrap();
        let err = w.set_clock_high_water(499).unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("backwards") && msg.contains("499") && msg.contains("500"),
            "expected typed regression error, got: {msg}"
        );
        // Idempotent same-value update is fine.
        w.set_clock_high_water(500).unwrap();
        assert_eq!(w.clock_high_water(), 500);
        let _ = std::fs::remove_file(&p);
    }

    /// Synthetic v1 WAL — exact byte layout the previous engine
    /// version wrote. Opening it must succeed and report
    /// `clock_high_water == 0` (the bytes were reserved-zero in v1).
    /// This is the "graceful upgrade" contract.
    #[test]
    fn v1_wal_opens_with_zero_clock() {
        let p = tmp_wal("v1_compat");
        // Hand-build a v1 WAL header. Frame body is intentionally
        // omitted — we're testing header parsing, not frame replay.
        let mut buf = vec![0u8; WAL_HEADER_SIZE];
        buf[0..8].copy_from_slice(WAL_MAGIC);
        buf[8..12].copy_from_slice(&1u32.to_le_bytes()); // version 1
        buf[12..16].copy_from_slice(&(PAGE_SIZE as u32).to_le_bytes());
        buf[16..20].copy_from_slice(&0xdead_beef_u32.to_le_bytes()); // salt
        buf[20..24].copy_from_slice(&7u32.to_le_bytes()); // checkpoint_seq
        // bytes 24..32 left as zero — v1's reserved bytes.
        std::fs::write(&p, &buf).unwrap();

        let w = Wal::open(&p).unwrap();
        assert_eq!(w.header().salt, 0xdead_beef);
        assert_eq!(w.header().checkpoint_seq, 7);
        assert_eq!(
            w.clock_high_water(),
            0,
            "v1 reserved bytes must read as clock=0"
        );
        let _ = std::fs::remove_file(&p);
    }

    /// Forward-versions we don't understand must error cleanly rather
    /// than silently misinterpreting bytes. Picks a version far above
    /// our `WAL_FORMAT_VERSION` so the test stays valid even after
    /// future bumps.
    #[test]
    fn unknown_future_version_is_rejected() {
        let p = tmp_wal("unknown_version");
        let mut buf = vec![0u8; WAL_HEADER_SIZE];
        buf[0..8].copy_from_slice(WAL_MAGIC);
        buf[8..12].copy_from_slice(&999u32.to_le_bytes());
        buf[12..16].copy_from_slice(&(PAGE_SIZE as u32).to_le_bytes());
        std::fs::write(&p, &buf).unwrap();
        let err = Wal::open(&p).unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("unsupported WAL format version") && msg.contains("999"),
            "unexpected error shape: {msg}"
        );
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn partial_trailing_frame_is_ignored() {
        // Write one valid frame, then append a half-frame's worth of
        // random bytes. The reader should stop cleanly at the valid
        // frame.
        let p = tmp_wal("partial");
        let mut w = Wal::create(&p).unwrap();
        w.append_frame(42, &page(42), Some(1)).unwrap();
        drop(w);
        {
            let mut f = OpenOptions::new().write(true).open(&p).unwrap();
            f.seek(SeekFrom::End(0)).unwrap();
            f.write_all(&[0xaa; 2000]).unwrap();
        }
        let mut w2 = Wal::open(&p).unwrap();
        assert_eq!(
            w2.read_page(42).unwrap().unwrap().as_ref(),
            page(42).as_ref()
        );
        assert_eq!(w2.frame_count(), 1);
        let _ = std::fs::remove_file(&p);
    }
}