ethrex_storage/
block_data_buffer.rs

1use std::collections::VecDeque;
2use std::sync::Arc;
3
4use ethrex_common::H256;
5use ethrex_common::seeded_hash::SeededHashMap;
6use ethrex_common::types::{Block, BlockBody, BlockHash, BlockHeader, BlockNumber, Code, Receipt};
7use ethrex_crypto::NativeCrypto;
8use rustc_hash::{FxHashMap, FxHashSet};
9
10/// Block data held in memory after `newPayload` returns, before the background
11/// flusher writes it to RocksDB. One per block hash (holds reorg siblings).
12#[derive(Debug, Clone)]
13pub struct BufferedBlock {
14    pub header: BlockHeader,
15    pub body: BlockBody,
16    pub number: BlockNumber,
17    pub receipts: Vec<Receipt>,
18}
19
20/// In-memory overlay for not-yet-flushed block data. Consulted before disk by
21/// every read of headers/bodies/numbers/receipts/codes/tx-locations.
22///
23/// Held behind `Arc<RwLock<Arc<BlockDataBuffer>>>` in `Store`. Readers clone
24/// the inner `Arc` under a brief read lock and then work lock-free; the single
25/// writer mutates a clone and RCU-swaps the `Arc` in. Both critical sections
26/// are O(1) pointer operations.
27#[derive(Debug, Clone)]
28pub struct BlockDataBuffer {
29    by_hash: FxHashMap<BlockHash, Arc<BufferedBlock>>,
30    by_number: FxHashMap<BlockNumber, Vec<BlockHash>>,
31    tx_index: FxHashMap<H256, Vec<(BlockNumber, BlockHash, u64)>>,
32    /// Content-addressed code introduced by buffered blocks, with the block
33    /// number that introduced it (for eviction). Seeded: a deployer can grind its code hash
34    /// into a chosen bucket (see `ethrex_common::seeded_hash`).
35    codes: SeededHashMap<H256, (Code, BlockNumber)>,
36    /// Hashes not yet known to be on disk, oldest first.
37    unflushed: VecDeque<BlockHash>,
38    /// Highest block number whose data is durably on disk.
39    flushed_upto: BlockNumber,
40}
41
42impl Default for BlockDataBuffer {
43    fn default() -> Self {
44        Self::new()
45    }
46}
47
48impl BlockDataBuffer {
49    pub fn new() -> Self {
50        Self {
51            by_hash: FxHashMap::default(),
52            by_number: FxHashMap::default(),
53            tx_index: FxHashMap::default(),
54            codes: SeededHashMap::default(),
55            unflushed: VecDeque::new(),
56            flushed_upto: 0,
57        }
58    }
59
60    /// Takes ownership of `block` and moves its header/body into the buffer (the
61    /// caller hands ownership over the flush channel and drops it right after, so
62    /// cloning would be pure waste).
63    pub fn insert(&mut self, block: Block, receipts: Vec<Receipt>, codes: Vec<(H256, Code)>) {
64        let hash = block.hash();
65        let number = block.header.number;
66        if self.by_hash.contains_key(&hash) {
67            return; // duplicate guard
68        }
69        for (index, tx) in block.body.transactions.iter().enumerate() {
70            self.tx_index
71                .entry(tx.hash(&NativeCrypto))
72                .or_default()
73                .push((number, hash, index as u64));
74        }
75        for (code_hash, code) in codes {
76            self.codes.entry(code_hash).or_insert((code, number));
77        }
78        self.by_number.entry(number).or_default().push(hash);
79        let Block { header, body } = block;
80        self.by_hash.insert(
81            hash,
82            Arc::new(BufferedBlock {
83                header,
84                body,
85                number,
86                receipts,
87            }),
88        );
89        self.unflushed.push_back(hash);
90    }
91
92    pub fn get_header(&self, hash: &BlockHash) -> Option<BlockHeader> {
93        self.by_hash.get(hash).map(|b| b.header.clone())
94    }
95
96    pub fn get_body(&self, hash: &BlockHash) -> Option<BlockBody> {
97        self.by_hash.get(hash).map(|b| b.body.clone())
98    }
99
100    pub fn get_number(&self, hash: &BlockHash) -> Option<BlockNumber> {
101        self.by_hash.get(hash).map(|b| b.number)
102    }
103
104    pub fn get_receipt(&self, hash: &BlockHash, index: u64) -> Option<Receipt> {
105        self.by_hash
106            .get(hash)
107            .and_then(|b| b.receipts.get(index as usize).cloned())
108    }
109
110    pub fn get_receipts(&self, hash: &BlockHash) -> Option<Vec<Receipt>> {
111        self.by_hash.get(hash).map(|b| b.receipts.clone())
112    }
113
114    pub fn get_code(&self, code_hash: &H256) -> Option<Code> {
115        self.codes.get(code_hash).map(|(c, _)| c.clone())
116    }
117
118    pub fn get_tx_locations(&self, tx_hash: &H256) -> Vec<(BlockNumber, BlockHash, u64)> {
119        self.tx_index.get(tx_hash).cloned().unwrap_or_default()
120    }
121
122    pub fn flushed_upto(&self) -> BlockNumber {
123        self.flushed_upto
124    }
125
126    pub fn set_flushed_upto(&mut self, n: BlockNumber) {
127        self.flushed_upto = self.flushed_upto.max(n);
128    }
129
130    /// All unflushed blocks, ascending by number (deterministic flush order).
131    pub fn flushable(&self) -> Vec<Arc<BufferedBlock>> {
132        let mut out: Vec<Arc<BufferedBlock>> = self
133            .unflushed
134            .iter()
135            .filter_map(|h| self.by_hash.get(h).cloned())
136            .collect();
137        out.sort_by_key(|b| b.number);
138        out
139    }
140
141    /// Codes introduced by the given block hashes (for the flush write tx).
142    pub fn codes_for(&self, hashes: &[BlockHash]) -> Vec<(H256, Code)> {
143        let numbers: FxHashSet<BlockNumber> =
144            hashes.iter().filter_map(|h| self.get_number(h)).collect();
145        self.codes
146            .iter()
147            .filter(|(_, (_, n))| numbers.contains(n))
148            .map(|(h, (c, _))| (*h, c.clone()))
149            .collect()
150    }
151
152    /// Drop all buffered data for blocks with number <= `new_flushed_upto` and
153    /// advance the durable marker. Called only after the disk write committed.
154    pub fn evict_flushed(&mut self, new_flushed_upto: BlockNumber) {
155        self.set_flushed_upto(new_flushed_upto);
156        let drop_hashes: Vec<BlockHash> = self
157            .by_hash
158            .iter()
159            .filter(|(_, b)| b.number <= new_flushed_upto)
160            .map(|(h, _)| *h)
161            .collect();
162        for h in &drop_hashes {
163            if let Some(b) = self.by_hash.remove(h) {
164                if let Some(v) = self.by_number.get_mut(&b.number) {
165                    v.retain(|x| x != h);
166                    if v.is_empty() {
167                        self.by_number.remove(&b.number);
168                    }
169                }
170                for tx in b.body.transactions.iter() {
171                    if let Some(v) = self.tx_index.get_mut(&tx.hash(&NativeCrypto)) {
172                        v.retain(|(_, bh, _)| bh != h);
173                        if v.is_empty() {
174                            self.tx_index.remove(&tx.hash(&NativeCrypto));
175                        }
176                    }
177                }
178            }
179        }
180        self.unflushed.retain(|h| self.by_hash.contains_key(h));
181        self.codes.retain(|_, (_, n)| *n > new_flushed_upto);
182    }
183}
ethrex_storage/block_data_buffer.rs

ethrex_storage/
block_data_buffer.rs
