ab_client_database/

lib.rs

//! Client database.
//!
//! ## High-level architecture overview
//!
//! The database operates on [`ClientDatabaseStorageBackend`], which is backed by [`AlignedPage`]s
//! that can be read or written. Pages contain `StorageItem`s, one storage item can occupy one or
//! more pages, but pages always belong to a single storage item. Pages are the smallest unit and
//! align nicely with the hardware architecture of modern SSDs. Each page starts with a prefix that
//! describes the contents of the page. `StorageItem` always starts at the multiple of the
//! `u128`/16 bytes, allowing for direct memory mapping onto target data structures.
//!
//! [`AlignedPage`]: crate::storage_backend::AlignedPage
//!
//! Individual pages are grouped into page groups (configurable via [`ClientDatabaseOptions`]). Page
//! groups can be permanent and ephemeral. Permanent page groups store information that is never
//! going to be deleted, like segment headers. Ephemeral page groups store the majority of the
//! information about blocks, blockchain state and other things that are being created all the time.
//! Once information in an ephemeral page group is too old and no longer needed, it can be
//! repurposed for a new permanent or ephemeral page group. There are different kinds of page groups
//! defined in `PageGroupKind`, and each variant has independent sequence numbers.
//!
//! Page groups are append-only, there is only one active permanent and one ephemeral page group.
//! They are appended with more pages containing storage items until there is no space to add a
//! complete storage item, after which the next page group is started.
//!
//! Ephemeral page groups can be freed only when they contain 100% outdated storage items.
//! Individual pages can't be freed.
//!
//! Each storage item has a sequence number and checksums that help to define the global ordering
//! and check whether a storage item was written fully. Upon restart, the page group containing the
//! latest storage items is found, and the latest fully written storage item is identified to
//! reconstruct the database state.
//!
//! Each page group starts with a `StorageItemPageGroupHeader` storage item for easier
//! identification.
//!
//! The database is typically contained in a single file (though in principle could be contained in
//! multiple if necessary). Before the database can be used, it needs to be formatted with a
//! specific size (it is possible to increase the size afterward) before it can be used. It is
//! expected (but depends on the storage backend) that the whole file size is pre-allocated on disk
//! and no writes will fail due to lack of disk space (which could be the case with a sparse file).

#![feature(
    const_block_items,
    const_convert,
    const_trait_impl,
    default_field_values,
    iter_collect_into,
    maybe_uninit_as_bytes,
    maybe_uninit_fill
)]

mod page_group;
pub mod storage_backend;
mod storage_backend_adapter;
#[cfg(test)]
mod test_utils;
#[cfg(test)]
mod tests;

use crate::page_group::temporary::StorageItemTemporary;
use crate::page_group::temporary::block::StorageItemTemporaryBlock;
use crate::page_group::temporary::segment_headers::StorageItemTemporarySegmentHeaders;
use crate::page_group::temporary::super_segment_headers::StorageItemTemporarySuperSegmentHeaders;
use crate::storage_backend::ClientDatabaseStorageBackend;
use crate::storage_backend_adapter::{
    StorageBackendAdapter, StorageItemHandlerArg, StorageItemHandlers, WriteLocation,
};
use ab_client_api::{
    BeaconChainInfo, BeaconChainInfoWrite, BlockDetails, ChainInfo, ChainInfoWrite,
    ContractSlotState, PersistBlockError, PersistSegmentHeadersError,
    PersistSuperSegmentHeadersError, ReadBlockError, ShardSegmentRoot, ShardSegmentRootsError,
};
use ab_core_primitives::block::body::BeaconChainBody;
use ab_core_primitives::block::body::owned::{GenericOwnedBlockBody, OwnedBeaconChainBody};
use ab_core_primitives::block::header::GenericBlockHeader;
use ab_core_primitives::block::header::owned::GenericOwnedBlockHeader;
use ab_core_primitives::block::owned::{GenericOwnedBlock, OwnedBeaconChainBlock};
use ab_core_primitives::block::{BlockNumber, BlockRoot, GenericBlock};
use ab_core_primitives::segments::{
    LocalSegmentIndex, SegmentHeader, SegmentIndex, SuperSegmentHeader, SuperSegmentIndex,
};
use ab_core_primitives::shard::RealShardKind;
use ab_core_primitives::trees::BlockMerkleMountainRange;
use ab_io_type::trivial_type::TrivialType;
use async_lock::{
    RwLock as AsyncRwLock, RwLockUpgradableReadGuard, RwLockWriteGuard as AsyncRwLockWriteGuard,
};
use rand::rngs::SysError;
use replace_with::replace_with_or_abort;
use smallvec::{SmallVec, smallvec};
use std::any::Any;
use std::collections::{HashMap, VecDeque};
use std::hash::{BuildHasherDefault, Hasher};
use std::num::{NonZeroU32, NonZeroUsize};
use std::ops::Deref;
use std::{fmt, io};
use tracing::error;
use triomphe::Arc;

/// Unique identifier for a database
#[derive(Debug, Copy, Clone, Eq, PartialEq, TrivialType)]
#[repr(C)]
pub struct DatabaseId([u8; 32]);

impl Deref for DatabaseId {
    type Target = [u8; 32];

    #[inline(always)]
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl AsRef<[u8]> for DatabaseId {
    #[inline(always)]
    fn as_ref(&self) -> &[u8] {
        &self.0
    }
}

impl DatabaseId {
    #[inline(always)]
    pub const fn new(bytes: [u8; 32]) -> Self {
        Self(bytes)
    }
}

#[derive(Default)]
struct BlockRootHasher(u64);

impl Hasher for BlockRootHasher {
    #[inline(always)]
    fn finish(&self) -> u64 {
        self.0
    }

    #[inline(always)]
    fn write(&mut self, bytes: &[u8]) {
        let Some(state) = bytes.as_chunks().0.first().copied().map(u64::from_le_bytes) else {
            return;
        };

        self.0 = state;
    }
}

#[derive(Debug)]
pub struct GenesisBlockBuilderResult<Block> {
    /// Genesis block
    pub block: Block,
    /// System contracts state in the genesis block
    pub system_contract_states: Arc<[ContractSlotState]>,
}

/// Options for [`ClientDatabase`].
///
/// Lifecycle of a block, by block offset from the best block with block number `n`, with
/// `block_confirmation_depth = 5` and `soft_confirmation_depth = 2` as an example:
///
/// ```text
/// block   block   fork       written   block details
/// offset  number  blocks     to disk   in memory
///         n + 1                                    next best block, being imported
///      0  n       accepted   no        yes         best block
///      1  n - 1   accepted   no        yes
///      2  n - 2   accepted   yes       yes         soft-confirmed <- soft_confirmation_depth
///      3  n - 3   accepted   yes       yes         soft-confirmed
///      4  n - 4   rejected   yes       yes         confirmed once n + 1 is imported
///      5  n - 5   rejected   yes       no          confirmed <- block_confirmation_depth
///    ...
/// ```
///
/// When block `n` is built, the block at `n - block_confirmation_depth` is confirmed (see
/// `ConsensusConstants::block_confirmation_depth`), and the database marks the same block as
/// confirmed. Fork blocks are only accepted above block offset `block_confirmation_depth - 1`. The
/// block there is only confirmed once `n + 1` is imported, but it already can't change: a fork
/// block next to it would start a reorg of `block_confirmation_depth` blocks, more than the largest
/// allowed reorg of `block_confirmation_depth - 1` blocks. Soft-confirmed blocks are only written
/// to disk early, a reorg can still revert them.
///
/// Importing `n + 1` on any fork makes it the best block and shifts all block offsets by one. Then
/// other blocks at block offset `block_confirmation_depth - 1` are pruned with their descendants,
/// the block at block offset `block_confirmation_depth` is confirmed, which drops its block details
/// (MMR and system contract states) from memory, and blocks that reached block offset
/// `soft_confirmation_depth` are written to disk. Block details of confirmed blocks are only left
/// on disk, [`ChainInfo::header_with_details()`] doesn't return them.
#[derive(Debug, Copy, Clone)]
#[expect(unevaluated_default_field_value, reason = "Intentional")]
pub struct ClientDatabaseOptions<GBB, StorageBackend> {
    /// Write buffer size.
    ///
    /// Larger buffer allows buffering more async writes for improved responsiveness but requires
    /// more RAM. Zero buffer size means all writes must be completed before returning from the
    /// operation that triggered it. Non-zero buffer means writes can happen in the background.
    ///
    /// The recommended value is 5.
    pub write_buffer_size: usize = 5,
    /// Block confirmation depth, the value of `ConsensusConstants::block_confirmation_depth`.
    ///
    /// Fork blocks are only accepted above block offset `block_confirmation_depth - 1` from the
    /// best block, other blocks at that block offset are pruned with their descendants, and blocks
    /// are confirmed at block offset `block_confirmation_depth` (see the block lifecycle in
    /// [`ClientDatabaseOptions`]).
    ///
    /// Must be at least 2.
    pub block_confirmation_depth: BlockNumber,
    /// Soft confirmation depth for blocks.
    ///
    /// Doesn't prevent forking on the consensus level but makes it extremely unlikely.
    ///
    /// This parameter determines how many blocks are retained in memory before being written to
    /// disk. Writing discarded blocks to disk is a waste of resources, so they are retained in
    /// memory before being soft-confirmed and written to disk for longer-term storage.
    ///
    /// Blocks are persisted as soon as they reach this depth. More recent blocks may be lost if the
    /// node is not shut down gracefully (crash, killed by the OS, etc.).
    ///
    /// A smaller number reduces memory usage while increasing the probability of unnecessary disk
    /// writes. A larger number increases memory usage and the number of blocks that may be lost
    /// this way, while avoiding unnecessary disk writes.
    ///
    /// The recommended value is 3 blocks.
    pub soft_confirmation_depth: BlockNumber = BlockNumber::from(3),
    /// Defines how many fork tips should be maintained in total.
    ///
    /// As natural forks occur, there may be more than one tip in existence, with only one of them
    /// being considered "canonical". This parameter defines how many of these tips to maintain in a
    /// sort of LRU style cache. Tips beyond this limit that were not extended for a long time will
    /// be pruned automatically.
    ///
    /// A larger number results in higher memory usage and higher complexity of pruning algorithms.
    ///
    /// The recommended value is 3 blocks.
    pub max_fork_tips: NonZeroUsize = NonZeroUsize::new(3).expect("Not zero; qed"),
    /// Max distance between fork tip and the best block.
    ///
    /// When forks are deeper than this, they will be pruned, even without reaching the
    /// `max_fork_tips` limit. This essentially means the tip was not extended for some time, and
    /// while it is theoretically possible for the chain to continue from this tip, the probability
    /// is so small that it is not worth storing it.
    ///
    /// A larger value results in higher memory usage and higher complexity of pruning algorithms.
    ///
    /// Fork blocks at block offset `block_confirmation_depth - 1` and deeper are pruned regardless
    /// (see the block lifecycle in [`ClientDatabaseOptions`]), so values of
    /// `block_confirmation_depth - 2` and larger all have the same effect as no limit.
    ///
    /// The recommended value is 5 blocks.
    pub max_fork_tip_distance: BlockNumber = BlockNumber::from(5),
    /// Genesis block builder is responsible to create genesis block and corresponding state for
    /// bootstrapping purposes.
    pub genesis_block_builder: GBB,
    /// Storage backend to use for storing and retrieving storage items
    pub storage_backend: StorageBackend,
}

/// Options for [`ClientDatabase`]
#[derive(Debug, Copy, Clone)]
pub struct ClientDatabaseFormatOptions {
    /// The number of [`AlignedPage`]s in a single page group.
    ///
    /// [`AlignedPage`]: crate::storage_backend::AlignedPage
    ///
    /// Each group always has a set of storage items with monotonically increasing sequence
    /// numbers. The database only frees page groups for reuse when all storage items there are
    /// no longer in use.
    ///
    /// A smaller number means storage can be reclaimed for reuse more quickly and higher
    /// concurrency during restart, but must not be too small that no storage item fits within a
    /// page group anymore. A larger number allows finding the range of sequence numbers that are
    /// already used and where potential write interruption happened on restart more efficiently,
    /// but will use more RAM in the process.
    ///
    /// The recommended size is 256 MiB unless a tiny database is used for testing purposes, where
    /// a smaller value might work too.
    ///
    /// A storage item must fit into a single page group together with the page group header, so
    /// this also bounds the size of the largest block that can be stored.
    pub page_group_size: NonZeroU32,
    /// By default, formatting will be aborted if the database appears to be already formatted.
    ///
    /// Setting this option to `true` skips the check and formats the database anyway.
    pub force: bool,
}

#[derive(Debug, thiserror::Error)]
pub enum ClientDatabaseError {
    /// Invalid block confirmation depth, it must be at least 2.
    ///
    /// Fork blocks are only accepted above block offset `block_confirmation_depth - 1`, smaller
    /// values leave no block offset for them.
    #[error("Invalid block confirmation depth, it must be at least 2")]
    InvalidBlockConfirmationDepth,
    /// Invalid soft confirmation depth, it must be smaller than block confirmation depth
    #[error("Invalid soft confirmation depth, it must be smaller than block confirmation depth")]
    InvalidSoftConfirmationDepth,
    /// Storage backend has canceled read request
    #[error("Storage backend has canceled read request")]
    ReadRequestCancelled,
    /// Storage backend read error
    #[error("Storage backend read error: {error}")]
    ReadError {
        /// Low-level error
        error: io::Error,
    },
    /// Unsupported database version
    #[error("Unsupported database version: {database_version}")]
    UnsupportedDatabaseVersion {
        /// Database version
        database_version: u8,
    },
    /// Page group size is too small, must be at least two pages
    #[error("Page group size is too small ({page_group_size}), must be at least two pages")]
    PageGroupSizeTooSmall {
        /// Page group size in pages
        page_group_size: u32,
    },
    /// Page group size is larger than the storage backend
    #[error(
        "Page group size is too large ({page_group_size}), storage backend only has {num_pages} \
        pages"
    )]
    PageGroupSizeTooLarge {
        /// Page group size in pages
        page_group_size: u32,
        /// Number of pages in the storage backend
        num_pages: u32,
    },
    /// Unexpected sequence number
    #[error(
        "Unexpected sequence number {actual} at page offset {page_offset} (expected \
        {expected})"
    )]
    UnexpectedSequenceNumber {
        /// Sequence number in the database
        actual: u64,
        /// Expected sequence number
        expected: u64,
        /// Page offset where storage item is found
        page_offset: u32,
    },
    /// Unexpected storage item
    #[error("Unexpected storage item at offset {page_offset}: {storage_item:?}")]
    UnexpectedStorageItem {
        /// First storage item
        storage_item: Box<dyn fmt::Debug + Send + Sync>,
        /// Page offset where storage item is found
        page_offset: u32,
    },
    /// Invalid block
    #[error("Invalid block at offset {page_offset}")]
    InvalidBlock {
        /// Page offset where storage item is found
        page_offset: u32,
    },
    /// Invalid segment headers
    #[error("Invalid segment headers at offset {page_offset}")]
    InvalidSegmentHeaders {
        /// Page offset where storage item is found
        page_offset: u32,
    },
    /// Failed to adjust ancestor block forks
    #[error("Failed to adjust ancestor block forks")]
    FailedToAdjustAncestorBlockForks,
    /// Database is not formatted yet
    #[error("Database is not formatted yet")]
    Unformatted,
    /// Non-permanent first page group
    #[error("Non-permanent first page group")]
    NonPermanentFirstPageGroup,
}

/// Error for [`ClientDatabase::format()`]
#[derive(Debug, thiserror::Error)]
pub enum ClientDatabaseFormatError {
    /// Storage backend has canceled read request
    #[error("Storage backend has canceled read request")]
    ReadRequestCancelled,
    /// Storage backend read error
    #[error("Storage backend read error: {error}")]
    ReadError {
        /// Low-level error
        error: io::Error,
    },
    /// Failed to generate database id
    #[error("Failed to generate database id")]
    FailedToGenerateDatabaseId {
        /// Low-level error
        #[from]
        error: SysError,
    },
    /// Page group size is too small, must be at least two pages
    #[error("Page group size is too small ({page_group_size}), must be at least two pages")]
    PageGroupSizeTooSmall {
        /// Page group size in pages
        page_group_size: u32,
    },
    /// Page group size is larger than the storage backend
    #[error(
        "Page group size is too large ({page_group_size}), storage backend only has {num_pages} \
        pages"
    )]
    PageGroupSizeTooLarge {
        /// Page group size in pages
        page_group_size: u32,
        /// Number of pages in the storage backend
        num_pages: u32,
    },
    /// Database is already formatted yet
    #[error("Database is already formatted yet")]
    AlreadyFormatted,
    /// Storage backend has canceled a writing request
    #[error("Storage backend has canceled a writing request")]
    WriteRequestCancelled,
    /// Storage item write error
    #[error("Storage item write error")]
    StorageItemWriteError {
        /// Low-level error
        #[from]
        error: io::Error,
    },
}

#[derive(Debug, Copy, Clone)]
struct ForkTip {
    number: BlockNumber,
    root: BlockRoot,
}

/// How far [`ClientDatabase::adjust_ancestor_block_forks()`] walks towards older blocks
#[derive(Debug, Copy, Clone)]
enum AncestorBlockForksWalk {
    /// Stop at the first ancestor that is already at fork offset `0`, relies on the invariant
    /// documented on [`StateData::blocks`]
    UntilCanonical,
    /// Walk block offsets up to and including `max_block_offset`, for when the invariant doesn't
    /// hold (in [`ClientDatabase::open()`])
    UpToBlockOffset { max_block_offset: usize },
}

enum FullBlock<'a, Block>
where
    Block: GenericOwnedBlock,
{
    InMemory(&'a Block),
    Persisted {
        header: &'a Block::Header,
        write_location: WriteLocation,
    },
}

#[derive(Debug)]
struct BeaconChainBlockDetails {
    // Shard segment roots, only present for beacon chain blocks
    shard_segment_roots: Arc<[ShardSegmentRoot]>,
}

impl BeaconChainBlockDetails {
    fn from_body(body: &BeaconChainBody<'_>) -> Self {
        let shard_segment_roots = body
            .intermediate_shard_blocks()
            .iter()
            .flat_map(|intermediate_shard_block_info| {
                let own_segments = intermediate_shard_block_info
                    .own_segments
                    .into_iter()
                    .flat_map({
                        let shard_index = intermediate_shard_block_info.header.prefix.shard_index;

                        move |own_segments| {
                            own_segments
                                .iter()
                                .map(move |(segment_index, &segment_root)| ShardSegmentRoot {
                                    shard_index,
                                    segment_index,
                                    segment_root,
                                })
                        }
                    });
                let child_shard_segment_roots = intermediate_shard_block_info
                    .leaf_shards_segments()
                    .flat_map(move |(_position, shard_index, own_segments)| {
                        own_segments
                            .iter()
                            .map(move |(segment_index, &segment_root)| ShardSegmentRoot {
                                shard_index,
                                segment_index,
                                segment_root,
                            })
                    });

                own_segments.chain(child_shard_segment_roots)
            })
            .collect();

        Self {
            shard_segment_roots,
        }
    }
}

/// Client database block contains details about the block state in the database.
///
/// Originally all blocks are stored in memory. Once a block is soft-confirmed (see
/// [`ClientDatabaseOptions::soft_confirmation_depth`]), it is persisted (likely on disk). Later, at
/// block offset `block_confirmation_depth` (see the block lifecycle in [`ClientDatabaseOptions`]),
/// it is marked as confirmed and its [`BlockDetails`] are dropped from memory, they are only left
/// on disk.
#[derive(Debug)]
enum ClientDatabaseBlock<Block>
where
    Block: GenericOwnedBlock,
{
    /// Block is stored in memory and wasn't persisted yet
    InMemory {
        block: Block,
        block_details: BlockDetails,
        /// Only present for beacon chain blocks
        beacon_chain_block_details: Option<BeaconChainBlockDetails>,
    },
    /// Block was persisted (likely on disk)
    Persisted {
        header: Block::Header,
        block_details: BlockDetails,
        /// Only present for beacon chain blocks
        beacon_chain_block_details: Option<BeaconChainBlockDetails>,
        write_location: WriteLocation,
    },
    /// Block was persisted (likely on disk) and confirmed, its [`BlockDetails`] are only on disk
    PersistedConfirmed {
        header: Block::Header,
        /// Only present for beacon chain blocks
        beacon_chain_block_details: Option<BeaconChainBlockDetails>,
        write_location: WriteLocation,
    },
}

impl<Block> ClientDatabaseBlock<Block>
where
    Block: GenericOwnedBlock,
{
    #[inline(always)]
    fn header(&self) -> &Block::Header {
        #[expect(
            clippy::rest_pattern_accessible_field,
            reason = "Do not need other fields"
        )]
        match self {
            Self::InMemory { block, .. } => block.header(),
            Self::Persisted { header, .. } | Self::PersistedConfirmed { header, .. } => header,
        }
    }

    #[inline(always)]
    fn full_block(&self) -> FullBlock<'_, Block> {
        #[expect(
            clippy::rest_pattern_accessible_field,
            reason = "Do not need other fields"
        )]
        match self {
            Self::InMemory { block, .. } => FullBlock::InMemory(block),
            Self::Persisted {
                header,
                write_location,
                ..
            }
            | Self::PersistedConfirmed {
                header,
                write_location,
                ..
            } => FullBlock::Persisted {
                header,
                write_location: *write_location,
            },
        }
    }

    #[inline(always)]
    fn block_details(&self) -> Option<&BlockDetails> {
        #[expect(
            clippy::rest_pattern_accessible_field,
            reason = "Do not need other fields"
        )]
        match self {
            Self::InMemory { block_details, .. } | Self::Persisted { block_details, .. } => {
                Some(block_details)
            }
            Self::PersistedConfirmed { .. } => None,
        }
    }

    #[inline(always)]
    fn beacon_chain_block_details(&self) -> Option<&BeaconChainBlockDetails> {
        #[expect(
            clippy::rest_pattern_accessible_field,
            reason = "Do not need other fields"
        )]
        match self {
            Self::InMemory {
                beacon_chain_block_details,
                ..
            }
            | Self::Persisted {
                beacon_chain_block_details,
                ..
            }
            | Self::PersistedConfirmed {
                beacon_chain_block_details,
                ..
            } => beacon_chain_block_details.as_ref(),
        }
    }
}

#[derive(Debug)]
struct StateData<Block>
where
    Block: GenericOwnedBlock,
{
    /// Tips of forks that have no descendants.
    ///
    /// The current best block is at the front, the rest are in the order from most recently
    /// updated towards the front to least recently at the back.
    fork_tips: VecDeque<ForkTip>,
    /// Map from block root to block number.
    ///
    /// Is meant to be used in conjunction with `headers` and `blocks` fields, which are indexed by
    /// block numbers.
    block_roots: HashMap<BlockRoot, BlockNumber, BuildHasherDefault<BlockRootHasher>>,
    /// List of blocks with the newest at the front.
    ///
    /// The first element of the first entry corresponds to the best block.
    ///
    /// It is expected that in most block numbers there will be exactly one block, some two,
    /// anything more than that will be very rare. The list of forks for a block number is
    /// organized in such a way that the first entry at every block number corresponds to the
    /// canonical version of the blockchain at any point in time.
    ///
    /// In other words, except while reading blocks in [`ClientDatabase::open()`], the block at
    /// block offset `n + 1` and fork offset `0` is always the parent of the block at block offset
    /// `n` and fork offset `0`. New best blocks establish this with
    /// [`ClientDatabase::adjust_ancestor_block_forks()`], other blocks are only appended at fork
    /// offset `1` or higher, and pruning never removes the block at fork offset `0`. Block offset
    /// [`ClientDatabaseInnerOptions::fork_pruning_block_offset()`] and deeper ones have no forks.
    ///
    /// A position withing this data structure is called "block offset". This is an ephemeral value
    /// and changes as new best blocks are added. Blocks at the same height are collectively called
    /// "block forks" and the position of the block within the same block height is called
    /// "fork offset". While fork offset `0` always corresponds to the canonical version of the
    /// blockchain, other offsets are not guaranteed to follow any particular ordering rules.
    blocks: VecDeque<SmallVec<[ClientDatabaseBlock<Block>; 2]>>,
}

#[derive(Debug)]
struct SegmentHeadersCache {
    segment_headers_cache: Vec<SegmentHeader>,
}

impl SegmentHeadersCache {
    #[inline(always)]
    fn last_segment_header(&self) -> Option<SegmentHeader> {
        self.segment_headers_cache.last().copied()
    }

    #[inline(always)]
    fn max_local_segment_index(&self) -> Option<LocalSegmentIndex> {
        self.segment_headers_cache
            .last()
            .map(|segment_header| segment_header.index.as_inner())
    }

    #[inline(always)]
    fn get_segment_header(&self, local_segment_index: LocalSegmentIndex) -> Option<SegmentHeader> {
        self.segment_headers_cache
            .get(u64::from(local_segment_index) as usize)
            .copied()
    }

    /// Returns actually added segments (some might have been skipped)
    fn add_segment_headers(
        &mut self,
        mut segment_headers: Vec<SegmentHeader>,
    ) -> Result<Vec<SegmentHeader>, PersistSegmentHeadersError> {
        self.segment_headers_cache.reserve(segment_headers.len());

        let mut maybe_last_local_segment_index = self.max_local_segment_index();

        if let Some(last_segment_index) = maybe_last_local_segment_index {
            // Skip already stored segment headers
            segment_headers
                .retain(|segment_header| segment_header.index.as_inner() > last_segment_index);
        }

        // Check all input segment headers to see which ones are not stored yet and verifying that
        // segment indices are monotonically increasing
        for segment_header in segment_headers.iter().copied() {
            let local_segment_index = segment_header.index.as_inner();
            if let Some(last_local_segment_index) = maybe_last_local_segment_index {
                if local_segment_index != last_local_segment_index + LocalSegmentIndex::ONE {
                    return Err(PersistSegmentHeadersError::MustFollowLastSegmentIndex {
                        local_segment_index,
                        last_local_segment_index,
                    });
                }

                self.segment_headers_cache.push(segment_header);
                maybe_last_local_segment_index.replace(local_segment_index);
            } else {
                if local_segment_index != LocalSegmentIndex::ZERO {
                    return Err(PersistSegmentHeadersError::FirstSegmentIndexZero {
                        local_segment_index,
                    });
                }

                self.segment_headers_cache.push(segment_header);
                maybe_last_local_segment_index.replace(local_segment_index);
            }
        }

        Ok(segment_headers)
    }
}

#[derive(Debug)]
struct SuperSegmentHeadersCache {
    super_segment_headers_cache: Vec<SuperSegmentHeader>,
}

impl SuperSegmentHeadersCache {
    #[inline(always)]
    fn last_super_segment_header(&self) -> Option<SuperSegmentHeader> {
        self.super_segment_headers_cache.last().copied()
    }

    #[inline]
    fn previous_super_segment_header(
        &self,
        target_block_number: BlockNumber,
    ) -> Option<SuperSegmentHeader> {
        let block_number = target_block_number.checked_sub(BlockNumber::ONE)?;
        let index = match self.super_segment_headers_cache.binary_search_by_key(
            &block_number,
            |super_segment_header| {
                super_segment_header
                    .target_beacon_chain_block_number
                    .as_inner()
            },
        ) {
            Ok(found_index) => found_index,
            Err(insert_index) => insert_index.checked_sub(1)?,
        };

        self.super_segment_headers_cache.get(index).copied()
    }

    #[inline(always)]
    fn get_super_segment_header(
        &self,
        local_segment_index: SuperSegmentIndex,
    ) -> Option<SuperSegmentHeader> {
        self.super_segment_headers_cache
            .get(u64::from(local_segment_index) as usize)
            .copied()
    }

    #[inline(always)]
    fn get_super_segment_header_for_segment_index(
        &self,
        segment_index: SegmentIndex,
    ) -> Option<SuperSegmentHeader> {
        let index = self
            .super_segment_headers_cache
            .binary_search_by_key(&segment_index, |super_segment_header| {
                super_segment_header.max_segment_index.as_inner()
            })
            .unwrap_or_else(|insert_index| insert_index);

        let super_segment_header = self.super_segment_headers_cache.get(index).copied()?;

        let max_segment_index = super_segment_header.max_segment_index.as_inner();
        let first_segment_index = max_segment_index
            - SegmentIndex::from(u64::from(super_segment_header.num_segments))
            + SegmentIndex::ONE;

        (first_segment_index..=max_segment_index)
            .contains(&segment_index)
            .then_some(super_segment_header)
    }

    /// Returns actually added super segments (some might have been skipped)
    fn add_super_segment_headers(
        &mut self,
        mut super_segment_headers: Vec<SuperSegmentHeader>,
    ) -> Result<Vec<SuperSegmentHeader>, PersistSuperSegmentHeadersError> {
        self.super_segment_headers_cache
            .reserve(super_segment_headers.len());

        let mut maybe_last_super_segment_index = self
            .super_segment_headers_cache
            .last()
            .map(|header| header.index.as_inner());

        if let Some(last_super_segment_index) = maybe_last_super_segment_index {
            // Skip already stored super segment headers
            super_segment_headers.retain(|super_segment_header| {
                super_segment_header.index.as_inner() > last_super_segment_index
            });
        }

        // Check all input super segment headers to see which ones are not stored yet and verifying
        // that super segment indices are monotonically increasing
        for super_segment_header in super_segment_headers.iter().copied() {
            let super_segment_index = super_segment_header.index.as_inner();
            if let Some(last_super_segment_index) = maybe_last_super_segment_index {
                if super_segment_index != last_super_segment_index + SuperSegmentIndex::ONE {
                    return Err(
                        PersistSuperSegmentHeadersError::MustFollowLastSegmentIndex {
                            super_segment_index,
                            last_super_segment_index,
                        },
                    );
                }

                self.super_segment_headers_cache.push(super_segment_header);
                maybe_last_super_segment_index.replace(super_segment_index);
            } else {
                if super_segment_index != SuperSegmentIndex::ZERO {
                    return Err(PersistSuperSegmentHeadersError::FirstSegmentIndexZero {
                        super_segment_index,
                    });
                }

                self.super_segment_headers_cache.push(super_segment_header);
                maybe_last_super_segment_index.replace(super_segment_index);
            }
        }

        Ok(super_segment_headers)
    }
}

// TODO: Hide implementation details
#[derive(Debug)]
struct State<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
{
    data: StateData<Block>,
    segment_headers_cache: SegmentHeadersCache,
    super_segment_headers_cache: SuperSegmentHeadersCache,
    storage_backend_adapter: AsyncRwLock<StorageBackendAdapter<StorageBackend>>,
}

impl<Block, StorageBackend> State<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
{
    #[inline(always)]
    fn best_tip(&self) -> &ForkTip {
        self.data
            .fork_tips
            .front()
            .expect("The best block is always present; qed")
    }

    #[inline(always)]
    fn best_block(&self) -> &ClientDatabaseBlock<Block> {
        self.data
            .blocks
            .front()
            .expect("The best block is always present; qed")
            .first()
            .expect("The best block is always present; qed")
    }
}

#[derive(Debug)]
struct BlockToPersist<'a, Block>
where
    Block: GenericOwnedBlock,
{
    block_offset: usize,
    fork_offset: usize,
    block: &'a Block,
    block_details: &'a BlockDetails,
}

#[derive(Debug)]
struct PersistedBlock {
    block_offset: usize,
    fork_offset: usize,
    write_location: WriteLocation,
}

#[derive(Debug)]
struct ClientDatabaseInnerOptions {
    block_confirmation_depth: BlockNumber,
    soft_confirmation_depth: BlockNumber,
    max_fork_tips: NonZeroUsize,
    max_fork_tip_distance: BlockNumber,
}

impl ClientDatabaseInnerOptions {
    /// Create options, checking that `block_confirmation_depth` is at least 2 and
    /// `soft_confirmation_depth` is smaller than `block_confirmation_depth`
    fn new(
        block_confirmation_depth: BlockNumber,
        soft_confirmation_depth: BlockNumber,
        max_fork_tips: NonZeroUsize,
        max_fork_tip_distance: BlockNumber,
    ) -> Result<Self, ClientDatabaseError> {
        if block_confirmation_depth < BlockNumber::from(2) {
            return Err(ClientDatabaseError::InvalidBlockConfirmationDepth);
        }

        if soft_confirmation_depth >= block_confirmation_depth {
            return Err(ClientDatabaseError::InvalidSoftConfirmationDepth);
        }

        Ok(Self {
            block_confirmation_depth,
            soft_confirmation_depth,
            max_fork_tips,
            max_fork_tip_distance,
        })
    }

    /// Block offset of the block that the next best block (block number `best_number + 1`)
    /// confirms, fork blocks are only accepted above it and other blocks there are pruned.
    ///
    /// With `block_confirmation_depth = 5` (see the block lifecycle in [`ClientDatabaseOptions`]):
    ///
    /// ```text
    /// block offset  block number  fork blocks
    ///            0  n             accepted
    ///            1  n - 1         accepted
    ///            2  n - 2         accepted
    ///            3  n - 3         accepted
    ///            4  n - 4         rejected, other blocks pruned,
    ///                             confirmed once n + 1 is imported
    /// ```
    #[inline(always)]
    fn fork_pruning_block_offset(&self) -> usize {
        let fork_pruning_block_offset = self
            .block_confirmation_depth
            .checked_sub(BlockNumber::ONE)
            .expect("Block confirmation depth is checked to be at least 2 in `Self::new()`; qed");

        u64::from(fork_pruning_block_offset) as usize
    }
}

#[derive(Debug)]
struct Inner<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
{
    state: AsyncRwLock<State<Block, StorageBackend>>,
    options: ClientDatabaseInnerOptions,
}

/// Client database
#[derive(Debug)]
pub struct ClientDatabase<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
{
    inner: Arc<Inner<Block, StorageBackend>>,
}

impl<Block, StorageBackend> Clone for ClientDatabase<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
{
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

#[expect(clippy::empty_drop, reason = "Not implemented yet")]
impl<Block, StorageBackend> Drop for ClientDatabase<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
{
    fn drop(&mut self) {
        // TODO: Persist things that were not persisted yet to reduce the data loss on shutdown.
        //  Blocks must be written in the same order as in `Self::insert_new_best_block()`,
        //  `Self::open()` relies on it.
    }
}

impl<Block, StorageBackend> ChainInfo<Block> for ClientDatabase<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
    StorageBackend: ClientDatabaseStorageBackend,
{
    #[inline]
    fn best_root(&self) -> BlockRoot {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        self.inner.state.read_blocking().best_tip().root
    }

    #[inline]
    fn best_header(&self) -> Block::Header {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        self.inner
            .state
            .read_blocking()
            .best_block()
            .header()
            .clone()
    }

    #[inline]
    fn best_header_with_details(&self) -> (Block::Header, BlockDetails) {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();
        let best_block = state.best_block();
        (
            best_block.header().clone(),
            best_block
                .block_details()
                .expect("Always present for the best block; qed")
                .clone(),
        )
    }

    // TODO: Add fast path when `descendant_block_root` is the best block
    #[inline]
    fn ancestor_header(
        &self,
        ancestor_block_number: BlockNumber,
        descendant_block_root: &BlockRoot,
    ) -> Option<Block::Header> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();
        let best_number = state.best_tip().number;

        let ancestor_block_offset =
            u64::from(best_number.checked_sub(ancestor_block_number)?) as usize;
        let ancestor_block_candidates = state.data.blocks.get(ancestor_block_offset)?;

        let descendant_block_number = *state.data.block_roots.get(descendant_block_root)?;
        if ancestor_block_number > descendant_block_number {
            return None;
        }
        let descendant_block_offset =
            u64::from(best_number.checked_sub(descendant_block_number)?) as usize;

        // Range of blocks where the first item is expected to contain a descendant
        let mut blocks_range_iter = state
            .data
            .blocks
            .iter()
            .enumerate()
            .skip(descendant_block_offset);

        let (_offset, descendant_block_candidates) = blocks_range_iter.next()?;
        let descendant_header = descendant_block_candidates
            .iter()
            .find(|block| &*block.header().header().root() == descendant_block_root)?
            .header();

        if ancestor_block_number == descendant_block_number {
            return Some(descendant_header.clone());
        }

        // If there are no forks at this level, then this is the canonical chain and ancestor
        // block number we're looking for is the first block at the corresponding block number.
        // Similarly, if there is just a single ancestor candidate and descendant exists, it must be
        // the one we care about.
        if descendant_block_candidates.len() == 1 || ancestor_block_candidates.len() == 1 {
            return ancestor_block_candidates
                .iter()
                .next()
                .map(|block| block.header().clone());
        }

        let mut parent_block_root = &descendant_header.header().prefix.parent_root;

        // Iterate over the blocks following descendant until ancestor is reached
        for (block_offset, parent_candidates) in blocks_range_iter {
            let parent_header = parent_candidates
                .iter()
                .find(|header| &*header.header().header().root() == parent_block_root)?
                .header();

            // When header offset matches, we found the header
            if block_offset == ancestor_block_offset {
                return Some(parent_header.clone());
            }

            parent_block_root = &parent_header.header().prefix.parent_root;
        }

        None
    }

    #[inline]
    fn header(&self, block_root: &BlockRoot) -> Option<Block::Header> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();
        let best_number = state.best_tip().number;

        let block_number = *state.data.block_roots.get(block_root)?;
        let block_offset = u64::from(best_number.checked_sub(block_number)?) as usize;
        let block_candidates = state.data.blocks.get(block_offset)?;

        block_candidates.iter().find_map(|block| {
            let header = block.header();

            if &*header.header().root() == block_root {
                Some(header.clone())
            } else {
                None
            }
        })
    }

    #[inline]
    fn header_with_details(&self, block_root: &BlockRoot) -> Option<(Block::Header, BlockDetails)> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();
        let best_number = state.best_tip().number;

        let block_number = *state.data.block_roots.get(block_root)?;
        let block_offset = u64::from(best_number.checked_sub(block_number)?) as usize;
        let block_candidates = state.data.blocks.get(block_offset)?;

        block_candidates.iter().find_map(|block| {
            let header = block.header();
            let block_details = block.block_details().cloned()?;

            if &*header.header().root() == block_root {
                Some((header.clone(), block_details))
            } else {
                None
            }
        })
    }

    #[inline]
    async fn block(&self, block_root: &BlockRoot) -> Result<Block, ReadBlockError> {
        let state = self.inner.state.read().await;
        let best_number = state.best_tip().number;

        let block_number = *state
            .data
            .block_roots
            .get(block_root)
            .ok_or(ReadBlockError::UnknownBlockRoot)?;
        let block_offset = u64::from(
            best_number
                .checked_sub(block_number)
                .expect("Known block roots always have valid block offset; qed"),
        ) as usize;
        let block_candidates = state
            .data
            .blocks
            .get(block_offset)
            .expect("Valid block offsets always have block entries; qed");

        for block_candidate in block_candidates {
            let header = block_candidate.header();

            if &*header.header().root() == block_root {
                return match block_candidate.full_block() {
                    FullBlock::InMemory(block) => Ok(block.clone()),
                    FullBlock::Persisted {
                        header,
                        write_location,
                    } => {
                        let storage_backend_adapter = state.storage_backend_adapter.read().await;

                        let storage_item = storage_backend_adapter
                            .read_storage_item::<StorageItemTemporary>(write_location)
                            .await?;

                        let storage_item_block = match storage_item {
                            StorageItemTemporary::Block(storage_item_block) => storage_item_block,
                            StorageItemTemporary::SegmentHeaders(_) => {
                                return Err(ReadBlockError::StorageItemReadError {
                                    error: io::Error::other(
                                        "Unexpected storage item: `SegmentHeaders`",
                                    ),
                                });
                            }
                            StorageItemTemporary::SuperSegmentHeaders(_) => {
                                return Err(ReadBlockError::StorageItemReadError {
                                    error: io::Error::other(
                                        "Unexpected storage item: `SuperSegmentHeaders`",
                                    ),
                                });
                            }
                        };

                        let StorageItemTemporaryBlock {
                            header: _,
                            body,
                            mmr_with_block: _,
                            system_contract_states: _,
                        } = storage_item_block;

                        Block::from_buffers(header.buffer().clone(), body)
                            .ok_or(ReadBlockError::FailedToDecode)
                    }
                };
            }
        }

        unreachable!("Known block root always has block candidate associated with it; qed")
    }

    #[inline]
    fn last_segment_header(&self) -> Option<SegmentHeader> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();
        state.segment_headers_cache.last_segment_header()
    }

    #[inline]
    fn get_segment_header(&self, segment_index: LocalSegmentIndex) -> Option<SegmentHeader> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();

        state
            .segment_headers_cache
            .get_segment_header(segment_index)
    }

    fn segment_headers_for_block(&self, block_number: BlockNumber) -> Vec<SegmentHeader> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();

        let Some(last_local_segment_index) = state.segment_headers_cache.max_local_segment_index()
        else {
            // Not initialized
            return Vec::new();
        };

        // Special case for the initial segment (for beacon chain genesis block)
        if Block::Block::SHARD_KIND == RealShardKind::BeaconChain
            && block_number == BlockNumber::ONE
        {
            // If there is a segment index present, and we store monotonically increasing segment
            // headers, then the first header exists
            return vec![
                state
                    .segment_headers_cache
                    .get_segment_header(LocalSegmentIndex::ZERO)
                    .expect("Segment headers are stored in monotonically increasing order; qed"),
            ];
        }

        if last_local_segment_index == LocalSegmentIndex::ZERO {
            // Genesis segment already included in block #1
            return Vec::new();
        }

        let mut current_local_segment_index = last_local_segment_index;
        loop {
            // If the current segment index present, and we store monotonically increasing segment
            // headers, then the current segment header exists as well.
            let current_segment_header = state
                .segment_headers_cache
                .get_segment_header(current_local_segment_index)
                .expect("Segment headers are stored in monotonically increasing order; qed");

            // The block immediately after the archived segment adding the confirmation depth
            let target_block_number = current_segment_header.last_archived_block.number()
                + BlockNumber::ONE
                + self.inner.options.block_confirmation_depth;
            if target_block_number == block_number {
                let mut headers_for_block = vec![current_segment_header];

                // Check block spanning multiple segments
                let last_archived_block_number = current_segment_header.last_archived_block.number;
                let mut local_segment_index = current_local_segment_index - LocalSegmentIndex::ONE;

                while let Some(segment_header) = state
                    .segment_headers_cache
                    .get_segment_header(local_segment_index)
                {
                    if segment_header.last_archived_block.number == last_archived_block_number {
                        headers_for_block.insert(0, segment_header);
                        local_segment_index -= LocalSegmentIndex::ONE;
                    } else {
                        break;
                    }
                }

                return headers_for_block;
            }

            // iterate segments further
            if target_block_number > block_number {
                // no need to check the initial segment
                if current_local_segment_index > LocalSegmentIndex::ONE {
                    current_local_segment_index -= LocalSegmentIndex::ONE;
                } else {
                    break;
                }
            } else {
                // No segment headers required
                return Vec::new();
            }
        }

        // No segment headers required
        Vec::new()
    }
}

impl<Block, StorageBackend> ChainInfoWrite<Block> for ClientDatabase<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
    StorageBackend: ClientDatabaseStorageBackend,
{
    async fn persist_block(
        &self,
        block: Block,
        block_details: BlockDetails,
    ) -> Result<(), PersistBlockError> {
        let mut state = self.inner.state.write().await;
        let best_number = state.best_tip().number;

        let header = block.header().header();

        let block_number = header.prefix.number;
        let block_root = *header.root();

        // A known block would be stored again as a separate fork block, pruning of which would
        // also remove the original block
        if state.data.block_roots.contains_key(&block_root) {
            return Err(PersistBlockError::AlreadyKnown);
        }

        if best_number == BlockNumber::ZERO && block_number != BlockNumber::ONE {
            // Special case when syncing on top of the fresh database
            Self::insert_first_block(&mut state.data, block, block_details);

            return Ok(());
        }

        if block_number == best_number + BlockNumber::ONE {
            return Self::insert_new_best_block(state, &self.inner, block, block_details).await;
        }

        let block_offset = u64::from(
            best_number
                .checked_sub(block_number)
                .ok_or(PersistBlockError::MissingParent)?,
        ) as usize;

        // A fork block at the fork pruning block offset or deeper could only become a part of the
        // best chain with a reorg of at least `block_confirmation_depth` blocks (see the block
        // lifecycle in `ClientDatabaseOptions`)
        if block_offset >= self.inner.options.fork_pruning_block_offset() {
            return Err(PersistBlockError::OutsideAcceptableRange);
        }

        // The parent may have been pruned after block import checked it and before this write
        // lock was taken
        if !state
            .data
            .block_roots
            .contains_key(&header.prefix.parent_root)
        {
            return Err(PersistBlockError::MissingParent);
        }

        let state = &mut *state;

        let block_forks = state.data.blocks.get_mut(block_offset).ok_or_else(|| {
            error!(
                %block_number,
                %block_offset,
                "Failed to store block fork, header offset is missing despite being within \
                acceptable range"
            );

            PersistBlockError::OutsideAcceptableRange
        })?;

        for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
            // Block's parent is no longer a fork tip, remove it
            if fork_tip.root == header.prefix.parent_root {
                state.data.fork_tips.remove(index);
                break;
            }
        }

        // Insert at position 1, which means the most recent tip, which doesn't correspond to
        // the best block
        state.data.fork_tips.insert(
            1,
            ForkTip {
                number: block_number,
                root: block_root,
            },
        );
        state.data.block_roots.insert(block_root, block_number);
        let beacon_chain_block_details = <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
            .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
        block_forks.push(ClientDatabaseBlock::InMemory {
            block,
            block_details,
            beacon_chain_block_details,
        });

        // Forks at or below soft confirmation depth are not persisted here, such forks are only
        // imported as part of a chain that is about to become the best chain and are persisted by
        // `Self::insert_new_best_block()` once it does

        Self::prune_outdated_fork_tips(best_number, &mut state.data, &self.inner.options);

        Ok(())
    }

    async fn persist_segment_headers(
        &self,
        segment_headers: Vec<SegmentHeader>,
    ) -> Result<(), PersistSegmentHeadersError> {
        let mut state = self.inner.state.write().await;

        let added_segment_headers = state
            .segment_headers_cache
            .add_segment_headers(segment_headers)?;

        if added_segment_headers.is_empty() {
            return Ok(());
        }

        // Convert write lock into upgradable read lock to allow reads, while preventing segment
        // headers modifications
        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
        //  are satisfied. If not, blocking read locks in other places will cause issues.
        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);

        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;

        storage_backend_adapter
            .write_storage_item(StorageItemTemporary::SegmentHeaders(
                StorageItemTemporarySegmentHeaders {
                    segment_headers: added_segment_headers,
                },
            ))
            .await?;

        Ok(())
    }
}

impl<StorageBackend> BeaconChainInfo for ClientDatabase<OwnedBeaconChainBlock, StorageBackend>
where
    StorageBackend: ClientDatabaseStorageBackend,
{
    fn shard_segment_roots(
        &self,
        block_number: BlockNumber,
    ) -> Result<Arc<[ShardSegmentRoot]>, ShardSegmentRootsError> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();
        let best_number = state.best_tip().number;

        let block_offset = u64::from(
            best_number
                .checked_sub(block_number)
                .ok_or(ShardSegmentRootsError::BlockMissing { block_number })?,
        ) as usize;

        let block = state
            .data
            .blocks
            .get(block_offset)
            .ok_or(ShardSegmentRootsError::BlockMissing { block_number })?
            .first()
            .expect("There is always at least one block candidate; qed");

        Ok(Arc::clone(
            &block
                .beacon_chain_block_details()
                .as_ref()
                .expect("Always present in the beacon chain block; qed")
                .shard_segment_roots,
        ))
    }

    #[inline]
    fn last_super_segment_header(&self) -> Option<SuperSegmentHeader> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();
        state
            .super_segment_headers_cache
            .last_super_segment_header()
    }

    #[inline]
    fn previous_super_segment_header(
        &self,
        block_number: BlockNumber,
    ) -> Option<SuperSegmentHeader> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();

        state
            .super_segment_headers_cache
            .previous_super_segment_header(block_number)
    }

    #[inline]
    fn get_super_segment_header(
        &self,
        super_segment_index: SuperSegmentIndex,
    ) -> Option<SuperSegmentHeader> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();

        state
            .super_segment_headers_cache
            .get_super_segment_header(super_segment_index)
    }

    fn get_super_segment_header_for_segment_index(
        &self,
        segment_index: SegmentIndex,
    ) -> Option<SuperSegmentHeader> {
        // Blocking read lock is fine because where a write lock is only taken for a short time and
        // most locks are read locks
        let state = self.inner.state.read_blocking();

        state
            .super_segment_headers_cache
            .get_super_segment_header_for_segment_index(segment_index)
    }
}

impl<StorageBackend> BeaconChainInfoWrite for ClientDatabase<OwnedBeaconChainBlock, StorageBackend>
where
    StorageBackend: ClientDatabaseStorageBackend,
{
    async fn persist_super_segment_header(
        &self,
        super_segment_header: SuperSegmentHeader,
    ) -> Result<bool, PersistSuperSegmentHeadersError> {
        let mut state = self.inner.state.write().await;

        let added_super_segment_headers = state
            .super_segment_headers_cache
            .add_super_segment_headers(vec![super_segment_header])?;

        if added_super_segment_headers.is_empty() {
            return Ok(false);
        }

        // Convert write lock into upgradable read lock to allow reads, while preventing super
        // segment headers modifications
        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
        //  are satisfied. If not, blocking read locks in other places will cause issues.
        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);

        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;

        storage_backend_adapter
            .write_storage_item(StorageItemTemporary::SuperSegmentHeaders(
                StorageItemTemporarySuperSegmentHeaders {
                    super_segment_headers: added_super_segment_headers,
                },
            ))
            .await?;

        Ok(true)
    }

    async fn persist_super_segment_headers(
        &self,
        super_segment_headers: Vec<SuperSegmentHeader>,
    ) -> Result<(), PersistSuperSegmentHeadersError> {
        let mut state = self.inner.state.write().await;

        let added_super_segment_headers = state
            .super_segment_headers_cache
            .add_super_segment_headers(super_segment_headers)?;

        if added_super_segment_headers.is_empty() {
            return Ok(());
        }

        // Convert write lock into upgradable read lock to allow reads, while preventing super
        // segment headers modifications
        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
        //  are satisfied. If not, blocking read locks in other places will cause issues.
        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);

        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;

        storage_backend_adapter
            .write_storage_item(StorageItemTemporary::SuperSegmentHeaders(
                StorageItemTemporarySuperSegmentHeaders {
                    super_segment_headers: added_super_segment_headers,
                },
            ))
            .await?;

        Ok(())
    }
}

impl<Block, StorageBackend> ClientDatabase<Block, StorageBackend>
where
    Block: GenericOwnedBlock,
    StorageBackend: ClientDatabaseStorageBackend,
{
    /// Open the existing database.
    ///
    /// NOTE: The database needs to be formatted with [`Self::format()`] before it can be used.
    ///
    /// Opening replays storage items in the order they were written. Every block read at a new
    /// highest block number is handled like a new best block: forks are adjusted towards it, other
    /// blocks at block offset `block_confirmation_depth - 1` are pruned, then the block at block
    /// offset `block_confirmation_depth` is confirmed (see the block lifecycle in
    /// [`ClientDatabaseOptions`]). The best block after opening is the last block written at the
    /// highest stored block number. It may be behind or on a different fork than the best block
    /// before shutdown, the node re-syncs from there.
    pub async fn open<GBB>(
        options: ClientDatabaseOptions<GBB, StorageBackend>,
    ) -> Result<Self, ClientDatabaseError>
    where
        GBB: FnOnce() -> GenesisBlockBuilderResult<Block>,
    {
        let ClientDatabaseOptions {
            write_buffer_size,
            block_confirmation_depth,
            soft_confirmation_depth,
            max_fork_tips,
            max_fork_tip_distance,
            genesis_block_builder,
            storage_backend,
        } = options;
        let options = ClientDatabaseInnerOptions::new(
            block_confirmation_depth,
            soft_confirmation_depth,
            max_fork_tips,
            max_fork_tip_distance,
        )?;

        let mut state_data = StateData {
            fork_tips: VecDeque::new(),
            block_roots: HashMap::default(),
            blocks: VecDeque::new(),
        };
        let mut segment_headers_cache = SegmentHeadersCache {
            segment_headers_cache: Vec::new(),
        };
        let mut super_segment_headers_cache = SuperSegmentHeadersCache {
            super_segment_headers_cache: Vec::new(),
        };

        // Deeper block offsets only have a single block, so ancestor walks stop there
        let fork_pruning_block_offset = options.fork_pruning_block_offset();

        let storage_item_handlers = StorageItemHandlers {
            permanent: |_arg| {
                // TODO
                Ok(())
            },
            temporary: |arg| {
                let StorageItemHandlerArg {
                    storage_item,
                    page_offset,
                    num_pages,
                } = arg;
                let storage_item_block = match storage_item {
                    StorageItemTemporary::Block(storage_item_block) => storage_item_block,
                    StorageItemTemporary::SegmentHeaders(segment_headers) => {
                        let num_segment_headers = segment_headers.segment_headers.len();
                        return match segment_headers_cache
                            .add_segment_headers(segment_headers.segment_headers)
                        {
                            Ok(_) => Ok(()),
                            Err(error) => {
                                error!(
                                    %page_offset,
                                    %num_segment_headers,
                                    %error,
                                    "Failed to add segment headers from storage item"
                                );

                                Err(ClientDatabaseError::InvalidSegmentHeaders { page_offset })
                            }
                        };
                    }
                    StorageItemTemporary::SuperSegmentHeaders(super_segment_headers) => {
                        let num_super_segment_headers =
                            super_segment_headers.super_segment_headers.len();
                        return match super_segment_headers_cache
                            .add_super_segment_headers(super_segment_headers.super_segment_headers)
                        {
                            Ok(_) => Ok(()),
                            Err(error) => {
                                error!(
                                    %page_offset,
                                    %num_super_segment_headers,
                                    %error,
                                    "Failed to add segment headers from storage item"
                                );

                                Err(ClientDatabaseError::InvalidSegmentHeaders { page_offset })
                            }
                        };
                    }
                };

                // TODO: It would be nice to not allocate body here since we'll not use it here
                //  anyway
                let StorageItemTemporaryBlock {
                    header,
                    body,
                    mmr_with_block,
                    system_contract_states,
                } = storage_item_block;

                let header = Block::Header::from_buffer(header).map_err(|_buffer| {
                    error!(%page_offset, "Failed to decode block header from bytes");

                    ClientDatabaseError::InvalidBlock { page_offset }
                })?;
                let body = Block::Body::from_buffer(body).map_err(|_buffer| {
                    error!(%page_offset, "Failed to decode block body from bytes");

                    ClientDatabaseError::InvalidBlock { page_offset }
                })?;

                let block_root = *header.header().root();
                let block_number = header.header().prefix.number;

                state_data.block_roots.insert(block_root, block_number);

                let maybe_best_number = state_data
                    .blocks
                    .front()
                    .and_then(|block_forks| block_forks.first())
                    .map(|best_block| {
                        // Type inference is not working here for some reason
                        let header: &Block::Header = best_block.header();

                        header.header().prefix.number
                    });

                let block_offset = if let Some(best_number) = maybe_best_number {
                    if block_number <= best_number {
                        u64::from(best_number - block_number) as usize
                    } else {
                        // The new best block must follow the previous best block
                        if block_number - best_number != BlockNumber::ONE {
                            error!(
                                %page_offset,
                                %best_number,
                                %block_number,
                                "Invalid new best block number, it must be only one block \
                                higher than the best block"
                            );

                            return Err(ClientDatabaseError::InvalidBlock { page_offset });
                        }

                        state_data.blocks.push_front(SmallVec::new());
                        // Will insert a new block at the front
                        0
                    }
                } else {
                    state_data.blocks.push_front(SmallVec::new());
                    // Will insert a new block at the front
                    0
                };

                let Some(block_forks) = state_data.blocks.get_mut(block_offset) else {
                    // Ignore the older block, other blocks at its height were already pruned
                    // anyway
                    return Ok(());
                };

                // Push a new block to the end of the list, we'll fix it up later
                let beacon_chain_block_details =
                    <dyn Any>::downcast_ref::<OwnedBeaconChainBody>(&body)
                        .map(|body| BeaconChainBlockDetails::from_body(body.body()));
                block_forks.push(ClientDatabaseBlock::Persisted {
                    header,
                    block_details: BlockDetails {
                        mmr_with_block,
                        system_contract_states,
                    },
                    beacon_chain_block_details,
                    write_location: WriteLocation {
                        page_offset,
                        num_pages,
                    },
                });

                // If a new block was inserted, prune forks and confirm a new canonical block to
                // prune extra in-memory information, like `Self::insert_new_best_block()` does
                // (see the block lifecycle in `ClientDatabaseOptions`)
                if block_offset == 0 && block_forks.len() == 1 {
                    // Block forks are in the order they were written in rather than in canonical
                    // order here. Any block at a new highest block number descends from the only
                    // block that was left at the fork pruning block offset when it was written, so
                    // the forks are adjusted towards the block just read first.
                    if state_data
                        .blocks
                        .get(fork_pruning_block_offset)
                        .is_some_and(|block_forks| block_forks.len() > 1)
                    {
                        if !Self::adjust_ancestor_block_forks(
                            &mut state_data.blocks,
                            block_root,
                            AncestorBlockForksWalk::UpToBlockOffset {
                                max_block_offset: fork_pruning_block_offset,
                            },
                        ) {
                            error!(
                                %page_offset,
                                %block_number,
                                "Failed to find ancestors of the block read from the database"
                            );

                            return Err(ClientDatabaseError::FailedToAdjustAncestorBlockForks);
                        }

                        Self::prune_non_canonical_blocks(
                            fork_pruning_block_offset,
                            &mut state_data,
                        );
                    }

                    Self::confirm_canonical_block(block_number, &mut state_data, &options);
                }

                Ok(())
            },
        };

        let storage_backend_adapter =
            StorageBackendAdapter::open(write_buffer_size, storage_item_handlers, storage_backend)
                .await?;

        // NOTE: Confirmation is not stored, a deep reorg right after opening may revert blocks that
        // were confirmed before shutdown, consumers like the archiver then fail loudly, as intended
        if let Some(best_block) = state_data.blocks.front().and_then(|block_forks| {
            // The best block is the last block written at the highest block number, block forks
            // are in the order they were written in
            block_forks.last()
        }) {
            // Type inference is not working here for some reason
            let header: &Block::Header = best_block.header();
            let header = header.header();
            let block_number = header.prefix.number;
            let block_root = *header.root();

            // Block forks are in the order they were written in, so all block offsets that may
            // have forks are adjusted
            if !Self::adjust_ancestor_block_forks(
                &mut state_data.blocks,
                block_root,
                AncestorBlockForksWalk::UpToBlockOffset {
                    max_block_offset: fork_pruning_block_offset,
                },
            ) {
                return Err(ClientDatabaseError::FailedToAdjustAncestorBlockForks);
            }

            // Store the best block as the first and only fork tip
            state_data.fork_tips.push_front(ForkTip {
                number: block_number,
                root: block_root,
            });
        } else {
            let GenesisBlockBuilderResult {
                block,
                system_contract_states,
            } = genesis_block_builder();

            // If the database is empty, initialize everything with the genesis block
            let header = block.header().header();
            let block_number = header.prefix.number;
            let block_root = *header.root();

            state_data.fork_tips.push_front(ForkTip {
                number: block_number,
                root: block_root,
            });
            state_data.block_roots.insert(block_root, block_number);
            let beacon_chain_block_details =
                <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
                    .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
            state_data
                .blocks
                .push_front(smallvec![ClientDatabaseBlock::InMemory {
                    block,
                    block_details: BlockDetails {
                        system_contract_states,
                        mmr_with_block: Arc::new({
                            let mut mmr = BlockMerkleMountainRange::new();
                            mmr.add_leaf(&block_root);
                            mmr
                        })
                    },
                    beacon_chain_block_details,
                }]);
        }

        let state = State {
            data: state_data,
            segment_headers_cache,
            super_segment_headers_cache,
            storage_backend_adapter: AsyncRwLock::new(storage_backend_adapter),
        };

        let inner = Inner {
            state: AsyncRwLock::new(state),
            options,
        };

        Ok(Self {
            inner: Arc::new(inner),
        })
    }

    /// Format a new database
    pub async fn format(
        storage_backend: &StorageBackend,
        options: ClientDatabaseFormatOptions,
    ) -> Result<(), ClientDatabaseFormatError> {
        StorageBackendAdapter::format(storage_backend, options).await
    }

    fn insert_first_block(state: &mut StateData<Block>, block: Block, block_details: BlockDetails) {
        // If the database is empty, initialize everything with the genesis block
        let header = block.header().header();
        let block_number = header.prefix.number;
        let block_root = *header.root();

        state.fork_tips.clear();
        state.fork_tips.push_front(ForkTip {
            number: block_number,
            root: block_root,
        });
        state.block_roots.clear();
        state.block_roots.insert(block_root, block_number);
        state.blocks.clear();
        let beacon_chain_block_details = <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
            .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
        state
            .blocks
            .push_front(smallvec![ClientDatabaseBlock::InMemory {
                block,
                block_details,
                beacon_chain_block_details,
            }]);
    }

    async fn insert_new_best_block(
        mut state: AsyncRwLockWriteGuard<'_, State<Block, StorageBackend>>,
        inner: &Inner<Block, StorageBackend>,
        block: Block,
        block_details: BlockDetails,
    ) -> Result<(), PersistBlockError> {
        let header = block.header().header();
        let block_number = header.prefix.number;
        let block_root = *header.root();
        let parent_root = header.prefix.parent_root;

        // Adjust the relative order of forks to ensure the first index always corresponds to
        // ancestors of the new best block
        if !Self::adjust_ancestor_block_forks(
            &mut state.data.blocks,
            parent_root,
            AncestorBlockForksWalk::UntilCanonical,
        ) {
            return Err(PersistBlockError::MissingParent);
        }

        // Store new block in the state
        {
            for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
                // Block's parent is no longer a fork tip, remove it
                if fork_tip.root == parent_root {
                    state.data.fork_tips.remove(index);
                    break;
                }
            }

            state.data.fork_tips.push_front(ForkTip {
                number: block_number,
                root: block_root,
            });
            state.data.block_roots.insert(block_root, block_number);
            let beacon_chain_block_details =
                <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
                    .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
            state
                .data
                .blocks
                .push_front(smallvec![ClientDatabaseBlock::InMemory {
                    block,
                    block_details: block_details.clone(),
                    beacon_chain_block_details,
                }]);
        }

        let options = &inner.options;

        // The new best block shifted all block offsets by one, other blocks that reached the fork
        // pruning block offset could only become a part of the best chain with a reorg of
        // `block_confirmation_depth` blocks now (see the block lifecycle in
        // `ClientDatabaseOptions`)
        Self::prune_non_canonical_blocks(options.fork_pruning_block_offset(), &mut state.data);
        Self::confirm_canonical_block(block_number, &mut state.data, options);
        Self::prune_outdated_fork_tips(block_number, &mut state.data, options);

        // Convert write lock into upgradable read lock to allow reads, while preventing concurrent
        // block modifications
        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
        //  are satisfied. If not, blocking read locks in other places will cause issues.
        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);

        let mut blocks_to_persist = Vec::new();
        for block_offset in u64::from(options.soft_confirmation_depth) as usize.. {
            let Some(fork_blocks) = state.data.blocks.get(block_offset) else {
                break;
            };

            let len_before = blocks_to_persist.len();
            fork_blocks
                .iter()
                .enumerate()
                .filter_map(|(fork_offset, client_database_block)| {
                    #[expect(
                        clippy::rest_pattern_accessible_field,
                        reason = "Do not need other fields"
                    )]
                    match client_database_block {
                        ClientDatabaseBlock::InMemory {
                            block,
                            block_details,
                            beacon_chain_block_details: _,
                        } => Some(BlockToPersist {
                            block_offset,
                            fork_offset,
                            block,
                            block_details,
                        }),
                        ClientDatabaseBlock::Persisted { .. }
                        | ClientDatabaseBlock::PersistedConfirmed { .. } => {
                            // Already persisted
                            None
                        }
                    }
                })
                .collect_into(&mut blocks_to_persist);

            if blocks_to_persist.len() == len_before {
                break;
            }
        }

        // Persist blocks from older to newer: the deepest block offset first and within each block
        // offset the canonical block at fork offset `0` last. The scan above stops at the first
        // block offset without blocks to persist, so a block is never written before its parent.
        // `Self::open()` relies on this order.
        let mut persisted_blocks = Vec::with_capacity(blocks_to_persist.len());
        {
            let mut storage_backend_adapter = state.storage_backend_adapter.write().await;

            for block_to_persist in blocks_to_persist.into_iter().rev() {
                let BlockToPersist {
                    block_offset,
                    fork_offset,
                    block,
                    block_details,
                } = block_to_persist;

                let write_location = storage_backend_adapter
                    .write_storage_item(StorageItemTemporary::Block(StorageItemTemporaryBlock {
                        header: block.header().buffer().clone(),
                        body: block.body().buffer().clone(),
                        mmr_with_block: Arc::clone(&block_details.mmr_with_block),
                        system_contract_states: Arc::clone(&block_details.system_contract_states),
                    }))
                    .await?;

                persisted_blocks.push(PersistedBlock {
                    block_offset,
                    fork_offset,
                    write_location,
                });
            }
        }

        // Convert blocks to persisted
        let mut state = RwLockUpgradableReadGuard::upgrade(state).await;
        for persisted_block in persisted_blocks {
            let PersistedBlock {
                block_offset,
                fork_offset,
                write_location,
            } = persisted_block;

            let block = state
                .data
                .blocks
                .get_mut(block_offset)
                .expect("Still holding the same lock since last check; qed")
                .get_mut(fork_offset)
                .expect("Still holding the same lock since last check; qed");

            replace_with_or_abort(block, |block| {
                if let ClientDatabaseBlock::InMemory {
                    block,
                    block_details,
                    beacon_chain_block_details,
                } = block
                {
                    let (header, _body) = block.split();

                    ClientDatabaseBlock::Persisted {
                        header,
                        block_details,
                        beacon_chain_block_details,
                        write_location,
                    }
                } else {
                    unreachable!("Still holding the same lock since last check; qed");
                }
            });
        }

        // TODO: Prune blocks that are no longer necessary
        // TODO: Prune unused page groups here or elsewhere? Page groups that contain blocks whose
        //  block number is within `block_confirmation_depth` of the highest stored block number
        //  must not be freed, `Self::open()` needs them to prune forks and find the canonical
        //  blocks to confirm. Once older page groups are freed, block numbers below that window
        //  may only have forks that are not ancestors of the best block left, `Self::open()`
        //  currently confirms such a fork and needs to drop it instead. Add a test for this
        //  together with freeing, it also pins the bound of
        //  `AncestorBlockForksWalk::UpToBlockOffset`: an unbounded walk fails to find the
        //  canonical ancestor at such a block number.

        Ok(())
    }

    /// Adjust the relative order of forks to ensure the first index always corresponds to
    /// `parent_block_root` and its ancestors.
    ///
    /// Returns `true` on success and `false` if one of the parents was not found.
    #[must_use = "The return value indicates whether all parent blocks were found"]
    fn adjust_ancestor_block_forks(
        blocks: &mut VecDeque<SmallVec<[ClientDatabaseBlock<Block>; 2]>>,
        mut parent_block_root: BlockRoot,
        walk: AncestorBlockForksWalk,
    ) -> bool {
        for (block_offset, parent_blocks) in blocks.iter_mut().enumerate() {
            if let AncestorBlockForksWalk::UpToBlockOffset { max_block_offset } = walk
                && block_offset > max_block_offset
            {
                break;
            }

            let Some(fork_offset_parent_block_root) =
                parent_blocks
                    .iter()
                    .enumerate()
                    .find_map(|(fork_offset, fork_block)| {
                        let fork_header = fork_block.header().header();
                        if *fork_header.root() == parent_block_root {
                            Some((fork_offset, fork_header.prefix.parent_root))
                        } else {
                            None
                        }
                    })
            else {
                return false;
            };

            let fork_offset;
            (fork_offset, parent_block_root) = fork_offset_parent_block_root;

            if fork_offset == 0 && matches!(walk, AncestorBlockForksWalk::UntilCanonical) {
                // All older ancestors are already at fork offset `0` too
                break;
            }

            parent_blocks.swap(0, fork_offset);
        }

        true
    }

    /// Prune outdated fork tips that are too deep and have not been updated for a long time.
    ///
    /// Note that actual headers, blocks and MMRs could remain if they are currently used by
    /// something or were already persisted on disk. With persisted blocks specifically, RAM usage
    /// implications are minimal, and we wouldn't want to re-download already stored blocks in case
    /// they end up being necessary later.
    fn prune_outdated_fork_tips(
        best_number: BlockNumber,
        state: &mut StateData<Block>,
        options: &ClientDatabaseInnerOptions,
    ) {
        let state = &mut *state;

        // These forks are just candidates because they will not be pruned if the reference count is
        // not 1, indicating they are still in use by something
        let mut candidate_forks_to_remove = Vec::with_capacity(options.max_fork_tips.get());

        // Prune forks that are too far away from the best block
        state.fork_tips.retain(|fork_tip| {
            if best_number - fork_tip.number > options.max_fork_tip_distance {
                candidate_forks_to_remove.push(*fork_tip);
                false
            } else {
                true
            }
        });
        // Prune forks that exceed the maximum number of forks
        if state.fork_tips.len() > options.max_fork_tips.get() {
            state
                .fork_tips
                .drain(options.max_fork_tips.get()..)
                .collect_into(&mut candidate_forks_to_remove);
        }

        // Prune all possible candidates
        candidate_forks_to_remove
            .retain(|fork_tip| !Self::prune_outdated_fork(best_number, fork_tip, state));
        // Return those that were not pruned back to the list of tips
        state.fork_tips.extend(candidate_forks_to_remove);
    }

    /// Returns `true` if the tip was pruned successfully and `false` if it should be returned to
    /// the list of fork tips
    #[must_use = "The return value indicates whether the fork tip was pruned"]
    fn prune_outdated_fork(
        best_number: BlockNumber,
        fork_tip: &ForkTip,
        state: &mut StateData<Block>,
    ) -> bool {
        let block_offset = u64::from(best_number - fork_tip.number) as usize;

        // Prune fork top and all its ancestors that are not used
        let mut block_root_to_prune = fork_tip.root;
        let mut pruned_tip = false;
        for block_offset in block_offset.. {
            let Some(fork_blocks) = state.blocks.get_mut(block_offset) else {
                if !pruned_tip {
                    error!(
                        %best_number,
                        ?fork_tip,
                        block_offset,
                        "Block offset was not present in the database, this is an implementation \
                        bug #1"
                    );
                }
                // No forks left to prune
                break;
            };

            if fork_blocks.len() == 1 {
                if !pruned_tip {
                    error!(
                        %best_number,
                        ?fork_tip,
                        block_offset,
                        "Block offset was not present in the database, this is an implementation \
                        bug #2"
                    );
                }

                // No forks left to prune
                break;
            }

            let Some((fork_offset, block)) = fork_blocks
                .iter()
                .enumerate()
                // Skip ancestor of the best block, it is certainly not a fork to be pruned
                .skip(1)
                .find(|(_fork_offset, block)| {
                    *block.header().header().root() == block_root_to_prune
                })
            else {
                if !pruned_tip {
                    error!(
                        %best_number,
                        ?fork_tip,
                        block_offset,
                        "Block offset was not present in the database, this is an implementation \
                        bug #3"
                    );
                }

                // Nothing left to prune
                break;
            };

            let header = block.header();

            // More than one instance means something somewhere is using this block, children are
            // not counted here and are checked separately below
            if header.ref_count() > 1 {
                break;
            }

            // Blocks that are already persisted
            #[expect(
                clippy::rest_pattern_accessible_field,
                reason = "Do not care about fields"
            )]
            match block {
                ClientDatabaseBlock::InMemory { .. } => {
                    // Prune
                }
                ClientDatabaseBlock::Persisted { .. }
                | ClientDatabaseBlock::PersistedConfirmed { .. } => {
                    // Already on disk, keep it in memory for later, but prune the tip
                    pruned_tip = true;
                    break;
                }
            }

            let parent_root = header.header().prefix.parent_root;
            state.block_roots.remove(&block_root_to_prune);
            fork_blocks.swap_remove(fork_offset);

            pruned_tip = true;

            // The parent is still needed by its other children
            if fork_blocks
                .iter()
                .any(|fork_block| fork_block.header().header().prefix.parent_root == parent_root)
            {
                break;
            }

            block_root_to_prune = parent_root;
        }

        pruned_tip
    }

    /// Confirm the block at block offset `block_confirmation_depth`.
    ///
    /// The block at that block offset must be an ancestor of the block at `best_number` that was
    /// just added, other blocks at that block offset were pruned by
    /// [`Self::prune_non_canonical_blocks()`] at
    /// [`ClientDatabaseInnerOptions::fork_pruning_block_offset()`] already.
    fn confirm_canonical_block(
        best_number: BlockNumber,
        state_data: &mut StateData<Block>,
        options: &ClientDatabaseInnerOptions,
    ) {
        // Blocks keep their `BlockDetails` in memory until they are confirmed here, and
        // `ChainInfo::header_with_details()` only returns them from memory. Block import needs them
        // for the parent of every block it imports. Fork blocks are only accepted above the fork
        // pruning block offset, so their parents are at most at it, which is exactly what is left
        // in memory:
        //
        // block offset                        block details
        // 0 ..= block_confirmation_depth - 2  in memory, fork blocks and their parents
        // block_confirmation_depth - 1        in memory, parents of the deepest fork blocks
        // block_confirmation_depth            dropped from memory here
        let block_offset = u64::from(options.block_confirmation_depth) as usize;

        let Some(fork_blocks) = state_data.blocks.get_mut(block_offset) else {
            // Nothing to confirm yet
            return;
        };

        // Mark the canonical block as confirmed
        let Some(canonical_block) = fork_blocks.first_mut() else {
            error!(
                %best_number,
                block_offset,
                "Have not found a canonical block to confirm, this is an implementation bug"
            );
            return;
        };

        replace_with_or_abort(canonical_block, |block| {
            #[expect(
                clippy::rest_pattern_accessible_field,
                reason = "Do not need other fields"
            )]
            match block {
                ClientDatabaseBlock::InMemory { .. } => {
                    error!(
                        %best_number,
                        block_offset,
                        header = ?block.header(),
                        "Block to be confirmed must not be in memory, this is an implementation bug"
                    );
                    block
                }
                ClientDatabaseBlock::Persisted {
                    header,
                    block_details: _,
                    beacon_chain_block_details,
                    write_location,
                } => ClientDatabaseBlock::PersistedConfirmed {
                    header,
                    beacon_chain_block_details,
                    write_location,
                },
                ClientDatabaseBlock::PersistedConfirmed { .. } => {
                    error!(
                        %best_number,
                        block_offset,
                        header = ?block.header(),
                        "Block to be confirmed must not be confirmed yet, this is an \
                        implementation bug"
                    );
                    block
                }
            }
        });
    }

    /// Prune blocks other than the one at fork offset `0` at `block_offset` with their descendants
    fn prune_non_canonical_blocks(block_offset: usize, state_data: &mut StateData<Block>) {
        let Some(fork_blocks) = state_data.blocks.get_mut(block_offset) else {
            return;
        };

        let mut block_roots_to_prune = fork_blocks
            .drain(1..)
            .map(|block| *block.header().header().root())
            .collect::<Vec<_>>();
        let mut current_block_offset = block_offset;
        while !block_roots_to_prune.is_empty() {
            // Prune fork tips (if any)
            state_data
                .fork_tips
                .retain(|fork_tip| !block_roots_to_prune.contains(&fork_tip.root));

            // Prune removed block roots
            for block_root in &block_roots_to_prune {
                state_data.block_roots.remove(block_root);
            }

            // Block offset for direct descendants
            if let Some(next_block_offset) = current_block_offset.checked_sub(1) {
                current_block_offset = next_block_offset;
            } else {
                // Reached the tip
                break;
            }

            let fork_blocks = state_data
                .blocks
                .get_mut(current_block_offset)
                .expect("Lower block offset always exists; qed");

            // Collect descendants of pruned blocks to prune them next
            block_roots_to_prune = fork_blocks
                .drain_filter(|block| {
                    let header = block.header().header();

                    block_roots_to_prune.contains(&header.prefix.parent_root)
                })
                .map(|block| *block.header().header().root())
                .collect();
        }
    }
}