starknet_committer: add mdbx storage (#8917)

Author: yoavGrs

Cargo.lock

@@ -14,7 +14,9 @@ workspace = true
 
 [dependencies]
 hex.workspace = true
+libmdbx.workspace = true
 lru.workspace = true
+page_size.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
 starknet-types-core.workspace = true

crates/starknet_patricia_storage/Cargo.toml

@@ -3,4 +3,5 @@ pub mod errors;
 pub mod map_storage;
 #[cfg(test)]
 pub mod map_storage_test;
+pub mod mdbx_storage;
 pub mod storage_trait;

crates/starknet_patricia_storage/src/lib.rs

@@ -0,0 +1,111 @@
+use std::path::Path;
+
+use libmdbx::{
+    Database as MdbxDb,
+    DatabaseFlags,
+    Geometry,
+    PageSize,
+    TableFlags,
+    WriteFlags,
+    WriteMap,
+};
+use page_size;
+
+use crate::storage_trait::{DbHashMap, DbKey, DbValue, PatriciaStorageResult, Storage};
+
+pub struct MdbxStorage {
+    db: MdbxDb<WriteMap>,
+}
+
+// Size in bytes.
+const MDBX_MIN_PAGESIZE: usize = 1 << 8; // 256 bytes
+const MDBX_MAX_PAGESIZE: usize = 1 << 16; // 64KB
+
+fn get_page_size(os_page_size: usize) -> PageSize {
+    let mut page_size = os_page_size.clamp(MDBX_MIN_PAGESIZE, MDBX_MAX_PAGESIZE);
+
+    // Page size must be power of two.
+    if !page_size.is_power_of_two() {
+        page_size = page_size.next_power_of_two() / 2;
+    }
+
+    PageSize::Set(page_size)
+}
+
+impl MdbxStorage {
+    pub fn open(path: &Path) -> PatriciaStorageResult<Self> {
+        // TODO(tzahi): geometry and related definitions are taken from apollo_storage. Check if
+        // there are better configurations for the committer and consider moving boh crates mdbx
+        // code to a common location.
+        let db = MdbxDb::<WriteMap>::new()
+            .set_geometry(Geometry {
+                size: Some(1 << 20..1 << 40),
+                growth_step: Some(1 << 32),
+                page_size: Some(get_page_size(page_size::get())),
+                ..Default::default()
+            })
+            .set_flags(DatabaseFlags {
+                // As DbKeys are hashed, there is no locality of pages in the database almost at
+                // all, so readahead will fill the RAM with garbage.
+                // See https://libmdbx.dqdkfa.ru/group__c__opening.html#gga9138119a904355d245777c4119534061a16a07f878f8053cc79990063ca9510e7
+                no_rdahead: true,
+                // LIFO policy for recycling a Garbage Collection items should be faster when using
+                // disks with write-back cache.
+                liforeclaim: true,
+                ..Default::default()
+            })
+            .open(path)?;
+        let txn = db.begin_rw_txn()?;
+        txn.create_table(None, TableFlags::empty())?;
+        txn.commit()?;
+        Ok(Self { db })
+    }
+}
+
+impl Storage for MdbxStorage {
+    fn get(&mut self, key: &DbKey) -> PatriciaStorageResult<Option<DbValue>> {
+        let txn = self.db.begin_ro_txn()?;
+        let table = txn.open_table(None)?;
+        Ok(txn.get(&table, &key.0)?.map(DbValue))
+    }
+
+    fn set(&mut self, key: DbKey, value: DbValue) -> PatriciaStorageResult<Option<DbValue>> {
+        let txn = self.db.begin_rw_txn()?;
+        let table = txn.open_table(None)?;
+        let prev_val = txn.get(&table, &key.0)?.map(DbValue);
+        txn.put(&table, key.0, value.0, WriteFlags::UPSERT)?;
+        txn.commit()?;
+        Ok(prev_val)
+    }
+
+    fn mget(&mut self, keys: &[&DbKey]) -> PatriciaStorageResult<Vec<Option<DbValue>>> {
+        let txn = self.db.begin_ro_txn()?;
+        let table = txn.open_table(None)?;
+        let mut res = Vec::with_capacity(keys.len());
+        for key in keys {
+            res.push(txn.get(&table, &key.0)?.map(DbValue));
+        }
+        Ok(res)
+    }
+
+    fn mset(&mut self, key_to_value: DbHashMap) -> PatriciaStorageResult<()> {
+        let txn = self.db.begin_rw_txn()?;
+        let table = txn.open_table(None)?;
+        for (key, value) in key_to_value {
+            txn.put(&table, key.0, value.0, WriteFlags::UPSERT)?;
+        }
+        txn.commit()?;
+        Ok(())
+    }
+
+    fn delete(&mut self, key: &DbKey) -> PatriciaStorageResult<Option<DbValue>> {
+        let txn = self.db.begin_rw_txn()?;
+        let table = txn.open_table(None)?;
+        let prev_val = txn.get(&table, &key.0)?.map(DbValue);
+        if prev_val.is_some() {
+            txn.del(&table, &key.0, None)?;
+            txn.commit()?;
+        }
+        Ok(prev_val)
+    }
+}

crates/starknet_patricia_storage/src/mdbx_storage.rs

@@ -13,7 +13,11 @@ pub type DbHashMap = HashMap<DbKey, DbValue>;
 
 /// An error that can occur when interacting with the database.
 #[derive(thiserror::Error, Debug)]
-pub enum PatriciaStorageError {}
+pub enum PatriciaStorageError {
+    /// An error that occurred in the database library.
+    #[error(transparent)]
+    Mdbx(#[from] libmdbx::Error),
+}
 
 pub type PatriciaStorageResult<T> = Result<T, PatriciaStorageError>;