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

forest/db/
parity_db.rs
