tx fees, policy: CBlockPolicyEstimator update from `CValidationInterface` notifications

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

Re-delegate calculation of `validForFeeEstimation` from validation to fee estimator.

Also clean up the validForFeeEstimation arg thats no longer needed in `CTxMempool`.

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

5 files changed, 98 insertions, 15 deletions

diff --git a/src/test/fuzz/package_eval.cpp b/src/test/fuzz/package_eval.cpp
index 021acc02f2..08b8be26e7 100644
--- a/src/test/fuzz/package_eval.cpp
+++ b/src/test/fuzz/package_eval.cpp
@@ -121,7 +121,6 @@ CTxMemPool MakeMempool(FuzzedDataProvider& fuzzed_data_provider, const NodeConte
     mempool_opts.expiry = std::chrono::hours{fuzzed_data_provider.ConsumeIntegralInRange<unsigned>(0, 999)};
     nBytesPerSigOp = fuzzed_data_provider.ConsumeIntegralInRange<unsigned>(1, 999);
 
-    mempool_opts.estimator = nullptr;
     mempool_opts.check_ratio = 1;
     mempool_opts.require_standard = fuzzed_data_provider.ConsumeBool();
 
diff --git a/src/test/fuzz/policy_estimator.cpp b/src/test/fuzz/policy_estimator.cpp
index e5e0b0d8da..4a41707edf 100644
--- a/src/test/fuzz/policy_estimator.cpp
+++ b/src/test/fuzz/policy_estimator.cpp
@@ -44,9 +44,16 @@ FUZZ_TARGET(policy_estimator, .init = initialize_policy_estimator)
                     return;
                 }
                 const CTransaction tx{*mtx};
-                block_policy_estimator.processTransaction(ConsumeTxMemPoolEntry(fuzzed_data_provider, tx), fuzzed_data_provider.ConsumeBool());
+                const CTxMemPoolEntry& entry = ConsumeTxMemPoolEntry(fuzzed_data_provider, tx);
+                const auto tx_info = NewMempoolTransactionInfo(entry.GetSharedTx(), entry.GetFee(),
+                                                               entry.GetTxSize(), entry.GetHeight(),
+                                                               /* m_from_disconnected_block */ false,
+                                                               /* m_submitted_in_package */ false,
+                                                               /* m_chainstate_is_current */ true,
+                                                               /* m_has_no_mempool_parents */ fuzzed_data_provider.ConsumeBool());
+                block_policy_estimator.processTransaction(tx_info);
                 if (fuzzed_data_provider.ConsumeBool()) {
-                    (void)block_policy_estimator.removeTx(tx.GetHash(), /*inBlock=*/fuzzed_data_provider.ConsumeBool());
+                    (void)block_policy_estimator.removeTx(tx.GetHash());
                 }
             },
             [&] {
@@ -69,7 +76,7 @@ FUZZ_TARGET(policy_estimator, .init = initialize_policy_estimator)
                 block_policy_estimator.processBlock(txs, fuzzed_data_provider.ConsumeIntegral<unsigned int>());
             },
             [&] {
-                (void)block_policy_estimator.removeTx(ConsumeUInt256(fuzzed_data_provider), /*inBlock=*/fuzzed_data_provider.ConsumeBool());
+                (void)block_policy_estimator.removeTx(ConsumeUInt256(fuzzed_data_provider));
             },
             [&] {
                 block_policy_estimator.FlushUnconfirmed();
diff --git a/src/test/fuzz/tx_pool.cpp b/src/test/fuzz/tx_pool.cpp
index 001b452722..cc816f213b 100644
--- a/src/test/fuzz/tx_pool.cpp
+++ b/src/test/fuzz/tx_pool.cpp
@@ -123,7 +123,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();
 
diff --git a/src/test/policyestimator_tests.cpp b/src/test/policyestimator_tests.cpp
index bc9c672560..13ec89663a 100644
--- a/src/test/policyestimator_tests.cpp
+++ b/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);
@@ -59,8 +60,23 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
         for (int j = 0; j < 10; j++) { // For each fee
             for (int k = 0; k < 4; k++) { // add 4 fee txs
                 tx.vin[0].prevout.n = 10000*blocknum+100*j+k; // make transaction unique
+                {
+                    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
+                    const int64_t virtual_size = GetVirtualTransactionSize(*MakeTransactionRef(tx));
+                    const NewMempoolTransactionInfo tx_info{NewMempoolTransactionInfo(MakeTransactionRef(tx),
+                                                                                      feeV[j],
+                                                                                      virtual_size,
+                                                                                      entry.nHeight,
+                                                                                      /* m_from_disconnected_block */ false,
+                                                                                      /* m_submitted_in_package */ false,
+                                                                                      /* m_chainstate_is_current */ true,
+                                                                                      /* m_has_no_mempool_parents */ true)};
+                    GetMainSignals().TransactionAddedToMempool(tx_info, mpool.GetAndIncrementSequence());
+                }
                 uint256 hash = tx.GetHash();
-                mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
                 txHashes[j].push_back(hash);
             }
         }
@@ -76,10 +92,17 @@ 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 +137,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++) {
@@ -130,14 +158,35 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
         for (int j = 0; j < 10; j++) { // For each fee multiple
             for (int k = 0; k < 4; k++) { // add 4 fee txs
                 tx.vin[0].prevout.n = 10000*blocknum+100*j+k;
+                {
+                    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
+                    const int64_t virtual_size = GetVirtualTransactionSize(*MakeTransactionRef(tx));
+                    const NewMempoolTransactionInfo tx_info{NewMempoolTransactionInfo(MakeTransactionRef(tx),
+                                                                                      feeV[j],
+                                                                                      virtual_size,
+                                                                                      entry.nHeight,
+                                                                                      /* m_from_disconnected_block */ false,
+                                                                                      /* m_submitted_in_package */ false,
+                                                                                      /* m_chainstate_is_current */ true,
+                                                                                      /* m_has_no_mempool_parents */ true)};
+                    GetMainSignals().TransactionAddedToMempool(tx_info, mpool.GetAndIncrementSequence());
+                }
                 uint256 hash = tx.GetHash();
-                mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
                 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 +201,16 @@ 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);
@@ -165,17 +222,39 @@ BOOST_AUTO_TEST_CASE(BlockPolicyEstimates)
         for (int j = 0; j < 10; j++) { // For each fee multiple
             for (int k = 0; k < 4; k++) { // add 4 fee txs
                 tx.vin[0].prevout.n = 10000*blocknum+100*j+k;
+                {
+                    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
+                    const int64_t virtual_size = GetVirtualTransactionSize(*MakeTransactionRef(tx));
+                    const NewMempoolTransactionInfo tx_info{NewMempoolTransactionInfo(MakeTransactionRef(tx),
+                                                                                      feeV[j],
+                                                                                      virtual_size,
+                                                                                      entry.nHeight,
+                                                                                      /* m_from_disconnected_block */ false,
+                                                                                      /* m_submitted_in_package */ false,
+                                                                                      /* m_chainstate_is_current */ true,
+                                                                                      /* m_has_no_mempool_parents */ true)};
+                    GetMainSignals().TransactionAddedToMempool(tx_info, mpool.GetAndIncrementSequence());
+                }
                 uint256 hash = tx.GetHash();
-                mpool.addUnchecked(entry.Fee(feeV[j]).Time(Now<NodeSeconds>()).Height(blocknum).FromTx(tx));
                 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);
diff --git a/src/test/util/txmempool.cpp b/src/test/util/txmempool.cpp
index 6d3bb01be8..379c3c9329 100644
--- a/src/test/util/txmempool.cpp
+++ b/src/test/util/txmempool.cpp
@@ -18,7 +18,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,