1#[cfg(feature = "rocksdb")]
2use crate::backend::rocksdb::RocksDBBackend;
3use crate::{
4 STORE_METADATA_FILENAME, STORE_SCHEMA_VERSION,
5 api::{
6 StorageBackend, StorageReadView, StorageWriteBatch,
7 tables::{
8 ACCOUNT_CODE_METADATA, ACCOUNT_CODES, ACCOUNT_FLATKEYVALUE, ACCOUNT_TRIE_NODES,
9 BAD_BLOCKS, BLOCK_ACCESS_LISTS, BLOCK_NUMBERS, BODIES, CANONICAL_BLOCK_HASHES,
10 CHAIN_DATA, EXECUTION_WITNESSES, FULLSYNC_HEADERS, HEADERS, INVALID_CHAINS,
11 MISC_VALUES, PENDING_BLOCKS, RECEIPTS_V2, SNAP_STATE, STATE_HISTORY,
12 STORAGE_FLATKEYVALUE, STORAGE_TRIE_NODES, TRANSACTION_LOCATIONS,
13 },
14 },
15 apply_prefix,
16 backend::in_memory::InMemoryBackend,
17 block_data_buffer::BlockDataBuffer,
18 error::StoreError,
19 journal::{FlatDiff, JournalEntry},
20 layering::{Overlay, OverlayCf, TrieLayerCache, TrieWrapper},
21 rlp::{BlockBodyRLP, BlockHeaderRLP, BlockRLP},
22 trie::{BackendTrieDB, BackendTrieDBLocked, classify_trie_key},
23 utils::{ChainDataIndex, SnapStateIndex},
24};
25
26use ethrex_common::{
27 Address, H256, U256,
28 seeded_hash::{SeededHashSet, SeededState},
29 types::{
30 AccountInfo, AccountState, AccountUpdate, Block, BlockBody, BlockHash, BlockHeader,
31 BlockNumber, ChainConfig, Code, CodeMetadata, ForkId, Genesis, GenesisAccount, Index,
32 Receipt, Transaction,
33 block_access_list::BlockAccessList,
34 block_execution_witness::{ExecutionWitness, RpcExecutionWitness},
35 },
36 utils::keccak,
37};
38use ethrex_crypto::{NativeCrypto, keccak::keccak_hash};
39use ethrex_rlp::{
40 decode::{RLPDecode, decode_bytes, decode_rlp_item},
41 encode::RLPEncode,
42};
43use ethrex_trie::{EMPTY_TRIE_HASH, Nibbles, Trie, TrieLogger, TrieNode, TrieWitness};
44use ethrex_trie::{Node, NodeRLP};
45use indexmap::IndexMap;
46use lru::LruCache;
47use rayon::prelude::*;
48use serde::{Deserialize, Serialize};
49use std::{
50 collections::{BTreeMap, HashMap, HashSet, hash_map::Entry},
51 fmt::Debug,
52 hash::BuildHasher,
53 io::Write,
54 num::NonZeroUsize,
55 path::{Path, PathBuf},
56 sync::{
57 Arc, Condvar, Mutex, RwLock,
58 atomic::{AtomicUsize, Ordering},
59 mpsc::{SyncSender, TryRecvError, sync_channel},
60 },
61 thread::JoinHandle,
62};
63use tracing::{debug, error, info, warn};
64
65pub const MAX_WITNESSES: u64 = 128;
67
68pub const DB_COMMIT_THRESHOLD: usize = 128;
72const IN_MEMORY_COMMIT_THRESHOLD: usize = 10000;
73
74pub const BATCH_COMMIT_THRESHOLD: usize = 4;
80
81const KEYS_PER_SHARD: usize = 256;
88const MAX_SHARDS: usize = 64;
89
90fn shard_count(n: usize) -> usize {
92 n.div_ceil(KEYS_PER_SHARD).clamp(1, MAX_SHARDS)
93}
94
95pub const MAX_ROCKSDB_BLOCK_CACHE_SIZE_BYTES: usize = 12 * 1024 * 1024 * 1024;
105
106pub const MIN_ROCKSDB_BLOCK_CACHE_SIZE_BYTES: usize = 512 * 1024 * 1024;
114
115pub const ROCKSDB_BLOCK_CACHE_MEMORY_PERCENT: usize = 40;
124
125pub fn default_rocksdb_block_cache_size() -> usize {
137 let physical = physical_memory_bytes();
138 let cgroup = cgroup_memory_limit_bytes();
139 let limit = [physical, cgroup].into_iter().flatten().min();
140 let size = rocksdb_block_cache_size_for(limit);
141 let source = match (physical, cgroup) {
146 (None, None) => "undetected, falling back to the ceiling",
147 (Some(_), None) => "physical memory",
148 (None, Some(_)) => "cgroup limit",
149 (Some(p), Some(c)) => {
150 if c <= p {
151 "cgroup limit"
152 } else {
153 "physical memory"
154 }
155 }
156 };
157 info!(
158 detected_limit_bytes = ?limit,
159 source,
160 cache_bytes = size,
161 "sized RocksDB block cache"
162 );
163 size
164}
165
166pub fn rocksdb_block_cache_size_for(memory_limit: Option<usize>) -> usize {
169 let Some(limit) = memory_limit else {
170 return MAX_ROCKSDB_BLOCK_CACHE_SIZE_BYTES;
171 };
172 (limit.saturating_mul(ROCKSDB_BLOCK_CACHE_MEMORY_PERCENT) / 100).clamp(
173 MIN_ROCKSDB_BLOCK_CACHE_SIZE_BYTES,
174 MAX_ROCKSDB_BLOCK_CACHE_SIZE_BYTES,
175 )
176}
177
178fn physical_memory_bytes() -> Option<usize> {
180 let meminfo = std::fs::read_to_string("/proc/meminfo").ok()?;
181 let kib: usize = meminfo
182 .lines()
183 .find_map(|line| line.strip_prefix("MemTotal:"))?
184 .split_whitespace()
185 .next()?
186 .parse()
187 .ok()?;
188 kib.checked_mul(1024)
189}
190
191fn cgroup_memory_limit_bytes() -> Option<usize> {
203 let cgroups = std::fs::read_to_string("/proc/self/cgroup").ok()?;
204 let v2 = cgroups
205 .lines()
206 .find_map(|line| line.strip_prefix("0::"))
207 .and_then(|relative| {
208 cgroup_limit_up_to_root(Path::new("/sys/fs/cgroup"), relative, "memory.max")
209 });
210 v2.or_else(|| {
211 let relative = cgroups.lines().find_map(|line| {
212 let mut fields = line.splitn(3, ':');
214 let controllers = fields.nth(1)?;
215 let path = fields.next()?;
216 controllers
217 .split(',')
218 .any(|controller| controller == "memory")
219 .then_some(path)
220 })?;
221 cgroup_limit_up_to_root(
222 Path::new("/sys/fs/cgroup/memory"),
223 relative,
224 "memory.limit_in_bytes",
225 )
226 })
227}
228
229fn cgroup_limit_up_to_root(mount: &Path, relative: &str, file: &str) -> Option<usize> {
232 let mut dir = mount.join(relative.trim_start_matches('/'));
233 let mut limit: Option<usize> = None;
234 loop {
235 if let Some(value) = std::fs::read_to_string(dir.join(file))
236 .ok()
237 .and_then(|contents| contents.trim().parse::<usize>().ok())
238 {
239 limit = Some(limit.map_or(value, |current| current.min(value)));
240 }
241 if dir == mount || !dir.pop() {
242 return limit;
243 }
244 }
245}
246
247#[derive(Debug, Clone, Copy)]
252pub struct StoreConfig {
253 pub rocksdb_block_cache_size: usize,
258 pub code_cache_size: Option<u64>,
260 pub persist_channel_capacity: usize,
265}
266
267impl StoreConfig {
268 pub fn with_rocksdb_block_cache_size(rocksdb_block_cache_size: usize) -> Self {
273 Self {
274 rocksdb_block_cache_size,
275 code_cache_size: None,
276 persist_channel_capacity: DEFAULT_PERSIST_CHANNEL_CAPACITY,
277 }
278 }
279}
280
281impl Default for StoreConfig {
282 fn default() -> Self {
283 Self::with_rocksdb_block_cache_size(default_rocksdb_block_cache_size())
284 }
285}
286
287#[derive(Debug, PartialEq)]
289enum FKVGeneratorControlMessage {
290 Stop,
291 Continue,
292}
293
294const CODE_CACHE_MAX_SIZE: u64 = 2 * 1024 * 1024 * 1024;
307
308const CODE_METADATA_CACHE_MAX_ENTRIES: usize = (16 * 1024 * 1024) / 64;
316
317const FLUSHED_UPTO_KEY: &[u8] = b"bodies_flushed_upto";
319
320const BAD_BLOCKS_KEY: &[u8] = b"bad_blocks";
322
323const MAX_BAD_BLOCKS: usize = 16;
325
326const CODE_CACHE_ENTRY_OVERHEAD: u64 = 160;
332
333struct CodeCache {
344 entries: IndexMap<H256, Code, SeededState>,
345 cache_size: u64,
346 max_size: u64,
347 rng: u64,
349}
350
351impl std::fmt::Debug for CodeCache {
354 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
355 f.debug_struct("CodeCache")
356 .field("len", &self.entries.len())
357 .field("cache_size", &self.cache_size)
358 .field("max_size", &self.max_size)
359 .finish_non_exhaustive()
360 }
361}
362
363impl Default for CodeCache {
364 fn default() -> Self {
365 Self {
366 entries: IndexMap::default(),
367 cache_size: 0,
368 max_size: CODE_CACHE_MAX_SIZE,
369 rng: std::collections::hash_map::RandomState::new().hash_one(0u64) | 1,
371 }
372 }
373}
374
375impl CodeCache {
376 fn entry_size(code: &Code) -> u64 {
378 (code.size() as u64).saturating_add(CODE_CACHE_ENTRY_OVERHEAD)
379 }
380
381 fn get(&self, code_hash: &H256) -> Result<Option<Code>, StoreError> {
382 Ok(self.entries.get(code_hash).cloned())
383 }
384
385 fn contains(&self, code_hash: &H256) -> bool {
386 self.entries.contains_key(code_hash)
387 }
388
389 fn insert(&mut self, code: &Code) -> Result<(), StoreError> {
390 if self.entries.contains_key(&code.hash) {
393 return Ok(());
394 }
395 let size = Self::entry_size(code);
398 if size > self.max_size {
399 return Ok(());
400 }
401
402 while self.cache_size + size > self.max_size && !self.entries.is_empty() {
405 let victim = (self.next_random() % self.entries.len() as u64) as usize;
406 if let Some((_, evicted)) = self.entries.swap_remove_index(victim) {
407 self.cache_size -= Self::entry_size(&evicted);
408 }
409 }
410 self.cache_size += size;
411 self.entries.insert(code.hash, code.clone());
412
413 let cache_len = self.entries.len();
414 let current_size = self.cache_size;
415 debug!(
416 "[ACCOUNT CODE CACHE] cache elements (): {cache_len}, total size: {current_size} bytes"
417 );
418 Ok(())
419 }
420
421 fn next_random(&mut self) -> u64 {
422 self.rng ^= self.rng >> 12;
423 self.rng ^= self.rng << 25;
424 self.rng ^= self.rng >> 27;
425 self.rng.wrapping_mul(0x2545_F491_4F6C_DD1D)
426 }
427}
428
429#[derive(Debug, Clone)]
435pub struct Store {
436 db_path: PathBuf,
438 backend: Arc<dyn StorageBackend>,
440 chain_config: ChainConfig,
442 trie_cache: Arc<RwLock<Arc<TrieLayerCache>>>,
444 flatkeyvalue_control_tx: std::sync::mpsc::SyncSender<FKVGeneratorControlMessage>,
446 block_data_buffer: Arc<RwLock<Arc<BlockDataBuffer>>>,
448 persist_tx: std::sync::mpsc::SyncSender<PersistMessage>,
452 pending_trie_roots: Arc<PendingTrieRoots>,
456 latest_block_header: LatestBlockHeaderCache,
459 last_computed_flatkeyvalue: Arc<RwLock<Vec<u8>>>,
461
462 account_code_cache: Arc<Mutex<CodeCache>>,
467
468 code_metadata_cache: Arc<Mutex<LruCache<H256, CodeMetadata, SeededState>>>,
475
476 fcu_lock: Arc<tokio::sync::Mutex<()>>,
479
480 safe_commit_root: Arc<RwLock<H256>>,
486
487 journal_pruning_paused: Arc<std::sync::atomic::AtomicBool>,
497
498 background_threads: Arc<ThreadList>,
499}
500
501#[derive(Debug, Default)]
502struct ThreadList {
503 list: Vec<JoinHandle<()>>,
504}
505
506impl Drop for ThreadList {
507 fn drop(&mut self) {
508 for handle in self.list.drain(..) {
509 let _ = handle.join();
510 }
511 }
512}
513
514pub type StorageTrieNodes = Vec<(H256, Vec<(Nibbles, Vec<u8>)>)>;
519type StorageTries = HashMap<Address, (TrieWitness, Trie)>;
520
521#[derive(Debug, Clone, Copy, PartialEq, Eq)]
525pub enum EngineType {
526 InMemory,
528 #[cfg(feature = "rocksdb")]
530 RocksDB,
531}
532
533pub struct UpdateBatch {
538 pub account_updates: Vec<TrieNode>,
540 pub storage_updates: Vec<(H256, Vec<TrieNode>)>,
542 pub blocks: Vec<Block>,
544 pub receipts: Vec<(H256, Vec<Receipt>)>,
546 pub code_updates: Vec<(H256, Code)>,
548 pub commit_depth: Option<usize>,
556 pub wait_for_flush: bool,
567}
568
569pub struct BackfilledBlock {
577 pub header: BlockHeader,
578 pub body: BlockBody,
579 pub receipts: Vec<Receipt>,
580 pub index_transactions: bool,
581}
582
583pub type StorageUpdates = Vec<(H256, Vec<(Nibbles, Vec<u8>)>)>;
585
586pub struct AccountUpdatesList {
591 pub state_trie_hash: H256,
593 pub state_updates: Vec<(Nibbles, Vec<u8>)>,
595 pub storage_updates: StorageUpdates,
597 pub code_updates: Vec<(H256, Code)>,
599}
600
601pub(crate) fn encode_tx_location_operand(
612 block_number: BlockNumber,
613 block_hash: BlockHash,
614 index: Index,
615) -> Vec<u8> {
616 vec![(block_number, block_hash, index)].encode_to_vec()
617}
618
619pub fn tx_locations_merge(
637 existing: Option<&[u8]>,
638 operands: impl IntoIterator<Item = impl AsRef<[u8]>>,
639) -> Option<Vec<u8>> {
640 fn fold_chunk(
644 list: &mut Vec<(BlockNumber, BlockHash, Index)>,
645 bytes: &[u8],
646 what: &str,
647 ) -> bool {
648 match <Vec<(BlockNumber, BlockHash, Index)>>::decode(bytes) {
649 Ok(entries) => {
650 for (bn, bh, idx) in entries {
651 list.retain(|(_, existing_bh, _)| *existing_bh != bh);
652 list.push((bn, bh, idx));
653 }
654 true
655 }
656 Err(e) => {
657 error!(
658 "tx_locations_merge: failed to decode {what} ({} bytes): {e}; \
659 aborting merge to avoid silent data loss",
660 bytes.len()
661 );
662 false
663 }
664 }
665 }
666
667 let mut list: Vec<(BlockNumber, BlockHash, Index)> = Vec::new();
668
669 if let Some(bytes) = existing
671 && !fold_chunk(&mut list, bytes, "existing value")
672 {
673 return None;
674 }
675 for op in operands {
676 if !fold_chunk(&mut list, op.as_ref(), "operand") {
677 return None;
678 }
679 }
680 Some(list.encode_to_vec())
681}
682
683impl Store {
684 pub async fn wait_for_persistence_idle(&self) -> Result<(), StoreError> {
695 let tx = self.persist_tx.clone();
696 tokio::task::spawn_blocking(move || {
697 let (ack_tx, ack_rx) = sync_channel::<Result<(), StoreError>>(1);
698 tx.send(PersistMessage::Ping(ack_tx))
699 .map_err(|e| StoreError::Custom(format!("wait_for_persistence_idle send: {e}")))?;
700 ack_rx
701 .recv()
702 .map_err(|e| StoreError::Custom(format!("wait_for_persistence_idle ack: {e}")))?
703 })
704 .await
705 .map_err(|e| StoreError::Custom(format!("wait_for_persistence_idle join: {e}")))?
706 }
707
708 pub async fn shutdown(&self) -> Result<(), StoreError> {
727 let tx = self.persist_tx.clone();
728 let backend = self.backend.clone();
729 tokio::task::spawn_blocking(move || {
730 let (ack_tx, ack_rx) = sync_channel::<Result<(), StoreError>>(1);
731 tx.send(PersistMessage::Shutdown { ack: ack_tx })
732 .map_err(|e| StoreError::Custom(format!("shutdown send: {e}")))?;
733 ack_rx
734 .recv()
735 .map_err(|e| StoreError::Custom(format!("shutdown ack: {e}")))??;
736 backend.flush()
739 })
740 .await
741 .map_err(|e| StoreError::Custom(format!("shutdown join: {e}")))?
742 }
743
744 pub async fn add_block(&self, block: Block) -> Result<(), StoreError> {
747 self.add_blocks(vec![block]).await
748 }
749
750 pub async fn add_blocks(&self, blocks: Vec<Block>) -> Result<(), StoreError> {
753 let db = self.backend.clone();
754 tokio::task::spawn_blocking(move || {
755 let mut tx = db.begin_write()?;
756
757 for block in blocks {
758 write_block_data(
759 tx.as_mut(),
760 block.header.number,
761 block.hash(),
762 &block.header,
763 &block.body,
764 )?;
765 }
766
767 tx.commit()
768 })
769 .await
770 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
771 }
772
773 pub async fn add_block_header(
775 &self,
776 block_hash: BlockHash,
777 block_header: BlockHeader,
778 ) -> Result<(), StoreError> {
779 let hash_key = block_hash.encode_to_vec();
780 let header_value = BlockHeaderRLP::from(block_header).into_vec();
781 self.write_async(HEADERS, hash_key, header_value).await
782 }
783
784 pub async fn add_block_headers(
786 &self,
787 block_headers: Vec<BlockHeader>,
788 ) -> Result<(), StoreError> {
789 let mut txn = self.backend.begin_write()?;
790
791 for header in block_headers {
792 let block_hash = header.hash();
793 let block_number = header.number;
794 let hash_key = block_hash.encode_to_vec();
795 let header_value = BlockHeaderRLP::from(header).into_vec();
796
797 txn.put(HEADERS, &hash_key, &header_value)?;
798
799 let number_key = block_number.to_le_bytes().to_vec();
800 txn.put(BLOCK_NUMBERS, &hash_key, &number_key)?;
801 }
802 txn.commit()?;
803 Ok(())
804 }
805
806 pub fn get_block_header(
808 &self,
809 block_number: BlockNumber,
810 ) -> Result<Option<BlockHeader>, StoreError> {
811 let latest = self.latest_block_header.get();
812 if block_number == latest.number {
813 return Ok(Some((*latest).clone()));
814 }
815 let Some(block_hash) = self.get_canonical_block_hash_sync(block_number)? else {
820 return Ok(None);
821 };
822 self.get_block_header_by_hash(block_hash)
823 }
824
825 pub async fn add_block_body(
827 &self,
828 block_hash: BlockHash,
829 block_body: BlockBody,
830 ) -> Result<(), StoreError> {
831 let hash_key = block_hash.encode_to_vec();
832 let body_value = BlockBodyRLP::from(block_body).into_vec();
833 self.write_async(BODIES, hash_key, body_value).await
834 }
835
836 pub async fn add_backfilled_blocks(
854 &self,
855 blocks: Vec<BackfilledBlock>,
856 new_earliest: BlockNumber,
857 ) -> Result<(), StoreError> {
858 let backend = self.backend.clone();
859 tokio::task::spawn_blocking(move || {
860 let mut txn = backend.begin_write()?;
861 for block in &blocks {
862 let hash = block.header.hash();
863 let hash_key = hash.encode_to_vec();
864 txn.put(
865 BODIES,
866 &hash_key,
867 &BlockBodyRLP::from(block.body.clone()).into_vec(),
868 )?;
869 for (index, receipt) in block.receipts.iter().enumerate() {
870 txn.put(
871 RECEIPTS_V2,
872 &receipt_key(&hash, index as u64),
873 &receipt.encode_storage(),
874 )?;
875 }
876 if block.index_transactions {
877 for (index, transaction) in block.body.transactions.iter().enumerate() {
878 txn.merge(
879 TRANSACTION_LOCATIONS,
880 transaction.hash(&NativeCrypto).as_bytes(),
881 &encode_tx_location_operand(block.header.number, hash, index as u64),
882 )?;
883 }
884 }
885 }
886 txn.put(
887 CHAIN_DATA,
888 &chain_data_key(ChainDataIndex::EarliestBlockNumber),
889 &new_earliest.to_le_bytes(),
890 )?;
891 txn.commit()
892 })
893 .await
894 .map_err(|e| StoreError::Custom(format!("Backfill write task panicked: {e}")))?
895 }
896
897 pub fn rocksdb_stats(&self) -> Option<crate::api::RocksDbStats> {
901 self.backend.db_stats()
902 }
903
904 pub async fn get_block_body(
906 &self,
907 block_number: BlockNumber,
908 ) -> Result<Option<BlockBody>, StoreError> {
909 let Some(block_hash) = self.get_canonical_block_hash_sync(block_number)? else {
910 return Ok(None);
911 };
912
913 self.get_block_body_by_hash(block_hash).await
914 }
915
916 pub async fn remove_block(&self, block_number: BlockNumber) -> Result<(), StoreError> {
918 let Some(hash) = self.get_canonical_block_hash_sync(block_number)? else {
919 return Ok(());
920 };
921
922 let backend = self.backend.clone();
923 tokio::task::spawn_blocking(move || {
924 let hash_key = hash.encode_to_vec();
925
926 let mut txn = backend.begin_write()?;
927 txn.delete(
928 CANONICAL_BLOCK_HASHES,
929 block_number.to_le_bytes().as_slice(),
930 )?;
931 txn.delete(BODIES, &hash_key)?;
932 txn.delete(HEADERS, &hash_key)?;
933 txn.delete(BLOCK_NUMBERS, &hash_key)?;
934 txn.commit()
935 })
936 .await
937 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
938 }
939
940 pub async fn get_block_bodies(
942 &self,
943 from: BlockNumber,
944 to: BlockNumber,
945 ) -> Result<Vec<Option<BlockBody>>, StoreError> {
946 let buffer = self.buffer()?;
948 let backend = self.backend.clone();
949 tokio::task::spawn_blocking(move || {
950 let numbers: Vec<BlockNumber> = (from..=to).collect();
951 let mut block_bodies = Vec::new();
952
953 let txn = backend.begin_read()?;
954 for number in numbers {
955 let Some(hash) = txn
956 .get(CANONICAL_BLOCK_HASHES, number.to_le_bytes().as_slice())?
957 .map(|bytes| H256::decode(bytes.as_slice()))
958 .transpose()?
959 else {
960 block_bodies.push(None);
961 continue;
962 };
963 if let Some(body) = buffer.get_body(&hash) {
966 block_bodies.push(Some(body));
967 continue;
968 }
969 let hash_key = hash.encode_to_vec();
970 let block_body_opt = txn
971 .get(BODIES, &hash_key)?
972 .map(|bytes| BlockBodyRLP::from_bytes(bytes).to())
973 .transpose()
974 .map_err(StoreError::from)?;
975
976 block_bodies.push(block_body_opt);
977 }
978
979 Ok(block_bodies)
980 })
981 .await
982 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
983 }
984
985 pub async fn get_block_bodies_by_hash(
987 &self,
988 hashes: Vec<BlockHash>,
989 ) -> Result<Vec<BlockBody>, StoreError> {
990 let buffer = self.buffer()?;
991 let backend = self.backend.clone();
992 tokio::task::spawn_blocking(move || {
994 let txn = backend.begin_read()?;
995 let mut block_bodies = Vec::new();
996 for hash in hashes {
997 if let Some(body) = buffer.get_body(&hash) {
1000 block_bodies.push(body);
1001 continue;
1002 }
1003 let hash_key = hash.encode_to_vec();
1004
1005 let Some(block_body) = txn
1006 .get(BODIES, &hash_key)?
1007 .map(|bytes| BlockBodyRLP::from_bytes(bytes).to())
1008 .transpose()
1009 .map_err(StoreError::from)?
1010 else {
1011 return Err(StoreError::Custom(format!(
1012 "Block body not found for hash: {hash}"
1013 )));
1014 };
1015 block_bodies.push(block_body);
1016 }
1017 Ok(block_bodies)
1018 })
1019 .await
1020 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
1021 }
1022
1023 pub async fn get_block_body_by_hash(
1025 &self,
1026 block_hash: BlockHash,
1027 ) -> Result<Option<BlockBody>, StoreError> {
1028 if let Some(b) = self.buffer()?.get_body(&block_hash) {
1029 return Ok(Some(b));
1030 }
1031 self.read_async(BODIES, block_hash.encode_to_vec())
1032 .await?
1033 .map(|bytes| BlockBodyRLP::from_bytes(bytes).to())
1034 .transpose()
1035 .map_err(StoreError::from)
1036 }
1037
1038 pub fn get_block_header_by_hash(
1039 &self,
1040 block_hash: BlockHash,
1041 ) -> Result<Option<BlockHeader>, StoreError> {
1042 let latest = self.latest_block_header.get();
1043 if block_hash == latest.hash() {
1044 return Ok(Some((*latest).clone()));
1045 }
1046 if let Some(h) = self.buffer()?.get_header(&block_hash) {
1047 return Ok(Some(h));
1048 }
1049 self.load_block_header_by_hash(block_hash)
1050 }
1051
1052 pub fn add_pending_block(&self, block: Block) -> Result<(), StoreError> {
1053 let block_hash = block.hash();
1054 let block_value = BlockRLP::from(block).into_vec();
1055 self.write(PENDING_BLOCKS, block_hash.as_bytes().to_vec(), block_value)
1056 }
1057
1058 pub async fn get_pending_block(
1059 &self,
1060 block_hash: BlockHash,
1061 ) -> Result<Option<Block>, StoreError> {
1062 self.read_async(PENDING_BLOCKS, block_hash.as_bytes().to_vec())
1063 .await?
1064 .map(|bytes| BlockRLP::from_bytes(bytes).to())
1065 .transpose()
1066 .map_err(StoreError::from)
1067 }
1068
1069 pub async fn add_block_number(
1071 &self,
1072 block_hash: BlockHash,
1073 block_number: BlockNumber,
1074 ) -> Result<(), StoreError> {
1075 let number_value = block_number.to_le_bytes().to_vec();
1076 self.write_async(BLOCK_NUMBERS, block_hash.encode_to_vec(), number_value)
1077 .await
1078 }
1079
1080 pub async fn get_block_number(
1082 &self,
1083 block_hash: BlockHash,
1084 ) -> Result<Option<BlockNumber>, StoreError> {
1085 if let Some(n) = self.buffer()?.get_number(&block_hash) {
1086 return Ok(Some(n));
1087 }
1088 self.read_async(BLOCK_NUMBERS, block_hash.encode_to_vec())
1089 .await?
1090 .map(decode_block_number)
1091 .transpose()
1092 }
1093
1094 pub async fn add_transaction_location(
1096 &self,
1097 transaction_hash: H256,
1098 block_number: BlockNumber,
1099 block_hash: BlockHash,
1100 index: Index,
1101 ) -> Result<(), StoreError> {
1102 self.add_transaction_locations(vec![(transaction_hash, block_number, block_hash, index)])
1103 .await
1104 }
1105
1106 pub async fn add_transaction_locations(
1108 &self,
1109 locations: Vec<(H256, BlockNumber, BlockHash, Index)>,
1110 ) -> Result<(), StoreError> {
1111 let db = self.backend.clone();
1112 tokio::task::spawn_blocking(move || {
1113 let mut tx = db.begin_write()?;
1114 for (tx_hash, block_number, block_hash, index) in locations {
1115 tx.merge(
1116 TRANSACTION_LOCATIONS,
1117 tx_hash.as_bytes(),
1118 &encode_tx_location_operand(block_number, block_hash, index),
1119 )?;
1120 }
1121 tx.commit()
1122 })
1123 .await
1124 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
1125 }
1126
1127 pub async fn get_transaction_location(
1129 &self,
1130 transaction_hash: H256,
1131 ) -> Result<Option<(BlockNumber, BlockHash, Index)>, StoreError> {
1132 let buffered = self.buffer()?.get_tx_locations(&transaction_hash);
1133 let db = self.backend.clone();
1134 tokio::task::spawn_blocking(move || {
1135 let tx = db.begin_read()?;
1136 let mut locations = buffered;
1137 if let Some(bytes) = tx.get(TRANSACTION_LOCATIONS, transaction_hash.as_bytes())? {
1138 locations.extend(<Vec<(BlockNumber, BlockHash, Index)>>::decode(&bytes)?);
1139 }
1140 for (block_number, block_hash, index) in locations {
1141 let canonical_hash = tx
1142 .get(
1143 CANONICAL_BLOCK_HASHES,
1144 block_number.to_le_bytes().as_slice(),
1145 )?
1146 .map(|bytes| H256::decode(bytes.as_slice()))
1147 .transpose()?;
1148 if canonical_hash == Some(block_hash) {
1149 return Ok(Some((block_number, block_hash, index)));
1150 }
1151 }
1152 Ok(None)
1153 })
1154 .await
1155 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
1156 }
1157
1158 pub async fn add_receipt(
1160 &self,
1161 block_hash: BlockHash,
1162 index: Index,
1163 receipt: Receipt,
1164 ) -> Result<(), StoreError> {
1165 let key = receipt_key(&block_hash, index);
1166 let value = receipt.encode_storage();
1170 self.write_async(RECEIPTS_V2, key, value).await
1171 }
1172
1173 pub async fn add_receipts(
1175 &self,
1176 block_hash: BlockHash,
1177 receipts: Vec<Receipt>,
1178 ) -> Result<(), StoreError> {
1179 let batch_items: Vec<_> = receipts
1180 .into_iter()
1181 .enumerate()
1182 .map(|(index, receipt)| {
1183 let key = receipt_key(&block_hash, index as u64);
1184 let value = receipt.encode_storage();
1185 (key, value)
1186 })
1187 .collect();
1188 self.write_batch_async(RECEIPTS_V2, batch_items).await
1189 }
1190
1191 pub async fn get_receipt(
1193 &self,
1194 block_number: BlockNumber,
1195 index: Index,
1196 ) -> Result<Option<Receipt>, StoreError> {
1197 let Some(block_hash) = self.get_canonical_block_hash(block_number).await? else {
1199 return Ok(None);
1200 };
1201 self.get_receipt_by_block_hash(block_hash, index).await
1202 }
1203
1204 async fn get_receipt_by_block_hash(
1206 &self,
1207 block_hash: BlockHash,
1208 index: Index,
1209 ) -> Result<Option<Receipt>, StoreError> {
1210 if let Some(r) = self.buffer()?.get_receipt(&block_hash, index) {
1211 return Ok(Some(r));
1212 }
1213 let key = receipt_key(&block_hash, index);
1214 self.read_async(RECEIPTS_V2, key)
1215 .await?
1216 .map(|bytes| Receipt::decode_storage(bytes.as_slice()))
1217 .transpose()
1218 .map_err(StoreError::from)
1219 }
1220
1221 pub fn code_cache_budget_bytes(&self) -> u64 {
1228 self.account_code_cache
1229 .lock()
1230 .map(|cache| cache.max_size)
1231 .unwrap_or(0)
1232 }
1233
1234 pub fn get_account_code(&self, code_hash: H256) -> Result<Option<Code>, StoreError> {
1241 if let Some(code) = self.buffer()?.get_code(&code_hash) {
1242 return Ok(Some(code));
1243 }
1244 if let Some(code) = self
1246 .account_code_cache
1247 .lock()
1248 .map_err(|_| StoreError::LockError)?
1249 .get(&code_hash)?
1250 {
1251 return Ok(Some(code));
1252 }
1253
1254 let Some(bytes) = self
1255 .backend
1256 .begin_read()?
1257 .get(ACCOUNT_CODES, code_hash.as_bytes())?
1258 else {
1259 return Ok(None);
1260 };
1261 let (bytecode_slice, jumpdests) = decode_bytes(&bytes)?;
1262 let (jumpdests, was_legacy) = decode_jumpdests(bytecode_slice, jumpdests)?;
1263 let code = Code::from_parts_unchecked(code_hash, bytecode_slice, jumpdests);
1264
1265 if was_legacy && let Ok(handle) = tokio::runtime::Handle::try_current() {
1272 let buf = encode_code(&code);
1273 let hash_key = code_hash.0.to_vec();
1274 let backend = self.backend.clone();
1275 handle.spawn(async move {
1276 if let Err(e) = async {
1277 let mut tx = backend.begin_write()?;
1278 tx.put(ACCOUNT_CODES, &hash_key, &buf)?;
1279 tx.commit()
1280 }
1281 .await
1282 {
1283 tracing::warn!("Failed to rewrite legacy jumpdests during backfill: {e}");
1284 }
1285 });
1286 }
1287
1288 self.account_code_cache
1290 .lock()
1291 .map_err(|_| StoreError::LockError)?
1292 .insert(&code)?;
1293
1294 Ok(Some(code))
1295 }
1296
1297 pub fn get_account_codes_batch(
1312 &self,
1313 code_hashes: &[H256],
1314 ) -> Result<Vec<Option<Code>>, StoreError> {
1315 let mut out: Vec<Option<Code>> = vec![None; code_hashes.len()];
1316 let mut pending: HashMap<H256, Vec<usize>> = HashMap::new();
1318
1319 {
1320 let buffer = self.buffer()?;
1321 let cache = self
1322 .account_code_cache
1323 .lock()
1324 .map_err(|_| StoreError::LockError)?;
1325 for (i, hash) in code_hashes.iter().enumerate() {
1326 if let Some(code) = buffer.get_code(hash) {
1327 out[i] = Some(code);
1328 } else if let Some(code) = cache.get(hash)? {
1329 out[i] = Some(code);
1330 } else {
1331 pending.entry(*hash).or_default().push(i);
1332 }
1333 }
1334 }
1335
1336 if pending.is_empty() {
1337 return Ok(out);
1338 }
1339
1340 let mut missing: Vec<H256> = pending.keys().copied().collect();
1341 missing.sort_unstable();
1342
1343 const KEYS_PER_SHARD: usize = 256;
1365 const MIN_SHARD_CAP: usize = 64;
1366 let parallelism = std::thread::available_parallelism().map_or(8, |p| p.get());
1367 let max_shards = parallelism.saturating_mul(2).max(MIN_SHARD_CAP);
1368 let shards = missing.len().div_ceil(KEYS_PER_SHARD).min(max_shards);
1369 let read_view = self.backend.begin_read()?;
1370 let decode = |hash: &H256, value: Option<Vec<u8>>| -> Result<Option<Code>, StoreError> {
1373 let Some(bytes) = value else { return Ok(None) };
1374 let (bytecode_slice, jumpdests) = decode_bytes(&bytes)?;
1375 let (jumpdests, _was_legacy) = decode_jumpdests(bytecode_slice, jumpdests)?;
1378 Ok(Some(Code::from_parts_unchecked(
1379 *hash,
1380 bytecode_slice,
1381 jumpdests,
1382 )))
1383 };
1384 let decoded: Vec<Result<Option<Code>, StoreError>> = if shards > parallelism {
1385 let chunk = missing.len().div_ceil(shards);
1386 let rv = read_view.as_ref();
1387 let read_shard = |hashes: &[H256]| -> Vec<Result<Option<Code>, StoreError>> {
1388 let keys: Vec<&[u8]> = hashes.iter().map(|h| h.as_bytes()).collect();
1389 rv.multi_get(ACCOUNT_CODES, &keys)
1390 .into_iter()
1391 .zip(hashes.iter())
1392 .map(|(value, hash)| decode(hash, value?))
1393 .collect()
1394 };
1395 std::thread::scope(|scope| {
1396 let handles: Vec<_> = missing
1397 .chunks(chunk)
1398 .map(|ck| (ck, scope.spawn(move || read_shard(ck))))
1399 .collect();
1400 handles
1401 .into_iter()
1402 .flat_map(|(shard, handle)| {
1403 handle.join().unwrap_or_else(|_| {
1408 shard
1409 .iter()
1410 .map(|_| {
1411 Err(StoreError::Custom(
1412 "account code shard panicked".to_string(),
1413 ))
1414 })
1415 .collect()
1416 })
1417 })
1418 .collect()
1419 })
1420 } else {
1421 let rv = read_view.as_ref();
1424 missing
1425 .par_iter()
1426 .map(|hash| decode(hash, rv.get(ACCOUNT_CODES, hash.as_bytes())?))
1427 .collect()
1428 };
1429
1430 let mut fetched: Vec<Code> = Vec::new();
1431 for (hash, code) in missing.iter().zip(decoded.into_iter()) {
1432 let Some(code) = code? else { continue };
1433 for &i in pending.get(hash).into_iter().flatten() {
1434 out[i] = Some(code.clone());
1435 }
1436 fetched.push(code);
1437 }
1438
1439 if !fetched.is_empty() {
1440 let mut cache = self
1441 .account_code_cache
1442 .lock()
1443 .map_err(|_| StoreError::LockError)?;
1444 for code in &fetched {
1445 cache.insert(code)?;
1446 }
1447 }
1448
1449 Ok(out)
1450 }
1451
1452 pub fn code_exists(&self, code_hash: H256) -> Result<bool, StoreError> {
1457 if self.buffer()?.get_code(&code_hash).is_some() {
1461 return Ok(true);
1462 }
1463 if self
1465 .account_code_cache
1466 .lock()
1467 .map_err(|_| StoreError::LockError)?
1468 .contains(&code_hash)
1469 {
1470 return Ok(true);
1471 }
1472 Ok(self
1474 .backend
1475 .begin_read()?
1476 .get(ACCOUNT_CODES, code_hash.as_bytes())?
1477 .is_some())
1478 }
1479
1480 pub fn get_code_metadata(&self, code_hash: H256) -> Result<Option<CodeMetadata>, StoreError> {
1485 use ethrex_common::constants::EMPTY_KECCAK_HASH;
1486
1487 if code_hash == *EMPTY_KECCAK_HASH {
1489 return Ok(Some(CodeMetadata { length: 0 }));
1490 }
1491
1492 if let Some(metadata) = self
1494 .code_metadata_cache
1495 .lock()
1496 .map_err(|_| StoreError::LockError)?
1497 .get(&code_hash)
1498 .copied()
1499 {
1500 return Ok(Some(metadata));
1501 }
1502
1503 let metadata = if let Some(bytes) = self
1505 .backend
1506 .begin_read()?
1507 .get(ACCOUNT_CODE_METADATA, code_hash.as_bytes())?
1508 {
1509 let length =
1510 u64::from_be_bytes(bytes.try_into().map_err(|_| {
1511 StoreError::Custom("Invalid metadata length encoding".to_string())
1512 })?);
1513 CodeMetadata { length }
1514 } else {
1515 let Some(code) = self.get_account_code(code_hash)? else {
1517 return Ok(None);
1518 };
1519 let metadata = CodeMetadata {
1520 length: code.len() as u64,
1521 };
1522
1523 if let Ok(handle) = tokio::runtime::Handle::try_current() {
1531 let metadata_buf = metadata.length.to_be_bytes().to_vec();
1532 let hash_key = code_hash.0.to_vec();
1533 let backend = self.backend.clone();
1534 handle.spawn(async move {
1535 if let Err(e) = async {
1536 let mut tx = backend.begin_write()?;
1537 tx.put(ACCOUNT_CODE_METADATA, &hash_key, &metadata_buf)?;
1538 tx.commit()
1539 }
1540 .await
1541 {
1542 tracing::warn!("Failed to write code metadata during backfill: {}", e);
1543 }
1544 });
1545 }
1546
1547 metadata
1548 };
1549
1550 self.code_metadata_cache
1552 .lock()
1553 .map_err(|_| StoreError::LockError)?
1554 .put(code_hash, metadata);
1555
1556 Ok(Some(metadata))
1557 }
1558
1559 pub async fn add_account_code(&self, code: Code) -> Result<(), StoreError> {
1561 let hash_key = code.hash.0.to_vec();
1562 let buf = encode_code(&code);
1563 let metadata_buf = (code.len() as u64).to_be_bytes();
1564
1565 let backend = self.backend.clone();
1567 tokio::task::spawn_blocking(move || {
1568 let mut tx = backend.begin_write()?;
1569 tx.put(ACCOUNT_CODES, &hash_key, &buf)?;
1570 tx.put(ACCOUNT_CODE_METADATA, &hash_key, &metadata_buf)?;
1571 tx.commit()
1572 })
1573 .await
1574 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
1575 }
1576
1577 pub async fn clear_snap_state(&self) -> Result<(), StoreError> {
1579 let db = self.backend.clone();
1580 tokio::task::spawn_blocking(move || db.clear_table(SNAP_STATE))
1581 .await
1582 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
1583 }
1584
1585 pub async fn get_transaction_by_hash(
1586 &self,
1587 transaction_hash: H256,
1588 ) -> Result<Option<Transaction>, StoreError> {
1589 let (_block_number, block_hash, index) =
1590 match self.get_transaction_location(transaction_hash).await? {
1591 Some(location) => location,
1592 None => return Ok(None),
1593 };
1594 self.get_transaction_by_location(block_hash, index).await
1595 }
1596
1597 pub async fn get_transaction_by_location(
1598 &self,
1599 block_hash: H256,
1600 index: u64,
1601 ) -> Result<Option<Transaction>, StoreError> {
1602 let block_body = match self.get_block_body_by_hash(block_hash).await? {
1603 Some(body) => body,
1604 None => return Ok(None),
1605 };
1606 let index: usize = index.try_into()?;
1607 Ok(block_body.transactions.get(index).cloned())
1608 }
1609
1610 pub async fn get_block_by_hash(
1611 &self,
1612 block_hash: BlockHash,
1613 ) -> Result<Option<Block>, StoreError> {
1614 let header = match self.get_block_header_by_hash(block_hash)? {
1615 Some(header) => header,
1616 None => return Ok(None),
1617 };
1618 let body = match self.get_block_body_by_hash(block_hash).await? {
1619 Some(body) => body,
1620 None => return Ok(None),
1621 };
1622 Ok(Some(Block::new(header, body)))
1623 }
1624
1625 pub async fn get_block_by_number(
1626 &self,
1627 block_number: BlockNumber,
1628 ) -> Result<Option<Block>, StoreError> {
1629 let Some(block_hash) = self.get_canonical_block_hash(block_number).await? else {
1630 return Ok(None);
1631 };
1632 self.get_block_by_hash(block_hash).await
1633 }
1634
1635 pub async fn get_canonical_block_hash(
1637 &self,
1638 block_number: BlockNumber,
1639 ) -> Result<Option<BlockHash>, StoreError> {
1640 let last = self.latest_block_header.get();
1641 if last.number == block_number {
1642 return Ok(Some(last.hash()));
1643 }
1644 let backend = self.backend.clone();
1645 tokio::task::spawn_blocking(move || {
1646 backend
1647 .begin_read()?
1648 .get(
1649 CANONICAL_BLOCK_HASHES,
1650 block_number.to_le_bytes().as_slice(),
1651 )?
1652 .map(|bytes| H256::decode(bytes.as_slice()))
1653 .transpose()
1654 .map_err(StoreError::from)
1655 })
1656 .await
1657 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
1658 }
1659
1660 pub async fn set_chain_config(&mut self, chain_config: &ChainConfig) -> Result<(), StoreError> {
1663 self.chain_config = *chain_config;
1664 let key = chain_data_key(ChainDataIndex::ChainConfig);
1665 let value = serde_json::to_string(chain_config)
1666 .map_err(|_| StoreError::Custom("Failed to serialize chain config".to_string()))?
1667 .into_bytes();
1668 self.write_async(CHAIN_DATA, key, value).await
1669 }
1670
1671 pub async fn update_earliest_block_number(
1673 &self,
1674 block_number: BlockNumber,
1675 ) -> Result<(), StoreError> {
1676 let key = chain_data_key(ChainDataIndex::EarliestBlockNumber);
1677 let value = block_number.to_le_bytes().to_vec();
1678 self.write_async(CHAIN_DATA, key, value).await
1679 }
1680
1681 async fn read_chain_data_block_number(
1683 &self,
1684 index: ChainDataIndex,
1685 ) -> Result<Option<BlockNumber>, StoreError> {
1686 self.read_async(CHAIN_DATA, chain_data_key(index))
1687 .await?
1688 .map(decode_block_number)
1689 .transpose()
1690 }
1691
1692 pub async fn get_earliest_block_number(&self) -> Result<BlockNumber, StoreError> {
1694 self.read_chain_data_block_number(ChainDataIndex::EarliestBlockNumber)
1695 .await?
1696 .ok_or(StoreError::MissingEarliestBlockNumber)
1697 }
1698
1699 pub async fn get_finalized_block_number(&self) -> Result<Option<BlockNumber>, StoreError> {
1701 self.read_chain_data_block_number(ChainDataIndex::FinalizedBlockNumber)
1702 .await
1703 }
1704
1705 pub async fn get_safe_block_number(&self) -> Result<Option<BlockNumber>, StoreError> {
1707 self.read_chain_data_block_number(ChainDataIndex::SafeBlockNumber)
1708 .await
1709 }
1710
1711 pub fn get_latest_block_number(&self) -> Result<BlockNumber, StoreError> {
1713 Ok(self.latest_block_header.get().number)
1714 }
1715
1716 pub async fn update_pending_block_number(
1718 &self,
1719 block_number: BlockNumber,
1720 ) -> Result<(), StoreError> {
1721 let key = chain_data_key(ChainDataIndex::PendingBlockNumber);
1722 let value = block_number.to_le_bytes().to_vec();
1723 self.write_async(CHAIN_DATA, key, value).await
1724 }
1725
1726 pub async fn get_pending_block_number(&self) -> Result<Option<BlockNumber>, StoreError> {
1728 self.read_chain_data_block_number(ChainDataIndex::PendingBlockNumber)
1729 .await
1730 }
1731
1732 async fn forkchoice_update_inner(
1747 &self,
1748 new_canonical_blocks: Vec<(BlockNumber, BlockHash)>,
1749 head_number: BlockNumber,
1750 head_hash: BlockHash,
1751 safe: Option<BlockNumber>,
1752 finalized: Option<BlockNumber>,
1753 ) -> Result<(), StoreError> {
1754 let latest = self.load_latest_block_number().await?.unwrap_or(0);
1755 let db = self.backend.clone();
1756 let journal_pruning_paused = self.journal_pruning_paused.clone();
1757 tokio::task::spawn_blocking(move || {
1758 let mut txn = db.begin_write()?;
1759
1760 for (block_number, block_hash) in new_canonical_blocks {
1761 let head_key = block_number.to_le_bytes();
1762 let head_value = block_hash.encode_to_vec();
1763 txn.put(CANONICAL_BLOCK_HASHES, &head_key, &head_value)?;
1764 }
1765
1766 for number in (head_number + 1)..=(latest) {
1773 txn.delete(CANONICAL_BLOCK_HASHES, number.to_le_bytes().as_slice())?;
1774 }
1775
1776 let head_key = head_number.to_le_bytes();
1778 let head_value = head_hash.encode_to_vec();
1779 txn.put(CANONICAL_BLOCK_HASHES, &head_key, &head_value)?;
1780
1781 let latest_key = chain_data_key(ChainDataIndex::LatestBlockNumber);
1783 txn.put(CHAIN_DATA, &latest_key, &head_number.to_le_bytes())?;
1784
1785 if let Some(safe) = safe {
1786 let safe_key = chain_data_key(ChainDataIndex::SafeBlockNumber);
1787 txn.put(CHAIN_DATA, &safe_key, &safe.to_le_bytes())?;
1788 }
1789
1790 if let Some(finalized) = finalized {
1791 let finalized_key = chain_data_key(ChainDataIndex::FinalizedBlockNumber);
1792
1793 let read = db.begin_read()?;
1803 let prev_finalized = match read.get(CHAIN_DATA, &finalized_key)? {
1804 Some(bytes) => {
1805 let arr: [u8; 8] = bytes.as_slice().try_into().map_err(|_| {
1806 StoreError::Custom(format!(
1807 "FinalizedBlockNumber has unexpected length {} (want 8)",
1808 bytes.len()
1809 ))
1810 })?;
1811 BlockNumber::from_le_bytes(arr)
1812 }
1813 None => 0,
1814 };
1815
1816 txn.put(CHAIN_DATA, &finalized_key, &finalized.to_le_bytes())?;
1817
1818 if finalized > prev_finalized
1848 && !journal_pruning_paused.load(std::sync::atomic::Ordering::Acquire)
1849 && let Some(lowest) = read.first_key(STATE_HISTORY)?
1850 {
1851 let lowest = state_history_key_number(&lowest)?;
1852 if lowest <= finalized {
1853 let highest = match read.last_key(STATE_HISTORY)? {
1855 Some(key) => state_history_key_number(&key)?,
1856 None => finalized,
1857 };
1858 let end = finalized.min(highest).saturating_add(1);
1859 txn.delete_range(STATE_HISTORY, &lowest.to_be_bytes(), &end.to_be_bytes())?;
1860 }
1861 }
1862 }
1863
1864 txn.commit()
1865 })
1866 .await
1867 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
1868 }
1869
1870 pub async fn get_receipts_for_block(
1871 &self,
1872 block_hash: &BlockHash,
1873 ) -> Result<Vec<Receipt>, StoreError> {
1874 self.get_receipts_for_block_from_index(block_hash, 0, None)
1875 .await
1876 }
1877
1878 pub async fn get_receipts_for_block_from_index(
1887 &self,
1888 block_hash: &BlockHash,
1889 start_index: u64,
1890 max_count: Option<usize>,
1891 ) -> Result<Vec<Receipt>, StoreError> {
1892 if let Some(all) = self.buffer()?.get_receipts(block_hash) {
1893 let start = start_index as usize;
1894 let slice = all.into_iter().skip(start);
1895 return Ok(match max_count {
1896 Some(max) => slice.take(max).collect(),
1897 None => slice.collect(),
1898 });
1899 }
1900 let backend = self.backend.clone();
1901 let block_hash = *block_hash;
1902
1903 tokio::task::spawn_blocking(move || {
1904 let txn = backend.begin_read()?;
1905 let prefix = block_hash.as_bytes().to_vec();
1906 let mut seek_key = prefix.clone();
1909 seek_key.extend_from_slice(&start_index.to_be_bytes());
1910 let iter = txn.prefix_iterator(RECEIPTS_V2, &seek_key)?;
1911 let mut receipts = Vec::new();
1912 for result in iter {
1913 let (k, v) = result?;
1914 if !k.starts_with(&prefix) {
1915 break;
1916 }
1917 if k.len() != 40 {
1918 continue;
1919 }
1920 receipts.push(Receipt::decode_storage(v.as_ref())?);
1921 if let Some(max) = max_count
1922 && receipts.len() >= max
1923 {
1924 break;
1925 }
1926 }
1927 Ok(receipts)
1928 })
1929 .await
1930 .map_err(|e| StoreError::Custom(format!("Task panicked: {e}")))?
1931 }
1932
1933 pub async fn set_header_download_checkpoint(
1937 &self,
1938 block_hash: BlockHash,
1939 ) -> Result<(), StoreError> {
1940 let key = snap_state_key(SnapStateIndex::HeaderDownloadCheckpoint);
1941 let value = block_hash.encode_to_vec();
1942 self.write_async(SNAP_STATE, key, value).await
1943 }
1944
1945 pub async fn get_header_download_checkpoint(&self) -> Result<Option<BlockHash>, StoreError> {
1947 let key = snap_state_key(SnapStateIndex::HeaderDownloadCheckpoint);
1948 self.backend
1949 .begin_read()?
1950 .get(SNAP_STATE, &key)?
1951 .map(|bytes| H256::decode(bytes.as_slice()))
1952 .transpose()
1953 .map_err(StoreError::from)
1954 }
1955
1956 pub async fn set_latest_valid_ancestor(
1962 &self,
1963 bad_block: BlockHash,
1964 latest_valid: BlockHash,
1965 ) -> Result<(), StoreError> {
1966 let value = latest_valid.encode_to_vec();
1967 self.write_async(INVALID_CHAINS, bad_block.as_bytes().to_vec(), value)
1968 .await
1969 }
1970
1971 pub async fn get_latest_valid_ancestor(
1974 &self,
1975 block: BlockHash,
1976 ) -> Result<Option<BlockHash>, StoreError> {
1977 self.read_async(INVALID_CHAINS, block.as_bytes().to_vec())
1978 .await?
1979 .map(|bytes| H256::decode(bytes.as_slice()))
1980 .transpose()
1981 .map_err(StoreError::from)
1982 }
1983
1984 pub async fn add_bad_block(&self, block: Block) -> Result<(), StoreError> {
1989 let mut bad_blocks = self.get_bad_blocks().await?;
1990 let block_number = block.header.number;
1991 let block_hash = block.hash();
1992 if bad_blocks
1993 .iter()
1994 .any(|b| b.header.number == block_number && b.hash() == block_hash)
1995 {
1996 return Ok(());
1997 }
1998 bad_blocks.push(block);
1999 bad_blocks.sort_by(|a, b| b.header.number.cmp(&a.header.number));
2000 bad_blocks.truncate(MAX_BAD_BLOCKS);
2001 self.write_async(
2002 BAD_BLOCKS,
2003 BAD_BLOCKS_KEY.to_vec(),
2004 bad_blocks.encode_to_vec(),
2005 )
2006 .await
2007 }
2008
2009 pub async fn get_bad_blocks(&self) -> Result<Vec<Block>, StoreError> {
2012 match self.read_async(BAD_BLOCKS, BAD_BLOCKS_KEY.to_vec()).await? {
2013 Some(bytes) => Vec::<Block>::decode(&bytes).map_err(StoreError::from),
2014 None => Ok(Vec::new()),
2015 }
2016 }
2017
2018 pub fn get_block_number_sync(
2020 &self,
2021 block_hash: BlockHash,
2022 ) -> Result<Option<BlockNumber>, StoreError> {
2023 if let Some(n) = self.buffer()?.get_number(&block_hash) {
2024 return Ok(Some(n));
2025 }
2026 let txn = self.backend.begin_read()?;
2027 txn.get(BLOCK_NUMBERS, &block_hash.encode_to_vec())?
2028 .map(decode_block_number)
2029 .transpose()
2030 }
2031
2032 pub fn get_canonical_block_hash_sync(
2034 &self,
2035 block_number: BlockNumber,
2036 ) -> Result<Option<BlockHash>, StoreError> {
2037 let last = self.latest_block_header.get();
2038 if last.number == block_number {
2039 return Ok(Some(last.hash()));
2040 }
2041 let txn = self.backend.begin_read()?;
2042 txn.get(
2043 CANONICAL_BLOCK_HASHES,
2044 block_number.to_le_bytes().as_slice(),
2045 )?
2046 .map(|bytes| H256::decode(bytes.as_slice()))
2047 .transpose()
2048 .map_err(StoreError::from)
2049 }
2050
2051 pub async fn write_storage_trie_nodes_batch(
2054 &self,
2055 storage_trie_nodes: StorageUpdates,
2056 ) -> Result<(), StoreError> {
2057 let mut txn = self.backend.begin_write()?;
2058 tokio::task::spawn_blocking(move || {
2059 for (address_hash, nodes) in storage_trie_nodes {
2060 for (node_path, node_data) in nodes {
2061 let key = apply_prefix(Some(address_hash), node_path);
2062 if node_data.is_empty() {
2063 txn.delete(STORAGE_TRIE_NODES, key.as_ref())?;
2064 } else {
2065 txn.put(STORAGE_TRIE_NODES, key.as_ref(), &node_data)?;
2066 }
2067 }
2068 }
2069 txn.commit()
2070 })
2071 .await
2072 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
2073 }
2074
2075 pub async fn write_account_code_batch(
2078 &self,
2079 account_codes: Vec<(H256, Code)>,
2080 ) -> Result<(), StoreError> {
2081 let mut code_batch_items = Vec::new();
2082 let mut metadata_batch_items = Vec::new();
2083
2084 for (code_hash, code) in account_codes {
2085 let buf = encode_code(&code);
2086 let metadata_buf = (code.len() as u64).to_be_bytes().to_vec();
2087 code_batch_items.push((code_hash.as_bytes().to_vec(), buf));
2088 metadata_batch_items.push((code_hash.as_bytes().to_vec(), metadata_buf));
2089 }
2090
2091 self.write_batch_async(ACCOUNT_CODES, code_batch_items)
2093 .await?;
2094 self.write_batch_async(ACCOUNT_CODE_METADATA, metadata_batch_items)
2095 .await
2096 }
2097
2098 fn buffer(&self) -> Result<Arc<BlockDataBuffer>, StoreError> {
2100 Ok(self
2101 .block_data_buffer
2102 .read()
2103 .map_err(|_| StoreError::LockError)?
2104 .clone())
2105 }
2106
2107 pub fn write(
2113 &self,
2114 table: &'static str,
2115 key: Vec<u8>,
2116 value: Vec<u8>,
2117 ) -> Result<(), StoreError> {
2118 let backend = self.backend.clone();
2119 let mut txn = backend.begin_write()?;
2120 txn.put(table, &key, &value)?;
2121 txn.commit()
2122 }
2123
2124 async fn write_async(
2127 &self,
2128 table: &'static str,
2129 key: Vec<u8>,
2130 value: Vec<u8>,
2131 ) -> Result<(), StoreError> {
2132 let backend = self.backend.clone();
2133
2134 tokio::task::spawn_blocking(move || {
2135 let mut txn = backend.begin_write()?;
2136 txn.put(table, &key, &value)?;
2137 txn.commit()
2138 })
2139 .await
2140 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
2141 }
2142
2143 pub async fn read_async(
2146 &self,
2147 table: &'static str,
2148 key: Vec<u8>,
2149 ) -> Result<Option<Vec<u8>>, StoreError> {
2150 let backend = self.backend.clone();
2151
2152 tokio::task::spawn_blocking(move || {
2153 let txn = backend.begin_read()?;
2154 txn.get(table, &key)
2155 })
2156 .await
2157 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
2158 }
2159
2160 pub fn read(&self, table: &'static str, key: Vec<u8>) -> Result<Option<Vec<u8>>, StoreError> {
2163 let backend = self.backend.clone();
2164 let txn = backend.begin_read()?;
2165 txn.get(table, &key)
2166 }
2167
2168 pub async fn write_batch_async(
2172 &self,
2173 table: &'static str,
2174 batch_ops: Vec<(Vec<u8>, Vec<u8>)>,
2175 ) -> Result<(), StoreError> {
2176 let backend = self.backend.clone();
2177
2178 tokio::task::spawn_blocking(move || {
2179 let mut txn = backend.begin_write()?;
2180 txn.put_batch(table, batch_ops)?;
2181 txn.commit()
2182 })
2183 .await
2184 .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
2185 }
2186
2187 pub fn write_batch(
2189 &self,
2190 table: &'static str,
2191 batch_ops: Vec<(Vec<u8>, Vec<u8>)>,
2192 ) -> Result<(), StoreError> {
2193 let backend = self.backend.clone();
2194 let mut txn = backend.begin_write()?;
2195 txn.put_batch(table, batch_ops)?;
2196 txn.commit()
2197 }
2198
2199 pub async fn add_fullsync_batch(&self, headers: Vec<BlockHeader>) -> Result<(), StoreError> {
2200 self.write_batch_async(
2201 FULLSYNC_HEADERS,
2202 headers
2203 .into_iter()
2204 .map(|header| (header.number.to_le_bytes().to_vec(), header.encode_to_vec()))
2205 .collect(),
2206 )
2207 .await
2208 }
2209
2210 pub async fn read_fullsync_batch(
2211 &self,
2212 start: BlockNumber,
2213 limit: u64,
2214 ) -> Result<Vec<Option<BlockHeader>>, StoreError> {
2215 let mut res = vec![];
2216 let read_tx = self.backend.begin_read()?;
2217 for key in start..start + limit {
2219 let header_opt = read_tx
2220 .get(FULLSYNC_HEADERS, &key.to_le_bytes())?
2221 .map(|header| BlockHeader::decode(&header))
2222 .transpose()?;
2223 res.push(header_opt);
2224 }
2225 Ok(res)
2226 }
2227
2228 pub async fn clear_fullsync_headers(&self) -> Result<(), StoreError> {
2229 self.backend.clear_table(FULLSYNC_HEADERS)
2230 }
2231
2232 pub fn delete(&self, table: &'static str, key: Vec<u8>) -> Result<(), StoreError> {
2234 let mut txn = self.backend.begin_write()?;
2235 txn.delete(table, &key)?;
2236 txn.commit()
2237 }
2238
2239 pub fn store_block_updates(&self, update_batch: UpdateBatch) -> Result<(), StoreError> {
2240 self.apply_updates(update_batch)
2241 }
2242
2243 fn batch_state_roots(
2250 &self,
2251 update_batch: &UpdateBatch,
2252 ) -> Result<(H256, H256, BlockNumber, H256), StoreError> {
2253 let parent_state_root = self
2254 .get_block_header_by_hash(
2255 update_batch
2256 .blocks
2257 .first()
2258 .ok_or(StoreError::UpdateBatchNoBlocks)?
2259 .header
2260 .parent_hash,
2261 )?
2262 .map(|header| header.state_root)
2263 .unwrap_or_default();
2264 let last_block = update_batch
2265 .blocks
2266 .last()
2267 .ok_or(StoreError::UpdateBatchNoBlocks)?;
2268 let last_state_root = last_block.header.state_root;
2269 let last_block_number = last_block.header.number;
2270 let last_block_hash = last_block.hash();
2271 Ok((
2272 parent_state_root,
2273 last_state_root,
2274 last_block_number,
2275 last_block_hash,
2276 ))
2277 }
2278
2279 fn apply_updates(&self, update_batch: UpdateBatch) -> Result<(), StoreError> {
2283 let (parent_state_root, last_state_root, last_block_number, last_block_hash) =
2284 self.batch_state_roots(&update_batch)?;
2285
2286 let UpdateBatch {
2287 account_updates,
2288 storage_updates,
2289 blocks,
2290 receipts,
2291 code_updates,
2292 commit_depth,
2293 wait_for_flush,
2294 } = update_batch;
2295
2296 self.pending_trie_roots.register(last_state_root)?;
2300
2301 let blocks_with_receipts: Vec<(Block, Vec<Receipt>)> = if blocks.len() == 1 {
2304 let block = blocks.into_iter().next().expect("len == 1");
2305 let hash = block.hash();
2306 let r = receipts
2307 .into_iter()
2308 .find(|(h, _)| *h == hash)
2309 .map(|(_, r)| r)
2310 .unwrap_or_default();
2311 vec![(block, r)]
2312 } else {
2313 let mut receipts_by_hash: std::collections::HashMap<BlockHash, Vec<Receipt>> =
2314 receipts.into_iter().collect();
2315 blocks
2316 .into_iter()
2317 .map(|b| {
2318 let r = receipts_by_hash.remove(&b.hash()).unwrap_or_default();
2319 (b, r)
2320 })
2321 .collect()
2322 };
2323
2324 let (ack_tx, ack_rx) = sync_channel(1);
2329 self.persist_tx
2330 .send(PersistMessage::Block(Box::new(BlockPersist {
2331 blocks: blocks_with_receipts,
2332 codes: code_updates,
2333 parent_state_root,
2334 child_state_root: last_state_root,
2335 account_updates,
2336 storage_updates,
2337 commit_depth,
2338 wait_for_flush,
2339 block_number: last_block_number,
2340 block_hash: last_block_hash,
2341 ack: ack_tx,
2342 })))
2343 .map_err(|e| StoreError::Custom(format!("failed to send block persist: {e}")))?;
2344 ack_rx
2345 .recv()
2346 .map_err(|e| StoreError::Custom(format!("block persist ack failed: {e}")))??;
2347
2348 Ok(())
2349 }
2350
2351 pub fn new(path: impl AsRef<Path>, engine_type: EngineType) -> Result<Self, StoreError> {
2356 Self::new_with_config(path, engine_type, StoreConfig::default())
2357 }
2358
2359 pub fn new_with_config(
2361 path: impl AsRef<Path>,
2362 engine_type: EngineType,
2363 config: StoreConfig,
2364 ) -> Result<Self, StoreError> {
2365 let code_cache_size = config.code_cache_size.unwrap_or(CODE_CACHE_MAX_SIZE);
2366 let store = Self::open_with_config(path, engine_type, config)?;
2367 if let Ok(mut cache) = store.account_code_cache.lock() {
2368 cache.max_size = code_cache_size;
2369 }
2370 info!(cache_bytes = code_cache_size, "sized bytecode cache");
2371 Ok(store)
2372 }
2373
2374 fn open_with_config(
2375 path: impl AsRef<Path>,
2376 engine_type: EngineType,
2377 #[cfg_attr(not(feature = "rocksdb"), allow(unused_variables))] config: StoreConfig,
2379 ) -> Result<Self, StoreError> {
2380 let db_path = path.as_ref().to_path_buf();
2381
2382 if engine_type != EngineType::InMemory {
2383 let version = read_store_schema_version(&db_path)?;
2384
2385 match version {
2386 None if db_path.exists() && dir_contains_legacy_db(&db_path)? => {
2387 return Err(StoreError::NotFoundDBVersion);
2389 }
2390 None => {
2391 init_metadata_file(&db_path)?;
2397 }
2398 Some(v) if v < 1 => {
2402 return Err(StoreError::IncompatibleDBVersion {
2404 found: v,
2405 expected: STORE_SCHEMA_VERSION,
2406 });
2407 }
2408 Some(v) if v > STORE_SCHEMA_VERSION => {
2409 return Err(StoreError::IncompatibleDBVersion {
2412 found: v,
2413 expected: STORE_SCHEMA_VERSION,
2414 });
2415 }
2416 #[cfg(feature = "rocksdb")]
2417 Some(v) if v < STORE_SCHEMA_VERSION => {
2418 let rocksdb = Arc::new(RocksDBBackend::open(
2422 &path,
2423 config.rocksdb_block_cache_size,
2424 )?);
2425 crate::migrations::run_pending_migrations(rocksdb.as_ref(), &db_path, v)?;
2426 rocksdb.drop_obsolete_cfs(&path);
2427 let backend: Arc<dyn crate::api::StorageBackend> = rocksdb;
2428 return Self::from_backend(
2429 backend,
2430 db_path,
2431 DB_COMMIT_THRESHOLD,
2432 config.persist_channel_capacity,
2433 );
2434 }
2435 Some(_) => {
2436 }
2441 }
2442 }
2443
2444 match engine_type {
2445 #[cfg(feature = "rocksdb")]
2446 EngineType::RocksDB => {
2447 let rocksdb = RocksDBBackend::open(&path, config.rocksdb_block_cache_size)?;
2448 rocksdb.drop_obsolete_cfs(&path);
2449 let backend: Arc<dyn StorageBackend> = Arc::new(rocksdb);
2450 Self::from_backend(
2451 backend,
2452 db_path,
2453 DB_COMMIT_THRESHOLD,
2454 config.persist_channel_capacity,
2455 )
2456 }
2457 EngineType::InMemory => {
2458 let backend = Arc::new(InMemoryBackend::open()?);
2459 Self::from_backend(
2460 backend,
2461 db_path,
2462 IN_MEMORY_COMMIT_THRESHOLD,
2463 config.persist_channel_capacity,
2464 )
2465 }
2466 }
2467 }
2468
2469 fn from_backend(
2470 backend: Arc<dyn StorageBackend>,
2471 db_path: PathBuf,
2472 commit_threshold: usize,
2473 persist_channel_capacity: usize,
2474 ) -> Result<Self, StoreError> {
2475 debug!("Initializing Store with {commit_threshold} in-memory diff-layers");
2476 let (fkv_tx, fkv_rx) = std::sync::mpsc::sync_channel(0);
2477 let persist_cap = persist_channel_capacity.max(1); let (persist_tx, persist_rx) = std::sync::mpsc::sync_channel(persist_cap);
2479
2480 let (last_written, initial_flushed_upto) = {
2481 let tx = backend.begin_read()?;
2482 let last_written = tx
2483 .get(MISC_VALUES, "last_written".as_bytes())?
2484 .unwrap_or_else(|| vec![0u8; 64]);
2485 let last_written = if last_written == [0xff] {
2486 vec![0xff; 64]
2487 } else {
2488 last_written
2489 };
2490 let initial_flushed_upto = match tx.get(MISC_VALUES, FLUSHED_UPTO_KEY)? {
2491 Some(bytes) => decode_flushed_upto(&bytes)?,
2492 None => 0,
2493 };
2494 (last_written, initial_flushed_upto)
2495 };
2496 let mut initial_buffer = BlockDataBuffer::new();
2497 initial_buffer.set_flushed_upto(initial_flushed_upto);
2498
2499 let mut background_threads = Vec::new();
2500 let safe_commit_root = Arc::new(RwLock::new(H256::zero()));
2501 let mut store = Self {
2502 db_path,
2503 backend,
2504 chain_config: Default::default(),
2505 latest_block_header: Default::default(),
2506 trie_cache: Arc::new(RwLock::new(Arc::new(TrieLayerCache::new_with_safe_commit(
2507 commit_threshold,
2508 safe_commit_root.clone(),
2509 )))),
2510 flatkeyvalue_control_tx: fkv_tx,
2511 block_data_buffer: Arc::new(RwLock::new(Arc::new(initial_buffer))),
2512 persist_tx,
2513 pending_trie_roots: Arc::new(PendingTrieRoots::default()),
2514 last_computed_flatkeyvalue: Arc::new(RwLock::new(last_written)),
2515 account_code_cache: Arc::new(Mutex::new(CodeCache::default())),
2516 code_metadata_cache: Arc::new(Mutex::new(LruCache::with_hasher(
2517 NonZeroUsize::new(CODE_METADATA_CACHE_MAX_ENTRIES).unwrap_or(NonZeroUsize::MIN),
2518 SeededState::default(),
2519 ))),
2520 fcu_lock: Arc::new(tokio::sync::Mutex::new(())),
2521 safe_commit_root,
2522 journal_pruning_paused: Arc::new(std::sync::atomic::AtomicBool::new(false)),
2523 background_threads: Default::default(),
2524 };
2525 let backend_clone = store.backend.clone();
2526 let last_computed_fkv = store.last_computed_flatkeyvalue.clone();
2527 background_threads.push(std::thread::spawn(move || {
2528 let rx = fkv_rx;
2529 loop {
2531 match rx.recv() {
2532 Ok(FKVGeneratorControlMessage::Continue) => break,
2533 Ok(FKVGeneratorControlMessage::Stop) => {}
2534 Err(std::sync::mpsc::RecvError) => {
2535 debug!("Closing FlatKeyValue generator.");
2536 return;
2537 }
2538 }
2539 }
2540
2541 let _ = flatkeyvalue_generator(&backend_clone, &last_computed_fkv, &rx)
2542 .inspect_err(|err| error!("Error while generating FlatKeyValue: {err}"));
2543 }));
2544 let persist_backend = store.backend.clone();
2547 let persist_buffer = store.block_data_buffer.clone();
2548 let persist_trie_cache = store.trie_cache.clone();
2549 let persist_pending_roots = store.pending_trie_roots.clone();
2550 let persist_fkv_ctl = store.flatkeyvalue_control_tx.clone();
2551 background_threads.push(std::thread::spawn(move || {
2552 let rx = persist_rx;
2553 let mut last_flush_result: Result<(), StoreError> = Ok(());
2556 loop {
2557 match rx.recv() {
2558 Ok(PersistMessage::Block(bp)) => {
2559 let bp = *bp;
2560 let staged = mutate_block_buffer(&persist_buffer, move |b| {
2563 let mut codes = Some(bp.codes);
2564 for (block, receipts) in bp.blocks {
2565 b.insert(block, receipts, codes.take().unwrap_or_default());
2566 }
2567 });
2568 if let Err(e) = staged {
2569 persist_pending_roots.clear(bp.child_state_root);
2574 let _ = bp.ack.send(Err(e));
2575 continue;
2576 }
2577 if !bp.wait_for_flush {
2586 let _ = bp
2587 .ack
2588 .send(std::mem::replace(&mut last_flush_result, Ok(())));
2589 }
2590 if let Err(err) = apply_trie_phase1(
2592 &persist_trie_cache,
2593 &persist_pending_roots,
2594 bp.parent_state_root,
2595 bp.child_state_root,
2596 bp.block_number,
2597 bp.block_hash,
2598 bp.account_updates,
2599 bp.storage_updates,
2600 ) {
2601 error!("persist worker trie phase-1 failed: {err}");
2602 if bp.wait_for_flush {
2603 let _ = bp.ack.send(Err(err));
2604 } else {
2605 last_flush_result = Err(err);
2606 }
2607 continue;
2608 }
2609 let flushed = flush_block_data(persist_backend.as_ref(), &persist_buffer)
2611 .inspect_err(|err| error!("flush_block_data failed: {err}"))
2612 .and_then(|_| {
2613 commit_trie_if_due(
2614 persist_backend.as_ref(),
2615 &persist_trie_cache,
2616 &persist_fkv_ctl,
2617 bp.parent_state_root,
2618 bp.commit_depth,
2619 bp.wait_for_flush,
2620 )
2621 });
2622 if bp.wait_for_flush {
2625 let prior = std::mem::replace(&mut last_flush_result, Ok(()));
2626 let _ = bp.ack.send(prior.and(flushed));
2627 } else {
2628 last_flush_result = flushed;
2629 }
2630 }
2631 Ok(PersistMessage::Commit(root)) => match persist_trie_cache.read() {
2632 Ok(guard) => {
2633 let trie = guard.clone();
2634 drop(guard);
2635 let _ = commit_to_disk(
2638 persist_backend.as_ref(),
2639 &persist_fkv_ctl,
2640 &persist_trie_cache,
2641 &trie,
2642 root,
2643 false,
2644 )
2645 .inspect_err(|err| error!("commit_to_disk failed: {err}"));
2646 }
2647 Err(_) => error!("trie cache lock poisoned during commit"),
2648 },
2649 Ok(PersistMessage::Ping(ack)) => {
2650 let _ = ack.send(std::mem::replace(&mut last_flush_result, Ok(())));
2654 }
2655 Ok(PersistMessage::Shutdown { ack }) => {
2656 let result = flush_block_data(persist_backend.as_ref(), &persist_buffer);
2663 let prior = std::mem::replace(&mut last_flush_result, Ok(()));
2664 let _ = ack.send(prior.and(result));
2665 return;
2667 }
2668 Err(_) => return,
2669 }
2670 }
2671 }));
2672 store.background_threads = Arc::new(ThreadList {
2673 list: background_threads,
2674 });
2675 Ok(store)
2676 }
2677
2678 pub async fn new_from_genesis(
2681 store_path: &Path,
2682 engine_type: EngineType,
2683 genesis_path: &str,
2684 ) -> Result<Self, StoreError> {
2685 Self::new_from_genesis_with_config(
2686 store_path,
2687 engine_type,
2688 genesis_path,
2689 StoreConfig::default(),
2690 )
2691 .await
2692 }
2693
2694 pub async fn new_from_genesis_with_config(
2697 store_path: &Path,
2698 engine_type: EngineType,
2699 genesis_path: &str,
2700 config: StoreConfig,
2701 ) -> Result<Self, StoreError> {
2702 let file = std::fs::File::open(genesis_path)
2703 .map_err(|error| StoreError::Custom(format!("Failed to open genesis file: {error}")))?;
2704 let reader = std::io::BufReader::new(file);
2705 let genesis: Genesis = serde_json::from_reader(reader)
2706 .map_err(|e| StoreError::Custom(format!("Failed to deserialize genesis file: {e}")))?;
2707 let mut store = Self::new_with_config(store_path, engine_type, config)?;
2708 store.add_initial_state(genesis).await?;
2709 Ok(store)
2710 }
2711
2712 pub async fn get_account_info(
2713 &self,
2714 block_number: BlockNumber,
2715 address: Address,
2716 ) -> Result<Option<AccountInfo>, StoreError> {
2717 match self.get_canonical_block_hash(block_number).await? {
2718 Some(block_hash) => self.get_account_info_by_hash(block_hash, address),
2719 None => Ok(None),
2720 }
2721 }
2722
2723 pub fn get_account_info_by_hash(
2724 &self,
2725 block_hash: BlockHash,
2726 address: Address,
2727 ) -> Result<Option<AccountInfo>, StoreError> {
2728 let Some(state_trie) = self.state_trie(block_hash)? else {
2729 return Ok(None);
2730 };
2731 let hashed_address = hash_address_fixed(&address);
2732
2733 let Some(encoded_state) = state_trie.get(hashed_address.as_bytes())? else {
2734 return Ok(None);
2735 };
2736
2737 let account_state = AccountState::decode(&encoded_state)?;
2738 Ok(Some(AccountInfo {
2739 code_hash: account_state.code_hash,
2740 balance: account_state.balance,
2741 nonce: account_state.nonce,
2742 }))
2743 }
2744
2745 pub fn get_account_state_by_acc_hash(
2746 &self,
2747 block_hash: BlockHash,
2748 account_hash: H256,
2749 ) -> Result<Option<AccountState>, StoreError> {
2750 let Some(state_trie) = self.state_trie(block_hash)? else {
2751 return Ok(None);
2752 };
2753 let Some(encoded_state) = state_trie.get(account_hash.as_bytes())? else {
2754 return Ok(None);
2755 };
2756 let account_state = AccountState::decode(&encoded_state)?;
2757 Ok(Some(account_state))
2758 }
2759
2760 pub async fn get_fork_id(&self) -> Result<ForkId, StoreError> {
2761 let chain_config = self.get_chain_config();
2762 let genesis_header = self
2763 .load_block_header(0)?
2764 .ok_or(StoreError::MissingEarliestBlockNumber)?;
2765 let block_header = self.latest_block_header.get();
2766
2767 Ok(ForkId::new(
2768 chain_config,
2769 genesis_header,
2770 block_header.timestamp,
2771 block_header.number,
2772 ))
2773 }
2774
2775 pub async fn get_code_by_account_address(
2776 &self,
2777 block_number: BlockNumber,
2778 address: Address,
2779 ) -> Result<Option<Code>, StoreError> {
2780 let Some(block_hash) = self.get_canonical_block_hash(block_number).await? else {
2781 return Ok(None);
2782 };
2783 let Some(state_trie) = self.state_trie(block_hash)? else {
2784 return Ok(None);
2785 };
2786 let hashed_address = hash_address_fixed(&address);
2787 let Some(encoded_state) = state_trie.get(hashed_address.as_bytes())? else {
2788 return Ok(None);
2789 };
2790 let account_state = AccountState::decode(&encoded_state)?;
2791 self.get_account_code(account_state.code_hash)
2792 }
2793
2794 pub async fn get_nonce_by_account_address(
2795 &self,
2796 block_number: BlockNumber,
2797 address: Address,
2798 ) -> Result<Option<u64>, StoreError> {
2799 let Some(block_hash) = self.get_canonical_block_hash(block_number).await? else {
2800 return Ok(None);
2801 };
2802 let Some(state_trie) = self.state_trie(block_hash)? else {
2803 return Ok(None);
2804 };
2805 let hashed_address = hash_address_fixed(&address);
2806 let Some(encoded_state) = state_trie.get(hashed_address.as_bytes())? else {
2807 return Ok(None);
2808 };
2809 let account_state = AccountState::decode(&encoded_state)?;
2810 Ok(Some(account_state.nonce))
2811 }
2812
2813 pub fn apply_account_updates_batch(
2816 &self,
2817 block_hash: BlockHash,
2818 account_updates: &[AccountUpdate],
2819 ) -> Result<Option<AccountUpdatesList>, StoreError> {
2820 let Some(mut state_trie) = self.state_trie(block_hash)? else {
2821 return Ok(None);
2822 };
2823
2824 Ok(Some(self.apply_account_updates_from_trie_batch(
2825 &mut state_trie,
2826 account_updates,
2827 )?))
2828 }
2829
2830 pub fn apply_account_updates_from_trie_batch<'a>(
2831 &self,
2832 state_trie: &mut Trie,
2833 account_updates: impl IntoIterator<Item = &'a AccountUpdate>,
2834 ) -> Result<AccountUpdatesList, StoreError> {
2835 let mut ret_storage_updates = Vec::new();
2836 let mut code_updates = Vec::new();
2837 let state_root = state_trie.hash_no_commit(&NativeCrypto);
2838 for update in account_updates {
2839 let hashed_address = hash_address_fixed(&update.address);
2840 if update.removed {
2841 state_trie.remove(hashed_address.as_bytes())?;
2843 continue;
2844 }
2845 let mut account_state = match state_trie.get(hashed_address.as_bytes())? {
2848 Some(encoded_state) => AccountState::decode(&encoded_state)?,
2849 None => AccountState::default(),
2850 };
2851 if update.removed_storage {
2852 account_state.storage_root = EMPTY_TRIE_HASH;
2853 }
2854 if let Some(info) = &update.info {
2855 account_state.nonce = info.nonce;
2856 account_state.balance = info.balance;
2857 account_state.code_hash = info.code_hash;
2858 if let Some(code) = &update.code {
2860 code_updates.push((info.code_hash, code.clone()));
2861 }
2862 }
2863 if !update.added_storage.is_empty() {
2865 let mut storage_trie =
2866 self.open_storage_trie(hashed_address, state_root, account_state.storage_root)?;
2867 for (storage_key, storage_value) in &update.added_storage {
2868 let hashed_key = hash_key(storage_key);
2869 if storage_value.is_zero() {
2870 storage_trie.remove(&hashed_key)?;
2871 } else {
2872 storage_trie.insert(hashed_key, storage_value.encode_to_vec())?;
2873 }
2874 }
2875 let (storage_hash, storage_updates) =
2876 storage_trie.collect_changes_since_last_hash(&NativeCrypto);
2877 account_state.storage_root = storage_hash;
2878 ret_storage_updates.push((hashed_address, storage_updates));
2879 }
2880 state_trie.insert(
2881 hashed_address.as_bytes().to_vec(),
2882 account_state.encode_to_vec(),
2883 )?;
2884 }
2885 let (state_trie_hash, state_updates) =
2886 state_trie.collect_changes_since_last_hash(&NativeCrypto);
2887
2888 Ok(AccountUpdatesList {
2889 state_trie_hash,
2890 state_updates,
2891 storage_updates: ret_storage_updates,
2892 code_updates,
2893 })
2894 }
2895
2896 pub fn apply_account_updates_from_trie_with_witness(
2899 &self,
2900 mut state_trie: Trie,
2901 account_updates: &[AccountUpdate],
2902 mut storage_tries: StorageTries,
2903 ) -> Result<(StorageTries, AccountUpdatesList), StoreError> {
2904 let mut ret_storage_updates = Vec::new();
2905
2906 let mut code_updates = Vec::new();
2907
2908 let state_root = state_trie.hash_no_commit(&NativeCrypto);
2909
2910 let mut ordered: Vec<&AccountUpdate> = account_updates.iter().collect();
2915 ordered.sort_unstable_by_key(|update| update.address);
2916
2917 for update in ordered {
2918 let hashed_address = hash_address(&update.address);
2919
2920 if update.removed {
2921 state_trie.remove(&hashed_address)?;
2923
2924 continue;
2925 }
2926
2927 let mut account_state = match state_trie.get(&hashed_address)? {
2930 Some(encoded_state) => AccountState::decode(&encoded_state)?,
2931 None => AccountState::default(),
2932 };
2933
2934 if update.removed_storage {
2935 account_state.storage_root = EMPTY_TRIE_HASH;
2936 }
2937
2938 if let Some(info) = &update.info {
2939 account_state.nonce = info.nonce;
2940
2941 account_state.balance = info.balance;
2942
2943 account_state.code_hash = info.code_hash;
2944
2945 if let Some(code) = &update.code {
2947 code_updates.push((info.code_hash, code.clone()));
2948 }
2949 }
2950
2951 if !update.added_storage.is_empty() {
2953 let (_witness, storage_trie) = match storage_tries.entry(update.address) {
2954 Entry::Occupied(value) => value.into_mut(),
2955 Entry::Vacant(vacant) => {
2956 let trie = self.open_storage_trie(
2957 H256::from_slice(&hashed_address),
2958 state_root,
2959 account_state.storage_root,
2960 )?;
2961 vacant.insert(TrieLogger::open_trie(trie))
2962 }
2963 };
2964
2965 for (storage_key, storage_value) in &update.added_storage {
2966 let hashed_key = hash_key(storage_key);
2967
2968 if storage_value.is_zero() {
2969 storage_trie.remove(&hashed_key)?;
2970 } else {
2971 storage_trie.insert(hashed_key, storage_value.encode_to_vec())?;
2972 }
2973 }
2974
2975 let (storage_hash, storage_updates) =
2976 storage_trie.collect_changes_since_last_hash(&NativeCrypto);
2977
2978 account_state.storage_root = storage_hash;
2979
2980 ret_storage_updates.push((H256::from_slice(&hashed_address), storage_updates));
2981 }
2982
2983 state_trie.insert(hashed_address, account_state.encode_to_vec())?;
2984 }
2985
2986 let (state_trie_hash, state_updates) =
2987 state_trie.collect_changes_since_last_hash(&NativeCrypto);
2988
2989 let account_updates_list = AccountUpdatesList {
2990 state_trie_hash,
2991 state_updates,
2992 storage_updates: ret_storage_updates,
2993 code_updates,
2994 };
2995
2996 Ok((storage_tries, account_updates_list))
2997 }
2998
2999 pub async fn setup_genesis_state_trie(
3001 &self,
3002 genesis_accounts: BTreeMap<Address, GenesisAccount>,
3003 ) -> Result<H256, StoreError> {
3004 let mut storage_trie_nodes = vec![];
3005 let mut genesis_state_trie = self.open_direct_state_trie(EMPTY_TRIE_HASH)?;
3006 for (address, account) in genesis_accounts {
3007 let hashed_address = hash_address(&address);
3008 let h256_hashed_address = H256::from_slice(&hashed_address);
3009
3010 let code = Code::from_bytecode(account.code, &NativeCrypto);
3012 let code_hash = code.hash;
3013 self.add_account_code(code).await?;
3014
3015 let mut storage_trie =
3017 self.open_direct_storage_trie(h256_hashed_address, EMPTY_TRIE_HASH)?;
3018 for (storage_key, storage_value) in account.storage {
3019 if !storage_value.is_zero() {
3020 let hashed_key = hash_key(&H256(storage_key.to_big_endian()));
3021 storage_trie.insert(hashed_key, storage_value.encode_to_vec())?;
3022 }
3023 }
3024
3025 let (storage_root, storage_nodes) =
3026 storage_trie.collect_changes_since_last_hash(&NativeCrypto);
3027
3028 storage_trie_nodes.extend(
3029 storage_nodes
3030 .into_iter()
3031 .map(|(path, n)| (apply_prefix(Some(h256_hashed_address), path).into_vec(), n)),
3032 );
3033
3034 let account_state = AccountState {
3036 nonce: account.nonce,
3037 balance: account.balance,
3038 storage_root,
3039 code_hash,
3040 };
3041 genesis_state_trie.insert(hashed_address, account_state.encode_to_vec())?;
3042 }
3043
3044 let (state_root, account_trie_nodes) =
3045 genesis_state_trie.collect_changes_since_last_hash(&NativeCrypto);
3046 let account_trie_nodes = account_trie_nodes
3047 .into_iter()
3048 .map(|(path, n)| (apply_prefix(None, path).into_vec(), n))
3049 .collect::<Vec<_>>();
3050
3051 let mut tx = self.backend.begin_write()?;
3052 tx.put_batch(ACCOUNT_TRIE_NODES, account_trie_nodes)?;
3053 tx.put_batch(STORAGE_TRIE_NODES, storage_trie_nodes)?;
3054 tx.commit()?;
3055
3056 Ok(state_root)
3057 }
3058
3059 fn make_witness_key(block_number: u64, block_hash: &BlockHash) -> Vec<u8> {
3061 let mut composite_key = Vec::with_capacity(8 + 32);
3062 composite_key.extend_from_slice(&block_number.to_be_bytes());
3063 composite_key.extend_from_slice(block_hash.as_bytes());
3064 composite_key
3065 }
3066
3067 pub fn store_witness(
3073 &self,
3074 block_hash: BlockHash,
3075 block_number: u64,
3076 witness: ExecutionWitness,
3077 ) -> Result<(), StoreError> {
3078 let rpc_witness = RpcExecutionWitness::try_from(witness)?;
3080 let key = Self::make_witness_key(block_number, &block_hash);
3081 let value = serde_json::to_vec(&rpc_witness)?;
3082 self.write(EXECUTION_WITNESSES, key, value)?;
3083 self.cleanup_old_witnesses(block_number)
3085 }
3086
3087 fn cleanup_old_witnesses(&self, latest_block_number: u64) -> Result<(), StoreError> {
3088 if latest_block_number <= MAX_WITNESSES {
3090 return Ok(());
3091 }
3092
3093 let threshold = latest_block_number - MAX_WITNESSES;
3094
3095 if let Some(oldest_block_number) = self.get_oldest_witness_number()? {
3096 let prefix = oldest_block_number.to_be_bytes();
3097 let mut to_delete = Vec::new();
3098
3099 {
3100 let read_txn = self.backend.begin_read()?;
3101 let iter = read_txn.prefix_iterator(EXECUTION_WITNESSES, &prefix)?;
3102
3103 for item in iter {
3105 let (key, _value) = item?;
3106 let mut block_number_bytes = [0u8; 8];
3107 block_number_bytes.copy_from_slice(&key[0..8]);
3108 let block_number = u64::from_be_bytes(block_number_bytes);
3109 if block_number > threshold {
3110 break;
3111 }
3112 to_delete.push(key.to_vec());
3113 }
3114 }
3115
3116 for key in to_delete {
3117 self.delete(EXECUTION_WITNESSES, key)?;
3118 }
3119 };
3120
3121 self.update_oldest_witness_number(threshold + 1)?;
3122
3123 Ok(())
3124 }
3125
3126 fn update_oldest_witness_number(&self, oldest_block_number: u64) -> Result<(), StoreError> {
3127 self.write(
3128 MISC_VALUES,
3129 b"oldest_witness_block_number".to_vec(),
3130 oldest_block_number.to_le_bytes().to_vec(),
3131 )?;
3132 Ok(())
3133 }
3134
3135 fn get_oldest_witness_number(&self) -> Result<Option<u64>, StoreError> {
3136 let Some(value) = self.read(MISC_VALUES, b"oldest_witness_block_number".to_vec())? else {
3137 return Ok(None);
3138 };
3139
3140 let array: [u8; 8] = value.as_slice().try_into().map_err(|_| {
3141 StoreError::Custom("Invalid oldest witness block number bytes".to_string())
3142 })?;
3143 Ok(Some(u64::from_le_bytes(array)))
3144 }
3145
3146 pub fn get_witness_json_bytes(
3152 &self,
3153 block_number: u64,
3154 block_hash: BlockHash,
3155 ) -> Result<Option<Vec<u8>>, StoreError> {
3156 let key = Self::make_witness_key(block_number, &block_hash);
3157 self.read(EXECUTION_WITNESSES, key)
3158 }
3159
3160 pub fn get_witness_by_number_and_hash(
3165 &self,
3166 block_number: u64,
3167 block_hash: BlockHash,
3168 ) -> Result<Option<RpcExecutionWitness>, StoreError> {
3169 let key = Self::make_witness_key(block_number, &block_hash);
3170 match self.read(EXECUTION_WITNESSES, key)? {
3171 Some(value) => {
3172 let witness: RpcExecutionWitness = serde_json::from_slice(&value)?;
3173 Ok(Some(witness))
3174 }
3175 None => Ok(None),
3176 }
3177 }
3178
3179 pub fn store_block_access_list(
3181 &self,
3182 block_hash: BlockHash,
3183 bal: &BlockAccessList,
3184 ) -> Result<(), StoreError> {
3185 let mut value = vec![];
3186 bal.encode(&mut value);
3187 self.store_block_access_list_encoded(block_hash, value)
3188 }
3189
3190 pub fn store_block_access_list_encoded(
3193 &self,
3194 block_hash: BlockHash,
3195 encoded: Vec<u8>,
3196 ) -> Result<(), StoreError> {
3197 self.write(BLOCK_ACCESS_LISTS, block_hash.as_bytes().to_vec(), encoded)
3198 }
3199
3200 pub fn get_block_access_list(
3202 &self,
3203 block_hash: BlockHash,
3204 ) -> Result<Option<BlockAccessList>, StoreError> {
3205 let key = block_hash.as_bytes().to_vec();
3206 match self.read(BLOCK_ACCESS_LISTS, key)? {
3207 Some(value) => {
3208 let bal = BlockAccessList::decode(&value)
3209 .map_err(|e| StoreError::Custom(format!("Failed to decode BAL: {e}")))?;
3210 Ok(Some(bal))
3211 }
3212 None => Ok(None),
3213 }
3214 }
3215
3216 pub fn iter_block_access_lists_by_hashes(
3221 &self,
3222 hashes: &[BlockHash],
3223 ) -> Result<Vec<Option<BlockAccessList>>, StoreError> {
3224 let mut out = Vec::with_capacity(hashes.len());
3225 for hash in hashes {
3226 out.push(self.get_block_access_list(*hash)?);
3227 }
3228 Ok(out)
3229 }
3230
3231 pub async fn add_initial_state(&mut self, genesis: Genesis) -> Result<(), StoreError> {
3232 self.add_initial_state_inner(genesis, false).await
3233 }
3234
3235 pub async fn add_initial_state_skip_validation(
3246 &mut self,
3247 genesis: Genesis,
3248 ) -> Result<(), StoreError> {
3249 self.add_initial_state_inner(genesis, true).await
3250 }
3251
3252 async fn add_initial_state_inner(
3253 &mut self,
3254 genesis: Genesis,
3255 skip_genesis_validation: bool,
3256 ) -> Result<(), StoreError> {
3257 debug!("Storing initial state from genesis");
3258
3259 let genesis_block = genesis.get_block();
3261 let genesis_block_number = genesis_block.header.number;
3262
3263 let genesis_hash = genesis_block.hash();
3264
3265 let stored_genesis_header = self.load_block_header(genesis_block_number)?;
3266
3267 self.set_chain_config(&genesis.config).await?;
3274
3275 if let Some(latest) = self.load_latest_block_number().await? {
3280 self.anchor_to_durable_head(latest).await?;
3281 }
3282
3283 match stored_genesis_header {
3284 Some(header) if skip_genesis_validation => {
3285 info!(
3286 stored_genesis = %header.hash(),
3287 "Skipping genesis state validation; trusting the genesis header and state already stored in the datadir"
3288 );
3289 return Ok(());
3290 }
3291 Some(header) if header.hash() == genesis_hash => {
3292 info!("Received genesis file matching a previously stored one, nothing to do");
3293 return Ok(());
3294 }
3295 Some(_) => {
3296 error!(
3297 "The chain configuration stored in the database is incompatible with the provided configuration. If you intended to switch networks, choose another datadir or clear the database (e.g., run `ethrex removedb`) and try again."
3298 );
3299 return Err(StoreError::IncompatibleChainConfig);
3300 }
3301 None => {
3302 self.add_block_header(genesis_hash, genesis_block.header.clone())
3303 .await?
3304 }
3305 }
3306 let genesis_state_root = self.setup_genesis_state_trie(genesis.alloc).await?;
3309 debug_assert_eq!(genesis_state_root, genesis_block.header.state_root);
3310
3311 info!(hash = %genesis_hash, "Storing genesis block");
3313
3314 self.add_block(genesis_block).await?;
3315 self.update_earliest_block_number(genesis_block_number)
3316 .await?;
3317 self.forkchoice_update(vec![], genesis_block_number, genesis_hash, None, None)
3318 .await?;
3319 Ok(())
3320 }
3321
3322 pub async fn load_initial_state(&self) -> Result<(), StoreError> {
3323 info!("Loading initial state from DB");
3324 let Some(latest) = self.load_latest_block_number().await? else {
3325 return Err(StoreError::MissingLatestBlockNumber);
3326 };
3327 self.anchor_to_durable_head(latest).await?;
3331 Ok(())
3332 }
3333
3334 pub fn get_storage_at(
3335 &self,
3336 block_number: BlockNumber,
3337 address: Address,
3338 storage_key: H256,
3339 ) -> Result<Option<U256>, StoreError> {
3340 match self.get_block_header(block_number)? {
3341 Some(header) => self.get_storage_at_root(header.state_root, address, storage_key),
3342 None => Ok(None),
3343 }
3344 }
3345
3346 pub fn get_storage_at_root(
3347 &self,
3348 state_root: H256,
3349 address: Address,
3350 storage_key: H256,
3351 ) -> Result<Option<U256>, StoreError> {
3352 let account_hash = hash_address_fixed(&address);
3353
3354 let read_view = self.backend.begin_read()?;
3356 let cache = self.gated_snapshot(state_root)?;
3357 let last_written = self.last_written()?;
3358 let use_fkv = Self::flatkeyvalue_computed_with_last_written(account_hash, &last_written)
3365 && !cache.overlay_serves(state_root);
3366
3367 let storage_root = if use_fkv {
3368 EMPTY_TRIE_HASH
3370 } else {
3371 let state_trie = self.open_state_trie_shared(
3372 state_root,
3373 read_view.clone(),
3374 cache.clone(),
3375 last_written.clone(),
3376 )?;
3377 let Some(encoded_account) = state_trie.get(account_hash.as_bytes())? else {
3378 return Ok(None);
3379 };
3380 let account = AccountState::decode(&encoded_account)?;
3381 account.storage_root
3382 };
3383 let storage_trie = self.open_storage_trie_shared(
3384 account_hash,
3385 state_root,
3386 storage_root,
3387 read_view,
3388 cache,
3389 last_written,
3390 )?;
3391
3392 let hashed_key = hash_key_fixed(&storage_key);
3393 storage_trie
3394 .get(&hashed_key)?
3395 .map(|rlp| U256::decode(&rlp).map_err(StoreError::RLPDecode))
3396 .transpose()
3397 }
3398
3399 pub fn get_storage_at_root_with_known_storage_root(
3404 &self,
3405 state_root: H256,
3406 account_hash: H256,
3407 storage_root: H256,
3408 storage_key: H256,
3409 ) -> Result<Option<U256>, StoreError> {
3410 let read_view = self.backend.begin_read()?;
3411 let cache = self.gated_snapshot(state_root)?;
3412 let last_written = self.last_written()?;
3413 let storage_root =
3420 if Self::flatkeyvalue_computed_with_last_written(account_hash, &last_written)
3421 && !cache.overlay_serves(state_root)
3422 {
3423 EMPTY_TRIE_HASH
3424 } else {
3425 storage_root
3426 };
3427 let storage_trie = self.open_storage_trie_shared(
3428 account_hash,
3429 state_root,
3430 storage_root,
3431 read_view,
3432 cache,
3433 last_written,
3434 )?;
3435
3436 let hashed_key = hash_key_fixed(&storage_key);
3437 storage_trie
3438 .get(&hashed_key)?
3439 .map(|rlp| U256::decode(&rlp).map_err(StoreError::RLPDecode))
3440 .transpose()
3441 }
3442
3443 pub fn get_chain_config(&self) -> ChainConfig {
3444 self.chain_config
3445 }
3446
3447 pub async fn get_latest_canonical_block_hash(&self) -> Result<Option<BlockHash>, StoreError> {
3448 Ok(Some(self.latest_block_header.get().hash()))
3449 }
3450
3451 pub async fn forkchoice_update(
3456 &self,
3457 new_canonical_blocks: Vec<(BlockNumber, BlockHash)>,
3458 head_number: BlockNumber,
3459 head_hash: BlockHash,
3460 safe: Option<BlockNumber>,
3461 finalized: Option<BlockNumber>,
3462 ) -> Result<(), StoreError> {
3463 let _guard = self.fcu_lock.lock().await;
3469
3470 let previous_head = self.latest_block_header.get();
3475 let new_head = self
3476 .get_block_header_by_hash(head_hash)?
3477 .ok_or_else(|| StoreError::MissingLatestBlockNumber)?;
3478 self.latest_block_header.update(new_head);
3479 if let Err(err) = self
3480 .forkchoice_update_inner(
3481 new_canonical_blocks,
3482 head_number,
3483 head_hash,
3484 safe,
3485 finalized,
3486 )
3487 .await
3488 {
3489 self.latest_block_header.update((*previous_head).clone());
3490 return Err(err);
3491 }
3492
3493 if let Some(root) = self.compute_safe_commit_root(head_number)? {
3499 if self.set_safe_commit_root(root)? {
3504 let tx = self.persist_tx.clone();
3505 let _ = tokio::task::spawn_blocking(move || tx.send(PersistMessage::Commit(root)))
3506 .await;
3507 }
3508 }
3509
3510 Ok(())
3511 }
3512
3513 pub(crate) fn set_safe_commit_root(&self, root: H256) -> Result<bool, StoreError> {
3521 let mut guard = self
3522 .safe_commit_root
3523 .write()
3524 .map_err(|_| StoreError::LockError)?;
3525 let changed = *guard != root;
3526 *guard = root;
3527 Ok(changed)
3528 }
3529
3530 pub(crate) fn compute_safe_commit_root(
3539 &self,
3540 head: BlockNumber,
3541 ) -> Result<Option<H256>, StoreError> {
3542 let commit_threshold = self
3543 .trie_cache
3544 .read()
3545 .map_err(|_| StoreError::LockError)?
3546 .commit_threshold;
3547 let Some(target) = head.checked_sub(commit_threshold as u64) else {
3548 return Ok(None);
3549 };
3550 let Some(hash) = self.get_canonical_block_hash_sync(target)? else {
3551 return Ok(None);
3552 };
3553 let Some(header) = self.get_block_header_by_hash(hash)? else {
3554 return Ok(None);
3555 };
3556 Ok(Some(header.state_root))
3557 }
3558
3559 pub fn state_trie(&self, block_hash: BlockHash) -> Result<Option<Trie>, StoreError> {
3561 let Some(header) = self.get_block_header_by_hash(block_hash)? else {
3562 return Ok(None);
3563 };
3564 Ok(Some(self.open_state_trie(header.state_root)?))
3565 }
3566
3567 pub fn storage_trie(
3569 &self,
3570 block_hash: BlockHash,
3571 address: Address,
3572 ) -> Result<Option<Trie>, StoreError> {
3573 let Some(header) = self.get_block_header_by_hash(block_hash)? else {
3574 return Ok(None);
3575 };
3576 let Some(state_trie) = self.state_trie(block_hash)? else {
3578 return Ok(None);
3579 };
3580 let hashed_address = hash_address_fixed(&address);
3581 let Some(encoded_account) = state_trie.get(hashed_address.as_bytes())? else {
3582 return Ok(None);
3583 };
3584 let account = AccountState::decode(&encoded_account)?;
3585 let storage_root = account.storage_root;
3587 Ok(Some(self.open_storage_trie(
3588 hashed_address,
3589 header.state_root,
3590 storage_root,
3591 )?))
3592 }
3593
3594 pub async fn get_account_state(
3595 &self,
3596 block_number: BlockNumber,
3597 address: Address,
3598 ) -> Result<Option<AccountState>, StoreError> {
3599 let Some(block_hash) = self.get_canonical_block_hash(block_number).await? else {
3600 return Ok(None);
3601 };
3602 let Some(state_trie) = self.state_trie(block_hash)? else {
3603 return Ok(None);
3604 };
3605 self.get_account_state_from_trie(&state_trie, address)
3606 }
3607
3608 pub fn get_account_state_by_root(
3609 &self,
3610 state_root: H256,
3611 address: Address,
3612 ) -> Result<Option<AccountState>, StoreError> {
3613 let state_trie = self.open_state_trie(state_root)?;
3614 self.get_account_state_from_trie(&state_trie, address)
3615 }
3616
3617 pub fn get_account_state_from_trie(
3618 &self,
3619 state_trie: &Trie,
3620 address: Address,
3621 ) -> Result<Option<AccountState>, StoreError> {
3622 let hashed_address = hash_address_fixed(&address);
3623 let Some(encoded_state) = state_trie.get(hashed_address.as_bytes())? else {
3624 return Ok(None);
3625 };
3626 Ok(Some(AccountState::decode(&encoded_state)?))
3627 }
3628
3629 pub fn get_account_states_batch_by_root(
3638 &self,
3639 state_root: H256,
3640 addresses: &[Address],
3641 ) -> Result<Vec<Option<AccountState>>, StoreError> {
3642 if addresses.is_empty() {
3643 return Ok(Vec::new());
3644 }
3645
3646 let last_written = self.last_written()?;
3647 let trie_cache = self
3648 .trie_cache
3649 .read()
3650 .map_err(|_| StoreError::LockError)?
3651 .clone();
3652
3653 let mut results: Vec<Option<AccountState>> = vec![None; addresses.len()];
3654 let leaf_paths: Vec<Vec<u8>> = addresses
3656 .iter()
3657 .map(|addr| {
3658 let hashed = hash_address_fixed(addr);
3659 Nibbles::from_bytes(hashed.as_bytes()).into_vec()
3660 })
3661 .collect();
3662
3663 let mut fkv_indices: Vec<usize> = Vec::new();
3664 let mut trie_indices: Vec<usize> = Vec::new();
3665
3666 let overlay_active = trie_cache.overlay_serves(state_root);
3682 let fkv_cursor: &[u8] = last_written.as_slice();
3683 for (i, path) in leaf_paths.iter().enumerate() {
3684 if let Some(value) = trie_cache.get(state_root, path.as_slice()) {
3685 if !value.is_empty() {
3686 results[i] = Some(AccountState::decode(&value)?);
3687 }
3688 continue;
3689 }
3690 if !overlay_active && fkv_cursor >= path.as_slice() {
3691 fkv_indices.push(i);
3692 } else {
3693 trie_indices.push(i);
3694 }
3695 }
3696
3697 if !fkv_indices.is_empty() {
3698 fkv_indices.sort_unstable_by(|&a, &b| leaf_paths[a].cmp(&leaf_paths[b]));
3705 let read_view = self.backend.begin_read()?;
3706 let keys: Vec<&[u8]> = fkv_indices
3707 .iter()
3708 .map(|&i| leaf_paths[i].as_slice())
3709 .collect();
3710 let shards = shard_count(keys.len());
3711 let raw: Vec<Result<Option<Vec<u8>>, StoreError>> = if shards <= 1 {
3712 read_view.multi_get(ACCOUNT_FLATKEYVALUE, &keys)
3713 } else {
3714 let chunk = keys.len().div_ceil(shards);
3715 let rv = read_view.as_ref();
3716 std::thread::scope(|scope| {
3717 let handles: Vec<_> = keys
3718 .chunks(chunk)
3719 .map(|ck| scope.spawn(move || rv.multi_get(ACCOUNT_FLATKEYVALUE, ck)))
3720 .collect();
3721 handles
3722 .into_iter()
3723 .flat_map(|h| {
3729 h.join().unwrap_or_else(|_| {
3730 vec![Err(StoreError::Custom(
3731 "account prefetch shard panicked".into(),
3732 ))]
3733 })
3734 })
3735 .collect()
3736 })
3737 };
3738 for (slot, res) in fkv_indices.iter().zip(raw.into_iter()) {
3739 let Some(encoded) = res? else { continue };
3740 if encoded.is_empty() {
3741 continue;
3742 }
3743 results[*slot] = Some(AccountState::decode(&encoded)?);
3744 }
3745 }
3746
3747 if !trie_indices.is_empty() {
3748 let state_trie = self.open_state_trie(state_root)?;
3754 let fetched: Result<Vec<(usize, Option<AccountState>)>, StoreError> = trie_indices
3755 .par_iter()
3756 .map(|&i| {
3757 self.get_account_state_from_trie(&state_trie, addresses[i])
3758 .map(|s| (i, s))
3759 })
3760 .collect();
3761 for (i, s) in fetched? {
3762 results[i] = s;
3763 }
3764 }
3765
3766 Ok(results)
3767 }
3768
3769 pub fn get_storage_values_batch_by_root(
3789 &self,
3790 state_root: H256,
3791 slots: &[(H256, H256, H256)],
3792 ) -> Result<Vec<Option<U256>>, StoreError> {
3793 if slots.is_empty() {
3794 return Ok(Vec::new());
3795 }
3796
3797 let last_written = self.last_written()?;
3798 let trie_cache = self
3799 .trie_cache
3800 .read()
3801 .map_err(|_| StoreError::LockError)?
3802 .clone();
3803
3804 let mut results: Vec<Option<U256>> = vec![None; slots.len()];
3805 let leaf_paths: Vec<Vec<u8>> = slots
3811 .iter()
3812 .map(|(account_hash, _storage_root, slot)| {
3813 let hashed_slot = hash_key_fixed(slot);
3814 apply_prefix(Some(*account_hash), Nibbles::from_bytes(&hashed_slot)).into_vec()
3815 })
3816 .collect();
3817
3818 let mut fkv_indices: Vec<usize> = Vec::new();
3819 let mut trie_indices: Vec<usize> = Vec::new();
3820
3821 let overlay_active = trie_cache.overlay_serves(state_root);
3835 let fkv_cursor: &[u8] = last_written.as_slice();
3836 for (i, path) in leaf_paths.iter().enumerate() {
3837 if let Some(value) = trie_cache.get(state_root, path.as_slice()) {
3842 if !value.is_empty() {
3843 results[i] = Some(U256::decode(&value).map_err(StoreError::RLPDecode)?);
3844 }
3845 continue;
3846 }
3847 if !overlay_active && fkv_cursor >= path.as_slice() {
3848 fkv_indices.push(i);
3849 } else {
3850 trie_indices.push(i);
3851 }
3852 }
3853
3854 if !fkv_indices.is_empty() {
3855 fkv_indices.sort_unstable_by(|&a, &b| leaf_paths[a].cmp(&leaf_paths[b]));
3860 let read_view = self.backend.begin_read()?;
3861 let keys: Vec<&[u8]> = fkv_indices
3862 .iter()
3863 .map(|&i| leaf_paths[i].as_slice())
3864 .collect();
3865 let shards = shard_count(keys.len());
3878 let raw: Vec<Result<Option<Vec<u8>>, StoreError>> = if shards <= 1 {
3879 read_view.multi_get(STORAGE_FLATKEYVALUE, &keys)
3880 } else {
3881 let chunk = keys.len().div_ceil(shards);
3882 let rv = read_view.as_ref();
3883 std::thread::scope(|scope| {
3884 let handles: Vec<_> = keys
3885 .chunks(chunk)
3886 .map(|ck| scope.spawn(move || rv.multi_get(STORAGE_FLATKEYVALUE, ck)))
3887 .collect();
3888 handles
3889 .into_iter()
3890 .flat_map(|h| {
3896 h.join().unwrap_or_else(|_| {
3897 vec![Err(StoreError::Custom(
3898 "storage prefetch shard panicked".into(),
3899 ))]
3900 })
3901 })
3902 .collect()
3903 })
3904 };
3905 for (slot_idx, res) in fkv_indices.iter().zip(raw.into_iter()) {
3906 let Some(encoded) = res? else { continue };
3907 if encoded.is_empty() {
3908 continue;
3909 }
3910 results[*slot_idx] = Some(U256::decode(&encoded).map_err(StoreError::RLPDecode)?);
3911 }
3912 }
3913
3914 if !trie_indices.is_empty() {
3915 let fetched: Result<Vec<(usize, Option<U256>)>, StoreError> = trie_indices
3923 .par_iter()
3924 .map(|&i| {
3925 let (account_hash, storage_root, slot) = slots[i];
3926 self.get_storage_at_root_with_known_storage_root(
3927 state_root,
3928 account_hash,
3929 storage_root,
3930 slot,
3931 )
3932 .map(|v| (i, v))
3933 })
3934 .collect();
3935 for (i, v) in fetched? {
3936 results[i] = v;
3937 }
3938 }
3939
3940 Ok(results)
3941 }
3942
3943 pub fn prefetch_trie_nodes(
3970 &self,
3971 storage_slots: &[(Address, H256)],
3972 accounts: &[Address],
3973 ) -> Result<usize, StoreError> {
3974 const MAX_PREFETCH_DEPTH: usize = 8;
3979
3980 let mut storage_keys: Vec<Vec<u8>> =
3983 Vec::with_capacity(storage_slots.len() * (MAX_PREFETCH_DEPTH + 1));
3984 for (address, slot) in storage_slots {
3985 let hashed_address = hash_address_fixed(address);
3986 let hashed_slot = hash_key_fixed(slot);
3987 let slot_nibbles = Nibbles::from_bytes(&hashed_slot);
3988 let max_d = MAX_PREFETCH_DEPTH.min(slot_nibbles.len());
3989 for d in 0..=max_d {
3990 let prefix = apply_prefix(Some(hashed_address), slot_nibbles.slice(0, d));
3991 storage_keys.push(prefix.into_vec());
3992 }
3993 }
3994
3995 let mut account_keys: Vec<Vec<u8>> =
3997 Vec::with_capacity(accounts.len() * (MAX_PREFETCH_DEPTH + 1));
3998 for address in accounts {
3999 let hashed_address = hash_address_fixed(address);
4000 let addr_nibbles = Nibbles::from_bytes(hashed_address.as_bytes());
4001 let max_d = MAX_PREFETCH_DEPTH.min(addr_nibbles.len());
4002 for d in 0..=max_d {
4003 account_keys.push(addr_nibbles.slice(0, d).into_vec());
4004 }
4005 }
4006
4007 let hits = self.warm_trie_node_keys(STORAGE_TRIE_NODES, storage_keys)?
4008 + self.warm_trie_node_keys(ACCOUNT_TRIE_NODES, account_keys)?;
4009 Ok(hits)
4010 }
4011
4012 fn warm_trie_node_keys(
4019 &self,
4020 table: &'static str,
4021 mut keys: Vec<Vec<u8>>,
4022 ) -> Result<usize, StoreError> {
4023 if keys.is_empty() {
4024 return Ok(0);
4025 }
4026 keys.sort_unstable();
4027 keys.dedup();
4028
4029 let shards = shard_count(keys.len());
4030 let read_view = self.backend.begin_read()?;
4031 let count_hits = |res: Vec<Result<Option<Vec<u8>>, StoreError>>| {
4032 res.into_iter().filter(|r| matches!(r, Ok(Some(_)))).count()
4033 };
4034 if shards <= 1 {
4035 let refs: Vec<&[u8]> = keys.iter().map(|k| k.as_slice()).collect();
4036 return Ok(count_hits(read_view.multi_get(table, &refs)));
4037 }
4038 let chunk = keys.len().div_ceil(shards);
4039 let rv = read_view.as_ref();
4040 let hits = std::thread::scope(|scope| {
4041 let handles: Vec<_> = keys
4042 .chunks(chunk)
4043 .map(|ck| {
4044 scope.spawn(move || {
4045 let refs: Vec<&[u8]> = ck.iter().map(|k| k.as_slice()).collect();
4046 count_hits(rv.multi_get(table, &refs))
4048 })
4049 })
4050 .collect();
4051 handles
4052 .into_iter()
4053 .map(|h| h.join().unwrap_or(0))
4058 .sum::<usize>()
4059 });
4060 Ok(hits)
4061 }
4062
4063 pub async fn get_account_proof(
4068 &self,
4069 state_root: H256,
4070 address: Address,
4071 storage_keys: &[H256],
4072 ) -> Result<Option<AccountProof>, StoreError> {
4073 let state_trie = self.open_state_trie(state_root)?;
4078 let address_path = hash_address_fixed(&address);
4079 let proof = state_trie.get_proof(address_path.as_bytes())?;
4080 let account_opt = state_trie
4081 .get(address_path.as_bytes())?
4082 .map(|encoded_state| AccountState::decode(&encoded_state))
4083 .transpose()?;
4084
4085 let mut storage_proof = Vec::with_capacity(storage_keys.len());
4086
4087 if let Some(account) = &account_opt {
4088 let storage_trie =
4089 self.open_storage_trie(address_path, state_root, account.storage_root)?;
4090
4091 for key in storage_keys {
4092 let hashed_key = hash_key(key);
4093 let proof = storage_trie.get_proof(&hashed_key)?;
4094 let value = storage_trie
4095 .get(&hashed_key)?
4096 .map(|rlp| U256::decode(&rlp).map_err(StoreError::RLPDecode))
4097 .transpose()?
4098 .unwrap_or_default();
4099
4100 let slot_proof = StorageSlotProof {
4101 proof,
4102 key: *key,
4103 value,
4104 };
4105 storage_proof.push(slot_proof);
4106 }
4107 } else {
4108 storage_proof.extend(storage_keys.iter().map(|key| StorageSlotProof {
4109 proof: Vec::new(),
4110 key: *key,
4111 value: U256::zero(),
4112 }));
4113 }
4114 let account = account_opt.unwrap_or_default();
4115 let account_proof = AccountProof {
4116 proof,
4117 account,
4118 storage_proof,
4119 };
4120 Ok(Some(account_proof))
4121 }
4122
4123 pub fn iter_accounts_from(
4126 &self,
4127 state_root: H256,
4128 starting_address: H256,
4129 ) -> Result<impl Iterator<Item = (H256, AccountState)>, StoreError> {
4130 let mut iter = self.open_locked_state_trie(state_root)?.into_iter();
4131 iter.advance(starting_address.0.to_vec())?;
4132 Ok(iter.content().map_while(|(path, value)| {
4133 Some((H256::from_slice(&path), AccountState::decode(&value).ok()?))
4134 }))
4135 }
4136
4137 pub fn iter_accounts(
4140 &self,
4141 state_root: H256,
4142 ) -> Result<impl Iterator<Item = (H256, AccountState)>, StoreError> {
4143 self.iter_accounts_from(state_root, H256::zero())
4144 }
4145
4146 pub fn iter_storage_from(
4149 &self,
4150 state_root: H256,
4151 hashed_address: H256,
4152 starting_slot: H256,
4153 ) -> Result<Option<impl Iterator<Item = (H256, U256)>>, StoreError> {
4154 let state_trie = self.open_locked_state_trie(state_root)?;
4155 let Some(account_rlp) = state_trie.get(hashed_address.as_bytes())? else {
4156 return Ok(None);
4157 };
4158 let storage_root = AccountState::decode(&account_rlp)?.storage_root;
4159 let mut iter = self
4160 .open_locked_storage_trie(hashed_address, state_root, storage_root)?
4161 .into_iter();
4162 iter.advance(starting_slot.0.to_vec())?;
4163 Ok(Some(iter.content().map_while(|(path, value)| {
4164 Some((H256::from_slice(&path), U256::decode(&value).ok()?))
4165 })))
4166 }
4167
4168 pub fn iter_storage(
4171 &self,
4172 state_root: H256,
4173 hashed_address: H256,
4174 ) -> Result<Option<impl Iterator<Item = (H256, U256)>>, StoreError> {
4175 self.iter_storage_from(state_root, hashed_address, H256::zero())
4176 }
4177
4178 pub fn get_account_range_proof(
4179 &self,
4180 state_root: H256,
4181 starting_hash: H256,
4182 last_hash: Option<H256>,
4183 ) -> Result<Vec<Vec<u8>>, StoreError> {
4184 let state_trie = self.open_state_trie(state_root)?;
4185 let mut proof = state_trie.get_proof(starting_hash.as_bytes())?;
4186 if let Some(last_hash) = last_hash {
4187 proof.extend_from_slice(&state_trie.get_proof(last_hash.as_bytes())?);
4188 }
4189 Ok(proof)
4190 }
4191
4192 pub fn get_storage_range_proof(
4193 &self,
4194 state_root: H256,
4195 hashed_address: H256,
4196 starting_hash: H256,
4197 last_hash: Option<H256>,
4198 ) -> Result<Option<Vec<Vec<u8>>>, StoreError> {
4199 let state_trie = self.open_state_trie(state_root)?;
4200 let Some(account_rlp) = state_trie.get(hashed_address.as_bytes())? else {
4201 return Ok(None);
4202 };
4203 let storage_root = AccountState::decode(&account_rlp)?.storage_root;
4204 let storage_trie = self.open_storage_trie(hashed_address, state_root, storage_root)?;
4205 let mut proof = storage_trie.get_proof(starting_hash.as_bytes())?;
4206 if let Some(last_hash) = last_hash {
4207 proof.extend_from_slice(&storage_trie.get_proof(last_hash.as_bytes())?);
4208 }
4209 Ok(Some(proof))
4210 }
4211
4212 pub fn get_trie_nodes(
4219 &self,
4220 state_root: H256,
4221 paths: Vec<Vec<u8>>,
4222 byte_limit: u64,
4223 ) -> Result<Vec<Vec<u8>>, StoreError> {
4224 let Some(account_path) = paths.first() else {
4225 return Ok(vec![]);
4226 };
4227 let state_trie = self.open_state_trie(state_root)?;
4228 if paths.len() == 1 {
4230 let node = state_trie.get_node(account_path)?;
4232 return Ok(vec![node]);
4233 }
4234 let Some(account_state) = state_trie
4236 .get(account_path)?
4237 .map(|ref rlp| AccountState::decode(rlp))
4238 .transpose()?
4239 else {
4240 return Ok(vec![]);
4241 };
4242 let Ok(hashed_address) = account_path.clone().try_into().map(H256) else {
4244 return Ok(vec![]);
4245 };
4246 let storage_trie =
4247 self.open_storage_trie(hashed_address, state_root, account_state.storage_root)?;
4248 let mut nodes = vec![];
4250 let mut bytes_used = 0;
4251 for path in paths.iter().skip(1) {
4255 if bytes_used >= byte_limit {
4256 break;
4257 }
4258 let node = storage_trie.get_node(path)?;
4259 bytes_used += node.len() as u64;
4260 nodes.push(node);
4261 }
4262 Ok(nodes)
4263 }
4264
4265 fn gated_snapshot(&self, state_root: H256) -> Result<Arc<TrieLayerCache>, StoreError> {
4274 self.pending_trie_roots.wait_until_ready(state_root)?;
4275 Ok(self
4276 .trie_cache
4277 .read()
4278 .map_err(|_| StoreError::LockError)?
4279 .clone())
4280 }
4281
4282 pub fn open_state_trie(&self, state_root: H256) -> Result<Trie, StoreError> {
4286 let trie_db = TrieWrapper::new(
4287 state_root,
4288 self.gated_snapshot(state_root)?,
4289 Box::new(BackendTrieDB::new_for_accounts(
4290 self.backend.clone(),
4291 self.last_written()?,
4292 )?),
4293 None,
4294 );
4295 Ok(Trie::open(Box::new(trie_db), state_root))
4296 }
4297
4298 pub fn open_direct_state_trie(&self, state_root: H256) -> Result<Trie, StoreError> {
4302 Ok(Trie::open(
4303 Box::new(BackendTrieDB::new_for_accounts(
4304 self.backend.clone(),
4305 self.last_written()?,
4306 )?),
4307 state_root,
4308 ))
4309 }
4310
4311 pub fn open_locked_state_trie(&self, state_root: H256) -> Result<Trie, StoreError> {
4315 let trie_db = TrieWrapper::new(
4316 state_root,
4317 self.gated_snapshot(state_root)?,
4318 Box::new(state_trie_locked_backend(
4319 self.backend.as_ref(),
4320 self.last_written()?,
4321 )?),
4322 None,
4323 );
4324 Ok(Trie::open(Box::new(trie_db), state_root))
4325 }
4326
4327 pub fn open_storage_trie(
4330 &self,
4331 account_hash: H256,
4332 state_root: H256,
4333 storage_root: H256,
4334 ) -> Result<Trie, StoreError> {
4335 let trie_db = TrieWrapper::new(
4336 state_root,
4337 self.gated_snapshot(state_root)?,
4338 Box::new(BackendTrieDB::new_for_storages(
4339 self.backend.clone(),
4340 self.last_written()?,
4341 )?),
4342 Some(account_hash),
4343 );
4344 Ok(Trie::open(Box::new(trie_db), storage_root))
4345 }
4346
4347 fn open_state_trie_shared(
4351 &self,
4352 state_root: H256,
4353 read_view: Arc<dyn StorageReadView>,
4354 cache: Arc<TrieLayerCache>,
4355 last_written: Vec<u8>,
4356 ) -> Result<Trie, StoreError> {
4357 let trie_db = TrieWrapper::new(
4358 state_root,
4359 cache,
4360 Box::new(BackendTrieDB::new_for_accounts_with_view(
4361 self.backend.clone(),
4362 read_view,
4363 last_written,
4364 )?),
4365 None,
4366 );
4367 Ok(Trie::open(Box::new(trie_db), state_root))
4368 }
4369
4370 fn open_storage_trie_shared(
4372 &self,
4373 account_hash: H256,
4374 state_root: H256,
4375 storage_root: H256,
4376 read_view: Arc<dyn StorageReadView>,
4377 cache: Arc<TrieLayerCache>,
4378 last_written: Vec<u8>,
4379 ) -> Result<Trie, StoreError> {
4380 let trie_db = TrieWrapper::new(
4381 state_root,
4382 cache,
4383 Box::new(BackendTrieDB::new_for_storages_with_view(
4384 self.backend.clone(),
4385 read_view,
4386 last_written,
4387 )?),
4388 Some(account_hash),
4389 );
4390 Ok(Trie::open(Box::new(trie_db), storage_root))
4391 }
4392
4393 pub fn open_direct_storage_trie(
4396 &self,
4397 account_hash: H256,
4398 storage_root: H256,
4399 ) -> Result<Trie, StoreError> {
4400 Ok(Trie::open(
4401 Box::new(BackendTrieDB::new_for_account_storage(
4402 self.backend.clone(),
4403 account_hash,
4404 self.last_written()?,
4405 )?),
4406 storage_root,
4407 ))
4408 }
4409
4410 pub fn open_locked_storage_trie(
4413 &self,
4414 account_hash: H256,
4415 state_root: H256,
4416 storage_root: H256,
4417 ) -> Result<Trie, StoreError> {
4418 let trie_db = TrieWrapper::new(
4419 state_root,
4420 self.gated_snapshot(state_root)?,
4421 Box::new(state_trie_locked_backend(
4422 self.backend.as_ref(),
4423 self.last_written()?,
4424 )?),
4425 Some(account_hash),
4426 );
4427 Ok(Trie::open(Box::new(trie_db), storage_root))
4428 }
4429
4430 pub fn has_state_root(&self, state_root: H256) -> Result<bool, StoreError> {
4431 if state_root == EMPTY_TRIE_HASH {
4433 return Ok(true);
4434 }
4435 let trie = self.open_state_trie(state_root)?;
4436 let Some(root) = trie.db().get(Nibbles::default())? else {
4438 return Ok(false);
4439 };
4440 let root_hash = ethrex_trie::Node::decode(&root)?
4441 .compute_hash(&NativeCrypto)
4442 .finalize(&NativeCrypto);
4443 Ok(state_root == root_hash)
4444 }
4445
4446 pub fn is_state_in_layer_cache(&self, state_root: H256) -> Result<bool, StoreError> {
4455 let trie = self
4456 .trie_cache
4457 .read()
4458 .map_err(|_| StoreError::LockError)?
4459 .clone();
4460 Ok(trie.contains(state_root))
4461 }
4462
4463 pub fn highest_state_history_block_number(&self) -> Result<Option<BlockNumber>, StoreError> {
4470 let read = self.backend.begin_read()?;
4471 let Some(key) = read.last_key(STATE_HISTORY)? else {
4472 return Ok(None);
4473 };
4474 let arr = <[u8; 8]>::try_from(key.as_slice()).map_err(|_| {
4475 StoreError::Custom(format!(
4476 "STATE_HISTORY key has unexpected length: {}",
4477 key.len()
4478 ))
4479 })?;
4480 Ok(Some(BlockNumber::from_be_bytes(arr)))
4481 }
4482
4483 pub fn get_state_history_block_hash(
4491 &self,
4492 block_number: BlockNumber,
4493 ) -> Result<Option<BlockHash>, StoreError> {
4494 let read = self.backend.begin_read()?;
4495 let Some(bytes) = read.get(STATE_HISTORY, &block_number.to_be_bytes())? else {
4496 return Ok(None);
4497 };
4498 JournalEntry::decode_block_hash(&bytes)
4499 .map(Some)
4500 .map_err(|e| StoreError::Custom(format!("STATE_HISTORY entry {block_number}: {e}")))
4501 }
4502
4503 pub fn lowest_state_history_block_number(&self) -> Result<Option<BlockNumber>, StoreError> {
4508 let read = self.backend.begin_read()?;
4509 let Some(key) = read.first_key(STATE_HISTORY)? else {
4510 return Ok(None);
4511 };
4512 let arr = <[u8; 8]>::try_from(key.as_slice()).map_err(|_| {
4513 StoreError::Custom(format!(
4514 "STATE_HISTORY key has unexpected length: {}",
4515 key.len()
4516 ))
4517 })?;
4518 Ok(Some(BlockNumber::from_be_bytes(arr)))
4519 }
4520
4521 pub fn reorg_retention(&self) -> Result<u64, StoreError> {
4526 let cache = self.trie_cache.read().map_err(|_| StoreError::LockError)?;
4527 Ok(cache.commit_threshold() as u64)
4528 }
4529
4530 #[doc(hidden)]
4534 pub fn put_state_history_entry_for_test(
4535 &self,
4536 block_number: BlockNumber,
4537 encoded: &[u8],
4538 ) -> Result<(), StoreError> {
4539 let mut tx = self.backend.begin_write()?;
4540 tx.put(STATE_HISTORY, &block_number.to_be_bytes(), encoded)?;
4541 tx.commit()?;
4542 Ok(())
4543 }
4544
4545 pub fn install_overlay_for_reorg(
4566 &self,
4567 from_block: BlockNumber,
4568 to_block: BlockNumber,
4569 expected_hash: impl Fn(BlockNumber) -> Option<H256>,
4570 ) -> Result<(), StoreError> {
4571 let overlay =
4573 Overlay::from_journal(self.backend.as_ref(), from_block, to_block, expected_hash)
4574 .map_err(|e| StoreError::Custom(format!("overlay construction failed: {e}")))?;
4575
4576 let threshold = {
4577 let current = self.trie_cache.read().map_err(|_| StoreError::LockError)?;
4578 current.commit_threshold()
4579 };
4580 let mut fresh =
4583 TrieLayerCache::new_with_safe_commit(threshold, self.safe_commit_root.clone());
4584 fresh.set_overlay(Arc::new(overlay));
4585
4586 self.rendezvous_persist_worker("install_overlay_for_reorg")?;
4589
4590 let mut guard = self.trie_cache.write().map_err(|_| StoreError::LockError)?;
4591 *guard = Arc::new(fresh);
4592 Ok(())
4593 }
4594
4595 fn rendezvous_persist_worker(&self, caller: &str) -> Result<(), StoreError> {
4608 let (ack_tx, ack_rx) = sync_channel::<Result<(), StoreError>>(1);
4609 self.persist_tx
4610 .send(PersistMessage::Ping(ack_tx))
4611 .map_err(|e| {
4612 StoreError::Custom(format!("{caller}: failed to ping persist worker: {e}"))
4613 })?;
4614 ack_rx.recv().map_err(|e| {
4615 StoreError::Custom(format!("{caller}: persist worker ping ack failed: {e}"))
4616 })??;
4617 Ok(())
4618 }
4619
4620 pub fn reorg_overlay_size_hint(&self) -> Result<(usize, usize), StoreError> {
4625 let guard = self.trie_cache.read().map_err(|_| StoreError::LockError)?;
4626 match guard.overlay() {
4627 Some(ov) => Ok((ov.len(), ov.byte_size())),
4628 None => Ok((0, 0)),
4629 }
4630 }
4631
4632 pub fn set_journal_pruning_paused(&self, paused: bool) {
4636 self.journal_pruning_paused
4637 .store(paused, std::sync::atomic::Ordering::Release);
4638 }
4639
4640 pub fn clear_reorg_overlay(&self) -> Result<(), StoreError> {
4644 let mut guard = self.trie_cache.write().map_err(|_| StoreError::LockError)?;
4645 let mut updated = (**guard).clone();
4646 updated.clear_overlay();
4647 *guard = Arc::new(updated);
4648 Ok(())
4649 }
4650
4651 pub fn abort_reorg(&self) -> Result<(), StoreError> {
4657 self.rendezvous_persist_worker("abort_reorg")?;
4665 let mut guard = self.trie_cache.write().map_err(|_| StoreError::LockError)?;
4666 let threshold = guard.commit_threshold();
4667 *guard = Arc::new(TrieLayerCache::new_with_safe_commit(
4668 threshold,
4669 self.safe_commit_root.clone(),
4670 ));
4671 Ok(())
4672 }
4673
4674 pub fn ancestors(&self, block_hash: BlockHash) -> AncestorIterator {
4677 AncestorIterator {
4678 store: self.clone(),
4679 next_hash: block_hash,
4680 }
4681 }
4682
4683 pub fn is_canonical_sync(&self, block_hash: BlockHash) -> Result<bool, StoreError> {
4685 let Some(block_number) = self.get_block_number_sync(block_hash)? else {
4686 return Ok(false);
4687 };
4688 Ok(self
4689 .get_canonical_block_hash_sync(block_number)?
4690 .is_some_and(|h| h == block_hash))
4691 }
4692
4693 pub fn generate_flatkeyvalue(&self) -> Result<(), StoreError> {
4694 self.flatkeyvalue_control_tx
4695 .send(FKVGeneratorControlMessage::Continue)
4696 .map_err(|_| StoreError::Custom("FlatKeyValue thread disconnected.".to_string()))
4697 }
4698
4699 pub fn create_checkpoint(&self, path: impl AsRef<Path>) -> Result<(), StoreError> {
4700 self.backend.create_checkpoint(path.as_ref())?;
4701 init_metadata_file(path.as_ref())?;
4702 Ok(())
4703 }
4704
4705 pub fn get_store_directory(&self) -> Result<PathBuf, StoreError> {
4706 Ok(self.db_path.clone())
4707 }
4708
4709 async fn load_latest_block_number(&self) -> Result<Option<BlockNumber>, StoreError> {
4711 self.read_chain_data_block_number(ChainDataIndex::LatestBlockNumber)
4712 .await
4713 }
4714
4715 fn load_canonical_block_hash(
4716 &self,
4717 block_number: BlockNumber,
4718 ) -> Result<Option<BlockHash>, StoreError> {
4719 let txn = self.backend.begin_read()?;
4720 txn.get(
4721 CANONICAL_BLOCK_HASHES,
4722 block_number.to_le_bytes().as_slice(),
4723 )?
4724 .map(|bytes| H256::decode(bytes.as_slice()))
4725 .transpose()
4726 .map_err(StoreError::from)
4727 }
4728
4729 fn load_block_header(
4730 &self,
4731 block_number: BlockNumber,
4732 ) -> Result<Option<BlockHeader>, StoreError> {
4733 let Some(block_hash) = self.load_canonical_block_hash(block_number)? else {
4734 return Ok(None);
4735 };
4736 self.load_block_header_by_hash(block_hash)
4737 }
4738
4739 fn load_block_header_by_hash(
4741 &self,
4742 block_hash: BlockHash,
4743 ) -> Result<Option<BlockHeader>, StoreError> {
4744 let txn = self.backend.begin_read()?;
4745 let hash_key = block_hash.encode_to_vec();
4746 let header_value = txn.get(HEADERS, hash_key.as_slice())?;
4747 let mut header = header_value
4748 .map(|bytes| BlockHeaderRLP::from_bytes(bytes).to())
4749 .transpose()
4750 .map_err(StoreError::from)?;
4751 header.as_mut().inspect(|h| {
4752 let _ = h.hash.set(block_hash);
4754 });
4755 Ok(header)
4756 }
4757
4758 pub fn last_written(&self) -> Result<Vec<u8>, StoreError> {
4759 let last_computed_flatkeyvalue = self
4760 .last_computed_flatkeyvalue
4761 .read()
4762 .map_err(|_| StoreError::LockError)?;
4763 Ok(last_computed_flatkeyvalue.clone())
4764 }
4765
4766 pub fn flatkeyvalue_fully_generated(&self) -> Result<bool, StoreError> {
4777 let tx = self.backend.begin_read()?;
4778 let marker = tx.get(MISC_VALUES, "last_written".as_bytes())?;
4779 Ok(Self::flatkeyvalue_generation_complete(marker.as_deref()))
4780 }
4781
4782 fn flatkeyvalue_generation_complete(marker: Option<&[u8]>) -> bool {
4786 marker == Some([0xff].as_slice())
4787 }
4788
4789 fn flatkeyvalue_computed_with_last_written(account: H256, last_written: &[u8]) -> bool {
4790 let account_nibbles = Nibbles::from_bytes(account.as_bytes());
4791 &last_written[0..64] > account_nibbles.as_ref()
4792 }
4793
4794 pub fn advance_flushed_upto(&self, block_number: BlockNumber) -> Result<(), StoreError> {
4807 if self
4808 .read_flushed_upto_opt()?
4809 .is_some_and(|current| current >= block_number)
4810 {
4811 return Ok(());
4812 }
4813 let mut tx = self.backend.begin_write()?;
4814 write_flushed_upto(tx.as_mut(), block_number)?;
4815 tx.commit()?;
4816 mutate_block_buffer(&self.block_data_buffer, |b| {
4817 b.set_flushed_upto(block_number)
4818 })
4819 }
4820
4821 pub fn read_flushed_upto(&self) -> Result<BlockNumber, StoreError> {
4826 Ok(self.read_flushed_upto_opt()?.unwrap_or(0))
4827 }
4828
4829 fn read_flushed_upto_opt(&self) -> Result<Option<BlockNumber>, StoreError> {
4832 let tx = self.backend.begin_read()?;
4833 match tx.get(MISC_VALUES, FLUSHED_UPTO_KEY)? {
4834 Some(bytes) => Ok(Some(decode_flushed_upto(&bytes)?)),
4835 None => Ok(None),
4836 }
4837 }
4838
4839 #[cfg(any(test, feature = "testing"))]
4842 pub fn buffer_block_for_test(&self, block: &Block) {
4843 mutate_block_buffer(&self.block_data_buffer, |b| {
4844 b.insert(block.clone(), vec![], vec![])
4845 })
4846 .expect("block_data_buffer lock poisoned");
4847 }
4848
4849 #[cfg(any(test, feature = "testing"))]
4852 pub fn flush_block_data_for_test(&self) -> Result<(), StoreError> {
4853 flush_block_data(self.backend.as_ref(), &self.block_data_buffer)
4854 }
4855
4856 #[cfg(any(test, feature = "testing"))]
4860 pub fn get_trie_node_for_test(
4861 &self,
4862 is_account: bool,
4863 key: &[u8],
4864 ) -> Result<Option<Vec<u8>>, StoreError> {
4865 let table = if is_account {
4866 ACCOUNT_TRIE_NODES
4867 } else {
4868 STORAGE_TRIE_NODES
4869 };
4870 self.backend.begin_read()?.get(table, key)
4871 }
4872
4873 #[cfg(any(test, feature = "testing"))]
4877 pub fn buffer_block_with_codes_for_test(&self, block: &Block, codes: Vec<(H256, Code)>) {
4878 mutate_block_buffer(&self.block_data_buffer, |b| {
4879 b.insert(block.clone(), vec![], codes)
4880 })
4881 .expect("block_data_buffer lock poisoned");
4882 }
4883
4884 #[cfg(any(test, feature = "testing"))]
4889 pub fn register_pending_root_for_test(&self, root: H256) -> Result<(), StoreError> {
4890 self.pending_trie_roots.register(root)
4891 }
4892
4893 #[cfg(any(test, feature = "testing"))]
4896 pub fn clear_pending_root_for_test(&self, root: H256) {
4897 self.pending_trie_roots.clear(root)
4898 }
4899
4900 async fn anchor_to_durable_head(&self, latest: BlockNumber) -> Result<(), StoreError> {
4919 let marker = self.read_flushed_upto_opt()?;
4920 let start = match marker {
4921 Some(flushed) => flushed.min(latest),
4922 None => latest,
4923 };
4924
4925 let mut head = start;
4926 let latest_block_header = loop {
4927 match self.load_block_header(head)? {
4928 Some(header) => break header,
4929 None if marker.is_none() => return Err(StoreError::MissingLatestBlockNumber),
4932 None if head == 0 => return Err(StoreError::MissingLatestBlockNumber),
4933 None => {
4934 warn!(
4935 "durable head {head}: canonical hash has no on-disk header \
4936 (reorg inside flush window); rewinding"
4937 );
4938 head -= 1;
4939 }
4940 }
4941 };
4942 self.latest_block_header.update(latest_block_header);
4943
4944 let reanchor = head != latest;
4949 let rewrite_marker = marker != Some(head);
4950 if reanchor || rewrite_marker {
4951 let mut tx = self.backend.begin_write()?;
4952 if reanchor {
4953 let latest_key = chain_data_key(ChainDataIndex::LatestBlockNumber);
4956 tx.put(CHAIN_DATA, &latest_key, &head.to_le_bytes())?;
4957 }
4958 if rewrite_marker {
4959 write_flushed_upto(tx.as_mut(), head)?;
4960 }
4961 tx.commit()?;
4962 }
4963 Ok(())
4964 }
4965}
4966
4967pub fn write_flushed_upto(
4971 tx: &mut dyn StorageWriteBatch,
4972 n: BlockNumber,
4973) -> Result<(), StoreError> {
4974 tx.put(MISC_VALUES, FLUSHED_UPTO_KEY, &n.to_le_bytes())
4975}
4976
4977fn state_history_key_number(key: &[u8]) -> Result<BlockNumber, StoreError> {
4985 let arr = <[u8; 8]>::try_from(key).map_err(|_| {
4986 StoreError::Custom(format!(
4987 "STATE_HISTORY key has unexpected length: {}",
4988 key.len()
4989 ))
4990 })?;
4991 Ok(BlockNumber::from_be_bytes(arr))
4992}
4993
4994fn decode_flushed_upto(bytes: &[u8]) -> Result<BlockNumber, StoreError> {
4995 let arr: [u8; 8] = bytes
4996 .try_into()
4997 .map_err(|_| StoreError::Custom("Invalid flushed_upto bytes".to_string()))?;
4998 Ok(BlockNumber::from_le_bytes(arr))
4999}
5000
5001fn mutate_block_buffer(
5007 buffer: &Arc<RwLock<Arc<BlockDataBuffer>>>,
5008 f: impl FnOnce(&mut BlockDataBuffer),
5009) -> Result<(), StoreError> {
5010 let mut new_buf = (*buffer.read().map_err(|_| StoreError::LockError)?.clone()).clone();
5011 f(&mut new_buf);
5012 *buffer.write().map_err(|_| StoreError::LockError)? = Arc::new(new_buf);
5013 Ok(())
5014}
5015
5016const DEFAULT_PERSIST_CHANNEL_CAPACITY: usize = 2;
5018
5019struct BlockPersist {
5025 blocks: Vec<(Block, Vec<Receipt>)>,
5026 codes: Vec<(H256, Code)>,
5027 parent_state_root: H256,
5028 child_state_root: H256,
5029 account_updates: TrieNodesUpdate,
5030 storage_updates: Vec<(H256, TrieNodesUpdate)>,
5031 commit_depth: Option<usize>,
5032 wait_for_flush: bool,
5033 block_number: BlockNumber,
5045 block_hash: H256,
5047 ack: std::sync::mpsc::SyncSender<Result<(), StoreError>>,
5048}
5049
5050enum PersistMessage {
5054 Block(Box<BlockPersist>),
5055 Commit(H256),
5061 Ping(std::sync::mpsc::SyncSender<Result<(), StoreError>>),
5062 Shutdown {
5067 ack: std::sync::mpsc::SyncSender<Result<(), StoreError>>,
5068 },
5069}
5070
5071fn write_block_data(
5076 tx: &mut dyn StorageWriteBatch,
5077 number: BlockNumber,
5078 hash: BlockHash,
5079 header: &BlockHeader,
5080 body: &BlockBody,
5081) -> Result<(), StoreError> {
5082 let hash_key = hash.encode_to_vec();
5083 tx.put(
5084 HEADERS,
5085 &hash_key,
5086 BlockHeaderRLP::from(header.clone()).bytes(),
5087 )?;
5088 tx.put(
5089 BODIES,
5090 &hash_key,
5091 BlockBodyRLP::from_bytes(body.encode_to_vec()).bytes(),
5092 )?;
5093 tx.put(BLOCK_NUMBERS, &hash_key, &number.to_le_bytes())?;
5094 for (index, transaction) in body.transactions.iter().enumerate() {
5095 tx.merge(
5096 TRANSACTION_LOCATIONS,
5097 transaction.hash(&NativeCrypto).as_bytes(),
5098 &encode_tx_location_operand(number, hash, index as u64),
5099 )?;
5100 }
5101 Ok(())
5102}
5103
5104fn flush_block_data(
5107 backend: &dyn StorageBackend,
5108 buffer: &Arc<RwLock<Arc<BlockDataBuffer>>>,
5109) -> Result<(), StoreError> {
5110 let snapshot = buffer.read().map_err(|_| StoreError::LockError)?.clone();
5111 let to_flush = snapshot.flushable();
5112 if to_flush.is_empty() {
5113 return Ok(());
5114 }
5115 let hashes: Vec<_> = to_flush.iter().map(|b| b.header.hash()).collect();
5116 let codes = snapshot.codes_for(&hashes);
5117 let mut max_number = snapshot.flushed_upto();
5118
5119 let mut tx = backend.begin_write()?;
5120 for b in &to_flush {
5121 let hash = b.header.hash();
5122 write_block_data(tx.as_mut(), b.number, hash, &b.header, &b.body)?;
5123 for (index, receipt) in b.receipts.iter().enumerate() {
5124 tx.put(
5125 RECEIPTS_V2,
5126 &receipt_key(&hash, index as u64),
5127 &receipt.encode_to_vec(),
5128 )?;
5129 }
5130 max_number = max_number.max(b.number);
5131 }
5132 for (code_hash, code) in codes {
5133 let buf = encode_code(&code);
5134 tx.put(ACCOUNT_CODES, code_hash.as_ref(), &buf)?;
5135 tx.put(
5136 ACCOUNT_CODE_METADATA,
5137 code_hash.as_ref(),
5138 &(code.len() as u64).to_be_bytes(),
5139 )?;
5140 }
5141 write_flushed_upto(tx.as_mut(), max_number)?;
5142 tx.commit()?;
5143
5144 mutate_block_buffer(buffer, |b| b.evict_flushed(max_number))
5146}
5147
5148type TrieNodesUpdate = Vec<(Nibbles, Vec<u8>)>;
5149
5150#[derive(Debug, Default)]
5156struct PendingTrieRoots {
5157 count: AtomicUsize,
5159 roots: Mutex<HashSet<H256>>,
5160 ready: Condvar,
5161}
5162
5163impl PendingTrieRoots {
5164 fn register(&self, root: H256) -> Result<(), StoreError> {
5168 let mut roots = self.roots.lock().map_err(|_| StoreError::LockError)?;
5169 if roots.insert(root) {
5170 self.count.fetch_add(1, Ordering::Release);
5171 }
5172 Ok(())
5173 }
5174
5175 fn clear(&self, root: H256) {
5180 let Ok(mut roots) = self.roots.lock() else {
5181 return;
5182 };
5183 if roots.remove(&root) {
5184 self.count.fetch_sub(1, Ordering::Release);
5185 self.ready.notify_all();
5186 }
5187 }
5188
5189 fn wait_until_ready(&self, root: H256) -> Result<(), StoreError> {
5192 if self.count.load(Ordering::Acquire) == 0 {
5193 return Ok(());
5194 }
5195 let mut roots = self.roots.lock().map_err(|_| StoreError::LockError)?;
5196 while roots.contains(&root) {
5197 roots = self.ready.wait(roots).map_err(|_| StoreError::LockError)?;
5198 }
5199 Ok(())
5200 }
5201}
5202
5203#[allow(clippy::too_many_arguments)]
5207fn apply_trie_phase1(
5208 trie_cache: &Arc<RwLock<Arc<TrieLayerCache>>>,
5209 pending_roots: &PendingTrieRoots,
5210 parent_state_root: H256,
5211 child_state_root: H256,
5212 block_number: BlockNumber,
5213 block_hash: H256,
5214 account_updates: TrieNodesUpdate,
5215 storage_updates: Vec<(H256, TrieNodesUpdate)>,
5216) -> Result<(), StoreError> {
5217 let build: Result<(), StoreError> = (|| {
5218 let new_layer = storage_updates
5219 .into_iter()
5220 .flat_map(|(account_hash, nodes)| {
5221 nodes
5222 .into_iter()
5223 .map(move |(path, node)| (apply_prefix(Some(account_hash), path), node))
5224 })
5225 .chain(account_updates)
5226 .collect();
5227 let trie = trie_cache
5228 .read()
5229 .map_err(|_| StoreError::LockError)?
5230 .clone();
5231 let mut trie_mut = (*trie).clone();
5232 trie_mut.put_batch(
5233 parent_state_root,
5234 child_state_root,
5235 block_number,
5236 block_hash,
5237 new_layer,
5238 );
5239 *trie_cache.write().map_err(|_| StoreError::LockError)? = Arc::new(trie_mut);
5240 Ok(())
5241 })();
5242 pending_roots.clear(child_state_root);
5247 build
5248}
5249
5250fn commit_trie_if_due(
5275 backend: &dyn StorageBackend,
5276 trie_cache: &Arc<RwLock<Arc<TrieLayerCache>>>,
5277 fkv_ctl: &SyncSender<FKVGeneratorControlMessage>,
5278 parent_state_root: H256,
5279 commit_depth: Option<usize>,
5280 is_batch: bool,
5281) -> Result<(), StoreError> {
5282 let trie = trie_cache
5283 .read()
5284 .map_err(|_| StoreError::LockError)?
5285 .clone();
5286 let commitable = match commit_depth {
5288 Some(depth) => trie.get_commitable_by_depth(parent_state_root, depth),
5289 None => trie.get_commitable(parent_state_root),
5290 };
5291 let Some(root) = commitable else {
5292 return Ok(());
5294 };
5295 commit_to_disk(backend, fkv_ctl, trie_cache, &trie, root, is_batch)
5296}
5297
5298fn commit_to_disk(
5308 backend: &dyn StorageBackend,
5309 fkv_ctl: &SyncSender<FKVGeneratorControlMessage>,
5310 trie_cache: &Arc<RwLock<Arc<TrieLayerCache>>>,
5311 trie: &Arc<TrieLayerCache>,
5312 root: H256,
5313 is_batch: bool,
5314) -> Result<(), StoreError> {
5315 if !trie.has_layer(root) {
5319 debug!(
5320 root = ?root,
5321 layers = trie.layer_count(),
5322 is_batch,
5323 "Skipping trie commit: state root has no in-memory layer. Expected when the block \
5324 did not change the state root (empty L2 blocks) or the root was already flushed."
5325 );
5326 return Ok(());
5327 }
5328
5329 let _ = fkv_ctl.send(FKVGeneratorControlMessage::Stop);
5332
5333 let mut trie_mut = (**trie).clone();
5335
5336 let read_view = backend.begin_read()?;
5350 let last_written = read_view
5351 .get(MISC_VALUES, "last_written".as_bytes())?
5352 .unwrap_or_default();
5353
5354 let mut write_tx = backend.begin_write()?;
5355
5356 let overlay_for_reconciliation = if !is_batch {
5366 trie.overlay().cloned()
5367 } else {
5368 None
5369 };
5370
5371 let root = if overlay_for_reconciliation.is_some() {
5381 trie.bottom_layer_root(root)
5382 } else {
5383 root
5384 };
5385
5386 let committed_layers = trie_mut.commit(root).ok_or_else(|| {
5396 StoreError::Custom(format!(
5397 "trie layer for state root {root:?} disappeared between the has_layer check and \
5398 commit (layers={}, is_batch={is_batch}): the TrieLayerCache clone lost the layer \
5399 map, so this block's state was not written",
5400 trie.layer_count(),
5401 ))
5402 })?;
5403
5404 debug_assert!(
5409 overlay_for_reconciliation.is_none() || committed_layers.len() == 1,
5410 "overlay-backed reconciliation must commit exactly one layer (T), got {}",
5411 committed_layers.len()
5412 );
5413
5414 let extra_writes: Vec<(Vec<u8>, Vec<u8>)> = match &overlay_for_reconciliation {
5420 Some(overlay) => {
5421 let layer_keys: SeededHashSet<&Vec<u8>> = committed_layers
5422 .iter()
5423 .flat_map(|l| l.nodes.iter().map(|(k, _)| k))
5424 .collect();
5425 overlay
5426 .iter_all_entries()
5427 .filter(|(_, key, _)| !layer_keys.contains(key))
5428 .map(|(_, key, value)| (key.clone(), value.clone().unwrap_or_default()))
5429 .collect()
5430 }
5431 None => Vec::new(),
5432 };
5433 let reorg_heights = overlay_for_reconciliation
5435 .as_ref()
5436 .map(|ov| (ov.to_block(), ov.from_block()));
5437
5438 let skip_journal_upto = if is_batch || overlay_for_reconciliation.is_some() {
5450 None
5451 } else {
5452 journal_skip_ceiling(read_view.as_ref())
5453 };
5454 let journals_layer = |block_number: BlockNumber| {
5455 !is_batch && skip_journal_upto.is_none_or(|upto| block_number > upto)
5456 };
5457 let track_overlay = committed_layers
5460 .iter()
5461 .any(|layer| journals_layer(layer.block_number));
5462
5463 let mut overlay: HashMap<Vec<u8>, Option<Vec<u8>>> = HashMap::new();
5474
5475 let mut result = Ok(());
5476 'layers: for layer in &committed_layers {
5477 let journal_layer = journals_layer(layer.block_number);
5486 let mut journal_account_trie: FlatDiff = Vec::new();
5487 let mut journal_storage_trie: FlatDiff = Vec::new();
5488 let mut journal_account_flat: FlatDiff = Vec::new();
5489 let mut journal_storage_flat: FlatDiff = Vec::new();
5490
5491 let is_reconciliation_layer = overlay_for_reconciliation.is_some();
5502 let extra: &[(Vec<u8>, Vec<u8>)] = if is_reconciliation_layer {
5503 &extra_writes
5504 } else {
5505 &[]
5506 };
5507
5508 let pivot_overlay = overlay_for_reconciliation.as_ref();
5516
5517 for (key, value) in layer.nodes.iter().chain(extra.iter()) {
5518 let (is_leaf, is_account) = classify_trie_key(key.len());
5519
5520 if is_leaf && key.as_slice() > last_written.as_slice() {
5525 continue;
5526 }
5527 let table = if is_leaf {
5528 if is_account {
5529 &ACCOUNT_FLATKEYVALUE
5530 } else {
5531 &STORAGE_FLATKEYVALUE
5532 }
5533 } else if is_account {
5534 &ACCOUNT_TRIE_NODES
5535 } else {
5536 &STORAGE_TRIE_NODES
5537 };
5538
5539 let prev_value = if journal_layer {
5543 match overlay.get(key) {
5544 Some(v) => Some(v.clone()),
5545 None => match pivot_overlay
5546 .and_then(|ov| ov.lookup(OverlayCf::classify_by_key_length(key.len()), key))
5547 {
5548 Some(v) => Some(v),
5549 None => match read_view.get(table, key) {
5550 Ok(v) => Some(v),
5551 Err(e) => {
5552 result = Err(e);
5553 break 'layers;
5554 }
5555 },
5556 },
5557 }
5558 } else {
5559 None
5560 };
5561
5562 let new_value = if value.is_empty() {
5563 result = write_tx.delete(table, key);
5564 None
5565 } else {
5566 result = write_tx.put(table, key, value);
5567 Some(value.clone())
5568 };
5569 if result.is_err() {
5570 break 'layers;
5571 }
5572
5573 if let Some(prev) = prev_value {
5576 let bucket = match (is_leaf, is_account) {
5577 (false, true) => &mut journal_account_trie,
5578 (false, false) => &mut journal_storage_trie,
5579 (true, true) => &mut journal_account_flat,
5580 (true, false) => &mut journal_storage_flat,
5581 };
5582 bucket.push((key.clone(), prev));
5583 }
5584 if track_overlay {
5587 overlay.insert(key.clone(), new_value);
5588 }
5589 }
5590
5591 if !is_batch && let Some((t, d)) = reorg_heights {
5596 debug_assert_eq!(
5597 layer.block_number, t,
5598 "first new-chain commit must be at the pivot's T height (overlay.to_block)"
5599 );
5600 if layer.block_number == t {
5607 let start = t.to_be_bytes();
5608 let end = d.saturating_add(1).to_be_bytes();
5609 result = write_tx.delete_range(STATE_HISTORY, &start, &end);
5610 if result.is_err() {
5611 break 'layers;
5612 }
5613 } else {
5614 error!(
5615 block_number = layer.block_number,
5616 t,
5617 d,
5618 "deep-reorg reconciliation skipped: committed layer not at pivot height; skipping STATE_HISTORY delete_range to avoid history gaps"
5619 );
5620 }
5621 }
5622
5623 if journal_layer {
5629 let entry = JournalEntry {
5630 block_hash: layer.block_hash,
5631 parent_state_root: layer.parent_state_root,
5632 account_trie_diff: journal_account_trie,
5633 storage_trie_diff: journal_storage_trie,
5634 account_flat_diff: journal_account_flat,
5635 storage_flat_diff: journal_storage_flat,
5636 };
5637 result = write_tx.put(
5638 STATE_HISTORY,
5639 &layer.block_number.to_be_bytes(),
5640 &entry.encode(),
5641 );
5642 if result.is_err() {
5643 break 'layers;
5644 }
5645 }
5646 }
5647
5648 if result.is_ok() {
5649 result = write_tx.commit();
5650 }
5651 let _ = fkv_ctl.send(FKVGeneratorControlMessage::Continue);
5653 result?;
5654
5655 if overlay_for_reconciliation.is_some() {
5658 trie_mut.clear_overlay();
5659 }
5660
5661 *trie_cache.write().map_err(|_| StoreError::LockError)? = Arc::new(trie_mut);
5667 Ok(())
5668}
5669
5670fn journal_skip_ceiling(read_view: &dyn StorageReadView) -> Option<BlockNumber> {
5693 let read_number = |index| {
5694 read_view
5695 .get(CHAIN_DATA, &chain_data_key(index))
5696 .ok()
5697 .flatten()
5698 .and_then(|bytes| <[u8; 8]>::try_from(bytes.as_slice()).ok())
5699 .map(BlockNumber::from_le_bytes)
5700 };
5701 let finalized = read_number(ChainDataIndex::FinalizedBlockNumber)?;
5702 let head = read_number(ChainDataIndex::LatestBlockNumber)?;
5703 match read_view.first_key(STATE_HISTORY) {
5704 Ok(None) => Some(finalized.min(head)),
5705 _ => None,
5706 }
5707}
5708
5709fn flatkeyvalue_generator(
5712 backend: &Arc<dyn StorageBackend>,
5713 last_computed_fkv: &RwLock<Vec<u8>>,
5714 control_rx: &std::sync::mpsc::Receiver<FKVGeneratorControlMessage>,
5715) -> Result<(), StoreError> {
5716 info!("Generation of FlatKeyValue started.");
5717 let initial_last_written = backend
5718 .begin_read()?
5719 .get(MISC_VALUES, "last_written".as_bytes())?
5720 .unwrap_or_default();
5721
5722 if initial_last_written.is_empty() {
5723 backend.clear_table(ACCOUNT_FLATKEYVALUE)?;
5725 backend.clear_table(STORAGE_FLATKEYVALUE)?;
5726 } else if initial_last_written == [0xff] {
5727 info!("FlatKeyValue already generated. Skipping.");
5729 return Ok(());
5730 }
5731
5732 loop {
5733 let read_tx = backend.begin_read()?;
5736 let root = read_tx
5737 .get(ACCOUNT_TRIE_NODES, &[])?
5738 .ok_or(StoreError::MissingLatestBlockNumber)?;
5739 let root: Node = ethrex_trie::Node::decode(&root)?;
5740 let state_root = root.compute_hash(&NativeCrypto).finalize(&NativeCrypto);
5741
5742 let last_written = read_tx
5743 .get(MISC_VALUES, "last_written".as_bytes())?
5744 .unwrap_or_default();
5745 let last_written_account = last_written
5746 .get(0..64)
5747 .map(|v| Nibbles::from_hex(v.to_vec()))
5748 .unwrap_or_default();
5749 let mut last_written_storage = last_written
5750 .get(66..130)
5751 .map(|v| Nibbles::from_hex(v.to_vec()))
5752 .unwrap_or_default();
5753
5754 debug!("Starting FlatKeyValue loop pivot={last_written:?} SR={state_root:x}");
5755
5756 let mut ctr = 0;
5757 let mut write_txn = backend.begin_write()?;
5758 let mut iter = Trie::open(
5759 Box::new(BackendTrieDB::new_for_accounts_with_view(
5760 backend.clone(),
5761 read_tx.clone(),
5762 last_written.clone(),
5763 )?),
5764 state_root,
5765 )
5766 .into_iter();
5767 if last_written_account > Nibbles::default() {
5768 iter.advance(last_written_account.to_bytes())?;
5769 }
5770 let res = iter.try_for_each(|(path, node)| -> Result<(), StoreError> {
5771 let Node::Leaf(node) = node else {
5772 return Ok(());
5773 };
5774 let account_state = AccountState::decode(&node.value)?;
5775 let account_hash = H256::from_slice(&path.to_bytes());
5776 write_txn.put(MISC_VALUES, "last_written".as_bytes(), path.as_ref())?;
5777 write_txn.put(ACCOUNT_FLATKEYVALUE, path.as_ref(), &node.value)?;
5778 ctr += 1;
5779 if ctr > 10_000 {
5780 write_txn.commit()?;
5781 write_txn = backend.begin_write()?;
5782 *last_computed_fkv
5783 .write()
5784 .map_err(|_| StoreError::LockError)? = path.as_ref().to_vec();
5785 ctr = 0;
5786 }
5787
5788 let mut iter_inner = Trie::open(
5789 Box::new(BackendTrieDB::new_for_account_storage_with_view(
5790 backend.clone(),
5791 read_tx.clone(),
5792 account_hash,
5793 path.as_ref().to_vec(),
5794 )?),
5795 account_state.storage_root,
5796 )
5797 .into_iter();
5798 if last_written_storage > Nibbles::default() {
5799 iter_inner.advance(last_written_storage.to_bytes())?;
5800 last_written_storage = Nibbles::default();
5801 }
5802 iter_inner.try_for_each(|(path, node)| -> Result<(), StoreError> {
5803 let Node::Leaf(node) = node else {
5804 return Ok(());
5805 };
5806 let key = apply_prefix(Some(account_hash), path);
5807 write_txn.put(MISC_VALUES, "last_written".as_bytes(), key.as_ref())?;
5808 write_txn.put(STORAGE_FLATKEYVALUE, key.as_ref(), &node.value)?;
5809 ctr += 1;
5810 if ctr > 10_000 {
5811 write_txn.commit()?;
5812 write_txn = backend.begin_write()?;
5813 *last_computed_fkv
5814 .write()
5815 .map_err(|_| StoreError::LockError)? = key.into_vec();
5816 ctr = 0;
5817 }
5818 fkv_check_for_stop_msg(control_rx)?;
5819 Ok(())
5820 })?;
5821 fkv_check_for_stop_msg(control_rx)?;
5822 Ok(())
5823 });
5824 match res {
5825 Err(StoreError::PivotChanged) => {
5826 match control_rx.recv() {
5827 Ok(FKVGeneratorControlMessage::Continue) => {}
5828 Ok(FKVGeneratorControlMessage::Stop) => {
5829 return Err(StoreError::Custom("Unexpected Stop message".to_string()));
5830 }
5831 Err(std::sync::mpsc::RecvError) => {
5833 info!("Store closed, stopping FlatKeyValue generation.");
5834 return Ok(());
5835 }
5836 }
5837 }
5838 Err(err) => return Err(err),
5839 Ok(()) => {
5840 write_txn.put(MISC_VALUES, "last_written".as_bytes(), &[0xff])?;
5841 write_txn.commit()?;
5842 *last_computed_fkv
5843 .write()
5844 .map_err(|_| StoreError::LockError)? = vec![0xff; 131];
5845 info!("FlatKeyValue generation finished.");
5846 return Ok(());
5847 }
5848 };
5849 }
5850}
5851
5852fn fkv_check_for_stop_msg(
5853 control_rx: &std::sync::mpsc::Receiver<FKVGeneratorControlMessage>,
5854) -> Result<(), StoreError> {
5855 match control_rx.try_recv() {
5856 Ok(FKVGeneratorControlMessage::Stop) | Err(TryRecvError::Disconnected) => {
5857 return Err(StoreError::PivotChanged);
5858 }
5859 Ok(FKVGeneratorControlMessage::Continue) => {
5860 return Err(StoreError::Custom(
5861 "Unexpected Continue message".to_string(),
5862 ));
5863 }
5864 Err(TryRecvError::Empty) => {}
5865 }
5866 Ok(())
5867}
5868
5869fn state_trie_locked_backend(
5870 backend: &dyn StorageBackend,
5871 last_written: Vec<u8>,
5872) -> Result<BackendTrieDBLocked, StoreError> {
5873 BackendTrieDBLocked::new(backend, last_written)
5875}
5876
5877pub struct AccountProof {
5878 pub proof: Vec<NodeRLP>,
5879 pub account: AccountState,
5880 pub storage_proof: Vec<StorageSlotProof>,
5881}
5882
5883pub struct StorageSlotProof {
5884 pub proof: Vec<NodeRLP>,
5885 pub key: H256,
5886 pub value: U256,
5887}
5888
5889pub struct AncestorIterator {
5890 store: Store,
5891 next_hash: BlockHash,
5892}
5893
5894impl Iterator for AncestorIterator {
5895 type Item = Result<(BlockHash, BlockHeader), StoreError>;
5896
5897 fn next(&mut self) -> Option<Self::Item> {
5898 let next_hash = self.next_hash;
5899 match self.store.get_block_header_by_hash(next_hash) {
5903 Ok(Some(header)) => {
5904 let ret_hash = self.next_hash;
5905 self.next_hash = header.parent_hash;
5906 Some(Ok((ret_hash, header)))
5907 }
5908 Ok(None) => None,
5909 Err(e) => Some(Err(e)),
5910 }
5911 }
5912}
5913
5914pub fn hash_address(address: &Address) -> Vec<u8> {
5915 keccak_hash(address.to_fixed_bytes()).to_vec()
5916}
5917
5918fn hash_address_fixed(address: &Address) -> H256 {
5919 keccak(address.to_fixed_bytes())
5920}
5921
5922pub fn hash_key(key: &H256) -> Vec<u8> {
5923 keccak_hash(key.to_fixed_bytes()).to_vec()
5924}
5925
5926pub fn hash_key_fixed(key: &H256) -> [u8; 32] {
5927 keccak_hash(key.to_fixed_bytes())
5928}
5929
5930fn chain_data_key(index: ChainDataIndex) -> Vec<u8> {
5931 (index as u8).encode_to_vec()
5932}
5933
5934fn decode_block_number(bytes: Vec<u8>) -> Result<BlockNumber, StoreError> {
5936 let array: [u8; 8] = bytes
5937 .try_into()
5938 .map_err(|_| StoreError::Custom("Invalid BlockNumber bytes".to_string()))?;
5939 Ok(BlockNumber::from_le_bytes(array))
5940}
5941
5942fn snap_state_key(index: SnapStateIndex) -> Vec<u8> {
5943 (index as u8).encode_to_vec()
5944}
5945
5946pub fn receipt_key(block_hash: &BlockHash, index: u64) -> Vec<u8> {
5948 let mut key = Vec::with_capacity(40);
5949 key.extend_from_slice(block_hash.as_bytes());
5950 key.extend_from_slice(&index.to_be_bytes());
5951 key
5952}
5953
5954pub fn encode_code(code: &Code) -> Vec<u8> {
5955 let jumpdests = code.jumpdests();
5956 let mut buf = Vec::with_capacity(6 + code.len() + jumpdests.len());
5957 code.code().encode(&mut buf);
5958 jumpdests.encode(&mut buf);
5959 buf
5960}
5961
5962fn decode_jumpdests(code: &[u8], encoded: &[u8]) -> Result<(Arc<[u8]>, bool), StoreError> {
5973 let (is_list, payload, _) = decode_rlp_item(encoded)?;
5974 if is_list {
5975 return Ok((Code::compute_jumpdests(code), true));
5976 }
5977 Ok((Arc::from(payload), false))
5978}
5979
5980#[derive(Debug, Default, Clone)]
5981struct LatestBlockHeaderCache {
5982 current: Arc<Mutex<Arc<BlockHeader>>>,
5983}
5984
5985impl LatestBlockHeaderCache {
5986 pub fn get(&self) -> Arc<BlockHeader> {
5987 self.current.lock().expect("poisoned mutex").clone()
5988 }
5989
5990 pub fn update(&self, header: BlockHeader) {
5991 let new = Arc::new(header);
5992 *self.current.lock().expect("poisoned mutex") = new;
5993 }
5994}
5995
5996#[derive(Debug, Serialize, Deserialize)]
5997pub struct StoreMetadata {
5998 pub schema_version: u64,
5999}
6000
6001impl StoreMetadata {
6002 pub fn new(schema_version: u64) -> Self {
6003 Self { schema_version }
6004 }
6005}
6006
6007fn read_store_schema_version(path: &Path) -> Result<Option<u64>, StoreError> {
6012 let metadata_path = path.join(STORE_METADATA_FILENAME);
6013 if !metadata_path.exists() {
6014 return Ok(None);
6015 }
6016 if !metadata_path.is_file() {
6017 return Err(StoreError::Custom(
6018 "store schema path exists but is not a file".to_string(),
6019 ));
6020 }
6021 let file_contents = std::fs::read_to_string(metadata_path)?;
6022 let metadata: StoreMetadata = serde_json::from_str(&file_contents)?;
6023 Ok(Some(metadata.schema_version))
6024}
6025
6026fn init_metadata_file(parent_path: &Path) -> Result<(), StoreError> {
6027 std::fs::create_dir_all(parent_path)?;
6028
6029 let metadata_path = parent_path.join(STORE_METADATA_FILENAME);
6030 let metadata = StoreMetadata::new(STORE_SCHEMA_VERSION);
6031 let serialized_metadata = serde_json::to_string_pretty(&metadata)?;
6032 let mut new_file = std::fs::File::create_new(metadata_path)?;
6033 new_file.write_all(serialized_metadata.as_bytes())?;
6034 Ok(())
6035}
6036
6037fn dir_contains_legacy_db(path: &Path) -> Result<bool, StoreError> {
6049 if path.join("CURRENT").is_file() {
6053 return Ok(true);
6054 }
6055 for entry in std::fs::read_dir(path)? {
6059 let entry = entry?;
6060 if !entry.file_type()?.is_file() {
6061 continue;
6062 }
6063 if entry.file_name().to_string_lossy().starts_with("MANIFEST-") {
6064 return Ok(true);
6065 }
6066 }
6067 Ok(false)
6068}
6069
6070pub fn has_valid_db(path: &Path) -> bool {
6074 let metadata_path = path.join(STORE_METADATA_FILENAME);
6075 if !metadata_path.is_file() {
6076 return false;
6077 }
6078 let Ok(contents) = std::fs::read_to_string(&metadata_path) else {
6079 return false;
6080 };
6081 let Ok(metadata) = serde_json::from_str::<StoreMetadata>(&contents) else {
6082 return false;
6083 };
6084 metadata.schema_version >= 1 && metadata.schema_version <= STORE_SCHEMA_VERSION
6085}
6086
6087pub fn read_chain_id_from_db(path: &Path) -> Option<u64> {
6096 if !has_valid_db(path) {
6097 return None;
6098 }
6099 #[cfg(feature = "rocksdb")]
6100 {
6101 let backend = match RocksDBBackend::open(path, MAX_ROCKSDB_BLOCK_CACHE_SIZE_BYTES) {
6104 Ok(backend) => backend,
6105 Err(e) => {
6106 warn!("Failed to open RocksDB at {path:?} to read chain ID: {e}");
6107 return None;
6108 }
6109 };
6110 let read = match backend.begin_read() {
6111 Ok(read) => read,
6112 Err(e) => {
6113 warn!("Failed to begin read transaction at {path:?}: {e}");
6114 return None;
6115 }
6116 };
6117 let key = chain_data_key(ChainDataIndex::ChainConfig);
6118 let bytes = match read.get(CHAIN_DATA, &key) {
6119 Ok(Some(bytes)) => bytes,
6120 Ok(None) => {
6121 warn!("Chain config entry not found in database at {path:?}");
6122 return None;
6123 }
6124 Err(e) => {
6125 warn!("Failed to read chain config from database at {path:?}: {e}");
6126 return None;
6127 }
6128 };
6129 #[derive(serde::Deserialize)]
6134 #[serde(rename_all = "camelCase")]
6135 struct ChainIdOnly {
6136 chain_id: u64,
6137 }
6138 match serde_json::from_slice::<ChainIdOnly>(&bytes) {
6139 Ok(partial) => Some(partial.chain_id),
6140 Err(e) => {
6141 warn!("Failed to deserialize chain ID from database at {path:?}: {e}");
6142 None
6143 }
6144 }
6145 }
6146 #[cfg(not(feature = "rocksdb"))]
6147 {
6148 let _ = path;
6149 None
6150 }
6151}
6152
6153#[cfg(test)]
6154mod account_code_tests {
6155 use super::*;
6156
6157 const JUMPDEST: u8 = 0x5b;
6158 const PUSH1: u8 = 0x60;
6159
6160 #[test]
6166 fn metadata_falls_back_to_the_bytecode_without_a_tokio_runtime() {
6167 use crate::backend::in_memory::InMemoryBackend;
6168
6169 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6170 let dir = tempfile::tempdir().unwrap();
6171 let store = Store::from_backend(
6172 backend.clone(),
6173 dir.path().to_path_buf(),
6174 1,
6175 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6176 )
6177 .unwrap();
6178
6179 let code = jumpdest_dense_code();
6182 let mut tx = backend.begin_write().unwrap();
6183 tx.put(ACCOUNT_CODES, code.hash.as_bytes(), &encode_code(&code))
6184 .unwrap();
6185 tx.commit().unwrap();
6186
6187 let metadata = store
6188 .get_code_metadata(code.hash)
6189 .expect("metadata read must not fail off-runtime");
6190 assert_eq!(metadata.map(|m| m.length), Some(code.len() as u64));
6191 }
6192
6193 fn jumpdest_dense_code() -> Code {
6197 let mut bytecode = vec![JUMPDEST; 24576];
6198 bytecode[0] = 0x00;
6199 Code::from_bytecode_unchecked(bytecode.into(), H256::zero())
6200 }
6201
6202 fn encode_code_legacy(code: &Code) -> Vec<u8> {
6205 let offsets: Vec<u32> = (0..code.len())
6206 .filter(|offset| code.is_valid_jumpdest(*offset))
6207 .map(|offset| offset as u32)
6208 .collect();
6209 let mut buf = Vec::new();
6210 code.code().encode(&mut buf);
6211 offsets.encode(&mut buf);
6212 buf
6213 }
6214
6215 #[test]
6216 fn encoded_value_carries_one_bitmap_bit_per_code_byte() {
6217 let code = jumpdest_dense_code();
6218 let encoded = encode_code(&code);
6219
6220 assert_eq!(encoded.len(), 6 + code.len() + code.len().div_ceil(8));
6221 }
6222
6223 #[test]
6224 fn round_trip_preserves_the_bitmap() {
6225 let code = jumpdest_dense_code();
6226 let encoded = encode_code(&code);
6227
6228 let (bytecode, jumpdests) = decode_bytes(&encoded).unwrap();
6229 assert_eq!(bytecode, code.code());
6230 let (rebuilt, was_legacy) = decode_jumpdests(bytecode, jumpdests).unwrap();
6231 assert_eq!(rebuilt.as_ref(), code.jumpdests());
6232 assert!(!was_legacy, "bitmap format must not be flagged legacy");
6233 }
6234
6235 #[test]
6242 fn legacy_offset_list_values_decode_to_the_same_bitmap() {
6243 for (bytecode, offsets) in [
6244 (vec![0x00, JUMPDEST, 0x00, JUMPDEST], vec![1u32, 3]),
6245 (vec![PUSH1, JUMPDEST, JUMPDEST], vec![2u32]),
6247 (vec![0x00; 32], vec![]),
6248 ] {
6249 let code = Code::from_bytecode_unchecked(bytecode.clone().into(), H256::zero());
6250 let legacy = encode_code_legacy(&code);
6251
6252 let mut expected = vec![0u8; bytecode.len().div_ceil(8)];
6253 for offset in &offsets {
6254 let index = usize::try_from(*offset).expect("offset fits usize");
6255 expected[index / 8] |= 1 << (index % 8);
6256 }
6257 if offsets.is_empty() {
6259 expected.clear();
6260 }
6261
6262 let (decoded_bytecode, jumpdests) = decode_bytes(&legacy).unwrap();
6263 assert_eq!(decoded_bytecode, code.code());
6264 let (rebuilt, was_legacy) = decode_jumpdests(decoded_bytecode, jumpdests).unwrap();
6265 assert_eq!(
6266 rebuilt.as_ref(),
6267 expected.as_slice(),
6268 "rebuilt bitmap disagrees with the legacy offsets for {bytecode:?}"
6269 );
6270 assert!(
6271 was_legacy,
6272 "legacy RLP-list format must be flagged for self-heal"
6273 );
6274 }
6275 }
6276
6277 #[test]
6280 fn repeated_inserts_do_not_inflate_the_accounted_size() {
6281 let code = jumpdest_dense_code();
6282 let mut cache = CodeCache::default();
6283
6284 cache.insert(&code).unwrap();
6285 let after_first = cache.cache_size;
6286 for _ in 0..8 {
6287 cache.insert(&code).unwrap();
6288 }
6289
6290 assert_eq!(cache.cache_size, after_first);
6291 assert_eq!(cache.entries.len(), 1);
6292 }
6293
6294 #[test]
6297 fn accounted_size_covers_the_bytecode_and_bitmap() {
6298 let code = jumpdest_dense_code();
6299 let mut cache = CodeCache::default();
6300 cache.insert(&code).unwrap();
6301
6302 assert!(cache.cache_size >= (code.len() + code.jumpdests().len()) as u64);
6303 }
6304
6305 const TEST_RNG_SEED: u64 = 0x9E37_79B9_7F4A_7C15;
6307
6308 #[test]
6309 fn cache_evicts_down_to_its_budget() {
6310 let mut cache = CodeCache {
6311 max_size: 128 * 1024,
6312 rng: TEST_RNG_SEED,
6313 ..Default::default()
6314 };
6315
6316 for i in 0..16u8 {
6317 let mut bytecode = vec![JUMPDEST; 24576];
6318 bytecode[0] = i;
6319 cache
6320 .insert(&Code::from_bytecode_unchecked(
6321 bytecode.into(),
6322 H256::from_low_u64_be(i.into()),
6323 ))
6324 .unwrap();
6325 }
6326
6327 assert!(cache.cache_size <= cache.max_size);
6328 assert!(cache.entries.len() < 16);
6329 }
6330
6331 fn numbered_code(i: u64, len: usize) -> Code {
6332 let mut bytecode = vec![JUMPDEST; len];
6333 bytecode[..8].copy_from_slice(&i.to_be_bytes());
6334 Code::from_bytecode_unchecked(bytecode.into(), H256::from_low_u64_be(i))
6335 }
6336
6337 #[test]
6341 fn a_cyclic_scan_larger_than_the_budget_still_hits() {
6342 let entry_size = CodeCache::entry_size(&numbered_code(0, 4096));
6343 let mut cache = CodeCache {
6344 max_size: 400 * entry_size,
6345 rng: TEST_RNG_SEED,
6346 ..Default::default()
6347 };
6348 let codes: Vec<Code> = (0..500).map(|i| numbered_code(i, 4096)).collect();
6350
6351 let (mut hits, mut reads) = (0u64, 0u64);
6352 for round in 0..20 {
6353 for code in &codes {
6354 let cached = cache.get(&code.hash).unwrap();
6355 if round >= 5 {
6356 reads += 1;
6357 hits += u64::from(cached.is_some());
6358 }
6359 if cached.is_none() {
6360 cache.insert(code).unwrap();
6361 }
6362 }
6363 }
6364
6365 assert!(cache.cache_size <= cache.max_size);
6366 assert!(
6367 hits * 2 > reads,
6368 "only {hits} of {reads} reads hit once the scan outgrew the budget"
6369 );
6370 }
6371
6372 #[test]
6374 fn eviction_keeps_the_entry_being_inserted() {
6375 let entry_size = CodeCache::entry_size(&numbered_code(0, 4096));
6376 let mut cache = CodeCache {
6377 max_size: 2 * entry_size,
6378 rng: TEST_RNG_SEED,
6379 ..Default::default()
6380 };
6381 for i in 0..64 {
6382 let code = numbered_code(i, 4096);
6383 cache.insert(&code).unwrap();
6384 assert!(cache.get(&code.hash).unwrap().is_some());
6385 assert!(cache.cache_size <= cache.max_size);
6386 }
6387 }
6388
6389 #[test]
6392 fn an_entry_larger_than_the_budget_is_not_cached() {
6393 let small = numbered_code(0, 1024);
6394 let mut cache = CodeCache {
6395 max_size: 4 * CodeCache::entry_size(&small),
6396 rng: TEST_RNG_SEED,
6397 ..Default::default()
6398 };
6399 cache.insert(&small).unwrap();
6400
6401 let large = numbered_code(1, 8192);
6402 assert!(CodeCache::entry_size(&large) > cache.max_size);
6403 cache.insert(&large).unwrap();
6404
6405 assert!(!cache.contains(&large.hash));
6406 assert!(cache.contains(&small.hash));
6407 assert_eq!(cache.cache_size, CodeCache::entry_size(&small));
6408 }
6409
6410 #[test]
6413 fn debug_output_leaves_out_the_cached_code() {
6414 let mut cache = CodeCache::default();
6415 cache.insert(&numbered_code(7, 4096)).unwrap();
6416 let printed = format!("{cache:?}");
6417 assert!(printed.len() < 200, "{printed}");
6418 }
6419
6420 #[test]
6423 fn small_contracts_are_charged_their_entry_overhead() {
6424 let mut cache = CodeCache {
6425 max_size: 160 * 1024,
6426 rng: TEST_RNG_SEED,
6427 ..Default::default()
6428 };
6429 let mut charged = 0;
6430 for i in 0..4096u64 {
6431 let mut bytecode = vec![JUMPDEST; 32];
6432 bytecode[..8].copy_from_slice(&i.to_be_bytes());
6433 let code = Code::from_bytecode_unchecked(bytecode.into(), H256::from_low_u64_be(i));
6434 charged = code.size() as u64 + CODE_CACHE_ENTRY_OVERHEAD;
6435 cache.insert(&code).unwrap();
6436 }
6437
6438 assert!(cache.cache_size <= cache.max_size);
6439 assert_eq!(cache.entries.len() as u64, cache.max_size / charged);
6440 }
6441
6442 #[test]
6445 fn default_cache_uses_the_configured_budget() {
6446 assert_eq!(CodeCache::default().max_size, CODE_CACHE_MAX_SIZE);
6447 }
6448
6449 #[test]
6452 fn store_uses_the_configured_code_cache_budget() {
6453 let config = StoreConfig {
6454 code_cache_size: Some(300 * 1024 * 1024),
6455 ..StoreConfig::with_rocksdb_block_cache_size(MIN_ROCKSDB_BLOCK_CACHE_SIZE_BYTES)
6456 };
6457 let store = Store::new_with_config("", EngineType::InMemory, config).unwrap();
6458 assert_eq!(store.code_cache_budget_bytes(), 300 * 1024 * 1024);
6459
6460 let config = StoreConfig::with_rocksdb_block_cache_size(MIN_ROCKSDB_BLOCK_CACHE_SIZE_BYTES);
6461 let store = Store::new_with_config("", EngineType::InMemory, config).unwrap();
6462 assert_eq!(store.code_cache_budget_bytes(), CODE_CACHE_MAX_SIZE);
6463 }
6464}
6465
6466#[cfg(test)]
6467mod state_history_tests {
6468 use super::*;
6469 use crate::api::tables::STATE_HISTORY;
6470 use crate::backend::in_memory::InMemoryBackend;
6471 use crate::journal::JournalEntry;
6472 use ethrex_common::types::{BlockBody, BlockHeader};
6473 use ethrex_trie::Nibbles;
6474 use std::time::{Duration, Instant};
6475
6476 fn make_block(number: BlockNumber, parent_hash: H256, state_root: H256) -> Block {
6477 let header = BlockHeader {
6478 number,
6479 parent_hash,
6480 state_root,
6481 ..Default::default()
6482 };
6483 Block::new(header, BlockBody::default())
6484 }
6485
6486 fn await_journal_entry(
6490 backend: &Arc<dyn StorageBackend>,
6491 block_number: BlockNumber,
6492 timeout: Duration,
6493 ) -> Option<Vec<u8>> {
6494 let key = block_number.to_be_bytes();
6495 let deadline = Instant::now() + timeout;
6496 loop {
6497 let read = backend.begin_read().ok()?;
6498 if let Ok(Some(v)) = read.get(STATE_HISTORY, &key) {
6499 return Some(v);
6500 }
6501 if Instant::now() >= deadline {
6502 return None;
6503 }
6504 std::thread::sleep(Duration::from_millis(10));
6505 }
6506 }
6507
6508 fn assert_no_journal_entry(backend: &Arc<dyn StorageBackend>, block_number: BlockNumber) {
6513 let window = Duration::from_millis(500);
6514 let key = block_number.to_be_bytes();
6515 let deadline = Instant::now() + window;
6516 loop {
6517 let read = backend.begin_read().expect("read view");
6518 let v = read.get(STATE_HISTORY, &key).expect("get");
6519 assert!(
6520 v.is_none(),
6521 "expected no STATE_HISTORY entry for block {block_number}, got {v:?}"
6522 );
6523 if Instant::now() >= deadline {
6524 return;
6525 }
6526 std::thread::sleep(Duration::from_millis(20));
6527 }
6528 }
6529
6530 #[test]
6536 fn journal_entry_written_per_block_in_regular_mode() {
6537 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6538 let dir = tempfile::tempdir().unwrap();
6539 let store = Store::from_backend(
6540 backend.clone(),
6541 dir.path().to_path_buf(),
6542 1,
6543 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6544 )
6545 .unwrap();
6546
6547 let state_root_1 = H256::repeat_byte(0x11);
6548 let block1 = make_block(1, H256::zero(), state_root_1);
6549 let block1_hash = block1.hash();
6550 store
6551 .store_block_updates(UpdateBatch {
6552 account_updates: vec![(Nibbles::from_raw(&[0x00, 0x01], false), vec![0xab, 0xcd])],
6553 storage_updates: vec![],
6554 blocks: vec![block1],
6555 receipts: vec![],
6556 code_updates: vec![],
6557 commit_depth: None,
6558 wait_for_flush: false,
6559 })
6560 .unwrap();
6561
6562 store.set_safe_commit_root(state_root_1).unwrap();
6565 let state_root_2 = H256::repeat_byte(0x22);
6566 let block2 = make_block(2, block1_hash, state_root_2);
6567 let block2_hash = block2.hash();
6568 store
6569 .store_block_updates(UpdateBatch {
6570 account_updates: vec![(Nibbles::from_raw(&[0x00, 0x02], false), vec![0xef, 0x11])],
6571 storage_updates: vec![],
6572 blocks: vec![block2],
6573 receipts: vec![],
6574 code_updates: vec![],
6575 commit_depth: None,
6576 wait_for_flush: false,
6577 })
6578 .unwrap();
6579
6580 let bytes = await_journal_entry(&backend, 1, Duration::from_secs(2))
6581 .expect("STATE_HISTORY entry for block 1 should appear after block 2 commits it");
6582 let entry = JournalEntry::decode(&bytes).unwrap();
6583 assert_eq!(entry.block_hash, block1_hash);
6584 assert_eq!(entry.parent_state_root, H256::zero());
6585 assert!(!entry.account_trie_diff.is_empty());
6586 let (path, prev) = &entry.account_trie_diff[0];
6587 assert_eq!(prev, &None, "first-time write means previous value is None");
6588 assert!(path.len() < 65);
6589
6590 store.set_safe_commit_root(state_root_2).unwrap();
6593 let state_root_3 = H256::repeat_byte(0x33);
6594 let block3 = make_block(3, block2_hash, state_root_3);
6595 store
6596 .store_block_updates(UpdateBatch {
6597 account_updates: vec![(Nibbles::from_raw(&[0x00, 0x03], false), vec![0x77])],
6598 storage_updates: vec![],
6599 blocks: vec![block3],
6600 receipts: vec![],
6601 code_updates: vec![],
6602 commit_depth: None,
6603 wait_for_flush: false,
6604 })
6605 .unwrap();
6606
6607 let bytes = await_journal_entry(&backend, 2, Duration::from_secs(2))
6608 .expect("STATE_HISTORY entry for block 2 should appear after block 3 commits it");
6609 let entry = JournalEntry::decode(&bytes).unwrap();
6610 assert_eq!(entry.block_hash, block2_hash);
6611 assert_eq!(entry.parent_state_root, state_root_1);
6612 assert!(!entry.account_trie_diff.is_empty());
6613 }
6614
6615 #[test]
6625 fn depth_gated_per_block_commit_journals_with_correct_pre_image() {
6626 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6627 let dir = tempfile::tempdir().unwrap();
6628 let store = Store::from_backend(
6629 backend.clone(),
6630 dir.path().to_path_buf(),
6631 1,
6632 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6633 )
6634 .unwrap();
6635
6636 let shared = Nibbles::from_raw(&[0x0a, 0x0b], false);
6638 let mut prev_hash = H256::zero();
6639 let mut hashes = vec![H256::zero()];
6640 let mut roots = vec![H256::zero()];
6641 for n in 1..=6u64 {
6642 let state_root = H256::repeat_byte(0xc0 | (n as u8));
6643 let block = make_block(n, prev_hash, state_root);
6644 prev_hash = block.hash();
6645 hashes.push(prev_hash);
6646 roots.push(state_root);
6647 store
6648 .store_block_updates(UpdateBatch {
6649 account_updates: vec![(shared.clone(), vec![0xb0 | (n as u8)])],
6650 storage_updates: vec![],
6651 blocks: vec![block],
6652 receipts: vec![],
6653 code_updates: vec![],
6654 commit_depth: Some(3),
6655 wait_for_flush: false,
6656 })
6657 .unwrap();
6658 }
6659
6660 let bytes = await_journal_entry(&backend, 1, Duration::from_secs(2))
6661 .expect("a depth-gated per-block commit must journal");
6662 let entry = JournalEntry::decode(&bytes).unwrap();
6663 assert_eq!(
6664 entry.block_hash, hashes[1],
6665 "entry 1 carries block 1's identity"
6666 );
6667 assert_eq!(entry.parent_state_root, H256::zero());
6668 assert_eq!(
6669 entry.account_trie_diff[0].1, None,
6670 "block 1 first-writes the key, so its pre-image is absence"
6671 );
6672
6673 let bytes =
6674 await_journal_entry(&backend, 2, Duration::from_secs(2)).expect("entry for block 2");
6675 let entry = JournalEntry::decode(&bytes).unwrap();
6676 assert_eq!(
6677 entry.block_hash, hashes[2],
6678 "entry 2 carries block 2's identity"
6679 );
6680 assert_eq!(
6681 entry.parent_state_root, roots[1],
6682 "entry 2's parent_state_root is block 1's state root"
6683 );
6684 assert_eq!(
6685 entry.account_trie_diff[0].1,
6686 Some(vec![0xb0 | 1u8]),
6687 "block 2's pre-image is the value block 1 wrote"
6688 );
6689 }
6690
6691 #[test]
6700 fn journal_skipped_in_batch_mode() {
6701 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6702 let dir = tempfile::tempdir().unwrap();
6703 let store = Store::from_backend(
6704 backend.clone(),
6705 dir.path().to_path_buf(),
6706 1,
6707 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6708 )
6709 .unwrap();
6710
6711 let mut prev_hash = H256::zero();
6712 for n in 1..=5u64 {
6713 let state_root = H256::repeat_byte(0xa0 | (n as u8));
6714 let block = make_block(n, prev_hash, state_root);
6715 prev_hash = block.hash();
6716 store
6717 .store_block_updates(UpdateBatch {
6718 account_updates: vec![(
6719 Nibbles::from_raw(&[n as u8], false),
6720 vec![0xde, 0xad, n as u8],
6721 )],
6722 storage_updates: vec![],
6723 blocks: vec![block],
6724 receipts: vec![],
6725 code_updates: vec![],
6726 commit_depth: Some(BATCH_COMMIT_THRESHOLD),
6727 wait_for_flush: true,
6728 })
6729 .unwrap();
6730 }
6731
6732 for n in 1..=5u64 {
6733 assert_no_journal_entry(&backend, n);
6734 }
6735 }
6736
6737 #[test]
6740 fn journal_storage_updates_appear_in_storage_diff() {
6741 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6742 let dir = tempfile::tempdir().unwrap();
6743 let store = Store::from_backend(
6744 backend.clone(),
6745 dir.path().to_path_buf(),
6746 1,
6747 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6748 )
6749 .unwrap();
6750
6751 let account_hash_a = H256::repeat_byte(0xa0);
6752 let account_hash_b = H256::repeat_byte(0xb0);
6753
6754 let state_root_1 = H256::repeat_byte(0x33);
6755 let block1 = make_block(1, H256::zero(), state_root_1);
6756 let block1_hash = block1.hash();
6757 store
6758 .store_block_updates(UpdateBatch {
6759 account_updates: vec![],
6760 storage_updates: vec![
6761 (
6762 account_hash_a,
6763 vec![(Nibbles::from_raw(&[0x05], false), vec![0x01])],
6764 ),
6765 (
6766 account_hash_b,
6767 vec![(Nibbles::from_raw(&[0x06], false), vec![0x02])],
6768 ),
6769 ],
6770 blocks: vec![block1],
6771 receipts: vec![],
6772 code_updates: vec![],
6773 commit_depth: None,
6774 wait_for_flush: false,
6775 })
6776 .unwrap();
6777
6778 store.set_safe_commit_root(state_root_1).unwrap();
6781 let state_root_2 = H256::repeat_byte(0x44);
6782 let block2 = make_block(2, block1_hash, state_root_2);
6783 store
6784 .store_block_updates(UpdateBatch {
6785 account_updates: vec![(Nibbles::from_raw(&[0xee], false), vec![0xff])],
6786 storage_updates: vec![],
6787 blocks: vec![block2],
6788 receipts: vec![],
6789 code_updates: vec![],
6790 commit_depth: None,
6791 wait_for_flush: false,
6792 })
6793 .unwrap();
6794
6795 let bytes = await_journal_entry(&backend, 1, Duration::from_secs(2))
6796 .expect("STATE_HISTORY entry for block 1");
6797 let entry = JournalEntry::decode(&bytes).unwrap();
6798 assert_eq!(entry.block_hash, block1_hash);
6799 assert_eq!(
6800 entry.storage_trie_diff.len(),
6801 2,
6802 "two distinct account hashes must produce two storage_trie entries"
6803 );
6804 for (_key, prev) in &entry.storage_trie_diff {
6805 assert_eq!(prev, &None, "first-time storage write has None pre-image");
6806 }
6807 assert!(entry.account_trie_diff.is_empty());
6808 }
6809
6810 fn seed_journal_entries(backend: &Arc<dyn StorageBackend>, block_numbers: &[BlockNumber]) {
6811 let mut tx = backend.begin_write().unwrap();
6812 for n in block_numbers {
6813 let entry = JournalEntry {
6814 block_hash: H256::repeat_byte(*n as u8),
6815 parent_state_root: H256::zero(),
6816 account_trie_diff: vec![(vec![*n as u8], None)],
6817 storage_trie_diff: vec![],
6818 account_flat_diff: vec![],
6819 storage_flat_diff: vec![],
6820 };
6821 tx.put(STATE_HISTORY, &n.to_be_bytes(), &entry.encode())
6822 .unwrap();
6823 }
6824 tx.commit().unwrap();
6825 }
6826
6827 fn journal_entry_exists(backend: &Arc<dyn StorageBackend>, block_number: BlockNumber) -> bool {
6828 backend
6829 .begin_read()
6830 .unwrap()
6831 .get(STATE_HISTORY, &block_number.to_be_bytes())
6832 .unwrap()
6833 .is_some()
6834 }
6835
6836 #[tokio::test]
6839 async fn finality_advance_prunes_journal_below_boundary() {
6840 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6841 let dir = tempfile::tempdir().unwrap();
6842 let store = Store::from_backend(
6843 backend.clone(),
6844 dir.path().to_path_buf(),
6845 1,
6846 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6847 )
6848 .unwrap();
6849
6850 seed_journal_entries(&backend, &(1..=10).collect::<Vec<_>>());
6851 for n in 1..=10 {
6852 assert!(journal_entry_exists(&backend, n), "seed entry {n} present");
6853 }
6854
6855 store
6856 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(5))
6857 .await
6858 .unwrap();
6859
6860 for n in 1..=5 {
6861 assert!(
6862 !journal_entry_exists(&backend, n),
6863 "entry {n} should have been pruned (<= finalized)"
6864 );
6865 }
6866 for n in 6..=10 {
6867 assert!(
6868 journal_entry_exists(&backend, n),
6869 "entry {n} should remain (> finalized)"
6870 );
6871 }
6872 }
6873
6874 #[tokio::test]
6876 async fn finality_no_op_does_not_prune() {
6877 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6878 let dir = tempfile::tempdir().unwrap();
6879 let store = Store::from_backend(
6880 backend.clone(),
6881 dir.path().to_path_buf(),
6882 1,
6883 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6884 )
6885 .unwrap();
6886
6887 store
6888 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(5))
6889 .await
6890 .unwrap();
6891
6892 seed_journal_entries(&backend, &(6..=10).collect::<Vec<_>>());
6893
6894 store
6896 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(5))
6897 .await
6898 .unwrap();
6899
6900 for n in 6..=10 {
6901 assert!(
6902 journal_entry_exists(&backend, n),
6903 "entry {n} should still exist after no-op finality update"
6904 );
6905 }
6906
6907 store
6909 .forkchoice_update_inner(vec![], 100, H256::zero(), None, None)
6910 .await
6911 .unwrap();
6912
6913 for n in 6..=10 {
6914 assert!(
6915 journal_entry_exists(&backend, n),
6916 "entry {n} should still exist when finalized is None"
6917 );
6918 }
6919 }
6920
6921 #[tokio::test]
6925 async fn malformed_finalized_returns_error_not_silent_zero() {
6926 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6927 let dir = tempfile::tempdir().unwrap();
6928 let store = Store::from_backend(
6929 backend.clone(),
6930 dir.path().to_path_buf(),
6931 1,
6932 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6933 )
6934 .unwrap();
6935
6936 let mut tx = backend.begin_write().unwrap();
6938 let finalized_key = chain_data_key(ChainDataIndex::FinalizedBlockNumber);
6939 tx.put(CHAIN_DATA, &finalized_key, &[0u8, 0, 0, 0]).unwrap();
6940 tx.commit().unwrap();
6941
6942 seed_journal_entries(&backend, &(1..=5).collect::<Vec<_>>());
6943
6944 let err = store
6945 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(3))
6946 .await
6947 .expect_err("malformed finalized must surface as an error");
6948 let msg = format!("{err}");
6949 assert!(
6950 msg.contains("FinalizedBlockNumber has unexpected length"),
6951 "expected length-mismatch error, got: {msg}"
6952 );
6953
6954 for n in 1..=5 {
6956 assert!(
6957 journal_entry_exists(&backend, n),
6958 "entry {n} must NOT be pruned when finalized read failed"
6959 );
6960 }
6961 }
6962
6963 #[tokio::test]
6970 async fn highest_state_history_block_number_finds_max() {
6971 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6972 let dir = tempfile::tempdir().unwrap();
6973 let store = Store::from_backend(
6974 backend.clone(),
6975 dir.path().to_path_buf(),
6976 1,
6977 DEFAULT_PERSIST_CHANNEL_CAPACITY,
6978 )
6979 .unwrap();
6980
6981 assert_eq!(store.highest_state_history_block_number().unwrap(), None);
6982
6983 seed_journal_entries(&backend, &[3, 7, 5, 11, 8]);
6984 assert_eq!(
6985 store.highest_state_history_block_number().unwrap(),
6986 Some(11),
6987 "max over present entries"
6988 );
6989 }
6990
6991 #[tokio::test]
6995 async fn journaled_block_hash_reads_back() {
6996 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
6997 let dir = tempfile::tempdir().unwrap();
6998 let store = Store::from_backend(
6999 backend.clone(),
7000 dir.path().to_path_buf(),
7001 1,
7002 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7003 )
7004 .unwrap();
7005
7006 seed_journal_entries(&backend, &[3, 11]);
7007 assert_eq!(
7008 store.get_state_history_block_hash(11).unwrap(),
7009 Some(H256::repeat_byte(11))
7010 );
7011 assert_eq!(store.get_state_history_block_hash(4).unwrap(), None);
7012 }
7013
7014 #[tokio::test]
7017 async fn advance_flushed_upto_only_moves_forward() {
7018 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7019 let dir = tempfile::tempdir().unwrap();
7020 let store = Store::from_backend(
7021 backend,
7022 dir.path().to_path_buf(),
7023 1,
7024 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7025 )
7026 .unwrap();
7027
7028 store.advance_flushed_upto(10).unwrap();
7029 assert_eq!(store.read_flushed_upto().unwrap(), 10);
7030 store.advance_flushed_upto(5).unwrap();
7031 assert_eq!(store.read_flushed_upto().unwrap(), 10);
7032 store.advance_flushed_upto(12).unwrap();
7033 assert_eq!(store.read_flushed_upto().unwrap(), 12);
7034 }
7035
7036 #[tokio::test]
7040 async fn lowest_state_history_block_number_finds_min() {
7041 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7042 let dir = tempfile::tempdir().unwrap();
7043 let store = Store::from_backend(
7044 backend.clone(),
7045 dir.path().to_path_buf(),
7046 1,
7047 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7048 )
7049 .unwrap();
7050
7051 assert_eq!(store.lowest_state_history_block_number().unwrap(), None);
7052
7053 seed_journal_entries(&backend, &[7, 3, 11, 5, 8]);
7054 assert_eq!(
7055 store.lowest_state_history_block_number().unwrap(),
7056 Some(3),
7057 "min over present entries"
7058 );
7059 }
7060
7061 #[tokio::test]
7065 async fn install_overlay_replaces_cache_with_fresh_one() {
7066 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7067 let dir = tempfile::tempdir().unwrap();
7068 let store = Store::from_backend(
7069 backend.clone(),
7070 dir.path().to_path_buf(),
7071 4,
7072 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7073 )
7074 .unwrap();
7075
7076 seed_journal_entries(&backend, &[3, 4]);
7077
7078 {
7080 let mut guard = store.trie_cache.write().unwrap();
7081 let mut updated = (**guard).clone();
7082 updated.put_batch(
7083 H256::zero(),
7084 H256::repeat_byte(0xff),
7085 42,
7086 H256::repeat_byte(0x42),
7087 vec![],
7088 );
7089 *guard = Arc::new(updated);
7090 }
7091 assert!(
7092 store
7093 .is_state_in_layer_cache(H256::repeat_byte(0xff))
7094 .unwrap()
7095 );
7096
7097 store
7098 .install_overlay_for_reorg(4, 3, |_| None)
7099 .expect("install must succeed");
7100
7101 assert!(
7102 !store
7103 .is_state_in_layer_cache(H256::repeat_byte(0xff))
7104 .unwrap(),
7105 "old cache layer must be gone after swap"
7106 );
7107 let cache = store.trie_cache.read().unwrap().clone();
7108 assert!(cache.overlay().is_some(), "overlay must be installed");
7109 assert_eq!(
7110 cache.commit_threshold(),
7111 4,
7112 "fresh cache must inherit threshold"
7113 );
7114 }
7115
7116 #[tokio::test]
7119 async fn install_overlay_failure_preserves_cache() {
7120 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7121 let dir = tempfile::tempdir().unwrap();
7122 let store = Store::from_backend(
7123 backend.clone(),
7124 dir.path().to_path_buf(),
7125 4,
7126 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7127 )
7128 .unwrap();
7129
7130 seed_journal_entries(&backend, &[5]);
7132 {
7133 let mut guard = store.trie_cache.write().unwrap();
7134 let mut updated = (**guard).clone();
7135 updated.put_batch(
7136 H256::zero(),
7137 H256::repeat_byte(0xaa),
7138 7,
7139 H256::repeat_byte(0x77),
7140 vec![],
7141 );
7142 *guard = Arc::new(updated);
7143 }
7144 assert!(
7145 store
7146 .is_state_in_layer_cache(H256::repeat_byte(0xaa))
7147 .unwrap()
7148 );
7149
7150 let err = store
7151 .install_overlay_for_reorg(5, 3, |_| None)
7152 .expect_err("missing entries 3 and 4 must abort");
7153 let msg = format!("{err}");
7154 assert!(
7155 msg.contains("overlay construction failed"),
7156 "error should explain construction failure: {msg}"
7157 );
7158
7159 assert!(
7161 store
7162 .is_state_in_layer_cache(H256::repeat_byte(0xaa))
7163 .unwrap(),
7164 "cache must survive a failed overlay install"
7165 );
7166 let cache = store.trie_cache.read().unwrap().clone();
7167 assert!(cache.overlay().is_none(), "no overlay should be installed");
7168 }
7169
7170 #[tokio::test]
7174 async fn abort_reorg_resets_cache_to_fresh() {
7175 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7176 let dir = tempfile::tempdir().unwrap();
7177 let store = Store::from_backend(
7178 backend.clone(),
7179 dir.path().to_path_buf(),
7180 4,
7181 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7182 )
7183 .unwrap();
7184
7185 seed_journal_entries(&backend, &[3, 4]);
7186 store.install_overlay_for_reorg(4, 3, |_| None).unwrap();
7187 {
7189 let mut guard = store.trie_cache.write().unwrap();
7190 let mut updated = (**guard).clone();
7191 updated.put_batch(
7192 H256::zero(),
7193 H256::repeat_byte(0xbb),
7194 4,
7195 H256::repeat_byte(0xb4),
7196 vec![],
7197 );
7198 *guard = Arc::new(updated);
7199 }
7200 assert!(
7201 store
7202 .is_state_in_layer_cache(H256::repeat_byte(0xbb))
7203 .unwrap()
7204 );
7205
7206 store.abort_reorg().unwrap();
7207
7208 assert!(
7209 !store
7210 .is_state_in_layer_cache(H256::repeat_byte(0xbb))
7211 .unwrap(),
7212 "partial new-chain layer must be discarded"
7213 );
7214 let cache = store.trie_cache.read().unwrap().clone();
7215 assert!(cache.overlay().is_none(), "overlay must be cleared");
7216 assert_eq!(
7217 cache.commit_threshold(),
7218 4,
7219 "fresh cache must inherit threshold"
7220 );
7221 }
7222
7223 #[tokio::test]
7226 async fn clear_reorg_overlay_removes_overlay_and_is_idempotent() {
7227 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7228 let dir = tempfile::tempdir().unwrap();
7229 let store = Store::from_backend(
7230 backend.clone(),
7231 dir.path().to_path_buf(),
7232 4,
7233 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7234 )
7235 .unwrap();
7236
7237 store.clear_reorg_overlay().unwrap();
7239 assert!(store.trie_cache.read().unwrap().overlay().is_none());
7240
7241 seed_journal_entries(&backend, &[7]);
7242 store.install_overlay_for_reorg(7, 7, |_| None).unwrap();
7243 assert!(store.trie_cache.read().unwrap().overlay().is_some());
7244
7245 store.clear_reorg_overlay().unwrap();
7246 assert!(store.trie_cache.read().unwrap().overlay().is_none());
7247
7248 store.clear_reorg_overlay().unwrap();
7250 assert!(store.trie_cache.read().unwrap().overlay().is_none());
7251 }
7252
7253 #[tokio::test]
7261 async fn deep_reorg_overlay_disables_flat_kv_fast_paths() {
7262 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7263
7264 {
7268 let mut tx = backend.begin_write().unwrap();
7269 tx.put(MISC_VALUES, "last_written".as_bytes(), &[0xff])
7270 .unwrap();
7271 tx.commit().unwrap();
7272 }
7273
7274 let dir = tempfile::tempdir().unwrap();
7275 let store = Store::from_backend(
7276 backend.clone(),
7277 dir.path().to_path_buf(),
7278 1,
7279 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7280 )
7281 .unwrap();
7282
7283 let address = Address::from_low_u64_be(0xbeef);
7288 let account_hash = hash_address_fixed(&address);
7289 let slot = H256::from_low_u64_be(1);
7290 let slot_hash = hash_key_fixed(&slot);
7291 let generator_account = AccountState {
7292 nonce: 7,
7293 balance: U256::from(999u64),
7294 storage_root: H256::zero(),
7295 code_hash: H256::zero(),
7296 };
7297 let generator_slot_value = U256::from(42u64);
7298 {
7299 let mut tx = backend.begin_write().unwrap();
7300 tx.put(
7301 ACCOUNT_FLATKEYVALUE,
7302 &Nibbles::from_bytes(account_hash.as_bytes()).into_vec(),
7303 &generator_account.encode_to_vec(),
7304 )
7305 .unwrap();
7306 tx.put(
7307 STORAGE_FLATKEYVALUE,
7308 &apply_prefix(Some(account_hash), Nibbles::from_bytes(&slot_hash)).into_vec(),
7309 &generator_slot_value.encode_to_vec(),
7310 )
7311 .unwrap();
7312 tx.commit().unwrap();
7313 }
7314
7315 let pivot_root = EMPTY_TRIE_HASH;
7320 {
7321 let mut tx = backend.begin_write().unwrap();
7322 for (n, parent_root) in [(10u64, pivot_root), (11, H256::repeat_byte(0x10))] {
7323 let entry = JournalEntry {
7324 block_hash: H256::repeat_byte(n as u8),
7325 parent_state_root: parent_root,
7326 account_trie_diff: vec![],
7327 storage_trie_diff: vec![],
7328 account_flat_diff: vec![],
7329 storage_flat_diff: vec![],
7330 };
7331 tx.put(STATE_HISTORY, &n.to_be_bytes(), &entry.encode())
7332 .unwrap();
7333 }
7334 tx.commit().unwrap();
7335 }
7336
7337 store
7338 .install_overlay_for_reorg(11, 10, |_| None)
7339 .expect("overlay install must succeed");
7340
7341 let accounts = store
7344 .get_account_states_batch_by_root(pivot_root, &[address])
7345 .unwrap();
7346 assert_eq!(
7347 accounts,
7348 vec![None],
7349 "batch account read must not serve the generator's stale flat-KV value"
7350 );
7351
7352 let slot_value = store
7353 .get_storage_at_root(pivot_root, address, slot)
7354 .unwrap();
7355 assert_eq!(
7356 slot_value, None,
7357 "storage read must not serve the generator's stale flat-KV value"
7358 );
7359
7360 let batched_slots = store
7361 .get_storage_values_batch_by_root(
7362 pivot_root,
7363 &[(hash_address_fixed(&address), EMPTY_TRIE_HASH, slot)],
7364 )
7365 .unwrap();
7366 assert_eq!(
7367 batched_slots,
7368 vec![None],
7369 "batch storage read must not serve the generator's stale flat-KV value"
7370 );
7371
7372 let state_trie = store.open_state_trie(pivot_root).unwrap();
7373 assert_eq!(
7374 state_trie.get(account_hash.as_bytes()).unwrap(),
7375 None,
7376 "trie read must not serve the generator's stale flat-KV value"
7377 );
7378
7379 let unserved_root = H256::repeat_byte(0x99);
7382 let accounts = store
7383 .get_account_states_batch_by_root(unserved_root, &[address])
7384 .unwrap();
7385 assert_eq!(
7386 accounts,
7387 vec![Some(generator_account)],
7388 "unserved roots must keep the flat-KV fast path"
7389 );
7390 let slot_value = store
7391 .get_storage_at_root(unserved_root, address, slot)
7392 .unwrap();
7393 assert_eq!(
7394 slot_value,
7395 Some(generator_slot_value),
7396 "unserved roots must keep the flat-KV fast path"
7397 );
7398
7399 let batched_slots = store
7400 .get_storage_values_batch_by_root(
7401 unserved_root,
7402 &[(hash_address_fixed(&address), EMPTY_TRIE_HASH, slot)],
7403 )
7404 .unwrap();
7405 assert_eq!(
7406 batched_slots,
7407 vec![Some(generator_slot_value)],
7408 "unserved roots must keep the flat-KV fast path"
7409 );
7410 }
7411
7412 #[tokio::test]
7420 async fn overlay_backed_commit_only_commits_bottom_layer_per_pass() {
7421 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7422 let dir = tempfile::tempdir().unwrap();
7423 let store = Store::from_backend(
7424 backend.clone(),
7425 dir.path().to_path_buf(),
7426 4,
7427 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7428 )
7429 .unwrap();
7430
7431 let pivot_root = H256::repeat_byte(0x09);
7434 {
7435 let mut tx = backend.begin_write().unwrap();
7436 for (n, parent_root) in [(10u64, pivot_root), (11, H256::repeat_byte(0x0a))] {
7437 let entry = JournalEntry {
7438 block_hash: H256::repeat_byte(n as u8),
7439 parent_state_root: parent_root,
7440 account_trie_diff: vec![(vec![0x00, n as u8], None)],
7441 storage_trie_diff: vec![],
7442 account_flat_diff: vec![],
7443 storage_flat_diff: vec![],
7444 };
7445 tx.put(STATE_HISTORY, &n.to_be_bytes(), &entry.encode())
7446 .unwrap();
7447 }
7448 tx.commit().unwrap();
7449 }
7450 store.install_overlay_for_reorg(11, 10, |_| None).unwrap();
7451
7452 let roots: Vec<H256> = (10..=12u8).map(H256::repeat_byte).collect();
7454 {
7455 let mut guard = store.trie_cache.write().unwrap();
7456 let mut updated = (**guard).clone();
7457 let mut parent = pivot_root;
7458 for (i, root) in roots.iter().enumerate() {
7459 let n = 10 + i as u64;
7460 updated.put_batch(
7461 parent,
7462 *root,
7463 n,
7464 H256::repeat_byte(n as u8),
7465 vec![(Nibbles::from_raw(&[0x01, n as u8], false), vec![n as u8])],
7466 );
7467 parent = *root;
7468 }
7469 *guard = Arc::new(updated);
7470 }
7471
7472 let trie = store.trie_cache.read().unwrap().clone();
7475 commit_to_disk(
7476 store.backend.as_ref(),
7477 &store.flatkeyvalue_control_tx,
7478 &store.trie_cache,
7479 &trie,
7480 roots[2],
7481 false,
7482 )
7483 .expect("bottom-layer commit must succeed (no debug_assert panic)");
7484
7485 assert!(
7487 !store.is_state_in_layer_cache(roots[0]).unwrap(),
7488 "bottom layer must be committed and pruned"
7489 );
7490 assert!(
7491 store.is_state_in_layer_cache(roots[1]).unwrap()
7492 && store.is_state_in_layer_cache(roots[2]).unwrap(),
7493 "upper layers must remain resident after the bottom-layer-only commit"
7494 );
7495 assert!(
7497 store.trie_cache.read().unwrap().overlay().is_none(),
7498 "reconciliation must clear the overlay"
7499 );
7500 assert!(
7502 journal_entry_exists(&backend, 10),
7503 "bottom layer T must be journaled"
7504 );
7505 assert!(!journal_entry_exists(&backend, 11));
7506 assert!(!journal_entry_exists(&backend, 12));
7507
7508 let entry_bytes = backend
7513 .begin_read()
7514 .unwrap()
7515 .get(STATE_HISTORY, &10u64.to_be_bytes())
7516 .unwrap()
7517 .expect("journal entry for block 10");
7518 let entry = JournalEntry::decode(&entry_bytes).unwrap();
7519 let diff_keys: Vec<&Vec<u8>> = entry.account_trie_diff.iter().map(|(k, _)| k).collect();
7520 assert!(
7521 diff_keys.contains(&&vec![0x00, 10]) && diff_keys.contains(&&vec![0x00, 11]),
7522 "bridge keys from the overlay must be folded into T's journal entry, got {diff_keys:?}"
7523 );
7524
7525 let trie = store.trie_cache.read().unwrap().clone();
7527 commit_to_disk(
7528 store.backend.as_ref(),
7529 &store.flatkeyvalue_control_tx,
7530 &store.trie_cache,
7531 &trie,
7532 roots[2],
7533 false,
7534 )
7535 .unwrap();
7536 assert!(!store.is_state_in_layer_cache(roots[1]).unwrap());
7537 assert!(!store.is_state_in_layer_cache(roots[2]).unwrap());
7538 assert!(journal_entry_exists(&backend, 11));
7539 assert!(journal_entry_exists(&backend, 12));
7540 }
7541
7542 #[tokio::test]
7547 async fn journal_pruning_pauses_during_reorg_and_catches_up_after() {
7548 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7549 let dir = tempfile::tempdir().unwrap();
7550 let store = Store::from_backend(
7551 backend.clone(),
7552 dir.path().to_path_buf(),
7553 1,
7554 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7555 )
7556 .unwrap();
7557
7558 seed_journal_entries(&backend, &(1..=5).collect::<Vec<_>>());
7559
7560 store.set_journal_pruning_paused(true);
7562 store
7563 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(3))
7564 .await
7565 .unwrap();
7566 for n in 1..=5 {
7567 assert!(
7568 journal_entry_exists(&backend, n),
7569 "entry {n} must survive pruning while the reorg pause is held"
7570 );
7571 }
7572
7573 store.set_journal_pruning_paused(false);
7575 store
7576 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(4))
7577 .await
7578 .unwrap();
7579 for n in 1..=4 {
7580 assert!(
7581 !journal_entry_exists(&backend, n),
7582 "entry {n} must be pruned once the pause is released"
7583 );
7584 }
7585 assert!(journal_entry_exists(&backend, 5));
7586 }
7587
7588 type RecordedRanges = Arc<Mutex<Vec<(BlockNumber, BlockNumber)>>>;
7590
7591 #[derive(Debug)]
7594 struct RangeDeleteRecorder {
7595 inner: Arc<dyn StorageBackend>,
7596 ranges: RecordedRanges,
7597 }
7598
7599 struct RecordingWriteBatch {
7600 inner: Box<dyn StorageWriteBatch + 'static>,
7601 ranges: RecordedRanges,
7602 }
7603
7604 fn block_number_key(key: &[u8]) -> BlockNumber {
7605 BlockNumber::from_be_bytes(key.try_into().expect("8-byte STATE_HISTORY key"))
7606 }
7607
7608 impl StorageBackend for RangeDeleteRecorder {
7609 fn clear_table(&self, table: &'static str) -> Result<(), StoreError> {
7610 self.inner.clear_table(table)
7611 }
7612 fn begin_read(&self) -> Result<Arc<dyn StorageReadView>, StoreError> {
7613 self.inner.begin_read()
7614 }
7615 fn begin_write(&self) -> Result<Box<dyn StorageWriteBatch + 'static>, StoreError> {
7616 Ok(Box::new(RecordingWriteBatch {
7617 inner: self.inner.begin_write()?,
7618 ranges: self.ranges.clone(),
7619 }))
7620 }
7621 fn begin_locked(
7622 &self,
7623 table_name: &'static str,
7624 ) -> Result<Box<dyn crate::api::StorageLockedView + 'static>, StoreError> {
7625 self.inner.begin_locked(table_name)
7626 }
7627 fn create_checkpoint(&self, path: &std::path::Path) -> Result<(), StoreError> {
7628 self.inner.create_checkpoint(path)
7629 }
7630 }
7631
7632 impl StorageWriteBatch for RecordingWriteBatch {
7633 fn put_batch(
7634 &mut self,
7635 table: &'static str,
7636 batch: Vec<(Vec<u8>, Vec<u8>)>,
7637 ) -> Result<(), StoreError> {
7638 self.inner.put_batch(table, batch)
7639 }
7640 fn delete(&mut self, table: &'static str, key: &[u8]) -> Result<(), StoreError> {
7641 self.inner.delete(table, key)
7642 }
7643 fn delete_range(
7644 &mut self,
7645 table: &'static str,
7646 start: &[u8],
7647 end: &[u8],
7648 ) -> Result<(), StoreError> {
7649 if table == STATE_HISTORY {
7650 self.ranges
7651 .lock()
7652 .unwrap()
7653 .push((block_number_key(start), block_number_key(end)));
7654 }
7655 self.inner.delete_range(table, start, end)
7656 }
7657 fn merge(
7658 &mut self,
7659 table: &'static str,
7660 key: &[u8],
7661 operand: &[u8],
7662 ) -> Result<(), StoreError> {
7663 self.inner.merge(table, key, operand)
7664 }
7665 fn commit(&mut self) -> Result<(), StoreError> {
7666 self.inner.commit()
7667 }
7668 }
7669
7670 fn recording_store() -> (
7671 Store,
7672 Arc<dyn StorageBackend>,
7673 RecordedRanges,
7674 tempfile::TempDir,
7675 ) {
7676 let ranges = Arc::new(Mutex::new(Vec::new()));
7677 let backend: Arc<dyn StorageBackend> = Arc::new(RangeDeleteRecorder {
7678 inner: Arc::new(InMemoryBackend::open().unwrap()),
7679 ranges: ranges.clone(),
7680 });
7681 let dir = tempfile::tempdir().unwrap();
7682 let store = Store::from_backend(
7683 backend.clone(),
7684 dir.path().to_path_buf(),
7685 1,
7686 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7687 )
7688 .unwrap();
7689 (store, backend, ranges, dir)
7690 }
7691
7692 #[tokio::test]
7697 async fn finality_pruning_writes_disjoint_ranges() {
7698 let (store, backend, ranges, _dir) = recording_store();
7699 seed_journal_entries(&backend, &[3, 7, 9, 12]);
7700
7701 for finalized in [5, 8, 10] {
7702 store
7703 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(finalized))
7704 .await
7705 .unwrap();
7706 }
7707
7708 assert_eq!(*ranges.lock().unwrap(), vec![(3, 6), (7, 9), (9, 11)]);
7709 assert!(journal_entry_exists(&backend, 12));
7710 }
7711
7712 #[tokio::test]
7715 async fn finality_advance_without_expired_entries_writes_no_range() {
7716 let (store, backend, ranges, _dir) = recording_store();
7717
7718 store
7720 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(5))
7721 .await
7722 .unwrap();
7723
7724 seed_journal_entries(&backend, &[20]);
7726 store
7727 .forkchoice_update_inner(vec![], 100, H256::zero(), None, Some(10))
7728 .await
7729 .unwrap();
7730
7731 assert!(ranges.lock().unwrap().is_empty());
7732 assert!(journal_entry_exists(&backend, 20));
7733 }
7734
7735 #[tokio::test]
7739 async fn finality_pruning_stays_disjoint_when_entries_land_below_finality() {
7740 let (store, backend, ranges, _dir) = recording_store();
7741 let lag = DB_COMMIT_THRESHOLD as u64;
7742 for head in 1..=(lag + 50) {
7743 if head > lag {
7744 seed_journal_entries(&backend, &[head - lag]);
7745 }
7746 store
7747 .forkchoice_update_inner(vec![], head, H256::zero(), None, Some(head))
7748 .await
7749 .unwrap();
7750 }
7751
7752 let ranges = ranges.lock().unwrap();
7753 assert_eq!(ranges.len(), 50);
7754 for pair in ranges.windows(2) {
7755 assert!(pair[0].1 <= pair[1].0, "overlapping ranges {pair:?}");
7756 }
7757 }
7758
7759 fn journal_store() -> (Store, Arc<dyn StorageBackend>, tempfile::TempDir) {
7760 let backend: Arc<dyn StorageBackend> = Arc::new(InMemoryBackend::open().unwrap());
7761 let dir = tempfile::tempdir().unwrap();
7762 let store = Store::from_backend(
7763 backend.clone(),
7764 dir.path().to_path_buf(),
7765 1,
7766 DEFAULT_PERSIST_CHANNEL_CAPACITY,
7767 )
7768 .unwrap();
7769 (store, backend, dir)
7770 }
7771
7772 fn set_head_and_finalized(
7775 backend: &Arc<dyn StorageBackend>,
7776 head: BlockNumber,
7777 finalized: BlockNumber,
7778 ) {
7779 let mut tx = backend.begin_write().unwrap();
7780 tx.put(
7781 CHAIN_DATA,
7782 &chain_data_key(ChainDataIndex::LatestBlockNumber),
7783 &head.to_le_bytes(),
7784 )
7785 .unwrap();
7786 tx.put(
7787 CHAIN_DATA,
7788 &chain_data_key(ChainDataIndex::FinalizedBlockNumber),
7789 &finalized.to_le_bytes(),
7790 )
7791 .unwrap();
7792 tx.commit().unwrap();
7793 }
7794
7795 fn store_block_rewriting(
7798 store: &Store,
7799 number: BlockNumber,
7800 parent_hash: H256,
7801 shared: &Nibbles,
7802 commit_depth: Option<usize>,
7803 ) -> (H256, H256) {
7804 let state_root = H256::repeat_byte(0xc0 | number as u8);
7805 let block = make_block(number, parent_hash, state_root);
7806 let hash = block.hash();
7807 store
7808 .store_block_updates(UpdateBatch {
7809 account_updates: vec![(shared.clone(), vec![number as u8])],
7810 storage_updates: vec![],
7811 blocks: vec![block],
7812 receipts: vec![],
7813 code_updates: vec![],
7814 commit_depth,
7815 wait_for_flush: false,
7816 })
7817 .unwrap();
7818 (hash, state_root)
7819 }
7820
7821 fn store_chain(
7824 store: &Store,
7825 count: BlockNumber,
7826 shared: &Nibbles,
7827 commit_depth: Option<usize>,
7828 ) {
7829 let mut parent_hash = H256::zero();
7830 for number in 1..=count {
7831 (parent_hash, _) =
7832 store_block_rewriting(store, number, parent_hash, shared, commit_depth);
7833 }
7834 }
7835
7836 fn journal_entry(backend: &Arc<dyn StorageBackend>, block_number: BlockNumber) -> JournalEntry {
7837 let bytes = backend
7838 .begin_read()
7839 .unwrap()
7840 .get(STATE_HISTORY, &block_number.to_be_bytes())
7841 .unwrap()
7842 .unwrap_or_else(|| panic!("journal entry for block {block_number}"));
7843 JournalEntry::decode(&bytes).unwrap()
7844 }
7845
7846 #[tokio::test]
7850 async fn final_layers_are_committed_without_journal_entries() {
7851 let (store, backend, _dir) = journal_store();
7852 set_head_and_finalized(&backend, 100, 3);
7853 let shared = Nibbles::from_raw(&[0x0a, 0x0b], false);
7854
7855 store_chain(&store, 6, &shared, Some(1));
7856 store.wait_for_persistence_idle().await.unwrap();
7857
7858 for n in 1..=3 {
7859 assert!(!journal_entry_exists(&backend, n), "block {n} is final");
7860 }
7861 for n in 4..=5 {
7862 assert!(journal_entry_exists(&backend, n), "block {n} is not final");
7863 }
7864 let entry = journal_entry(&backend, 4);
7865 assert_eq!(entry.account_trie_diff.len(), 1);
7866 assert_eq!(
7867 entry.account_trie_diff[0].1,
7868 Some(vec![3]),
7869 "block 4's pre-image is what skipped block 3 wrote"
7870 );
7871 }
7872
7873 #[tokio::test]
7877 async fn layer_journaled_after_skipped_ones_in_one_commit_sees_their_writes() {
7878 let (store, backend, _dir) = journal_store();
7879 set_head_and_finalized(&backend, 100, 2);
7880 let shared = Nibbles::from_raw(&[0x0a, 0x0b], false);
7881
7882 let mut parent_hash = H256::zero();
7884 let mut roots = vec![H256::zero()];
7885 for number in 1..=4 {
7886 let (hash, root) = store_block_rewriting(&store, number, parent_hash, &shared, None);
7887 parent_hash = hash;
7888 roots.push(root);
7889 }
7890 store.set_safe_commit_root(roots[3]).unwrap();
7893 store_block_rewriting(&store, 5, parent_hash, &shared, None);
7894 store.wait_for_persistence_idle().await.unwrap();
7895
7896 assert!(!journal_entry_exists(&backend, 1));
7897 assert!(!journal_entry_exists(&backend, 2));
7898 let entry = journal_entry(&backend, 3);
7899 assert_eq!(entry.account_trie_diff.len(), 1);
7900 assert_eq!(
7901 entry.account_trie_diff[0].1,
7902 Some(vec![2]),
7903 "block 3's pre-image is what skipped block 2 staged in the same batch"
7904 );
7905 }
7906
7907 #[tokio::test]
7911 async fn layers_above_the_head_are_journaled_even_when_final() {
7912 let (store, backend, _dir) = journal_store();
7913 set_head_and_finalized(&backend, 2, 100);
7914 let shared = Nibbles::from_raw(&[0x0a, 0x0b], false);
7915
7916 store_chain(&store, 6, &shared, Some(1));
7917 store.wait_for_persistence_idle().await.unwrap();
7918
7919 for n in 1..=2 {
7920 assert!(
7921 !journal_entry_exists(&backend, n),
7922 "block {n} is final and not above the head"
7923 );
7924 }
7925 for n in 3..=5 {
7926 assert!(
7927 journal_entry_exists(&backend, n),
7928 "block {n} is above the head"
7929 );
7930 }
7931 }
7932
7933 #[tokio::test]
7937 async fn final_layers_are_journaled_while_older_entries_survive() {
7938 let (store, backend, _dir) = journal_store();
7939 seed_journal_entries(&backend, &[0]);
7940 set_head_and_finalized(&backend, 100, 100);
7941 let shared = Nibbles::from_raw(&[0x0a, 0x0b], false);
7942
7943 store_chain(&store, 4, &shared, Some(1));
7944 store.wait_for_persistence_idle().await.unwrap();
7945
7946 for n in 1..=3 {
7947 assert!(
7948 journal_entry_exists(&backend, n),
7949 "block {n} follows a surviving entry"
7950 );
7951 }
7952 }
7953
7954 #[tokio::test]
7958 async fn chain_finalizing_its_head_writes_no_journal_and_no_range_deletes() {
7959 let (store, backend, ranges, _dir) = recording_store();
7960 let shared = Nibbles::from_raw(&[0x0a, 0x0b], false);
7961
7962 let mut parent_hash = H256::zero();
7963 for number in 1..=20 {
7964 let (hash, _) = store_block_rewriting(&store, number, parent_hash, &shared, Some(4));
7965 store
7966 .forkchoice_update_inner(vec![], number, hash, None, Some(number))
7967 .await
7968 .unwrap();
7969 parent_hash = hash;
7970 }
7971 store.wait_for_persistence_idle().await.unwrap();
7972
7973 for n in 1..=20 {
7974 assert!(
7975 !journal_entry_exists(&backend, n),
7976 "block {n} was final when committed"
7977 );
7978 }
7979 assert!(ranges.lock().unwrap().is_empty());
7980 }
7981}
7982
7983#[cfg(test)]
7984mod merge_tests {
7985 use super::*;
7986
7987 fn h256(b: u8) -> H256 {
7988 H256::from_low_u64_be(b as u64)
7989 }
7990
7991 fn op(bn: BlockNumber, bh: H256, idx: Index) -> Vec<u8> {
7992 encode_tx_location_operand(bn, bh, idx)
7993 }
7994
7995 fn decode(v: &[u8]) -> Vec<(BlockNumber, BlockHash, Index)> {
7996 <Vec<(BlockNumber, BlockHash, Index)>>::decode(v).unwrap()
7997 }
7998
7999 #[test]
8000 fn single_operand_on_empty_base() {
8001 let out = tx_locations_merge(None, vec![op(100, h256(0x10), 0)]).unwrap();
8002 assert_eq!(decode(&out), vec![(100, h256(0x10), 0)]);
8003 }
8004
8005 #[test]
8006 fn operand_appended_to_existing_base() {
8007 let base = vec![(100u64, h256(0x10), 0u64)].encode_to_vec();
8008 let out = tx_locations_merge(Some(&base), vec![op(101, h256(0x11), 5)]).unwrap();
8009 let mut got = decode(&out);
8010 got.sort();
8011 let mut want = vec![(100, h256(0x10), 0), (101, h256(0x11), 5)];
8012 want.sort();
8013 assert_eq!(got, want);
8014 }
8015
8016 #[test]
8017 fn multiple_operands_combined() {
8018 let out = tx_locations_merge(
8019 None,
8020 vec![
8021 op(100, h256(0x10), 0),
8022 op(100, h256(0x11), 1),
8023 op(101, h256(0x12), 2),
8024 ],
8025 )
8026 .unwrap();
8027 assert_eq!(decode(&out).len(), 3);
8028 }
8029
8030 #[test]
8031 fn same_block_hash_is_deduped() {
8032 let out =
8034 tx_locations_merge(None, vec![op(100, h256(0x10), 0), op(100, h256(0x10), 7)]).unwrap();
8035 assert_eq!(decode(&out), vec![(100, h256(0x10), 7)]);
8036 }
8037
8038 #[test]
8039 fn malformed_operand_aborts_merge() {
8040 let out = tx_locations_merge(None, vec![vec![0xff, 0xff], op(100, h256(0x10), 0)]);
8043 assert!(out.is_none(), "merge must abort on a malformed operand");
8044 }
8045
8046 #[test]
8047 fn malformed_base_value_aborts_merge() {
8048 let out = tx_locations_merge(Some(&[0xff, 0xff]), vec![op(100, h256(0x10), 0)]);
8049 assert!(out.is_none(), "merge must abort on a corrupt base value");
8050 }
8051
8052 #[test]
8059 fn partial_merge_result_is_a_valid_operand() {
8060 let partial =
8062 tx_locations_merge(None, vec![op(100, h256(0x10), 0), op(101, h256(0x11), 1)]).unwrap();
8063
8064 let base = vec![(99u64, h256(0x09), 9u64)].encode_to_vec();
8068 let out = tx_locations_merge(Some(&base), vec![partial]).unwrap();
8069
8070 let mut got = decode(&out);
8071 got.sort();
8072 let mut want = vec![
8073 (99, h256(0x09), 9),
8074 (100, h256(0x10), 0),
8075 (101, h256(0x11), 1),
8076 ];
8077 want.sort();
8078 assert_eq!(
8079 got, want,
8080 "no entries may be lost when re-merging a partial result"
8081 );
8082 }
8083
8084 #[test]
8087 fn operand_encoding_matches_value_encoding() {
8088 let operand = op(100, h256(0x10), 3);
8089 assert_eq!(decode(&operand), vec![(100, h256(0x10), 3)]);
8091 }
8092
8093 #[test]
8095 fn chained_partial_merges() {
8096 let p1 = tx_locations_merge(None, vec![op(1, h256(0x01), 0)]).unwrap();
8097 let p2 = tx_locations_merge(None, vec![p1, op(2, h256(0x02), 0)]).unwrap();
8098 let p3 = tx_locations_merge(None, vec![p2, op(3, h256(0x03), 0)]).unwrap();
8099 let out = tx_locations_merge(None, vec![p3]).unwrap();
8100 assert_eq!(decode(&out).len(), 3);
8101 }
8102}
8103
8104#[cfg(test)]
8105mod backfill_write_tests {
8106 use super::*;
8107 use ethrex_common::types::{EIP1559Transaction, TxType};
8108
8109 fn test_header(number: BlockNumber) -> BlockHeader {
8110 BlockHeader {
8111 number,
8112 ..Default::default()
8113 }
8114 }
8115
8116 fn test_tx(nonce: u64) -> Transaction {
8117 Transaction::EIP1559Transaction(EIP1559Transaction {
8118 nonce,
8119 ..Default::default()
8120 })
8121 }
8122
8123 fn test_receipt() -> Receipt {
8124 Receipt {
8125 tx_type: TxType::EIP1559,
8126 succeeded: true,
8127 cumulative_gas_used: 21_000,
8128 logs: vec![],
8129 payer: None,
8130 frame_receipts: None,
8131 }
8132 }
8133
8134 #[tokio::test]
8138 async fn writes_data_indexes_txs_and_lowers_frontier() {
8139 let store = Store::new("", EngineType::InMemory).expect("in-memory store");
8140 store.update_earliest_block_number(100).await.unwrap();
8141
8142 let header = test_header(99);
8143 let hash = header.hash();
8144 let tx = test_tx(7);
8145 let body = BlockBody {
8146 transactions: vec![tx.clone()],
8147 ommers: vec![],
8148 withdrawals: Some(vec![]),
8149 };
8150 store.add_block_headers(vec![header.clone()]).await.unwrap();
8154 store
8155 .forkchoice_update(vec![(99, hash)], 99, hash, None, None)
8156 .await
8157 .unwrap();
8158
8159 store
8160 .add_backfilled_blocks(
8161 vec![BackfilledBlock {
8162 header,
8163 body,
8164 receipts: vec![test_receipt()],
8165 index_transactions: true,
8166 }],
8167 99,
8168 )
8169 .await
8170 .unwrap();
8171
8172 assert_eq!(
8173 store.get_earliest_block_number().await.unwrap(),
8174 99,
8175 "frontier must be lowered atomically with the data write"
8176 );
8177 assert!(
8178 store.get_block_body_by_hash(hash).await.unwrap().is_some(),
8179 "body must be stored"
8180 );
8181 assert_eq!(
8182 store.get_receipts_for_block(&hash).await.unwrap().len(),
8183 1,
8184 "receipts must be stored"
8185 );
8186 assert_eq!(
8187 store
8188 .get_transaction_location(tx.hash(&NativeCrypto))
8189 .await
8190 .unwrap(),
8191 Some((99, hash, 0)),
8192 "indexed transaction must be locatable"
8193 );
8194 }
8195
8196 #[tokio::test]
8200 async fn skips_tx_index_when_disabled() {
8201 let store = Store::new("", EngineType::InMemory).expect("in-memory store");
8202 store.update_earliest_block_number(100).await.unwrap();
8203
8204 let header = test_header(99);
8205 let hash = header.hash();
8206 let tx = test_tx(7);
8207 let body = BlockBody {
8208 transactions: vec![tx.clone()],
8209 ommers: vec![],
8210 withdrawals: Some(vec![]),
8211 };
8212 store.add_block_headers(vec![header.clone()]).await.unwrap();
8213 store
8214 .forkchoice_update(vec![(99, hash)], 99, hash, None, None)
8215 .await
8216 .unwrap();
8217
8218 store
8219 .add_backfilled_blocks(
8220 vec![BackfilledBlock {
8221 header,
8222 body,
8223 receipts: vec![test_receipt()],
8224 index_transactions: false,
8225 }],
8226 99,
8227 )
8228 .await
8229 .unwrap();
8230
8231 assert!(
8232 store.get_block_body_by_hash(hash).await.unwrap().is_some(),
8233 "body must still be stored"
8234 );
8235 assert_eq!(
8236 store.get_receipts_for_block(&hash).await.unwrap().len(),
8237 1,
8238 "receipts must still be stored"
8239 );
8240 assert_eq!(
8241 store
8242 .get_transaction_location(tx.hash(&NativeCrypto))
8243 .await
8244 .unwrap(),
8245 None,
8246 "transaction index must be skipped when index_transactions is false"
8247 );
8248 }
8249
8250 #[tokio::test]
8254 async fn historical_getters_transition_from_null_to_present() {
8255 let store = Store::new("", EngineType::InMemory).expect("in-memory store");
8256 store.update_earliest_block_number(100).await.unwrap();
8257
8258 let header = test_header(99);
8259 let hash = header.hash();
8260 let tx = test_tx(7);
8261 let tx_hash = tx.hash(&NativeCrypto);
8262 let body = BlockBody {
8263 transactions: vec![tx],
8264 ommers: vec![],
8265 withdrawals: Some(vec![]),
8266 };
8267 store.add_block_headers(vec![header.clone()]).await.unwrap();
8269 store
8270 .forkchoice_update(vec![(99, hash)], 99, hash, None, None)
8271 .await
8272 .unwrap();
8273
8274 assert!(
8276 store.get_block_body(99).await.unwrap().is_none(),
8277 "body absent before backfill"
8278 );
8279 assert!(
8280 store
8281 .get_transaction_by_hash(tx_hash)
8282 .await
8283 .unwrap()
8284 .is_none(),
8285 "tx unlocatable before backfill"
8286 );
8287 assert!(
8288 store.get_receipt(99, 0).await.unwrap().is_none(),
8289 "receipt absent before backfill"
8290 );
8291
8292 store
8293 .add_backfilled_blocks(
8294 vec![BackfilledBlock {
8295 header,
8296 body,
8297 receipts: vec![test_receipt()],
8298 index_transactions: true,
8299 }],
8300 99,
8301 )
8302 .await
8303 .unwrap();
8304
8305 assert!(
8307 store.get_block_body(99).await.unwrap().is_some(),
8308 "body present after backfill"
8309 );
8310 assert!(
8311 store
8312 .get_transaction_by_hash(tx_hash)
8313 .await
8314 .unwrap()
8315 .is_some(),
8316 "tx locatable after backfill"
8317 );
8318 assert!(
8319 store.get_receipt(99, 0).await.unwrap().is_some(),
8320 "receipt present after backfill"
8321 );
8322 }
8323
8324 #[tokio::test]
8327 async fn resume_fills_contiguously_without_gaps() {
8328 let store = Store::new("", EngineType::InMemory).expect("in-memory store");
8329 store.update_earliest_block_number(100).await.unwrap();
8330
8331 let make_batch = |range: std::ops::Range<u64>| -> Vec<BackfilledBlock> {
8332 range
8333 .rev()
8334 .map(|n| BackfilledBlock {
8335 header: test_header(n),
8336 body: BlockBody {
8337 transactions: vec![],
8338 ommers: vec![],
8339 withdrawals: Some(vec![]),
8340 },
8341 receipts: vec![],
8342 index_transactions: false,
8343 })
8344 .collect()
8345 };
8346
8347 store
8349 .add_backfilled_blocks(make_batch(90..100), 90)
8350 .await
8351 .unwrap();
8352 assert_eq!(store.get_earliest_block_number().await.unwrap(), 90);
8353
8354 let resume_from = store.get_earliest_block_number().await.unwrap();
8356 assert_eq!(resume_from, 90, "resume reads the persisted frontier");
8357
8358 store
8360 .add_backfilled_blocks(make_batch(80..90), 80)
8361 .await
8362 .unwrap();
8363 assert_eq!(store.get_earliest_block_number().await.unwrap(), 80);
8364
8365 for n in 80..100 {
8367 assert!(
8368 store
8369 .get_block_body_by_hash(test_header(n).hash())
8370 .await
8371 .unwrap()
8372 .is_some(),
8373 "block {n} must have a body (no gap left by resume)"
8374 );
8375 }
8376 }
8377}
8378
8379#[cfg(test)]
8382#[cfg(feature = "rocksdb")]
8383mod schema_version_guard_tests {
8384 use super::*;
8385
8386 fn write_marker(dir: &Path, schema_version: u64) {
8387 let metadata = StoreMetadata::new(schema_version);
8388 std::fs::write(
8389 dir.join(STORE_METADATA_FILENAME),
8390 serde_json::to_string_pretty(&metadata).unwrap(),
8391 )
8392 .unwrap();
8393 }
8394
8395 fn read_marker(dir: &Path) -> u64 {
8396 read_store_schema_version(dir).unwrap().unwrap()
8397 }
8398
8399 #[test]
8400 fn a_database_ahead_of_the_binary_is_refused_and_left_untouched() {
8401 let dir = tempfile::tempdir().unwrap();
8407 let newer = STORE_SCHEMA_VERSION + 1;
8408 write_marker(dir.path(), newer);
8409
8410 let result =
8411 Store::new_with_config(dir.path(), EngineType::RocksDB, StoreConfig::default());
8412
8413 assert!(
8414 matches!(
8415 result,
8416 Err(StoreError::IncompatibleDBVersion { found, expected })
8417 if found == newer && expected == STORE_SCHEMA_VERSION
8418 ),
8419 "expected IncompatibleDBVersion, got {:?}",
8420 result.err()
8421 );
8422 assert_eq!(read_marker(dir.path()), newer);
8423 }
8424
8425 #[test]
8426 fn a_zero_schema_version_is_incompatible_not_a_failed_migration() {
8427 let dir = tempfile::tempdir().unwrap();
8430 write_marker(dir.path(), 0);
8431
8432 let result =
8433 Store::new_with_config(dir.path(), EngineType::RocksDB, StoreConfig::default());
8434
8435 assert!(
8436 matches!(
8437 result,
8438 Err(StoreError::IncompatibleDBVersion { found: 0, expected })
8439 if expected == STORE_SCHEMA_VERSION
8440 ),
8441 "expected IncompatibleDBVersion, got {:?}",
8442 result.err()
8443 );
8444 }
8445}
8446
8447#[cfg(test)]
8448mod datadir_tests {
8449 use super::*;
8450 use std::fs;
8451
8452 #[test]
8453 fn empty_dir_has_no_existing_db() {
8454 let dir = tempfile::tempdir().unwrap();
8455 assert!(!dir_contains_legacy_db(dir.path()).unwrap());
8456 }
8457
8458 #[test]
8459 fn dir_with_only_unrelated_files_has_no_existing_db() {
8460 let dir = tempfile::tempdir().unwrap();
8463 fs::write(dir.path().join("jwt.hex"), "0xdeadbeef").unwrap();
8464 fs::write(dir.path().join("LOG"), "noise").unwrap();
8465 assert!(!dir_contains_legacy_db(dir.path()).unwrap());
8466 }
8467
8468 #[test]
8469 fn dir_with_rocksdb_markers_has_existing_db() {
8470 let dir = tempfile::tempdir().unwrap();
8472 fs::write(dir.path().join("CURRENT"), "MANIFEST-000001\n").unwrap();
8473 assert!(dir_contains_legacy_db(dir.path()).unwrap());
8474
8475 let dir2 = tempfile::tempdir().unwrap();
8476 fs::write(dir2.path().join("MANIFEST-000007"), "x").unwrap();
8477 assert!(dir_contains_legacy_db(dir2.path()).unwrap());
8478 }
8479
8480 #[test]
8481 fn dir_with_marker_named_subdirectories_has_no_existing_db() {
8482 let dir = tempfile::tempdir().unwrap();
8485 fs::create_dir(dir.path().join("CURRENT")).unwrap();
8486 fs::create_dir(dir.path().join("MANIFEST-000001")).unwrap();
8487 assert!(!dir_contains_legacy_db(dir.path()).unwrap());
8488 }
8489}
8490
8491#[cfg(test)]
8492mod flatkeyvalue_completeness_tests {
8493 use super::*;
8494
8495 #[test]
8500 fn only_the_one_byte_ff_sentinel_counts_as_complete() {
8501 assert!(Store::flatkeyvalue_generation_complete(Some(&[0xff])));
8503
8504 assert!(!Store::flatkeyvalue_generation_complete(None)); assert!(!Store::flatkeyvalue_generation_complete(Some(&[]))); assert!(!Store::flatkeyvalue_generation_complete(Some(&[0u8; 64]))); assert!(!Store::flatkeyvalue_generation_complete(Some(&[0x0a; 64]))); assert!(!Store::flatkeyvalue_generation_complete(Some(&[0xff; 64])));
8512 assert!(!Store::flatkeyvalue_generation_complete(Some(&[
8513 0xff, 0xff
8514 ])));
8515 }
8516}
8517
8518#[cfg(test)]
8519mod cgroup_memory_limit_tests {
8520 use super::*;
8521 use std::fs;
8522
8523 fn cgroup_tree(root: &Path, relative: &str, file: &str, limits: &[(&str, &str)]) {
8526 fs::create_dir_all(root.join(relative)).expect("create cgroup tree");
8527 for (dir, contents) in limits {
8528 fs::write(root.join(dir).join(file), contents).expect("write limit");
8529 }
8530 }
8531
8532 #[test]
8536 fn ancestor_limit_applies() {
8537 let tmp = tempfile::tempdir().expect("tempdir");
8538 let root = tmp.path();
8539 cgroup_tree(
8540 root,
8541 "system.slice/ethrex.service",
8542 "memory.max",
8543 &[("", "max"), ("system.slice", "4294967296")],
8544 );
8545
8546 assert_eq!(
8547 cgroup_limit_up_to_root(root, "/system.slice/ethrex.service", "memory.max"),
8548 Some(4 * 1024 * 1024 * 1024)
8549 );
8550 }
8551
8552 #[test]
8554 fn smallest_limit_in_the_chain_wins() {
8555 let tmp = tempfile::tempdir().expect("tempdir");
8556 let root = tmp.path();
8557 cgroup_tree(
8558 root,
8559 "kubepods/pod123/container",
8560 "memory.max",
8561 &[
8562 ("kubepods", "8589934592"),
8563 ("kubepods/pod123", "2147483648"),
8564 ("kubepods/pod123/container", "4294967296"),
8565 ],
8566 );
8567
8568 assert_eq!(
8569 cgroup_limit_up_to_root(root, "kubepods/pod123/container", "memory.max"),
8570 Some(2 * 1024 * 1024 * 1024)
8571 );
8572 }
8573
8574 #[test]
8577 fn unlimited_chain_yields_no_limit() {
8578 let tmp = tempfile::tempdir().expect("tempdir");
8579 let root = tmp.path();
8580 cgroup_tree(root, "docker/abc", "memory.max", &[("", "max")]);
8581
8582 assert_eq!(
8583 cgroup_limit_up_to_root(root, "docker/abc", "memory.max"),
8584 None
8585 );
8586 assert_eq!(
8589 cgroup_limit_up_to_root(root, "not/here", "memory.max"),
8590 None
8591 );
8592 }
8593
8594 #[test]
8596 fn root_limit_applies_to_the_root_cgroup() {
8597 let tmp = tempfile::tempdir().expect("tempdir");
8598 let root = tmp.path();
8599 cgroup_tree(root, "", "memory.limit_in_bytes", &[("", "536870912")]);
8600
8601 assert_eq!(
8602 cgroup_limit_up_to_root(root, "/", "memory.limit_in_bytes"),
8603 Some(512 * 1024 * 1024)
8604 );
8605 }
8606}
8607
8608#[cfg(test)]
8609mod in_memory_receipts_tests {
8610 use super::*;
8611 use ethrex_common::types::TxType;
8612
8613 #[tokio::test]
8617 async fn in_memory_store_returns_every_receipt_of_a_block() {
8618 let store = Store::new("", EngineType::InMemory).expect("in-memory store");
8619 let receipt = |gas| Receipt::new(TxType::Legacy, true, gas, vec![]);
8620 let block = BlockHash::repeat_byte(1);
8621 let next_block = BlockHash::repeat_byte(2);
8622 store
8623 .add_receipts(block, vec![receipt(1), receipt(2), receipt(3)])
8624 .await
8625 .unwrap();
8626 store
8627 .add_receipts(next_block, vec![receipt(4)])
8628 .await
8629 .unwrap();
8630
8631 let gas = |receipts: Vec<Receipt>| {
8632 receipts
8633 .iter()
8634 .map(|receipt| receipt.cumulative_gas_used)
8635 .collect::<Vec<_>>()
8636 };
8637 assert_eq!(
8638 gas(store.get_receipts_for_block(&block).await.unwrap()),
8639 [1, 2, 3]
8640 );
8641 assert_eq!(
8642 gas(store
8643 .get_receipts_for_block_from_index(&block, 1, None)
8644 .await
8645 .unwrap()),
8646 [2, 3]
8647 );
8648 assert_eq!(
8649 gas(store
8650 .get_receipts_for_block_from_index(&block, 1, Some(1))
8651 .await
8652 .unwrap()),
8653 [2]
8654 );
8655 assert_eq!(
8656 gas(store.get_receipts_for_block(&next_block).await.unwrap()),
8657 [4]
8658 );
8659 }
8660}
8661
8662#[cfg(test)]
8663mod witness_pruning_tests {
8664 use super::*;
8665
8666 #[test]
8670 fn storing_a_witness_prunes_every_block_below_the_window() {
8671 let store = Store::new("", EngineType::InMemory).expect("in-memory store");
8672 let fork = BlockHash::repeat_byte;
8673 let store_at = |number: u64, hash: BlockHash| {
8674 store
8675 .store_witness(hash, number, ExecutionWitness::default())
8676 .unwrap()
8677 };
8678 let kept = |number: u64, hash: BlockHash| {
8679 store
8680 .get_witness_json_bytes(number, hash)
8681 .unwrap()
8682 .is_some()
8683 };
8684
8685 for (number, hash) in [(2, fork(1)), (3, fork(1)), (3, fork(2)), (4, fork(1))] {
8686 store_at(number, hash);
8687 }
8688 store_at(MAX_WITNESSES + 1, fork(1));
8691 assert!(kept(2, fork(1)) && kept(3, fork(1)) && kept(3, fork(2)) && kept(4, fork(1)));
8692
8693 store_at(MAX_WITNESSES + 3, fork(1));
8696 assert!(!kept(2, fork(1)));
8697 assert!(!kept(3, fork(1)));
8698 assert!(!kept(3, fork(2)));
8699 assert!(kept(4, fork(1)));
8700 assert!(kept(MAX_WITNESSES + 1, fork(1)));
8701 assert!(kept(MAX_WITNESSES + 3, fork(1)));
8702 }
8703}