Adjust DB transactions to be one per log batch (sigp#331)

Author: Zacholme7

Cargo.lock

@@ -1,4 +1,4 @@
-use rusqlite::params;
+use rusqlite::{Transaction, params};
 use ssv_types::{Cluster, ClusterId, OperatorId, Share, ValidatorMetadata};
 use types::{Address, PublicKeyBytes};
 
@@ -13,10 +13,8 @@ impl NetworkDatabase {
         cluster: Cluster,
         validator: ValidatorMetadata,
         shares: Vec<Share>,
+        tx: &Transaction<'_>,
     ) -> Result<(), DatabaseError> {
-        let mut conn = self.connection()?;
-        let tx = conn.transaction()?;
-
         // Insert the top level cluster data if it does not exist, and the associated validator
         // metadata
         tx.prepare_cached(SQL[&SqlStatement::InsertCluster])?
@@ -51,12 +49,9 @@ impl NetworkDatabase {
             // Insert the cluster member and the share
             tx.prepare_cached(SQL[&SqlStatement::InsertClusterMember])?
                 .execute(params![*share.cluster_id, *share.operator_id])?;
-            self.insert_share(&tx, share, &validator.public_key)
+            self.insert_share(tx, share, &validator.public_key)
         })?;
 
-        // Commit all operations to the db
-        tx.commit()?;
-
         self.modify_state(|state| {
             // If we are a member in this cluster, store membership and our share
             if let Some(share) = our_share {
@@ -96,9 +91,13 @@ impl NetworkDatabase {
     }
 
     /// Mark the cluster as liquidated or active
-    pub fn update_status(&self, cluster_id: ClusterId, status: bool) -> Result<(), DatabaseError> {
-        let conn = self.connection()?;
-        conn.prepare_cached(SQL[&SqlStatement::UpdateClusterStatus])?
+    pub fn update_status(
+        &self,
+        cluster_id: ClusterId,
+        status: bool,
+        tx: &Transaction<'_>,
+    ) -> Result<(), DatabaseError> {
+        tx.prepare_cached(SQL[&SqlStatement::UpdateClusterStatus])?
             .execute(params![
                 status,      // status of the cluster (liquidated = false, active = true)
                 *cluster_id  // Id of the cluster
@@ -118,10 +117,13 @@ impl NetworkDatabase {
     /// Delete a validator from a cluster. This will cascade and remove all corresponding share
     /// data for this validator. If this validator is the last one in the cluster, the cluster
     /// and all corresponding cluster members will also be removed
-    pub fn delete_validator(&self, validator_pubkey: &PublicKeyBytes) -> Result<(), DatabaseError> {
+    pub fn delete_validator(
+        &self,
+        validator_pubkey: &PublicKeyBytes,
+        tx: &Transaction<'_>,
+    ) -> Result<(), DatabaseError> {
         // Remove from database
-        let conn = self.connection()?;
-        conn.prepare_cached(SQL[&SqlStatement::DeleteValidator])?
+        tx.prepare_cached(SQL[&SqlStatement::DeleteValidator])?
             .execute(params![validator_pubkey.to_string()])?;
 
         self.modify_state(|state| {
@@ -150,10 +152,13 @@ impl NetworkDatabase {
     }
 
     /// Bump the nonce of the owner
-    pub fn bump_and_get_nonce(&self, owner: &Address) -> Result<u16, DatabaseError> {
+    pub fn bump_and_get_nonce(
+        &self,
+        owner: &Address,
+        tx: &Transaction<'_>,
+    ) -> Result<u16, DatabaseError> {
         // bump the nonce in the db
-        let conn = self.connection()?;
-        conn.prepare_cached(SQL[&SqlStatement::BumpNonce])?
+        tx.prepare_cached(SQL[&SqlStatement::BumpNonce])?
             .execute(params![owner.to_string()])?;
 
         let mut nonce = 0;

anchor/database/src/cluster_operations.rs

@@ -7,7 +7,7 @@ use std::{
 
 use openssl::{pkey::Public, rsa::Rsa};
 use r2d2_sqlite::SqliteConnectionManager;
-use rusqlite::params;
+use rusqlite::{Transaction, params};
 use ssv_types::{Cluster, ClusterId, CommitteeId, Operator, OperatorId, Share, ValidatorMetadata};
 use tokio::sync::{
     watch,
@@ -166,9 +166,12 @@ impl NetworkDatabase {
 
     /// Update the last processed block number in the database
     /// Also, trigger a notification for other code to act on the new state
-    pub fn processed_block(&self, block_number: u64) -> Result<(), DatabaseError> {
-        let conn = self.connection()?;
-        conn.prepare_cached(SQL[&SqlStatement::UpdateBlockNumber])?
+    pub fn processed_block(
+        &self,
+        block_number: u64,
+        tx: &Transaction<'_>,
+    ) -> Result<(), DatabaseError> {
+        tx.prepare_cached(SQL[&SqlStatement::UpdateBlockNumber])?
             .execute(params![block_number])?;
         self.state
             .send_modify(|state| state.single_state.last_processed_block = block_number);
@@ -213,7 +216,7 @@ impl NetworkDatabase {
     }
 
     // Open a new connection
-    fn connection(&self) -> Result<PoolConn, DatabaseError> {
+    pub fn connection(&self) -> Result<PoolConn, DatabaseError> {
         Ok(self.conn_pool.get()?)
     }
 

anchor/database/src/lib.rs

@@ -1,13 +1,17 @@
 use base64::prelude::*;
-use rusqlite::params;
+use rusqlite::{Transaction, params};
 use ssv_types::{Operator, OperatorId};
 
 use super::{DatabaseError, NetworkDatabase, PubkeyOrId, SQL, SqlStatement};
 
 /// Implements all operator related functionality on the database
 impl NetworkDatabase {
     /// Insert a new Operator into the database
-    pub fn insert_operator(&self, operator: &Operator) -> Result<(), DatabaseError> {
+    pub fn insert_operator(
+        &self,
+        operator: &Operator,
+        tx: &Transaction<'_>,
+    ) -> Result<(), DatabaseError> {
         // 1ake sure that this operator does not already exist
         if self.state().operator_exists(&operator.id) {
             return Err(DatabaseError::NotFound(format!(
@@ -24,8 +28,7 @@ impl NetworkDatabase {
         let encoded = BASE64_STANDARD.encode(pem_key.clone());
 
         // Insert into the database
-        let conn = self.connection()?;
-        conn.prepare_cached(SQL[&SqlStatement::InsertOperator])?
+        tx.prepare_cached(SQL[&SqlStatement::InsertOperator])?
             .execute(params![
                 *operator.id,               // The id of the registered operator
                 encoded,                    // RSA public key
@@ -56,7 +59,11 @@ impl NetworkDatabase {
     }
 
     /// Delete an operator
-    pub fn delete_operator(&self, id: OperatorId) -> Result<(), DatabaseError> {
+    pub fn delete_operator(
+        &self,
+        id: OperatorId,
+        tx: &Transaction<'_>,
+    ) -> Result<(), DatabaseError> {
         // Make sure that this operator exists
         if !self.state().operator_exists(&id) {
             return Err(DatabaseError::NotFound(format!(
@@ -67,8 +74,7 @@ impl NetworkDatabase {
 
         // Remove from db and in memory. This should cascade to delete this operator from all of the
         // clusters that it is in and all of the shares that it owns
-        let conn = self.connection()?;
-        conn.prepare_cached(SQL[&SqlStatement::DeleteOperator])?
+        tx.prepare_cached(SQL[&SqlStatement::DeleteOperator])?
             .execute(params![*id])?;
 
         self.state.send_modify(|state| {

anchor/database/src/operator_operations.rs

@@ -8,15 +8,15 @@ mod cluster_database_tests {
     // Test inserting a cluster into the database
     fn test_insert_retrieve_cluster() {
         let fixture = TestFixture::new();
-        assertions::cluster::exists_in_db(&fixture.db, &fixture.cluster);
+
+        let mut conn = fixture.db.connection().unwrap();
+        let tx = conn.transaction().unwrap();
+
+        assertions::cluster::exists_in_db(&fixture.cluster, &tx);
         assertions::cluster::exists_in_memory(&fixture.db, &fixture.cluster);
         assertions::validator::exists_in_memory(&fixture.db, &fixture.validator);
-        assertions::validator::exists_in_db(&fixture.db, &fixture.validator);
-        assertions::share::exists_in_db(
-            &fixture.db,
-            &fixture.validator.public_key,
-            &fixture.shares,
-        );
+        assertions::validator::exists_in_db(&fixture.validator, &tx);
+        assertions::share::exists_in_db(&fixture.validator.public_key, &fixture.shares, &tx);
     }
 
     #[test]
@@ -25,14 +25,18 @@ mod cluster_database_tests {
     fn test_delete_last_validator() {
         let fixture = TestFixture::new();
         let pubkey = fixture.validator.public_key;
-        assert!(fixture.db.delete_validator(&pubkey).is_ok());
+
+        let mut conn = fixture.db.connection().unwrap();
+        let tx = conn.transaction().unwrap();
+
+        assert!(fixture.db.delete_validator(&pubkey, &tx).is_ok());
 
         // Since there was only one validator in the cluster, everything should be removed
-        assertions::cluster::exists_not_in_db(&fixture.db, fixture.cluster.cluster_id);
+        assertions::cluster::exists_not_in_db(fixture.cluster.cluster_id, &tx);
         assertions::cluster::exists_not_in_memory(&fixture.db, fixture.cluster.cluster_id);
-        assertions::validator::exists_not_in_db(&fixture.db, &fixture.validator);
+        assertions::validator::exists_not_in_db(&fixture.validator, &tx);
         assertions::validator::exists_not_in_memory(&fixture.db, &fixture.validator);
-        assertions::share::exists_not_in_db(&fixture.db, &pubkey);
+        assertions::share::exists_not_in_db(&pubkey, &tx);
         assertions::share::exists_not_in_memory(&fixture.db, &pubkey);
     }
 
@@ -43,17 +47,20 @@ mod cluster_database_tests {
         let mut cluster = fixture.cluster;
         let new_fee_recipient = Address::random();
 
+        let mut conn = fixture.db.connection().unwrap();
+        let tx = conn.transaction().unwrap();
+
         // Update fee recipient
         assert!(
             fixture
                 .db
-                .update_fee_recipient(cluster.owner, new_fee_recipient)
+                .update_fee_recipient(cluster.owner, new_fee_recipient, &tx)
                 .is_ok()
         );
 
         // assertions will compare the data
         cluster.fee_recipient = new_fee_recipient;
-        assertions::cluster::exists_in_db(&fixture.db, &cluster);
+        assertions::cluster::exists_in_db(&cluster, &tx);
         assertions::cluster::exists_in_memory(&fixture.db, &cluster);
     }
 
@@ -68,9 +75,11 @@ mod cluster_database_tests {
             OperatorId(1),
             &fixture.validator.public_key,
         )];
+        let mut conn = fixture.db.connection().unwrap();
+        let tx = conn.transaction().unwrap();
         fixture
             .db
-            .insert_validator(cluster, metadata, shares)
+            .insert_validator(cluster, metadata, shares, &tx)
             .expect_err("Insertion should fail");
     }
 
@@ -80,25 +89,30 @@ mod cluster_database_tests {
         let fixture = TestFixture::new();
         let mut cluster = fixture.cluster;
 
+        let mut conn = fixture.db.connection().unwrap();
+        let tx = conn.transaction().unwrap();
+
         // Test updating to liquidated
         fixture
             .db
-            .update_status(cluster.cluster_id, true)
+            .update_status(cluster.cluster_id, true, &tx)
             .expect("Failed to update cluster status");
 
         // verify in memory and db
         cluster.liquidated = true;
-        assertions::cluster::exists_in_db(&fixture.db, &cluster);
+        assertions::cluster::exists_in_db(&cluster, &tx);
         assertions::cluster::exists_in_memory(&fixture.db, &cluster);
     }
 
     #[test]
     // Test inserting a cluster that already exists
     fn test_duplicate_cluster_insert() {
         let fixture = TestFixture::new();
+        let mut conn = fixture.db.connection().unwrap();
+        let tx = conn.transaction().unwrap();
         fixture
             .db
-            .insert_validator(fixture.cluster, fixture.validator, fixture.shares)
+            .insert_validator(fixture.cluster, fixture.validator, fixture.shares, &tx)
             .expect_err("Expected failure when inserting cluster that already exists");
     }
 
@@ -108,26 +122,35 @@ mod cluster_database_tests {
         let fixture = TestFixture::new();
         let mut cluster = fixture.cluster;
 
+        let mut conn = fixture.db.connection().unwrap();
+        let tx = conn.transaction().unwrap();
+
         // Confirm that the fee recipient was inserted when the cluster was made
-        let fee_recipient = fixture.db.fee_recipient_for_owner(&cluster.owner).unwrap();
+        let fee_recipient = fixture
+            .db
+            .fee_recipient_for_owner(&cluster.owner, &tx)
+            .unwrap();
         assert_eq!(fee_recipient, Some(cluster.fee_recipient));
 
         // Update fee recipient
         let new_fee_recipient = Address::random();
         assert!(
             fixture
                 .db
-                .update_fee_recipient(cluster.owner, new_fee_recipient)
+                .update_fee_recipient(cluster.owner, new_fee_recipient, &tx)
                 .is_ok()
         );
 
         // Confirm that fee recipient was updated
         cluster.fee_recipient = new_fee_recipient;
-        assertions::cluster::exists_in_db(&fixture.db, &cluster);
+        assertions::cluster::exists_in_db(&cluster, &tx);
         assertions::cluster::exists_in_memory(&fixture.db, &cluster);
 
         // Confirm that we have set the correct fee recipient for the owner
-        let stored_fee_recipient = fixture.db.fee_recipient_for_owner(&cluster.owner).unwrap();
+        let stored_fee_recipient = fixture
+            .db
+            .fee_recipient_for_owner(&cluster.owner, &tx)
+            .unwrap();
         assert_eq!(stored_fee_recipient, Some(new_fee_recipient));
     }
 }