forest/db/
parity_db.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4mod gc;
5pub use gc::*;
6
7use super::{
8    BLOCK_BLOOM_LEN, EthBlockBloomStore, EthMappingsStore, PersistentStore, SettingsStore,
9    decode_block_bloom, encode_block_bloom,
10};
11use crate::blocks::TipsetKey;
12use crate::db::{DBStatistics, parity_db_config::ParityDbConfig};
13use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
14use crate::prelude::*;
15use crate::rpc::eth::types::EthHash;
16use crate::shim::clock::ChainEpoch;
17use crate::utils::{broadcast::has_subscribers, multihash::prelude::*};
18use bytes::Bytes;
19use fvm_ipld_encoding::DAG_CBOR;
20use parity_db::{CompressionType, Db, Operation, Options};
21use parking_lot::RwLock;
22use std::path::PathBuf;
23use strum::{Display, EnumIter, FromRepr, IntoEnumIterator};
24use tracing::warn;
25
26/// This is specific to Forest's `ParityDb` usage.
27/// It is used to determine which column to use for a given entry type.
28#[derive(Copy, Clone, Debug, Display, PartialEq, FromRepr, EnumIter)]
29#[repr(u8)]
30pub enum DbColumn {
31    /// Column for storing IPLD data with `Blake2b256` hash and `DAG_CBOR` codec.
32    /// Most entries in the `blockstore` will be stored in this column.
33    GraphDagCborBlake2b256,
34    /// Column for storing other IPLD data (different codec or hash function).
35    /// It allows for key retrieval at the cost of degraded performance. Given that
36    /// there will be a small number of entries in this column, the performance
37    /// degradation is negligible.
38    GraphFull,
39    /// Column for storing Forest-specific settings.
40    Settings,
41    /// Column for storing Ethereum mappings.
42    EthMappings,
43    /// Column for storing IPLD data that has to be ignored by the garbage collector.
44    /// Anything stored in this column can be considered permanent, unless manually
45    /// deleted.
46    PersistentGraph,
47    /// Column for storing per-tipset Ethereum block logs blooms.
48    EthBlockBloom,
49}
50
51impl DbColumn {
52    fn create_column_options(compression: CompressionType) -> Vec<parity_db::ColumnOptions> {
53        DbColumn::iter()
54            .map(|col| {
55                match col {
56                    DbColumn::GraphDagCborBlake2b256 | DbColumn::PersistentGraph => {
57                        parity_db::ColumnOptions {
58                            preimage: true,
59                            compression,
60                            ..Default::default()
61                        }
62                    }
63                    DbColumn::GraphFull => parity_db::ColumnOptions {
64                        preimage: true,
65                        // This is needed for key retrieval.
66                        btree_index: true,
67                        compression,
68                        ..Default::default()
69                    },
70                    DbColumn::Settings => parity_db::ColumnOptions {
71                        // explicitly disable preimage for settings column
72                        // othewise we are not able to overwrite entries
73                        preimage: false,
74                        // This is needed for key retrieval.
75                        btree_index: true,
76                        compression,
77                        ..Default::default()
78                    },
79                    DbColumn::EthMappings => parity_db::ColumnOptions {
80                        preimage: false,
81                        btree_index: false,
82                        compression,
83                        ..Default::default()
84                    },
85                    DbColumn::EthBlockBloom => parity_db::ColumnOptions {
86                        preimage: false,
87                        btree_index: true,
88                        compression,
89                        ..Default::default()
90                    },
91                }
92            })
93            .collect()
94    }
95}
96
97type WriteOpsBroadcastTxSender = tokio::sync::broadcast::Sender<Vec<(Cid, Bytes)>>;
98
99pub struct ParityDb {
100    pub db: parity_db::Db,
101    statistics_enabled: bool,
102    // This is needed to maintain backwards-compatibility for pre-persistent-column migrations.
103    disable_persistent_fallback: bool,
104    write_ops_broadcast_tx: RwLock<Option<WriteOpsBroadcastTxSender>>,
105}
106
107impl ParityDb {
108    pub fn to_options(path: impl Into<PathBuf>, config: &ParityDbConfig) -> Options {
109        Options {
110            path: path.into(),
111            sync_wal: true,
112            sync_data: true,
113            stats: config.enable_statistics,
114            salt: None,
115            columns: DbColumn::create_column_options(CompressionType::Lz4),
116            compression_threshold: [(0, 128)].into_iter().collect(),
117        }
118    }
119
120    pub fn open(path: impl Into<PathBuf>, config: &ParityDbConfig) -> anyhow::Result<Self> {
121        let opts = Self::to_options(path.into(), config);
122        Self::open_with_options(&opts)
123    }
124
125    pub fn open_with_options(options: &Options) -> anyhow::Result<Self> {
126        Ok(Self {
127            db: Db::open_or_create(options)?,
128            statistics_enabled: options.stats,
129            disable_persistent_fallback: false,
130            write_ops_broadcast_tx: RwLock::new(None),
131        })
132    }
133
134    /// Returns an appropriate column variant based on the information
135    /// in the Cid.
136    fn choose_column(cid: &Cid) -> DbColumn {
137        match cid.codec() {
138            DAG_CBOR if cid.hash().code() == u64::from(MultihashCode::Blake2b256) => {
139                DbColumn::GraphDagCborBlake2b256
140            }
141            _ => DbColumn::GraphFull,
142        }
143    }
144
145    fn read_from_column<K>(&self, key: K, column: DbColumn) -> anyhow::Result<Option<Vec<u8>>>
146    where
147        K: AsRef<[u8]>,
148    {
149        self.db
150            .get(column as u8, key.as_ref())
151            .with_context(|| format!("error from column {column}"))
152    }
153
154    fn write_to_column<K, V>(&self, key: K, value: V, column: DbColumn) -> anyhow::Result<()>
155    where
156        K: AsRef<[u8]>,
157        V: AsRef<[u8]>,
158    {
159        let tx = [(column as u8, key.as_ref(), Some(value.as_ref().to_vec()))];
160        self.db
161            .commit(tx)
162            .with_context(|| format!("error writing to column {column}"))
163    }
164}
165
166impl SettingsStore for ParityDb {
167    fn read_bin(&self, key: &str) -> anyhow::Result<Option<Vec<u8>>> {
168        self.read_from_column(key.as_bytes(), DbColumn::Settings)
169    }
170
171    fn write_bin(&self, key: &str, value: &[u8]) -> anyhow::Result<()> {
172        self.write_to_column(key.as_bytes(), value, DbColumn::Settings)
173    }
174
175    fn exists(&self, key: &str) -> anyhow::Result<bool> {
176        self.db
177            .get_size(DbColumn::Settings as u8, key.as_bytes())
178            .map(|size| size.is_some())
179            .context("error checking if key exists")
180    }
181
182    fn setting_keys(&self) -> anyhow::Result<Vec<String>> {
183        let mut iter = self.db.iter(DbColumn::Settings as u8)?;
184        let mut keys = vec![];
185        while let Some((key, _)) = iter.next()? {
186            keys.push(String::from_utf8(key)?);
187        }
188        Ok(keys)
189    }
190}
191
192impl super::HeaviestTipsetKeyProvider for ParityDb {
193    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
194        super::SettingsStoreExt::read_obj::<TipsetKey>(self, super::setting_keys::HEAD_KEY)
195    }
196
197    fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> {
198        super::SettingsStoreExt::write_obj(self, super::setting_keys::HEAD_KEY, tsk)
199    }
200}
201
202impl EthMappingsStore for ParityDb {
203    fn read_bin(&self, key: &EthHash) -> anyhow::Result<Option<Vec<u8>>> {
204        self.read_from_column(key.0.as_bytes(), DbColumn::EthMappings)
205    }
206
207    fn write_bin(&self, key: &EthHash, value: &[u8]) -> anyhow::Result<()> {
208        self.write_to_column(key.0.as_bytes(), value, DbColumn::EthMappings)
209    }
210
211    fn exists(&self, key: &EthHash) -> anyhow::Result<bool> {
212        self.db
213            .get_size(DbColumn::EthMappings as u8, key.0.as_bytes())
214            .map(|size| size.is_some())
215            .context("error checking if key exists")
216    }
217
218    fn get_message_cids(&self) -> anyhow::Result<Vec<(Cid, u64)>> {
219        let mut cids = Vec::new();
220
221        self.db
222            .iter_column_while(DbColumn::EthMappings as u8, |val| {
223                if let Ok(value) = fvm_ipld_encoding::from_slice::<(Cid, u64)>(&val.value) {
224                    cids.push(value);
225                }
226                true
227            })?;
228
229        Ok(cids)
230    }
231
232    fn delete(&self, keys: Vec<EthHash>) -> anyhow::Result<()> {
233        Ok(self.db.commit_changes(keys.into_iter().map(|key| {
234            let bytes = key.0.as_bytes().to_vec();
235            (DbColumn::EthMappings as u8, Operation::Dereference(bytes))
236        }))?)
237    }
238
239    fn tipset_key_by_epoch(&self, epoch: ChainEpoch) -> anyhow::Result<Option<TipsetKey>> {
240        let key = epoch.to_le_bytes();
241        if let Some(bytes) = self.read_from_column(key, DbColumn::EthMappings)? {
242            Ok(Some(fvm_ipld_encoding::from_slice(&bytes)?))
243        } else {
244            Ok(None)
245        }
246    }
247
248    fn delete_tipset_key_at_epoch(&self, epoch: ChainEpoch) -> anyhow::Result<()> {
249        Ok(self.db.commit_changes(std::iter::once(epoch).map(|epoch| {
250            (
251                DbColumn::EthMappings as u8,
252                Operation::Dereference(epoch.to_le_bytes().to_vec()),
253            )
254        }))?)
255    }
256
257    fn set_tipset_key_at_epoch_raw(
258        &self,
259        epoch: ChainEpoch,
260        tsk: &TipsetKey,
261    ) -> anyhow::Result<()> {
262        let key = epoch.to_le_bytes();
263        let bytes = fvm_ipld_encoding::to_vec(tsk)?;
264        self.write_to_column(key, bytes, DbColumn::EthMappings)
265    }
266}
267
268impl EthBlockBloomStore for ParityDb {
269    fn read_bloom(&self, key: &Cid) -> anyhow::Result<Option<[u8; BLOCK_BLOOM_LEN]>> {
270        Ok(self
271            .read_from_column(key.to_bytes(), DbColumn::EthBlockBloom)?
272            .and_then(|entry| decode_block_bloom(&entry).map(|(_, bloom)| *bloom)))
273    }
274
275    fn write_bloom(
276        &self,
277        key: &Cid,
278        height: ChainEpoch,
279        bloom: &[u8; BLOCK_BLOOM_LEN],
280    ) -> anyhow::Result<()> {
281        self.write_to_column(
282            key.to_bytes(),
283            encode_block_bloom(height, bloom),
284            DbColumn::EthBlockBloom,
285        )
286    }
287
288    fn delete_blooms_before_height(&self, height: ChainEpoch) -> anyhow::Result<()> {
289        let mut stale = Vec::new();
290        let mut iter = self.db.iter(DbColumn::EthBlockBloom as u8)?;
291        while let Some((key, entry)) = iter.next()? {
292            // Drop rows below the cutoff and any that fail to decode (corrupt/garbage).
293            if decode_block_bloom(&entry).is_none_or(|(h, _)| h < height) {
294                stale.push(key);
295            }
296        }
297        Ok(self.db.commit_changes(
298            stale
299                .into_iter()
300                .map(|key| (DbColumn::EthBlockBloom as u8, Operation::Dereference(key))),
301        )?)
302    }
303}
304
305impl Blockstore for ParityDb {
306    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
307        let column = Self::choose_column(k);
308        let res = self.read_from_column(k.to_bytes(), column)?;
309        if res.is_some() {
310            return Ok(res);
311        }
312        self.get_persistent(k)
313    }
314
315    fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
316        let column = Self::choose_column(k);
317        // We can put the data directly into the database without any encoding.
318        self.write_to_column(k.to_bytes(), block, column)?;
319        match &*self.write_ops_broadcast_tx.read() {
320            Some(tx) if has_subscribers(tx) => {
321                let _ = tx.send(vec![(*k, Bytes::copy_from_slice(block))]);
322            }
323            _ => {}
324        }
325
326        Ok(())
327    }
328
329    fn put_many_keyed<D, I>(&self, blocks: I) -> anyhow::Result<()>
330    where
331        Self: Sized,
332        D: AsRef<[u8]>,
333        I: IntoIterator<Item = (Cid, D)>,
334    {
335        let tx_opt = &*self.write_ops_broadcast_tx.read();
336        let has_subscribers = tx_opt.as_ref().map(has_subscribers).unwrap_or_default();
337        let mut values_for_subscriber = vec![];
338        let values = blocks.into_iter().map(|(k, v)| {
339            let column = Self::choose_column(&k);
340            let v = v.as_ref().to_vec();
341            if has_subscribers {
342                values_for_subscriber.push((k, Bytes::copy_from_slice(&v)));
343            }
344            (column, k.to_bytes(), v)
345        });
346        let tx = values
347            .into_iter()
348            .map(|(col, k, v)| (col as u8, Operation::Set(k, v)));
349        self.db.commit_changes(tx).context("error bulk writing")?;
350        if let Some(tx) = tx_opt {
351            let _ = tx.send(values_for_subscriber);
352        }
353        Ok(())
354    }
355}
356
357impl PersistentStore for ParityDb {
358    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
359        self.write_to_column(k.to_bytes(), block, DbColumn::PersistentGraph)
360    }
361}
362
363impl BitswapStoreRead for ParityDb {
364    fn contains(&self, cid: &Cid) -> anyhow::Result<bool> {
365        // We need to check both columns because we don't know which one
366        // the data is in. The order is important because most data will
367        // be in the [`DbColumn::GraphDagCborBlake2b256`] column and so
368        // it directly affects performance. If this assumption ever changes
369        // then this code should be modified accordingly.
370        for column in [DbColumn::GraphDagCborBlake2b256, DbColumn::GraphFull] {
371            if self
372                .db
373                .get_size(column as u8, &cid.to_bytes())
374                .context("error checking if key exists")?
375                .is_some()
376            {
377                return Ok(true);
378            }
379        }
380        Ok(false)
381    }
382
383    fn get(&self, cid: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
384        Blockstore::get(self, cid)
385    }
386}
387
388impl BitswapStoreReadWrite for ParityDb {
389    type Hashes = MultihashCode;
390
391    fn insert(&self, block: &crate::libp2p_bitswap::Block64<Self::Hashes>) -> anyhow::Result<()> {
392        self.put_keyed(block.cid(), block.data())
393    }
394}
395
396impl DBStatistics for ParityDb {
397    fn get_statistics(&self) -> Option<String> {
398        if !self.statistics_enabled {
399            return None;
400        }
401
402        let mut buf = Vec::new();
403        if let Err(err) = self.db.write_stats_text(&mut buf, None) {
404            warn!("Unable to write database statistics: {err}");
405            return None;
406        }
407
408        match String::from_utf8(buf) {
409            Ok(stats) => Some(stats),
410            Err(e) => {
411                warn!("Malformed statistics: {e}");
412                None
413            }
414        }
415    }
416}
417
418type Op = (u8, Operation<Vec<u8>, Vec<u8>>);
419
420impl ParityDb {
421    /// Removes a record.
422    ///
423    /// # Arguments
424    /// * `key` - record identifier
425    #[allow(dead_code)]
426    pub fn dereference_operation(key: &Cid) -> Op {
427        let column = Self::choose_column(key);
428        (column as u8, Operation::Dereference(key.to_bytes()))
429    }
430
431    /// Updates/inserts a record.
432    ///
433    /// # Arguments
434    /// * `column` - column identifier
435    /// * `key` - record identifier
436    /// * `value` - record contents
437    pub fn set_operation(column: u8, key: Vec<u8>, value: Vec<u8>) -> Op {
438        (column, Operation::Set(key, value))
439    }
440
441    // Get data from persistent graph column.
442    fn get_persistent(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
443        if self.disable_persistent_fallback {
444            return Ok(None);
445        }
446        self.read_from_column(k.to_bytes(), DbColumn::PersistentGraph)
447    }
448}
449
450impl super::BlockstoreWriteOpsSubscribable for ParityDb {
451    fn subscribe_write_ops(
452        &self,
453    ) -> anyhow::Result<tokio::sync::broadcast::Receiver<Vec<(Cid, bytes::Bytes)>>> {
454        // Use double checks to avoid concurrent first-time subscribers to create duplicate channels
455        if let Some(tx) = self.write_ops_broadcast_tx.read().as_ref() {
456            return Ok(tx.subscribe());
457        }
458        let mut tx_lock = self.write_ops_broadcast_tx.write();
459        if let Some(tx) = tx_lock.as_ref() {
460            return Ok(tx.subscribe());
461        }
462        let (tx, rx) = tokio::sync::broadcast::channel(65536);
463        *tx_lock = Some(tx);
464        Ok(rx)
465    }
466
467    fn unsubscribe_write_ops(&self) {
468        self.write_ops_broadcast_tx.write().take();
469    }
470}
471
472#[cfg(test)]
473mod test {
474    use super::*;
475    use crate::db::{BlockstoreWriteOpsSubscribable, tests::db_utils::parity::TempParityDB};
476    use fvm_ipld_encoding::IPLD_RAW;
477    use nom::AsBytes;
478    use std::ops::Deref;
479
480    #[test]
481    fn write_read_different_columns_test() {
482        let db = TempParityDB::new();
483        let data = [
484            b"h'nglui mglw'nafh".to_vec(),
485            b"Cthulhu".to_vec(),
486            b"R'lyeh wgah'nagl fhtagn!!".to_vec(),
487        ];
488        let cids = [
489            Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&data[0])),
490            Cid::new_v1(DAG_CBOR, MultihashCode::Sha2_256.digest(&data[1])),
491            Cid::new_v1(IPLD_RAW, MultihashCode::Blake2b256.digest(&data[1])),
492        ];
493
494        let cases = [
495            (DbColumn::GraphDagCborBlake2b256, cids[0], &data[0]),
496            (DbColumn::GraphFull, cids[1], &data[1]),
497            (DbColumn::GraphFull, cids[2], &data[2]),
498        ];
499
500        for (_, cid, data) in cases {
501            db.put_keyed(&cid, data).unwrap();
502        }
503
504        for (column, cid, data) in cases {
505            let actual = db
506                .read_from_column(cid.to_bytes(), column)
507                .unwrap()
508                .expect("data not found");
509            assert_eq!(data, actual.as_bytes());
510
511            // assert that the data is NOT in the other column
512            let other_column = match column {
513                DbColumn::GraphDagCborBlake2b256 => DbColumn::GraphFull,
514                DbColumn::GraphFull => DbColumn::GraphDagCborBlake2b256,
515                DbColumn::Settings => panic!("invalid column for IPLD data"),
516                DbColumn::EthMappings => panic!("invalid column for IPLD data"),
517                DbColumn::EthBlockBloom => panic!("invalid column for IPLD data"),
518                DbColumn::PersistentGraph => panic!("invalid column for GC enabled IPLD data"),
519            };
520            let actual = db.read_from_column(cid.to_bytes(), other_column).unwrap();
521            assert!(actual.is_none());
522
523            // Blockstore API usage should be transparent
524            let actual = fvm_ipld_blockstore::Blockstore::get(db.as_ref(), &cid)
525                .unwrap()
526                .expect("data not found");
527            assert_eq!(data, actual.as_slice());
528        }
529
530        // Check non-IPLD column as well
531        db.write_to_column(b"dagon", b"bloop", DbColumn::Settings)
532            .unwrap();
533        let actual = db
534            .read_from_column(b"dagon", DbColumn::Settings)
535            .unwrap()
536            .expect("data not found");
537        assert_eq!(b"bloop", actual.as_bytes());
538    }
539
540    #[test]
541    fn choose_column_test() {
542        let data = [0u8; 32];
543        let cases = [
544            (
545                Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&data)),
546                DbColumn::GraphDagCborBlake2b256,
547            ),
548            (
549                Cid::new_v1(
550                    fvm_ipld_encoding::CBOR,
551                    MultihashCode::Blake2b256.digest(&data),
552                ),
553                DbColumn::GraphFull,
554            ),
555            (
556                Cid::new_v1(DAG_CBOR, MultihashCode::Sha2_256.digest(&data)),
557                DbColumn::GraphFull,
558            ),
559        ];
560
561        for (cid, expected) in cases {
562            let actual = ParityDb::choose_column(&cid);
563            assert_eq!(expected, actual);
564        }
565    }
566
567    #[test]
568    fn persistent_tests() {
569        let db = TempParityDB::new();
570        let data = [
571            b"h'nglui mglw'nafh".to_vec(),
572            b"Cthulhu".to_vec(),
573            b"R'lyeh wgah'nagl fhtagn!!".to_vec(),
574        ];
575
576        let persistent_data = data
577            .clone()
578            .into_iter()
579            .map(|mut entry| {
580                entry.push(255);
581                entry
582            })
583            .collect_vec();
584
585        let cids = [
586            Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&data[0])),
587            Cid::new_v1(DAG_CBOR, MultihashCode::Sha2_256.digest(&data[1])),
588            Cid::new_v1(IPLD_RAW, MultihashCode::Blake2b256.digest(&data[1])),
589        ];
590
591        for idx in 0..3 {
592            let cid = &cids[idx];
593            let persistent_entry = &persistent_data[idx];
594            let data_entry = &data[idx];
595            db.put_keyed_persistent(cid, persistent_entry).unwrap();
596            // Check that we get persistent data if the data is otherwise absent from the GC enabled
597            // storage.
598            assert_eq!(
599                Blockstore::get(db.deref(), cid).unwrap(),
600                Some(persistent_entry.clone())
601            );
602            assert!(
603                db.read_from_column(cid.to_bytes(), DbColumn::PersistentGraph)
604                    .unwrap()
605                    .is_some()
606            );
607            db.put_keyed(cid, data_entry).unwrap();
608            assert_eq!(
609                Blockstore::get(db.deref(), cid).unwrap(),
610                Some(data_entry.clone())
611            );
612        }
613    }
614
615    #[test]
616    fn subscription_tests() {
617        let db = TempParityDB::new();
618        assert!(db.write_ops_broadcast_tx.read().is_none());
619        let data = [
620            b"h'nglui mglw'nafh".to_vec(),
621            b"Cthulhu".to_vec(),
622            b"R'lyeh wgah'nagl fhtagn!!".to_vec(),
623        ];
624
625        let cids = [
626            Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&data[0])),
627            Cid::new_v1(DAG_CBOR, MultihashCode::Sha2_256.digest(&data[1])),
628            Cid::new_v1(IPLD_RAW, MultihashCode::Blake2b256.digest(&data[1])),
629        ];
630
631        let mut rx1 = db.subscribe_write_ops().unwrap();
632        let mut rx2 = db.subscribe_write_ops().unwrap();
633
634        assert!(has_subscribers(
635            db.write_ops_broadcast_tx.read().as_ref().unwrap()
636        ));
637
638        for (idx, cid) in cids.iter().enumerate() {
639            let data_entry = &data[idx];
640            db.put_keyed(cid, data_entry).unwrap();
641            let expected = vec![(*cid, Bytes::copy_from_slice(data_entry))];
642            assert_eq!(rx1.blocking_recv().unwrap(), expected);
643            assert_eq!(rx2.blocking_recv().unwrap(), expected);
644        }
645
646        drop(rx1);
647        drop(rx2);
648
649        assert!(!has_subscribers(
650            db.write_ops_broadcast_tx.read().as_ref().unwrap()
651        ));
652
653        db.unsubscribe_write_ops();
654
655        assert!(db.write_ops_broadcast_tx.read().is_none());
656    }
657}
forest/db/parity_db.rs

forest/db/
parity_db.rs
