Redb slasher backend impl

Makefile

@@ -15,7 +15,7 @@ PINNED_NIGHTLY ?= nightly
 CLIPPY_PINNED_NIGHTLY=nightly-2022-05-19
 
 # List of features to use when cross-compiling. Can be overridden via the environment.
-CROSS_FEATURES ?= gnosis,slasher-lmdb,slasher-mdbx,jemalloc
+CROSS_FEATURES ?= gnosis,slasher-lmdb,slasher-mdbx,slasher-redb,jemalloc
 
 # Cargo profile for Cross builds. Default is for local builds, CI uses an override.
 CROSS_PROFILE ?= release

lighthouse/Cargo.toml

@@ -22,6 +22,8 @@ gnosis = []
 slasher-mdbx = ["slasher/mdbx"]
 # Support slasher LMDB backend.
 slasher-lmdb = ["slasher/lmdb"]
+# Support slasher redb backend.
+slasher-redb = ["slasher/redb"]
 # Deprecated. This is now enabled by default on non windows targets.
 jemalloc = []
 

lighthouse/tests/beacon_node.rs

@@ -2242,6 +2242,8 @@ fn slasher_broadcast_flag_false() {
             assert!(!slasher_config.broadcast);
         });
 }
+
+#[cfg(all(feature = "lmdb"))]
 #[test]
 fn slasher_backend_override_to_default() {
     // Hard to test this flag because all but one backend is disabled by default and the backend

slasher/Cargo.toml

@@ -8,11 +8,13 @@ edition = { workspace = true }
 default = ["lmdb"]
 mdbx = ["dep:mdbx"]
 lmdb = ["lmdb-rkv", "lmdb-rkv-sys"]
+redb = ["dep:redb"]
 portable = ["types/portable"]
 
 [dependencies]
 bincode = { workspace = true }
 byteorder = { workspace = true }
+derivative = { workspace = true }
 ethereum_ssz = { workspace = true }
 ethereum_ssz_derive = { workspace = true }
 flate2 = { version = "1.0.14", features = ["zlib"], default-features = false }
@@ -36,6 +38,8 @@ mdbx = { package = "libmdbx", git = "https://github.com/sigp/libmdbx-rs", tag =
 lmdb-rkv = { git = "https://github.com/sigp/lmdb-rs", rev = "f33845c6469b94265319aac0ed5085597862c27e", optional = true }
 lmdb-rkv-sys = { git = "https://github.com/sigp/lmdb-rs", rev = "f33845c6469b94265319aac0ed5085597862c27e", optional = true }
 
+redb = { version = "2.1", optional = true }
+
 [dev-dependencies]
 maplit = { workspace = true }
 rayon = { workspace = true }

slasher/src/config.rs

@@ -15,16 +15,19 @@ pub const DEFAULT_MAX_DB_SIZE: usize = 512 * 1024; // 512 GiB
 pub const DEFAULT_ATTESTATION_ROOT_CACHE_SIZE: NonZeroUsize = new_non_zero_usize(100_000);
 pub const DEFAULT_BROADCAST: bool = false;
 
-#[cfg(all(feature = "mdbx", not(feature = "lmdb")))]
+#[cfg(all(feature = "mdbx", not(any(feature = "lmdb", feature = "redb"))))]
 pub const DEFAULT_BACKEND: DatabaseBackend = DatabaseBackend::Mdbx;
 #[cfg(feature = "lmdb")]
 pub const DEFAULT_BACKEND: DatabaseBackend = DatabaseBackend::Lmdb;
-#[cfg(not(any(feature = "mdbx", feature = "lmdb")))]
+#[cfg(all(feature = "redb", not(any(feature = "mdbx", feature = "lmdb"))))]
+pub const DEFAULT_BACKEND: DatabaseBackend = DatabaseBackend::Redb;
+#[cfg(not(any(feature = "mdbx", feature = "lmdb", feature = "redb")))]
 pub const DEFAULT_BACKEND: DatabaseBackend = DatabaseBackend::Disabled;
 
 pub const MAX_HISTORY_LENGTH: usize = 1 << 16;
 pub const MEGABYTE: usize = 1 << 20;
 pub const MDBX_DATA_FILENAME: &str = "mdbx.dat";
+pub const REDB_DATA_FILENAME: &str = "slasher.redb";
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Config {
@@ -64,6 +67,8 @@ pub enum DatabaseBackend {
     Mdbx,
     #[cfg(feature = "lmdb")]
     Lmdb,
+    #[cfg(feature = "redb")]
+    Redb,
     Disabled,
 }
 

slasher/src/database.rs

@@ -1,6 +1,7 @@
 pub mod interface;
 mod lmdb_impl;
 mod mdbx_impl;
+mod redb_impl;
 
 use crate::{
     metrics, AttesterRecord, AttesterSlashingStatus, CompactAttesterRecord, Config, Error,
@@ -489,8 +490,7 @@ impl<E: EthSpec> SlasherDB<E> {
         }
 
         // Store the new indexed attestation at the end of the current table.
-        let db = &self.databases.indexed_attestation_db;
-        let mut cursor = txn.cursor(db)?;
+        let mut cursor = txn.cursor(&self.databases.indexed_attestation_db)?;
 
         let indexed_att_id = match cursor.last_key()? {
             // First ID is 1 so that 0 can be used to represent `null` in `CompactAttesterRecord`.
@@ -504,7 +504,6 @@ impl<E: EthSpec> SlasherDB<E> {
 
         cursor.put(attestation_key.as_ref(), &data)?;
         drop(cursor);
-
         // Update the (epoch, hash) to ID mapping.
         self.put_indexed_attestation_id(txn, &id_key, attestation_key)?;
 
@@ -743,21 +742,17 @@ impl<E: EthSpec> SlasherDB<E> {
             return Ok(());
         }
 
-        loop {
-            let (key_bytes, _) = cursor.get_current()?.ok_or(Error::MissingProposerKey)?;
-
-            let (slot, _) = ProposerKey::parse(key_bytes)?;
+        let should_delete = |key: &[u8]| -> Result<bool, Error> {
+            let mut should_delete = false;
+            let (slot, _) = ProposerKey::parse(Cow::from(key))?;
             if slot < min_slot {
-                cursor.delete_current()?;
-
-                // End the loop if there is no next entry.
-                if cursor.next_key()?.is_none() {
-                    break;
-                }
-            } else {
-                break;
+                should_delete = true;
             }
-        }
+
+            Ok(should_delete)
+        };
+
+        cursor.delete_while(should_delete)?;
 
         Ok(())
     }
@@ -771,37 +766,27 @@ impl<E: EthSpec> SlasherDB<E> {
             .saturating_add(1u64)
             .saturating_sub(self.config.history_length as u64);
 
-        // Collect indexed attestation IDs to delete.
-        let mut indexed_attestation_ids = vec![];
-
         let mut cursor = txn.cursor(&self.databases.indexed_attestation_id_db)?;
 
         // Position cursor at first key, bailing out if the database is empty.
         if cursor.first_key()?.is_none() {
             return Ok(());
         }
 
-        loop {
-            let (key_bytes, value) = cursor
-                .get_current()?
-                .ok_or(Error::MissingIndexedAttestationIdKey)?;
-
-            let (target_epoch, _) = IndexedAttestationIdKey::parse(key_bytes)?;
-
+        let should_delete = |key: &[u8]| -> Result<bool, Error> {
+            let (target_epoch, _) = IndexedAttestationIdKey::parse(Cow::from(key))?;
             if target_epoch < min_epoch {
-                indexed_attestation_ids.push(IndexedAttestationId::new(
-                    IndexedAttestationId::parse(value)?,
-                ));
+                return Ok(true);
+            }
 
-                cursor.delete_current()?;
+            Ok(false)
+        };
 
-                if cursor.next_key()?.is_none() {
-                    break;
-                }
-            } else {
-                break;
-            }
-        }
+        let indexed_attestation_ids = cursor
+            .delete_while(should_delete)?
+            .into_iter()
+            .map(|id| IndexedAttestationId::parse(id).map(IndexedAttestationId::new))
+            .collect::<Result<Vec<IndexedAttestationId>, Error>>()?;
         drop(cursor);
 
         // Delete the indexed attestations.