Replace by fee on mempool (kaspanet#499)

* Replace by fee on mempool with tests

* Add a custom RemovalReason -- ReplacedByFee

* Let `MinerManager` handle replace by fee (RBF) for RPC and P2P

* Refines success conditions and process of RBF policies

* Add an RPC `submit_transaction_replacement` method

* fix fmt

* Fix CLI Build WASM32 error

* Avoid breaking wRPC

* Extend some tests coverage to all priority, orphan & RBF policy combinations

* Let RBF fail early or at least before checking transaction scripts in consensus

* Cleaning

* More cleaning

* Use contextual instead of compute mass in RBF eviction rule

* Avoid collision with another PR

* Avoid collision with another PR (2)

* Extended test coverage of RBF

* Extended test coverage of RBF (2)

* Rename `TransactionBatchValidationArgs` to `TransactionValidationBatchArgs`

* Add comments

* Assert instead of condition

* Add an `RbfPolicy` parameter to mining manager tx validate_and_insert_... fns

* Infer RBF policy from unorphaned transaction

* Apply the RBF policy to all orphan-related cases

* In Rbf allowed mode, check feerate threshold vs all double spends (i.e., compare to the max)

* Minor hashset optimization

* Rename: fee_per_mass -> feerate

* Use rbf policy arg for post processing step as well

* Renames and comments

* Relaxation: fail gracefully if rbf replaced tx is missing (also fixes an edge case where mempool duplication resulted in this scenario)

* Tx id is appended by the caller anyway

---------

Co-authored-by: Tiram <[email protected]>
Co-authored-by: Michael Sutton <[email protected]>

consensus/core/src/api/args.rs

@@ -0,0 +1,47 @@
+use std::collections::HashMap;
+
+use crate::tx::TransactionId;
+
+/// A struct provided to consensus for transaction validation processing calls
+#[derive(Clone, Debug, Default)]
+pub struct TransactionValidationArgs {
+    /// Optional fee/mass threshold above which a bound transaction in not rejected
+    pub feerate_threshold: Option<f64>,
+}
+
+impl TransactionValidationArgs {
+    pub fn new(feerate_threshold: Option<f64>) -> Self {
+        Self { feerate_threshold }
+    }
+}
+
+/// A struct provided to consensus for transactions validation batch processing calls
+pub struct TransactionValidationBatchArgs {
+    tx_args: HashMap<TransactionId, TransactionValidationArgs>,
+}
+
+impl TransactionValidationBatchArgs {
+    const DEFAULT_ARGS: TransactionValidationArgs = TransactionValidationArgs { feerate_threshold: None };
+
+    pub fn new() -> Self {
+        Self { tx_args: HashMap::new() }
+    }
+
+    /// Set some fee/mass threshold for transaction `transaction_id`.
+    pub fn set_feerate_threshold(&mut self, transaction_id: TransactionId, feerate_threshold: f64) {
+        self.tx_args
+            .entry(transaction_id)
+            .and_modify(|x| x.feerate_threshold = Some(feerate_threshold))
+            .or_insert(TransactionValidationArgs::new(Some(feerate_threshold)));
+    }
+
+    pub fn get(&self, transaction_id: &TransactionId) -> &TransactionValidationArgs {
+        self.tx_args.get(transaction_id).unwrap_or(&Self::DEFAULT_ARGS)
+    }
+}
+
+impl Default for TransactionValidationBatchArgs {
+    fn default() -> Self {
+        Self::new()
+    }
+}

consensus/core/src/api/mod.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 
 use crate::{
     acceptance_data::AcceptanceData,
+    api::args::{TransactionValidationArgs, TransactionValidationBatchArgs},
     block::{Block, BlockTemplate, TemplateBuildMode, TemplateTransactionSelector, VirtualStateApproxId},
     blockstatus::BlockStatus,
     coinbase::MinerData,
@@ -25,6 +26,7 @@ use kaspa_hashes::Hash;
 
 pub use self::stats::{BlockCount, ConsensusStats};
 
+pub mod args;
 pub mod counters;
 pub mod stats;
 
@@ -62,14 +64,18 @@ pub trait ConsensusApi: Send + Sync {
     }
 
     /// Populates the mempool transaction with maximally found UTXO entry data and proceeds to full transaction
-    /// validation if all are found. If validation is successful, also [`transaction.calculated_fee`] is expected to be populated.
-    fn validate_mempool_transaction(&self, transaction: &mut MutableTransaction) -> TxResult<()> {
+    /// validation if all are found. If validation is successful, also `transaction.calculated_fee` is expected to be populated.
+    fn validate_mempool_transaction(&self, transaction: &mut MutableTransaction, args: &TransactionValidationArgs) -> TxResult<()> {
         unimplemented!()
     }
 
     /// Populates the mempool transactions with maximally found UTXO entry data and proceeds to full transactions
-    /// validation if all are found. If validation is successful, also [`transaction.calculated_fee`] is expected to be populated.
-    fn validate_mempool_transactions_in_parallel(&self, transactions: &mut [MutableTransaction]) -> Vec<TxResult<()>> {
+    /// validation if all are found. If validation is successful, also `transaction.calculated_fee` is expected to be populated.
+    fn validate_mempool_transactions_in_parallel(
+        &self,
+        transactions: &mut [MutableTransaction],
+        args: &TransactionValidationBatchArgs,
+    ) -> Vec<TxResult<()>> {
         unimplemented!()
     }
 

consensus/core/src/errors/tx.rs

@@ -88,6 +88,11 @@ pub enum TxRuleError {
 
     #[error("calculated contextual mass (including storage mass) {0} is not equal to the committed mass field {1}")]
     WrongMass(u64, u64),
+
+    /// [`TxRuleError::FeerateTooLow`] is not a consensus error but a mempool error triggered by the
+    /// fee/mass RBF validation rule
+    #[error("fee rate per contextual mass gram is not greater than the fee rate of the replaced transaction")]
+    FeerateTooLow,
 }
 
 pub type TxResult<T> = std::result::Result<T, TxRuleError>;

consensus/core/src/tx.rs

@@ -406,6 +406,19 @@ impl<T: AsRef<Transaction>> MutableTransaction<T> {
             *entry = None;
         }
     }
+
+    /// Returns the calculated feerate. The feerate is calculated as the amount of fee
+    /// this transactions pays per gram of the full contextual (compute & storage) mass. The
+    /// function returns a value when calculated fee exists and the contextual mass is greater
+    /// than zero, otherwise `None` is returned.
+    pub fn calculated_feerate(&self) -> Option<f64> {
+        let contextual_mass = self.tx.as_ref().mass();
+        if contextual_mass > 0 {
+            self.calculated_fee.map(|fee| fee as f64 / contextual_mass as f64)
+        } else {
+            None
+        }
+    }
 }
 
 impl<T: AsRef<Transaction>> AsRef<Transaction> for MutableTransaction<T> {

consensus/src/consensus/mod.rs

@@ -40,7 +40,11 @@ use crate::{
 };
 use kaspa_consensus_core::{
     acceptance_data::AcceptanceData,
-    api::{stats::BlockCount, BlockValidationFutures, ConsensusApi, ConsensusStats},
+    api::{
+        args::{TransactionValidationArgs, TransactionValidationBatchArgs},
+        stats::BlockCount,
+        BlockValidationFutures, ConsensusApi, ConsensusStats,
+    },
     block::{Block, BlockTemplate, TemplateBuildMode, TemplateTransactionSelector, VirtualStateApproxId},
     blockhash::BlockHashExtensions,
     blockstatus::BlockStatus,
@@ -49,9 +53,10 @@ use kaspa_consensus_core::{
     errors::{
         coinbase::CoinbaseResult,
         consensus::{ConsensusError, ConsensusResult},
+        difficulty::DifficultyError,
+        pruning::PruningImportError,
         tx::TxResult,
     },
-    errors::{difficulty::DifficultyError, pruning::PruningImportError},
     header::Header,
     muhash::MuHashExtensions,
     network::NetworkType,
@@ -418,13 +423,17 @@ impl ConsensusApi for Consensus {
         BlockValidationFutures { block_task: Box::pin(block_task), virtual_state_task: Box::pin(virtual_state_task) }
     }
 
-    fn validate_mempool_transaction(&self, transaction: &mut MutableTransaction) -> TxResult<()> {
-        self.virtual_processor.validate_mempool_transaction(transaction)?;
+    fn validate_mempool_transaction(&self, transaction: &mut MutableTransaction, args: &TransactionValidationArgs) -> TxResult<()> {
+        self.virtual_processor.validate_mempool_transaction(transaction, args)?;
         Ok(())
     }
 
-    fn validate_mempool_transactions_in_parallel(&self, transactions: &mut [MutableTransaction]) -> Vec<TxResult<()>> {
-        self.virtual_processor.validate_mempool_transactions_in_parallel(transactions)
+    fn validate_mempool_transactions_in_parallel(
+        &self,
+        transactions: &mut [MutableTransaction],
+        args: &TransactionValidationBatchArgs,
+    ) -> Vec<TxResult<()>> {
+        self.virtual_processor.validate_mempool_transactions_in_parallel(transactions, args)
     }
 
     fn populate_mempool_transaction(&self, transaction: &mut MutableTransaction) -> TxResult<()> {

consensus/src/pipeline/virtual_processor/processor.rs

@@ -48,6 +48,7 @@ use crate::{
 };
 use kaspa_consensus_core::{
     acceptance_data::AcceptanceData,
+    api::args::{TransactionValidationArgs, TransactionValidationBatchArgs},
     block::{BlockTemplate, MutableBlock, TemplateBuildMode, TemplateTransactionSelector},
     blockstatus::BlockStatus::{StatusDisqualifiedFromChain, StatusUTXOValid},
     coinbase::MinerData,
@@ -757,23 +758,28 @@ impl VirtualStateProcessor {
         virtual_utxo_view: &impl UtxoView,
         virtual_daa_score: u64,
         virtual_past_median_time: u64,
+        args: &TransactionValidationArgs,
     ) -> TxResult<()> {
         self.transaction_validator.validate_tx_in_isolation(&mutable_tx.tx)?;
         self.transaction_validator.utxo_free_tx_validation(&mutable_tx.tx, virtual_daa_score, virtual_past_median_time)?;
-        self.validate_mempool_transaction_in_utxo_context(mutable_tx, virtual_utxo_view, virtual_daa_score)?;
+        self.validate_mempool_transaction_in_utxo_context(mutable_tx, virtual_utxo_view, virtual_daa_score, args)?;
         Ok(())
     }
 
-    pub fn validate_mempool_transaction(&self, mutable_tx: &mut MutableTransaction) -> TxResult<()> {
+    pub fn validate_mempool_transaction(&self, mutable_tx: &mut MutableTransaction, args: &TransactionValidationArgs) -> TxResult<()> {
         let virtual_read = self.virtual_stores.read();
         let virtual_state = virtual_read.state.get().unwrap();
         let virtual_utxo_view = &virtual_read.utxo_set;
         let virtual_daa_score = virtual_state.daa_score;
         let virtual_past_median_time = virtual_state.past_median_time;
-        self.validate_mempool_transaction_impl(mutable_tx, virtual_utxo_view, virtual_daa_score, virtual_past_median_time)
+        self.validate_mempool_transaction_impl(mutable_tx, virtual_utxo_view, virtual_daa_score, virtual_past_median_time, args)
     }
 
-    pub fn validate_mempool_transactions_in_parallel(&self, mutable_txs: &mut [MutableTransaction]) -> Vec<TxResult<()>> {
+    pub fn validate_mempool_transactions_in_parallel(
+        &self,
+        mutable_txs: &mut [MutableTransaction],
+        args: &TransactionValidationBatchArgs,
+    ) -> Vec<TxResult<()>> {
         let virtual_read = self.virtual_stores.read();
         let virtual_state = virtual_read.state.get().unwrap();
         let virtual_utxo_view = &virtual_read.utxo_set;
@@ -784,7 +790,13 @@ impl VirtualStateProcessor {
             mutable_txs
                 .par_iter_mut()
                 .map(|mtx| {
-                    self.validate_mempool_transaction_impl(mtx, &virtual_utxo_view, virtual_daa_score, virtual_past_median_time)
+                    self.validate_mempool_transaction_impl(
+                        mtx,
+                        &virtual_utxo_view,
+                        virtual_daa_score,
+                        virtual_past_median_time,
+                        args.get(&mtx.id()),
+                    )
                 })
                 .collect::<Vec<TxResult<()>>>()
         })

consensus/src/pipeline/virtual_processor/utxo_validation.rs

@@ -15,6 +15,7 @@ use crate::{
 };
 use kaspa_consensus_core::{
     acceptance_data::{AcceptedTxEntry, MergesetBlockAcceptanceData},
+    api::args::TransactionValidationArgs,
     coinbase::*,
     hashing,
     header::Header,
@@ -248,7 +249,7 @@ impl VirtualStateProcessor {
             }
         }
         let populated_tx = PopulatedTransaction::new(transaction, entries);
-        let res = self.transaction_validator.validate_populated_transaction_and_get_fee(&populated_tx, pov_daa_score, flags);
+        let res = self.transaction_validator.validate_populated_transaction_and_get_fee(&populated_tx, pov_daa_score, flags, None);
         match res {
             Ok(calculated_fee) => Ok(ValidatedTransaction::new(populated_tx, calculated_fee)),
             Err(tx_rule_error) => {
@@ -290,6 +291,7 @@ impl VirtualStateProcessor {
         mutable_tx: &mut MutableTransaction,
         utxo_view: &impl UtxoView,
         pov_daa_score: u64,
+        args: &TransactionValidationArgs,
     ) -> TxResult<()> {
         self.populate_mempool_transaction_in_utxo_context(mutable_tx, utxo_view)?;
 
@@ -308,10 +310,12 @@ impl VirtualStateProcessor {
         mutable_tx.tx.set_mass(contextual_mass);
 
         // At this point we know all UTXO entries are populated, so we can safely pass the tx as verifiable
+        let mass_and_feerate_threshold = args.feerate_threshold.map(|threshold| (contextual_mass, threshold));
         let calculated_fee = self.transaction_validator.validate_populated_transaction_and_get_fee(
             &mutable_tx.as_verifiable(),
             pov_daa_score,
             TxValidationFlags::SkipMassCheck, // we can skip the mass check since we just set it
+            mass_and_feerate_threshold,
         )?;
         mutable_tx.calculated_fee = Some(calculated_fee);
         Ok(())

consensus/src/processes/transaction_validator/transaction_validator_populated.rs

@@ -27,10 +27,12 @@ impl TransactionValidator {
         tx: &impl VerifiableTransaction,
         pov_daa_score: u64,
         flags: TxValidationFlags,
+        mass_and_feerate_threshold: Option<(u64, f64)>,
     ) -> TxResult<u64> {
         self.check_transaction_coinbase_maturity(tx, pov_daa_score)?;
         let total_in = self.check_transaction_input_amounts(tx)?;
         let total_out = Self::check_transaction_output_values(tx, total_in)?;
+        let fee = total_in - total_out;
         if flags != TxValidationFlags::SkipMassCheck && pov_daa_score > self.storage_mass_activation_daa_score {
             // Storage mass hardfork was activated
             self.check_mass_commitment(tx)?;
@@ -40,14 +42,31 @@ impl TransactionValidator {
             }
         }
         Self::check_sequence_lock(tx, pov_daa_score)?;
+
+        // The following call is not a consensus check (it could not be one in the first place since it uses floating number)
+        // but rather a mempool Replace by Fee validation rule. It was placed here purposely for avoiding unneeded script checks.
+        Self::check_feerate_threshold(fee, mass_and_feerate_threshold)?;
+
         match flags {
             TxValidationFlags::Full | TxValidationFlags::SkipMassCheck => {
                 Self::check_sig_op_counts(tx)?;
                 self.check_scripts(tx)?;
             }
             TxValidationFlags::SkipScriptChecks => {}
         }
-        Ok(total_in - total_out)
+        Ok(fee)
+    }
+
+    fn check_feerate_threshold(fee: u64, mass_and_feerate_threshold: Option<(u64, f64)>) -> TxResult<()> {
+        // An actual check can only occur if some mass and threshold are provided,
+        // otherwise, the check does not verify anything and exits successfully.
+        if let Some((contextual_mass, feerate_threshold)) = mass_and_feerate_threshold {
+            assert!(contextual_mass > 0);
+            if fee as f64 / contextual_mass as f64 <= feerate_threshold {
+                return Err(TxRuleError::FeerateTooLow);
+            }
+        }
+        Ok(())
     }
 
     fn check_transaction_coinbase_maturity(&self, tx: &impl VerifiableTransaction, pov_daa_score: u64) -> TxResult<()> {

crypto/txscript/src/standard.rs

@@ -100,9 +100,9 @@ pub mod test_helpers {
         (script_public_key, redeem_script)
     }
 
-    // Creates a transaction that spends the first output of provided transaction.
-    // Assumes that the output being spent has opTrueScript as it's scriptPublicKey.
-    // Creates the value of the spent output minus provided `fee` (in sompi).
+    /// Creates a transaction that spends the first output of provided transaction.
+    /// Assumes that the output being spent has opTrueScript as its scriptPublicKey.
+    /// Creates the value of the spent output minus provided `fee` (in sompi).
     pub fn create_transaction(tx_to_spend: &Transaction, fee: u64) -> Transaction {
         let (script_public_key, redeem_script) = op_true_script();
         let signature_script = pay_to_script_hash_signature_script(redeem_script, vec![]).expect("the script is canonical");
@@ -111,6 +111,42 @@ pub mod test_helpers {
         let output = TransactionOutput::new(tx_to_spend.outputs[0].value - fee, script_public_key);
         Transaction::new(TX_VERSION, vec![input], vec![output], 0, SUBNETWORK_ID_NATIVE, 0, vec![])
     }
+
+    /// Creates a transaction that spends the outputs of specified indexes (if they exist) of every provided transaction and returns an optional change.
+    /// Assumes that the outputs being spent have opTrueScript as their scriptPublicKey.
+    ///
+    /// If some change is provided, creates two outputs, first one with the value of the spent outputs minus `change`
+    /// and `fee` (in sompi) and second one of `change` amount.
+    ///
+    /// If no change is provided, creates only one output with the value of the spent outputs minus and `fee` (in sompi)
+    pub fn create_transaction_with_change<'a>(
+        txs_to_spend: impl Iterator<Item = &'a Transaction>,
+        output_indexes: Vec<usize>,
+        change: Option<u64>,
+        fee: u64,
+    ) -> Transaction {
+        let (script_public_key, redeem_script) = op_true_script();
+        let signature_script = pay_to_script_hash_signature_script(redeem_script, vec![]).expect("the script is canonical");
+        let mut inputs_value: u64 = 0;
+        let mut inputs = vec![];
+        for tx_to_spend in txs_to_spend {
+            for i in output_indexes.iter().copied() {
+                if i < tx_to_spend.outputs.len() {
+                    let previous_outpoint = TransactionOutpoint::new(tx_to_spend.id(), i as u32);
+                    inputs.push(TransactionInput::new(previous_outpoint, signature_script.clone(), MAX_TX_IN_SEQUENCE_NUM, 1));
+                    inputs_value += tx_to_spend.outputs[i].value;
+                }
+            }
+        }
+        let outputs = match change {
+            Some(change) => vec![
+                TransactionOutput::new(inputs_value - fee - change, script_public_key.clone()),
+                TransactionOutput::new(change, script_public_key),
+            ],
+            None => vec![TransactionOutput::new(inputs_value - fee, script_public_key.clone())],
+        };
+        Transaction::new(TX_VERSION, inputs, outputs, 0, SUBNETWORK_ID_NATIVE, 0, vec![])
+    }
 }
 
 #[cfg(test)]

mining/errors/src/mempool.rs

@@ -4,7 +4,7 @@ use kaspa_consensus_core::{
 };
 use thiserror::Error;
 
-#[derive(Error, Debug, Clone)]
+#[derive(Error, Debug, Clone, PartialEq, Eq)]
 pub enum RuleError {
     /// A consensus transaction rule error
     ///
@@ -24,9 +24,15 @@ pub enum RuleError {
     #[error("transaction {0} is already in the mempool")]
     RejectDuplicate(TransactionId),
 
-    #[error("output {0} already spent by transaction {1} in the memory pool")]
+    #[error("output {0} already spent by transaction {1} in the mempool")]
     RejectDoubleSpendInMempool(TransactionOutpoint, TransactionId),
 
+    #[error("replace by fee found no double spending transaction in the mempool")]
+    RejectRbfNoDoubleSpend,
+
+    #[error("replace by fee found more than one double spending transaction in the mempool")]
+    RejectRbfTooManyDoubleSpendingTransactions,
+
     /// New behavior: a transaction is rejected if the mempool is full
     #[error("number of high-priority transactions in mempool ({0}) has reached the maximum allowed ({1})")]
     RejectMempoolIsFull(usize, u64),
@@ -95,7 +101,7 @@ impl From<TxRuleError> for RuleError {
 
 pub type RuleResult<T> = std::result::Result<T, RuleError>;
 
-#[derive(Error, Debug, Clone)]
+#[derive(Error, Debug, Clone, PartialEq, Eq)]
 pub enum NonStandardError {
     #[error("transaction version {1} is not in the valid range of {2}-{3}")]
     RejectVersion(TransactionId, u16, u16, u16),