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.blockchain_db.read().get(k).cloned().or(self
191            .blockchain_persistent_db
192            .read()
193            .get(k)
194            .cloned()))
195    }
196
197    fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
198        self.blockchain_db.write().insert(*k, block.to_vec());
199        Ok(())
200    }
201}
202
203impl PersistentStore for MemoryDB {
204    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
205        self.blockchain_persistent_db
206            .write()
207            .insert(*k, block.to_vec());
208        Ok(())
209    }
210}
211
212impl BitswapStoreRead for MemoryDB {
213    fn contains(&self, cid: &Cid) -> anyhow::Result<bool> {
214        Ok(self.blockchain_db.read().contains_key(cid))
215    }
216
217    fn get(&self, cid: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
218        Blockstore::get(self, cid)
219    }
220}
221
222impl BitswapStoreReadWrite for MemoryDB {
223    type Hashes = MultihashCode;
224
225    fn insert(&self, block: &crate::libp2p_bitswap::Block64<Self::Hashes>) -> anyhow::Result<()> {
226        self.put_keyed(block.cid(), block.data())
227    }
228}
229
230impl super::HeaviestTipsetKeyProvider for MemoryDB {
231    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
232        SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)
233    }
234
235    fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> {
236        SettingsStoreExt::write_obj(self, crate::db::setting_keys::HEAD_KEY, tsk)
237    }
238}
239
240#[derive(Debug, Default, derive_more::Deref)]
241/// A memory blockstore that preserves the insertion order
242pub struct IndexMapBlockstore {
243    inner: RwLock<IndexMap<Cid, Vec<u8>>>,
244}
245
246impl IndexMapBlockstore {
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(&mut *writer, roots, frames).await?;
275        writer.flush().await?;
276        Ok(())
277    }
278}
279
280impl Blockstore for IndexMapBlockstore {
281    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
282        Ok(self.read().get(k).cloned())
283    }
284
285    fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
286        self.write().insert(*k, block.to_vec());
287        Ok(())
288    }
289}
290
291#[cfg(test)]
292mod tests {
293    use super::*;
294    use crate::{
295        db::{car::ForestCar, setting_keys::HEAD_KEY},
296        utils::cid::CidCborExt as _,
297    };
298    use fil_actors_shared::fvm_ipld_hamt::Hamt;
299    use fvm_ipld_encoding::DAG_CBOR;
300    use multihash_codetable::Code::Blake2b256;
301    use nunny::vec as nonempty;
302
303    #[tokio::test]
304    async fn test_export_forest_car() {
305        let db = MemoryDB::default();
306        let record1 = b"non-persistent";
307        let key1 = Cid::new_v1(DAG_CBOR, Blake2b256.digest(record1.as_slice()));
308        db.put_keyed(&key1, record1.as_slice()).unwrap();
309
310        let record2 = b"persistent";
311        let key2 = Cid::new_v1(DAG_CBOR, Blake2b256.digest(record2.as_slice()));
312        db.put_keyed_persistent(&key2, record2.as_slice()).unwrap();
313
314        let mut car_db_bytes = vec![];
315        assert!(
316            db.export_forest_car(&mut car_db_bytes)
317                .await
318                .unwrap_err()
319                .to_string()
320                .contains("chain head is not tracked and cannot be exported")
321        );
322
323        db.write_obj(HEAD_KEY, &TipsetKey::from(nonempty![key1]))
324            .unwrap();
325
326        car_db_bytes.clear();
327        db.export_forest_car(&mut car_db_bytes).await.unwrap();
328
329        let car = ForestCar::new(car_db_bytes).unwrap();
330        assert_eq!(car.head_tipset_key(), &nonempty![key1]);
331        assert!(car.has(&key1).unwrap());
332        assert!(car.has(&key2).unwrap());
333    }
334
335    #[test]
336    fn block_bloom_encode_decode() {
337        let bloom = [0xab; 256];
338        let entry = encode_block_bloom(42, &bloom);
339        let (height, decoded) = decode_block_bloom(&entry).unwrap();
340        assert_eq!(height, 42);
341        assert_eq!(decoded, &bloom);
342        assert!(decode_block_bloom(&[0, 1, 2]).is_none());
343    }
344
345    #[tokio::test]
346    async fn test_index_map_blockstore() {
347        const BIT_WIDTH: u32 = 5;
348
349        let db = IndexMapBlockstore::default();
350        // similate tipset lookup table
351        let mut hamt: Hamt<_, TipsetKey, ChainEpoch> = Hamt::new_with_bit_width(&db, BIT_WIDTH);
352        let checkpoints = [
353            (
354                0,
355                TipsetKey::from(nunny::vec![Cid::from_cbor_blake2b256(&"1").unwrap()]),
356            ),
357            (
358                5,
359                TipsetKey::from(nunny::vec![Cid::from_cbor_blake2b256(&"5").unwrap()]),
360            ),
361            (
362                10,
363                TipsetKey::from(nunny::vec![Cid::from_cbor_blake2b256(&"10").unwrap()]),
364            ),
365        ];
366        for (epoch, tsk) in checkpoints.iter().cloned() {
367            hamt.set(epoch, tsk).unwrap();
368        }
369        let hamt_root = hamt.flush().unwrap();
370        assert!(db.has(&hamt_root).unwrap(), "hamt root should present");
371
372        // export with invalid roots should fail
373        let mut car = vec![];
374        db.export_forest_car(
375            nunny::vec![Cid::from_cbor_blake2b256(&"invalid").unwrap()],
376            &mut car,
377        )
378        .await
379        .unwrap_err();
380
381        // export with hamt root should succeed
382        let mut car_bytes = vec![];
383        let car_roots = nunny::vec![hamt_root];
384        db.export_forest_car(car_roots.clone(), &mut car_bytes)
385            .await
386            .unwrap();
387
388        let car: ForestCar<Vec<u8>> = ForestCar::new(car_bytes).unwrap();
389        let hamt_from_car: Hamt<_, TipsetKey, ChainEpoch> =
390            Hamt::load_with_bit_width(&hamt_root, &car, BIT_WIDTH).unwrap();
391
392        let checkpoints_from_memdb_hamt = {
393            let mut v = vec![];
394            hamt.for_each_cacheless(|epoch, tsk| {
395                v.push((*epoch, tsk.clone()));
396                anyhow::Ok(())
397            })
398            .unwrap();
399            v
400        };
401        let checkpoints_from_car_hamt = {
402            let mut v = vec![];
403            hamt_from_car
404                .for_each_cacheless(|epoch, tsk| {
405                    v.push((*epoch, tsk.clone()));
406                    anyhow::Ok(())
407                })
408                .unwrap();
409            v
410        };
411        // Cannot compare the 2 hamts as they use different DB types
412        assert_eq!(checkpoints_from_memdb_hamt, checkpoints_from_car_hamt);
413        // The hamt iteration order is different from the insertion order
414        assert_eq!(
415            HashMap::from_iter(checkpoints),
416            HashMap::from_iter(checkpoints_from_car_hamt)
417        );
418    }
419}
forest/db/memory.rs

forest/db/
memory.rs
