darkfid/registry: guard state behind a single lock

bin/darkfid/src/registry/mod.rs

@@ -57,166 +57,37 @@ use model::{
     PowRewardV1Zk,
 };
 
-/// Atomic pointer to the DarkFi node miners registry.
-pub type DarkfiMinersRegistryPtr = Arc<DarkfiMinersRegistry>;
+/// Atomic pointer to the DarkFi node miners registry state.
+pub type DarkfiMinersRegistryStatePtr = Arc<RwLock<DarkfiMinersRegistryState>>;
 
-/// DarkFi node miners registry.
-pub struct DarkfiMinersRegistry {
-    /// Blockchain network
-    pub network: Network,
+/// DarkFi node miners registry state.
+pub struct DarkfiMinersRegistryState {
     /// PowRewardV1 ZK data
     pub powrewardv1_zk: PowRewardV1Zk,
     /// Mining block templates of each wallet config
-    pub block_templates: RwLock<HashMap<String, BlockTemplate>>,
+    pub block_templates: HashMap<String, BlockTemplate>,
     /// Active native clients mapped to their job information.
     /// This client information includes their wallet template key,
     /// recipient configuration, current mining job key(job id) and
     /// its connection publisher. For native jobs the job key is the
     /// hex encoded header hash.
-    pub jobs: RwLock<HashMap<String, MinerClient>>,
+    pub jobs: HashMap<String, MinerClient>,
     /// Active merge mining jobs mapped to the wallet template they
     /// represent. The key(job id) is the the header template hash.
-    pub mm_jobs: RwLock<HashMap<String, String>>,
-    /// Submission lock so we can queue up submissions process
-    pub submit_lock: RwLock<()>,
-    /// Stratum JSON-RPC background task
-    stratum_rpc_task: StoppableTaskPtr,
-    /// Stratum JSON-RPC connection tracker
-    pub stratum_rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
-    /// HTTP JSON-RPC background task
-    mm_rpc_task: StoppableTaskPtr,
-    /// HTTP JSON-RPC connection tracker
-    pub mm_rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
+    pub mm_jobs: HashMap<String, String>,
 }
 
-impl DarkfiMinersRegistry {
-    /// Initialize a DarkFi node miners registry.
-    pub async fn init(
-        network: Network,
-        validator: &ValidatorPtr,
-    ) -> Result<DarkfiMinersRegistryPtr> {
-        info!(
-            target: "darkfid::registry::mod::DarkfiMinersRegistry::init",
-            "Initializing a new DarkFi node miners registry..."
-        );
-
+impl DarkfiMinersRegistryState {
+    pub async fn new(validator: &ValidatorPtr) -> Result<DarkfiMinersRegistryStatePtr> {
         // Generate the PowRewardV1 ZK data
         let powrewardv1_zk = PowRewardV1Zk::new(validator).await?;
 
-        // Generate the stratum JSON-RPC background task and its
-        // connections tracker.
-        let stratum_rpc_task = StoppableTask::new();
-        let stratum_rpc_connections = Mutex::new(HashSet::new());
-
-        // Generate the HTTP JSON-RPC background task and its
-        // connections tracker.
-        let mm_rpc_task = StoppableTask::new();
-        let mm_rpc_connections = Mutex::new(HashSet::new());
-
-        info!(
-            target: "darkfid::registry::mod::DarkfiMinersRegistry::init",
-            "DarkFi node miners registry generated successfully!"
-        );
-
-        Ok(Arc::new(Self {
-            network,
+        Ok(Arc::new(RwLock::new(Self {
             powrewardv1_zk,
-            block_templates: RwLock::new(HashMap::new()),
-            jobs: RwLock::new(HashMap::new()),
-            mm_jobs: RwLock::new(HashMap::new()),
-            submit_lock: RwLock::new(()),
-            stratum_rpc_task,
-            stratum_rpc_connections,
-            mm_rpc_task,
-            mm_rpc_connections,
-        }))
-    }
-
-    /// Start the DarkFi node miners registry for provided DarkFi node
-    /// instance.
-    pub fn start(
-        &self,
-        executor: &ExecutorPtr,
-        node: &DarkfiNodePtr,
-        stratum_rpc_settings: &Option<RpcSettings>,
-        mm_rpc_settings: &Option<RpcSettings>,
-    ) -> Result<()> {
-        info!(
-            target: "darkfid::registry::mod::DarkfiMinersRegistry::start",
-            "Starting the DarkFi node miners registry..."
-        );
-
-        // Start the stratum server JSON-RPC task
-        if let Some(stratum_rpc) = stratum_rpc_settings {
-            info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::start", "Starting Stratum JSON-RPC server");
-            let node_ = node.clone();
-            self.stratum_rpc_task.clone().start(
-                listen_and_serve::<StratumRpcHandler>(stratum_rpc.clone(), node.clone(), None, executor.clone()),
-                |res| async move {
-                    match res {
-                        Ok(()) | Err(Error::RpcServerStopped) => <DarkfiNode as RequestHandler<StratumRpcHandler>>::stop_connections(&node_).await,
-                        Err(e) => error!(target: "darkfid::registry::mod::DarkfiMinersRegistry::start", "Failed starting Stratum JSON-RPC server: {e}"),
-                    }
-                },
-                Error::RpcServerStopped,
-                executor.clone(),
-            );
-        } else {
-            // Create a dummy task
-            self.stratum_rpc_task.clone().start(
-                async { Ok(()) },
-                |_| async { /* Do nothing */ },
-                Error::RpcServerStopped,
-                executor.clone(),
-            );
-        }
-
-        // Start the merge mining JSON-RPC task
-        if let Some(mm_rpc) = mm_rpc_settings {
-            info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::start", "Starting merge mining JSON-RPC server");
-            let node_ = node.clone();
-            self.mm_rpc_task.clone().start(
-                listen_and_serve::<MmRpcHandler>(mm_rpc.clone(), node.clone(), None, executor.clone()),
-                |res| async move {
-                    match res {
-                        Ok(()) | Err(Error::RpcServerStopped) => <DarkfiNode as RequestHandler<MmRpcHandler>>::stop_connections(&node_).await,
-                        Err(e) => error!(target: "darkfid::registry::mod::DarkfiMinersRegistry::start", "Failed starting merge mining JSON-RPC server: {e}"),
-                    }
-                },
-                Error::RpcServerStopped,
-                executor.clone(),
-            );
-        } else {
-            // Create a dummy task
-            self.mm_rpc_task.clone().start(
-                async { Ok(()) },
-                |_| async { /* Do nothing */ },
-                Error::RpcServerStopped,
-                executor.clone(),
-            );
-        }
-
-        info!(
-            target: "darkfid::registry::mod::DarkfiMinersRegistry::start",
-            "DarkFi node miners registry started successfully!"
-        );
-
-        Ok(())
-    }
-
-    /// Stop the DarkFi node miners registry.
-    pub async fn stop(&self) {
-        info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::stop", "Terminating DarkFi node miners registry...");
-
-        // Stop the Stratum JSON-RPC task
-        info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::stop", "Stopping Stratum JSON-RPC server...");
-        self.stratum_rpc_task.stop().await;
-
-        // Stop the merge mining JSON-RPC task
-        info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::stop", "Stopping merge mining JSON-RPC server...");
-        self.mm_rpc_task.stop().await;
-
-        info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::stop", "DarkFi node miners registry terminated successfully!");
+            block_templates: HashMap::new(),
+            jobs: HashMap::new(),
+            mm_jobs: HashMap::new(),
+        })))
     }
 
     /// Create a registry record for provided wallet config. If the
@@ -226,16 +97,13 @@ impl DarkfiMinersRegistry {
     /// Note: Always remember to purge new trees from the database if
     /// not needed.
     async fn create_template(
-        &self,
+        &mut self,
         validator: &Validator,
         wallet: &String,
         config: &MinerRewardsRecipientConfig,
     ) -> Result<BlockTemplate> {
-        // Grab a lock over current templates
-        let mut block_templates = self.block_templates.write().await;
-
         // Check if a template already exists for this wallet
-        if let Some(block_template) = block_templates.get(wallet) {
+        if let Some(block_template) = self.block_templates.get(wallet) {
             return Ok(block_template.clone())
         }
 
@@ -253,7 +121,7 @@ impl DarkfiMinersRegistry {
         .await?;
 
         // Create the new registry record
-        block_templates.insert(wallet.clone(), block_template.clone());
+        self.block_templates.insert(wallet.clone(), block_template.clone());
 
         // Print the new template wallet information
         let recipient_str = format!("{}", config.recipient);
@@ -274,14 +142,11 @@ impl DarkfiMinersRegistry {
 
     /// Register a new miner and create its job.
     pub async fn register_miner(
-        &self,
+        &mut self,
         validator: &Validator,
         wallet: &String,
         config: &MinerRewardsRecipientConfig,
     ) -> Result<(String, String, JsonValue, JsonSubscriber)> {
-        // Grab a lock over current native jobs
-        let mut jobs = self.jobs.write().await;
-
         // Create wallet template
         let block_template = self.create_template(validator, wallet, config).await?;
 
@@ -289,21 +154,18 @@ impl DarkfiMinersRegistry {
         let (job_id, job) = block_template.job_notification();
         let (client_id, client) = MinerClient::new(wallet, config, &job_id);
         let publisher = client.publisher.clone();
-        jobs.insert(client_id.clone(), client);
+        self.jobs.insert(client_id.clone(), client);
 
         Ok((client_id, job_id, job, publisher))
     }
 
     /// Register a new merge miner and create its job.
     pub async fn register_merge_miner(
-        &self,
+        &mut self,
         validator: &Validator,
         wallet: &String,
         config: &MinerRewardsRecipientConfig,
     ) -> Result<(String, f64)> {
-        // Grab a lock over current mm jobs
-        let mut jobs = self.mm_jobs.write().await;
-
         // Create wallet template
         let block_template = self.create_template(validator, wallet, config).await?;
 
@@ -311,7 +173,7 @@ impl DarkfiMinersRegistry {
         // create the job record.
         let block_template_hash = block_template.block.header.template_hash().as_string();
         let difficulty = block_template.difficulty;
-        jobs.insert(block_template_hash.clone(), wallet.clone());
+        self.mm_jobs.insert(block_template_hash.clone(), wallet.clone());
 
         Ok((block_template_hash, difficulty))
     }
@@ -346,22 +208,13 @@ impl DarkfiMinersRegistry {
         Ok(())
     }
 
-    /// Refresh outdated jobs in the provided registry maps based on
-    /// provided validator state.
-    ///
-    /// Note: Always remember to purge new trees from the database if
-    /// not needed.
-    pub async fn refresh_jobs(
-        &self,
-        block_templates: &mut HashMap<String, BlockTemplate>,
-        jobs: &mut HashMap<String, MinerClient>,
-        mm_jobs: &mut HashMap<String, String>,
-        validator: &Validator,
-    ) -> Result<()> {
+    /// Refresh outdated jobs in the registry based on provided
+    /// validator state.
+    pub async fn refresh(&mut self, validator: &Validator) -> Result<()> {
         // Find inactive native jobs and drop them
         let mut dropped_jobs = vec![];
         let mut active_templates = HashSet::new();
-        for (client_id, client) in jobs.iter() {
+        for (client_id, client) in self.jobs.iter() {
             // Clear inactive client publisher subscribers. If none
             // exists afterwards, the client is considered inactive so
             // we mark it for drop.
@@ -373,7 +226,7 @@ impl DarkfiMinersRegistry {
             // Mark client block template as active
             active_templates.insert(client.wallet.clone());
         }
-        jobs.retain(|client_id, _| !dropped_jobs.contains(client_id));
+        self.jobs.retain(|client_id, _| !dropped_jobs.contains(client_id));
 
         // Grab validator best current fork and its last proposal for
         // checks.
@@ -383,10 +236,10 @@ impl DarkfiMinersRegistry {
         // Find mm jobs not extending the best current fork and drop
         // them.
         let mut dropped_mm_jobs = vec![];
-        for (job_id, wallet) in mm_jobs.iter() {
+        for (job_id, wallet) in self.mm_jobs.iter() {
             // Grab its wallet template. Its safe to unwrap here since
             // we know the job exists.
-            let block_template = block_templates.get(wallet).unwrap();
+            let block_template = self.block_templates.get(wallet).unwrap();
 
             // Check if it extends current best fork
             if block_template.block.header.previous == last_proposal {
@@ -398,22 +251,22 @@ impl DarkfiMinersRegistry {
             // it for drop.
             dropped_mm_jobs.push(job_id.clone());
         }
-        mm_jobs.retain(|job_id, _| !dropped_mm_jobs.contains(job_id));
+        self.mm_jobs.retain(|job_id, _| !dropped_mm_jobs.contains(job_id));
 
         // Drop inactive templates. Merge miners will create a new
         // template and job on next poll.
-        block_templates.retain(|wallet, _| active_templates.contains(wallet));
+        self.block_templates.retain(|wallet, _| active_templates.contains(wallet));
 
         // Return if no wallets templates exists.
-        if block_templates.is_empty() {
+        if self.block_templates.is_empty() {
             return Ok(())
         }
 
         // Iterate over active clients to refresh their jobs, if needed
-        for (job_id, client) in jobs.iter_mut() {
+        for (job_id, client) in self.jobs.iter_mut() {
             // Grab its wallet template. Its safe to unwrap here since
             // we know the job exists.
-            let block_template = block_templates.get_mut(&client.wallet).unwrap();
+            let block_template = self.block_templates.get_mut(&client.wallet).unwrap();
 
             // Check if it extends current best fork
             if block_template.block.header.previous == last_proposal {
@@ -474,32 +327,11 @@ impl DarkfiMinersRegistry {
         Ok(())
     }
 
-    /// Refresh outdated jobs in the registry based on provided
-    /// validator state.
-    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;
-        let mut jobs = self.jobs.write().await;
-        let mut mm_jobs = self.mm_jobs.write().await;
-
-        // Refresh jobs
-        self.refresh_jobs(&mut block_templates, &mut jobs, &mut mm_jobs, validator).await?;
-
-        // Release registry locks
-        drop(block_templates);
-        drop(jobs);
-        drop(mm_jobs);
-        drop(submit_lock);
-
-        Ok(())
-    }
-
     /// Auxilliary function to retrieve all current block templates
     /// newly opened trees.
-    pub fn new_trees(&self, block_templates: &HashMap<String, BlockTemplate>) -> BTreeSet<IVec> {
+    pub fn new_trees(&self) -> BTreeSet<IVec> {
         let mut new_trees = BTreeSet::new();
-        for block_template in block_templates.values() {
+        for block_template in self.block_templates.values() {
             for new_tree in &block_template.new_trees {
                 new_trees.insert(new_tree.clone());
             }
@@ -509,12 +341,9 @@ impl DarkfiMinersRegistry {
 
     /// Auxilliary function to retrieve all current block templates
     /// transactions hashes.
-    pub fn proposed_transactions(
-        &self,
-        block_templates: &HashMap<String, BlockTemplate>,
-    ) -> HashSet<TransactionHash> {
+    pub fn proposed_transactions(&self) -> HashSet<TransactionHash> {
         let mut proposed_txs = HashSet::new();
-        for block_template in block_templates.values() {
+        for block_template in self.block_templates.values() {
             for tx in &block_template.block.txs {
                 proposed_txs.insert(tx.hash());
             }
@@ -522,3 +351,149 @@ impl DarkfiMinersRegistry {
         proposed_txs
     }
 }
+
+/// Atomic pointer to the DarkFi node miners registry.
+pub type DarkfiMinersRegistryPtr = Arc<DarkfiMinersRegistry>;
+
+/// DarkFi node miners registry.
+pub struct DarkfiMinersRegistry {
+    /// Blockchain network
+    pub network: Network,
+    /// Registry state
+    pub state: DarkfiMinersRegistryStatePtr,
+    /// Stratum JSON-RPC background task
+    stratum_rpc_task: StoppableTaskPtr,
+    /// Stratum JSON-RPC connection tracker
+    pub stratum_rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
+    /// HTTP JSON-RPC background task
+    mm_rpc_task: StoppableTaskPtr,
+    /// HTTP JSON-RPC connection tracker
+    pub mm_rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
+}
+
+impl DarkfiMinersRegistry {
+    /// Initialize a DarkFi node miners registry.
+    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 registry state
+        let state = DarkfiMinersRegistryState::new(validator).await?;
+
+        // Generate the stratum JSON-RPC background task and its
+        // connections tracker.
+        let stratum_rpc_task = StoppableTask::new();
+        let stratum_rpc_connections = Mutex::new(HashSet::new());
+
+        // Generate the HTTP JSON-RPC background task and its
+        // connections tracker.
+        let mm_rpc_task = StoppableTask::new();
+        let mm_rpc_connections = Mutex::new(HashSet::new());
+
+        info!(
+            target: "darkfid::registry::mod::DarkfiMinersRegistry::init",
+            "DarkFi node miners registry generated successfully!"
+        );
+
+        Ok(Arc::new(Self {
+            network,
+            state,
+            stratum_rpc_task,
+            stratum_rpc_connections,
+            mm_rpc_task,
+            mm_rpc_connections,
+        }))
+    }
+
+    /// Start the DarkFi node miners registry for provided DarkFi node
+    /// instance.
+    pub fn start(
+        &self,
+        executor: &ExecutorPtr,
+        node: &DarkfiNodePtr,
+        stratum_rpc_settings: &Option<RpcSettings>,
+        mm_rpc_settings: &Option<RpcSettings>,
+    ) -> Result<()> {
+        info!(
+            target: "darkfid::registry::mod::DarkfiMinersRegistry::start",
+            "Starting the DarkFi node miners registry..."
+        );
+
+        // Start the stratum server JSON-RPC task
+        if let Some(stratum_rpc) = stratum_rpc_settings {
+            info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::start", "Starting Stratum JSON-RPC server");
+            let node_ = node.clone();
+            self.stratum_rpc_task.clone().start(
+                listen_and_serve::<StratumRpcHandler>(stratum_rpc.clone(), node.clone(), None, executor.clone()),
+                |res| async move {
+                    match res {
+                        Ok(()) | Err(Error::RpcServerStopped) => <DarkfiNode as RequestHandler<StratumRpcHandler>>::stop_connections(&node_).await,
+                        Err(e) => error!(target: "darkfid::registry::mod::DarkfiMinersRegistry::start", "Failed starting Stratum JSON-RPC server: {e}"),
+                    }
+                },
+                Error::RpcServerStopped,
+                executor.clone(),
+            );
+        } else {
+            // Create a dummy task
+            self.stratum_rpc_task.clone().start(
+                async { Ok(()) },
+                |_| async { /* Do nothing */ },
+                Error::RpcServerStopped,
+                executor.clone(),
+            );
+        }
+
+        // Start the merge mining JSON-RPC task
+        if let Some(mm_rpc) = mm_rpc_settings {
+            info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::start", "Starting merge mining JSON-RPC server");
+            let node_ = node.clone();
+            self.mm_rpc_task.clone().start(
+                listen_and_serve::<MmRpcHandler>(mm_rpc.clone(), node.clone(), None, executor.clone()),
+                |res| async move {
+                    match res {
+                        Ok(()) | Err(Error::RpcServerStopped) => <DarkfiNode as RequestHandler<MmRpcHandler>>::stop_connections(&node_).await,
+                        Err(e) => error!(target: "darkfid::registry::mod::DarkfiMinersRegistry::start", "Failed starting merge mining JSON-RPC server: {e}"),
+                    }
+                },
+                Error::RpcServerStopped,
+                executor.clone(),
+            );
+        } else {
+            // Create a dummy task
+            self.mm_rpc_task.clone().start(
+                async { Ok(()) },
+                |_| async { /* Do nothing */ },
+                Error::RpcServerStopped,
+                executor.clone(),
+            );
+        }
+
+        info!(
+            target: "darkfid::registry::mod::DarkfiMinersRegistry::start",
+            "DarkFi node miners registry started successfully!"
+        );
+
+        Ok(())
+    }
+
+    /// Stop the DarkFi node miners registry.
+    pub async fn stop(&self) {
+        info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::stop", "Terminating DarkFi node miners registry...");
+
+        // Stop the Stratum JSON-RPC task
+        info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::stop", "Stopping Stratum JSON-RPC server...");
+        self.stratum_rpc_task.stop().await;
+
+        // Stop the merge mining JSON-RPC task
+        info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::stop", "Stopping merge mining JSON-RPC server...");
+        self.mm_rpc_task.stop().await;
+
+        info!(target: "darkfid::registry::mod::DarkfiMinersRegistry::stop", "DarkFi node miners registry terminated successfully!");
+    }
+}

bin/darkfid/src/rpc/stratum.rs

@@ -190,17 +190,23 @@ impl DarkfiNode {
             target: "darkfid::rpc::rpc_stratum::stratum_login",
             "[RPC-STRATUM] Got login from {wallet} ({agent})",
         );
-        let (client_id, job_id, job, publisher) =
-            match self.registry.register_miner(&validator, wallet, &config).await {
-                Ok(p) => p,
-                Err(e) => {
-                    error!(
-                        target: "darkfid::rpc::rpc_stratum::stratum_login",
-                        "[RPC-STRATUM] Failed to register miner: {e}",
-                    );
-                    return JsonResponse::new(JsonValue::from(HashMap::new()), id).into()
-                }
-            };
+        let (client_id, job_id, job, publisher) = match self
+            .registry
+            .state
+            .write()
+            .await
+            .register_miner(&validator, wallet, &config)
+            .await
+        {
+            Ok(p) => p,
+            Err(e) => {
+                error!(
+                    target: "darkfid::rpc::rpc_stratum::stratum_login",
+                    "[RPC-STRATUM] Failed to register miner: {e}",
+                );
+                return JsonResponse::new(JsonValue::from(HashMap::new()), id).into()
+            }
+        };
 
         // Now we have the new job, we ship it to RPC
         info!(
@@ -246,9 +252,6 @@ impl DarkfiNode {
             return miner_status_response(id, "rejected")
         }
 
-        // Grab registry submissions lock
-        let submit_lock = self.registry.submit_lock.write().await;
-
         // Parse request params
         let Some(params) = params.get::<HashMap<String, JsonValue>>() else {
             return JsonError::new(InvalidParams, None, id).into()
@@ -263,8 +266,8 @@ impl DarkfiNode {
         };
 
         // If we don't know about this client, we can just abort here
-        let mut jobs = self.registry.jobs.write().await;
-        let Some(client) = jobs.get(client_id) else {
+        let mut registry = self.registry.state.write().await;
+        let Some(client) = registry.jobs.get(client_id) else {
             return miner_status_response(id, "rejected")
         };
 
@@ -281,11 +284,11 @@ impl DarkfiNode {
         if &client.job != job_id {
             return miner_status_response(id, "rejected")
         }
+        let wallet = client.wallet.clone();
 
         // If this client job wallet template doesn't exist, we can
         // just abort here.
-        let mut block_templates = self.registry.block_templates.write().await;
-        let Some(block_template) = block_templates.get_mut(&client.wallet) else {
+        let Some(block_template) = registry.block_templates.get(&wallet) else {
             return miner_status_response(id, "rejected")
         };
 
@@ -328,9 +331,13 @@ impl DarkfiNode {
         block.header.nonce = nonce;
         block.sign(&block_template.secret);
 
+        // Keep the template in memory so we can safely refernce the
+        // registry.
+        let mut block_template = block_template.clone();
+
         // Submit the new block through the registry
         if let Err(e) =
-            self.registry.submit(&mut validator, &self.subscribers, &self.p2p_handler, block).await
+            registry.submit(&mut validator, &self.subscribers, &self.p2p_handler, block).await
         {
             error!(
                 target: "darkfid::rpc::rpc_stratum::stratum_submit",
@@ -338,34 +345,19 @@ impl DarkfiNode {
             );
 
             // Try to refresh the jobs before returning error
-            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, &validator)
-                .await
-            {
+            if let Err(e) = registry.refresh(&validator).await {
                 error!(
                     target: "darkfid::rpc::rpc_stratum::stratum_submit",
                     "[RPC-STRATUM] Error refreshing registry jobs: {e}",
                 );
             }
 
-            // Release all locks
-            drop(block_templates);
-            drop(jobs);
-            drop(mm_jobs);
-            drop(submit_lock);
-
             return miner_status_response(id, "rejected")
         }
 
         // Mark block as submitted
         block_template.submitted = true;
-
-        // Release all locks
-        drop(block_templates);
-        drop(jobs);
-        drop(submit_lock);
+        registry.block_templates.insert(wallet, block_template);
 
         miner_status_response(id, "OK")
     }
@@ -396,7 +388,7 @@ impl DarkfiNode {
         };
 
         // If we don't know about this client job, we can just abort here
-        if !self.registry.jobs.read().await.contains_key(client_id) {
+        if !self.registry.state.read().await.jobs.contains_key(client_id) {
             return server_error(RpcError::MinerUnknownClient, id, None)
         };
 

bin/darkfid/src/rpc/tx.rs

@@ -193,26 +193,11 @@ impl DarkfiNode {
             return server_error(RpcError::NotSynced, id, None)
         }
 
-        // Grab node registry locks
-        let submit_lock = self.registry.submit_lock.write().await;
-        let block_templates = self.registry.block_templates.write().await;
-        let jobs = self.registry.jobs.write().await;
-        let mm_jobs = self.registry.mm_jobs.write().await;
+        // Retrieve registry transactions
+        let registry_txs = self.registry.state.read().await.proposed_transactions();
 
         // Purge all unproposed pending transactions from the database
-        let result = validator
-            .consensus
-            .purge_unproposed_pending_txs(self.registry.proposed_transactions(&block_templates))
-            .await;
-
-        // Release registry locks
-        drop(block_templates);
-        drop(jobs);
-        drop(mm_jobs);
-        drop(submit_lock);
-
-        // Check result
-        if let Err(e) = result {
+        if let Err(e) = validator.consensus.purge_unproposed_pending_txs(registry_txs).await {
             error!(target: "darkfid::rpc::tx_clean_pending", "Failed removing pending txs: {e}");
             return JsonError::new(InternalError, None, id).into()
         };

bin/darkfid/src/rpc/xmr.rs

@@ -186,7 +186,8 @@ impl DarkfiNode {
         };
 
         // Check if we already have this job
-        if self.registry.mm_jobs.read().await.contains_key(&aux_hash.to_string()) {
+        let mut registry = self.registry.state.write().await;
+        if registry.mm_jobs.contains_key(&aux_hash.to_string()) {
             return JsonResponse::new(JsonValue::from(HashMap::new()), id).into()
         }
 
@@ -226,7 +227,7 @@ impl DarkfiNode {
 
         // Register the new merge miner
         let (job_id, difficulty) =
-            match self.registry.register_merge_miner(&validator, wallet, &config).await {
+            match registry.register_merge_miner(&validator, wallet, &config).await {
                 Ok(p) => p,
                 Err(e) => {
                     error!(
@@ -291,9 +292,6 @@ impl DarkfiNode {
             return miner_status_response(id, "rejected")
         }
 
-        // Grab registry submissions lock
-        let submit_lock = self.registry.submit_lock.write().await;
-
         // Parse request params
         let Some(params) = params.get::<HashMap<String, JsonValue>>() else {
             return JsonError::new(InvalidParams, None, id).into()
@@ -311,15 +309,15 @@ impl DarkfiNode {
         }
 
         // If we don't know about this mm job, we can just abort here
-        let mut mm_jobs = self.registry.mm_jobs.write().await;
-        let Some(wallet) = mm_jobs.get(aux_hash) else {
+        let mut registry = self.registry.state.write().await;
+        let Some(wallet) = registry.mm_jobs.get(aux_hash) else {
             return miner_status_response(id, "rejected")
         };
+        let wallet = wallet.clone();
 
         // If this job wallet template doesn't exist, we can just
         // abort here.
-        let mut block_templates = self.registry.block_templates.write().await;
-        let Some(block_template) = block_templates.get_mut(wallet) else {
+        let Some(block_template) = registry.block_templates.get(&wallet) else {
             return miner_status_response(id, "rejected")
         };
 
@@ -423,9 +421,13 @@ impl DarkfiNode {
         block.header.pow_data = PowData::Monero(monero_pow_data);
         block.sign(&block_template.secret);
 
+        // Keep the template in memory so we can safely refernce the
+        // registry.
+        let mut block_template = block_template.clone();
+
         // Submit the new block through the registry
         if let Err(e) =
-            self.registry.submit(&mut validator, &self.subscribers, &self.p2p_handler, block).await
+            registry.submit(&mut validator, &self.subscribers, &self.p2p_handler, block).await
         {
             error!(
                 target: "darkfid::rpc::rpc_xmr::xmr_merge_mining_submit_solution",
@@ -433,36 +435,19 @@ impl DarkfiNode {
             );
 
             // Try to refresh the jobs before returning error
-            let mut jobs = self.registry.jobs.write().await;
-            if let Err(e) = self
-                .registry
-                .refresh_jobs(&mut block_templates, &mut jobs, &mut mm_jobs, &validator)
-                .await
-            {
+            if let Err(e) = registry.refresh(&validator).await {
                 error!(
                     target: "darkfid::rpc::rpc_xmr::xmr_merge_mining_submit_solution",
                     "[RPC-XMR] Error refreshing registry jobs: {e}",
                 );
             }
 
-            // Release all locks
-            drop(block_templates);
-            drop(jobs);
-            drop(mm_jobs);
-            drop(submit_lock);
-            drop(validator);
-
             return miner_status_response(id, "rejected")
         }
 
         // Mark block as submitted
         block_template.submitted = true;
-
-        // Release all locks
-        drop(block_templates);
-        drop(mm_jobs);
-        drop(submit_lock);
-        drop(validator);
+        registry.block_templates.insert(wallet, block_template);
 
         miner_status_response(id, "accepted")
     }

bin/darkfid/src/task/consensus.rs

@@ -29,10 +29,7 @@ use darkfi_serial::serialize_async;
 use tracing::{error, info};
 
 use crate::{
-    task::{
-        garbage_collect::{garbage_collect_task, purge_unreferenced_trees},
-        sync_task,
-    },
+    task::{garbage_collect::garbage_collect_task, sync_task},
     DarkfiNodePtr,
 };
 
@@ -196,7 +193,8 @@ async fn consensus_task(
         };
 
         // Refresh mining registry
-        if let Err(e) = node.registry.refresh(&validator).await {
+        let mut registry = node.registry.state.write().await;
+        if let Err(e) = registry.refresh(&validator).await {
             error!(target: "darkfid", "Failed refreshing mining block templates: {e}")
         }
 
@@ -204,11 +202,12 @@ async fn consensus_task(
             continue
         }
 
-        // Grab the append lock so no other proposal gets processed
-        // while the node is purging all unreferenced contract trees
-        // from the database.
-        purge_unreferenced_trees(&validator, &node.registry).await;
-        drop(validator);
+        // Purge all unreferenced contract trees fro the database
+        if let Err(e) =
+            validator.consensus.purge_unreferenced_trees(&mut registry.new_trees()).await
+        {
+            error!(target: "darkfid::task::garbage_collect::purge_unreferenced_trees", "Purging unreferenced contract trees from the database failed: {e}");
+        }
 
         let mut notif_blocks = Vec::with_capacity(confirmed.len());
         for block in confirmed {

bin/darkfid/src/task/garbage_collect.rs

@@ -16,15 +16,11 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use darkfi::{
-    error::TxVerifyFailed,
-    validator::{verification::verify_transactions, Validator},
-    Error, Result,
-};
+use darkfi::{error::TxVerifyFailed, validator::verification::verify_transactions, Error, Result};
 use darkfi_sdk::crypto::MerkleTree;
 use tracing::{debug, error, info};
 
-use crate::{DarkfiMinersRegistryPtr, DarkfiNodePtr};
+use crate::DarkfiNodePtr;
 
 /// Async task used for purging erroneous pending transactions from the nodes mempool.
 pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
@@ -172,28 +168,3 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
     info!(target: "darkfid::task::garbage_collect_task", "Garbage collection finished successfully!");
     Ok(())
 }
-
-/// Auxiliary function to purge all unreferenced contract trees from
-/// the node database.
-pub async fn purge_unreferenced_trees(validator: &Validator, registry: &DarkfiMinersRegistryPtr) {
-    // Grab node registry locks
-    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) = validator
-        .consensus
-        .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}");
-    }
-
-    // Release registry locks
-    drop(block_templates);
-    drop(jobs);
-    drop(mm_jobs);
-    drop(submit_lock);
-}

bin/darkfid/src/task/unknown_proposal.rs

@@ -386,7 +386,7 @@ async fn handle_reorg(
     };
 
     // Refresh mining registry
-    if let Err(e) = node.registry.refresh(&validator).await {
+    if let Err(e) = node.registry.state.write().await.refresh(&validator).await {
         error!(target: "darkfid::task::handle_reorg", "Failed refreshing mining block templates: {e}")
     }