1#![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#[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 pub block: Block,
148 pub system_contract_states: StdArc<[ContractSlotState]>,
150}
151
152#[derive(Debug, Copy, Clone)]
154pub struct ClientDatabaseOptions<GBB, StorageBackend> {
155 pub write_buffer_size: usize = 5,
163 pub block_confirmation_depth: BlockNumber,
169 pub soft_confirmation_depth: BlockNumber = BlockNumber::from(3),
183 pub max_fork_tips: NonZeroUsize = NonZeroUsize::new(3).expect("Not zero; qed"),
194 pub max_fork_tip_distance: BlockNumber = BlockNumber::from(5),
205 pub genesis_block_builder: GBB,
208 pub storage_backend: StorageBackend,
210}
211
212#[derive(Debug, Copy, Clone)]
214pub struct ClientDatabaseFormatOptions {
215 pub page_group_size: NonZeroU32,
232 pub force: bool,
236}
237
238#[derive(Debug, thiserror::Error)]
239pub enum ClientDatabaseError {
240 #[error("Invalid soft confirmation depth, it must be smaller than confirmation depth k")]
242 InvalidSoftConfirmationDepth,
243 #[error("Invalid max fork tip distance, it must be smaller or equal to confirmation depth k")]
245 InvalidMaxForkTipDistance,
246 #[error("Storage backend has canceled read request")]
248 ReadRequestCancelled,
249 #[error("Storage backend read error: {error}")]
251 ReadError {
252 error: io::Error,
254 },
255 #[error("Unsupported database version: {database_version}")]
257 UnsupportedDatabaseVersion {
258 database_version: u8,
260 },
261 #[error("Page group size is too small ({page_group_size}), must be at least two pages")]
263 PageGroupSizeTooSmall {
264 page_group_size: u32,
266 },
267 #[error(
269 "Unexpected sequence number {actual} at page offset {page_offset} (expected \
270 {expected})"
271 )]
272 UnexpectedSequenceNumber {
273 actual: u64,
275 expected: u64,
277 page_offset: u32,
279 },
280 #[error("Unexpected storage item at offset {page_offset}: {storage_item:?}")]
282 UnexpectedStorageItem {
283 storage_item: Box<dyn fmt::Debug + Send + Sync>,
285 page_offset: u32,
287 },
288 #[error("Invalid block at offset {page_offset}")]
290 InvalidBlock {
291 page_offset: u32,
293 },
294 #[error("Invalid segment headers at offset {page_offset}")]
296 InvalidSegmentHeaders {
297 page_offset: u32,
299 },
300 #[error("Failed to adjust ancestor block forks")]
302 FailedToAdjustAncestorBlockForks,
303 #[error("Database is not formatted yet")]
305 Unformatted,
306 #[error("Non-permanent first page group")]
308 NonPermanentFirstPageGroup,
309}
310
311#[derive(Debug, thiserror::Error)]
313pub enum ClientDatabaseFormatError {
314 #[error("Storage backend has canceled read request")]
316 ReadRequestCancelled,
317 #[error("Storage backend read error: {error}")]
319 ReadError {
320 error: io::Error,
322 },
323 #[error("Failed to generate database id")]
325 FailedToGenerateDatabaseId {
326 #[from]
328 error: SysError,
329 },
330 #[error("Database is already formatted yet")]
332 AlreadyFormatted,
333 #[error("Storage backend has canceled a writing request")]
335 WriteRequestCancelled,
336 #[error("Storage item write error")]
338 StorageItemWriteError {
339 #[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: 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#[derive(Debug)]
419enum ClientDatabaseBlock<Block>
420where
421 Block: GenericOwnedBlock,
422{
423 InMemory {
425 block: Block,
426 block_details: BlockDetails,
427 beacon_chain_block_details: Option<BeaconChainBlockDetails>,
429 },
430 Persisted {
432 header: Block::Header,
433 block_details: BlockDetails,
434 beacon_chain_block_details: Option<BeaconChainBlockDetails>,
436 write_location: WriteLocation,
437 },
438 PersistedConfirmed {
441 header: Block::Header,
442 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 fork_tips: VecDeque<ForkTip>,
535 block_roots: HashMap<BlockRoot, BlockNumber, BuildHasherDefault<BlockRootHasher>>,
540 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 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 segment_headers
594 .retain(|segment_header| segment_header.index.as_inner() > last_segment_index);
595 }
596
597 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 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 super_segment_headers.retain(|super_segment_header| {
709 super_segment_header.index.as_inner() > last_super_segment_index
710 });
711 }
712
713 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#[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#[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 }
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 self.inner.state.read_blocking().best_tip().root
855 }
856
857 #[inline]
858 fn best_header(&self) -> Block::Header {
859 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 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 #[inline]
886 fn ancestor_header(
887 &self,
888 ancestor_block_number: BlockNumber,
889 descendant_block_root: &BlockRoot,
890 ) -> Option<Block::Header> {
891 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 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 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 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 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 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 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 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 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 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 return Vec::new();
1098 };
1099
1100 if Block::Block::SHARD_KIND == RealShardKind::BeaconChain
1102 && block_number == BlockNumber::ONE
1103 {
1104 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 return Vec::new();
1117 }
1118
1119 let mut current_local_segment_index = last_local_segment_index;
1120 loop {
1121 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 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 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 if target_block_number > block_number {
1156 if current_local_segment_index > LocalSegmentIndex::ONE {
1158 current_local_segment_index -= LocalSegmentIndex::ONE;
1159 } else {
1160 break;
1161 }
1162 } else {
1163 return Vec::new();
1165 }
1166 }
1167
1168 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 0
1607 }
1608 } else {
1609 state_data.blocks.push_front(SmallVec::new());
1610 0
1612 };
1613
1614 let Some(block_forks) = state_data.blocks.get_mut(block_offset) else {
1615 return Ok(());
1618 };
1619
1620 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 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 block_forks.last()
1655 }) {
1656 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 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 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 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 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 if !Self::adjust_ancestor_block_forks(&mut state.data.blocks, parent_root) {
1770 return Err(PersistBlockError::MissingParent);
1771 }
1772
1773 {
1775 for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
1776 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 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 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 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 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 Ok(())
1926 }
1927
1928 #[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 break;
1943 }
1944
1945 let Some(parent_blocks) = ancestor_blocks.next() else {
1946 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 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 let mut candidate_forks_to_remove = Vec::with_capacity(options.max_fork_tips.get());
1991
1992 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 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 candidate_forks_to_remove
2011 .retain(|fork_tip| !Self::prune_outdated_fork(best_number, fork_tip, state));
2012 state.fork_tips.extend(candidate_forks_to_remove);
2014 }
2015
2016 #[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 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 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 break;
2057 }
2058
2059 let Some((fork_offset, block)) = fork_blocks
2060 .iter()
2061 .enumerate()
2062 .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 break;
2080 };
2081
2082 if block.header().ref_count() > 1 {
2084 break;
2085 }
2086
2087 #[expect(
2089 clippy::rest_pattern_accessible_field,
2090 reason = "Do not care about fields"
2091 )]
2092 match block {
2093 ClientDatabaseBlock::InMemory { .. } => {
2094 }
2096 ClientDatabaseBlock::Persisted { .. }
2097 | ClientDatabaseBlock::PersistedConfirmed { .. } => {
2098 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 fn confirm_canonical_block(
2117 best_number: BlockNumber,
2118 state_data: &mut StateData<Block>,
2119 options: &ClientDatabaseInnerOptions,
2120 ) {
2121 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 return;
2129 };
2130
2131 {
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 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 state_data
2190 .fork_tips
2191 .retain(|fork_tip| !block_roots_to_prune.contains(&fork_tip.root));
2192
2193 for block_root in &block_roots_to_prune {
2195 state_data.block_roots.remove(block_root);
2196 }
2197
2198 if let Some(next_block_offset) = current_block_offset.checked_sub(1) {
2200 current_block_offset = next_block_offset;
2201 } else {
2202 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 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}