Sqlite slasher backend impl

lighthouse/Cargo.toml

@@ -7,7 +7,7 @@ autotests = false
 rust-version = "1.69.0"
 
 [features]
-default = ["slasher-lmdb"]
+default = ["slasher-sqlite"]
 # Writes debugging .ssz files to /tmp during block processing.
 write_ssz_files = ["beacon_node/write_ssz_files"]
 # Compiles the BLS crypto code so that the binary is portable across machines.
@@ -24,6 +24,8 @@ gnosis = []
 slasher-mdbx = ["slasher/mdbx"]
 # Support slasher LMDB backend.
 slasher-lmdb = ["slasher/lmdb"]
+# Support slasher LMDB backend.
+slasher-sqlite = ["slasher/sqlite"]
 # Use jemalloc.
 jemalloc = ["malloc_utils/jemalloc"]
 

lighthouse/tests/beacon_node.rs

@@ -2069,11 +2069,11 @@ fn slasher_backend_override_to_default() {
     CommandLineTest::new()
         .flag("slasher", None)
         .flag("slasher-max-db-size", Some("1"))
-        .flag("slasher-backend", Some("lmdb"))
+        .flag("slasher-backend", Some("sqlite"))
         .run_with_zero_port()
         .with_config(|config| {
             let slasher_config = config.slasher.as_ref().unwrap();
-            assert_eq!(slasher_config.backend, slasher::DatabaseBackend::Lmdb);
+            assert_eq!(slasher_config.backend, slasher::DatabaseBackend::Sqlite);
         });
 }
 

slasher/Cargo.toml

@@ -5,13 +5,15 @@ authors = ["Michael Sproul <[email protected]>"]
 edition = { workspace = true }
 
 [features]
-default = ["lmdb"]
+default = ["sqlite"]
 mdbx = ["dep:mdbx"]
 lmdb = ["lmdb-rkv", "lmdb-rkv-sys"]
+sqlite = ["dep:rusqlite"]
 
 [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 }
@@ -31,6 +33,10 @@ tree_hash_derive = { workspace = true }
 types = { workspace = true }
 strum = { workspace = true }
 
+rusqlite = { version = "*", optional = true}
+r2d2 = "0.8.10"
+r2d2_sqlite = "0.21.0"
+
 # MDBX is pinned at the last version with Windows and macOS support.
 mdbx = { package = "libmdbx", git = "https://github.com/sigp/libmdbx-rs", tag = "v0.1.4", optional = true }
 lmdb-rkv = { git = "https://github.com/sigp/lmdb-rs", rev = "f33845c6469b94265319aac0ed5085597862c27e", optional = true }

slasher/src/config.rs

@@ -17,7 +17,9 @@ pub const DEFAULT_BROADCAST: bool = false;
 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(feature = "sqlite")]
+pub const DEFAULT_BACKEND: DatabaseBackend = DatabaseBackend::Sqlite;
+#[cfg(not(any(feature = "mdbx", feature = "lmdb", feature = "sqlite")))]
 pub const DEFAULT_BACKEND: DatabaseBackend = DatabaseBackend::Disabled;
 
 pub const MAX_HISTORY_LENGTH: usize = 1 << 16;
@@ -62,6 +64,8 @@ pub enum DatabaseBackend {
     Mdbx,
     #[cfg(feature = "lmdb")]
     Lmdb,
+    #[cfg(feature = "sqlite")]
+    Sqlite,
     Disabled,
 }
 

slasher/src/database.rs

@@ -1,6 +1,7 @@
 pub mod interface;
 mod lmdb_impl;
 mod mdbx_impl;
+mod sqlite_impl;
 
 use crate::{
     metrics, AttesterRecord, AttesterSlashingStatus, CompactAttesterRecord, Config, Error,
@@ -272,7 +273,6 @@ impl<E: EthSpec> SlasherDB<E> {
         };
 
         db = db.migrate()?;
-
         let mut txn = db.begin_rw_txn()?;
         if let Some(on_disk_config) = db.load_config(&mut txn)? {
             let current_disk_config = db.config.disk_config();
@@ -448,9 +448,8 @@ 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 indexed_att_id = match cursor.last_key()? {
+        let indexed_att_id = match txn.last_key(db)? {
             // First ID is 1 so that 0 can be used to represent `null` in `CompactAttesterRecord`.
             None => 1,
             Some(key_bytes) => IndexedAttestationId::parse(key_bytes)? + 1,
@@ -459,8 +458,7 @@ impl<E: EthSpec> SlasherDB<E> {
         let attestation_key = IndexedAttestationId::new(indexed_att_id);
         let data = indexed_attestation.as_ssz_bytes();
 
-        cursor.put(attestation_key.as_ref(), &data)?;
-        drop(cursor);
+        txn.put(db, attestation_key.as_ref(), &data)?;
 
         // Update the (epoch, hash) to ID mapping.
         self.put_indexed_attestation_id(txn, &id_key, attestation_key)?;
@@ -692,28 +690,24 @@ impl<E: EthSpec> SlasherDB<E> {
             .saturating_sub(self.config.history_length)
             .start_slot(E::slots_per_epoch());
 
-        let mut cursor = txn.cursor(&self.databases.proposers_db)?;
+        let db = &self.databases.proposers_db;
 
         // Position cursor at first key, bailing out if the database is empty.
-        if cursor.first_key()?.is_none() {
+        if txn.first_key(db)?.is_none() {
             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)
+        };
+
+        txn.delete_while(&self.databases.proposers_db, should_delete)?;
 
         Ok(())
     }
@@ -727,38 +721,31 @@ 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)?;
+        let db = &self.databases.indexed_attestation_id_db;
 
         // Position cursor at first key, bailing out if the database is empty.
-        if cursor.first_key()?.is_none() {
+        if txn.first_key(db)?.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;
-            }
-        }
-        drop(cursor);
+        let indexed_attestation_ids: Vec<IndexedAttestationId> = txn
+            .delete_while(&self.databases.proposers_db, should_delete)?
+            .into_iter()
+            .map(|value| {
+                IndexedAttestationId::new(
+                    IndexedAttestationId::parse(Cow::from(value)).unwrap_or_default(),
+                )
+            })
+            .collect();
 
         // Delete the indexed attestations.
         // Optimisation potential: use a cursor here.