validator: guard everything behind a single lock

bin/darkfid/src/lib.rs

@@ -138,7 +138,7 @@ impl Darkfid {
         let p2p_handler = DarkfidP2pHandler::init(net_settings, ex).await?;
 
         // Initialize the miners registry
-        let registry = DarkfiMinersRegistry::init(network, &validator)?;
+        let registry = DarkfiMinersRegistry::init(network, &validator).await?;
 
         // Grab blockchain network configured transactions batch size for garbage collection
         let txs_batch_size = match txs_batch_size {
@@ -301,7 +301,8 @@ impl Darkfid {
 
         // Flush sled database data
         info!(target: "darkfid::Darkfid::stop", "Flushing sled database...");
-        let flushed_bytes = self.node.validator.blockchain.sled_db.flush_async().await?;
+        let flushed_bytes =
+            self.node.validator.read().await.blockchain.sled_db.flush_async().await?;
         info!(target: "darkfid::Darkfid::stop", "Flushed {flushed_bytes} bytes");
 
         info!(target: "darkfid::Darkfid::stop", "Darkfi daemon terminated successfully!");

bin/darkfid/src/main.rs

@@ -197,7 +197,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     if let Some(height) = args.reset {
         info!(target: "darkfid", "Node will reset validator state to height: {height}");
         let validator = Validator::new(&sled_db, &config).await?;
-        validator.reset_to_height(height).await?;
+        validator.write().await.reset_to_height(height).await?;
         info!(target: "darkfid", "Validator state reset successfully!");
         return Ok(())
     }
@@ -206,7 +206,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     if args.purge_sync {
         info!(target: "darkfid", "Node will purge all pending sync headers.");
         let validator = Validator::new(&sled_db, &config).await?;
-        validator.blockchain.headers.remove_all_sync()?;
+        validator.read().await.blockchain.headers.remove_all_sync()?;
         info!(target: "darkfid", "Validator pending sync headers purged successfully!");
         return Ok(())
     }
@@ -215,7 +215,11 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     if args.validate {
         info!(target: "darkfid", "Node will validate existing blockchain state.");
         let validator = Validator::new(&sled_db, &config).await?;
-        validator.validate_blockchain(config.pow_target, config.pow_fixed_difficulty).await?;
+        validator
+            .read()
+            .await
+            .validate_blockchain(config.pow_target, config.pow_fixed_difficulty)
+            .await?;
         info!(target: "darkfid", "Validator blockchain state validated successfully!");
         return Ok(())
     }
@@ -225,6 +229,8 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
         info!(target: "darkfid", "Node will rebuild difficulties of existing blockchain state.");
         let validator = Validator::new(&sled_db, &config).await?;
         validator
+            .read()
+            .await
             .rebuild_block_difficulties(config.pow_target, config.pow_fixed_difficulty)
             .await?;
         info!(target: "darkfid", "Validator difficulties rebuilt successfully!");

bin/darkfid/src/proto/protocol_proposal.rs

@@ -189,7 +189,8 @@ async fn handle_receive_proposal(
         };
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let mut validator = validator.write().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_proposal::handle_receive_proposal",
                 "Node still syncing blockchain, skipping..."

bin/darkfid/src/proto/protocol_sync.rs

@@ -452,7 +452,8 @@ async fn handle_receive_tip_request(
         debug!(target: "darkfid::proto::protocol_sync::handle_receive_tip_request", "Received request: {request:?}");
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let validator = validator.read().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
                 "Node still syncing blockchain"
@@ -548,7 +549,8 @@ async fn handle_receive_header_request(
         };
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let validator = validator.read().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_sync::handle_receive_header_request",
                 "Node still syncing blockchain, skipping..."
@@ -599,7 +601,8 @@ async fn handle_receive_request(
         };
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let validator = validator.read().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_sync::handle_receive_request",
                 "Node still syncing blockchain, skipping..."
@@ -660,7 +663,8 @@ async fn handle_receive_fork_request(
         };
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let validator = validator.read().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_sync::handle_receive_fork_request",
                 "Node still syncing blockchain, skipping..."
@@ -715,7 +719,8 @@ async fn handle_receive_fork_header_hash_request(
         };
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let validator = validator.read().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_sync::handle_receive_fork_header_hash_request",
                 "Node still syncing blockchain, skipping..."
@@ -800,7 +805,8 @@ async fn handle_receive_fork_headers_request(
         };
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let validator = validator.read().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_sync::handle_receive_fork_headers_request",
                 "Node still syncing blockchain, skipping..."
@@ -897,7 +903,8 @@ async fn handle_receive_fork_proposals_request(
         };
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let validator = validator.read().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_sync::handle_receive_fork_proposals_request",
                 "Node still syncing blockchain, skipping..."

bin/darkfid/src/proto/protocol_tx.rs

@@ -122,7 +122,8 @@ async fn handle_receive_tx(
         };
 
         // Check if node has finished syncing its blockchain
-        if !*validator.synced.read().await {
+        let mut validator = validator.write().await;
+        if !validator.synced {
             debug!(
                 target: "darkfid::proto::protocol_tx::handle_receive_tx",
                 "Node still syncing blockchain, skipping..."

bin/darkfid/src/registry/mod.rs

@@ -35,7 +35,7 @@ use darkfi::{
     },
     system::{ExecutorPtr, StoppableTask, StoppableTaskPtr},
     util::encoding::base64,
-    validator::{consensus::Proposal, ValidatorPtr},
+    validator::{consensus::Proposal, Validator, ValidatorPtr},
     Error, Result,
 };
 use darkfi_sdk::{
@@ -91,14 +91,17 @@ pub struct DarkfiMinersRegistry {
 
 impl DarkfiMinersRegistry {
     /// Initialize a DarkFi node miners registry.
-    pub fn init(network: Network, validator: &ValidatorPtr) -> Result<DarkfiMinersRegistryPtr> {
+    pub async fn init(
+        network: Network,
+        validator: &ValidatorPtr,
+    ) -> Result<DarkfiMinersRegistryPtr> {
         info!(
             target: "darkfid::registry::mod::DarkfiMinersRegistry::init",
             "Initializing a new DarkFi node miners registry..."
         );
 
         // Generate the PowRewardV1 ZK data
-        let powrewardv1_zk = PowRewardV1Zk::new(validator)?;
+        let powrewardv1_zk = PowRewardV1Zk::new(validator).await?;
 
         // Generate the stratum JSON-RPC background task and its
         // connections tracker.
@@ -224,7 +227,7 @@ impl DarkfiMinersRegistry {
     /// not needed.
     async fn create_template(
         &self,
-        validator: &ValidatorPtr,
+        validator: &Validator,
         wallet: &String,
         config: &MinerRewardsRecipientConfig,
     ) -> Result<BlockTemplate> {
@@ -272,7 +275,7 @@ impl DarkfiMinersRegistry {
     /// Register a new miner and create its job.
     pub async fn register_miner(
         &self,
-        validator: &ValidatorPtr,
+        validator: &Validator,
         wallet: &String,
         config: &MinerRewardsRecipientConfig,
     ) -> Result<(String, String, JsonValue, JsonSubscriber)> {
@@ -294,7 +297,7 @@ impl DarkfiMinersRegistry {
     /// Register a new merge miner and create its job.
     pub async fn register_merge_miner(
         &self,
-        validator: &ValidatorPtr,
+        validator: &Validator,
         wallet: &String,
         config: &MinerRewardsRecipientConfig,
     ) -> Result<(String, f64)> {
@@ -316,7 +319,7 @@ impl DarkfiMinersRegistry {
     /// Submit provided block to the provided node.
     pub async fn submit(
         &self,
-        validator: &ValidatorPtr,
+        validator: &mut Validator,
         subscribers: &HashMap<&'static str, JsonSubscriber>,
         p2p_handler: &DarkfidP2pHandlerPtr,
         block: BlockInfo,
@@ -353,7 +356,7 @@ impl DarkfiMinersRegistry {
         block_templates: &mut HashMap<String, BlockTemplate>,
         jobs: &mut HashMap<String, MinerClient>,
         mm_jobs: &mut HashMap<String, String>,
-        validator: &ValidatorPtr,
+        validator: &Validator,
     ) -> Result<()> {
         // Find inactive native jobs and drop them
         let mut dropped_jobs = vec![];
@@ -473,7 +476,7 @@ impl DarkfiMinersRegistry {
 
     /// Refresh outdated jobs in the registry based on provided
     /// validator state.
-    pub async fn refresh(&self, validator: &ValidatorPtr) -> Result<()> {
+    pub async fn refresh(&self, validator: &Validator) -> Result<()> {
         // Grab registry locks
         let submit_lock = self.submit_lock.write().await;
         let mut block_templates = self.block_templates.write().await;

bin/darkfid/src/registry/model.rs

@@ -231,12 +231,13 @@ pub struct PowRewardV1Zk {
 }
 
 impl PowRewardV1Zk {
-    pub fn new(validator: &ValidatorPtr) -> Result<Self> {
+    pub async fn new(validator: &ValidatorPtr) -> Result<Self> {
         info!(
             target: "darkfid::registry::model::PowRewardV1Zk::new",
             "Generating PowRewardV1 ZkCircuit and ProvingKey...",
         );
 
+        let validator = validator.read().await;
         let (zkbin, _) = validator.blockchain.contracts.get_zkas(
             &validator.blockchain.sled_db,
             &MONEY_CONTRACT_ID,

bin/darkfid/src/rpc/blockchain.rs

@@ -63,7 +63,13 @@ impl DarkfiNode {
 
         let block_height = *params[0].get::<f64>().unwrap() as u32;
 
-        let blocks = match self.validator.blockchain.get_blocks_by_heights(&[block_height]) {
+        let blocks = match self
+            .validator
+            .read()
+            .await
+            .blockchain
+            .get_blocks_by_heights(&[block_height])
+        {
             Ok(v) => v,
             Err(e) => {
                 error!(target: "darkfid::rpc::blockchain_get_block", "Failed fetching block by height: {e}");
@@ -109,7 +115,7 @@ impl DarkfiNode {
             Err(_) => return JsonError::new(ParseError, None, id).into(),
         };
 
-        let txs = match self.validator.blockchain.transactions.get(&[tx_hash], true) {
+        let txs = match self.validator.read().await.blockchain.transactions.get(&[tx_hash], true) {
             Ok(txs) => txs,
             Err(e) => {
                 error!(target: "darkfid::rpc::blockchain_get_tx", "Failed fetching tx by hash: {e}");
@@ -152,7 +158,9 @@ impl DarkfiNode {
             return JsonResponse::new(JsonValue::Array(vec![1_f64.into(), 1_f64.into()]), id).into()
         }
 
-        let Ok(diff) = self.validator.blockchain.blocks.get_difficulty(&[height], true) else {
+        let Ok(diff) =
+            self.validator.read().await.blockchain.blocks.get_difficulty(&[height], true)
+        else {
             return server_error(RpcError::UnknownBlockHeight, id, None)
         };
 
@@ -184,7 +192,7 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        let Ok((height, hash)) = self.validator.blockchain.last() else {
+        let Ok((height, hash)) = self.validator.read().await.blockchain.last() else {
             return JsonError::new(InternalError, None, id).into()
         };
 
@@ -221,7 +229,8 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        let Ok(next_block_height) = self.validator.best_fork_next_block_height().await else {
+        let Ok(next_block_height) = self.validator.read().await.best_fork_next_block_height().await
+        else {
             return JsonError::new(InternalError, None, id).into()
         };
 
@@ -247,7 +256,7 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        let block_target = self.validator.consensus.module.read().await.target;
+        let block_target = self.validator.read().await.consensus.module.target;
 
         JsonResponse::new(JsonValue::Number(block_target as f64), id).into()
     }
@@ -355,14 +364,16 @@ impl DarkfiNode {
             }
         };
 
-        let Ok(zkas_db) = self.validator.blockchain.contracts.lookup(
-            &self.validator.blockchain.sled_db,
+        let validator = self.validator.read().await;
+        let Ok(zkas_db) = validator.blockchain.contracts.lookup(
+            &validator.blockchain.sled_db,
             &contract_id,
             SMART_CONTRACT_ZKAS_DB_NAME,
         ) else {
             error!(target: "darkfid::rpc::blockchain_lookup_zkas", "Did not find zkas db for ContractId: {contract_id}");
             return server_error(RpcError::ContractZkasDbNotFound, id, None)
         };
+        drop(validator);
 
         let mut ret = vec![];
 
@@ -417,7 +428,7 @@ impl DarkfiNode {
             return server_error(RpcError::ParseError, id, None)
         };
 
-        let Ok(bincode) = self.validator.blockchain.contracts.get(contract_id) else {
+        let Ok(bincode) = self.validator.read().await.blockchain.contracts.get(contract_id) else {
             return server_error(RpcError::ContractWasmNotFound, id, None)
         };
 
@@ -457,8 +468,9 @@ impl DarkfiNode {
 
         let tree_name = params[1].get::<String>().unwrap();
 
-        match self.validator.blockchain.contracts.get_state_tree_records(
-            &self.validator.blockchain.sled_db,
+        let validator = self.validator.read().await;
+        match validator.blockchain.contracts.get_state_tree_records(
+            &validator.blockchain.sled_db,
             &contract_id,
             tree_name,
         ) {
@@ -521,8 +533,9 @@ impl DarkfiNode {
             return server_error(RpcError::ParseError, id, None)
         };
 
-        match self.validator.blockchain.contracts.get_state_tree_value(
-            &self.validator.blockchain.sled_db,
+        let validator = self.validator.read().await;
+        match validator.blockchain.contracts.get_state_tree_value(
+            &validator.blockchain.sled_db,
             &contract_id,
             tree_name,
             &key,

bin/darkfid/src/rpc/stratum.rs

@@ -122,7 +122,8 @@ impl DarkfiNode {
     //     }
     pub async fn stratum_login(&self, id: u16, params: JsonValue) -> JsonResult {
         // Check if node is synced before responding
-        if !*self.validator.synced.read().await {
+        let validator = self.validator.read().await;
+        if !validator.synced {
             return JsonResponse::new(JsonValue::from(HashMap::new()), id).into()
         }
 
@@ -190,7 +191,7 @@ impl DarkfiNode {
             "[RPC-STRATUM] Got login from {wallet} ({agent})",
         );
         let (client_id, job_id, job, publisher) =
-            match self.registry.register_miner(&self.validator, wallet, &config).await {
+            match self.registry.register_miner(&validator, wallet, &config).await {
                 Ok(p) => p,
                 Err(e) => {
                     error!(
@@ -240,7 +241,8 @@ impl DarkfiNode {
     // <-- {"jsonrpc": "2.0", "result": {"status": "OK"}, "id": 1}
     pub async fn stratum_submit(&self, id: u16, params: JsonValue) -> JsonResult {
         // Check if node is synced before responding
-        if !*self.validator.synced.read().await {
+        let mut validator = self.validator.write().await;
+        if !validator.synced {
             return miner_status_response(id, "rejected")
         }
 
@@ -328,7 +330,7 @@ impl DarkfiNode {
 
         // Submit the new block through the registry
         if let Err(e) =
-            self.registry.submit(&self.validator, &self.subscribers, &self.p2p_handler, block).await
+            self.registry.submit(&mut validator, &self.subscribers, &self.p2p_handler, block).await
         {
             error!(
                 target: "darkfid::rpc::rpc_stratum::stratum_submit",
@@ -339,7 +341,7 @@ impl DarkfiNode {
             let mut mm_jobs = self.registry.mm_jobs.write().await;
             if let Err(e) = self
                 .registry
-                .refresh_jobs(&mut block_templates, &mut jobs, &mut mm_jobs, &self.validator)
+                .refresh_jobs(&mut block_templates, &mut jobs, &mut mm_jobs, &validator)
                 .await
             {
                 error!(

bin/darkfid/src/rpc/tx.rs

@@ -48,7 +48,8 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !*self.validator.synced.read().await {
+        let mut validator = self.validator.write().await;
+        if !validator.synced {
             error!(target: "darkfid::rpc::tx_simulate", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
@@ -72,12 +73,8 @@ impl DarkfiNode {
         };
 
         // Simulate state transition
-        let result = self.validator.append_tx(&tx, false).await;
-        if result.is_err() {
-            error!(
-                target: "darkfid::rpc::tx_simulate", "Failed to validate state transition: {}",
-                result.err().unwrap()
-            );
+        if let Err(e) = validator.append_tx(&tx, false).await {
+            error!(target: "darkfid::rpc::tx_simulate", "Failed to validate state transition: {e}");
             return server_error(RpcError::TxSimulationFail, id, None)
         };
 
@@ -101,7 +98,8 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !*self.validator.synced.read().await {
+        let mut validator = self.validator.write().await;
+        if !validator.synced {
             error!(target: "darkfid::rpc::tx_broadcast", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
@@ -125,7 +123,7 @@ impl DarkfiNode {
         };
 
         // We'll perform the state transition check here.
-        if let Err(e) = self.validator.append_tx(&tx, true).await {
+        if let Err(e) = validator.append_tx(&tx, true).await {
             error!(target: "darkfid::rpc::tx_broadcast", "Failed to append transaction to mempool: {e}");
             return server_error(RpcError::TxSimulationFail, id, None)
         };
@@ -153,12 +151,13 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !*self.validator.synced.read().await {
+        let validator = self.validator.read().await;
+        if !validator.synced {
             error!(target: "darkfid::rpc::tx_pending", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
 
-        let pending_txs = match self.validator.blockchain.get_pending_txs() {
+        let pending_txs = match validator.blockchain.get_pending_txs() {
             Ok(v) => v,
             Err(e) => {
                 error!(target: "darkfid::rpc::tx_pending", "Failed fetching pending txs: {e}");
@@ -188,7 +187,8 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !*self.validator.synced.read().await {
+        let mut validator = self.validator.write().await;
+        if !validator.synced {
             error!(target: "darkfid::rpc::tx_clean_pending", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
@@ -200,8 +200,7 @@ impl DarkfiNode {
         let mm_jobs = self.registry.mm_jobs.write().await;
 
         // Purge all unproposed pending transactions from the database
-        let result = self
-            .validator
+        let result = validator
             .consensus
             .purge_unproposed_pending_txs(self.registry.proposed_transactions(&block_templates))
             .await;
@@ -236,7 +235,8 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !*self.validator.synced.read().await {
+        let validator = self.validator.read().await;
+        if !validator.synced {
             error!(target: "darkfid::rpc::tx_calculate_fee", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
@@ -263,7 +263,7 @@ impl DarkfiNode {
         let include_fee = params[1].get::<bool>().unwrap();
 
         // Simulate state transition
-        let result = self.validator.calculate_fee(&tx, *include_fee).await;
+        let result = validator.calculate_fee(&tx, *include_fee).await;
         if result.is_err() {
             error!(
                 target: "darkfid::rpc::tx_calculate_fee", "Failed to validate state transition: {}",

bin/darkfid/src/rpc/xmr.rs

@@ -102,7 +102,7 @@ impl DarkfiNode {
         }
 
         // Grab genesis block to use as chain identifier
-        let (_, genesis_hash) = match self.validator.blockchain.genesis() {
+        let (_, genesis_hash) = match self.validator.read().await.blockchain.genesis() {
             Ok(v) => v,
             Err(e) => {
                 error!(
@@ -164,7 +164,8 @@ impl DarkfiNode {
     //     }
     pub async fn xmr_merge_mining_get_aux_block(&self, id: u16, params: JsonValue) -> JsonResult {
         // Check if node is synced before responding to p2pool
-        if !*self.validator.synced.read().await {
+        let validator = self.validator.read().await;
+        if !validator.synced {
             return JsonResponse::new(JsonValue::from(HashMap::new()), id).into()
         }
 
@@ -225,7 +226,7 @@ impl DarkfiNode {
 
         // Register the new merge miner
         let (job_id, difficulty) =
-            match self.registry.register_merge_miner(&self.validator, wallet, &config).await {
+            match self.registry.register_merge_miner(&validator, wallet, &config).await {
                 Ok(p) => p,
                 Err(e) => {
                     error!(
@@ -285,7 +286,8 @@ impl DarkfiNode {
     // <-- {"jsonrpc":"2.0", "result": {"status": "accepted"}, "id": 1}
     pub async fn xmr_merge_mining_submit_solution(&self, id: u16, params: JsonValue) -> JsonResult {
         // Check if node is synced before responding to p2pool
-        if !*self.validator.synced.read().await {
+        let mut validator = self.validator.write().await;
+        if !validator.synced {
             return miner_status_response(id, "rejected")
         }
 
@@ -423,7 +425,7 @@ impl DarkfiNode {
 
         // Submit the new block through the registry
         if let Err(e) =
-            self.registry.submit(&self.validator, &self.subscribers, &self.p2p_handler, block).await
+            self.registry.submit(&mut validator, &self.subscribers, &self.p2p_handler, block).await
         {
             error!(
                 target: "darkfid::rpc::rpc_xmr::xmr_merge_mining_submit_solution",
@@ -434,7 +436,7 @@ impl DarkfiNode {
             let mut jobs = self.registry.jobs.write().await;
             if let Err(e) = self
                 .registry
-                .refresh_jobs(&mut block_templates, &mut jobs, &mut mm_jobs, &self.validator)
+                .refresh_jobs(&mut block_templates, &mut jobs, &mut mm_jobs, &validator)
                 .await
             {
                 error!(
@@ -448,6 +450,7 @@ impl DarkfiNode {
             drop(jobs);
             drop(mm_jobs);
             drop(submit_lock);
+            drop(validator);
 
             return miner_status_response(id, "rejected")
         }
@@ -459,6 +462,7 @@ impl DarkfiNode {
         drop(block_templates);
         drop(mm_jobs);
         drop(submit_lock);
+        drop(validator);
 
         miner_status_response(id, "accepted")
     }

bin/darkfid/src/task/consensus.rs

@@ -56,11 +56,12 @@ pub async fn consensus_init_task(
     // Check current canonical blockchain for curruption
     // TODO: create a restore method reverting each block backwards
     //       until its healthy again
-    node.validator.consensus.healthcheck().await?;
+    let mut validator = node.validator.write().await;
+    validator.consensus.healthcheck().await?;
 
     // Check if network genesis is in the future.
     let current = Timestamp::current_time().inner();
-    let genesis = node.validator.consensus.module.read().await.genesis.inner();
+    let genesis = validator.consensus.module.genesis.inner();
     if current < genesis {
         let diff = genesis - current;
         info!(target: "darkfid::task::consensus_init_task", "Waiting for network genesis: {diff} seconds");
@@ -69,7 +70,8 @@ pub async fn consensus_init_task(
 
     // Generate a new fork to be able to extend
     info!(target: "darkfid::task::consensus_init_task", "Generating new empty fork...");
-    node.validator.consensus.generate_empty_fork().await?;
+    validator.consensus.generate_empty_fork().await?;
+    drop(validator);
 
     // Sync blockchain
     let comms_timeout =
@@ -99,7 +101,7 @@ pub async fn consensus_init_task(
         }
         checkpoint
     } else {
-        *node.validator.synced.write().await = true;
+        node.validator.write().await.synced = true;
         None
     };
 
@@ -109,7 +111,7 @@ pub async fn consensus_init_task(
             Ok(_) => return Ok(()),
             Err(Error::NetworkNotConnected) => {
                 // Sync node again
-                *node.validator.synced.write().await = false;
+                node.validator.write().await.synced = false;
                 if !config.skip_sync {
                     loop {
                         match sync_task(&node, checkpoint).await {
@@ -122,7 +124,7 @@ pub async fn consensus_init_task(
                         }
                     }
                 } else {
-                    *node.validator.synced.write().await = true;
+                    node.validator.write().await.synced = true;
                 }
             }
             Err(e) => return Err(e),
@@ -181,7 +183,8 @@ async fn consensus_task(
         subscription.receive().await;
 
         // Check if we can confirm anything and broadcast them
-        let confirmed = match node.validator.confirmation().await {
+        let mut validator = node.validator.write().await;
+        let confirmed = match validator.confirmation().await {
             Ok(f) => f,
             Err(e) => {
                 error!(
@@ -193,7 +196,7 @@ async fn consensus_task(
         };
 
         // Refresh mining registry
-        if let Err(e) = node.registry.refresh(&node.validator).await {
+        if let Err(e) = node.registry.refresh(&validator).await {
             error!(target: "darkfid", "Failed refreshing mining block templates: {e}")
         }
 
@@ -204,9 +207,8 @@ async fn consensus_task(
         // Grab the append lock so no other proposal gets processed
         // while the node is purging all unreferenced contract trees
         // from the database.
-        let append_lock = node.validator.consensus.append_lock.write().await;
-        purge_unreferenced_trees(node).await;
-        drop(append_lock);
+        purge_unreferenced_trees(&validator, &node.registry).await;
+        drop(validator);
 
         let mut notif_blocks = Vec::with_capacity(confirmed.len());
         for block in confirmed {

bin/darkfid/src/task/garbage_collect.rs

@@ -16,11 +16,15 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use darkfi::{error::TxVerifyFailed, validator::verification::verify_transactions, Error, Result};
+use darkfi::{
+    error::TxVerifyFailed,
+    validator::{verification::verify_transactions, Validator},
+    Error, Result,
+};
 use darkfi_sdk::crypto::MerkleTree;
 use tracing::{debug, error, info};
 
-use crate::DarkfiNodePtr;
+use crate::{DarkfiMinersRegistryPtr, DarkfiNodePtr};
 
 /// Async task used for purging erroneous pending transactions from the nodes mempool.
 pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
@@ -28,8 +32,9 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
 
     // Grab all current unproposed transactions.  We verify them in batches,
     // to not load them all in memory.
+    let validator = node.validator.read().await;
     let (mut last_checked, mut txs) =
-        match node.validator.blockchain.transactions.get_after_pending(0, node.txs_batch_size) {
+        match validator.blockchain.transactions.get_after_pending(0, node.txs_batch_size) {
             Ok(pair) => pair,
             Err(e) => {
                 error!(
@@ -46,6 +51,10 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
         return Ok(())
     }
 
+    // Grab configured target
+    let target = validator.consensus.module.target;
+    drop(validator);
+
     while !txs.is_empty() {
         // Verify each one against current forks
         for tx in txs {
@@ -54,10 +63,10 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
             let mut valid = false;
 
             // Grab a lock over current consensus forks state
-            let mut forks = node.validator.consensus.forks.write().await;
+            let mut validator = node.validator.write().await;
 
             // Iterate over them to verify transaction validity in their overlays
-            for fork in forks.iter_mut() {
+            for fork in validator.consensus.forks.iter_mut() {
                 // Clone forks' overlay
                 let overlay = match fork.overlay.lock().unwrap().full_clone() {
                     Ok(o) => o,
@@ -104,7 +113,7 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
                 let result = verify_transactions(
                     &overlay,
                     next_block_height,
-                    node.validator.consensus.module.read().await.target,
+                    target,
                     &tx_vec,
                     &mut MerkleTree::new(1),
                     false,
@@ -128,13 +137,10 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
                 }
             }
 
-            // Drop forks lock
-            drop(forks);
-
             // Remove transaction if its invalid for all the forks
             if !valid {
                 debug!(target: "darkfid::task::garbage_collect_task", "Removing invalid transaction: {tx_hash}");
-                if let Err(e) = node.validator.blockchain.remove_pending_txs_hashes(&[tx_hash]) {
+                if let Err(e) = validator.blockchain.remove_pending_txs_hashes(&[tx_hash]) {
                     error!(
                         target: "darkfid::task::garbage_collect_task",
                         "Removing invalid transaction {tx_hash} failed: {e}"
@@ -146,6 +152,8 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
         // Grab next batch
         (last_checked, txs) = match node
             .validator
+            .read()
+            .await
             .blockchain
             .transactions
             .get_after_pending(last_checked + node.txs_batch_size as u64, node.txs_batch_size)
@@ -167,18 +175,17 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
 
 /// Auxiliary function to purge all unreferenced contract trees from
 /// the node database.
-pub async fn purge_unreferenced_trees(node: &DarkfiNodePtr) {
+pub async fn purge_unreferenced_trees(validator: &Validator, registry: &DarkfiMinersRegistryPtr) {
     // Grab node registry locks
-    let submit_lock = node.registry.submit_lock.write().await;
-    let block_templates = node.registry.block_templates.write().await;
-    let jobs = node.registry.jobs.write().await;
-    let mm_jobs = node.registry.mm_jobs.write().await;
+    let submit_lock = registry.submit_lock.write().await;
+    let block_templates = registry.block_templates.write().await;
+    let jobs = registry.jobs.write().await;
+    let mm_jobs = registry.mm_jobs.write().await;
 
     // Purge all unreferenced contract trees from the database
-    if let Err(e) = node
-        .validator
+    if let Err(e) = validator
         .consensus
-        .purge_unreferenced_trees(&mut node.registry.new_trees(&block_templates))
+        .purge_unreferenced_trees(&mut registry.new_trees(&block_templates))
         .await
     {
         error!(target: "darkfid::task::garbage_collect::purge_unreferenced_trees", "Purging unreferenced contract trees from the database failed: {e}");

bin/darkfid/src/task/sync.rs

@@ -46,27 +46,29 @@ pub async fn sync_task(node: &DarkfiNodePtr, checkpoint: Option<(u32, HeaderHash
     let block_sub = node.subscribers.get("blocks").unwrap();
 
     // Grab last known block header, including existing pending sync ones
-    let mut last = node.validator.blockchain.last()?;
+    let validator = node.validator.read().await;
+    let mut last = validator.blockchain.last()?;
 
     // If checkpoint is not reached, purge headers and start syncing from scratch
     if let Some(checkpoint) = checkpoint {
         if checkpoint.0 > last.0 {
-            node.validator.blockchain.headers.remove_all_sync()?;
+            validator.blockchain.headers.remove_all_sync()?;
         }
     }
 
     // Check sync headers first record is the next one
-    if let Some(next) = node.validator.blockchain.headers.get_first_sync()? {
+    if let Some(next) = validator.blockchain.headers.get_first_sync()? {
         if next.height == last.0 + 1 {
             // Grab last sync header to continue syncing from
-            if let Some(last_sync) = node.validator.blockchain.headers.get_last_sync()? {
+            if let Some(last_sync) = validator.blockchain.headers.get_last_sync()? {
                 last = (last_sync.height, last_sync.hash());
             }
         } else {
             // Purge headers and start syncing from scratch
-            node.validator.blockchain.headers.remove_all_sync()?;
+            validator.blockchain.headers.remove_all_sync()?;
         }
     }
+    drop(validator);
     info!(target: "darkfid::task::sync_task", "Last known block: {} - {}", last.0, last.1);
 
     // Grab the most common tip and the corresponding peers
@@ -76,7 +78,7 @@ pub async fn sync_task(node: &DarkfiNodePtr, checkpoint: Option<(u32, HeaderHash
     // If the most common tip is the empty tip, we skip syncing
     // further and will reorg if needed when a new proposal arrives.
     if common_tip_hash == [0u8; 32] {
-        *node.validator.synced.write().await = true;
+        node.validator.write().await.synced = true;
         info!(target: "darkfid::task::sync_task", "Blockchain synced!");
         return Ok(())
     }
@@ -123,7 +125,8 @@ pub async fn sync_task(node: &DarkfiNodePtr, checkpoint: Option<(u32, HeaderHash
     sync_best_fork(node, &common_tip_peers, &last.1).await;
 
     // Perform confirmation
-    let confirmed = node.validator.confirmation().await?;
+    let mut validator = node.validator.write().await;
+    let confirmed = validator.confirmation().await?;
     if !confirmed.is_empty() {
         // Notify subscriber
         let mut notif_blocks = Vec::with_capacity(confirmed.len());
@@ -133,7 +136,7 @@ pub async fn sync_task(node: &DarkfiNodePtr, checkpoint: Option<(u32, HeaderHash
         block_sub.notify(JsonValue::Array(notif_blocks)).await;
     }
 
-    *node.validator.synced.write().await = true;
+    validator.synced = true;
     info!(target: "darkfid::task::sync_task", "Blockchain synced!");
     Ok(())
 }
@@ -301,6 +304,7 @@ async fn retrieve_headers(
     // We subtract 1 since tip_height is increased by one
     let total = tip_height - last_known - 1;
     let mut last_tip_height = tip_height;
+    let validator = node.validator.read().await;
     'headers_loop: loop {
         // Check if all our peers are failing
         let mut count = 0;
@@ -356,14 +360,14 @@ async fn retrieve_headers(
             }
 
             // Store the headers
-            node.validator.blockchain.headers.insert_sync(&response_headers)?;
+            validator.blockchain.headers.insert_sync(&response_headers)?;
             last_tip_height = response_headers[0].height;
-            info!(target: "darkfid::task::sync::retrieve_headers", "Headers received: {}/{total}", node.validator.blockchain.headers.len_sync());
+            info!(target: "darkfid::task::sync::retrieve_headers", "Headers received: {}/{total}", validator.blockchain.headers.len_sync());
         }
     }
 
     // Check if we retrieved any new headers
-    if node.validator.blockchain.headers.is_empty_sync() {
+    if validator.blockchain.headers.is_empty_sync() {
         return Ok(());
     }
 
@@ -373,19 +377,19 @@ async fn retrieve_headers(
     // to not load them all in memory.
     info!(target: "darkfid::task::sync::retrieve_headers", "Verifying headers sequence...");
     let mut verified_headers = 0;
-    let total = node.validator.blockchain.headers.len_sync();
+    let total = validator.blockchain.headers.len_sync();
     // First we verify the first `BATCH` sequence, using the last known header
     // as the first sync header previous.
-    let last_known = node.validator.consensus.best_fork_last_header().await?;
-    let mut headers = node.validator.blockchain.headers.get_after_sync(0, BATCH)?;
+    let last_known = validator.consensus.best_fork_last_header().await?;
+    let mut headers = validator.blockchain.headers.get_after_sync(0, BATCH)?;
     if headers[0].previous != last_known.1 || headers[0].height != last_known.0 + 1 {
-        node.validator.blockchain.headers.remove_all_sync()?;
+        validator.blockchain.headers.remove_all_sync()?;
         return Err(Error::BlockIsInvalid(headers[0].hash().as_string()))
     }
     verified_headers += 1;
     for (index, header) in headers[1..].iter().enumerate() {
         if header.previous != headers[index].hash() || header.height != headers[index].height + 1 {
-            node.validator.blockchain.headers.remove_all_sync()?;
+            validator.blockchain.headers.remove_all_sync()?;
             return Err(Error::BlockIsInvalid(header.hash().as_string()))
         }
         verified_headers += 1;
@@ -394,12 +398,12 @@ async fn retrieve_headers(
 
     // Now we verify the rest sequences
     let mut last_checked = headers.last().unwrap().clone();
-    headers = node.validator.blockchain.headers.get_after_sync(last_checked.height, BATCH)?;
+    headers = validator.blockchain.headers.get_after_sync(last_checked.height, BATCH)?;
     while !headers.is_empty() {
         if headers[0].previous != last_checked.hash() ||
             headers[0].height != last_checked.height + 1
         {
-            node.validator.blockchain.headers.remove_all_sync()?;
+            validator.blockchain.headers.remove_all_sync()?;
             return Err(Error::BlockIsInvalid(headers[0].hash().as_string()))
         }
         verified_headers += 1;
@@ -407,13 +411,13 @@ async fn retrieve_headers(
             if header.previous != headers[index].hash() ||
                 header.height != headers[index].height + 1
             {
-                node.validator.blockchain.headers.remove_all_sync()?;
+                validator.blockchain.headers.remove_all_sync()?;
                 return Err(Error::BlockIsInvalid(header.hash().as_string()))
             }
             verified_headers += 1;
         }
         last_checked = headers.last().unwrap().clone();
-        headers = node.validator.blockchain.headers.get_after_sync(last_checked.height, BATCH)?;
+        headers = validator.blockchain.headers.get_after_sync(last_checked.height, BATCH)?;
         info!(target: "darkfid::task::sync::retrieve_headers", "Headers verified: {verified_headers}/{total}");
     }
 
@@ -444,7 +448,8 @@ async fn retrieve_blocks(
     }
 
     let mut received_blocks = 0;
-    let total = node.validator.blockchain.headers.len_sync();
+    let mut validator = node.validator.write().await;
+    let total = validator.blockchain.headers.len_sync();
     'blocks_loop: loop {
         // Check if all our peers are failing
         let mut count = 0;
@@ -469,7 +474,7 @@ async fn retrieve_blocks(
             };
 
             // Grab first `BATCH` headers
-            let headers = node.validator.blockchain.headers.get_after_sync(0, BATCH)?;
+            let headers = validator.blockchain.headers.get_after_sync(0, BATCH)?;
             if headers.is_empty() {
                 break 'blocks_loop
             }
@@ -508,16 +513,14 @@ async fn retrieve_blocks(
             received_blocks += response.blocks.len();
             if checkpoint_blocks {
                 if let Err(e) =
-                    node.validator.add_checkpoint_blocks(&response.blocks, &headers_hashes).await
+                    validator.add_checkpoint_blocks(&response.blocks, &headers_hashes).await
                 {
                     debug!(target: "darkfid::task::sync::retrieve_blocks", "Error while adding checkpoint blocks: {e}");
                     continue
                 };
             } else {
                 for block in &response.blocks {
-                    if let Err(e) =
-                        node.validator.append_proposal(&Proposal::new(block.clone())).await
-                    {
+                    if let Err(e) = validator.append_proposal(&Proposal::new(block.clone())).await {
                         debug!(target: "darkfid::task::sync::retrieve_blocks", "Error while appending proposal: {e}");
                         continue 'peers_loop
                     };
@@ -526,7 +529,7 @@ async fn retrieve_blocks(
             last_received = (*synced_headers.last().unwrap(), *headers_hashes.last().unwrap());
 
             // Remove synced headers
-            node.validator.blockchain.headers.remove_sync(&synced_headers)?;
+            validator.blockchain.headers.remove_sync(&synced_headers)?;
 
             if checkpoint_blocks {
                 // Notify subscriber
@@ -540,7 +543,7 @@ async fn retrieve_blocks(
                 block_sub.notify(JsonValue::Array(notif_blocks)).await;
             } else {
                 // Perform confirmation for received blocks
-                let confirmed = node.validator.confirmation().await?;
+                let confirmed = validator.confirmation().await?;
                 if !confirmed.is_empty() {
                     // Notify subscriber
                     let mut notif_blocks = Vec::with_capacity(confirmed.len());
@@ -596,8 +599,9 @@ async fn sync_best_fork(node: &DarkfiNodePtr, peers: &[ChannelPtr], last_tip: &H
 
     // Verify and store retrieved proposals
     debug!(target: "darkfid::task::sync::sync_best_fork", "Processing received proposals");
+    let mut validator = node.validator.write().await;
     for proposal in &response.proposals {
-        if let Err(e) = node.validator.append_proposal(proposal).await {
+        if let Err(e) = validator.append_proposal(proposal).await {
             debug!(target: "darkfid::task::sync::sync_best_fork", "Error while appending proposal: {e}");
             return
         };

bin/darkfid/src/task/unknown_proposal.rs

@@ -35,7 +35,7 @@ use darkfi::{
         pow::PoWModule,
         utils::{best_fork_index, header_rank},
         verification::verify_fork_proposal,
-        ValidatorPtr,
+        Validator, ValidatorPtr,
     },
     Error::{Custom, DatabaseError, PoWInvalidOutHash, ProposalAlreadyExists},
     Result,
@@ -132,7 +132,7 @@ async fn handle_unknown_proposal(node: &DarkfiNodePtr, channel: u32, proposal: &
     };
 
     // Grab last known block to create the request and execute it
-    let last = match node.validator.blockchain.last() {
+    let last = match node.validator.read().await.blockchain.last() {
         Ok(l) => l,
         Err(e) => {
             error!(target: "darkfid::task::handle_unknown_proposal", "Blockchain last retriaval failed: {e}");
@@ -193,7 +193,7 @@ async fn handle_unknown_proposal(node: &DarkfiNodePtr, channel: u32, proposal: &
     // Process response proposals
     for proposal in &response.proposals {
         // Append proposal
-        match node.validator.append_proposal(proposal).await {
+        match node.validator.write().await.append_proposal(proposal).await {
             Ok(()) => { /* Do nothing */ }
             // Skip already existing proposals
             Err(ProposalAlreadyExists) => continue,
@@ -261,10 +261,11 @@ async fn handle_reorg(
         };
 
     // Create a new PoW module from last common height
+    let validator = node.validator.read().await;
     let module = match PoWModule::new(
-        node.validator.consensus.blockchain.clone(),
-        node.validator.consensus.module.read().await.target,
-        node.validator.consensus.module.read().await.fixed_difficulty.clone(),
+        validator.consensus.blockchain.clone(),
+        validator.consensus.module.target,
+        validator.consensus.module.fixed_difficulty.clone(),
         Some(last_common_height + 1),
     ) {
         Ok(m) => m,
@@ -277,7 +278,7 @@ async fn handle_reorg(
     // Grab last common height ranks
     let last_difficulty = match last_common_height {
         0 => {
-            let genesis_timestamp = match node.validator.blockchain.genesis_block() {
+            let genesis_timestamp = match validator.blockchain.genesis_block() {
                 Ok(b) => b.header.timestamp,
                 Err(e) => {
                     error!(target: "darkfid::task::handle_reorg", "Retrieving genesis block failed: {e}");
@@ -286,7 +287,7 @@ async fn handle_reorg(
             };
             BlockDifficulty::genesis(genesis_timestamp)
         }
-        _ => match node.validator.blockchain.blocks.get_difficulty(&[last_common_height], true) {
+        _ => match validator.blockchain.blocks.get_difficulty(&[last_common_height], true) {
             Ok(d) => d[0].clone().unwrap(),
             Err(e) => {
                 error!(target: "darkfid::task::handle_reorg", "Retrieving block difficulty failed: {e}");
@@ -294,6 +295,7 @@ async fn handle_reorg(
             }
         },
     };
+    drop(validator);
 
     // Retrieve the headers of the hashes sequence and its ranking
     let (targets_rank, hashes_rank) = match retrieve_peer_headers_sequence_ranking(
@@ -315,20 +317,19 @@ async fn handle_reorg(
         }
     };
 
-    // Grab the append lock so no other proposal gets processed while
-    // we are verifying the sequence.
-    let append_lock = node.validator.consensus.append_lock.write().await;
+    // Grab the validator lock so no other proposal gets processed
+    // while we are verifying the sequence.
+    let mut validator = node.validator.write().await;
 
     // Check if the sequence ranks higher than our current best fork
-    let mut forks = node.validator.consensus.forks.write().await;
-    let index = match best_fork_index(&forks) {
+    let index = match best_fork_index(&validator.consensus.forks) {
         Ok(i) => i,
         Err(e) => {
             debug!(target: "darkfid::task::handle_reorg", "Retrieving best fork index failed: {e}");
             return false
         }
     };
-    let best_fork = &forks[index];
+    let best_fork = &validator.consensus.forks[index];
     if targets_rank < best_fork.targets_rank ||
         (targets_rank == best_fork.targets_rank && hashes_rank <= best_fork.hashes_rank)
     {
@@ -338,7 +339,7 @@ async fn handle_reorg(
 
     // Generate the peer fork and retrieve its ranking
     let peer_fork = match retrieve_peer_fork(
-        &node.validator,
+        &validator,
         (&last_common_height, &module, &last_difficulty),
         channel,
         proposal,
@@ -368,17 +369,15 @@ async fn handle_reorg(
 
     // Execute the reorg
     info!(target: "darkfid::task::handle_reorg", "Peer fork ranks higher than our current best fork, executing reorg...");
-    if let Err(e) = node.validator.blockchain.reset_to_height(last_common_height) {
+    if let Err(e) = validator.blockchain.reset_to_height(last_common_height) {
         error!(target: "darkfid::task::handle_reorg", "Applying full inverse diff failed: {e}");
         return false
     };
-    *node.validator.consensus.module.write().await = module;
-    *forks = vec![peer_fork];
-    drop(forks);
-    drop(append_lock);
+    validator.consensus.module = module;
+    validator.consensus.forks = vec![peer_fork];
 
     // Check if we can confirm anything and broadcast them
-    let confirmed = match node.validator.confirmation().await {
+    let confirmed = match validator.confirmation().await {
         Ok(f) => f,
         Err(e) => {
             error!(target: "darkfid::task::handle_reorg", "Confirmation failed: {e}");
@@ -387,7 +386,7 @@ async fn handle_reorg(
     };
 
     // Refresh mining registry
-    if let Err(e) = node.registry.refresh(&node.validator).await {
+    if let Err(e) = node.registry.refresh(&validator).await {
         error!(target: "darkfid::task::handle_reorg", "Failed refreshing mining block templates: {e}")
     }
 
@@ -444,7 +443,7 @@ async fn retrieve_peer_header_hashes(
         };
 
         // Check if we know this header
-        let headers = match validator.blockchain.blocks.get_order(&[height], false) {
+        let headers = match validator.read().await.blockchain.blocks.get_order(&[height], false) {
             Ok(h) => h,
             Err(e) => return Err(DatabaseError(format!("Retrieving headers failed: {e}"))),
         };
@@ -587,7 +586,7 @@ async fn retrieve_peer_headers_sequence_ranking(
 /// and its ranking, based on provided last common information.
 async fn retrieve_peer_fork(
     // Validator pointer
-    validator: &ValidatorPtr,
+    validator: &Validator,
     // Last common header height, PoW module and difficulty
     last_common_info: (&u32, &PoWModule, &BlockDifficulty),
     // Peer channel and its communications timeout

bin/darkfid/src/tests/harness.rs

@@ -127,8 +127,8 @@ impl Harness {
     }
 
     pub async fn validate_chains(&self, total_blocks: usize) -> Result<()> {
-        let alice = &self.alice.validator;
-        let bob = &self.bob.validator;
+        let alice = &self.alice.validator.read().await;
+        let bob = &self.bob.validator.read().await;
 
         alice
             .validate_blockchain(self.config.pow_target, self.config.pow_fixed_difficulty.clone())
@@ -147,8 +147,8 @@ impl Harness {
     }
 
     pub async fn validate_fork_chains(&self, total_forks: usize, fork_sizes: Vec<usize>) {
-        let alice = &self.alice.validator.consensus.forks.read().await;
-        let bob = &self.bob.validator.consensus.forks.read().await;
+        let alice = &self.alice.validator.read().await.consensus.forks;
+        let bob = &self.bob.validator.read().await.consensus.forks;
 
         let alice_forks_len = alice.len();
         assert_eq!(alice_forks_len, bob.len());
@@ -172,7 +172,7 @@ impl Harness {
         // and then we broadcast it to rest nodes
         for block in blocks {
             let proposal = Proposal::new(block.clone());
-            self.alice.validator.append_proposal(&proposal).await?;
+            self.alice.validator.write().await.append_proposal(&proposal).await?;
             let message = ProposalMessage(proposal);
             self.alice.p2p_handler.p2p.broadcast(&message).await;
         }
@@ -180,8 +180,8 @@ impl Harness {
         // Sleep a bit so blocks can be propagated and then
         // trigger confirmation check to Alice and Bob
         sleep(10).await;
-        self.alice.validator.confirmation().await?;
-        self.bob.validator.confirmation().await?;
+        self.alice.validator.write().await.confirmation().await?;
+        self.bob.validator.write().await.confirmation().await?;
 
         Ok(())
     }
@@ -263,7 +263,7 @@ impl Harness {
             &fork.module,
             &block,
             &previous,
-            self.alice.validator.verify_fees,
+            self.alice.validator.read().await.verify_fees,
         )
         .await?;
         fork.append_proposal(&Proposal::new(block.clone())).await?;
@@ -297,22 +297,22 @@ pub async fn generate_node(
     subscribers.insert("dnet", JsonSubscriber::new("dnet.subscribe_events"));
 
     let p2p_handler = DarkfidP2pHandler::init(settings, ex).await?;
-    let registry = DarkfiMinersRegistry::init(Network::Mainnet, &validator)?;
+    let registry = DarkfiMinersRegistry::init(Network::Mainnet, &validator).await?;
     let node =
         DarkfiNode::new(validator.clone(), p2p_handler.clone(), registry, 50, subscribers.clone())
             .await?;
 
     p2p_handler.start(ex, &node).await?;
 
-    node.validator.consensus.generate_empty_fork().await?;
+    node.validator.write().await.consensus.generate_empty_fork().await?;
 
     if !skip_sync {
         sync_task(&node, checkpoint).await?;
     } else {
-        *node.validator.synced.write().await = true;
+        node.validator.write().await.synced = true;
     }
 
-    node.validator.purge_pending_txs().await?;
+    node.validator.write().await.purge_pending_txs().await?;
 
     Ok(node)
 }

bin/darkfid/src/tests/mod.rs

@@ -59,7 +59,7 @@ async fn sync_blocks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     let th = Harness::new(config, true, &ex).await?;
 
     // Generate a fork to create new blocks
-    let mut fork = th.alice.validator.consensus.forks.read().await[0].full_clone()?;
+    let mut fork = th.alice.validator.read().await.consensus.forks[0].full_clone()?;
 
     // Generate next blocks
     let block1 = th.generate_next_block(&mut fork).await?;
@@ -76,31 +76,22 @@ async fn sync_blocks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     // Extend current fork sequence
     let block5 = th.generate_next_block(&mut fork).await?;
     // Create a new fork extending canonical
-    fork = Fork::new(
-        th.alice.validator.consensus.blockchain.clone(),
-        th.alice.validator.consensus.module.read().await.clone(),
-    )
-    .await?;
+    let alice = th.alice.validator.read().await;
+    fork = Fork::new(alice.consensus.blockchain.clone(), alice.consensus.module.clone()).await?;
     // Append block3 to fork and generate the next one
-    verify_block(
-        &fork.overlay,
-        &fork.diffs,
-        &fork.module,
-        &block3,
-        &block2,
-        th.alice.validator.verify_fees,
-    )
-    .await?;
+    verify_block(&fork.overlay, &fork.diffs, &fork.module, &block3, &block2, alice.verify_fees)
+        .await?;
+    drop(alice);
     let block6 = th.generate_next_block(&mut fork).await?;
     // Add them to nodes
     th.add_blocks(&[block5, block6]).await?;
 
     // Grab current best fork index
-    let forks = th.alice.validator.consensus.forks.read().await;
+    let alice = th.alice.validator.read().await;
     // If index corresponds to the small fork, confirmation
     // did not occur, as it's size is not over the threshold.
-    let small_best = best_fork_index(&forks)? == 1;
-    drop(forks);
+    let small_best = best_fork_index(&alice.consensus.forks)? == 1;
+    drop(alice);
     if small_best {
         // Nodes must have one fork with 3 blocks and one with 2 blocks
         th.validate_fork_chains(2, vec![3, 2]).await;
@@ -125,19 +116,18 @@ async fn sync_blocks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     )
     .await?;
     // Verify node synced
-    let alice = &th.alice.validator;
-    let charlie = &charlie.validator;
-    assert_eq!(alice.blockchain.len(), charlie.blockchain.len());
-    assert!(charlie.blockchain.headers.is_empty_sync());
+    let alice = th.alice.validator.read().await;
+    let charlie_validator = charlie.validator.read().await;
+    assert_eq!(alice.blockchain.len(), charlie_validator.blockchain.len());
+    assert!(charlie_validator.blockchain.headers.is_empty_sync());
     // Node must have just the best fork
-    let forks = alice.consensus.forks.read().await;
-    let best_fork = &forks[best_fork_index(&forks)?];
-    let charlie_forks = charlie.consensus.forks.read().await;
-    assert_eq!(charlie_forks.len(), 1);
-    assert_eq!(charlie_forks[0].proposals.len(), best_fork.proposals.len());
-    assert_eq!(charlie_forks[0].diffs.len(), best_fork.diffs.len());
-    drop(forks);
-    drop(charlie_forks);
+    let index = best_fork_index(&alice.consensus.forks)?;
+    let best_fork = &alice.consensus.forks[index];
+    assert_eq!(charlie_validator.consensus.forks.len(), 1);
+    assert_eq!(charlie_validator.consensus.forks[0].proposals.len(), best_fork.proposals.len());
+    assert_eq!(charlie_validator.consensus.forks[0].diffs.len(), best_fork.diffs.len());
+    drop(charlie_validator);
+    drop(alice);
 
     // Extend the small fork sequence and add it to nodes
     th.add_blocks(&[th.generate_next_block(&mut fork).await?]).await?;
@@ -145,23 +135,23 @@ async fn sync_blocks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     // Nodes must have two forks with 2 blocks each
     th.validate_fork_chains(2, vec![2, 2]).await;
     // Check charlie has the correct forks
-    let charlie_forks = charlie.consensus.forks.read().await;
+    let charlie_validator = charlie.validator.read().await;
     if small_best {
         // If Charlie already had the small fork as its best,
         // it will have a single fork with 3 blocks.
-        assert_eq!(charlie_forks.len(), 1);
-        assert_eq!(charlie_forks[0].proposals.len(), 3);
-        assert_eq!(charlie_forks[0].diffs.len(), 3);
+        assert_eq!(charlie_validator.consensus.forks.len(), 1);
+        assert_eq!(charlie_validator.consensus.forks[0].proposals.len(), 3);
+        assert_eq!(charlie_validator.consensus.forks[0].diffs.len(), 3);
     } else {
         // Charlie didn't originaly have the fork, but it
         // should be synced when its proposal was received
-        assert_eq!(charlie_forks.len(), 2);
-        assert_eq!(charlie_forks[0].proposals.len(), 2);
-        assert_eq!(charlie_forks[0].diffs.len(), 2);
-        assert_eq!(charlie_forks[1].proposals.len(), 2);
-        assert_eq!(charlie_forks[1].diffs.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks[0].proposals.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks[0].diffs.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks[1].proposals.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks[1].diffs.len(), 2);
     }
-    drop(charlie_forks);
+    drop(charlie_validator);
 
     // Since the don't know if the second fork was the best,
     // we extend it until it becomes best and a confirmation
@@ -169,34 +159,34 @@ async fn sync_blocks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     loop {
         th.add_blocks(&[th.generate_next_block(&mut fork).await?]).await?;
         // Check if confirmation occured
-        if th.alice.validator.blockchain.len() > 4 {
+        if th.alice.validator.read().await.blockchain.len() > 4 {
             break
         }
     }
 
     // Nodes must have executed confirmation, so we validate their chains
     th.validate_chains(4 + (fork.proposals.len() - 2)).await?;
-    let bob = &th.bob.validator;
-    let last = alice.blockchain.last()?.1;
+    let last = th.alice.validator.read().await.blockchain.last()?.1;
     assert_eq!(last, fork.proposals[fork.proposals.len() - 3]);
-    assert_eq!(last, bob.blockchain.last()?.1);
+    assert_eq!(last, th.bob.validator.read().await.blockchain.last()?.1);
     // Nodes must have one fork with 2 blocks
     th.validate_fork_chains(1, vec![2]).await;
-    let last_proposal = alice.consensus.forks.read().await[0].proposals[1];
+    let alice = &th.alice.validator.read().await;
+    let last_proposal = alice.consensus.forks[0].proposals[1];
     assert_eq!(last_proposal, *fork.proposals.last().unwrap());
-    assert_eq!(last_proposal, bob.consensus.forks.read().await[0].proposals[1]);
+    assert_eq!(last_proposal, th.bob.validator.read().await.consensus.forks[0].proposals[1]);
 
     // Same for Charlie
+    let mut charlie = charlie.validator.write().await;
     charlie.confirmation().await?;
     charlie.validate_blockchain(pow_target, pow_fixed_difficulty).await?;
     assert_eq!(alice.blockchain.len(), charlie.blockchain.len());
     assert!(charlie.blockchain.headers.is_empty_sync());
     assert_eq!(last, charlie.blockchain.last()?.1);
-    let charlie_forks = charlie.consensus.forks.read().await;
-    assert_eq!(charlie_forks.len(), 1);
-    assert_eq!(charlie_forks[0].proposals.len(), 2);
-    assert_eq!(charlie_forks[0].diffs.len(), 2);
-    assert_eq!(last_proposal, charlie_forks[0].proposals[1]);
+    assert_eq!(charlie.consensus.forks.len(), 1);
+    assert_eq!(charlie.consensus.forks[0].proposals.len(), 2);
+    assert_eq!(charlie.consensus.forks[0].diffs.len(), 2);
+    assert_eq!(last_proposal, charlie.consensus.forks[0].proposals[1]);
 
     // Thanks for reading
     Ok(())

bin/darkfid/src/tests/sync_forks.rs

@@ -43,7 +43,7 @@ async fn sync_forks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     let th = Harness::new(config, true, &ex).await?;
 
     // Generate 3 forks
-    let mut fork0 = th.alice.validator.consensus.forks.read().await[0].full_clone()?;
+    let mut fork0 = th.alice.validator.read().await.consensus.forks[0].full_clone()?;
     let mut fork1 = fork0.full_clone()?;
     let mut fork2 = fork1.full_clone()?;
 
@@ -74,36 +74,37 @@ async fn sync_forks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     let charlie = generate_node(&th.vks, &th.validator_config, &settings, &ex, false, None).await?;
 
     // Verify node synced the best fork
-    let forks = th.alice.validator.consensus.forks.read().await;
-    let best_fork = &forks[best_fork_index(&forks)?];
-    let charlie_forks = charlie.validator.consensus.forks.read().await;
-    assert_eq!(charlie_forks.len(), 1);
-    assert_eq!(charlie_forks[0].proposals.len(), best_fork.proposals.len());
+    let alice = th.alice.validator.read().await;
+    let index = best_fork_index(&alice.consensus.forks)?;
+    let best_fork = &alice.consensus.forks[index];
+    let charlie_validator = charlie.validator.read().await;
+    assert_eq!(charlie_validator.consensus.forks.len(), 1);
+    assert_eq!(charlie_validator.consensus.forks[0].proposals.len(), best_fork.proposals.len());
     let small_best = best_fork.proposals.len() == 1;
-    drop(forks);
-    drop(charlie_forks);
+    drop(charlie_validator);
+    drop(alice);
 
     // Extend the small fork sequences and add it to nodes
     th.add_blocks(&[th.generate_next_block(&mut fork1).await?]).await?;
     th.add_blocks(&[th.generate_next_block(&mut fork2).await?]).await?;
 
     // Check charlie has the correct forks
-    let charlie_forks = charlie.validator.consensus.forks.read().await;
+    let charlie_validator = charlie.validator.read().await;
     if small_best {
         // If Charlie already had a small fork as its best,
         // it will have two forks with 2 blocks each.
-        assert_eq!(charlie_forks.len(), 2);
-        assert_eq!(charlie_forks[0].proposals.len(), 2);
-        assert_eq!(charlie_forks[1].proposals.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks[0].proposals.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks[1].proposals.len(), 2);
     } else {
         // Charlie didn't originaly have the forks, but they
         // should be synced when their proposals were received
-        assert_eq!(charlie_forks.len(), 3);
-        assert_eq!(charlie_forks[0].proposals.len(), 3);
-        assert_eq!(charlie_forks[1].proposals.len(), 2);
-        assert_eq!(charlie_forks[2].proposals.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks.len(), 3);
+        assert_eq!(charlie_validator.consensus.forks[0].proposals.len(), 3);
+        assert_eq!(charlie_validator.consensus.forks[1].proposals.len(), 2);
+        assert_eq!(charlie_validator.consensus.forks[2].proposals.len(), 2);
     }
-    drop(charlie_forks);
+    drop(charlie_validator);
 
     // Thanks for reading
     Ok(())

bin/darkfid/src/tests/unproposed_txs.rs

@@ -60,10 +60,10 @@ async fn simulate_unproposed_txs(
     };
 
     // Create chain test harness using created configuration
-    let blockchain_test_harness = Harness::new(config, false, &ex).await?;
+    let th = Harness::new(config, false, &ex).await?;
 
     // Get validator and generate the fork
-    let validator = blockchain_test_harness.alice.validator.clone();
+    let mut validator = th.alice.validator.write().await;
     validator.consensus.generate_empty_fork().await?;
 
     // Create contract test harness
@@ -87,9 +87,8 @@ async fn simulate_unproposed_txs(
     }
 
     // Obtain fork
-    let mut forks = validator.consensus.forks.write().await;
-    let index = best_fork_index(&forks)?;
-    let best_fork = &mut forks[index];
+    let index = best_fork_index(&validator.consensus.forks)?;
+    let best_fork = &mut validator.consensus.forks[index];
 
     // Retrieve unproposed transactions
     let (tx, total_gas_used, _) = best_fork.unproposed_txs(current_block_height, false).await?;

src/contract/dao/tests/integration.rs

@@ -436,7 +436,7 @@ async fn execute_transfer_proposal(
 
     // Grab creation blockwindow
     let block_target =
-        th.holders.get_mut(&Holder::Dao).unwrap().validator.consensus.module.read().await.target;
+        th.holders.get_mut(&Holder::Dao).unwrap().validator.read().await.consensus.module.target;
     let creation_blockwindow = blockwindow(*current_block_height, block_target);
 
     let (tx, params, fee_params, proposal_info) = th
@@ -609,7 +609,7 @@ async fn execute_generic_proposal(
 
     // Grab creation blockwindow
     let block_target =
-        th.holders.get_mut(&Holder::Dao).unwrap().validator.consensus.module.read().await.target;
+        th.holders.get_mut(&Holder::Dao).unwrap().validator.read().await.consensus.module.target;
     let creation_blockwindow = blockwindow(*current_block_height, block_target);
 
     let (tx, params, fee_params, proposal_info) = th

src/contract/money/tests/delayed_tx.rs

@@ -136,17 +136,18 @@ fn delayed_tx() -> Result<()> {
 
         // First we verify the fee-less transaction to see how much gas it uses for execution
         // and verification.
-        let gas_used = wallet
-            .validator
+        let validator = wallet.validator.read().await;
+        let gas_used = validator
             .add_test_transactions(
                 &[tx],
                 current_block_height,
-                wallet.validator.consensus.module.read().await.target,
+                validator.consensus.module.target,
                 false,
                 false,
             )
             .await?
             .0;
+        drop(validator);
 
         // Compute the required fee
         let required_fee = compute_fee(&(gas_used + FEE_CALL_GAS));

src/contract/test-harness/src/dao_exec.rs

@@ -152,7 +152,7 @@ impl TestHarness {
             xfer_params.inputs.iter().map(|input| input.value_commit).sum()
         );
 
-        let block_target = dao_wallet.validator.consensus.module.read().await.target;
+        let block_target = dao_wallet.validator.read().await.consensus.module.target;
         let current_blockwindow = blockwindow(block_height, block_target);
         let exec_builder = DaoExecCall {
             proposal: proposal.clone(),
@@ -276,7 +276,7 @@ impl TestHarness {
 
         // Create the exec call
         let exec_signature_secret = SecretKey::random(&mut OsRng);
-        let block_target = wallet.validator.consensus.module.read().await.target;
+        let block_target = wallet.validator.read().await.consensus.module.target;
         let current_blockwindow = blockwindow(block_height, block_target);
         let exec_builder = DaoExecCall {
             proposal: proposal.clone(),

src/contract/test-harness/src/dao_propose.rs

@@ -75,7 +75,7 @@ impl TestHarness {
 
         // Useful code snippet to dump a sled contract DB
         /*{
-            let blockchain = &wallet.validator.blockchain;
+            let blockchain = &wallet.validator.read().await.blockchain;
             let contracts = &blockchain.contracts;
             let tree = contracts
                 .lookup(&blockchain.sled_db, &MONEY_CONTRACT_ID, "nullifier_roots")
@@ -119,7 +119,7 @@ impl TestHarness {
             },
         ];
 
-        let block_target = wallet.validator.consensus.module.read().await.target;
+        let block_target = wallet.validator.read().await.consensus.module.target;
         let creation_blockwindow = blockwindow(block_height, block_target);
         let proposal = DaoProposal {
             auth_calls,
@@ -215,7 +215,7 @@ impl TestHarness {
 
         // Useful code snippet to dump a sled contract DB
         /*{
-            let blockchain = &wallet.validator.blockchain;
+            let blockchain = &wallet.validator.read().await.blockchain;
             let contracts = &blockchain.contracts;
             let tree = contracts
                 .lookup(&blockchain.sled_db, &MONEY_CONTRACT_ID, "nullifier_roots")
@@ -237,7 +237,7 @@ impl TestHarness {
                 .unwrap(),
         };
 
-        let block_target = wallet.validator.consensus.module.read().await.target;
+        let block_target = wallet.validator.read().await.consensus.module.target;
         let creation_blockwindow = blockwindow(block_height, block_target);
         let proposal = DaoProposal {
             auth_calls: vec![],

src/contract/test-harness/src/dao_vote.rs

@@ -74,7 +74,7 @@ impl TestHarness {
             merkle_path: snapshot_money_merkle_tree.witness(vote_owncoin.leaf_position, 0).unwrap(),
         };
 
-        let block_target = wallet.validator.consensus.module.read().await.target;
+        let block_target = wallet.validator.read().await.consensus.module.target;
         let current_blockwindow = blockwindow(block_height, block_target);
         let call = DaoVoteCall {
             money_null_smt: wallet.money_null_smt_snapshot.as_ref().unwrap(),

src/contract/test-harness/src/lib.rs

@@ -220,19 +220,20 @@ impl Wallet {
             let _ = benchmark_wasm_calls(callname, &self.validator, &tx, block_height).await;
         }
 
-        self.validator
+        let validator = self.validator.read().await;
+        validator
             .add_test_transactions(
                 slice::from_ref(&tx),
                 block_height,
-                self.validator.consensus.module.read().await.target,
+                validator.consensus.module.target,
                 true,
-                self.validator.verify_fees,
+                validator.verify_fees,
             )
             .await?;
 
         // Write the data
         {
-            let blockchain = &self.validator.blockchain;
+            let blockchain = &validator.blockchain;
             let txs = &blockchain.transactions;
             txs.insert(slice::from_ref(&tx)).expect("insert tx");
             txs.insert_location(&[tx.hash()], block_height).expect("insert loc");
@@ -329,12 +330,13 @@ impl TestHarness {
 
 async fn benchmark_wasm_calls(
     callname: &str,
-    validator: &Validator,
+    validator: &ValidatorPtr,
     tx: &Transaction,
     block_height: u32,
 ) -> Result<()> {
     let mut file = std::fs::OpenOptions::new().create(true).append(true).open("bench.csv")?;
 
+    let validator = validator.read().await;
     for (idx, call) in tx.calls.iter().enumerate() {
         let overlay = BlockchainOverlay::new(&validator.blockchain).expect("blockchain overlay");
         let wasm = overlay.lock().unwrap().contracts.get(call.data.contract_id)?;
@@ -343,7 +345,7 @@ async fn benchmark_wasm_calls(
             overlay.clone(),
             call.data.contract_id,
             block_height,
-            validator.consensus.module.read().await.target,
+            validator.consensus.module.target,
             tx.hash(),
             idx as u8,
         )

src/contract/test-harness/src/money_fee.rs

@@ -221,12 +221,12 @@ impl TestHarness {
         // First we verify the fee-less transaction to see how much gas it uses for execution
         // and verification.
         let wallet = self.holders.get(holder).unwrap();
-        let gas_used = wallet
-            .validator
+        let validator = wallet.validator.read().await;
+        let gas_used = validator
             .add_test_transactions(
                 &[tx],
                 block_height,
-                wallet.validator.consensus.module.read().await.target,
+                validator.consensus.module.target,
                 false,
                 false,
             )

src/contract/test-harness/src/money_pow_reward.rs

@@ -53,7 +53,7 @@ impl TestHarness {
         let (mint_pk, mint_zkbin) = self.proving_keys.get(MONEY_CONTRACT_ZKAS_MINT_NS_V1).unwrap();
 
         // Reference the last block in the holder's blockchain
-        let last_block = wallet.validator.blockchain.last_block()?;
+        let last_block = wallet.validator.read().await.blockchain.last_block()?;
 
         // If there's a set reward recipient, use it, otherwise reward the holder
         let recipient = if let Some(holder) = recipient {
@@ -110,7 +110,8 @@ impl TestHarness {
 
         // Fetch the last block in the blockchain
         let wallet = self.holders.get(miner).unwrap();
-        let previous = wallet.validator.blockchain.last_block()?;
+        let validator = wallet.validator.read().await;
+        let previous = validator.blockchain.last_block()?;
 
         // We increment timestamp so we don't have to use sleep
         let timestamp = previous.header.timestamp.checked_add(1.into())?;
@@ -130,15 +131,16 @@ impl TestHarness {
         block.append_txs(vec![tx]);
 
         // Compute block contracts states monotree root
-        let overlay = BlockchainOverlay::new(&wallet.validator.blockchain)?;
+        let overlay = BlockchainOverlay::new(&validator.blockchain)?;
         let _ = apply_producer_transaction(
             &overlay,
             block.header.height,
-            wallet.validator.consensus.module.read().await.target,
+            validator.consensus.module.target,
             block.txs.last().unwrap(),
             &mut MerkleTree::new(1),
         )
         .await?;
+        drop(validator);
         let diff = overlay.lock().unwrap().overlay.lock().unwrap().diff(&[])?;
         block.header.state_root = overlay.lock().unwrap().contracts.update_state_monotree(&diff)?;
 
@@ -149,7 +151,7 @@ impl TestHarness {
         let mut found_owncoins = vec![];
         for holder in holders {
             let wallet = self.holders.get_mut(holder).unwrap();
-            wallet.validator.add_test_blocks(&[block.clone()]).await?;
+            wallet.validator.write().await.add_test_blocks(&[block.clone()]).await?;
             wallet.money_merkle_tree.append(MerkleNode::from(params.output.coin.inner()));
 
             // Attempt to decrypt the note to see if this is a coin for the holder

src/validator/consensus.rs

@@ -22,7 +22,6 @@ use darkfi_sdk::{crypto::MerkleTree, tx::TransactionHash};
 use darkfi_serial::{async_trait, deserialize, SerialDecodable, SerialEncodable};
 use num_bigint::BigUint;
 use sled_overlay::{database::SledDbOverlayStateDiff, sled::IVec};
-use smol::lock::RwLock;
 use tracing::{debug, info, warn};
 
 use crate::{
@@ -51,11 +50,9 @@ pub struct Consensus {
     /// Fork size(length) after which it can be confirmed
     pub confirmation_threshold: usize,
     /// Fork chains containing block proposals
-    pub forks: RwLock<Vec<Fork>>,
+    pub forks: Vec<Fork>,
     /// Canonical blockchain PoW module state
-    pub module: RwLock<PoWModule>,
-    /// Lock to restrict when proposals appends can happen
-    pub append_lock: RwLock<()>,
+    pub module: PoWModule,
 }
 
 impl Consensus {
@@ -66,50 +63,37 @@ impl Consensus {
         pow_target: u32,
         pow_fixed_difficulty: Option<BigUint>,
     ) -> Result<Self> {
-        let forks = RwLock::new(vec![]);
-
-        let module = RwLock::new(PoWModule::new(
-            blockchain.clone(),
-            pow_target,
-            pow_fixed_difficulty,
-            None,
-        )?);
-
-        let append_lock = RwLock::new(());
+        let module = PoWModule::new(blockchain.clone(), pow_target, pow_fixed_difficulty, None)?;
 
-        Ok(Self { blockchain, confirmation_threshold, forks, module, append_lock })
+        Ok(Self { blockchain, confirmation_threshold, forks: vec![], module })
     }
 
     /// Generate a new empty fork.
-    pub async fn generate_empty_fork(&self) -> Result<()> {
+    pub async fn generate_empty_fork(&mut self) -> Result<()> {
         debug!(target: "validator::consensus::generate_empty_fork", "Generating new empty fork...");
-        let mut forks = self.forks.write().await;
         // Check if we already have an empty fork
-        for fork in forks.iter() {
+        for fork in &self.forks {
             if fork.proposals.is_empty() {
                 debug!(target: "validator::consensus::generate_empty_fork", "An empty fork already exists.");
-                drop(forks);
                 return Ok(())
             }
         }
-        let fork = Fork::new(self.blockchain.clone(), self.module.read().await.clone()).await?;
-        forks.push(fork);
-        drop(forks);
+        let fork = Fork::new(self.blockchain.clone(), self.module.clone()).await?;
+        self.forks.push(fork);
         debug!(target: "validator::consensus::generate_empty_fork", "Fork generated!");
+
         Ok(())
     }
 
     /// Given a proposal, the node verifys it and finds which fork it extends.
     /// If the proposal extends the canonical blockchain, a new fork chain is created.
-    pub async fn append_proposal(&self, proposal: &Proposal, verify_fees: bool) -> Result<()> {
+    pub async fn append_proposal(&mut self, proposal: &Proposal, verify_fees: bool) -> Result<()> {
         debug!(target: "validator::consensus::append_proposal", "Appending proposal {}", proposal.hash);
 
         // Check if proposal already exists
-        let lock = self.forks.read().await;
-        for fork in lock.iter() {
+        for fork in &self.forks {
             for p in fork.proposals.iter().rev() {
                 if p == &proposal.hash {
-                    drop(lock);
                     debug!(target: "validator::consensus::append_proposal", "Proposal {} already exists", proposal.hash);
                     return Err(Error::ProposalAlreadyExists)
                 }
@@ -120,12 +104,10 @@ impl Consensus {
             self.blockchain.blocks.get_order(&[proposal.block.header.height], true)
         {
             if canonical_headers[0].unwrap() == proposal.hash {
-                drop(lock);
                 debug!(target: "validator::consensus::append_proposal", "Proposal {} already exists", proposal.hash);
                 return Err(Error::ProposalAlreadyExists)
             }
         }
-        drop(lock);
 
         // Verify proposal and grab corresponding fork
         let (mut fork, index) = verify_proposal(self, proposal, verify_fees).await?;
@@ -138,21 +120,20 @@ impl Consensus {
 
         // If a fork index was found, replace forks with the mutated one,
         // otherwise push the new fork.
-        let mut lock = self.forks.write().await;
         match index {
             Some(i) => {
-                if i < lock.len() && lock[i].proposals == fork.proposals[..fork.proposals.len() - 1]
+                if i < self.forks.len() &&
+                    self.forks[i].proposals == fork.proposals[..fork.proposals.len() - 1]
                 {
-                    lock[i] = fork;
+                    self.forks[i] = fork;
                 } else {
-                    lock.push(fork);
+                    self.forks.push(fork);
                 }
             }
             None => {
-                lock.push(fork);
+                self.forks.push(fork);
             }
         }
-        drop(lock);
 
         info!(target: "validator::consensus::append_proposal", "Appended proposal {}", proposal.hash);
 
@@ -165,11 +146,8 @@ impl Consensus {
     /// a new fork is created. Additionally, we return the fork index if a new fork
     /// was not created, so caller can replace the fork.
     pub async fn find_extended_fork(&self, proposal: &Proposal) -> Result<(Fork, Option<usize>)> {
-        // Grab a lock over current forks
-        let forks = self.forks.read().await;
-
         // Check if proposal extends any fork
-        let found = find_extended_fork_index(&forks, proposal);
+        let found = find_extended_fork_index(&self.forks, proposal);
         if found.is_err() {
             if let Err(Error::ProposalAlreadyExists) = found {
                 return Err(Error::ProposalAlreadyExists)
@@ -184,26 +162,26 @@ impl Consensus {
             }
 
             // Check if we have an empty fork to use
-            for (f_index, fork) in forks.iter().enumerate() {
+            for (f_index, fork) in self.forks.iter().enumerate() {
                 if fork.proposals.is_empty() {
-                    return Ok((forks[f_index].full_clone()?, Some(f_index)))
+                    return Ok((self.forks[f_index].full_clone()?, Some(f_index)))
                 }
             }
 
             // Generate a new fork extending canonical
-            let fork = Fork::new(self.blockchain.clone(), self.module.read().await.clone()).await?;
+            let fork = Fork::new(self.blockchain.clone(), self.module.clone()).await?;
             return Ok((fork, None))
         }
 
         let (f_index, p_index) = found.unwrap();
-        let original_fork = &forks[f_index];
+        let original_fork = &self.forks[f_index];
         // Check if proposal extends fork at last proposal
         if p_index == (original_fork.proposals.len() - 1) {
             return Ok((original_fork.full_clone()?, Some(f_index)))
         }
 
         // Rebuild fork
-        let mut fork = Fork::new(self.blockchain.clone(), self.module.read().await.clone()).await?;
+        let mut fork = Fork::new(self.blockchain.clone(), self.module.clone()).await?;
         fork.proposals = original_fork.proposals[..p_index + 1].to_vec();
         fork.diffs = original_fork.diffs[..p_index + 1].to_vec();
 
@@ -227,9 +205,6 @@ impl Consensus {
             fork.hashes_rank += hash_distance_sq;
         }
 
-        // Drop forks lock
-        drop(forks);
-
         Ok((fork, None))
     }
 
@@ -245,21 +220,16 @@ impl Consensus {
         debug!(target: "validator::consensus::confirmation", "Started confirmation check");
 
         // Grab best fork
-        let forks = self.forks.read().await;
-        let index = best_fork_index(&forks)?;
-        let fork = &forks[index];
+        let index = best_fork_index(&self.forks)?;
+        let fork = &self.forks[index];
 
         // Check its length
         let length = fork.proposals.len();
         if length < self.confirmation_threshold {
             debug!(target: "validator::consensus::confirmation", "Nothing to confirme yet, best fork size: {length}");
-            drop(forks);
             return Ok(None)
         }
 
-        // Drop forks lock
-        drop(forks);
-
         Ok(Some(index))
     }
 
@@ -270,12 +240,9 @@ impl Consensus {
         height: u32,
         fork_header: &HeaderHash,
     ) -> Result<Option<HeaderHash>> {
-        // Grab a lock over current forks
-        let forks = self.forks.read().await;
-
         // Find the fork containing the provided header
         let mut found = None;
-        'outer: for (index, fork) in forks.iter().enumerate() {
+        'outer: for (index, fork) in self.forks.iter().enumerate() {
             for p in fork.proposals.iter().rev() {
                 if p == fork_header {
                     found = Some(index);
@@ -284,16 +251,13 @@ impl Consensus {
             }
         }
         if found.is_none() {
-            drop(forks);
             return Ok(None)
         }
         let index = found.unwrap();
 
         // Grab header if it exists
-        let header = forks[index].overlay.lock().unwrap().blocks.get_order(&[height], false)?[0];
-
-        // Drop forks lock
-        drop(forks);
+        let header =
+            self.forks[index].overlay.lock().unwrap().blocks.get_order(&[height], false)?[0];
 
         Ok(header)
     }
@@ -306,12 +270,9 @@ impl Consensus {
         headers: &[HeaderHash],
         fork_header: &HeaderHash,
     ) -> Result<Vec<Header>> {
-        // Grab a lock over current forks
-        let forks = self.forks.read().await;
-
         // Find the fork containing the provided header
         let mut found = None;
-        'outer: for (index, fork) in forks.iter().enumerate() {
+        'outer: for (index, fork) in self.forks.iter().enumerate() {
             for p in fork.proposals.iter().rev() {
                 if p == fork_header {
                     found = Some(index);
@@ -319,16 +280,10 @@ impl Consensus {
                 }
             }
         }
-        let Some(index) = found else {
-            drop(forks);
-            return Ok(vec![])
-        };
+        let Some(index) = found else { return Ok(vec![]) };
 
         // Grab headers
-        let headers = forks[index].overlay.lock().unwrap().get_headers_by_hash(headers)?;
-
-        // Drop forks lock
-        drop(forks);
+        let headers = self.forks[index].overlay.lock().unwrap().get_headers_by_hash(headers)?;
 
         Ok(headers)
     }
@@ -341,12 +296,9 @@ impl Consensus {
         headers: &[HeaderHash],
         fork_header: &HeaderHash,
     ) -> Result<Vec<Proposal>> {
-        // Grab a lock over current forks
-        let forks = self.forks.read().await;
-
         // Find the fork containing the provided header
         let mut found = None;
-        'outer: for (index, fork) in forks.iter().enumerate() {
+        'outer: for (index, fork) in self.forks.iter().enumerate() {
             for p in fork.proposals.iter().rev() {
                 if p == fork_header {
                     found = Some(index);
@@ -354,21 +306,15 @@ impl Consensus {
                 }
             }
         }
-        let Some(index) = found else {
-            drop(forks);
-            return Ok(vec![])
-        };
+        let Some(index) = found else { return Ok(vec![]) };
 
         // Grab proposals
-        let blocks = forks[index].overlay.lock().unwrap().get_blocks_by_hash(headers)?;
+        let blocks = self.forks[index].overlay.lock().unwrap().get_blocks_by_hash(headers)?;
         let mut proposals = Vec::with_capacity(blocks.len());
         for block in blocks {
             proposals.push(Proposal::new(block));
         }
 
-        // Drop forks lock
-        drop(forks);
-
         Ok(proposals)
     }
 
@@ -382,9 +328,6 @@ impl Consensus {
         fork_tip: Option<HeaderHash>,
         limit: u32,
     ) -> Result<Vec<Proposal>> {
-        // Grab a lock over current forks
-        let forks = self.forks.read().await;
-
         // Create return vector
         let mut proposals = vec![];
 
@@ -392,7 +335,7 @@ impl Consensus {
         let index = match fork_tip {
             Some(fork_tip) => {
                 let mut found = None;
-                'outer: for (index, fork) in forks.iter().enumerate() {
+                'outer: for (index, fork) in self.forks.iter().enumerate() {
                     for p in fork.proposals.iter().rev() {
                         if p == &fork_tip {
                             found = Some(index);
@@ -401,25 +344,23 @@ impl Consensus {
                     }
                 }
                 if found.is_none() {
-                    drop(forks);
                     return Ok(proposals)
                 }
                 found.unwrap()
             }
-            None => best_fork_index(&forks)?,
+            None => best_fork_index(&self.forks)?,
         };
 
         // Check tip exists
-        let Ok(existing_tips) = forks[index].overlay.lock().unwrap().get_blocks_by_hash(&[tip])
+        let Ok(existing_tips) =
+            self.forks[index].overlay.lock().unwrap().get_blocks_by_hash(&[tip])
         else {
-            drop(forks);
             return Ok(proposals)
         };
 
         // Check tip is not far behind
-        let last_block_height = forks[index].overlay.lock().unwrap().last()?.0;
+        let last_block_height = self.forks[index].overlay.lock().unwrap().last()?.0;
         if last_block_height - existing_tips[0].header.height >= limit {
-            drop(forks);
             return Ok(proposals)
         }
 
@@ -429,15 +370,15 @@ impl Consensus {
         for block in blocks {
             proposals.push(Proposal::new(block));
         }
-        let blocks =
-            forks[index].overlay.lock().unwrap().get_blocks_by_hash(&forks[index].proposals)?;
+        let blocks = self.forks[index]
+            .overlay
+            .lock()
+            .unwrap()
+            .get_blocks_by_hash(&self.forks[index].proposals)?;
         for block in blocks {
             proposals.push(Proposal::new(block));
         }
 
-        // Drop forks lock
-        drop(forks);
-
         Ok(proposals)
     }
 
@@ -445,17 +386,15 @@ impl Consensus {
     /// based on next block height.
     /// If no forks exist, returns the canonical key.
     pub async fn current_mining_randomx_key(&self) -> Result<HeaderHash> {
-        // Grab a lock over current forks
-        let forks = self.forks.read().await;
-
         // Grab next block height and current keys.
         // If no forks exist, use canonical keys
-        let (next_block_height, rx_keys) = if forks.is_empty() {
+        let (next_block_height, rx_keys) = if self.forks.is_empty() {
             let (next_block_height, _) = self.blockchain.last()?;
-            (next_block_height + 1, self.module.read().await.darkfi_rx_keys)
+            (next_block_height + 1, self.module.darkfi_rx_keys)
         } else {
             // Grab best fork and its last proposal
-            let fork = &forks[best_fork_index(&forks)?];
+            let index = best_fork_index(&self.forks)?;
+            let fork = &self.forks[index];
             let last = fork.last_proposal()?;
             (last.block.header.height + 1, fork.module.darkfi_rx_keys)
         };
@@ -474,29 +413,24 @@ impl Consensus {
 
     /// Auxiliary function to grab best current fork full clone.
     pub async fn best_current_fork(&self) -> Result<Fork> {
-        let forks = self.forks.read().await;
-        let index = best_fork_index(&forks)?;
-        forks[index].full_clone()
+        let index = best_fork_index(&self.forks)?;
+        self.forks[index].full_clone()
     }
 
     /// Auxiliary function to retrieve current best fork last header.
     /// If no forks exist, grab the last header from canonical.
     pub async fn best_fork_last_header(&self) -> Result<(u32, HeaderHash)> {
-        // Grab a lock over current forks
-        let forks = self.forks.read().await;
-
         // Check if node has any forks
-        if forks.is_empty() {
-            drop(forks);
+        if self.forks.is_empty() {
             return self.blockchain.last()
         }
 
         // Grab best fork
-        let fork = &forks[best_fork_index(&forks)?];
+        let index = best_fork_index(&self.forks)?;
+        let fork = &self.forks[index];
 
         // Grab its last header
         let last = fork.last_proposal()?;
-        drop(forks);
         Ok((last.block.header.height, last.hash))
     }
 
@@ -510,14 +444,11 @@ impl Consensus {
     /// Note: Always remember to purge new trees from the database if
     /// not needed.
     pub async fn reset_forks(
-        &self,
+        &mut self,
         prefix: &[HeaderHash],
         confirmed_fork_index: &usize,
         confirmed_txs: &[Transaction],
     ) -> Result<()> {
-        // Grab a lock over current forks
-        let mut forks = self.forks.write().await;
-
         // Find all the forks that start with the provided prefix,
         // excluding confirmed fork index, and remove their prefixed
         // proposals, and their corresponding diffs. If the fork is not
@@ -525,10 +456,10 @@ impl Consensus {
         let excess = prefix.len();
         let prefix_last_index = excess - 1;
         let prefix_last = prefix.last().unwrap();
-        let mut keep = vec![true; forks.len()];
+        let mut keep = vec![true; self.forks.len()];
         let confirmed_txs_hashes: Vec<TransactionHash> =
             confirmed_txs.iter().map(|tx| tx.hash()).collect();
-        for (index, fork) in forks.iter_mut().enumerate() {
+        for (index, fork) in self.forks.iter_mut().enumerate() {
             if &index == confirmed_fork_index {
                 // Remove confirmed proposals txs from fork's mempool
                 fork.mempool.retain(|tx| !confirmed_txs_hashes.contains(tx));
@@ -559,49 +490,40 @@ impl Consensus {
 
         // Drop invalid forks
         let mut iter = keep.iter();
-        forks.retain(|_| *iter.next().unwrap());
+        self.forks.retain(|_| *iter.next().unwrap());
 
         // Remove confirmed proposals txs from the unporposed txs sled tree
         self.blockchain.remove_pending_txs_hashes(&confirmed_txs_hashes)?;
 
-        // Drop forks lock
-        drop(forks);
-
         Ok(())
     }
 
     /// Auxiliary function to fully purge current forks and leave only a new empty fork.
-    pub async fn purge_forks(&self) -> Result<()> {
+    pub async fn purge_forks(&mut self) -> Result<()> {
         debug!(target: "validator::consensus::purge_forks", "Purging current forks...");
-        let mut forks = self.forks.write().await;
-        *forks = vec![Fork::new(self.blockchain.clone(), self.module.read().await.clone()).await?];
-        drop(forks);
+        self.forks = vec![Fork::new(self.blockchain.clone(), self.module.clone()).await?];
         debug!(target: "validator::consensus::purge_forks", "Forks purged!");
+
         Ok(())
     }
 
     /// Auxiliary function to reset PoW module.
-    pub async fn reset_pow_module(&self) -> Result<()> {
+    pub async fn reset_pow_module(&mut self) -> Result<()> {
         debug!(target: "validator::consensus::reset_pow_module", "Resetting PoW module...");
-
-        let mut module = self.module.write().await;
-        *module = PoWModule::new(
+        self.module = PoWModule::new(
             self.blockchain.clone(),
-            module.target,
-            module.fixed_difficulty.clone(),
+            self.module.target,
+            self.module.fixed_difficulty.clone(),
             None,
         )?;
-        drop(module);
         debug!(target: "validator::consensus::reset_pow_module", "PoW module reset successfully!");
+
         Ok(())
     }
 
     /// Auxiliary function to check current contracts states
     /// Monotree(SMT) validity in all active forks and canonical.
     pub async fn healthcheck(&self) -> Result<()> {
-        // Grab a lock over current forks
-        let lock = self.forks.read().await;
-
         // Grab current canonical contracts states monotree root
         let state_root = self.blockchain.contracts.get_state_monotree_root()?;
 
@@ -615,7 +537,7 @@ impl Consensus {
         }
 
         // Check each fork health
-        for fork in lock.iter() {
+        for fork in &self.forks {
             fork.healthcheck()?;
         }
 
@@ -628,18 +550,15 @@ impl Consensus {
         &self,
         referenced_trees: &mut BTreeSet<IVec>,
     ) -> Result<()> {
-        // Grab a lock over current forks
-        let lock = self.forks.read().await;
-
         // Check if we have forks
-        if lock.is_empty() {
+        if self.forks.is_empty() {
             // If no forks exist, build a new one so we retrieve the
             // native/protected trees references.
-            let fork = Fork::new(self.blockchain.clone(), self.module.read().await.clone()).await?;
+            let fork = Fork::new(self.blockchain.clone(), self.module.clone()).await?;
             fork.referenced_trees(referenced_trees);
         } else {
             // Iterate over current forks to retrieve referenced trees
-            for fork in lock.iter() {
+            for fork in &self.forks {
                 fork.referenced_trees(referenced_trees);
             }
         }
@@ -669,14 +588,11 @@ impl Consensus {
     /// Auxiliary function to purge all unproposed pending
     /// transactions from the database.
     pub async fn purge_unproposed_pending_txs(
-        &self,
+        &mut self,
         mut proposed_txs: HashSet<TransactionHash>,
     ) -> Result<()> {
-        // Grab a lock over current forks
-        let mut forks = self.forks.write().await;
-
         // Iterate over all forks to find proposed txs
-        for fork in forks.iter() {
+        for fork in &self.forks {
             // Grab all current proposals transactions hashes
             let proposals_txs =
                 fork.overlay.lock().unwrap().get_blocks_txs_hashes(&fork.proposals)?;
@@ -687,7 +603,7 @@ impl Consensus {
 
         // Iterate over all forks again to remove unproposed txs from
         // their mempools.
-        for fork in forks.iter_mut() {
+        for fork in self.forks.iter_mut() {
             fork.mempool.retain(|tx| proposed_txs.contains(tx));
         }
 

src/validator/mod.rs

@@ -78,7 +78,7 @@ pub struct ValidatorConfig {
 }
 
 /// Atomic pointer to validator.
-pub type ValidatorPtr = Arc<Validator>;
+pub type ValidatorPtr = Arc<RwLock<Validator>>;
 
 /// This struct represents a DarkFi validator node.
 pub struct Validator {
@@ -87,7 +87,7 @@ pub struct Validator {
     /// Hot/Live data used by the consensus algorithm
     pub consensus: Consensus,
     /// Flag signalling if the node is synced
-    pub synced: RwLock<bool>,
+    pub synced: bool,
     /// Flag to enable tx fee verification
     pub verify_fees: bool,
 }
@@ -129,12 +129,12 @@ impl Validator {
         )?;
 
         // Create the actual state
-        let state = Arc::new(Self {
+        let state = Arc::new(RwLock::new(Self {
             blockchain,
             consensus,
-            synced: RwLock::new(false),
+            synced: false,
             verify_fees: config.verify_fees,
-        });
+        }));
 
         info!(target: "validator::new", "Finished initializing validator");
         Ok(state)
@@ -149,9 +149,8 @@ impl Validator {
     /// not needed.
     pub async fn calculate_fee(&self, tx: &Transaction, verify_fee: bool) -> Result<u64> {
         // Grab the best fork to verify against
-        let forks = self.consensus.forks.read().await;
-        let fork = forks[best_fork_index(&forks)?].full_clone()?;
-        drop(forks);
+        let index = best_fork_index(&self.consensus.forks)?;
+        let fork = self.consensus.forks[index].full_clone()?;
 
         // Map of ZK proof verifying keys for the transaction
         let mut vks: HashMap<[u8; 32], HashMap<String, VerifyingKey>> = HashMap::new();
@@ -166,7 +165,7 @@ impl Validator {
         let verify_result = verify_transaction(
             &fork.overlay,
             next_block_height,
-            self.consensus.module.read().await.target,
+            self.consensus.module.target,
             tx,
             &mut MerkleTree::new(1),
             &mut vks,
@@ -182,7 +181,7 @@ impl Validator {
     ///
     /// Note: Always remember to purge new trees from the database if
     /// not needed.
-    pub async fn append_tx(&self, tx: &Transaction, write: bool) -> Result<()> {
+    pub async fn append_tx(&mut self, tx: &Transaction, write: bool) -> Result<()> {
         let tx_hash = tx.hash();
 
         // Check if we have already seen this tx
@@ -199,11 +198,8 @@ impl Validator {
         let tx_vec = [tx.clone()];
         let mut valid = false;
 
-        // Grab a lock over current consensus forks state
-        let mut forks = self.consensus.forks.write().await;
-
         // Iterate over node forks to verify transaction validity in their overlays
-        for fork in forks.iter_mut() {
+        for fork in self.consensus.forks.iter_mut() {
             // Clone fork state
             let fork_clone = fork.full_clone()?;
 
@@ -214,7 +210,7 @@ impl Validator {
             let verify_result = verify_transactions(
                 &fork_clone.overlay,
                 next_block_height,
-                self.consensus.module.read().await.target,
+                self.consensus.module.target,
                 &tx_vec,
                 &mut MerkleTree::new(1),
                 self.verify_fees,
@@ -236,9 +232,6 @@ impl Validator {
             }
         }
 
-        // Drop forks lock
-        drop(forks);
-
         // Return error if transaction is not valid for any fork
         if !valid {
             return Err(TxVerifyFailed::ErroneousTxs(tx_vec.to_vec()).into())
@@ -258,7 +251,7 @@ impl Validator {
     ///
     /// Note: Always remember to purge new trees from the database if
     /// not needed.
-    pub async fn purge_pending_txs(&self) -> Result<()> {
+    pub async fn purge_pending_txs(&mut self) -> Result<()> {
         info!(target: "validator::purge_pending_txs", "Removing invalid transactions from pending transactions store...");
 
         // Check if any pending transactions exist
@@ -268,9 +261,6 @@ impl Validator {
             return Ok(())
         }
 
-        // Grab a lock over current consensus forks state
-        let mut forks = self.consensus.forks.write().await;
-
         let mut removed_txs = vec![];
         for tx in pending_txs {
             let tx_hash = tx.hash();
@@ -278,7 +268,7 @@ impl Validator {
             let mut valid = false;
 
             // Iterate over node forks to verify transaction validity in their overlays
-            for fork in forks.iter_mut() {
+            for fork in self.consensus.forks.iter_mut() {
                 // Clone fork state
                 let fork_clone = fork.full_clone()?;
 
@@ -289,7 +279,7 @@ impl Validator {
                 let verify_result = verify_transactions(
                     &fork_clone.overlay,
                     next_block_height,
-                    self.consensus.module.read().await.target,
+                    self.consensus.module.target,
                     &tx_vec,
                     &mut MerkleTree::new(1),
                     self.verify_fees,
@@ -316,9 +306,6 @@ impl Validator {
             }
         }
 
-        // Drop forks lock
-        drop(forks);
-
         if removed_txs.is_empty() {
             info!(target: "validator::purge_pending_txs", "No erroneous transactions found");
             return Ok(())
@@ -329,48 +316,31 @@ impl Validator {
         Ok(())
     }
 
-    /// The node locks its consensus state and tries to append provided proposal.
-    pub async fn append_proposal(&self, proposal: &Proposal) -> Result<()> {
-        // Grab append lock so we restrict concurrent calls of this function
-        let append_lock = self.consensus.append_lock.write().await;
-
-        // Execute append
-        let result = self.consensus.append_proposal(proposal, self.verify_fees).await;
-
-        // Release append lock
-        drop(append_lock);
-
-        result
+    /// The node tries to append provided proposal to its consensus
+    /// state.
+    pub async fn append_proposal(&mut self, proposal: &Proposal) -> Result<()> {
+        self.consensus.append_proposal(proposal, self.verify_fees).await
     }
 
     /// The node checks if best fork can be confirmed.
     /// If proposals can be confirmed, node appends them to canonical,
     /// and resets the current forks.
-    pub async fn confirmation(&self) -> Result<Vec<BlockInfo>> {
-        // Grab append lock so no new proposals can be appended while
-        // we execute confirmation
-        let append_lock = self.consensus.append_lock.write().await;
-
+    pub async fn confirmation(&mut self) -> Result<Vec<BlockInfo>> {
         info!(target: "validator::confirmation", "Performing confirmation check");
 
         // Grab best fork index that can be confirmed
         let confirmed_fork = match self.consensus.confirmation().await {
             Ok(f) => f,
-            Err(e) => {
-                drop(append_lock);
-                return Err(e)
-            }
+            Err(e) => return Err(e),
         };
         if confirmed_fork.is_none() {
             info!(target: "validator::confirmation", "No proposals can be confirmed");
-            drop(append_lock);
             return Ok(vec![])
         }
 
         // Grab the actual best fork
         let confirmed_fork = confirmed_fork.unwrap();
-        let mut forks = self.consensus.forks.write().await;
-        let fork = &mut forks[confirmed_fork];
+        let fork = &mut self.consensus.forks[confirmed_fork];
 
         // Find the excess over confirmation threshold
         let excess = (fork.proposals.len() - self.consensus.confirmation_threshold) + 1;
@@ -388,7 +358,7 @@ impl Validator {
             fork.overlay.lock().unwrap().get_blocks_by_hash(&confirmed_proposals)?;
 
         // Apply confirmed proposals diffs and update PoW module
-        let mut module = self.consensus.module.write().await;
+        let mut module = self.consensus.module.clone();
         let mut confirmed_txs = vec![];
         let mut state_inverse_diffs_heights = vec![];
         let mut state_inverse_diffs = vec![];
@@ -402,8 +372,7 @@ impl Validator {
             state_inverse_diffs_heights.push(confirmed_blocks[index].header.height);
             state_inverse_diffs.push(diffs[index].inverse());
         }
-        drop(module);
-        drop(forks);
+        self.consensus.module = module;
 
         // Store the block inverse diffs
         self.blockchain
@@ -414,9 +383,6 @@ impl Validator {
         self.consensus.reset_forks(&confirmed_proposals, &confirmed_fork, &confirmed_txs).await?;
         info!(target: "validator::confirmation", "Confirmation completed!");
 
-        // Release append lock
-        drop(append_lock);
-
         Ok(confirmed_blocks)
     }
 
@@ -432,7 +398,7 @@ impl Validator {
     /// holding the updated module. Always remember to purge new trees
     /// from the database if not needed.
     pub async fn add_checkpoint_blocks(
-        &self,
+        &mut self,
         blocks: &[BlockInfo],
         headers: &[HeaderHash],
     ) -> Result<()> {
@@ -450,7 +416,7 @@ impl Validator {
         let mut current_hashes_rank = last_difficulty.ranks.hashes_rank;
 
         // Grab current PoW module to validate each block
-        let mut module = self.consensus.module.read().await.clone();
+        let mut module = self.consensus.module.clone();
 
         // Keep track of all blocks transactions to remove them from pending txs store
         let mut removed_txs = vec![];
@@ -525,11 +491,10 @@ impl Validator {
         self.blockchain.remove_pending_txs(&removed_txs)?;
 
         // Update PoW module
-        *self.consensus.module.write().await = module.clone();
+        self.consensus.module = module.clone();
 
         // Update forks
-        *self.consensus.forks.write().await =
-            vec![Fork::new(self.blockchain.clone(), module).await?];
+        self.consensus.forks = vec![Fork::new(self.blockchain.clone(), module).await?];
 
         Ok(())
     }
@@ -540,7 +505,7 @@ impl Validator {
     /// Note: this function should only be used in tests when we don't
     /// want to perform consensus logic and always remember to purge
     /// new trees from the database if not needed.
-    pub async fn add_test_blocks(&self, blocks: &[BlockInfo]) -> Result<()> {
+    pub async fn add_test_blocks(&mut self, blocks: &[BlockInfo]) -> Result<()> {
         debug!(target: "validator::add_test_blocks", "Instantiating BlockchainOverlay");
         let overlay = BlockchainOverlay::new(&self.blockchain)?;
 
@@ -553,7 +518,7 @@ impl Validator {
         let mut current_hashes_rank = last_difficulty.ranks.hashes_rank;
 
         // Grab current PoW module to validate each block
-        let mut module = self.consensus.module.read().await.clone();
+        let mut module = self.consensus.module.clone();
 
         // Keep track of all blocks transactions to remove them from pending txs store
         let mut removed_txs = vec![];
@@ -633,7 +598,7 @@ impl Validator {
         self.purge_pending_txs().await?;
 
         // Update PoW module
-        *self.consensus.module.write().await = module;
+        self.consensus.module = module;
 
         Ok(())
     }
@@ -834,21 +799,17 @@ impl Validator {
 
     /// Auxiliary function to retrieve current best fork next block height.
     pub async fn best_fork_next_block_height(&self) -> Result<u32> {
-        let forks = self.consensus.forks.read().await;
-        let fork = &forks[best_fork_index(&forks)?];
+        let index = best_fork_index(&self.consensus.forks)?;
+        let fork = &self.consensus.forks[index];
         let next_block_height = fork.get_next_block_height()?;
-        drop(forks);
 
         Ok(next_block_height)
     }
 
     /// Auxiliary function to reset the validator blockchain and consensus states
     /// to the provided block height.
-    pub async fn reset_to_height(&self, height: u32) -> Result<()> {
+    pub async fn reset_to_height(&mut self, height: u32) -> Result<()> {
         info!(target: "validator::reset_to_height", "Resetting validator to height: {height}");
-        // Grab append lock so no new proposals can be appended while we execute a reset
-        let append_lock = self.consensus.append_lock.write().await;
-
         // Reset our databasse to provided height
         self.blockchain.reset_to_height(height)?;
 
@@ -858,9 +819,6 @@ impl Validator {
         // Purge current forks
         self.consensus.purge_forks().await?;
 
-        // Release append lock
-        drop(append_lock);
-
         info!(target: "validator::reset_to_height", "Validator reset successfully!");
 
         Ok(())
@@ -875,9 +833,6 @@ impl Validator {
         pow_fixed_difficulty: Option<BigUint>,
     ) -> Result<()> {
         info!(target: "validator::rebuild_block_difficulties", "Rebuilding validator block difficulties...");
-        // Grab append lock so no new proposals can be appended while we execute the rebuild
-        let append_lock = self.consensus.append_lock.write().await;
-
         // Clear the block difficulties tree
         self.blockchain.blocks.difficulty.clear()?;
 
@@ -945,9 +900,6 @@ impl Validator {
         // Flush the database
         self.blockchain.sled_db.flush()?;
 
-        // Release append lock
-        drop(append_lock);
-
         info!(target: "validator::rebuild_block_difficulties", "Validator block difficulties rebuilt successfully!");
 
         Ok(())