/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2026 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
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;
/// 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...");
'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
}
// 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);
// 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> = 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
}
};
// 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::(record) {
Ok((h, t)) => (h, t),
Err(e) => {
error!(target: "darkfid::task::garbage_collect_task", "Failed parsing pending tx: {e}");
continue
}
};
// 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
}
// 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();
}
// 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.");
}
}
}