tx fees, policy: CBlockPolicyEstimator update from `CValidationInte…

…rface` notifications

`CBlockPolicyEstimator` will implement `CValidationInterface` and subscribe to
its notification to process tx added to the mempool and removed.

Co-authored-by: Matt Corallo <git@bluematt.me>

src/init.cpp

@@ -348,6 +348,10 @@ void Shutdown(NodeContext& node)
     node.chainman.reset();
     node.scheduler.reset();
 
+    if (node.fee_estimator) {
+        UnregisterValidationInterface(node.fee_estimator.get());
+    }
+
     try {
         if (!fs::remove(GetPidFile(*node.args))) {
             LogPrintf("%s: Unable to remove PID file: File does not exist\n", __func__);
@@ -1233,6 +1237,7 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
         // Flush estimates to disk periodically
         CBlockPolicyEstimator* fee_estimator = node.fee_estimator.get();
         node.scheduler->scheduleEvery([fee_estimator] { fee_estimator->FlushFeeEstimates(); }, FEE_FLUSH_INTERVAL);
+        RegisterValidationInterface(fee_estimator);
     }
 
     // Check port numbers
@@ -1452,7 +1457,6 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
     assert(!node.chainman);
 
     CTxMemPool::Options mempool_opts{
-        .estimator = node.fee_estimator.get(),
         .check_ratio = chainparams.DefaultConsistencyChecks() ? 1 : 0,
     };
     auto result{ApplyArgsManOptions(args, chainparams, mempool_opts)};

src/kernel/mempool_entry.h

@@ -190,6 +190,10 @@ struct NewMempoolTransactionInfo {
      */
     int64_t m_virtual_transaction_size;
     unsigned int txHeight;
+    bool m_from_disconnected_block;
+    bool m_submitted_in_package;
+    bool m_chainstate_is_current;
+    bool m_has_no_mempool_parents;
 };
 
 #endif // BITCOIN_KERNEL_MEMPOOL_ENTRY_H

src/kernel/mempool_options.h

@@ -13,8 +13,6 @@
 #include <cstdint>
 #include <optional>
 
-class CBlockPolicyEstimator;
-
 /** Default for -maxmempool, maximum megabytes of mempool memory usage */
 static constexpr unsigned int DEFAULT_MAX_MEMPOOL_SIZE_MB{300};
 /** Default for -maxmempool when blocksonly is set */
@@ -33,8 +31,6 @@ namespace kernel {
  * Most of the time, this struct should be referenced as CTxMemPool::Options.
  */
 struct MemPoolOptions {
-    /* Used to estimate appropriate transaction fees. */
-    CBlockPolicyEstimator* estimator{nullptr};
     /* The ratio used to determine how often sanity checks will run.  */
     int check_ratio{0};
     int64_t max_size_bytes{DEFAULT_MAX_MEMPOOL_SIZE_MB * 1'000'000};

src/policy/fees.cpp

@@ -502,11 +502,6 @@ void TxConfirmStats::removeTx(unsigned int entryHeight, unsigned int nBestSeenHe
     }
 }
 
-// This function is called from CTxMemPool::removeUnchecked to ensure
-// txs removed from the mempool for any reason are no longer
-// tracked. Txs that were part of a block have already been removed in
-// processBlockTx to ensure they are never double tracked, but it is
-// of no harm to try to remove them again.
 bool CBlockPolicyEstimator::removeTx(uint256 hash, bool inBlock)
 {
     LOCK(m_cs_fee_estimator);
@@ -566,7 +561,7 @@ CBlockPolicyEstimator::CBlockPolicyEstimator(const fs::path& estimation_filepath
 
 CBlockPolicyEstimator::~CBlockPolicyEstimator() = default;
 
-void CBlockPolicyEstimator::processTransaction(const CTransactionRef& tx, const NewMempoolTransactionInfo& tx_info, bool validForFeeEstimation)
+void CBlockPolicyEstimator::processTransaction(const CTransactionRef& tx, const NewMempoolTransactionInfo& tx_info)
 {
     LOCK(m_cs_fee_estimator);
     unsigned int txHeight = tx_info.txHeight;
@@ -583,6 +578,7 @@ void CBlockPolicyEstimator::processTransaction(const CTransactionRef& tx, const
         // It will be synced next time a block is processed.
         return;
     }
+    bool validForFeeEstimation = !tx_info.m_from_disconnected_block && !tx_info.m_submitted_in_package && tx_info.m_chainstate_is_current && tx_info.m_has_no_mempool_parents;
     // Only want to be updating estimates when our blockchain is synced,
     // otherwise we'll miscalculate how many blocks its taking to get included.
     if (!validForFeeEstimation) {
@@ -635,7 +631,7 @@ bool CBlockPolicyEstimator::processBlockTx(unsigned int nBlockHeight, const CTra
 }
 
 void CBlockPolicyEstimator::processBlock(unsigned int nBlockHeight,
-                                         std::vector<CTransactionRef>& txs_removed_for_block)
+                                         const std::vector<CTransactionRef>& txs_removed_for_block)
 {
     LOCK(m_cs_fee_estimator);
     if (nBlockHeight <= nBestSeenHeight) {
@@ -1037,6 +1033,21 @@ std::chrono::hours CBlockPolicyEstimator::GetFeeEstimatorFileAge()
     return std::chrono::duration_cast<std::chrono::hours>(now - file_time);
 }
 
+void CBlockPolicyEstimator::TransactionAddedToMempool(const CTransactionRef& tx, const NewMempoolTransactionInfo& tx_info, uint64_t mempool_sequence)
+{
+    processTransaction(tx, tx_info);
+}
+
+void CBlockPolicyEstimator::TransactionRemovedFromMempool(const CTransactionRef& tx, MemPoolRemovalReason reason, uint64_t mempool_sequence)
+{
+    removeTx(tx->GetHash(), false);
+}
+
+void CBlockPolicyEstimator::MempoolBlockConnect(const std::vector<CTransactionRef>& txs_removed_for_block, unsigned int nBlockHeight)
+{
+    processBlock(nBlockHeight, txs_removed_for_block);
+}
+
 static std::set<double> MakeFeeSet(const CFeeRate& min_incremental_fee,
                                    double max_filter_fee_rate,
                                    double fee_filter_spacing)

src/policy/fees.h

@@ -13,6 +13,7 @@
 #include <threadsafety.h>
 #include <uint256.h>
 #include <util/fs.h>
+#include <validationinterface.h>
 
 #include <array>
 #include <chrono>
@@ -144,7 +145,7 @@ struct FeeCalculation
  * a certain number of blocks.  Every time a block is added to the best chain, this class records
  * stats on the transactions included in that block
  */
-class CBlockPolicyEstimator
+class CBlockPolicyEstimator : public CValidationInterface
 {
 private:
     /** Track confirm delays up to 12 blocks for short horizon */
@@ -199,15 +200,15 @@ class CBlockPolicyEstimator
 public:
     /** Create new BlockPolicyEstimator and initialize stats tracking classes with default values */
     CBlockPolicyEstimator(const fs::path& estimation_filepath, const bool read_stale_estimates);
-    ~CBlockPolicyEstimator();
+    virtual ~CBlockPolicyEstimator();
 
     /** Process all the transactions that have been included in a block */
     void processBlock(unsigned int nBlockHeight,
-                      std::vector<CTransactionRef>& txs_removed_for_block)
+                      const std::vector<CTransactionRef>& txs_removed_for_block)
         EXCLUSIVE_LOCKS_REQUIRED(!m_cs_fee_estimator);
 
     /** Process a transaction accepted to the mempool*/
-    void processTransaction(const CTransactionRef& tx, const NewMempoolTransactionInfo& tx_info, bool validForFeeEstimation)
+    void processTransaction(const CTransactionRef& tx, const NewMempoolTransactionInfo& tx_info)
         EXCLUSIVE_LOCKS_REQUIRED(!m_cs_fee_estimator);
 
     /** Remove a transaction from the mempool tracking stats*/
@@ -261,6 +262,15 @@ class CBlockPolicyEstimator
     /** Calculates the age of the file, since last modified */
     std::chrono::hours GetFeeEstimatorFileAge();
 
+protected:
+    /** Overridden from CValidationInterface. */
+    void TransactionAddedToMempool(const CTransactionRef& tx, const NewMempoolTransactionInfo& tx_info, uint64_t mempool_sequence) override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_cs_fee_estimator);
+    void TransactionRemovedFromMempool(const CTransactionRef& tx, MemPoolRemovalReason reason, uint64_t mempool_sequence) override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_cs_fee_estimator);
+    void MempoolBlockConnect(const std::vector<CTransactionRef>& txs_removed_for_block, unsigned int nBlockHeight) override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_cs_fee_estimator);
+
 private:
     mutable Mutex m_cs_fee_estimator;
 

src/test/fuzz/policy_estimator.cpp

@@ -12,6 +12,7 @@
 #include <test/fuzz/util/mempool.h>
 #include <test/util/setup_common.h>
 #include <txmempool.h>
+#include <validationinterface.h>
 
 #include <cstdint>
 #include <optional>
@@ -42,8 +43,9 @@ FUZZ_TARGET(policy_estimator, .init = initialize_policy_estimator)
                 }
                 const CTransaction tx{*mtx};
                 const CTxMemPoolEntry entry = ConsumeTxMemPoolEntry(fuzzed_data_provider, tx);
-                const NewMempoolTransactionInfo tx_info = {entry.GetFee(), entry.GetTxSize(), entry.GetHeight()};
-                block_policy_estimator.processTransaction(entry.GetSharedTx(), tx_info, fuzzed_data_provider.ConsumeBool());
+                const NewMempoolTransactionInfo tx_info = {entry.GetFee(), entry.GetTxSize(), entry.GetHeight(),
+                                                           fuzzed_data_provider.ConsumeBool(), fuzzed_data_provider.ConsumeBool(), fuzzed_data_provider.ConsumeBool(), fuzzed_data_provider.ConsumeBool()};
+                block_policy_estimator.processTransaction(entry.GetSharedTx(), tx_info);
                 if (fuzzed_data_provider.ConsumeBool()) {
                     (void)block_policy_estimator.removeTx(tx.GetHash(), /*inBlock=*/fuzzed_data_provider.ConsumeBool());
                 }

src/test/fuzz/tx_pool.cpp

@@ -124,7 +124,6 @@ CTxMemPool MakeMempool(FuzzedDataProvider& fuzzed_data_provider, const NodeConte
     CTxMemPool::Options mempool_opts{MemPoolOptionsForTest(node)};
 
     // ...override specific options for this specific fuzz suite
-    mempool_opts.estimator = nullptr;
     mempool_opts.check_ratio = 1;
     mempool_opts.require_standard = fuzzed_data_provider.ConsumeBool();
 

src/test/policyestimator_tests.cpp

@@ -8,6 +8,7 @@
 #include <txmempool.h>
 #include <uint256.h>
 #include <util/time.h>
+#include <validationinterface.h>
 
 #include <test/util/setup_common.h>
 
@@ -19,7 +20,7 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
 {
     CBlockPolicyEstimator& feeEst = *Assert(m_node.fee_estimator);
     CTxMemPool& mpool = *Assert(m_node.mempool);
-    LOCK2(cs_main, mpool.cs);
+    RegisterValidationInterface(&feeEst);
     TestMemPoolEntryHelper entry;
     CAmount basefee(2000);
     CAmount deltaFee(100);
@@ -60,7 +61,22 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
             for (int k = 0; k < 4; k++) { // add 4 fee txs
                 tx.vin[0].prevout.n = 10000*blocknum+100*j+k; // make transaction unique
                 uint256 hash = tx.GetHash();
-                mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
+                {
+                    LOCK2(cs_main, mpool.cs);
+                    mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
+                    // Since TransactionAddedToMempool callbacks are generated in ATMP,
+                    // not addUnchecked, we cheat and create one manually here
+                    int64_t virtual_size = GetVirtualTransactionSize(*MakeTransactionRef(tx));
+                    GetMainSignals().TransactionAddedToMempool(MakeTransactionRef(tx),
+                                                               {feeV[j],
+                                                                virtual_size,
+                                                                entry.nHeight,
+                                                                false,
+                                                                false,
+                                                                true,
+                                                                true},
+                                                               mpool.GetAndIncrementSequence());
+                }
                 txHashes[j].push_back(hash);
             }
         }
@@ -76,10 +92,15 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
                 txHashes[9-h].pop_back();
             }
         }
-        mpool.removeForBlock(block, ++blocknum);
+        {
+            LOCK(mpool.cs);
+            mpool.removeForBlock(block, ++blocknum);
+        }
         block.clear();
         // Check after just a few txs that combining buckets works as expected
         if (blocknum == 3) {
+            // Wait for fee estimator to catch up
+            SyncWithValidationInterfaceQueue();
             // At this point we should need to combine 3 buckets to get enough data points
             // So estimateFee(1) should fail and estimateFee(2) should return somewhere around
             // 9*baserate.  estimateFee(2) %'s are 100,100,90 = average 97%
@@ -114,8 +135,13 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
 
     // Mine 50 more blocks with no transactions happening, estimates shouldn't change
     // We haven't decayed the moving average enough so we still have enough data points in every bucket
-    while (blocknum < 250)
+    while (blocknum < 250) {
+        LOCK(mpool.cs);
         mpool.removeForBlock(block, ++blocknum);
+    }
+
+    // Wait for fee estimator to catch up
+    SyncWithValidationInterfaceQueue();
 
     BOOST_CHECK(feeEst.estimateFee(1) == CFeeRate(0));
     for (int i = 2; i < 10;i++) {
@@ -131,13 +157,28 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
             for (int k = 0; k < 4; k++) { // add 4 fee txs
                 tx.vin[0].prevout.n = 10000*blocknum+100*j+k;
                 uint256 hash = tx.GetHash();
-                mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
+                {
+                    LOCK2(cs_main, mpool.cs);
+                    mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
+
+                    // Since TransactionAddedToMempool callbacks are generated in ATMP,
+                    // not addUnchecked, we cheat and create one manually here
+                    int64_t virtual_size = GetVirtualTransactionSize(*MakeTransactionRef(tx));
+                    GetMainSignals().TransactionAddedToMempool(MakeTransactionRef(tx), {feeV[j], virtual_size, entry.nHeight, false, false, true, true}, mpool.GetAndIncrementSequence());
+                }
                 txHashes[j].push_back(hash);
             }
         }
-        mpool.removeForBlock(block, ++blocknum);
+        {
+            LOCK(mpool.cs);
+            mpool.removeForBlock(block, ++blocknum);
+        }
     }
 
+
+    // Wait for fee estimator to catch up
+    SyncWithValidationInterfaceQueue();
+
     for (int i = 1; i < 10;i++) {
         BOOST_CHECK(feeEst.estimateFee(i) == CFeeRate(0) || feeEst.estimateFee(i).GetFeePerK() > origFeeEst[i-1] - deltaFee);
     }
@@ -152,8 +193,15 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
             txHashes[j].pop_back();
         }
     }
-    mpool.removeForBlock(block, 266);
+    {
+        LOCK(mpool.cs);
+        mpool.removeForBlock(block, 266);
+    }
     block.clear();
+
+    // Wait for fee estimator to catch up
+    SyncWithValidationInterfaceQueue();
+
     BOOST_CHECK(feeEst.estimateFee(1) == CFeeRate(0));
     for (int i = 2; i < 10;i++) {
         BOOST_CHECK(feeEst.estimateFee(i) == CFeeRate(0) || feeEst.estimateFee(i).GetFeePerK() > origFeeEst[i-1] - deltaFee);
@@ -166,20 +214,35 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
             for (int k = 0; k < 4; k++) { // add 4 fee txs
                 tx.vin[0].prevout.n = 10000*blocknum+100*j+k;
                 uint256 hash = tx.GetHash();
-                mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
+                {
+                    LOCK2(cs_main, mpool.cs);
+                    mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
+                    // Since TransactionAddedToMempool callbacks are generated in ATMP,
+                    // not addUnchecked, we cheat and create one manually here
+                    int64_t virtual_size = GetVirtualTransactionSize(*MakeTransactionRef(tx));
+                    GetMainSignals().TransactionAddedToMempool(MakeTransactionRef(tx), {feeV[j], virtual_size, entry.nHeight, false, false, true, true}, mpool.GetAndIncrementSequence());
+                }
                 CTransactionRef ptx = mpool.get(hash);
                 if (ptx)
                     block.push_back(ptx);
 
             }
         }
-        mpool.removeForBlock(block, ++blocknum);
+        {
+            LOCK(mpool.cs);
+            mpool.removeForBlock(block, ++blocknum);
+        }
         block.clear();
     }
+
+    // Wait for fee estimator to catch up
+    SyncWithValidationInterfaceQueue();
+
     BOOST_CHECK(feeEst.estimateFee(1) == CFeeRate(0));
     for (int i = 2; i < 9; i++) { // At 9, the original estimate was already at the bottom (b/c scale = 2)
         BOOST_CHECK(feeEst.estimateFee(i).GetFeePerK() < origFeeEst[i-1] - deltaFee);
     }
+    UnregisterValidationInterface(&feeEst);
 }
 
 BOOST_AUTO_TEST_SUITE_END()

src/test/util/txmempool.cpp

@@ -17,7 +17,6 @@ using node::NodeContext;
 CTxMemPool::Options MemPoolOptionsForTest(const NodeContext& node)
 {
     CTxMemPool::Options mempool_opts{
-        .estimator = node.fee_estimator.get(),
         // Default to always checking mempool regardless of
         // chainparams.DefaultConsistencyChecks for tests
         .check_ratio = 1,