chore: adjust to the sequential processing changes in snarkVM

Signed-off-by: ljedrz <[email protected]>

Author: ljedrz

cli/src/commands/start.rs

@@ -55,7 +55,10 @@ use std::{
     path::PathBuf,
     sync::{Arc, atomic::AtomicBool},
 };
-use tokio::runtime::{self, Runtime};
+use tokio::{
+    runtime::{self, Runtime},
+    task,
+};
 use tracing::warn;
 use ureq::http;
 
@@ -617,7 +620,8 @@ impl Start {
         let cdn = self.parse_cdn::<N>().with_context(|| "Failed to parse given CDN URL")?;
 
         // Parse the genesis block.
-        let genesis = self.parse_genesis::<N>()?;
+        let start = self.clone();
+        let genesis = task::spawn_blocking(move || start.parse_genesis::<N>()).await??;
         // Parse the private key of the node.
         let account = self.parse_private_key::<N>()?;
         // Parse the node type.

node/bft/src/sync/mod.rs

@@ -1054,8 +1054,9 @@ mod tests {
 
         // Create a genesis block with a seeded RNG to reproduce the same genesis private keys.
         let seed: u64 = rng.r#gen();
-        let genesis_rng = &mut TestRng::from_seed(seed);
-        let genesis = VM::from(store).unwrap().genesis_beacon(account.private_key(), genesis_rng).unwrap();
+        let vm = VM::from(store).unwrap();
+        let genesis_pk = *account.private_key();
+        let genesis = spawn_blocking!(vm.genesis_beacon(&genesis_pk, &mut TestRng::from_seed(seed))).unwrap();
 
         // Extract the private keys from the genesis committee by using the same RNG to sample private keys.
         let genesis_rng = &mut TestRng::from_seed(seed);
@@ -1067,7 +1068,8 @@ mod tests {
         ];
 
         // Initialize the ledger with the genesis block.
-        let ledger = CurrentLedger::load(genesis.clone(), StorageMode::new_test(None)).unwrap();
+        let genesis_clone = genesis.clone();
+        let ledger = spawn_blocking!(CurrentLedger::load(genesis_clone, StorageMode::new_test(None))).unwrap();
         // Initialize the ledger.
         let core_ledger = Arc::new(CoreLedgerService::new(ledger.clone(), Default::default()));
 
@@ -1180,10 +1182,13 @@ mod tests {
             subdag_map.insert(commit_round, leader_cert_map.clone());
             subdag_map.insert(commit_round - 1, previous_cert_map.clone());
             let subdag = Subdag::from(subdag_map.clone())?;
-            core_ledger.prepare_advance_to_next_quorum_block(subdag, Default::default())?
+            let ledger = core_ledger.clone();
+            spawn_blocking!(ledger.prepare_advance_to_next_quorum_block(subdag, Default::default()))?
         };
         // Insert block 1.
-        core_ledger.advance_to_next_block(&block_1)?;
+        let ledger = core_ledger.clone();
+        let block = block_1.clone();
+        spawn_blocking!(ledger.advance_to_next_block(&block))?;
 
         // Create block 2.
         let leader_round_2 = commit_round + 2;
@@ -1207,10 +1212,13 @@ mod tests {
             subdag_map_2.insert(leader_round_2 - 1, previous_cert_map_2.clone());
             subdag_map_2.insert(leader_round_2 - 2, prev_commit_cert_map_2.clone());
             let subdag_2 = Subdag::from(subdag_map_2.clone())?;
-            core_ledger.prepare_advance_to_next_quorum_block(subdag_2, Default::default())?
+            let ledger = core_ledger.clone();
+            spawn_blocking!(ledger.prepare_advance_to_next_quorum_block(subdag_2, Default::default()))?
         };
         // Insert block 2.
-        core_ledger.advance_to_next_block(&block_2)?;
+        let ledger = core_ledger.clone();
+        let block = block_2.clone();
+        spawn_blocking!(ledger.advance_to_next_block(&block))?;
 
         // Create block 3
         let leader_round_3 = commit_round + 4;
@@ -1234,24 +1242,25 @@ mod tests {
             subdag_map_3.insert(leader_round_3 - 1, previous_cert_map_3.clone());
             subdag_map_3.insert(leader_round_3 - 2, prev_commit_cert_map_3.clone());
             let subdag_3 = Subdag::from(subdag_map_3.clone())?;
-            core_ledger.prepare_advance_to_next_quorum_block(subdag_3, Default::default())?
+            let ledger = core_ledger.clone();
+            spawn_blocking!(ledger.prepare_advance_to_next_quorum_block(subdag_3, Default::default()))?
         };
         // Insert block 3.
-        core_ledger.advance_to_next_block(&block_3)?;
+        let ledger = core_ledger.clone();
+        let block = block_3.clone();
+        spawn_blocking!(ledger.advance_to_next_block(&block))?;
 
         // Initialize the syncing ledger.
-        let storage_mode = StorageMode::new_test(None);
-        let syncing_ledger = Arc::new(CoreLedgerService::new(
-            CurrentLedger::load(genesis, storage_mode.clone()).unwrap(),
-            Default::default(),
-        ));
+        let syncing_ledger = spawn_blocking!(CurrentLedger::load(genesis, StorageMode::new_test(None))).unwrap();
+        let syncing_ledger = Arc::new(CoreLedgerService::new(syncing_ledger, Default::default()));
         // Initialize the gateway.
+        let storage_mode = StorageMode::new_test(None);
         let gateway =
             Gateway::new(account.clone(), storage.clone(), syncing_ledger.clone(), None, &[], storage_mode, None)?;
         // Initialize the block synchronization logic.
         let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone()));
         // Initialize the sync module.
-        let sync = Sync::new(gateway.clone(), storage.clone(), syncing_ledger.clone(), block_sync);
+        let sync = Sync::new(gateway, storage, syncing_ledger.clone(), block_sync);
         // Try to sync block 1.
         sync.sync_storage_with_block(block_1).await?;
         assert_eq!(syncing_ledger.latest_block_height(), 1);
@@ -1282,8 +1291,9 @@ mod tests {
 
         // Create a genesis block with a seeded RNG to reproduce the same genesis private keys.
         let seed: u64 = rng.r#gen();
-        let genesis_rng = &mut TestRng::from_seed(seed);
-        let genesis = VM::from(store).unwrap().genesis_beacon(account.private_key(), genesis_rng).unwrap();
+        let vm = VM::from(store).unwrap();
+        let genesis_pk = *account.private_key();
+        let genesis = spawn_blocking!(vm.genesis_beacon(&genesis_pk, &mut TestRng::from_seed(seed))).unwrap();
 
         // Extract the private keys from the genesis committee by using the same RNG to sample private keys.
         let genesis_rng = &mut TestRng::from_seed(seed);
@@ -1294,13 +1304,13 @@ mod tests {
             PrivateKey::new(genesis_rng)?,
         ];
         // Initialize the ledger with the genesis block.
-        let ledger = CurrentLedger::load(genesis.clone(), StorageMode::new_test(None)).unwrap();
+        let ledger = spawn_blocking!(CurrentLedger::load(genesis, StorageMode::new_test(None))).unwrap();
         // Initialize the ledger.
-        let core_ledger = Arc::new(CoreLedgerService::new(ledger.clone(), Default::default()));
+        let core_ledger = Arc::new(CoreLedgerService::new(ledger, Default::default()));
         // Sample rounds of batch certificates starting at the genesis round from a static set of 4 authors.
         let (round_to_certificates_map, committee) = {
             // Initialize the committee.
-            let committee = ledger.latest_committee().unwrap();
+            let committee = core_ledger.current_committee().unwrap();
             // Initialize a mapping from the round number to the set of batch certificates in the round.
             let mut round_to_certificates_map: HashMap<u64, IndexSet<BatchCertificate<CurrentNetwork>>> =
                 HashMap::new();
@@ -1369,10 +1379,13 @@ mod tests {
             subdag_map.insert(commit_round, leader_cert_map.clone());
             subdag_map.insert(commit_round - 1, previous_cert_map.clone());
             let subdag = Subdag::from(subdag_map.clone())?;
-            core_ledger.prepare_advance_to_next_quorum_block(subdag, Default::default())?
+            let ledger = core_ledger.clone();
+            spawn_blocking!(ledger.prepare_advance_to_next_quorum_block(subdag, Default::default()))?
         };
         // Insert block 1.
-        core_ledger.advance_to_next_block(&block_1)?;
+        let ledger = core_ledger.clone();
+        let block = block_1.clone();
+        spawn_blocking!(ledger.advance_to_next_block(&block))?;
 
         // Create block 2.
         let leader_round_2 = commit_round + 2;
@@ -1389,10 +1402,13 @@ mod tests {
             subdag_map_2.insert(leader_round_2, leader_cert_map_2.clone());
             subdag_map_2.insert(leader_round_2 - 1, previous_cert_map_2.clone());
             let subdag_2 = Subdag::from(subdag_map_2.clone())?;
-            core_ledger.prepare_advance_to_next_quorum_block(subdag_2, Default::default())?
+            let ledger = core_ledger.clone();
+            spawn_blocking!(ledger.prepare_advance_to_next_quorum_block(subdag_2, Default::default()))?
         };
         // Insert block 2.
-        core_ledger.advance_to_next_block(&block_2)?;
+        let ledger = core_ledger.clone();
+        let block = block_2.clone();
+        spawn_blocking!(ledger.advance_to_next_block(&block))?;
 
         // Create block 3
         let leader_round_3 = commit_round + 4;
@@ -1409,10 +1425,13 @@ mod tests {
             subdag_map_3.insert(leader_round_3, leader_cert_map_3.clone());
             subdag_map_3.insert(leader_round_3 - 1, previous_cert_map_3.clone());
             let subdag_3 = Subdag::from(subdag_map_3.clone())?;
-            core_ledger.prepare_advance_to_next_quorum_block(subdag_3, Default::default())?
+            let ledger = core_ledger.clone();
+            spawn_blocking!(ledger.prepare_advance_to_next_quorum_block(subdag_3, Default::default()))?
         };
         // Insert block 3.
-        core_ledger.advance_to_next_block(&block_3)?;
+        let ledger = core_ledger.clone();
+        let block = block_3.clone();
+        spawn_blocking!(ledger.advance_to_next_block(&block))?;
 
         /*
             Check that the pending certificates are computed correctly.
@@ -1443,6 +1462,7 @@ mod tests {
         }
         // Check that the set of pending certificates is equal to the set of candidate pending certificates.
         assert_eq!(pending_certificates, candidate_pending_certificates);
+
         Ok(())
     }
 }

node/bft/tests/bft_e2e.rs

@@ -31,15 +31,19 @@ async fn test_state_coherence() {
     const N: u16 = 4;
     const TRANSMISSION_INTERVAL_MS: u64 = 10;
 
-    let mut network = TestNetwork::new(TestNetworkConfig {
-        num_nodes: N,
-        bft: true,
-        connect_all: true,
-        fire_transmissions: Some(TRANSMISSION_INTERVAL_MS),
-        // Set this to Some(0..=4) to see the logs.
-        log_level: Some(0),
-        log_connections: true,
-    });
+    let mut network = tokio::task::spawn_blocking(|| {
+        TestNetwork::new(TestNetworkConfig {
+            num_nodes: N,
+            bft: true,
+            connect_all: true,
+            fire_transmissions: Some(TRANSMISSION_INTERVAL_MS),
+            // Set this to Some(0..=4) to see the logs.
+            log_level: Some(0),
+            log_connections: true,
+        })
+    })
+    .await
+    .unwrap();
 
     network.start().await;
 
@@ -52,15 +56,19 @@ async fn test_resync() {
     // Start N nodes, connect them and start the cannons for each.
     const N: u16 = 4;
     const TRANSMISSION_INTERVAL_MS: u64 = 10;
-    let mut network = TestNetwork::new(TestNetworkConfig {
-        num_nodes: N,
-        bft: true,
-        connect_all: true,
-        fire_transmissions: Some(TRANSMISSION_INTERVAL_MS),
-        // Set this to Some(0..=4) to see the logs.
-        log_level: Some(0),
-        log_connections: false,
-    });
+    let mut network = tokio::task::spawn_blocking(|| {
+        TestNetwork::new(TestNetworkConfig {
+            num_nodes: N,
+            bft: true,
+            connect_all: true,
+            fire_transmissions: Some(TRANSMISSION_INTERVAL_MS),
+            // Set this to Some(0..=4) to see the logs.
+            log_level: Some(0),
+            log_connections: false,
+        })
+    })
+    .await
+    .unwrap();
     network.start().await;
 
     // Let the nodes advance through the rounds.
@@ -98,15 +106,19 @@ async fn test_quorum_threshold() {
     const N: u16 = 4;
     const TRANSMISSION_INTERVAL_MS: u64 = 10;
 
-    let mut network = TestNetwork::new(TestNetworkConfig {
-        num_nodes: N,
-        bft: true,
-        connect_all: false,
-        fire_transmissions: None,
-        // Set this to Some(0..=4) to see the logs.
-        log_level: None,
-        log_connections: true,
-    });
+    let mut network = tokio::task::spawn_blocking(|| {
+        TestNetwork::new(TestNetworkConfig {
+            num_nodes: N,
+            bft: true,
+            connect_all: false,
+            fire_transmissions: None,
+            // Set this to Some(0..=4) to see the logs.
+            log_level: None,
+            log_connections: true,
+        })
+    })
+    .await
+    .unwrap();
     network.start().await;
 
     // Check each node is at round 1 (0 is genesis).
@@ -142,23 +154,28 @@ async fn test_quorum_threshold() {
 
     // Check the nodes reach quorum and advance through the rounds.
     const TARGET_ROUND: u64 = 4;
-    deadline!(Duration::from_secs(20), move || { network.is_round_reached(TARGET_ROUND) });
+    let net = network.clone();
+    deadline!(Duration::from_secs(20), move || { net.is_round_reached(TARGET_ROUND) });
 }
 
 #[tokio::test(flavor = "multi_thread")]
 async fn test_quorum_break() {
     // Start N nodes, connect them and start the cannons for each.
     const N: u16 = 4;
     const TRANSMISSION_INTERVAL_MS: u64 = 10;
-    let mut network = TestNetwork::new(TestNetworkConfig {
-        num_nodes: N,
-        bft: true,
-        connect_all: true,
-        fire_transmissions: Some(TRANSMISSION_INTERVAL_MS),
-        // Set this to Some(0..=4) to see the logs.
-        log_level: None,
-        log_connections: true,
-    });
+    let mut network = tokio::task::spawn_blocking(|| {
+        TestNetwork::new(TestNetworkConfig {
+            num_nodes: N,
+            bft: true,
+            connect_all: true,
+            fire_transmissions: Some(TRANSMISSION_INTERVAL_MS),
+            // Set this to Some(0..=4) to see the logs.
+            log_level: None,
+            log_connections: true,
+        })
+    })
+    .await
+    .unwrap();
     network.start().await;
 
     // Check the nodes have started advancing through the rounds.
@@ -186,15 +203,19 @@ async fn test_leader_election_consistency() {
     // Start N nodes, connect them and start the cannons for each.
     const N: u16 = 4;
     const CANNON_INTERVAL_MS: u64 = 10;
-    let mut network = TestNetwork::new(TestNetworkConfig {
-        num_nodes: N,
-        bft: true,
-        connect_all: true,
-        fire_transmissions: Some(CANNON_INTERVAL_MS),
-        // Set this to Some(0..=4) to see the logs.
-        log_level: None,
-        log_connections: true,
-    });
+    let mut network = tokio::task::spawn_blocking(|| {
+        TestNetwork::new(TestNetworkConfig {
+            num_nodes: N,
+            bft: true,
+            connect_all: true,
+            fire_transmissions: Some(CANNON_INTERVAL_MS),
+            // Set this to Some(0..=4) to see the logs.
+            log_level: None,
+            log_connections: true,
+        })
+    })
+    .await
+    .unwrap();
     network.start().await;
 
     // Wait for starting round to be reached
@@ -238,15 +259,19 @@ async fn test_transient_break() {
     // Start N nodes, connect them and start the cannons for each.
     const N: u16 = 4;
     const TRANSMISSION_INTERVAL_MS: u64 = 10;
-    let mut network = TestNetwork::new(TestNetworkConfig {
-        num_nodes: N,
-        bft: true,
-        connect_all: true,
-        fire_transmissions: Some(TRANSMISSION_INTERVAL_MS),
-        // Set this to Some(0..=4) to see the logs.
-        log_level: Some(6),
-        log_connections: false,
-    });
+    let mut network = tokio::task::spawn_blocking(|| {
+        TestNetwork::new(TestNetworkConfig {
+            num_nodes: N,
+            bft: true,
+            connect_all: true,
+            fire_transmissions: Some(TRANSMISSION_INTERVAL_MS),
+            // Set this to Some(0..=4) to see the logs.
+            log_level: Some(6),
+            log_connections: false,
+        })
+    })
+    .await
+    .unwrap();
     network.start().await;
 
     // Check the nodes have started advancing through the rounds.