dig_blockstore/

store.rs

//! `BlockStore` — RocksDB-backed persistent block and chain state.
//!
//! # Architecture
//!
//! This module is the primary entry point for all block persistence in the DIG L2
//! network. It mirrors the storage patterns established by the **Chia blockchain**'s
//! `BlockStore` in `chia-blockchain/chia/consensus/block_store.py`, adapted for Rust
//! and RocksDB instead of Python/SQLite:
//!
//! | Chia Python pattern | DIG Rust equivalent |
//! |---------------------|---------------------|
//! | `full_blocks` SQLite table | [`CF_BLOCKS`](crate::CF_BLOCKS) column family (zstd-compressed bincode) |
//! | `block_records` SQLite table | In-memory [`BlockRecord`](crate::BlockRecord) cache (never persisted; [`TYP-004`]) |
//! | `block_cache: LRUCache[bytes32, FullBlock]` | [`ShardedBlockCache`](crate::cache::sharded::ShardedBlockCache) (sharded LRU, [`CAC-001`]) |
//! | `current_peak` single-row | [`META_TIP`](crate::META_TIP) in [`CF_METADATA`](crate::CF_METADATA) (40-byte [`ChainTip`]) |
//! | `INSERT OR IGNORE` idempotency | [`put_block`](BlockStore::put_block) existence check → `Ok(false)` |
//! | `BlockHeightMap` bytearray | [`CF_CANONICAL`](crate::CF_CANONICAL) + `canonical.bin` mmap ([`CAN-001`](../docs/requirements/domains/canonical_chain/specs/CAN-001.md), [`crate::canonical::mmap`](crate::canonical::mmap)) |
//!
//! # Column family ownership
//!
//! - [`CF_BLOCKS`]: Compressed full block bodies keyed by header hash ([`SER-001`]).
//! - [`CF_HEADERS`]: Uncompressed bincode headers keyed by header hash ([`SER-002`]).
//! - [`CF_CANONICAL`]: Dense height→hash index for the canonical chain ([`CAN-001`]).
//! - [`CF_METADATA`]: Tip, genesis hash, schema version, zstd dictionary ([`TYP-002`]).
//! - [`CF_ATTESTED`]: [`AttestedBlock`](dig_block::AttestedBlock) rows via [`BlockStore::put_attestation`] ([`BLK-009`](../docs/requirements/domains/block_storage/specs/BLK-009.md)).
//! - [`CF_CHECKPOINTS`]: Checkpoint storage ([`CKP-*`](../docs/requirements/IMPLEMENTATION_ORDER.md) Phase 9).
//!
//! # Three-tier read path
//!
//! Every `get_*` method follows a consistent tiered lookup:
//!
//! 1. **In-memory cache** — sharded LRU for blocks/headers, `HashMap` for records.
//!    Cache hits return clones with zero RocksDB I/O.
//! 2. **RocksDB column family** — on miss, raw bytes are fetched, deserialized, and
//!    inserted back into the cache (read-through).
//! 3. **Absent** — `Ok(None)` when the key does not exist at any tier.
//!
//! # Concurrency model
//!
//! `BlockStore` uses `&self` for all public methods (no `&mut self`). Interior
//! mutability is provided by:
//! - [`parking_lot::RwLock`] for tip and zstd dictionary (read-heavy, rare writes).
//! - [`parking_lot::Mutex`] for the record cache (short critical sections).
//! - [`std::sync::atomic::AtomicUsize`] for instrumentation counters (lock-free).
//! - [`Arc<DB>`] for the RocksDB handle (thread-safe by design).
//!
//! # Requirements trace
//!
//! - [`STR-004`](../docs/requirements/domains/crate_structure/specs/STR-004.md) — constructors and lifecycle.
//! - [`BLK-001`](../docs/requirements/domains/block_storage/specs/BLK-001.md) — `put_block` / `put`.
//! - [`BLK-002`](../docs/requirements/domains/block_storage/specs/BLK-002.md) — `get_block` with block cache.
//! - [`BLK-003`](../docs/requirements/domains/block_storage/specs/BLK-003.md) — `get_header` with header cache.
//! - [`BLK-004`](../docs/requirements/domains/block_storage/specs/BLK-004.md) — `get_record` with layered caching.
//! - [`BLK-005`](../docs/requirements/domains/block_storage/specs/BLK-005.md) — `get_blocks_by_hash` batch retrieval.
//! - [`BLK-006`](../docs/requirements/domains/block_storage/specs/BLK-006.md) — `stream_blocks_in_range` sequential readahead.
//! - [`BLK-007`](../docs/requirements/domains/block_storage/specs/BLK-007.md) — async read wrappers (`get_block_async`, …).
//! - [`BLK-008`](../docs/requirements/domains/block_storage/specs/BLK-008.md) — async write pipeline (`put_pipelined`, batched `WriteBatch`).
//! - [`BLK-009`](../docs/requirements/domains/block_storage/specs/BLK-009.md) — `put_attestation` / `get_attestation` on [`CF_ATTESTED`].
//! - [`BLK-010`](../docs/requirements/domains/block_storage/specs/BLK-010.md) — `update_status` on in-memory [`BlockRecord`] only.
//! - [`BLK-011`](../docs/requirements/domains/block_storage/specs/BLK-011.md) — `has_block` lightweight existence by hash.
//! - [`BLK-012`](../docs/requirements/domains/block_storage/specs/BLK-012.md) — `stats` aggregate [`StorageStats`](crate::types::StorageStats) snapshot.
//! - [`BLK-013`](../docs/requirements/domains/block_storage/specs/BLK-013.md) — `flush` / `compact` maintenance on the shared [`rocksdb::DB`].
//! - [`BLK-014`](../docs/requirements/domains/block_storage/specs/BLK-014.md) — `get_blocks_in_range` and sync `get_block_by_height` over [`CF_CANONICAL`].
//! - [`BLK-015`](../docs/requirements/domains/block_storage/specs/BLK-015.md) — `get_records_in_range` / `get_record_by_height` (header-derived, no block bodies).
//! - [`CAN-001`](../docs/requirements/domains/canonical_chain/specs/CAN-001.md) — dual-layer canonical index (`canonical.bin` + [`CF_CANONICAL`]).
//! - [`CAN-003`](../docs/requirements/domains/canonical_chain/specs/CAN-003.md) — [`set_canonical`](BlockStore::set_canonical) for existing stored blocks.
//! - [`CAN-004`](../docs/requirements/domains/canonical_chain/specs/CAN-004.md) — [`set_canonical_batch`](BlockStore::set_canonical_batch) (single [`WriteBatch`](rocksdb::WriteBatch) for reorg-scale promotion).
//! - [`SER-001`](../docs/requirements/domains/serialization/specs/SER-001.md) — bincode + zstd block serialization.
//! - [`SER-002`](../docs/requirements/domains/serialization/specs/SER-002.md) — bincode-only header serialization.
//! - [`SER-005`](../docs/requirements/domains/serialization/specs/SER-005.md) — dictionary training and persistence.
//! - [`CAC-001`](../docs/requirements/domains/caching/specs/CAC-001_sharded_block_cache.md) — sharded block LRU.
//! - [`CAC-002`](../docs/requirements/domains/caching/specs/CAC-002_sharded_header_cache.md) — sharded header LRU.
//! - [`CAC-006`](../docs/requirements/domains/caching/specs/CAC-006_cache_warming_on_startup.md) — startup warming.
//!
//! **Spec:** `docs/resources/SPEC.md` §15.1 (constructors), §16 (crate boundary).

use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;

use tokio::sync::mpsc;

use chia_protocol::Bytes32;
use dig_block::{AttestedBlock, BlockStatus, L2Block, L2BlockHeader};
use parking_lot::{Mutex, RwLock};
use rocksdb::{Direction, IteratorMode, Options, WriteBatch, DB};

use crate::cache::sharded::{ShardedBlockCache, ShardedHeaderCache, ShardedLruCache};
use crate::canonical::mmap::CanonicalBin;
use crate::cf_options;
use crate::compression::resolve_zstd_dictionary;
use crate::constants::{
    CF_ATTESTED, CF_BLOCKS, CF_CANONICAL, CF_CHECKPOINTS, CF_HEADERS, CF_METADATA,
    META_GENESIS_HASH, META_MIN_HEIGHT, META_TIP,
};
use crate::encoding::{hash_key, height_key};
use crate::error::{
    BlockStoreError, ERR_ASYNC_JOIN_PREFIX, ERR_INIT_GENESIS_ALREADY_INITIALIZED,
    ERR_INIT_GENESIS_READ_ONLY, ERR_MUTATION_READ_ONLY, ERR_OPEN_READONLY_PATH_MISSING_PREFIX,
    ERR_UPDATE_STATUS_RECORD_NOT_CACHED_PREFIX,
};
use crate::pipeline::PipelineJob;
use crate::types::{BlockRecord, ChainTip, ReorgResult, StorageStats};
use crate::BlockStoreConfig;

pub use crate::pipeline::StreamBlocksInRange;

/// Shared RocksDB + cache state behind [`BlockStore`].
///
/// **Why a separate type ([`BLK-007`](../docs/requirements/domains/block_storage/specs/BLK-007.md)):** [`BlockStore`]
/// wraps this struct in [`Arc`] so [`BlockStore::clone`] is a single refcount increment. Async helpers move a clone
/// into [`tokio::task::spawn_blocking`] while preserving one logical store (atomics + caches stay shared).
///
/// Public API remains on [`BlockStore`] via [`std::ops::Deref`]. The type is `pub` so [`Deref::Target`] is
/// well-formed; external crates should still depend on [`BlockStore`] methods only ([`BLK-008`](../docs/requirements/domains/block_storage/specs/BLK-008.md) will deepen `Arc` sharing).
#[doc(hidden)]
pub struct BlockStoreInner {
    /// RocksDB handle shared across all operations. Thread-safe via RocksDB's internal locking.
    /// All six column families ([`TYP-001`]) are created at open time.
    pub(crate) db: Arc<DB>,
    /// When `true`, all mutation APIs (`put_block`, `init_genesis`, [`BlockStore::put_attestation`](BlockStore::put_attestation)) return
    /// [`BlockStoreError::Serialization`] with [`ERR_MUTATION_READ_ONLY`].
    /// Set by [`BlockStore::open_readonly`]; cannot be toggled after construction.
    pub(crate) read_only: bool,
    /// In-memory copy of the chain tip from [`META_TIP`] in [`CF_METADATA`].
    /// Updated atomically after [`BlockStore::init_genesis`] and future tip-advance APIs.
    /// Reads use [`RwLock::read`] (very cheap with parking_lot); writes are rare (new blocks).
    pub(crate) tip: RwLock<Option<ChainTip>>,
    /// Count of blocks verified present during cache warming at last [`BlockStore::open`].
    /// Exposed via [`BlockStore::warm_blocks_loaded_count`] for startup diagnostics.
    pub(crate) warm_blocks_loaded: AtomicUsize,
    /// Zstd level for [`BlockStore::serialize_block`] / plain fallback ([`SER-001`](../docs/requirements/domains/serialization/specs/SER-001.md) §6).
    pub(crate) compression_level: i32,
    /// When true and [`Self::zstd_dict`] is [`Some`], compress with dictionary ([`SER-005`](../docs/requirements/domains/serialization/specs/SER-005.md) precursor).
    pub(crate) use_compression_dict: bool,
    /// Cap passed to [`zstd::bulk::Decompressor::decompress`] ([`SER-001`](../docs/requirements/domains/serialization/specs/SER-001.md) implementation notes).
    pub(crate) max_decompressed_block_bytes: usize,
    /// Trained dictionary loaded from [`META_ZSTD_DICT`] or [`BlockStoreConfig::zstd_dictionary_override`].
    ///
    /// **[`SER-005`](../docs/requirements/domains/serialization/specs/SER-005.md):** [`RwLock`] lets
    /// [`BlockStore::maybe_train_dictionary`] publish the first trained dictionary **after** the write that crosses
    /// [`DICT_TRAINING_THRESHOLD`](crate::constants::DICT_TRAINING_THRESHOLD) while keeping [`BlockStore`] on an
    /// immutable `&self` API surface (matches `put`-style ergonomics slated for [`BLK-001`](../docs/requirements/domains/block_storage/specs/BLK-001.md)).
    pub(crate) zstd_dict: RwLock<Option<Arc<Vec<u8>>>>,
    /// [`BlockRecord`] rows derived on write; **never** persisted ([`BLK-001`](../docs/requirements/domains/block_storage/specs/BLK-001.md), [`CAC-003`](../docs/requirements/domains/caching/specs/CAC-003.md) precursor).
    ///
    /// **Concurrency:** [`parking_lot::Mutex`] keeps inserts from [`BlockStore::put_block`] / [`BlockStore::init_genesis`] and
    /// lookups from [`BlockStore::get_record`] safe without `&mut self`.
    pub(crate) record_cache: Mutex<HashMap<Bytes32, BlockRecord>>,
    /// Sharded LRU of deserialized [`L2Block`] values ([`BLK-002`](../docs/requirements/domains/block_storage/specs/BLK-002.md)).
    pub(crate) block_cache: Arc<ShardedBlockCache>,
    /// Count of RocksDB `get_cf` calls against [`CF_BLOCKS`] issued from [`BlockStore::get_block`] **after** a cache miss.
    ///
    /// **Rationale:** Proves AC §2 “no I/O on hit” in `tests/blk_002_tests.rs`; cheap atomic hot path on miss only.
    /// **Not incremented** by [`BlockStore::get_blocks_by_hash`] (that path uses [`DB::multi_get_cf`](rocksdb::DB::multi_get_cf); see [`BlockStore::cf_blocks_multi_get_batch_count`]).
    pub(crate) cf_blocks_physical_gets: AtomicUsize,
    /// Count of [`rocksdb::DB::multi_get_cf`] **batch invocations** from [`BlockStore::get_blocks_by_hash`] when the input
    /// contains at least one block-cache miss ([`BLK-005`](../docs/requirements/domains/block_storage/specs/BLK-005.md) AC §3).
    ///
    /// **Semantics:** Increments by **at most one per `get_blocks_by_hash` call** that performs RocksDB I/O (all misses
    /// share one `multi_get_cf` round-trip). Stays at zero when every hash hits [`BlockStore::block_cache`] or when `hashes` is empty.
    pub(crate) cf_blocks_multi_get_batches: AtomicUsize,
    /// Count of [`DB::get_cf_opt`](rocksdb::DB::get_cf_opt) calls against [`CF_BLOCKS`] from [`StreamBlocksInRange`]
    /// ([`BLK-006`](../docs/requirements/domains/block_storage/specs/BLK-006.md)) after a block-cache miss.
    ///
    /// **Rationale:** Distinct from [`BlockStore::cf_blocks_physical_get_count`] ([`get_block`](BlockStore::get_block)) so tests can
    /// prove cache hits in a streamed range skip redundant block-blob reads ([`tests/blk_006_tests.rs`]).
    pub(crate) cf_blocks_stream_physical_gets: AtomicUsize,
    /// Copy of [`BlockStoreConfig::readahead_size`](crate::BlockStoreConfig::readahead_size) at open time ([`BLK-006`](../docs/requirements/domains/block_storage/specs/BLK-006.md) AC §4).
    pub(crate) readahead_size: usize,
    /// Sharded LRU of [`L2BlockHeader`] values ([`BLK-003`](../docs/requirements/domains/block_storage/specs/BLK-003.md)).
    ///
    /// **Separate** from [`Self::block_cache`] per BLK-003 implementation notes (tunables: [`BlockStoreConfig::header_cache_capacity`](crate::BlockStoreConfig::header_cache_capacity)).
    pub(crate) header_cache: Arc<ShardedHeaderCache>,
    /// Count of RocksDB `get_cf` calls against [`CF_HEADERS`] after **both** [`Self::header_cache`] and
    /// [`Self::record_cache`] miss — incremented by [`BlockStore::get_header`] and by [`BlockStore::get_record`] ([`BLK-003`](../docs/requirements/domains/block_storage/specs/BLK-003.md), [`BLK-004`](../docs/requirements/domains/block_storage/specs/BLK-004.md)).
    pub(crate) cf_headers_physical_gets: AtomicUsize,
    /// Max jobs per RocksDB [`WriteBatch`] flush ([`BlockStoreConfig::write_pipeline_batch_size`](crate::BlockStoreConfig::write_pipeline_batch_size)).
    pub(crate) pipeline_batch_size: usize,
    /// Partial-batch flush timer ([`BlockStoreConfig::write_pipeline_flush_ms`](crate::BlockStoreConfig::write_pipeline_flush_ms)).
    pub(crate) pipeline_flush_ms: u64,
    /// Bounded channel depth ([`BlockStoreConfig::write_pipeline_channel_capacity`](crate::BlockStoreConfig::write_pipeline_channel_capacity)).
    pub(crate) pipeline_channel_capacity: usize,
    /// Count of successful [`DB::write`](rocksdb::DB::write) calls issued **only** by the pipeline worker ([`tests/blk_008_tests.rs`]).
    pub(crate) pipeline_write_batches: AtomicUsize,
    /// Dense height→hash mmap sidecar (`canonical.bin`) kept in lockstep with [`CF_CANONICAL`] ([`CAN-001`](../docs/requirements/domains/canonical_chain/specs/CAN-001.md)).
    ///
    /// **Reads:** [`parking_lot::RwLock::read`] for [`Self::get_hash_by_height`] (hot path). **Writes:**
    /// [`RwLock::write`] after every successful RocksDB batch that touches the canonical index (`init_genesis`, [`BlockStore::put_block`], pipeline flush).
    pub(crate) canonical_bin: RwLock<CanonicalBin>,
    /// In-memory height→hash cache for hot canonical heights ([`CAC-004`](../docs/requirements/domains/caching/specs/CAC-004_canonical_height_index_cache.md)).
    ///
    /// **BTreeMap** provides O(log n) lookup and ordered iteration for range queries.
    /// Populated from `set_canonical`, `put_block(canonical=true)`, and `get_hash_by_height` read-through.
    /// Evicted on rollback. Bounded by `canonical_height_cache_capacity` (default 10,000).
    /// Cached `META_MIN_HEIGHT` for fast access by rollback validation and compaction filter ([`PRN-004`]).
    /// Loaded from CF_METADATA at startup; updated with `Release` ordering after prune succeeds.
    /// Shared with the compaction filter ([`PRN-003`]) when `enable_compaction_pruning` is true.
    /// The filter reads this with `Acquire` ordering; `prune_before_height` writes with `Release`.
    pub(crate) min_retained_height_cached: Arc<AtomicU64>,
    pub(crate) canonical_height_cache: RwLock<std::collections::BTreeMap<u64, Bytes32>>,
    /// Max entries before the lowest-height entry is evicted from [`Self::canonical_height_cache`].
    pub(crate) canonical_height_cache_capacity: usize,
    /// Hash→height reverse lookup cache ([`CAC-005`](../docs/requirements/domains/caching/specs/CAC-005_hash_to_height_reverse_cache.md)).
    ///
    /// Sharded LRU with `u64` values (block height). Populated on `put_block`, header reads,
    /// and block reads. Used by `find_common_ancestor` and `set_canonical` to avoid header
    /// deserialization solely for height extraction.
    pub(crate) hash_to_height_cache: Arc<ShardedLruCache<u64>>,
}

/// Primary handle for all block persistence APIs.
///
/// # Chia blockchain analogy
///
/// This struct corresponds to `BlockStore` in `chia-blockchain/chia/consensus/block_store.py`.
/// Where Chia uses a single SQLite `full_blocks` table with Python LRU caches, DIG uses
/// RocksDB column families with Rust sharded LRU caches for higher throughput under
/// concurrent access. The API surface mirrors Chia's: `add_full_block` → [`put_block`](Self::put_block),
/// `get_full_block` → [`get_block`](Self::get_block), `get_block_record` → [`get_record`](Self::get_record).
///
/// # Ownership
///
/// Thin [`Arc`] around [`BlockStoreInner`]: cheap [`Clone`] for [`tokio::task::spawn_blocking`] dispatch ([`BLK-007`](../docs/requirements/domains/block_storage/specs/BLK-007.md)).
/// Field access on `&BlockStore` transparently reaches [`BlockStoreInner`] via [`std::ops::Deref`].
///
/// **Write pipeline ([`BLK-008`](../docs/requirements/domains/block_storage/specs/BLK-008.md)):** [`Self::pipeline_tx`]
/// holds the lazy [`mpsc::Sender`] **outside** [`BlockStoreInner`]. The worker task also keeps an [`Arc`] to `inner`
/// for RocksDB; if the sender lived on `inner`, dropping all [`BlockStore`] handles would still leave the sender
/// alive (circular retention), the channel would never close, and AC §8 “flush on shutdown” would not run.
///
/// # Construction
///
/// Use [`BlockStore::open`] for read-write access or [`BlockStore::open_readonly`] for read-only
/// access to an existing database. After construction, call [`BlockStore::init_genesis`] once
/// to initialize a new chain.
pub struct BlockStore {
    pub(crate) inner: Arc<BlockStoreInner>,
    /// Lazy bounded ingress for [`Self::put_pipelined`] — **not** stored on [`BlockStoreInner`] (see struct docs).
    pub(crate) pipeline_tx: Arc<tokio::sync::Mutex<Option<mpsc::Sender<PipelineJob>>>>,
}

impl Clone for BlockStore {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            pipeline_tx: self.pipeline_tx.clone(),
        }
    }
}

impl std::ops::Deref for BlockStore {
    type Target = BlockStoreInner;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

impl BlockStore {
    /// Open or create a store at `config.path` with all column families ([`STR-004`](../docs/requirements/domains/crate_structure/specs/STR-004.md), [`TYP-008`](../docs/requirements/domains/storage_types/specs/TYP-008.md)).
    pub fn open(config: BlockStoreConfig) -> Result<Self, BlockStoreError> {
        let compression_level = config.compression_level;
        let use_compression_dict = config.use_compression_dict;
        let max_decompressed_block_bytes = config.max_decompressed_block_bytes;
        let zstd_dictionary_override = config.zstd_dictionary_override.clone();
        // ERR-001 has no `Io` variant; surface directory creation failures as [`BlockStoreError::Serialization`]
        // until the taxonomy adds filesystem errors ([`ERR-001`](../docs/requirements/domains/error_types/specs/ERR-001_blockstoreerror_enum.md)).
        std::fs::create_dir_all(&config.path).map_err(|e| {
            BlockStoreError::Serialization(format!(
                "filesystem error creating database directory {}: {e}",
                config.path.display()
            ))
        })?;
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.create_missing_column_families(true);
        // PRN-003: create shared AtomicU64 for compaction filter BEFORE DB open
        let prune_threshold = if config.enable_compaction_pruning {
            Some(Arc::new(AtomicU64::new(0)))
        } else {
            None
        };
        let cfs = cf_options::column_family_descriptors(&config, prune_threshold.clone());
        let db = DB::open_cf_descriptors(&opts, &config.path, cfs)?;
        let db = Arc::new(db);
        let canonical_bin =
            RwLock::new(CanonicalBin::open_synced(&db, config.path.as_path(), true)?);
        let zstd_dict =
            resolve_zstd_dictionary(&db, use_compression_dict, zstd_dictionary_override)?;
        let tip = load_tip(&db)?;
        let warm_cache_on_open = config.warm_cache_on_open;
        let warm_cache_depth = config.warm_cache_depth;
        let readahead_size = config.readahead_size;
        let shards = config.cache_shards.max(1);
        let block_cache = Arc::new(ShardedBlockCache::new(config.block_cache_capacity, shards));
        let header_cache = Arc::new(ShardedHeaderCache::new(
            config.header_cache_capacity,
            shards,
        ));
        let store = Self {
            inner: Arc::new(BlockStoreInner {
                db,
                read_only: false,
                tip: RwLock::new(tip),
                warm_blocks_loaded: AtomicUsize::new(0),
                compression_level,
                use_compression_dict,
                max_decompressed_block_bytes,
                zstd_dict: RwLock::new(zstd_dict),
                record_cache: Mutex::new(HashMap::new()),
                block_cache,
                cf_blocks_physical_gets: AtomicUsize::new(0),
                cf_blocks_multi_get_batches: AtomicUsize::new(0),
                cf_blocks_stream_physical_gets: AtomicUsize::new(0),
                readahead_size,
                header_cache,
                cf_headers_physical_gets: AtomicUsize::new(0),
                pipeline_batch_size: config.write_pipeline_batch_size.max(1),
                pipeline_flush_ms: config.write_pipeline_flush_ms.max(1),
                pipeline_channel_capacity: config.write_pipeline_channel_capacity.max(1),
                pipeline_write_batches: AtomicUsize::new(0),
                canonical_bin,
                min_retained_height_cached: prune_threshold
                    .unwrap_or_else(|| Arc::new(AtomicU64::new(0))),
                canonical_height_cache: RwLock::new(std::collections::BTreeMap::new()),
                canonical_height_cache_capacity: config.canonical_height_cache_capacity,
                hash_to_height_cache: Arc::new(ShardedLruCache::new(
                    config.hash_to_height_cache_capacity,
                    shards,
                )),
            }),
            pipeline_tx: Arc::new(tokio::sync::Mutex::new(None)),
        };
        // PRN-004: load persisted min_retained_height into AtomicU64
        if let Ok(Some(h)) = store.read_min_retained_height() {
            store.min_retained_height_cached.store(h, Ordering::Release);
        }
        // CAC-006: warm ALL caches after full construction. get_block_by_height and
        // get_record_by_height auto-populate block_cache, header_cache, record_cache,
        // canonical_height_cache (CAC-004), and hash_to_height_cache (CAC-005).
        if warm_cache_on_open {
            let warmed = store.warm_caches(warm_cache_depth);
            store.warm_blocks_loaded.store(warmed, Ordering::Relaxed);
        }
        Ok(store)
    }

    /// Open an existing database read-only; fails if `path` does not exist ([`STR-004`](../docs/requirements/domains/crate_structure/specs/STR-004.md)).
    pub fn open_readonly(path: impl AsRef<Path>) -> Result<Self, BlockStoreError> {
        let path = path.as_ref();
        if !path.exists() {
            return Err(BlockStoreError::Serialization(format!(
                "{ERR_OPEN_READONLY_PATH_MISSING_PREFIX}{}",
                path.display()
            )));
        }
        let opts = Options::default();
        // CF option structs must match how the DB was created; tests use STR-005 `test_config`, which
        // mirrors [`BlockStoreConfig::default`] for `enable_blob_db` ([`TYP-003`](../../docs/requirements/domains/storage_types/specs/TYP-003.md)).
        let readonly_cfg = BlockStoreConfig {
            path: path.to_path_buf(),
            ..BlockStoreConfig::default()
        };
        let compression_level = readonly_cfg.compression_level;
        let use_compression_dict = readonly_cfg.use_compression_dict;
        let max_decompressed_block_bytes = readonly_cfg.max_decompressed_block_bytes;
        let zstd_dictionary_override = readonly_cfg.zstd_dictionary_override.clone();
        let cfs = cf_options::column_family_descriptors(&readonly_cfg, None);
        let db = DB::open_cf_descriptors_read_only(&opts, path, cfs, false)?;
        let db = Arc::new(db);
        let canonical_bin = RwLock::new(CanonicalBin::open_synced(&db, path, false)?);
        let zstd_dict =
            resolve_zstd_dictionary(&db, use_compression_dict, zstd_dictionary_override)?;
        let tip = load_tip(&db)?;
        let readahead_size = readonly_cfg.readahead_size;
        let shards = readonly_cfg.cache_shards.max(1);
        let block_cache = Arc::new(ShardedBlockCache::new(
            readonly_cfg.block_cache_capacity,
            shards,
        ));
        let header_cache = Arc::new(ShardedHeaderCache::new(
            readonly_cfg.header_cache_capacity,
            shards,
        ));
        let store = Self {
            inner: Arc::new(BlockStoreInner {
                db,
                read_only: true,
                tip: RwLock::new(tip),
                warm_blocks_loaded: AtomicUsize::new(0),
                compression_level,
                use_compression_dict,
                max_decompressed_block_bytes,
                zstd_dict: RwLock::new(zstd_dict),
                record_cache: Mutex::new(HashMap::new()),
                block_cache,
                cf_blocks_physical_gets: AtomicUsize::new(0),
                cf_blocks_multi_get_batches: AtomicUsize::new(0),
                cf_blocks_stream_physical_gets: AtomicUsize::new(0),
                readahead_size,
                header_cache,
                cf_headers_physical_gets: AtomicUsize::new(0),
                pipeline_batch_size: readonly_cfg.write_pipeline_batch_size.max(1),
                pipeline_flush_ms: readonly_cfg.write_pipeline_flush_ms.max(1),
                pipeline_channel_capacity: readonly_cfg.write_pipeline_channel_capacity.max(1),
                pipeline_write_batches: AtomicUsize::new(0),
                canonical_bin,
                min_retained_height_cached: Arc::new(AtomicU64::new(0)),
                canonical_height_cache: RwLock::new(std::collections::BTreeMap::new()),
                canonical_height_cache_capacity: readonly_cfg.canonical_height_cache_capacity,
                hash_to_height_cache: Arc::new(ShardedLruCache::new(
                    readonly_cfg.hash_to_height_cache_capacity,
                    shards,
                )),
            }),
            pipeline_tx: Arc::new(tokio::sync::Mutex::new(None)),
        };
        // PRN-004: load persisted min_retained_height
        if let Ok(Some(h)) = store.read_min_retained_height() {
            store.min_retained_height_cached.store(h, Ordering::Release);
        }
        Ok(store)
    }

    /// Initialize genesis: empty store only; atomic [`WriteBatch`] ([`STR-004`](../docs/requirements/domains/crate_structure/specs/STR-004.md)).
    pub fn init_genesis(&self, block: &L2Block) -> Result<(), BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_INIT_GENESIS_READ_ONLY.into(),
            ));
        }
        let meta = self.cf(CF_METADATA)?;
        if self.db.get_cf(meta, META_TIP.as_bytes())?.is_some()
            || self
                .db
                .get_cf(meta, META_GENESIS_HASH.as_bytes())?
                .is_some()
        {
            return Err(BlockStoreError::Serialization(
                ERR_INIT_GENESIS_ALREADY_INITIALIZED.into(),
            ));
        }
        let hash = block.hash();
        if block.height() != 0 {
            return Err(BlockStoreError::Serialization(format!(
                "init_genesis: genesis block height must be 0, got {}",
                block.height()
            )));
        }
        // [`SER-001`](../docs/requirements/domains/serialization/specs/SER-001.md): bincode + zstd (dictionary when configured).
        let compressed = self.serialize_block(block)?;
        // [`SER-002`](../docs/requirements/domains/serialization/specs/SER-002.md): headers are bincode-only in `CF_HEADERS`.
        let header_bytes = Self::serialize_header(&block.header)?;
        let tip = ChainTip { hash, height: 0 };
        let mut batch = WriteBatch::default();
        let cf_b = self.cf(CF_BLOCKS)?;
        let cf_h = self.cf(CF_HEADERS)?;
        let cf_c = self.cf(CF_CANONICAL)?;
        // [`hash_key`] returns `[u8; 32]`; use `.as_slice()` (not `.as_ref()`) so RocksDB keys resolve to
        // `&[u8]` without ambiguous `AsRef` when the `bitcoin` crate is also in the dependency graph.
        batch.put_cf(cf_b, hash_key(&hash).as_slice(), &compressed);
        batch.put_cf(cf_h, hash_key(&hash).as_slice(), &header_bytes);
        batch.put_cf(cf_c, height_key(0), hash_key(&hash).as_slice());
        batch.put_cf(meta, META_TIP.as_bytes(), tip.to_bytes().as_slice());
        batch.put_cf(meta, META_GENESIS_HASH.as_bytes(), hash.as_ref());
        self.db.write(batch)?;
        self.canonical_bin.write().extend_write(0, &hash)?;
        *self.tip.write() = Some(tip);
        let record = BlockRecord::from_header(&block.header, BlockStatus::Validated);
        self.record_cache.lock().insert(hash, record);
        self.block_cache.insert(hash, block.clone());
        self.header_cache.insert(hash, block.header.clone());
        self.maybe_train_dictionary()?;
        Ok(())
    }

    /// **Diagnostics / tests:** Disable the mmap acceleration layer so height→hash resolution uses [`CF_CANONICAL`]
    /// only until the next [`BlockStore::open`] ([`CAN-001`](../docs/requirements/domains/canonical_chain/specs/CAN-001.md) test plan: mmap fallback).
    ///
    /// **Production:** Do not call — the next process restart re-syncs `canonical.bin` from RocksDB anyway.
    pub fn disable_canonical_bin_acceleration(&self) {
        self.canonical_bin.write().disable();
    }

    /// Current chain tip — hash and height of the highest canonical block.
    ///
    /// Returns the in-memory cached copy loaded from [`META_TIP`](crate::META_TIP) at startup
    /// and updated by [`BlockStore::set_tip`], [`BlockStore::init_genesis`], and future
    /// `extend_chain` / `rollback_to_height` APIs.
    ///
    /// # Performance
    ///
    /// This is a hot-path accessor queried on every block ingestion for parent-hash validation.
    /// The [`parking_lot::RwLock::read`] is lock-free on the uncontended fast path (~2-5ns).
    /// No RocksDB I/O occurs.
    ///
    /// # Chia analogy
    ///
    /// Corresponds to `BlockStore.get_peak()` in Chia's `block_store.py`.
    ///
    /// **Requirement:** [`CAN-007`](../docs/requirements/domains/canonical_chain/specs/CAN-007.md).
    pub fn tip(&self) -> Option<ChainTip> {
        *self.tip.read()
    }

    /// Convenience accessor for the current canonical chain height.
    ///
    /// Returns `tip().map(|t| t.height)` — `None` when the store has no tip (before genesis),
    /// `Some(0)` after genesis, `Some(n)` after extending to height `n`.
    ///
    /// **Requirement:** [`CAN-007`](../docs/requirements/domains/canonical_chain/specs/CAN-007.md) § Accessors.
    #[must_use]
    pub fn height(&self) -> Option<u64> {
        self.tip().map(|t| t.height)
    }

    /// Persist a new chain tip to [`CF_METADATA`](crate::CF_METADATA) / [`META_TIP`](crate::META_TIP)
    /// and update the in-memory cache.
    ///
    /// # Encoding
    ///
    /// The value written is exactly 40 bytes: `hash (32 bytes, raw Bytes32) || height (8 bytes, little-endian u64)`.
    /// This matches [`ChainTip::to_bytes()`](crate::ChainTip::to_bytes) and the
    /// [`TYP-006`](../docs/requirements/domains/storage_types/specs/TYP-006.md) wire format.
    ///
    /// # Ordering
    ///
    /// RocksDB write is performed **before** updating the in-memory `RwLock`. If the write
    /// fails, the in-memory tip remains unchanged (no stale state visible to concurrent readers).
    ///
    /// # Errors
    ///
    /// - [`BlockStoreError::Serialization`] with [`ERR_MUTATION_READ_ONLY`](crate::ERR_MUTATION_READ_ONLY)
    ///   when called on a read-only handle.
    /// - [`BlockStoreError::RocksDb`] on write failure.
    ///
    /// # Update points ([`CAN-007`](../docs/requirements/domains/canonical_chain/specs/CAN-007.md))
    ///
    /// | Operation | New Tip |
    /// |-----------|---------|
    /// | `extend_chain` (CAN-005) | Newly added block |
    /// | `rollback_to_height` (ROR-001) | Block at target height |
    /// | `apply_reorg` (ROR-003) | Last block in new chain |
    /// | `init_genesis` (STR-004) | Genesis block (height 0) |
    pub fn set_tip(&self, tip: ChainTip) -> Result<(), BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_MUTATION_READ_ONLY.into(),
            ));
        }
        let cf = self.cf(CF_METADATA)?;
        // Write the 40-byte encoding to CF_METADATA before updating in-memory state.
        // On failure, the in-memory tip remains at the old value (no stale state).
        self.db
            .put_cf(cf, META_TIP.as_bytes(), tip.to_bytes().as_slice())?;
        *self.tip.write() = Some(tip);
        Ok(())
    }

    /// Blocks successfully verified present while warming on last [`Self::open`] ([`STR-004`](../docs/requirements/domains/crate_structure/specs/STR-004.md) / [`CAC-006`](../docs/requirements/domains/caching/specs/CAC-006_cache_warming_on_startup.md)).
    pub fn warm_blocks_loaded_count(&self) -> usize {
        self.warm_blocks_loaded.load(Ordering::Relaxed)
    }

    /// Serialize a block header for [`CF_HEADERS`] ([`SER-002`](../docs/requirements/domains/serialization/specs/SER-002.md)).
    ///
    /// **Write path (normative):** `L2BlockHeader` → [`bincode::serialize`] → raw bytes (no zstd). Headers are
    /// small and read on every chain walk; skipping compression avoids framing overhead and decode latency
    /// on the hot path ([`NORMATIVE.md` § SER-002](../docs/requirements/domains/serialization/NORMATIVE.md)).
    ///
    /// **Errors:** [`BlockStoreError::Serialization`] — same variant as corrupt block payloads so upper
    /// layers can treat “bytes unusable” uniformly until ERR-* adds finer codes.
    ///
    /// **Write path:** [`Self::put_block`] / [`Self::init_genesis`] insert fresh values so steady-state reads hit RAM.
    pub fn get_block(&self, hash: &Bytes32) -> Result<Option<L2Block>, BlockStoreError> {
        if let Some(block) = self.block_cache.get_clone(hash) {
            return Ok(Some(block));
        }
        let cf = self.cf(CF_BLOCKS)?;
        self.cf_blocks_physical_gets.fetch_add(1, Ordering::Relaxed);
        let raw_opt = self.db.get_cf(cf, hash_key(hash).as_slice())?;
        let Some(raw) = raw_opt else {
            return Ok(None);
        };
        let block = self.deserialize_block(&raw)?;
        self.block_cache.insert(*hash, block.clone());
        self.header_cache.insert(*hash, block.header.clone());
        // CAC-005 (API-002): populate hash→height on block read-through
        self.hash_to_height_cache.insert(*hash, block.height());
        Ok(Some(block))
    }

    /// **[`BLK-011`](../docs/requirements/domains/block_storage/specs/BLK-011.md)** — Whether any persisted row exists for `hash` **without** decoding zstd or bincode ([`NORMATIVE.md` § BLK-011](../docs/requirements/domains/block_storage/NORMATIVE.md#blk-011-has-block-has_block)).
    ///
    /// **Cache first (AC §2):** [`Self::block_cache`] and [`Self::header_cache`] are consulted via [`ShardedLruCache::contains`](crate::cache::sharded::ShardedLruCache::contains) ([`LruCache::peek`](lru::LruCache::peek) — no LRU promotion).
    ///
    /// **RocksDB (AC §1):** If both caches miss, probe [`CF_HEADERS`] then [`CF_BLOCKS`] using [`hash_key`](crate::encoding::hash_key) (same key layout as [`Self::put_block`]). The second probe covers edge cases where only a body row exists; normal [`Self::put_block`] writes both families together ([`BLK-001`](../docs/requirements/domains/block_storage/specs/BLK-001.md)).
    ///
    /// **No deserialize / decompress (AC §3):** Uses only [`DB::get_cf`](rocksdb::DB::get_cf) presence checks — returned bytes are discarded without calling [`Self::deserialize_block`] or [`Self::deserialize_header`].
    ///
    /// **Instrumentation:** This path does **not** increment [`Self::cf_blocks_physical_get_count`] (that counter remains exclusive to [`Self::get_block`]) so tests can prove cache-fast paths avoid the heavy read API ([`BLK-002`](../docs/requirements/domains/block_storage/specs/BLK-002.md) counter semantics).
    pub fn has_block(&self, hash: &Bytes32) -> Result<bool, BlockStoreError> {
        if self.block_cache.contains(hash) || self.header_cache.contains(hash) {
            return Ok(true);
        }
        let key = hash_key(hash);
        let cf_h = self.cf(CF_HEADERS)?;
        if self.db.get_cf(cf_h, key.as_slice())?.is_some() {
            return Ok(true);
        }
        let cf_b = self.cf(CF_BLOCKS)?;
        Ok(self.db.get_cf(cf_b, key.as_slice())?.is_some())
    }

    /// **[`BLK-012`](../docs/requirements/domains/block_storage/specs/BLK-012.md)** — Aggregate [`StorageStats`](crate::types::StorageStats) for monitoring / diagnostics ([`TYP-007`](../docs/requirements/domains/storage_types/specs/TYP-007.md), [`NORMATIVE` BLK-012](../docs/requirements/domains/block_storage/NORMATIVE.md#blk-012-storage-statistics-stats)).
    ///
    /// **Row counts:** Each `*_count` field is the number of keys in the corresponding column family from a linear
    /// [`DB::iterator_cf`](rocksdb::DB::iterator_cf) scan ([`CF_BLOCKS`], [`CF_HEADERS`], [`CF_CANONICAL`],
    /// [`CF_CHECKPOINTS`], [`CF_ATTESTED`]). This is **exact** for current store sizes (typical node counts) and
    /// matches NORMATIVE wording (“reflects the total number of entries”). If full scans become too costly at scale,
    /// a future revision may offer `rocksdb.estimate-num-keys` behind configuration with documented error bounds.
    ///
    /// **Tip / pruning:** [`StorageStats::tip_height`] mirrors [`Self::tip`] (RAM snapshot loaded from [`META_TIP`] on open, updated by [`Self::init_genesis`] today). [`Self::put_block`] does not yet advance the tip ([`CAN-007`](../docs/requirements/domains/canonical_chain/specs/CAN-007.md)); operators should not assume `tip_height == max(block heights)` until chain-tip APIs land. [`StorageStats::min_height`] reads
    /// [`META_MIN_HEIGHT`] in [`CF_METADATA`] as **8 bytes little-endian** `u64` ([`storage_types/NORMATIVE`](../docs/requirements/domains/storage_types/NORMATIVE.md)); missing key means no prune watermark yet ([`PRN-004`](../docs/requirements/domains/pruning/specs/PRN-004_min_retained_height_tracking.md)).
    ///
    /// **Disk estimate:** [`StorageStats::total_size_bytes`] sums per-CF RocksDB property `rocksdb.estimate-live-data-size`
    /// (live SST + memtable footprint estimate). It is **not** a byte-exact `du` of the directory; callers should treat
    /// it as an order-of-magnitude health signal; use [`Self::flush`] / [`Self::compact`] before relying on
    /// filesystem-level durability or space reclamation ([`BLK-013`](../docs/requirements/domains/block_storage/specs/BLK-013.md)).
    pub fn stats(&self) -> Result<StorageStats, BlockStoreError> {
        Ok(StorageStats {
            block_count: self.count_cf_entries(CF_BLOCKS)?,
            canonical_block_count: self.count_cf_entries(CF_CANONICAL)?,
            header_count: self.count_cf_entries(CF_HEADERS)?,
            checkpoint_count: self.count_cf_entries(CF_CHECKPOINTS)?,
            attested_count: self.count_cf_entries(CF_ATTESTED)?,
            tip_height: self.tip().map(|t| t.height),
            min_height: self.read_min_retained_height()?,
            total_size_bytes: self.sum_cf_live_data_size_estimates()?,
        })
    }

    /// Exact key count for `cf_name` — used by [`Self::stats`] ([`BLK-012`](../docs/requirements/domains/block_storage/specs/BLK-012.md)).
    fn count_cf_entries(&self, cf_name: &'static str) -> Result<u64, BlockStoreError> {
        let cf = self.cf(cf_name)?;
        let mut n = 0u64;
        for entry in self.db.iterator_cf(cf, IteratorMode::Start) {
            let (_k, _v) = entry?;
            n += 1;
        }
        Ok(n)
    }

    /// Sum RocksDB `rocksdb.estimate-live-data-size` across all user column families ([`BLK-012`](../docs/requirements/domains/block_storage/specs/BLK-012.md) § Field Population).
    fn sum_cf_live_data_size_estimates(&self) -> Result<u64, BlockStoreError> {
        const PROP: &str = "rocksdb.estimate-live-data-size";
        let mut sum = 0u64;
        for name in [
            CF_BLOCKS,
            CF_HEADERS,
            CF_CANONICAL,
            CF_METADATA,
            CF_ATTESTED,
            CF_CHECKPOINTS,
        ] {
            let cf = self.cf(name)?;
            if let Some(v) = self.db.property_int_value_cf(cf, PROP)? {
                sum = sum.saturating_add(v);
            }
        }
        Ok(sum)
    }

    /// Read persisted minimum retained height, if pruning has written [`META_MIN_HEIGHT`].
    fn read_min_retained_height(&self) -> Result<Option<u64>, BlockStoreError> {
        let meta = self.cf(CF_METADATA)?;
        let Some(bytes) = self.db.get_cf(meta, META_MIN_HEIGHT.as_bytes())? else {
            return Ok(None);
        };
        let arr: [u8; 8] = bytes.as_slice().try_into().map_err(|_| {
            BlockStoreError::Serialization(format!(
                "stats: META_MIN_HEIGHT value must be exactly 8 bytes (little-endian u64), got {} bytes",
                bytes.len()
            ))
        })?;
        Ok(Some(u64::from_le_bytes(arr)))
    }

    /// **[`BLK-013`](../docs/requirements/domains/block_storage/specs/BLK-013.md)** — Persist buffered engine state
    /// ([`NORMATIVE` BLK-013](../docs/requirements/domains/block_storage/NORMATIVE.md#blk-013-flush-and-compact)).
    ///
    /// **Semantics:** First `rocksdb::DB::flush_wal(true)` so the write-ahead log is **synced** through the OS to
    /// stable storage, then [`rocksdb::DB::flush`] to flush **all** column-family
    /// memtables to SST files. Together this matches operators’ “make my recent writes durable” intent while staying
    /// close to the BLK-013 spec snippet (which only showed `flush()` — WAL sync is required by NORMATIVE item 1’s
    /// “WAL flush” wording).
    ///
    /// **Logical state:** Does not mutate dig-blockstore caches, tip, or row keys — only RocksDB I/O.
    ///
    /// **Errors:** Any [`rocksdb::Error`] maps to [`BlockStoreError::RocksDb`] ([`ERR-002`](../docs/requirements/domains/error_types/specs/ERR-002_error_from_conversions.md)).
    pub fn flush(&self) -> Result<(), BlockStoreError> {
        self.db.flush_wal(true)?;
        self.db.flush()?;
        Ok(())
    }

    /// **[`BLK-013`](../docs/requirements/domains/block_storage/specs/BLK-013.md)** — Request manual compaction on
    /// **every** column family in [`crate::constants::ALL_COLUMN_FAMILIES`] ([`TYP-001`](../docs/requirements/domains/storage_types/specs/TYP-001.md)).
    ///
    /// **Implementation:** For each family, [`DB::compact_range_cf`](rocksdb::DB::compact_range_cf) with a `None`
    /// key range compacts the **entire** keyspace (RocksDB schedules background work). The rust-rocksdb binding
    /// returns `()` from `compact_range_cf` (errors surface asynchronously); callers use this for **space reclamation**
    /// and read amplification tuning, not as a transactional barrier.
    ///
    /// **Logical state:** Compaction does not delete live keys written by [`Self::put_block`] / [`Self::init_genesis`];
    /// it merges SSTables. Same error mapping as [`Self::flush`] if future APIs gain fallible compaction entry points.
    pub fn compact(&self) -> Result<(), BlockStoreError> {
        for &name in crate::constants::ALL_COLUMN_FAMILIES {
            let cf = self.cf(name)?;
            self.db.compact_range_cf(cf, None::<&[u8]>, None::<&[u8]>);
        }
        Ok(())
    }

    /// Batch-fetch blocks by hash ([`BLK-005`](../docs/requirements/domains/block_storage/specs/BLK-005.md)).
    ///
    /// **Algorithm**
    /// 1. For each input hash in order, clone from [`Self::block_cache`] when present ([`CAC-001`](../docs/requirements/domains/caching/specs/CAC-001_sharded_block_cache.md)).
    /// 2. Collect all cache misses; if non-empty, issue **one** [`rocksdb::DB::multi_get_cf`] over [`CF_BLOCKS`]
    ///    (same `(cf, key)` pattern as [`Self::get_block`], [`SER-001`](../docs/requirements/domains/serialization/specs/SER-001.md) payloads).
    /// 3. For each returned blob: [`Self::deserialize_block`], then insert into [`Self::block_cache`] and [`Self::header_cache`]
    ///    (mirrors single-key read-through in [`Self::get_block`]).
    ///
    /// **Ordering:** Output `Vec` index `i` always corresponds to `hashes[i]` (per NORMATIVE BLK-005 §5).
    ///
    /// **Missing keys:** `Ok(None)` at that index; RocksDB row absent still consumes one slot in the `multi_get` result vector.
    ///
    /// **Empty input:** Returns `Ok(vec![])` without touching RocksDB.
    ///
    /// **Chunking:** Very large batches stay single-call for now ([`BLK-005.md`](../docs/requirements/domains/block_storage/specs/BLK-005.md) implementation notes); future work may split to bound peak memory.
    pub fn get_blocks_by_hash(
        &self,
        hashes: &[Bytes32],
    ) -> Result<Vec<Option<L2Block>>, BlockStoreError> {
        let mut results: Vec<Option<L2Block>> = vec![None; hashes.len()];
        let mut miss_indices: Vec<usize> = Vec::new();
        for (i, hash) in hashes.iter().enumerate() {
            if let Some(block) = self.block_cache.get_clone(hash) {
                results[i] = Some(block);
            } else {
                miss_indices.push(i);
            }
        }
        if miss_indices.is_empty() {
            return Ok(results);
        }
        let cf = self.cf(CF_BLOCKS)?;
        self.cf_blocks_multi_get_batches
            .fetch_add(1, Ordering::Relaxed);
        let keys: Vec<[u8; 32]> = miss_indices
            .iter()
            .map(|&idx| *hash_key(&hashes[idx]))
            .collect();
        let db_results = self
            .db
            .multi_get_cf(keys.iter().map(|k| (cf, k.as_slice())));
        for (j, db_result) in db_results.into_iter().enumerate() {
            let idx = miss_indices[j];
            let maybe_raw = db_result?;
            let Some(raw) = maybe_raw else {
                continue;
            };
            let block = self.deserialize_block(&raw)?;
            self.block_cache.insert(hashes[idx], block.clone());
            self.header_cache.insert(hashes[idx], block.header.clone());
            results[idx] = Some(block);
        }
        Ok(results)
    }

    /// Drop a single entry from the in-memory block LRU — **no RocksDB writes** ([`BLK-002`](../docs/requirements/domains/block_storage/specs/BLK-002.md) test plan: simulate eviction).
    pub fn invalidate_block_cache_entry(&self, hash: &Bytes32) {
        self.block_cache.remove(hash);
    }

    /// Look up the canonical block at `height` ([`CAN-006`](../docs/requirements/domains/canonical_chain/specs/CAN-006.md) precursor, [`BLK-014`](../docs/requirements/domains/block_storage/specs/BLK-014.md) building block).
    ///
    /// **Algorithm:** [`Self::get_hash_by_height`] (mmap then [`CF_CANONICAL`]) → [`Self::get_block`]
    /// ([`BLK-002`](../docs/requirements/domains/block_storage/specs/BLK-002.md) decompress + cache).
    ///
    /// **Returns:** `Ok(None)` when the height index is absent **or** when the hash is indexed but the body row is
    /// missing (same as [`Self::get_block`] returning `None`).
    ///
    /// **Threading:** Safe on any thread; performs synchronous RocksDB + zstd work — use [`Self::get_block_by_height_async`]
    /// from async contexts that must not block the runtime ([`BLK-007`](../docs/requirements/domains/block_storage/specs/BLK-007.md)).
    pub fn get_block_by_height(&self, height: u64) -> Result<Option<L2Block>, BlockStoreError> {
        let Some(hash) = self.get_hash_by_height(height)? else {
            return Ok(None);
        };
        self.get_block(&hash)
    }

    /// **[`BLK-014`](../docs/requirements/domains/block_storage/specs/BLK-014.md)** — Collect canonical [`L2Block`]s for
    /// heights in `[start_height, end_height]` inclusive ([`NORMATIVE` BLK-014](../docs/requirements/domains/block_storage/NORMATIVE.md#blk-014-get-blocks-in-range-get_blocks_in_range)).
    ///
    /// **Semantics:** Ascending height order; `start_height > end_height` ⇒ empty `Vec` (not an error); any height with
    /// no canonical row or no retrievable body is **omitted** (gaps and “beyond tip” behave the same — fewer results).
    ///
    /// **vs [`Self::stream_blocks_in_range`] ([`BLK-006`]):** This API eagerly builds a `Vec` with simple point lookups per height.
    /// [`StreamBlocksInRange`] is better for large scans (single readahead iterator over [`CF_CANONICAL`]).
    pub fn get_blocks_in_range(
        &self,
        start_height: u64,
        end_height: u64,
    ) -> Result<Vec<L2Block>, BlockStoreError> {
        if start_height > end_height {
            return Ok(Vec::new());
        }
        let mut blocks = Vec::with_capacity((end_height - start_height + 1) as usize);
        for height in start_height..=end_height {
            if let Some(block) = self.get_block_by_height(height)? {
                blocks.push(block);
            }
        }
        Ok(blocks)
    }

    /// Look up the canonical [`BlockRecord`] at `height` ([`BLK-015`](../docs/requirements/domains/block_storage/specs/BLK-015.md), [`BLK-004`](../docs/requirements/domains/block_storage/specs/BLK-004.md)).
    ///
    /// **Resolution:** Same [`CF_CANONICAL`] → [`Bytes32`] step as [`Self::get_block_by_height`], then [`Self::get_record`]
    /// so misses load **bincode headers only** from [`CF_HEADERS`] ([`SER-002`](../docs/requirements/domains/serialization/specs/SER-002.md)) — no zstd frame read from [`CF_BLOCKS`].
    ///
    /// **Returns:** `Ok(None)` when the height index is missing or when neither [`CF_HEADERS`] nor caches can supply a header.
    ///
    /// **Canonical resolution:** Same [`Self::get_hash_by_height`] dual layer as [`Self::get_block_by_height`] ([`CAN-001`](../docs/requirements/domains/canonical_chain/specs/CAN-001.md)).
    pub fn get_record_by_height(
        &self,
        height: u64,
    ) -> Result<Option<BlockRecord>, BlockStoreError> {
        let Some(hash) = self.get_hash_by_height(height)? else {
            return Ok(None);
        };
        self.get_record(&hash)
    }

    /// Look up the canonical header at `height` ([`CAN-006`](../docs/requirements/domains/canonical_chain/specs/CAN-006.md)).
    ///
    /// **Algorithm:** [`Self::get_hash_by_height`] → [`Self::get_header`]
    /// ([`BLK-003`](../docs/requirements/domains/block_storage/specs/BLK-003.md) cache + bincode).
    ///
    /// **Returns:** `Ok(None)` when the height is not canonical or the header row is absent.
    ///
    /// **Lighter than `get_block_by_height`:** Headers are uncompressed bincode (~700 bytes)
    /// versus full block bodies (zstd decompression + larger payload). Use this when only
    /// header fields are needed (e.g., parent-hash walks, timestamp checks).
    pub fn get_header_by_height(
        &self,
        height: u64,
    ) -> Result<Option<L2BlockHeader>, BlockStoreError> {
        let Some(hash) = self.get_hash_by_height(height)? else {
            return Ok(None);
        };
        self.get_header(&hash)
    }

    /// Collect canonical block hashes for all heights in the given epoch.
    ///
    /// **Algorithm:** Uses [`dig_epoch::first_height_in_epoch`] and
    /// [`dig_epoch::epoch_checkpoint_height`] to derive the inclusive `[start, end]`
    /// height range, then calls [`Self::get_hash_by_height`] for each height. Stops
    /// early when a height returns `None` (chain hasn't reached that height yet).
    ///
    /// **Returns:** A `Vec<Bytes32>` containing one hash per canonical height in the epoch,
    /// in ascending height order. May be shorter than `BLOCKS_PER_EPOCH` if the chain is
    /// still growing into the epoch, or empty if the epoch is entirely beyond the chain tip.
    ///
    /// **Requirement:** [`CAN-006`](../docs/requirements/domains/canonical_chain/specs/CAN-006.md) § Epoch Block Hashes.
    pub fn get_epoch_block_hashes(&self, epoch: u64) -> Result<Vec<Bytes32>, BlockStoreError> {
        let start = dig_epoch::first_height_in_epoch(epoch);
        let end = dig_epoch::epoch_checkpoint_height(epoch);
        let mut hashes = Vec::new();
        for height in start..=end {
            if let Some(hash) = self.get_hash_by_height(height)? {
                hashes.push(hash);
            } else {
                break; // chain hasn't reached this height yet
            }
        }
        Ok(hashes)
    }

    /// **[`BLK-015`](../docs/requirements/domains/block_storage/specs/BLK-015.md)** — Materialize canonical [`BlockRecord`]s for `[start_height, end_height]` inclusive ([`NORMATIVE` BLK-015](../docs/requirements/domains/block_storage/NORMATIVE.md#blk-015-get-records-in-range-get_records_in_range)).
    ///
    /// **Semantics:** Matches [`Self::get_blocks_in_range`] ordering and gap rules ([`BLK-014`](../docs/requirements/domains/block_storage/specs/BLK-014.md)), but each row comes from [`Self::get_record_by_height`] so operators avoid zstd decompression on the hot path ([`BLK-015`](../docs/requirements/domains/block_storage/specs/BLK-015.md) § Specification).
    ///
    /// **Cache interaction:** [`Self::get_record`] may insert derived rows into [`Self::record_cache`] / [`Self::header_cache`];
    /// repeated scans therefore become cheaper, mirroring single-hash lookups ([`CAC-003`](../docs/requirements/domains/caching/specs/CAC-003.md) precursor).
    pub fn get_records_in_range(
        &self,
        start_height: u64,
        end_height: u64,
    ) -> Result<Vec<BlockRecord>, BlockStoreError> {
        if start_height > end_height {
            return Ok(Vec::new());
        }
        let mut records = Vec::with_capacity((end_height - start_height + 1) as usize);
        for height in start_height..=end_height {
            if let Some(record) = self.get_record_by_height(height)? {
                records.push(record);
            }
        }
        Ok(records)
    }

    /// How many times [`Self::get_block`] reached RocksDB [`CF_BLOCKS`] after a cache miss (includes `Ok(None)` probes).
    ///
    /// **Tests / ops:** [`tests/blk_002_tests.rs`] asserts hits add zero; misses increment exactly once per call.
    pub fn cf_blocks_physical_get_count(&self) -> u64 {
        self.cf_blocks_physical_gets.load(Ordering::Relaxed) as u64
    }

    /// How many times [`Self::get_blocks_by_hash`] invoked [`rocksdb::DB::multi_get_cf`] because at least one hash missed
    /// [`Self::block_cache`] ([`BLK-005`](../docs/requirements/domains/block_storage/specs/BLK-005.md); see [`tests/blk_005_tests.rs`]).
    #[inline]
    pub fn cf_blocks_multi_get_batch_count(&self) -> u64 {
        self.cf_blocks_multi_get_batches.load(Ordering::Relaxed) as u64
    }

    /// RocksDB readahead hint (bytes) copied from [`BlockStoreConfig::readahead_size`](crate::BlockStoreConfig::readahead_size) at open ([`BLK-006`](../docs/requirements/domains/block_storage/specs/BLK-006.md) AC §4).
    #[must_use]
    pub fn readahead_size(&self) -> usize {
        self.readahead_size
    }

    /// How many times [`StreamBlocksInRange`] issued [`DB::get_cf_opt`](rocksdb::DB::get_cf_opt) against [`CF_BLOCKS`]
    /// after a block-cache miss while streaming ([`BLK-006`](../docs/requirements/domains/block_storage/specs/BLK-006.md); [`tests/blk_006_tests.rs`]).
    #[must_use]
    pub fn cf_blocks_stream_physical_get_count(&self) -> u64 {
        self.cf_blocks_stream_physical_gets.load(Ordering::Relaxed) as u64
    }

    /// Retrieve a block header by hash ([`BLK-003`](../docs/requirements/domains/block_storage/specs/BLK-003.md)).
    ///
    /// **Order:** [`Self::header_cache`] → on miss, `get_cf` [`CF_HEADERS`] → [`Self::deserialize_header`] (**no zstd**;
    /// [`SER-002`](../docs/requirements/domains/serialization/specs/SER-002.md)).
    ///
    /// **Write path:** [`Self::put_block`] / [`Self::init_genesis`] insert headers in parallel with block bodies.
    pub fn get_header(&self, hash: &Bytes32) -> Result<Option<L2BlockHeader>, BlockStoreError> {
        if let Some(header) = self.header_cache.get_clone(hash) {
            return Ok(Some(header));
        }
        let cf = self.cf(CF_HEADERS)?;
        self.cf_headers_physical_gets
            .fetch_add(1, Ordering::Relaxed);
        let raw_opt = self.db.get_cf(cf, hash_key(hash).as_slice())?;
        let Some(raw) = raw_opt else {
            return Ok(None);
        };
        let header = Self::deserialize_header(&raw)?;
        self.header_cache.insert(*hash, header.clone());
        // CAC-005 (API-002): populate hash→height on header read-through
        self.hash_to_height_cache.insert(*hash, header.height);
        Ok(Some(header))
    }

    /// Drop one header from the in-memory LRU ([`BLK-003`](../docs/requirements/domains/block_storage/specs/BLK-003.md) tests / future invalidation).
    pub fn invalidate_header_cache_entry(&self, hash: &Bytes32) {
        self.header_cache.remove(hash);
    }

    /// Count of RocksDB [`CF_HEADERS`] `get_cf` calls from [`Self::get_header`] or [`Self::get_record`]
    /// when the in-memory header **and** record caches do not already supply the header/record ([`BLK-003`](../docs/requirements/domains/block_storage/specs/BLK-003.md), [`BLK-004`](../docs/requirements/domains/block_storage/specs/BLK-004.md)).
    ///
    /// **Note:** [`Self::get_record`] consults [`Self::header_cache`] before touching RocksDB, so a record-cache
    /// miss with a warm header cache does **not** increment this counter (still satisfies “derive from header”).
    pub fn cf_headers_physical_get_count(&self) -> u64 {
        self.cf_headers_physical_gets.load(Ordering::Relaxed) as u64
    }

    /// **[`BLK-001`](../docs/requirements/domains/block_storage/specs/BLK-001.md)** — Primary name in
    /// [`IMPLEMENTATION_ORDER.md`](../docs/requirements/IMPLEMENTATION_ORDER.md) Phase 5.
    ///
    /// **Pipeline:** zstd payload → [`CF_BLOCKS`], bincode header → [`CF_HEADERS`], optional height index →
    /// [`CF_CANONICAL`]; [`BlockRecord`] is derived with [`BlockStatus::Validated`] and stored only in
    /// [`Self::record_cache`] ([`TYP-004`](../docs/requirements/domains/storage_types/specs/TYP-004.md) persistence rule).
    ///
    /// **Idempotency:** If the block hash already exists in `CF_BLOCKS`, returns `Ok(false)` and performs no writes
    /// ([`start.md`](../docs/prompt/start.md) hard requirement §9).
    ///
    /// **[`SER-005`](../docs/requirements/domains/serialization/specs/SER-005.md):** A successful insert that makes
    /// [`Self::block_count`] reach [`DICT_TRAINING_THRESHOLD`] triggers **one-time** dictionary training when
    /// [`BlockStoreConfig::use_compression_dict`](crate::BlockStoreConfig) is `true`.
    pub fn put_block(&self, block: &L2Block, canonical: bool) -> Result<bool, BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_MUTATION_READ_ONLY.into(),
            ));
        }
        let hash = block.hash();
        let cf_b = self.cf(CF_BLOCKS)?;
        if self.db.get_cf(cf_b, hash_key(&hash).as_slice())?.is_some() {
            return Ok(false);
        }
        let compressed = self.serialize_block(block)?;
        let header_bytes = Self::serialize_header(&block.header)?;
        let mut batch = WriteBatch::default();
        let cf_h = self.cf(CF_HEADERS)?;
        batch.put_cf(cf_b, hash_key(&hash).as_slice(), &compressed);
        batch.put_cf(cf_h, hash_key(&hash).as_slice(), &header_bytes);
        if canonical {
            let cf_c = self.cf(CF_CANONICAL)?;
            batch.put_cf(cf_c, height_key(block.height()), hash_key(&hash).as_slice());
        }
        self.db.write(batch)?;
        if canonical {
            self.canonical_bin
                .write()
                .extend_write(block.height(), &hash)?;
            // CAC-004: populate height→hash cache for canonical blocks
            self.insert_canonical_height_cache(block.height(), hash);
        }
        let record = BlockRecord::from_header(&block.header, BlockStatus::Validated);
        self.record_cache.lock().insert(hash, record);
        self.block_cache.insert(hash, block.clone());
        self.header_cache.insert(hash, block.header.clone());
        // CAC-005: populate hash→height cache for all blocks (canonical or not)
        self.hash_to_height_cache.insert(hash, block.height());
        self.maybe_train_dictionary()?;
        Ok(true)
    }

    /// Alias for [`Self::put_block`] — matches the BLK-001 normative snippet name `put` ([`NORMATIVE.md` § BLK-001](../docs/requirements/domains/block_storage/NORMATIVE.md)).
    #[inline]
    pub fn put(&self, block: &L2Block, canonical: bool) -> Result<bool, BlockStoreError> {
        self.put_block(block, canonical)
    }

    // -----------------------------------------------------------------------
    // Rollback & Reorg (ROR domain)
    // -----------------------------------------------------------------------

    /// Public accessor for the pruning floor: minimum retained block height.
    ///
    /// Returns `0` when no pruning has occurred (all heights retained).
    /// After [`PRN-001`] runs, this reflects the `META_MIN_HEIGHT` value in `CF_METADATA`.
    ///
    /// **Requirement:** [`ROR-005`](../docs/requirements/domains/rollback_reorg/specs/ROR-005.md).
    /// Public accessor for the pruning floor: minimum retained block height.
    ///
    /// Returns `0` when no pruning has occurred (all heights retained).
    /// Uses the cached [`AtomicU64`] for fast lock-free access ([`PRN-004`]).
    pub fn min_retained_height(&self) -> Result<u64, BlockStoreError> {
        Ok(self.min_retained_height_cached.load(Ordering::Acquire))
    }

    /// Read-only preview: which canonical blocks would be reverted by a rollback to `target_height`.
    ///
    /// Returns hashes in **descending** height order (tip first). Returns empty `Vec` when
    /// no tip is set or `target_height >= tip.height`. Does NOT modify any state.
    ///
    /// **Requirement:** [`ROR-006`](../docs/requirements/domains/rollback_reorg/specs/ROR-006.md).
    pub fn blocks_to_revert(&self, target_height: u64) -> Result<Vec<Bytes32>, BlockStoreError> {
        let Some(current_tip) = self.tip() else {
            return Ok(Vec::new());
        };
        if target_height >= current_tip.height {
            return Ok(Vec::new());
        }
        let mut reverted = Vec::new();
        for h in (target_height + 1..=current_tip.height).rev() {
            if let Some(hash) = self.get_hash_by_height(h)? {
                reverted.push(hash);
            }
        }
        Ok(reverted)
    }

    /// Revert the canonical chain to `target_height`, removing higher heights from
    /// the canonical index and updating the tip.
    ///
    /// # Validation ([`ROR-005`](../docs/requirements/domains/rollback_reorg/specs/ROR-005.md))
    ///
    /// Checked in order before any mutation:
    /// 1. **NoTip** — no chain tip set.
    /// 2. **RollbackAboveTip** — `target_height > tip.height`.
    /// 3. **RollbackBelowMin** — `target_height < min_retained_height()`.
    ///
    /// # Mutation ([`ROR-001`](../docs/requirements/domains/rollback_reorg/specs/ROR-001.md))
    ///
    /// 1. Collect reverted hashes from tip down to `target_height + 1`.
    /// 2. `WriteBatch` deletes on CF_CANONICAL for each reverted height.
    /// 3. Truncate `canonical.bin` to `(target_height + 1) * 32`.
    /// 4. Update tip to the block at `target_height`.
    /// 5. Mark reverted blocks as non-canonical in record_cache.
    /// 6. Evict reverted heights from canonical_height_cache.
    ///
    /// # Returns
    ///
    /// Hashes of reverted blocks in **descending** height order (tip first).
    /// Returns empty Vec for a no-op rollback at the current tip.
    /// Block data in CF_BLOCKS is NOT deleted (fork preservation per ROR-004).
    pub fn rollback_to_height(&self, target_height: u64) -> Result<Vec<Bytes32>, BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_MUTATION_READ_ONLY.into(),
            ));
        }
        // ROR-005: boundary validation (in order: NoTip, AboveTip, BelowMin)
        let current_tip = self.tip().ok_or(BlockStoreError::NoTip)?;
        if target_height > current_tip.height {
            return Err(BlockStoreError::RollbackAboveTip {
                target: target_height,
                tip: current_tip.height,
            });
        }
        let min_height = self.min_retained_height()?;
        if target_height < min_height {
            return Err(BlockStoreError::RollbackBelowMin {
                target: target_height,
                min: min_height,
            });
        }
        // No-op: rollback at current tip
        if target_height == current_tip.height {
            return Ok(Vec::new());
        }

        // Collect hashes to revert (descending order)
        let mut reverted = Vec::new();
        for h in (target_height + 1..=current_tip.height).rev() {
            if let Some(hash) = self.get_hash_by_height(h)? {
                reverted.push(hash);
            }
        }

        // WriteBatch: delete CF_CANONICAL entries for reverted heights
        let cf_c = self.cf(CF_CANONICAL)?;
        let mut batch = WriteBatch::default();
        for h in target_height + 1..=current_tip.height {
            batch.delete_cf(cf_c, height_key(h));
        }
        self.db.write(batch)?;

        // Truncate canonical.bin (mmap) to (target_height + 1) * 32
        self.canonical_bin
            .write()
            .truncate_to_height(target_height)?;

        // Update tip to block at target_height
        if let Some(target_hash) = self.get_hash_by_height(target_height)? {
            self.set_tip(ChainTip {
                hash: target_hash,
                height: target_height,
            })?;
        }

        // Mark reverted blocks as non-canonical in record_cache
        {
            let mut cache = self.record_cache.lock();
            for hash in &reverted {
                if let Some(r) = cache.get_mut(hash) {
                    r.in_canonical_chain = false;
                }
            }
        }

        // Evict reverted heights from canonical_height_cache (CAC-004)
        {
            let mut hcache = self.canonical_height_cache.write();
            for h in target_height + 1..=current_tip.height {
                hcache.remove(&h);
            }
        }

        Ok(reverted)
    }

    /// Atomically rollback the canonical chain to `ancestor_height` and re-canonicalize
    /// the blocks in `new_chain_hashes`.
    ///
    /// # Algorithm ([`ROR-003`](../docs/requirements/domains/rollback_reorg/specs/ROR-003.md))
    ///
    /// 1. **Validate:** NoTip → error. EmptyReorgChain → error. Each hash in `new_chain_hashes`
    ///    must be in the store (BlockNotInStore if not).
    /// 2. **WriteBatch (atomic):**
    ///    - Delete CF_CANONICAL entries for heights `ancestor_height + 1` .. `current_tip.height`.
    ///    - Put new canonical entries for each hash in `new_chain_hashes` (height from record).
    ///    - Write new tip (last hash in `new_chain_hashes`) to META_TIP.
    /// 3. **Post-commit:**
    ///    - Truncate `canonical.bin` to `ancestor_height`, then write new hashes.
    ///    - Update record_cache: reverted → `in_canonical_chain=false`, applied → `true`.
    ///    - Update in-memory tip.
    ///    - Evict/update canonical_height_cache.
    ///
    /// # Returns
    ///
    /// [`ReorgResult`] with `reverted` (descending), `applied` (ascending), and `new_tip`.
    pub fn apply_reorg(
        &self,
        ancestor_height: u64,
        new_chain_hashes: &[Bytes32],
    ) -> Result<ReorgResult, BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_MUTATION_READ_ONLY.into(),
            ));
        }
        let current_tip = self.tip().ok_or(BlockStoreError::NoTip)?;
        if new_chain_hashes.is_empty() {
            return Err(BlockStoreError::EmptyReorgChain);
        }

        // Validate all new chain hashes exist and collect their records
        let mut new_records: Vec<(Bytes32, BlockRecord)> =
            Vec::with_capacity(new_chain_hashes.len());
        for hash in new_chain_hashes {
            let record = self
                .get_record(hash)?
                .ok_or(BlockStoreError::BlockNotInStore(*hash))?;
            new_records.push((*hash, record));
        }

        let cf_c = self.cf(CF_CANONICAL)?;
        let cf_meta = self.cf(CF_METADATA)?;
        let mut batch = WriteBatch::default();

        // Phase 1: Rollback — delete old canonical entries above ancestor_height
        let mut reverted = Vec::new();
        for h in (ancestor_height + 1..=current_tip.height).rev() {
            if let Some(hash) = self.get_hash_by_height(h)? {
                reverted.push(hash);
            }
            batch.delete_cf(cf_c, height_key(h));
        }

        // Phase 2: Apply new chain
        for (hash, record) in &new_records {
            batch.put_cf(cf_c, height_key(record.height), hash_key(hash).as_slice());
        }

        // Phase 3: Write new tip in the same batch
        let new_tip_hash = new_chain_hashes
            .last()
            .copied()
            .expect("non-empty checked above");
        let new_tip_height = new_records.last().expect("non-empty").1.height;
        let new_tip = ChainTip {
            hash: new_tip_hash,
            height: new_tip_height,
        };
        batch.put_cf(cf_meta, META_TIP.as_bytes(), new_tip.to_bytes().as_slice());

        // Atomic commit
        self.db.write(batch)?;

        // Post-commit: update mmap
        self.canonical_bin
            .write()
            .truncate_to_height(ancestor_height)?;
        for (hash, record) in &new_records {
            self.canonical_bin
                .write()
                .extend_write(record.height, hash)?;
        }

        // Post-commit: update record cache
        {
            let mut cache = self.record_cache.lock();
            for hash in &reverted {
                if let Some(r) = cache.get_mut(hash) {
                    r.in_canonical_chain = false;
                }
            }
            for (hash, _) in &new_records {
                if let Some(r) = cache.get_mut(hash) {
                    r.in_canonical_chain = true;
                }
            }
        }

        // Post-commit: update canonical_height_cache
        {
            let mut hcache = self.canonical_height_cache.write();
            for h in ancestor_height + 1..=current_tip.height {
                hcache.remove(&h);
            }
            for (hash, record) in &new_records {
                hcache.insert(record.height, *hash);
            }
        }

        // Post-commit: update in-memory tip
        *self.tip.write() = Some(new_tip);

        Ok(ReorgResult {
            reverted,
            applied: new_chain_hashes.to_vec(),
            new_tip,
        })
    }

    /// Walk the `parent_hash` chain from `hash` backward, returning the first block
    /// that is the canonical block at its height.
    ///
    /// # Algorithm ([`ROR-002`](../docs/requirements/domains/rollback_reorg/specs/ROR-002.md))
    ///
    /// For up to `max_depth` steps:
    /// 1. Load the [`BlockRecord`] for `current_hash` (cache or CF_HEADERS derive).
    /// 2. Check if `get_hash_by_height(record.height) == current_hash` — if so, this
    ///    block is canonical and is the common ancestor.
    /// 3. Otherwise, follow `record.parent_hash` and repeat.
    ///
    /// # Returns
    ///
    /// - `Ok(Some((hash, height)))` — the first canonical ancestor found.
    /// - `Ok(None)` — hash not in store, parent chain broken, or `max_depth` exceeded.
    ///
    /// # Use case
    ///
    /// When a new block arrives whose parent is not the current tip, call this with the
    /// new block’s parent hash to find where the fork diverged from the canonical chain.
    /// The result feeds into [`apply_reorg`](Self) (ROR-003) as the `ancestor_height`.
    ///
    /// # Read-only
    ///
    /// This method does not modify any state. Safe to call concurrently.
    pub fn find_common_ancestor(
        &self,
        hash: &Bytes32,
        max_depth: u64,
    ) -> Result<Option<(Bytes32, u64)>, BlockStoreError> {
        let mut current_hash = *hash;
        for _ in 0..max_depth {
            // CAC-005 (API-002): try hash_to_height_cache first to avoid header deserialization
            // when we only need the height for the canonical check.
            let record = match self.get_record(&current_hash)? {
                Some(r) => r,
                None => return Ok(None), // block not in store or chain broken
            };
            let height = record.height;
            // Populate hash→height cache on access (read-through for future lookups)
            self.hash_to_height_cache.insert(current_hash, height);
            // Check if this block is canonical at its height
            if let Some(canonical_hash) = self.get_hash_by_height(height)? {
                if canonical_hash == current_hash {
                    return Ok(Some((current_hash, height)));
                }
            }
            // Walk backwards via parent_hash
            current_hash = record.parent_hash;
        }
        Ok(None) // exceeded max_depth
    }

    /// **[`BLK-009`](../docs/requirements/domains/block_storage/specs/BLK-009.md)** — Persist an [`AttestedBlock`] under the block’s hash key.
    ///
    /// **Key:** [`hash_key`](crate::encoding::hash_key)(`hash`) — raw 32 bytes in [`CF_ATTESTED`] ([`KEY-001`](../docs/requirements/domains/key_encoding/specs/KEY-001_hash_keys.md)), identical key shape to [`CF_BLOCKS`] / [`CF_HEADERS`].
    ///
    /// **Value:** [`bincode::serialize`] of `attested` (uncompressed; attestations are small per BLK-009 implementation notes).
    ///
    /// **Hash vs payload:** Callers normally pass `hash == attested.hash()`; this method does **not** verify that invariant so
    /// tests and migration tooling can stage rows independently of body presence in [`CF_BLOCKS`].
    ///
    /// **Overwrite (AC §4):** A second call with the same `hash` replaces the previous value (`DB::put_cf`).
    ///
    /// **Read-only:** Returns [`BlockStoreError::Serialization`] with [`ERR_MUTATION_READ_ONLY`] — same contract as [`Self::put_block`].
    pub fn put_attestation(
        &self,
        hash: &Bytes32,
        attested: &AttestedBlock,
    ) -> Result<(), BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_MUTATION_READ_ONLY.into(),
            ));
        }
        let bytes = bincode::serialize(attested)
            .map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
        let cf = self.cf(CF_ATTESTED)?;
        self.db.put_cf(cf, hash_key(hash).as_slice(), &bytes)?;
        Ok(())
    }

    /// **[`BLK-009`](../docs/requirements/domains/block_storage/specs/BLK-009.md)** — Read [`AttestedBlock`] bytes from [`CF_ATTESTED`].
    ///
    /// **Miss:** [`Ok(None)]` when no row exists (AC §3).
    ///
    /// **No attestation cache (yet):** Each call performs a RocksDB `get_cf` + bincode decode (BLK-009 notes; hot paths may add [`CAC-*`] later).
    ///
    /// **Corrupt rows:** Malformed bincode surfaces as [`BlockStoreError::Serialization`] so operators can distinguish “missing” vs “bad bytes”.
    pub fn get_attestation(
        &self,
        hash: &Bytes32,
    ) -> Result<Option<AttestedBlock>, BlockStoreError> {
        let cf = self.cf(CF_ATTESTED)?;
        let raw = match self.db.get_cf(cf, hash_key(hash).as_slice())? {
            Some(b) => b,
            None => return Ok(None),
        };
        let attested: AttestedBlock = bincode::deserialize(&raw).map_err(|e| {
            BlockStoreError::Serialization(format!(
                "get_attestation: bincode deserialize failed: {e}"
            ))
        })?;
        Ok(Some(attested))
    }

    // -----------------------------------------------------------------------
    // Checkpoint Storage (CKP domain)
    // -----------------------------------------------------------------------

    /// Persist a [`StoredCheckpoint`] to [`CF_CHECKPOINTS`] keyed by epoch.
    ///
    /// **Key:** [`epoch_key`](crate::encoding::epoch_key)(`checkpoint.checkpoint.epoch`) — 8-byte big-endian.
    /// **Value:** [`bincode::serialize`] of the full [`StoredCheckpoint`].
    /// **Idempotent:** overwrites any existing checkpoint at the same epoch.
    ///
    /// **Requirement:** [`CKP-001`](../docs/requirements/domains/checkpoint_storage/specs/CKP-001_put_checkpoint.md).
    pub fn put_checkpoint(
        &self,
        checkpoint: &crate::StoredCheckpoint,
    ) -> Result<(), BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_MUTATION_READ_ONLY.into(),
            ));
        }
        let epoch = checkpoint.checkpoint.epoch;
        let key = crate::encoding::epoch_key(epoch);
        let value = checkpoint
            .encode_bincode()
            .map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
        let cf = self.cf(CF_CHECKPOINTS)?;
        self.db.put_cf(cf, key.as_slice(), &value)?;
        Ok(())
    }

    /// Retrieve a [`StoredCheckpoint`] by epoch from [`CF_CHECKPOINTS`].
    ///
    /// Returns `Ok(None)` if no checkpoint exists for the given epoch.
    ///
    /// **Requirement:** [`CKP-002`](../docs/requirements/domains/checkpoint_storage/specs/CKP-002_get_checkpoint.md).
    pub fn get_checkpoint(
        &self,
        epoch: u64,
    ) -> Result<Option<crate::StoredCheckpoint>, BlockStoreError> {
        let cf = self.cf(CF_CHECKPOINTS)?;
        let key = crate::encoding::epoch_key(epoch);
        let Some(bytes) = self.db.get_cf(cf, key.as_slice())? else {
            return Ok(None);
        };
        let checkpoint = crate::StoredCheckpoint::decode_bincode(&bytes)
            .map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
        Ok(Some(checkpoint))
    }

    /// Retrieve the most recent checkpoint (highest epoch) via reverse iterator.
    ///
    /// Returns `Ok(None)` if no checkpoints are stored.
    ///
    /// **Requirement:** [`CKP-003`](../docs/requirements/domains/checkpoint_storage/specs/CKP-003_get_latest_checkpoint.md).
    pub fn get_latest_checkpoint(
        &self,
    ) -> Result<Option<crate::StoredCheckpoint>, BlockStoreError> {
        let cf = self.cf(CF_CHECKPOINTS)?;
        let mut iter = self.db.iterator_cf(cf, IteratorMode::End);
        let Some(item) = iter.next() else {
            return Ok(None);
        };
        let (_key, value) = item?;
        let checkpoint = crate::StoredCheckpoint::decode_bincode(&value)
            .map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
        Ok(Some(checkpoint))
    }

    /// Retrieve all checkpoints within an epoch range `[start_epoch, end_epoch]` inclusive.
    ///
    /// Returns empty `Vec` if no checkpoints exist in the range. If `start_epoch > end_epoch`,
    /// returns empty (no error).
    ///
    /// **Requirement:** [`CKP-004`](../docs/requirements/domains/checkpoint_storage/specs/CKP-004_get_checkpoints_in_range.md).
    pub fn get_checkpoints_in_range(
        &self,
        start_epoch: u64,
        end_epoch: u64,
    ) -> Result<Vec<crate::StoredCheckpoint>, BlockStoreError> {
        if start_epoch > end_epoch {
            return Ok(Vec::new());
        }
        let cf = self.cf(CF_CHECKPOINTS)?;
        let start_key = crate::encoding::epoch_key(start_epoch);
        let mode = IteratorMode::From(&start_key, Direction::Forward);
        let iter = self.db.iterator_cf(cf, mode);
        let mut result = Vec::new();
        for item in iter {
            let (key_bytes, value) = item?;
            if key_bytes.len() != 8 {
                continue;
            }
            let key_arr: [u8; 8] = key_bytes.as_ref().try_into().unwrap_or([0; 8]);
            let epoch = crate::encoding::decode_epoch_key(&key_arr);
            if epoch > end_epoch {
                break;
            }
            let checkpoint = crate::StoredCheckpoint::decode_bincode(&value)
                .map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
            result.push(checkpoint);
        }
        Ok(result)
    }

    // -----------------------------------------------------------------------
    // Pruning (PRN domain)
    // -----------------------------------------------------------------------

    /// Remove all blocks, headers, attestations, and canonical entries below `height`.
    ///
    /// # Algorithm ([`PRN-001`](../docs/requirements/domains/pruning/specs/PRN-001_prune_before_height.md))
    ///
    /// 1. Iterate `CF_CANONICAL` from `min_retained_height` to `height - 1`, collecting hashes.
    /// 2. Also scan `CF_HEADERS` for non-canonical blocks below `height` ([`PRN-005`]).
    /// 3. Single `WriteBatch` deletes from CF_BLOCKS, CF_HEADERS, CF_ATTESTED, CF_CANONICAL.
    /// 4. Update `META_MIN_HEIGHT` in the same batch.
    /// 5. Post-commit: evict from all caches, update `AtomicU64`.
    ///
    /// # Returns
    ///
    /// Count of blocks pruned (canonical + non-canonical).
    pub fn prune_before_height(&self, height: u64) -> Result<usize, BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_MUTATION_READ_ONLY.into(),
            ));
        }
        let current_min = self.min_retained_height()?;
        if height <= current_min {
            return Ok(0);
        }

        let cf_b = self.cf(CF_BLOCKS)?;
        let cf_h = self.cf(CF_HEADERS)?;
        let cf_a = self.cf(CF_ATTESTED)?;
        let cf_c = self.cf(CF_CANONICAL)?;
        let cf_meta = self.cf(CF_METADATA)?;

        let mut batch = WriteBatch::default();
        let mut pruned_hashes: Vec<Bytes32> = Vec::new();

        // Phase 1: Canonical blocks below target height
        for h in current_min..height {
            if let Some(hash) = self.get_hash_by_height(h)? {
                batch.delete_cf(cf_b, hash_key(&hash).as_slice());
                batch.delete_cf(cf_h, hash_key(&hash).as_slice());
                batch.delete_cf(cf_a, hash_key(&hash).as_slice());
                batch.delete_cf(cf_c, height_key(h));
                pruned_hashes.push(hash);
            }
        }

        // Phase 2 (PRN-005): Non-canonical blocks — scan CF_HEADERS for blocks below height
        // that were NOT already collected in the canonical pass.
        let canonical_set: std::collections::HashSet<Bytes32> =
            pruned_hashes.iter().copied().collect();
        let header_iter = self.db.iterator_cf(cf_h, IteratorMode::Start);
        for item in header_iter {
            let (key_bytes, value_bytes) = item?;
            if key_bytes.len() != 32 {
                continue;
            }
            let arr: [u8; 32] = key_bytes.as_ref().try_into().unwrap_or([0; 32]);
            let hash = Bytes32::new(arr);
            if canonical_set.contains(&hash) {
                continue; // already handled in canonical pass
            }
            // Deserialize header to check height
            if let Ok(header) = Self::deserialize_header(&value_bytes) {
                if header.height < height {
                    batch.delete_cf(cf_b, hash_key(&hash).as_slice());
                    batch.delete_cf(cf_h, hash_key(&hash).as_slice());
                    batch.delete_cf(cf_a, hash_key(&hash).as_slice());
                    pruned_hashes.push(hash);
                }
            }
        }

        // Update META_MIN_HEIGHT in the same batch
        batch.put_cf(cf_meta, META_MIN_HEIGHT.as_bytes(), height.to_le_bytes());

        let count = pruned_hashes.len();
        self.db.write(batch)?;

        // Post-commit: update AtomicU64
        self.min_retained_height_cached
            .store(height, Ordering::Release);

        // Post-commit: evict from all caches
        for hash in &pruned_hashes {
            self.block_cache.remove(hash);
            self.header_cache.remove(hash);
            self.record_cache.lock().remove(hash);
            self.hash_to_height_cache.remove(hash);
        }
        // Evict canonical height cache entries below height
        {
            let mut hcache = self.canonical_height_cache.write();
            let to_remove: Vec<u64> = hcache.range(..height).map(|(&h, _)| h).collect();
            for h in to_remove {
                hcache.remove(&h);
            }
        }

        Ok(count)
    }

    /// Remove all checkpoints with epoch < `epoch` from CF_CHECKPOINTS.
    ///
    /// Returns the count of pruned checkpoints.
    ///
    /// **Requirement:** [`PRN-002`](../docs/requirements/domains/pruning/specs/PRN-002_prune_checkpoints_before_epoch.md).
    pub fn prune_checkpoints_before_epoch(&self, epoch: u64) -> Result<usize, BlockStoreError> {
        if self.read_only {
            return Err(BlockStoreError::Serialization(
                ERR_MUTATION_READ_ONLY.into(),
            ));
        }
        if epoch == 0 {
            return Ok(0);
        }
        let cf = self.cf(CF_CHECKPOINTS)?;
        let mut batch = WriteBatch::default();
        let mut count = 0usize;
        let iter = self.db.iterator_cf(cf, IteratorMode::Start);
        for item in iter {
            let (key_bytes, _value) = item?;
            if key_bytes.len() != 8 {
                continue;
            }
            let key_arr: [u8; 8] = key_bytes.as_ref().try_into().unwrap_or([0; 8]);
            let e = crate::encoding::decode_epoch_key(&key_arr);
            if e >= epoch {
                break;
            }
            batch.delete_cf(cf, key_bytes.as_ref());
            count += 1;
        }
        if count > 0 {
            self.db.write(batch)?;
        }
        Ok(count)
    }

    /// Look up [`BlockRecord`] by hash ([`BLK-004`](../docs/requirements/domains/block_storage/specs/BLK-004.md)).
    ///
    /// **Order**
    /// 1. [`Self::record_cache`] (Mutex map) — clone on hit; **no** RocksDB I/O.
    /// 2. [`Self::header_cache`] — if the header is already deserialized (e.g. after [`Self::put_block`] or
    ///    [`Self::get_header`]), derive [`BlockRecord::from_header`] with [`BlockStatus::Validated`] and insert into
    ///    the record cache; **no** RocksDB `get_cf` on [`CF_HEADERS`].
    /// 3. Else load raw bytes from [`CF_HEADERS`], increment [`Self::cf_headers_physical_gets`], deserialize via
    ///    [`Self::deserialize_header`], warm [`Self::header_cache`] + record cache.
    ///
    /// **Persistence:** [`BlockRecord`] is never written to any column family ([`TYP-004`](../docs/requirements/domains/storage_types/specs/TYP-004.md)); only headers live under [`CF_HEADERS`].
    ///
    /// **Read-only stores:** Record/header RAM caches start empty; the first lookup may read [`CF_HEADERS`] and populate both caches without mutating on-disk layout beyond normal reads.
    pub fn get_record(&self, hash: &Bytes32) -> Result<Option<BlockRecord>, BlockStoreError> {
        {
            let guard = self.record_cache.lock();
            if let Some(r) = guard.get(hash) {
                return Ok(Some(r.clone()));
            }
        }
        if let Some(header) = self.header_cache.get_clone(hash) {
            let record = BlockRecord::from_header(&header, BlockStatus::Validated);
            self.record_cache.lock().insert(*hash, record.clone());
            return Ok(Some(record));
        }
        let cf = self.cf(CF_HEADERS)?;
        self.cf_headers_physical_gets
            .fetch_add(1, Ordering::Relaxed);
        let Some(bytes) = self.db.get_cf(cf, hash_key(hash).as_slice())? else {
            return Ok(None);
        };
        let header = Self::deserialize_header(&bytes)?;
        self.header_cache.insert(*hash, header.clone());
        let record = BlockRecord::from_header(&header, BlockStatus::Validated);
        self.record_cache.lock().insert(*hash, record.clone());
        Ok(Some(record))
    }

    /// Remove one hash from the in-memory [`BlockRecord`] map — **no RocksDB writes** ([`BLK-004`](../docs/requirements/domains/block_storage/specs/BLK-004.md) test plan: simulate record-cache eviction).
    pub fn invalidate_record_cache_entry(&self, hash: &Bytes32) {
        let mut guard = self.record_cache.lock();
        let _ = guard.remove(hash);
    }

    /// **[`BLK-010`](../docs/requirements/domains/block_storage/specs/BLK-010.md)** — Set [`BlockRecord::status`] for a hash already present in [`Self::record_cache`].
    ///
    /// **No disk I/O:** [`BlockRecord`] is cache-only ([`TYP-004`](../docs/requirements/domains/storage_types/specs/TYP-004.md)); this method never touches [`rocksdb::WriteBatch`] or [`DB::put_cf`](rocksdb::DB::put_cf).
    ///
    /// **`in_canonical_chain`:** Recomputed from [`BlockStatus::is_canonical`](dig_block::BlockStatus::is_canonical) so the row stays aligned with [`BlockRecord::from_header`](crate::types::BlockRecord::from_header) ([`TYP-004`](../docs/requirements/domains/storage_types/specs/TYP-004.md) module docs).
    ///
    /// **Precondition:** The hash must already exist in the record cache (typically after [`Self::put_block`] or a cache-warming [`Self::get_record`]). Otherwise returns [`BlockStoreError::Serialization`] whose message starts with
    /// [`ERR_UPDATE_STATUS_RECORD_NOT_CACHED_PREFIX`](crate::error::ERR_UPDATE_STATUS_RECORD_NOT_CACHED_PREFIX)
    /// ([`ERR-001`](../docs/requirements/domains/error_types/specs/ERR-001_blockstoreerror_enum.md) caps the public enum at thirteen variants, so this uses the same stable-prefix pattern as read-only mutation guards).
    pub fn update_status(
        &self,
        hash: &Bytes32,
        status: BlockStatus,
    ) -> Result<(), BlockStoreError> {
        let mut guard = self.record_cache.lock();
        let record = guard.get_mut(hash).ok_or_else(|| {
            BlockStoreError::Serialization(format!(
                "{ERR_UPDATE_STATUS_RECORD_NOT_CACHED_PREFIX}{hash}"
            ))
        })?;
        record.status = status;
        record.in_canonical_chain = status.is_canonical();
        Ok(())
    }

    /// Async retrieval by hash ([`BLK-007`](../docs/requirements/domains/block_storage/specs/BLK-007.md) AC §1, §4, §5).
    ///
    /// **Hot path:** [`Self::block_cache`] hits return cloned blocks **before** any `.await`, so the generated
    /// future can complete as [`Poll::Ready`] on the first [`poll`](std::future::Future::poll) without scheduling
    /// [`tokio::task::spawn_blocking`] (NORMATIVE BLK-007 §1–2).
    ///
    /// **Cold path:** Delegates to [`Self::get_block`] on the blocking pool so RocksDB + zstd never run on a
    /// cooperative tokio worker thread.
    pub async fn get_block_async(
        &self,
        hash: &Bytes32,
    ) -> Result<Option<L2Block>, BlockStoreError> {
        if let Some(block) = self.block_cache.get_clone(hash) {
            return Ok(Some(block));
        }
        let store = self.clone();
        let hash = *hash;
        tokio::task::spawn_blocking(move || store.get_block(&hash))
            .await
            .map_err(Self::map_spawn_join)?
    }

    /// Async header retrieval ([`BLK-007`](../docs/requirements/domains/block_storage/specs/BLK-007.md) AC §2, §4–5).
    pub async fn get_header_async(
        &self,
        hash: &Bytes32,
    ) -> Result<Option<L2BlockHeader>, BlockStoreError> {
        if let Some(header) = self.header_cache.get_clone(hash) {
            return Ok(Some(header));
        }
        let store = self.clone();
        let hash = *hash;
        tokio::task::spawn_blocking(move || store.get_header(&hash))
            .await
            .map_err(Self::map_spawn_join)?
    }

    /// Async canonical-height lookup followed by block load ([`BLK-007`](../docs/requirements/domains/block_storage/specs/BLK-007.md) AC §3).
    ///
    /// **Always `spawn_blocking`:** height→hash uses [`CF_CANONICAL`] I/O; per BLK-007 implementation notes this
    /// stays on the blocking pool even when the block body would hit [`Self::block_cache`], avoiding partial
    /// “async hits” that still touch RocksDB in the sync prelude.
    pub async fn get_block_by_height_async(
        &self,
        height: u64,
    ) -> Result<Option<L2Block>, BlockStoreError> {
        let store = self.clone();
        tokio::task::spawn_blocking(move || store.get_block_by_height(height))
            .await
            .map_err(Self::map_spawn_join)?
    }

    /// Maps a failed [`tokio::task::spawn_blocking`] join handle onto [`BlockStoreError::Serialization`].
    ///
    /// **Why not a dedicated enum variant:** [`ERR-001`](../docs/requirements/domains/error_types/specs/ERR-001_blockstoreerror_enum.md) caps
    /// the error surface at thirteen variants; join failures are rare operational faults surfaced with [`ERR_ASYNC_JOIN_PREFIX`].
    #[inline]
    fn map_spawn_join(err: tokio::task::JoinError) -> BlockStoreError {
        BlockStoreError::Serialization(format!("{ERR_ASYNC_JOIN_PREFIX}{err}"))
    }

    /// Resolve a column family handle by name, or error if the DB was not opened with it.
    ///
    /// This is a thin wrapper around [`DB::cf_handle`](rocksdb::DB::cf_handle) that converts
    /// the `Option<&ColumnFamily>` to our error type. In practice this should never fail
    /// because [`BlockStore::open`] creates all six families via [`cf_options::column_family_descriptors`],
    /// but defensive coding prevents silent `None` dereferences if the CF list drifts.
    pub(crate) fn cf(&self, name: &'static str) -> Result<&rocksdb::ColumnFamily, BlockStoreError> {
        self.db
            .cf_handle(name)
            .ok_or_else(|| BlockStoreError::Serialization(format!("missing column family {name}")))
    }
}

/// Load the current chain tip from [`CF_METADATA`] / [`META_TIP`].
///
/// The tip is a 40-byte value encoding `hash (32 bytes) || height (8 bytes LE)`,
/// decoded via [`ChainTip::from_bytes`]. Returns `None` for a brand-new database
/// that has not yet had [`BlockStore::init_genesis`] called.
///
/// # Chia analogy
///
/// Corresponds to reading `current_peak` from the `block_store` metadata in Chia's
/// `BlockStore.get_peak()`. The DIG version uses a fixed-width binary encoding
/// instead of SQLite row access.
///
/// # Called by
///
/// [`BlockStore::open`] and [`BlockStore::open_readonly`] to populate the in-memory
/// [`BlockStore::tip`] field at startup.
fn load_tip(db: &DB) -> Result<Option<ChainTip>, BlockStoreError> {
    let meta = db
        .cf_handle(CF_METADATA)
        .ok_or_else(|| BlockStoreError::Serialization("missing CF_METADATA".into()))?;
    let Some(raw) = db.get_cf(meta, META_TIP.as_bytes())? else {
        return Ok(None);
    };
    ChainTip::from_bytes(&raw).map(Some)
}

// NOTE: The old `warm_recent_blocks` free function has been replaced by
// `BlockStoreInner::warm_caches` (CAC-006) which runs AFTER full construction
// and populates ALL caches (block, header, record, height index, hash-to-height)
// instead of only counting block existence.