Merge bitcoin/bitcoin#27625: p2p: Stop relaying non-mempool txs

faa2976 Remove mapRelay (MarcoFalke)
fccecd7 net_processing: relay txs from m_most_recent_block (Anthony Towns)

Pull request description:

  `mapRelay` (used to relay announced transactions that are no longer in the mempool) has issues:

  * It doesn't have an absolute memory limit, only an implicit one based on the rate of transaction announcements
  * <strike>It doesn't have a use-case</strike> EDIT: see below

  Fix all issues by removing `mapRelay`.

  For more context, on why a transaction may have been removed from the mempool, see https://github.com/bitcoin/bitcoin/blob/c2f2abd0a4f4bd18bfca41b632d21d803479f3f4/src/txmempool.h#L228-L238

  For my rationale on why it is fine to not relay them:

  Reason | | Rationale
  -- | -- | --
  `EXPIRY` | Expired from mempool | Mempool expiry is by default 2 weeks and can not be less than 1 hour, so a transaction can not be in `mapRelay` while expiring, unless a re-broadcast happened. This should be fine, because the transaction will be re-added to the mempool and potentially announced/relayed on the next re-broadcast.
  `SIZELIMIT` | Removed in size limiting | A low fee transaction, which will be relayed by a different peer after `GETDATA_TX_INTERVAL` or after we sent a `notfound` message. Assuming it ever made it to another peer, otherwise it will happen on re-broadcast (same as with `EXPIRY` above).
  `REORG` | Removed for reorganization | Block races are rare, so reorgs should be rarer. Also, the transaction is likely to be re-accepted via the `disconnectpool` later on. If not, it seems fine to let the originating wallet deal with rebroadcast in this case.
  `BLOCK` | Removed for block | EDIT: Needed for compact block relay, see bitcoin/bitcoin#27625 (comment)
  `CONFLICT` | Removed for conflict with in-block transaction | The peer won't be able to add the tx to the mempool anyway, unless it is on a different block, in which case it seems fine to let the originating wallet take care of the rebroadcast (if needed).
  `REPLACED` | Removed for replacement | EDIT: Also needed for compact block relay, see bitcoin/bitcoin#27625 (comment) ?

ACKs for top commit:
  sdaftuar:
    ACK faa2976
  ajtowns:
    ACK faa2976
  glozow:
    code review ACK faa2976

Tree-SHA512: 64ae3e387b001bf6bd5b6c938e7317f4361f9bc0b8cc5d8f63a16cda2408d2f634a22f8157dfcd8957502ef358208292ec91e7d70c9c2d8a8c47cc0114ecfebd

Author: fanquake

src/net_processing.cpp

@@ -51,9 +51,7 @@
 #include <optional>
 #include <typeinfo>
 
-/** 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). */
+/** How long a transaction has to be in the mempool before it can unconditionally be relayed. */
 static constexpr auto UNCONDITIONAL_RELAY_DELAY = 2min;
 /** Headers download timeout.
  *  Timeout = base + per_header * (expected number of headers) */
@@ -851,6 +849,7 @@ class PeerManagerImpl final : public PeerManager
     std::shared_ptr<const CBlock> m_most_recent_block GUARDED_BY(m_most_recent_block_mutex);
     std::shared_ptr<const CBlockHeaderAndShortTxIDs> m_most_recent_compact_block GUARDED_BY(m_most_recent_block_mutex);
     uint256 m_most_recent_block_hash GUARDED_BY(m_most_recent_block_mutex);
+    std::unique_ptr<const std::map<uint256, CTransactionRef>> m_most_recent_block_txs GUARDED_BY(m_most_recent_block_mutex);
 
     // Data about the low-work headers synchronization, aggregated from all peers' HeadersSyncStates.
     /** Mutex guarding the other m_headers_presync_* variables. */
@@ -910,7 +909,7 @@ class PeerManagerImpl final : public PeerManager
 
     /** Determine whether or not a peer can request a transaction, and return it (or nullptr if not found or not allowed). */
     CTransactionRef FindTxForGetData(const Peer::TxRelay& tx_relay, const GenTxid& gtxid, const std::chrono::seconds mempool_req, const std::chrono::seconds now)
-        EXCLUSIVE_LOCKS_REQUIRED(NetEventsInterface::g_msgproc_mutex);
+        EXCLUSIVE_LOCKS_REQUIRED(!m_most_recent_block_mutex, NetEventsInterface::g_msgproc_mutex);
 
     void ProcessGetData(CNode& pfrom, Peer& peer, const std::atomic<bool>& interruptMsgProc)
         EXCLUSIVE_LOCKS_REQUIRED(!m_most_recent_block_mutex, peer.m_getdata_requests_mutex, NetEventsInterface::g_msgproc_mutex)
@@ -919,12 +918,6 @@ class PeerManagerImpl final : public PeerManager
     /** Process a new block. Perform any post-processing housekeeping */
     void ProcessBlock(CNode& node, const std::shared_ptr<const CBlock>& block, bool force_processing, bool min_pow_checked);
 
-    /** Relay map (txid or wtxid -> CTransactionRef) */
-    typedef std::map<uint256, CTransactionRef> MapRelay;
-    MapRelay mapRelay GUARDED_BY(NetEventsInterface::g_msgproc_mutex);
-    /** Expiration-time ordered list of (expire time, relay map entry) pairs. */
-    std::deque<std::pair<std::chrono::microseconds, MapRelay::iterator>> g_relay_expiration GUARDED_BY(NetEventsInterface::g_msgproc_mutex);
-
     /**
      * When a peer sends us a valid block, instruct it to announce blocks to us
      * using CMPCTBLOCK if possible by adding its nodeid to the end of
@@ -1927,10 +1920,17 @@ void PeerManagerImpl::NewPoWValidBlock(const CBlockIndex *pindex, const std::sha
         std::async(std::launch::deferred, [&] { return msgMaker.Make(NetMsgType::CMPCTBLOCK, *pcmpctblock); })};
 
     {
+        auto most_recent_block_txs = std::make_unique<std::map<uint256, CTransactionRef>>();
+        for (const auto& tx : pblock->vtx) {
+            most_recent_block_txs->emplace(tx->GetHash(), tx);
+            most_recent_block_txs->emplace(tx->GetWitnessHash(), tx);
+        }
+
         LOCK(m_most_recent_block_mutex);
         m_most_recent_block_hash = hashBlock;
         m_most_recent_block = pblock;
         m_most_recent_compact_block = pcmpctblock;
+        m_most_recent_block_txs = std::move(most_recent_block_txs);
     }
 
     m_connman.ForEachNode([this, pindex, &lazy_ser, &hashBlock](CNode* pnode) EXCLUSIVE_LOCKS_REQUIRED(::cs_main) {
@@ -2301,13 +2301,17 @@ CTransactionRef PeerManagerImpl::FindTxForGetData(const Peer::TxRelay& tx_relay,
         }
     }
 
-    // Otherwise, the transaction must have been announced recently.
-    if (tx_relay.m_recently_announced_invs.contains(gtxid.GetHash())) {
-        // 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(gtxid.GetHash());
-        if (mi != mapRelay.end()) return mi->second;
+    // Otherwise, the transaction might have been announced recently.
+    bool recent = tx_relay.m_recently_announced_invs.contains(gtxid.GetHash());
+    if (recent && txinfo.tx) return std::move(txinfo.tx);
+
+    // Or it might be from the most recent block
+    {
+        LOCK(m_most_recent_block_mutex);
+        if (m_most_recent_block_txs != nullptr) {
+            auto it = m_most_recent_block_txs->find(gtxid.GetHash());
+            if (it != m_most_recent_block_txs->end()) return it->second;
+        }
     }
 
     return {};
@@ -5778,7 +5782,6 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
                             continue;
                         }
                         auto txid = txinfo.tx->GetHash();
-                        auto wtxid = txinfo.tx->GetWitnessHash();
                         // Peer told you to not send transactions at that feerate? Don't bother sending it.
                         if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) {
                             continue;
@@ -5788,24 +5791,6 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
                         tx_relay->m_recently_announced_invs.insert(hash);
                         vInv.push_back(inv);
                         nRelayedTransactions++;
-                        {
-                            // Expire old relay messages
-                            while (!g_relay_expiration.empty() && g_relay_expiration.front().first < current_time)
-                            {
-                                mapRelay.erase(g_relay_expiration.front().second);
-                                g_relay_expiration.pop_front();
-                            }
-
-                            auto ret = mapRelay.emplace(txid, std::move(txinfo.tx));
-                            if (ret.second) {
-                                g_relay_expiration.emplace_back(current_time + RELAY_TX_CACHE_TIME, ret.first);
-                            }
-                            // Add wtxid-based lookup into mapRelay as well, so that peers can request by wtxid
-                            auto ret2 = mapRelay.emplace(wtxid, ret.first->second);
-                            if (ret2.second) {
-                                g_relay_expiration.emplace_back(current_time + RELAY_TX_CACHE_TIME, ret2.first);
-                            }
-                        }
                         if (vInv.size() == MAX_INV_SZ) {
                             m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv));
                             vInv.clear();

test/functional/p2p_leak_tx.py

@@ -4,8 +4,8 @@
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 """Test transaction upload"""
 
-from test_framework.messages import msg_getdata, CInv, MSG_TX
-from test_framework.p2p import p2p_lock, P2PDataStore
+from test_framework.messages import msg_getdata, CInv, MSG_TX, MSG_WTX
+from test_framework.p2p import p2p_lock, P2PDataStore, P2PTxInvStore
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
@@ -27,6 +27,7 @@ def run_test(self):
         self.miniwallet = MiniWallet(self.gen_node)
 
         self.test_tx_in_block()
+        self.test_notfound_on_replaced_tx()
         self.test_notfound_on_unannounced_tx()
 
     def test_tx_in_block(self):
@@ -45,8 +46,36 @@ def test_tx_in_block(self):
         inbound_peer.send_and_ping(want_tx)
         assert_equal(inbound_peer.last_message.get("tx").tx.getwtxid(), wtxid)
 
+    def test_notfound_on_replaced_tx(self):
+        self.gen_node.disconnect_p2ps()
+        inbound_peer = self.gen_node.add_p2p_connection(P2PTxInvStore())
+
+        self.log.info("Transaction tx_a is broadcast")
+        tx_a = self.miniwallet.send_self_transfer(from_node=self.gen_node)
+        inbound_peer.wait_for_broadcast(txns=[tx_a["wtxid"]])
+
+        tx_b = tx_a["tx"]
+        tx_b.vout[0].nValue -= 9000
+        self.gen_node.sendrawtransaction(tx_b.serialize().hex())
+
+        self.log.info("Re-request of tx_a after replacement is answered with notfound")
+        req_vec = [
+            CInv(t=MSG_TX, h=int(tx_a["txid"], 16)),
+            CInv(t=MSG_WTX, h=int(tx_a["wtxid"], 16)),
+        ]
+        want_tx = msg_getdata()
+        want_tx.inv = req_vec
+        with p2p_lock:
+            inbound_peer.last_message.pop("notfound", None)
+            inbound_peer.last_message.pop("tx", None)
+        inbound_peer.send_and_ping(want_tx)
+
+        assert_equal(inbound_peer.last_message.get("notfound").vec, req_vec)
+        assert "tx" not in inbound_peer.last_message
+
     def test_notfound_on_unannounced_tx(self):
         self.log.info("Check that we don't leak txs to inbound peers that we haven't yet announced to")
+        self.gen_node.disconnect_p2ps()
         inbound_peer = self.gen_node.add_p2p_connection(P2PNode())  # An "attacking" inbound peer
 
         MAX_REPEATS = 100