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

forest/db/
memory.rs
