From 70bcba1e6b7a7123b2f48e1c77b9afb9bbb11c26 Mon Sep 17 00:00:00 2001 From: Eitan Seri-Levi Date: Mon, 1 Jul 2024 03:36:40 +0200 Subject: [PATCH] Redb slasher backend impl (#4529) * initial redb impl * redb impl * remove phantom data * fixed table definition * fighting the borrow checker * a rough draft that doesnt cause lifetime issues * refactoring * refactor * refactor * passing unit tests * refactor * refactor * refactor * commit * move everything to one database * remove panics, ready for a review * merge * a working redb impl * passing a ref of txn to cursor * this tries to create a second write transaction when initializing cursor. breaks everything * Use 2 lifetimes and subtyping Also fixes a bug in last_key caused by rev and next_back cancelling out * Move table into cursor * Merge remote-tracking branch 'origin/unstable' into redb-slasher-backend-impl * changes based on feedback * update lmdb * fix lifetime issues * moving everything from Cursor to Transaction * update * upgrade to redb 2.0 * Merge branch 'unstable' of https://github.com/sigp/lighthouse into redb-slasher-backend-impl * bring back cursor * Merge branch 'unstable' of https://github.com/sigp/lighthouse into redb-slasher-backend-impl * fix delete while * linting * linting * switch to lmdb * update redb to v2.1 * build fixes, remove unwrap or default * another build error * hopefully this is the last build error * fmt * cargo.toml * fix mdbx * Merge branch 'unstable' of https://github.com/sigp/lighthouse into redb-slasher-backend-impl * Remove a collect * Merge remote-tracking branch 'origin/unstable' into redb-slasher-backend-impl * Merge branch 'redb-slasher-backend-impl' of https://github.com/eserilev/lighthouse into redb-slasher-backend-impl * re-enable test * fix failing slasher test * Merge remote-tracking branch 'origin/unstable' into redb-slasher-backend-impl * Rename DB file to `slasher.redb` --- Cargo.lock | 11 ++ Makefile | 2 +- lighthouse/Cargo.toml | 2 + lighthouse/tests/beacon_node.rs | 2 + slasher/Cargo.toml | 4 + slasher/src/config.rs | 9 +- slasher/src/database.rs | 59 +++--- slasher/src/database/interface.rs | 80 ++++++-- slasher/src/database/lmdb_impl.rs | 25 ++- slasher/src/database/mdbx_impl.rs | 25 ++- slasher/src/database/redb_impl.rs | 276 ++++++++++++++++++++++++++++ slasher/src/error.rs | 38 ++++ slasher/src/lib.rs | 2 +- slasher/tests/attester_slashings.rs | 2 +- slasher/tests/backend.rs | 20 +- slasher/tests/proposer_slashings.rs | 2 +- slasher/tests/random.rs | 2 +- slasher/tests/wrap_around.rs | 2 +- 18 files changed, 495 insertions(+), 68 deletions(-) create mode 100644 slasher/src/database/redb_impl.rs diff --git a/Cargo.lock b/Cargo.lock index d0ada21c807..f23777bc4c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6678,6 +6678,15 @@ dependencies = [ "yasna", ] +[[package]] +name = "redb" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed7508e692a49b6b2290b56540384ccae9b1fb4d77065640b165835b56ffe3bb" +dependencies = [ + "libc", +] + [[package]] name = "redox_syscall" version = "0.2.16" @@ -7635,6 +7644,7 @@ version = "0.1.0" dependencies = [ "bincode", "byteorder", + "derivative", "ethereum_ssz", "ethereum_ssz_derive", "filesystem", @@ -7650,6 +7660,7 @@ dependencies = [ "parking_lot 0.12.3", "rand", "rayon", + "redb", "safe_arith", "serde", "slog", diff --git a/Makefile b/Makefile index 3f8e688df18..7d144e55fb8 100644 --- a/Makefile +++ b/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 diff --git a/lighthouse/Cargo.toml b/lighthouse/Cargo.toml index 67c3dc260ef..912602776af 100644 --- a/lighthouse/Cargo.toml +++ b/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 = [] diff --git a/lighthouse/tests/beacon_node.rs b/lighthouse/tests/beacon_node.rs index caadf208e32..cd499f2adae 100644 --- a/lighthouse/tests/beacon_node.rs +++ b/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 diff --git a/slasher/Cargo.toml b/slasher/Cargo.toml index 563c4599d8b..01a8b9fb002 100644 --- a/slasher/Cargo.toml +++ b/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 } diff --git a/slasher/src/config.rs b/slasher/src/config.rs index 1851e2e4418..33d68fa0e5d 100644 --- a/slasher/src/config.rs +++ b/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, } diff --git a/slasher/src/database.rs b/slasher/src/database.rs index 801abe9283b..4f4729a123f 100644 --- a/slasher/src/database.rs +++ b/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 SlasherDB { } // 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 SlasherDB { 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 SlasherDB { 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 { + 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,9 +766,6 @@ impl SlasherDB { .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. @@ -781,27 +773,20 @@ impl SlasherDB { 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 { + 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::, Error>>()?; drop(cursor); // Delete the indexed attestations. diff --git a/slasher/src/database/interface.rs b/slasher/src/database/interface.rs index 5bb920383c3..46cf9a4a0c3 100644 --- a/slasher/src/database/interface.rs +++ b/slasher/src/database/interface.rs @@ -7,6 +7,8 @@ use std::path::PathBuf; use crate::database::lmdb_impl; #[cfg(feature = "mdbx")] use crate::database::mdbx_impl; +#[cfg(feature = "redb")] +use crate::database::redb_impl; #[derive(Debug)] pub enum Environment { @@ -14,6 +16,8 @@ pub enum Environment { Mdbx(mdbx_impl::Environment), #[cfg(feature = "lmdb")] Lmdb(lmdb_impl::Environment), + #[cfg(feature = "redb")] + Redb(redb_impl::Environment), Disabled, } @@ -23,6 +27,8 @@ pub enum RwTransaction<'env> { Mdbx(mdbx_impl::RwTransaction<'env>), #[cfg(feature = "lmdb")] Lmdb(lmdb_impl::RwTransaction<'env>), + #[cfg(feature = "redb")] + Redb(redb_impl::RwTransaction<'env>), Disabled(PhantomData<&'env ()>), } @@ -32,6 +38,8 @@ pub enum Database<'env> { Mdbx(mdbx_impl::Database<'env>), #[cfg(feature = "lmdb")] Lmdb(lmdb_impl::Database<'env>), + #[cfg(feature = "redb")] + Redb(redb_impl::Database<'env>), Disabled(PhantomData<&'env ()>), } @@ -54,6 +62,8 @@ pub enum Cursor<'env> { Mdbx(mdbx_impl::Cursor<'env>), #[cfg(feature = "lmdb")] Lmdb(lmdb_impl::Cursor<'env>), + #[cfg(feature = "redb")] + Redb(redb_impl::Cursor<'env>), Disabled(PhantomData<&'env ()>), } @@ -67,6 +77,8 @@ impl Environment { DatabaseBackend::Mdbx => mdbx_impl::Environment::new(config).map(Environment::Mdbx), #[cfg(feature = "lmdb")] DatabaseBackend::Lmdb => lmdb_impl::Environment::new(config).map(Environment::Lmdb), + #[cfg(feature = "redb")] + DatabaseBackend::Redb => redb_impl::Environment::new(config).map(Environment::Redb), DatabaseBackend::Disabled => Err(Error::SlasherDatabaseBackendDisabled), } } @@ -77,6 +89,8 @@ impl Environment { Self::Mdbx(env) => env.create_databases(), #[cfg(feature = "lmdb")] Self::Lmdb(env) => env.create_databases(), + #[cfg(feature = "redb")] + Self::Redb(env) => env.create_databases(), _ => Err(Error::MismatchedDatabaseVariant), } } @@ -87,6 +101,8 @@ impl Environment { Self::Mdbx(env) => env.begin_rw_txn().map(RwTransaction::Mdbx), #[cfg(feature = "lmdb")] Self::Lmdb(env) => env.begin_rw_txn().map(RwTransaction::Lmdb), + #[cfg(feature = "redb")] + Self::Redb(env) => env.begin_rw_txn().map(RwTransaction::Redb), _ => Err(Error::MismatchedDatabaseVariant), } } @@ -98,6 +114,8 @@ impl Environment { Self::Mdbx(env) => env.filenames(config), #[cfg(feature = "lmdb")] Self::Lmdb(env) => env.filenames(config), + #[cfg(feature = "redb")] + Self::Redb(env) => env.filenames(config), _ => vec![], } } @@ -106,7 +124,7 @@ impl Environment { impl<'env> RwTransaction<'env> { pub fn get + ?Sized>( &'env self, - db: &Database<'env>, + db: &'env Database, key: &K, ) -> Result>, Error> { match (self, db) { @@ -114,6 +132,8 @@ impl<'env> RwTransaction<'env> { (Self::Mdbx(txn), Database::Mdbx(db)) => txn.get(db, key), #[cfg(feature = "lmdb")] (Self::Lmdb(txn), Database::Lmdb(db)) => txn.get(db, key), + #[cfg(feature = "redb")] + (Self::Redb(txn), Database::Redb(db)) => txn.get(db, key), _ => Err(Error::MismatchedDatabaseVariant), } } @@ -129,6 +149,8 @@ impl<'env> RwTransaction<'env> { (Self::Mdbx(txn), Database::Mdbx(db)) => txn.put(db, key, value), #[cfg(feature = "lmdb")] (Self::Lmdb(txn), Database::Lmdb(db)) => txn.put(db, key, value), + #[cfg(feature = "redb")] + (Self::Redb(txn), Database::Redb(db)) => txn.put(db, key, value), _ => Err(Error::MismatchedDatabaseVariant), } } @@ -139,26 +161,32 @@ impl<'env> RwTransaction<'env> { (Self::Mdbx(txn), Database::Mdbx(db)) => txn.del(db, key), #[cfg(feature = "lmdb")] (Self::Lmdb(txn), Database::Lmdb(db)) => txn.del(db, key), + #[cfg(feature = "redb")] + (Self::Redb(txn), Database::Redb(db)) => txn.del(db, key), _ => Err(Error::MismatchedDatabaseVariant), } } - pub fn cursor<'a>(&'a mut self, db: &Database) -> Result, Error> { - match (self, db) { + pub fn commit(self) -> Result<(), Error> { + match self { #[cfg(feature = "mdbx")] - (Self::Mdbx(txn), Database::Mdbx(db)) => txn.cursor(db).map(Cursor::Mdbx), + Self::Mdbx(txn) => txn.commit(), #[cfg(feature = "lmdb")] - (Self::Lmdb(txn), Database::Lmdb(db)) => txn.cursor(db).map(Cursor::Lmdb), + Self::Lmdb(txn) => txn.commit(), + #[cfg(feature = "redb")] + Self::Redb(txn) => txn.commit(), _ => Err(Error::MismatchedDatabaseVariant), } } - pub fn commit(self) -> Result<(), Error> { - match self { + pub fn cursor<'a>(&'a mut self, db: &'a Database) -> Result, Error> { + match (self, db) { #[cfg(feature = "mdbx")] - Self::Mdbx(txn) => txn.commit(), + (Self::Mdbx(txn), Database::Mdbx(db)) => txn.cursor(db).map(Cursor::Mdbx), #[cfg(feature = "lmdb")] - Self::Lmdb(txn) => txn.commit(), + (Self::Lmdb(txn), Database::Lmdb(db)) => txn.cursor(db).map(Cursor::Lmdb), + #[cfg(feature = "redb")] + (Self::Redb(txn), Database::Redb(db)) => txn.cursor(db).map(Cursor::Redb), _ => Err(Error::MismatchedDatabaseVariant), } } @@ -172,6 +200,8 @@ impl<'env> Cursor<'env> { Cursor::Mdbx(cursor) => cursor.first_key(), #[cfg(feature = "lmdb")] Cursor::Lmdb(cursor) => cursor.first_key(), + #[cfg(feature = "redb")] + Cursor::Redb(cursor) => cursor.first_key(), _ => Err(Error::MismatchedDatabaseVariant), } } @@ -183,6 +213,8 @@ impl<'env> Cursor<'env> { Cursor::Mdbx(cursor) => cursor.last_key(), #[cfg(feature = "lmdb")] Cursor::Lmdb(cursor) => cursor.last_key(), + #[cfg(feature = "redb")] + Cursor::Redb(cursor) => cursor.last_key(), _ => Err(Error::MismatchedDatabaseVariant), } } @@ -193,37 +225,47 @@ impl<'env> Cursor<'env> { Cursor::Mdbx(cursor) => cursor.next_key(), #[cfg(feature = "lmdb")] Cursor::Lmdb(cursor) => cursor.next_key(), + #[cfg(feature = "redb")] + Cursor::Redb(cursor) => cursor.next_key(), _ => Err(Error::MismatchedDatabaseVariant), } } - /// Get the key value pair at the current position. - pub fn get_current(&mut self) -> Result, Error> { + pub fn delete_current(&mut self) -> Result<(), Error> { match self { #[cfg(feature = "mdbx")] - Cursor::Mdbx(cursor) => cursor.get_current(), + Cursor::Mdbx(cursor) => cursor.delete_current(), #[cfg(feature = "lmdb")] - Cursor::Lmdb(cursor) => cursor.get_current(), + Cursor::Lmdb(cursor) => cursor.delete_current(), + #[cfg(feature = "redb")] + Cursor::Redb(cursor) => cursor.delete_current(), _ => Err(Error::MismatchedDatabaseVariant), } } - pub fn delete_current(&mut self) -> Result<(), Error> { + pub fn put, V: AsRef<[u8]>>(&mut self, key: K, value: V) -> Result<(), Error> { match self { #[cfg(feature = "mdbx")] - Cursor::Mdbx(cursor) => cursor.delete_current(), + Self::Mdbx(cursor) => cursor.put(key, value), #[cfg(feature = "lmdb")] - Cursor::Lmdb(cursor) => cursor.delete_current(), + Self::Lmdb(cursor) => cursor.put(key, value), + #[cfg(feature = "redb")] + Self::Redb(cursor) => cursor.put(key, value), _ => Err(Error::MismatchedDatabaseVariant), } } - pub fn put, V: AsRef<[u8]>>(&mut self, key: K, value: V) -> Result<(), Error> { + pub fn delete_while( + &mut self, + f: impl Fn(&[u8]) -> Result, + ) -> Result>, Error> { match self { #[cfg(feature = "mdbx")] - Self::Mdbx(cursor) => cursor.put(key, value), + Self::Mdbx(txn) => txn.delete_while(f), #[cfg(feature = "lmdb")] - Self::Lmdb(cursor) => cursor.put(key, value), + Self::Lmdb(txn) => txn.delete_while(f), + #[cfg(feature = "redb")] + Self::Redb(txn) => txn.delete_while(f), _ => Err(Error::MismatchedDatabaseVariant), } } diff --git a/slasher/src/database/lmdb_impl.rs b/slasher/src/database/lmdb_impl.rs index 78deaf17676..20d89a36fb0 100644 --- a/slasher/src/database/lmdb_impl.rs +++ b/slasher/src/database/lmdb_impl.rs @@ -100,7 +100,7 @@ impl Environment { impl<'env> RwTransaction<'env> { pub fn get + ?Sized>( &'env self, - db: &Database<'env>, + db: &'env Database, key: &K, ) -> Result>, Error> { Ok(self.txn.get(db.db, key).optional()?.map(Cow::Borrowed)) @@ -182,6 +182,29 @@ impl<'env> Cursor<'env> { .put(&key, &value, RwTransaction::write_flags())?; Ok(()) } + + pub fn delete_while( + &mut self, + f: impl Fn(&[u8]) -> Result, + ) -> Result>, Error> { + let mut result = vec![]; + + loop { + let (key_bytes, value) = self.get_current()?.ok_or(Error::MissingKey)?; + + if f(&key_bytes)? { + result.push(value); + self.delete_current()?; + if self.next_key()?.is_none() { + break; + } + } else { + break; + } + } + + Ok(result) + } } /// Mix-in trait for loading values from LMDB that may or may not exist. diff --git a/slasher/src/database/mdbx_impl.rs b/slasher/src/database/mdbx_impl.rs index d25f17e7acf..e249de963f6 100644 --- a/slasher/src/database/mdbx_impl.rs +++ b/slasher/src/database/mdbx_impl.rs @@ -113,7 +113,7 @@ impl<'env> RwTransaction<'env> { pub fn get + ?Sized>( &'env self, - db: &Database<'env>, + db: &'env Database, key: &K, ) -> Result>, Error> { Ok(self.txn.get(&db.db, key.as_ref())?) @@ -183,4 +183,27 @@ impl<'env> Cursor<'env> { .put(key.as_ref(), value.as_ref(), RwTransaction::write_flags())?; Ok(()) } + + pub fn delete_while( + &mut self, + f: impl Fn(&[u8]) -> Result, + ) -> Result>, Error> { + let mut result = vec![]; + + loop { + let (key_bytes, value) = self.get_current()?.ok_or(Error::MissingKey)?; + + if f(&key_bytes)? { + result.push(value); + self.delete_current()?; + if self.next_key()?.is_none() { + break; + } + } else { + break; + } + } + + Ok(result) + } } diff --git a/slasher/src/database/redb_impl.rs b/slasher/src/database/redb_impl.rs new file mode 100644 index 00000000000..da7b4e38eda --- /dev/null +++ b/slasher/src/database/redb_impl.rs @@ -0,0 +1,276 @@ +#![cfg(feature = "redb")] +use crate::{ + config::REDB_DATA_FILENAME, + database::{ + interface::{Key, OpenDatabases, Value}, + *, + }, + Config, Error, +}; +use derivative::Derivative; +use redb::{ReadableTable, TableDefinition}; +use std::{borrow::Cow, path::PathBuf}; + +#[derive(Debug)] +pub struct Environment { + _db_count: usize, + db: redb::Database, +} + +#[derive(Debug)] +pub struct Database<'env> { + table_name: String, + _phantom: PhantomData<&'env ()>, +} + +#[derive(Derivative)] +#[derivative(Debug)] +pub struct RwTransaction<'env> { + #[derivative(Debug = "ignore")] + txn: redb::WriteTransaction, + _phantom: PhantomData<&'env ()>, +} + +#[derive(Derivative)] +#[derivative(Debug)] +pub struct Cursor<'env> { + #[derivative(Debug = "ignore")] + txn: &'env redb::WriteTransaction, + db: &'env Database<'env>, + current_key: Option>, +} + +impl Environment { + pub fn new(config: &Config) -> Result { + let db_path = config.database_path.join(REDB_DATA_FILENAME); + let database = redb::Database::create(db_path)?; + + Ok(Environment { + _db_count: MAX_NUM_DBS, + db: database, + }) + } + + pub fn create_databases(&self) -> Result { + let indexed_attestation_db = self.create_table(INDEXED_ATTESTATION_DB)?; + let indexed_attestation_id_db = self.create_table(INDEXED_ATTESTATION_ID_DB)?; + let attesters_db = self.create_table(ATTESTERS_DB)?; + let attesters_max_targets_db = self.create_table(ATTESTERS_MAX_TARGETS_DB)?; + let min_targets_db = self.create_table(MIN_TARGETS_DB)?; + let max_targets_db = self.create_table(MAX_TARGETS_DB)?; + let current_epochs_db = self.create_table(CURRENT_EPOCHS_DB)?; + let proposers_db = self.create_table(PROPOSERS_DB)?; + let metadata_db = self.create_table(METADATA_DB)?; + + Ok(OpenDatabases { + indexed_attestation_db, + indexed_attestation_id_db, + attesters_db, + attesters_max_targets_db, + min_targets_db, + max_targets_db, + current_epochs_db, + proposers_db, + metadata_db, + }) + } + + pub fn create_table<'env>( + &'env self, + table_name: &'env str, + ) -> Result, Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(table_name); + let tx = self.db.begin_write()?; + tx.open_table(table_definition)?; + tx.commit()?; + + Ok(crate::Database::Redb(Database { + table_name: table_name.to_string(), + _phantom: PhantomData, + })) + } + + pub fn filenames(&self, config: &Config) -> Vec { + vec![config.database_path.join(BASE_DB)] + } + + pub fn begin_rw_txn(&self) -> Result { + let mut txn = self.db.begin_write()?; + txn.set_durability(redb::Durability::Eventual); + Ok(RwTransaction { + txn, + _phantom: PhantomData, + }) + } +} + +impl<'env> RwTransaction<'env> { + pub fn get + ?Sized>( + &'env self, + db: &'env Database, + key: &K, + ) -> Result>, Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&db.table_name); + let table = self.txn.open_table(table_definition)?; + let result = table.get(key.as_ref())?; + if let Some(access_guard) = result { + let value = access_guard.value().to_vec(); + Ok(Some(Cow::from(value))) + } else { + Ok(None) + } + } + + pub fn put, V: AsRef<[u8]>>( + &mut self, + db: &Database, + key: K, + value: V, + ) -> Result<(), Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&db.table_name); + let mut table = self.txn.open_table(table_definition)?; + table.insert(key.as_ref(), value.as_ref())?; + + Ok(()) + } + + pub fn del>(&mut self, db: &Database, key: K) -> Result<(), Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&db.table_name); + let mut table = self.txn.open_table(table_definition)?; + table.remove(key.as_ref())?; + + Ok(()) + } + + pub fn commit(self) -> Result<(), Error> { + self.txn.commit()?; + Ok(()) + } + + pub fn cursor<'a>(&'a mut self, db: &'a Database) -> Result, Error> { + Ok(Cursor { + txn: &self.txn, + db, + current_key: None, + }) + } +} + +impl<'env> Cursor<'env> { + pub fn first_key(&mut self) -> Result, Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&self.db.table_name); + let table = self.txn.open_table(table_definition)?; + let first = table + .iter()? + .next() + .map(|x| x.map(|(key, _)| key.value().to_vec())); + + if let Some(owned_key) = first { + let owned_key = owned_key?; + self.current_key = Some(Cow::from(owned_key)); + Ok(self.current_key.clone()) + } else { + Ok(None) + } + } + + pub fn last_key(&mut self) -> Result>, Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&self.db.table_name); + let table = self.txn.open_table(table_definition)?; + let last = table + .iter()? + .next_back() + .map(|x| x.map(|(key, _)| key.value().to_vec())); + + if let Some(owned_key) = last { + let owned_key = owned_key?; + self.current_key = Some(Cow::from(owned_key)); + return Ok(self.current_key.clone()); + } + Ok(None) + } + + pub fn get_current(&self) -> Result, Value<'env>)>, Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&self.db.table_name); + let table = self.txn.open_table(table_definition)?; + if let Some(key) = &self.current_key { + let result = table.get(key.as_ref())?; + + if let Some(access_guard) = result { + let value = access_guard.value().to_vec(); + return Ok(Some((key.clone(), Cow::from(value)))); + } + } + Ok(None) + } + + pub fn next_key(&mut self) -> Result>, Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&self.db.table_name); + let table = self.txn.open_table(table_definition)?; + if let Some(current_key) = &self.current_key { + let range: std::ops::RangeFrom<&[u8]> = current_key..; + + let next = table + .range(range)? + .next() + .map(|x| x.map(|(key, _)| key.value().to_vec())); + + if let Some(owned_key) = next { + let owned_key = owned_key?; + self.current_key = Some(Cow::from(owned_key)); + return Ok(self.current_key.clone()); + } + } + Ok(None) + } + + pub fn delete_current(&self) -> Result<(), Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&self.db.table_name); + let mut table = self.txn.open_table(table_definition)?; + if let Some(key) = &self.current_key { + table.remove(key.as_ref())?; + } + Ok(()) + } + + pub fn delete_while( + &self, + f: impl Fn(&[u8]) -> Result, + ) -> Result>, Error> { + let mut deleted_values = vec![]; + if let Some(current_key) = &self.current_key { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&self.db.table_name); + + let mut table = self.txn.open_table(table_definition)?; + + let deleted = + table.extract_from_if(current_key.as_ref().., |key, _| f(key).unwrap_or(false))?; + + deleted.for_each(|result| { + if let Ok(item) = result { + let value = item.1.value().to_vec(); + deleted_values.push(Cow::from(value)); + } + }) + }; + Ok(deleted_values) + } + + pub fn put, V: AsRef<[u8]>>(&mut self, key: K, value: V) -> Result<(), Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(&self.db.table_name); + let mut table = self.txn.open_table(table_definition)?; + table.insert(key.as_ref(), value.as_ref())?; + + Ok(()) + } +} diff --git a/slasher/src/error.rs b/slasher/src/error.rs index 8d3295b22a8..b2e32f3dcd2 100644 --- a/slasher/src/error.rs +++ b/slasher/src/error.rs @@ -8,6 +8,8 @@ pub enum Error { DatabaseMdbxError(mdbx::Error), #[cfg(feature = "lmdb")] DatabaseLmdbError(lmdb::Error), + #[cfg(feature = "redb")] + DatabaseRedbError(redb::Error), SlasherDatabaseBackendDisabled, MismatchedDatabaseVariant, DatabaseIOError(io::Error), @@ -67,6 +69,7 @@ pub enum Error { MissingIndexedAttestationId, MissingIndexedAttestationIdKey, InconsistentAttestationDataRoot, + MissingKey, } #[cfg(feature = "mdbx")] @@ -89,6 +92,41 @@ impl From for Error { } } +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::TableError) -> Self { + Error::DatabaseRedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::TransactionError) -> Self { + Error::DatabaseRedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::DatabaseError) -> Self { + Error::DatabaseRedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::StorageError) -> Self { + Error::DatabaseRedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::CommitError) -> Self { + Error::DatabaseRedbError(e.into()) + } +} + impl From for Error { fn from(e: io::Error) -> Self { Error::DatabaseIOError(e) diff --git a/slasher/src/lib.rs b/slasher/src/lib.rs index 4d58fa77029..d3a26337d6a 100644 --- a/slasher/src/lib.rs +++ b/slasher/src/lib.rs @@ -1,6 +1,6 @@ #![deny(missing_debug_implementations)] #![cfg_attr( - not(any(feature = "mdbx", feature = "lmdb")), + not(any(feature = "mdbx", feature = "lmdb", feature = "redb")), allow(unused, clippy::drop_non_drop) )] diff --git a/slasher/tests/attester_slashings.rs b/slasher/tests/attester_slashings.rs index 902141d9710..cc6e57d95d7 100644 --- a/slasher/tests/attester_slashings.rs +++ b/slasher/tests/attester_slashings.rs @@ -1,4 +1,4 @@ -#![cfg(any(feature = "mdbx", feature = "lmdb"))] +#![cfg(any(feature = "mdbx", feature = "lmdb", feature = "redb"))] use logging::test_logger; use maplit::hashset; diff --git a/slasher/tests/backend.rs b/slasher/tests/backend.rs index fd1a6ae14f6..c24b861b18b 100644 --- a/slasher/tests/backend.rs +++ b/slasher/tests/backend.rs @@ -1,4 +1,4 @@ -#![cfg(feature = "lmdb")] +#![cfg(any(feature = "lmdb", feature = "redb"))] use slasher::{config::MDBX_DATA_FILENAME, Config, DatabaseBackend, DatabaseBackendOverride}; use std::fs::File; @@ -41,7 +41,7 @@ fn no_override_with_existing_mdbx_db() { } #[test] -#[cfg(all(not(feature = "mdbx"), feature = "lmdb"))] +#[cfg(all(not(feature = "mdbx"), feature = "lmdb", not(feature = "redb")))] fn failed_override_with_existing_mdbx_db() { let tempdir = tempdir().unwrap(); let mut config = Config::new(tempdir.path().into()); @@ -55,3 +55,19 @@ fn failed_override_with_existing_mdbx_db() { ); assert_eq!(config.backend, DatabaseBackend::Lmdb); } + +#[test] +#[cfg(feature = "redb")] +fn failed_override_with_existing_mdbx_db() { + let tempdir = tempdir().unwrap(); + let mut config = Config::new(tempdir.path().into()); + + let filename = config.database_path.join(MDBX_DATA_FILENAME); + File::create(&filename).unwrap(); + + assert_eq!( + config.override_backend(), + DatabaseBackendOverride::Failure(filename) + ); + assert_eq!(config.backend, DatabaseBackend::Redb); +} diff --git a/slasher/tests/proposer_slashings.rs b/slasher/tests/proposer_slashings.rs index 2d2738087dc..6d2a1f5176b 100644 --- a/slasher/tests/proposer_slashings.rs +++ b/slasher/tests/proposer_slashings.rs @@ -1,4 +1,4 @@ -#![cfg(any(feature = "mdbx", feature = "lmdb"))] +#![cfg(any(feature = "mdbx", feature = "lmdb", feature = "redb"))] use logging::test_logger; use slasher::{ diff --git a/slasher/tests/random.rs b/slasher/tests/random.rs index ebfe0ef4e97..0aaaa63f65c 100644 --- a/slasher/tests/random.rs +++ b/slasher/tests/random.rs @@ -1,4 +1,4 @@ -#![cfg(any(feature = "mdbx", feature = "lmdb"))] +#![cfg(any(feature = "mdbx", feature = "lmdb", feature = "redb"))] use logging::test_logger; use rand::prelude::*; diff --git a/slasher/tests/wrap_around.rs b/slasher/tests/wrap_around.rs index 9a42aeb60ba..2ec56bc7d5d 100644 --- a/slasher/tests/wrap_around.rs +++ b/slasher/tests/wrap_around.rs @@ -1,4 +1,4 @@ -#![cfg(any(feature = "mdbx", feature = "lmdb"))] +#![cfg(any(feature = "mdbx", feature = "lmdb", feature = "redb"))] use logging::test_logger; use slasher::{