Merge pull request #786 from subspace/parity-db-migration

Parity db migration

Author: i1i1

Cargo.lock

@@ -86,3 +86,5 @@ codegen-units = 1
 # TODO: Remove once chacha20poly1305 0.10 appears in libp2p's dependencies
 chacha20poly1305 = { git = "https://github.com/RustCrypto/AEADs", rev = "06dbfb5571687fd1bbe9d3c9b2193a1ba17f8e99" }
 libp2p = { git = "https://github.com/subspace/rust-libp2p", branch = "subspace-v3" }
+# TODO: Remove once parity-db 0.3.17 will be released (see https://github.com/paritytech/parity-db/issues/89)
+parity-db = { git = "https://github.com/paritytech/parity-db/", rev = "d324a26daabda11a2307de607d0a614a567b44ba" }

Cargo.toml

@@ -20,8 +20,6 @@ pub enum ArchivingError {
     GetBlockError(u32),
     #[error("jsonrpsee error: {0}")]
     RpcError(Box<dyn std::error::Error + Send + Sync>),
-    #[error("Last block retrieval from plot, rocksdb error: {0}")]
-    LastBlock(rocksdb::Error),
     #[error("Error joining task: {0}")]
     JoinTask(tokio::task::JoinError),
     #[error("Archiver instantiation error: {0}")]

crates/subspace-farmer/src/archiving.rs

@@ -7,9 +7,9 @@ use crate::plot::{PieceOffset, Plot};
 use arc_swap::ArcSwapOption;
 use databases::{CommitmentDatabases, CreateDbEntryResult, DbEntry};
 use event_listener_primitives::{Bag, HandlerId};
+use parity_db::Db;
 use parking_lot::Mutex;
 use rayon::prelude::*;
-use rocksdb::{WriteBatch, DB};
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
@@ -28,11 +28,13 @@ const TAGS_WRITE_BATCH_SIZE: usize = 16 * 1024 * 1024 / (TAG_SIZE + PIECE_OFFSET
 #[derive(Debug, Error)]
 pub enum CommitmentError {
     #[error("Metadata DB error: {0}")]
-    MetadataDb(rocksdb::Error),
+    MetadataDb(parity_db::Error),
     #[error("Commitment DB error: {0}")]
-    CommitmentDb(rocksdb::Error),
+    CommitmentDb(parity_db::Error),
     #[error("Plot error: {0}")]
     Plot(io::Error),
+    #[error("Migration error: {0}")]
+    Migrate(io::Error),
 }
 
 #[derive(Debug, Copy, Clone)]
@@ -130,9 +132,23 @@ impl Commitments {
             self.inner.current.swap(current);
             self.inner.next.swap(next);
 
-            let db_path = self.inner.base_directory.join(hex::encode(salt));
+            let options = parity_db::Options {
+                path: self.inner.base_directory.join(hex::encode(salt)),
+                columns: vec![parity_db::ColumnOptions {
+                    preimage: false,
+                    btree_index: true,
+                    uniform: false,
+                    ref_counted: false,
+                    compression: parity_db::CompressionType::NoCompression,
+                    compression_threshold: 4096,
+                }],
+                sync_wal: true,
+                sync_data: true,
+                stats: false,
+                salt: None,
+            };
             db_entry.lock().replace(Arc::new(
-                DB::open_default(db_path).map_err(CommitmentError::CommitmentDb)?,
+                Db::open_or_create(&options).map_err(CommitmentError::CommitmentDb)?,
             ));
         }
 
@@ -173,11 +189,12 @@ impl Commitments {
                 if tags_with_offset.len() == tags_with_offset.capacity() {
                     tags_with_offset.sort_by(|(tag_a, _), (tag_b, _)| tag_a.cmp(tag_b));
 
-                    let mut batch = WriteBatch::default();
-                    for (tag, offset) in &tags_with_offset {
-                        batch.put(tag, offset);
-                    }
-                    db.write(batch).map_err(CommitmentError::CommitmentDb)?;
+                    db.commit(
+                        tags_with_offset
+                            .iter()
+                            .map(|(tag, offset)| (0, tag, Some(offset.to_vec()))),
+                    )
+                    .map_err(CommitmentError::CommitmentDb)?;
 
                     tags_with_offset.clear();
                 }
@@ -195,11 +212,12 @@ impl Commitments {
                 if let Some(db) = db_guard.as_ref() {
                     tags_with_offset.sort_by(|(tag_a, _), (tag_b, _)| tag_a.cmp(tag_b));
 
-                    let mut batch = WriteBatch::default();
-                    for (tag, offset) in &tags_with_offset {
-                        batch.put(tag, offset);
-                    }
-                    db.write(batch).map_err(CommitmentError::CommitmentDb)?;
+                    db.commit(
+                        tags_with_offset
+                            .iter()
+                            .map(|(tag, offset)| (0, tag, Some(offset.to_vec()))),
+                    )
+                    .map_err(CommitmentError::CommitmentDb)?;
                 }
             }
 
@@ -240,12 +258,12 @@ impl Commitments {
             let db_guard = db_entry.lock();
 
             if let Some(db) = db_guard.as_ref() {
-                let mut batch = WriteBatch::default();
-                for piece in pieces {
-                    let tag = create_tag(piece, salt);
-                    batch.delete(tag);
-                }
-                db.write(batch).map_err(CommitmentError::CommitmentDb)?;
+                db.commit(
+                    pieces
+                        .iter()
+                        .map(|piece| (0, create_tag(piece, salt), None)),
+                )
+                .map_err(CommitmentError::CommitmentDb)?;
             }
         }
 
@@ -276,11 +294,12 @@ impl Commitments {
 
                 tags_with_offset.sort_by(|(tag_a, _), (tag_b, _)| tag_a.cmp(tag_b));
 
-                let mut batch = WriteBatch::default();
-                for (tag, piece_offset) in tags_with_offset {
-                    batch.put(tag, piece_offset.to_le_bytes());
-                }
-                db.write(batch).map_err(CommitmentError::CommitmentDb)?;
+                db.commit(
+                    tags_with_offset
+                        .into_iter()
+                        .map(|(tag, offset)| (0, tag, Some(offset.to_le_bytes().to_vec()))),
+                )
+                .map_err(CommitmentError::CommitmentDb)?;
             };
         }
 
@@ -315,12 +334,17 @@ impl Commitments {
                 return Vec::new();
             }
         };
-        let iter = db.raw_iterator();
+        let iter = match db.iter(0) {
+            Ok(iter) => iter,
+            Err(_) => {
+                return Vec::new();
+            }
+        };
 
         // Take the best out of 10 solutions
         let mut solutions = SolutionIterator::new(iter, target, range)
-            .take(limit)
-            .collect::<Vec<_>>();
+            .map(|iter| iter.take(limit).collect::<Vec<_>>())
+            .unwrap_or_default();
         let target = u64::from_be_bytes(target);
         solutions.sort_by_key(|(tag, _)| {
             let tag = u64::from_be_bytes(*tag);
@@ -374,7 +398,7 @@ enum SolutionIteratorState {
 }
 
 pub(crate) struct SolutionIterator<'a> {
-    iter: rocksdb::DBRawIterator<'a>,
+    iter: parity_db::BTreeIterator<'a>,
     state: SolutionIteratorState,
     /// Lower bound of solution range
     lower: u64,
@@ -383,7 +407,11 @@ pub(crate) struct SolutionIterator<'a> {
 }
 
 impl<'a> SolutionIterator<'a> {
-    pub fn new(mut iter: rocksdb::DBRawIterator<'a>, target: Tag, range: u64) -> Self {
+    pub fn new(
+        mut iter: parity_db::BTreeIterator<'a>,
+        target: Tag,
+        range: u64,
+    ) -> parity_db::Result<Self> {
         let (lower, is_lower_overflowed) = u64::from_be_bytes(target).overflowing_sub(range / 2);
         let (upper, is_upper_overflowed) = u64::from_be_bytes(target).overflowing_add(range / 2);
 
@@ -394,29 +422,27 @@ impl<'a> SolutionIterator<'a> {
         );
 
         let state = if is_lower_overflowed || is_upper_overflowed {
-            iter.seek_to_first();
+            iter.seek_to_first()?;
             SolutionIteratorState::OverflowStart
         } else {
-            iter.seek(lower.to_be_bytes());
+            iter.seek(&lower.to_be_bytes())?;
             SolutionIteratorState::NoOverflow
         };
-        Self {
+        Ok(Self {
             iter,
             state,
             lower,
             upper,
-        }
+        })
     }
 
     fn next_entry(&mut self) -> Option<(Tag, PieceOffset)> {
-        self.iter
-            .key()
-            .map(|tag| tag.try_into().unwrap())
-            .map(|tag| {
-                let offset = u64::from_le_bytes(self.iter.value().unwrap().try_into().unwrap());
-                self.iter.next();
-                (tag, offset)
-            })
+        self.iter.next().ok().flatten().map(|(tag, offset)| {
+            (
+                tag.try_into().unwrap(),
+                u64::from_le_bytes(offset.try_into().unwrap()),
+            )
+        })
     }
 }
 
@@ -433,7 +459,7 @@ impl<'a> Iterator for SolutionIterator<'a> {
                 .filter(|(tag, _)| u64::from_be_bytes(*tag) <= self.upper)
                 .or_else(|| {
                     self.state = SolutionIteratorState::OverflowEnd;
-                    self.iter.seek(self.lower.to_be_bytes());
+                    self.iter.seek(&self.lower.to_be_bytes()).ok()?;
                     self.next()
                 }),
             SolutionIteratorState::OverflowEnd => self.next_entry(),

crates/subspace-farmer/src/commitments.rs

@@ -1,8 +1,8 @@
 use crate::commitments::metadata::{CommitmentMetadata, CommitmentStatus};
 use crate::commitments::CommitmentError;
 use lru::LruCache;
+use parity_db::{ColumnOptions, CompressionType, Db, Options};
 use parking_lot::Mutex;
-use rocksdb::{Options, DB};
 use std::fmt;
 use std::ops::Deref;
 use std::path::PathBuf;
@@ -20,7 +20,7 @@ pub(super) struct CreateDbEntryResult {
 
 pub(super) struct DbEntry {
     salt: Salt,
-    db: Mutex<Option<Arc<DB>>>,
+    db: Mutex<Option<Arc<Db>>>,
 }
 
 impl fmt::Debug for DbEntry {
@@ -30,7 +30,7 @@ impl fmt::Debug for DbEntry {
 }
 
 impl Deref for DbEntry {
-    type Target = Mutex<Option<Arc<DB>>>;
+    type Target = Mutex<Option<Arc<Db>>>;
 
     fn deref(&self) -> &Self::Target {
         &self.db
@@ -52,6 +52,11 @@ pub(super) struct CommitmentDatabases {
 
 impl CommitmentDatabases {
     pub(super) fn new(base_directory: PathBuf) -> Result<Self, CommitmentError> {
+        if rocksdb::DB::open_default(base_directory.join("metadata")).is_ok() {
+            std::fs::remove_dir_all(&base_directory).map_err(CommitmentError::Migrate)?;
+            std::fs::create_dir(&base_directory).map_err(CommitmentError::Migrate)?;
+        }
+
         let mut metadata = CommitmentMetadata::new(base_directory.join("metadata"))?;
         let mut databases = LruCache::new(COMMITMENTS_CACHE_SIZE);
 
@@ -74,8 +79,22 @@ impl CommitmentDatabases {
 
             // Open databases that were fully created during previous run
             for salt in metadata.keys() {
-                let db = DB::open(&Options::default(), base_directory.join(hex::encode(salt)))
-                    .map_err(CommitmentError::CommitmentDb)?;
+                let options = Options {
+                    path: base_directory.join(hex::encode(salt)),
+                    columns: vec![ColumnOptions {
+                        preimage: false,
+                        btree_index: true,
+                        uniform: false,
+                        ref_counted: false,
+                        compression: CompressionType::NoCompression,
+                        compression_threshold: 4096,
+                    }],
+                    sync_wal: true,
+                    sync_data: true,
+                    stats: false,
+                    salt: None,
+                };
+                let db = Db::open_or_create(&options).map_err(CommitmentError::CommitmentDb)?;
                 databases.put(
                     *salt,
                     Arc::new(DbEntry {