obj-core 1.1.1

//! Hot backup primitives (M11 #92).
//!
//! [`backup_pager_to_path`] is the obj-core entry point the obj
//! crate's `Db::backup_to` dispatches through. The function takes a
//! `&Pager<F>` (the caller has already pinned a [`ReaderSnapshot`])
//! and writes a self-contained `.obj` file at `dest`. Writers may
//! continue against the source pager throughout; their post-snapshot
//! commits do not appear in the destination.
//!
//! See `docs/format.md` § Hot backup for the algorithm this module
//! implements.
//!
//! # Power-of-ten posture
//!
//! - **Rule 2.** The page-copy loop is bounded by
//!   `source.page_count()`; the WAL-overlay loop is bounded by the
//!   snapshot's frozen-view size (which is itself bounded by
//!   `source.page_count()`).
//! - **Rule 4.** The driver is short; per-phase helpers
//!   (`copy_main_file`, `overlay_frozen_view`, `patch_destination_header`)
//!   factor the work.
//! - **Rule 7.** No `unwrap` / `expect` in the production path.
//!   Every syscall is `?`-propagated. On any mid-backup error the
//!   destination file is removed best-effort so a half-written
//!   backup does not linger.

#![forbid(unsafe_code)]

use std::path::Path;

use crate::error::{Error, Result};
use crate::pager::header::{decode_header, encode_header, FileHeader};
use crate::pager::page::{Page, PageId, PAGE_SIZE};
use crate::pager::{Pager, ReaderSnapshot};
use crate::platform::{remove_file_if_exists, FileBackend, FileHandle, SyncMode};

/// Build a self-contained `.obj` file at `dest` carrying the state
/// of `source` as of `snapshot.pinned_lsn()`.
///
/// `snapshot` MUST have been taken against `source` and not yet
/// dropped; the pin keeps the source's WAL frames at-or-below the
/// pinned LSN from being reclaimed while the backup runs.
///
/// # Algorithm
///
/// 1. Refuse to overwrite an existing `dest` (`create_new`).
/// 2. Copy main-file pages `0..source.page_count()` byte-for-byte.
/// 3. Overlay every frame in the snapshot's frozen WAL view onto
///    the destination at its page-id offset.
/// 4. If the snapshot's frozen view carries a page-0 header frame,
///    overlay that on top of the main-file copy of page 0.
/// 5. Patch the destination header: zero `wal_salt`, recompute the
///    header CRC32C.
/// 6. `sync_data(SyncMode::Full)` on the destination.
///
/// On any mid-backup error the destination file is removed best-
/// effort so a half-written backup does not linger.
///
/// # Errors
///
/// - [`Error::BackupDestinationExists`] if `dest` already exists.
/// - [`Error::BackupNotSupportedForMemoryPager`] if `source` is an
///   in-memory pager.
/// - [`Error::Io`] on any syscall failure during the copy.
/// - [`Error::InvalidFormat`] / [`Error::Corruption`] propagated
///   from the source header decode (the source's header bytes are
///   re-encoded with the WAL-staged values applied).
pub fn backup_pager_to_path<F: FileBackend>(
    source: &Pager<F>,
    snapshot: &ReaderSnapshot<F>,
    dest: impl AsRef<Path>,
) -> Result<()> {
    let dest_path = dest.as_ref().to_path_buf();
    if source.is_memory_backed() {
        return Err(Error::BackupNotSupportedForMemoryPager);
    }
    if dest_path.exists() {
        return Err(Error::BackupDestinationExists { path: dest_path });
    }
    let result = run_backup(source, snapshot, &dest_path);
    if result.is_err() {
        // Best-effort cleanup; ignore the result so the original
        // error is the one the caller sees.
        let _ = remove_file_if_exists(&dest_path);
    }
    result
}

fn run_backup<F: FileBackend>(
    source: &Pager<F>,
    snapshot: &ReaderSnapshot<F>,
    dest_path: &Path,
) -> Result<()> {
    let dest_handle = FileHandle::create_new(dest_path)?;
    let page_count = source.page_count();
    dest_handle.set_len(
        page_count
            .checked_mul(PAGE_SIZE as u64)
            .ok_or(Error::InvalidArgument("backup: file size overflow"))?,
    )?;
    // #91: the source main file may be SHORTER than `page_count` — a
    // committed growing transaction advances `page_count` while its
    // fresh pages still live only in the WAL view, until the next
    // checkpoint. Gate the byte-for-byte main-file copy by the source's
    // PHYSICAL high-water (the real on-disk length), NOT by `page_count`;
    // reading a slot the file does not yet hold would `UnexpectedEof`.
    // The pages in `[physical, page_count)` are filled by
    // `overlay_frozen_view` from the snapshot's frozen WAL view (the
    // backup's pinned snapshot guarantees those frames are still
    // resident — checkpoint defers while the pin is live). The
    // destination is still sized to the full `page_count` above, so the
    // overlay lands those fresh pages at their correct offsets.
    let physical_page_count = source.main_physical_page_count()?;
    copy_main_file(source, &dest_handle, physical_page_count)?;
    overlay_frozen_view(snapshot, &dest_handle, page_count)?;
    overlay_frozen_header(snapshot, &dest_handle)?;
    patch_destination_header(&dest_handle)?;
    dest_handle.sync_data(SyncMode::Full)?;
    Ok(())
}

/// Copy every page in `0..physical_page_count` from the source pager's
/// main file to `dest`. Reads bypass the WAL overlay — that's the
/// snapshot's job in [`overlay_frozen_view`].
///
/// #91: the bound is the source's PHYSICAL high-water, not its
/// `page_count`. Pages in `[physical_page_count, page_count)` exist
/// only in the WAL view and are filled by [`overlay_frozen_view`];
/// reading them off the (too-short) main file here would
/// `UnexpectedEof`.
///
/// Power-of-ten Rule 2: bounded by `physical_page_count` (itself
/// bounded by `page_count`); Rule 3: a single page-sized scratch
/// buffer is reused across the loop.
fn copy_main_file<F: FileBackend>(
    source: &Pager<F>,
    dest: &FileHandle,
    physical_page_count: u64,
) -> Result<()> {
    // Page 0 first — it's the header, copied as-is here and patched
    // below in `patch_destination_header`.
    let mut buf = Page::zeroed();
    let page_size_u64 = PAGE_SIZE as u64;
    let mut id_raw: u64 = 0;
    while id_raw < physical_page_count {
        let off = id_raw
            .checked_mul(page_size_u64)
            .ok_or(Error::InvalidArgument("backup: byte-offset overflow"))?;
        if id_raw == 0 {
            source.read_main_file_page_zero(buf.as_bytes_mut())?;
        } else {
            let pid = PageId::new(id_raw)
                .ok_or(Error::InvalidArgument("backup: zero page id (impossible)"))?;
            let page = source.read_main_file_page(pid)?;
            buf.as_bytes_mut().copy_from_slice(page.as_bytes());
        }
        dest.write_all_at(buf.as_bytes(), off)?;
        id_raw = id_raw
            .checked_add(1)
            .ok_or(Error::InvalidArgument("backup: page id overflow"))?;
    }
    Ok(())
}

/// Overlay the snapshot's frozen WAL view onto `dest`. After this
/// returns, every page-id `<= page_count` whose body the snapshot
/// would observe via [`ReaderSnapshot::read_page`] carries that
/// observed body in `dest`.
fn overlay_frozen_view<F: FileBackend>(
    snapshot: &ReaderSnapshot<F>,
    dest: &FileHandle,
    page_count: u64,
) -> Result<()> {
    let page_size_u64 = PAGE_SIZE as u64;
    for (pid, page) in snapshot.frozen_pages() {
        if pid.get() >= page_count {
            // The snapshot pinned a frame for a page id that no
            // longer exists in the source (post-snapshot truncate
            // is impossible in M11, but defensive). Skip it.
            continue;
        }
        let off = pid
            .get()
            .checked_mul(page_size_u64)
            .ok_or(Error::InvalidArgument("backup: byte-offset overflow"))?;
        dest.write_all_at(page.as_bytes(), off)?;
    }
    Ok(())
}

/// Overlay the snapshot's frozen page-0 header (if any) on top of
/// `dest`. This places the WAL-staged catalog root / freelist head /
/// page count into the destination's header BEFORE
/// [`patch_destination_header`] zeros the WAL salt.
fn overlay_frozen_header<F: FileBackend>(
    snapshot: &ReaderSnapshot<F>,
    dest: &FileHandle,
) -> Result<()> {
    if let Some(header_page) = snapshot.frozen_header() {
        dest.write_all_at(header_page.as_bytes(), 0)?;
    }
    Ok(())
}

/// Re-encode the destination's page-0 header with `wal_salt` zeroed
/// and the header CRC recomputed. After this the destination is
/// self-consistent: a `Db::open(dest)` will see no WAL salt and
/// will create a fresh empty WAL on first open.
fn patch_destination_header(dest: &FileHandle) -> Result<()> {
    let mut page = Page::zeroed();
    dest.read_exact_at(page.as_bytes_mut(), 0)?;
    let mut header: FileHeader = decode_header(&page)?;
    header.wal_salt = [0u8; 16];
    encode_header(&header, &mut page);
    dest.write_all_at(page.as_bytes(), 0)?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pager::checksum::write_page_trailer;
    use crate::pager::{Config, Pager};
    use tempfile::TempDir;

    fn pid(n: u64) -> PageId {
        PageId::new(n).expect("non-zero")
    }

    /// A page body with a marker byte and a valid CRC32C trailer — the
    /// shape every real caller (B-tree / catalog) passes to `write_page`.
    fn stamped(marker: u8) -> Page {
        let mut p = Page::zeroed();
        p.as_bytes_mut()[0] = marker;
        write_page_trailer(&mut p);
        p
    }

    /// #91 guardrail 4: a backup taken in the window BETWEEN a growing
    /// commit and its next checkpoint must round-trip the fresh pages —
    /// even though the source main file is physically SHORTER than its
    /// `page_count` (the fresh bodies live only in the WAL view). The
    /// `copy_main_file` loop is gated by the source's PHYSICAL high-water
    /// (so it never reads past the short file's EOF) and
    /// `overlay_frozen_view` fills the WAL-resident fresh pages.
    #[test]
    fn backup_between_growing_commit_and_checkpoint_round_trips() {
        let dir = TempDir::new().expect("tmp");
        let src = dir.path().join("src.obj");
        let dst = dir.path().join("backup.obj");
        // checkpoint_threshold = MAX so the growing commit is NOT
        // auto-checkpointed: the source stays in the #91 window.
        let cfg = Config::default().with_checkpoint_threshold(u64::MAX);

        let (a, b) = {
            let mut p = Pager::open(&src, cfg).expect("open source");
            p.begin_txn();
            // Page `a`: alloc + write + commit + CHECKPOINT, so it lands
            // physically on the main file.
            let a = p.alloc_page().expect("alloc a");
            p.write_page(a, &stamped(0xA1)).expect("write a");
            let _ = p.commit().expect("commit a");
            p.checkpoint().expect("checkpoint a");
            // Page `b`: alloc + write + commit, NO checkpoint — `b` is
            // beyond the physical high-water, resident only in the WAL.
            let b = p.alloc_page().expect("alloc b");
            p.write_page(b, &stamped(0xB2)).expect("write b");
            let _ = p.commit().expect("commit b");

            // Confirm the window: the source main file does NOT physically
            // cover `b` (this is what makes the guardrail load-bearing).
            let physical = p.main_physical_page_count().expect("physical");
            assert!(
                b.get() >= physical,
                "test premise: fresh page `b` must be beyond the physical \
                 high-water (b={}, physical={physical})",
                b.get(),
            );

            // Pin a snapshot and back up in this exact window. The pin
            // keeps `b`'s WAL frame from being reclaimed.
            let snap = p.reader_snapshot().expect("snap");
            backup_pager_to_path(&p, &snap, &dst).expect("backup");
            (a.get(), b.get())
        };

        // Reopen the backup: both pages must read back their bodies. `a`
        // came from the copied main file; `b` came from the overlaid
        // frozen WAL view. Neither path may `UnexpectedEof`.
        let mut bp = Pager::open(&dst, Config::default()).expect("open backup");
        let ra = bp.read_page(pid(a)).expect("read a from backup");
        assert_eq!(ra.as_bytes()[0], 0xA1, "checkpointed page survives backup");
        let rb = bp.read_page(pid(b)).expect("read b from backup");
        assert_eq!(
            rb.as_bytes()[0],
            0xB2,
            "WAL-resident fresh page survives backup via overlay_frozen_view",
        );
    }
}