forest/db/
memory.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use super::{
5    BLOCK_BLOOM_LEN, EthBlockBloomStore, EthMappingsStore, SettingsStore, SettingsStoreExt,
6    decode_block_bloom, encode_block_bloom,
7};
8use crate::blocks::TipsetKey;
9use crate::db::PersistentStore;
10use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
11use crate::prelude::*;
12use crate::rpc::eth::types::EthHash;
13use crate::shim::clock::ChainEpoch;
14use crate::utils::db::car_stream::CarBlock;
15use crate::utils::multihash::prelude::*;
16use ahash::HashMap;
17use indexmap::IndexMap;
18use nunny::Vec as NonEmpty;
19use parking_lot::RwLock;
20use tokio::io::AsyncWriteExt as _;
21
22#[derive(Debug, Default)]
23pub struct MemoryDB {
24    blockchain_db: RwLock<HashMap<Cid, Vec<u8>>>,
25    blockchain_persistent_db: RwLock<HashMap<Cid, Vec<u8>>>,
26    settings_db: RwLock<HashMap<String, Vec<u8>>>,
27    pub eth_mappings_db: RwLock<HashMap<EthHash, Vec<u8>>>,
28    pub ts_lookup_db: RwLock<HashMap<ChainEpoch, TipsetKey>>,
29    pub eth_block_bloom_db: RwLock<HashMap<Cid, Vec<u8>>>,
30}
31
32impl MemoryDB {
33    pub fn blockstore_len(&self) -> usize {
34        self.blockchain_db.read().len() + self.blockchain_persistent_db.read().len()
35    }
36
37    pub fn blockstore_size_bytes(&self) -> usize {
38        self.blockchain_db
39            .read()
40            .iter()
41            .chain(self.blockchain_persistent_db.read().iter())
42            .map(|(k, v)| k.to_bytes().len() + v.len())
43            .sum()
44    }
45
46    pub async fn export_forest_car<W: tokio::io::AsyncWrite + Unpin>(
47        &self,
48        writer: &mut W,
49    ) -> anyhow::Result<()> {
50        let roots =
51            SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)?
52                .context("chain head is not tracked and cannot be exported")?
53                .into_cids();
54        self.export_forest_car_with_roots(roots, writer).await
55    }
56
57    pub async fn export_forest_car_with_roots<W: tokio::io::AsyncWrite + Unpin>(
58        &self,
59        roots: NonEmpty<Cid>,
60        writer: &mut W,
61    ) -> anyhow::Result<()> {
62        let blocks = {
63            let blockchain_db = self.blockchain_db.read();
64            let blockchain_persistent_db = self.blockchain_persistent_db.read();
65            blockchain_db
66                .iter()
67                .chain(blockchain_persistent_db.iter())
68                // Sort to make the result CAR deterministic
69                .sorted_by_key(|&(&cid, _)| cid)
70                .map(|(&cid, data)| {
71                    anyhow::Ok(CarBlock {
72                        cid,
73                        data: data.clone().into(),
74                    })
75                })
76                .collect_vec()
77        };
78        let frames =
79            crate::db::car::forest::Encoder::compress_stream_default(futures::stream::iter(blocks));
80        crate::db::car::forest::Encoder::write(writer, roots, frames).await
81    }
82}
83
84impl SettingsStore for MemoryDB {
85    fn read_bin(&self, key: &str) -> anyhow::Result<Option<Vec<u8>>> {
86        Ok(self.settings_db.read().get(key).cloned())
87    }
88
89    fn write_bin(&self, key: &str, value: &[u8]) -> anyhow::Result<()> {
90        self.settings_db
91            .write()
92            .insert(key.to_owned(), value.to_vec());
93        Ok(())
94    }
95
96    fn exists(&self, key: &str) -> anyhow::Result<bool> {
97        Ok(self.settings_db.read().contains_key(key))
98    }
99
100    fn setting_keys(&self) -> anyhow::Result<Vec<String>> {
101        Ok(self.settings_db.read().keys().cloned().collect_vec())
102    }
103}
104
105impl EthMappingsStore for MemoryDB {
106    fn read_bin(&self, key: &EthHash) -> anyhow::Result<Option<Vec<u8>>> {
107        Ok(self.eth_mappings_db.read().get(key).cloned())
108    }
109
110    fn write_bin(&self, key: &EthHash, value: &[u8]) -> anyhow::Result<()> {
111        self.eth_mappings_db
112            .write()
113            .insert(key.to_owned(), value.to_vec());
114        Ok(())
115    }
116
117    fn exists(&self, key: &EthHash) -> anyhow::Result<bool> {
118        Ok(self.eth_mappings_db.read().contains_key(key))
119    }
120
121    fn get_message_cids(&self) -> anyhow::Result<Vec<(Cid, u64)>> {
122        let cids = self
123            .eth_mappings_db
124            .read()
125            .values()
126            .filter_map(|value| fvm_ipld_encoding::from_slice::<(Cid, u64)>(value).ok())
127            .collect();
128
129        Ok(cids)
130    }
131
132    fn delete(&self, keys: Vec<EthHash>) -> anyhow::Result<()> {
133        let mut lock = self.eth_mappings_db.write();
134        for hash in keys.iter() {
135            lock.remove(hash);
136        }
137        Ok(())
138    }
139
140    fn tipset_key_by_epoch(&self, epoch: ChainEpoch) -> anyhow::Result<Option<TipsetKey>> {
141        Ok(self.ts_lookup_db.read().get(&epoch).cloned())
142    }
143
144    fn delete_tipset_key_at_epoch(&self, epoch: ChainEpoch) -> anyhow::Result<()> {
145        self.ts_lookup_db.write().remove(&epoch);
146        Ok(())
147    }
148
149    fn set_tipset_key_at_epoch_raw(
150        &self,
151        epoch: ChainEpoch,
152        tsk: &TipsetKey,
153    ) -> anyhow::Result<()> {
154        self.ts_lookup_db.write().insert(epoch, tsk.clone());
155        Ok(())
156    }
157}
158
159impl EthBlockBloomStore for MemoryDB {
160    fn read_bloom(&self, key: &Cid) -> anyhow::Result<Option<[u8; BLOCK_BLOOM_LEN]>> {
161        Ok(self
162            .eth_block_bloom_db
163            .read()
164            .get(key)
165            .and_then(|entry| decode_block_bloom(entry).map(|(_, bloom)| *bloom)))
166    }
167
168    fn write_bloom(
169        &self,
170        key: &Cid,
171        height: ChainEpoch,
172        bloom: &[u8; BLOCK_BLOOM_LEN],
173    ) -> anyhow::Result<()> {
174        self.eth_block_bloom_db
175            .write()
176            .insert(*key, encode_block_bloom(height, bloom));
177        Ok(())
178    }
179
180    fn delete_blooms_before_height(&self, height: ChainEpoch) -> anyhow::Result<()> {
181        self.eth_block_bloom_db
182            .write()
183            .retain(|_, entry| decode_block_bloom(entry).is_some_and(|(h, _)| h >= height));
184        Ok(())
185    }
186}
187
188impl Blockstore for MemoryDB {
189    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
190        Ok(self
191            .blockchain_db
192            .read()
193            .get(k)
194            .cloned()
195            .or_else(|| self.blockchain_persistent_db.read().get(k).cloned()))
196    }
197
198    fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
199        self.blockchain_db.write().insert(*k, block.to_vec());
200        Ok(())
201    }
202}
203
204impl PersistentStore for MemoryDB {
205    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
206        self.blockchain_persistent_db
207            .write()
208            .insert(*k, block.to_vec());
209        Ok(())
210    }
211}
212
213impl BitswapStoreRead for MemoryDB {
214    fn contains(&self, cid: &Cid) -> anyhow::Result<bool> {
215        Ok(self.blockchain_db.read().contains_key(cid))
216    }
217
218    fn get(&self, cid: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
219        Blockstore::get(self, cid)
220    }
221}
222
223impl BitswapStoreReadWrite for MemoryDB {
224    type Hashes = MultihashCode;
225
226    fn insert(&self, block: &crate::libp2p_bitswap::Block64<Self::Hashes>) -> anyhow::Result<()> {
227        self.put_keyed(block.cid(), block.data())
228    }
229}
230
231impl super::HeaviestTipsetKeyProvider for MemoryDB {
232    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
233        SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)
234    }
235
236    fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> {
237        SettingsStoreExt::write_obj(self, crate::db::setting_keys::HEAD_KEY, tsk)
238    }
239}
240
241#[derive(Debug, Default, derive_more::Deref)]
242/// A memory blockstore that preserves the insertion order
243pub struct IndexMapBlockstore {
244    inner: RwLock<IndexMap<Cid, Vec<u8>>>,
245}
246
247impl IndexMapBlockstore {
248    pub async fn export_forest_car<W: tokio::io::AsyncWrite + Unpin>(
249        &self,
250        roots: NonEmpty<Cid>,
251        writer: &mut W,
252    ) -> anyhow::Result<()> {
253        let blocks = {
254            let inner = self.inner.read();
255            let invalid_roots = roots
256                .iter()
257                .filter(|&c| !inner.contains_key(c))
258                .collect_vec();
259            anyhow::ensure!(
260                invalid_roots.is_empty(),
261                "All roots should present in the blockstore, invalid roots: {invalid_roots:?}"
262            );
263            inner
264                .iter()
265                .map(|(&cid, data)| {
266                    anyhow::Ok(CarBlock {
267                        cid,
268                        data: data.clone().into(),
269                    })
270                })
271                .collect_vec()
272        };
273        let frames =
274            crate::db::car::forest::Encoder::compress_stream_default(futures::stream::iter(blocks));
275        crate::db::car::forest::Encoder::write(&mut *writer, roots, frames).await?;
276        writer.flush().await?;
277        Ok(())
278    }
279}
280
281impl Blockstore for IndexMapBlockstore {
282    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
283        Ok(self.read().get(k).cloned())
284    }
285
286    fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
287        self.write().insert(*k, block.to_vec());
288        Ok(())
289    }
290}
291
292#[cfg(test)]
293mod tests {
294    use super::*;
295    use crate::{
296        db::{car::ForestCar, setting_keys::HEAD_KEY},
297        utils::cid::CidCborExt as _,
298    };
299    use fil_actors_shared::fvm_ipld_hamt::Hamt;
300    use fvm_ipld_encoding::DAG_CBOR;
301    use multihash_codetable::Code::Blake2b256;
302    use nunny::vec as nonempty;
303
304    #[tokio::test]
305    async fn test_export_forest_car() {
306        let db = MemoryDB::default();
307        let record1 = b"non-persistent";
308        let key1 = Cid::new_v1(DAG_CBOR, Blake2b256.digest(record1.as_slice()));
309        db.put_keyed(&key1, record1.as_slice()).unwrap();
310
311        let record2 = b"persistent";
312        let key2 = Cid::new_v1(DAG_CBOR, Blake2b256.digest(record2.as_slice()));
313        db.put_keyed_persistent(&key2, record2.as_slice()).unwrap();
314
315        let mut car_db_bytes = vec![];
316        assert!(
317            db.export_forest_car(&mut car_db_bytes)
318                .await
319                .unwrap_err()
320                .to_string()
321                .contains("chain head is not tracked and cannot be exported")
322        );
323
324        db.write_obj(HEAD_KEY, &TipsetKey::from(nonempty![key1]))
325            .unwrap();
326
327        car_db_bytes.clear();
328        db.export_forest_car(&mut car_db_bytes).await.unwrap();
329
330        let car = ForestCar::new(car_db_bytes).unwrap();
331        assert_eq!(car.head_tipset_key(), &nonempty![key1]);
332        assert!(car.has(&key1).unwrap());
333        assert!(car.has(&key2).unwrap());
334    }
335
336    #[test]
337    fn block_bloom_encode_decode() {
338        let bloom = [0xab; 256];
339        let entry = encode_block_bloom(42, &bloom);
340        let (height, decoded) = decode_block_bloom(&entry).unwrap();
341        assert_eq!(height, 42);
342        assert_eq!(decoded, &bloom);
343        assert!(decode_block_bloom(&[0, 1, 2]).is_none());
344    }
345
346    #[tokio::test]
347    async fn test_index_map_blockstore() {
348        const BIT_WIDTH: u32 = 5;
349
350        let db = IndexMapBlockstore::default();
351        // similate tipset lookup table
352        let mut hamt: Hamt<_, TipsetKey, ChainEpoch> = Hamt::new_with_bit_width(&db, BIT_WIDTH);
353        let checkpoints = [
354            (
355                0,
356                TipsetKey::from(nunny::vec![Cid::from_cbor_blake2b256(&"1").unwrap()]),
357            ),
358            (
359                5,
360                TipsetKey::from(nunny::vec![Cid::from_cbor_blake2b256(&"5").unwrap()]),
361            ),
362            (
363                10,
364                TipsetKey::from(nunny::vec![Cid::from_cbor_blake2b256(&"10").unwrap()]),
365            ),
366        ];
367        for (epoch, tsk) in checkpoints.iter().cloned() {
368            hamt.set(epoch, tsk).unwrap();
369        }
370        let hamt_root = hamt.flush().unwrap();
371        assert!(db.has(&hamt_root).unwrap(), "hamt root should present");
372
373        // export with invalid roots should fail
374        let mut car = vec![];
375        db.export_forest_car(
376            nunny::vec![Cid::from_cbor_blake2b256(&"invalid").unwrap()],
377            &mut car,
378        )
379        .await
380        .unwrap_err();
381
382        // export with hamt root should succeed
383        let mut car_bytes = vec![];
384        let car_roots = nunny::vec![hamt_root];
385        db.export_forest_car(car_roots.clone(), &mut car_bytes)
386            .await
387            .unwrap();
388
389        let car: ForestCar<Vec<u8>> = ForestCar::new(car_bytes).unwrap();
390        let hamt_from_car: Hamt<_, TipsetKey, ChainEpoch> =
391            Hamt::load_with_bit_width(&hamt_root, &car, BIT_WIDTH).unwrap();
392
393        let checkpoints_from_memdb_hamt = {
394            let mut v = vec![];
395            hamt.for_each_cacheless(|epoch, tsk| {
396                v.push((*epoch, tsk.clone()));
397                anyhow::Ok(())
398            })
399            .unwrap();
400            v
401        };
402        let checkpoints_from_car_hamt = {
403            let mut v = vec![];
404            hamt_from_car
405                .for_each_cacheless(|epoch, tsk| {
406                    v.push((*epoch, tsk.clone()));
407                    anyhow::Ok(())
408                })
409                .unwrap();
410            v
411        };
412        // Cannot compare the 2 hamts as they use different DB types
413        assert_eq!(checkpoints_from_memdb_hamt, checkpoints_from_car_hamt);
414        // The hamt iteration order is different from the insertion order
415        assert_eq!(
416            HashMap::from_iter(checkpoints),
417            HashMap::from_iter(checkpoints_from_car_hamt)
418        );
419    }
420}
forest/db/memory.rs

forest/db/
memory.rs
