reddb_server/storage/unified/store/

impl_pages.rs

use super::*;
use parking_lot::RwLock;

const CHECKPOINT_BATCH_BYTES: usize = 1024 * 1024;
const CHECKPOINT_BATCH_ROWS: usize = 512;

fn restore_published_tree(
    pager: &Arc<Pager>,
    root: u32,
    proofs: Option<&reddb_file::page_proof::CatalogProofs>,
) -> Result<BTree, StoreError> {
    let proof = proofs.and_then(|proofs| proofs.get(&root)).copied();
    let tree = match (root, proof) {
        (0, _) if proofs.is_some() => BTree::new_verified(Arc::clone(pager)),
        (_, Some(proof)) => BTree::with_verified_root(Arc::clone(pager), root, proof),
        (_, None) if proofs.is_some() => {
            return Err(page_integrity(
                root,
                None,
                "missing published tree generation proof",
            ));
        }
        _ => return Ok(BTree::with_root(Arc::clone(pager), root)),
    };
    tree.map_err(|error| page_integrity(root, None, error.to_string()))
}

#[derive(Debug)]
struct SystemCollectionCollision {
    survivor: String,
    loser: String,
    canonical: String,
    shared_physical_state: bool,
    resume_counter_drift: bool,
    merged_resume: Option<reddb_file::native_resume::CollectionResume>,
}

#[derive(Debug, Default)]
struct CatalogLoadOutcome {
    migrated: bool,
    collisions: Vec<SystemCollectionCollision>,
}

fn merge_shared_alias_resume(
    legacy: &reddb_file::native_resume::CollectionResume,
    canonical: &reddb_file::native_resume::CollectionResume,
) -> Option<reddb_file::native_resume::CollectionResume> {
    let (data_root_match, key_index_root_match, secondary_index_state_match) =
        shared_alias_resume_matches(legacy, canonical);
    if !data_root_match || !key_index_root_match || !secondary_index_state_match {
        return None;
    }
    let mut merged = legacy.clone();
    // The shared roots bind the exact durable data and derived indexes. Resume
    // counters are advisory allocation state; retaining each high-water mark
    // prevents identity reuse after an interrupted name migration. The exact
    // entity count is read from the shared data tree while loading.
    merged.entities = legacy.entities.max(canonical.entities);
    merged.entity_id_next = legacy.entity_id_next.max(canonical.entity_id_next);
    merged.row_id_next = legacy.row_id_next.max(canonical.row_id_next);
    merged.sequence_id_next = legacy.sequence_id_next.max(canonical.sequence_id_next);
    Some(merged)
}

fn shared_alias_resume_matches(
    legacy: &reddb_file::native_resume::CollectionResume,
    canonical: &reddb_file::native_resume::CollectionResume,
) -> (bool, bool, bool) {
    (
        legacy.root_page == canonical.root_page,
        legacy.key_index_root == canonical.key_index_root,
        legacy.secondary_indices == canonical.secondary_indices
            && legacy.metadata_expiries == canonical.metadata_expiries,
    )
}

/// Keep encoded payload memory proportional to a small batch, not a collection.
/// A single large value retains the existing B-tree value-size contract.
struct CheckpointTreeBuilder {
    tree: Arc<BTree>,
    records: Vec<(Vec<u8>, Vec<u8>)>,
    bytes: usize,
}

impl CheckpointTreeBuilder {
    fn new(pager: &Arc<Pager>) -> Result<Self, StoreError> {
        pager
            .fence_required_capabilities(
                reddb_file::file_format::FILE_CAPABILITY_TREE_PROOFS
                    | reddb_file::file_format::FILE_CAPABILITY_DWB_PUBLICATION,
            )
            .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
        Ok(Self {
            tree: UnifiedStore::new_derived_btree(pager)?,
            records: Vec::new(),
            bytes: 0,
        })
    }

    fn push(&mut self, record: (Vec<u8>, Vec<u8>)) -> Result<(), StoreError> {
        let bytes = record.0.len() + record.1.len();
        if !self.records.is_empty()
            && (self.bytes.saturating_add(bytes) > CHECKPOINT_BATCH_BYTES
                || self.records.len() >= CHECKPOINT_BATCH_ROWS)
        {
            self.flush()?;
        }
        self.bytes += bytes;
        self.records.push(record);
        #[cfg(test)]
        CHECKPOINT_BUFFERED_BYTES.with(|peak| peak.set(peak.get().max(self.bytes)));
        Ok(())
    }

    fn flush(&mut self) -> Result<(), StoreError> {
        self.tree
            .bulk_insert_sorted(&self.records)
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        self.records.clear();
        self.bytes = 0;
        Ok(())
    }

    fn copy_prefix(&mut self, previous: &BTree, count: usize) -> Result<(), StoreError> {
        let mut cursor = previous
            .cursor_first()
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        for _ in 0..count {
            let record = cursor
                .next()
                .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?
                .ok_or_else(|| StoreError::Serialization("checkpoint prefix disappeared".into()))?;
            self.push(record)?;
        }
        Ok(())
    }
}

#[cfg(test)]
thread_local! {
    static CHECKPOINT_SERIALIZED_ROWS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    static REPLAY_SERIALIZED_ROWS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    static CHECKPOINT_BUFFERED_BYTES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

fn checkpoint_visibility_matches(
    previous: &CheckpointVersion,
    current: &CheckpointVersion,
) -> bool {
    (previous.visibility_revision == current.visibility_revision
        || previous.excluded_xids == current.excluded_xids)
        && previous.pending_vault == current.pending_vault
}

// ── Pager-meta overflow chain (gh-477) ──────────────────────────────────────
// When the serialized collection registry + cross-refs exceed a single page,
// page 1 carries a native metadata-overflow header pointing at an overflow chain of
// `PageType::Overflow` pages. Single-page metadata keeps the historical
// bit-identical layout (`METADATA_MAGIC = "RDM2"` written directly at the
// content offset).
//
// Page 1 (overflow form), starting at `HEADER_SIZE`:
//   [0..4]   native metadata-overflow magic
//   [4..8]   format_version (u32, mirrors inner payload version for debug)
//   [8..12]  total_payload_bytes (u32)
//   [12..16] next_overflow_page_id (u32, > 0)
//   [16..]   first payload chunk (up to META_V3_FIRST_PAYLOAD_CAP bytes)
//
// Overflow continuation page, starting at `HEADER_SIZE`:
//   [0..4]   next_overflow_page_id (u32, 0 if last)
//   [4..8]   chunk_bytes (u32)
//   [8..]    chunk payload (up to META_V3_OVERFLOW_PAYLOAD_CAP bytes)
const META_PAGE_CONTENT_CAP: usize =
    crate::storage::engine::PAGE_SIZE - crate::storage::engine::HEADER_SIZE;
const META_V3_PAGE1_HEADER: usize = reddb_file::METADATA_OVERFLOW_HEADER_BYTES;
const META_V3_OVERFLOW_HEADER: usize = reddb_file::METADATA_OVERFLOW_CONTINUATION_HEADER_BYTES;
const META_V3_FIRST_PAYLOAD_CAP: usize = META_PAGE_CONTENT_CAP - META_V3_PAGE1_HEADER;
const META_V3_OVERFLOW_PAYLOAD_CAP: usize = META_PAGE_CONTENT_CAP - META_V3_OVERFLOW_HEADER;

fn existing_overflow_chain(pager: &Pager) -> Result<Vec<u32>, PagerError> {
    let cs = crate::storage::engine::HEADER_SIZE;
    let page = pager.read_manifest_page()?;
    let mut digest = reddb_file::page_proof::CatalogDigest::default();
    digest.add_page(page.as_bytes());
    let bytes = page.as_bytes();
    if bytes.len() < cs + META_V3_PAGE1_HEADER {
        return Ok(Vec::new());
    }
    let Some(header) =
        reddb_file::decode_native_metadata_overflow_header(&bytes[cs..]).map_err(|err| {
            PagerError::InvalidDatabase(format!("invalid metadata overflow header: {err}"))
        })?
    else {
        verify_catalog_digest(pager, digest.finish())?;
        return Ok(Vec::new());
    };
    let mut next = header.next_overflow_page_id;
    let mut pages = Vec::new();
    let mut seen = std::collections::HashSet::new();
    while next != 0 {
        if !seen.insert(next) || pages.len() >= pager.page_count()? as usize {
            return Err(PagerError::InvalidDatabase(
                "cyclic metadata overflow chain".into(),
            ));
        }
        let ov = pager.read_page(next)?;
        digest.add_page(ov.as_bytes());
        if ov.page_type()? != crate::storage::engine::PageType::Overflow {
            return Err(PagerError::InvalidDatabase(
                "invalid metadata overflow page type".into(),
            ));
        }
        let ob = ov.as_bytes();
        let nn = reddb_file::decode_native_metadata_overflow_continuation_header(&ob[cs..])
            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?
            .next_overflow_page_id;
        pages.push(next);
        next = nn;
    }
    verify_catalog_digest(pager, digest.finish())?;
    Ok(pages)
}

fn verify_catalog_digest(pager: &Pager, digest: [u8; 32]) -> Result<(), PagerError> {
    let reference = pager.manifest_reference()?;
    if reference.page_id != 0 && reference.digest != digest {
        return Err(PagerError::InvalidDatabase(format!(
            "catalog page {} generation proof does not match the superblock",
            reference.page_id
        )));
    }
    Ok(())
}

fn publish_paged_metadata(pager: &Pager, meta_data: &[u8]) -> Result<(), PagerError> {
    let _ownership = pager.lock_ownership_mutations();
    let mut retired = existing_overflow_chain(pager)?;
    let previous_root = pager.manifest_page_id()?;
    if previous_root >= reddb_engine::storage::pager_io::FIRST_ALLOCATABLE_PAGE {
        retired.push(previous_root);
    }
    if !retired.is_empty() {
        pager.record_retirement_intent(&retired)?;
    }
    pager.fence_required_capabilities(reddb_file::FILE_CAPABILITY_CATALOG_ROOT)?;
    let root = pager
        .allocate_page(crate::storage::engine::PageType::Header)?
        .page_id();
    let (page, digest) = build_meta_page1_with_overflow(pager, meta_data, root)?;
    pager.write_page(root, page)?;
    #[cfg(test)]
    if let Ok(prefix) = std::env::var("REDDB_TEST_CATALOG_HOME_IO_PREFIX") {
        crate::storage::engine::pager::test_fail_next_pager_write(
            Some(u64::from(root) * crate::storage::engine::PAGE_SIZE as u64),
            prefix.parse().expect("catalog home prefix"),
        );
    }
    // Publish only after all children, data pages and proof paths are durable.
    pager.sync()?;
    #[cfg(test)]
    if std::env::var_os("REDDB_TEST_EXIT_BEFORE_CATALOG_ROOT_PUBLICATION").is_some() {
        std::process::exit(197);
    }
    pager.publish_manifest_reference(reddb_file::page_proof::PageReference {
        page_id: root,
        digest,
    })?;
    #[cfg(test)]
    if std::env::var_os("REDDB_TEST_EXIT_AFTER_CATALOG_ROOT_PUBLICATION").is_some() {
        std::process::exit(198);
    }
    if !retired.is_empty() {
        // Metadata readers hold the ownership fence. Both superblocks have
        // forgotten these pages, so no reader or fallback can still name them.
        pager.free_retired_pages(&retired)?;
        pager.sync()?;
    }
    Ok(())
}

fn build_meta_page1_with_overflow(
    pager: &Pager,
    meta_data: &[u8],
    root_page_id: u32,
) -> Result<(crate::storage::engine::Page, [u8; 32]), PagerError> {
    use crate::storage::engine::{Page, PageType, HEADER_SIZE};

    let mut page1 = Page::new(PageType::Header, root_page_id);
    let cs = HEADER_SIZE;

    if meta_data.len() <= META_PAGE_CONTENT_CAP {
        // Single-page: bit-identical to the historical layout.
        let buf = page1.as_bytes_mut();
        buf[cs..cs + meta_data.len()].copy_from_slice(meta_data);
        page1.update_checksum();
        let digest = reddb_file::page_proof::page_digest(page1.as_bytes());
        return Ok((page1, digest));
    }

    // Multi-page overflow form. Split the inner payload into the first chunk
    // (held on page 1) followed by zero-or-more continuation chunks chained
    // through `PageType::Overflow` pages.
    let first_chunk = &meta_data[..META_V3_FIRST_PAYLOAD_CAP];
    let mut tail = &meta_data[META_V3_FIRST_PAYLOAD_CAP..];
    let mut chunks: Vec<&[u8]> = Vec::new();
    while !tail.is_empty() {
        let take = tail.len().min(META_V3_OVERFLOW_PAYLOAD_CAP);
        chunks.push(&tail[..take]);
        tail = &tail[take..];
    }

    let mut overflow_pages: Vec<Page> = Vec::with_capacity(chunks.len());
    let mut overflow_ids: Vec<u32> = Vec::with_capacity(chunks.len());
    for _ in 0..chunks.len() {
        let pg = pager.allocate_page(PageType::Overflow)?;
        overflow_ids.push(pg.page_id());
        overflow_pages.push(pg);
    }

    for i in 0..chunks.len() {
        let next = if i + 1 < chunks.len() {
            overflow_ids[i + 1]
        } else {
            0u32
        };
        let len = chunks[i].len() as u32;
        let buf = overflow_pages[i].as_bytes_mut();
        reddb_file::encode_native_metadata_overflow_continuation_header(
            &mut buf[cs..cs + META_V3_OVERFLOW_HEADER],
            reddb_file::NativeMetadataOverflowContinuationHeader {
                next_overflow_page_id: next,
                chunk_bytes: len,
            },
        )
        .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
        buf[cs + 8..cs + 8 + chunks[i].len()].copy_from_slice(chunks[i]);
    }

    // Mirror the inner format_version for debug-friendly hex dumps.
    let format_version = reddb_file::decode_native_paged_metadata_header(meta_data)
        .ok()
        .flatten()
        .map_or(0, |header| header.format_version);

    let buf = page1.as_bytes_mut();
    reddb_file::encode_native_metadata_overflow_header(
        &mut buf[cs..cs + META_V3_PAGE1_HEADER],
        reddb_file::NativeMetadataOverflowHeader {
            format_version,
            total_payload_bytes: meta_data.len() as u32,
            next_overflow_page_id: overflow_ids[0],
        },
    )
    .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
    buf[cs + META_V3_PAGE1_HEADER..cs + META_V3_PAGE1_HEADER + first_chunk.len()]
        .copy_from_slice(first_chunk);

    page1.update_checksum();
    let mut digest = reddb_file::page_proof::CatalogDigest::default();
    digest.add_page(page1.as_bytes());
    for mut page in overflow_pages {
        page.update_checksum();
        digest.add_page(page.as_bytes());
        pager.write_page(page.page_id(), page)?;
    }
    Ok((page1, digest.finish()))
}

fn storage_integrity(
    zone: &str,
    id: impl Into<String>,
    collection: Option<String>,
    detail: impl Into<String>,
) -> StoreError {
    StoreError::StorageIntegrity(crate::api::StorageIntegrityError::new(
        zone, id, collection, detail,
    ))
}

fn page_integrity(
    page_id: u32,
    collection: Option<String>,
    detail: impl Into<String>,
) -> StoreError {
    storage_integrity("page", page_id.to_string(), collection, detail)
}

/// Assemble the full metadata payload from the internal manifest zone: page 1
/// plus its overflow chain when the native overflow wrapper is present. Returns
/// the bytes that the metadata parser would see starting from the content
/// offset of page 1. Single-page metadata returns the raw page content
/// (including trailing zero-pad), so the legacy parser sees the same bytes it
/// always saw.
///
/// The zone is rooted by the superblock and checksummed per ADR 0074 §2: a
/// failing checksum surfaces the didactic manifest-zone integrity error
/// directly — there is no sidecar copy to fall back to since ADR 0038 §4
/// phase 1 retired the shadows.
fn read_meta_payload(pager: &Pager) -> Result<Option<Vec<u8>>, StoreError> {
    let _ownership = pager.lock_ownership_mutations();
    let cs = crate::storage::engine::HEADER_SIZE;
    let meta_page = pager.read_manifest_page().map_err(|err| {
        storage_integrity(
            "manifest",
            "manifest",
            None,
            format!("internal manifest zone read failed ({err})"),
        )
    })?;
    let mut digest = reddb_file::page_proof::CatalogDigest::default();
    digest.add_page(meta_page.as_bytes());
    let bytes = meta_page.as_bytes();
    if bytes.len() < cs + 4 {
        return Ok(Some(bytes.get(cs..).unwrap_or(&[]).to_vec()));
    }
    let header = match reddb_file::decode_native_metadata_overflow_header(&bytes[cs..])
        .map_err(|err| page_integrity(1, None, err.to_string()))?
    {
        Some(header) => header,
        None => {
            verify_catalog_digest(pager, digest.finish())
                .map_err(|error| page_integrity(meta_page.page_id(), None, error.to_string()))?;
            return Ok(Some(bytes[cs..].to_vec()));
        }
    };
    if bytes.len() < cs + META_V3_PAGE1_HEADER {
        return Err(page_integrity(
            1,
            None,
            "metadata overflow header is truncated",
        ));
    }
    let total = header.total_payload_bytes as usize;
    if total == 0 || total > 64 * 1024 * 1024 {
        return Err(page_integrity(
            meta_page.page_id(),
            None,
            "catalog exceeds metadata budget",
        ));
    }
    let mut next = header.next_overflow_page_id;
    let mut payload: Vec<u8> = Vec::with_capacity(total);
    let first_take = total.min(META_V3_FIRST_PAYLOAD_CAP);
    payload.extend_from_slice(
        &bytes[cs + META_V3_PAGE1_HEADER..cs + META_V3_PAGE1_HEADER + first_take],
    );
    let mut seen = std::collections::HashSet::new();
    while next != 0 && payload.len() < total {
        if !seen.insert(next)
            || seen.len()
                >= pager
                    .page_count()
                    .map_err(|err| page_integrity(next, None, err.to_string()))?
                    as usize
        {
            return Err(page_integrity(next, None, "cyclic metadata overflow chain"));
        }
        let ov = pager
            .read_page(next)
            .map_err(|err| page_integrity(next, None, err.to_string()))?;
        digest.add_page(ov.as_bytes());
        if ov
            .page_type()
            .map_err(|err| page_integrity(next, None, err.to_string()))?
            != crate::storage::engine::PageType::Overflow
        {
            return Err(page_integrity(
                next,
                None,
                "invalid metadata overflow page type",
            ));
        }
        let ob = ov.as_bytes();
        if ob.len() < cs + META_V3_OVERFLOW_HEADER {
            return Err(page_integrity(
                next,
                None,
                "metadata overflow continuation is truncated",
            ));
        }
        let continuation =
            reddb_file::decode_native_metadata_overflow_continuation_header(&ob[cs..])
                .map_err(|err| page_integrity(next, None, err.to_string()))?;
        let nn = continuation.next_overflow_page_id;
        let len = continuation.chunk_bytes as usize;
        let remaining = total - payload.len();
        let take = len.min(remaining).min(META_V3_OVERFLOW_PAYLOAD_CAP);
        payload.extend_from_slice(
            &ob[cs + META_V3_OVERFLOW_HEADER..cs + META_V3_OVERFLOW_HEADER + take],
        );
        next = nn;
    }
    if payload.len() != total || next != 0 {
        return Err(page_integrity(
            1,
            None,
            "metadata overflow chain is incomplete",
        ));
    }
    verify_catalog_digest(pager, digest.finish())
        .map_err(|error| page_integrity(meta_page.page_id(), None, error.to_string()))?;
    Ok(Some(payload))
}

impl UnifiedStore {
    pub(super) fn new_derived_btree(pager: &Arc<Pager>) -> Result<Arc<BTree>, StoreError> {
        let header = pager
            .header()
            .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
        let tree = if header.required_capabilities
            & reddb_file::file_format::FILE_CAPABILITY_TREE_PROOFS
            != 0
        {
            BTree::new_verified(Arc::clone(pager))
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?
        } else {
            BTree::new(Arc::clone(pager))
        };
        Ok(Arc::new(tree))
    }

    pub(super) fn immutable_paged_base(&self) -> Result<bool, StoreError> {
        if self.config.resident_memory_bytes_max.is_some() {
            return Ok(true);
        }
        let Some(pager) = &self.pager else {
            return Ok(false);
        };
        let header = pager
            .header()
            .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
        Ok(
            header.required_capabilities & reddb_file::file_format::FILE_CAPABILITY_TREE_PROOFS
                != 0,
        )
    }

    /// Offline reconstruction uses the same catalog-chain proof and publication
    /// ordering as the live store, with an exclusively owned destination pager.
    pub(crate) fn publish_native_catalog(pager: &Pager, metadata: &[u8]) -> Result<(), PagerError> {
        publish_paged_metadata(pager, metadata)
    }

    pub(crate) fn fence_required_capabilities(&self, capabilities: u64) -> Result<(), StoreError> {
        if capabilities == 0 {
            return Ok(());
        }
        let Some(pager) = &self.pager else {
            return Ok(());
        };
        pager
            .fence_required_capabilities(capabilities)
            .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))
    }

    /// Compare and encode one borrowed row at a time. Begin a new tree only at
    /// the first difference, copying its already-verified prefix in bounded
    /// batches. Unchanged trees require no replacement allocation.
    fn checkpoint_tree(
        &self,
        pager: &Arc<Pager>,
        manager: &SegmentManager,
        previous: Option<&Arc<BTree>>,
        mut encode: impl FnMut(&UnifiedEntity, Option<Metadata>) -> Option<Vec<u8>>,
    ) -> Result<Arc<BTree>, StoreError> {
        let previous = previous.filter(|tree| tree.has_page_proofs());
        if let Some((base, overlay)) = manager.backing_overlay(&mut encode) {
            use reddb_engine::storage::collection_read::{CollectionRead, MergedCollection};
            let merged = MergedCollection::new(base.as_ref(), overlay);
            let mut cursor = merged.cursor()?;
            let mut builder = CheckpointTreeBuilder::new(pager)?;
            while let Some(record) = cursor.next_record(
                usize::try_from(self.config.resident_memory_bytes_max.unwrap_or(u64::MAX))
                    .unwrap_or(usize::MAX),
            )? {
                builder.push((record.id.raw().to_be_bytes().to_vec(), record.bytes))?;
            }
            builder.flush()?;
            return Ok(builder.tree);
        }
        let mut cursor = previous
            .map(|tree| tree.cursor_first())
            .transpose()
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        let mut matched = 0;
        let mut builder = None;
        let mut failure = None;
        manager.for_each_entity_with_metadata_in_id_order(|entity, metadata| {
            let Some(value) = encode(entity, metadata) else {
                return true;
            };
            let record = (entity.id.raw().to_be_bytes().to_vec(), value);
            let result = (|| {
                if builder.is_none() {
                    if let Some(cursor) = cursor.as_mut() {
                        if cursor
                            .next()
                            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?
                            .as_ref()
                            == Some(&record)
                        {
                            matched += 1;
                            return Ok(());
                        }
                    }
                    let mut replacement = CheckpointTreeBuilder::new(pager)?;
                    if let Some(previous) = previous {
                        replacement.copy_prefix(previous, matched)?;
                    }
                    builder = Some(replacement);
                }
                builder
                    .as_mut()
                    .expect("replacement initialized")
                    .push(record)
            })();
            if let Err(error) = result {
                failure = Some(error);
                return false;
            }
            true
        });
        if let Some(error) = failure {
            return Err(error);
        }
        if builder.is_none() {
            if let (Some(previous), Some(cursor)) = (previous, cursor.as_mut()) {
                if cursor
                    .next()
                    .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?
                    .is_none()
                {
                    return Ok(Arc::clone(previous));
                }
            }
            let mut replacement = CheckpointTreeBuilder::new(pager)?;
            if let Some(previous) = previous {
                replacement.copy_prefix(previous, matched)?;
            }
            builder = Some(replacement);
        }
        let mut builder = builder.expect("replacement or reused tree");
        builder.flush()?;
        Ok(builder.tree)
    }

    /// Publication first, reuse second. The immediately preceding generation
    /// stays in the durable ledger for one more publication. Cursors and cloned
    /// tree handles defer retirement further without holding the writer fence.
    fn publish_reclaimable_trees(
        &self,
        pager: &Pager,
        metadata: &mut Vec<u8>,
        live: &HashMap<String, Arc<BTree>>,
        keys: &HashMap<String, PublishedKeyIndex>,
    ) -> Result<(), StoreError> {
        let mut next_xid_summary = None;
        let publishes_proofs = pager
            .header()
            .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?
            .required_capabilities
            & reddb_file::file_format::FILE_CAPABILITY_TREE_PROOFS
            != 0;
        if publishes_proofs {
            pager
                .fence_required_capabilities(
                    reddb_file::file_format::FILE_CAPABILITY_DWB_PUBLICATION,
                )
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
        }
        let mut retired = self.retired_btrees.lock();
        let mut roots = Vec::new();
        let mut reclaim = Vec::new();
        let mut remove = std::collections::HashSet::new();
        for (index, entry) in retired.iter_mut().enumerate() {
            if live
                .values()
                .chain(keys.values().flat_map(PublishedKeyIndex::trees))
                .any(|tree| Arc::ptr_eq(tree, &entry.tree))
            {
                remove.insert(index);
                continue;
            }
            if entry.published
                && Arc::strong_count(&entry.tree) == 1
                && !entry.tree.has_retirement_readers()
            {
                // Deliberately unsafe experimental control: keep the old
                // root in the catalog while releasing its pages. Never enable
                // this outside the isolated recovery test.
                #[cfg(test)]
                if std::env::var_os("REDDB_EXPERIMENTAL_RETAIN_RELEASED_ROOT").is_some() {
                    roots.push(entry.tree.root_page_id());
                }
                reclaim.extend(
                    entry
                        .tree
                        .retired_pages()
                        .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?,
                );
                remove.insert(index);
            } else if entry.tree.root_page_id() != 0 {
                if publishes_proofs && !entry.tree.has_page_proofs() {
                    let prepared = entry.tree.prepare_legacy_page_proofs().map_err(|error| {
                        page_integrity(entry.tree.root_page_id(), None, error.to_string())
                    })?;
                    entry.tree = Arc::new(prepared);
                }
                roots.push(entry.tree.root_page_id());
            }
        }
        roots.sort_unstable();
        roots.dedup();
        reddb_file::encode_native_paged_retired_trees(metadata, &roots)
            .map_err(|e| StoreError::Serialization(e.to_string()))?;
        let (pending_pages, published_pages) = pager
            .deferred_reclamation_pages()
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        let readers = live
            .values()
            .chain(keys.values().flat_map(PublishedKeyIndex::trees))
            .any(|tree| tree.has_retirement_readers())
            || retired
                .iter()
                .any(|entry| entry.tree.has_retirement_readers());
        let released_pages = if readers { Vec::new() } else { published_pages };
        let released: std::collections::HashSet<_> = released_pages.iter().copied().collect();
        let retained_pages: Vec<_> = pending_pages
            .into_iter()
            .filter(|id| !released.contains(id))
            .collect();
        reddb_file::encode_native_paged_deferred_pages(metadata, &retained_pages)
            .map_err(|e| StoreError::Serialization(e.to_string()))?;
        let active_epoch = self.ownership_active_epoch.load(Ordering::Acquire);
        let catalog_epoch = if active_epoch == 0 {
            self.ownership_catalog_epoch.load(Ordering::Acquire)
        } else {
            active_epoch
        };
        if catalog_epoch != 0 {
            reddb_file::native_store::encode_native_paged_ownership_epoch(metadata, catalog_epoch)
                .map_err(|e| StoreError::Serialization(e.to_string()))?;
        }
        if publishes_proofs {
            let mut proofs = reddb_file::page_proof::CatalogProofs::new();
            for tree in live
                .values()
                .chain(keys.values().flat_map(PublishedKeyIndex::trees))
                .chain(
                    retired
                        .iter()
                        .enumerate()
                        .filter(|(index, _)| !remove.contains(index))
                        .map(|(_, entry)| &entry.tree),
                )
            {
                if let Some((root, proof)) = tree
                    .seal_page_proofs()
                    .map_err(|error| page_integrity(tree.root_page_id(), None, error.to_string()))?
                {
                    if root != 0 {
                        if proofs.insert(root, proof).is_some_and(|old| old != proof) {
                            return Err(page_integrity(
                                root,
                                None,
                                "conflicting tree generation proofs",
                            ));
                        }
                    }
                }
            }
            if live
                .values()
                .chain(keys.values().flat_map(PublishedKeyIndex::trees))
                .any(|tree| tree.root_page_id() != 0 && !proofs.contains_key(&tree.root_page_id()))
            {
                return Err(page_integrity(1, None, "live tree has no generation proof"));
            }
            reddb_file::page_proof::encode_catalog(metadata, &proofs)
                .map_err(|error| StoreError::Serialization(error.to_string()))?;
        }
        if self.config.resident_memory_bytes_max.is_some() {
            pager
                .fence_required_capabilities(reddb_file::FILE_CAPABILITY_DISK_RESIDENCY)
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
            use reddb_file::native_resume::{
                encode_catalog_resume, CatalogResume, CollectionResume,
            };
            let mut states = std::collections::BTreeMap::new();
            let mut entity_id_next = self.resident.next_entity_id.load(Ordering::SeqCst);
            let collections = self.resident.collections.read();
            let metadata_expiry_prepared = collections
                .values()
                .all(|manager| manager.metadata_expiry_projection_ready());
            let publishes_rows = {
                let summary = self.recovery_xid_summary.read();
                live.iter().any(|(name, tree)| {
                    let root = tree.root_page_id();
                    root != 0
                        && summary.as_ref().and_then(|state| state.roots.get(name)) != Some(&root)
                })
            };
            let xid_high_water_mark =
                self.base_xid_high_water_mark(&collections).map(|mut high| {
                    // Ordinary metadata flushes retain the old immutable-base
                    // bound; WAL overlays are observed at boot. Only replacement
                    // data roots require a borrowed visit over resident headers.
                    // The publication fence excludes concurrent header mutations.
                    if publishes_rows {
                        for manager in collections.values() {
                            manager.for_each_entity(|entity| {
                                high = high.max(entity.max_transaction_id());
                                true
                            });
                        }
                    }
                    if let Some(snapshots) = self.snapshot_manager.get() {
                        high = high.max(snapshots.peek_next_xid().saturating_sub(1));
                    }
                    high
                });
            for (name, manager) in collections.iter() {
                let tree = live.get(name);
                let (entity_next, row_id_next) = manager.identity_counters();
                entity_id_next = entity_id_next.max(entity_next);
                let root_page = tree.map_or(0, |tree| tree.root_page_id());
                let entities = if root_page == 0 {
                    0
                } else if let Some(count) = manager.backing_count_at_root(root_page) {
                    count
                } else {
                    // Changed projections may exclude pending/aborted rows. Count
                    // published physical keys, not the live manager's cardinality.
                    let mut cursor = tree
                        .expect("invariant: nonzero root has a tree")
                        .cursor_first()
                        .map_err(|error| {
                            StoreError::Io(std::io::Error::other(error.to_string()))
                        })?;
                    let mut count = 0_u64;
                    while cursor
                        .skip_next()
                        .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?
                    {
                        count += 1;
                    }
                    count
                };
                let metadata_expiries = if metadata_expiry_prepared {
                    let mut expiries = Vec::new();
                    if let Some(tree) = tree {
                        for (deadline_ms, entity_id) in manager.metadata_expiry_entries() {
                            if tree
                                .get(&entity_id.raw().to_be_bytes())
                                .map_err(|error| {
                                    StoreError::Io(std::io::Error::other(error.to_string()))
                                })?
                                .is_some()
                            {
                                expiries.push(reddb_file::native_resume::MetadataExpiryResume {
                                    deadline_ms,
                                    entity_id: entity_id.raw(),
                                });
                            }
                        }
                    }
                    expiries
                } else {
                    Vec::new()
                };
                states.insert(
                    name.clone(),
                    CollectionResume {
                        root_page,
                        entities,
                        entity_id_next: entity_next,
                        row_id_next,
                        sequence_id_next: manager.sequence_id_next(),
                        key_index_root: tree.and_then(|tree| {
                            keys.get(name)
                                .filter(|index| {
                                    index.indexed && index.source.ptr_eq(&Arc::downgrade(tree))
                                })
                                .map(|index| index.tree.root_page_id())
                        }),
                        secondary_indices: tree
                            .and_then(|tree| {
                                keys.get(name)
                                    .filter(|index| index.source.ptr_eq(&Arc::downgrade(tree)))
                            })
                            .map(|index| {
                                index
                                    .secondary
                                    .iter()
                                    .map(|(name, index)| (name.clone(), index.resume()))
                                    .collect()
                            })
                            .unwrap_or_default(),
                        metadata_expiries,
                    },
                );
            }
            let resume = CatalogResume {
                entity_id_next,
                xid_high_water_mark,
                metadata_expiry_prepared,
                collections: states,
            };
            encode_catalog_resume(metadata, &resume)
                .map_err(|error| StoreError::Serialization(error.to_string()))?;
            next_xid_summary = xid_high_water_mark.map(|high_water_mark| {
                super::recovery_xid::RecoveryXidSummary {
                    high_water_mark,
                    roots: resume
                        .collections
                        .into_iter()
                        .map(|(name, state)| (name, state.root_page))
                        .collect(),
                }
            });
        }
        // Record the exact pages before the catalog can forget their owners.
        // Recovery resolves this intent against the durable catalog epoch.
        let mut retirement_intent = reclaim.clone();
        retirement_intent.extend_from_slice(&released_pages);
        if !retirement_intent.is_empty() {
            pager
                .record_retirement_intent(&retirement_intent)
                .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        }
        // Do not mutate retirement state or the free list before both the new
        // children and the manifest that forgets old roots have been synced.
        publish_paged_metadata(pager, metadata)
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        // PAGE_RESULT_CACHE_FILE_VERSION predates the disk-resident cache
        // catalog. Publish the complete resume metadata first, then promote
        // the header so the next boot may trust and reuse its derived roots.
        if self.config.resident_memory_bytes_max.is_some()
            && pager
                .header()
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?
                .version
                == reddb_file::file_format::PAGE_RESULT_CACHE_FILE_VERSION
        {
            pager
                .fence_file_format_version(
                    reddb_file::file_format::PAGE_RESULT_CACHE_DISK_FILE_VERSION,
                )
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
        }
        *self.recovery_xid_summary.write() = next_xid_summary;
        self.ownership_catalog_epoch
            .store(catalog_epoch, Ordering::Release);
        #[cfg(test)]
        if !reclaim.is_empty()
            && std::env::var_os("REDDB_EXPERIMENTAL_EXIT_BEFORE_RETIRED_FREE").is_some()
        {
            eprintln!(
                "experimental exit before retired free pages={}",
                reclaim.len()
            );
            std::process::exit(184);
        }
        let mut index = 0;
        retired.retain_mut(|entry| {
            let keep = !remove.contains(&index);
            entry.published = true;
            index += 1;
            keep
        });
        if !reclaim.is_empty() {
            pager
                .free_retired_pages(&reclaim)
                .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
            #[cfg(test)]
            if std::env::var_os("REDDB_EXPERIMENTAL_EXIT_AFTER_RETIRED_ENQUEUE").is_some() {
                eprintln!(
                    "experimental exit after retired enqueue pages={}",
                    reclaim.len()
                );
                std::process::exit(186);
            }
            pager
                .sync()
                .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
            #[cfg(test)]
            if std::env::var_os("REDDB_EXPERIMENTAL_EXIT_AFTER_RETIRED_FREE").is_some() {
                eprintln!(
                    "experimental exit after retired free pages={}",
                    reclaim.len()
                );
                std::process::exit(185);
            }
        }
        pager
            .complete_deferred_publication(&retained_pages, &released_pages)
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        if !released_pages.is_empty() {
            pager
                .sync()
                .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        }
        Ok(())
    }

    pub(super) fn retire_btree(&self, tree: Arc<BTree>) {
        let mut retired = self.retired_btrees.lock();
        if tree.root_page_id() != 0 && !retired.iter().any(|entry| Arc::ptr_eq(&entry.tree, &tree))
        {
            retired.push(RetiredBTree {
                tree,
                published: false,
            });
        }
    }

    /// A completed inline insert already made this tree match its manager.
    /// Advance the reuse proof only for that single, fenced mutation. Any
    /// untracked edit or segment publication keeps the full comparison path.
    pub(crate) fn note_paged_insert(
        &self,
        collection: &str,
        manager: &Arc<SegmentManager>,
        before: &[(u64, u64)],
        id: EntityId,
    ) {
        let after = manager.checkpoint_revisions();
        if before.len() != after.len() {
            return;
        }
        let mut changed = 0;
        for ((old_id, old_revision), (new_id, new_revision)) in before.iter().zip(&after) {
            if old_id != new_id {
                return;
            }
            if old_revision != new_revision {
                if old_revision.checked_add(1) != Some(*new_revision) {
                    return;
                }
                changed += 1;
            }
        }
        if changed != 1 {
            return;
        }
        let mut versions = self.checkpoint_versions.write();
        let Some(previous) = versions.get_mut(collection) else {
            return;
        };
        if previous.manager.ptr_eq(&Arc::downgrade(manager))
            && previous.segments == before
            && previous.format_version == self.format_version()
            && !previous.pending_vault.contains(&id)
            && !previous.excluded_xids.contains(&0)
        {
            previous.segments = after;
        }
    }

    /// Successful physical deletes already changed both the manager and tree.
    /// Reuse remains valid only when their exact revision delta accounts for
    /// this batch, with no intervening edits or segment replacement.
    pub(crate) fn note_paged_delete(
        &self,
        collection: &str,
        manager: &Arc<SegmentManager>,
        before: &[(u64, u64)],
        deleted: &[EntityId],
    ) {
        let after = manager.checkpoint_revisions();
        if before.len() != after.len() {
            return;
        }
        let mut changes = 0_u64;
        for ((old_id, old_revision), (new_id, new_revision)) in before.iter().zip(&after) {
            if old_id != new_id {
                return;
            }
            let Some(delta) = new_revision.checked_sub(*old_revision) else {
                return;
            };
            let Some(total) = changes.checked_add(delta) else {
                return;
            };
            changes = total;
        }
        let Ok(expected) = u64::try_from(deleted.len()) else {
            return;
        };
        if changes != expected {
            return;
        }
        let mut versions = self.checkpoint_versions.write();
        let Some(previous) = versions.get_mut(collection) else {
            return;
        };
        if !previous.manager.ptr_eq(&Arc::downgrade(manager))
            || previous.format_version != self.format_version()
            || deleted.iter().any(|id| previous.pending_vault.contains(id))
            || previous.excluded_xids.contains(&0)
        {
            return;
        }
        if previous.segments != before {
            let Some(deferred) = previous.deferred_rows.as_ref() else {
                return;
            };
            let Some(changed) = manager.checkpoint_mutations_since(&previous.segments) else {
                return;
            };
            let removed: std::collections::HashSet<_> = deleted.iter().copied().collect();
            let pending: std::collections::BTreeSet<_> = deferred
                .iter()
                .copied()
                .chain(changed)
                .filter(|id| !removed.contains(id))
                .collect();
            if pending.len() > crate::storage::unified::segment::CHECKPOINT_MUTATIONS_MAX {
                return;
            }
            // GC already applied its deletes to the tree. Carry earlier dirty
            // IDs forward instead of spending their journal on these deletes.
            // Sparse projection still requires exact WAL proof for every
            // differing row, so an unlogged manager edit cannot be concealed.
            previous.has_unprojected_rows = !pending.is_empty();
            previous.deferred_rows = Some(Arc::new(pending.into_iter().collect()));
        }
        previous.segments = after;
    }

    pub(crate) fn mark_paged_registry_dirty(&self) {
        self.resident.mark_paged_registry_dirty()
    }

    /// Get (or lazily create) the per-collection B-tree under a *read*
    /// lock whenever possible. Returns a cloned `Arc<BTree>` so callers
    /// can mutate the tree without holding the outer map's RwLock —
    /// previously every insert serialised on `btree_indices.write()`,
    /// costing ~60% of the concurrent-insert throughput ceiling.
    pub(crate) fn get_or_create_btree(&self, collection: &str) -> Option<Arc<BTree>> {
        let collection = Self::canonical_collection(collection);
        let pager = self.pager.as_ref()?;
        if let Some(btree) = self.btree_indices.read().get(collection).cloned() {
            return Some(btree);
        }
        let mut write = self.btree_indices.write();
        let btree = write
            .entry(collection.to_string())
            .or_insert_with(|| Arc::new(BTree::new(Arc::clone(pager))))
            .clone();
        Some(btree)
    }

    pub(crate) fn flush_paged_state(&self) -> Result<(), StoreError> {
        if self.is_read_only() {
            return Ok(());
        }
        let Some(pager) = &self.pager else {
            return Ok(());
        };

        if self.resident.paged_registry_dirty.load(Ordering::Acquire) {
            self.flush_paged_registry()?;
            self.resident
                .paged_registry_dirty
                .store(false, Ordering::Release);
            return Ok(());
        }

        pager
            .flush()
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))
    }

    pub(crate) fn flush_paged_registry(&self) -> Result<(), StoreError> {
        if self.is_read_only() {
            return Ok(());
        }
        let _vault_guard = self.vault_publication_lock.lock();
        let Some(pager) = &self.pager else {
            return Ok(());
        };
        let _ownership_guard = pager.lock_ownership_mutations();
        struct RegistryPublication<'a> {
            pager: &'a Pager,
            active: bool,
        }
        impl Drop for RegistryPublication<'_> {
            fn drop(&mut self) {
                if self.active {
                    self.pager.fence_interrupted_ownership();
                }
            }
        }
        let epoch = if let Some(commit) = &self.commit {
            if self.ownership_active_epoch.load(Ordering::Acquire) == 0 {
                commit.force_sync().map_err(StoreError::Io)?;
                let epoch = commit
                    .next_page_ownership_epoch(self)
                    .map_err(StoreError::Io)?;
                commit
                    .begin_page_ownership_epoch(self, epoch)
                    .map_err(StoreError::Io)?;
                Some(epoch)
            } else {
                None
            }
        } else {
            None
        };
        let mut publication = RegistryPublication {
            pager,
            active: epoch.is_some(),
        };

        match pager.read_manifest_page() {
            Ok(_) => {}
            Err(PagerError::PageNotFound(_)) => {
                let meta_page = pager
                    .allocate_page(crate::storage::engine::PageType::Header)
                    .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
                pager
                    .write_page(meta_page.page_id(), meta_page)
                    .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
            }
            Err(e) => {
                return Err(StoreError::Io(std::io::Error::other(e.to_string())));
            }
        }

        let format_version = self.format_version().max(STORE_VERSION_V11);
        self.set_format_version(format_version);

        let collections = self.resident.collections.read();
        let btree_indices = self.btree_indices.read();
        let mut collection_roots = Vec::with_capacity(collections.len());
        for (name, _) in collections.iter() {
            let root_page = btree_indices
                .get(name)
                .map_or(0, |btree| btree.root_page_id());
            collection_roots.push((name.clone(), root_page));
        }
        drop(btree_indices);
        drop(collections);

        let mut meta_data = Vec::with_capacity(4096);
        reddb_file::encode_native_paged_metadata_header(
            &mut meta_data,
            reddb_file::NativePagedMetadataHeader {
                format_version,
                collection_count: collection_roots.len() as u32,
            },
        );
        for (name, root_page) in &collection_roots {
            reddb_file::encode_native_paged_collection_root(&mut meta_data, name, *root_page);
        }

        let excluded = self
            .snapshot_manager
            .get()
            .map_or_else(Default::default, |manager| {
                manager.checkpoint_excluded_state().1
            });
        let hidden_graph_ids = self
            .resident
            .snapshot_excluded_graph_ids(&Default::default(), &excluded);
        let cross_refs = self.resident.cross_refs.read();
        let total_refs: usize = cross_refs
            .iter()
            .map(|(source, refs)| {
                if hidden_graph_ids.contains(source) {
                    0
                } else {
                    refs.iter()
                        .filter(|(target, _, _)| !hidden_graph_ids.contains(target))
                        .count()
                }
            })
            .sum();
        meta_data.extend_from_slice(&(total_refs as u32).to_le_bytes());
        for (source_id, refs) in cross_refs.iter() {
            if hidden_graph_ids.contains(source_id) {
                continue;
            }
            for (target_id, ref_type, collection) in refs {
                if hidden_graph_ids.contains(target_id) {
                    continue;
                }
                reddb_file::encode_native_paged_cross_ref(
                    &mut meta_data,
                    source_id.raw(),
                    target_id.raw(),
                    ref_type.to_byte(),
                    reddb_engine::storage::system_collection_names::canonical_or_self(collection),
                );
            }
        }

        self.publish_reclaimable_trees(
            pager,
            &mut meta_data,
            &self.btree_indices.read(),
            &self.key_indices.read(),
        )?;
        if let Some(epoch) = epoch {
            self.commit
                .as_ref()
                .expect("registry epoch has a coordinator")
                .finish_count_batch_ownership_epoch(self, epoch)
                .map_err(StoreError::Io)?;
            publication.active = false;
        }
        Ok(())
    }

    /// Get a reference to the underlying pager (if in paged mode).
    pub fn pager(&self) -> Option<&Arc<Pager>> {
        self.pager.as_ref()
    }

    pub(crate) fn ensure_pager_writeback_available(&self) -> Result<(), StoreError> {
        self.pager.as_ref().map_or(Ok(()), |pager| {
            pager
                .check_writeback_available()
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))
        })
    }

    pub(crate) fn is_read_only(&self) -> bool {
        self.pager
            .as_ref()
            .is_some_and(|pager| pager.is_read_only())
    }

    /// Bytes of page-cache memory currently occupied by resident pages
    /// (ADR 0073 §2). Slots are fixed size, so occupancy times the page size
    /// *is* the footprint; a store without a pager occupies none.
    pub fn page_cache_bytes_in_use(&self) -> u64 {
        self.pager.as_ref().map_or(0, |pager| {
            pager.cache_len() as u64 * crate::storage::memory_pools::PAGE_CACHE_PAGE_SIZE_BYTES
        })
    }

    /// Bytes of WAL buffer memory held by this store's group-commit
    /// coordinator: queued records plus the writer's fixed append buffer.
    pub fn wal_buffer_bytes_in_use(&self) -> u64 {
        self.commit
            .as_ref()
            .map_or(0, |commit| commit.buffered_bytes())
    }

    /// Approximate resident bytes across every collection's segment arena
    /// (ADR 0073 §2). Same number `stats().total_memory_bytes` reports, without
    /// building the per-collection stats map a sampler would throw away.
    pub fn segment_memory_bytes(&self) -> u64 {
        let managers: Vec<Arc<SegmentManager>> = self
            .resident
            .collections
            .read()
            .values()
            .map(Arc::clone)
            .collect();

        managers
            .iter()
            .map(|manager| manager.resident_bytes())
            .fold(0, u64::saturating_add)
    }

    /// Borrow the immutable store configuration. Runtime hooks (e.g. the
    /// `auto_index_id` first-insert hook in `MutationEngine`) read knobs
    /// off this struct without going through the legacy global config tree.
    pub fn config(&self) -> &UnifiedStoreConfig {
        &self.config
    }

    pub fn with_config(config: UnifiedStoreConfig) -> Self {
        Self {
            config,
            resident: reddb_engine::storage::resident_store::ResidentStore::new(),
            format_version: AtomicU32::new(STORE_VERSION_V11),
            pager: None,
            db_path: None,
            operational_manifest: parking_lot::Mutex::new(None),
            btree_indices: RwLock::new(HashMap::new()),
            key_indices: RwLock::new(HashMap::new()),
            secondary_index_specs: RwLock::new(HashMap::new()),
            count_builds: RwLock::new(HashMap::new()),
            count_projection_seeds: parking_lot::Mutex::new(HashMap::new()),
            queue_dedup_cursors: parking_lot::Mutex::new(HashMap::new()),
            recovery_xid_summary: RwLock::new(None),
            checkpoint_versions: RwLock::new(HashMap::new()),
            checkpoint_durable_rows: RwLock::new(super::checkpoint_delta::DurableRows::default()),
            retired_btrees: parking_lot::Mutex::new(Vec::new()),
            ownership_catalog_epoch: AtomicU64::new(0),
            ownership_active_epoch: AtomicU64::new(0),
            detached_native_blob_pages: parking_lot::Mutex::new(Default::default()),
            commit: None,
            replayed_turbo_inserts: parking_lot::Mutex::new(HashMap::new()),
            replayed_probabilistic_deltas: parking_lot::Mutex::new(Vec::new()),
            result_cache_epoch: Arc::new(result_cache_epoch::ResultCacheEpochState::default()),
            snapshot_manager: std::sync::OnceLock::new(),
            embedded_commit_session: std::sync::OnceLock::new(),
            #[cfg(test)]
            test_fail_deferred_wal_append: std::sync::atomic::AtomicBool::new(false),
            #[cfg(test)]
            test_versioned_update_after_old: parking_lot::Mutex::new(None),
            #[cfg(test)]
            test_fail_versioned_update_phase: std::sync::atomic::AtomicU8::new(0),
            #[cfg(test)]
            test_require_deferred_publication_fence: std::sync::atomic::AtomicBool::new(false),
            vault_publication_lock: parking_lot::ReentrantMutex::new(()),
            row_publications: Default::default(),
            pending_vault_versions: RwLock::new(std::collections::HashSet::new()),
        }
    }

    /// Open or create a page-based database
    ///
    /// This uses the page engine for ACID durability with B-tree indices.
    /// The database file uses 16 KiB pages with checksums and efficient caching.
    ///
    /// # Arguments
    ///
    /// * `path` - Path to the database file (e.g., "data.rdb")
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let store = UnifiedStore::open("security.rdb")?;
    /// store.create_collection("hosts")?;
    /// // ... operations ...
    /// store.persist()?; // Flush to disk
    /// ```
    pub fn open(path: impl AsRef<Path>) -> Result<Self, StoreError> {
        Self::open_with_config(path, UnifiedStoreConfig::default())
    }

    pub fn open_with_config(
        path: impl AsRef<Path>,
        config: UnifiedStoreConfig,
    ) -> Result<Self, StoreError> {
        Self::open_with_config_for_recovery(path, config, false, None, None)
    }

    /// Recover a native database into a private read view. No WAL writer,
    /// checkpoint, manifest publication or allocator reclamation is attached.
    pub fn open_read_only_with_config(
        path: impl AsRef<Path>,
        mut config: UnifiedStoreConfig,
    ) -> Result<Self, StoreError> {
        let path = path.as_ref();
        config.embedded_wal_path = None;
        let pager = Pager::open(
            path,
            PagerConfig {
                read_only: true,
                create: false,
                disk_residency: config.resident_memory_bytes_max.is_some(),
                cache_size: config
                    .page_cache_slots
                    .unwrap_or_else(|| PagerConfig::default().cache_size),
                ..PagerConfig::default()
            },
        )
        .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
        pager
            .enable_deferred_reclamation()
            .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
        let mut store = Self::with_config(config);
        store.pager = Some(Arc::new(pager));
        store.db_path = Some(path.to_path_buf());
        let wal_path = reddb_file::layout::unified_wal_path(path);
        StoreCommitCoordinator::restore_checkpoint_page_images_from_path(&store, &wal_path)
            .map_err(StoreError::Io)?;
        let catalog = store.load_from_pages(false, None)?;
        StoreCommitCoordinator::replay_read_only(&store, &wal_path).map_err(StoreError::Io)?;
        store.load_result_cache_epoch()?;
        store.require_normalized_red_config_keys()?;
        let pending_drops = store.legacy_pending_collection_drops()?.unwrap_or_default();
        for name in pending_drops {
            if !catalog
                .collisions
                .iter()
                .any(|collision| collision.loser == name)
                && store.get_collection(&name).is_some()
            {
                store.drop_collection_in_memory(&name)?;
            }
        }
        Ok(store)
    }

    /// Only for an isolated, operator-selected recovery copy. The caller must
    /// first validate every WAL action addressing the discarded identity.
    pub(crate) fn open_with_config_for_recovery(
        path: impl AsRef<Path>,
        config: UnifiedStoreConfig,
        retain_canonical_system_config: bool,
        boot_wal: Option<&crate::storage::compact::boot_system_wal::BootSystemWalPlan>,
        catalog_plan: Option<&crate::storage::compact::catalog_aliases::CanonicalCatalogPlan>,
    ) -> Result<Self, StoreError> {
        let path = path.as_ref();
        let mut pager_config = PagerConfig {
            disk_residency: config.resident_memory_bytes_max.is_some(),
            ..PagerConfig::default()
        };
        // ADR 0073 §2 — the page cache is pre-sized from its budget share.
        // The pager's own `cache_size` default survives only for direct
        // library callers who never resolved a budget.
        if let Some(slots) = config.page_cache_slots {
            pager_config.cache_size = slots;
        }
        // Tunables via env — experimental, used by the benchmark harness
        // to compare durability profiles head-to-head with Postgres.
        // REDDB_DOUBLE_WRITE=0 requests skipping the double-write buffer,
        // which otherwise adds two fsyncs per pager flush (one on DWB, one
        // on the main file). The pager honors this only when the actual
        // data file is proven to live on a CoW filesystem with atomic page
        // writes; otherwise it fails closed and keeps DWB enabled.
        if matches!(
            std::env::var("REDDB_DOUBLE_WRITE").ok().as_deref(),
            Some("0") | Some("false") | Some("off")
        ) {
            pager_config.double_write = false;
        }
        let pager_open_started = std::time::Instant::now();
        let pager = Pager::open(path, pager_config)
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        pager
            .enable_deferred_reclamation()
            .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
        trace_startup_phase(path, "paged_pager_open", pager_open_started);

        let header = pager
            .header()
            .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
        if config.resident_memory_bytes_max.is_none() && header.requires_disk_residency() {
            return Err(StoreError::Io(std::io::Error::new(std::io::ErrorKind::Unsupported,
                "this database was prepared for disk residency; reopen with the disk-resident storage profile")));
        }

        let wal_path = reddb_file::layout::unified_wal_path(path);
        // Explicit transactions and WAL-only vault versions need committed
        // batches in every durability mode, including Strict's inline fsync.
        let commit = Some(Arc::new(
            StoreCommitCoordinator::open(wal_path, config.durability_mode, config.group_commit)
                .map_err(StoreError::Io)?,
        ));

        let store = Self {
            config,
            resident: reddb_engine::storage::resident_store::ResidentStore::new(),
            format_version: AtomicU32::new(STORE_VERSION_V11),
            pager: Some(Arc::new(pager)),
            db_path: Some(path.to_path_buf()),
            operational_manifest: parking_lot::Mutex::new(None),
            btree_indices: RwLock::new(HashMap::new()),
            key_indices: RwLock::new(HashMap::new()),
            secondary_index_specs: RwLock::new(HashMap::new()),
            count_builds: RwLock::new(HashMap::new()),
            count_projection_seeds: parking_lot::Mutex::new(HashMap::new()),
            queue_dedup_cursors: parking_lot::Mutex::new(HashMap::new()),
            recovery_xid_summary: RwLock::new(None),
            checkpoint_versions: RwLock::new(HashMap::new()),
            checkpoint_durable_rows: RwLock::new(super::checkpoint_delta::DurableRows::default()),
            retired_btrees: parking_lot::Mutex::new(Vec::new()),
            ownership_catalog_epoch: AtomicU64::new(0),
            ownership_active_epoch: AtomicU64::new(0),
            detached_native_blob_pages: parking_lot::Mutex::new(Default::default()),
            commit,
            replayed_turbo_inserts: parking_lot::Mutex::new(HashMap::new()),
            replayed_probabilistic_deltas: parking_lot::Mutex::new(Vec::new()),
            result_cache_epoch: Arc::new(result_cache_epoch::ResultCacheEpochState::default()),
            snapshot_manager: std::sync::OnceLock::new(),
            embedded_commit_session: std::sync::OnceLock::new(),
            #[cfg(test)]
            test_fail_deferred_wal_append: std::sync::atomic::AtomicBool::new(false),
            #[cfg(test)]
            test_versioned_update_after_old: parking_lot::Mutex::new(None),
            #[cfg(test)]
            test_fail_versioned_update_phase: std::sync::atomic::AtomicU8::new(0),
            #[cfg(test)]
            test_require_deferred_publication_fence: std::sync::atomic::AtomicBool::new(false),
            vault_publication_lock: parking_lot::ReentrantMutex::new(()),
            row_publications: Default::default(),
            pending_vault_versions: RwLock::new(std::collections::HashSet::new()),
        };

        // An interrupted structural edit may have reached home pages without
        // publishing its catalog. Restore those pages before following roots.
        let checkpoint_recovery_started = std::time::Instant::now();
        if let Some(commit) = &store.commit {
            commit
                .restore_checkpoint_page_images(&store)
                .map_err(StoreError::Io)?;
        }
        trace_startup_phase(
            path,
            "paged_checkpoint_recovery",
            checkpoint_recovery_started,
        );
        // Load existing data from pages if database exists
        let page_load_started = std::time::Instant::now();
        let catalog_load = store.load_from_pages(retain_canonical_system_config, catalog_plan)?;
        trace_startup_phase(path, "paged_catalog_and_roots", page_load_started);
        let wal_replay_started = std::time::Instant::now();
        if let Some(commit) = &store.commit {
            commit
                .replay_into(&store, boot_wal)
                .map_err(StoreError::Io)?;
        }
        store.load_result_cache_epoch()?;
        let normalized_config_keys = store.normalize_red_config_keys()?;
        if normalized_config_keys > 0 {
            tracing::warn!(
                migrated = normalized_config_keys,
                "normalized legacy red.config keys before boot config consumers"
            );
        }
        trace_startup_phase(path, "paged_wal_replay", wal_replay_started);
        if catalog_load.migrated {
            // Publish only the renamed catalog keys. Existing row, key and
            // secondary-index roots remain attached and are not rebuilt.
            store.mark_paged_registry_dirty();
            store.flush_paged_state()?;
        }
        let manifest_recovery_started = std::time::Instant::now();
        store.recover_operational_manifest(&catalog_load.collisions)?;
        if store.config.resident_memory_bytes_max.is_some() {
            let missing_key_index = {
                let trees = store.btree_indices.read();
                let keys = store.key_indices.read();
                trees.iter().any(|(name, tree)| {
                    !keys
                        .get(name)
                        .is_some_and(|index| index.source.ptr_eq(&Arc::downgrade(tree)))
                })
            };
            if missing_key_index {
                store.persist()?;
            }
        }
        trace_startup_phase(
            path,
            "paged_manifest_and_derived_roots",
            manifest_recovery_started,
        );

        Ok(store)
    }

    /// Replay changes the authoritative managers first. Build replacement trees
    /// only for changed collections, leaving every published tree untouched until
    /// all replacement pages are durable. Retain the WAL for auxiliary recovery.
    pub(crate) fn reconcile_replayed_btrees(&self) -> Result<(), StoreError> {
        let Some(pager) = &self.pager else {
            return Ok(());
        };
        let collections = self.resident.collections.read();
        let mut indices = self.btree_indices.write();
        let previous_versions = self.checkpoint_versions.read();
        let raw_replay =
            self.snapshot_manager.get().is_none() && self.pending_vault_versions.read().is_empty();
        for (name, manager) in collections.iter() {
            let mut reusable = None;
            if raw_replay {
                if let (Some(previous), Some(tree)) = (
                    previous_versions.get(name),
                    indices.get(name).filter(|tree| tree.has_page_proofs()),
                ) {
                    let mut current = CheckpointVersion {
                        manager: Arc::downgrade(manager),
                        segments: manager.checkpoint_revisions(),
                        visibility_revision: 0,
                        untracked_visibility_revision: 0,
                        excluded_xids: Arc::default(),
                        pending_vault: Arc::default(),
                        format_version: self.format_version(),
                        deferred_rows: Some(Arc::default()),
                        has_unprojected_rows: false,
                    };
                    if previous.manager.ptr_eq(&current.manager)
                        && previous.format_version == current.format_version
                    {
                        if previous.segments == current.segments
                            && !previous.has_unprojected_rows
                            && checkpoint_visibility_matches(previous, &current)
                        {
                            reusable = Some(Arc::clone(tree));
                        } else {
                            // The replay ownership epoch protects in-place
                            // changes with the same undo/publication protocol
                            // used by normal sparse checkpoints. Missing WAL
                            // proof, old formats and exhausted journals retain
                            // the full comparison/rebuild path below.
                            reusable = self.sparse_checkpoint_tree(
                                name,
                                manager,
                                tree,
                                previous,
                                &mut current,
                            )?;
                        }
                    }
                }
            }
            let tree = match reusable {
                Some(tree) => tree,
                None => {
                    self.checkpoint_tree(pager, manager, indices.get(name), |entity, metadata| {
                        #[cfg(test)]
                        REPLAY_SERIALIZED_ROWS.with(|count| count.set(count.get() + 1));
                        Some(Self::serialize_entity_record(
                            entity,
                            metadata.as_ref(),
                            self.format_version(),
                        ))
                    })?
                }
            };
            if indices
                .get(name)
                .is_some_and(|previous| Arc::ptr_eq(previous, &tree))
            {
                continue;
            }
            if let Some(previous) = indices.insert(name.clone(), tree) {
                self.retire_btree(previous);
            }
            self.mark_paged_registry_dirty();
            #[cfg(test)]
            if std::env::var_os("REDDB_EXPERIMENTAL_EXIT_DURING_REPLAY_REBUILD").is_some() {
                if std::env::var_os("REDDB_EXPERIMENTAL_REPLAY_NO_SYNC").is_none() {
                    pager.sync().map_err(|error| {
                        StoreError::Io(std::io::Error::other(error.to_string()))
                    })?;
                }
                eprintln!("experimental replay rebuild exit before manifest");
                std::process::exit(182);
            }
        }
        drop(previous_versions);
        drop(indices);
        drop(collections);
        self.flush_paged_state()?;
        #[cfg(test)]
        if std::env::var_os("REDDB_EXPERIMENTAL_EXIT_AFTER_REPLAY_PUBLICATION").is_some() {
            eprintln!("experimental replay exit after manifest");
            std::process::exit(182);
        }
        // Opening has no concurrent writers. Replay has now durably reconciled
        // these raw trees; do not immediately rebuild them at the first flush.
        if self.snapshot_manager.get().is_none() && self.pending_vault_versions.read().is_empty() {
            let collections = self.resident.collections.read();
            let mut versions = self.checkpoint_versions.write();
            for (name, manager) in collections.iter() {
                versions.insert(
                    name.clone(),
                    CheckpointVersion {
                        manager: Arc::downgrade(manager),
                        segments: manager.checkpoint_revisions(),
                        visibility_revision: 0,
                        untracked_visibility_revision: 0,
                        excluded_xids: Arc::default(),
                        pending_vault: Arc::default(),
                        format_version: self.format_version(),
                        deferred_rows: Some(Arc::default()),
                        has_unprojected_rows: false,
                    },
                );
            }
        }
        Ok(())
    }

    fn legacy_pending_collection_drops(&self) -> Result<Option<Vec<String>>, StoreError> {
        match self.try_get_config("storage.legacy_manifest_imported")? {
            Some(Value::Boolean(true)) => return Ok(None),
            None => {}
            Some(_) => {
                return Err(StoreError::Serialization(
                    "invalid legacy manifest import marker".into(),
                ))
            }
        }
        let Some(path) = &self.db_path else {
            return Ok(None);
        };
        self.operational_manifest_for_path(path)
            .read_only_pending_drops_if_present()
            .map_err(|err| storage_integrity("manifest", "current", None, err.to_string()))
    }

    fn recover_operational_manifest(
        &self,
        collisions: &[SystemCollectionCollision],
    ) -> Result<(), StoreError> {
        let Some(pending) = self.legacy_pending_collection_drops()? else {
            return Ok(());
        };
        // Consume legacy drop intents once, without rewriting their manifest.
        // WAL replay restores both the drops and this marker after a crash.
        for name in pending {
            if !collisions.iter().any(|collision| collision.loser == name)
                && self.get_collection(&name).is_some()
            {
                self.drop_collection(&name)?;
            }
        }
        self.try_set_config_tree(
            "storage.legacy_manifest_imported",
            &crate::serde_json::Value::Bool(true),
        )?;
        Ok(())
    }

    /// Read the commit boundary before following potentially interrupted trees.
    pub(super) fn published_ownership_epoch(&self) -> Result<u64, StoreError> {
        let Some(pager) = &self.pager else {
            return Ok(0);
        };
        let Some(content) = read_meta_payload(pager)? else {
            return Ok(0);
        };
        reddb_file::native_store::decode_native_paged_catalog_ownership_epoch(&content)
            .map(|epoch| epoch.unwrap_or(0))
            .map_err(|error| StoreError::Serialization(error.to_string()))
    }

    /// Read the B-tree indices and reconstruct collections from pages.
    fn load_from_pages(
        &self,
        retain_canonical_system_config: bool,
        catalog_plan: Option<&crate::storage::compact::catalog_aliases::CanonicalCatalogPlan>,
    ) -> Result<CatalogLoadOutcome, StoreError> {
        let pager = match &self.pager {
            Some(p) => p,
            None => return Ok(CatalogLoadOutcome::default()), // No pager, nothing to load
        };

        // Get page count
        let page_count = pager.page_count().map_err(|e| {
            StoreError::Io(std::io::Error::other(format!(
                "failed to read page count: {}",
                e
            )))
        })?;
        if page_count <= 1 {
            // Empty database (only header page)
            return Ok(CatalogLoadOutcome::default());
        }

        // Read metadata starting from page 1 (collections registry). The
        // helper transparently follows the `RDM3` overflow chain when the
        // metadata blob spans multiple pages and falls back to the legacy
        // `<data>-meta` shadow when page 1 itself is corrupted.
        if let Some(content_vec) = read_meta_payload(pager)? {
            let content: &[u8] = &content_vec;
            let header = pager
                .header()
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
            let prepared = header.requires_disk_residency();
            let proofs = reddb_file::native_store::decode_native_paged_catalog_tree_proofs(content)
                .map_err(|error| StoreError::Serialization(error.to_string()))?;
            if proofs.is_some()
                && (header.required_capabilities
                    & reddb_file::file_format::FILE_CAPABILITY_TREE_PROOFS
                    == 0
                    || pager
                        .manifest_reference()
                        .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?
                        .page_id
                        == 0)
            {
                return Err(page_integrity(
                    1,
                    None,
                    "tree proofs require an independently sealed catalog and reader capability",
                ));
            }
            if proofs.as_ref().is_some_and(|proofs| {
                proofs.iter().any(|(&root, proof)| {
                    root >= header.page_count || proof.reference.page_id >= header.page_count
                })
            }) {
                return Err(page_integrity(
                    1,
                    None,
                    "tree proof points outside allocation",
                ));
            }
            let resume = if prepared {
                reddb_file::native_resume::decode_catalog_resume(content)
                    .map_err(|error| StoreError::Serialization(error.to_string()))?
            } else {
                None
            };
            let mut outcome = CatalogLoadOutcome::default();
            let mut ignored_collections = std::collections::HashSet::new();
            let recovery_config_identities = resume.as_ref().map_or([false; 2], |resume| {
                [
                    resume.collections.contains_key("red_config"),
                    resume.collections.contains_key("red.config"),
                ]
            });
            if retain_canonical_system_config
                && !recovery_config_identities.iter().any(|present| *present)
            {
                return Err(page_integrity(
                    1,
                    None,
                    "canonical config recovery requires a resume identity",
                ));
            }
            if let Some(resume) = &resume {
                for &(legacy, canonical) in
                    reddb_engine::storage::system_collection_names::LEGACY_NAMES
                {
                    let (Some(legacy_state), Some(canonical_state)) = (
                        resume.collections.get(legacy),
                        resume.collections.get(canonical),
                    ) else {
                        continue;
                    };
                    let merged_resume = merge_shared_alias_resume(legacy_state, canonical_state);
                    let shared_physical_state = merged_resume.is_some();
                    let resume_counter_drift = legacy_state != canonical_state;
                    let (survivor, loser) = match (
                        legacy_state.entities == 0,
                        canonical_state.entities == 0,
                    ) {
                        _ if retain_canonical_system_config && legacy == "red_config" => {
                            (canonical, legacy)
                        }
                        _ if catalog_plan.is_some_and(|plan| plan.retires(legacy, canonical)) => {
                            (canonical, legacy)
                        }
                        (false, true) => (legacy, canonical),
                        (true, false) | (true, true) => (canonical, legacy),
                        (false, false) if shared_physical_state => (canonical, legacy),
                        (false, false) => {
                            let (
                                data_root_match,
                                key_index_root_match,
                                secondary_index_state_match,
                            ) = shared_alias_resume_matches(legacy_state, canonical_state);
                            return Err(page_integrity(
                                1,
                                None,
                                format!(
                                    "catalog contains both populated legacy collection '{legacy}' and canonical collection '{canonical}' (data_root_match={data_root_match}, key_index_root_match={key_index_root_match}, secondary_index_state_match={secondary_index_state_match})"
                                ),
                            ));
                        }
                    };
                    ignored_collections.insert(loser.to_string());
                    outcome.collisions.push(SystemCollectionCollision {
                        survivor: survivor.to_string(),
                        loser: loser.to_string(),
                        canonical: canonical.to_string(),
                        shared_physical_state,
                        resume_counter_drift,
                        merged_resume,
                    });
                }
                let live_roots = resume
                    .collections
                    .iter()
                    .filter(|(name, _)| !ignored_collections.contains(name.as_str()))
                    .flat_map(|(_, state)| {
                        std::iter::once(state.root_page)
                            .chain(state.key_index_root)
                            .chain(
                                state
                                    .secondary_indices
                                    .values()
                                    .map(|index| index.root_page),
                            )
                    })
                    .filter(|root| *root != 0)
                    .collect::<std::collections::HashSet<_>>();
                for collision in &outcome.collisions {
                    if collision.shared_physical_state {
                        continue;
                    }
                    let loser_state = resume
                        .collections
                        .get(&collision.loser)
                        .expect("collision loser is present");
                    if std::iter::once(loser_state.root_page)
                        .chain(loser_state.key_index_root)
                        .chain(
                            loser_state
                                .secondary_indices
                                .values()
                                .map(|index| index.root_page),
                        )
                        .filter(|root| *root != 0)
                        .any(|root| live_roots.contains(&root))
                    {
                        return Err(page_integrity(
                            1,
                            None,
                            format!(
                                "empty collection '{}' aliases a live collection root",
                                collision.loser
                            ),
                        ));
                    }
                }
                reddb_engine::storage::system_collection_names::migration_plan(
                    resume
                        .collections
                        .keys()
                        .map(String::as_str)
                        .filter(|name| !ignored_collections.contains(*name)),
                )
                .map_err(|(legacy, canonical)| {
                    page_integrity(
                        1,
                        None,
                        format!(
                            "catalog contains both legacy collection '{legacy}' and canonical collection '{canonical}'"
                        ),
                    )
                })?;
                for state in resume.collections.values() {
                    if !state.secondary_indices.is_empty()
                        && header.version
                            < reddb_file::file_format::PAGE_SECONDARY_INDEX_FILE_VERSION
                    {
                        return Err(page_integrity(
                            1,
                            None,
                            "secondary roots require the file format fence",
                        ));
                    }
                    for root in std::iter::once(state.root_page)
                        .chain(state.key_index_root)
                        .chain(
                            state
                                .secondary_indices
                                .values()
                                .map(|index| index.root_page),
                        )
                    {
                        if root >= page_count {
                            return Err(page_integrity(
                                root,
                                None,
                                "published root exceeds the database page count",
                            ));
                        }
                    }
                }
                self.resident
                    .next_entity_id
                    .fetch_max(resume.entity_id_next, Ordering::SeqCst);
            }
            *self.recovery_xid_summary.write() = resume.as_ref().and_then(|resume| {
                resume.xid_high_water_mark.map(|high_water_mark| {
                    super::recovery_xid::RecoveryXidSummary {
                        high_water_mark,
                        roots: resume
                            .collections
                            .iter()
                            .filter(|(name, _)| !ignored_collections.contains(name.as_str()))
                            .map(|(name, state)| {
                                (
                                    reddb_engine::storage::system_collection_names::canonical_or_self(name)
                                        .to_owned(),
                                    state.root_page,
                                )
                            })
                            .collect(),
                    }
                })
            });
            if content.len() >= 4 {
                let mut pos = 0;
                let mut format_version = STORE_VERSION_V1;

                let collection_count = if let Some(header) =
                    reddb_file::decode_native_paged_metadata_header(content)
                        .map_err(|err| StoreError::Serialization(err.to_string()))?
                {
                    format_version = header.format_version;
                    pos += reddb_file::METADATA_HEADER_BYTES;
                    header.collection_count as usize
                } else {
                    let count = u32::from_le_bytes([
                        content[pos],
                        content[pos + 1],
                        content[pos + 2],
                        content[pos + 3],
                    ]) as usize;
                    pos += 4;
                    count
                };

                self.set_format_version(format_version);

                if pos > content.len() {
                    return Ok(CatalogLoadOutcome::default());
                }

                if resume.is_none() {
                    // Old native catalogs have no advisory entity counters. A
                    // zero root is the only proof that an alias owns no rows;
                    // two nonzero roots still require explicit reconciliation.
                    let mut roots = std::collections::HashMap::new();
                    let mut root_pos = pos;
                    for _ in 0..collection_count {
                        let root =
                            reddb_file::decode_native_paged_collection_root(content, &mut root_pos)
                                .map_err(|err| page_integrity(1, None, err.to_string()))?;
                        if roots.insert(root.collection, root.root_page).is_some() {
                            return Err(page_integrity(
                                1,
                                None,
                                "duplicate raw collection identity",
                            ));
                        }
                    }
                    for &(legacy, canonical) in
                        reddb_engine::storage::system_collection_names::LEGACY_NAMES
                    {
                        let (Some(&legacy_root), Some(&canonical_root)) =
                            (roots.get(legacy), roots.get(canonical))
                        else {
                            continue;
                        };
                        let (survivor, loser) = match (legacy_root == 0, canonical_root == 0) {
                            (false, true) => (legacy, canonical),
                            (true, _) => (canonical, legacy),
                            (false, false) => continue,
                        };
                        ignored_collections.insert(loser.to_owned());
                        outcome.collisions.push(SystemCollectionCollision {
                            survivor: survivor.to_owned(),
                            loser: loser.to_owned(),
                            canonical: canonical.to_owned(),
                            shared_physical_state: false,
                            resume_counter_drift: false,
                            merged_resume: None,
                        });
                    }
                }
                let mut source_names = std::collections::HashMap::<String, String>::new();
                let mut migrated_catalog = !outcome.collisions.is_empty();
                let mut empty_duplicate_roots = std::collections::HashSet::new();
                let mut recovery_config_roots = [0u8; 2];
                // Older catalogs have no resume state, so the bounded loader
                // must already decode every physical entity to restore IDs.
                // Reuse that complete pass for the transaction floor too.
                let mut scanned_xid_high_water_mark = 0u64;
                let mut count_seed_bytes_remaining = self
                    .config
                    .resident_memory_bytes_max
                    .and_then(|limit| usize::try_from(limit).ok())
                    .unwrap_or(0)
                    / 2;

                // Read collection names and their B-tree root pages
                for _ in 0..collection_count {
                    let root = reddb_file::decode_native_paged_collection_root(content, &mut pos)
                        .map_err(|err| page_integrity(1, None, err.to_string()))?;
                    // Root page ID for this collection's B-tree
                    let root_page = root.root_page;
                    let stored_name = root.collection;
                    if retain_canonical_system_config {
                        let config_index = match stored_name.as_str() {
                            "red_config" => Some(0),
                            "red.config" => Some(1),
                            _ => None,
                        };
                        if let Some(index) = config_index {
                            recovery_config_roots[index] =
                                recovery_config_roots[index].saturating_add(1);
                            let Some(state) = resume
                                .as_ref()
                                .and_then(|resume| resume.collections.get(&stored_name))
                            else {
                                return Err(page_integrity(
                                    1,
                                    None,
                                    "config collection has no matching resume identity",
                                ));
                            };
                            if state.root_page != root_page {
                                return Err(page_integrity(
                                    1,
                                    None,
                                    "config collection root differs from resume catalog",
                                ));
                            }
                        }
                    }
                    if ignored_collections.contains(&stored_name) {
                        if resume.is_none() && root_page == 0 {
                            continue;
                        }
                        let shared_physical_state = outcome
                            .collisions
                            .iter()
                            .find(|collision| collision.loser == stored_name)
                            .is_some_and(|collision| collision.shared_physical_state);
                        if shared_physical_state {
                            continue;
                        }
                        let state = resume
                            .as_ref()
                            .and_then(|resume| resume.collections.get(&stored_name))
                            .expect("ignored collection was proven by catalog resume");
                        for retired_root in std::iter::once(state.root_page)
                            .chain(state.key_index_root)
                            .chain(
                                state
                                    .secondary_indices
                                    .values()
                                    .map(|index| index.root_page),
                            )
                            .filter(|root| *root != 0)
                        {
                            empty_duplicate_roots.insert(retired_root);
                        }
                        continue;
                    }
                    let name = reddb_engine::storage::system_collection_names::canonical_or_self(
                        &stored_name,
                    )
                    .to_owned();
                    if let Some(previous) = source_names.insert(name.clone(), stored_name.clone()) {
                        if previous != stored_name {
                            return Err(page_integrity(
                                1,
                                None,
                                format!(
                                    "catalog contains both legacy collection '{previous}' and canonical collection '{name}'"
                                ),
                            ));
                        }
                    }
                    migrated_catalog |= name != stored_name;

                    // Hydrate the collection in memory only. Loading must
                    // not emit WAL entries or rewrite the on-disk registry
                    // before the existing B-tree roots are attached.
                    let _ = self.create_collection_in_memory(&name);

                    let shared_alias = outcome.collisions.iter().find(|collision| {
                        collision.survivor == stored_name && collision.shared_physical_state
                    });
                    let counters = shared_alias
                        .and_then(|collision| collision.merged_resume.as_ref())
                        .or_else(|| {
                            resume
                                .as_ref()
                                .and_then(|resume| resume.collections.get(&stored_name))
                        });
                    let manager = self.get_collection(&name);
                    let mut count_seed =
                        if counters.is_none() && self.commit.is_some() && !self.is_read_only() {
                            let schema =
                                manager.as_ref().and_then(|manager| manager.column_schema());
                            super::count_index::CountProjectionSeedCollector::new(
                                schema,
                                count_seed_bytes_remaining,
                            )
                        } else {
                            None
                        };
                    if let (Some(counters), Some(manager)) = (counters, self.get_collection(&name))
                    {
                        manager.restore_identity_counters(
                            counters.entity_id_next,
                            counters.row_id_next,
                        );
                        manager.restore_sequence_id_next(counters.sequence_id_next);
                        let expiry_entries = counters
                            .metadata_expiries
                            .iter()
                            .map(|entry| (entry.deadline_ms, EntityId::new(entry.entity_id)))
                            .collect::<Vec<_>>();
                        manager.restore_metadata_expiry_projection(
                            &expiry_entries,
                            resume
                                .as_ref()
                                .is_some_and(|resume| resume.metadata_expiry_prepared),
                        );
                    }
                    // Load B-tree with root page if it exists
                    if root_page > 0 || counters.is_some() {
                        let btree = restore_published_tree(pager, root_page, proofs.as_ref())?;

                        // Load all entities from B-tree into the collection
                        let mut cursor = btree.cursor_first().map_err(|err| {
                            page_integrity(root_page, Some(name.clone()), err.to_string())
                        })?;
                        let hydrate_read_only =
                            self.is_read_only() && self.config.resident_memory_bytes_max.is_none();
                        let mut persisted_entities = if hydrate_read_only {
                            0
                        } else {
                            counters.map_or(0, |state| state.entities)
                        };
                        while counters.is_none() || hydrate_read_only {
                            let next = cursor
                                .next_with_budget(
                                    self.config
                                        .resident_memory_bytes_max
                                        .and_then(|limit| usize::try_from(limit).ok())
                                        .unwrap_or(usize::MAX),
                                )
                                .map_err(|err| {
                                    page_integrity(root_page, Some(name.clone()), err.to_string())
                                })?;
                            let Some((key, value)) = next else {
                                break;
                            };
                            persisted_entities += 1;
                            // Deserialize entity from value bytes
                            let key_bytes: [u8; 8] = key.as_slice().try_into().map_err(|_| {
                                page_integrity(
                                    root_page,
                                    Some(name.clone()),
                                    "physical entity key must contain eight bytes",
                                )
                            })?;
                            let record = reddb_engine::storage::collection_read::EncodedRecord {
                                id: EntityId::new(u64::from_be_bytes(key_bytes)),
                                bytes: value,
                            };
                            let decoded = if let Some(limit) = self.config.resident_memory_bytes_max
                            {
                                record.decode_with_budget(
                                    self.format_version(),
                                    usize::try_from(limit).unwrap_or(usize::MAX),
                                )
                            } else {
                                record.decode(self.format_version())
                            };
                            let (entity, metadata) = decoded.map_err(|err| {
                                page_integrity(root_page, Some(name.clone()), err.to_string())
                            })?;
                            scanned_xid_high_water_mark =
                                scanned_xid_high_water_mark.max(entity.max_transaction_id());
                            let discard_count_seed = if let Some(seed) = count_seed.as_mut() {
                                match seed.push(&entity) {
                                    Ok(complete) => !complete,
                                    Err(error) => {
                                        return Err(page_integrity(
                                            root_page,
                                            Some(name.clone()),
                                            error.to_string(),
                                        ));
                                    }
                                }
                            } else {
                                false
                            };
                            if discard_count_seed {
                                count_seed = None;
                            }
                            if let Some(m) = &manager {
                                let id = entity.id;
                                if let EntityKind::TableRow { row_id, .. } = &entity.kind {
                                    m.register_row_id(*row_id);
                                }
                                if self.config.resident_memory_bytes_max.is_some() {
                                    m.restore_metadata_expiry(&entity, metadata.as_ref());
                                    self.register_entity_id(id);
                                    m.restore_sequence_id_next(
                                        entity.sequence_id.saturating_add(1),
                                    );
                                    m.restore_identity_counters(
                                        id.raw().saturating_add(1),
                                        m.identity_counters().1,
                                    );
                                    continue;
                                }
                                self.resident.context_index.index_entity(&name, &entity);
                                m.restore_snapshot_entity(entity.clone())?;
                                if let Some(metadata) = metadata {
                                    let _ = m.set_metadata(id, metadata);
                                }
                                self.register_entity_id(id);
                                if self.config.auto_index_refs {
                                    self.index_cross_refs(&entity, &name)?;
                                }
                            }
                        }

                        if let Some(seed) = count_seed.take() {
                            count_seed_bytes_remaining =
                                count_seed_bytes_remaining.saturating_sub(seed.bytes());
                            self.remember_count_projection_seed(&name, root_page, seed);
                        }

                        // Store the B-tree for future lookups
                        let btree = Arc::new(btree);
                        if shared_alias.is_some_and(|collision| collision.resume_counter_drift) {
                            persisted_entities = u64::try_from(btree.count().map_err(|err| {
                                page_integrity(root_page, Some(name.clone()), err.to_string())
                            })?)
                            .map_err(|_| {
                                page_integrity(
                                    root_page,
                                    Some(name.clone()),
                                    "shared collection entity count exceeds u64",
                                )
                            })?;
                        }
                        let key_tree = counters
                            .and_then(|state| state.key_index_root)
                            .map(|root| {
                                restore_published_tree(pager, root, proofs.as_ref()).map(Arc::new)
                            })
                            .transpose()?;
                        if let Some(counters) = counters {
                            self.key_indices.write().insert(
                                name.clone(),
                                PublishedKeyIndex {
                                    indexed: key_tree.is_some(),
                                    secondary: counters.secondary_indices.iter().map(|(name, index)| {
                                        Ok((name.clone(), super::secondary_index::PublishedSecondaryIndex {
                                            identity: super::secondary_index::SecondaryIndexIdentity {
                                                codec_version: index.codec_version,
                                                definition: index.definition,
                                            },
                                            tree: Arc::new(restore_published_tree(pager, index.root_page, proofs.as_ref())?),
                                        }))
                                    }).collect::<Result<_, StoreError>>()?,
                                    source: Arc::downgrade(&btree),
                                    tree: match &key_tree { Some(tree) => Arc::clone(tree), None => Arc::new(restore_published_tree(pager, 0, proofs.as_ref())?) },
                                },
                            );
                        }
                        if let (Some(manager), Some(limit)) =
                            (&manager, self.config.resident_memory_bytes_max)
                        {
                            manager.install_backing(
                                Arc::new(
                                    match key_tree {
                                        Some(keys) => reddb_engine::storage::collection_read::PagedCollection::with_key_index(Arc::clone(&btree), keys),
                                        None => reddb_engine::storage::collection_read::PagedCollection::new(Arc::clone(&btree)),
                                    },
                                ),
                                btree.root_page_id(),
                                format_version,
                                usize::try_from(limit).unwrap_or(usize::MAX),
                                persisted_entities,
                            );
                        }
                        self.btree_indices.write().insert(name.clone(), btree);
                        if format_version == STORE_VERSION_V11 {
                            if let Some(manager) = &manager {
                                self.checkpoint_versions.write().insert(
                                    name,
                                    CheckpointVersion {
                                        manager: Arc::downgrade(manager),
                                        segments: manager.checkpoint_revisions(),
                                        visibility_revision: 0,
                                        untracked_visibility_revision: 0,
                                        excluded_xids: Arc::default(),
                                        pending_vault: Arc::default(),
                                        format_version,
                                        deferred_rows: Some(Arc::default()),
                                        has_unprojected_rows: false,
                                    },
                                );
                            }
                        }
                    }
                }

                let expected_config_roots = recovery_config_identities.map(u8::from);
                if retain_canonical_system_config && recovery_config_roots != expected_config_roots
                {
                    return Err(page_integrity(
                        1,
                        None,
                        "canonical config recovery requires exactly one root for each resume identity",
                    ));
                }
                if retain_canonical_system_config
                    && (format_version < STORE_VERSION_V2 || pos + 4 > content.len())
                {
                    return Err(page_integrity(
                        1,
                        None,
                        "canonical config recovery requires a complete cross-reference catalog",
                    ));
                }
                if format_version >= STORE_VERSION_V2 && pos + 4 <= content.len() {
                    let cross_ref_count = u32::from_le_bytes([
                        content[pos],
                        content[pos + 1],
                        content[pos + 2],
                        content[pos + 3],
                    ]) as usize;
                    pos += 4;

                    for _ in 0..cross_ref_count {
                        let cross_ref =
                            match reddb_file::decode_native_paged_cross_ref(content, &mut pos) {
                                Ok(cross_ref) => cross_ref,
                                Err(error) if retain_canonical_system_config => {
                                    return Err(page_integrity(1, None, error.to_string()));
                                }
                                Err(_) => break,
                            };
                        if retain_canonical_system_config
                            && cross_ref.target_collection == "red_config"
                        {
                            return Err(page_integrity(
                                1,
                                None,
                                "legacy red_config has cross-references and cannot be retired",
                            ));
                        }
                        let source_id = EntityId::new(cross_ref.source_id);
                        let target_id = EntityId::new(cross_ref.target_id);
                        let ref_type = RefType::from_byte(cross_ref.ref_type);
                        let target_collection =
                            reddb_engine::storage::system_collection_names::canonical_owned(
                                cross_ref.target_collection,
                            );

                        self.resident
                            .cross_refs
                            .write()
                            .entry(source_id)
                            .or_default()
                            .push((target_id, ref_type, target_collection.clone()));

                        if let Some((collection, mut entity)) = self.get_any(source_id) {
                            let exists = entity.cross_refs().iter().any(|xref| {
                                xref.target == target_id
                                    && xref.ref_type == ref_type
                                    && xref.target_collection == target_collection
                            });
                            if !exists {
                                entity.cross_refs_mut().push(CrossRef::new(
                                    source_id,
                                    target_id,
                                    target_collection.clone(),
                                    ref_type,
                                ));
                                if let Some(manager) = self.get_collection(&collection) {
                                    let _ = manager.update(entity);
                                }
                            }
                        }
                    }
                }
                let retired = reddb_file::decode_native_paged_retired_trees(content, &mut pos)
                    .map_err(|e| StoreError::Serialization(e.to_string()))?;
                let already_retired = retired
                    .iter()
                    .copied()
                    .collect::<std::collections::HashSet<_>>();
                let mut live_roots: std::collections::HashSet<_> = self
                    .btree_indices
                    .read()
                    .values()
                    .map(|tree| tree.root_page_id())
                    .collect();
                live_roots.extend(
                    self.key_indices
                        .read()
                        .values()
                        .flat_map(PublishedKeyIndex::trees)
                        .map(|tree| tree.root_page_id()),
                );
                for root in retired {
                    if live_roots.contains(&root)
                        || root
                            >= pager
                                .page_count()
                                .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?
                    {
                        return Err(page_integrity(
                            root,
                            None,
                            "retired root aliases a live or missing page",
                        ));
                    }
                    self.retired_btrees.lock().push(RetiredBTree {
                        tree: Arc::new(restore_published_tree(pager, root, proofs.as_ref())?),
                        published: true,
                    });
                }
                for root in empty_duplicate_roots
                    .iter()
                    .copied()
                    .filter(|root| !already_retired.contains(root))
                {
                    self.retire_btree(Arc::new(restore_published_tree(
                        pager,
                        root,
                        proofs.as_ref(),
                    )?));
                }
                if proofs.as_ref().is_some_and(|proofs| {
                    proofs.iter().any(|(root, proof)| {
                        let owns = |id| {
                            live_roots.contains(id)
                                || already_retired.contains(id)
                                || empty_duplicate_roots.contains(id)
                        };
                        !owns(root) || owns(&proof.reference.page_id)
                    })
                }) {
                    return Err(page_integrity(
                        1,
                        None,
                        "unowned or aliased tree generation proof",
                    ));
                }
                let deferred = reddb_file::decode_native_paged_deferred_pages(content, &mut pos)
                    .map_err(|e| StoreError::Serialization(e.to_string()))?;
                if deferred
                    .iter()
                    .any(|id| live_roots.contains(id) || empty_duplicate_roots.contains(id))
                {
                    return Err(page_integrity(1, None, "deferred page aliases a live root"));
                }
                pager
                    .restore_deferred_reclamation(&deferred)
                    .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
                let ownership_epoch =
                    reddb_file::native_store::decode_native_paged_ownership_epoch(
                        content, &mut pos,
                    )
                    .map_err(|e| StoreError::Serialization(e.to_string()))?;
                self.ownership_catalog_epoch
                    .store(ownership_epoch.unwrap_or(0), Ordering::Release);
                if migrated_catalog {
                    self.mark_paged_registry_dirty();
                }
                if self.format_version() < STORE_VERSION_V11 {
                    self.set_format_version(STORE_VERSION_V11);
                }
                if self.config.resident_memory_bytes_max.is_some()
                    && resume.is_none()
                    && !migrated_catalog
                    && !retain_canonical_system_config
                {
                    self.remember_startup_xid_high_water_mark(scanned_xid_high_water_mark);
                }
                outcome.migrated = migrated_catalog;
                return Ok(outcome);
            }
        }

        if retain_canonical_system_config {
            return Err(page_integrity(
                1,
                None,
                "canonical config recovery requires a complete catalog",
            ));
        }
        if self.format_version() < STORE_VERSION_V11 {
            self.set_format_version(STORE_VERSION_V11);
        }

        Ok(CatalogLoadOutcome::default())
    }

    /// Deserialize an entity from binary bytes
    pub(crate) fn deserialize_entity(
        data: &[u8],
        format_version: u32,
    ) -> Result<UnifiedEntity, StoreError> {
        let mut pos = 0;
        Self::read_entity_binary(data, &mut pos, format_version)
            .map_err(|e| StoreError::Serialization(e.to_string()))
    }

    /// Serialize an entity to binary bytes
    pub(crate) fn serialize_entity(entity: &UnifiedEntity, format_version: u32) -> Vec<u8> {
        // Pre-allocate ~256 bytes to cover the typical 15-column
        // typed row without any Vec growth. Bulk insert calls this
        // millions of times per bench run; saving 2-3 reallocs per
        // entity amortises.
        let mut buf = Vec::with_capacity(256);
        Self::write_entity_binary(&mut buf, entity, format_version);
        buf
    }

    pub(crate) fn serialize_entity_record(
        entity: &UnifiedEntity,
        metadata: Option<&Metadata>,
        format_version: u32,
    ) -> Vec<u8> {
        reddb_engine::storage::metadata_codec::serialize_entity_record(
            entity,
            metadata,
            format_version,
        )
    }

    pub(crate) fn deserialize_entity_record(
        data: &[u8],
        format_version: u32,
    ) -> Result<(UnifiedEntity, Option<Metadata>), StoreError> {
        reddb_engine::storage::metadata_codec::deserialize_entity_record(data, format_version)
            .map_err(StoreError::Serialization)
    }

    /// Persist all data to page-based storage
    ///
    /// Writes all entities to B-tree pages and flushes to disk.
    /// This provides ACID durability guarantees.
    pub fn persist(&self) -> Result<(), StoreError> {
        self.persist_inner(false)
    }

    pub(crate) fn persist_after_runtime_snapshots(&self) -> Result<(), StoreError> {
        self.persist_inner(true)
    }

    fn persist_inner(&self, runtime_snapshots_durable: bool) -> Result<(), StoreError> {
        if self.is_read_only() {
            return Ok(());
        }
        self.ensure_pager_writeback_available()?;
        let _vault_guard = self.vault_publication_lock.lock();
        let pending_vault_versions = self.pending_vault_versions.read();
        let pager = match &self.pager {
            Some(p) => p,
            None => {
                // No pager attached - use binary file fallback if path available
                if let Some(path) = &self.db_path {
                    return self
                        .save_to_file(path)
                        .map_err(|e| StoreError::Serialization(e.to_string()));
                }
                return Err(StoreError::Io(std::io::Error::other(
                    "No pager or path configured for persistence",
                )));
            }
        };
        // Prevent unrelated allocations or frees from being attributed to
        // this checkpoint's ownership epoch.
        let _ownership_guard = pager.lock_ownership_mutations();

        match pager.read_manifest_page() {
            Ok(_) => {}
            Err(PagerError::PageNotFound(_)) => {
                let meta_page = pager
                    .allocate_page(crate::storage::engine::PageType::Header)
                    .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
                pager
                    .write_page(meta_page.page_id(), meta_page)
                    .map_err(|e| StoreError::Io(std::io::Error::other(e.to_string())))?;
            }
            Err(e) => {
                return Err(StoreError::Io(std::io::Error::other(e.to_string())));
            }
        }

        if let Some(commit) = &self.commit {
            commit.force_sync().map_err(StoreError::Io)?;
        }

        struct InterruptedOwnership<'a> {
            pager: &'a Pager,
            active: bool,
        }
        impl Drop for InterruptedOwnership<'_> {
            fn drop(&mut self) {
                if self.active {
                    self.pager.fence_interrupted_ownership();
                }
            }
        }
        let mut ownership = InterruptedOwnership {
            pager,
            active: self.commit.is_some(),
        };
        let epoch = if let Some(commit) = &self.commit {
            let epoch = commit
                .next_page_ownership_epoch(self)
                .map_err(StoreError::Io)?;
            commit
                .begin_page_ownership_epoch(self, epoch)
                .map_err(StoreError::Io)?;
            Some(epoch)
        } else {
            None
        };

        let (visibility_revision, excluded, untracked_visibility_revision) =
            self.snapshot_manager.get().map_or_else(
                || (0, Default::default(), 0),
                |manager| manager.checkpoint_projection_state(),
            );
        let hidden_graph_ids = self
            .resident
            .snapshot_excluded_graph_ids(&pending_vault_versions, &excluded);
        let collections = self.resident.collections.read();
        let mut btree_indices = self.btree_indices.write();

        // Collect collection names and their B-tree root pages
        let mut collection_roots: Vec<(String, u32)> = Vec::new();

        // Build unpublished replacements. Deleting from the current trees here
        // would destroy the last checkpoint if a write fails before publication.
        let mut rebuilt_indices = HashMap::with_capacity(collections.len());
        let previous_versions = self.checkpoint_versions.read();
        let mut checkpoint_versions = HashMap::with_capacity(collections.len());
        let pending_vault = Arc::new(pending_vault_versions.clone());
        let excluded = Arc::new(excluded);
        for (name, manager) in collections.iter() {
            let mut version = CheckpointVersion {
                manager: Arc::downgrade(manager),
                segments: manager.checkpoint_revisions(),
                visibility_revision,
                untracked_visibility_revision,
                excluded_xids: Arc::clone(&excluded),
                pending_vault: Arc::clone(&pending_vault),
                format_version: self.format_version(),
                deferred_rows: Some(Arc::default()),
                has_unprojected_rows: false,
            };
            if let (Some(previous), Some(tree)) = (
                previous_versions.get(name),
                btree_indices
                    .get(name)
                    .filter(|tree| tree.has_page_proofs()),
            ) {
                if hidden_graph_ids.is_empty()
                    && previous.manager.ptr_eq(&version.manager)
                    && previous.segments == version.segments
                    && previous.format_version == version.format_version
                    && !previous.has_unprojected_rows
                    && checkpoint_visibility_matches(previous, &version)
                {
                    version.deferred_rows = previous.deferred_rows.clone();
                    collection_roots.push((name.clone(), tree.root_page_id()));
                    rebuilt_indices.insert(name.clone(), Arc::clone(tree));
                    checkpoint_versions.insert(name.clone(), version);
                    continue;
                }
            }
            if let (Some(previous), Some(tree)) = (
                previous_versions.get(name),
                btree_indices
                    .get(name)
                    .filter(|tree| tree.has_page_proofs()),
            ) {
                if hidden_graph_ids.is_empty()
                    && previous.manager.ptr_eq(&version.manager)
                    && previous.format_version == version.format_version
                {
                    if let Some(tree) =
                        self.sparse_checkpoint_tree(name, manager, tree, previous, &mut version)?
                    {
                        collection_roots.push((name.clone(), tree.root_page_id()));
                        rebuilt_indices.insert(name.clone(), tree);
                        checkpoint_versions.insert(name.clone(), version);
                        continue;
                    }
                }
            }
            let btree = self.checkpoint_tree(
                pager,
                manager,
                btree_indices.get(name),
                |entity, metadata| {
                    #[cfg(test)]
                    CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(count.get() + 1));
                    let hidden = pending_vault_versions.contains(&entity.id)
                        || excluded.contains(&entity.xmin);
                    let normalize_xmax = excluded.contains(&entity.xmax);
                    let normalize_branch = entity.branch_version().is_some_and(|branch| {
                        branch
                            .deletions
                            .iter()
                            .any(|(_, xid)| excluded.contains(xid))
                    });
                    let hide_refs = entity.cross_refs().iter().any(|reference| {
                        hidden_graph_ids.contains(&reference.source)
                            || hidden_graph_ids.contains(&reference.target)
                    });
                    if hidden || normalize_xmax || normalize_branch || hide_refs {
                        version.remember_deferred_row(entity.id);
                    }
                    if hidden {
                        return None;
                    }
                    let mut normalized = std::borrow::Cow::Borrowed(entity);
                    if normalize_xmax {
                        normalized.to_mut().xmax = 0;
                    }
                    if normalize_branch {
                        normalized
                            .to_mut()
                            .retain_branch_deletions(|xid| !excluded.contains(&xid));
                    }
                    if hide_refs {
                        normalized.to_mut().cross_refs_mut().retain(|reference| {
                            !hidden_graph_ids.contains(&reference.source)
                                && !hidden_graph_ids.contains(&reference.target)
                        });
                    }
                    Some(Self::serialize_entity_record(
                        &normalized,
                        metadata.as_ref(),
                        self.format_version(),
                    ))
                },
            )?;
            checkpoint_versions.insert(name.clone(), version);

            collection_roots.push((name.clone(), btree.root_page_id()));
            rebuilt_indices.insert(name.clone(), btree);
        }

        // Write collection metadata to page 1
        let mut meta_data = Vec::with_capacity(4096);

        let format_version = self.format_version().max(STORE_VERSION_V11);
        self.set_format_version(format_version);

        reddb_file::encode_native_paged_metadata_header(
            &mut meta_data,
            reddb_file::NativePagedMetadataHeader {
                format_version,
                collection_count: collection_roots.len() as u32,
            },
        );

        // Write each collection's name and B-tree root page
        for (name, root_page) in &collection_roots {
            reddb_file::encode_native_paged_collection_root(&mut meta_data, name, *root_page);
        }

        // Write cross-reference metadata
        let cross_refs = self.resident.cross_refs.read();
        let total_refs: usize = cross_refs
            .iter()
            .map(|(source, refs)| {
                if hidden_graph_ids.contains(source) {
                    0
                } else {
                    refs.iter()
                        .filter(|(target, _, _)| !hidden_graph_ids.contains(target))
                        .count()
                }
            })
            .sum();
        meta_data.extend_from_slice(&(total_refs as u32).to_le_bytes());
        for (source_id, refs) in cross_refs.iter() {
            if hidden_graph_ids.contains(source_id) {
                continue;
            }
            for (target_id, ref_type, collection) in refs {
                if hidden_graph_ids.contains(target_id) {
                    continue;
                }
                reddb_file::encode_native_paged_cross_ref(
                    &mut meta_data,
                    source_id.raw(),
                    target_id.raw(),
                    ref_type.to_byte(),
                    collection,
                );
            }
        }

        for (name, previous) in btree_indices.iter() {
            if !rebuilt_indices
                .get(name)
                .is_some_and(|tree| Arc::ptr_eq(tree, previous))
            {
                self.retire_btree(Arc::clone(previous));
            }
        }
        #[cfg(test)]
        if std::env::var_os("REDDB_EXPERIMENTAL_EXIT_DURING_CHECKPOINT_REBUILD").is_some() {
            pager
                .sync()
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
            eprintln!("experimental checkpoint exit before manifest");
            std::process::exit(183);
        }
        let rebuilt_keys = self.checkpoint_key_indices(pager, &rebuilt_indices)?;
        #[cfg(test)]
        if std::env::var_os("REDDB_TEST_EXIT_AFTER_SECONDARY_INDEX_BUILD").is_some() {
            pager
                .sync()
                .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
            std::process::exit(196);
        }
        self.publish_reclaimable_trees(pager, &mut meta_data, &rebuilt_indices, &rebuilt_keys)?;

        #[cfg(test)]
        if std::env::var_os("REDDB_TEST_EXIT_AFTER_UNDO_CATALOG_PUBLICATION").is_some() {
            std::process::exit(195);
        }

        // Only the now-durable catalog may replace the live tree handles. The
        // previous generation is retained in the manifest retirement ledger.
        *btree_indices = rebuilt_indices;
        *self.key_indices.write() = rebuilt_keys;
        drop(previous_versions);
        *self.checkpoint_versions.write() = checkpoint_versions;

        if let Some(commit) = &self.commit {
            commit
                .finish_page_ownership_epoch(self, epoch.expect("WAL ownership epoch"))
                .map_err(StoreError::Io)?;
            ownership.active = false;
            if runtime_snapshots_durable {
                commit
                    .truncate_after_runtime_checkpoint()
                    .map_err(StoreError::Io)?;
            } else {
                commit.truncate().map_err(StoreError::Io)?;
            }
        }
        *self.checkpoint_durable_rows.write() = Default::default();

        if let Some(limit) = self.config.resident_memory_bytes_max {
            let mut versions = self.checkpoint_versions.write();
            for (name, manager) in collections.iter() {
                if versions
                    .get(name)
                    .is_some_and(|version| !version.has_unprojected_rows)
                {
                    if let Some(tree) = btree_indices.get(name) {
                        manager.install_backing(
                            Arc::new(
                                {
                                    let keys = self.key_indices.read();
                                    let index = keys.get(name).expect("invariant: prepared checkpoint classifies its key index");
                                    if index.indexed {
                                        reddb_engine::storage::collection_read::PagedCollection::with_key_index(Arc::clone(tree), Arc::clone(&index.tree))
                                    } else {
                                        reddb_engine::storage::collection_read::PagedCollection::new(Arc::clone(tree))
                                    }
                                },
                            ),
                            tree.root_page_id(),
                            self.format_version(),
                            usize::try_from(limit).unwrap_or(usize::MAX),
                            manager.count() as u64,
                        );
                        if let Some(version) = versions.get_mut(name) {
                            version.segments = manager.checkpoint_revisions();
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Check if the store is using page-based persistence
    pub fn is_paged(&self) -> bool {
        self.pager.is_some()
    }

    /// Current root page for a collection's primary B-tree, if one has
    /// been materialized in this store.
    pub(crate) fn collection_root_page(&self, collection: &str) -> Option<u32> {
        self.btree_indices
            .read()
            .get(Self::canonical_collection(collection))
            .map(|btree| btree.root_page_id())
            .filter(|root| *root != 0)
    }

    /// Keep the journal replay baseline for the lifetime of this store, including
    /// persistent resident layouts whose paged `db_path` is absent.
    pub(crate) fn operational_manifest_for_path(
        &self,
        path: &Path,
    ) -> reddb_file::OperationalManifest {
        let mut cached = self.operational_manifest.lock();
        if let Some((current, manifest)) = cached.as_ref() {
            if current == path {
                return manifest.clone();
            }
        }
        let manifest = reddb_file::OperationalManifest::for_db_path(path);
        *cached = Some((path.to_owned(), manifest.clone()));
        manifest
    }

    /// Get the database file path (if using paged mode)
    pub fn db_path(&self) -> Option<&Path> {
        self.db_path.as_deref()
    }
}

#[cfg(test)]
mod checkpoint_work_tests {
    use super::*;
    use reddb_file::PAGED_PAGE_SIZE as PAGE_SIZE;

    struct SystemNameProjection;

    impl super::super::secondary_index::SecondaryIndexProjector for SystemNameProjection {
        fn project(
            &self,
            entity: &UnifiedEntity,
            key: &mut [u8; super::super::secondary_index::SECONDARY_KEY_BYTES_MAX],
        ) -> Result<Option<usize>, StoreError> {
            key[..8].copy_from_slice(&entity.id.raw().to_be_bytes());
            Ok(Some(8))
        }
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn interrupted_inline_sparse_checkpoint_replays_durable_rows() {
        const CHILD: &str = "REDDB_TEST_INLINE_SPARSE_PATH";
        if let Some(path) = std::env::var_os(CHILD) {
            let insert = std::env::var_os("REDDB_TEST_INLINE_SPARSE_INSERT").is_some();
            let multi = std::env::var_os("REDDB_TEST_INLINE_SPARSE_MULTI").is_some();
            let store = UnifiedStore::open(path).expect("open store");
            let initial_count = if multi { 600 } else { 1 };
            for id in 1..=initial_count {
                store.insert_auto("rows", row(id, 1)).expect("seed");
            }
            store.persist().expect("baseline");
            let manager = store.get_collection("rows").expect("collection");
            let ids: Vec<u64> = if multi {
                (1..=initial_count).collect()
            } else {
                vec![if insert { 2 } else { 1 }]
            };
            let mut actions = Vec::new();
            for id in ids {
                if insert {
                    manager.insert(row(id, 9)).expect("insert version");
                } else {
                    manager.update(row(id, 9)).expect("update version");
                }
                let entity = manager.get(EntityId::new(id)).expect("changed entity");
                actions.push(StoreWalAction::upsert_entity(
                    "rows",
                    &entity,
                    None,
                    store.format_version(),
                ));
            }
            let references = store
                .commit
                .as_ref()
                .expect("coordinator")
                .append_checkpoint_actions(&actions)
                .expect("durable WAL");
            store.remember_checkpoint_rows(actions, references);
            store.persist().expect("exit after sparse write");
            panic!("inline sparse crash hook was not reached");
        }
        for (insert, multi) in [(false, false), (true, false), (false, true)] {
            let directory = tempfile::tempdir().expect("temporary store");
            let path = directory.path().join("data.rdb");
            let mut command =
                std::process::Command::new(std::env::current_exe().expect("test executable"));
            command
                .args(["--exact", "storage::unified::store::impl_pages::checkpoint_work_tests::interrupted_inline_sparse_checkpoint_replays_durable_rows", "--nocapture"])
                .env(CHILD, &path)
                .env("REDDB_TEST_EXIT_AFTER_INLINE_SPARSE_WRITE", "1");
            if insert {
                command.env("REDDB_TEST_INLINE_SPARSE_INSERT", "1");
            }
            if multi {
                command.env("REDDB_TEST_INLINE_SPARSE_MULTI", "1");
            }
            let output = command.output().expect("crash child");
            assert_eq!(
                output.status.code(),
                Some(190),
                "insert={insert} multi={multi}: {}",
                String::from_utf8_lossy(&output.stderr)
            );
            let recovered = UnifiedStore::open(&path).expect("recover");
            let ids: Vec<u64> = if multi {
                (1..=600).collect()
            } else {
                vec![if insert { 2 } else { 1 }]
            };
            for id in ids {
                assert_eq!(
                    recovered
                        .get("rows", EntityId::new(id))
                        .expect("recovered row")
                        .data
                        .as_row()
                        .expect("row")
                        .columns,
                    vec![Value::Integer(9)]
                );
            }
            assert_eq!(
                recovered.get_collection("rows").expect("rows").count(),
                if multi {
                    600
                } else if insert {
                    2
                } else {
                    1
                }
            );
        }
    }

    #[test]
    fn catalog_ownership_epoch_survives_reopen_and_later_publication() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        {
            let store = UnifiedStore::open(&path).expect("open store");
            store.create_collection("rows").expect("create collection");
            store.persist().expect("publish ownership epoch");
            assert_eq!(store.ownership_catalog_epoch.load(Ordering::Acquire), 1);
            store.persist().expect("publish next ownership epoch");
        }
        let reopened = UnifiedStore::open(&path).expect("reopen store");
        assert_eq!(reopened.ownership_catalog_epoch.load(Ordering::Acquire), 2);
    }

    #[test]
    fn metadata_overflow_with_missing_continuation_fails_closed() {
        use crate::storage::engine::{Page, PageType, HEADER_SIZE};

        let directory = tempfile::tempdir().expect("temporary store");
        let store = UnifiedStore::open(directory.path().join("data.rdb")).expect("paged store");
        let pager = store.pager.as_ref().unwrap();
        let valid_payload = vec![42; META_PAGE_CONTENT_CAP * 3];
        let (valid_manifest, _) = build_meta_page1_with_overflow(pager, &valid_payload, 1).unwrap();
        pager.write_page(1, valid_manifest).unwrap();
        assert!(!existing_overflow_chain(pager).unwrap().is_empty());
        assert_eq!(read_meta_payload(pager).unwrap().unwrap(), valid_payload);

        let missing = pager.page_count().unwrap() + 100;
        let mut manifest = Page::new(PageType::Header, 1);
        reddb_file::encode_native_metadata_overflow_header(
            &mut manifest.as_bytes_mut()[HEADER_SIZE..],
            reddb_file::NativeMetadataOverflowHeader {
                format_version: STORE_VERSION_V11,
                total_payload_bytes: META_PAGE_CONTENT_CAP as u32 + 1,
                next_overflow_page_id: missing,
            },
        )
        .unwrap();
        pager.write_page(1, manifest).unwrap();

        assert!(
            existing_overflow_chain(pager).is_err(),
            "a missing continuation must not be treated as an empty chain"
        );

        let wrong_type = pager.allocate_page(PageType::BTreeLeaf).unwrap();
        let mut manifest = Page::new(PageType::Header, 1);
        reddb_file::encode_native_metadata_overflow_header(
            &mut manifest.as_bytes_mut()[HEADER_SIZE..],
            reddb_file::NativeMetadataOverflowHeader {
                format_version: STORE_VERSION_V11,
                total_payload_bytes: META_PAGE_CONTENT_CAP as u32 + 1,
                next_overflow_page_id: wrong_type.page_id(),
            },
        )
        .unwrap();
        pager.write_page(1, manifest).unwrap();
        assert!(
            existing_overflow_chain(pager).is_err(),
            "a live page must not be mistaken for metadata overflow"
        );
        assert!(
            read_meta_payload(pager).is_err(),
            "a live page must not be read as metadata overflow"
        );

        let mut cycle = pager.allocate_page(PageType::Overflow).unwrap();
        let cycle_id = cycle.page_id();
        reddb_file::encode_native_metadata_overflow_continuation_header(
            &mut cycle.as_bytes_mut()[HEADER_SIZE..],
            reddb_file::NativeMetadataOverflowContinuationHeader {
                next_overflow_page_id: cycle_id,
                chunk_bytes: 1,
            },
        )
        .unwrap();
        pager.write_page(cycle_id, cycle).unwrap();
        let mut manifest = Page::new(PageType::Header, 1);
        reddb_file::encode_native_metadata_overflow_header(
            &mut manifest.as_bytes_mut()[HEADER_SIZE..],
            reddb_file::NativeMetadataOverflowHeader {
                format_version: STORE_VERSION_V11,
                total_payload_bytes: META_V3_FIRST_PAYLOAD_CAP as u32 + 3,
                next_overflow_page_id: cycle_id,
            },
        )
        .unwrap();
        pager.write_page(1, manifest).unwrap();
        assert!(
            read_meta_payload(pager).is_err(),
            "cyclic overflow must not be accepted"
        );

        let mut manifest = Page::new(PageType::Header, 1);
        reddb_file::encode_native_metadata_overflow_header(
            &mut manifest.as_bytes_mut()[HEADER_SIZE..],
            reddb_file::NativeMetadataOverflowHeader {
                format_version: STORE_VERSION_V11,
                total_payload_bytes: META_PAGE_CONTENT_CAP as u32 + 1,
                next_overflow_page_id: 0,
            },
        )
        .unwrap();
        pager.write_page(1, manifest).unwrap();
        assert!(
            read_meta_payload(pager).is_err(),
            "a truncated metadata payload must not be accepted"
        );
    }

    #[test]
    fn retired_tree_pages_remain_pinned_by_a_cursor() {
        let directory = tempfile::tempdir().expect("temporary store");
        let store = UnifiedStore::open(&directory.path().join("data.rdb")).expect("paged store");
        for id in 1..=100 {
            store.insert_auto("rows", row(id, 1)).expect("seed");
        }
        store.persist().expect("initial checkpoint");
        let tree = store
            .btree_indices
            .read()
            .get("rows")
            .cloned()
            .expect("tree");
        let mut cursor = tree.cursor_first().expect("cursor");
        drop(tree);
        let manager = store.get_collection("rows").expect("rows");
        for value in 2..=6 {
            manager.update(row(1, value)).expect("change");
            store.persist().expect("checkpoint with old reader");
        }
        let mut count = 0;
        while let Some((_, bytes)) = cursor.next().expect("pinned tree must stay readable") {
            let (entity, _) =
                UnifiedStore::deserialize_entity_record(&bytes, store.format_version())
                    .expect("decode");
            assert_eq!(
                entity.data.as_row().expect("row").columns,
                [Value::Integer(1)]
            );
            count += 1;
        }
        assert_eq!(count, 100);
        drop(cursor);
        store.persist().expect("reclaim after reader release");
        assert!(store.retired_btrees.lock().is_empty());
    }

    #[test]
    fn checkpoint_payload_buffer_is_bounded_for_a_large_collection() {
        let directory = tempfile::tempdir().expect("temporary store");
        let store = UnifiedStore::open(&directory.path().join("data.rdb")).expect("paged store");
        store.create_collection("rows").expect("collection");
        let manager = store.get_collection("rows").expect("rows");
        for id in (1..=1200).rev() {
            manager
                .insert(UnifiedEntity::table_row(
                    EntityId::new(id),
                    "rows",
                    id,
                    vec![Value::text("x".repeat(4096))],
                ))
                .expect("seed");
        }
        CHECKPOINT_BUFFERED_BYTES.with(|peak| peak.set(0));
        store.persist().expect("bounded checkpoint");
        assert!(CHECKPOINT_BUFFERED_BYTES.with(|peak| peak.get()) <= CHECKPOINT_BATCH_BYTES);
        assert!(CHECKPOINT_BUFFERED_BYTES.with(|peak| peak.get()) > 0);
        assert_eq!(store.get_collection("rows").expect("rows").count(), 1200);
    }

    #[test]
    fn unchanged_checkpoint_does_not_serialize_collection_rows() {
        let directory = tempfile::tempdir().expect("temporary store");
        let store = UnifiedStore::open(&directory.path().join("data.rdb")).expect("paged store");
        store.create_collection("rows").expect("collection");
        for id in 1..=40 {
            store
                .insert(
                    "rows",
                    UnifiedEntity::table_row(
                        EntityId::new(id),
                        "rows",
                        id,
                        vec![Value::Integer(id as i64)],
                    ),
                )
                .expect("seed row");
        }
        store.persist().expect("initial checkpoint");
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("unchanged checkpoint");
        assert_eq!(
            CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()),
            0,
            "unchanged collections must not hold the writer fence while serializing every row"
        );
        drop(store);
        let reopened = UnifiedStore::open(&directory.path().join("data.rdb")).expect("reopen");
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        reopened.persist().expect("unchanged recovered generation");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);
    }

    #[test]
    fn sparse_wal_replay_does_not_reencode_complete_collections() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        for (name, first) in [("rows", 1), ("cold", 129)] {
            store.create_collection(name).expect("collection");
            for id in first..first + 128 {
                store
                    .insert(
                        name,
                        UnifiedEntity::table_row(
                            EntityId::new(id),
                            name,
                            id,
                            vec![Value::text(format!("{id}:{}", "x".repeat(8192)))],
                        ),
                    )
                    .expect("seed");
            }
        }
        store.persist().expect("durable baseline");
        let action = super::super::commit::StoreWalAction::upsert_entity(
            "rows",
            &row(1, 42),
            None,
            STORE_VERSION_V11,
        );
        // The acknowledged WAL image has not reached its home B-tree yet.
        store
            .commit
            .as_ref()
            .unwrap()
            .append_actions(&[action])
            .unwrap();
        drop(store);

        REPLAY_SERIALIZED_ROWS.with(|count| count.set(0));
        let recovered = UnifiedStore::open(&path).expect("replay committed change");
        assert_eq!(
            recovered
                .get("rows", EntityId::new(1))
                .unwrap()
                .data
                .as_row()
                .unwrap()
                .columns,
            vec![Value::Integer(42)]
        );
        for name in ["rows", "cold"] {
            assert_eq!(recovered.get_collection(name).unwrap().count(), 128);
        }
        assert_eq!(
            REPLAY_SERIALIZED_ROWS.with(|count| count.get()),
            0,
            "one WAL-backed row must not reencode its whole collection or unrelated cold rows"
        );
        for (name, first) in [("rows", 2), ("cold", 129)] {
            let end = if name == "rows" { 129 } else { 257 };
            for id in first..end {
                assert_eq!(
                    recovered
                        .get(name, EntityId::new(id))
                        .unwrap()
                        .data
                        .as_row()
                        .unwrap()
                        .columns,
                    vec![Value::text(format!("{id}:{}", "x".repeat(8192)))],
                );
            }
        }

        drop(recovered);
        let reopened = UnifiedStore::open(&path).expect("durable sparse replay");
        assert_eq!(
            reopened
                .get("rows", EntityId::new(1))
                .unwrap()
                .data
                .as_row()
                .unwrap()
                .columns,
            vec![Value::Integer(42)]
        );
    }

    fn replay_test_payload(bytes: usize) -> String {
        let mut state = 0x1234_5678_u32;
        (0..bytes)
            .map(|_| {
                state ^= state << 13;
                state ^= state >> 17;
                state ^= state << 5;
                char::from(b'!' + (state % 90) as u8)
            })
            .collect()
    }

    #[test]
    fn wal_replay_with_exhausted_row_proof_keeps_full_validation() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("data.rdb");
        let config = || UnifiedStoreConfig::default().with_page_cache_slots(4);
        let store = UnifiedStore::open_with_config(&path, config()).unwrap();
        for id in 1..=16 {
            store.insert_auto("rows", row(id, 1)).unwrap();
        }
        store.persist().unwrap();
        // Larger than the retained-image allowance, so replay must fall back.
        let payload = replay_test_payload(64 * 1024);
        let changed =
            UnifiedEntity::table_row(EntityId::new(1), "rows", 1, vec![Value::text(payload)]);
        let action = StoreWalAction::upsert_entity("rows", &changed, None, store.format_version());
        store
            .commit
            .as_ref()
            .unwrap()
            .append_actions(&[action])
            .unwrap();
        drop(store);
        REPLAY_SERIALIZED_ROWS.with(|count| count.set(0));
        let recovered = UnifiedStore::open_with_config(&path, config()).unwrap();
        assert!(REPLAY_SERIALIZED_ROWS.with(|count| count.get()) >= 16);
        assert_eq!(
            recovered
                .get("rows", EntityId::new(1))
                .unwrap()
                .data
                .as_row()
                .unwrap()
                .columns,
            changed.data.as_row().unwrap().columns
        );
        drop(recovered);
        let reopened = UnifiedStore::open_with_config(&path, config()).unwrap();
        assert_eq!(reopened.get_collection("rows").unwrap().count(), 16);
        assert_eq!(
            reopened
                .get("rows", EntityId::new(1))
                .unwrap()
                .data
                .as_row()
                .unwrap()
                .columns,
            changed.data.as_row().unwrap().columns
        );
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn interrupted_sparse_replay_preserves_rows_and_bounds_growth() {
        const CHILD: &str = "REDDB_TEST_SPARSE_REPLAY_PATH";
        if let Some(path) = std::env::var_os(CHILD) {
            let _store = UnifiedStore::open(path).unwrap();
            panic!("replay crash hook was not reached");
        }
        for overflow in [false, true] {
            for (hook, code) in [
                ("REDDB_TEST_EXIT_AFTER_SPARSE_BATCH", 193),
                ("REDDB_EXPERIMENTAL_EXIT_AFTER_REPLAY_PUBLICATION", 182),
            ] {
                let directory = tempfile::tempdir().unwrap();
                let path = directory.path().join("data.rdb");
                let store = UnifiedStore::open(&path).unwrap();
                for id in 1..=128 {
                    store.insert_auto("rows", row(id, 1)).unwrap();
                }
                store.persist().unwrap();
                let changed = if overflow {
                    UnifiedEntity::table_row(
                        EntityId::new(1),
                        "rows",
                        1,
                        vec![Value::text(replay_test_payload(16 * 1024))],
                    )
                } else {
                    row(1, 42)
                };
                let action =
                    StoreWalAction::upsert_entity("rows", &changed, None, store.format_version());
                store
                    .commit
                    .as_ref()
                    .unwrap()
                    .append_actions(&[action])
                    .unwrap();
                drop(store);
                let mut first_size = 0;
                for cycle in 0..12 {
                    let output = std::process::Command::new(std::env::current_exe().unwrap())
                        .args(["--exact", "storage::unified::store::impl_pages::checkpoint_work_tests::interrupted_sparse_replay_preserves_rows_and_bounds_growth", "--nocapture"])
                        .env(CHILD, &path).env(hook, "1").output().unwrap();
                    assert_eq!(
                        output.status.code(),
                        Some(code),
                        "overflow={overflow} hook={hook} cycle={cycle}: {}",
                        String::from_utf8_lossy(&output.stderr)
                    );
                    let size = std::fs::metadata(&path).unwrap().len();
                    if cycle == 0 {
                        first_size = size;
                    }
                    assert!(
                        size <= first_size + 2 * crate::storage::engine::PAGE_SIZE as u64,
                        "repeated recovery must reuse reservations: {first_size} -> {size}"
                    );
                }
                for _ in 0..2 {
                    let recovered = UnifiedStore::open(&path).unwrap();
                    assert_eq!(recovered.get_collection("rows").unwrap().count(), 128);
                    for id in 1..=128 {
                        let expected = if id == 1 { changed.clone() } else { row(id, 1) };
                        assert_eq!(
                            recovered
                                .get("rows", EntityId::new(id))
                                .unwrap()
                                .data
                                .as_row()
                                .unwrap()
                                .columns,
                            expected.data.as_row().unwrap().columns
                        );
                    }
                }
            }
        }
    }

    #[test]
    fn paged_reopen_preserves_stored_sequence_identity() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        store.create_collection("rows").expect("collection");
        for id in [30, 10, 20] {
            store.insert("rows", row(id, id as i64)).expect("insert");
        }
        store.persist().expect("checkpoint");
        let sequences = [10, 20, 30].map(|id| {
            store
                .get("rows", EntityId::new(id))
                .expect("row")
                .sequence_id
        });
        drop(store);
        let reopened = UnifiedStore::open(&path).expect("reopen");
        let recovered = [10, 20, 30].map(|id| {
            reopened
                .get("rows", EntityId::new(id))
                .expect("row")
                .sequence_id
        });
        assert_eq!(
            recovered, sequences,
            "opening a checkpoint must not rewrite stored row identities"
        );
    }

    #[test]
    fn auto_insert_persists_the_assigned_sequence_identity() {
        let directory = tempfile::tempdir().expect("temporary store");
        let store = UnifiedStore::open(&directory.path().join("data.rdb")).expect("paged store");
        for id in 1u64..=3 {
            store
                .insert_auto("rows", row(id, id as i64))
                .expect("insert");
        }
        let manager = store.get_collection("rows").expect("rows");
        let tree = store
            .btree_indices
            .read()
            .get("rows")
            .cloned()
            .expect("tree");
        for id in 1u64..=3 {
            let bytes = tree
                .get(&id.to_be_bytes())
                .expect("lookup")
                .expect("stored row");
            let (persisted, _) =
                UnifiedStore::deserialize_entity_record(&bytes, store.format_version())
                    .expect("decode");
            assert_eq!(
                persisted.sequence_id,
                manager
                    .get(EntityId::new(id))
                    .expect("live row")
                    .sequence_id
            );
        }
    }

    #[test]
    fn completed_inline_insert_reuses_history_but_not_untracked_edits() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        store.insert_auto("rows", row(1, 1)).expect("seed");
        store.persist().expect("checkpoint");
        store.insert_auto("rows", row(2, 2)).expect("append");
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("inline append checkpoint");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);
        store
            .get_collection("rows")
            .expect("rows")
            .update(row(1, 99))
            .expect("untracked edit");
        store.insert_auto("rows", row(3, 3)).expect("next append");
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("untracked edit checkpoint");
        assert!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()) > 0);
        drop(store);
        let reopened = UnifiedStore::open(&path).expect("reopen");
        assert_eq!(
            reopened
                .get("rows", EntityId::new(1))
                .expect("row")
                .data
                .as_row()
                .expect("table")
                .columns,
            vec![Value::Integer(99)]
        );
    }

    #[test]
    fn replayed_trees_are_reusable_after_durable_reconciliation() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        store.insert_auto("rows", row(1, 1)).expect("seed");
        store.persist().expect("checkpoint");
        store
            .insert_auto("rows", row(2, 2))
            .expect("WAL-backed append");
        drop(store);
        let reopened = UnifiedStore::open(&path).expect("replay");
        assert!(reopened.get("rows", EntityId::new(2)).is_some());
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        reopened.persist().expect("checkpoint reconciled tree");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);
    }

    #[test]
    fn legacy_system_catalog_and_wal_migrate_without_cold_boot_rebuild() {
        use crate::storage::wal::{WalRecord, WalWriter};

        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let config = UnifiedStoreConfig {
            page_cache_slots: Some(32),
            resident_memory_bytes_max: Some(2 * 1024 * 1024),
            ..UnifiedStoreConfig::default()
        };
        let index_identity = super::super::secondary_index::SecondaryIndexIdentity {
            codec_version: 1,
            definition: [91; 32],
        };
        let baseline_index_root;
        {
            let store = UnifiedStore::open_with_config(&path, config.clone()).expect("paged store");
            store
                .create_collection("red_config")
                .expect("system collection");
            store
                .insert_auto(
                    "red_config",
                    UnifiedEntity::table_row(
                        EntityId::new(1),
                        "red_config",
                        1,
                        vec![Value::text("checkpoint")],
                    ),
                )
                .expect("legacy-compatible insert");
            store
                .register_secondary_index(
                    "red_config",
                    "by_id",
                    index_identity.clone(),
                    Arc::new(SystemNameProjection),
                )
                .expect("secondary index");
            store.persist().expect("baseline checkpoint");
            baseline_index_root = store
                .secondary_index("red.config", "by_id", &index_identity)
                .expect("published secondary index")
                .root_page_id();
        }

        // Recreate a pre-migration catalog and operational manifest. The two
        // spellings are deliberately equal-length so the test can preserve
        // every root and checksum-protected page layout except the name.
        {
            let pager = Pager::open(
                &path,
                PagerConfig {
                    disk_residency: true,
                    ..PagerConfig::default()
                },
            )
            .expect("open pager");
            let mut page = pager.read_page(1).expect("catalog page");
            let bytes = page.as_bytes_mut();
            let canonical = b"red.config";
            let legacy = b"red_config";
            let mut replacements = 0;
            for offset in 0..=bytes.len() - canonical.len() {
                if &bytes[offset..offset + canonical.len()] == canonical {
                    bytes[offset..offset + legacy.len()].copy_from_slice(legacy);
                    replacements += 1;
                }
            }
            assert!(replacements >= 1, "catalog must contain the canonical name");
            pager.write_page(1, page).expect("write legacy catalog");
            pager.sync().expect("sync legacy catalog");
        }
        crate::storage::operational_manifest::OperationalManifest::for_db_path(&path)
            .rename_collections(&[("red.config", "red_config")])
            .expect("restore legacy operational name");

        // Append an acknowledged transaction written by a legacy binary. Its
        // payload must replay into the canonical manager during migration.
        let wal_upsert = reddb_file::encode_store_wal_action_frame(
            &reddb_file::StoreWalActionFrame::UpsertEntityRecord {
                collection: "red_config".to_string(),
                record: UnifiedStore::serialize_entity_record(
                    &UnifiedEntity::table_row(
                        EntityId::new(2),
                        "red_config",
                        2,
                        vec![Value::text("legacy WAL")],
                    ),
                    None,
                    STORE_VERSION_CURRENT,
                ),
            },
        )
        .expect("legacy WAL action");
        let wal_delete = reddb_file::encode_store_wal_action_frame(
            &reddb_file::StoreWalActionFrame::DeleteEntityRecord {
                collection: "red_config".to_string(),
                entity_id: EntityId::new(1).raw(),
            },
        )
        .expect("legacy WAL delete");
        let wal_path = reddb_file::layout::unified_wal_path(&path);
        {
            let mut writer = WalWriter::open(&wal_path).expect("open WAL");
            writer
                .append(&WalRecord::TxCommitBatch {
                    tx_id: 77,
                    actions: vec![wal_upsert, wal_delete],
                })
                .expect("append legacy transaction");
            writer.sync().expect("sync legacy transaction");
        }

        let (root_after_recovery, pages_after_recovery) = {
            let recovered =
                UnifiedStore::open_with_config(&path, config.clone()).expect("migrate and recover");
            assert_eq!(recovered.list_collections(), vec!["red.config".to_string()]);
            assert!(
                recovered.get("red.config", EntityId::new(1)).is_none(),
                "legacy WAL deletes must target the canonical collection"
            );
            assert_eq!(
                recovered
                    .get("red.config", EntityId::new(2))
                    .expect("WAL row")
                    .kind
                    .collection(),
                "red.config"
            );
            let root = recovered
                .collection_root_page("red.config")
                .expect("published canonical root");
            assert_eq!(
                recovered
                    .secondary_index("red.config", "by_id", &index_identity)
                    .expect("recovered secondary index")
                    .root_page_id(),
                baseline_index_root,
                "name migration must reuse the secondary-index root"
            );
            let pages = recovered
                .pager()
                .expect("pager")
                .page_count()
                .expect("page count");
            (root, pages)
        };

        // Simulate a crash/import boundary where the catalog rename is
        // durable but the operational manifest still has the legacy key.
        crate::storage::operational_manifest::OperationalManifest::for_db_path(&path)
            .rename_collections(&[("red.config", "red_config")])
            .expect("restore stale manifest identity");

        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        let cold = UnifiedStore::open_with_config(&path, config).expect("second cold boot");
        assert_eq!(
            cold.collection_root_page("red.config"),
            Some(root_after_recovery),
            "an idempotent cold boot must reuse the published data root"
        );
        assert_eq!(
            cold.pager()
                .expect("pager")
                .page_count()
                .expect("page count"),
            pages_after_recovery,
            "cold boot must not allocate replacement index trees"
        );
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);
        assert_eq!(
            cold.secondary_index("red.config", "by_id", &index_identity)
                .expect("secondary index after cold boot")
                .root_page_id(),
            baseline_index_root
        );
        assert!(
            crate::storage::operational_manifest::OperationalManifest::for_db_path(&path)
                .recover_or_bootstrap(&["red.config".to_string()])
                .expect("canonical manifest")
                .is_empty()
        );
    }

    struct DuplicateSystemCatalogFixture {
        _directory: tempfile::TempDir,
        path: std::path::PathBuf,
        config: UnifiedStoreConfig,
        pages_before: u32,
        legacy_root: Option<u32>,
        legacy_key_root: Option<u32>,
        legacy_secondary_root: Option<u32>,
        index_identity: super::super::secondary_index::SecondaryIndexIdentity,
    }

    fn duplicate_system_catalog_fixture(
        legacy_populated: bool,
        canonical_populated: bool,
        index_legacy: bool,
    ) -> DuplicateSystemCatalogFixture {
        duplicate_system_catalog_fixture_with_ref(
            legacy_populated,
            canonical_populated,
            index_legacy,
            false,
        )
    }

    fn duplicate_system_catalog_fixture_with_ref(
        legacy_populated: bool,
        canonical_populated: bool,
        index_legacy: bool,
        cross_reference_legacy: bool,
    ) -> DuplicateSystemCatalogFixture {
        duplicate_system_catalog_fixture_with_key(
            legacy_populated,
            canonical_populated,
            index_legacy,
            cross_reference_legacy,
            None,
        )
    }

    fn config_recovery_row(id: u64, key: &str, value: i64) -> UnifiedEntity {
        UnifiedEntity::new(
            EntityId::new(id),
            EntityKind::TableRow {
                table: Arc::from("red.config"),
                row_id: id,
            },
            EntityData::Row(crate::storage::RowData::with_names(
                vec![Value::text(key), Value::Integer(value)],
                vec!["key".into(), "value".into()],
            )),
        )
    }

    fn legacy_config_recovery_row(id: u64, key: &str, value: i64) -> UnifiedEntity {
        UnifiedEntity::new(
            EntityId::new(id),
            EntityKind::TableRow {
                table: Arc::from("red_config"),
                row_id: id,
            },
            EntityData::Row(crate::storage::RowData::with_names(
                vec![Value::text(key), Value::Integer(value)],
                vec!["key".into(), "value".into()],
            )),
        )
    }

    fn authorize_fixture_dwb(path: &Path, frame: &[u8]) {
        use reddb_file::{file_format as f, PAGED_SUPERBLOCK_SLOT_SIZE};
        use std::io::{Seek, SeekFrom, Write};

        let bytes = std::fs::read(path).expect("fixture headers");
        let selected = reddb_file::select_paged_superblock(&bytes).expect("valid superblock");
        let start = selected.copy_index * PAGED_SUPERBLOCK_SLOT_SIZE;
        let mut image = [0; PAGED_SUPERBLOCK_SLOT_SIZE];
        image.copy_from_slice(&bytes[start..start + PAGED_SUPERBLOCK_SLOT_SIZE]);
        f::set_database_header_dwb_publication(
            &mut image,
            Some(f::DwbPublication {
                phase: f::DwbPhase::Ready,
                frame_bytes: u32::try_from(frame.len()).expect("bounded fixture frame"),
                digest: f::dwb_frame_digest(frame),
            }),
        )
        .expect("known-current DWB authority");
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .open(path)
            .expect("fixture header writer");
        for step in 1..=reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
            let copy = (selected.copy_index + step) % reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT;
            reddb_file::seal_paged_superblock_slot(
                &mut image,
                copy,
                selected.generation + step as u64,
            )
            .expect("fixture header seal");
            file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
                copy,
            )))
            .expect("fixture header offset");
            file.write_all(&image).expect("persist DWB authority");
            file.sync_all().expect("durable DWB authority");
        }
    }

    fn duplicate_system_catalog_fixture_with_key(
        legacy_populated: bool,
        canonical_populated: bool,
        index_legacy: bool,
        cross_reference_legacy: bool,
        config_key: Option<&str>,
    ) -> DuplicateSystemCatalogFixture {
        let config_keys = config_key.into_iter().collect::<Vec<_>>();
        duplicate_system_catalog_fixture_with_keys(
            legacy_populated,
            canonical_populated,
            index_legacy,
            cross_reference_legacy,
            &config_keys,
        )
    }

    fn duplicate_system_catalog_fixture_with_keys(
        legacy_populated: bool,
        canonical_populated: bool,
        index_legacy: bool,
        cross_reference_legacy: bool,
        config_keys: &[&str],
    ) -> DuplicateSystemCatalogFixture {
        duplicate_system_catalog_fixture_with_aliases(
            legacy_populated,
            canonical_populated,
            index_legacy,
            cross_reference_legacy,
            config_keys,
            None,
        )
    }

    fn duplicate_system_catalog_fixture_with_aliases(
        legacy_populated: bool,
        canonical_populated: bool,
        index_legacy: bool,
        cross_reference_legacy: bool,
        config_keys: &[&str],
        alias_case: Option<&str>,
    ) -> DuplicateSystemCatalogFixture {
        system_catalog_fixture_with_aliases(
            legacy_populated,
            canonical_populated,
            index_legacy,
            cross_reference_legacy,
            config_keys,
            alias_case,
            true,
        )
    }

    fn single_legacy_system_catalog_fixture_with_keys(
        config_keys: &[&str],
    ) -> DuplicateSystemCatalogFixture {
        system_catalog_fixture_with_aliases(true, false, false, false, config_keys, None, false)
    }

    fn single_empty_legacy_system_catalog_fixture() -> DuplicateSystemCatalogFixture {
        system_catalog_fixture_with_aliases(false, false, false, false, &[], None, false)
    }

    fn system_catalog_fixture_with_aliases(
        legacy_populated: bool,
        canonical_populated: bool,
        index_legacy: bool,
        cross_reference_legacy: bool,
        config_keys: &[&str],
        alias_case: Option<&str>,
        canonical_identity: bool,
    ) -> DuplicateSystemCatalogFixture {
        assert!(canonical_identity || !canonical_populated);
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let config = UnifiedStoreConfig {
            page_cache_slots: Some(32),
            resident_memory_bytes_max: Some(2 * 1024 * 1024),
            ..UnifiedStoreConfig::default()
        };
        let index_identity = super::super::secondary_index::SecondaryIndexIdentity {
            codec_version: 1,
            definition: [73; 32],
        };
        let (legacy_root, legacy_key_root, legacy_secondary_root);
        {
            let store = UnifiedStore::open_with_config(&path, config.clone()).expect("paged store");
            store
                .create_collection("cfg_legacy")
                .expect("legacy source collection");
            if legacy_populated {
                if config_keys.is_empty() {
                    store
                        .insert_auto(
                            "cfg_legacy",
                            UnifiedEntity::table_row(
                                EntityId::new(1),
                                "red_config",
                                1,
                                vec![Value::text("legacy")],
                            ),
                        )
                        .expect("legacy row");
                } else {
                    for (offset, key) in config_keys.iter().enumerate() {
                        let id = u64::try_from(offset).unwrap() + 1;
                        let entity = if canonical_identity {
                            config_recovery_row(id, key, i64::try_from(id).unwrap())
                        } else {
                            legacy_config_recovery_row(id, key, i64::try_from(id).unwrap())
                        };
                        store
                            .insert_auto("cfg_legacy", entity)
                            .expect("legacy config row");
                    }
                }
            }
            if canonical_identity {
                store
                    .create_collection("cfg.canon_")
                    .expect("canonical source collection");
            }
            if canonical_identity && canonical_populated {
                if config_keys.is_empty() {
                    store
                        .insert_auto(
                            "cfg.canon_",
                            UnifiedEntity::table_row(
                                EntityId::new(2),
                                "red.config",
                                2,
                                vec![Value::text("canonical")],
                            ),
                        )
                        .expect("canonical row");
                } else {
                    let base = u64::try_from(config_keys.len()).unwrap() + 1;
                    for (offset, key) in config_keys.iter().enumerate() {
                        let id = base + u64::try_from(offset).unwrap();
                        store
                            .insert_auto(
                                "cfg.canon_",
                                config_recovery_row(id, key, i64::try_from(id).unwrap()),
                            )
                            .expect("canonical config row");
                    }
                }
            }
            if cross_reference_legacy {
                store
                    .add_cross_ref(
                        "cfg.canon_",
                        EntityId::new(2),
                        "cfg_legacy",
                        EntityId::new(1),
                        RefType::RowToNode,
                        1.0,
                    )
                    .expect("reference to legacy config");
            }
            if index_legacy {
                store
                    .register_secondary_index(
                        "cfg_legacy",
                        "by_id",
                        index_identity.clone(),
                        Arc::new(SystemNameProjection),
                    )
                    .expect("legacy secondary index");
            }

            if let Some(case) = alias_case {
                use crate::storage::{EntityData, RowData};
                let pairs = [
                    (
                        "old_stats",
                        "new.query.stats_catalog",
                        "red_stats",
                        "red.query.stats_catalog",
                    ),
                    (
                        "old_index_registry",
                        "new.index.registry",
                        "red_index_registry",
                        "red.index.registry",
                    ),
                    (
                        "old_queue_meta",
                        "new.queue.meta",
                        "red_queue_meta",
                        "red.queue.meta",
                    ),
                    (
                        "old_materialized_view_defs",
                        "new.materialized_view.definitions",
                        "red_materialized_view_defs",
                        "red.materialized_view.definitions",
                    ),
                ];
                for (pair_index, (old, new, legacy, canonical)) in pairs.iter().enumerate() {
                    for (is_old, name, table) in [(true, *old, *legacy), (false, *new, *canonical)]
                    {
                        store.create_collection(name).unwrap();
                        let text = |s: &str| Value::text(s);
                        let mut fields = match pair_index {
                            0 => vec![
                                ("kind", text("table")),
                                ("table", text("users")),
                                ("row_count", Value::Integer(if is_old { 1 } else { 2 })),
                                ("avg_row_size", Value::Integer(20)),
                                ("page_count", Value::Integer(1)),
                            ],
                            1 => vec![
                                ("collection", text("users")),
                                (
                                    "name",
                                    text(if is_old {
                                        "legacy_only"
                                    } else {
                                        "canonical_only"
                                    }),
                                ),
                                ("columns", text("id")),
                                ("method", text("hash")),
                                ("resolution", Value::Integer(0)),
                                ("unique", Value::Boolean(false)),
                                ("dropped", Value::Boolean(false)),
                            ],
                            2 => vec![
                                ("kind", text("queue_group")),
                                ("queue", text("jobs")),
                                ("group", text("workers")),
                                (
                                    "created_at_ns",
                                    Value::UnsignedInteger(if is_old { 1 } else { 2 }),
                                ),
                            ],
                            _ => vec![
                                ("name", text("view")),
                                ("refresh_every_ms", Value::UnsignedInteger(1000)),
                                ("retention_duration_ms", Value::UnsignedInteger(0)),
                                ("source_collections", text("users")),
                                (
                                    "source_sql",
                                    text(if case == "view-conflict" && !is_old {
                                        "SELECT 2"
                                    } else {
                                        "SELECT 1"
                                    }),
                                ),
                            ],
                        };
                        if case == "queue-delivery" && pair_index == 2 && is_old {
                            fields[0].1 = text("queue_pending");
                        }
                        let id = if is_old {
                            10000 + pair_index as u64
                        } else {
                            100 + pair_index as u64
                        };
                        let entity = UnifiedEntity::new(
                            EntityId::new(id),
                            EntityKind::TableRow {
                                table: Arc::from(table),
                                row_id: id,
                            },
                            EntityData::Row(RowData {
                                columns: vec![],
                                named: Some(
                                    fields.into_iter().map(|(k, v)| (k.to_owned(), v)).collect(),
                                ),
                                schema: None,
                            }),
                        );
                        store.insert_auto(name, entity).unwrap();
                    }
                }
            }
            store.persist().expect("checkpoint duplicate catalog");
            if !canonical_identity && legacy_populated && !config_keys.is_empty() {
                let tree = store
                    .btree_indices
                    .read()
                    .get("cfg_legacy")
                    .cloned()
                    .expect("legacy config tree");
                for (offset, key) in config_keys.iter().enumerate() {
                    let id = u64::try_from(offset).unwrap() + 1;
                    let entity = legacy_config_recovery_row(id, key, i64::try_from(id).unwrap());
                    let mut record =
                        UnifiedStore::serialize_entity_record(&entity, None, STORE_VERSION_CURRENT);
                    let canonical = b"red.config";
                    let legacy = b"red_config";
                    let mut replacements = 0;
                    for offset in 0..=record.len() - canonical.len() {
                        if &record[offset..offset + canonical.len()] == canonical {
                            record[offset..offset + legacy.len()].copy_from_slice(legacy);
                            replacements += 1;
                        }
                    }
                    assert_eq!(
                        replacements, 1,
                        "persisted fixture row must carry one historical table identity"
                    );
                    tree.upsert(&id.to_be_bytes(), &record)
                        .expect("write authenticated historical config row");
                }
                store
                    .flush_paged_registry()
                    .expect("publish historical config generation");
            }
            legacy_root = store.collection_root_page("cfg_legacy");
            legacy_key_root = store
                .key_indices
                .read()
                .get("cfg_legacy")
                .map(|index| index.tree.root_page_id());
            legacy_secondary_root = index_legacy.then(|| {
                store
                    .secondary_index("cfg_legacy", "by_id", &index_identity)
                    .expect("legacy secondary root")
                    .root_page_id()
            });
        }

        // Recreate the production shape: a populated legacy collection and an
        // empty canonical collection published by an intermediate binary.
        // Every spelling has the same byte length, so roots and checksummed
        // page layouts stay unchanged while only the catalog identities move.
        {
            let pager = Pager::open(
                &path,
                PagerConfig {
                    disk_residency: true,
                    ..PagerConfig::default()
                },
            )
            .expect("open pager");
            let mut metadata = read_meta_payload(&pager)
                .expect("read catalog payload")
                .expect("published catalog payload");
            let bytes = metadata.as_mut_slice();
            let mut names = vec![("cfg_legacy", "red_config")];
            if canonical_identity {
                names.push(("cfg.canon_", "red.config"));
            }
            if alias_case.is_some() {
                names.extend([
                    ("old_stats", "red_stats"),
                    ("new.query.stats_catalog", "red.query.stats_catalog"),
                    ("old_index_registry", "red_index_registry"),
                    ("new.index.registry", "red.index.registry"),
                    ("old_queue_meta", "red_queue_meta"),
                    ("new.queue.meta", "red.queue.meta"),
                    ("old_materialized_view_defs", "red_materialized_view_defs"),
                    (
                        "new.materialized_view.definitions",
                        "red.materialized_view.definitions",
                    ),
                ]);
            }
            for (source, destination) in names {
                let (source, destination) = (source.as_bytes(), destination.as_bytes());
                let mut replacements = 0;
                for offset in 0..=bytes.len() - source.len() {
                    if &bytes[offset..offset + source.len()] == source {
                        bytes[offset..offset + destination.len()].copy_from_slice(destination);
                        replacements += 1;
                    }
                }
                assert!(replacements >= 1, "catalog must contain the source name");
            }
            publish_paged_metadata(&pager, &metadata).expect("write duplicate catalog");
            pager.sync().expect("sync duplicate catalog");
            if alias_case.is_some() {
                let mut metadata = read_meta_payload(&pager).unwrap().unwrap();
                let resume = reddb_file::native_resume::decode_catalog_resume(&metadata)
                    .unwrap()
                    .unwrap();
                let trailer = metadata
                    .windows(4)
                    .rposition(|bytes| bytes == b"RDRS")
                    .unwrap();
                metadata.truncate(trailer);
                reddb_file::native_resume::encode_catalog_resume(&mut metadata, &resume).unwrap();
                publish_paged_metadata(&pager, &metadata).unwrap();
                pager.sync().unwrap();
            }
        }
        let mut config_names = vec![("cfg_legacy", "red_config")];
        if canonical_identity {
            config_names.push(("cfg.canon_", "red.config"));
        }
        crate::storage::operational_manifest::OperationalManifest::for_db_path(&path)
            .rename_collections(&config_names)
            .expect("publish duplicate operational names");

        if alias_case.is_some() {
            crate::storage::operational_manifest::OperationalManifest::for_db_path(&path)
                .rename_collections(&[
                    ("old_stats", "red_stats"),
                    ("new.query.stats_catalog", "red.query.stats_catalog"),
                    ("old_index_registry", "red_index_registry"),
                    ("new.index.registry", "red.index.registry"),
                    ("old_queue_meta", "red_queue_meta"),
                    ("new.queue.meta", "red.queue.meta"),
                    ("old_materialized_view_defs", "red_materialized_view_defs"),
                    (
                        "new.materialized_view.definitions",
                        "red.materialized_view.definitions",
                    ),
                ])
                .unwrap();
        }

        let pages_before = Pager::open(
            &path,
            PagerConfig {
                disk_residency: true,
                ..PagerConfig::default()
            },
        )
        .expect("inspect pager")
        .page_count()
        .expect("page count");
        DuplicateSystemCatalogFixture {
            _directory: directory,
            path,
            config,
            pages_before,
            legacy_root,
            legacy_key_root,
            legacy_secondary_root,
            index_identity,
        }
    }

    fn remove_fixture_catalog_resume(fixture: &DuplicateSystemCatalogFixture) {
        let pager = Pager::open(
            &fixture.path,
            PagerConfig {
                disk_residency: true,
                ..PagerConfig::default()
            },
        )
        .expect("fixture pager");
        let mut metadata = read_meta_payload(&pager).unwrap().unwrap();
        assert!(reddb_file::native_resume::decode_catalog_resume(&metadata)
            .unwrap()
            .is_some());
        let trailer = metadata
            .windows(4)
            .rposition(|bytes| bytes == b"RDRS")
            .unwrap();
        metadata.truncate(trailer);
        publish_paged_metadata(&pager, &metadata).unwrap();
        pager.sync().unwrap();
        assert!(reddb_file::native_resume::decode_catalog_resume(
            &read_meta_payload(&pager).unwrap().unwrap()
        )
        .unwrap()
        .is_none());
    }

    #[test]
    fn no_resume_catalog_keeps_populated_config_beside_zero_root_alias() {
        for (legacy_populated, canonical_populated) in
            [(true, false), (false, true), (false, false)]
        {
            let fixture =
                duplicate_system_catalog_fixture(legacy_populated, canonical_populated, false);
            remove_fixture_catalog_resume(&fixture);
            let store = UnifiedStore::open_with_config(&fixture.path, fixture.config.clone())
                .expect("resolve physically empty alias without resume counters");
            assert_eq!(store.list_collections(), vec!["red.config".to_owned()]);
            assert_eq!(
                store.get("red.config", EntityId::new(1)).is_some(),
                legacy_populated
            );
            assert_eq!(
                store.get("red.config", EntityId::new(2)).is_some(),
                canonical_populated
            );
            if legacy_populated {
                assert_eq!(
                    store.collection_root_page("red.config"),
                    fixture.legacy_root
                );
            }
            store.persist().expect("checkpoint canonical catalog");
            drop(store);
            let cold = UnifiedStore::open_with_config(&fixture.path, fixture.config)
                .expect("cold reopen canonical catalog");
            assert_eq!(cold.list_collections(), vec!["red.config".to_owned()]);
            assert_eq!(
                cold.get("red.config", EntityId::new(1)).is_some(),
                legacy_populated
            );
            assert_eq!(
                cold.get("red.config", EntityId::new(2)).is_some(),
                canonical_populated
            );
        }
    }

    #[test]
    fn no_resume_catalog_refuses_two_populated_config_roots() {
        let fixture = duplicate_system_catalog_fixture(true, true, false);
        remove_fixture_catalog_resume(&fixture);
        let error = UnifiedStore::open_with_config(&fixture.path, fixture.config)
            .err()
            .expect("two populated roots require reconciliation");
        assert!(error.to_string().contains("both legacy collection"));
    }

    fn alias_duplicate_system_catalog_state_with(
        fixture: &DuplicateSystemCatalogFixture,
        mutate_canonical: impl FnOnce(&mut reddb_file::native_resume::CollectionResume),
    ) {
        let pager = Pager::open(
            &fixture.path,
            PagerConfig {
                disk_residency: true,
                ..PagerConfig::default()
            },
        )
        .expect("open duplicate catalog pager");
        let mut metadata = read_meta_payload(&pager)
            .expect("read duplicate catalog")
            .expect("published duplicate catalog");
        let mut resume = reddb_file::native_resume::decode_catalog_resume(&metadata)
            .expect("decode duplicate resume")
            .expect("prepared duplicate resume");
        let legacy_state = resume
            .collections
            .get("red_config")
            .expect("legacy resume state")
            .clone();
        let mut canonical_state = legacy_state.clone();
        mutate_canonical(&mut canonical_state);
        resume
            .collections
            .insert("red.config".to_string(), canonical_state);

        let header = reddb_file::decode_native_paged_metadata_header(&metadata)
            .expect("decode metadata header")
            .expect("native metadata header");
        let mut position = reddb_file::METADATA_HEADER_BYTES;
        let mut replaced_root = false;
        for _ in 0..header.collection_count {
            let root = reddb_file::decode_native_paged_collection_root(&metadata, &mut position)
                .expect("decode collection root");
            if root.collection == "red.config" {
                metadata[position - 4..position]
                    .copy_from_slice(&legacy_state.root_page.to_le_bytes());
                replaced_root = true;
            }
        }
        assert!(replaced_root, "canonical catalog root must be present");

        let trailer = metadata
            .windows(4)
            .rposition(|bytes| bytes == b"RDRS")
            .expect("catalog resume trailer");
        metadata.truncate(trailer);
        reddb_file::native_resume::encode_catalog_resume(&mut metadata, &resume)
            .expect("encode aliased resume");
        publish_paged_metadata(&pager, &metadata).expect("publish aliased catalog");
    }

    fn alias_duplicate_system_catalog_state(fixture: &DuplicateSystemCatalogFixture) {
        alias_duplicate_system_catalog_state_with(fixture, |_| {});
    }

    #[test]
    fn populated_legacy_system_collection_replaces_empty_canonical_duplicate() {
        let fixture = duplicate_system_catalog_fixture(true, false, true);
        let recovered = UnifiedStore::open_with_config(&fixture.path, fixture.config.clone())
            .expect("resolve empty duplicate");
        assert_eq!(recovered.list_collections(), vec!["red.config".to_string()]);
        assert_eq!(
            recovered
                .get("red.config", EntityId::new(1))
                .expect("preserved legacy row")
                .kind
                .collection(),
            "red.config"
        );
        assert_eq!(
            recovered.collection_root_page("red.config"),
            fixture.legacy_root,
            "the populated data root must be reused"
        );
        assert_eq!(
            recovered
                .key_indices
                .read()
                .get("red.config")
                .map(|index| index.tree.root_page_id()),
            fixture.legacy_key_root,
            "the populated key-index root must be reused"
        );
        assert_eq!(
            recovered
                .secondary_index("red.config", "by_id", &fixture.index_identity)
                .expect("surviving secondary index")
                .root_page_id(),
            fixture
                .legacy_secondary_root
                .expect("fixture secondary root")
        );
        drop(recovered);

        let cold = UnifiedStore::open_with_config(&fixture.path, fixture.config)
            .expect("idempotent cold boot");
        assert_eq!(cold.list_collections(), vec!["red.config".to_string()]);
        assert_eq!(
            cold.pager()
                .expect("pager")
                .page_count()
                .expect("page count"),
            fixture.pages_before,
            "resolving an empty duplicate must not extend the file"
        );
    }

    #[test]
    fn populated_canonical_system_collection_replaces_empty_legacy_duplicate() {
        let fixture = duplicate_system_catalog_fixture(false, true, false);
        let recovered = UnifiedStore::open_with_config(&fixture.path, fixture.config)
            .expect("retain populated canonical duplicate");
        assert_eq!(recovered.list_collections(), vec!["red.config".to_string()]);
        assert_eq!(
            recovered
                .get("red.config", EntityId::new(2))
                .expect("preserved canonical row")
                .kind
                .collection(),
            "red.config"
        );
        assert!(recovered.get("red.config", EntityId::new(1)).is_none());
    }

    #[test]
    fn empty_canonical_system_collection_replaces_empty_legacy_duplicate() {
        let fixture = duplicate_system_catalog_fixture(false, false, false);
        let recovered = UnifiedStore::open_with_config(&fixture.path, fixture.config)
            .expect("prefer canonical empty duplicate");
        assert_eq!(recovered.list_collections(), vec!["red.config".to_string()]);
        assert_eq!(
            recovered
                .get_collection("red.config")
                .expect("canonical collection")
                .count(),
            0
        );
    }

    #[test]
    fn populated_system_collection_duplicates_still_fail_closed() {
        let fixture = duplicate_system_catalog_fixture(true, true, false);
        let error = match UnifiedStore::open_with_config(&fixture.path, fixture.config) {
            Ok(_) => panic!("two populated identities are ambiguous"),
            Err(error) => error,
        };
        assert!(
            error
                .to_string()
                .contains("both populated legacy collection 'red_config' and canonical collection 'red.config'"),
            "unexpected error: {error}"
        );
        assert!(
            error.to_string().contains(
                "data_root_match=false, key_index_root_match=true, secondary_index_state_match=true"
            ),
            "the error must identify which physical state diverged without exposing page IDs: {error}"
        );
    }

    #[test]
    fn explicit_compaction_promotes_populated_legacy_system_config() {
        let reported_keys = super::super::config_keys::tests::REPORTED_DATABASE_KEYS;
        let fixture =
            duplicate_system_catalog_fixture_with_keys(true, false, false, false, reported_keys);
        let destination = fixture._directory.path().join("canonical-system-config");

        let report = crate::storage::compact::compact_paged_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect("rewrite duplicate system config");
        assert!(report.records > 0);
        assert_eq!(report.reconciled_system_collections, 1);
        assert_eq!(report.imported_system_records, reported_keys.len());

        let rewritten_path = destination.join("data.rdb");
        let rewritten = UnifiedStore::open_with_config(&rewritten_path, fixture.config.clone())
            .expect("open canonical rewrite");
        assert_eq!(rewritten.list_collections(), vec!["red.config".to_string()]);
        for (offset, legacy) in reported_keys.iter().enumerate() {
            let canonical = legacy
                .strip_prefix("red.config.")
                .or_else(|| legacy.strip_prefix("red."))
                .unwrap_or(legacy);
            assert_eq!(
                rewritten.get_config(canonical),
                Some(Value::Integer(i64::try_from(offset).unwrap() + 1)),
                "the legacy config value must survive as {canonical}"
            );
        }
        assert_eq!(rewritten.legacy_red_config_key_count().unwrap(), 0);
        drop(rewritten);

        let cold = UnifiedStore::open_with_config(&rewritten_path, fixture.config)
            .expect("cold reopen canonical rewrite");
        assert_eq!(cold.list_collections(), vec!["red.config".to_string()]);
        assert!(
            cold.get_collection("red.config")
                .expect("canonical collection")
                .count()
                >= reported_keys.len()
        );
        for (offset, legacy) in reported_keys.iter().enumerate() {
            let canonical = legacy
                .strip_prefix("red.config.")
                .or_else(|| legacy.strip_prefix("red."))
                .unwrap_or(legacy);
            assert_eq!(
                cold.get_config(canonical),
                Some(Value::Integer(i64::try_from(offset).unwrap() + 1)),
                "cold reopen lost {canonical}"
            );
        }
        assert!(
            crate::storage::operational_manifest::OperationalManifest::for_db_path(&rewritten_path)
                .recover_or_bootstrap(&["red.config".to_string()])
                .expect("canonical operational manifest")
                .is_empty()
        );
    }

    #[test]
    fn explicit_compaction_refuses_two_populated_system_config_trees() {
        let fixture = duplicate_system_catalog_fixture(true, true, false);
        let destination = fixture._directory.path().join("ambiguous-system-config");

        let error = crate::storage::compact::compact_paged_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect_err("two populated config trees must not be discarded");
        assert!(error.to_string().contains("refusing lossy migration"));
        assert!(!destination.exists());
    }

    #[test]
    fn canonical_system_config_recovery_promotes_single_legacy_root_with_pending_wal() {
        use crate::storage::wal::{WalRecord, WalWriter};

        let fixture = single_legacy_system_catalog_fixture_with_keys(&["red.auth.require_auth"]);
        let wal = reddb_file::unified_wal_path(&fixture.path);
        let action = reddb_file::encode_store_wal_action_frame(
            &reddb_file::StoreWalActionFrame::CreateCollection {
                name: "pending_collection".into(),
            },
        )
        .expect("encode pending action");
        {
            let mut writer = WalWriter::open(&wal).expect("open WAL");
            writer
                .append(&WalRecord::TxCommitBatch {
                    tx_id: 999,
                    actions: vec![action],
                })
                .expect("append committed WAL");
            writer.sync().expect("sync WAL");
        }
        let original = std::fs::read(&fixture.path).expect("source bytes");
        let original_wal = std::fs::read(&wal).expect("WAL bytes");
        let destination = fixture
            ._directory
            .path()
            .join("recovered-single-legacy-config");

        let report = crate::storage::compact::recover_paged_wal_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect("recover pending WAL and promote the legacy config root");

        assert_eq!(report.reconciled_system_collections, 1);
        assert_eq!(report.imported_system_records, 1);
        assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
        assert_eq!(std::fs::read(&wal).unwrap(), original_wal);
        for _ in 0..2 {
            let recovered = UnifiedStore::open_with_config(
                destination.join("data.rdb"),
                fixture.config.clone(),
            )
            .expect("open recovered store");
            let mut collections = recovered.list_collections();
            collections.sort();
            assert_eq!(collections, ["pending_collection", "red.config"]);
            assert_eq!(
                recovered.get_config("auth.require_auth"),
                Some(Value::Integer(1))
            );
            let mut recovered_rows = 0;
            recovered
                .get_collection("red.config")
                .expect("canonical config collection")
                .try_for_each_entity(4 * 1024 * 1024, |entity| {
                    recovered_rows += 1;
                    assert_eq!(entity.kind.collection(), "red.config");
                    true
                })
                .expect("scan recovered config rows");
            assert_eq!(recovered_rows, 1);
            assert_eq!(recovered.legacy_red_config_key_count().unwrap(), 0);
        }
    }

    #[test]
    fn canonical_system_config_recovery_promotes_empty_legacy_root() {
        use crate::storage::wal::{WalRecord, WalWriter};

        let fixture = single_empty_legacy_system_catalog_fixture();
        let wal = reddb_file::unified_wal_path(&fixture.path);
        let action = reddb_file::encode_store_wal_action_frame(
            &reddb_file::StoreWalActionFrame::CreateCollection {
                name: "pending_collection".into(),
            },
        )
        .expect("encode pending action");
        {
            let mut writer = WalWriter::open(&wal).expect("open WAL");
            writer
                .append(&WalRecord::TxCommitBatch {
                    tx_id: 1000,
                    actions: vec![action],
                })
                .expect("append committed WAL");
            writer.sync().expect("sync WAL");
        }
        let destination = fixture
            ._directory
            .path()
            .join("recovered-empty-legacy-config");

        let report = crate::storage::compact::recover_paged_wal_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect("recover and promote the empty legacy config root");

        assert_eq!(report.reconciled_system_collections, 1);
        assert_eq!(report.imported_system_records, 0);
        for _ in 0..2 {
            let recovered = UnifiedStore::open_with_config(
                destination.join("data.rdb"),
                fixture.config.clone(),
            )
            .expect("cold reopen recovered store");
            let mut collections = recovered.list_collections();
            collections.sort();
            assert_eq!(collections, ["pending_collection", "red.config"]);
            assert_eq!(
                recovered
                    .get_collection("red.config")
                    .expect("canonical config root")
                    .count(),
                0
            );
        }
    }

    #[test]
    fn canonical_system_config_compaction_promotes_empty_legacy_root() {
        let fixture = single_empty_legacy_system_catalog_fixture();
        let destination = fixture
            ._directory
            .path()
            .join("compacted-empty-legacy-config");

        let report = crate::storage::compact::compact_paged_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect("compact and promote the empty legacy config root");

        assert_eq!(report.reconciled_system_collections, 1);
        assert_eq!(report.imported_system_records, 0);
        let compacted =
            UnifiedStore::open_with_config(destination.join("data.rdb"), fixture.config.clone())
                .expect("cold reopen compacted store");
        assert_eq!(compacted.list_collections(), ["red.config"]);
        assert_eq!(
            compacted
                .get_collection("red.config")
                .expect("canonical config root")
                .count(),
            0
        );
    }

    #[test]
    fn canonical_system_config_recovery_replays_pending_unrelated_wal() {
        use crate::storage::wal::{WalRecord, WalWriter};

        let fixture = duplicate_system_catalog_fixture(true, true, true);
        let wal = reddb_file::unified_wal_path(&fixture.path);
        let action = reddb_file::encode_store_wal_action_frame(
            &reddb_file::StoreWalActionFrame::CreateCollection {
                name: "pending_collection".into(),
            },
        )
        .expect("encode pending action");
        let canonical_update = reddb_file::encode_store_wal_action_frame(
            &reddb_file::StoreWalActionFrame::UpsertEntityRecord {
                collection: "red.config".into(),
                record: UnifiedStore::serialize_entity_record(
                    &UnifiedEntity::table_row(
                        EntityId::new(3),
                        "red.config",
                        3,
                        vec![Value::text("canonical WAL")],
                    ),
                    None,
                    STORE_VERSION_CURRENT,
                ),
            },
        )
        .expect("encode canonical update");
        {
            let mut writer = WalWriter::open(&wal).expect("open WAL");
            writer
                .append(&WalRecord::TxCommitBatch {
                    tx_id: 999,
                    actions: vec![action, canonical_update],
                })
                .expect("append committed WAL");
            writer.sync().expect("sync WAL");
        }
        let original = std::fs::read(&fixture.path).expect("source bytes");
        let original_wal = std::fs::read(&wal).expect("WAL bytes");
        let destination = fixture._directory.path().join("recovered-canonical-config");

        let default_destination = fixture._directory.path().join("default-recovery");
        let error = crate::storage::compact::recover_paged_wal(&fixture.path, &default_destination)
            .expect_err("ordinary recovery must reject divergent identities");
        assert!(error
            .to_string()
            .contains("both populated legacy collection"));

        crate::storage::compact::recover_paged_wal_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect("recover pending WAL and retain canonical config");

        assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
        assert_eq!(std::fs::read(&wal).unwrap(), original_wal);
        for _ in 0..2 {
            let recovered = UnifiedStore::open_with_config(
                destination.join("data.rdb"),
                fixture.config.clone(),
            )
            .expect("open recovered store");
            assert!(recovered.get_collection("pending_collection").is_some());
            assert!(recovered.get("red.config", EntityId::new(1)).is_none());
            assert!(recovered.get("red.config", EntityId::new(2)).is_some());
            assert!(recovered.get("red.config", EntityId::new(3)).is_some());
        }
    }

    #[test]
    fn canonical_system_catalog_recovery_reconciles_all_metadata_and_keeps_missing_index() {
        use crate::storage::wal::{WalRecord, WalWriter};
        for case in [
            "valid",
            "view-conflict",
            "queue-delivery",
            "legacy-wal",
            "canonical-wal",
        ] {
            let fixture = duplicate_system_catalog_fixture_with_aliases(
                true,
                true,
                true,
                false,
                &["system.pid"],
                Some(case),
            );
            let wal = reddb_file::unified_wal_path(&fixture.path);
            {
                let mut writer = WalWriter::open(&wal).unwrap();
                let name = match case {
                    "legacy-wal" => "red_index_registry",
                    "canonical-wal" => "red.index.registry",
                    _ => "durable_rows",
                };
                writer
                    .append(&WalRecord::TxCommitBatch {
                        tx_id: 1001,
                        actions: vec![reddb_file::encode_store_wal_action_frame(
                            &reddb_file::StoreWalActionFrame::CreateCollection {
                                name: name.into(),
                            },
                        )
                        .unwrap()],
                    })
                    .unwrap();
                writer.sync().unwrap();
            }
            let original = std::fs::read(&fixture.path).unwrap();
            let original_wal = std::fs::read(&wal).unwrap();
            let baseline = UnifiedStore::open_with_config_for_recovery(
                &fixture.path,
                fixture.config.clone(),
                true,
                None,
                None,
            )
            .err()
            .expect("unproved aliases must still fail");
            assert!(baseline.to_string().contains("red_stats"), "{baseline}");
            assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
            let destination = fixture._directory.path().join("catalog-recovered");
            let result = crate::storage::compact::recover_paged_wal_canonical_system_config(
                &fixture.path,
                &destination,
            );
            assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
            assert_eq!(std::fs::read(&wal).unwrap(), original_wal);
            if case != "valid" {
                let error = result.expect_err(case).to_string();
                let expected = match case {
                    "view-conflict" => "conflicting canonical definition",
                    "queue-delivery" => "unsupported legacy metadata",
                    _ => "WAL changes metadata",
                };
                assert!(error.contains(expected), "{case}: {error}");
                assert!(!destination.join("data.rdb").exists());
                continue;
            }
            let report = result.expect("recover all supported aliases together");
            assert_eq!(report.reconciled_system_collections, 4);
            assert_eq!(report.imported_system_records, 1);
            for _ in 0..2 {
                let store = UnifiedStore::open_with_config(
                    destination.join("data.rdb"),
                    fixture.config.clone(),
                )
                .unwrap();
                assert!(store.get_collection("durable_rows").is_some());
                let mut imported = Vec::new();
                store
                    .get_collection("red.index.registry")
                    .unwrap()
                    .try_for_each_entity(4 * 1024 * 1024, |entity| {
                        if entity.data.as_row().and_then(|r| r.get_field("name"))
                            == Some(&Value::text("legacy_only"))
                        {
                            imported.push(entity.clone());
                        }
                        true
                    })
                    .unwrap();
                assert_eq!(imported.len(), 1, "legacy-only index must survive");
                assert!(
                    imported[0].id.raw() > 10003,
                    "all retired identities must be reserved first"
                );
                assert!(
                    matches!(imported[0].kind, EntityKind::TableRow{row_id,..} if row_id>10001)
                );
                assert_eq!(imported[0].logical_id(), imported[0].id);
                assert!(store
                    .get("red.index.registry", EntityId::new(10001))
                    .is_none());
                assert!(store
                    .get("red.index.registry", EntityId::new(101))
                    .is_some());
                assert!(store
                    .get("red.query.stats_catalog", EntityId::new(100))
                    .is_some());
                assert!(store
                    .get("red.query.stats_catalog", EntityId::new(10000))
                    .is_none());
                assert_eq!(store.get_collection("red.queue.meta").unwrap().count(), 1);
                assert_eq!(
                    store
                        .get_collection("red.materialized_view.definitions")
                        .unwrap()
                        .count(),
                    1
                );
                assert!(
                    store
                        .get_collection("red.index.registry")
                        .unwrap()
                        .identity_counters()
                        .0
                        > 10001
                );
            }
        }
    }

    #[test]
    fn canonical_system_config_recovery_retires_boot_wal_without_losing_other_actions() {
        use crate::storage::wal::{WalRecord, WalWriter};
        use reddb_file::StoreWalActionFrame as Action;
        for pending_dwb in [false, true] {
            let fixture = duplicate_system_catalog_fixture_with_key(
                true,
                true,
                true,
                false,
                Some("system.pid"),
            );
            let encode = |action| reddb_file::encode_store_wal_action_frame(&action).unwrap();
            let upsert = |id, value| {
                encode(Action::UpsertEntityRecord {
                    collection: "red_config".into(),
                    record: UnifiedStore::serialize_entity_record(
                        &config_recovery_row(id, "system.pid", value),
                        None,
                        STORE_VERSION_CURRENT,
                    ),
                })
            };
            let delete = |id| {
                encode(Action::DeleteEntityRecord {
                    collection: "red_config".into(),
                    entity_id: id,
                })
            };
            let wal = reddb_file::unified_wal_path(&fixture.path);
            {
                let mut writer = WalWriter::open(&wal).unwrap();
                writer
                    .append(&WalRecord::TxCommitBatch {
                        tx_id: 1001,
                        actions: vec![
                            upsert(10000, 3),
                            delete(1),
                            encode(Action::CreateCollection {
                                name: "durable_rows".into(),
                            }),
                        ],
                    })
                    .unwrap();
                writer
                    .append(&WalRecord::TxCommitBatch {
                        tx_id: 1002,
                        actions: vec![
                            upsert(10001, 4),
                            delete(10000),
                            encode(Action::UpsertEntityRecord {
                                collection: "durable_rows".into(),
                                record: UnifiedStore::serialize_entity_record(
                                    &UnifiedEntity::table_row(
                                        EntityId::new(3),
                                        "durable_rows",
                                        1,
                                        vec![Value::text("must survive")],
                                    ),
                                    None,
                                    STORE_VERSION_CURRENT,
                                ),
                            }),
                        ],
                    })
                    .unwrap();
                writer.sync().unwrap();
            }
            if pending_dwb {
                use std::io::{Seek, SeekFrom, Write};
                let bytes = std::fs::read(&fixture.path).unwrap();
                let root = fixture.legacy_root.unwrap();
                let offset = root as usize * PAGE_SIZE;
                let page: &[u8; PAGE_SIZE] = bytes[offset..offset + PAGE_SIZE].try_into().unwrap();
                let frame = reddb_file::encode_paged_dwb_frame([(root, page)]);
                let mut file = std::fs::OpenOptions::new()
                    .write(true)
                    .open(&fixture.path)
                    .unwrap();
                file.seek(SeekFrom::Start(3 * PAGE_SIZE as u64)).unwrap();
                file.write_all(&frame).unwrap();
                // Make raw-source identity inspection impossible: only the staged
                // DWB replay can restore the legacy configuration leaf.
                file.seek(SeekFrom::Start(offset as u64)).unwrap();
                file.write_all(&[0u8; PAGE_SIZE]).unwrap();
                file.sync_all().unwrap();
                authorize_fixture_dwb(&fixture.path, &frame);
            }
            let original = std::fs::read(&fixture.path).unwrap();
            let original_wal = std::fs::read(&wal).unwrap();
            let destination = fixture._directory.path().join("boot-config-recovered");
            let report = crate::storage::compact::recover_paged_wal_canonical_system_config(
                &fixture.path,
                &destination,
            )
            .expect("retire only proved boot metadata actions on the isolated copy");
            assert_eq!(report.retired_boot_system_wal_actions, 4);
            assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
            assert_eq!(std::fs::read(&wal).unwrap(), original_wal);
            for _ in 0..2 {
                let recovered = UnifiedStore::open_with_config(
                    destination.join("data.rdb"),
                    fixture.config.clone(),
                )
                .unwrap();
                let manager = recovered.get_collection("red.config").unwrap();
                assert!(manager.identity_counters().0 > 10001);
                assert!(manager.identity_counters().1 > 10001);
                assert_eq!(recovered.get_config("system.pid"), Some(Value::Integer(2)));
                assert!(recovered.get("red.config", EntityId::new(1)).is_none());
                assert!(recovered.get("red.config", EntityId::new(10000)).is_none());
                assert!(recovered.get("red.config", EntityId::new(10001)).is_none());
                assert_eq!(
                    recovered
                        .get("durable_rows", EntityId::new(3))
                        .unwrap()
                        .data
                        .as_row()
                        .unwrap()
                        .columns,
                    vec![Value::text("must survive")]
                );
            }
        }
    }

    #[test]
    fn canonical_system_config_recovery_requires_stable_boot_identity_evidence() {
        use crate::storage::wal::{WalRecord, WalWriter};
        let fixture =
            duplicate_system_catalog_fixture_with_key(true, true, false, false, Some("system.pid"));
        let original = std::fs::read(&fixture.path).unwrap();
        let wal = reddb_file::unified_wal_path(&fixture.path);
        {
            let mut writer = WalWriter::open(&wal).unwrap();
            writer
                .append(&WalRecord::TxCommitBatch {
                    tx_id: 1004,
                    actions: vec![reddb_file::encode_store_wal_action_frame(
                        &reddb_file::StoreWalActionFrame::DeleteEntityRecord {
                            collection: "red_config".into(),
                            entity_id: 1,
                        },
                    )
                    .unwrap()],
                })
                .unwrap();
            writer
                .append(&WalRecord::FullPageImage {
                    tx_id: 1004,
                    page_id: 1,
                    ckpt_epoch: 1,
                    data: original[PAGE_SIZE..2 * PAGE_SIZE].to_vec(),
                })
                .unwrap();
            writer.sync().unwrap();
        }
        let original_wal = std::fs::read(&wal).unwrap();
        let destination = fixture._directory.path().join("unproved-checkpoint");
        let error = crate::storage::compact::recover_paged_wal_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect_err("undo images must be recovered before proving legacy identities");
        assert!(error
            .to_string()
            .contains("requires a stable checkpoint image"));
        assert!(!destination.exists());
        assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
        assert_eq!(std::fs::read(&wal).unwrap(), original_wal);
    }

    #[test]
    fn canonical_system_config_recovery_rejects_unproved_boot_wal() {
        use crate::storage::wal::{WalRecord, WalWriter};
        use reddb_file::StoreWalActionFrame as Action;
        for (case, checkpoint_key, wal_key, delete_id) in [
            ("unknown-delete", "system.pid", "system.pid", Some(90000)),
            (
                "user-key",
                "system.customer_setting",
                "system.customer_setting",
                None,
            ),
            (
                "identity-reuse",
                "red.config.customer_setting",
                "system.pid",
                None,
            ),
        ] {
            let fixture = duplicate_system_catalog_fixture_with_key(
                true,
                true,
                false,
                false,
                Some(checkpoint_key),
            );
            let action = if let Some(entity_id) = delete_id {
                Action::DeleteEntityRecord {
                    collection: "red_config".into(),
                    entity_id,
                }
            } else {
                Action::UpsertEntityRecord {
                    collection: "red_config".into(),
                    record: UnifiedStore::serialize_entity_record(
                        &config_recovery_row(1, wal_key, 99),
                        None,
                        STORE_VERSION_CURRENT,
                    ),
                }
            };
            let wal = reddb_file::unified_wal_path(&fixture.path);
            {
                let mut writer = WalWriter::open(&wal).unwrap();
                writer
                    .append(&WalRecord::TxCommitBatch {
                        tx_id: 1003,
                        actions: vec![reddb_file::encode_store_wal_action_frame(&action).unwrap()],
                    })
                    .unwrap();
                writer.sync().unwrap();
            }
            let original = std::fs::read(&fixture.path).unwrap();
            let original_wal = std::fs::read(&wal).unwrap();
            let destination = fixture._directory.path().join(case);
            let error = crate::storage::compact::recover_paged_wal_canonical_system_config(
                &fixture.path,
                &destination,
            )
            .expect_err(case);
            assert!(
                error
                    .to_string()
                    .contains("WAL references divergent legacy red_config"),
                "{case}: {error}"
            );
            assert!(!destination.exists());
            assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
            assert_eq!(std::fs::read(&wal).unwrap(), original_wal);
        }
    }

    #[test]
    fn canonical_system_config_recovery_rejects_legacy_wal_actions() {
        use crate::storage::wal::{WalRecord, WalWriter};

        let fixture = duplicate_system_catalog_fixture(true, true, false);
        let wal = reddb_file::unified_wal_path(&fixture.path);
        let action = reddb_file::encode_store_wal_action_frame(
            &reddb_file::StoreWalActionFrame::DeleteEntityRecord {
                collection: "red_config".into(),
                entity_id: 1,
            },
        )
        .unwrap();
        {
            let mut writer = WalWriter::open(&wal).unwrap();
            writer
                .append(&WalRecord::TxCommitBatch {
                    tx_id: 999,
                    actions: vec![action],
                })
                .unwrap();
            writer.sync().unwrap();
        }
        let original = std::fs::read(&fixture.path).unwrap();
        let original_wal = std::fs::read(&wal).unwrap();
        let destination = fixture._directory.path().join("rejected-legacy-wal");
        let error = crate::storage::compact::recover_paged_wal_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect_err("legacy WAL must not be replayed against the canonical tree");
        assert!(error
            .to_string()
            .contains("WAL references divergent legacy red_config"));
        assert!(!destination.exists());
        assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
        assert_eq!(std::fs::read(&wal).unwrap(), original_wal);
    }

    #[test]
    fn canonical_system_config_recovery_rejects_legacy_cross_references() {
        use crate::storage::wal::{WalRecord, WalWriter};

        let fixture = duplicate_system_catalog_fixture_with_ref(true, true, false, true);
        let wal = reddb_file::unified_wal_path(&fixture.path);
        {
            let mut writer = WalWriter::open(&wal).unwrap();
            writer.append(&WalRecord::Begin { tx_id: 999 }).unwrap();
            writer.sync().unwrap();
        }
        let original = std::fs::read(&fixture.path).unwrap();
        let original_wal = std::fs::read(&wal).unwrap();
        let destination = fixture._directory.path().join("rejected-legacy-reference");
        let error = crate::storage::compact::recover_paged_wal_canonical_system_config(
            &fixture.path,
            &destination,
        )
        .expect_err("legacy references must not be retargeted silently");
        assert!(error
            .to_string()
            .contains("legacy red_config has cross-references"));
        assert!(!destination.join("data.rdb").exists());
        assert_eq!(std::fs::read(&fixture.path).unwrap(), original);
        assert_eq!(std::fs::read(&wal).unwrap(), original_wal);
    }

    #[test]
    fn populated_system_collection_aliases_reconcile_without_rebuild() {
        let fixture = duplicate_system_catalog_fixture(true, false, true);
        alias_duplicate_system_catalog_state(&fixture);

        let recovered = UnifiedStore::open_with_config(&fixture.path, fixture.config.clone())
            .expect("reconcile identical populated aliases");
        assert_eq!(recovered.list_collections(), vec!["red.config".to_string()]);
        assert_eq!(
            recovered
                .get("red.config", EntityId::new(1))
                .expect("preserved aliased row")
                .kind
                .collection(),
            "red.config"
        );
        assert_eq!(
            recovered.collection_root_page("red.config"),
            fixture.legacy_root,
            "the shared data root must be reused"
        );
        assert_eq!(
            recovered
                .key_indices
                .read()
                .get("red.config")
                .map(|index| index.tree.root_page_id()),
            fixture.legacy_key_root,
            "the shared key-index root must be reused"
        );
        assert_eq!(
            recovered
                .secondary_index("red.config", "by_id", &fixture.index_identity)
                .expect("shared secondary index")
                .root_page_id(),
            fixture
                .legacy_secondary_root
                .expect("fixture secondary root")
        );
        drop(recovered);

        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        let cold = UnifiedStore::open_with_config(&fixture.path, fixture.config)
            .expect("idempotent cold boot");
        assert_eq!(cold.list_collections(), vec!["red.config".to_string()]);
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);
        assert_eq!(
            cold.pager()
                .expect("pager")
                .page_count()
                .expect("page count"),
            fixture.pages_before,
            "reconciling a physical alias must not extend the file"
        );
    }

    #[test]
    fn populated_system_collection_aliases_merge_resume_counter_drift() {
        let fixture = duplicate_system_catalog_fixture(true, false, false);
        alias_duplicate_system_catalog_state_with(&fixture, |canonical| {
            canonical.entities += 7;
            canonical.entity_id_next += 100;
            canonical.row_id_next += 200;
            canonical.sequence_id_next += 300;
        });

        let recovered = UnifiedStore::open_with_config(&fixture.path, fixture.config.clone())
            .expect("reconcile shared roots with conservative counters");
        let manager = recovered
            .get_collection("red.config")
            .expect("canonical collection");
        assert_eq!(manager.count(), 1, "the shared tree owns the exact count");
        assert_eq!(manager.identity_counters(), (102, 202));
        assert_eq!(manager.sequence_id_next(), 301);
        assert_eq!(
            recovered.collection_root_page("red.config"),
            fixture.legacy_root,
            "counter reconciliation must keep the shared data root"
        );
        drop(manager);
        drop(recovered);

        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        let cold = UnifiedStore::open_with_config(&fixture.path, fixture.config)
            .expect("idempotent cold boot after counter reconciliation");
        assert_eq!(cold.list_collections(), vec!["red.config".to_string()]);
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);
        assert_eq!(
            cold.pager()
                .expect("pager")
                .page_count()
                .expect("page count"),
            fixture.pages_before,
            "counter reconciliation must not extend the file"
        );
    }

    fn row(id: u64, value: i64) -> UnifiedEntity {
        UnifiedEntity::table_row(EntityId::new(id), "rows", id, vec![Value::Integer(value)])
    }

    #[test]
    fn paged_checkpoint_excludes_uncommitted_graph_versions() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("graph.rdb");
        let store = UnifiedStore::open(&path).unwrap();
        let snapshots = Arc::new(crate::storage::transaction::snapshot::SnapshotManager::new());
        assert!(store.snapshot_manager.set(Arc::clone(&snapshots)).is_ok());
        store.create_collection("graph").unwrap();
        let node = UnifiedEntity::graph_node(EntityId::new(1), "a", "person", HashMap::new());
        let edge =
            UnifiedEntity::graph_edge(EntityId::new(2), "knows", "1", "1", 1.0, HashMap::new());
        store.insert("graph", node).unwrap();
        store.insert("graph", edge).unwrap();
        store.persist().unwrap();

        let xid = snapshots.begin();
        let manager = store.get_collection("graph").unwrap();
        for id in [1, 2] {
            let mut entity = manager.get(EntityId::new(id)).unwrap();
            entity.xmax = xid;
            manager.update(entity).unwrap();
        }
        let mut node = UnifiedEntity::graph_node(EntityId::new(3), "b", "person", HashMap::new());
        let mut edge =
            UnifiedEntity::graph_edge(EntityId::new(4), "knows", "1", "3", 1.0, HashMap::new());
        node.xmin = xid;
        edge.xmin = xid;
        store.insert("graph", node).unwrap();
        store.insert("graph", edge).unwrap();
        store
            .add_cross_ref(
                "graph",
                EntityId::new(1),
                "graph",
                EntityId::new(4),
                RefType::RelatedTo,
                1.0,
            )
            .unwrap();
        store.persist().unwrap();
        snapshots.rollback(xid);
        store.persist().unwrap();
        drop(manager);
        drop(store);

        let recovered = UnifiedStore::open(&path).unwrap();
        let manager = recovered.get_collection("graph").unwrap();
        assert_eq!(manager.get(EntityId::new(1)).unwrap().xmax, 0);
        assert_eq!(manager.get(EntityId::new(2)).unwrap().xmax, 0);
        assert!(manager.get(EntityId::new(3)).is_none());
        assert!(manager.get(EntityId::new(4)).is_none());
        assert!(!manager
            .get(EntityId::new(1))
            .unwrap()
            .cross_refs()
            .iter()
            .any(|reference| reference.target == EntityId::new(4)));
        assert!(!recovered
            .get_refs_from(EntityId::new(1))
            .iter()
            .any(|(target, _, _)| *target == EntityId::new(4)));
    }

    #[test]
    fn paged_checkpoint_restores_graph_references_after_commit() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("graph.rdb");
        let store = UnifiedStore::open(&path).unwrap();
        let snapshots = Arc::new(crate::storage::transaction::snapshot::SnapshotManager::new());
        assert!(store.snapshot_manager.set(Arc::clone(&snapshots)).is_ok());
        store.create_collection("graph").unwrap();
        store
            .insert(
                "graph",
                UnifiedEntity::graph_node(EntityId::new(1), "a", "person", HashMap::new()),
            )
            .unwrap();
        store.persist().unwrap();
        let xid = snapshots.begin();
        let mut edge =
            UnifiedEntity::graph_edge(EntityId::new(2), "knows", "1", "1", 1.0, HashMap::new());
        edge.xmin = xid;
        store.insert("graph", edge).unwrap();
        store
            .add_cross_ref(
                "graph",
                EntityId::new(1),
                "graph",
                EntityId::new(2),
                RefType::RelatedTo,
                1.0,
            )
            .unwrap();
        store.persist().unwrap();
        snapshots.commit(xid);
        store.persist().unwrap();
        drop(store);

        let recovered = UnifiedStore::open(&path).unwrap();
        let node = recovered.get("graph", EntityId::new(1)).unwrap();
        assert!(recovered.get("graph", EntityId::new(2)).is_some());
        assert!(node
            .cross_refs()
            .iter()
            .any(|reference| reference.target == EntityId::new(2)));
        assert!(recovered
            .get_refs_from(EntityId::new(1))
            .iter()
            .any(|(target, _, _)| *target == EntityId::new(2)));
    }

    #[test]
    fn checkpoint_reuse_tracks_manager_updates_metadata_deletes_and_bulk_inserts() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        store.create_collection("rows").expect("collection");
        store.insert("rows", row(1, 1)).expect("seed");
        store.persist().expect("initial checkpoint");
        let manager = store.get_collection("rows").expect("rows");
        manager.update(row(1, 2)).expect("direct update");
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("updated row checkpoint");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 1);
        let mut metadata = Metadata::new();
        metadata.set("verified", MetadataValue::Bool(true));
        manager
            .set_metadata(EntityId::new(1), metadata)
            .expect("metadata");
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("metadata-only checkpoint");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 1);
        manager
            .bulk_insert(vec![row(2, 2), row(3, 3)])
            .expect("bulk insert");
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("bulk insert checkpoint");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 3);
        manager.delete(EntityId::new(2)).expect("delete");
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("changed checkpoint");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 2);
        drop(store);
        let recovered = UnifiedStore::open(&path).expect("recover changed rows");
        let manager = recovered.get_collection("rows").expect("rows");
        assert_eq!(manager.count(), 2);
        assert_eq!(
            manager
                .get(EntityId::new(1))
                .expect("row")
                .data
                .as_row()
                .expect("table")
                .columns,
            vec![Value::Integer(2)]
        );
        assert!(manager.get(EntityId::new(2)).is_none());
        assert!(manager
            .get_metadata(EntityId::new(1))
            .expect("metadata")
            .get("verified")
            .is_some());
    }

    #[test]
    fn unrelated_visibility_changes_do_not_serialize_collection_rows() {
        let directory = tempfile::tempdir().expect("temporary store");
        let store = UnifiedStore::open(&directory.path().join("data.rdb")).expect("paged store");
        let snapshots = Arc::new(crate::storage::transaction::snapshot::SnapshotManager::new());
        assert!(store.snapshot_manager.set(Arc::clone(&snapshots)).is_ok());
        store.create_collection("rows").expect("collection");
        for id in 1..=40 {
            store.insert("rows", row(id, id as i64)).expect("seed");
        }
        store.persist().expect("initial checkpoint");
        let unrelated = snapshots.begin();
        snapshots.rollback(unrelated);
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("unrelated rollback checkpoint");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);

        store
            .pending_vault_versions
            .write()
            .insert(EntityId::new(100));
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store.persist().expect("unrelated vault checkpoint");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);
        store.pending_vault_versions.write().clear();
        CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(0));
        store
            .persist()
            .expect("unrelated vault publication checkpoint");
        assert_eq!(CHECKPOINT_SERIALIZED_ROWS.with(|count| count.get()), 0);
    }

    #[test]
    fn low_level_rollback_of_a_non_active_xid_invalidates_sparse_visibility() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        let snapshots = Arc::new(crate::storage::transaction::snapshot::SnapshotManager::new());
        assert!(store.snapshot_manager.set(Arc::clone(&snapshots)).is_ok());
        store.create_collection("rows").expect("collection");
        let xid = snapshots.begin();
        let mut entity = row(1, 1);
        entity.xmin = xid;
        store.insert("rows", entity).expect("committed row");
        snapshots.commit(xid);
        store.persist().expect("committed baseline");
        snapshots.rollback(xid);
        store
            .persist()
            .expect("untracked outcome requires validation");
        drop(store);
        let recovered = UnifiedStore::open(&path).expect("recover");
        assert_eq!(recovered.get_collection("rows").expect("rows").count(), 0);
    }

    #[test]
    fn exhausted_deferred_row_cache_keeps_full_validation() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        let snapshots = Arc::new(crate::storage::transaction::snapshot::SnapshotManager::new());
        assert!(store.snapshot_manager.set(Arc::clone(&snapshots)).is_ok());
        store.create_collection("rows").expect("collection");
        let xid = snapshots.begin();
        for id in 1..=1025 {
            let mut entity = row(id, id as i64);
            entity.xmin = xid;
            store.insert("rows", entity).expect("pending row");
        }
        store.persist().expect("exclude pending rows");
        assert!(store.checkpoint_versions.read()["rows"]
            .deferred_rows
            .is_none());
        snapshots.commit(xid);
        store
            .persist()
            .expect("publish committed rows conservatively");
        drop(store);
        let recovered = UnifiedStore::open(&path).expect("recover");
        assert_eq!(
            recovered.get_collection("rows").expect("rows").count(),
            1025
        );
    }

    #[test]
    fn rollback_invalidates_reuse_without_a_segment_mutation() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        let snapshots = Arc::new(crate::storage::transaction::snapshot::SnapshotManager::new());
        assert!(store.snapshot_manager.set(Arc::clone(&snapshots)).is_ok());
        store.create_collection("rows").expect("collection");
        let xid = snapshots.begin();
        let mut pending = row(1, 1);
        pending.xmin = xid;
        store.insert("rows", pending).expect("version");
        store.persist().expect("before rollback");
        snapshots.rollback(xid);
        store.persist().expect("after rollback");
        drop(store);
        let recovered = UnifiedStore::open(&path).expect("recover");
        assert_eq!(recovered.get_collection("rows").expect("rows").count(), 0);
    }

    #[test]
    fn rolled_back_delete_is_normalized_without_a_segment_mutation() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        let snapshots = Arc::new(crate::storage::transaction::snapshot::SnapshotManager::new());
        assert!(store.snapshot_manager.set(Arc::clone(&snapshots)).is_ok());
        store.create_collection("rows").expect("collection");
        let xid = snapshots.begin();
        let mut deleted = row(1, 1);
        deleted.xmax = xid;
        store.insert("rows", deleted).expect("version");
        store.persist().expect("before rollback");
        snapshots.rollback(xid);
        store.persist().expect("after rollback");
        drop(store);
        let recovered = UnifiedStore::open(&path).expect("recover");
        let manager = recovered.get_collection("rows").expect("rows");
        assert_eq!(manager.get(EntityId::new(1)).expect("retained row").xmax, 0);
    }

    #[test]
    fn publishing_a_pending_vault_version_invalidates_reuse() {
        let directory = tempfile::tempdir().expect("temporary store");
        let path = directory.path().join("data.rdb");
        let store = UnifiedStore::open(&path).expect("paged store");
        store.create_collection("rows").expect("collection");
        store.insert("rows", row(1, 1)).expect("row");
        store
            .pending_vault_versions
            .write()
            .insert(EntityId::new(1));
        store.persist().expect("exclude pending vault version");
        store
            .pending_vault_versions
            .write()
            .remove(&EntityId::new(1));
        store.persist().expect("include published vault version");
        drop(store);
        let recovered = UnifiedStore::open(&path).expect("recover");
        assert_eq!(recovered.get_collection("rows").expect("rows").count(), 1);
    }
}