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        #[expect(
455            clippy::rest_pattern_accessible_field,
456            reason = "Do not need other fields"
457        )]
458        match self {
459            Self::InMemory { block, .. } => block.header(),
460            Self::Persisted { header, .. } | Self::PersistedConfirmed { header, .. } => header,
461        }
462    }
463
464    #[inline(always)]
465    fn full_block(&self) -> FullBlock<'_, Block> {
466        #[expect(
467            clippy::rest_pattern_accessible_field,
468            reason = "Do not need other fields"
469        )]
470        match self {
471            Self::InMemory { block, .. } => FullBlock::InMemory(block),
472            Self::Persisted {
473                header,
474                write_location,
475                ..
476            }
477            | Self::PersistedConfirmed {
478                header,
479                write_location,
480                ..
481            } => FullBlock::Persisted {
482                header,
483                write_location: *write_location,
484            },
485        }
486    }
487
488    #[inline(always)]
489    fn block_details(&self) -> Option<&BlockDetails> {
490        #[expect(
491            clippy::rest_pattern_accessible_field,
492            reason = "Do not need other fields"
493        )]
494        match self {
495            Self::InMemory { block_details, .. } | Self::Persisted { block_details, .. } => {
496                Some(block_details)
497            }
498            Self::PersistedConfirmed { .. } => None,
499        }
500    }
501
502    #[inline(always)]
503    fn beacon_chain_block_details(&self) -> Option<&BeaconChainBlockDetails> {
504        #[expect(
505            clippy::rest_pattern_accessible_field,
506            reason = "Do not need other fields"
507        )]
508        match self {
509            Self::InMemory {
510                beacon_chain_block_details,
511                ..
512            }
513            | Self::Persisted {
514                beacon_chain_block_details,
515                ..
516            }
517            | Self::PersistedConfirmed {
518                beacon_chain_block_details,
519                ..
520            } => beacon_chain_block_details.as_ref(),
521        }
522    }
523}
524
525#[derive(Debug)]
526struct StateData<Block>
527where
528    Block: GenericOwnedBlock,
529{
530    /// Tips of forks that have no descendants.
531    ///
532    /// The current best block is at the front, the rest are in the order from most recently
533    /// updated towards the front to least recently at the back.
534    fork_tips: VecDeque<ForkTip>,
535    /// Map from block root to block number.
536    ///
537    /// Is meant to be used in conjunction with `headers` and `blocks` fields, which are indexed by
538    /// block numbers.
539    block_roots: HashMap<BlockRoot, BlockNumber, BuildHasherDefault<BlockRootHasher>>,
540    /// List of blocks with the newest at the front.
541    ///
542    /// The first element of the first entry corresponds to the best block.
543    ///
544    /// It is expected that in most block numbers there will be exactly one block, some two,
545    /// anything more than that will be very rare. The list of forks for a block number is
546    /// organized in such a way that the first entry at every block number corresponds to the
547    /// canonical version of the blockchain at any point in time.
548    ///
549    /// A position withing this data structure is called "block offset". This is an ephemeral value
550    /// and changes as new best blocks are added. Blocks at the same height are collectively called
551    /// "block forks" and the position of the block within the same block height is called
552    /// "fork offset". While fork offset `0` always corresponds to the canonical version of the
553    /// blockchain, other offsets are not guaranteed to follow any particular ordering rules.
554    blocks: VecDeque<SmallVec<[ClientDatabaseBlock<Block>; 2]>>,
555}
556
557#[derive(Debug)]
558struct SegmentHeadersCache {
559    segment_headers_cache: Vec<SegmentHeader>,
560}
561
562impl SegmentHeadersCache {
563    #[inline(always)]
564    fn last_segment_header(&self) -> Option<SegmentHeader> {
565        self.segment_headers_cache.last().copied()
566    }
567
568    #[inline(always)]
569    fn max_local_segment_index(&self) -> Option<LocalSegmentIndex> {
570        self.segment_headers_cache
571            .last()
572            .map(|segment_header| segment_header.index.as_inner())
573    }
574
575    #[inline(always)]
576    fn get_segment_header(&self, local_segment_index: LocalSegmentIndex) -> Option<SegmentHeader> {
577        self.segment_headers_cache
578            .get(u64::from(local_segment_index) as usize)
579            .copied()
580    }
581
582    /// Returns actually added segments (some might have been skipped)
583    fn add_segment_headers(
584        &mut self,
585        mut segment_headers: Vec<SegmentHeader>,
586    ) -> Result<Vec<SegmentHeader>, PersistSegmentHeadersError> {
587        self.segment_headers_cache.reserve(segment_headers.len());
588
589        let mut maybe_last_local_segment_index = self.max_local_segment_index();
590
591        if let Some(last_segment_index) = maybe_last_local_segment_index {
592            // Skip already stored segment headers
593            segment_headers
594                .retain(|segment_header| segment_header.index.as_inner() > last_segment_index);
595        }
596
597        // Check all input segment headers to see which ones are not stored yet and verifying that
598        // segment indices are monotonically increasing
599        for segment_header in segment_headers.iter().copied() {
600            let local_segment_index = segment_header.index.as_inner();
601            if let Some(last_local_segment_index) = maybe_last_local_segment_index {
602                if local_segment_index != last_local_segment_index + LocalSegmentIndex::ONE {
603                    return Err(PersistSegmentHeadersError::MustFollowLastSegmentIndex {
604                        local_segment_index,
605                        last_local_segment_index,
606                    });
607                }
608
609                self.segment_headers_cache.push(segment_header);
610                maybe_last_local_segment_index.replace(local_segment_index);
611            } else {
612                if local_segment_index != LocalSegmentIndex::ZERO {
613                    return Err(PersistSegmentHeadersError::FirstSegmentIndexZero {
614                        local_segment_index,
615                    });
616                }
617
618                self.segment_headers_cache.push(segment_header);
619                maybe_last_local_segment_index.replace(local_segment_index);
620            }
621        }
622
623        Ok(segment_headers)
624    }
625}
626
627#[derive(Debug)]
628struct SuperSegmentHeadersCache {
629    super_segment_headers_cache: Vec<SuperSegmentHeader>,
630}
631
632impl SuperSegmentHeadersCache {
633    #[inline(always)]
634    fn last_super_segment_header(&self) -> Option<SuperSegmentHeader> {
635        self.super_segment_headers_cache.last().copied()
636    }
637
638    #[inline]
639    fn previous_super_segment_header(
640        &self,
641        target_block_number: BlockNumber,
642    ) -> Option<SuperSegmentHeader> {
643        let block_number = target_block_number.checked_sub(BlockNumber::ONE)?;
644        let index = match self.super_segment_headers_cache.binary_search_by_key(
645            &block_number,
646            |super_segment_header| {
647                super_segment_header
648                    .target_beacon_chain_block_number
649                    .as_inner()
650            },
651        ) {
652            Ok(found_index) => found_index,
653            Err(insert_index) => insert_index.checked_sub(1)?,
654        };
655
656        self.super_segment_headers_cache.get(index).copied()
657    }
658
659    #[inline(always)]
660    fn get_super_segment_header(
661        &self,
662        local_segment_index: SuperSegmentIndex,
663    ) -> Option<SuperSegmentHeader> {
664        self.super_segment_headers_cache
665            .get(u64::from(local_segment_index) as usize)
666            .copied()
667    }
668
669    #[inline(always)]
670    fn get_super_segment_header_for_segment_index(
671        &self,
672        segment_index: SegmentIndex,
673    ) -> Option<SuperSegmentHeader> {
674        let index = self
675            .super_segment_headers_cache
676            .binary_search_by_key(&segment_index, |super_segment_header| {
677                super_segment_header.max_segment_index.as_inner()
678            })
679            .unwrap_or_else(|insert_index| insert_index);
680
681        let super_segment_header = self.super_segment_headers_cache.get(index).copied()?;
682
683        let max_segment_index = super_segment_header.max_segment_index.as_inner();
684        let first_segment_index = max_segment_index
685            - SegmentIndex::from(u64::from(super_segment_header.num_segments))
686            + SegmentIndex::ONE;
687
688        (first_segment_index..=max_segment_index)
689            .contains(&segment_index)
690            .then_some(super_segment_header)
691    }
692
693    /// Returns actually added super segments (some might have been skipped)
694    fn add_super_segment_headers(
695        &mut self,
696        mut super_segment_headers: Vec<SuperSegmentHeader>,
697    ) -> Result<Vec<SuperSegmentHeader>, PersistSuperSegmentHeadersError> {
698        self.super_segment_headers_cache
699            .reserve(super_segment_headers.len());
700
701        let mut maybe_last_super_segment_index = self
702            .super_segment_headers_cache
703            .last()
704            .map(|header| header.index.as_inner());
705
706        if let Some(last_super_segment_index) = maybe_last_super_segment_index {
707            // Skip already stored super segment headers
708            super_segment_headers.retain(|super_segment_header| {
709                super_segment_header.index.as_inner() > last_super_segment_index
710            });
711        }
712
713        // Check all input super segment headers to see which ones are not stored yet and verifying
714        // that super segment indices are monotonically increasing
715        for super_segment_header in super_segment_headers.iter().copied() {
716            let super_segment_index = super_segment_header.index.as_inner();
717            if let Some(last_super_segment_index) = maybe_last_super_segment_index {
718                if super_segment_index != last_super_segment_index + SuperSegmentIndex::ONE {
719                    return Err(
720                        PersistSuperSegmentHeadersError::MustFollowLastSegmentIndex {
721                            super_segment_index,
722                            last_super_segment_index,
723                        },
724                    );
725                }
726
727                self.super_segment_headers_cache.push(super_segment_header);
728                maybe_last_super_segment_index.replace(super_segment_index);
729            } else {
730                if super_segment_index != SuperSegmentIndex::ZERO {
731                    return Err(PersistSuperSegmentHeadersError::FirstSegmentIndexZero {
732                        super_segment_index,
733                    });
734                }
735
736                self.super_segment_headers_cache.push(super_segment_header);
737                maybe_last_super_segment_index.replace(super_segment_index);
738            }
739        }
740
741        Ok(super_segment_headers)
742    }
743}
744
745// TODO: Hide implementation details
746#[derive(Debug)]
747struct State<Block, StorageBackend>
748where
749    Block: GenericOwnedBlock,
750{
751    data: StateData<Block>,
752    segment_headers_cache: SegmentHeadersCache,
753    super_segment_headers_cache: SuperSegmentHeadersCache,
754    storage_backend_adapter: AsyncRwLock<StorageBackendAdapter<StorageBackend>>,
755}
756
757impl<Block, StorageBackend> State<Block, StorageBackend>
758where
759    Block: GenericOwnedBlock,
760{
761    #[inline(always)]
762    fn best_tip(&self) -> &ForkTip {
763        self.data
764            .fork_tips
765            .front()
766            .expect("The best block is always present; qed")
767    }
768
769    #[inline(always)]
770    fn best_block(&self) -> &ClientDatabaseBlock<Block> {
771        self.data
772            .blocks
773            .front()
774            .expect("The best block is always present; qed")
775            .first()
776            .expect("The best block is always present; qed")
777    }
778}
779
780#[derive(Debug)]
781struct BlockToPersist<'a, Block>
782where
783    Block: GenericOwnedBlock,
784{
785    block_offset: usize,
786    fork_offset: usize,
787    block: &'a Block,
788    block_details: &'a BlockDetails,
789}
790
791#[derive(Debug)]
792struct PersistedBlock {
793    block_offset: usize,
794    fork_offset: usize,
795    write_location: WriteLocation,
796}
797
798#[derive(Debug)]
799struct ClientDatabaseInnerOptions {
800    block_confirmation_depth: BlockNumber,
801    soft_confirmation_depth: BlockNumber,
802    max_fork_tips: NonZeroUsize,
803    max_fork_tip_distance: BlockNumber,
804}
805
806#[derive(Debug)]
807struct Inner<Block, StorageBackend>
808where
809    Block: GenericOwnedBlock,
810{
811    state: AsyncRwLock<State<Block, StorageBackend>>,
812    options: ClientDatabaseInnerOptions,
813}
814
815/// Client database
816#[derive(Debug)]
817pub struct ClientDatabase<Block, StorageBackend>
818where
819    Block: GenericOwnedBlock,
820{
821    inner: Arc<Inner<Block, StorageBackend>>,
822}
823
824impl<Block, StorageBackend> Clone for ClientDatabase<Block, StorageBackend>
825where
826    Block: GenericOwnedBlock,
827{
828    fn clone(&self) -> Self {
829        Self {
830            inner: self.inner.clone(),
831        }
832    }
833}
834
835#[expect(clippy::empty_drop, reason = "Not implemented yet")]
836impl<Block, StorageBackend> Drop for ClientDatabase<Block, StorageBackend>
837where
838    Block: GenericOwnedBlock,
839{
840    fn drop(&mut self) {
841        // TODO: Persist things that were not persisted yet to reduce the data loss on shutdown
842    }
843}
844
845impl<Block, StorageBackend> ChainInfo<Block> for ClientDatabase<Block, StorageBackend>
846where
847    Block: GenericOwnedBlock,
848    StorageBackend: ClientDatabaseStorageBackend,
849{
850    #[inline]
851    fn best_root(&self) -> BlockRoot {
852        // Blocking read lock is fine because where a write lock is only taken for a short time and
853        // most locks are read locks
854        self.inner.state.read_blocking().best_tip().root
855    }
856
857    #[inline]
858    fn best_header(&self) -> Block::Header {
859        // Blocking read lock is fine because where a write lock is only taken for a short time and
860        // most locks are read locks
861        self.inner
862            .state
863            .read_blocking()
864            .best_block()
865            .header()
866            .clone()
867    }
868
869    #[inline]
870    fn best_header_with_details(&self) -> (Block::Header, BlockDetails) {
871        // Blocking read lock is fine because where a write lock is only taken for a short time and
872        // most locks are read locks
873        let state = self.inner.state.read_blocking();
874        let best_block = state.best_block();
875        (
876            best_block.header().clone(),
877            best_block
878                .block_details()
879                .expect("Always present for the best block; qed")
880                .clone(),
881        )
882    }
883
884    // TODO: Add fast path when `descendant_block_root` is the best block
885    #[inline]
886    fn ancestor_header(
887        &self,
888        ancestor_block_number: BlockNumber,
889        descendant_block_root: &BlockRoot,
890    ) -> Option<Block::Header> {
891        // Blocking read lock is fine because where a write lock is only taken for a short time and
892        // most locks are read locks
893        let state = self.inner.state.read_blocking();
894        let best_number = state.best_tip().number;
895
896        let ancestor_block_offset =
897            u64::from(best_number.checked_sub(ancestor_block_number)?) as usize;
898        let ancestor_block_candidates = state.data.blocks.get(ancestor_block_offset)?;
899
900        let descendant_block_number = *state.data.block_roots.get(descendant_block_root)?;
901        if ancestor_block_number > descendant_block_number {
902            return None;
903        }
904        let descendant_block_offset =
905            u64::from(best_number.checked_sub(descendant_block_number)?) as usize;
906
907        // Range of blocks where the first item is expected to contain a descendant
908        let mut blocks_range_iter = state
909            .data
910            .blocks
911            .iter()
912            .enumerate()
913            .skip(descendant_block_offset);
914
915        let (_offset, descendant_block_candidates) = blocks_range_iter.next()?;
916        let descendant_header = descendant_block_candidates
917            .iter()
918            .find(|block| &*block.header().header().root() == descendant_block_root)?
919            .header()
920            .header();
921
922        // If there are no forks at this level, then this is the canonical chain and ancestor
923        // block number we're looking for is the first block at the corresponding block number.
924        // Similarly, if there is just a single ancestor candidate and descendant exists, it must be
925        // the one we care about.
926        if descendant_block_candidates.len() == 1 || ancestor_block_candidates.len() == 1 {
927            return ancestor_block_candidates
928                .iter()
929                .next()
930                .map(|block| block.header().clone());
931        }
932
933        let mut parent_block_root = &descendant_header.prefix.parent_root;
934
935        // Iterate over the blocks following descendant until ancestor is reached
936        for (block_offset, parent_candidates) in blocks_range_iter {
937            let parent_header = parent_candidates
938                .iter()
939                .find(|header| &*header.header().header().root() == parent_block_root)?
940                .header();
941
942            // When header offset matches, we found the header
943            if block_offset == ancestor_block_offset {
944                return Some(parent_header.clone());
945            }
946
947            parent_block_root = &parent_header.header().prefix.parent_root;
948        }
949
950        None
951    }
952
953    #[inline]
954    fn header(&self, block_root: &BlockRoot) -> Option<Block::Header> {
955        // Blocking read lock is fine because where a write lock is only taken for a short time and
956        // most locks are read locks
957        let state = self.inner.state.read_blocking();
958        let best_number = state.best_tip().number;
959
960        let block_number = *state.data.block_roots.get(block_root)?;
961        let block_offset = u64::from(best_number.checked_sub(block_number)?) as usize;
962        let block_candidates = state.data.blocks.get(block_offset)?;
963
964        block_candidates.iter().find_map(|block| {
965            let header = block.header();
966
967            if &*header.header().root() == block_root {
968                Some(header.clone())
969            } else {
970                None
971            }
972        })
973    }
974
975    #[inline]
976    fn header_with_details(&self, block_root: &BlockRoot) -> Option<(Block::Header, BlockDetails)> {
977        // Blocking read lock is fine because where a write lock is only taken for a short time and
978        // most locks are read locks
979        let state = self.inner.state.read_blocking();
980        let best_number = state.best_tip().number;
981
982        let block_number = *state.data.block_roots.get(block_root)?;
983        let block_offset = u64::from(best_number.checked_sub(block_number)?) as usize;
984        let block_candidates = state.data.blocks.get(block_offset)?;
985
986        block_candidates.iter().find_map(|block| {
987            let header = block.header();
988            let block_details = block.block_details().cloned()?;
989
990            if &*header.header().root() == block_root {
991                Some((header.clone(), block_details))
992            } else {
993                None
994            }
995        })
996    }
997
998    #[inline]
999    async fn block(&self, block_root: &BlockRoot) -> Result<Block, ReadBlockError> {
1000        let state = self.inner.state.read().await;
1001        let best_number = state.best_tip().number;
1002
1003        let block_number = *state
1004            .data
1005            .block_roots
1006            .get(block_root)
1007            .ok_or(ReadBlockError::UnknownBlockRoot)?;
1008        let block_offset = u64::from(
1009            best_number
1010                .checked_sub(block_number)
1011                .expect("Known block roots always have valid block offset; qed"),
1012        ) as usize;
1013        let block_candidates = state
1014            .data
1015            .blocks
1016            .get(block_offset)
1017            .expect("Valid block offsets always have block entries; qed");
1018
1019        for block_candidate in block_candidates {
1020            let header = block_candidate.header();
1021
1022            if &*header.header().root() == block_root {
1023                return match block_candidate.full_block() {
1024                    FullBlock::InMemory(block) => Ok(block.clone()),
1025                    FullBlock::Persisted {
1026                        header,
1027                        write_location,
1028                    } => {
1029                        let storage_backend_adapter = state.storage_backend_adapter.read().await;
1030
1031                        let storage_item = storage_backend_adapter
1032                            .read_storage_item::<StorageItemTemporary>(write_location)
1033                            .await?;
1034
1035                        let storage_item_block = match storage_item {
1036                            StorageItemTemporary::Block(storage_item_block) => storage_item_block,
1037                            StorageItemTemporary::SegmentHeaders(_) => {
1038                                return Err(ReadBlockError::StorageItemReadError {
1039                                    error: io::Error::other(
1040                                        "Unexpected storage item: `SegmentHeaders`",
1041                                    ),
1042                                });
1043                            }
1044                            StorageItemTemporary::SuperSegmentHeaders(_) => {
1045                                return Err(ReadBlockError::StorageItemReadError {
1046                                    error: io::Error::other(
1047                                        "Unexpected storage item: `SuperSegmentHeaders`",
1048                                    ),
1049                                });
1050                            }
1051                        };
1052
1053                        let StorageItemTemporaryBlock {
1054                            header: _,
1055                            body,
1056                            mmr_with_block: _,
1057                            system_contract_states: _,
1058                        } = storage_item_block;
1059
1060                        Block::from_buffers(header.buffer().clone(), body)
1061                            .ok_or(ReadBlockError::FailedToDecode)
1062                    }
1063                };
1064            }
1065        }
1066
1067        unreachable!("Known block root always has block candidate associated with it; qed")
1068    }
1069
1070    #[inline]
1071    fn last_segment_header(&self) -> Option<SegmentHeader> {
1072        // Blocking read lock is fine because where a write lock is only taken for a short time and
1073        // most locks are read locks
1074        let state = self.inner.state.read_blocking();
1075        state.segment_headers_cache.last_segment_header()
1076    }
1077
1078    #[inline]
1079    fn get_segment_header(&self, segment_index: LocalSegmentIndex) -> Option<SegmentHeader> {
1080        // Blocking read lock is fine because where a write lock is only taken for a short time and
1081        // most locks are read locks
1082        let state = self.inner.state.read_blocking();
1083
1084        state
1085            .segment_headers_cache
1086            .get_segment_header(segment_index)
1087    }
1088
1089    fn segment_headers_for_block(&self, block_number: BlockNumber) -> Vec<SegmentHeader> {
1090        // Blocking read lock is fine because where a write lock is only taken for a short time and
1091        // most locks are read locks
1092        let state = self.inner.state.read_blocking();
1093
1094        let Some(last_local_segment_index) = state.segment_headers_cache.max_local_segment_index()
1095        else {
1096            // Not initialized
1097            return Vec::new();
1098        };
1099
1100        // Special case for the initial segment (for beacon chain genesis block)
1101        if Block::Block::SHARD_KIND == RealShardKind::BeaconChain
1102            && block_number == BlockNumber::ONE
1103        {
1104            // If there is a segment index present, and we store monotonically increasing segment
1105            // headers, then the first header exists
1106            return vec![
1107                state
1108                    .segment_headers_cache
1109                    .get_segment_header(LocalSegmentIndex::ZERO)
1110                    .expect("Segment headers are stored in monotonically increasing order; qed"),
1111            ];
1112        }
1113
1114        if last_local_segment_index == LocalSegmentIndex::ZERO {
1115            // Genesis segment already included in block #1
1116            return Vec::new();
1117        }
1118
1119        let mut current_local_segment_index = last_local_segment_index;
1120        loop {
1121            // If the current segment index present, and we store monotonically increasing segment
1122            // headers, then the current segment header exists as well.
1123            let current_segment_header = state
1124                .segment_headers_cache
1125                .get_segment_header(current_local_segment_index)
1126                .expect("Segment headers are stored in monotonically increasing order; qed");
1127
1128            // The block immediately after the archived segment adding the confirmation depth
1129            let target_block_number = current_segment_header.last_archived_block.number()
1130                + BlockNumber::ONE
1131                + self.inner.options.block_confirmation_depth;
1132            if target_block_number == block_number {
1133                let mut headers_for_block = vec![current_segment_header];
1134
1135                // Check block spanning multiple segments
1136                let last_archived_block_number = current_segment_header.last_archived_block.number;
1137                let mut local_segment_index = current_local_segment_index - LocalSegmentIndex::ONE;
1138
1139                while let Some(segment_header) = state
1140                    .segment_headers_cache
1141                    .get_segment_header(local_segment_index)
1142                {
1143                    if segment_header.last_archived_block.number == last_archived_block_number {
1144                        headers_for_block.insert(0, segment_header);
1145                        local_segment_index -= LocalSegmentIndex::ONE;
1146                    } else {
1147                        break;
1148                    }
1149                }
1150
1151                return headers_for_block;
1152            }
1153
1154            // iterate segments further
1155            if target_block_number > block_number {
1156                // no need to check the initial segment
1157                if current_local_segment_index > LocalSegmentIndex::ONE {
1158                    current_local_segment_index -= LocalSegmentIndex::ONE;
1159                } else {
1160                    break;
1161                }
1162            } else {
1163                // No segment headers required
1164                return Vec::new();
1165            }
1166        }
1167
1168        // No segment headers required
1169        Vec::new()
1170    }
1171}
1172
1173impl<Block, StorageBackend> ChainInfoWrite<Block> for ClientDatabase<Block, StorageBackend>
1174where
1175    Block: GenericOwnedBlock,
1176    StorageBackend: ClientDatabaseStorageBackend,
1177{
1178    async fn persist_block(
1179        &self,
1180        block: Block,
1181        block_details: BlockDetails,
1182    ) -> Result<(), PersistBlockError> {
1183        let mut state = self.inner.state.write().await;
1184        let best_number = state.best_tip().number;
1185
1186        let header = block.header().header();
1187
1188        let block_number = header.prefix.number;
1189
1190        if best_number == BlockNumber::ZERO && block_number != BlockNumber::ONE {
1191            // Special case when syncing on top of the fresh database
1192            Self::insert_first_block(&mut state.data, block, block_details);
1193
1194            return Ok(());
1195        }
1196
1197        if block_number == best_number + BlockNumber::ONE {
1198            return Self::insert_new_best_block(state, &self.inner, block, block_details).await;
1199        }
1200
1201        let block_offset = u64::from(
1202            best_number
1203                .checked_sub(block_number)
1204                .ok_or(PersistBlockError::MissingParent)?,
1205        ) as usize;
1206
1207        if block_offset >= u64::from(self.inner.options.block_confirmation_depth) as usize {
1208            return Err(PersistBlockError::OutsideAcceptableRange);
1209        }
1210
1211        let state = &mut *state;
1212
1213        let block_forks = state.data.blocks.get_mut(block_offset).ok_or_else(|| {
1214            error!(
1215                %block_number,
1216                %block_offset,
1217                "Failed to store block fork, header offset is missing despite being within \
1218                acceptable range"
1219            );
1220
1221            PersistBlockError::OutsideAcceptableRange
1222        })?;
1223
1224        for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
1225            // Block's parent is no longer a fork tip, remove it
1226            if fork_tip.root == header.prefix.parent_root {
1227                state.data.fork_tips.remove(index);
1228                break;
1229            }
1230        }
1231
1232        let block_root = *header.root();
1233        // Insert at position 1, which means the most recent tip, which doesn't correspond to
1234        // the best block
1235        state.data.fork_tips.insert(
1236            1,
1237            ForkTip {
1238                number: block_number,
1239                root: block_root,
1240            },
1241        );
1242        state.data.block_roots.insert(block_root, block_number);
1243        let beacon_chain_block_details = <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1244            .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1245        block_forks.push(ClientDatabaseBlock::InMemory {
1246            block,
1247            block_details,
1248            beacon_chain_block_details,
1249        });
1250
1251        Self::prune_outdated_fork_tips(block_number, &mut state.data, &self.inner.options);
1252
1253        Ok(())
1254    }
1255
1256    async fn persist_segment_headers(
1257        &self,
1258        segment_headers: Vec<SegmentHeader>,
1259    ) -> Result<(), PersistSegmentHeadersError> {
1260        let mut state = self.inner.state.write().await;
1261
1262        let added_segment_headers = state
1263            .segment_headers_cache
1264            .add_segment_headers(segment_headers)?;
1265
1266        if added_segment_headers.is_empty() {
1267            return Ok(());
1268        }
1269
1270        // Convert write lock into upgradable read lock to allow reads, while preventing segment
1271        // headers modifications
1272        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
1273        //  are satisfied. If not, blocking read locks in other places will cause issues.
1274        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1275
1276        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1277
1278        storage_backend_adapter
1279            .write_storage_item(StorageItemTemporary::SegmentHeaders(
1280                StorageItemTemporarySegmentHeaders {
1281                    segment_headers: added_segment_headers,
1282                },
1283            ))
1284            .await?;
1285
1286        Ok(())
1287    }
1288}
1289
1290impl<StorageBackend> BeaconChainInfo for ClientDatabase<OwnedBeaconChainBlock, StorageBackend>
1291where
1292    StorageBackend: ClientDatabaseStorageBackend,
1293{
1294    fn shard_segment_roots(
1295        &self,
1296        block_number: BlockNumber,
1297    ) -> Result<StdArc<[ShardSegmentRoot]>, ShardSegmentRootsError> {
1298        // Blocking read lock is fine because where a write lock is only taken for a short time and
1299        // most locks are read locks
1300        let state = self.inner.state.read_blocking();
1301        let best_number = state.best_tip().number;
1302
1303        let block_offset = u64::from(
1304            best_number
1305                .checked_sub(block_number)
1306                .ok_or(ShardSegmentRootsError::BlockMissing { block_number })?,
1307        ) as usize;
1308
1309        let block = state
1310            .data
1311            .blocks
1312            .get(block_offset)
1313            .ok_or(ShardSegmentRootsError::BlockMissing { block_number })?
1314            .first()
1315            .expect("There is always at least one block candidate; qed");
1316
1317        Ok(StdArc::clone(
1318            &block
1319                .beacon_chain_block_details()
1320                .as_ref()
1321                .expect("Always present in the beacon chain block; qed")
1322                .shard_segment_roots,
1323        ))
1324    }
1325
1326    #[inline]
1327    fn last_super_segment_header(&self) -> Option<SuperSegmentHeader> {
1328        // Blocking read lock is fine because where a write lock is only taken for a short time and
1329        // most locks are read locks
1330        let state = self.inner.state.read_blocking();
1331        state
1332            .super_segment_headers_cache
1333            .last_super_segment_header()
1334    }
1335
1336    #[inline]
1337    fn previous_super_segment_header(
1338        &self,
1339        block_number: BlockNumber,
1340    ) -> Option<SuperSegmentHeader> {
1341        // Blocking read lock is fine because where a write lock is only taken for a short time and
1342        // most locks are read locks
1343        let state = self.inner.state.read_blocking();
1344
1345        state
1346            .super_segment_headers_cache
1347            .previous_super_segment_header(block_number)
1348    }
1349
1350    #[inline]
1351    fn get_super_segment_header(
1352        &self,
1353        super_segment_index: SuperSegmentIndex,
1354    ) -> Option<SuperSegmentHeader> {
1355        // Blocking read lock is fine because where a write lock is only taken for a short time and
1356        // most locks are read locks
1357        let state = self.inner.state.read_blocking();
1358
1359        state
1360            .super_segment_headers_cache
1361            .get_super_segment_header(super_segment_index)
1362    }
1363
1364    fn get_super_segment_header_for_segment_index(
1365        &self,
1366        segment_index: SegmentIndex,
1367    ) -> Option<SuperSegmentHeader> {
1368        // Blocking read lock is fine because where a write lock is only taken for a short time and
1369        // most locks are read locks
1370        let state = self.inner.state.read_blocking();
1371
1372        state
1373            .super_segment_headers_cache
1374            .get_super_segment_header_for_segment_index(segment_index)
1375    }
1376}
1377
1378impl<StorageBackend> BeaconChainInfoWrite for ClientDatabase<OwnedBeaconChainBlock, StorageBackend>
1379where
1380    StorageBackend: ClientDatabaseStorageBackend,
1381{
1382    async fn persist_super_segment_header(
1383        &self,
1384        super_segment_header: SuperSegmentHeader,
1385    ) -> Result<bool, PersistSuperSegmentHeadersError> {
1386        let mut state = self.inner.state.write().await;
1387
1388        let added_super_segment_headers = state
1389            .super_segment_headers_cache
1390            .add_super_segment_headers(vec![super_segment_header])?;
1391
1392        if added_super_segment_headers.is_empty() {
1393            return Ok(false);
1394        }
1395
1396        // Convert write lock into upgradable read lock to allow reads, while preventing super
1397        // segment headers modifications
1398        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
1399        //  are satisfied. If not, blocking read locks in other places will cause issues.
1400        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1401
1402        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1403
1404        storage_backend_adapter
1405            .write_storage_item(StorageItemTemporary::SuperSegmentHeaders(
1406                StorageItemTemporarySuperSegmentHeaders {
1407                    super_segment_headers: added_super_segment_headers,
1408                },
1409            ))
1410            .await?;
1411
1412        Ok(true)
1413    }
1414
1415    async fn persist_super_segment_headers(
1416        &self,
1417        super_segment_headers: Vec<SuperSegmentHeader>,
1418    ) -> Result<(), PersistSuperSegmentHeadersError> {
1419        let mut state = self.inner.state.write().await;
1420
1421        let added_super_segment_headers = state
1422            .super_segment_headers_cache
1423            .add_super_segment_headers(super_segment_headers)?;
1424
1425        if added_super_segment_headers.is_empty() {
1426            return Ok(());
1427        }
1428
1429        // Convert write lock into upgradable read lock to allow reads, while preventing super
1430        // segment headers modifications
1431        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
1432        //  are satisfied. If not, blocking read locks in other places will cause issues.
1433        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1434
1435        let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1436
1437        storage_backend_adapter
1438            .write_storage_item(StorageItemTemporary::SuperSegmentHeaders(
1439                StorageItemTemporarySuperSegmentHeaders {
1440                    super_segment_headers: added_super_segment_headers,
1441                },
1442            ))
1443            .await?;
1444
1445        Ok(())
1446    }
1447}
1448
1449impl<Block, StorageBackend> ClientDatabase<Block, StorageBackend>
1450where
1451    Block: GenericOwnedBlock,
1452    StorageBackend: ClientDatabaseStorageBackend,
1453{
1454    /// Open the existing database.
1455    ///
1456    /// NOTE: The database needs to be formatted with [`Self::format()`] before it can be used.
1457    pub async fn open<GBB>(
1458        options: ClientDatabaseOptions<GBB, StorageBackend>,
1459    ) -> Result<Self, ClientDatabaseError>
1460    where
1461        GBB: FnOnce() -> GenesisBlockBuilderResult<Block>,
1462    {
1463        let ClientDatabaseOptions {
1464            write_buffer_size,
1465            block_confirmation_depth,
1466            soft_confirmation_depth,
1467            max_fork_tips,
1468            max_fork_tip_distance,
1469            genesis_block_builder,
1470            storage_backend,
1471        } = options;
1472        if soft_confirmation_depth >= block_confirmation_depth {
1473            return Err(ClientDatabaseError::InvalidSoftConfirmationDepth);
1474        }
1475
1476        if max_fork_tip_distance > block_confirmation_depth {
1477            return Err(ClientDatabaseError::InvalidMaxForkTipDistance);
1478        }
1479
1480        let mut state_data = StateData {
1481            fork_tips: VecDeque::new(),
1482            block_roots: HashMap::default(),
1483            blocks: VecDeque::new(),
1484        };
1485        let mut segment_headers_cache = SegmentHeadersCache {
1486            segment_headers_cache: Vec::new(),
1487        };
1488        let mut super_segment_headers_cache = SuperSegmentHeadersCache {
1489            super_segment_headers_cache: Vec::new(),
1490        };
1491
1492        let options = ClientDatabaseInnerOptions {
1493            block_confirmation_depth,
1494            soft_confirmation_depth,
1495            max_fork_tips,
1496            max_fork_tip_distance,
1497        };
1498
1499        let storage_item_handlers = StorageItemHandlers {
1500            permanent: |_arg| {
1501                // TODO
1502                Ok(())
1503            },
1504            temporary: |arg| {
1505                let StorageItemHandlerArg {
1506                    storage_item,
1507                    page_offset,
1508                    num_pages,
1509                } = arg;
1510                let storage_item_block = match storage_item {
1511                    StorageItemTemporary::Block(storage_item_block) => storage_item_block,
1512                    StorageItemTemporary::SegmentHeaders(segment_headers) => {
1513                        let num_segment_headers = segment_headers.segment_headers.len();
1514                        return match segment_headers_cache
1515                            .add_segment_headers(segment_headers.segment_headers)
1516                        {
1517                            Ok(_) => Ok(()),
1518                            Err(error) => {
1519                                error!(
1520                                    %page_offset,
1521                                    %num_segment_headers,
1522                                    %error,
1523                                    "Failed to add segment headers from storage item"
1524                                );
1525
1526                                Err(ClientDatabaseError::InvalidSegmentHeaders { page_offset })
1527                            }
1528                        };
1529                    }
1530                    StorageItemTemporary::SuperSegmentHeaders(super_segment_headers) => {
1531                        let num_super_segment_headers =
1532                            super_segment_headers.super_segment_headers.len();
1533                        return match super_segment_headers_cache
1534                            .add_super_segment_headers(super_segment_headers.super_segment_headers)
1535                        {
1536                            Ok(_) => Ok(()),
1537                            Err(error) => {
1538                                error!(
1539                                    %page_offset,
1540                                    %num_super_segment_headers,
1541                                    %error,
1542                                    "Failed to add segment headers from storage item"
1543                                );
1544
1545                                Err(ClientDatabaseError::InvalidSegmentHeaders { page_offset })
1546                            }
1547                        };
1548                    }
1549                };
1550
1551                // TODO: It would be nice to not allocate body here since we'll not use it here
1552                //  anyway
1553                let StorageItemTemporaryBlock {
1554                    header,
1555                    body,
1556                    mmr_with_block,
1557                    system_contract_states,
1558                } = storage_item_block;
1559
1560                let header = Block::Header::from_buffer(header).map_err(|_buffer| {
1561                    error!(%page_offset, "Failed to decode block header from bytes");
1562
1563                    ClientDatabaseError::InvalidBlock { page_offset }
1564                })?;
1565                let body = Block::Body::from_buffer(body).map_err(|_buffer| {
1566                    error!(%page_offset, "Failed to decode block body from bytes");
1567
1568                    ClientDatabaseError::InvalidBlock { page_offset }
1569                })?;
1570
1571                let block_root = *header.header().root();
1572                let block_number = header.header().prefix.number;
1573
1574                state_data.block_roots.insert(block_root, block_number);
1575
1576                let maybe_best_number = state_data
1577                    .blocks
1578                    .front()
1579                    .and_then(|block_forks| block_forks.first())
1580                    .map(|best_block| {
1581                        // Type inference is not working here for some reason
1582                        let header: &Block::Header = best_block.header();
1583
1584                        header.header().prefix.number
1585                    });
1586
1587                let block_offset = if let Some(best_number) = maybe_best_number {
1588                    if block_number <= best_number {
1589                        u64::from(best_number - block_number) as usize
1590                    } else {
1591                        // The new best block must follow the previous best block
1592                        if block_number - best_number != BlockNumber::ONE {
1593                            error!(
1594                                %page_offset,
1595                                %best_number,
1596                                %block_number,
1597                                "Invalid new best block number, it must be only one block \
1598                                higher than the best block"
1599                            );
1600
1601                            return Err(ClientDatabaseError::InvalidBlock { page_offset });
1602                        }
1603
1604                        state_data.blocks.push_front(SmallVec::new());
1605                        // Will insert a new block at the front
1606                        0
1607                    }
1608                } else {
1609                    state_data.blocks.push_front(SmallVec::new());
1610                    // Will insert a new block at the front
1611                    0
1612                };
1613
1614                let Some(block_forks) = state_data.blocks.get_mut(block_offset) else {
1615                    // Ignore the older block, other blocks at its height were already pruned
1616                    // anyway
1617                    return Ok(());
1618                };
1619
1620                // Push a new block to the end of the list, we'll fix it up later
1621                let beacon_chain_block_details =
1622                    <dyn Any>::downcast_ref::<OwnedBeaconChainBody>(&body)
1623                        .map(|body| BeaconChainBlockDetails::from_body(body.body()));
1624                block_forks.push(ClientDatabaseBlock::Persisted {
1625                    header,
1626                    block_details: BlockDetails {
1627                        mmr_with_block,
1628                        system_contract_states,
1629                    },
1630                    beacon_chain_block_details,
1631                    write_location: WriteLocation {
1632                        page_offset,
1633                        num_pages,
1634                    },
1635                });
1636
1637                // If a new block was inserted, confirm a new canonical block to prune extra
1638                // in-memory information
1639                if block_offset == 0 && block_forks.len() == 1 {
1640                    Self::confirm_canonical_block(block_number, &mut state_data, &options);
1641                }
1642
1643                Ok(())
1644            },
1645        };
1646
1647        let storage_backend_adapter =
1648            StorageBackendAdapter::open(write_buffer_size, storage_item_handlers, storage_backend)
1649                .await?;
1650
1651        if let Some(best_block) = state_data.blocks.front().and_then(|block_forks| {
1652            // The best block is last in the list here because that is how it was inserted while
1653            // reading from the database
1654            block_forks.last()
1655        }) {
1656            // Type inference is not working here for some reason
1657            let header: &Block::Header = best_block.header();
1658            let header = header.header();
1659            let block_number = header.prefix.number;
1660            let block_root = *header.root();
1661
1662            if !Self::adjust_ancestor_block_forks(&mut state_data.blocks, block_root) {
1663                return Err(ClientDatabaseError::FailedToAdjustAncestorBlockForks);
1664            }
1665
1666            // Store the best block as the first and only fork tip
1667            state_data.fork_tips.push_front(ForkTip {
1668                number: block_number,
1669                root: block_root,
1670            });
1671        } else {
1672            let GenesisBlockBuilderResult {
1673                block,
1674                system_contract_states,
1675            } = genesis_block_builder();
1676
1677            // If the database is empty, initialize everything with the genesis block
1678            let header = block.header().header();
1679            let block_number = header.prefix.number;
1680            let block_root = *header.root();
1681
1682            state_data.fork_tips.push_front(ForkTip {
1683                number: block_number,
1684                root: block_root,
1685            });
1686            state_data.block_roots.insert(block_root, block_number);
1687            let beacon_chain_block_details =
1688                <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1689                    .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1690            state_data
1691                .blocks
1692                .push_front(smallvec![ClientDatabaseBlock::InMemory {
1693                    block,
1694                    block_details: BlockDetails {
1695                        system_contract_states,
1696                        mmr_with_block: Arc::new({
1697                            let mut mmr = BlockMerkleMountainRange::new();
1698                            mmr.add_leaf(&block_root);
1699                            mmr
1700                        })
1701                    },
1702                    beacon_chain_block_details,
1703                }]);
1704        }
1705
1706        let state = State {
1707            data: state_data,
1708            segment_headers_cache,
1709            super_segment_headers_cache,
1710            storage_backend_adapter: AsyncRwLock::new(storage_backend_adapter),
1711        };
1712
1713        let inner = Inner {
1714            state: AsyncRwLock::new(state),
1715            options,
1716        };
1717
1718        Ok(Self {
1719            inner: Arc::new(inner),
1720        })
1721    }
1722
1723    /// Format a new database
1724    pub async fn format(
1725        storage_backend: &StorageBackend,
1726        options: ClientDatabaseFormatOptions,
1727    ) -> Result<(), ClientDatabaseFormatError> {
1728        StorageBackendAdapter::format(storage_backend, options).await
1729    }
1730
1731    fn insert_first_block(state: &mut StateData<Block>, block: Block, block_details: BlockDetails) {
1732        // If the database is empty, initialize everything with the genesis block
1733        let header = block.header().header();
1734        let block_number = header.prefix.number;
1735        let block_root = *header.root();
1736
1737        state.fork_tips.clear();
1738        state.fork_tips.push_front(ForkTip {
1739            number: block_number,
1740            root: block_root,
1741        });
1742        state.block_roots.clear();
1743        state.block_roots.insert(block_root, block_number);
1744        state.blocks.clear();
1745        let beacon_chain_block_details = <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1746            .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1747        state
1748            .blocks
1749            .push_front(smallvec![ClientDatabaseBlock::InMemory {
1750                block,
1751                block_details,
1752                beacon_chain_block_details,
1753            }]);
1754    }
1755
1756    async fn insert_new_best_block(
1757        mut state: AsyncRwLockWriteGuard<'_, State<Block, StorageBackend>>,
1758        inner: &Inner<Block, StorageBackend>,
1759        block: Block,
1760        block_details: BlockDetails,
1761    ) -> Result<(), PersistBlockError> {
1762        let header = block.header().header();
1763        let block_number = header.prefix.number;
1764        let block_root = *header.root();
1765        let parent_root = header.prefix.parent_root;
1766
1767        // Adjust the relative order of forks to ensure the first index always corresponds to
1768        // ancestors of the new best block
1769        if !Self::adjust_ancestor_block_forks(&mut state.data.blocks, parent_root) {
1770            return Err(PersistBlockError::MissingParent);
1771        }
1772
1773        // Store new block in the state
1774        {
1775            for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
1776                // Block's parent is no longer a fork tip, remove it
1777                if fork_tip.root == parent_root {
1778                    state.data.fork_tips.remove(index);
1779                    break;
1780                }
1781            }
1782
1783            state.data.fork_tips.push_front(ForkTip {
1784                number: block_number,
1785                root: block_root,
1786            });
1787            state.data.block_roots.insert(block_root, block_number);
1788            let beacon_chain_block_details =
1789                <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1790                    .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1791            state
1792                .data
1793                .blocks
1794                .push_front(smallvec![ClientDatabaseBlock::InMemory {
1795                    block,
1796                    block_details: block_details.clone(),
1797                    beacon_chain_block_details,
1798                }]);
1799        }
1800
1801        let options = &inner.options;
1802
1803        Self::confirm_canonical_block(block_number, &mut state.data, options);
1804        Self::prune_outdated_fork_tips(block_number, &mut state.data, options);
1805
1806        // Convert write lock into upgradable read lock to allow reads, while preventing concurrent
1807        // block modifications
1808        // TODO: This assumes both guarantees in https://github.com/smol-rs/async-lock/issues/100
1809        //  are satisfied. If not, blocking read locks in other places will cause issues.
1810        let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1811
1812        let mut blocks_to_persist = Vec::new();
1813        for block_offset in u64::from(options.soft_confirmation_depth) as usize.. {
1814            let Some(fork_blocks) = state.data.blocks.get(block_offset) else {
1815                break;
1816            };
1817
1818            let len_before = blocks_to_persist.len();
1819            fork_blocks
1820                .iter()
1821                .enumerate()
1822                .filter_map(|(fork_offset, client_database_block)| {
1823                    #[expect(
1824                        clippy::rest_pattern_accessible_field,
1825                        reason = "Do not need other fields"
1826                    )]
1827                    match client_database_block {
1828                        ClientDatabaseBlock::InMemory {
1829                            block,
1830                            block_details,
1831                            beacon_chain_block_details: _,
1832                        } => Some(BlockToPersist {
1833                            block_offset,
1834                            fork_offset,
1835                            block,
1836                            block_details,
1837                        }),
1838                        ClientDatabaseBlock::Persisted { .. }
1839                        | ClientDatabaseBlock::PersistedConfirmed { .. } => {
1840                            // Already persisted
1841                            None
1842                        }
1843                    }
1844                })
1845                .collect_into(&mut blocks_to_persist);
1846
1847            if blocks_to_persist.len() == len_before {
1848                break;
1849            }
1850        }
1851
1852        // Persist blocks from older to newer
1853        let mut persisted_blocks = Vec::with_capacity(blocks_to_persist.len());
1854        {
1855            let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1856
1857            for block_to_persist in blocks_to_persist.into_iter().rev() {
1858                let BlockToPersist {
1859                    block_offset,
1860                    fork_offset,
1861                    block,
1862                    block_details,
1863                } = block_to_persist;
1864
1865                let write_location = storage_backend_adapter
1866                    .write_storage_item(StorageItemTemporary::Block(StorageItemTemporaryBlock {
1867                        header: block.header().buffer().clone(),
1868                        body: block.body().buffer().clone(),
1869                        mmr_with_block: Arc::clone(&block_details.mmr_with_block),
1870                        system_contract_states: StdArc::clone(
1871                            &block_details.system_contract_states,
1872                        ),
1873                    }))
1874                    .await?;
1875
1876                persisted_blocks.push(PersistedBlock {
1877                    block_offset,
1878                    fork_offset,
1879                    write_location,
1880                });
1881            }
1882        }
1883
1884        // Convert blocks to persisted
1885        let mut state = RwLockUpgradableReadGuard::upgrade(state).await;
1886        for persisted_block in persisted_blocks {
1887            let PersistedBlock {
1888                block_offset,
1889                fork_offset,
1890                write_location,
1891            } = persisted_block;
1892
1893            let block = state
1894                .data
1895                .blocks
1896                .get_mut(block_offset)
1897                .expect("Still holding the same lock since last check; qed")
1898                .get_mut(fork_offset)
1899                .expect("Still holding the same lock since last check; qed");
1900
1901            replace_with_or_abort(block, |block| {
1902                if let ClientDatabaseBlock::InMemory {
1903                    block,
1904                    block_details,
1905                    beacon_chain_block_details,
1906                } = block
1907                {
1908                    let (header, _body) = block.split();
1909
1910                    ClientDatabaseBlock::Persisted {
1911                        header,
1912                        block_details,
1913                        beacon_chain_block_details,
1914                        write_location,
1915                    }
1916                } else {
1917                    unreachable!("Still holding the same lock since last check; qed");
1918                }
1919            });
1920        }
1921
1922        // TODO: Prune blocks that are no longer necessary
1923        // TODO: Prune unused page groups here or elsewhere?
1924
1925        Ok(())
1926    }
1927
1928    /// Adjust the relative order of forks to ensure the first index always corresponds to
1929    /// `parent_block_root` and its ancestors.
1930    ///
1931    /// Returns `true` on success and `false` if one of the parents was not found.
1932    #[must_use]
1933    fn adjust_ancestor_block_forks(
1934        blocks: &mut VecDeque<SmallVec<[ClientDatabaseBlock<Block>; 2]>>,
1935        mut parent_block_root: BlockRoot,
1936    ) -> bool {
1937        let mut ancestor_blocks = blocks.iter_mut();
1938
1939        loop {
1940            if ancestor_blocks.len() == 1 {
1941                // Nothing left to adjust with a single fork
1942                break;
1943            }
1944
1945            let Some(parent_blocks) = ancestor_blocks.next() else {
1946                // No more parent headers present
1947                break;
1948            };
1949
1950            let Some(fork_offset_parent_block_root) =
1951                parent_blocks
1952                    .iter()
1953                    .enumerate()
1954                    .find_map(|(fork_offset, fork_block)| {
1955                        let fork_header = fork_block.header().header();
1956                        if *fork_header.root() == parent_block_root {
1957                            Some((fork_offset, fork_header.prefix.parent_root))
1958                        } else {
1959                            None
1960                        }
1961                    })
1962            else {
1963                return false;
1964            };
1965
1966            let fork_offset;
1967            (fork_offset, parent_block_root) = fork_offset_parent_block_root;
1968
1969            parent_blocks.swap(0, fork_offset);
1970        }
1971
1972        true
1973    }
1974
1975    /// Prune outdated fork tips that are too deep and have not been updated for a long time.
1976    ///
1977    /// Note that actual headers, blocks and MMRs could remain if they are currently used by
1978    /// something or were already persisted on disk. With persisted blocks specifically, RAM usage
1979    /// implications are minimal, and we wouldn't want to re-download already stored blocks in case
1980    /// they end up being necessary later.
1981    fn prune_outdated_fork_tips(
1982        best_number: BlockNumber,
1983        state: &mut StateData<Block>,
1984        options: &ClientDatabaseInnerOptions,
1985    ) {
1986        let state = &mut *state;
1987
1988        // These forks are just candidates because they will not be pruned if the reference count is
1989        // not 1, indicating they are still in use by something
1990        let mut candidate_forks_to_remove = Vec::with_capacity(options.max_fork_tips.get());
1991
1992        // Prune forks that are too far away from the best block
1993        state.fork_tips.retain(|fork_tip| {
1994            if best_number - fork_tip.number > options.max_fork_tip_distance {
1995                candidate_forks_to_remove.push(*fork_tip);
1996                false
1997            } else {
1998                true
1999            }
2000        });
2001        // Prune forks that exceed the maximum number of forks
2002        if state.fork_tips.len() > options.max_fork_tips.get() {
2003            state
2004                .fork_tips
2005                .drain(options.max_fork_tips.get()..)
2006                .collect_into(&mut candidate_forks_to_remove);
2007        }
2008
2009        // Prune all possible candidates
2010        candidate_forks_to_remove
2011            .retain(|fork_tip| !Self::prune_outdated_fork(best_number, fork_tip, state));
2012        // Return those that were not pruned back to the list of tips
2013        state.fork_tips.extend(candidate_forks_to_remove);
2014    }
2015
2016    /// Returns `true` if the tip was pruned successfully and `false` if it should be returned to
2017    /// the list of fork tips
2018    #[must_use]
2019    fn prune_outdated_fork(
2020        best_number: BlockNumber,
2021        fork_tip: &ForkTip,
2022        state: &mut StateData<Block>,
2023    ) -> bool {
2024        let block_offset = u64::from(best_number - fork_tip.number) as usize;
2025
2026        // Prune fork top and all its ancestors that are not used
2027        let mut block_root_to_prune = fork_tip.root;
2028        let mut pruned_tip = false;
2029        for block_offset in block_offset.. {
2030            let Some(fork_blocks) = state.blocks.get_mut(block_offset) else {
2031                if !pruned_tip {
2032                    error!(
2033                        %best_number,
2034                        ?fork_tip,
2035                        block_offset,
2036                        "Block offset was not present in the database, this is an implementation \
2037                        bug #1"
2038                    );
2039                }
2040                // No forks left to prune
2041                break;
2042            };
2043
2044            if fork_blocks.len() == 1 {
2045                if !pruned_tip {
2046                    error!(
2047                        %best_number,
2048                        ?fork_tip,
2049                        block_offset,
2050                        "Block offset was not present in the database, this is an implementation \
2051                        bug #2"
2052                    );
2053                }
2054
2055                // No forks left to prune
2056                break;
2057            }
2058
2059            let Some((fork_offset, block)) = fork_blocks
2060                .iter()
2061                .enumerate()
2062                // Skip ancestor of the best block, it is certainly not a fork to be pruned
2063                .skip(1)
2064                .find(|(_fork_offset, block)| {
2065                    *block.header().header().root() == block_root_to_prune
2066                })
2067            else {
2068                if !pruned_tip {
2069                    error!(
2070                        %best_number,
2071                        ?fork_tip,
2072                        block_offset,
2073                        "Block offset was not present in the database, this is an implementation \
2074                        bug #3"
2075                    );
2076                }
2077
2078                // Nothing left to prune
2079                break;
2080            };
2081
2082            // More than one instance means something somewhere is using or depends on this block
2083            if block.header().ref_count() > 1 {
2084                break;
2085            }
2086
2087            // Blocks that are already persisted
2088            #[expect(
2089                clippy::rest_pattern_accessible_field,
2090                reason = "Do not care about fields"
2091            )]
2092            match block {
2093                ClientDatabaseBlock::InMemory { .. } => {
2094                    // Prune
2095                }
2096                ClientDatabaseBlock::Persisted { .. }
2097                | ClientDatabaseBlock::PersistedConfirmed { .. } => {
2098                    // Already on disk, keep it in memory for later, but prune the tip
2099                    pruned_tip = true;
2100                    break;
2101                }
2102            }
2103
2104            state.block_roots.get_mut(&block_root_to_prune);
2105            block_root_to_prune = block.header().header().prefix.parent_root;
2106            fork_blocks.swap_remove(fork_offset);
2107
2108            pruned_tip = true;
2109        }
2110
2111        pruned_tip
2112    }
2113
2114    /// Confirm a block at confirmation depth k and prune any other blocks at the same depth with
2115    /// their descendants
2116    fn confirm_canonical_block(
2117        best_number: BlockNumber,
2118        state_data: &mut StateData<Block>,
2119        options: &ClientDatabaseInnerOptions,
2120    ) {
2121        // `+1` means it effectively confirms parent blocks instead. This is done to keep the parent
2122        // of the confirmed block with its MMR in memory due to confirmed blocks not storing their
2123        // MMRs, which might be needed for reorgs at the lowest possible depth.
2124        let block_offset = u64::from(options.block_confirmation_depth + BlockNumber::ONE) as usize;
2125
2126        let Some(fork_blocks) = state_data.blocks.get_mut(block_offset) else {
2127            // Nothing to confirm yet
2128            return;
2129        };
2130
2131        // Mark the canonical block as confirmed
2132        {
2133            let Some(canonical_block) = fork_blocks.first_mut() else {
2134                error!(
2135                    %best_number,
2136                    block_offset,
2137                    "Have not found a canonical block to confirm, this is an implementation bug"
2138                );
2139                return;
2140            };
2141
2142            replace_with_or_abort(canonical_block, |block| {
2143                #[expect(
2144                    clippy::rest_pattern_accessible_field,
2145                    reason = "Do not need other fields"
2146                )]
2147                match block {
2148                    ClientDatabaseBlock::InMemory { .. } => {
2149                        error!(
2150                            %best_number,
2151                            block_offset,
2152                            header = ?block.header(),
2153                            "Block to be confirmed must not be in memory, this is an implementation bug"
2154                        );
2155                        block
2156                    }
2157                    ClientDatabaseBlock::Persisted {
2158                        header,
2159                        block_details: _,
2160                        beacon_chain_block_details,
2161                        write_location,
2162                    } => ClientDatabaseBlock::PersistedConfirmed {
2163                        header,
2164                        beacon_chain_block_details,
2165                        write_location,
2166                    },
2167                    ClientDatabaseBlock::PersistedConfirmed { .. } => {
2168                        error!(
2169                            %best_number,
2170                            block_offset,
2171                            header = ?block.header(),
2172                            "Block to be confirmed must not be confirmed yet, this is an \
2173                            implementation bug"
2174                        );
2175                        block
2176                    }
2177                }
2178            });
2179        }
2180
2181        // Prune the rest of the blocks and their descendants
2182        let mut block_roots_to_prune = fork_blocks
2183            .drain(1..)
2184            .map(|block| *block.header().header().root())
2185            .collect::<Vec<_>>();
2186        let mut current_block_offset = block_offset;
2187        while !block_roots_to_prune.is_empty() {
2188            // Prune fork tips (if any)
2189            state_data
2190                .fork_tips
2191                .retain(|fork_tip| !block_roots_to_prune.contains(&fork_tip.root));
2192
2193            // Prune removed block roots
2194            for block_root in &block_roots_to_prune {
2195                state_data.block_roots.remove(block_root);
2196            }
2197
2198            // Block offset for direct descendants
2199            if let Some(next_block_offset) = current_block_offset.checked_sub(1) {
2200                current_block_offset = next_block_offset;
2201            } else {
2202                // Reached the tip
2203                break;
2204            }
2205
2206            let fork_blocks = state_data
2207                .blocks
2208                .get_mut(current_block_offset)
2209                .expect("Lower block offset always exists; qed");
2210
2211            // Collect descendants of pruned blocks to prune them next
2212            block_roots_to_prune = fork_blocks
2213                .drain_filter(|block| {
2214                    let header = block.header().header();
2215
2216                    block_roots_to_prune.contains(&header.prefix.parent_root)
2217                })
2218                .map(|block| *block.header().header().root())
2219                .collect();
2220        }
2221    }
2222}