reddb_file/

embedded.rs

//! Embedded single-file `.rdb` artifact.
//!
//! This module models the promoted path where one durable database artifact
//! carries its superblock pair, internal manifest, WAL reservation, and current
//! store snapshot inside the `.rdb` file itself.

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

use fs2::FileExt;

mod group_commit;
pub use group_commit::{
    EmbeddedGroupCommitOptions, EmbeddedGroupCommitSession, PendingEmbeddedAppend,
};

pub type RdbFileResult<T> = Result<T, RdbFileError>;

pub const DEFAULT_FORMAT_VERSION: u32 = 1;

#[derive(Debug)]
pub enum RdbFileError {
    InvalidOperation(String),
    Io(std::io::Error),
    /// A named zone of the `.rdb` cannot be trusted, so the store does not
    /// open. ADR 0074 §2 requires the zone be named and §4 requires the
    /// operator be pointed at `red salvage` rather than left with a panic.
    ZoneUnrecoverable {
        zone: &'static str,
        path: PathBuf,
    },
}

impl std::fmt::Display for RdbFileError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::InvalidOperation(msg) => write!(f, "INVALID_OPERATION: {msg}"),
            Self::Io(err) => write!(f, "io error: {err}"),
            Self::ZoneUnrecoverable { zone, path } => write!(
                f,
                "{zone} zone of {} failed validation, so the store will not be opened \
                 (opening it could only return data the zone can no longer vouch for). \
                 Run the `SCRUB` statement to classify the fault and `red salvage` to \
                 extract every entity the damage did not touch; `red salvage` never writes \
                 into the damaged file (ADR 0074 §2/§4).",
                path.display()
            ),
        }
    }
}

impl std::error::Error for RdbFileError {}

impl From<std::io::Error> for RdbFileError {
    fn from(err: std::io::Error) -> Self {
        Self::Io(err)
    }
}

fn crc32(data: &[u8]) -> u32 {
    let mut hasher = crc32fast::Hasher::new();
    hasher.update(data);
    hasher.finalize()
}

pub const EMBEDDED_RDB_SUPERBLOCK_SIZE: u64 = 4096;
pub const EMBEDDED_RDB_SUPERBLOCK_0_OFFSET: u64 = 0;
pub const EMBEDDED_RDB_SUPERBLOCK_1_OFFSET: u64 = EMBEDDED_RDB_SUPERBLOCK_SIZE;

/// The manifest zone is a ping-pong pair, exactly like the superblock zone.
///
/// A single in-place manifest could not satisfy "never torn": overwriting it
/// leaves a window in which the durable superblock still names the old
/// checksum while the bytes are already the new ones. Publishing into the
/// *inactive* slot and only then pointing a fresh superblock at it means every
/// valid superblock always references an intact manifest — pre-update or
/// post-update, never a mixture.
pub const EMBEDDED_RDB_MANIFEST_SLOT_SIZE: u64 = 4096;
pub const EMBEDDED_RDB_MANIFEST_0_OFFSET: u64 = EMBEDDED_RDB_SUPERBLOCK_SIZE * 2;
pub const EMBEDDED_RDB_MANIFEST_1_OFFSET: u64 =
    EMBEDDED_RDB_MANIFEST_0_OFFSET + EMBEDDED_RDB_MANIFEST_SLOT_SIZE;
pub const EMBEDDED_RDB_MANIFEST_ZONE_END: u64 =
    EMBEDDED_RDB_MANIFEST_1_OFFSET + EMBEDDED_RDB_MANIFEST_SLOT_SIZE;

const SUPERBLOCK_MAGIC: &[u8; 8] = b"RDBSBLK1";
const MANIFEST_MAGIC: &[u8; 8] = b"RDBMNFS1";
const SUPERBLOCK_VERSION: u32 = 2;
const MANIFEST_VERSION: u32 = 2;
const LEGACY_SUPERBLOCK_VERSION: u32 = 1;
const LEGACY_MANIFEST_VERSION: u32 = 1;
const CHECKSUM_LEN: usize = 4;
const MANIFEST_REGION_BYTES: u64 = EMBEDDED_RDB_MANIFEST_SLOT_SIZE;
const WAL_REGION_BYTES: u64 = 64 * 1024;
const SNAPSHOT_ALIGNMENT: u64 = 4096;
const SNAPSHOT_MAGIC: &[u8; 4] = b"RDST";
const WAL_FRAME_MAGIC: &[u8; 8] = b"RDBEWAL1";
const WAL_FRAME_VERSION: u16 = 2;
const WAL_FRAME_HEADER_BYTES: usize = 8 + 2 + 2 + 8 + 4 + 4 + 4 + 4;
const LEGACY_WAL_FRAME_HEADER_BYTES: usize = 8 + 4 + 4;
const CRASH_INJECT_ENV: &str = "REDDB_EMBEDDED_RDB_CRASH_AT";

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EmbeddedRdbManifest {
    pub version: u32,
    pub wal_region_offset: u64,
    pub wal_region_bytes: u64,
    pub wal_recovery_boundary: u64,
    pub wal_live_bytes: u64,
    pub snapshot_offset: u64,
    pub snapshot_bytes: u64,
    pub snapshot_checksum: u32,
    pub created_at_unix_ms: u128,
    pub checksum: u32,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EmbeddedRdbSuperblock {
    pub copy_index: u8,
    pub generation: u64,
    pub format_version: u32,
    pub manifest_offset: u64,
    pub manifest_len: u64,
    pub manifest_checksum: u32,
    pub wal_region_offset: u64,
    pub wal_region_bytes: u64,
    pub wal_recovery_boundary: u64,
    pub wal_live_bytes: u64,
    pub snapshot_offset: u64,
    pub snapshot_bytes: u64,
    pub snapshot_checksum: u32,
    pub checksum: u32,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EmbeddedRdbOpen {
    pub path: PathBuf,
    pub selected_superblock: EmbeddedRdbSuperblock,
    pub manifest: EmbeddedRdbManifest,
}

#[derive(Debug, Default)]
struct WalScan {
    payloads: Vec<Vec<u8>>,
    next_sequence: u64,
    previous_frame_crc: u32,
    valid_bytes: u64,
}

#[derive(Default)]
struct EmbeddedPathState {
    tail: Option<CachedWalTail>,
}

struct CachedWalTail {
    // Published only after validating this generation's snapshot, or after
    // appending to an unchanged file with an already validated snapshot.
    stamp: FileMutationStamp,
    superblock: EmbeddedRdbSuperblock,
    next_sequence: u64,
    previous_frame_crc: u32,
    valid_bytes: u64,
}

// The superblock detects cooperating writers/checkpoints. File identity and
// mutation times also invalidate replacement or in-place edits. On platforms
// without this identity vocabulary, append keeps the validated full scan.
#[derive(PartialEq, Eq)]
struct FileMutationStamp {
    device: u64,
    inode: u64,
    len: u64,
    modified: (i64, i64),
    changed: (i64, i64),
}

fn file_mutation_stamp(file: &File) -> std::io::Result<Option<FileMutationStamp>> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        let metadata = file.metadata()?;
        Ok(Some(FileMutationStamp {
            device: metadata.dev(),
            inode: metadata.ino(),
            len: metadata.len(),
            modified: (metadata.mtime(), metadata.mtime_nsec()),
            changed: (metadata.ctime(), metadata.ctime_nsec()),
        }))
    }
    #[cfg(not(unix))]
    {
        let _ = file;
        Ok(None)
    }
}

/// Process-lifetime writer fence for an embedded `.rdb`.
///
/// The per-operation lock on the data file only serializes individual appends
/// and checkpoints. A writer process keeps the whole store in memory and
/// checkpoints that image on close, so two processes writing the same file
/// would each publish their own image and the last one to close would
/// silently discard the other's commits. Holding this lock for the writer's
/// lifetime makes the second opener fail instead.
///
/// It lives in a sibling lock file, not on the data file: the per-operation
/// lock takes a fresh handle, which under POSIX `flock` would contend with a
/// lock held through another handle of the same process, and under Windows
/// `LockFileEx` would be denied outright (see `open_inner_with_file`).
pub struct EmbeddedRdbWriterLock {
    #[cfg_attr(not(unix), allow(dead_code))]
    path: PathBuf,
    _file: File,
}

impl EmbeddedRdbWriterLock {
    pub fn acquire(data_path: impl AsRef<Path>) -> RdbFileResult<Self> {
        let data_path = data_path.as_ref();
        let path = crate::layout::embedded_writer_lock_path(data_path);
        if let Some(parent) = path.parent() {
            if !parent.as_os_str().is_empty() {
                fs::create_dir_all(parent)?;
            }
        }
        loop {
            let file = OpenOptions::new()
                .create(true)
                .truncate(false)
                .read(true)
                .write(true)
                .open(&path)?;
            if let Err(err) = file.try_lock_exclusive() {
                if err.raw_os_error() == fs2::lock_contended_error().raw_os_error() {
                    return Err(RdbFileError::InvalidOperation(format!(
                        "{} is already open for writing by another process; an embedded \
                         single-file store has exactly one writer process",
                        data_path.display()
                    )));
                }
                return Err(err.into());
            }
            // The previous owner unlinks the lock file on release. If that
            // happened between our open and our lock, we hold an orphaned
            // inode no later opener can see, so start over on the live path.
            if lock_file_is_live(&file, &path)? {
                return Ok(Self { path, _file: file });
            }
        }
    }
}

impl Drop for EmbeddedRdbWriterLock {
    fn drop(&mut self) {
        // Unlink while still holding the lock so a closed store is back to
        // one operator-visible artifact; `acquire` re-checks the inode to
        // stay correct across this. Without an inode identity check the
        // file is left in place instead.
        #[cfg(unix)]
        {
            let _ = fs::remove_file(&self.path);
        }
    }
}

fn lock_file_is_live(file: &File, path: &Path) -> RdbFileResult<bool> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        let held = file.metadata()?;
        match fs::metadata(path) {
            Ok(current) => Ok(held.dev() == current.dev() && held.ino() == current.ino()),
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(err) => Err(err.into()),
        }
    }
    #[cfg(not(unix))]
    {
        let _ = (file, path);
        Ok(true)
    }
}

pub struct EmbeddedRdbArtifact;

impl EmbeddedRdbArtifact {
    pub fn create(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
        Self::create_with_snapshot(path, &[])
    }

    pub fn create_with_snapshot(
        path: impl AsRef<Path>,
        snapshot: &[u8],
    ) -> RdbFileResult<EmbeddedRdbOpen> {
        let path = path.as_ref();
        if let Some(parent) = path.parent() {
            if !parent.as_os_str().is_empty() {
                fs::create_dir_all(parent)?;
            }
        }

        let created_at_unix_ms = now_unix_ms();
        let wal_region_offset = EMBEDDED_RDB_MANIFEST_ZONE_END;
        let snapshot_offset = wal_region_offset + WAL_REGION_BYTES;
        let manifest = EmbeddedRdbManifest {
            version: MANIFEST_VERSION,
            wal_region_offset,
            wal_region_bytes: WAL_REGION_BYTES,
            wal_recovery_boundary: wal_region_offset,
            wal_live_bytes: 0,
            snapshot_offset,
            snapshot_bytes: snapshot.len() as u64,
            snapshot_checksum: crc32(snapshot),
            created_at_unix_ms,
            checksum: 0,
        };
        let manifest_bytes = encode_manifest(manifest);
        let manifest_checksum = trailer_checksum(&manifest_bytes);

        let mut file = OpenOptions::new()
            .create(true)
            .truncate(true)
            .read(true)
            .write(true)
            .open(path)?;
        file.set_len(snapshot_offset + snapshot.len() as u64)?;
        write_at(&mut file, EMBEDDED_RDB_MANIFEST_0_OFFSET, &manifest_bytes)?;
        if !snapshot.is_empty() {
            write_at(&mut file, snapshot_offset, snapshot)?;
        }

        let base = EmbeddedRdbSuperblock {
            copy_index: 0,
            generation: 1,
            format_version: DEFAULT_FORMAT_VERSION,
            manifest_offset: EMBEDDED_RDB_MANIFEST_0_OFFSET,
            manifest_len: manifest_bytes.len() as u64,
            manifest_checksum,
            wal_region_offset,
            wal_region_bytes: WAL_REGION_BYTES,
            wal_recovery_boundary: wal_region_offset,
            wal_live_bytes: 0,
            snapshot_offset,
            snapshot_bytes: snapshot.len() as u64,
            snapshot_checksum: crc32(snapshot),
            checksum: 0,
        };
        Self::write_superblock_copy(&mut file, &base)?;
        Self::write_superblock_copy(
            &mut file,
            &EmbeddedRdbSuperblock {
                copy_index: 1,
                generation: 2,
                ..base
            },
        )?;
        file.sync_all()?;

        Self::open(path)
    }

    pub fn open(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
        Self::open_inner(path, true)
    }

    fn open_inner(
        path: impl AsRef<Path>,
        validate_snapshot_refs: bool,
    ) -> RdbFileResult<EmbeddedRdbOpen> {
        let path = path.as_ref();
        let mut file = File::open(path)?;
        // Readers wait for an in-flight publication. On Windows an unlocked
        // ReadFile races the writer's mandatory byte-range lock and fails 33.
        FileExt::lock_shared(&file)?;
        Self::open_inner_with_file(&mut file, path, validate_snapshot_refs)
    }

    /// Read the store's roots through an already-open handle.
    ///
    /// The writer path holds an exclusive lock on the data file and must do
    /// all of its I/O through that one handle: `fs2` maps onto `LockFileEx`
    /// on Windows, whose byte-range locks are *mandatory* and scoped to the
    /// handle that took them, so a second `File::open` of the same path is
    /// denied with `ERROR_LOCK_VIOLATION` (os error 33). That is what made
    /// every embedded write fail on Windows while reads kept working
    /// (issues #2265, #2266). POSIX `flock` is advisory and scoped to the
    /// open file description, which is why it never reproduced on Linux.
    ///
    /// Every helper here seeks explicitly, so sharing one cursor across the
    /// reads and writes of a single operation is safe.
    fn open_inner_with_file(
        file: &mut File,
        path: &Path,
        validate_snapshot_refs: bool,
    ) -> RdbFileResult<EmbeddedRdbOpen> {
        // An I/O error propagates as an I/O error; only bytes that were read
        // and failed to decode count towards "the zone is unrecoverable".
        let mut superblocks: Vec<EmbeddedRdbSuperblock> = [
            read_superblock_copy(file, 0)?,
            read_superblock_copy(file, 1)?,
        ]
        .into_iter()
        .flatten()
        .collect();
        superblocks.sort_by_key(|superblock| std::cmp::Reverse(superblock.generation));

        if superblocks.is_empty() {
            return Err(RdbFileError::ZoneUnrecoverable {
                zone: "superblock",
                path: path.to_path_buf(),
            });
        }

        for selected_superblock in superblocks {
            // The manifest a valid superblock names is always intact: it was
            // fsynced into an inactive slot before this generation existed. A
            // checksum failure here is therefore bit rot, not a torn update,
            // and ADR 0074 §2 says it fails the open by name — never falls
            // back to a stale root that would resurrect superseded state.
            let mut manifest = read_manifest(file, selected_superblock).map_err(|_| {
                RdbFileError::ZoneUnrecoverable {
                    zone: "manifest",
                    path: path.to_path_buf(),
                }
            })?;
            manifest.wal_recovery_boundary = selected_superblock.wal_recovery_boundary;
            manifest.wal_live_bytes = selected_superblock.wal_live_bytes;
            if validate_snapshot_refs && !snapshot_reference_valid(file, &manifest)? {
                continue;
            }
            return Ok(EmbeddedRdbOpen {
                path: path.to_path_buf(),
                selected_superblock,
                manifest,
            });
        }

        Err(RdbFileError::ZoneUnrecoverable {
            zone: "snapshot",
            path: path.to_path_buf(),
        })
    }

    pub fn wal_payloads_encoded_len(payloads: &[Vec<u8>]) -> RdbFileResult<u64> {
        let mut len = 0u64;
        for payload in payloads {
            let payload_len = u32::try_from(payload.len()).map_err(|_| {
                RdbFileError::InvalidOperation("embedded wal payload too large".into())
            })?;
            let frame_len = WAL_FRAME_HEADER_BYTES as u64 + payload_len as u64;
            len = len.checked_add(frame_len).ok_or_else(|| {
                RdbFileError::InvalidOperation("embedded wal encoded length overflow".into())
            })?;
        }
        Ok(len)
    }

    pub fn write_snapshot_with_wal_capacity(
        path: impl AsRef<Path>,
        snapshot: &[u8],
        min_wal_bytes: u64,
    ) -> RdbFileResult<EmbeddedRdbOpen> {
        let path = path.as_ref();
        group_commit::wait_for_pending_appends(path)?;
        let path_lock = embedded_path_lock(path);
        let mut path_guard = path_lock
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        path_guard.tail = None;
        // One handle for the whole operation: it carries the exclusive lock,
        // and on Windows that lock would deny any *other* handle this module
        // opened to the same path (see `open_inner_with_file`).
        let mut file = OpenOptions::new().read(true).write(true).open(path)?;
        file.lock_exclusive()?;

        let open = Self::open_inner_with_file(&mut file, path, true)?;
        let wal_scan = match wal_region_start_checked(&open)? {
            Some(wal_start) => scan_wal_with_file(&mut file, &open, wal_start)?,
            None => WalScan {
                next_sequence: 1,
                ..WalScan::default()
            },
        };
        let checkpoint_boundary = wal_boundary_after_live_bytes(&open, wal_scan.valid_bytes)?;
        let wal_region_bytes =
            grow_wal_region_bytes(open.manifest.wal_region_bytes, min_wal_bytes)?;
        let snapshot_checksum = crc32(snapshot);
        // Nothing to publish when the WAL is empty, its region need not grow,
        // and the durable image already holds exactly these bytes. A close
        // used to checkpoint the same image two or three times, each a full
        // write plus fsyncs.
        if open.manifest.wal_live_bytes == 0
            && wal_scan.valid_bytes == 0
            && wal_region_bytes == open.manifest.wal_region_bytes
            && open.manifest.snapshot_bytes == snapshot.len() as u64
            && open.manifest.snapshot_checksum == snapshot_checksum
            && published_snapshot_matches(&mut file, &open, snapshot)?
        {
            FileExt::unlock(&file)?;
            return Ok(open);
        }
        let snapshot_offset = next_snapshot_offset(&mut file, &open, wal_region_bytes, snapshot)?;
        let manifest = EmbeddedRdbManifest {
            wal_region_bytes,
            wal_recovery_boundary: checkpoint_boundary,
            wal_live_bytes: 0,
            snapshot_offset,
            snapshot_bytes: snapshot.len() as u64,
            snapshot_checksum,
            checksum: 0,
            ..open.manifest
        };
        let manifest_bytes = encode_manifest(manifest);
        let manifest_checksum = trailer_checksum(&manifest_bytes);

        // The image may land in a hole below a live snapshot, so only ever
        // extend here: shrinking before publish would cut a live image.
        let snapshot_end = snapshot_offset + snapshot.len() as u64;
        if file.metadata()?.len() < snapshot_end {
            file.set_len(snapshot_end)?;
        }
        if !snapshot.is_empty() {
            write_at(&mut file, snapshot_offset, snapshot)?;
        }
        crash_inject("snapshot_after_image_write");
        file.sync_data()?;
        crash_inject("snapshot_after_image_sync");

        // Publish the manifest into the slot the live superblock does NOT
        // reference, and make it durable before any superblock names it. Until
        // the superblock write below lands, the durable state still roots
        // through the old manifest slot, which these bytes never touched.
        let next_manifest_offset = inactive_manifest_offset(open.selected_superblock)?;
        write_at(&mut file, next_manifest_offset, &manifest_bytes)?;
        crash_inject("snapshot_after_manifest_write");
        file.sync_data()?;
        crash_inject("snapshot_after_manifest_sync");

        let next_copy_index = if open.selected_superblock.copy_index == 0 {
            1
        } else {
            0
        };
        let next_superblock = EmbeddedRdbSuperblock {
            copy_index: next_copy_index,
            generation: open.selected_superblock.generation.saturating_add(1),
            manifest_offset: next_manifest_offset,
            manifest_len: manifest_bytes.len() as u64,
            manifest_checksum,
            wal_region_bytes,
            wal_recovery_boundary: checkpoint_boundary,
            wal_live_bytes: 0,
            snapshot_offset,
            snapshot_bytes: snapshot.len() as u64,
            snapshot_checksum,
            checksum: 0,
            ..open.selected_superblock
        };
        Self::write_superblock_copy(&mut file, &next_superblock)?;
        crash_inject("snapshot_after_superblock_write");
        file.sync_all()?;

        // Published: the two superblock copies now name this image and the
        // previous one. Anything past both (and past the WAL region) is
        // unreachable, so give it back instead of growing on every checkpoint.
        let retained_end = snapshot_end
            .max(open.manifest.snapshot_offset + open.manifest.snapshot_bytes)
            .max(open.manifest.wal_region_offset + wal_region_bytes);
        if file.metadata()?.len() > retained_end {
            file.set_len(retained_end)?;
            file.sync_all()?;
        }
        // Release before re-opening: `Self::open` takes its own handle, and
        // on Windows it could not open one while this lock is held.
        FileExt::unlock(&file)?;
        Self::open(path)
    }

    pub fn open_strict_manifest(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
        let path = path.as_ref();
        let mut file = File::open(path)?;
        FileExt::lock_shared(&file)?;
        let selected_superblock = [
            read_superblock_copy(&mut file, 0)?,
            read_superblock_copy(&mut file, 1)?,
        ]
        .into_iter()
        .flatten()
        .max_by_key(|superblock| superblock.generation)
        .ok_or_else(|| RdbFileError::InvalidOperation("no valid embedded superblock".into()))?;

        let mut manifest = read_manifest(&mut file, selected_superblock)?;
        manifest.wal_recovery_boundary = selected_superblock.wal_recovery_boundary;
        // The superblock is the WAL authority: appends update it without
        // rewriting the manifest zone, so a stale manifest wal_live_bytes
        // would make the scan see an empty region and drop live records.
        manifest.wal_live_bytes = selected_superblock.wal_live_bytes;
        Ok(EmbeddedRdbOpen {
            path: path.to_path_buf(),
            selected_superblock,
            manifest,
        })
    }

    pub fn read_snapshot(open: &EmbeddedRdbOpen) -> RdbFileResult<Option<Vec<u8>>> {
        if open.manifest.snapshot_bytes == 0 {
            return Ok(None);
        }
        let mut file = File::open(&open.path)?;
        FileExt::lock_shared(&file)?;
        let mut bytes = vec![0u8; open.manifest.snapshot_bytes as usize];
        file.seek(SeekFrom::Start(open.manifest.snapshot_offset))?;
        file.read_exact(&mut bytes)?;
        let checksum = crc32(&bytes);
        if checksum != open.manifest.snapshot_checksum {
            return Err(RdbFileError::InvalidOperation(format!(
                "embedded snapshot checksum mismatch: stored {:#010x}, computed {:#010x}",
                open.manifest.snapshot_checksum, checksum
            )));
        }
        if bytes.len() >= SNAPSHOT_MAGIC.len() && &bytes[..SNAPSHOT_MAGIC.len()] != SNAPSHOT_MAGIC {
            return Err(RdbFileError::InvalidOperation(
                "invalid embedded snapshot magic".into(),
            ));
        }
        Ok(Some(bytes))
    }

    pub fn write_snapshot(
        path: impl AsRef<Path>,
        snapshot: &[u8],
    ) -> RdbFileResult<EmbeddedRdbOpen> {
        Self::write_snapshot_with_wal_capacity(path, snapshot, 0)
    }

    pub fn read_wal_payloads(open: &EmbeddedRdbOpen) -> RdbFileResult<Vec<Vec<u8>>> {
        Ok(scan_wal(open)?.payloads)
    }

    pub fn append_wal_payloads(
        path: impl AsRef<Path>,
        payloads: &[Vec<u8>],
    ) -> RdbFileResult<EmbeddedRdbOpen> {
        let path = path.as_ref();
        if payloads.is_empty() {
            return Self::open(path);
        }

        let path_lock = embedded_path_lock(path);
        let mut path_guard = path_lock
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        // Take before any fallible I/O: a failed append must leave no cached
        // state that could describe a partially published mutation.
        let cached_tail = path_guard.tail.take();
        // Single locked handle for the whole append — see
        // `write_snapshot_with_wal_capacity` and `open_inner_with_file`.
        let mut file = OpenOptions::new().read(true).write(true).open(path)?;
        file.lock_exclusive()?;

        let open = Self::open_inner_with_file(&mut file, path, false)?;
        let stamp = file_mutation_stamp(&file)?;
        let cached_tail = cached_tail.filter(|tail| {
            stamp.as_ref() == Some(&tail.stamp) && tail.superblock == open.selected_superblock
        });
        let snapshot_validated = cached_tail.is_some();
        let wal_scan = match cached_tail {
            Some(tail) => WalScan {
                next_sequence: tail.next_sequence,
                previous_frame_crc: tail.previous_frame_crc,
                valid_bytes: tail.valid_bytes,
                payloads: Vec::new(),
            },
            None => match wal_region_start_checked(&open)? {
                Some(wal_start) => scan_wal_with_file(&mut file, &open, wal_start)?,
                None => WalScan {
                    next_sequence: 1,
                    ..WalScan::default()
                },
            },
        };
        // Recovery may salvage a valid prefix, but appending after a corrupt
        // published frame would acknowledge an unreachable new commit.
        if wal_scan.valid_bytes != open.manifest.wal_live_bytes {
            return Err(RdbFileError::ZoneUnrecoverable {
                zone: "wal",
                path: path.to_path_buf(),
            });
        }
        let mut sequence = wal_scan.next_sequence;
        let mut previous_frame_crc = wal_scan.previous_frame_crc;
        let mut encoded = Vec::new();
        for payload in payloads {
            let (frame, frame_crc) = encode_wal_frame(sequence, previous_frame_crc, payload)?;
            encoded.extend_from_slice(&frame);
            previous_frame_crc = frame_crc;
            sequence = sequence.saturating_add(1);
        }

        let encoded_len = encoded.len() as u64;
        let free_bytes = open
            .manifest
            .wal_region_bytes
            .checked_sub(wal_scan.valid_bytes)
            .ok_or_else(|| {
                RdbFileError::InvalidOperation("embedded wal live bytes overflow".into())
            })?;
        if encoded_len > free_bytes {
            return Err(RdbFileError::InvalidOperation(
                "embedded wal region full".into(),
            ));
        }
        let next_boundary =
            wal_boundary_after_live_bytes(&open, wal_scan.valid_bytes + encoded_len)?;

        write_circular_wal_bytes(&mut file, &open, &encoded)?;
        crash_inject("wal_after_frame_write");
        file.sync_data()?;
        crash_inject("wal_after_frame_sync");

        let next_copy_index = if open.selected_superblock.copy_index == 0 {
            1
        } else {
            0
        };
        let next_superblock = EmbeddedRdbSuperblock {
            copy_index: next_copy_index,
            generation: open.selected_superblock.generation.saturating_add(1),
            wal_recovery_boundary: next_boundary,
            wal_live_bytes: wal_scan.valid_bytes + encoded_len,
            checksum: 0,
            ..open.selected_superblock
        };
        Self::write_superblock_copy(&mut file, &next_superblock)?;
        crash_inject("wal_after_superblock_write");
        file.sync_all()?;
        // A warm append writes only WAL bytes and the alternate superblock:
        // it does not change the snapshot. The matching identity/mutation
        // stamp above proves no intervening file edit since its validation.
        // Cache misses (including unsupported platforms) retain the full
        // checksum pass. Normal open/read/scrub always verify snapshot bytes;
        // this reuse is not a substitute for detecting bit rot on read.
        let next_open = Self::open_inner_with_file(&mut file, path, !snapshot_validated)?;
        let next_stamp = file_mutation_stamp(&file)?;
        FileExt::unlock(&file)?;
        if let Some(stamp) = next_stamp
            .filter(|_| next_open.selected_superblock.generation == next_superblock.generation)
        {
            path_guard.tail = Some(CachedWalTail {
                stamp,
                superblock: next_open.selected_superblock,
                next_sequence: sequence,
                previous_frame_crc,
                valid_bytes: wal_scan.valid_bytes + encoded_len,
            });
        }
        Ok(next_open)
    }

    pub fn write_superblock_copy(
        file: &mut File,
        superblock: &EmbeddedRdbSuperblock,
    ) -> RdbFileResult<()> {
        let offset = superblock_offset(superblock.copy_index)?;
        write_at(file, offset, &encode_superblock(*superblock)?)?;
        Ok(())
    }
}

fn read_superblock_copy(
    file: &mut File,
    copy_index: u8,
) -> RdbFileResult<Option<EmbeddedRdbSuperblock>> {
    let offset = superblock_offset(copy_index)?;
    let mut bytes = vec![0u8; EMBEDDED_RDB_SUPERBLOCK_SIZE as usize];
    file.seek(SeekFrom::Start(offset))?;
    file.read_exact(&mut bytes)?;
    Ok(decode_superblock(copy_index, &bytes).ok())
}

/// The manifest slot the given superblock does not reference.
///
/// With two superblock copies and two manifest slots this is always the slot
/// belonging to the *stale* superblock copy — the same copy the update is about
/// to overwrite. That coincidence is what makes the pair safe: a crash can only
/// ever clobber the manifest of a superblock that was already being replaced.
fn inactive_manifest_offset(superblock: EmbeddedRdbSuperblock) -> RdbFileResult<u64> {
    match superblock.manifest_offset {
        EMBEDDED_RDB_MANIFEST_0_OFFSET => Ok(EMBEDDED_RDB_MANIFEST_1_OFFSET),
        EMBEDDED_RDB_MANIFEST_1_OFFSET => Ok(EMBEDDED_RDB_MANIFEST_0_OFFSET),
        other => Err(RdbFileError::InvalidOperation(format!(
            "superblock names a manifest offset outside the zone: {other}"
        ))),
    }
}

fn read_manifest(
    file: &mut File,
    superblock: EmbeddedRdbSuperblock,
) -> RdbFileResult<EmbeddedRdbManifest> {
    if superblock.manifest_len < CHECKSUM_LEN as u64
        || superblock.manifest_len > MANIFEST_REGION_BYTES
    {
        return Err(RdbFileError::InvalidOperation(format!(
            "invalid embedded manifest length {}",
            superblock.manifest_len
        )));
    }
    // A superblock that points outside the manifest zone is a misdirected or
    // forged write; never chase the pointer.
    if !matches!(
        superblock.manifest_offset,
        EMBEDDED_RDB_MANIFEST_0_OFFSET | EMBEDDED_RDB_MANIFEST_1_OFFSET
    ) {
        return Err(RdbFileError::InvalidOperation(format!(
            "superblock names a manifest offset outside the zone: {}",
            superblock.manifest_offset
        )));
    }

    let mut bytes = vec![0u8; superblock.manifest_len as usize];
    file.seek(SeekFrom::Start(superblock.manifest_offset))?;
    file.read_exact(&mut bytes)?;
    let checksum = trailer_checksum(&bytes);
    if checksum != superblock.manifest_checksum {
        return Err(RdbFileError::InvalidOperation(format!(
            "embedded manifest checksum mismatch: stored {:#010x}, computed {:#010x}",
            superblock.manifest_checksum, checksum
        )));
    }
    decode_manifest(&bytes)
}

fn snapshot_reference_valid(
    file: &mut File,
    manifest: &EmbeddedRdbManifest,
) -> RdbFileResult<bool> {
    if manifest.snapshot_bytes == 0 {
        return Ok(true);
    }
    let snapshot_end = manifest
        .snapshot_offset
        .checked_add(manifest.snapshot_bytes)
        .ok_or_else(|| RdbFileError::InvalidOperation("embedded snapshot end overflow".into()))?;
    if snapshot_end > file.metadata()?.len() {
        return Ok(false);
    }

    let mut bytes = vec![0u8; manifest.snapshot_bytes as usize];
    file.seek(SeekFrom::Start(manifest.snapshot_offset))?;
    if file.read_exact(&mut bytes).is_err() {
        return Ok(false);
    }
    if crc32(&bytes) != manifest.snapshot_checksum {
        return Ok(false);
    }
    if bytes.len() >= SNAPSHOT_MAGIC.len() && &bytes[..SNAPSHOT_MAGIC.len()] != SNAPSHOT_MAGIC {
        return Ok(false);
    }
    Ok(true)
}

/// Byte-for-byte comparison with the published image; the checksum match
/// that gates this call is not proof of equality on its own.
fn published_snapshot_matches(
    file: &mut File,
    open: &EmbeddedRdbOpen,
    snapshot: &[u8],
) -> RdbFileResult<bool> {
    if snapshot.is_empty() {
        return Ok(true);
    }
    let mut published = vec![0u8; snapshot.len()];
    file.seek(SeekFrom::Start(open.manifest.snapshot_offset))?;
    file.read_exact(&mut published)?;
    Ok(published == snapshot)
}

fn grow_wal_region_bytes(current: u64, min_required: u64) -> RdbFileResult<u64> {
    let mut next = current.max(WAL_REGION_BYTES);
    while next < min_required {
        next = next.checked_mul(2).ok_or_else(|| {
            RdbFileError::InvalidOperation("embedded wal region size overflow".into())
        })?;
    }
    Ok(next)
}

fn embedded_path_lock(path: &Path) -> Arc<Mutex<EmbeddedPathState>> {
    static LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<EmbeddedPathState>>>>> =
        OnceLock::new();
    let key = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
    let mut locks = LOCKS
        .get_or_init(|| Mutex::new(HashMap::new()))
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    locks
        .entry(key)
        .or_insert_with(|| Arc::new(Mutex::new(EmbeddedPathState::default())))
        .clone()
}

/// Lowest aligned offset past the WAL region where `snapshot` overlaps no
/// image either superblock copy still names.
///
/// Both images stay intact until the new superblock is durable: the selected
/// one is the published state, and the other copy's is the fallback when the
/// newer superblock or its image fails validation. Any other bytes past the
/// WAL region are unreachable and may be reused.
fn next_snapshot_offset(
    file: &mut File,
    open: &EmbeddedRdbOpen,
    wal_region_bytes: u64,
    snapshot: &[u8],
) -> RdbFileResult<u64> {
    let base = open
        .manifest
        .wal_region_offset
        .checked_add(wal_region_bytes)
        .ok_or_else(|| {
            RdbFileError::InvalidOperation("embedded snapshot offset overflow".into())
        })?;
    if open.manifest.snapshot_bytes == 0 && snapshot.is_empty() {
        return Ok(base);
    }

    let other_copy = if open.selected_superblock.copy_index == 0 {
        1
    } else {
        0
    };
    let mut live = vec![(open.manifest.snapshot_offset, open.manifest.snapshot_bytes)];
    if let Some(other) = read_superblock_copy(file, other_copy)? {
        live.push((other.snapshot_offset, other.snapshot_bytes));
    }
    let mut live_ranges = Vec::with_capacity(live.len());
    for (offset, bytes) in live {
        if bytes == 0 {
            continue;
        }
        let end = offset.checked_add(bytes).ok_or_else(|| {
            RdbFileError::InvalidOperation("embedded snapshot end overflow".into())
        })?;
        live_ranges.push((offset, end));
    }

    let len = snapshot.len() as u64;
    let mut candidates = vec![align_up(base, SNAPSHOT_ALIGNMENT)?];
    for &(_, end) in &live_ranges {
        candidates.push(align_up(end.max(base), SNAPSHOT_ALIGNMENT)?);
    }
    candidates.sort_unstable();
    for &candidate in &candidates {
        let end = candidate.checked_add(len).ok_or_else(|| {
            RdbFileError::InvalidOperation("embedded snapshot end overflow".into())
        })?;
        if live_ranges
            .iter()
            .all(|&(start, live_end)| end <= start || candidate >= live_end)
        {
            return Ok(candidate);
        }
    }
    // The largest candidate starts past every live image, so it always fits.
    Ok(candidates[candidates.len() - 1])
}

fn align_up(value: u64, alignment: u64) -> RdbFileResult<u64> {
    if alignment == 0 {
        return Ok(value);
    }
    let remainder = value % alignment;
    if remainder == 0 {
        return Ok(value);
    }
    value
        .checked_add(alignment - remainder)
        .ok_or_else(|| RdbFileError::InvalidOperation("embedded alignment overflow".into()))
}

/// Validate the WAL geometry recorded in the manifest and return the region
/// start. Shared by [`scan_wal`] and the writer path so both reject the same
/// malformed manifests.
fn wal_region_start_checked(open: &EmbeddedRdbOpen) -> RdbFileResult<Option<u64>> {
    let wal_start = open.manifest.wal_region_offset;
    let wal_end = wal_start
        .checked_add(open.manifest.wal_region_bytes)
        .ok_or_else(|| RdbFileError::InvalidOperation("embedded wal end overflow".into()))?;
    let append_boundary = open.manifest.wal_recovery_boundary;
    if append_boundary < wal_start || append_boundary > wal_end {
        return Err(RdbFileError::InvalidOperation(format!(
            "invalid embedded wal boundary {append_boundary}"
        )));
    }
    if open.manifest.wal_live_bytes > open.manifest.wal_region_bytes {
        return Err(RdbFileError::InvalidOperation(format!(
            "invalid embedded wal live bytes {}",
            open.manifest.wal_live_bytes
        )));
    }
    if open.manifest.wal_live_bytes == 0 {
        return Ok(None);
    }
    Ok(Some(wal_start))
}

/// Scan the WAL, opening the file for reading. Callers already holding a
/// handle (the writer path) use [`scan_wal_with_file`] instead.
fn scan_wal(open: &EmbeddedRdbOpen) -> RdbFileResult<WalScan> {
    let wal_start = open.manifest.wal_region_offset;
    let wal_end = wal_start
        .checked_add(open.manifest.wal_region_bytes)
        .ok_or_else(|| RdbFileError::InvalidOperation("embedded wal end overflow".into()))?;
    let append_boundary = open.manifest.wal_recovery_boundary;
    if append_boundary < wal_start || append_boundary > wal_end {
        return Err(RdbFileError::InvalidOperation(format!(
            "invalid embedded wal boundary {append_boundary}"
        )));
    }
    if open.manifest.wal_live_bytes > open.manifest.wal_region_bytes {
        return Err(RdbFileError::InvalidOperation(format!(
            "invalid embedded wal live bytes {}",
            open.manifest.wal_live_bytes
        )));
    }
    if open.manifest.wal_live_bytes == 0 {
        return Ok(WalScan {
            next_sequence: 1,
            ..WalScan::default()
        });
    }

    let mut file = File::open(&open.path)?;
    FileExt::lock_shared(&file)?;
    scan_wal_with_file(&mut file, open, wal_start)
}

/// Scan the WAL through an already-open handle — see
/// [`EmbeddedRdbArtifact::open_inner_with_file`] for why the writer path
/// cannot open a second one.
fn scan_wal_with_file(
    file: &mut File,
    open: &EmbeddedRdbOpen,
    wal_start: u64,
) -> RdbFileResult<WalScan> {
    let file_len = file.metadata()?.len();
    if file_len <= wal_start {
        return Ok(WalScan {
            next_sequence: 1,
            ..WalScan::default()
        });
    }
    let bytes = read_circular_wal_bytes(file, open, file_len)?;
    Ok(scan_wal_bytes(&bytes))
}

fn read_circular_wal_bytes(
    file: &mut File,
    open: &EmbeddedRdbOpen,
    file_len: u64,
) -> RdbFileResult<Vec<u8>> {
    let live_start = wal_live_start_relative(open)?;
    let region_bytes = open.manifest.wal_region_bytes;
    let live_bytes = open.manifest.wal_live_bytes;
    let mut remaining = live_bytes.min(file_len.saturating_sub(open.manifest.wal_region_offset));
    let mut relative = live_start;
    let mut bytes = Vec::with_capacity(remaining as usize);
    while remaining > 0 {
        let contiguous = (region_bytes - relative).min(remaining);
        let offset = open.manifest.wal_region_offset + relative;
        file.seek(SeekFrom::Start(offset))?;
        let mut chunk = vec![0u8; contiguous as usize];
        file.read_exact(&mut chunk)?;
        bytes.extend_from_slice(&chunk);
        remaining -= contiguous;
        relative = 0;
    }
    Ok(bytes)
}

fn write_circular_wal_bytes(
    file: &mut File,
    open: &EmbeddedRdbOpen,
    bytes: &[u8],
) -> RdbFileResult<()> {
    let mut written = 0usize;
    let mut relative = wal_append_relative(open)?;
    while written < bytes.len() {
        if relative == open.manifest.wal_region_bytes {
            relative = 0;
        }
        let contiguous = (open.manifest.wal_region_bytes - relative) as usize;
        let chunk_len = contiguous.min(bytes.len() - written);
        write_at(
            file,
            open.manifest.wal_region_offset + relative,
            &bytes[written..written + chunk_len],
        )?;
        written += chunk_len;
        relative += chunk_len as u64;
    }
    Ok(())
}

fn wal_append_relative(open: &EmbeddedRdbOpen) -> RdbFileResult<u64> {
    open.manifest
        .wal_recovery_boundary
        .checked_sub(open.manifest.wal_region_offset)
        .ok_or_else(|| RdbFileError::InvalidOperation("embedded wal boundary underflow".into()))
}

fn wal_live_start_relative(open: &EmbeddedRdbOpen) -> RdbFileResult<u64> {
    let append = wal_append_relative(open)?;
    let region = open.manifest.wal_region_bytes;
    if region == 0 {
        return Err(RdbFileError::InvalidOperation(
            "embedded wal region is empty".into(),
        ));
    }
    Ok((append + region - (open.manifest.wal_live_bytes % region)) % region)
}

fn wal_boundary_after_live_bytes(open: &EmbeddedRdbOpen, live_bytes: u64) -> RdbFileResult<u64> {
    if live_bytes > open.manifest.wal_region_bytes {
        return Err(RdbFileError::InvalidOperation(format!(
            "embedded wal live bytes {live_bytes} exceed region size {}",
            open.manifest.wal_region_bytes
        )));
    }
    let start = if open.manifest.wal_live_bytes == 0 {
        wal_append_relative(open)?
    } else {
        wal_live_start_relative(open)?
    };
    let relative = (start + live_bytes) % open.manifest.wal_region_bytes;
    open.manifest
        .wal_region_offset
        .checked_add(relative)
        .ok_or_else(|| RdbFileError::InvalidOperation("embedded wal boundary overflow".into()))
}

fn scan_wal_bytes(bytes: &[u8]) -> WalScan {
    let mut scan = WalScan {
        next_sequence: 1,
        ..WalScan::default()
    };
    let mut cursor = 0usize;
    while cursor < bytes.len() {
        let Some(frame) = decode_next_wal_frame(bytes, cursor, &scan) else {
            break;
        };
        scan.payloads.push(frame.payload);
        scan.next_sequence = scan.next_sequence.saturating_add(1);
        scan.previous_frame_crc = frame.frame_crc;
        cursor = frame.end;
        scan.valid_bytes = cursor as u64;
    }
    scan
}

struct DecodedWalFrame {
    payload: Vec<u8>,
    frame_crc: u32,
    end: usize,
}

fn decode_next_wal_frame(bytes: &[u8], start: usize, scan: &WalScan) -> Option<DecodedWalFrame> {
    let remaining = bytes.len().checked_sub(start)?;
    if remaining < WAL_FRAME_MAGIC.len() {
        return None;
    }
    if &bytes[start..start + WAL_FRAME_MAGIC.len()] != WAL_FRAME_MAGIC {
        return None;
    }
    if remaining < WAL_FRAME_MAGIC.len() + 2 {
        return None;
    }
    let version_offset = start + WAL_FRAME_MAGIC.len();
    let version = u16::from_le_bytes(bytes[version_offset..version_offset + 2].try_into().ok()?);
    if version == WAL_FRAME_VERSION {
        decode_v2_wal_frame(bytes, start, scan)
    } else {
        decode_legacy_wal_frame(bytes, start)
    }
}

fn decode_v2_wal_frame(bytes: &[u8], start: usize, scan: &WalScan) -> Option<DecodedWalFrame> {
    if bytes.len().checked_sub(start)? < WAL_FRAME_HEADER_BYTES {
        return None;
    }
    let header_len_offset = start + 10;
    let header_len = u16::from_le_bytes(
        bytes[header_len_offset..header_len_offset + 2]
            .try_into()
            .ok()?,
    ) as usize;
    if header_len != WAL_FRAME_HEADER_BYTES {
        return None;
    }
    let sequence_offset = start + 12;
    let sequence = u64::from_le_bytes(
        bytes[sequence_offset..sequence_offset + 8]
            .try_into()
            .ok()?,
    );
    if sequence != scan.next_sequence {
        return None;
    }
    let payload_len_offset = start + 20;
    let payload_len = u32::from_le_bytes(
        bytes[payload_len_offset..payload_len_offset + 4]
            .try_into()
            .ok()?,
    ) as usize;
    let payload_crc_offset = start + 24;
    let payload_crc = u32::from_le_bytes(
        bytes[payload_crc_offset..payload_crc_offset + 4]
            .try_into()
            .ok()?,
    );
    let previous_frame_crc_offset = start + 28;
    let previous_frame_crc = u32::from_le_bytes(
        bytes[previous_frame_crc_offset..previous_frame_crc_offset + 4]
            .try_into()
            .ok()?,
    );
    if previous_frame_crc != scan.previous_frame_crc {
        return None;
    }
    let header_crc_offset = start + 32;
    let header_crc = u32::from_le_bytes(
        bytes[header_crc_offset..header_crc_offset + 4]
            .try_into()
            .ok()?,
    );
    if header_crc != crc32(&bytes[start..header_crc_offset]) {
        return None;
    }
    let payload_start = start.checked_add(header_len)?;
    let end = payload_start.checked_add(payload_len)?;
    if end > bytes.len() {
        return None;
    }
    let payload = bytes[payload_start..end].to_vec();
    if crc32(&payload) != payload_crc {
        return None;
    }
    Some(DecodedWalFrame {
        payload,
        frame_crc: crc32(&bytes[start..end]),
        end,
    })
}

fn decode_legacy_wal_frame(bytes: &[u8], start: usize) -> Option<DecodedWalFrame> {
    if bytes.len().checked_sub(start)? < LEGACY_WAL_FRAME_HEADER_BYTES {
        return None;
    }
    let payload_len_offset = start + WAL_FRAME_MAGIC.len();
    let payload_len = u32::from_le_bytes(
        bytes[payload_len_offset..payload_len_offset + 4]
            .try_into()
            .ok()?,
    ) as usize;
    let payload_crc_offset = payload_len_offset + 4;
    let payload_crc = u32::from_le_bytes(
        bytes[payload_crc_offset..payload_crc_offset + 4]
            .try_into()
            .ok()?,
    );
    let payload_start = start.checked_add(LEGACY_WAL_FRAME_HEADER_BYTES)?;
    let end = payload_start.checked_add(payload_len)?;
    if end > bytes.len() {
        return None;
    }
    let payload = bytes[payload_start..end].to_vec();
    if crc32(&payload) != payload_crc {
        return None;
    }
    Some(DecodedWalFrame {
        payload,
        frame_crc: crc32(&bytes[start..end]),
        end,
    })
}

fn encode_wal_frame(
    sequence: u64,
    previous_frame_crc: u32,
    payload: &[u8],
) -> RdbFileResult<(Vec<u8>, u32)> {
    let payload_len = u32::try_from(payload.len())
        .map_err(|_| RdbFileError::InvalidOperation("embedded wal payload too large".into()))?;
    let mut frame = Vec::with_capacity(WAL_FRAME_HEADER_BYTES + payload.len());
    frame.extend_from_slice(WAL_FRAME_MAGIC);
    frame.extend_from_slice(&WAL_FRAME_VERSION.to_le_bytes());
    frame.extend_from_slice(&(WAL_FRAME_HEADER_BYTES as u16).to_le_bytes());
    frame.extend_from_slice(&sequence.to_le_bytes());
    frame.extend_from_slice(&payload_len.to_le_bytes());
    frame.extend_from_slice(&crc32(payload).to_le_bytes());
    frame.extend_from_slice(&previous_frame_crc.to_le_bytes());
    let header_crc = crc32(&frame);
    frame.extend_from_slice(&header_crc.to_le_bytes());
    frame.extend_from_slice(payload);
    let frame_crc = crc32(&frame);
    Ok((frame, frame_crc))
}

fn encode_superblock(superblock: EmbeddedRdbSuperblock) -> RdbFileResult<Vec<u8>> {
    let mut bytes = vec![0u8; EMBEDDED_RDB_SUPERBLOCK_SIZE as usize];
    let mut cursor = 0usize;
    put_bytes(&mut bytes, &mut cursor, SUPERBLOCK_MAGIC);
    put_u32(&mut bytes, &mut cursor, SUPERBLOCK_VERSION);
    put_u8(&mut bytes, &mut cursor, superblock.copy_index);
    put_u64(&mut bytes, &mut cursor, superblock.generation);
    put_u32(&mut bytes, &mut cursor, superblock.format_version);
    put_u64(&mut bytes, &mut cursor, superblock.manifest_offset);
    put_u64(&mut bytes, &mut cursor, superblock.manifest_len);
    put_u32(&mut bytes, &mut cursor, superblock.manifest_checksum);
    put_u64(&mut bytes, &mut cursor, superblock.wal_region_offset);
    put_u64(&mut bytes, &mut cursor, superblock.wal_region_bytes);
    put_u64(&mut bytes, &mut cursor, superblock.wal_recovery_boundary);
    put_u64(&mut bytes, &mut cursor, superblock.wal_live_bytes);
    put_u64(&mut bytes, &mut cursor, superblock.snapshot_offset);
    put_u64(&mut bytes, &mut cursor, superblock.snapshot_bytes);
    put_u32(&mut bytes, &mut cursor, superblock.snapshot_checksum);

    let checksum_offset = bytes.len() - CHECKSUM_LEN;
    let checksum = crc32(&bytes[..checksum_offset]);
    bytes[checksum_offset..].copy_from_slice(&checksum.to_le_bytes());
    Ok(bytes)
}

fn decode_superblock(copy_index: u8, bytes: &[u8]) -> RdbFileResult<EmbeddedRdbSuperblock> {
    if bytes.len() != EMBEDDED_RDB_SUPERBLOCK_SIZE as usize {
        return Err(RdbFileError::InvalidOperation(
            "invalid embedded superblock size".into(),
        ));
    }
    let checksum_offset = bytes.len() - CHECKSUM_LEN;
    let stored_checksum = u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap());
    let computed_checksum = crc32(&bytes[..checksum_offset]);
    if stored_checksum != computed_checksum {
        return Err(RdbFileError::InvalidOperation(
            "embedded superblock checksum mismatch".into(),
        ));
    }

    let mut cursor = 0usize;
    if take_bytes(bytes, &mut cursor, SUPERBLOCK_MAGIC.len())? != SUPERBLOCK_MAGIC {
        return Err(RdbFileError::InvalidOperation(
            "invalid embedded superblock magic".into(),
        ));
    }
    let version = take_u32(bytes, &mut cursor)?;
    if version != SUPERBLOCK_VERSION && version != LEGACY_SUPERBLOCK_VERSION {
        return Err(RdbFileError::InvalidOperation(format!(
            "unsupported embedded superblock version {version}"
        )));
    }
    let stored_copy_index = take_u8(bytes, &mut cursor)?;
    if stored_copy_index != copy_index {
        return Err(RdbFileError::InvalidOperation(
            "embedded superblock copy index mismatch".into(),
        ));
    }

    let generation = take_u64(bytes, &mut cursor)?;
    let format_version = take_u32(bytes, &mut cursor)?;
    let manifest_offset = take_u64(bytes, &mut cursor)?;
    let manifest_len = take_u64(bytes, &mut cursor)?;
    let manifest_checksum = take_u32(bytes, &mut cursor)?;
    let wal_region_offset = take_u64(bytes, &mut cursor)?;
    let wal_region_bytes = take_u64(bytes, &mut cursor)?;
    let wal_recovery_boundary = take_u64(bytes, &mut cursor)?;
    let wal_live_bytes = if version == SUPERBLOCK_VERSION {
        take_u64(bytes, &mut cursor)?
    } else {
        wal_recovery_boundary.saturating_sub(wal_region_offset)
    };

    Ok(EmbeddedRdbSuperblock {
        copy_index: stored_copy_index,
        generation,
        format_version,
        manifest_offset,
        manifest_len,
        manifest_checksum,
        wal_region_offset,
        wal_region_bytes,
        wal_recovery_boundary,
        wal_live_bytes,
        snapshot_offset: take_u64(bytes, &mut cursor)?,
        snapshot_bytes: take_u64(bytes, &mut cursor)?,
        snapshot_checksum: take_u32(bytes, &mut cursor)?,
        checksum: stored_checksum,
    })
}

fn encode_manifest(manifest: EmbeddedRdbManifest) -> Vec<u8> {
    let mut bytes = vec![0u8; 8 + 4 + 8 + 8 + 8 + 8 + 8 + 8 + 4 + 16 + CHECKSUM_LEN];
    let mut cursor = 0usize;
    put_bytes(&mut bytes, &mut cursor, MANIFEST_MAGIC);
    put_u32(&mut bytes, &mut cursor, manifest.version);
    put_u64(&mut bytes, &mut cursor, manifest.wal_region_offset);
    put_u64(&mut bytes, &mut cursor, manifest.wal_region_bytes);
    put_u64(&mut bytes, &mut cursor, manifest.wal_recovery_boundary);
    put_u64(&mut bytes, &mut cursor, manifest.wal_live_bytes);
    put_u64(&mut bytes, &mut cursor, manifest.snapshot_offset);
    put_u64(&mut bytes, &mut cursor, manifest.snapshot_bytes);
    put_u32(&mut bytes, &mut cursor, manifest.snapshot_checksum);
    put_u128(&mut bytes, &mut cursor, manifest.created_at_unix_ms);

    let checksum_offset = bytes.len() - CHECKSUM_LEN;
    let checksum = crc32(&bytes[..checksum_offset]);
    bytes[checksum_offset..].copy_from_slice(&checksum.to_le_bytes());
    bytes
}

fn decode_manifest(bytes: &[u8]) -> RdbFileResult<EmbeddedRdbManifest> {
    let checksum_offset = bytes
        .len()
        .checked_sub(CHECKSUM_LEN)
        .ok_or_else(|| RdbFileError::InvalidOperation("embedded manifest too short".into()))?;
    let stored_checksum = u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap());
    let computed_checksum = crc32(&bytes[..checksum_offset]);
    if stored_checksum != computed_checksum {
        return Err(RdbFileError::InvalidOperation(
            "embedded manifest checksum mismatch".into(),
        ));
    }

    let mut cursor = 0usize;
    if take_bytes(bytes, &mut cursor, MANIFEST_MAGIC.len())? != MANIFEST_MAGIC {
        return Err(RdbFileError::InvalidOperation(
            "invalid embedded manifest magic".into(),
        ));
    }
    let version = take_u32(bytes, &mut cursor)?;
    if version != MANIFEST_VERSION && version != LEGACY_MANIFEST_VERSION {
        return Err(RdbFileError::InvalidOperation(format!(
            "unsupported embedded manifest version {version}"
        )));
    }
    let wal_region_offset = take_u64(bytes, &mut cursor)?;
    let wal_region_bytes = take_u64(bytes, &mut cursor)?;
    let wal_recovery_boundary = take_u64(bytes, &mut cursor)?;
    let wal_live_bytes = if version == MANIFEST_VERSION {
        take_u64(bytes, &mut cursor)?
    } else {
        wal_recovery_boundary.saturating_sub(wal_region_offset)
    };
    Ok(EmbeddedRdbManifest {
        version,
        wal_region_offset,
        wal_region_bytes,
        wal_recovery_boundary,
        wal_live_bytes,
        snapshot_offset: take_u64(bytes, &mut cursor)?,
        snapshot_bytes: take_u64(bytes, &mut cursor)?,
        snapshot_checksum: take_u32(bytes, &mut cursor)?,
        created_at_unix_ms: take_u128(bytes, &mut cursor)?,
        checksum: stored_checksum,
    })
}

fn trailer_checksum(bytes: &[u8]) -> u32 {
    let checksum_offset = bytes.len() - CHECKSUM_LEN;
    u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap())
}

fn superblock_offset(copy_index: u8) -> RdbFileResult<u64> {
    match copy_index {
        0 => Ok(EMBEDDED_RDB_SUPERBLOCK_0_OFFSET),
        1 => Ok(EMBEDDED_RDB_SUPERBLOCK_1_OFFSET),
        _ => Err(RdbFileError::InvalidOperation(format!(
            "invalid embedded superblock copy index {copy_index}"
        ))),
    }
}

fn write_at(file: &mut File, offset: u64, bytes: &[u8]) -> RdbFileResult<()> {
    file.seek(SeekFrom::Start(offset))?;
    file.write_all(bytes)?;
    Ok(())
}

fn crash_inject(point: &str) {
    if std::env::var(CRASH_INJECT_ENV).ok().as_deref() == Some(point) {
        std::process::exit(173);
    }
    if crate::buggify!(CRASH_INJECT_ENV, point) {
        std::process::exit(173);
    }
}

fn now_unix_ms() -> u128 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_millis())
        .unwrap_or(0)
}

fn put_bytes(target: &mut [u8], cursor: &mut usize, value: &[u8]) {
    target[*cursor..*cursor + value.len()].copy_from_slice(value);
    *cursor += value.len();
}

fn put_u8(target: &mut [u8], cursor: &mut usize, value: u8) {
    target[*cursor] = value;
    *cursor += 1;
}

fn put_u32(target: &mut [u8], cursor: &mut usize, value: u32) {
    put_bytes(target, cursor, &value.to_le_bytes());
}

fn put_u64(target: &mut [u8], cursor: &mut usize, value: u64) {
    put_bytes(target, cursor, &value.to_le_bytes());
}

fn put_u128(target: &mut [u8], cursor: &mut usize, value: u128) {
    put_bytes(target, cursor, &value.to_le_bytes());
}

fn take_bytes<'a>(bytes: &'a [u8], cursor: &mut usize, len: usize) -> RdbFileResult<&'a [u8]> {
    let end = cursor.checked_add(len).ok_or_else(|| {
        RdbFileError::InvalidOperation("embedded artifact cursor overflow".into())
    })?;
    if end > bytes.len() {
        return Err(RdbFileError::InvalidOperation(
            "embedded artifact truncated".into(),
        ));
    }
    let value = &bytes[*cursor..end];
    *cursor = end;
    Ok(value)
}

fn take_u8(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u8> {
    Ok(take_bytes(bytes, cursor, 1)?[0])
}

fn take_u32(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u32> {
    Ok(u32::from_le_bytes(
        take_bytes(bytes, cursor, 4)?.try_into().unwrap(),
    ))
}

fn take_u64(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u64> {
    Ok(u64::from_le_bytes(
        take_bytes(bytes, cursor, 8)?.try_into().unwrap(),
    ))
}

fn take_u128(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u128> {
    Ok(u128::from_le_bytes(
        take_bytes(bytes, cursor, 16)?.try_into().unwrap(),
    ))
}