darkfid/registry: split jobs into native and mm for simpler handling and only generate new templates if they don't extend best current fork

bin/darkfid/src/registry/mod.rs

@@ -64,13 +64,15 @@ pub struct DarkfiMinersRegistry {
     pub powrewardv1_zk: PowRewardV1Zk,
     /// Mining block templates of each wallet config
     pub block_templates: RwLock<HashMap<String, BlockTemplate>>,
-    /// Active mining jobs mapped to the wallet template they
-    /// represent. For native jobs the key(job id) is the hex
-    /// encoded header hash, while for merge mining jobs it's
-    /// the header template hash.
-    pub jobs: RwLock<HashMap<String, String>>,
-    /// Active native clients mapped to their information.
-    pub clients: RwLock<HashMap<String, MinerClient>>,
+    /// 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>>,
+    /// 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
@@ -114,7 +116,7 @@ impl DarkfiMinersRegistry {
             powrewardv1_zk,
             block_templates: RwLock::new(HashMap::new()),
             jobs: RwLock::new(HashMap::new()),
-            clients: RwLock::new(HashMap::new()),
+            mm_jobs: RwLock::new(HashMap::new()),
             submit_lock: RwLock::new(()),
             stratum_rpc_task,
             stratum_rpc_connections,
@@ -272,24 +274,20 @@ impl DarkfiMinersRegistry {
         validator: &ValidatorPtr,
         wallet: &String,
         config: &MinerRewardsRecipientConfig,
-    ) -> Result<(String, BlockTemplate, JsonSubscriber)> {
-        // Grab a lock over current jobs and clients
+    ) -> Result<(String, String, JsonValue, JsonSubscriber)> {
+        // Grab a lock over current native jobs
         let mut jobs = self.jobs.write().await;
-        let mut clients = self.clients.write().await;
 
         // Create wallet template
         let block_template = self.create_template(validator, wallet, config).await?;
 
-        // Grab the hex encoded block hash and create the job record
-        let block_hash = hex::encode(block_template.block.header.hash().inner()).to_string();
-        jobs.insert(block_hash.clone(), wallet.clone());
-
-        // Create the client record
-        let (client_id, client) = MinerClient::new(wallet, config, &block_hash);
+        // Grab the hex encoded block hash and create the client job record
+        let (job_id, job) = block_template.job_notification();
+        let (client_id, client) = MinerClient::new(wallet, config, &job_id);
         let publisher = client.publisher.clone();
-        clients.insert(client_id.clone(), client);
+        jobs.insert(client_id.clone(), client);
 
-        Ok((client_id, block_template, publisher))
+        Ok((client_id, job_id, job, publisher))
     }
 
     /// Register a new merge miner and create its job.
@@ -299,8 +297,8 @@ impl DarkfiMinersRegistry {
         wallet: &String,
         config: &MinerRewardsRecipientConfig,
     ) -> Result<(String, f64)> {
-        // Grab a lock over current jobs
-        let mut jobs = self.jobs.write().await;
+        // 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?;
@@ -346,97 +344,77 @@ impl DarkfiMinersRegistry {
     /// Refresh outdated jobs in the registry based on provided
     /// validator state.
     pub async fn refresh(&self, validator: &ValidatorPtr) -> Result<()> {
-        // Grab locks
+        // Grab registry locks
         let submit_lock = self.submit_lock.write().await;
-        let mut clients = self.clients.write().await;
         let mut jobs = self.jobs.write().await;
+        let mut mm_jobs = self.mm_jobs.write().await;
         let mut block_templates = self.block_templates.write().await;
 
-        // Find inactive clients
-        let mut dropped_clients = vec![];
-        let mut active_clients_jobs = vec![];
-        for (client_id, client) in clients.iter() {
+        // 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() {
+            // Clear inactive client publisher subscribers. If none
+            // exists afterwards, the client is considered inactive so
+            // we mark it for drop.
             if client.publisher.publisher.clear_inactive().await {
-                dropped_clients.push(client_id.clone());
+                dropped_jobs.push(client_id.clone());
                 continue
             }
-            active_clients_jobs.push(client.job.clone());
-        }
 
-        // Drop inactive clients and their jobs
-        for client_id in dropped_clients {
-            // Its safe to unwrap here since the client key is from the
-            // previous loop.
-            let client = clients.remove(&client_id).unwrap();
-            let wallet = jobs.remove(&client.job).unwrap();
-            block_templates.remove(&wallet);
+            // Mark client block template as active
+            active_templates.insert(client.wallet.clone());
         }
+        jobs.retain(|client_id, _| !dropped_jobs.contains(client_id));
 
-        // Return if no clients exists. Merge miners will create a new
-        // template and job on next poll.
-        if clients.is_empty() {
-            *jobs = HashMap::new();
-            *block_templates = HashMap::new();
-            return Ok(())
-        }
-
-        // Find inactive jobs (not referenced by clients)
-        let mut dropped_jobs = vec![];
-        let mut active_wallets = vec![];
-        for (job, wallet) in jobs.iter() {
-            if !active_clients_jobs.contains(job) {
-                dropped_jobs.push(job.clone());
+        // Grab validator best current fork and its last proposal for
+        // checks.
+        let extended_fork = validator.best_current_fork().await?;
+        let last_proposal = extended_fork.last_proposal()?.hash;
+
+        // 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() {
+            // Grab its wallet template. Its safe to unwrap here since
+            // we know the job exists.
+            let block_template = block_templates.get(wallet).unwrap();
+
+            // Check if it extends current best fork
+            if block_template.block.header.previous == last_proposal {
+                active_templates.insert(wallet.clone());
                 continue
             }
-            active_wallets.push(wallet.clone());
-        }
 
-        // Drop inactive jobs
-        for job in dropped_jobs {
-            jobs.remove(&job);
+            // This mm job doesn't extend current best fork so we mark
+            // it for drop.
+            dropped_mm_jobs.push(job_id.clone());
         }
+        mm_jobs.retain(|job_id, _| !dropped_mm_jobs.contains(job_id));
 
-        // Return if no jobs exists. Merge miners will create a new
+        // Drop inactive templates. Merge miners will create a new
         // template and job on next poll.
-        if jobs.is_empty() {
-            *block_templates = HashMap::new();
-            return Ok(())
-        }
+        block_templates.retain(|wallet, _| active_templates.contains(wallet));
 
-        // Find inactive wallets templates
-        let mut dropped_wallets = vec![];
-        for wallet in block_templates.keys() {
-            if !active_wallets.contains(wallet) {
-                dropped_wallets.push(wallet.clone());
-            }
-        }
-
-        // Drop inactive wallets templates
-        for wallet in dropped_wallets {
-            block_templates.remove(&wallet);
-        }
-
-        // Return if no wallets templates exists. Merge miners will
-        // create a new template and job on next poll.
+        // Return if no wallets templates exists.
         if block_templates.is_empty() {
             return Ok(())
         }
 
-        // Grab validator best current fork
-        let extended_fork = validator.best_current_fork().await?;
+        // Iterate over active clients to refresh their jobs, if needed
+        for (_, client) in 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();
+
+            // Check if it extends current best fork
+            if block_template.block.header.previous == last_proposal {
+                continue
+            }
 
-        // Iterate over active clients to refresh their jobs
-        for (_, client) in clients.iter_mut() {
             // Clone the fork so each client generates over a new one
             let mut extended_fork = extended_fork.full_clone()?;
 
-            // Drop its current job. Its safe to unwrap here since we
-            // know the job exists.
-            let wallet = jobs.remove(&client.job).unwrap();
-            // Drop its current template. Its safe to unwrap here since
-            // we know the template exists.
-            block_templates.remove(&wallet);
-
             // Generate the next block template
             let result = generate_next_block_template(
                 &mut extended_fork,
@@ -451,7 +429,7 @@ impl DarkfiMinersRegistry {
             extended_fork.overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
 
             // Check result
-            let block_template = result?;
+            *block_template = result?;
 
             // Print the updated template wallet information
             let recipient_str = format!("{}", client.config.recipient);
@@ -470,10 +448,6 @@ impl DarkfiMinersRegistry {
             // Create the new job notification
             let (job, notification) = block_template.job_notification();
 
-            // Create the new registry records
-            block_templates.insert(wallet.clone(), block_template);
-            jobs.insert(job.clone(), wallet);
-
             // Update the client record
             client.job = job;
 
@@ -481,8 +455,9 @@ impl DarkfiMinersRegistry {
             client.publisher.notify(notification).await;
         }
 
-        // Release all locks
+        // Release registry locks
         drop(block_templates);
+        drop(mm_jobs);
         drop(jobs);
         drop(submit_lock);
 

bin/darkfid/src/registry/model.rs

@@ -169,6 +169,8 @@ impl BlockTemplate {
 /// Auxiliary structure representing a native miner client record.
 #[derive(Debug, Clone)]
 pub struct MinerClient {
+    /// Miner wallet template key
+    pub wallet: String,
     /// Miner recipient configuration
     pub config: MinerRewardsRecipientConfig,
     /// Current mining job
@@ -184,7 +186,15 @@ impl MinerClient {
         hasher.update(&NanoTimestamp::current_time().inner().to_le_bytes());
         let client_id = hex::encode(hasher.finalize().as_bytes()).to_string();
         let publisher = JsonSubscriber::new("job");
-        (client_id, Self { config: config.clone(), job: job.to_owned(), publisher })
+        (
+            client_id,
+            Self {
+                wallet: String::from(wallet),
+                config: config.clone(),
+                job: job.to_owned(),
+                publisher,
+            },
+        )
     }
 }
 

bin/darkfid/src/rpc/mod.rs

@@ -91,16 +91,6 @@ impl RequestHandler<DefaultRpcHandler> for DarkfiNode {
             "tx.clean_pending" => self.tx_clean_pending(req.id, req.params).await,
             "tx.calculate_fee" => self.tx_calculate_fee(req.id, req.params).await,
 
-            // TODO: drop
-            // =============
-            // Miner methods
-            // =============
-            /*
-            "miner.get_current_mining_randomx_key" => self.miner_get_current_mining_randomx_key(req.id, req.params).await,
-            "miner.get_header" => self.miner_get_header(req.id, req.params).await,
-            "miner.submit_solution" => self.miner_submit_solution(req.id, req.params).await,
-            */
-
             // ==============
             // Invalid method
             // ==============

bin/darkfid/src/rpc/rpc_stratum.rs

@@ -159,7 +159,7 @@ impl DarkfiNode {
             target: "darkfid::rpc::rpc_stratum::stratum_login",
             "[RPC-STRATUM] Got login from {wallet} ({agent})",
         );
-        let (client_id, block_template, publisher) =
+        let (client_id, job_id, job, publisher) =
             match self.registry.register_miner(&self.validator, wallet, &config).await {
                 Ok(p) => p,
                 Err(e) => {
@@ -172,7 +172,6 @@ impl DarkfiNode {
             };
 
         // Now we have the new job, we ship it to RPC
-        let (job_id, job) = block_template.job_notification();
         info!(
             target: "darkfid::rpc::rpc_stratum::stratum_login",
             "[RPC-STRATUM] Created new mining job for client {client_id}: {job_id}"
@@ -222,8 +221,8 @@ impl DarkfiNode {
         };
 
         // If we don't know about this client, we can just abort here
-        let clients = self.registry.clients.read().await;
-        let Some(client) = clients.get(client_id) else {
+        let jobs = self.registry.jobs.read().await;
+        let Some(client) = jobs.get(client_id) else {
             return server_error(RpcError::MinerUnknownClient, id, None)
         };
 
@@ -235,20 +234,16 @@ impl DarkfiNode {
             return server_error(RpcError::MinerInvalidJobId, id, None)
         };
 
-        // If we don't know about this job or it doesn't match the
-        // client one, we can just abort here
+        // If this job doesn't match the client one, we can just abort
+        // here.
         if &client.job != job_id {
             return server_error(RpcError::MinerUnknownJob, id, None)
         }
-        let jobs = self.registry.jobs.read().await;
-        let Some(wallet) = jobs.get(job_id) else {
-            return server_error(RpcError::MinerUnknownJob, id, None)
-        };
 
-        // If this job wallet template doesn't exist, we can just
-        // abort here.
+        // 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(wallet) else {
+        let Some(block_template) = block_templates.get_mut(&client.wallet) else {
             return server_error(RpcError::MinerUnknownJob, id, None)
         };
 
@@ -364,8 +359,8 @@ impl DarkfiNode {
             return server_error(RpcError::MinerInvalidClientId, id, None)
         };
 
-        // If we don't know about this client, we can just abort here
-        if !self.registry.clients.read().await.contains_key(client_id) {
+        // If we don't know about this client job, we can just abort here
+        if !self.registry.jobs.read().await.contains_key(client_id) {
             return server_error(RpcError::MinerUnknownClient, id, None)
         };
 

bin/darkfid/src/rpc/rpc_xmr.rs

@@ -155,7 +155,7 @@ impl DarkfiNode {
         };
 
         // Check if we already have this job
-        if self.registry.jobs.read().await.contains_key(&aux_hash.to_string()) {
+        if self.registry.mm_jobs.read().await.contains_key(&aux_hash.to_string()) {
             return JsonResponse::new(JsonValue::from(HashMap::new()), id).into()
         }
 
@@ -266,8 +266,8 @@ impl DarkfiNode {
             return server_error(RpcError::MinerInvalidAuxHash, id, None)
         }
 
-        // If we don't know about this job, we can just abort here
-        let jobs = self.registry.jobs.read().await;
+        // If we don't know about this mm job, we can just abort here
+        let jobs = self.registry.mm_jobs.read().await;
         let Some(wallet) = jobs.get(aux_hash) else {
             return server_error(RpcError::MinerUnknownJob, id, None)
         };