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 match self {
455 Self::InMemory { block, .. } => block.header(),
456 Self::Persisted { header, .. } | Self::PersistedConfirmed { header, .. } => header,
457 }
458 }
459
460 #[inline(always)]
461 fn full_block(&self) -> FullBlock<'_, Block> {
462 match self {
463 Self::InMemory { block, .. } => FullBlock::InMemory(block),
464 Self::Persisted {
465 header,
466 write_location,
467 ..
468 }
469 | Self::PersistedConfirmed {
470 header,
471 write_location,
472 ..
473 } => FullBlock::Persisted {
474 header,
475 write_location: *write_location,
476 },
477 }
478 }
479
480 #[inline(always)]
481 fn block_details(&self) -> Option<&BlockDetails> {
482 match self {
483 Self::InMemory { block_details, .. } | Self::Persisted { block_details, .. } => {
484 Some(block_details)
485 }
486 Self::PersistedConfirmed { .. } => None,
487 }
488 }
489
490 #[inline(always)]
491 fn beacon_chain_block_details(&self) -> Option<&BeaconChainBlockDetails> {
492 match self {
493 Self::InMemory {
494 beacon_chain_block_details,
495 ..
496 }
497 | Self::Persisted {
498 beacon_chain_block_details,
499 ..
500 }
501 | Self::PersistedConfirmed {
502 beacon_chain_block_details,
503 ..
504 } => beacon_chain_block_details.as_ref(),
505 }
506 }
507}
508
509#[derive(Debug)]
510struct StateData<Block>
511where
512 Block: GenericOwnedBlock,
513{
514 fork_tips: VecDeque<ForkTip>,
519 block_roots: HashMap<BlockRoot, BlockNumber, BuildHasherDefault<BlockRootHasher>>,
524 blocks: VecDeque<SmallVec<[ClientDatabaseBlock<Block>; 2]>>,
539}
540
541#[derive(Debug)]
542struct SegmentHeadersCache {
543 segment_headers_cache: Vec<SegmentHeader>,
544}
545
546impl SegmentHeadersCache {
547 #[inline(always)]
548 fn last_segment_header(&self) -> Option<SegmentHeader> {
549 self.segment_headers_cache.last().copied()
550 }
551
552 #[inline(always)]
553 fn max_local_segment_index(&self) -> Option<LocalSegmentIndex> {
554 self.segment_headers_cache
555 .last()
556 .map(|segment_header| segment_header.index.as_inner())
557 }
558
559 #[inline(always)]
560 fn get_segment_header(&self, local_segment_index: LocalSegmentIndex) -> Option<SegmentHeader> {
561 self.segment_headers_cache
562 .get(u64::from(local_segment_index) as usize)
563 .copied()
564 }
565
566 fn add_segment_headers(
568 &mut self,
569 mut segment_headers: Vec<SegmentHeader>,
570 ) -> Result<Vec<SegmentHeader>, PersistSegmentHeadersError> {
571 self.segment_headers_cache.reserve(segment_headers.len());
572
573 let mut maybe_last_local_segment_index = self.max_local_segment_index();
574
575 if let Some(last_segment_index) = maybe_last_local_segment_index {
576 segment_headers
578 .retain(|segment_header| segment_header.index.as_inner() > last_segment_index);
579 }
580
581 for segment_header in segment_headers.iter().copied() {
584 let local_segment_index = segment_header.index.as_inner();
585 if let Some(last_local_segment_index) = maybe_last_local_segment_index {
586 if local_segment_index != last_local_segment_index + LocalSegmentIndex::ONE {
587 return Err(PersistSegmentHeadersError::MustFollowLastSegmentIndex {
588 local_segment_index,
589 last_local_segment_index,
590 });
591 }
592
593 self.segment_headers_cache.push(segment_header);
594 maybe_last_local_segment_index.replace(local_segment_index);
595 } else {
596 if local_segment_index != LocalSegmentIndex::ZERO {
597 return Err(PersistSegmentHeadersError::FirstSegmentIndexZero {
598 local_segment_index,
599 });
600 }
601
602 self.segment_headers_cache.push(segment_header);
603 maybe_last_local_segment_index.replace(local_segment_index);
604 }
605 }
606
607 Ok(segment_headers)
608 }
609}
610
611#[derive(Debug)]
612struct SuperSegmentHeadersCache {
613 super_segment_headers_cache: Vec<SuperSegmentHeader>,
614}
615
616impl SuperSegmentHeadersCache {
617 #[inline(always)]
618 fn last_super_segment_header(&self) -> Option<SuperSegmentHeader> {
619 self.super_segment_headers_cache.last().copied()
620 }
621
622 #[inline]
623 fn previous_super_segment_header(
624 &self,
625 target_block_number: BlockNumber,
626 ) -> Option<SuperSegmentHeader> {
627 let block_number = target_block_number.checked_sub(BlockNumber::ONE)?;
628 let index = match self.super_segment_headers_cache.binary_search_by_key(
629 &block_number,
630 |super_segment_header| {
631 super_segment_header
632 .target_beacon_chain_block_number
633 .as_inner()
634 },
635 ) {
636 Ok(found_index) => found_index,
637 Err(insert_index) => insert_index.checked_sub(1)?,
638 };
639
640 self.super_segment_headers_cache.get(index).copied()
641 }
642
643 #[inline(always)]
644 fn get_super_segment_header(
645 &self,
646 local_segment_index: SuperSegmentIndex,
647 ) -> Option<SuperSegmentHeader> {
648 self.super_segment_headers_cache
649 .get(u64::from(local_segment_index) as usize)
650 .copied()
651 }
652
653 #[inline(always)]
654 fn get_super_segment_header_for_segment_index(
655 &self,
656 segment_index: SegmentIndex,
657 ) -> Option<SuperSegmentHeader> {
658 let index = self
659 .super_segment_headers_cache
660 .binary_search_by_key(&segment_index, |super_segment_header| {
661 super_segment_header.max_segment_index.as_inner()
662 })
663 .unwrap_or_else(|insert_index| insert_index);
664
665 let super_segment_header = self.super_segment_headers_cache.get(index).copied()?;
666
667 let max_segment_index = super_segment_header.max_segment_index.as_inner();
668 let first_segment_index = max_segment_index
669 - SegmentIndex::from(u64::from(super_segment_header.num_segments))
670 + SegmentIndex::ONE;
671
672 (first_segment_index..=max_segment_index)
673 .contains(&segment_index)
674 .then_some(super_segment_header)
675 }
676
677 fn add_super_segment_headers(
679 &mut self,
680 mut super_segment_headers: Vec<SuperSegmentHeader>,
681 ) -> Result<Vec<SuperSegmentHeader>, PersistSuperSegmentHeadersError> {
682 self.super_segment_headers_cache
683 .reserve(super_segment_headers.len());
684
685 let mut maybe_last_super_segment_index = self
686 .super_segment_headers_cache
687 .last()
688 .map(|header| header.index.as_inner());
689
690 if let Some(last_super_segment_index) = maybe_last_super_segment_index {
691 super_segment_headers.retain(|super_segment_header| {
693 super_segment_header.index.as_inner() > last_super_segment_index
694 });
695 }
696
697 for super_segment_header in super_segment_headers.iter().copied() {
700 let super_segment_index = super_segment_header.index.as_inner();
701 if let Some(last_super_segment_index) = maybe_last_super_segment_index {
702 if super_segment_index != last_super_segment_index + SuperSegmentIndex::ONE {
703 return Err(
704 PersistSuperSegmentHeadersError::MustFollowLastSegmentIndex {
705 super_segment_index,
706 last_super_segment_index,
707 },
708 );
709 }
710
711 self.super_segment_headers_cache.push(super_segment_header);
712 maybe_last_super_segment_index.replace(super_segment_index);
713 } else {
714 if super_segment_index != SuperSegmentIndex::ZERO {
715 return Err(PersistSuperSegmentHeadersError::FirstSegmentIndexZero {
716 super_segment_index,
717 });
718 }
719
720 self.super_segment_headers_cache.push(super_segment_header);
721 maybe_last_super_segment_index.replace(super_segment_index);
722 }
723 }
724
725 Ok(super_segment_headers)
726 }
727}
728
729#[derive(Debug)]
731struct State<Block, StorageBackend>
732where
733 Block: GenericOwnedBlock,
734{
735 data: StateData<Block>,
736 segment_headers_cache: SegmentHeadersCache,
737 super_segment_headers_cache: SuperSegmentHeadersCache,
738 storage_backend_adapter: AsyncRwLock<StorageBackendAdapter<StorageBackend>>,
739}
740
741impl<Block, StorageBackend> State<Block, StorageBackend>
742where
743 Block: GenericOwnedBlock,
744{
745 #[inline(always)]
746 fn best_tip(&self) -> &ForkTip {
747 self.data
748 .fork_tips
749 .front()
750 .expect("The best block is always present; qed")
751 }
752
753 #[inline(always)]
754 fn best_block(&self) -> &ClientDatabaseBlock<Block> {
755 self.data
756 .blocks
757 .front()
758 .expect("The best block is always present; qed")
759 .first()
760 .expect("The best block is always present; qed")
761 }
762}
763
764#[derive(Debug)]
765struct BlockToPersist<'a, Block>
766where
767 Block: GenericOwnedBlock,
768{
769 block_offset: usize,
770 fork_offset: usize,
771 block: &'a Block,
772 block_details: &'a BlockDetails,
773}
774
775#[derive(Debug)]
776struct PersistedBlock {
777 block_offset: usize,
778 fork_offset: usize,
779 write_location: WriteLocation,
780}
781
782#[derive(Debug)]
783struct ClientDatabaseInnerOptions {
784 block_confirmation_depth: BlockNumber,
785 soft_confirmation_depth: BlockNumber,
786 max_fork_tips: NonZeroUsize,
787 max_fork_tip_distance: BlockNumber,
788}
789
790#[derive(Debug)]
791struct Inner<Block, StorageBackend>
792where
793 Block: GenericOwnedBlock,
794{
795 state: AsyncRwLock<State<Block, StorageBackend>>,
796 options: ClientDatabaseInnerOptions,
797}
798
799#[derive(Debug)]
801pub struct ClientDatabase<Block, StorageBackend>
802where
803 Block: GenericOwnedBlock,
804{
805 inner: Arc<Inner<Block, StorageBackend>>,
806}
807
808impl<Block, StorageBackend> Clone for ClientDatabase<Block, StorageBackend>
809where
810 Block: GenericOwnedBlock,
811{
812 fn clone(&self) -> Self {
813 Self {
814 inner: self.inner.clone(),
815 }
816 }
817}
818
819#[expect(clippy::empty_drop, reason = "Not implemented yet")]
820impl<Block, StorageBackend> Drop for ClientDatabase<Block, StorageBackend>
821where
822 Block: GenericOwnedBlock,
823{
824 fn drop(&mut self) {
825 }
827}
828
829impl<Block, StorageBackend> ChainInfo<Block> for ClientDatabase<Block, StorageBackend>
830where
831 Block: GenericOwnedBlock,
832 StorageBackend: ClientDatabaseStorageBackend,
833{
834 #[inline]
835 fn best_root(&self) -> BlockRoot {
836 self.inner.state.read_blocking().best_tip().root
839 }
840
841 #[inline]
842 fn best_header(&self) -> Block::Header {
843 self.inner
846 .state
847 .read_blocking()
848 .best_block()
849 .header()
850 .clone()
851 }
852
853 #[inline]
854 fn best_header_with_details(&self) -> (Block::Header, BlockDetails) {
855 let state = self.inner.state.read_blocking();
858 let best_block = state.best_block();
859 (
860 best_block.header().clone(),
861 best_block
862 .block_details()
863 .expect("Always present for the best block; qed")
864 .clone(),
865 )
866 }
867
868 #[inline]
870 fn ancestor_header(
871 &self,
872 ancestor_block_number: BlockNumber,
873 descendant_block_root: &BlockRoot,
874 ) -> Option<Block::Header> {
875 let state = self.inner.state.read_blocking();
878 let best_number = state.best_tip().number;
879
880 let ancestor_block_offset =
881 u64::from(best_number.checked_sub(ancestor_block_number)?) as usize;
882 let ancestor_block_candidates = state.data.blocks.get(ancestor_block_offset)?;
883
884 let descendant_block_number = *state.data.block_roots.get(descendant_block_root)?;
885 if ancestor_block_number > descendant_block_number {
886 return None;
887 }
888 let descendant_block_offset =
889 u64::from(best_number.checked_sub(descendant_block_number)?) as usize;
890
891 let mut blocks_range_iter = state
893 .data
894 .blocks
895 .iter()
896 .enumerate()
897 .skip(descendant_block_offset);
898
899 let (_offset, descendant_block_candidates) = blocks_range_iter.next()?;
900 let descendant_header = descendant_block_candidates
901 .iter()
902 .find(|block| &*block.header().header().root() == descendant_block_root)?
903 .header()
904 .header();
905
906 if descendant_block_candidates.len() == 1 || ancestor_block_candidates.len() == 1 {
911 return ancestor_block_candidates
912 .iter()
913 .next()
914 .map(|block| block.header().clone());
915 }
916
917 let mut parent_block_root = &descendant_header.prefix.parent_root;
918
919 for (block_offset, parent_candidates) in blocks_range_iter {
921 let parent_header = parent_candidates
922 .iter()
923 .find(|header| &*header.header().header().root() == parent_block_root)?
924 .header();
925
926 if block_offset == ancestor_block_offset {
928 return Some(parent_header.clone());
929 }
930
931 parent_block_root = &parent_header.header().prefix.parent_root;
932 }
933
934 None
935 }
936
937 #[inline]
938 fn header(&self, block_root: &BlockRoot) -> Option<Block::Header> {
939 let state = self.inner.state.read_blocking();
942 let best_number = state.best_tip().number;
943
944 let block_number = *state.data.block_roots.get(block_root)?;
945 let block_offset = u64::from(best_number.checked_sub(block_number)?) as usize;
946 let block_candidates = state.data.blocks.get(block_offset)?;
947
948 block_candidates.iter().find_map(|block| {
949 let header = block.header();
950
951 if &*header.header().root() == block_root {
952 Some(header.clone())
953 } else {
954 None
955 }
956 })
957 }
958
959 #[inline]
960 fn header_with_details(&self, block_root: &BlockRoot) -> Option<(Block::Header, BlockDetails)> {
961 let state = self.inner.state.read_blocking();
964 let best_number = state.best_tip().number;
965
966 let block_number = *state.data.block_roots.get(block_root)?;
967 let block_offset = u64::from(best_number.checked_sub(block_number)?) as usize;
968 let block_candidates = state.data.blocks.get(block_offset)?;
969
970 block_candidates.iter().find_map(|block| {
971 let header = block.header();
972 let block_details = block.block_details().cloned()?;
973
974 if &*header.header().root() == block_root {
975 Some((header.clone(), block_details))
976 } else {
977 None
978 }
979 })
980 }
981
982 #[inline]
983 async fn block(&self, block_root: &BlockRoot) -> Result<Block, ReadBlockError> {
984 let state = self.inner.state.read().await;
985 let best_number = state.best_tip().number;
986
987 let block_number = *state
988 .data
989 .block_roots
990 .get(block_root)
991 .ok_or(ReadBlockError::UnknownBlockRoot)?;
992 let block_offset = u64::from(
993 best_number
994 .checked_sub(block_number)
995 .expect("Known block roots always have valid block offset; qed"),
996 ) as usize;
997 let block_candidates = state
998 .data
999 .blocks
1000 .get(block_offset)
1001 .expect("Valid block offsets always have block entries; qed");
1002
1003 for block_candidate in block_candidates {
1004 let header = block_candidate.header();
1005
1006 if &*header.header().root() == block_root {
1007 return match block_candidate.full_block() {
1008 FullBlock::InMemory(block) => Ok(block.clone()),
1009 FullBlock::Persisted {
1010 header,
1011 write_location,
1012 } => {
1013 let storage_backend_adapter = state.storage_backend_adapter.read().await;
1014
1015 let storage_item = storage_backend_adapter
1016 .read_storage_item::<StorageItemTemporary>(write_location)
1017 .await?;
1018
1019 let storage_item_block = match storage_item {
1020 StorageItemTemporary::Block(storage_item_block) => storage_item_block,
1021 StorageItemTemporary::SegmentHeaders(_) => {
1022 return Err(ReadBlockError::StorageItemReadError {
1023 error: io::Error::other(
1024 "Unexpected storage item: `SegmentHeaders`",
1025 ),
1026 });
1027 }
1028 StorageItemTemporary::SuperSegmentHeaders(_) => {
1029 return Err(ReadBlockError::StorageItemReadError {
1030 error: io::Error::other(
1031 "Unexpected storage item: `SuperSegmentHeaders`",
1032 ),
1033 });
1034 }
1035 };
1036
1037 let StorageItemTemporaryBlock {
1038 header: _,
1039 body,
1040 mmr_with_block: _,
1041 system_contract_states: _,
1042 } = storage_item_block;
1043
1044 Block::from_buffers(header.buffer().clone(), body)
1045 .ok_or(ReadBlockError::FailedToDecode)
1046 }
1047 };
1048 }
1049 }
1050
1051 unreachable!("Known block root always has block candidate associated with it; qed")
1052 }
1053
1054 #[inline]
1055 fn last_segment_header(&self) -> Option<SegmentHeader> {
1056 let state = self.inner.state.read_blocking();
1059 state.segment_headers_cache.last_segment_header()
1060 }
1061
1062 #[inline]
1063 fn get_segment_header(&self, segment_index: LocalSegmentIndex) -> Option<SegmentHeader> {
1064 let state = self.inner.state.read_blocking();
1067
1068 state
1069 .segment_headers_cache
1070 .get_segment_header(segment_index)
1071 }
1072
1073 fn segment_headers_for_block(&self, block_number: BlockNumber) -> Vec<SegmentHeader> {
1074 let state = self.inner.state.read_blocking();
1077
1078 let Some(last_local_segment_index) = state.segment_headers_cache.max_local_segment_index()
1079 else {
1080 return Vec::new();
1082 };
1083
1084 if Block::Block::SHARD_KIND == RealShardKind::BeaconChain
1086 && block_number == BlockNumber::ONE
1087 {
1088 return vec![
1091 state
1092 .segment_headers_cache
1093 .get_segment_header(LocalSegmentIndex::ZERO)
1094 .expect("Segment headers are stored in monotonically increasing order; qed"),
1095 ];
1096 }
1097
1098 if last_local_segment_index == LocalSegmentIndex::ZERO {
1099 return Vec::new();
1101 }
1102
1103 let mut current_local_segment_index = last_local_segment_index;
1104 loop {
1105 let current_segment_header = state
1108 .segment_headers_cache
1109 .get_segment_header(current_local_segment_index)
1110 .expect("Segment headers are stored in monotonically increasing order; qed");
1111
1112 let target_block_number = current_segment_header.last_archived_block.number()
1114 + BlockNumber::ONE
1115 + self.inner.options.block_confirmation_depth;
1116 if target_block_number == block_number {
1117 let mut headers_for_block = vec![current_segment_header];
1118
1119 let last_archived_block_number = current_segment_header.last_archived_block.number;
1121 let mut local_segment_index = current_local_segment_index - LocalSegmentIndex::ONE;
1122
1123 while let Some(segment_header) = state
1124 .segment_headers_cache
1125 .get_segment_header(local_segment_index)
1126 {
1127 if segment_header.last_archived_block.number == last_archived_block_number {
1128 headers_for_block.insert(0, segment_header);
1129 local_segment_index -= LocalSegmentIndex::ONE;
1130 } else {
1131 break;
1132 }
1133 }
1134
1135 return headers_for_block;
1136 }
1137
1138 if target_block_number > block_number {
1140 if current_local_segment_index > LocalSegmentIndex::ONE {
1142 current_local_segment_index -= LocalSegmentIndex::ONE;
1143 } else {
1144 break;
1145 }
1146 } else {
1147 return Vec::new();
1149 }
1150 }
1151
1152 Vec::new()
1154 }
1155}
1156
1157impl<Block, StorageBackend> ChainInfoWrite<Block> for ClientDatabase<Block, StorageBackend>
1158where
1159 Block: GenericOwnedBlock,
1160 StorageBackend: ClientDatabaseStorageBackend,
1161{
1162 async fn persist_block(
1163 &self,
1164 block: Block,
1165 block_details: BlockDetails,
1166 ) -> Result<(), PersistBlockError> {
1167 let mut state = self.inner.state.write().await;
1168 let best_number = state.best_tip().number;
1169
1170 let header = block.header().header();
1171
1172 let block_number = header.prefix.number;
1173
1174 if best_number == BlockNumber::ZERO && block_number != BlockNumber::ONE {
1175 Self::insert_first_block(&mut state.data, block, block_details);
1177
1178 return Ok(());
1179 }
1180
1181 if block_number == best_number + BlockNumber::ONE {
1182 return Self::insert_new_best_block(state, &self.inner, block, block_details).await;
1183 }
1184
1185 let block_offset = u64::from(
1186 best_number
1187 .checked_sub(block_number)
1188 .ok_or(PersistBlockError::MissingParent)?,
1189 ) as usize;
1190
1191 if block_offset >= u64::from(self.inner.options.block_confirmation_depth) as usize {
1192 return Err(PersistBlockError::OutsideAcceptableRange);
1193 }
1194
1195 let state = &mut *state;
1196
1197 let block_forks = state.data.blocks.get_mut(block_offset).ok_or_else(|| {
1198 error!(
1199 %block_number,
1200 %block_offset,
1201 "Failed to store block fork, header offset is missing despite being within \
1202 acceptable range"
1203 );
1204
1205 PersistBlockError::OutsideAcceptableRange
1206 })?;
1207
1208 for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
1209 if fork_tip.root == header.prefix.parent_root {
1211 state.data.fork_tips.remove(index);
1212 break;
1213 }
1214 }
1215
1216 let block_root = *header.root();
1217 state.data.fork_tips.insert(
1220 1,
1221 ForkTip {
1222 number: block_number,
1223 root: block_root,
1224 },
1225 );
1226 state.data.block_roots.insert(block_root, block_number);
1227 let beacon_chain_block_details = <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1228 .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1229 block_forks.push(ClientDatabaseBlock::InMemory {
1230 block,
1231 block_details,
1232 beacon_chain_block_details,
1233 });
1234
1235 Self::prune_outdated_fork_tips(block_number, &mut state.data, &self.inner.options);
1236
1237 Ok(())
1238 }
1239
1240 async fn persist_segment_headers(
1241 &self,
1242 segment_headers: Vec<SegmentHeader>,
1243 ) -> Result<(), PersistSegmentHeadersError> {
1244 let mut state = self.inner.state.write().await;
1245
1246 let added_segment_headers = state
1247 .segment_headers_cache
1248 .add_segment_headers(segment_headers)?;
1249
1250 if added_segment_headers.is_empty() {
1251 return Ok(());
1252 }
1253
1254 let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1259
1260 let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1261
1262 storage_backend_adapter
1263 .write_storage_item(StorageItemTemporary::SegmentHeaders(
1264 StorageItemTemporarySegmentHeaders {
1265 segment_headers: added_segment_headers,
1266 },
1267 ))
1268 .await?;
1269
1270 Ok(())
1271 }
1272}
1273
1274impl<StorageBackend> BeaconChainInfo for ClientDatabase<OwnedBeaconChainBlock, StorageBackend>
1275where
1276 StorageBackend: ClientDatabaseStorageBackend,
1277{
1278 fn shard_segment_roots(
1279 &self,
1280 block_number: BlockNumber,
1281 ) -> Result<StdArc<[ShardSegmentRoot]>, ShardSegmentRootsError> {
1282 let state = self.inner.state.read_blocking();
1285 let best_number = state.best_tip().number;
1286
1287 let block_offset = u64::from(
1288 best_number
1289 .checked_sub(block_number)
1290 .ok_or(ShardSegmentRootsError::BlockMissing { block_number })?,
1291 ) as usize;
1292
1293 let block = state
1294 .data
1295 .blocks
1296 .get(block_offset)
1297 .ok_or(ShardSegmentRootsError::BlockMissing { block_number })?
1298 .first()
1299 .expect("There is always at least one block candidate; qed");
1300
1301 Ok(StdArc::clone(
1302 &block
1303 .beacon_chain_block_details()
1304 .as_ref()
1305 .expect("Always present in the beacon chain block; qed")
1306 .shard_segment_roots,
1307 ))
1308 }
1309
1310 #[inline]
1311 fn last_super_segment_header(&self) -> Option<SuperSegmentHeader> {
1312 let state = self.inner.state.read_blocking();
1315 state
1316 .super_segment_headers_cache
1317 .last_super_segment_header()
1318 }
1319
1320 #[inline]
1321 fn previous_super_segment_header(
1322 &self,
1323 block_number: BlockNumber,
1324 ) -> Option<SuperSegmentHeader> {
1325 let state = self.inner.state.read_blocking();
1328
1329 state
1330 .super_segment_headers_cache
1331 .previous_super_segment_header(block_number)
1332 }
1333
1334 #[inline]
1335 fn get_super_segment_header(
1336 &self,
1337 super_segment_index: SuperSegmentIndex,
1338 ) -> Option<SuperSegmentHeader> {
1339 let state = self.inner.state.read_blocking();
1342
1343 state
1344 .super_segment_headers_cache
1345 .get_super_segment_header(super_segment_index)
1346 }
1347
1348 fn get_super_segment_header_for_segment_index(
1349 &self,
1350 segment_index: SegmentIndex,
1351 ) -> Option<SuperSegmentHeader> {
1352 let state = self.inner.state.read_blocking();
1355
1356 state
1357 .super_segment_headers_cache
1358 .get_super_segment_header_for_segment_index(segment_index)
1359 }
1360}
1361
1362impl<StorageBackend> BeaconChainInfoWrite for ClientDatabase<OwnedBeaconChainBlock, StorageBackend>
1363where
1364 StorageBackend: ClientDatabaseStorageBackend,
1365{
1366 async fn persist_super_segment_header(
1367 &self,
1368 super_segment_header: SuperSegmentHeader,
1369 ) -> Result<bool, PersistSuperSegmentHeadersError> {
1370 let mut state = self.inner.state.write().await;
1371
1372 let added_super_segment_headers = state
1373 .super_segment_headers_cache
1374 .add_super_segment_headers(vec![super_segment_header])?;
1375
1376 if added_super_segment_headers.is_empty() {
1377 return Ok(false);
1378 }
1379
1380 let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1385
1386 let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1387
1388 storage_backend_adapter
1389 .write_storage_item(StorageItemTemporary::SuperSegmentHeaders(
1390 StorageItemTemporarySuperSegmentHeaders {
1391 super_segment_headers: added_super_segment_headers,
1392 },
1393 ))
1394 .await?;
1395
1396 Ok(true)
1397 }
1398
1399 async fn persist_super_segment_headers(
1400 &self,
1401 super_segment_headers: Vec<SuperSegmentHeader>,
1402 ) -> Result<(), PersistSuperSegmentHeadersError> {
1403 let mut state = self.inner.state.write().await;
1404
1405 let added_super_segment_headers = state
1406 .super_segment_headers_cache
1407 .add_super_segment_headers(super_segment_headers)?;
1408
1409 if added_super_segment_headers.is_empty() {
1410 return Ok(());
1411 }
1412
1413 let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1418
1419 let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1420
1421 storage_backend_adapter
1422 .write_storage_item(StorageItemTemporary::SuperSegmentHeaders(
1423 StorageItemTemporarySuperSegmentHeaders {
1424 super_segment_headers: added_super_segment_headers,
1425 },
1426 ))
1427 .await?;
1428
1429 Ok(())
1430 }
1431}
1432
1433impl<Block, StorageBackend> ClientDatabase<Block, StorageBackend>
1434where
1435 Block: GenericOwnedBlock,
1436 StorageBackend: ClientDatabaseStorageBackend,
1437{
1438 pub async fn open<GBB>(
1442 options: ClientDatabaseOptions<GBB, StorageBackend>,
1443 ) -> Result<Self, ClientDatabaseError>
1444 where
1445 GBB: FnOnce() -> GenesisBlockBuilderResult<Block>,
1446 {
1447 let ClientDatabaseOptions {
1448 write_buffer_size,
1449 block_confirmation_depth,
1450 soft_confirmation_depth,
1451 max_fork_tips,
1452 max_fork_tip_distance,
1453 genesis_block_builder,
1454 storage_backend,
1455 } = options;
1456 if soft_confirmation_depth >= block_confirmation_depth {
1457 return Err(ClientDatabaseError::InvalidSoftConfirmationDepth);
1458 }
1459
1460 if max_fork_tip_distance > block_confirmation_depth {
1461 return Err(ClientDatabaseError::InvalidMaxForkTipDistance);
1462 }
1463
1464 let mut state_data = StateData {
1465 fork_tips: VecDeque::new(),
1466 block_roots: HashMap::default(),
1467 blocks: VecDeque::new(),
1468 };
1469 let mut segment_headers_cache = SegmentHeadersCache {
1470 segment_headers_cache: Vec::new(),
1471 };
1472 let mut super_segment_headers_cache = SuperSegmentHeadersCache {
1473 super_segment_headers_cache: Vec::new(),
1474 };
1475
1476 let options = ClientDatabaseInnerOptions {
1477 block_confirmation_depth,
1478 soft_confirmation_depth,
1479 max_fork_tips,
1480 max_fork_tip_distance,
1481 };
1482
1483 let storage_item_handlers = StorageItemHandlers {
1484 permanent: |_arg| {
1485 Ok(())
1487 },
1488 temporary: |arg| {
1489 let StorageItemHandlerArg {
1490 storage_item,
1491 page_offset,
1492 num_pages,
1493 } = arg;
1494 let storage_item_block = match storage_item {
1495 StorageItemTemporary::Block(storage_item_block) => storage_item_block,
1496 StorageItemTemporary::SegmentHeaders(segment_headers) => {
1497 let num_segment_headers = segment_headers.segment_headers.len();
1498 return match segment_headers_cache
1499 .add_segment_headers(segment_headers.segment_headers)
1500 {
1501 Ok(_) => Ok(()),
1502 Err(error) => {
1503 error!(
1504 %page_offset,
1505 %num_segment_headers,
1506 %error,
1507 "Failed to add segment headers from storage item"
1508 );
1509
1510 Err(ClientDatabaseError::InvalidSegmentHeaders { page_offset })
1511 }
1512 };
1513 }
1514 StorageItemTemporary::SuperSegmentHeaders(super_segment_headers) => {
1515 let num_super_segment_headers =
1516 super_segment_headers.super_segment_headers.len();
1517 return match super_segment_headers_cache
1518 .add_super_segment_headers(super_segment_headers.super_segment_headers)
1519 {
1520 Ok(_) => Ok(()),
1521 Err(error) => {
1522 error!(
1523 %page_offset,
1524 %num_super_segment_headers,
1525 %error,
1526 "Failed to add segment headers from storage item"
1527 );
1528
1529 Err(ClientDatabaseError::InvalidSegmentHeaders { page_offset })
1530 }
1531 };
1532 }
1533 };
1534
1535 let StorageItemTemporaryBlock {
1538 header,
1539 body,
1540 mmr_with_block,
1541 system_contract_states,
1542 } = storage_item_block;
1543
1544 let header = Block::Header::from_buffer(header).map_err(|_buffer| {
1545 error!(%page_offset, "Failed to decode block header from bytes");
1546
1547 ClientDatabaseError::InvalidBlock { page_offset }
1548 })?;
1549 let body = Block::Body::from_buffer(body).map_err(|_buffer| {
1550 error!(%page_offset, "Failed to decode block body from bytes");
1551
1552 ClientDatabaseError::InvalidBlock { page_offset }
1553 })?;
1554
1555 let block_root = *header.header().root();
1556 let block_number = header.header().prefix.number;
1557
1558 state_data.block_roots.insert(block_root, block_number);
1559
1560 let maybe_best_number = state_data
1561 .blocks
1562 .front()
1563 .and_then(|block_forks| block_forks.first())
1564 .map(|best_block| {
1565 let header: &Block::Header = best_block.header();
1567
1568 header.header().prefix.number
1569 });
1570
1571 let block_offset = if let Some(best_number) = maybe_best_number {
1572 if block_number <= best_number {
1573 u64::from(best_number - block_number) as usize
1574 } else {
1575 if block_number - best_number != BlockNumber::ONE {
1577 error!(
1578 %page_offset,
1579 %best_number,
1580 %block_number,
1581 "Invalid new best block number, it must be only one block \
1582 higher than the best block"
1583 );
1584
1585 return Err(ClientDatabaseError::InvalidBlock { page_offset });
1586 }
1587
1588 state_data.blocks.push_front(SmallVec::new());
1589 0
1591 }
1592 } else {
1593 state_data.blocks.push_front(SmallVec::new());
1594 0
1596 };
1597
1598 let Some(block_forks) = state_data.blocks.get_mut(block_offset) else {
1599 return Ok(());
1602 };
1603
1604 let beacon_chain_block_details =
1606 <dyn Any>::downcast_ref::<OwnedBeaconChainBody>(&body)
1607 .map(|body| BeaconChainBlockDetails::from_body(body.body()));
1608 block_forks.push(ClientDatabaseBlock::Persisted {
1609 header,
1610 block_details: BlockDetails {
1611 mmr_with_block,
1612 system_contract_states,
1613 },
1614 beacon_chain_block_details,
1615 write_location: WriteLocation {
1616 page_offset,
1617 num_pages,
1618 },
1619 });
1620
1621 if block_offset == 0 && block_forks.len() == 1 {
1624 Self::confirm_canonical_block(block_number, &mut state_data, &options);
1625 }
1626
1627 Ok(())
1628 },
1629 };
1630
1631 let storage_backend_adapter =
1632 StorageBackendAdapter::open(write_buffer_size, storage_item_handlers, storage_backend)
1633 .await?;
1634
1635 if let Some(best_block) = state_data.blocks.front().and_then(|block_forks| {
1636 block_forks.last()
1639 }) {
1640 let header: &Block::Header = best_block.header();
1642 let header = header.header();
1643 let block_number = header.prefix.number;
1644 let block_root = *header.root();
1645
1646 if !Self::adjust_ancestor_block_forks(&mut state_data.blocks, block_root) {
1647 return Err(ClientDatabaseError::FailedToAdjustAncestorBlockForks);
1648 }
1649
1650 state_data.fork_tips.push_front(ForkTip {
1652 number: block_number,
1653 root: block_root,
1654 });
1655 } else {
1656 let GenesisBlockBuilderResult {
1657 block,
1658 system_contract_states,
1659 } = genesis_block_builder();
1660
1661 let header = block.header().header();
1663 let block_number = header.prefix.number;
1664 let block_root = *header.root();
1665
1666 state_data.fork_tips.push_front(ForkTip {
1667 number: block_number,
1668 root: block_root,
1669 });
1670 state_data.block_roots.insert(block_root, block_number);
1671 let beacon_chain_block_details =
1672 <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1673 .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1674 state_data
1675 .blocks
1676 .push_front(smallvec![ClientDatabaseBlock::InMemory {
1677 block,
1678 block_details: BlockDetails {
1679 system_contract_states,
1680 mmr_with_block: Arc::new({
1681 let mut mmr = BlockMerkleMountainRange::new();
1682 mmr.add_leaf(&block_root);
1683 mmr
1684 })
1685 },
1686 beacon_chain_block_details,
1687 }]);
1688 }
1689
1690 let state = State {
1691 data: state_data,
1692 segment_headers_cache,
1693 super_segment_headers_cache,
1694 storage_backend_adapter: AsyncRwLock::new(storage_backend_adapter),
1695 };
1696
1697 let inner = Inner {
1698 state: AsyncRwLock::new(state),
1699 options,
1700 };
1701
1702 Ok(Self {
1703 inner: Arc::new(inner),
1704 })
1705 }
1706
1707 pub async fn format(
1709 storage_backend: &StorageBackend,
1710 options: ClientDatabaseFormatOptions,
1711 ) -> Result<(), ClientDatabaseFormatError> {
1712 StorageBackendAdapter::format(storage_backend, options).await
1713 }
1714
1715 fn insert_first_block(state: &mut StateData<Block>, block: Block, block_details: BlockDetails) {
1716 let header = block.header().header();
1718 let block_number = header.prefix.number;
1719 let block_root = *header.root();
1720
1721 state.fork_tips.clear();
1722 state.fork_tips.push_front(ForkTip {
1723 number: block_number,
1724 root: block_root,
1725 });
1726 state.block_roots.clear();
1727 state.block_roots.insert(block_root, block_number);
1728 state.blocks.clear();
1729 let beacon_chain_block_details = <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1730 .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1731 state
1732 .blocks
1733 .push_front(smallvec![ClientDatabaseBlock::InMemory {
1734 block,
1735 block_details,
1736 beacon_chain_block_details,
1737 }]);
1738 }
1739
1740 async fn insert_new_best_block(
1741 mut state: AsyncRwLockWriteGuard<'_, State<Block, StorageBackend>>,
1742 inner: &Inner<Block, StorageBackend>,
1743 block: Block,
1744 block_details: BlockDetails,
1745 ) -> Result<(), PersistBlockError> {
1746 let header = block.header().header();
1747 let block_number = header.prefix.number;
1748 let block_root = *header.root();
1749 let parent_root = header.prefix.parent_root;
1750
1751 if !Self::adjust_ancestor_block_forks(&mut state.data.blocks, parent_root) {
1754 return Err(PersistBlockError::MissingParent);
1755 }
1756
1757 {
1759 for (index, fork_tip) in state.data.fork_tips.iter_mut().enumerate() {
1760 if fork_tip.root == parent_root {
1762 state.data.fork_tips.remove(index);
1763 break;
1764 }
1765 }
1766
1767 state.data.fork_tips.push_front(ForkTip {
1768 number: block_number,
1769 root: block_root,
1770 });
1771 state.data.block_roots.insert(block_root, block_number);
1772 let beacon_chain_block_details =
1773 <dyn Any>::downcast_ref::<OwnedBeaconChainBlock>(&block)
1774 .map(|block| BeaconChainBlockDetails::from_body(block.body.body()));
1775 state
1776 .data
1777 .blocks
1778 .push_front(smallvec![ClientDatabaseBlock::InMemory {
1779 block,
1780 block_details: block_details.clone(),
1781 beacon_chain_block_details,
1782 }]);
1783 }
1784
1785 let options = &inner.options;
1786
1787 Self::confirm_canonical_block(block_number, &mut state.data, options);
1788 Self::prune_outdated_fork_tips(block_number, &mut state.data, options);
1789
1790 let state = AsyncRwLockWriteGuard::downgrade_to_upgradable(state);
1795
1796 let mut blocks_to_persist = Vec::new();
1797 for block_offset in u64::from(options.soft_confirmation_depth) as usize.. {
1798 let Some(fork_blocks) = state.data.blocks.get(block_offset) else {
1799 break;
1800 };
1801
1802 let len_before = blocks_to_persist.len();
1803 fork_blocks
1804 .iter()
1805 .enumerate()
1806 .filter_map(|(fork_offset, client_database_block)| {
1807 match client_database_block {
1808 ClientDatabaseBlock::InMemory {
1809 block,
1810 block_details,
1811 beacon_chain_block_details: _,
1812 } => Some(BlockToPersist {
1813 block_offset,
1814 fork_offset,
1815 block,
1816 block_details,
1817 }),
1818 ClientDatabaseBlock::Persisted { .. }
1819 | ClientDatabaseBlock::PersistedConfirmed { .. } => {
1820 None
1822 }
1823 }
1824 })
1825 .collect_into(&mut blocks_to_persist);
1826
1827 if blocks_to_persist.len() == len_before {
1828 break;
1829 }
1830 }
1831
1832 let mut persisted_blocks = Vec::with_capacity(blocks_to_persist.len());
1834 {
1835 let mut storage_backend_adapter = state.storage_backend_adapter.write().await;
1836
1837 for block_to_persist in blocks_to_persist.into_iter().rev() {
1838 let BlockToPersist {
1839 block_offset,
1840 fork_offset,
1841 block,
1842 block_details,
1843 } = block_to_persist;
1844
1845 let write_location = storage_backend_adapter
1846 .write_storage_item(StorageItemTemporary::Block(StorageItemTemporaryBlock {
1847 header: block.header().buffer().clone(),
1848 body: block.body().buffer().clone(),
1849 mmr_with_block: Arc::clone(&block_details.mmr_with_block),
1850 system_contract_states: StdArc::clone(
1851 &block_details.system_contract_states,
1852 ),
1853 }))
1854 .await?;
1855
1856 persisted_blocks.push(PersistedBlock {
1857 block_offset,
1858 fork_offset,
1859 write_location,
1860 });
1861 }
1862 }
1863
1864 let mut state = RwLockUpgradableReadGuard::upgrade(state).await;
1866 for persisted_block in persisted_blocks {
1867 let PersistedBlock {
1868 block_offset,
1869 fork_offset,
1870 write_location,
1871 } = persisted_block;
1872
1873 let block = state
1874 .data
1875 .blocks
1876 .get_mut(block_offset)
1877 .expect("Still holding the same lock since last check; qed")
1878 .get_mut(fork_offset)
1879 .expect("Still holding the same lock since last check; qed");
1880
1881 replace_with_or_abort(block, |block| {
1882 if let ClientDatabaseBlock::InMemory {
1883 block,
1884 block_details,
1885 beacon_chain_block_details,
1886 } = block
1887 {
1888 let (header, _body) = block.split();
1889
1890 ClientDatabaseBlock::Persisted {
1891 header,
1892 block_details,
1893 beacon_chain_block_details,
1894 write_location,
1895 }
1896 } else {
1897 unreachable!("Still holding the same lock since last check; qed");
1898 }
1899 });
1900 }
1901
1902 Ok(())
1906 }
1907
1908 #[must_use]
1913 fn adjust_ancestor_block_forks(
1914 blocks: &mut VecDeque<SmallVec<[ClientDatabaseBlock<Block>; 2]>>,
1915 mut parent_block_root: BlockRoot,
1916 ) -> bool {
1917 let mut ancestor_blocks = blocks.iter_mut();
1918
1919 loop {
1920 if ancestor_blocks.len() == 1 {
1921 break;
1923 }
1924
1925 let Some(parent_blocks) = ancestor_blocks.next() else {
1926 break;
1928 };
1929
1930 let Some(fork_offset_parent_block_root) =
1931 parent_blocks
1932 .iter()
1933 .enumerate()
1934 .find_map(|(fork_offset, fork_block)| {
1935 let fork_header = fork_block.header().header();
1936 if *fork_header.root() == parent_block_root {
1937 Some((fork_offset, fork_header.prefix.parent_root))
1938 } else {
1939 None
1940 }
1941 })
1942 else {
1943 return false;
1944 };
1945
1946 let fork_offset;
1947 (fork_offset, parent_block_root) = fork_offset_parent_block_root;
1948
1949 parent_blocks.swap(0, fork_offset);
1950 }
1951
1952 true
1953 }
1954
1955 fn prune_outdated_fork_tips(
1962 best_number: BlockNumber,
1963 state: &mut StateData<Block>,
1964 options: &ClientDatabaseInnerOptions,
1965 ) {
1966 let state = &mut *state;
1967
1968 let mut candidate_forks_to_remove = Vec::with_capacity(options.max_fork_tips.get());
1971
1972 state.fork_tips.retain(|fork_tip| {
1974 if best_number - fork_tip.number > options.max_fork_tip_distance {
1975 candidate_forks_to_remove.push(*fork_tip);
1976 false
1977 } else {
1978 true
1979 }
1980 });
1981 if state.fork_tips.len() > options.max_fork_tips.get() {
1983 state
1984 .fork_tips
1985 .drain(options.max_fork_tips.get()..)
1986 .collect_into(&mut candidate_forks_to_remove);
1987 }
1988
1989 candidate_forks_to_remove
1991 .retain(|fork_tip| !Self::prune_outdated_fork(best_number, fork_tip, state));
1992 state.fork_tips.extend(candidate_forks_to_remove);
1994 }
1995
1996 #[must_use]
1999 fn prune_outdated_fork(
2000 best_number: BlockNumber,
2001 fork_tip: &ForkTip,
2002 state: &mut StateData<Block>,
2003 ) -> bool {
2004 let block_offset = u64::from(best_number - fork_tip.number) as usize;
2005
2006 let mut block_root_to_prune = fork_tip.root;
2008 let mut pruned_tip = false;
2009 for block_offset in block_offset.. {
2010 let Some(fork_blocks) = state.blocks.get_mut(block_offset) else {
2011 if !pruned_tip {
2012 error!(
2013 %best_number,
2014 ?fork_tip,
2015 block_offset,
2016 "Block offset was not present in the database, this is an implementation \
2017 bug #1"
2018 );
2019 }
2020 break;
2022 };
2023
2024 if fork_blocks.len() == 1 {
2025 if !pruned_tip {
2026 error!(
2027 %best_number,
2028 ?fork_tip,
2029 block_offset,
2030 "Block offset was not present in the database, this is an implementation \
2031 bug #2"
2032 );
2033 }
2034
2035 break;
2037 }
2038
2039 let Some((fork_offset, block)) = fork_blocks
2040 .iter()
2041 .enumerate()
2042 .skip(1)
2044 .find(|(_fork_offset, block)| {
2045 *block.header().header().root() == block_root_to_prune
2046 })
2047 else {
2048 if !pruned_tip {
2049 error!(
2050 %best_number,
2051 ?fork_tip,
2052 block_offset,
2053 "Block offset was not present in the database, this is an implementation \
2054 bug #3"
2055 );
2056 }
2057
2058 break;
2060 };
2061
2062 if block.header().ref_count() > 1 {
2064 break;
2065 }
2066
2067 match block {
2069 ClientDatabaseBlock::InMemory { .. } => {
2070 }
2072 ClientDatabaseBlock::Persisted { .. }
2073 | ClientDatabaseBlock::PersistedConfirmed { .. } => {
2074 pruned_tip = true;
2076 break;
2077 }
2078 }
2079
2080 state.block_roots.get_mut(&block_root_to_prune);
2081 block_root_to_prune = block.header().header().prefix.parent_root;
2082 fork_blocks.swap_remove(fork_offset);
2083
2084 pruned_tip = true;
2085 }
2086
2087 pruned_tip
2088 }
2089
2090 fn confirm_canonical_block(
2093 best_number: BlockNumber,
2094 state_data: &mut StateData<Block>,
2095 options: &ClientDatabaseInnerOptions,
2096 ) {
2097 let block_offset = u64::from(options.block_confirmation_depth + BlockNumber::ONE) as usize;
2101
2102 let Some(fork_blocks) = state_data.blocks.get_mut(block_offset) else {
2103 return;
2105 };
2106
2107 {
2109 let Some(canonical_block) = fork_blocks.first_mut() else {
2110 error!(
2111 %best_number,
2112 block_offset,
2113 "Have not found a canonical block to confirm, this is an implementation bug"
2114 );
2115 return;
2116 };
2117
2118 replace_with_or_abort(canonical_block, |block| match block {
2119 ClientDatabaseBlock::InMemory { .. } => {
2120 error!(
2121 %best_number,
2122 block_offset,
2123 header = ?block.header(),
2124 "Block to be confirmed must not be in memory, this is an implementation bug"
2125 );
2126 block
2127 }
2128 ClientDatabaseBlock::Persisted {
2129 header,
2130 block_details: _,
2131 beacon_chain_block_details,
2132 write_location,
2133 } => ClientDatabaseBlock::PersistedConfirmed {
2134 header,
2135 beacon_chain_block_details,
2136 write_location,
2137 },
2138 ClientDatabaseBlock::PersistedConfirmed { .. } => {
2139 error!(
2140 %best_number,
2141 block_offset,
2142 header = ?block.header(),
2143 "Block to be confirmed must not be confirmed yet, this is an \
2144 implementation bug"
2145 );
2146 block
2147 }
2148 });
2149 }
2150
2151 let mut block_roots_to_prune = fork_blocks
2153 .drain(1..)
2154 .map(|block| *block.header().header().root())
2155 .collect::<Vec<_>>();
2156 let mut current_block_offset = block_offset;
2157 while !block_roots_to_prune.is_empty() {
2158 state_data
2160 .fork_tips
2161 .retain(|fork_tip| !block_roots_to_prune.contains(&fork_tip.root));
2162
2163 for block_root in &block_roots_to_prune {
2165 state_data.block_roots.remove(block_root);
2166 }
2167
2168 if let Some(next_block_offset) = current_block_offset.checked_sub(1) {
2170 current_block_offset = next_block_offset;
2171 } else {
2172 break;
2174 }
2175
2176 let fork_blocks = state_data
2177 .blocks
2178 .get_mut(current_block_offset)
2179 .expect("Lower block offset always exists; qed");
2180
2181 block_roots_to_prune = fork_blocks
2183 .drain_filter(|block| {
2184 let header = block.header().header();
2185
2186 block_roots_to_prune.contains(&header.prefix.parent_root)
2187 })
2188 .map(|block| *block.header().header().root())
2189 .collect();
2190 }
2191 }
2192}