diff --git a/src/bench/mempool_eviction.cpp b/src/bench/mempool_eviction.cpp index b9cfdf3914f9..dced2861aa8b 100644 --- a/src/bench/mempool_eviction.cpp +++ b/src/bench/mempool_eviction.cpp @@ -12,11 +12,12 @@ static void AddTx(const CTransactionRef& tx, const CAmount& nFee, CTxMemPool& po { int64_t nTime = 0; unsigned int nHeight = 1; + uint64_t sequence = 0; bool spendsCoinbase = false; unsigned int sigOps = 1; LockPoints lp; pool.addUnchecked(CTxMemPoolEntry( - tx, nFee, nTime, nHeight, + tx, nFee, nTime, nHeight, sequence, spendsCoinbase, sigOps, lp)); } diff --git a/src/bench/mempool_stress.cpp b/src/bench/mempool_stress.cpp index 75009c879361..100ab1401849 100644 --- a/src/bench/mempool_stress.cpp +++ b/src/bench/mempool_stress.cpp @@ -14,10 +14,11 @@ static void AddTx(const CTransactionRef& tx, CTxMemPool& pool) EXCLUSIVE_LOCKS_R { int64_t nTime = 0; unsigned int nHeight = 1; + uint64_t sequence = 0; bool spendsCoinbase = false; unsigned int sigOpCost = 4; LockPoints lp; - pool.addUnchecked(CTxMemPoolEntry(tx, 1000, nTime, nHeight, spendsCoinbase, sigOpCost, lp)); + pool.addUnchecked(CTxMemPoolEntry(tx, 1000, nTime, nHeight, sequence, spendsCoinbase, sigOpCost, lp)); } struct Available { diff --git a/src/bench/rpc_mempool.cpp b/src/bench/rpc_mempool.cpp index 39930b862358..8d853f5389c7 100644 --- a/src/bench/rpc_mempool.cpp +++ b/src/bench/rpc_mempool.cpp @@ -14,7 +14,7 @@ static void AddTx(const CTransactionRef& tx, const CAmount& fee, CTxMemPool& pool) EXCLUSIVE_LOCKS_REQUIRED(cs_main, pool.cs) { LockPoints lp; - pool.addUnchecked(CTxMemPoolEntry(tx, fee, /*time=*/0, /*entry_height=*/1, /*spends_coinbase=*/false, /*sigops_count=*/1, lp)); + pool.addUnchecked(CTxMemPoolEntry(tx, fee, /*time=*/0, /*entry_height=*/1, /*entry_sequence=*/0, /*spends_coinbase=*/false, /*sigops_count=*/1, lp)); } static void RpcMempool(benchmark::Bench& bench) diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 1d634907c5bc..eceedf46058a 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -102,8 +102,6 @@ static const unsigned int MAX_GETDATA_SZ = 1000; /** How long to cache transactions in mapRelay for normal relay */ static constexpr auto RELAY_TX_CACHE_TIME = 15min; -/** How long a transaction has to be in the mempool before it can unconditionally be relayed (even when not in mapRelay). */ -static constexpr auto UNCONDITIONAL_RELAY_DELAY = 2min; /** Headers download timeout. * Timeout = base + per_header * (expected number of headers) */ static constexpr auto HEADERS_DOWNLOAD_TIMEOUT_BASE = 15min; @@ -181,13 +179,6 @@ static constexpr auto OUTBOUND_INVENTORY_BROADCAST_INTERVAL{2s}; static constexpr unsigned int INVENTORY_BROADCAST_PER_SECOND = 7; /** Maximum number of inventory items to send per transmission. */ static constexpr unsigned int INVENTORY_BROADCAST_MAX_PER_1MB_BLOCK = 4 * INVENTORY_BROADCAST_PER_SECOND * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL); -/** The number of most recently announced transactions a peer can request. */ -static constexpr unsigned int INVENTORY_MAX_RECENT_RELAY = 3500; -/** Verify that INVENTORY_MAX_RECENT_RELAY is enough to cache everything typically - * relayed before unconditional relay from the mempool kicks in. This is only a - * lower bound, and it should be larger to account for higher inv rate to outbound - * peers, and random variations in the broadcast mechanism. */ -static_assert(INVENTORY_MAX_RECENT_RELAY >= INVENTORY_BROADCAST_PER_SECOND * UNCONDITIONAL_RELAY_DELAY / std::chrono::seconds{1}, "INVENTORY_RELAY_MAX too low"); /** Maximum number of compact filters that may be requested with one getcfilters. See BIP 157. */ static constexpr uint32_t MAX_GETCFILTERS_SIZE = 1000; /** Maximum number of cf hashes that may be requested with one getcfheaders. See BIP 157. */ @@ -312,11 +303,12 @@ struct Peer { * permitted if the peer has NetPermissionFlags::Mempool or we advertise * NODE_BLOOM. See BIP35. */ bool m_send_mempool GUARDED_BY(m_tx_inventory_mutex){false}; - /** The last time a BIP35 `mempool` request was serviced. */ - std::atomic m_last_mempool_req{0s}; /** The next time after which we will send an `inv` message containing * transaction announcements to this peer. */ std::chrono::microseconds m_next_inv_send_time GUARDED_BY(NetEventsInterface::g_msgproc_mutex){0}; + /** The mempool sequence num at which we sent the last `inv` message to this peer. + * Can relay txs with lower sequence numbers than this (see CTxMempool::info_for_relay). */ + uint64_t m_last_inv_sequence GUARDED_BY(NetEventsInterface::g_msgproc_mutex){1}; }; /** @@ -583,9 +575,6 @@ struct CNodeState { //! Whether this peer is an inbound connection const bool m_is_inbound; - //! A rolling bloom filter of all announced tx CInvs to this peer. - CRollingBloomFilter m_recently_announced_invs = CRollingBloomFilter{INVENTORY_MAX_RECENT_RELAY, 0.000001}; - CNodeState(bool is_inbound) : m_is_inbound(is_inbound) {} }; @@ -1010,7 +999,8 @@ class PeerManagerImpl final : public PeerManager std::atomic m_last_tip_update{0s}; /** Determine whether or not a peer can request a transaction, and return it (or nullptr if not found or not allowed). */ - CTransactionRef FindTxForGetData(const CNode* peer, const uint256& txid, const std::chrono::seconds mempool_req, const std::chrono::seconds now) LOCKS_EXCLUDED(cs_main); + CTransactionRef FindTxForGetData(const Peer::TxRelay& tx_relay, const uint256& txid) + EXCLUSIVE_LOCKS_REQUIRED(!m_most_recent_block_mutex, NetEventsInterface::g_msgproc_mutex); void ProcessGetData(CNode& pfrom, Peer& peer, const std::atomic& interruptMsgProc) EXCLUSIVE_LOCKS_REQUIRED(peer.m_getdata_requests_mutex) LOCKS_EXCLUDED(::cs_main); @@ -2660,29 +2650,23 @@ void PeerManagerImpl::ProcessGetBlockData(CNode& pfrom, Peer& peer, const CInv& } } -//! Determine whether or not a peer can request a transaction, and return it (or nullptr if not found or not allowed). -CTransactionRef PeerManagerImpl::FindTxForGetData(const CNode* peer, const uint256& txid, const std::chrono::seconds mempool_req, const std::chrono::seconds now) +CTransactionRef PeerManagerImpl::FindTxForGetData(const Peer::TxRelay& tx_relay, const uint256& txid) { - auto txinfo = m_mempool.info(txid); + // If a tx was in the mempool prior to the last INV for this peer, permit the request. + auto txinfo = m_mempool.info_for_relay(txid, tx_relay.m_last_inv_sequence); if (txinfo.tx) { - // If a TX could have been INVed in reply to a MEMPOOL request, - // or is older than UNCONDITIONAL_RELAY_DELAY, permit the request - // unconditionally. - if ((mempool_req.count() && txinfo.m_time <= mempool_req) || txinfo.m_time <= now - UNCONDITIONAL_RELAY_DELAY) { - return std::move(txinfo.tx); - } + return std::move(txinfo.tx); } + // Or it might be from the most recent block { - LOCK(cs_main); - - // Otherwise, the transaction must have been announced recently. - if (State(peer->GetId())->m_recently_announced_invs.contains(txid)) { - // If it was, it can be relayed from either the mempool... - if (txinfo.tx) return std::move(txinfo.tx); - // ... or the relay pool. - auto mi = mapRelay.find(txid); - if (mi != mapRelay.end()) return mi->second; + LOCK(m_most_recent_block_mutex); + if (m_most_recent_block) { + for (const auto& tx : m_most_recent_block->vtx) { + if (tx->GetHash() == txid) { + return tx; + } + } } } @@ -2699,11 +2683,6 @@ void PeerManagerImpl::ProcessGetData(CNode& pfrom, Peer& peer, const std::atomic std::vector vNotFound; const CNetMsgMaker msgMaker(pfrom.GetCommonVersion()); - const auto now{GetTime()}; - // Get last mempool request time - const auto mempool_req = tx_relay != nullptr ? tx_relay->m_last_mempool_req.load() - : std::chrono::seconds::min(); - // Process as many TX items from the front of the getdata queue as // possible, since they're common and it's efficient to batch process // them. @@ -2731,7 +2710,7 @@ void PeerManagerImpl::ProcessGetData(CNode& pfrom, Peer& peer, const std::atomic bool push = false; if (inv.IsGenTxMsg()) { - CTransactionRef tx = FindTxForGetData(&pfrom, inv.hash, mempool_req, now); + CTransactionRef tx = FindTxForGetData(*tx_relay, inv.hash); if (tx) { CCoinJoinBroadcastTx dstx; if (inv.IsMsgDstx()) { @@ -2744,30 +2723,6 @@ void PeerManagerImpl::ProcessGetData(CNode& pfrom, Peer& peer, const std::atomic } m_mempool.RemoveUnbroadcastTx(tx->GetHash()); push = true; - - // As we're going to send tx, make sure its unconfirmed parents are made requestable. - std::vector parent_ids_to_add; - { - LOCK(m_mempool.cs); - auto txiter = m_mempool.GetIter(tx->GetHash()); - if (txiter) { - const CTxMemPoolEntry::Parents& parents = (*txiter)->GetMemPoolParentsConst(); - parent_ids_to_add.reserve(parents.size()); - for (const CTxMemPoolEntry& parent : parents) { - if (parent.GetTime() > now - UNCONDITIONAL_RELAY_DELAY) { - parent_ids_to_add.push_back(parent.GetTx().GetHash()); - } - } - } - } - - for (const uint256& parent_txid : parent_ids_to_add) { - // Relaying a transaction with a recent but unconfirmed parent. - if (WITH_LOCK(tx_relay->m_tx_inventory_mutex, return !tx_relay->m_tx_inventory_known_filter.contains(parent_txid))) { - LOCK(cs_main); - State(pfrom.GetId())->m_recently_announced_invs.insert(parent_txid); - } - } } } @@ -6054,7 +6009,6 @@ bool PeerManagerImpl::SendMessages(CNode* pto) auto queueAndMaybePushInv = [this, pto, peer, &vInv, &msgMaker](const CInv& invIn) { LogPrint(BCLog::NET, "SendMessages -- queued inv: %s index=%d peer=%d\n", invIn.ToString(), vInv.size(), pto->GetId()); - // Responses to MEMPOOL requests bypass the m_recently_announced_invs filter. vInv.push_back(invIn); if (vInv.size() == MAX_INV_SZ) { LogPrint(BCLog::NET, "SendMessages -- pushing invs: count=%d peer=%d\n", vInv.size(), pto->GetId()); @@ -6119,7 +6073,6 @@ bool PeerManagerImpl::SendMessages(CNode* pto) tx_relay->m_tx_inventory_known_filter.insert(chainlockHash); queueAndMaybePushInv(CInv(MSG_CLSIG, chainlockHash)); } - tx_relay->m_last_mempool_req = std::chrono::duration_cast(current_time); } // Determine transactions to relay @@ -6161,7 +6114,6 @@ bool PeerManagerImpl::SendMessages(CNode* pto) } if (tx_relay->m_bloom_filter && !tx_relay->m_bloom_filter->IsRelevantAndUpdate(*txinfo.tx)) continue; // Send - State(pto->GetId())->m_recently_announced_invs.insert(hash); nRelayedTransactions++; { // Expire old relay messages @@ -6180,6 +6132,10 @@ bool PeerManagerImpl::SendMessages(CNode* pto) tx_relay->m_tx_inventory_known_filter.insert(hash); queueAndMaybePushInv(CInv(nInvType, hash)); } + + // Ensure we'll respond to GETDATA requests for anything we've just announced + LOCK(m_mempool.cs); + tx_relay->m_last_inv_sequence = m_mempool.GetSequence(); } } { diff --git a/src/node/interfaces.cpp b/src/node/interfaces.cpp index 493c8606c86c..0402ebea2a48 100644 --- a/src/node/interfaces.cpp +++ b/src/node/interfaces.cpp @@ -925,7 +925,7 @@ class ChainImpl : public Chain { if (!m_node.mempool) return true; LockPoints lp; - CTxMemPoolEntry entry(tx, 0, 0, 0, false, 0, lp); + CTxMemPoolEntry entry(tx, 0, 0, 0, 0, false, 0, lp); CTxMemPool::setEntries ancestors; auto limit_ancestor_count = gArgs.GetIntArg("-limitancestorcount", DEFAULT_ANCESTOR_LIMIT); auto limit_ancestor_size = gArgs.GetIntArg("-limitancestorsize", DEFAULT_ANCESTOR_SIZE_LIMIT) * 1000; diff --git a/src/test/util/setup_common.cpp b/src/test/util/setup_common.cpp index 457d52460cd9..eb051f36f41f 100644 --- a/src/test/util/setup_common.cpp +++ b/src/test/util/setup_common.cpp @@ -656,7 +656,7 @@ std::vector TestChainSetup::PopulateMempool(FastRandomContext& if (submit) { LOCK2(m_node.mempool->cs, cs_main); LockPoints lp; - m_node.mempool->addUnchecked(CTxMemPoolEntry(ptx, 1000, 0, 1, false, 4, lp)); + m_node.mempool->addUnchecked(CTxMemPoolEntry(ptx, 1000, 0, 1, 0, false, 4, lp)); } --num_transactions; } @@ -670,7 +670,7 @@ CTxMemPoolEntry TestMemPoolEntryHelper::FromTx(const CMutableTransaction& tx) co CTxMemPoolEntry TestMemPoolEntryHelper::FromTx(const CTransactionRef& tx) const { - return CTxMemPoolEntry(tx, nFee, TicksSinceEpoch(time), nHeight, + return CTxMemPoolEntry(tx, nFee, TicksSinceEpoch(time), nHeight, 0, spendsCoinbase, sigOpCount, lp); } diff --git a/src/txmempool.cpp b/src/txmempool.cpp index 32f91fc65c06..a14e789d6f4d 100644 --- a/src/txmempool.cpp +++ b/src/txmempool.cpp @@ -49,13 +49,14 @@ bool TestLockPointValidity(CChain& active_chain, const LockPoints& lp) } CTxMemPoolEntry::CTxMemPoolEntry(const CTransactionRef& tx, CAmount fee, - int64_t time, unsigned int entry_height, + int64_t time, unsigned int entry_height, uint64_t entry_sequence, bool spends_coinbase, int64_t sigops_count, LockPoints lp) : tx{tx}, nFee{fee}, nTxSize(tx->GetTotalSize()), nUsageSize{RecursiveDynamicUsage(tx)}, nTime{time}, + entry_sequence{entry_sequence}, entryHeight{entry_height}, spendsCoinbase{spends_coinbase}, sigOpCount{sigops_count}, @@ -1327,6 +1328,17 @@ TxMempoolInfo CTxMemPool::info(const uint256& hash) const return GetInfo(i); } +TxMempoolInfo CTxMemPool::info_for_relay(const uint256& txid, uint64_t last_sequence) const +{ + LOCK(cs); + indexed_transaction_set::const_iterator i = mapTx.find(txid); + if (i != mapTx.end() && i->GetSequence() < last_sequence) { + return GetInfo(i); + } else { + return TxMempoolInfo(); + } +} + bool CTxMemPool::existsProviderTxConflict(const CTransaction &tx) const { assert(m_dmnman); diff --git a/src/txmempool.h b/src/txmempool.h index 25eedd282b9a..624d6e0f26af 100644 --- a/src/txmempool.h +++ b/src/txmempool.h @@ -111,6 +111,7 @@ class CTxMemPoolEntry const size_t nTxSize; //!< ... and avoid recomputing tx size const size_t nUsageSize; //!< ... and total memory usage const int64_t nTime; //!< Local time when entering the mempool + const uint64_t entry_sequence; //!< Sequence number used to determine whether this transaction is too recent for relay const unsigned int entryHeight; //!< Chain height when entering the mempool const bool spendsCoinbase; //!< keep track of transactions that spend a coinbase const int64_t sigOpCount; //!< Legacy sig ops plus P2SH sig op count @@ -132,7 +133,7 @@ class CTxMemPoolEntry public: CTxMemPoolEntry(const CTransactionRef& tx, CAmount fee, - int64_t time, unsigned int entry_height, + int64_t time, unsigned int entry_height, uint64_t entry_sequence, bool spends_coinbase, int64_t sigops_count, LockPoints lp); @@ -142,6 +143,7 @@ class CTxMemPoolEntry size_t GetTxSize() const; std::chrono::seconds GetTime() const { return std::chrono::seconds{nTime}; } unsigned int GetHeight() const { return entryHeight; } + uint64_t GetSequence() const { return entry_sequence; } int64_t GetSigOpCount() const { return sigOpCount; } CAmount GetModifiedFee() const { return m_modified_fee; } size_t DynamicMemoryUsage() const { return nUsageSize; } @@ -793,6 +795,10 @@ class CTxMemPool CTransactionRef get(const uint256& hash) const; TxMempoolInfo info(const uint256& hash) const; + + /** Returns info for a transaction if its entry_sequence < last_sequence */ + TxMempoolInfo info_for_relay(const uint256& txid, uint64_t last_sequence) const; + std::vector infoAll() const; /** diff --git a/src/validation.cpp b/src/validation.cpp index 823c2ca5d354..caa56c52678c 100644 --- a/src/validation.cpp +++ b/src/validation.cpp @@ -853,7 +853,8 @@ bool MemPoolAccept::PreChecks(ATMPArgs& args, Workspace& ws) } } - entry.reset(new CTxMemPoolEntry(ptx, ws.m_base_fees, nAcceptTime, m_active_chainstate.m_chain.Height(), + entry.reset(new CTxMemPoolEntry(ptx, ws.m_base_fees, nAcceptTime, m_active_chainstate.m_chain.Height(), + bypass_limits ? 0 : m_pool.GetSequence(), fSpendsCoinbase, nSigOps, lp)); ws.m_vsize = entry->GetTxSize(); diff --git a/test/functional/mempool_reorg.py b/test/functional/mempool_reorg.py index f32024c20e2d..0a91dac9125c 100755 --- a/test/functional/mempool_reorg.py +++ b/test/functional/mempool_reorg.py @@ -8,6 +8,17 @@ that spend (directly or indirectly) coinbase transactions. """ +import time + +from test_framework.messages import ( + CInv, + MSG_TX, + msg_getdata, +) +from test_framework.p2p import ( + P2PTxInvStore, + p2p_lock, +) from test_framework.test_framework import BitcoinTestFramework from test_framework.util import assert_equal, assert_raises_rpc_error from test_framework.wallet import MiniWallet @@ -22,8 +33,84 @@ def set_test_params(self): [] ] + def test_reorg_relay(self): + self.log.info("Test that transactions from disconnected blocks are available for relay immediately") + # Prevent time from moving forward + self.nodes[1].setmocktime(int(time.time())) + self.connect_nodes(0, 1) + self.generate(self.wallet, 3) + + # Disconnect node0 and node1 to create different chains. + self.disconnect_nodes(0, 1) + # Connect a peer to node1, which doesn't have immediate tx relay + peer1 = self.nodes[1].add_p2p_connection(P2PTxInvStore()) + + # Create a transaction that is included in a block. + tx_disconnected = self.wallet.send_self_transfer(from_node=self.nodes[1]) + self.generate(self.nodes[1], 1, sync_fun=self.no_op) + + # Create a transaction and submit it to node1's mempool. + tx_before_reorg = self.wallet.send_self_transfer(from_node=self.nodes[1]) + + # Create a child of that transaction and submit it to node1's mempool. + tx_child = self.wallet.send_self_transfer(utxo_to_spend=tx_disconnected["new_utxo"], from_node=self.nodes[1]) + assert_equal(self.nodes[1].getmempoolentry(tx_child["txid"])["ancestorcount"], 1) + assert_equal(len(peer1.get_invs()), 0) + + # node0 has a longer chain in which tx_disconnected was not confirmed. + self.generate(self.nodes[0], 3, sync_fun=self.no_op) + + # Reconnect the nodes and sync chains. node0's chain should win. + self.connect_nodes(0, 1) + self.sync_blocks() + + # Child now has an ancestor from the disconnected block + assert_equal(self.nodes[1].getmempoolentry(tx_child["txid"])["ancestorcount"], 2) + assert_equal(self.nodes[1].getmempoolentry(tx_before_reorg["txid"])["ancestorcount"], 1) + + # peer1 should not have received an inv for any of the transactions during this time, as not + # enough time has elapsed for those transactions to be announced. Likewise, it cannot + # request very recent, unanounced transactions. + assert_equal(len(peer1.get_invs()), 0) + # It's too early to request these two transactions + requests_too_recent = msg_getdata([CInv(t=MSG_TX, h=int(tx["txid"], 16)) for tx in [tx_before_reorg, tx_child]]) + peer1.send_and_ping(requests_too_recent) + for _ in range(len(requests_too_recent.inv)): + peer1.sync_with_ping() + with p2p_lock: + assert "tx" not in peer1.last_message + assert "notfound" in peer1.last_message + + # Request the tx from the disconnected block + request_disconnected_tx = msg_getdata([CInv(t=MSG_TX, h=int(tx_disconnected["txid"], 16))]) + peer1.send_and_ping(request_disconnected_tx) + + # The tx from the disconnected block was never announced, and it entered the mempool later + # than the transactions that are too recent. + assert_equal(len(peer1.get_invs()), 0) + with p2p_lock: + # However, the node will answer requests for the tx from the recently-disconnected block. + assert_equal(peer1.last_message["tx"].tx.sha256, int(tx_disconnected["txid"], 16)) + + self.nodes[1].setmocktime(int(time.time()) + 30) + peer1.sync_with_ping() + # the transactions are now announced + assert_equal(len(peer1.get_invs()), 3) + for _ in range(3): + # make sure all tx requests have been responded to + peer1.sync_with_ping() + last_tx_received = peer1.last_message["tx"] + + tx_after_reorg = self.wallet.send_self_transfer(from_node=self.nodes[1]) + request_after_reorg = msg_getdata([CInv(t=MSG_TX, h=int(tx_after_reorg["txid"], 16))]) + assert tx_after_reorg["txid"] in self.nodes[1].getrawmempool() + peer1.send_and_ping(request_after_reorg) + with p2p_lock: + assert_equal(peer1.last_message["tx"], last_tx_received) + def run_test(self): - wallet = MiniWallet(self.nodes[0]) + self.wallet = MiniWallet(self.nodes[0]) + wallet = self.wallet # Start with a 200 block chain assert_equal(self.nodes[0].getblockcount(), 200) @@ -104,6 +191,8 @@ def run_test(self): assert_equal(set(self.nodes[0].getrawmempool()), set()) self.sync_all() + self.test_reorg_relay() + if __name__ == '__main__': MempoolCoinbaseTest().main()