|
|
@@ -16,158 +16,154 @@
|
|
|
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
*/
|
|
|
|
|
|
-use darkfi::{error::TxVerifyFailed, validator::verification::verify_transactions, Error, Result};
|
|
|
+use std::collections::HashMap;
|
|
|
+
|
|
|
+use darkfi::{
|
|
|
+ blockchain::parse_record, tx::Transaction, validator::verification::verify_transaction,
|
|
|
+ zk::VerifyingKey, Result,
|
|
|
+};
|
|
|
use darkfi_sdk::{crypto::MerkleTree, tx::TransactionHash};
|
|
|
+use smol::channel::Receiver;
|
|
|
use tracing::{debug, error, info};
|
|
|
|
|
|
use crate::DarkfiNodePtr;
|
|
|
|
|
|
-/// Async task used for purging erroneous pending transactions from the nodes mempool.
|
|
|
-pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
|
|
|
+/// Auxiliary macro to check if channel receiver is empty so we can
|
|
|
+/// abort current iteration.
|
|
|
+macro_rules! trigger_queue_check {
|
|
|
+ ($receiver:ident, $label:tt) => {
|
|
|
+ if !$receiver.is_empty() {
|
|
|
+ continue $label
|
|
|
+ }
|
|
|
+ };
|
|
|
+}
|
|
|
+
|
|
|
+/// Async task used for purging unreferenced trees and erroneous
|
|
|
+/// pending transactions from the nodes mempool.
|
|
|
+pub async fn garbage_collect_task(receiver: Receiver<()>, node: DarkfiNodePtr) -> Result<()> {
|
|
|
info!(target: "darkfid::task::garbage_collect_task", "Starting garbage collection task...");
|
|
|
|
|
|
- // 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 = TransactionHash::none();
|
|
|
- let mut txs = match validator
|
|
|
- .blockchain
|
|
|
- .transactions
|
|
|
- .get_after_pending(&last_checked, node.txs_batch_size)
|
|
|
- {
|
|
|
- Ok(txs) => txs,
|
|
|
- Err(e) => {
|
|
|
- error!(
|
|
|
- target: "darkfid::task::garbage_collect_task",
|
|
|
- "Uproposed transactions retrieval failed: {e}"
|
|
|
- );
|
|
|
- return Ok(())
|
|
|
+ 'outer: loop {
|
|
|
+ // Wait for a new trigger
|
|
|
+ if let Err(e) = receiver.recv().await {
|
|
|
+ error!(target: "darkfid::task::garbage_collect_task", "recv fail: {e}");
|
|
|
+ continue
|
|
|
+ };
|
|
|
+
|
|
|
+ // Purge all unreferenced contract trees from the database
|
|
|
+ trigger_queue_check!(receiver, 'outer);
|
|
|
+ debug!(target: "darkfid::task::garbage_collect_task", "Starting garbage collection iteration...");
|
|
|
+ if let Err(e) = node
|
|
|
+ .validator
|
|
|
+ .read()
|
|
|
+ .await
|
|
|
+ .consensus
|
|
|
+ .purge_unreferenced_trees(&mut node.registry.state.read().await.new_trees())
|
|
|
+ .await
|
|
|
+ {
|
|
|
+ error!(target: "darkfid::task::garbage_collect_task", "Purging unreferenced contract trees from the database failed: {e}");
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ debug!(target: "darkfid::task::garbage_collect_task", "Unreferenced trees purged successfully, retrieving pending transactions...");
|
|
|
+
|
|
|
+ // Check if our mempool is empty
|
|
|
+ trigger_queue_check!(receiver, 'outer);
|
|
|
+ let validator = node.validator.read().await;
|
|
|
+ if validator.blockchain.transactions.pending.is_empty() {
|
|
|
+ debug!(target: "darkfid::task::garbage_collect_task", "No pending transactions to process");
|
|
|
+ continue
|
|
|
}
|
|
|
- };
|
|
|
|
|
|
- // Check if we have transactions to process
|
|
|
- if txs.is_empty() {
|
|
|
- info!(target: "darkfid::task::garbage_collect_task", "Garbage collection finished successfully!");
|
|
|
- return Ok(())
|
|
|
- }
|
|
|
+ // Grab validator current best fork and an iterator over its
|
|
|
+ // pending transactions so we don't hold the validator lock.
|
|
|
+ let pending = validator.blockchain.transactions.pending.iter();
|
|
|
+ let fork = match validator.best_current_fork().await {
|
|
|
+ Ok(f) => f,
|
|
|
+ Err(e) => {
|
|
|
+ error!(target: "darkfid::task::garbage_collect_task", "Retrieving validator current best fork failed: {e}");
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ };
|
|
|
+ let verify_fees = validator.verify_fees;
|
|
|
+ drop(validator);
|
|
|
|
|
|
- // 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 {
|
|
|
- last_checked = tx.hash();
|
|
|
- let tx_vec = [tx.clone()];
|
|
|
- let mut valid = false;
|
|
|
-
|
|
|
- // Grab a lock over current consensus forks state
|
|
|
- let mut validator = node.validator.write().await;
|
|
|
-
|
|
|
- // Iterate over them to verify transaction validity in their overlays
|
|
|
- for fork in validator.consensus.forks.iter_mut() {
|
|
|
- // Clone forks' overlay
|
|
|
- let overlay = match fork.overlay.lock().unwrap().full_clone() {
|
|
|
- Ok(o) => o,
|
|
|
- Err(e) => {
|
|
|
- error!(
|
|
|
- target: "darkfid::task::garbage_collect_task",
|
|
|
- "Overlay full clone creation failed: {e}"
|
|
|
- );
|
|
|
- return Err(e)
|
|
|
- }
|
|
|
- };
|
|
|
+ // Transactions Merkle tree
|
|
|
+ trigger_queue_check!(receiver, 'outer);
|
|
|
+ let mut tree = MerkleTree::new(1);
|
|
|
+
|
|
|
+ // Map of ZK proof verifying keys for the current transactions
|
|
|
+ // batch.
|
|
|
+ let mut vks: HashMap<[u8; 32], HashMap<String, VerifyingKey>> = HashMap::new();
|
|
|
+
|
|
|
+ // Grab forks' next block height
|
|
|
+ let next_block_height = match fork.get_next_block_height() {
|
|
|
+ Ok(h) => h,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "darkfid::task::garbage_collect_task",
|
|
|
+ "Next fork block height retrieval failed: {e}"
|
|
|
+ );
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ };
|
|
|
|
|
|
- // Grab all current proposals transactions hashes
|
|
|
- let proposals_txs =
|
|
|
- match overlay.lock().unwrap().get_blocks_txs_hashes(&fork.proposals) {
|
|
|
- Ok(txs) => txs,
|
|
|
- Err(e) => {
|
|
|
- error!(
|
|
|
- target: "darkfid::task::garbage_collect_task",
|
|
|
- "Proposal transactions retrieval failed: {e}"
|
|
|
- );
|
|
|
- return Err(e)
|
|
|
- }
|
|
|
- };
|
|
|
-
|
|
|
- // If the hash is contained in the proposals transactions vec, skip it
|
|
|
- if proposals_txs.contains(&last_checked) {
|
|
|
+ // Iterate over all pending transactions
|
|
|
+ for record in pending {
|
|
|
+ trigger_queue_check!(receiver, 'outer);
|
|
|
+ let record = match record {
|
|
|
+ Ok(r) => r,
|
|
|
+ Err(e) => {
|
|
|
+ error!(target: "darkfid::task::garbage_collect_task", "Failed retrieving pending tx: {e}");
|
|
|
+ continue 'outer
|
|
|
+ }
|
|
|
+ };
|
|
|
+ let (tx_hash, tx) = match parse_record::<TransactionHash, Transaction>(record) {
|
|
|
+ Ok((h, t)) => (h, t),
|
|
|
+ Err(e) => {
|
|
|
+ error!(target: "darkfid::task::garbage_collect_task", "Failed parsing pending tx: {e}");
|
|
|
continue
|
|
|
}
|
|
|
-
|
|
|
- // Grab forks' next block height
|
|
|
- let next_block_height = match fork.get_next_block_height() {
|
|
|
- Ok(h) => h,
|
|
|
- Err(e) => {
|
|
|
- error!(
|
|
|
- target: "darkfid::task::garbage_collect_task",
|
|
|
- "Next fork block height retrieval failed: {e}"
|
|
|
- );
|
|
|
- return Err(e)
|
|
|
- }
|
|
|
+ };
|
|
|
+
|
|
|
+ // If the transaction has already been proposed, remove it
|
|
|
+ trigger_queue_check!(receiver, 'outer);
|
|
|
+ debug!(target: "darkfid::task::garbage_collect_task", "Checking transaction: {tx_hash}");
|
|
|
+ if fork.overlay.lock().unwrap().transactions.contains(&tx_hash)? {
|
|
|
+ debug!(target: "darkfid::task::garbage_collect_task", "Transaction {tx_hash} has already been proposed, removing...");
|
|
|
+ if let Err(e) = fork.blockchain.remove_pending_txs_hashes(&[tx_hash]) {
|
|
|
+ error!(target: "darkfid::task::garbage_collect_task", "Failed removing pending tx: {e}");
|
|
|
};
|
|
|
+ continue
|
|
|
+ }
|
|
|
|
|
|
- // Verify transaction
|
|
|
- let result = verify_transactions(
|
|
|
- &overlay,
|
|
|
- next_block_height,
|
|
|
- target,
|
|
|
- &tx_vec,
|
|
|
- &mut MerkleTree::new(1),
|
|
|
- false,
|
|
|
- )
|
|
|
- .await;
|
|
|
-
|
|
|
- // Check result
|
|
|
- match result {
|
|
|
- Ok(_) => valid = true,
|
|
|
- Err(Error::TxVerifyFailed(TxVerifyFailed::ErroneousTxs(_))) => {
|
|
|
- /* Do nothing */
|
|
|
- }
|
|
|
- Err(e) => {
|
|
|
- error!(
|
|
|
- target: "darkfid::task::garbage_collect_task",
|
|
|
- "Verifying transaction {last_checked} failed: {e}"
|
|
|
- );
|
|
|
- return Err(e)
|
|
|
- }
|
|
|
- }
|
|
|
+ // Update the verifying keys map
|
|
|
+ trigger_queue_check!(receiver, 'outer);
|
|
|
+ for call in &tx.calls {
|
|
|
+ vks.entry(call.data.contract_id.to_bytes()).or_default();
|
|
|
}
|
|
|
|
|
|
- // Remove transaction if its invalid for all the forks
|
|
|
- if !valid {
|
|
|
- debug!(target: "darkfid::task::garbage_collect_task", "Removing invalid transaction: {last_checked}");
|
|
|
- if let Err(e) = validator.blockchain.remove_pending_txs_hashes(&[last_checked]) {
|
|
|
- error!(
|
|
|
- target: "darkfid::task::garbage_collect_task",
|
|
|
- "Removing invalid transaction {last_checked} failed: {e}"
|
|
|
- );
|
|
|
+ // Verify the transaction against current state
|
|
|
+ trigger_queue_check!(receiver, 'outer);
|
|
|
+ fork.overlay.lock().unwrap().checkpoint();
|
|
|
+ let result = verify_transaction(
|
|
|
+ &fork.overlay,
|
|
|
+ next_block_height,
|
|
|
+ fork.module.target,
|
|
|
+ &tx,
|
|
|
+ &mut tree,
|
|
|
+ &mut vks,
|
|
|
+ verify_fees,
|
|
|
+ )
|
|
|
+ .await;
|
|
|
+ fork.overlay.lock().unwrap().revert_to_checkpoint();
|
|
|
+ if let Err(e) = result {
|
|
|
+ debug!(target: "darkfid::task::garbage_collect_task", "Pending transaction {tx_hash} verification failed: {e}");
|
|
|
+ if let Err(e) = fork.blockchain.remove_pending_txs_hashes(&[tx_hash]) {
|
|
|
+ error!(target: "darkfid::task::garbage_collect_task", "Failed removing pending tx: {e}");
|
|
|
};
|
|
|
+ continue
|
|
|
}
|
|
|
+ debug!(target: "darkfid::task::garbage_collect_task", "Pending transaction {tx_hash} verification successfully.");
|
|
|
}
|
|
|
-
|
|
|
- // Grab next batch
|
|
|
- txs = match node
|
|
|
- .validator
|
|
|
- .read()
|
|
|
- .await
|
|
|
- .blockchain
|
|
|
- .transactions
|
|
|
- .get_after_pending(&last_checked, node.txs_batch_size)
|
|
|
- {
|
|
|
- Ok(txs) => txs,
|
|
|
- Err(e) => {
|
|
|
- error!(
|
|
|
- target: "darkfid::task::garbage_collect_task",
|
|
|
- "Uproposed transactions next batch retrieval failed: {e}"
|
|
|
- );
|
|
|
- break
|
|
|
- }
|
|
|
- };
|
|
|
}
|
|
|
-
|
|
|
- info!(target: "darkfid::task::garbage_collect_task", "Garbage collection finished successfully!");
|
|
|
- Ok(())
|
|
|
}
|