add clear unused transaction logic

roles/jd-server/src/lib/job_declarator/message_handler.rs

@@ -7,14 +7,17 @@ use roles_logic_sv2::{
         ProvideMissingTransactions, ProvideMissingTransactionsSuccess, SubmitSolutionJd,
     },
     parsers::JobDeclaration,
+    utils::Mutex,
 };
-use std::{convert::TryInto, io::Cursor};
+use std::{convert::TryInto, io::Cursor, sync::Arc};
 use stratum_common::bitcoin::{Transaction, Txid};
 pub type SendTo = SendTo_<JobDeclaration<'static>, ()>;
+use crate::mempool::JDsMempool;
+
 use super::{signed_token, TransactionState};
 use roles_logic_sv2::{errors::Error, parsers::PoolMessages as AllMessages};
 use stratum_common::bitcoin::consensus::Decodable;
-use tracing::info;
+use tracing::{debug, info};
 
 use super::JobDeclaratorDownstream;
 
@@ -61,6 +64,9 @@ impl ParseClientJobDeclarationMessages for JobDeclaratorDownstream {
     }
 
     fn handle_declare_mining_job(&mut self, message: DeclareMiningJob) -> Result<SendTo, Error> {
+        if let Some(old_mining_job) = self.declared_mining_job.0.take() {
+            clear_declared_mining_job(old_mining_job, &message, self.mempool.clone())?;
+        }
         // the transactions that are present in the mempool are stored here, that is sent to the
         // mempool which use the rpc client to retrieve the whole data for each transaction.
         // The unknown transactions is a vector that contains the transactions that are not in the
@@ -219,3 +225,55 @@ impl ParseClientJobDeclarationMessages for JobDeclaratorDownstream {
         Ok(SendTo::None(Some(m)))
     }
 }
+
+pub fn clear_declared_mining_job(
+    old_mining_job: DeclareMiningJob,
+    new_mining_job: &DeclareMiningJob,
+    mempool: Arc<Mutex<JDsMempool>>,
+) -> Result<(), Error> {
+    let old_transactions = old_mining_job.tx_short_hash_list.inner_as_ref();
+    let new_transactions = new_mining_job.tx_short_hash_list.inner_as_ref();
+
+    if old_transactions.is_empty() {
+        info!("No transactions to remove from mempool");
+        return Ok(());
+    }
+
+    let nonce = old_mining_job.tx_short_hash_nonce;
+
+    let result = mempool.safe_lock(|mempool_| -> Result<(), Error> {
+        let short_ids_map = mempool_
+            .to_short_ids(nonce)
+            .ok_or(Error::JDSMissingTransactions)?;
+
+        for short_id in old_transactions {
+            if !new_transactions.contains(&short_id) {
+                if let Some(transaction_with_hash) = short_ids_map.get(short_id) {
+                    let txid = transaction_with_hash.id;
+                    match mempool_.mempool.get_mut(&txid) {
+                        Some(Some((transaction, counter))) => {
+                            if *counter > 1 {
+                                *counter -= 1;
+                                info!("Fat transaction {:?} decreased mempool counter because job id {:?} was dropped", txid, old_mining_job.request_id);
+                            } else {
+                                mempool_.mempool.remove(&txid);
+                                info!("Fat transaction {:?} in job with request id {:?} removed from mempool", txid, old_mining_job.request_id);
+                            }
+                        },
+                        Some(None) => info!("Thin transaction {:?} in job with request id {:?} removed from mempool", txid, old_mining_job.request_id),
+                        None => {},
+                    }
+                } else {
+                    debug!("Transaction with short id {:?} not found in mempool while clearing old jobs", short_id);
+                }
+            }
+        }
+        Ok(())
+    });
+
+    if let Err(err) = result {
+        return Err(Error::PoisonLock(err.to_string()));
+    }
+
+    Ok(())
+}

roles/jd-server/src/lib/job_declarator/mod.rs

@@ -145,7 +145,7 @@ impl JobDeclaratorDownstream {
                     .ok_or(Box::new(JdsError::ImpossibleToReconstructBlock(
                         "Txid found in jds mempool but transactions not present".to_string(),
                     )))?;
-                transactions_list.push(tx);
+                transactions_list.push(tx.0);
             } else {
                 return Err(Box::new(JdsError::ImpossibleToReconstructBlock(
                     "Unknown transaction".to_string(),

roles/jd-server/src/lib/mempool/mod.rs

@@ -12,12 +12,12 @@ use stratum_common::{bitcoin, bitcoin::hash_types::Txid};
 #[derive(Clone, Debug)]
 pub struct TransactionWithHash {
     pub id: Txid,
-    pub tx: Option<Transaction>,
+    pub tx: Option<(Transaction, u32)>,
 }
 
 #[derive(Clone, Debug)]
 pub struct JDsMempool {
-    pub mempool: HashMap<Txid, Option<Transaction>>,
+    pub mempool: HashMap<Txid, Option<(Transaction, u32)>>,
     auth: mini_rpc_client::Auth,
     url: String,
     new_block_receiver: Receiver<String>,
@@ -50,7 +50,7 @@ impl JDsMempool {
         new_block_receiver: Receiver<String>,
     ) -> Self {
         let auth = mini_rpc_client::Auth::new(username, password);
-        let empty_mempool: HashMap<Txid, Option<Transaction>> = HashMap::new();
+        let empty_mempool: HashMap<Txid, Option<(Transaction, u32)>> = HashMap::new();
         JDsMempool {
             mempool: empty_mempool,
             auth,
@@ -82,28 +82,31 @@ impl JDsMempool {
                     .get_raw_transaction(&txid.to_string(), None)
                     .await
                     .map_err(JdsMempoolError::Rpc)?;
-                let _ =
-                    self_.safe_lock(|a| a.mempool.insert(transaction.txid(), Some(transaction)));
+                let _ = self_
+                    .safe_lock(|a| a.mempool.insert(transaction.txid(), Some((transaction, 1))));
             }
         }
 
         // fill in the mempool the transactions given in input
         for transaction in transactions {
-            let _ = self_.safe_lock(|a| a.mempool.insert(transaction.txid(), Some(transaction)));
+            let _ =
+                self_.safe_lock(|a| a.mempool.insert(transaction.txid(), Some((transaction, 1))));
         }
         Ok(())
     }
 
     pub async fn update_mempool(self_: Arc<Mutex<Self>>) -> Result<(), JdsMempoolError> {
-
         let client = self_
             .safe_lock(|x| x.get_client())?
             .ok_or(JdsMempoolError::NoClient)?;
 
         let mempool = client.get_raw_mempool().await?;
         let raw_mempool_txids: Result<Vec<Txid>, _> = mempool
             .into_iter()
-            .map(|id| Txid::from_str(&id).map_err(|err| JdsMempoolError::Rpc(RpcError::Deserialization(err.to_string()))))
+            .map(|id| {
+                Txid::from_str(&id)
+                    .map_err(|err| JdsMempoolError::Rpc(RpcError::Deserialization(err.to_string())))
+            })
             .collect();
 
         let raw_mempool_txids = raw_mempool_txids?;