Skip to main content

ab_client_database/
lib.rs

1//! Client database.
2//!
3//! ## High-level architecture overview
4//!
5//! The database operates on [`ClientDatabaseStorageBackend`], which is backed by [`AlignedPage`]s
6//! that can be read or written. Pages contain `StorageItem`s, one storage item can occupy one or
7//! more pages, but pages always belong to a single storage item. Pages are the smallest unit and
8//! align nicely with the hardware architecture of modern SSDs. Each page starts with a prefix that
9//! describes the contents of the page. `StorageItem` always starts at the multiple of the
10//! `u128`/16 bytes, allowing for direct memory mapping onto target data structures.
11//!
12//! [`AlignedPage`]: crate::storage_backend::AlignedPage
13//!
14//! Individual pages are grouped into page groups (configurable via [`ClientDatabaseOptions`]). Page
15//! groups can be permanent and ephemeral. Permanent page groups store information that is never
16//! going to be deleted, like segment headers. Ephemeral page groups store the majority of the
17//! information about blocks, blockchain state and other things that are being created all the time.
18//! Once information in an ephemeral page group is too old and no longer needed, it can be
19//! repurposed for a new permanent or ephemeral page group. There are different kinds of page groups
20//! defined in `PageGroupKind`, and each variant has independent sequence numbers.
21//!
22//! Page groups are append-only, there is only one active permanent and one ephemeral page group.
23//! They are appended with more pages containing storage items until there is no space to add a
24//! complete storage item, after which the next page group is started.
25//!
26//! Ephemeral page groups can be freed only when they contain 100% outdated storage items.
27//! Individual pages can't be freed.
28//!
29//! Each storage item has a sequence number and checksums that help to define the global ordering
30//! and check whether a storage item was written fully. Upon restart, the page group containing the
31//! latest storage items is found, and the latest fully written storage item is identified to
32//! reconstruct the database state.
33//!
34//! Each page group starts with a `StorageItemPageGroupHeader` storage item for easier
35//! identification.
36//!
37//! The database is typically contained in a single file (though in principle could be contained in
38//! multiple if necessary). Before the database can be used, it needs to be formatted with a
39//! specific size (it is possible to increase the size afterward) before it can be used. It is
40//! expected (but depends on the storage backend) that the whole file size is pre-allocated on disk
41//! and no writes will fail due to lack of disk space (which could be the case with a sparse file).
42
43#![feature(
44    const_block_items,
45    const_convert,
46    const_trait_impl,
47    default_field_values,
48    get_mut_unchecked,
49    iter_collect_into,
50    maybe_uninit_fill
51)]
52
53mod page_group;
54pub mod storage_backend;
55mod storage_backend_adapter;
56
57use crate::page_group::temporary::StorageItemTemporary;
58use crate::page_group::temporary::block::StorageItemTemporaryBlock;
59use crate::page_group::temporary::segment_headers::StorageItemTemporarySegmentHeaders;
60use crate::page_group::temporary::super_segment_headers::StorageItemTemporarySuperSegmentHeaders;
61use crate::storage_backend::ClientDatabaseStorageBackend;
62use crate::storage_backend_adapter::{
63    StorageBackendAdapter, StorageItemHandlerArg, StorageItemHandlers, WriteLocation,
64};
65use ab_client_api::{
66    BeaconChainInfo, BeaconChainInfoWrite, BlockDetails, BlockMerkleMountainRange, ChainInfo,
67    ChainInfoWrite, ContractSlotState, PersistBlockError, PersistSegmentHeadersError,
68    PersistSuperSegmentHeadersError, ReadBlockError, ShardSegmentRoot, ShardSegmentRootsError,
69};
70use ab_core_primitives::block::body::BeaconChainBody;
71use ab_core_primitives::block::body::owned::{GenericOwnedBlockBody, OwnedBeaconChainBody};
72use ab_core_primitives::block::header::GenericBlockHeader;
73use ab_core_primitives::block::header::owned::GenericOwnedBlockHeader;
74use ab_core_primitives::block::owned::{GenericOwnedBlock, OwnedBeaconChainBlock};
75use ab_core_primitives::block::{BlockNumber, BlockRoot, GenericBlock};
76use ab_core_primitives::segments::{
77    LocalSegmentIndex, SegmentHeader, SegmentIndex, SuperSegmentHeader, SuperSegmentIndex,
78};
79use ab_core_primitives::shard::RealShardKind;
80use ab_io_type::trivial_type::TrivialType;
81use async_lock::{
82    RwLock as AsyncRwLock, RwLockUpgradableReadGuard, RwLockWriteGuard as AsyncRwLockWriteGuard,
83};
84use rand::rngs::SysError;
85use rclite::Arc;
86use replace_with::replace_with_or_abort;
87use smallvec::{SmallVec, smallvec};
88use std::any::Any;
89use std::collections::{HashMap, VecDeque};
90use std::hash::{BuildHasherDefault, Hasher};
91use std::num::{NonZeroU32, NonZeroUsize};
92use std::ops::Deref;
93use std::sync::Arc as StdArc;
94use std::{fmt, io};
95use tracing::error;
96
97/// Unique identifier for a database
98#[derive(Debug, Copy, Clone, Eq, PartialEq, TrivialType)]
99#[repr(C)]
100pub struct DatabaseId([u8; 32]);
101
102impl Deref for DatabaseId {
103    type Target = [u8; 32];
104
105    #[inline(always)]
106    fn deref(&self) -> &Self::Target {
107        &self.0
108    }
109}
110
111impl AsRef<[u8]> for DatabaseId {
112    #[inline(always)]
113    fn as_ref(&self) -> &[u8] {
114        &self.0
115    }
116}
117
118impl DatabaseId {
119    #[inline(always)]
120    pub const fn new(bytes: [u8; 32]) -> Self {
121        Self(bytes)
122    }
123}
124
125#[derive(Default)]
126struct BlockRootHasher(u64);
127
128impl Hasher for BlockRootHasher {
129    #[inline(always)]
130    fn finish(&self) -> u64 {
131        self.0
132    }
133
134    #[inline(always)]
135    fn write(&mut self, bytes: &[u8]) {
136        let Some(state) = bytes.as_chunks().0.first().copied().map(u64::from_le_bytes) else {
137            return;
138        };
139
140        self.0 = state;
141    }
142}
143
144#[derive(Debug)]
145pub struct GenesisBlockBuilderResult<Block> {
146    /// Genesis block
147    pub block: Block,
148    /// System contracts state in the genesis block
149    pub system_contract_states: StdArc<[ContractSlotState]>,
150}
151
152/// Options for [`ClientDatabase`]
153#[derive(Debug, Copy, Clone)]
154pub struct ClientDatabaseOptions<GBB, StorageBackend> {
155    /// Write buffer size.
156    ///
157    /// Larger buffer allows buffering more async writes for improved responsiveness but requires
158    /// more RAM. Zero buffer size means all writes must be completed before returning from the
159    /// operation that triggered it. Non-zero buffer means writes can happen in the background.
160    ///
161    /// The recommended value is 5.
162    pub write_buffer_size: usize = 5,
163    /// Blocks at this depth are considered to be "confirmed" and irreversible from the consensus
164    /// perspective.
165    ///
166    /// This parameter allows establishing a final canonical order of blocks and eliminating any
167    /// potential forks at a specified depth and beyond.
168    pub block_confirmation_depth: BlockNumber,
169    /// Soft confirmation depth for blocks.
170    ///
171    /// Doesn't prevent forking on the consensus level but makes it extremely unlikely.
172    ///
173    /// This parameter determines how many blocks are retained in memory before being written to
174    /// disk. Writing discarded blocks to disk is a waste of resources, so they are retained in
175    /// memory before being soft-confirmed and written to disk for longer-term storage.
176    ///
177    /// A smaller number reduces memory usage while increasing the probability of unnecessary disk
178    /// writes. A larger number increases memory usage, while avoiding unnecessary disk writes, but
179    /// also increases the chance of recent blocks not being retained on disk in case of a crash.
180    ///
181    /// The recommended value is 3 blocks.
182    pub soft_confirmation_depth: BlockNumber = BlockNumber::from(3),
183    /// Defines how many fork tips should be maintained in total.
184    ///
185    /// As natural forks occur, there may be more than one tip in existence, with only one of them
186    /// being considered "canonical". This parameter defines how many of these tips to maintain in a
187    /// sort of LRU style cache. Tips beyond this limit that were not extended for a long time will
188    /// be pruned automatically.
189    ///
190    /// A larger number results in higher memory usage and higher complexity of pruning algorithms.
191    ///
192    /// The recommended value is 3 blocks.
193    pub max_fork_tips: NonZeroUsize = NonZeroUsize::new(3).expect("Not zero; qed"),
194    /// Max distance between fork tip and the best block.
195    ///
196    /// When forks are this deep, they will be pruned, even without reaching the `max_fork_tips`
197    /// limit. This essentially means the tip was not extended for some time, and while it is
198    /// theoretically possible for the chain to continue from this tip, the probability is so small
199    /// that it is not worth storing it.
200    ///
201    /// A larger value results in higher memory usage and higher complexity of pruning algorithms.
202    ///
203    /// The recommended value is 5 blocks.
204    pub max_fork_tip_distance: BlockNumber = BlockNumber::from(5),
205    /// Genesis block builder is responsible to create genesis block and corresponding state for
206    /// bootstrapping purposes.
207    pub genesis_block_builder: GBB,
208    /// Storage backend to use for storing and retrieving storage items
209    pub storage_backend: StorageBackend,
210}
211
212/// Options for [`ClientDatabase`]
213#[derive(Debug, Copy, Clone)]
214pub struct ClientDatabaseFormatOptions {
215    /// The number of [`AlignedPage`]s in a single page group.
216    ///
217    /// [`AlignedPage`]: crate::storage_backend::AlignedPage
218    ///
219    /// Each group always has a set of storage items with monotonically increasing sequence
220    /// numbers. The database only frees page groups for reuse when all storage items there are
221    /// no longer in use.
222    ///
223    /// A smaller number means storage can be reclaimed for reuse more quickly and higher
224    /// concurrency during restart, but must not be too small that no storage item fits within a
225    /// page group anymore. A larger number allows finding the range of sequence numbers that are
226    /// already used and where potential write interruption happened on restart more efficiently,
227    /// but will use more RAM in the process.
228    ///
229    /// The recommended size is 256 MiB unless a tiny database is used for testing purposes, where
230    /// a smaller value might work too.
231    pub page_group_size: NonZeroU32,
232    /// By default, formatting will be aborted if the database appears to be already formatted.
233    ///
234    /// Setting this option to `true` skips the check and formats the database anyway.
235    pub force: bool,
236}
237
238#[derive(Debug, thiserror::Error)]
239pub enum ClientDatabaseError {
240    /// Invalid soft confirmation depth, it must be smaller than confirmation depth k
241    #[error("Invalid soft confirmation depth, it must be smaller than confirmation depth k")]
242    InvalidSoftConfirmationDepth,
243    /// Invalid max fork tip distance, it must be smaller or equal to confirmation depth k
244    #[error("Invalid max fork tip distance, it must be smaller or equal to confirmation depth k")]
245    InvalidMaxForkTipDistance,
246    /// Storage backend has canceled read request
247    #[error("Storage backend has canceled read request")]
248    ReadRequestCancelled,
249    /// Storage backend read error
250    #[error("Storage backend read error: {error}")]
251    ReadError {
252        /// Low-level error
253        error: io::Error,
254    },
255    /// Unsupported database version
256    #[error("Unsupported database version: {database_version}")]
257    UnsupportedDatabaseVersion {
258        /// Database version
259        database_version: u8,
260    },
261    /// Page group size is too small, must be at least two pages
262    #[error("Page group size is too small ({page_group_size}), must be at least two pages")]
263    PageGroupSizeTooSmall {
264        /// Page group size in pages
265        page_group_size: u32,
266    },
267    /// Unexpected sequence number
268    #[error(
269        "Unexpected sequence number {actual} at page offset {page_offset} (expected \
270        {expected})"
271    )]
272    UnexpectedSequenceNumber {
273        /// Sequence number in the database
274        actual: u64,
275        /// Expected sequence number
276        expected: u64,
277        /// Page offset where storage item is found
278        page_offset: u32,
279    },
280    /// Unexpected storage item
281    #[error("Unexpected storage item at offset {page_offset}: {storage_item:?}")]
282    UnexpectedStorageItem {
283        /// First storage item
284        storage_item: Box<dyn fmt::Debug + Send + Sync>,
285        /// Page offset where storage item is found
286        page_offset: u32,
287    },
288    /// Invalid block
289    #[error("Invalid block at offset {page_offset}")]
290    InvalidBlock {
291        /// Page offset where storage item is found
292        page_offset: u32,
293    },
294    /// Invalid segment headers
295    #[error("Invalid segment headers at offset {page_offset}")]
296    InvalidSegmentHeaders {
297        /// Page offset where storage item is found
298        page_offset: u32,
299    },
300    /// Failed to adjust ancestor block forks
301    #[error("Failed to adjust ancestor block forks")]
302    FailedToAdjustAncestorBlockForks,
303    /// Database is not formatted yet
304    #[error("Database is not formatted yet")]
305    Unformatted,
306    /// Non-permanent first page group
307    #[error("Non-permanent first page group")]
308    NonPermanentFirstPageGroup,
309}
310
311/// Error for [`ClientDatabase::format()`]
312#[derive(Debug, thiserror::Error)]
313pub enum ClientDatabaseFormatError {
314    /// Storage backend has canceled read request
315    #[error("Storage backend has canceled read request")]
316    ReadRequestCancelled,
317    /// Storage backend read error
318    #[error("Storage backend read error: {error}")]
319    ReadError {
320        /// Low-level error
321        error: io::Error,
322    },
323    /// Failed to generate database id
324    #[error("Failed to generate database id")]
325    FailedToGenerateDatabaseId {
326        /// Low-level error
327        #[from]
328        error: SysError,
329    },
330    /// Database is already formatted yet
331    #[error("Database is already formatted yet")]
332    AlreadyFormatted,
333    /// Storage backend has canceled a writing request
334    #[error("Storage backend has canceled a writing request")]
335    WriteRequestCancelled,
336    /// Storage item write error
337    #[error("Storage item write error")]
338    StorageItemWriteError {
339        /// Low-level error
340        #[from]
341        error: io::Error,
342    },
343}
344
345#[derive(Debug, Copy, Clone)]
346struct ForkTip {
347    number: BlockNumber,
348    root: BlockRoot,
349}
350
351enum FullBlock<'a, Block>
352where
353    Block: GenericOwnedBlock,
354{
355    InMemory(&'a Block),
356    Persisted {
357        header: &'a Block::Header,
358        write_location: WriteLocation,
359    },
360}
361
362#[derive(Debug)]
363struct BeaconChainBlockDetails {
364    // Shard segment roots, only present for beacon chain blocks
365    shard_segment_roots: StdArc<[ShardSegmentRoot]>,
366}
367
368impl BeaconChainBlockDetails {
369    fn from_body(body: &BeaconChainBody<'_>) -> Self {
370        let shard_segment_roots = body
371            .intermediate_shard_blocks()
372            .iter()
373            .flat_map(|intermediate_shard_block_info| {
374                let own_segments = intermediate_shard_block_info
375                    .own_segments
376                    .into_iter()
377                    .flat_map({
378                        let shard_index = intermediate_shard_block_info.header.prefix.shard_index;
379
380                        move |own_segments| {
381                            (own_segments.first_local_segment_index..)
382                                .zip(own_segments.segment_roots)
383                                .map(move |(segment_index, &segment_root)| ShardSegmentRoot {
384                                    shard_index,
385                                    segment_index,
386                                    segment_root,
387                                })
388                        }
389                    });
390                let child_shard_segment_roots = intermediate_shard_block_info
391                    .leaf_shards_segments()
392                    .flat_map(move |(shard_index, own_segments)| {
393                        (own_segments.first_local_segment_index..)
394                            .zip(own_segments.segment_roots)
395                            .map(move |(segment_index, &segment_root)| ShardSegmentRoot {
396                                shard_index,
397                                segment_index,
398                                segment_root,
399                            })
400                    });
401
402                own_segments.chain(child_shard_segment_roots)
403            })
404            .collect();
405
406        Self {
407            shard_segment_roots,
408        }
409    }
410}
411
412/// Client database block contains details about the block state in the database.
413///
414/// Originally all blocks are stored in memory. Once a block is soft-confirmed (see
415/// [`ClientDatabaseOptions::soft_confirmation_depth`]), it is persisted (likely on disk). Later
416///  when it is "confirmed" fully (see [`ClientDatabaseOptions::soft_confirmation_depth`]), it
417/// becomes irreversible.
418#[derive(Debug)]
419enum ClientDatabaseBlock<Block>
420where
421    Block: GenericOwnedBlock,
422{
423    /// Block is stored in memory and wasn't persisted yet
424    InMemory {
425        block: Block,
426        block_details: BlockDetails,
427        /// Only present for beacon chain blocks
428        beacon_chain_block_details: Option<BeaconChainBlockDetails>,
429    },
430    /// Block was persisted (likely on disk)
431    Persisted {
432        header: Block::Header,
433        block_details: BlockDetails,
434        /// Only present for beacon chain blocks
435        beacon_chain_block_details: Option<BeaconChainBlockDetails>,
436        write_location: WriteLocation,
437    },
438    /// Block was persisted (likely on disk) and is irreversibly "confirmed" from the consensus
439    /// perspective
440    PersistedConfirmed {
441        header: Block::Header,
442        /// Only present for beacon chain blocks
443        beacon_chain_block_details: Option<BeaconChainBlockDetails>,
444        write_location: WriteLocation,
445    },
446}
447
448impl<Block> ClientDatabaseBlock<Block>
449where
450    Block: GenericOwnedBlock,
451{
452    #[inline(always)]
453    fn header(&self) -> &Block::Header {
454        match self {
455            Self::InMemory { block, .. } => block.header(),
456            Self::Persisted { header, .. } | Self::PersistedConfirmed { header, .. } => header,
457        }
458    }
459
460    #[inline(always)]
461    fn full_block(&self) -> FullBlock<'_, Block> {
462        match self {
463            Self::InMemory { block, .. } => FullBlock::InMemory(block),
464            Self::Persisted {
465                header,
466                write_location,
467                ..
468            }
469            | Self::PersistedConfirmed {
470                header,
471                write_location,
472                ..
473            } => FullBlock::Persisted {
474                header,
475                write_location: *write_location,
476            },
477        }
478    }
479
480    #[inline(always)]
481    fn block_details(&self) -> Option<&BlockDetails> {
482        match self {
483            Self::InMemory { block_details, .. } | Self::Persisted { block_details, .. } => {
484                Some(block_details)
485            }
486            Self::PersistedConfirmed { .. } => None,
487        }
488    }
489
490    #[inline(always)]
491    fn beacon_chain_block_details(&self) -> Option<&BeaconChainBlockDetails> {
492        match self {
493            Self::InMemory {
494                beacon_chain_block_details,
495                ..
496            }
497            | Self::Persisted {
498                beacon_chain_block_details,
499                ..
500            }
501            | Self::PersistedConfirmed {
502                beacon_chain_block_details,
503                ..
504            } => beacon_chain_block_details.as_ref(),
505        }
506    }
507}
508
509#[derive(Debug)]
510struct StateData<Block>
511where
512    Block: GenericOwnedBlock,
513{
514    /// Tips of forks that have no descendants.
515    ///
516    /// The current best block is at the front, the rest are in the order from most recently
517    /// updated towards the front to least recently at the back.
518    fork_tips: VecDeque<ForkTip>,
519    /// Map from block root to block number.
520    ///
521    /// Is meant to be used in conjunction with `headers` and `blocks` fields, which are indexed by
522    /// block numbers.
523    block_roots: HashMap<BlockRoot, BlockNumber, BuildHasherDefault<BlockRootHasher>>,
524    /// List of blocks with the newest at the front.
525    ///
526    /// The first element of the first entry corresponds to the best block.
527    ///
528    /// It is expected that in most block numbers there will be exactly one block, some two,
529    /// anything more than that will be very rare. The list of forks for a block number is
530    /// organized in such a way that the first entry at every block number corresponds to the
531    /// canonical version of the blockchain at any point in time.
532    ///
533    /// A position withing this data structure is called "block offset". This is an ephemeral value
534    /// and changes as new best blocks are added. Blocks at the same height are collectively called
535    /// "block forks" and the position of the block within the same block height is called
536    /// "fork offset". While fork offset `0` always corresponds to the canonical version of the
537    /// blockchain, other offsets are not guaranteed to follow any particular ordering rules.
538    blocks: VecDeque<SmallVec<[ClientDatabaseBlock<Block>; 2]>>,
539}
540
541#[derive(Debug)]
542struct SegmentHeadersCache {
543    segment_headers_cache: Vec<SegmentHeader>,
544}
545
546impl SegmentHeadersCache {
547    #[inline(always)]
548    fn last_segment_header(&self) -> Option<SegmentHeader> {
549        self.segment_headers_cache.last().copied()
550    }
551
552    #[inline(always)]
553    fn max_local_segment_index(&self) -> Option<LocalSegmentIndex> {
554        self.segment_headers_cache
555            .last()
556            .map(|segment_header| segment_header.index.as_inner())
557    }
558
559    #[inline(always)]
560    fn get_segment_header(&self, local_segment_index: LocalSegmentIndex) -> Option<SegmentHeader> {
561        self.segment_headers_cache
562            .get(u64::from(local_segment_index) as usize)
563            .copied()
564    }
565
566    /// Returns actually added segments (some might have been skipped)
567    fn add_segment_headers(
568        &mut self,
569        mut segment_headers: Vec<SegmentHeader>,
570    ) -> Result<Vec<SegmentHeader>, PersistSegmentHeadersError> {
571        self.segment_headers_cache.reserve(segment_headers.len());
572
573        let mut maybe_last_local_segment_index = self.max_local_segment_index();
574
575        if let Some(last_segment_index) = maybe_last_local_segment_index {
576            // Skip already stored segment headers
577            segment_headers
578                .retain(|segment_header| segment_header.index.as_inner() > last_segment_index);
579        }
580
581        // Check all input segment headers to see which ones are not stored yet and verifying that
582        // segment indices are monotonically increasing
583        for segment_header in segment_headers.iter().copied() {
584            let local_segment_index = segment_header.index.as_inner();
585            if let Some(last_local_segment_index) = maybe_last_local_segment_index {
586                if local_segment_index != last_local_segment_index + LocalSegmentIndex::ONE {
587                    return Err(PersistSegmentHeadersError::MustFollowLastSegmentIndex {
588                        local_segment_index,
589                        last_local_segment_index,
590                    });
591                }
592
593                self.segment_headers_cache.push(segment_header);
594                maybe_last_local_segment_index.replace(local_segment_index);
595            } else {
596                if local_segment_index != LocalSegmentIndex::ZERO {
597                    return Err(PersistSegmentHeadersError::FirstSegmentIndexZero {
598                        local_segment_index,
599                    });
600                }
601
602                self.segment_headers_cache.push(segment_header);
603                maybe_last_local_segment_index.replace(local_segment_index);
604            }
605        }
606
607        Ok(segment_headers)
608    }
609}
610
611#[derive(Debug)]
612struct SuperSegmentHeadersCache {
613    super_segment_headers_cache: Vec<SuperSegmentHeader>,
614}
615
616impl SuperSegmentHeadersCache {
617    #[inline(always)]
618    fn last_super_segment_header(&self) -> Option<SuperSegmentHeader> {
619        self.super_segment_headers_cache.last().copied()
620    }
621
622    #[inline]
623    fn previous_super_segment_header(
624        &self,
625        target_block_number: BlockNumber,
626    ) -> Option<SuperSegmentHeader> {
627        let block_number = target_block_number.checked_sub(BlockNumber::ONE)?;
628        let index = match self.super_segment_headers_cache.binary_search_by_key(
629            &block_number,
630            |super_segment_header| {
631                super_segment_header
632                    .target_beacon_chain_block_number
633                    .as_inner()
634            },
635        ) {
636            Ok(found_index) => found_index,
637            Err(insert_index) => insert_index.checked_sub(1)?,
638        };
639
640        self.super_segment_headers_cache.get(index).copied()
641    }
642
643    #[inline(always)]
644    fn get_super_segment_header(
645        &self,
646        local_segment_index: SuperSegmentIndex,
647    ) -> Option<SuperSegmentHeader> {
648        self.super_segment_headers_cache
649            .get(u64::from(local_segment_index) as usize)
650            .copied()
651    }
652
653    #[inline(always)]
654    fn get_super_segment_header_for_segment_index(
655        &self,
656        segment_index: SegmentIndex,
657    ) -> Option<SuperSegmentHeader> {
658        let index = self
659            .super_segment_headers_cache
660            .binary_search_by_key(&segment_index, |super_segment_header| {
661                super_segment_header.max_segment_index.as_inner()
662            })
663            .unwrap_or_else(|insert_index| insert_index);
664
665        let super_segment_header = self.super_segment_headers_cache.get(index).copied()?;
666
667        let max_segment_index = super_segment_header.max_segment_index.as_inner();
668        let first_segment_index = max_segment_index
669            - SegmentIndex::from(u64::from(super_segment_header.num_segments))
670            + SegmentIndex::ONE;
671
672        (first_segment_index..=max_segment_index)
673            .contains(&segment_index)
674            .then_some(super_segment_header)
675    }
676
677    /// Returns actually added super segments (some might have been skipped)
678    fn add_super_segment_headers(
679        &mut self,
680        mut super_segment_headers: Vec<SuperSegmentHeader>,
681    ) -> Result<Vec<SuperSegmentHeader>, PersistSuperSegmentHeadersError> {
682        self.super_segment_headers_cache
683            .reserve(super_segment_headers.len());
684
685        let mut maybe_last_super_segment_index = self
686            .super_segment_headers_cache
687            .last()
688            .map(|header| header.index.as_inner());
689
690        if let Some(last_super_segment_index) = maybe_last_super_segment_index {
691            // Skip already stored super segment headers
692            super_segment_headers.retain(|super_segment_header| {
693                super_segment_header.index.as_inner() > last_super_segment_index
694            });
695        }
696
697        // Check all input super segment headers to see which ones are not stored yet and verifying
698        // that super segment indices are monotonically increasing
699        for super_segment_header in super_segment_headers.iter().copied() {
700            let super_segment_index = super_segment_header.index.as_inner();
701            if let Some(last_super_segment_index) = maybe_last_super_segment_index {
702                if super_segment_index != last_super_segment_index + SuperSegmentIndex::ONE {
703                    return Err(
704                        PersistSuperSegmentHeadersError::MustFollowLastSegmentIndex {
705                            super_segment_index,
706                            last_super_segment_index,
707                        },
708                    );
709                }
710
711                self.super_segment_headers_cache.push(super_segment_header);
712                maybe_last_super_segment_index.replace(super_segment_index);
713            } else {
714                if super_segment_index != SuperSegmentIndex::ZERO {
715                    return Err(PersistSuperSegmentHeadersError::FirstSegmentIndexZero {
716                        super_segment_index,
717                    });
718                }
719
720                self.super_segment_headers_cache.push(super_segment_header);
721                maybe_last_super_segment_index.replace(super_segment_index);
722            }
723        }
724
725        Ok(super_segment_headers)
726    }
727}
728
729// TODO: Hide implementation details
730#[derive(Debug)]
731struct State<Block, StorageBackend>
732where
733    Block: GenericOwnedBlock,
734{
735    data: StateData<Block>,
736    segment_headers_cache: SegmentHeadersCache,
737    super_segment_headers_cache: SuperSegmentHeadersCache,
738    storage_backend_adapter: AsyncRwLock<StorageBackendAdapter<StorageBackend>>,
739}
740
741impl<Block, StorageBackend> State<Block, StorageBackend>
742where
743    Block: GenericOwnedBlock,
744{
745    #[inline(always)]
746    fn best_tip(&self) -> &ForkTip {
747        self.data
748            .fork_tips
749            .front()
750            .expect("The best block is always present; qed")
751    }
752
753    #[inline(always)]
754    fn best_block(&self) -> &ClientDatabaseBlock<Block> {
755        self.data
756            .blocks
757            .front()
758            .expect("The best block is always present; qed")
759            .first()
760            .expect("The best block is always present; qed")
761    }
762}
763
764#[derive(Debug)]
765struct BlockToPersist<'a, Block>
766where
767    Block: GenericOwnedBlock,
768{
769    block_offset: usize,
770    fork_offset: usize,
771    block: &'a Block,
772    block_details: &'a BlockDetails,
773}
774
775#[derive(Debug)]
776struct PersistedBlock {
777    block_offset: usize,
778    fork_offset: usize,
779    write_location: WriteLocation,
780}
781
782#[derive(Debug)]
783struct ClientDatabaseInnerOptions {
784    block_confirmation_depth: BlockNumber,
785    soft_confirmation_depth: BlockNumber,
786    max_fork_tips: NonZeroUsize,
787    max_fork_tip_distance: BlockNumber,
788}
789
790#[derive(Debug)]
791struct Inner<Block, StorageBackend>
792where
793    Block: GenericOwnedBlock,
794{
795    state: AsyncRwLock<State<Block, StorageBackend>>,
796    options: ClientDatabaseInnerOptions,
797}
798
799/// Client database
800#[derive(Debug)]
801pub struct ClientDatabase<Block, StorageBackend>
802where
803    Block: GenericOwnedBlock,
804{
805    inner: Arc<Inner<Block, StorageBackend>>,
806}
807
808impl<Block, StorageBackend> Clone for ClientDatabase<Block, StorageBackend>
809where
810    Block: GenericOwnedBlock,
811{
812    fn clone(&self) -> Self {
813        Self {
814            inner: self.inner.clone(),
815        }
816    }
817}
818
819#[expect(clippy::empty_drop, reason = "Not implemented yet")]
820impl<Block, StorageBackend> Drop for ClientDatabase<Block, StorageBackend>
821where
822    Block: GenericOwnedBlock,
823{
824    fn drop(&mut self) {
825        // TODO: Persist things that were not persisted yet to reduce the data loss on shutdown
826    }
827}
828
829impl<Block, StorageBackend> ChainInfo<Block> for ClientDatabase<Block, StorageBackend>
830where
831    Block: GenericOwnedBlock,
832    StorageBackend: ClientDatabaseStorageBackend,
833{
834    #[inline]
835    fn best_root(&self) -> BlockRoot {
836        // Blocking read lock is fine because where a write lock is only taken for a short time and
837        // most locks are read locks
838        self.inner.state.read_blocking().best_tip().root
839    }
840
841    #[inline]
842    fn best_header(&self) -> Block::Header {
843        // Blocking read lock is fine because where a write lock is only taken for a short time and
844        // most locks are read locks
845        self.inner
846            .state
847            .read_blocking()
848            .best_block()
849            .header()
850            .clone()
851    }
852
853    #[inline]
854    fn best_header_with_details(&self) -> (Block::Header, BlockDetails) {
855        // Blocking read lock is fine because where a write lock is only taken for a short time and
856        // most locks are read locks
857        let state = self.inner.state.read_blocking();
858        let best_block = state.best_block();
859        (
860            best_block.header().clone(),
861            best_block
862                .block_details()
863                .expect("Always present for the best block; qed")
864                .clone(),
865        )
866    }
867
868    // TODO: Add fast path when `descendant_block_root` is the best block
869    #[inline]
870    fn ancestor_header(
871        &self,
872        ancestor_block_number: BlockNumber,
873        descendant_block_root: &BlockRoot,
874    ) -> Option<Block::Header> {
875        // Blocking read lock is fine because where a write lock is only taken for a short time and
876        // most locks are read locks
877        let state = self.inner.state.read_blocking();
878        let best_number = state.best_tip().number;
879
880        let ancestor_block_offset =
881            u64::from(best_number.checked_sub(ancestor_block_number)?) as usize;
882        let ancestor_block_candidates = state.data.blocks.get(ancestor_block_offset)?;
883
884        let descendant_block_number = *state.data.block_roots.get(descendant_block_root)?;
885        if ancestor_block_number > descendant_block_number {
886            return None;
887        }
888        let descendant_block_offset =
889            u64::from(best_number.checked_sub(descendant_block_number)?) as usize;
890
891        // Range of blocks where the first item is expected to contain a descendant
892        let mut blocks_range_iter = state
893            .data
894            .blocks
895            .iter()
896            .enumerate()
897            .skip(descendant_block_offset);
898
899        let (_offset, descendant_block_candidates) = blocks_range_iter.next()?;
900        let descendant_header = descendant_block_candidates
901            .iter()
902            .find(|block| &*block.header().header().root() == descendant_block_root)?
903            .header()
904            .header();
905
906        // If there are no forks at this level, then this is the canonical chain and ancestor
907        // block number we're looking for is the first block at the corresponding block number.
908        // Similarly, if there is just a single ancestor candidate and descendant exists, it must be
909        // the one we care about.
910        if descendant_block_candidates.len() == 1 || ancestor_block_candidates.len() == 1 {
911            return ancestor_block_candidates
912                .iter()
913                .next()
914                .map(|block| block.header().clone());
915        }
916
917        let mut parent_block_root = &descendant_header.prefix.parent_root;
918
919        // Iterate over the blocks following descendant until ancestor is reached
920        for (block_offset, parent_candidates) in blocks_range_iter {
921            let parent_header = parent_candidates
922                .iter()
923                .find(|header| &*header.header().header().root() == parent_block_root)?
924                .header();
925
926            // When header offset matches, we found the header
927            if block_offset == ancestor_block_offset {
928                return Some(parent_header.clone());
929            }
930
931            parent_block_root = &parent_header.header().prefix.parent_root;
932        }
933
934        None
935    }
936
937    #[inline]
938    fn header(&self, block_root: &BlockRoot) -> Option<Block::Header> {
939        // Blocking read lock is fine because where a write lock is only taken for a short time and
940        // most locks are read locks
941        let state = self.inner.state.read_blocking();
942        let best_number = state.best_tip().number;
943
944        let block_number = *state.data.block_roots.get(block_root)?;
945        let block_offset = u64::from(best_number.checked_sub(block_number)?) as usize;
946        let block_candidates = state.data.blocks.get(block_offset)?;
947
948        block_candidates.iter().find_map(|block| {
949            let header = block.header();
950
951            if &*header.header().root() == block_root {
952                Some(header.clone())
953            } else {
954                None
955            }
956        })
957    }
958
959    #[inline]
960    fn header_with_details(&self, block_root: &BlockRoot) -> Option<(Block::Header, BlockDetails)> {
961        // Blocking read lock is fine because where a write lock is only taken for a short time and
962        // most locks are read locks
963        let state = self.inner.state.read_blocking();
964        let best_number = state.best_tip().number;
965
966        let block_number = *state.data.block_roots.get(block_root)?;
967        let block_offset = u64::from(best_number.checked_sub(block_number)?) as usize;
968        let block_candidates = state.data.blocks.get(block_offset)?;
969
970        block_candidates.iter().find_map(|block| {
971            let header = block.header();
972            let block_details = block.block_details().cloned()?;
973
974            if &*header.header().root() == block_root {
975                Some((header.clone(), block_details))
976            } else {
977                None
978            }
979        })
980    }
981
982    #[inline]
983    async fn block(&self, block_root: &BlockRoot) -> Result<Block, ReadBlockError> {
984        let state = self.inner.state.read().await;
985        let best_number = state.best_tip().number;
986
987        let block_number = *state
988            .data
989            .block_roots
990            .get(block_root)
991            .ok_or(ReadBlockError::UnknownBlockRoot)?;
992        let block_offset = u64::from(
993            best_number
994                .checked_sub(block_number)
995                .expect("Known block roots always have valid block offset; qed"),
996        ) as usize;
997        let block_candidates = state
998            .data
999            .blocks
1000            .get(block_offset)
1001            .expect("Valid block offsets always have block entries; qed");
1002
1003        for block_candidate in block_candidates {
1004            let header = block_candidate.header();
1005
1006            if &*header.header().root() == block_root {
1007                return match block_candidate.full_block() {
1008                    FullBlock::InMemory(block) => Ok(block.clone()),
1009                    FullBlock::Persisted {
1010                        header,
1011                        write_location,
1012                    } => {
1013                        let storage_backend_adapter = state.storage_backend_adapter.read().await;
1014
1015                        let storage_item = storage_backend_adapter
1016                            .read_storage_item::<StorageItemTemporary>(write_location)
1017                            .await?;
1018
1019                        let storage_item_block = match storage_item {
1020                            StorageItemTemporary::Block(storage_item_block) => storage_item_block,
1021                            StorageItemTemporary::SegmentHeaders(_) => {
1022                                return Err(ReadBlockError::StorageItemReadError {
1023                                    error: io::Error::other(
1024                                        "Unexpected storage item: `SegmentHeaders`",
1025                                    ),
1026                                });
1027                            }
1028                            StorageItemTemporary::SuperSegmentHeaders(_) => {
1029                                return Err(ReadBlockError::StorageItemReadError {
1030                                    error: io::Error::other(
1031                                        "Unexpected storage item: `SuperSegmentHeaders`",
1032                                    ),
1033                                });
1034                            }
1035                        };
1036
1037                        let StorageItemTemporaryBlock {
1038                            header: _,
1039                            body,
1040                            mmr_with_block: _,
1041                            system_contract_states: _,
1042                        } = storage_item_block;
1043
1044                        Block::from_buffers(header.buffer().clone(), body)
1045                            .ok_or(ReadBlockError::FailedToDecode)
1046                    }
1047                };
1048            }
1049        }
1050
1051        unreachable!("Known block root always has block candidate associated with it; qed")
1052    }
1053
1054    #[inline]
1055    fn last_segment_header(&self) -> Option<SegmentHeader> {
1056        // Blocking read lock is fine because where a write lock is only taken for a short time and
1057        // most locks are read locks
1058        let state = self.inner.state.read_blocking();
1059        state.segment_headers_cache.last_segment_header()
1060    }
1061
1062    #[inline]
1063    fn get_segment_header(&self, segment_index: LocalSegmentIndex) -> Option<SegmentHeader> {
1064        // Blocking read lock is fine because where a write lock is only taken for a short time and
1065        // most locks are read locks
1066        let state = self.inner.state.read_blocking();
1067
1068        state
1069            .segment_headers_cache
1070            .get_segment_header(segment_index)
1071    }
1072
1073    fn segment_headers_for_block(&self, block_number: BlockNumber) -> Vec<SegmentHeader> {
1074        // Blocking read lock is fine because where a write lock is only taken for a short time and
1075        // most locks are read locks
1076        let state = self.inner.state.read_blocking();
1077
1078        let Some(last_local_segment_index) = state.segment_headers_cache.max_local_segment_index()
1079        else {
1080            // Not initialized
1081            return Vec::new();
1082        };
1083
1084        // Special case for the initial segment (for beacon chain genesis block)
1085        if Block::Block::SHARD_KIND == RealShardKind::BeaconChain
1086            && block_number == BlockNumber::ONE
1087        {
1088            // If there is a segment index present, and we store monotonically increasing segment
1089            // headers, then the first header exists
1090            return vec![
1091                state
1092                    .segment_headers_cache
1093                    .get_segment_header(LocalSegmentIndex::ZERO)
1094                    .expect("Segment headers are stored in monotonically increasing order; qed"),
1095            ];
1096        }
1097
1098        if last_local_segment_index == LocalSegmentIndex::ZERO {
1099            // Genesis segment already included in block #1
1100            return Vec::new();
1101        }
1102
1103        let mut current_local_segment_index = last_local_segment_index;
1104        loop {
1105            // If the current segment index present, and we store monotonically increasing segment
1106            // headers, then the current segment header exists as well.
1107            let current_segment_header = state
1108                .segment_headers_cache
1109                .get_segment_header(current_local_segment_index)
1110                .expect("Segment headers are stored in monotonically increasing order; qed");
1111
1112            // The block immediately after the archived segment adding the confirmation depth
1113            let target_block_number = current_segment_header.last_archived_block.number()
1114                + BlockNumber::ONE
1115                + self.inner.options.block_confirmation_depth;
1116            if target_block_number == block_number {
1117                let mut headers_for_block = vec![current_segment_header];
1118
1119                // Check block spanning multiple segments
1120                let last_archived_block_number = current_segment_header.last_archived_block.number;
1121                let mut local_segment_index = current_local_segment_index - LocalSegmentIndex::ONE;
1122
1123                while let Some(segment_header) = state
1124                    .segment_headers_cache
1125                    .get_segment_header(local_segment_index)
1126                {
1127                    if segment_header.last_archived_block.number == last_archived_block_number {
1128                        headers_for_block.insert(0, segment_header);
1129                        local_segment_index -= LocalSegmentIndex::ONE;
1130                    } else {
1131                        break;
1132                    }
1133                }
1134
1135                return headers_for_block;
1136            }
1137
1138            // iterate segments further
1139            if target_block_number > block_number {
1140                // no need to check the initial segment
1141                if current_local_segment_index > LocalSegmentIndex::ONE {
1142                    current_local_segment_index -= LocalSegmentIndex::ONE;
1143                } else {
1144                    break;
1145                }
1146            } else {
1147                // No segment headers required
1148                return Vec::new();
1149            }
1150        }
1151
1152        // No segment headers required
1153        Vec::new()
1154    }
1155}
1156
1157impl<Block, StorageBackend> ChainInfoWrite<Block> for ClientDatabase<Block, StorageBackend>
1158where
1159    Block: GenericOwnedBlock,
1160    StorageBackend: ClientDatabaseStorageBackend,
1161{
1162    async fn persist_block(
1163        &self,
1164        block: Block,
1165        block_details: BlockDetails,
1166    ) -> Result<(), PersistBlockError> {
1167        let mut state = self.inner.state.write().await;
1168        let best_number = state.best_tip().number;
1169
1170        let header = block.header().header();
1171
1172        let block_number = header.prefix.number;
1173
1174        if best_number == BlockNumber::ZERO && block_number != BlockNumber::ONE {
1175            // Special case when syncing on top of the fresh database
1176            Self::insert_first_block(&mut state.data, block, block_details);
1177
1178            return Ok(());
1179        }
1180
1181        if block_number == best_number + BlockNumber::ONE {
1182            return Self::insert_new_best_block(state, &self.inner, block, block_details).await;
1183        }
1184
1185        let block_offset = u64::from(
1186            best_number
1187                .checked_sub(block_number)
1188                .ok_or(PersistBlockError::MissingParent)?,
1189        ) as usize;
1190
1191        if block_offset >= u64::from(self.inner.options.block_confirmation_depth) as usize {
1192            return Err(PersistBlockError::OutsideAcceptableRange);
1193        }
1194
1195        let state = &mut *state;
1196
1197        let block_forks = state.data.blocks.get_mut(block_offset).ok_or_else(|| {
1198            error!(
1199                %block_number,
1200                %block_offset,
1201                "Failed to store block fork, header offset is missing despite being within \
1202                acceptable range"
1203            );
1204
1205            PersistBlockError::OutsideAcceptableRange
1206        })?;
1207
1208        for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
1209            // Block's parent is no longer a fork tip, remove it
1210            if fork_tip.root == header.prefix.parent_root {
1211                state.data.fork_tips.remove(index);
1212                break;
1213            }
1214        }
1215
1216        let block_root = *header.root();
1217        // Insert at position 1, which means the most recent tip, which doesn't correspond to
1218        // the best block
1219        state.data.fork_tips.insert(
1220            1,
1221            ForkTip {
1222                number: block_number,
1223                root: block_root,
1224            },
1225        );
1226        state.data.block_roots.insert(block_root, block_number);
1227        let beacon_chain_block_details = <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1228            .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1229        block_forks.push(ClientDatabaseBlock::InMemory {
1230            block,
1231            block_details,
1232            beacon_chain_block_details,
1233        });
1234
1235        Self::prune_outdated_fork_tips(block_number, &mut state.data, &self.inner.options);
1236
1237        Ok(())
1238    }
1239
1240    async fn persist_segment_headers(
1241        &self,
1242        segment_headers: Vec<SegmentHeader>,
1243    ) -> Result<(), PersistSegmentHeadersError> {
1244        let mut state = self.inner.state.write().await;
1245
1246        let added_segment_headers = state
1247            .segment_headers_cache
1248            .add_segment_headers(segment_headers)?;
1249
1250        if added_segment_headers.is_empty() {
1251            return Ok(());
1252        }
1253
1254        // Convert write lock into upgradable read lock to allow reads, while preventing segment
1255        // headers modifications
1256        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
1257        //  are satisfied. If not, blocking read locks in other places will cause issues.
1258        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1259
1260        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1261
1262        storage_backend_adapter
1263            .write_storage_item(StorageItemTemporary::SegmentHeaders(
1264                StorageItemTemporarySegmentHeaders {
1265                    segment_headers: added_segment_headers,
1266                },
1267            ))
1268            .await?;
1269
1270        Ok(())
1271    }
1272}
1273
1274impl<StorageBackend> BeaconChainInfo for ClientDatabase<OwnedBeaconChainBlock, StorageBackend>
1275where
1276    StorageBackend: ClientDatabaseStorageBackend,
1277{
1278    fn shard_segment_roots(
1279        &self,
1280        block_number: BlockNumber,
1281    ) -> Result<StdArc<[ShardSegmentRoot]>, ShardSegmentRootsError> {
1282        // Blocking read lock is fine because where a write lock is only taken for a short time and
1283        // most locks are read locks
1284        let state = self.inner.state.read_blocking();
1285        let best_number = state.best_tip().number;
1286
1287        let block_offset = u64::from(
1288            best_number
1289                .checked_sub(block_number)
1290                .ok_or(ShardSegmentRootsError::BlockMissing { block_number })?,
1291        ) as usize;
1292
1293        let block = state
1294            .data
1295            .blocks
1296            .get(block_offset)
1297            .ok_or(ShardSegmentRootsError::BlockMissing { block_number })?
1298            .first()
1299            .expect("There is always at least one block candidate; qed");
1300
1301        Ok(StdArc::clone(
1302            &block
1303                .beacon_chain_block_details()
1304                .as_ref()
1305                .expect("Always present in the beacon chain block; qed")
1306                .shard_segment_roots,
1307        ))
1308    }
1309
1310    #[inline]
1311    fn last_super_segment_header(&self) -> Option<SuperSegmentHeader> {
1312        // Blocking read lock is fine because where a write lock is only taken for a short time and
1313        // most locks are read locks
1314        let state = self.inner.state.read_blocking();
1315        state
1316            .super_segment_headers_cache
1317            .last_super_segment_header()
1318    }
1319
1320    #[inline]
1321    fn previous_super_segment_header(
1322        &self,
1323        block_number: BlockNumber,
1324    ) -> Option<SuperSegmentHeader> {
1325        // Blocking read lock is fine because where a write lock is only taken for a short time and
1326        // most locks are read locks
1327        let state = self.inner.state.read_blocking();
1328
1329        state
1330            .super_segment_headers_cache
1331            .previous_super_segment_header(block_number)
1332    }
1333
1334    #[inline]
1335    fn get_super_segment_header(
1336        &self,
1337        super_segment_index: SuperSegmentIndex,
1338    ) -> Option<SuperSegmentHeader> {
1339        // Blocking read lock is fine because where a write lock is only taken for a short time and
1340        // most locks are read locks
1341        let state = self.inner.state.read_blocking();
1342
1343        state
1344            .super_segment_headers_cache
1345            .get_super_segment_header(super_segment_index)
1346    }
1347
1348    fn get_super_segment_header_for_segment_index(
1349        &self,
1350        segment_index: SegmentIndex,
1351    ) -> Option<SuperSegmentHeader> {
1352        // Blocking read lock is fine because where a write lock is only taken for a short time and
1353        // most locks are read locks
1354        let state = self.inner.state.read_blocking();
1355
1356        state
1357            .super_segment_headers_cache
1358            .get_super_segment_header_for_segment_index(segment_index)
1359    }
1360}
1361
1362impl<StorageBackend> BeaconChainInfoWrite for ClientDatabase<OwnedBeaconChainBlock, StorageBackend>
1363where
1364    StorageBackend: ClientDatabaseStorageBackend,
1365{
1366    async fn persist_super_segment_header(
1367        &self,
1368        super_segment_header: SuperSegmentHeader,
1369    ) -> Result<bool, PersistSuperSegmentHeadersError> {
1370        let mut state = self.inner.state.write().await;
1371
1372        let added_super_segment_headers = state
1373            .super_segment_headers_cache
1374            .add_super_segment_headers(vec![super_segment_header])?;
1375
1376        if added_super_segment_headers.is_empty() {
1377            return Ok(false);
1378        }
1379
1380        // Convert write lock into upgradable read lock to allow reads, while preventing super
1381        // segment headers modifications
1382        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
1383        //  are satisfied. If not, blocking read locks in other places will cause issues.
1384        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1385
1386        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1387
1388        storage_backend_adapter
1389            .write_storage_item(StorageItemTemporary::SuperSegmentHeaders(
1390                StorageItemTemporarySuperSegmentHeaders {
1391                    super_segment_headers: added_super_segment_headers,
1392                },
1393            ))
1394            .await?;
1395
1396        Ok(true)
1397    }
1398
1399    async fn persist_super_segment_headers(
1400        &self,
1401        super_segment_headers: Vec<SuperSegmentHeader>,
1402    ) -> Result<(), PersistSuperSegmentHeadersError> {
1403        let mut state = self.inner.state.write().await;
1404
1405        let added_super_segment_headers = state
1406            .super_segment_headers_cache
1407            .add_super_segment_headers(super_segment_headers)?;
1408
1409        if added_super_segment_headers.is_empty() {
1410            return Ok(());
1411        }
1412
1413        // Convert write lock into upgradable read lock to allow reads, while preventing super
1414        // segment headers modifications
1415        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
1416        //  are satisfied. If not, blocking read locks in other places will cause issues.
1417        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1418
1419        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1420
1421        storage_backend_adapter
1422            .write_storage_item(StorageItemTemporary::SuperSegmentHeaders(
1423                StorageItemTemporarySuperSegmentHeaders {
1424                    super_segment_headers: added_super_segment_headers,
1425                },
1426            ))
1427            .await?;
1428
1429        Ok(())
1430    }
1431}
1432
1433impl<Block, StorageBackend> ClientDatabase<Block, StorageBackend>
1434where
1435    Block: GenericOwnedBlock,
1436    StorageBackend: ClientDatabaseStorageBackend,
1437{
1438    /// Open the existing database.
1439    ///
1440    /// NOTE: The database needs to be formatted with [`Self::format()`] before it can be used.
1441    pub async fn open<GBB>(
1442        options: ClientDatabaseOptions<GBB, StorageBackend>,
1443    ) -> Result<Self, ClientDatabaseError>
1444    where
1445        GBB: FnOnce() -> GenesisBlockBuilderResult<Block>,
1446    {
1447        let ClientDatabaseOptions {
1448            write_buffer_size,
1449            block_confirmation_depth,
1450            soft_confirmation_depth,
1451            max_fork_tips,
1452            max_fork_tip_distance,
1453            genesis_block_builder,
1454            storage_backend,
1455        } = options;
1456        if soft_confirmation_depth >= block_confirmation_depth {
1457            return Err(ClientDatabaseError::InvalidSoftConfirmationDepth);
1458        }
1459
1460        if max_fork_tip_distance > block_confirmation_depth {
1461            return Err(ClientDatabaseError::InvalidMaxForkTipDistance);
1462        }
1463
1464        let mut state_data = StateData {
1465            fork_tips: VecDeque::new(),
1466            block_roots: HashMap::default(),
1467            blocks: VecDeque::new(),
1468        };
1469        let mut segment_headers_cache = SegmentHeadersCache {
1470            segment_headers_cache: Vec::new(),
1471        };
1472        let mut super_segment_headers_cache = SuperSegmentHeadersCache {
1473            super_segment_headers_cache: Vec::new(),
1474        };
1475
1476        let options = ClientDatabaseInnerOptions {
1477            block_confirmation_depth,
1478            soft_confirmation_depth,
1479            max_fork_tips,
1480            max_fork_tip_distance,
1481        };
1482
1483        let storage_item_handlers = StorageItemHandlers {
1484            permanent: |_arg| {
1485                // TODO
1486                Ok(())
1487            },
1488            temporary: |arg| {
1489                let StorageItemHandlerArg {
1490                    storage_item,
1491                    page_offset,
1492                    num_pages,
1493                } = arg;
1494                let storage_item_block = match storage_item {
1495                    StorageItemTemporary::Block(storage_item_block) => storage_item_block,
1496                    StorageItemTemporary::SegmentHeaders(segment_headers) => {
1497                        let num_segment_headers = segment_headers.segment_headers.len();
1498                        return match segment_headers_cache
1499                            .add_segment_headers(segment_headers.segment_headers)
1500                        {
1501                            Ok(_) => Ok(()),
1502                            Err(error) => {
1503                                error!(
1504                                    %page_offset,
1505                                    %num_segment_headers,
1506                                    %error,
1507                                    "Failed to add segment headers from storage item"
1508                                );
1509
1510                                Err(ClientDatabaseError::InvalidSegmentHeaders { page_offset })
1511                            }
1512                        };
1513                    }
1514                    StorageItemTemporary::SuperSegmentHeaders(super_segment_headers) => {
1515                        let num_super_segment_headers =
1516                            super_segment_headers.super_segment_headers.len();
1517                        return match super_segment_headers_cache
1518                            .add_super_segment_headers(super_segment_headers.super_segment_headers)
1519                        {
1520                            Ok(_) => Ok(()),
1521                            Err(error) => {
1522                                error!(
1523                                    %page_offset,
1524                                    %num_super_segment_headers,
1525                                    %error,
1526                                    "Failed to add segment headers from storage item"
1527                                );
1528
1529                                Err(ClientDatabaseError::InvalidSegmentHeaders { page_offset })
1530                            }
1531                        };
1532                    }
1533                };
1534
1535                // TODO: It would be nice to not allocate body here since we'll not use it here
1536                //  anyway
1537                let StorageItemTemporaryBlock {
1538                    header,
1539                    body,
1540                    mmr_with_block,
1541                    system_contract_states,
1542                } = storage_item_block;
1543
1544                let header = Block::Header::from_buffer(header).map_err(|_buffer| {
1545                    error!(%page_offset, "Failed to decode block header from bytes");
1546
1547                    ClientDatabaseError::InvalidBlock { page_offset }
1548                })?;
1549                let body = Block::Body::from_buffer(body).map_err(|_buffer| {
1550                    error!(%page_offset, "Failed to decode block body from bytes");
1551
1552                    ClientDatabaseError::InvalidBlock { page_offset }
1553                })?;
1554
1555                let block_root = *header.header().root();
1556                let block_number = header.header().prefix.number;
1557
1558                state_data.block_roots.insert(block_root, block_number);
1559
1560                let maybe_best_number = state_data
1561                    .blocks
1562                    .front()
1563                    .and_then(|block_forks| block_forks.first())
1564                    .map(|best_block| {
1565                        // Type inference is not working here for some reason
1566                        let header: &Block::Header = best_block.header();
1567
1568                        header.header().prefix.number
1569                    });
1570
1571                let block_offset = if let Some(best_number) = maybe_best_number {
1572                    if block_number <= best_number {
1573                        u64::from(best_number - block_number) as usize
1574                    } else {
1575                        // The new best block must follow the previous best block
1576                        if block_number - best_number != BlockNumber::ONE {
1577                            error!(
1578                                %page_offset,
1579                                %best_number,
1580                                %block_number,
1581                                "Invalid new best block number, it must be only one block \
1582                                higher than the best block"
1583                            );
1584
1585                            return Err(ClientDatabaseError::InvalidBlock { page_offset });
1586                        }
1587
1588                        state_data.blocks.push_front(SmallVec::new());
1589                        // Will insert a new block at the front
1590                        0
1591                    }
1592                } else {
1593                    state_data.blocks.push_front(SmallVec::new());
1594                    // Will insert a new block at the front
1595                    0
1596                };
1597
1598                let Some(block_forks) = state_data.blocks.get_mut(block_offset) else {
1599                    // Ignore the older block, other blocks at its height were already pruned
1600                    // anyway
1601                    return Ok(());
1602                };
1603
1604                // Push a new block to the end of the list, we'll fix it up later
1605                let beacon_chain_block_details =
1606                    <dyn Any>::downcast_ref::<OwnedBeaconChainBody>(&body)
1607                        .map(|body| BeaconChainBlockDetails::from_body(body.body()));
1608                block_forks.push(ClientDatabaseBlock::Persisted {
1609                    header,
1610                    block_details: BlockDetails {
1611                        mmr_with_block,
1612                        system_contract_states,
1613                    },
1614                    beacon_chain_block_details,
1615                    write_location: WriteLocation {
1616                        page_offset,
1617                        num_pages,
1618                    },
1619                });
1620
1621                // If a new block was inserted, confirm a new canonical block to prune extra
1622                // in-memory information
1623                if block_offset == 0 && block_forks.len() == 1 {
1624                    Self::confirm_canonical_block(block_number, &mut state_data, &options);
1625                }
1626
1627                Ok(())
1628            },
1629        };
1630
1631        let storage_backend_adapter =
1632            StorageBackendAdapter::open(write_buffer_size, storage_item_handlers, storage_backend)
1633                .await?;
1634
1635        if let Some(best_block) = state_data.blocks.front().and_then(|block_forks| {
1636            // The best block is last in the list here because that is how it was inserted while
1637            // reading from the database
1638            block_forks.last()
1639        }) {
1640            // Type inference is not working here for some reason
1641            let header: &Block::Header = best_block.header();
1642            let header = header.header();
1643            let block_number = header.prefix.number;
1644            let block_root = *header.root();
1645
1646            if !Self::adjust_ancestor_block_forks(&mut state_data.blocks, block_root) {
1647                return Err(ClientDatabaseError::FailedToAdjustAncestorBlockForks);
1648            }
1649
1650            // Store the best block as the first and only fork tip
1651            state_data.fork_tips.push_front(ForkTip {
1652                number: block_number,
1653                root: block_root,
1654            });
1655        } else {
1656            let GenesisBlockBuilderResult {
1657                block,
1658                system_contract_states,
1659            } = genesis_block_builder();
1660
1661            // If the database is empty, initialize everything with the genesis block
1662            let header = block.header().header();
1663            let block_number = header.prefix.number;
1664            let block_root = *header.root();
1665
1666            state_data.fork_tips.push_front(ForkTip {
1667                number: block_number,
1668                root: block_root,
1669            });
1670            state_data.block_roots.insert(block_root, block_number);
1671            let beacon_chain_block_details =
1672                <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1673                    .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1674            state_data
1675                .blocks
1676                .push_front(smallvec![ClientDatabaseBlock::InMemory {
1677                    block,
1678                    block_details: BlockDetails {
1679                        system_contract_states,
1680                        mmr_with_block: Arc::new({
1681                            let mut mmr = BlockMerkleMountainRange::new();
1682                            mmr.add_leaf(&block_root);
1683                            mmr
1684                        })
1685                    },
1686                    beacon_chain_block_details,
1687                }]);
1688        }
1689
1690        let state = State {
1691            data: state_data,
1692            segment_headers_cache,
1693            super_segment_headers_cache,
1694            storage_backend_adapter: AsyncRwLock::new(storage_backend_adapter),
1695        };
1696
1697        let inner = Inner {
1698            state: AsyncRwLock::new(state),
1699            options,
1700        };
1701
1702        Ok(Self {
1703            inner: Arc::new(inner),
1704        })
1705    }
1706
1707    /// Format a new database
1708    pub async fn format(
1709        storage_backend: &StorageBackend,
1710        options: ClientDatabaseFormatOptions,
1711    ) -> Result<(), ClientDatabaseFormatError> {
1712        StorageBackendAdapter::format(storage_backend, options).await
1713    }
1714
1715    fn insert_first_block(state: &mut StateData<Block>, block: Block, block_details: BlockDetails) {
1716        // If the database is empty, initialize everything with the genesis block
1717        let header = block.header().header();
1718        let block_number = header.prefix.number;
1719        let block_root = *header.root();
1720
1721        state.fork_tips.clear();
1722        state.fork_tips.push_front(ForkTip {
1723            number: block_number,
1724            root: block_root,
1725        });
1726        state.block_roots.clear();
1727        state.block_roots.insert(block_root, block_number);
1728        state.blocks.clear();
1729        let beacon_chain_block_details = <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1730            .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1731        state
1732            .blocks
1733            .push_front(smallvec![ClientDatabaseBlock::InMemory {
1734                block,
1735                block_details,
1736                beacon_chain_block_details,
1737            }]);
1738    }
1739
1740    async fn insert_new_best_block(
1741        mut state: AsyncRwLockWriteGuard<'_, State<Block, StorageBackend>>,
1742        inner: &Inner<Block, StorageBackend>,
1743        block: Block,
1744        block_details: BlockDetails,
1745    ) -> Result<(), PersistBlockError> {
1746        let header = block.header().header();
1747        let block_number = header.prefix.number;
1748        let block_root = *header.root();
1749        let parent_root = header.prefix.parent_root;
1750
1751        // Adjust the relative order of forks to ensure the first index always corresponds to
1752        // ancestors of the new best block
1753        if !Self::adjust_ancestor_block_forks(&mut state.data.blocks, parent_root) {
1754            return Err(PersistBlockError::MissingParent);
1755        }
1756
1757        // Store new block in the state
1758        {
1759            for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
1760                // Block's parent is no longer a fork tip, remove it
1761                if fork_tip.root == parent_root {
1762                    state.data.fork_tips.remove(index);
1763                    break;
1764                }
1765            }
1766
1767            state.data.fork_tips.push_front(ForkTip {
1768                number: block_number,
1769                root: block_root,
1770            });
1771            state.data.block_roots.insert(block_root, block_number);
1772            let beacon_chain_block_details =
1773                <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1774                    .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1775            state
1776                .data
1777                .blocks
1778                .push_front(smallvec![ClientDatabaseBlock::InMemory {
1779                    block,
1780                    block_details: block_details.clone(),
1781                    beacon_chain_block_details,
1782                }]);
1783        }
1784
1785        let options = &inner.options;
1786
1787        Self::confirm_canonical_block(block_number, &mut state.data, options);
1788        Self::prune_outdated_fork_tips(block_number, &mut state.data, options);
1789
1790        // Convert write lock into upgradable read lock to allow reads, while preventing concurrent
1791        // block modifications
1792        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
1793        //  are satisfied. If not, blocking read locks in other places will cause issues.
1794        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1795
1796        let mut blocks_to_persist = Vec::new();
1797        for block_offset in u64::from(options.soft_confirmation_depth) as usize.. {
1798            let Some(fork_blocks) = state.data.blocks.get(block_offset) else {
1799                break;
1800            };
1801
1802            let len_before = blocks_to_persist.len();
1803            fork_blocks
1804                .iter()
1805                .enumerate()
1806                .filter_map(|(fork_offset, client_database_block)| {
1807                    match client_database_block {
1808                        ClientDatabaseBlock::InMemory {
1809                            block,
1810                            block_details,
1811                            beacon_chain_block_details: _,
1812                        } => Some(BlockToPersist {
1813                            block_offset,
1814                            fork_offset,
1815                            block,
1816                            block_details,
1817                        }),
1818                        ClientDatabaseBlock::Persisted { .. }
1819                        | ClientDatabaseBlock::PersistedConfirmed { .. } => {
1820                            // Already persisted
1821                            None
1822                        }
1823                    }
1824                })
1825                .collect_into(&mut blocks_to_persist);
1826
1827            if blocks_to_persist.len() == len_before {
1828                break;
1829            }
1830        }
1831
1832        // Persist blocks from older to newer
1833        let mut persisted_blocks = Vec::with_capacity(blocks_to_persist.len());
1834        {
1835            let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1836
1837            for block_to_persist in blocks_to_persist.into_iter().rev() {
1838                let BlockToPersist {
1839                    block_offset,
1840                    fork_offset,
1841                    block,
1842                    block_details,
1843                } = block_to_persist;
1844
1845                let write_location = storage_backend_adapter
1846                    .write_storage_item(StorageItemTemporary::Block(StorageItemTemporaryBlock {
1847                        header: block.header().buffer().clone(),
1848                        body: block.body().buffer().clone(),
1849                        mmr_with_block: Arc::clone(&block_details.mmr_with_block),
1850                        system_contract_states: StdArc::clone(
1851                            &block_details.system_contract_states,
1852                        ),
1853                    }))
1854                    .await?;
1855
1856                persisted_blocks.push(PersistedBlock {
1857                    block_offset,
1858                    fork_offset,
1859                    write_location,
1860                });
1861            }
1862        }
1863
1864        // Convert blocks to persisted
1865        let mut state = RwLockUpgradableReadGuard::upgrade(state).await;
1866        for persisted_block in persisted_blocks {
1867            let PersistedBlock {
1868                block_offset,
1869                fork_offset,
1870                write_location,
1871            } = persisted_block;
1872
1873            let block = state
1874                .data
1875                .blocks
1876                .get_mut(block_offset)
1877                .expect("Still holding the same lock since last check; qed")
1878                .get_mut(fork_offset)
1879                .expect("Still holding the same lock since last check; qed");
1880
1881            replace_with_or_abort(block, |block| {
1882                if let ClientDatabaseBlock::InMemory {
1883                    block,
1884                    block_details,
1885                    beacon_chain_block_details,
1886                } = block
1887                {
1888                    let (header, _body) = block.split();
1889
1890                    ClientDatabaseBlock::Persisted {
1891                        header,
1892                        block_details,
1893                        beacon_chain_block_details,
1894                        write_location,
1895                    }
1896                } else {
1897                    unreachable!("Still holding the same lock since last check; qed");
1898                }
1899            });
1900        }
1901
1902        // TODO: Prune blocks that are no longer necessary
1903        // TODO: Prune unused page groups here or elsewhere?
1904
1905        Ok(())
1906    }
1907
1908    /// Adjust the relative order of forks to ensure the first index always corresponds to
1909    /// `parent_block_root` and its ancestors.
1910    ///
1911    /// Returns `true` on success and `false` if one of the parents was not found.
1912    #[must_use]
1913    fn adjust_ancestor_block_forks(
1914        blocks: &mut VecDeque<SmallVec<[ClientDatabaseBlock<Block>; 2]>>,
1915        mut parent_block_root: BlockRoot,
1916    ) -> bool {
1917        let mut ancestor_blocks = blocks.iter_mut();
1918
1919        loop {
1920            if ancestor_blocks.len() == 1 {
1921                // Nothing left to adjust with a single fork
1922                break;
1923            }
1924
1925            let Some(parent_blocks) = ancestor_blocks.next() else {
1926                // No more parent headers present
1927                break;
1928            };
1929
1930            let Some(fork_offset_parent_block_root) =
1931                parent_blocks
1932                    .iter()
1933                    .enumerate()
1934                    .find_map(|(fork_offset, fork_block)| {
1935                        let fork_header = fork_block.header().header();
1936                        if *fork_header.root() == parent_block_root {
1937                            Some((fork_offset, fork_header.prefix.parent_root))
1938                        } else {
1939                            None
1940                        }
1941                    })
1942            else {
1943                return false;
1944            };
1945
1946            let fork_offset;
1947            (fork_offset, parent_block_root) = fork_offset_parent_block_root;
1948
1949            parent_blocks.swap(0, fork_offset);
1950        }
1951
1952        true
1953    }
1954
1955    /// Prune outdated fork tips that are too deep and have not been updated for a long time.
1956    ///
1957    /// Note that actual headers, blocks and MMRs could remain if they are currently used by
1958    /// something or were already persisted on disk. With persisted blocks specifically, RAM usage
1959    /// implications are minimal, and we wouldn't want to re-download already stored blocks in case
1960    /// they end up being necessary later.
1961    fn prune_outdated_fork_tips(
1962        best_number: BlockNumber,
1963        state: &mut StateData<Block>,
1964        options: &ClientDatabaseInnerOptions,
1965    ) {
1966        let state = &mut *state;
1967
1968        // These forks are just candidates because they will not be pruned if the reference count is
1969        // not 1, indicating they are still in use by something
1970        let mut candidate_forks_to_remove = Vec::with_capacity(options.max_fork_tips.get());
1971
1972        // Prune forks that are too far away from the best block
1973        state.fork_tips.retain(|fork_tip| {
1974            if best_number - fork_tip.number > options.max_fork_tip_distance {
1975                candidate_forks_to_remove.push(*fork_tip);
1976                false
1977            } else {
1978                true
1979            }
1980        });
1981        // Prune forks that exceed the maximum number of forks
1982        if state.fork_tips.len() > options.max_fork_tips.get() {
1983            state
1984                .fork_tips
1985                .drain(options.max_fork_tips.get()..)
1986                .collect_into(&mut candidate_forks_to_remove);
1987        }
1988
1989        // Prune all possible candidates
1990        candidate_forks_to_remove
1991            .retain(|fork_tip| !Self::prune_outdated_fork(best_number, fork_tip, state));
1992        // Return those that were not pruned back to the list of tips
1993        state.fork_tips.extend(candidate_forks_to_remove);
1994    }
1995
1996    /// Returns `true` if the tip was pruned successfully and `false` if it should be returned to
1997    /// the list of fork tips
1998    #[must_use]
1999    fn prune_outdated_fork(
2000        best_number: BlockNumber,
2001        fork_tip: &ForkTip,
2002        state: &mut StateData<Block>,
2003    ) -> bool {
2004        let block_offset = u64::from(best_number - fork_tip.number) as usize;
2005
2006        // Prune fork top and all its ancestors that are not used
2007        let mut block_root_to_prune = fork_tip.root;
2008        let mut pruned_tip = false;
2009        for block_offset in block_offset.. {
2010            let Some(fork_blocks) = state.blocks.get_mut(block_offset) else {
2011                if !pruned_tip {
2012                    error!(
2013                        %best_number,
2014                        ?fork_tip,
2015                        block_offset,
2016                        "Block offset was not present in the database, this is an implementation \
2017                        bug #1"
2018                    );
2019                }
2020                // No forks left to prune
2021                break;
2022            };
2023
2024            if fork_blocks.len() == 1 {
2025                if !pruned_tip {
2026                    error!(
2027                        %best_number,
2028                        ?fork_tip,
2029                        block_offset,
2030                        "Block offset was not present in the database, this is an implementation \
2031                        bug #2"
2032                    );
2033                }
2034
2035                // No forks left to prune
2036                break;
2037            }
2038
2039            let Some((fork_offset, block)) = fork_blocks
2040                .iter()
2041                .enumerate()
2042                // Skip ancestor of the best block, it is certainly not a fork to be pruned
2043                .skip(1)
2044                .find(|(_fork_offset, block)| {
2045                    *block.header().header().root() == block_root_to_prune
2046                })
2047            else {
2048                if !pruned_tip {
2049                    error!(
2050                        %best_number,
2051                        ?fork_tip,
2052                        block_offset,
2053                        "Block offset was not present in the database, this is an implementation \
2054                        bug #3"
2055                    );
2056                }
2057
2058                // Nothing left to prune
2059                break;
2060            };
2061
2062            // More than one instance means something somewhere is using or depends on this block
2063            if block.header().ref_count() > 1 {
2064                break;
2065            }
2066
2067            // Blocks that are already persisted
2068            match block {
2069                ClientDatabaseBlock::InMemory { .. } => {
2070                    // Prune
2071                }
2072                ClientDatabaseBlock::Persisted { .. }
2073                | ClientDatabaseBlock::PersistedConfirmed { .. } => {
2074                    // Already on disk, keep it in memory for later, but prune the tip
2075                    pruned_tip = true;
2076                    break;
2077                }
2078            }
2079
2080            state.block_roots.get_mut(&block_root_to_prune);
2081            block_root_to_prune = block.header().header().prefix.parent_root;
2082            fork_blocks.swap_remove(fork_offset);
2083
2084            pruned_tip = true;
2085        }
2086
2087        pruned_tip
2088    }
2089
2090    /// Confirm a block at confirmation depth k and prune any other blocks at the same depth with
2091    /// their descendants
2092    fn confirm_canonical_block(
2093        best_number: BlockNumber,
2094        state_data: &mut StateData<Block>,
2095        options: &ClientDatabaseInnerOptions,
2096    ) {
2097        // `+1` means it effectively confirms parent blocks instead. This is done to keep the parent
2098        // of the confirmed block with its MMR in memory due to confirmed blocks not storing their
2099        // MMRs, which might be needed for reorgs at the lowest possible depth.
2100        let block_offset = u64::from(options.block_confirmation_depth + BlockNumber::ONE) as usize;
2101
2102        let Some(fork_blocks) = state_data.blocks.get_mut(block_offset) else {
2103            // Nothing to confirm yet
2104            return;
2105        };
2106
2107        // Mark the canonical block as confirmed
2108        {
2109            let Some(canonical_block) = fork_blocks.first_mut() else {
2110                error!(
2111                    %best_number,
2112                    block_offset,
2113                    "Have not found a canonical block to confirm, this is an implementation bug"
2114                );
2115                return;
2116            };
2117
2118            replace_with_or_abort(canonical_block, |block| match block {
2119                ClientDatabaseBlock::InMemory { .. } => {
2120                    error!(
2121                        %best_number,
2122                        block_offset,
2123                        header = ?block.header(),
2124                        "Block to be confirmed must not be in memory, this is an implementation bug"
2125                    );
2126                    block
2127                }
2128                ClientDatabaseBlock::Persisted {
2129                    header,
2130                    block_details: _,
2131                    beacon_chain_block_details,
2132                    write_location,
2133                } => ClientDatabaseBlock::PersistedConfirmed {
2134                    header,
2135                    beacon_chain_block_details,
2136                    write_location,
2137                },
2138                ClientDatabaseBlock::PersistedConfirmed { .. } => {
2139                    error!(
2140                        %best_number,
2141                        block_offset,
2142                        header = ?block.header(),
2143                        "Block to be confirmed must not be confirmed yet, this is an \
2144                        implementation bug"
2145                    );
2146                    block
2147                }
2148            });
2149        }
2150
2151        // Prune the rest of the blocks and their descendants
2152        let mut block_roots_to_prune = fork_blocks
2153            .drain(1..)
2154            .map(|block| *block.header().header().root())
2155            .collect::<Vec<_>>();
2156        let mut current_block_offset = block_offset;
2157        while !block_roots_to_prune.is_empty() {
2158            // Prune fork tips (if any)
2159            state_data
2160                .fork_tips
2161                .retain(|fork_tip| !block_roots_to_prune.contains(&fork_tip.root));
2162
2163            // Prune removed block roots
2164            for block_root in &block_roots_to_prune {
2165                state_data.block_roots.remove(block_root);
2166            }
2167
2168            // Block offset for direct descendants
2169            if let Some(next_block_offset) = current_block_offset.checked_sub(1) {
2170                current_block_offset = next_block_offset;
2171            } else {
2172                // Reached the tip
2173                break;
2174            }
2175
2176            let fork_blocks = state_data
2177                .blocks
2178                .get_mut(current_block_offset)
2179                .expect("Lower block offset always exists; qed");
2180
2181            // Collect descendants of pruned blocks to prune them next
2182            block_roots_to_prune = fork_blocks
2183                .drain_filter(|block| {
2184                    let header = block.header().header();
2185
2186                    block_roots_to_prune.contains(&header.prefix.parent_root)
2187                })
2188                .map(|block| *block.header().header().root())
2189                .collect();
2190        }
2191    }
2192}