WIP; Add new rocksdb ChainDB implementation as alternative to Hammersbald

.gitignore

@@ -2,3 +2,5 @@
 target
 *.iml
 Cargo.lock
+client.db.*
+client.db

Cargo.toml

@@ -7,19 +7,21 @@ homepage = "https://github.com/rust-bitcoin/murmel/"
 repository = "https://github.com/rust-bitcoin/murmel/"
 documentation = "https://github.com/rust-bitcoin/murmel/"
 description = "Murmel Bitcoin node"
-keywords = [ "bitcoin" ]
+keywords = ["bitcoin"]
 readme = "README.md"
 edition = "2018"
 
+[features]
+default = ["hammersbald"]
+
 [lib]
 name = "murmel"
 path = "src/lib.rs"
 
 [dependencies]
-lightning = { version ="0.0.9", optional=true }
-bitcoin = { version= "0.21", features=["use-serde"]}
+lightning = { version = "0.0.9", optional = true }
+bitcoin = { version = "0.21", features = ["use-serde"] }
 bitcoin_hashes = "0.7"
-hammersbald = { version= "2.4", features=["bitcoin_support"]}
 mio = "0.6"
 rand = "0.7"
 log = "0.4"
@@ -28,8 +30,13 @@ byteorder = "1.2"
 lru-cache = "0.1.1"
 futures-preview = "=0.3.0-alpha.18"
 futures-timer = "0.3"
-serde="1"
-serde_derive="1"
+serde = "1"
+serde_derive = "1"
+serde_cbor = "0.10"
+
+## optional
+hammersbald = { version = "2.4", features = ["bitcoin_support"], optional = true }
+rocksdb = { version = "0.15.0", default-features = false, features = ["lz4"], optional = true }
 
 [dev-dependencies]
 rustc-serialize = "0.3"

src/chaindb.rs

@@ -14,176 +14,66 @@
 // limitations under the License.
 //
 //!
-//! # Blockchain DB for a node
+//! # Blockchain DB API for a node
 //!
 
 use std::sync::{Arc, RwLock};
-use std::path::Path;
 
-use bitcoin::{BitcoinHash, Network};
+use bitcoin::BitcoinHash;
 use bitcoin::blockdata::block::BlockHeader;
-use bitcoin::blockdata::constants::genesis_block;
 
 use bitcoin_hashes::sha256d;
-use hammersbald::{BitcoinAdaptor, HammersbaldAPI, persistent, transient};
 
 use crate::error::Error;
-use crate::headercache::{CachedHeader, HeaderCache};
-use log::{debug, info, warn, error};
+use crate::headercache::CachedHeader;
+
 use serde_derive::{Serialize, Deserialize};
 
 /// Shared handle to a database storing the block chain
 /// protected by an RwLock
-pub type SharedChainDB = Arc<RwLock<ChainDB>>;
-
-/// Database storing the block chain
-pub struct ChainDB {
-    db: BitcoinAdaptor,
-    headercache: HeaderCache,
-    network: Network,
-}
-
-impl ChainDB {
-    /// Create an in-memory database instance
-    pub fn mem(network: Network) -> Result<ChainDB, Error> {
-        info!("working with in memory chain db");
-        let db = BitcoinAdaptor::new(transient(2)?);
-        let headercache = HeaderCache::new(network);
-        Ok(ChainDB { db, network, headercache })
-    }
+pub type SharedChainDB = Arc<RwLock<Box<dyn ChainDB>>>;
 
-    /// Create or open a persistent database instance identified by the path
-    pub fn new(path: &Path, network: Network) -> Result<ChainDB, Error> {
-        let basename = path.to_str().unwrap().to_string();
-        let db = BitcoinAdaptor::new(persistent((basename.clone()).as_str(), 100, 2)?);
-        let headercache = HeaderCache::new(network);
-        Ok(ChainDB { db, network, headercache })
-    }
+/// Blockchain DB API for a client node.
+pub trait ChainDB: Send + Sync {
 
-    /// Initialize caches
-    pub fn init(&mut self) -> Result<(), Error> {
-        self.init_headers()?;
-        Ok(())
-    }
+    /// Initialize caches.
+    fn init(&mut self) -> Result<(), Error>;
 
     /// Batch updates. Updates are permanent after finishing a batch.
-    pub fn batch(&mut self) -> Result<(), Error> {
-        self.db.batch()?;
-        Ok(())
-    }
+    fn batch(&mut self) -> Result<(), Error>;
 
-    fn init_headers(&mut self) -> Result<(), Error> {
-        if let Some(tip) = self.fetch_header_tip()? {
-            info!("reading stored header chain from tip {}", tip);
-            if self.fetch_header(&tip)?.is_some() {
-                let mut h = tip;
-                while let Some(stored) = self.fetch_header(&h)? {
-                    debug!("read stored header {}", &stored.bitcoin_hash());
-                    self.headercache.add_header_unchecked(&h, &stored);
-                    if stored.header.prev_blockhash != sha256d::Hash::default() {
-                        h = stored.header.prev_blockhash;
-                    } else {
-                        break;
-                    }
-                }
-                self.headercache.reverse_trunk();
-                info!("read {} headers", self.headercache.len());
-            } else {
-                warn!("unable to read header for tip {}", tip);
-                self.init_to_genesis()?;
-            }
-        } else {
-            info!("no header tip found");
-            self.init_to_genesis()?;
-        }
-        Ok(())
-    }
+    /// Store a header.
+    fn add_header(&mut self, header: &BlockHeader) -> Result<Option<(StoredHeader, Option<Vec<sha256d::Hash>>, Option<Vec<sha256d::Hash>>)>, Error>;
 
-    fn init_to_genesis(&mut self) -> Result<(), Error> {
-        let genesis = genesis_block(self.network).header;
-        if let Some((cached, _, _)) = self.headercache.add_header(&genesis)? {
-            info!("initialized with genesis header {}", genesis.bitcoin_hash());
-            self.db.put_hash_keyed(&cached.stored)?;
-            self.db.batch()?;
-            self.store_header_tip(&cached.bitcoin_hash())?;
-            self.db.batch()?;
-        } else {
-            error!("failed to initialize with genesis header");
-            return Err(Error::NoTip);
-        }
-        Ok(())
-    }
+    /// Return position of hash on trunk if hash is on trunk.
+    fn pos_on_trunk(&self, hash: &sha256d::Hash) -> Option<u32>;
 
-    /// Store a header
-    pub fn add_header(&mut self, header: &BlockHeader) -> Result<Option<(StoredHeader, Option<Vec<sha256d::Hash>>, Option<Vec<sha256d::Hash>>)>, Error> {
-        if let Some((cached, unwinds, forward)) = self.headercache.add_header(header)? {
-            self.db.put_hash_keyed(&cached.stored)?;
-            if let Some(forward) = forward.clone() {
-                if forward.len() > 0 {
-                    self.store_header_tip(forward.last().unwrap())?;
-                }
-            }
-            return Ok(Some((cached.stored, unwinds, forward)));
-        }
-        Ok(None)
-    }
+    /// Iterate trunk [from .. tip].
+    fn iter_trunk<'a>(&'a self, from: u32) -> Box<dyn Iterator<Item=&'a CachedHeader> + 'a>;
 
-    /// return position of hash on trunk if hash is on trunk
-    pub fn pos_on_trunk(&self, hash: &sha256d::Hash) -> Option<u32> {
-        self.headercache.pos_on_trunk(hash)
-    }
+    /// Iterate trunk [genesis .. from] in reverse order from is the tip if not specified.
+    fn iter_trunk_rev<'a>(&'a self, from: Option<u32>) -> Box<dyn Iterator<Item=&'a CachedHeader> + 'a>;
 
-    /// iterate trunk [from .. tip]
-    pub fn iter_trunk<'a>(&'a self, from: u32) -> impl Iterator<Item=&'a CachedHeader> + 'a {
-        self.headercache.iter_trunk(from)
-    }
-
-    /// iterate trunk [genesis .. from] in reverse order from is the tip if not specified
-    pub fn iter_trunk_rev<'a>(&'a self, from: Option<u32>) -> impl Iterator<Item=&'a CachedHeader> + 'a {
-        self.headercache.iter_trunk_rev(from)
-    }
-
-    /// retrieve the id of the block/header with most work
-    pub fn header_tip(&self) -> Option<CachedHeader> {
-        self.headercache.tip()
-    }
-
-    /// Fetch a header by its id from cache
-    pub fn get_header(&self, id: &sha256d::Hash) -> Option<CachedHeader> {
-        self.headercache.get_header(id)
-    }
+    /// Retrieve the id of the block/header with most work.
+    fn header_tip(&self) -> Option<CachedHeader>;
 
-    /// Fetch a header by its id from cache
-    pub fn get_header_for_height(&self, height: u32) -> Option<CachedHeader> {
-        self.headercache.get_header_for_height(height)
-    }
+    /// Fetch a header by its id from cache.
+    fn get_header(&self, id: &sha256d::Hash) -> Option<CachedHeader>;
 
-    /// locator for getheaders message
-    pub fn header_locators(&self) -> Vec<sha256d::Hash> {
-        self.headercache.locator_hashes()
-    }
+    /// Fetch a header by its id from cache.
+    fn get_header_for_height(&self, height: u32) -> Option<CachedHeader>;
 
-    /// Store the header id with most work
-    pub fn store_header_tip(&mut self, tip: &sha256d::Hash) -> Result<(), Error> {
-        self.db.put_keyed_encodable(HEADER_TIP_KEY, tip)?;
-        Ok(())
-    }
+    /// Locator for getheaders message.
+    fn header_locators(&self) -> Vec<sha256d::Hash>;
 
-    /// Find header id with most work
-    pub fn fetch_header_tip(&self) -> Result<Option<sha256d::Hash>, Error> {
-        Ok(self.db.get_keyed_decodable::<sha256d::Hash>(HEADER_TIP_KEY)?.map(|(_, h)| h.clone()))
-    }
+    /// Store the header id with most work.
+    fn store_header_tip(&mut self, tip: &sha256d::Hash) -> Result<(), Error>;
 
-    /// Read header from the DB
-    pub fn fetch_header(&self, id: &sha256d::Hash) -> Result<Option<StoredHeader>, Error> {
-        Ok(self.db.get_hash_keyed::<StoredHeader>(id)?.map(|(_, header)| header))
-    }
+    /// Find header id with most work.
+    fn fetch_header_tip(&self) -> Result<Option<sha256d::Hash>, Error>;
 
-    /// Shutdown db
-    pub fn shutdown(&mut self) {
-        self.db.shutdown();
-        debug!("shutdown chain db")
-    }
+    /// Read header from the DB.
+    fn fetch_header(&self, id: &sha256d::Hash) -> Result<Option<StoredHeader>, Error>;
 }
 
 /// A header enriched with information about its position on the blockchain
@@ -204,45 +94,4 @@ impl BitcoinHash for StoredHeader {
     }
 }
 
-const HEADER_TIP_KEY: &[u8] = &[0u8; 1];
-
-#[cfg(test)]
-mod test {
-    use bitcoin::{Network, BitcoinHash};
-    use bitcoin_hashes::sha256d::Hash;
-    use bitcoin::blockdata::constants::genesis_block;
-
-    use crate::chaindb::ChainDB;
-
-    #[test]
-    fn init_tip_header() {
-        let network = Network::Testnet;
-        let genesis_header = genesis_block(network).header;
-
-        let mut chaindb = ChainDB::mem(network).unwrap();
-        chaindb.init().unwrap();
-        chaindb.init().unwrap();
-
-        let header_tip = chaindb.header_tip();
-        assert!(header_tip.is_some(), "failed to get header for tip");
-        assert!(header_tip.unwrap().stored.bitcoin_hash().eq(&genesis_header.bitcoin_hash()))
-    }
-
-    #[test]
-    fn init_recover_if_missing_tip_header() {
-        let network = Network::Testnet;
-        let genesis_header = genesis_block(network).header;
-
-        let mut chaindb = ChainDB::mem(network).unwrap();
-        let missing_tip_header_hash: Hash = "6cfb35868c4465b7c289d7d5641563aa973db6a929655282a7bf95c8257f53ef".parse().unwrap();
-        chaindb.store_header_tip(&missing_tip_header_hash).unwrap();
-
-        chaindb.init().unwrap();
-
-        let header_tip = chaindb.header_tip();
-        assert!(header_tip.is_some(), "failed to get header for tip");
-        assert!(header_tip.unwrap().stored.bitcoin_hash().eq(&genesis_header.bitcoin_hash()))
-    }
-}
-
 

src/constructor.rs

@@ -24,7 +24,7 @@ use bitcoin::{
         constants::Network
     }
 };
-use crate::chaindb::{ChainDB, SharedChainDB};
+
 use crate::dispatcher::Dispatcher;
 use crate::dns::dns_seed;
 use crate::error::Error;
@@ -55,6 +55,7 @@ use bitcoin::network::message::NetworkMessage;
 use bitcoin::network::message::RawNetworkMessage;
 use crate::p2p::BitcoinP2PConfig;
 use std::time::Duration;
+use crate::chaindb::{SharedChainDB, ChainDB};
 
 const MAX_PROTOCOL_VERSION: u32 = 70001;
 const USER_AGENT: &'static str = concat!("/Murmel:", env!("CARGO_PKG_VERSION"), '/');
@@ -67,18 +68,35 @@ pub struct Constructor {
 }
 
 impl Constructor {
-    /// open DBs
+    /// open DB
     pub fn open_db(path: Option<&Path>, network: Network, _birth: u64) -> Result<SharedChainDB, Error> {
-        let mut chaindb =
-            if let Some(path) = path {
-                ChainDB::new(path, network)?
-            } else {
-                ChainDB::mem(network)?
-            };
+        let mut chaindb = Constructor::new_db(path, network)?;
         chaindb.init()?;
         Ok(Arc::new(RwLock::new(chaindb)))
     }
 
+    /// new Hammersbald DB
+    #[cfg(feature = "default")]
+    fn new_db(path: Option<&Path>, network:Network) -> Result<Box<dyn ChainDB>, Error> {
+        use crate::hammersbald::Hammersbald;
+        if let Some(path) = path {
+            Hammersbald::new(path, network)
+        } else {
+            Hammersbald::mem(network)
+        }
+    }
+
+    /// new RocksDB
+    #[cfg(feature = "rocksdb")]
+    fn new_db(path: Option<&Path>, network:Network) -> Result<Box<dyn ChainDB>, Error> {
+        use crate::rocksdb::RocksDB;
+        if let Some(path) = path {
+            RocksDB::new(path, network)
+        } else {
+            Err(Error::RocksDB("RocksDB doesn't support mem DB, path is required.".to_string()))
+        }
+    }
+
     /// Construct the stack
     pub fn new(network: Network, listen: Vec<SocketAddr>, chaindb: SharedChainDB) -> Result<Constructor, Error> {
         const BACK_PRESSURE: usize = 10;