diff --git a/src/Makefile.am b/src/Makefile.am index 3395d9e602fb..9c77ee4340e5 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -406,6 +406,7 @@ BITCOIN_CORE_H = \ txdb.h \ txmempool.h \ txorphanage.h \ + txrequest.h \ undo.h \ unordered_lru_cache.h \ util/asmap.h \ @@ -648,6 +649,7 @@ libbitcoin_node_a_SOURCES = \ txdb.cpp \ txmempool.cpp \ txorphanage.cpp \ + txrequest.cpp \ validation.cpp \ validationinterface.cpp \ versionbits.cpp \ diff --git a/src/Makefile.test.include b/src/Makefile.test.include index e98111bed99c..10f2a7cce4ad 100644 --- a/src/Makefile.test.include +++ b/src/Makefile.test.include @@ -197,6 +197,7 @@ BITCOIN_TESTS =\ test/txindex_tests.cpp \ test/txpackage_tests.cpp \ test/txreconciliation_tests.cpp \ + test/txrequest_tests.cpp \ test/txvalidation_tests.cpp \ test/txvalidationcache_tests.cpp \ test/uint256_tests.cpp \ @@ -380,6 +381,7 @@ test_fuzz_fuzz_SOURCES = \ test/fuzz/tx_out.cpp \ test/fuzz/tx_pool.cpp \ test/fuzz/txorphan.cpp \ + test/fuzz/txrequest.cpp \ test/fuzz/utxo_snapshot.cpp \ test/fuzz/utxo_total_supply.cpp \ test/fuzz/validation_load_mempool.cpp \ diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index d30a7bd376ad..bc0dd30133df 100644 --- a/src/coinjoin/server.cpp +++ b/src/coinjoin/server.cpp @@ -192,6 +192,7 @@ void CCoinJoinServer::ProcessDSQUEUE(NodeId from, CDataStream& vRecv) LogPrint(BCLog::COINJOIN, "DSQUEUE -- new CoinJoin queue, masternode=%s, queue=%s\n", dmn->proTxHash.ToString(), dsq.ToString()); if (!m_queueman.TryAddQueue(dsq)) return; + WITH_LOCK(cs_main, m_peer_manager->PeerForgetObjectRequest(CInv{MSG_DSQ, dsq.GetHash()})); m_peer_manager->PeerRelayDSQ(dsq); } } diff --git a/src/governance/net_governance.cpp b/src/governance/net_governance.cpp index 0b77ca586872..44465e4d5bf6 100644 --- a/src/governance/net_governance.cpp +++ b/src/governance/net_governance.cpp @@ -195,7 +195,9 @@ void NetGovernance::ProcessMessage(CNode& peer, const std::string& msg_type, CDa return; } - if (!WITH_LOCK(::cs_main, return m_gov_manager.ProcessObject(peer.GetLogString(), nHash, govobj))) { + if (WITH_LOCK(::cs_main, return m_gov_manager.ProcessObject(peer.GetLogString(), nHash, govobj))) { + WITH_LOCK(::cs_main, m_peer_manager->PeerForgetObjectRequest(CInv{MSG_GOVERNANCE_OBJECT, nHash})); + } else { // apply node's ban score m_peer_manager->PeerMisbehaving(peer.GetId(), 20); } @@ -244,6 +246,7 @@ void NetGovernance::ProcessMessage(CNode& peer, const std::string& msg_type, CDa uint256 hashToRequest; if (m_gov_manager.ProcessVote(vote, exception, hashToRequest)) { LogPrint(BCLog::GOBJECT, "MNGOVERNANCEOBJECTVOTE -- %s new\n", strHash); + WITH_LOCK(::cs_main, m_peer_manager->PeerForgetObjectRequest(CInv{MSG_GOVERNANCE_OBJECT_VOTE, nHash})); m_node_sync.BumpAssetLastTime("MNGOVERNANCEOBJECTVOTE"); if (!m_node_sync.IsSynced()) { diff --git a/src/instantsend/net_instantsend.cpp b/src/instantsend/net_instantsend.cpp index 2cb7d660fce1..08b67aea937f 100644 --- a/src/instantsend/net_instantsend.cpp +++ b/src/instantsend/net_instantsend.cpp @@ -288,6 +288,8 @@ void NetInstantSend::ProcessMessage(CNode& pfrom, const std::string& msg_type, C islock->txid.ToString(), hash.ToString(), from); m_is_manager.EnqueueInstantSendLock(from, hash, std::move(islock)); + // Enqueued locks count as AlreadyHave, no need to request them from anyone anymore. + WITH_LOCK(::cs_main, m_peer_manager->PeerForgetObjectRequest(CInv{MSG_ISDLOCK, hash})); } } diff --git a/src/msg_result.h b/src/msg_result.h index a90fc7fc11c5..1cccda1bbb2e 100644 --- a/src/msg_result.h +++ b/src/msg_result.h @@ -51,7 +51,7 @@ struct MessageProcessingResult //! @m_transactions will relay transactions to peers which is ready to accept it (some peers does not accept transactions) std::vector m_transactions; - //! @m_to_erase triggers EraseObjectRequest from PeerManager for this inventory if not nullopt + //! @m_to_erase completes the sending peer's pending request for this inventory if not nullopt std::optional m_to_erase; MessageProcessingResult() = default; diff --git a/src/net.h b/src/net.h index 6b41bd15c454..c1d87ceaca33 100644 --- a/src/net.h +++ b/src/net.h @@ -16,7 +16,6 @@ #include #include #include -#include #include #include #include diff --git a/src/net_processing.cpp b/src/net_processing.cpp index b8d059ae1abd..86b34a495a5f 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -33,6 +33,7 @@ #include #include #include +#include #include #include #include @@ -79,20 +80,20 @@ using node::fImporting; using node::fPruneMode; using node::fReindex; -/** Maximum number of in-flight objects from a peer */ -static constexpr int32_t MAX_PEER_OBJECT_IN_FLIGHT = 100; -/** Maximum number of announced objects from a peer */ -static constexpr int32_t MAX_PEER_OBJECT_ANNOUNCEMENTS = 2 * MAX_INV_SZ; -/** How many microseconds to delay requesting transactions from inbound peers */ -static constexpr auto INBOUND_PEER_TX_DELAY{2s}; +/** Maximum number of in-flight object requests from a peer. It is not a hard limit, but the + * threshold at which point the OVERLOADED_PEER_OBJECT_DELAY kicks in. */ +static constexpr int32_t MAX_PEER_OBJECT_REQUEST_IN_FLIGHT = 100; +/** Maximum number of announced objects from a peer. + * Unlike Bitcoin, this is not reduced to 5000: governance vote sync legitimately announces up to + * MAX_INV_SZ objects from a single peer (see CGovernanceManager). */ +static constexpr int32_t MAX_PEER_OBJECT_ANNOUNCEMENTS = std::max(5000, 2 * MAX_INV_SZ); +/** How long to delay requesting transactions from non-preferred peers */ +static constexpr auto NONPREF_PEER_TX_DELAY{2s}; +/** How long to delay requesting objects from overloaded peers (see + * MAX_PEER_OBJECT_REQUEST_IN_FLIGHT). */ +static constexpr auto OVERLOADED_PEER_OBJECT_DELAY{2s}; /** How long to wait before downloading a transaction from an additional peer */ static constexpr auto GETDATA_TX_INTERVAL{60s}; -/** Maximum delay for transaction requests to avoid biasing some peers over others. */ -static constexpr auto MAX_GETDATA_RANDOM_DELAY{2s}; -/** How long to wait (expiry * factor microseconds) before expiring an in-flight getdata request to a peer */ -static constexpr int64_t TX_EXPIRY_INTERVAL_FACTOR = 10; -static_assert(INBOUND_PEER_TX_DELAY >= MAX_GETDATA_RANDOM_DELAY, -"To preserve security, MAX_GETDATA_RANDOM_DELAY should not exceed INBOUND_PEER_DELAY"); /** Limit to avoid sending big packets. Not used in processing incoming GETDATA for compatibility */ static const unsigned int MAX_GETDATA_SZ = 1000; @@ -509,69 +510,6 @@ struct CNodeState { //! Time of last new block announcement int64_t m_last_block_announcement{0}; - /* - * State associated with objects download. - * - * Tx download algorithm: - * - * When inv comes in, queue up (process_time, inv) inside the peer's - * CNodeState (m_object_process_time) as long as m_object_announced for the peer - * isn't too big (MAX_PEER_OBJECT_ANNOUNCEMENTS). - * - * The process_time for a objects is set to nNow for outbound peers, - * nNow + 2 seconds for inbound peers. This is the time at which we'll - * consider trying to request the objects from the peer in - * SendMessages(). The delay for inbound peers is to allow outbound peers - * a chance to announce before we request from inbound peers, to prevent - * an adversary from using inbound connections to blind us to a - * objects (InvBlock). - * - * When we call SendMessages() for a given peer, - * we will loop over the objects in m_object_process_time, looking - * at the objects whose process_time <= nNow. We'll request each - * such objects that we don't have already and that hasn't been - * requested from another peer recently, up until we hit the - * MAX_PEER_OBJECT_IN_FLIGHT limit for the peer. Then we'll update - * g_already_asked_for for each requested inv, storing the time of the - * GETDATA request. We use g_already_asked_for to coordinate objects - * requests amongst our peers. - * - * For objects that we still need but we have already recently - * requested from some other peer, we'll reinsert (process_time, inv) - * back into the peer's m_object_process_time at the point in the future at - * which the most recent GETDATA request would time out (ie - * GetObjectInterval + the request time stored in g_already_asked_for). - * We add an additional delay for inbound peers, again to prefer - * attempting download from outbound peers first. - * We also add an extra small random delay up to 2 seconds - * to avoid biasing some peers over others. (e.g., due to fixed ordering - * of peer processing in ThreadMessageHandler). - * - * When we receive a objects from a peer, we remove the inv from the - * peer's m_object_in_flight set and from their recently announced set - * (m_object_announced). We also clear g_already_asked_for for that entry, so - * that if somehow the objects is not accepted but also not added to - * the reject filter, then we will eventually redownload from other - * peers. - */ - struct ObjectDownloadState { - /* Track when to attempt download of announced objects (process - * time in micros -> inv) - */ - std::multimap m_object_process_time; - - //! Store all the objects a peer has recently announced - std::set m_object_announced; - - //! Store objects which were requested by us, with timestamp - std::map m_object_in_flight; - - //! Periodically check for stuck getdata requests - std::chrono::microseconds m_check_expiry_timer{0}; - }; - - ObjectDownloadState m_object_download; - //! Whether this peer is an inbound connection const bool m_is_inbound; @@ -655,6 +593,7 @@ class PeerManagerImpl final : public PeerManager bool PeerIsBanned(const NodeId node_id) override EXCLUSIVE_LOCKS_REQUIRED(cs_main, !m_peer_mutex); void PeerEraseObjectRequest(const NodeId nodeid, const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(::cs_main); bool PeerConsumeObjectRequest(NodeId nodeid, const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(::cs_main); + void PeerForgetObjectRequest(const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(::cs_main); void PeerPushInventory(NodeId nodeid, const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void PeerRelayInv(const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void PeerRelayInvFiltered(const CInv& inv, const CTransaction& relatedTx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); @@ -681,12 +620,15 @@ class PeerManagerImpl final : public PeerManager */ void RelayInvFiltered(const CInv& inv, const uint256& relatedTxHash) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); - void EraseObjectRequest(NodeId nodeid, const CInv& inv) EXCLUSIVE_LOCKS_REQUIRED(::cs_main); - bool ConsumeObjectRequest(NodeId nodeid, const CInv& inv) EXCLUSIVE_LOCKS_REQUIRED(::cs_main); - - void RequestObject(NodeId nodeid, const CInv& inv, std::chrono::microseconds current_time, bool fForce = false) + /** Register with m_object_request that an inv has been received from a peer, computing the + * request delay from the peer's preferredness and in-flight load. */ + void AddObjectAnnouncement(const CNode& node, const CInv& inv, std::chrono::microseconds current_time) EXCLUSIVE_LOCKS_REQUIRED(::cs_main); + /** Delete all announcements of a transaction across all peers, under both inv types it may + * have been announced with (MSG_TX and MSG_DSTX). */ + void ForgetTx(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(::cs_main); + /** Helper to process result of external handlers of message */ void PostProcessMessage(MessageProcessingResult&& ret, NodeId node) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); @@ -1105,6 +1047,11 @@ class PeerManagerImpl final : public PeerManager /** Storage for orphan information */ TxOrphanage m_orphanage; + /** Tracks announced inventories (transactions and all Dash-specific object types), and which + * peer to request them from next. All policy (preferredness, delays, per-type expiry) is + * decided by the callers; see AddObjectAnnouncement and the getdata section of SendMessages. */ + TxRequestTracker m_object_request GUARDED_BY(::cs_main); + void AddToCompactExtraTransactions(const CTransactionRef& tx) EXCLUSIVE_LOCKS_REQUIRED(g_msgproc_mutex); /** Orphan/conflicted/etc transactions that are kept for compact block reconstruction. @@ -1117,10 +1064,6 @@ class PeerManagerImpl final : public PeerManager std::vector> m_handlers; }; -// Keeps track of the time (in microseconds) when transactions were requested last time -unordered_limitedmap g_already_asked_for(MAX_INV_SZ, MAX_INV_SZ * 2); -unordered_limitedmap g_erased_object_requests(MAX_INV_SZ, MAX_INV_SZ * 2); - const CNodeState* PeerManagerImpl::State(NodeId pnode) const EXCLUSIVE_LOCKS_REQUIRED(cs_main) { std::map::const_iterator it = m_node_states.find(pnode); @@ -1573,63 +1516,6 @@ void PeerManagerImpl::PushNodeVersion(CNode& pnode, const Peer& peer) } } -void PeerManagerImpl::EraseObjectRequest(NodeId nodeid, const CInv& inv) -{ - AssertLockHeld(cs_main); - - CNodeState* state = State(nodeid); - if (state == nullptr) - return; - - LogPrint(BCLog::NET, "%s -- inv=(%s)\n", __func__, inv.ToString()); - g_already_asked_for.erase(inv.hash); - g_erased_object_requests.insert(std::make_pair(inv.hash, GetTime())); - - state->m_object_download.m_object_announced.erase(inv); - state->m_object_download.m_object_in_flight.erase(inv); -} - -bool PeerManagerImpl::ConsumeObjectRequest(NodeId nodeid, const CInv& inv) -{ - AssertLockHeld(cs_main); - - CNodeState* state = State(nodeid); - if (state == nullptr) - return false; - - LogPrint(BCLog::NET, "%s -- inv=(%s)\n", __func__, inv.ToString()); - auto& object_download = state->m_object_download; - const bool announced = object_download.m_object_announced.erase(inv) != 0; - const bool in_flight = object_download.m_object_in_flight.erase(inv) != 0; - // Any leftover m_object_process_time entry for this inv is left in place (removing it here - // would be an O(n) scan). The SendMessages drain skips queued entries whose per-peer - // announced/in-flight state was already consumed, so no redundant GETDATA is issued -- and we - // deliberately do not set the global g_erased_object_requests marker (avoids poisoning it via - // an unsolicited push). - return announced || in_flight; -} - -std::chrono::microseconds GetObjectRequestTime(const CInv& inv) EXCLUSIVE_LOCKS_REQUIRED(cs_main) -{ - AssertLockHeld(cs_main); - auto it = g_already_asked_for.find(inv.hash); - if (it != g_already_asked_for.end()) { - return it->second; - } - return {}; -} - -void UpdateObjectRequestTime(const CInv& inv, std::chrono::microseconds request_time) EXCLUSIVE_LOCKS_REQUIRED(cs_main) -{ - AssertLockHeld(cs_main); - auto it = g_already_asked_for.find(inv.hash); - if (it == g_already_asked_for.end()) { - g_already_asked_for.insert(std::make_pair(inv.hash, request_time)); - } else { - g_already_asked_for.update(it, request_time); - } -} - std::chrono::microseconds GetObjectInterval(int invType) { // some messages need to be re-requested faster when the first announcing peer did not answer to GETDATA @@ -1646,82 +1532,46 @@ std::chrono::microseconds GetObjectInterval(int invType) } } -std::chrono::microseconds GetObjectExpiryInterval(int invType) +void PeerManagerImpl::AddObjectAnnouncement(const CNode& node, const CInv& inv, std::chrono::microseconds current_time) { - return GetObjectInterval(invType) * TX_EXPIRY_INTERVAL_FACTOR; -} + AssertLockHeld(cs_main); -std::chrono::microseconds GetObjectRandomDelay(int invType) -{ - if (invType == MSG_TX) { - return GetRandMicros(MAX_GETDATA_RANDOM_DELAY); - } - return {}; -} + const CNodeState* state = State(node.GetId()); + if (state == nullptr) return; -std::chrono::microseconds CalculateObjectGetDataTime(const CInv& inv, std::chrono::microseconds current_time, bool is_masternode, bool use_inbound_delay) EXCLUSIVE_LOCKS_REQUIRED(cs_main) -{ - AssertLockHeld(cs_main); - std::chrono::microseconds process_time; - const auto last_request_time = GetObjectRequestTime(inv); - // First time requesting this tx - if (last_request_time.count() == 0) { - process_time = current_time; - } else { - // Randomize the delay to avoid biasing some peers over others (such as due to - // fixed ordering of peer processing in ThreadMessageHandler) - process_time = last_request_time + GetObjectInterval(inv.type) + GetObjectRandomDelay(inv.type); + if (m_object_request.Count(node.GetId()) >= MAX_PEER_OBJECT_ANNOUNCEMENTS) { + // Too many queued announcements from this peer + return; } - // We delay processing announcements from inbound peers - if (inv.IsMsgTx() && !is_masternode && use_inbound_delay) process_time += INBOUND_PEER_TX_DELAY; - - return process_time; + // Decide the TxRequestTracker parameters for this announcement: + // - "preferred": if fPreferredDownload is set (= outbound, or PF_NOBAN permission) + // - "reqtime": current time plus delays for: + // - NONPREF_PEER_TX_DELAY for announcements of transactions (both MSG_TX and MSG_DSTX) from + // non-preferred connections. Dash-specific object types are never delayed (they carry + // time-sensitive consensus data), and neither is anything in masternode mode. + // - OVERLOADED_PEER_OBJECT_DELAY for announcements from peers which have at least + // MAX_PEER_OBJECT_REQUEST_IN_FLIGHT requests in flight. + const bool preferred = state->fPreferredDownload; + auto delay{0us}; + if (inv.IsGenTxMsg() && !preferred && m_nodeman == nullptr) delay += NONPREF_PEER_TX_DELAY; + const bool overloaded = m_object_request.CountInFlight(node.GetId()) >= MAX_PEER_OBJECT_REQUEST_IN_FLIGHT; + if (overloaded) delay += OVERLOADED_PEER_OBJECT_DELAY; + + m_object_request.ReceivedInv(node.GetId(), inv, preferred, current_time + delay); } -void PeerManagerImpl::RequestObject(NodeId nodeid, const CInv& inv, std::chrono::microseconds current_time, bool fForce) +void PeerManagerImpl::ForgetTx(const uint256& txid) { AssertLockHeld(cs_main); - - CNodeState* state = State(nodeid); - if (state == nullptr) - return; - - CNodeState::ObjectDownloadState& peer_download_state = state->m_object_download; - if (peer_download_state.m_object_announced.size() >= MAX_PEER_OBJECT_ANNOUNCEMENTS || - peer_download_state.m_object_process_time.size() >= MAX_PEER_OBJECT_ANNOUNCEMENTS || - peer_download_state.m_object_announced.count(inv)) { - // Too many queued announcements from this peer, or we already have - // this announcement - return; - } - peer_download_state.m_object_announced.insert(inv); - - // Calculate the time to try requesting this transaction. Use - // fPreferredDownload as a proxy for outbound peers. - std::chrono::microseconds process_time = CalculateObjectGetDataTime(inv, current_time, /*is_masternode=*/m_nodeman != nullptr, - !state->fPreferredDownload); - - peer_download_state.m_object_process_time.emplace(process_time, inv); - - if (fForce) { - // make sure this object is actually requested ASAP - g_erased_object_requests.erase(inv.hash); - g_already_asked_for.erase(inv.hash); - } - - LogPrint(BCLog::NET, "%s -- inv=(%s), current_time=%d, process_time=%d, delta=%d\n", __func__, inv.ToString(), current_time.count(), process_time.count(), (process_time - current_time).count()); + m_object_request.ForgetTxHash(CInv(MSG_TX, txid)); + m_object_request.ForgetTxHash(CInv(MSG_DSTX, txid)); } size_t PeerManagerImpl::GetRequestedObjectCount(NodeId nodeid) const { AssertLockHeld(cs_main); - - const CNodeState* state = State(nodeid); - if (state == nullptr) - return 0; - - return state->m_object_download.m_object_process_time.size(); + return m_object_request.Count(nodeid); } void PeerManagerImpl::AddExtraHandler(std::unique_ptr&& handler) @@ -1783,6 +1633,7 @@ void PeerManagerImpl::InitializeNode(CNode& node, ServiceFlags our_services) { { LOCK(cs_main); m_node_states.emplace_hint(m_node_states.end(), std::piecewise_construct, std::forward_as_tuple(nodeid), std::forward_as_tuple(node.IsInboundConn())); + assert(m_object_request.Count(nodeid) == 0); } PeerRef peer = std::make_shared(nodeid, our_services); { @@ -1848,6 +1699,7 @@ void PeerManagerImpl::FinalizeNode(const CNode& node) { } } m_orphanage.EraseForPeer(nodeid); + m_object_request.DisconnectedPeer(nodeid); if (m_txreconciliation) m_txreconciliation->ForgetPeer(nodeid); m_num_preferred_download_peers -= state->fPreferredDownload; m_peers_downloading_from -= (!state->vBlocksInFlight.empty()); @@ -1864,6 +1716,7 @@ void PeerManagerImpl::FinalizeNode(const CNode& node) { assert(m_peers_downloading_from == 0); assert(m_outbound_peers_with_protect_from_disconnect == 0); assert(m_orphanage.Size() == 0); + assert(m_object_request.Size() == 0); } } // cs_main @@ -2190,6 +2043,10 @@ void PeerManagerImpl::BlockConnected(const std::shared_ptr& pblock while (have_candidates) { have_candidates = ProcessOrphanTx(/*node_id=*/-1); } + for (const auto& ptx : pblock->vtx) { + // Confirmed transactions no longer need to be requested. + ForgetTx(ptx->GetHash()); + } } m_orphanage.EraseForBlock(*pblock); @@ -2480,13 +2337,22 @@ void PeerManagerImpl::AskPeersForTransaction(const uint256& txid) } } { - CInv inv(MSG_TX, txid); LOCK(cs_main); + const auto current_time{GetTime()}; + // Clear any COMPLETED leftovers first (they would suppress the new announcements), so the + // transaction is requested ASAP even if earlier requests for it failed. + m_object_request.ForgetTxHash(CInv(MSG_TX, txid)); + m_object_request.ForgetTxHash(CInv(MSG_DSTX, txid)); for (PeerRef& peer : peersToAsk) { + // The peer may have been disconnected (and its tracker state wiped by DisconnectedPeer) + // after we collected it above but before we took cs_main. Registering an announcement + // for a gone peer would leave a candidate that is never requested and could block the + // live fallback peers, so skip it. + if (State(peer->m_id) == nullptr) continue; LogPrintf("PeerManagerImpl::%s -- txid=%s: asking other peer %d for correct TX\n", __func__, txid.ToString(), peer->m_id); - RequestObject(peer->m_id, inv, GetTime(), /*fForce=*/true); + m_object_request.ReceivedInv(peer->m_id, CInv(MSG_TX, txid), /*preferred=*/true, current_time); } } } @@ -3376,6 +3242,7 @@ bool PeerManagerImpl::ProcessOrphanTx(NodeId node_id) _RelayTransaction(porphanTx->GetHash()); m_orphanage.AddChildrenToWorkSet(*porphanTx, node_id); m_orphanage.EraseTx(orphanHash); + ForgetTx(orphanHash); break; } else if (state.GetResult() != TxValidationResult::TX_MISSING_INPUTS) { if (state.IsInvalid()) { @@ -3391,6 +3258,7 @@ bool PeerManagerImpl::ProcessOrphanTx(NodeId node_id) LogPrint(BCLog::MEMPOOL, " removed orphan tx %s\n", orphanHash.ToString()); m_recent_rejects.insert(orphanHash); m_orphanage.EraseTx(orphanHash); + ForgetTx(orphanHash); break; } } @@ -3759,12 +3627,15 @@ void PeerManagerImpl::PostProcessMessage(MessageProcessingResult&& result, NodeI if (peer) Misbehaving(*peer, result.m_error->score, result.m_error->message); } if (result.m_to_erase) { - WITH_LOCK(cs_main, EraseObjectRequest(node, result.m_to_erase.value())); + WITH_LOCK(cs_main, m_object_request.ReceivedResponse(node, result.m_to_erase.value())); } for (const auto& tx : result.m_transactions) { WITH_LOCK(cs_main, _RelayTransaction(tx)); } for (const auto& inv : result.m_inventory) { + // An inv being relayed is available locally, so there is no need to request it from + // anyone anymore. + WITH_LOCK(cs_main, m_object_request.ForgetTxHash(inv)); RelayInv(inv); } } @@ -4486,7 +4357,7 @@ void PeerManagerImpl::ProcessMessage( } bool allowWhileInIBD = allowWhileInIBDObjs.count(inv.type); if (allowWhileInIBD || !m_chainman.ActiveChainstate().IsInitialBlockDownload()) { - RequestObject(pfrom.GetId(), inv, current_time); + AddObjectAnnouncement(pfrom, inv, current_time); } } } @@ -4798,7 +4669,11 @@ void PeerManagerImpl::ProcessMessage( CInv inv(nInvType, tx.GetHash()); { LOCK(cs_main); - EraseObjectRequest(pfrom.GetId(), inv); + // A MSG_TX request may be answered with a DSTX message and vice versa (a getdata for + // either type serves the underlying transaction), so complete whichever announcement + // type the request was tracked under. + m_object_request.ReceivedResponse(pfrom.GetId(), CInv(MSG_TX, txid)); + m_object_request.ReceivedResponse(pfrom.GetId(), CInv(MSG_DSTX, txid)); } // Process custom logic, no matter if tx will be accepted to mempool later or not @@ -4842,6 +4717,7 @@ void PeerManagerImpl::ProcessMessage( m_dstxman.AddDSTX(dstx); } + ForgetTx(tx.GetHash()); _RelayTransaction(tx.GetHash()); m_orphanage.AddChildrenToWorkSet(tx, peer->m_id); @@ -4879,18 +4755,22 @@ void PeerManagerImpl::ProcessMessage( const auto current_time{GetTime()}; for (const uint256& parent_txid : unique_parents) { - CInv _inv(MSG_TX, parent_txid); + // We don't know if the parent was a regular or a mixing tx, so ask for it as + // MSG_DSTX: a getdata for MSG_DSTX is answered with the DSTX wrapper when the + // peer has one and with a plain TX otherwise, while a MSG_TX getdata would + // lose the metadata of a zero-fee DSTX. Announcing both types would get the + // parent fetched twice, as the tracker keys announcements on the full inv. + CInv _inv(MSG_DSTX, parent_txid); AddKnownInv(*peer, _inv.hash); - if (!AlreadyHave(_inv)) RequestObject(pfrom.GetId(), _inv, current_time); - // We don't know if the previous tx was a regular or a mixing one, try both - CInv _inv2(MSG_DSTX, parent_txid); - AddKnownInv(*peer, _inv2.hash); - if (!AlreadyHave(_inv2)) RequestObject(pfrom.GetId(), _inv2, current_time); + if (!AlreadyHave(_inv)) AddObjectAnnouncement(pfrom, _inv, current_time); } if (m_orphanage.AddTx(ptx, pfrom.GetId())) { AddToCompactExtraTransactions(ptx); } + // Once added to the orphan pool, a tx is considered AlreadyHave, and we shouldn't + // request it anymore. + ForgetTx(tx.GetHash()); // DoS prevention: do not allow m_orphans to grow unbounded (see CVE-2012-3789) unsigned int nMaxOrphanTxSize = (unsigned int)std::max((int64_t)0, gArgs.GetIntArg("-maxorphantxsize", DEFAULT_MAX_ORPHAN_TRANSACTIONS_SIZE)) * 1000000; @@ -4900,10 +4780,12 @@ void PeerManagerImpl::ProcessMessage( // We will continue to reject this tx since it has rejected // parents so avoid re-requesting it from other peers. m_recent_rejects.insert(tx.GetHash()); + ForgetTx(tx.GetHash()); m_llmq_ctx->isman->TransactionIsRemoved(ptx); } } else { m_recent_rejects.insert(tx.GetHash()); + ForgetTx(tx.GetHash()); if (RecursiveDynamicUsage(*ptx) < 100000) { AddToCompactExtraTransactions(ptx); } @@ -5566,13 +5448,14 @@ void PeerManagerImpl::ProcessMessage( uint256 hash = spork.GetHash(); CInv spork_inv{MSG_SPORK, hash}; - WITH_LOCK(::cs_main, EraseObjectRequest(pfrom.GetId(), spork_inv)); + WITH_LOCK(::cs_main, m_object_request.ReceivedResponse(pfrom.GetId(), spork_inv)); auto opt_signer = m_sporkman.GetValidSporkSigner(spork); if (!opt_signer) { Misbehaving(*peer, 100, strprintf("invalid spork received. peer=%d", pfrom.GetId())); return; } if (m_sporkman.ProcessSpork(spork, *opt_signer, strprintf(" peer=%d", pfrom.GetId()))) { + WITH_LOCK(::cs_main, m_object_request.ForgetTxHash(spork_inv)); RelayInv(spork_inv); } return; @@ -5613,28 +5496,22 @@ void PeerManagerImpl::ProcessMessage( return; } if (msg_type == NetMsgType::NOTFOUND) { - // Remove the NOTFOUND transactions from the peer + // Remove the NOTFOUND objects from the peer std::vector vInv; vRecv >> vInv; - if (vInv.size() > MAX_PEER_OBJECT_IN_FLIGHT + MAX_BLOCKS_IN_TRANSIT_PER_PEER) { + // A malicious peer can send a NOTFOUND entry per tracked announcement, so bound the + // message size by the announcement cap rather than the (soft) in-flight limit. + if (vInv.size() > MAX_PEER_OBJECT_ANNOUNCEMENTS + MAX_BLOCKS_IN_TRANSIT_PER_PEER) { Misbehaving(*peer, 20, strprintf("notfound message size = %u", vInv.size())); return; } LOCK(cs_main); - CNodeState *state = State(pfrom.GetId()); for (CInv &inv : vInv) { if (inv.IsKnownType()) { - // If we receive a NOTFOUND message for a txid we requested, erase - // it from our data structures for this peer. - auto in_flight_it = state->m_object_download.m_object_in_flight.find(inv); - if (in_flight_it == state->m_object_download.m_object_in_flight.end()) { - // Skip any further work if this is a spurious NOTFOUND - // message. - continue; - } - state->m_object_download.m_object_in_flight.erase(in_flight_it); - state->m_object_download.m_object_announced.erase(inv); + // If we receive a NOTFOUND message for an inv we requested, mark the announcement + // as completed, so a fallback peer can be tried. + m_object_request.ReceivedResponse(pfrom.GetId(), inv); } } return; @@ -5664,7 +5541,7 @@ void PeerManagerImpl::ProcessMessage( chainlock::ChainLockSig clsig; vRecv >> clsig; const uint256& hash = ::SerializeHash(clsig); - WITH_LOCK(::cs_main, EraseObjectRequest(pfrom.GetId(), CInv{MSG_CLSIG, hash})); + WITH_LOCK(::cs_main, m_object_request.ReceivedResponse(pfrom.GetId(), CInv{MSG_CLSIG, hash})); PostProcessMessage(m_clhandler.ProcessNewChainLock(pfrom.GetId(), clsig, *m_llmq_ctx->qman, hash), pfrom.GetId()); } return; // CLSIG @@ -6680,74 +6557,28 @@ bool PeerManagerImpl::SendMessages(CNode* pto) // Message: getdata (non-blocks) // - // For robustness, expire old requests after a long timeout, so that - // we can resume downloading objects from a peer even if they - // were unresponsive in the past. - // Eventually we should consider disconnecting peers, but this is - // conservative. - if (state.m_object_download.m_check_expiry_timer <= current_time) { - for (auto it=state.m_object_download.m_object_in_flight.begin(); it != state.m_object_download.m_object_in_flight.end();) { - if (it->second <= current_time - GetObjectExpiryInterval(it->first.type)) { - LogPrint(BCLog::NET, "timeout of inflight object %s from peer=%d\n", it->first.ToString(), pto->GetId()); - state.m_object_download.m_object_announced.erase(it->first); - state.m_object_download.m_object_in_flight.erase(it++); - } else { - ++it; - } - } - // On average, we do this check every GetObjectExpiryInterval. Randomize - // so that we're not doing this for all peers at the same time. - state.m_object_download.m_check_expiry_timer = current_time + GetObjectExpiryInterval(MSG_TX)/2 + GetRandMicros(GetObjectExpiryInterval(MSG_TX)); - } - - // DASH this code also handles non-TXs (Dash specific messages) - auto& object_process_time = state.m_object_download.m_object_process_time; - while (!object_process_time.empty() && object_process_time.begin()->first <= current_time && state.m_object_download.m_object_in_flight.size() < MAX_PEER_OBJECT_IN_FLIGHT) { - const CInv inv = object_process_time.begin()->second; - // Erase this entry from object_process_time (it may be added back for - // processing at a later time, see below) - object_process_time.erase(object_process_time.begin()); - // Drop stale entries whose per-peer request was already consumed or received (no - // longer announced or in-flight), so a consumed announcement is not re-requested. - // This covers ConsumeObjectRequest, which erases that state without setting the - // global g_erased_object_requests marker checked below. - if (state.m_object_download.m_object_announced.count(inv) == 0 && - state.m_object_download.m_object_in_flight.count(inv) == 0) { - continue; - } - if (g_erased_object_requests.count(inv.hash)) { - LogPrint(BCLog::NET, "%s -- GETDATA skipping inv=(%s), peer=%d\n", __func__, inv.ToString(), pto->GetId()); - state.m_object_download.m_object_announced.erase(inv); - state.m_object_download.m_object_in_flight.erase(inv); - continue; - } + // DASH unlike Bitcoin, this loop requests all Dash-specific object types too. The request + // expiry doubles as the fallback-to-another-peer trigger, so time-sensitive object types + // use a shorter per-type interval (see GetObjectInterval). + std::vector> expired; + auto requestable = m_object_request.GetRequestable(pto->GetId(), current_time, &expired); + for (const auto& entry : expired) { + LogPrint(BCLog::NET, "timeout of inflight object %s from peer=%d\n", entry.second.ToString(), entry.first); + } + for (const CInv& inv : requestable) { if (!AlreadyHave(inv)) { - // If this object was last requested more than GetObjectInterval ago, - // then request. - const auto last_request_time = GetObjectRequestTime(inv); - if (last_request_time <= current_time - GetObjectInterval(inv.type)) { - LogPrint(BCLog::NET, "Requesting %s peer=%d\n", inv.ToString(), pto->GetId()); - vGetData.push_back(inv); - if (vGetData.size() >= MAX_GETDATA_SZ) { - m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); - vGetData.clear(); - } - UpdateObjectRequestTime(inv, current_time); - state.m_object_download.m_object_in_flight.emplace(inv, current_time); - } else { - // This object is in flight from someone else; queue - // up processing to happen after the download times out - // (with a slight delay for inbound peers, to prefer - // requests to outbound peers). - const auto next_process_time = CalculateObjectGetDataTime(inv, current_time, is_masternode, !state.fPreferredDownload); - object_process_time.emplace(next_process_time, inv); - LogPrint(BCLog::NET, "%s -- GETDATA re-queue inv=(%s), next_process_time=%d, delta=%d, peer=%d\n", __func__, inv.ToString(), next_process_time.count(), (next_process_time - current_time).count(), pto->GetId()); + LogPrint(BCLog::NET, "Requesting %s peer=%d\n", inv.ToString(), pto->GetId()); + vGetData.push_back(inv); + if (vGetData.size() >= MAX_GETDATA_SZ) { + m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); + vGetData.clear(); } + m_object_request.RequestedTx(pto->GetId(), inv, current_time + GetObjectInterval(inv.type)); } else { - // We have already seen this object, no need to download. - state.m_object_download.m_object_announced.erase(inv); - state.m_object_download.m_object_in_flight.erase(inv); - LogPrint(BCLog::NET, "%s -- GETDATA already seen inv=(%s), peer=%d\n", __func__, inv.ToString(), pto->GetId()); + // We have already seen this object, no need to download. This is for belated + // announcements of objects which arrived via another peer; the tracker has no + // direct means to remove them once the object is received elsewhere. + m_object_request.ForgetTxHash(inv); } } @@ -6773,12 +6604,20 @@ bool PeerManagerImpl::PeerIsBanned(const NodeId node_id) void PeerManagerImpl::PeerEraseObjectRequest(const NodeId nodeid, const CInv& inv) { - EraseObjectRequest(nodeid, inv); + // Completing only this peer's announcement is deliberate: an invalid or unusable object must + // not stop us from fetching it from honest peers. Cleanup across peers happens once the object + // is accepted and AlreadyHave(inv) turns true. + m_object_request.ReceivedResponse(nodeid, inv); } bool PeerManagerImpl::PeerConsumeObjectRequest(NodeId nodeid, const CInv& inv) { - return ConsumeObjectRequest(nodeid, inv); + return m_object_request.ReceivedResponse(nodeid, inv); +} + +void PeerManagerImpl::PeerForgetObjectRequest(const CInv& inv) +{ + m_object_request.ForgetTxHash(inv); } void PeerManagerImpl::PeerPushInventory(NodeId nodeid, const CInv& inv) diff --git a/src/net_processing.h b/src/net_processing.h index 0abe1fcfd6f8..965a50ea8c24 100644 --- a/src/net_processing.h +++ b/src/net_processing.h @@ -64,14 +64,20 @@ class PeerManagerInternal public: virtual void PeerMisbehaving(const NodeId pnode, const int howmuch, const std::string& message = "") = 0; virtual bool PeerIsBanned(const NodeId node_id) = 0; - /** Erase a pending object request for a peer and update global request tracking. */ + /** Complete this peer's pending announcement of the inv, so it is not requested from them + * again. Announcements of the same inv by other peers are unaffected: an invalid or unusable + * object must not stop us from fetching it from honest peers. + * Requires ::cs_main (see the PeerManagerImpl override). */ virtual void PeerEraseObjectRequest(const NodeId nodeid, const CInv& inv) = 0; - /** Consume this peer's pending request for the inv -- erase the matching per-peer announced / - * in-flight state -- and return whether such a request existed (the peer announced the inv, or - * we requested it from the peer). Any queued m_object_process_time entry is left in place; the - * SendMessages drain skips it once the announced/in-flight state is gone. + /** Consume this peer's pending request for the inv and return whether such a request existed + * (the peer announced the inv, or we requested it from the peer). Consuming completes the + * announcement, so a second call without a re-announcement in between returns false. * Requires ::cs_main (see the PeerManagerImpl override). */ virtual bool PeerConsumeObjectRequest(NodeId nodeid, const CInv& inv) = 0; + /** Delete all peers' announcements of the inv. Call once the object is accepted (AlreadyHave + * turns true), so it is not requested from anyone anymore. + * Requires ::cs_main (see the PeerManagerImpl override). */ + virtual void PeerForgetObjectRequest(const CInv& inv) = 0; virtual void PeerPushInventory(NodeId nodeid, const CInv& inv) = 0; virtual void PeerRelayInv(const CInv& inv) = 0; virtual void PeerRelayInvFiltered(const CInv& inv, const CTransaction& relatedTx) = 0; diff --git a/src/protocol.h b/src/protocol.h index 8252d8ab48b2..6e39faa23dd0 100644 --- a/src/protocol.h +++ b/src/protocol.h @@ -536,6 +536,7 @@ class CInv SERIALIZE_METHODS(CInv, obj) { READWRITE(obj.type, obj.hash); } friend bool operator<(const CInv& a, const CInv& b); + friend bool operator==(const CInv& a, const CInv& b) { return a.type == b.type && a.hash == b.hash; } bool IsKnownType() const; std::string GetCommand() const; diff --git a/src/test/fuzz/txrequest.cpp b/src/test/fuzz/txrequest.cpp new file mode 100644 index 000000000000..35767d555f6d --- /dev/null +++ b/src/test/fuzz/txrequest.cpp @@ -0,0 +1,380 @@ +// Copyright (c) 2020 The Bitcoin Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +namespace { + +constexpr int MAX_TXHASHES = 16; +constexpr int MAX_PEERS = 16; + +//! Randomly generated invs used in this test (length is 2 * MAX_TXHASHES; MSG_TX and MSG_DSTX +//! announcements sharing each hash, to exercise same-hash-different-type independence). +constexpr int MAX_INVS = 2 * MAX_TXHASHES; +CInv INVS[MAX_INVS]; + +//! Precomputed random durations (positive and negative, each ~exponentially distributed). +std::chrono::microseconds DELAYS[256]; + +struct Initializer +{ + Initializer() + { + for (uint8_t txhash = 0; txhash < MAX_TXHASHES; txhash += 1) { + uint256 hash; + CSHA256().Write(&txhash, 1).Finalize(hash.begin()); + INVS[2 * txhash] = CInv{MSG_TX, hash}; + INVS[2 * txhash + 1] = CInv{MSG_DSTX, hash}; + } + int i = 0; + // DELAYS[N] for N=0..15 is just N microseconds. + for (; i < 16; ++i) { + DELAYS[i] = std::chrono::microseconds{i}; + } + // DELAYS[N] for N=16..127 has randomly-looking but roughly exponentially increasing values up to + // 198.416453 seconds. + for (; i < 128; ++i) { + int diff_bits = ((i - 10) * 2) / 9; + uint64_t diff = 1 + (CSipHasher(0, 0).Write(i).Finalize() >> (64 - diff_bits)); + DELAYS[i] = DELAYS[i - 1] + std::chrono::microseconds{diff}; + } + // DELAYS[N] for N=128..255 are negative delays with the same magnitude as N=0..127. + for (; i < 256; ++i) { + DELAYS[i] = -DELAYS[255 - i]; + } + } +} g_initializer; + +/** Tester class for TxRequestTracker + * + * It includes a naive reimplementation of its behavior, for a limited set + * of MAX_INVS distinct invs, and MAX_PEERS peer identifiers. + * + * All of the public member functions perform the same operation on + * an actual TxRequestTracker and on the state of the reimplementation. + * The output of GetRequestable is compared with the expected value + * as well. + * + * Check() calls the TxRequestTracker's sanity check, plus compares the + * output of the constant accessors (Size(), CountLoad(), CountTracked()) + * with expected values. + */ +class Tester +{ + //! TxRequestTracker object being tested. + TxRequestTracker m_tracker; + + //! States for inv/peer combinations in the naive data structure. + enum class State { + NOTHING, //!< Absence of this inv/peer combination + + // Note that this implementation does not distinguish between DELAYED/READY/BEST variants of CANDIDATE. + CANDIDATE, + REQUESTED, + COMPLETED, + }; + + //! Sequence numbers, incremented whenever a new CANDIDATE is added. + uint64_t m_current_sequence{0}; + + //! List of future 'events' (all inserted reqtimes/exptimes). This is used to implement AdvanceToEvent. + std::priority_queue, + std::greater> m_events; + + //! Information about an inv/peer combination. + struct Announcement + { + std::chrono::microseconds m_time; + uint64_t m_sequence; + State m_state{State::NOTHING}; + bool m_preferred; + uint64_t m_priority; //!< Precomputed priority. + }; + + //! Information about all inv/peer combinations. + Announcement m_announcements[MAX_INVS][MAX_PEERS]; + + //! The current time; can move forward and backward. + std::chrono::microseconds m_now{244466666}; + + //! Delete invs whose only announcements are COMPLETED. + void Cleanup(int inv) + { + bool all_nothing = true; + for (int peer = 0; peer < MAX_PEERS; ++peer) { + const Announcement& ann = m_announcements[inv][peer]; + if (ann.m_state != State::NOTHING) { + if (ann.m_state != State::COMPLETED) return; + all_nothing = false; + } + } + if (all_nothing) return; + for (int peer = 0; peer < MAX_PEERS; ++peer) { + m_announcements[inv][peer].m_state = State::NOTHING; + } + } + + //! Find the current best peer to request from for an inv (or -1 if none). + int GetSelected(int inv) const + { + int ret = -1; + uint64_t ret_priority = 0; + for (int peer = 0; peer < MAX_PEERS; ++peer) { + const Announcement& ann = m_announcements[inv][peer]; + // Return -1 if there already is a (non-expired) in-flight request. + if (ann.m_state == State::REQUESTED) return -1; + // If it's a viable candidate, see if it has lower priority than the best one so far. + if (ann.m_state == State::CANDIDATE && ann.m_time <= m_now) { + if (ret == -1 || ann.m_priority > ret_priority) { + std::tie(ret, ret_priority) = std::tie(peer, ann.m_priority); + } + } + } + return ret; + } + +public: + Tester() : m_tracker(true) {} + + std::chrono::microseconds Now() const { return m_now; } + + void AdvanceTime(std::chrono::microseconds offset) + { + m_now += offset; + while (!m_events.empty() && m_events.top() <= m_now) m_events.pop(); + } + + void AdvanceToEvent() + { + while (!m_events.empty() && m_events.top() <= m_now) m_events.pop(); + if (!m_events.empty()) { + m_now = m_events.top(); + m_events.pop(); + } + } + + void DisconnectedPeer(int peer) + { + // Apply to naive structure: all announcements for that peer are wiped. + for (int inv = 0; inv < MAX_INVS; ++inv) { + if (m_announcements[inv][peer].m_state != State::NOTHING) { + m_announcements[inv][peer].m_state = State::NOTHING; + Cleanup(inv); + } + } + + // Call TxRequestTracker's implementation. + m_tracker.DisconnectedPeer(peer); + } + + void ForgetTxHash(int inv) + { + // Apply to naive structure: all announcements for that inv are wiped. + for (int peer = 0; peer < MAX_PEERS; ++peer) { + m_announcements[inv][peer].m_state = State::NOTHING; + } + Cleanup(inv); + + // Call TxRequestTracker's implementation. + m_tracker.ForgetTxHash(INVS[inv]); + } + + void ReceivedInv(int peer, int inv, bool preferred, std::chrono::microseconds reqtime) + { + // Apply to naive structure: if no announcement for invnum/peer combination + // already, create a new CANDIDATE; otherwise do nothing. + Announcement& ann = m_announcements[inv][peer]; + if (ann.m_state == State::NOTHING) { + ann.m_preferred = preferred; + ann.m_state = State::CANDIDATE; + ann.m_time = reqtime; + ann.m_sequence = m_current_sequence++; + ann.m_priority = m_tracker.ComputePriority(INVS[inv], peer, ann.m_preferred); + + // Add event so that AdvanceToEvent can quickly jump to the point where its reqtime passes. + if (reqtime > m_now) m_events.push(reqtime); + } + + // Call TxRequestTracker's implementation. + m_tracker.ReceivedInv(peer, INVS[inv], preferred, reqtime); + } + + void RequestedTx(int peer, int inv, std::chrono::microseconds exptime) + { + // Apply to naive structure: if a CANDIDATE announcement exists for peer/inv, + // convert it to REQUESTED, and change any existing REQUESTED announcement for the same inv to COMPLETED. + if (m_announcements[inv][peer].m_state == State::CANDIDATE) { + for (int peer2 = 0; peer2 < MAX_PEERS; ++peer2) { + if (m_announcements[inv][peer2].m_state == State::REQUESTED) { + m_announcements[inv][peer2].m_state = State::COMPLETED; + } + } + m_announcements[inv][peer].m_state = State::REQUESTED; + m_announcements[inv][peer].m_time = exptime; + } + + // Add event so that AdvanceToEvent can quickly jump to the point where its exptime passes. + if (exptime > m_now) m_events.push(exptime); + + // Call TxRequestTracker's implementation. + m_tracker.RequestedTx(peer, INVS[inv], exptime); + } + + void ReceivedResponse(int peer, int inv) + { + // Apply to naive structure: convert CANDIDATE/REQUESTED to COMPLETED. The response completes an + // announcement precisely if one existed that wasn't already COMPLETED. + const State old_state = m_announcements[inv][peer].m_state; + const bool expected_completed = old_state == State::CANDIDATE || old_state == State::REQUESTED; + if (old_state != State::NOTHING) { + m_announcements[inv][peer].m_state = State::COMPLETED; + Cleanup(inv); + } + + // Call TxRequestTracker's implementation, and compare its return value with the naive expectation. + const bool completed = m_tracker.ReceivedResponse(peer, INVS[inv]); + assert(completed == expected_completed); + } + + void GetRequestable(int peer) + { + // Implement using naive structure: + + //! list of (sequence number, inv) pairs. + std::vector> result; + std::vector> expected_expired; + for (int inv = 0; inv < MAX_INVS; ++inv) { + // Mark any expired REQUESTED announcements as COMPLETED. + for (int peer2 = 0; peer2 < MAX_PEERS; ++peer2) { + Announcement& ann2 = m_announcements[inv][peer2]; + if (ann2.m_state == State::REQUESTED && ann2.m_time <= m_now) { + expected_expired.emplace_back(peer2, INVS[inv]); + ann2.m_state = State::COMPLETED; + break; + } + } + // And delete invs with only COMPLETED announcements left. + Cleanup(inv); + // CANDIDATEs for which this announcement has the highest priority get returned. + const Announcement& ann = m_announcements[inv][peer]; + if (ann.m_state == State::CANDIDATE && GetSelected(inv) == peer) { + result.emplace_back(ann.m_sequence, inv); + } + } + // Sort the results by sequence number. + std::sort(result.begin(), result.end()); + std::sort(expected_expired.begin(), expected_expired.end()); + + // Compare with TxRequestTracker's implementation. + std::vector> expired; + const auto actual = m_tracker.GetRequestable(peer, m_now, &expired); + std::sort(expired.begin(), expired.end()); + assert(expired == expected_expired); + + m_tracker.PostGetRequestableSanityCheck(m_now); + assert(result.size() == actual.size()); + for (size_t pos = 0; pos < actual.size(); ++pos) { + assert(INVS[result[pos].second] == actual[pos]); + } + } + + void Check() + { + // Compare CountTracked and CountLoad with naive structure. + size_t total = 0; + for (int peer = 0; peer < MAX_PEERS; ++peer) { + size_t tracked = 0; + size_t inflight = 0; + size_t candidates = 0; + for (int inv = 0; inv < MAX_INVS; ++inv) { + tracked += m_announcements[inv][peer].m_state != State::NOTHING; + inflight += m_announcements[inv][peer].m_state == State::REQUESTED; + candidates += m_announcements[inv][peer].m_state == State::CANDIDATE; + } + assert(m_tracker.Count(peer) == tracked); + assert(m_tracker.CountInFlight(peer) == inflight); + assert(m_tracker.CountCandidates(peer) == candidates); + total += tracked; + } + // Compare Size. + assert(m_tracker.Size() == total); + + // Invoke internal consistency check of TxRequestTracker object. + m_tracker.SanityCheck(); + } +}; +} // namespace + +FUZZ_TARGET(txrequest) +{ + // Tester object (which encapsulates a TxRequestTracker). + Tester tester; + + // Decode the input as a sequence of instructions with parameters + auto it = buffer.begin(); + while (it != buffer.end()) { + int cmd = *(it++) % 11; + int peer, invnum, delaynum; + switch (cmd) { + case 0: // Make time jump to the next event (m_time of CANDIDATE or REQUESTED) + tester.AdvanceToEvent(); + break; + case 1: // Change time + delaynum = it == buffer.end() ? 0 : *(it++); + tester.AdvanceTime(DELAYS[delaynum]); + break; + case 2: // Query for requestable txs + peer = it == buffer.end() ? 0 : *(it++) % MAX_PEERS; + tester.GetRequestable(peer); + break; + case 3: // Peer went offline + peer = it == buffer.end() ? 0 : *(it++) % MAX_PEERS; + tester.DisconnectedPeer(peer); + break; + case 4: // No longer need tx + invnum = it == buffer.end() ? 0 : *(it++); + tester.ForgetTxHash(invnum % MAX_INVS); + break; + case 5: // Received immediate preferred inv + case 6: // Same, but non-preferred. + peer = it == buffer.end() ? 0 : *(it++) % MAX_PEERS; + invnum = it == buffer.end() ? 0 : *(it++); + tester.ReceivedInv(peer, invnum % MAX_INVS, cmd & 1, + std::chrono::microseconds::min()); + break; + case 7: // Received delayed preferred inv + case 8: // Same, but non-preferred. + peer = it == buffer.end() ? 0 : *(it++) % MAX_PEERS; + invnum = it == buffer.end() ? 0 : *(it++); + delaynum = it == buffer.end() ? 0 : *(it++); + tester.ReceivedInv(peer, invnum % MAX_INVS, cmd & 1, + tester.Now() + DELAYS[delaynum]); + break; + case 9: // Requested tx from peer + peer = it == buffer.end() ? 0 : *(it++) % MAX_PEERS; + invnum = it == buffer.end() ? 0 : *(it++); + delaynum = it == buffer.end() ? 0 : *(it++); + tester.RequestedTx(peer, invnum % MAX_INVS, tester.Now() + DELAYS[delaynum]); + break; + case 10: // Received response + peer = it == buffer.end() ? 0 : *(it++) % MAX_PEERS; + invnum = it == buffer.end() ? 0 : *(it++); + tester.ReceivedResponse(peer, invnum % MAX_INVS); + break; + default: + assert(false); + } + } + tester.Check(); +} diff --git a/src/test/net_tests.cpp b/src/test/net_tests.cpp index e8d75346ec5f..7d4a48116554 100644 --- a/src/test/net_tests.cpp +++ b/src/test/net_tests.cpp @@ -94,23 +94,19 @@ BOOST_AUTO_TEST_CASE(peer_requested_object_authorizes_and_erases_per_peer_state) ProcessInv(*m_node.peerman, *peer, announced_inv); // The announcement is queued for a GETDATA that hasn't been sent yet. BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 1U); - // Consuming clears the peer's announced/in-flight state and returns true exactly once. + // Consuming completes the peer's announcement and returns true exactly once. BOOST_CHECK(WITH_LOCK(::cs_main, return m_node.peerman->PeerConsumeObjectRequest(peer->GetId(), announced_inv))); BOOST_CHECK(!WITH_LOCK(::cs_main, return m_node.peerman->PeerConsumeObjectRequest(peer->GetId(), announced_inv))); - // The queued GETDATA is left in place by the consume, but SendMessages must skip it (the - // per-peer request was consumed) rather than re-requesting, and drains the stale entry. - // Since this path never sets g_erased_object_requests, that skip relies on the drain-side - // announced/in-flight check. + // A consumed announcement must not be requested by SendMessages. SetMockTime(GetTime() + 61s); m_node.peerman->SendMessages(peer.get()); BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 0U); - // Not re-requested: no in-flight entry was created for the consumed announcement. + // Not re-requested: consuming did not resurrect the announcement. BOOST_CHECK(!WITH_LOCK(::cs_main, return m_node.peerman->PeerConsumeObjectRequest(peer->GetId(), announced_inv))); // Authorization must also survive getdata scheduling. After SendMessages issues the - // GETDATA the inv is in-flight (and still announced); PeerConsumeObjectRequest returns true - // for either state, so this checks that a requested inv authorizes -- not the in-flight - // branch specifically. + // GETDATA the announcement is in the requested state; PeerConsumeObjectRequest returns true + // for both announced and requested states, so this checks that a requested inv authorizes. const CInv requested_inv{MSG_SPORK, uint256S("02")}; ProcessInv(*m_node.peerman, *peer, requested_inv); SetMockTime(GetTime() + 61s); @@ -121,9 +117,8 @@ BOOST_AUTO_TEST_CASE(peer_requested_object_authorizes_and_erases_per_peer_state) const CInv unsolicited_inv{MSG_SPORK, uint256S("03")}; BOOST_CHECK(!WITH_LOCK(::cs_main, return m_node.peerman->PeerConsumeObjectRequest(peer->GetId(), unsolicited_inv))); - // A failed per-peer authorization check must not poison the global erased-object marker: - // a later legitimate announcement for the same hash must still be requested (GETDATA - // scheduled) and therefore authorize. + // A failed authorization check must leave no trace: a later legitimate announcement for the + // same hash must still be requested (GETDATA scheduled) and therefore authorize. ProcessInv(*m_node.peerman, *peer, unsolicited_inv); SetMockTime(GetTime() + 61s); m_node.peerman->SendMessages(peer.get()); diff --git a/src/test/txrequest_tests.cpp b/src/test/txrequest_tests.cpp new file mode 100644 index 000000000000..816391165be9 --- /dev/null +++ b/src/test/txrequest_tests.cpp @@ -0,0 +1,763 @@ +// Copyright (c) 2020 The Bitcoin Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + + +#include +#include + +#include +#include + +#include +#include +#include +#include + +#include + +std::string v2s(const std::vector& v) { + std::string out; + for (auto i : v) { + if (!out.empty()) out += ", "; + out += i.ToString(); + } + return strprintf("v[%d] = {%s}", v.size(), out); +} +BOOST_FIXTURE_TEST_SUITE(txrequest_tests, BasicTestingSetup) + +namespace { + +constexpr std::chrono::microseconds MIN_TIME = std::chrono::microseconds::min(); +constexpr std::chrono::microseconds MAX_TIME = std::chrono::microseconds::max(); +constexpr std::chrono::microseconds MICROSECOND = std::chrono::microseconds{1}; +constexpr std::chrono::microseconds NO_TIME = std::chrono::microseconds{0}; + +/** An Action is a function to call at a particular (simulated) timestamp. */ +using Action = std::pair>; + +/** Object that stores actions from multiple interleaved scenarios, and data shared across them. + * + * The Scenario below is used to fill this. + */ +struct Runner +{ + /** The TxRequestTracker being tested. */ + TxRequestTracker txrequest; + + /** List of actions to be executed (in order of increasing timestamp). */ + std::vector actions; + + /** Which node ids have been assigned already (to prevent reuse). */ + std::set peerset; + + /** Which txhashes have been assigned already (to prevent reuse). */ + std::set txhashset; + + /** Which (peer, inv) combinations are known to be expired. These need to be accumulated here instead of + * checked directly in the GetRequestable return value to avoid introducing a dependency between the various + * parallel tests. */ + std::multiset> expired; +}; + +std::chrono::microseconds RandomTime8s() { return std::chrono::microseconds{1 + InsecureRandBits(23)}; } +std::chrono::microseconds RandomTime1y() { return std::chrono::microseconds{1 + InsecureRandBits(45)}; } + +/** A proxy for a Runner that helps build a sequence of consecutive test actions on a TxRequestTracker. + * + * Each Scenario is a proxy through which actions for the (sequential) execution of various tests are added to a + * Runner. The actions from multiple scenarios are then run concurrently, resulting in these tests being performed + * against a TxRequestTracker in parallel. Every test has its own unique txhashes and NodeIds which are not + * reused in other tests, and thus they should be independent from each other. Running them in parallel however + * means that we verify the behavior (w.r.t. one test's txhashes and NodeIds) even when the state of the data + * structure is more complicated due to the presence of other tests. + */ +class Scenario +{ + Runner& m_runner; + std::chrono::microseconds m_now; + std::string m_testname; + +public: + Scenario(Runner& runner, std::chrono::microseconds starttime) : m_runner(runner), m_now(starttime) {} + + /** Set a name for the current test, to give more clear error messages. */ + void SetTestName(std::string testname) + { + m_testname = std::move(testname); + } + + /** Advance this Scenario's time; this affects the timestamps newly scheduled events get. */ + void AdvanceTime(std::chrono::microseconds amount) + { + assert(amount.count() >= 0); + m_now += amount; + } + + /** Schedule a ForgetTxHash call at the Scheduler's current time. */ + void ForgetTxHash(const CInv& txhash) + { + auto& runner = m_runner; + runner.actions.emplace_back(m_now, [=,&runner]() { + runner.txrequest.ForgetTxHash(txhash); + runner.txrequest.SanityCheck(); + }); + } + + /** Schedule a ReceivedInv call at the Scheduler's current time. */ + void ReceivedInv(NodeId peer, const CInv& gtxid, bool pref, std::chrono::microseconds reqtime) + { + auto& runner = m_runner; + runner.actions.emplace_back(m_now, [=,&runner]() { + runner.txrequest.ReceivedInv(peer, gtxid, pref, reqtime); + runner.txrequest.SanityCheck(); + }); + } + + /** Schedule a DisconnectedPeer call at the Scheduler's current time. */ + void DisconnectedPeer(NodeId peer) + { + auto& runner = m_runner; + runner.actions.emplace_back(m_now, [=,&runner]() { + runner.txrequest.DisconnectedPeer(peer); + runner.txrequest.SanityCheck(); + }); + } + + /** Schedule a RequestedTx call at the Scheduler's current time. */ + void RequestedTx(NodeId peer, const CInv& txhash, std::chrono::microseconds exptime) + { + auto& runner = m_runner; + runner.actions.emplace_back(m_now, [=,&runner]() { + runner.txrequest.RequestedTx(peer, txhash, exptime); + runner.txrequest.SanityCheck(); + }); + } + + /** Schedule a ReceivedResponse call at the Scheduler's current time. + * If expected_completed is provided, verify that the call reports completing an announcement + * (or not) accordingly. */ + void ReceivedResponse(NodeId peer, const CInv& txhash, std::optional expected_completed = std::nullopt) + { + auto& runner = m_runner; + runner.actions.emplace_back(m_now, [=,&runner]() { + const bool completed = runner.txrequest.ReceivedResponse(peer, txhash); + if (expected_completed.has_value()) { + BOOST_CHECK_EQUAL(completed, *expected_completed); + } + runner.txrequest.SanityCheck(); + }); + } + + /** Schedule calls to verify the TxRequestTracker's state at the Scheduler's current time. + * + * @param peer The peer whose state will be inspected. + * @param expected The expected return value for GetRequestable(peer) + * @param candidates The expected return value CountCandidates(peer) + * @param inflight The expected return value CountInFlight(peer) + * @param completed The expected return value of Count(peer), minus candidates and inflight. + * @param checkname An arbitrary string to include in error messages, for test identificatrion. + * @param offset Offset with the current time to use (must be <= 0). This allows simulations of time going + * backwards (but note that the ordering of this event only follows the scenario's m_now. + */ + void Check(NodeId peer, const std::vector& expected, size_t candidates, size_t inflight, + size_t completed, const std::string& checkname, + std::chrono::microseconds offset = std::chrono::microseconds{0}) + { + const auto comment = m_testname + " " + checkname; + auto& runner = m_runner; + const auto now = m_now; + assert(offset.count() <= 0); + runner.actions.emplace_back(m_now, [=,&runner]() { + std::vector> expired_now; + auto ret = runner.txrequest.GetRequestable(peer, now + offset, &expired_now); + for (const auto& entry : expired_now) runner.expired.insert(entry); + runner.txrequest.SanityCheck(); + runner.txrequest.PostGetRequestableSanityCheck(now + offset); + size_t total = candidates + inflight + completed; + size_t real_total = runner.txrequest.Count(peer); + size_t real_candidates = runner.txrequest.CountCandidates(peer); + size_t real_inflight = runner.txrequest.CountInFlight(peer); + BOOST_CHECK_MESSAGE(real_total == total, strprintf("[" + comment + "] total %i (%i expected)", real_total, total)); + BOOST_CHECK_MESSAGE(real_inflight == inflight, strprintf("[" + comment + "] inflight %i (%i expected)", real_inflight, inflight)); + BOOST_CHECK_MESSAGE(real_candidates == candidates, strprintf("[" + comment + "] candidates %i (%i expected)", real_candidates, candidates)); + + if (ret != expected) { + std::cerr << " ret: " << v2s(ret) << std::endl; + std::cerr << "expected: " << v2s(expected) << std::endl; + } + BOOST_CHECK_MESSAGE(ret == expected, "[" + comment + "] mismatching requestables"); + }); + } + + /** Verify that an announcement for the inv by peer has expired some time before this check is scheduled. + * + * Every expected expiration should be accounted for through exactly one call to this function. + */ + void CheckExpired(NodeId peer, const CInv& gtxid) + { + const auto& testname = m_testname; + auto& runner = m_runner; + runner.actions.emplace_back(m_now, [=,&runner]() { + auto it = runner.expired.find(std::pair{peer, gtxid}); + BOOST_CHECK_MESSAGE(it != runner.expired.end(), "[" + testname + "] missing expiration"); + if (it != runner.expired.end()) runner.expired.erase(it); + }); + } + + /** Generate a random txhash, whose priorities for certain peers are constrained. + * + * For example, NewTxHash({{p1,p2,p3},{p2,p4,p5}}) will generate a txhash T such that both: + * - priority(p1,T) > priority(p2,T) > priority(p3,T) + * - priority(p2,T) > priority(p4,T) > priority(p5,T) + * where priority is the predicted internal TxRequestTracker's priority, assuming all announcements + * are within the same preferredness class. + */ + CInv NewTxHash(const std::vector>& orders = {}) + { + CInv ret; + bool ok; + do { + ret = CInv{InsecureRandBool() ? MSG_TX : MSG_QUORUM_FINAL_COMMITMENT, InsecureRand256()}; + ok = true; + for (const auto& order : orders) { + for (size_t pos = 1; pos < order.size(); ++pos) { + uint64_t prio_prev = m_runner.txrequest.ComputePriority(ret, order[pos - 1], true); + uint64_t prio_cur = m_runner.txrequest.ComputePriority(ret, order[pos], true); + if (prio_prev <= prio_cur) { + ok = false; + break; + } + } + if (!ok) break; + } + if (ok) { + ok = m_runner.txhashset.insert(ret).second; + } + } while(!ok); + return ret; + } + + /** Generate a random GenTxid; the txhash follows NewTxHash; the is_wtxid flag is random. */ + CInv NewGTxid(const std::vector>& orders = {}) + { + return NewTxHash(orders); + } + + /** Generate a new random NodeId to use as peer. The same NodeId is never returned twice + * (across all Scenarios combined). */ + NodeId NewPeer() + { + bool ok; + NodeId ret; + do { + ret = InsecureRandBits(63); + ok = m_runner.peerset.insert(ret).second; + } while(!ok); + return ret; + } + + std::chrono::microseconds Now() const { return m_now; } +}; + +/** Add to scenario a test with a single tx announced by a single peer. + * + * config is an integer in [0, 32), which controls which variant of the test is used. + */ +void BuildSingleTest(Scenario& scenario, int config) +{ + auto peer = scenario.NewPeer(); + auto gtxid = scenario.NewGTxid(); + bool immediate = config & 1; + bool preferred = config & 2; + auto delay = immediate ? NO_TIME : RandomTime8s(); + + scenario.SetTestName(strprintf("Single(config=%i)", config)); + + // Receive an announcement, either immediately requestable or delayed. + scenario.ReceivedInv(peer, gtxid, preferred, immediate ? MIN_TIME : scenario.Now() + delay); + if (immediate) { + scenario.Check(peer, {gtxid}, 1, 0, 0, "s1"); + } else { + scenario.Check(peer, {}, 1, 0, 0, "s2"); + scenario.AdvanceTime(delay - MICROSECOND); + scenario.Check(peer, {}, 1, 0, 0, "s3"); + scenario.AdvanceTime(MICROSECOND); + scenario.Check(peer, {gtxid}, 1, 0, 0, "s4"); + } + + if (config >> 3) { // We'll request the transaction + scenario.AdvanceTime(RandomTime8s()); + auto expiry = RandomTime8s(); + scenario.Check(peer, {gtxid}, 1, 0, 0, "s5"); + scenario.RequestedTx(peer, gtxid, scenario.Now() + expiry); + scenario.Check(peer, {}, 0, 1, 0, "s6"); + + if ((config >> 3) == 1) { // The request will time out + scenario.AdvanceTime(expiry - MICROSECOND); + scenario.Check(peer, {}, 0, 1, 0, "s7"); + scenario.AdvanceTime(MICROSECOND); + scenario.Check(peer, {}, 0, 0, 0, "s8"); + scenario.CheckExpired(peer, gtxid); + return; + } else { + scenario.AdvanceTime(std::chrono::microseconds{InsecureRandRange(expiry.count())}); + scenario.Check(peer, {}, 0, 1, 0, "s9"); + if ((config >> 3) == 3) { // A response will arrive for the transaction + scenario.ReceivedResponse(peer, gtxid, true); + // A duplicate response does not complete anything anymore. + scenario.ReceivedResponse(peer, gtxid, false); + scenario.Check(peer, {}, 0, 0, 0, "s10"); + return; + } + } + } + + if (config & 4) { // The peer will go offline + scenario.DisconnectedPeer(peer); + } else { // The transaction is no longer needed + scenario.ForgetTxHash(gtxid); + } + scenario.Check(peer, {}, 0, 0, 0, "s11"); +} + +/** Add to scenario a test with a single tx announced by two peers, to verify the + * right peer is selected for requests. + * + * config is an integer in [0, 32), which controls which variant of the test is used. + */ +void BuildPriorityTest(Scenario& scenario, int config) +{ + scenario.SetTestName(strprintf("Priority(config=%i)", config)); + + // Two peers. They will announce in order {peer1, peer2}. + auto peer1 = scenario.NewPeer(), peer2 = scenario.NewPeer(); + // Construct a transaction that under random rules would be preferred by peer2 or peer1, + // depending on configuration. + bool prio1 = config & 1; + auto gtxid = prio1 ? scenario.NewGTxid({{peer1, peer2}}) : scenario.NewGTxid({{peer2, peer1}}); + bool pref1 = config & 2, pref2 = config & 4; + + scenario.ReceivedInv(peer1, gtxid, pref1, MIN_TIME); + scenario.Check(peer1, {gtxid}, 1, 0, 0, "p1"); + if (InsecureRandBool()) { + scenario.AdvanceTime(RandomTime8s()); + scenario.Check(peer1, {gtxid}, 1, 0, 0, "p2"); + } + + scenario.ReceivedInv(peer2, gtxid, pref2, MIN_TIME); + bool stage2_prio = + // At this point, peer2 will be given priority if: + // - It is preferred and peer1 is not + (pref2 && !pref1) || + // - They're in the same preference class, + // and the randomized priority favors peer2 over peer1. + (pref1 == pref2 && !prio1); + NodeId priopeer = stage2_prio ? peer2 : peer1, otherpeer = stage2_prio ? peer1 : peer2; + scenario.Check(otherpeer, {}, 1, 0, 0, "p3"); + scenario.Check(priopeer, {gtxid}, 1, 0, 0, "p4"); + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.Check(otherpeer, {}, 1, 0, 0, "p5"); + scenario.Check(priopeer, {gtxid}, 1, 0, 0, "p6"); + + // We possibly request from the selected peer. + if (config & 8) { + scenario.RequestedTx(priopeer, gtxid, MAX_TIME); + scenario.Check(priopeer, {}, 0, 1, 0, "p7"); + scenario.Check(otherpeer, {}, 1, 0, 0, "p8"); + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + } + + // The peer which was selected (or requested from) now goes offline, or a NOTFOUND is received from them. + if (config & 16) { + scenario.DisconnectedPeer(priopeer); + } else { + scenario.ReceivedResponse(priopeer, gtxid, true); + } + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.Check(priopeer, {}, 0, 0, !(config & 16), "p8"); + scenario.Check(otherpeer, {gtxid}, 1, 0, 0, "p9"); + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + + // Now the other peer goes offline. + scenario.DisconnectedPeer(otherpeer); + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.Check(peer1, {}, 0, 0, 0, "p10"); + scenario.Check(peer2, {}, 0, 0, 0, "p11"); +} + +/** Add to scenario a randomized test in which N peers announce the same transaction, to verify + * the order in which they are requested. */ +void BuildBigPriorityTest(Scenario& scenario, int peers) +{ + scenario.SetTestName(strprintf("BigPriority(peers=%i)", peers)); + + // We will have N peers announce the same transaction. + std::map preferred; + std::vector pref_peers, npref_peers; + int num_pref = InsecureRandRange(peers + 1) ; // Some preferred, ... + int num_npref = peers - num_pref; // some not preferred. + for (int i = 0; i < num_pref; ++i) { + pref_peers.push_back(scenario.NewPeer()); + preferred[pref_peers.back()] = true; + } + for (int i = 0; i < num_npref; ++i) { + npref_peers.push_back(scenario.NewPeer()); + preferred[npref_peers.back()] = false; + } + // Make a list of all peers, in order of intended request order (concatenation of pref_peers and npref_peers). + std::vector request_order; + for (int i = 0; i < num_pref; ++i) request_order.push_back(pref_peers[i]); + for (int i = 0; i < num_npref; ++i) request_order.push_back(npref_peers[i]); + + // Determine the announcement order randomly. + std::vector announce_order = request_order; + Shuffle(announce_order.begin(), announce_order.end(), g_insecure_rand_ctx); + + // Find a gtxid whose txhash prioritization is consistent with the required ordering within pref_peers and + // within npref_peers. + auto gtxid = scenario.NewGTxid({pref_peers, npref_peers}); + + // Decide reqtimes in opposite order of the expected request order. This means that as time passes we expect the + // to-be-requested-from-peer will change every time a subsequent reqtime is passed. + std::map reqtimes; + auto reqtime = scenario.Now(); + for (int i = peers - 1; i >= 0; --i) { + reqtime += RandomTime8s(); + reqtimes[request_order[i]] = reqtime; + } + + // Actually announce from all peers simultaneously (but in announce_order). + for (const auto peer : announce_order) { + scenario.ReceivedInv(peer, gtxid, preferred[peer], reqtimes[peer]); + } + for (const auto peer : announce_order) { + scenario.Check(peer, {}, 1, 0, 0, "b1"); + } + + // Let time pass and observe the to-be-requested-from peer change, from nonpreferred to preferred, and from + // high priority to low priority within each class. + for (int i = peers - 1; i >= 0; --i) { + scenario.AdvanceTime(reqtimes[request_order[i]] - scenario.Now() - MICROSECOND); + scenario.Check(request_order[i], {}, 1, 0, 0, "b2"); + scenario.AdvanceTime(MICROSECOND); + scenario.Check(request_order[i], {gtxid}, 1, 0, 0, "b3"); + } + + // Peers now in random order go offline, or send NOTFOUNDs. At every point in time the new to-be-requested-from + // peer should be the best remaining one, so verify this after every response. + for (int i = 0; i < peers; ++i) { + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + const int pos = InsecureRandRange(request_order.size()); + const auto peer = request_order[pos]; + request_order.erase(request_order.begin() + pos); + if (InsecureRandBool()) { + scenario.DisconnectedPeer(peer); + scenario.Check(peer, {}, 0, 0, 0, "b4"); + } else { + scenario.ReceivedResponse(peer, gtxid, true); + scenario.Check(peer, {}, 0, 0, request_order.size() > 0, "b5"); + } + if (request_order.size()) { + scenario.Check(request_order[0], {gtxid}, 1, 0, 0, "b6"); + } + } + + // Everything is gone in the end. + for (const auto peer : announce_order) { + scenario.Check(peer, {}, 0, 0, 0, "b7"); + } +} + +/** Add to scenario a test with one peer announcing two transactions, to verify they are + * fetched in announcement order. + * + * config is an integer in [0, 4) inclusive, and selects the variant of the test. + */ +void BuildRequestOrderTest(Scenario& scenario, int config) +{ + scenario.SetTestName(strprintf("RequestOrder(config=%i)", config)); + + auto peer = scenario.NewPeer(); + auto gtxid1 = scenario.NewGTxid(); + auto gtxid2 = scenario.NewGTxid(); + + auto reqtime2 = scenario.Now() + RandomTime8s(); + auto reqtime1 = reqtime2 + RandomTime8s(); + + scenario.ReceivedInv(peer, gtxid1, config & 1, reqtime1); + // Simulate time going backwards by giving the second announcement an earlier reqtime. + scenario.ReceivedInv(peer, gtxid2, config & 2, reqtime2); + + scenario.AdvanceTime(reqtime2 - MICROSECOND - scenario.Now()); + scenario.Check(peer, {}, 2, 0, 0, "o1"); + scenario.AdvanceTime(MICROSECOND); + scenario.Check(peer, {gtxid2}, 2, 0, 0, "o2"); + scenario.AdvanceTime(reqtime1 - MICROSECOND - scenario.Now()); + scenario.Check(peer, {gtxid2}, 2, 0, 0, "o3"); + scenario.AdvanceTime(MICROSECOND); + // Even with time going backwards in between announcements, the return value of GetRequestable is in + // announcement order. + scenario.Check(peer, {gtxid1, gtxid2}, 2, 0, 0, "o4"); + + scenario.DisconnectedPeer(peer); + scenario.Check(peer, {}, 0, 0, 0, "o5"); +} + +/** Add to scenario a test that verifies behavior related to two inv types with the same hash + * being announced. Unlike upstream's txid/wtxid pair, distinct inv types are tracked as independent + * announcements, so both can be requested concurrently. + * + * config is an integer in [0, 4) inclusive, and selects the variant of the test used. +*/ +void BuildWtxidTest(Scenario& scenario, int config) +{ + scenario.SetTestName(strprintf("Wtxid(config=%i)", config)); + + auto peerT = scenario.NewPeer(); + auto peerW = scenario.NewPeer(); + auto txhash = scenario.NewTxHash().hash; + CInv txid{MSG_TX, txhash}; + CInv wtxid{MSG_QUORUM_FINAL_COMMITMENT, txhash}; + + auto reqtimeT = InsecureRandBool() ? MIN_TIME : scenario.Now() + RandomTime8s(); + auto reqtimeW = InsecureRandBool() ? MIN_TIME : scenario.Now() + RandomTime8s(); + // Announce txid first or wtxid first. + if (config & 1) { + scenario.ReceivedInv(peerT, txid, config & 2, reqtimeT); + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.ReceivedInv(peerW, wtxid, !(config & 2), reqtimeW); + } else { + scenario.ReceivedInv(peerW, wtxid, !(config & 2), reqtimeW); + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.ReceivedInv(peerT, txid, config & 2, reqtimeT); + } + + // Let time pass if needed, and check that both announcements are to-be-requested + // independently (distinct inv types are separate announcements). + auto max_reqtime = std::max(reqtimeT, reqtimeW); + if (max_reqtime > scenario.Now()) scenario.AdvanceTime(max_reqtime - scenario.Now()); + if (config & 2) { + scenario.Check(peerT, {txid}, 1, 0, 0, "w1"); + scenario.Check(peerW, {wtxid}, 1, 0, 0, "w2"); + } else { + scenario.Check(peerT, {txid}, 1, 0, 0, "w3"); + scenario.Check(peerW, {wtxid}, 1, 0, 0, "w4"); + } + + // Let the preferred announcement be requested. It's not going to be delivered. + auto expiry = RandomTime8s(); + if (config & 2) { + scenario.RequestedTx(peerT, txid, scenario.Now() + expiry); + + scenario.Check(peerT, {}, 0, 1, 0, "w5"); + scenario.Check(peerW, {wtxid}, 1, 0, 0, "w6"); + } else { + scenario.RequestedTx(peerW, wtxid, scenario.Now() + expiry); + + scenario.Check(peerT, {txid}, 1, 0, 0, "w7"); + scenario.Check(peerW, {}, 0, 1, 0, "w8"); + } + + // After reaching expiration time of the preferred announcement, verify that the + // remaining one is requestable + scenario.AdvanceTime(expiry); + if (config & 2) { + scenario.Check(peerT, {}, 0, 0, 0, "w9"); + scenario.Check(peerW, {wtxid}, 1, 0, 0, "w10"); + scenario.CheckExpired(peerT, txid); + } else { + scenario.Check(peerT, {txid}, 1, 0, 0, "w11"); + scenario.Check(peerW, {}, 0, 0, 0, "w12"); + scenario.CheckExpired(peerW, wtxid); + } + + // Once the object arrives, the caller forgets each inv type it may have been + // announced under, and all announcements are gone. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.ForgetTxHash(CInv(MSG_TX, txhash)); + scenario.ForgetTxHash(CInv(MSG_QUORUM_FINAL_COMMITMENT, txhash)); + scenario.Check(peerT, {}, 0, 0, 0, "w13"); + scenario.Check(peerW, {}, 0, 0, 0, "w14"); +} + +/** Add to scenario a test that exercises clocks that go backwards. */ +void BuildTimeBackwardsTest(Scenario& scenario) +{ + auto peer1 = scenario.NewPeer(); + auto peer2 = scenario.NewPeer(); + auto gtxid = scenario.NewGTxid({{peer1, peer2}}); + + // Announce from peer2. + auto reqtime = scenario.Now() + RandomTime8s(); + scenario.ReceivedInv(peer2, gtxid, true, reqtime); + scenario.Check(peer2, {}, 1, 0, 0, "r1"); + scenario.AdvanceTime(reqtime - scenario.Now()); + scenario.Check(peer2, {gtxid}, 1, 0, 0, "r2"); + // Check that if the clock goes backwards by 1us, the transaction would stop being requested. + scenario.Check(peer2, {}, 1, 0, 0, "r3", -MICROSECOND); + // But it reverts to being requested if time goes forward again. + scenario.Check(peer2, {gtxid}, 1, 0, 0, "r4"); + + // Announce from peer1. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.ReceivedInv(peer1, gtxid, true, MAX_TIME); + scenario.Check(peer2, {gtxid}, 1, 0, 0, "r5"); + scenario.Check(peer1, {}, 1, 0, 0, "r6"); + + // Request from peer1. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + auto expiry = scenario.Now() + RandomTime8s(); + scenario.RequestedTx(peer1, gtxid, expiry); + scenario.Check(peer1, {}, 0, 1, 0, "r7"); + scenario.Check(peer2, {}, 1, 0, 0, "r8"); + + // Expiration passes. + scenario.AdvanceTime(expiry - scenario.Now()); + scenario.Check(peer1, {}, 0, 0, 1, "r9"); + scenario.Check(peer2, {gtxid}, 1, 0, 0, "r10"); // Request goes back to peer2. + scenario.CheckExpired(peer1, gtxid); + scenario.Check(peer1, {}, 0, 0, 1, "r11", -MICROSECOND); // Going back does not unexpire. + scenario.Check(peer2, {gtxid}, 1, 0, 0, "r12", -MICROSECOND); + + // Peer2 goes offline, meaning no viable announcements remain. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.DisconnectedPeer(peer2); + scenario.Check(peer1, {}, 0, 0, 0, "r13"); + scenario.Check(peer2, {}, 0, 0, 0, "r14"); +} + +/** Add to scenario a test that involves RequestedTx() calls for txhashes not returned by GetRequestable. */ +void BuildWeirdRequestsTest(Scenario& scenario) +{ + auto peer1 = scenario.NewPeer(); + auto peer2 = scenario.NewPeer(); + auto gtxid1 = scenario.NewGTxid({{peer1, peer2}}); + auto gtxid2 = scenario.NewGTxid({{peer2, peer1}}); + + // Announce gtxid1 by peer1. + scenario.ReceivedInv(peer1, gtxid1, true, MIN_TIME); + scenario.Check(peer1, {gtxid1}, 1, 0, 0, "q1"); + + // Announce gtxid2 by peer2. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.ReceivedInv(peer2, gtxid2, true, MIN_TIME); + scenario.Check(peer1, {gtxid1}, 1, 0, 0, "q2"); + scenario.Check(peer2, {gtxid2}, 1, 0, 0, "q3"); + + // We request gtxid2 from *peer1* - no effect. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.RequestedTx(peer1, gtxid2, MAX_TIME); + scenario.Check(peer1, {gtxid1}, 1, 0, 0, "q4"); + scenario.Check(peer2, {gtxid2}, 1, 0, 0, "q5"); + + // Now request gtxid1 from peer1 - marks it as REQUESTED. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + auto expiryA = scenario.Now() + RandomTime8s(); + scenario.RequestedTx(peer1, gtxid1, expiryA); + scenario.Check(peer1, {}, 0, 1, 0, "q6"); + scenario.Check(peer2, {gtxid2}, 1, 0, 0, "q7"); + + // Request it a second time - nothing happens, as it's already REQUESTED. + auto expiryB = expiryA + RandomTime8s(); + scenario.RequestedTx(peer1, gtxid1, expiryB); + scenario.Check(peer1, {}, 0, 1, 0, "q8"); + scenario.Check(peer2, {gtxid2}, 1, 0, 0, "q9"); + + // Also announce gtxid1 from peer2 now, so that the txhash isn't forgotten when the peer1 request expires. + scenario.ReceivedInv(peer2, gtxid1, true, MIN_TIME); + scenario.Check(peer1, {}, 0, 1, 0, "q10"); + scenario.Check(peer2, {gtxid2}, 2, 0, 0, "q11"); + + // When reaching expiryA, it expires (not expiryB, which is later). + scenario.AdvanceTime(expiryA - scenario.Now()); + scenario.Check(peer1, {}, 0, 0, 1, "q12"); + scenario.Check(peer2, {gtxid2, gtxid1}, 2, 0, 0, "q13"); + scenario.CheckExpired(peer1, gtxid1); + + // Requesting it yet again from peer1 doesn't do anything, as it's already COMPLETED. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.RequestedTx(peer1, gtxid1, MAX_TIME); + scenario.Check(peer1, {}, 0, 0, 1, "q14"); + scenario.Check(peer2, {gtxid2, gtxid1}, 2, 0, 0, "q15"); + + // Now announce gtxid2 from peer1. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.ReceivedInv(peer1, gtxid2, true, MIN_TIME); + scenario.Check(peer1, {}, 1, 0, 1, "q16"); + scenario.Check(peer2, {gtxid2, gtxid1}, 2, 0, 0, "q17"); + + // And request it from peer1 (weird as peer2 has the preference). + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.RequestedTx(peer1, gtxid2, MAX_TIME); + scenario.Check(peer1, {}, 0, 1, 1, "q18"); + scenario.Check(peer2, {gtxid1}, 2, 0, 0, "q19"); + + // If peer2 now (normally) requests gtxid2, the existing request by peer1 becomes COMPLETED. + if (InsecureRandBool()) scenario.AdvanceTime(RandomTime8s()); + scenario.RequestedTx(peer2, gtxid2, MAX_TIME); + scenario.Check(peer1, {}, 0, 0, 2, "q20"); + scenario.Check(peer2, {gtxid1}, 1, 1, 0, "q21"); + + // If peer2 goes offline, no viable announcements remain. + scenario.DisconnectedPeer(peer2); + scenario.Check(peer1, {}, 0, 0, 0, "q22"); + scenario.Check(peer2, {}, 0, 0, 0, "q23"); +} + +void TestInterleavedScenarios() +{ + // Create a list of functions which add tests to scenarios. + std::vector> builders; + // Add instances of every test, for every configuration. + for (int n = 0; n < 64; ++n) { + builders.emplace_back([n](Scenario& scenario){ BuildWtxidTest(scenario, n); }); + builders.emplace_back([n](Scenario& scenario){ BuildRequestOrderTest(scenario, n & 3); }); + builders.emplace_back([n](Scenario& scenario){ BuildSingleTest(scenario, n & 31); }); + builders.emplace_back([n](Scenario& scenario){ BuildPriorityTest(scenario, n & 31); }); + builders.emplace_back([n](Scenario& scenario){ BuildBigPriorityTest(scenario, (n & 7) + 1); }); + builders.emplace_back([](Scenario& scenario){ BuildTimeBackwardsTest(scenario); }); + builders.emplace_back([](Scenario& scenario){ BuildWeirdRequestsTest(scenario); }); + } + // Randomly shuffle all those functions. + Shuffle(builders.begin(), builders.end(), g_insecure_rand_ctx); + + Runner runner; + auto starttime = RandomTime1y(); + // Construct many scenarios, and run (up to) 10 randomly-chosen tests consecutively in each. + while (builders.size()) { + // Introduce some variation in the start time of each scenario, so they don't all start off + // concurrently, but get a more random interleaving. + auto scenario_start = starttime + RandomTime8s() + RandomTime8s() + RandomTime8s(); + Scenario scenario(runner, scenario_start); + for (int j = 0; builders.size() && j < 10; ++j) { + builders.back()(scenario); + builders.pop_back(); + } + } + // Sort all the actions from all those scenarios chronologically, resulting in the actions from + // distinct scenarios to become interleaved. Use stable_sort so that actions from one scenario + // aren't reordered w.r.t. each other. + std::stable_sort(runner.actions.begin(), runner.actions.end(), [](const Action& a1, const Action& a2) { + return a1.first < a2.first; + }); + + // Run all actions from all scenarios, in order. + for (auto& action : runner.actions) { + action.second(); + } + + BOOST_CHECK_EQUAL(runner.txrequest.Size(), 0U); + BOOST_CHECK(runner.expired.empty()); +} + +} // namespace + +BOOST_AUTO_TEST_CASE(TxRequestTest) +{ + for (int i = 0; i < 5; ++i) { + TestInterleavedScenarios(); + } +} + +BOOST_AUTO_TEST_SUITE_END() diff --git a/src/txrequest.cpp b/src/txrequest.cpp new file mode 100644 index 000000000000..fa970cd531dc --- /dev/null +++ b/src/txrequest.cpp @@ -0,0 +1,747 @@ +// Copyright (c) 2020 The Bitcoin Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#include + +#include +#include +#include +#include + +#include +#include + +#include +#include +#include + +#include + +namespace { + +/** The various states a (txhash,peer) pair can be in. + * + * Note that CANDIDATE is split up into 3 substates (DELAYED, BEST, READY), allowing more efficient implementation. + * Also note that the sorting order of ByTxHashView relies on the specific order of values in this enum. + * + * Expected behaviour is: + * - When first announced by a peer, the state is CANDIDATE_DELAYED until reqtime is reached. + * - Announcements that have reached their reqtime but not been requested will be either CANDIDATE_READY or + * CANDIDATE_BEST. Neither of those has an expiration time; they remain in that state until they're requested or + * no longer needed. CANDIDATE_READY announcements are promoted to CANDIDATE_BEST when they're the best one left. + * - When requested, an announcement will be in state REQUESTED until expiry is reached. + * - If expiry is reached, or the peer replies to the request (either with NOTFOUND or the tx), the state becomes + * COMPLETED. + */ +enum class State : uint8_t { + /** A CANDIDATE announcement whose reqtime is in the future. */ + CANDIDATE_DELAYED, + /** A CANDIDATE announcement that's not CANDIDATE_DELAYED or CANDIDATE_BEST. */ + CANDIDATE_READY, + /** The best CANDIDATE for a given txhash; only if there is no REQUESTED announcement already for that txhash. + * The CANDIDATE_BEST is the highest-priority announcement among all CANDIDATE_READY (and _BEST) ones for that + * txhash. */ + CANDIDATE_BEST, + /** A REQUESTED announcement. */ + REQUESTED, + /** A COMPLETED announcement. */ + COMPLETED, +}; + +//! Type alias for sequence numbers. +using SequenceNumber = uint64_t; + +/** An announcement. This is the data we track for each inv that is announced to us by each peer. */ +struct Announcement { + /** The inv that was announced. */ + const CInv m_inv; + /** For CANDIDATE_{DELAYED,BEST,READY} the reqtime; for REQUESTED the expiry. */ + std::chrono::microseconds m_time; + /** What peer the request was from. */ + const NodeId m_peer; + /** What sequence number this announcement has. */ + const SequenceNumber m_sequence : 59; + /** Whether the request is preferred. */ + const bool m_preferred : 1; + + /** What state this announcement is in. + * This is a uint8_t instead of a State to silence a GCC warning in versions prior to 8.4 and 9.3. + * See https://gcc.gnu.org/bugzilla/show_bug.cgi?id=61414 */ + uint8_t m_state : 3 {static_cast(State::CANDIDATE_DELAYED)}; + + /** Convert m_state to a State enum. */ + State GetState() const { return static_cast(m_state); } + + /** Convert a State enum to a uint8_t and store it in m_state. */ + void SetState(State state) { m_state = static_cast(state); } + + /** Whether this announcement is selected. There can be at most 1 selected peer per txhash. */ + bool IsSelected() const + { + return GetState() == State::CANDIDATE_BEST || GetState() == State::REQUESTED; + } + + /** Whether this announcement is waiting for a certain time to pass. */ + bool IsWaiting() const + { + return GetState() == State::REQUESTED || GetState() == State::CANDIDATE_DELAYED; + } + + /** Whether this announcement can feasibly be selected if the current IsSelected() one disappears. */ + bool IsSelectable() const + { + return GetState() == State::CANDIDATE_READY || GetState() == State::CANDIDATE_BEST; + } + + /** Construct a new announcement from scratch, initially in CANDIDATE_DELAYED state. */ + Announcement(const CInv& inv, NodeId peer, bool preferred, std::chrono::microseconds reqtime, + SequenceNumber sequence) : + m_inv(inv), m_time(reqtime), m_peer(peer), m_sequence(sequence), m_preferred(preferred) {} +}; + +//! Type alias for priorities. +using Priority = uint64_t; + +/** A functor with embedded salt that computes priority of an announcement. + * + * Higher priorities are selected first. + */ +class PriorityComputer { + const uint64_t m_k0, m_k1; +public: + explicit PriorityComputer(bool deterministic) : + m_k0{deterministic ? 0 : GetRand(0xFFFFFFFFFFFFFFFF)}, + m_k1{deterministic ? 0 : GetRand(0xFFFFFFFFFFFFFFFF)} {} + + Priority operator()(const CInv& inv, NodeId peer, bool preferred) const + { + uint64_t low_bits = CSipHasher(m_k0, m_k1).Write(inv.hash.begin(), inv.hash.size()).Write(inv.type).Write(peer).Finalize() >> 1; + return low_bits | uint64_t{preferred} << 63; + } + + Priority operator()(const Announcement& ann) const + { + return operator()(ann.m_inv, ann.m_peer, ann.m_preferred); + } +}; + +// Definitions for the 3 indexes used in the main data structure. +// +// Each index has a By* type to identify it, a By*View data type to represent the view of announcement it is sorted +// by, and an By*ViewExtractor type to convert an announcement into the By*View type. +// See https://www.boost.org/doc/libs/1_58_0/libs/multi_index/doc/reference/key_extraction.html#key_extractors +// for more information about the key extraction concept. + +// The ByPeer index is sorted by (peer, state == CANDIDATE_BEST, txhash) +// +// Uses: +// * Looking up existing announcements by peer/txhash, by checking both (peer, false, txhash) and +// (peer, true, txhash). +// * Finding all CANDIDATE_BEST announcements for a given peer in GetRequestable. +struct ByPeer {}; +using ByPeerView = std::tuple; +struct ByPeerViewExtractor +{ + using result_type = ByPeerView; + result_type operator()(const Announcement& ann) const + { + return ByPeerView{ann.m_peer, ann.GetState() == State::CANDIDATE_BEST, ann.m_inv}; + } +}; + +// The ByTxHash index is sorted by (txhash, state, priority). +// +// Note: priority == 0 whenever state != CANDIDATE_READY. +// +// Uses: +// * Deleting all announcements with a given txhash in ForgetTxHash. +// * Finding the best CANDIDATE_READY to convert to CANDIDATE_BEST, when no other CANDIDATE_READY or REQUESTED +// announcement exists for that txhash. +// * Determining when no more non-COMPLETED announcements for a given txhash exist, so the COMPLETED ones can be +// deleted. +struct ByTxHash {}; +using ByTxHashView = std::tuple; +class ByTxHashViewExtractor { + const PriorityComputer& m_computer; +public: + ByTxHashViewExtractor(const PriorityComputer& computer) : m_computer(computer) {} + using result_type = ByTxHashView; + result_type operator()(const Announcement& ann) const + { + const Priority prio = (ann.GetState() == State::CANDIDATE_READY) ? m_computer(ann) : 0; + return ByTxHashView{ann.m_inv, ann.GetState(), prio}; + } +}; + +enum class WaitState { + //! Used for announcements that need efficient testing of "is their timestamp in the future?". + FUTURE_EVENT, + //! Used for announcements whose timestamp is not relevant. + NO_EVENT, + //! Used for announcements that need efficient testing of "is their timestamp in the past?". + PAST_EVENT, +}; + +WaitState GetWaitState(const Announcement& ann) +{ + if (ann.IsWaiting()) return WaitState::FUTURE_EVENT; + if (ann.IsSelectable()) return WaitState::PAST_EVENT; + return WaitState::NO_EVENT; +} + +// The ByTime index is sorted by (wait_state, time). +// +// All announcements with a timestamp in the future can be found by iterating the index forward from the beginning. +// All announcements with a timestamp in the past can be found by iterating the index backwards from the end. +// +// Uses: +// * Finding CANDIDATE_DELAYED announcements whose reqtime has passed, and REQUESTED announcements whose expiry has +// passed. +// * Finding CANDIDATE_READY/BEST announcements whose reqtime is in the future (when the clock time went backwards). +struct ByTime {}; +using ByTimeView = std::pair; +struct ByTimeViewExtractor +{ + using result_type = ByTimeView; + result_type operator()(const Announcement& ann) const + { + return ByTimeView{GetWaitState(ann), ann.m_time}; + } +}; + +/** Data type for the main data structure (Announcement objects with ByPeer/ByTxHash/ByTime indexes). */ +using Index = boost::multi_index_container< + Announcement, + boost::multi_index::indexed_by< + boost::multi_index::ordered_unique, ByPeerViewExtractor>, + boost::multi_index::ordered_non_unique, ByTxHashViewExtractor>, + boost::multi_index::ordered_non_unique, ByTimeViewExtractor> + > +>; + +/** Helper type to simplify syntax of iterator types. */ +template +using Iter = typename Index::index::type::iterator; + +/** Per-peer statistics object. */ +struct PeerInfo { + size_t m_total = 0; //!< Total number of announcements for this peer. + size_t m_completed = 0; //!< Number of COMPLETED announcements for this peer. + size_t m_requested = 0; //!< Number of REQUESTED announcements for this peer. +}; + +/** Per-txhash statistics object. Only used for sanity checking. */ +struct TxHashInfo +{ + //! Number of CANDIDATE_DELAYED announcements for this txhash. + size_t m_candidate_delayed = 0; + //! Number of CANDIDATE_READY announcements for this txhash. + size_t m_candidate_ready = 0; + //! Number of CANDIDATE_BEST announcements for this txhash (at most one). + size_t m_candidate_best = 0; + //! Number of REQUESTED announcements for this txhash (at most one; mutually exclusive with CANDIDATE_BEST). + size_t m_requested = 0; + //! The priority of the CANDIDATE_BEST announcement if one exists, or max() otherwise. + Priority m_priority_candidate_best = std::numeric_limits::max(); + //! The highest priority of all CANDIDATE_READY announcements (or min() if none exist). + Priority m_priority_best_candidate_ready = std::numeric_limits::min(); + //! All peers we have an announcement for this txhash for. + std::vector m_peers; +}; + +/** Compare two PeerInfo objects. Only used for sanity checking. */ +bool operator==(const PeerInfo& a, const PeerInfo& b) +{ + return std::tie(a.m_total, a.m_completed, a.m_requested) == + std::tie(b.m_total, b.m_completed, b.m_requested); +}; + +/** (Re)compute the PeerInfo map from the index. Only used for sanity checking. */ +std::unordered_map RecomputePeerInfo(const Index& index) +{ + std::unordered_map ret; + for (const Announcement& ann : index) { + PeerInfo& info = ret[ann.m_peer]; + ++info.m_total; + info.m_requested += (ann.GetState() == State::REQUESTED); + info.m_completed += (ann.GetState() == State::COMPLETED); + } + return ret; +} + +/** Compute the TxHashInfo map. Only used for sanity checking. */ +std::map ComputeTxHashInfo(const Index& index, const PriorityComputer& computer) +{ + std::map ret; + for (const Announcement& ann : index) { + TxHashInfo& info = ret[ann.m_inv]; + // Classify how many announcements of each state we have for this txhash. + info.m_candidate_delayed += (ann.GetState() == State::CANDIDATE_DELAYED); + info.m_candidate_ready += (ann.GetState() == State::CANDIDATE_READY); + info.m_candidate_best += (ann.GetState() == State::CANDIDATE_BEST); + info.m_requested += (ann.GetState() == State::REQUESTED); + // And track the priority of the best CANDIDATE_READY/CANDIDATE_BEST announcements. + if (ann.GetState() == State::CANDIDATE_BEST) { + info.m_priority_candidate_best = computer(ann); + } + if (ann.GetState() == State::CANDIDATE_READY) { + info.m_priority_best_candidate_ready = std::max(info.m_priority_best_candidate_ready, computer(ann)); + } + // Also keep track of which peers this txhash has an announcement for (so we can detect duplicates). + info.m_peers.push_back(ann.m_peer); + } + return ret; +} + +} // namespace + +/** Actual implementation for TxRequestTracker's data structure. */ +class TxRequestTracker::Impl { + //! The current sequence number. Increases for every announcement. This is used to sort txhashes returned by + //! GetRequestable in announcement order. + SequenceNumber m_current_sequence{0}; + + //! This tracker's priority computer. + const PriorityComputer m_computer; + + //! This tracker's main data structure. See SanityCheck() for the invariants that apply to it. + Index m_index; + + //! Map with this tracker's per-peer statistics. + std::unordered_map m_peerinfo; + +public: + void SanityCheck() const + { + // Recompute m_peerdata from m_index. This verifies the data in it as it should just be caching statistics + // on m_index. It also verifies the invariant that no PeerInfo announcements with m_total==0 exist. + assert(m_peerinfo == RecomputePeerInfo(m_index)); + + // Calculate per-txhash statistics from m_index, and validate invariants. + for (auto& item : ComputeTxHashInfo(m_index, m_computer)) { + TxHashInfo& info = item.second; + + // Cannot have only COMPLETED peer (txhash should have been forgotten already) + assert(info.m_candidate_delayed + info.m_candidate_ready + info.m_candidate_best + info.m_requested > 0); + + // Can have at most 1 CANDIDATE_BEST/REQUESTED peer + assert(info.m_candidate_best + info.m_requested <= 1); + + // If there are any CANDIDATE_READY announcements, there must be exactly one CANDIDATE_BEST or REQUESTED + // announcement. + if (info.m_candidate_ready > 0) { + assert(info.m_candidate_best + info.m_requested == 1); + } + + // If there is both a CANDIDATE_READY and a CANDIDATE_BEST announcement, the CANDIDATE_BEST one must be + // at least as good (equal or higher priority) as the best CANDIDATE_READY. + if (info.m_candidate_ready && info.m_candidate_best) { + assert(info.m_priority_candidate_best >= info.m_priority_best_candidate_ready); + } + + // No txhash can have been announced by the same peer twice. + std::sort(info.m_peers.begin(), info.m_peers.end()); + assert(std::adjacent_find(info.m_peers.begin(), info.m_peers.end()) == info.m_peers.end()); + } + } + + void PostGetRequestableSanityCheck(std::chrono::microseconds now) const + { + for (const Announcement& ann : m_index) { + if (ann.IsWaiting()) { + // REQUESTED and CANDIDATE_DELAYED must have a time in the future (they should have been converted + // to COMPLETED/CANDIDATE_READY respectively). + assert(ann.m_time > now); + } else if (ann.IsSelectable()) { + // CANDIDATE_READY and CANDIDATE_BEST cannot have a time in the future (they should have remained + // CANDIDATE_DELAYED, or should have been converted back to it if time went backwards). + assert(ann.m_time <= now); + } + } + } + +private: + //! Wrapper around Index::...::erase that keeps m_peerinfo up to date. + template + Iter Erase(Iter it) + { + auto peerit = m_peerinfo.find(it->m_peer); + peerit->second.m_completed -= it->GetState() == State::COMPLETED; + peerit->second.m_requested -= it->GetState() == State::REQUESTED; + if (--peerit->second.m_total == 0) m_peerinfo.erase(peerit); + return m_index.get().erase(it); + } + + //! Wrapper around Index::...::modify that keeps m_peerinfo up to date. + template + void Modify(Iter it, Modifier modifier) + { + auto peerit = m_peerinfo.find(it->m_peer); + peerit->second.m_completed -= it->GetState() == State::COMPLETED; + peerit->second.m_requested -= it->GetState() == State::REQUESTED; + m_index.get().modify(it, std::move(modifier)); + peerit->second.m_completed += it->GetState() == State::COMPLETED; + peerit->second.m_requested += it->GetState() == State::REQUESTED; + } + + //! Convert a CANDIDATE_DELAYED announcement into a CANDIDATE_READY. If this makes it the new best + //! CANDIDATE_READY (and no REQUESTED exists) and better than the CANDIDATE_BEST (if any), it becomes the new + //! CANDIDATE_BEST. + void PromoteCandidateReady(Iter it) + { + assert(it != m_index.get().end()); + assert(it->GetState() == State::CANDIDATE_DELAYED); + // Convert CANDIDATE_DELAYED to CANDIDATE_READY first. + Modify(it, [](Announcement& ann){ ann.SetState(State::CANDIDATE_READY); }); + // The following code relies on the fact that the ByTxHash is sorted by txhash, and then by state (first + // _DELAYED, then _READY, then _BEST/REQUESTED). Within the _READY announcements, the best one (highest + // priority) comes last. Thus, if an existing _BEST exists for the same txhash that this announcement may + // be preferred over, it must immediately follow the newly created _READY. + auto it_next = std::next(it); + if (it_next == m_index.get().end() || it_next->m_inv != it->m_inv || + it_next->GetState() == State::COMPLETED) { + // This is the new best CANDIDATE_READY, and there is no IsSelected() announcement for this txhash + // already. + Modify(it, [](Announcement& ann){ ann.SetState(State::CANDIDATE_BEST); }); + } else if (it_next->GetState() == State::CANDIDATE_BEST) { + Priority priority_old = m_computer(*it_next); + Priority priority_new = m_computer(*it); + if (priority_new > priority_old) { + // There is a CANDIDATE_BEST announcement already, but this one is better. + Modify(it_next, [](Announcement& ann){ ann.SetState(State::CANDIDATE_READY); }); + Modify(it, [](Announcement& ann){ ann.SetState(State::CANDIDATE_BEST); }); + } + } + } + + //! Change the state of an announcement to something non-IsSelected(). If it was IsSelected(), the next best + //! announcement will be marked CANDIDATE_BEST. + void ChangeAndReselect(Iter it, State new_state) + { + assert(new_state == State::COMPLETED || new_state == State::CANDIDATE_DELAYED); + assert(it != m_index.get().end()); + if (it->IsSelected() && it != m_index.get().begin()) { + auto it_prev = std::prev(it); + // The next best CANDIDATE_READY, if any, immediately precedes the REQUESTED or CANDIDATE_BEST + // announcement in the ByTxHash index. + if (it_prev->m_inv == it->m_inv && it_prev->GetState() == State::CANDIDATE_READY) { + // If one such CANDIDATE_READY exists (for this txhash), convert it to CANDIDATE_BEST. + Modify(it_prev, [](Announcement& ann){ ann.SetState(State::CANDIDATE_BEST); }); + } + } + Modify(it, [new_state](Announcement& ann){ ann.SetState(new_state); }); + } + + //! Check if 'it' is the only announcement for a given txhash that isn't COMPLETED. + bool IsOnlyNonCompleted(Iter it) + { + assert(it != m_index.get().end()); + assert(it->GetState() != State::COMPLETED); // Not allowed to call this on COMPLETED announcements. + + // This announcement has a predecessor that belongs to the same txhash. Due to ordering, and the + // fact that 'it' is not COMPLETED, its predecessor cannot be COMPLETED here. + if (it != m_index.get().begin() && std::prev(it)->m_inv == it->m_inv) return false; + + // This announcement has a successor that belongs to the same txhash, and is not COMPLETED. + if (std::next(it) != m_index.get().end() && std::next(it)->m_inv == it->m_inv && + std::next(it)->GetState() != State::COMPLETED) return false; + + return true; + } + + /** Convert any announcement to a COMPLETED one. If there are no non-COMPLETED announcements left for this + * txhash, they are deleted. If this was a REQUESTED announcement, and there are other CANDIDATEs left, the + * best one is made CANDIDATE_BEST. Returns whether the announcement still exists. */ + bool MakeCompleted(Iter it) + { + assert(it != m_index.get().end()); + + // Nothing to be done if it's already COMPLETED. + if (it->GetState() == State::COMPLETED) return true; + + if (IsOnlyNonCompleted(it)) { + // This is the last non-COMPLETED announcement for this txhash. Delete all. + CInv txhash = it->m_inv; + do { + it = Erase(it); + } while (it != m_index.get().end() && it->m_inv == txhash); + return false; + } + + // Mark the announcement COMPLETED, and select the next best announcement (the first CANDIDATE_READY) if + // needed. + ChangeAndReselect(it, State::COMPLETED); + + return true; + } + + //! Make the data structure consistent with a given point in time: + //! - REQUESTED annoucements with expiry <= now are turned into COMPLETED. + //! - CANDIDATE_DELAYED announcements with reqtime <= now are turned into CANDIDATE_{READY,BEST}. + //! - CANDIDATE_{READY,BEST} announcements with reqtime > now are turned into CANDIDATE_DELAYED. + void SetTimePoint(std::chrono::microseconds now, std::vector>* expired) + { + if (expired) expired->clear(); + + // Iterate over all CANDIDATE_DELAYED and REQUESTED from old to new, as long as they're in the past, + // and convert them to CANDIDATE_READY and COMPLETED respectively. + while (!m_index.empty()) { + auto it = m_index.get().begin(); + if (it->GetState() == State::CANDIDATE_DELAYED && it->m_time <= now) { + PromoteCandidateReady(m_index.project(it)); + } else if (it->GetState() == State::REQUESTED && it->m_time <= now) { + if (expired) expired->emplace_back(it->m_peer, it->m_inv); + MakeCompleted(m_index.project(it)); + } else { + break; + } + } + + while (!m_index.empty()) { + // If time went backwards, we may need to demote CANDIDATE_BEST and CANDIDATE_READY announcements back + // to CANDIDATE_DELAYED. This is an unusual edge case, and unlikely to matter in production. However, + // it makes it much easier to specify and test TxRequestTracker::Impl's behaviour. + auto it = std::prev(m_index.get().end()); + if (it->IsSelectable() && it->m_time > now) { + ChangeAndReselect(m_index.project(it), State::CANDIDATE_DELAYED); + } else { + break; + } + } + } + +public: + Impl(bool deterministic) : + m_computer(deterministic), + // Explicitly initialize m_index as we need to pass a reference to m_computer to ByTxHashViewExtractor. + m_index(boost::make_tuple( + boost::make_tuple(ByPeerViewExtractor(), std::less()), + boost::make_tuple(ByTxHashViewExtractor(m_computer), std::less()), + boost::make_tuple(ByTimeViewExtractor(), std::less()) + )) {} + + // Disable copying and assigning (a default copy won't work due the stateful ByTxHashViewExtractor). + Impl(const Impl&) = delete; + Impl& operator=(const Impl&) = delete; + + void DisconnectedPeer(NodeId peer) + { + auto& index = m_index.get(); + auto it = index.lower_bound(ByPeerView{peer, false, CInv{}}); + while (it != index.end() && it->m_peer == peer) { + // Check what to continue with after this iteration. 'it' will be deleted in what follows, so we need to + // decide what to continue with afterwards. There are a number of cases to consider: + // - std::next(it) is end() or belongs to a different peer. In that case, this is the last iteration + // of the loop (denote this by setting it_next to end()). + // - 'it' is not the only non-COMPLETED announcement for its txhash. This means it will be deleted, but + // no other Announcement objects will be modified. Continue with std::next(it) if it belongs to the + // same peer, but decide this ahead of time (as 'it' may change position in what follows). + // - 'it' is the only non-COMPLETED announcement for its txhash. This means it will be deleted along + // with all other announcements for the same txhash - which may include std::next(it). However, other + // than 'it', no announcements for the same peer can be affected (due to (peer, txhash) uniqueness). + // In other words, the situation where std::next(it) is deleted can only occur if std::next(it) + // belongs to a different peer but the same txhash as 'it'. This is covered by the first bulletpoint + // already, and we'll have set it_next to end(). + auto it_next = (std::next(it) == index.end() || std::next(it)->m_peer != peer) ? index.end() : + std::next(it); + // If the announcement isn't already COMPLETED, first make it COMPLETED (which will mark other + // CANDIDATEs as CANDIDATE_BEST, or delete all of a txhash's announcements if no non-COMPLETED ones are + // left). + if (MakeCompleted(m_index.project(it))) { + // Then actually delete the announcement (unless it was already deleted by MakeCompleted). + Erase(it); + } + it = it_next; + } + } + + void ForgetTxHash(const CInv& txhash) + { + auto it = m_index.get().lower_bound(ByTxHashView{txhash, State::CANDIDATE_DELAYED, 0}); + while (it != m_index.get().end() && it->m_inv == txhash) { + it = Erase(it); + } + } + + void ReceivedInv(NodeId peer, const CInv& gtxid, bool preferred, + std::chrono::microseconds reqtime) + { + // Bail out if we already have a CANDIDATE_BEST announcement for this (txhash, peer) combination. The case + // where there is a non-CANDIDATE_BEST announcement already will be caught by the uniqueness property of the + // ByPeer index when we try to emplace the new object below. + if (m_index.get().count(ByPeerView{peer, true, gtxid})) return; + + // Try creating the announcement with CANDIDATE_DELAYED state (which will fail due to the uniqueness + // of the ByPeer index if a non-CANDIDATE_BEST announcement already exists with the same txhash and peer). + // Bail out in that case. + auto ret = m_index.get().emplace(gtxid, peer, preferred, reqtime, m_current_sequence); + if (!ret.second) return; + + // Update accounting metadata. + ++m_peerinfo[peer].m_total; + ++m_current_sequence; + } + + //! Find the invs to request now from peer. + std::vector GetRequestable(NodeId peer, std::chrono::microseconds now, + std::vector>* expired) + { + // Move time. + SetTimePoint(now, expired); + + // Find all CANDIDATE_BEST announcements for this peer. + std::vector selected; + auto it_peer = m_index.get().lower_bound(ByPeerView{peer, true, CInv{}}); + while (it_peer != m_index.get().end() && it_peer->m_peer == peer && + it_peer->GetState() == State::CANDIDATE_BEST) { + selected.emplace_back(&*it_peer); + ++it_peer; + } + // Sort by sequence number. + std::sort(selected.begin(), selected.end(), [](const Announcement* a, const Announcement* b) { + return a->m_sequence < b->m_sequence; + }); + + // Return the announced invs. + std::vector ret; + ret.reserve(selected.size()); + std::transform(selected.begin(), selected.end(), std::back_inserter(ret), [](const Announcement* ann) { + return ann->m_inv; + }); + return ret; + } + + void RequestedTx(NodeId peer, const CInv& txhash, std::chrono::microseconds expiry) + { + auto it = m_index.get().find(ByPeerView{peer, true, txhash}); + if (it == m_index.get().end()) { + // There is no CANDIDATE_BEST announcement, look for a _READY or _DELAYED instead. If the caller only + // ever invokes RequestedTx with the values returned by GetRequestable, and no other non-const functions + // other than ForgetTxHash and GetRequestable in between, this branch will never execute (as txhashes + // returned by GetRequestable always correspond to CANDIDATE_BEST announcements). + + it = m_index.get().find(ByPeerView{peer, false, txhash}); + if (it == m_index.get().end() || (it->GetState() != State::CANDIDATE_DELAYED && + it->GetState() != State::CANDIDATE_READY)) { + // There is no CANDIDATE announcement tracked for this peer, so we have nothing to do. Either this + // txhash wasn't tracked at all (and the caller should have called ReceivedInv), or it was already + // requested and/or completed for other reasons and this is just a superfluous RequestedTx call. + return; + } + + // Look for an existing CANDIDATE_BEST or REQUESTED with the same txhash. We only need to do this if the + // found announcement had a different state than CANDIDATE_BEST. If it did, invariants guarantee that no + // other CANDIDATE_BEST or REQUESTED can exist. + auto it_old = m_index.get().lower_bound(ByTxHashView{txhash, State::CANDIDATE_BEST, 0}); + if (it_old != m_index.get().end() && it_old->m_inv == txhash) { + if (it_old->GetState() == State::CANDIDATE_BEST) { + // The data structure's invariants require that there can be at most one CANDIDATE_BEST or one + // REQUESTED announcement per txhash (but not both simultaneously), so we have to convert any + // existing CANDIDATE_BEST to another CANDIDATE_* when constructing another REQUESTED. + // It doesn't matter whether we pick CANDIDATE_READY or _DELAYED here, as SetTimePoint() + // will correct it at GetRequestable() time. If time only goes forward, it will always be + // _READY, so pick that to avoid extra work in SetTimePoint(). + Modify(it_old, [](Announcement& ann) { ann.SetState(State::CANDIDATE_READY); }); + } else if (it_old->GetState() == State::REQUESTED) { + // As we're no longer waiting for a response to the previous REQUESTED announcement, convert it + // to COMPLETED. This also helps guaranteeing progress. + Modify(it_old, [](Announcement& ann) { ann.SetState(State::COMPLETED); }); + } + } + } + + Modify(it, [expiry](Announcement& ann) { + ann.SetState(State::REQUESTED); + ann.m_time = expiry; + }); + } + + bool ReceivedResponse(NodeId peer, const CInv& txhash) + { + // We need to search the ByPeer index for both (peer, false, txhash) and (peer, true, txhash). + auto it = m_index.get().find(ByPeerView{peer, false, txhash}); + if (it == m_index.get().end()) { + it = m_index.get().find(ByPeerView{peer, true, txhash}); + } + if (it == m_index.get().end() || it->GetState() == State::COMPLETED) return false; + MakeCompleted(m_index.project(it)); + return true; + } + + size_t CountInFlight(NodeId peer) const + { + auto it = m_peerinfo.find(peer); + if (it != m_peerinfo.end()) return it->second.m_requested; + return 0; + } + + size_t CountCandidates(NodeId peer) const + { + auto it = m_peerinfo.find(peer); + if (it != m_peerinfo.end()) return it->second.m_total - it->second.m_requested - it->second.m_completed; + return 0; + } + + size_t Count(NodeId peer) const + { + auto it = m_peerinfo.find(peer); + if (it != m_peerinfo.end()) return it->second.m_total; + return 0; + } + + //! Count how many announcements are being tracked in total across all peers and transactions. + size_t Size() const { return m_index.size(); } + + uint64_t ComputePriority(const CInv& txhash, NodeId peer, bool preferred) const + { + // Return Priority as a uint64_t as Priority is internal. + return uint64_t{m_computer(txhash, peer, preferred)}; + } + +}; + +TxRequestTracker::TxRequestTracker(bool deterministic) : + m_impl{std::make_unique(deterministic)} {} + +TxRequestTracker::~TxRequestTracker() = default; + +void TxRequestTracker::ForgetTxHash(const CInv& txhash) { m_impl->ForgetTxHash(txhash); } +void TxRequestTracker::DisconnectedPeer(NodeId peer) { m_impl->DisconnectedPeer(peer); } +size_t TxRequestTracker::CountInFlight(NodeId peer) const { return m_impl->CountInFlight(peer); } +size_t TxRequestTracker::CountCandidates(NodeId peer) const { return m_impl->CountCandidates(peer); } +size_t TxRequestTracker::Count(NodeId peer) const { return m_impl->Count(peer); } +size_t TxRequestTracker::Size() const { return m_impl->Size(); } +void TxRequestTracker::SanityCheck() const { m_impl->SanityCheck(); } + +void TxRequestTracker::PostGetRequestableSanityCheck(std::chrono::microseconds now) const +{ + m_impl->PostGetRequestableSanityCheck(now); +} + +void TxRequestTracker::ReceivedInv(NodeId peer, const CInv& gtxid, bool preferred, + std::chrono::microseconds reqtime) +{ + m_impl->ReceivedInv(peer, gtxid, preferred, reqtime); +} + +void TxRequestTracker::RequestedTx(NodeId peer, const CInv& txhash, std::chrono::microseconds expiry) +{ + m_impl->RequestedTx(peer, txhash, expiry); +} + +bool TxRequestTracker::ReceivedResponse(NodeId peer, const CInv& txhash) +{ + return m_impl->ReceivedResponse(peer, txhash); +} + +std::vector TxRequestTracker::GetRequestable(NodeId peer, std::chrono::microseconds now, + std::vector>* expired) +{ + return m_impl->GetRequestable(peer, now, expired); +} + +uint64_t TxRequestTracker::ComputePriority(const CInv& txhash, NodeId peer, bool preferred) const +{ + return m_impl->ComputePriority(txhash, peer, preferred); +} diff --git a/src/txrequest.h b/src/txrequest.h new file mode 100644 index 000000000000..5b4709be86d5 --- /dev/null +++ b/src/txrequest.h @@ -0,0 +1,219 @@ +// Copyright (c) 2020 The Bitcoin Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#ifndef BITCOIN_TXREQUEST_H +#define BITCOIN_TXREQUEST_H + +#include // For NodeId +#include // For CInv + +#include +#include + +#include + +/** Data structure to keep track of, and schedule, transaction and other object downloads from peers. + * + * Unlike upstream Bitcoin, Dash requests many object types via inv/getdata (transactions, governance + * objects and votes, InstantSend locks, ChainLocks, sporks, quorum messages, ...), so announcements are + * tracked per inv (type and hash) rather than per txid/wtxid. Where the specification below says + * "transaction", read "object announced through an inv". + * + * === Specification === + * + * We keep track of which peers have announced which transactions, and use that to determine which requests + * should go to which peer, when, and in what order. + * + * The following information is tracked per peer/inv combination ("announcement"): + * - Which peer announced it (through their NodeId) + * - The inv it was announced through (the object's type and hash; called "txhash" in what follows) + * - What the earliest permitted time is that that transaction can be requested from that peer (called "reqtime"). + * - Whether it's from a "preferred" peer or not. Which announcements get this flag is determined by the caller, but + * this is designed for outbound peers, or other peers that we have a higher level of trust in. Even when the + * peers' preferredness changes, the preferred flag of existing announcements from that peer won't change. + * - Whether or not the transaction was requested already, and if so, when it times out (called "expiry"). + * - Whether or not the transaction request failed already (timed out, or invalid transaction or NOTFOUND was + * received). + * + * Transaction requests are then assigned to peers, following these rules: + * + * - No transaction is requested as long as another request for the same txhash is outstanding (it needs to fail + * first by passing expiry, or a NOTFOUND or invalid transaction has to be received for it). + * + * Rationale: to avoid wasting bandwidth on multiple copies of the same transaction. Note that this only works + * per inv, so if the same object is announced under two different inv types (e.g. MSG_TX and + * MSG_DSTX for the same txid), we have no means to prevent fetching both (the caller can however + * mitigate this by announcing only one type, or by delaying one, see further). + * + * - The same transaction is never requested twice from the same peer, unless the announcement was forgotten in + * between, and re-announced. Announcements are forgotten only: + * - If a peer goes offline, all its announcements are forgotten. + * - If a transaction has been successfully received, or is otherwise no longer needed, the caller can call + * ForgetTxHash, which removes all announcements across all peers with the specified txhash. + * - If for a given txhash only already-failed announcements remain, they are all forgotten. + * + * Rationale: giving a peer multiple chances to announce a transaction would allow them to bias requests in their + * favor, worsening transaction censoring attacks. The flip side is that as long as an attacker manages + * to prevent us from receiving a transaction, failed announcements (including those from honest peers) + * will linger longer, increasing memory usage somewhat. The impact of this is limited by imposing a + * cap on the number of tracked announcements per peer. As failed requests in response to announcements + * from honest peers should be rare, this almost solely hinders attackers. + * Transaction censoring attacks can be done by announcing transactions quickly while not answering + * requests for them. See https://allquantor.at/blockchainbib/pdf/miller2015topology.pdf for more + * information. + * + * - Transactions are not requested from a peer until its reqtime has passed. + * + * Rationale: enable the calling code to define a delay for less-than-ideal peers, so that (presumed) better + * peers have a chance to give their announcement first. + * + * - If multiple viable candidate peers exist according to the above rules, pick a peer as follows: + * + * - If any preferred peers are available, non-preferred peers are not considered for what follows. + * + * Rationale: preferred peers are more trusted by us, so are less likely to be under attacker control. + * + * - Pick a uniformly random peer among the candidates. + * + * Rationale: random assignments are hard to influence for attackers. + * + * Together these rules strike a balance between being fast in non-adverserial conditions and minimizing + * susceptibility to censorship attacks. An attacker that races the network: + * - Will be unsuccessful if all preferred connections are honest (and there is at least one preferred connection). + * - If there are P preferred connections of which Ph>=1 are honest, the attacker can delay us from learning + * about a transaction by k expiration periods, where k ~ 1 + NHG(N=P-1,K=P-Ph-1,r=1), which has mean + * P/(Ph+1) (where NHG stands for Negative Hypergeometric distribution). The "1 +" is due to the fact that the + * attacker can be the first to announce through a preferred connection in this scenario, which very likely means + * they get the first request. + * - If all P preferred connections are to the attacker, and there are NP non-preferred connections of which NPh>=1 + * are honest, where we assume that the attacker can disconnect and reconnect those connections, the distribution + * becomes k ~ P + NB(p=1-NPh/NP,r=1) (where NB stands for Negative Binomial distribution), which has mean + * P-1+NP/NPh. + * + * Complexity: + * - Memory usage is proportional to the total number of tracked announcements (Size()) plus the number of + * peers with a nonzero number of tracked announcements. + * - CPU usage is generally logarithmic in the total number of tracked announcements, plus the number of + * announcements affected by an operation (amortized O(1) per announcement). + */ +class TxRequestTracker { + // Avoid littering this header file with implementation details. + class Impl; + const std::unique_ptr m_impl; + +public: + //! Construct a TxRequestTracker. + explicit TxRequestTracker(bool deterministic = false); + ~TxRequestTracker(); + + // Conceptually, the data structure consists of a collection of "announcements", one for each peer/txhash + // combination: + // + // - CANDIDATE announcements represent transactions that were announced by a peer, and that become available for + // download after their reqtime has passed. + // + // - REQUESTED announcements represent transactions that have been requested, and which we're awaiting a + // response for from that peer. Their expiry value determines when the request times out. + // + // - COMPLETED announcements represent transactions that have been requested from a peer, and a NOTFOUND or a + // transaction was received in response (valid or not), or they timed out. They're only kept around to + // prevent requesting them again. If only COMPLETED announcements for a given txhash remain (so no CANDIDATE + // or REQUESTED ones), all of them are deleted (this is an invariant, and maintained by all operations below). + // + // The operations below manipulate the data structure. + + /** Adds a new CANDIDATE announcement. + * + * Does nothing if one already exists for that (inv, peer) combination (whether it's CANDIDATE, REQUESTED, or + * COMPLETED). Note that the inv type participates in uniqueness, so announcements for the same hash under + * two different types (e.g. MSG_TX and MSG_DSTX) from the same peer are tracked separately. The new + * announcement is given the specified preferred and reqtime values. + */ + void ReceivedInv(NodeId peer, const CInv& gtxid, bool preferred, + std::chrono::microseconds reqtime); + + /** Deletes all announcements for a given peer. + * + * It should be called when a peer goes offline. + */ + void DisconnectedPeer(NodeId peer); + + /** Deletes all announcements for a given inv, across all peers. + * + * This should be called when an object is no longer needed. Announcements for the same hash under a + * different inv type are separate and have to be forgotten separately. The caller should ensure that new + * announcements for the same inv will not trigger new ReceivedInv calls, at least in the short term after + * this call. + */ + void ForgetTxHash(const CInv& txhash); + + /** Find the invs to request now from peer. + * + * It does the following: + * - Convert all REQUESTED announcements (for all txhashes/peers) with (expiry <= now) to COMPLETED ones. + * These are returned in expired, if non-nullptr. + * - Requestable announcements are selected: CANDIDATE announcements from the specified peer with + * (reqtime <= now) for which no existing REQUESTED announcement with the same txhash from a different peer + * exists, and for which the specified peer is the best choice among all (reqtime <= now) CANDIDATE + * announcements with the same txhash (subject to preferredness rules, and tiebreaking using a deterministic + * salted hash of peer and txhash). + * - The announced invs of the selected announcements are returned in + * announcement order (even if multiple were added at the same time, or when the clock went backwards while + * they were being added). This is done to minimize disruption from dependent transactions being requested + * out of order: if multiple dependent transactions are announced simultaneously by one peer, and end up + * being requested from them, the requests will happen in announcement order. + */ + std::vector GetRequestable(NodeId peer, std::chrono::microseconds now, + std::vector>* expired = nullptr); + + /** Marks a transaction as requested, with a specified expiry. + * + * If no CANDIDATE announcement for the provided peer and txhash exists, this call has no effect. Otherwise: + * - That announcement is converted to REQUESTED. + * - If any other REQUESTED announcement for the same txhash already existed, it means an unexpected request + * was made (GetRequestable will never advise doing so). In this case it is converted to COMPLETED, as we're + * no longer waiting for a response to it. + */ + void RequestedTx(NodeId peer, const CInv& txhash, std::chrono::microseconds expiry); + + /** Converts a CANDIDATE or REQUESTED announcement to a COMPLETED one. If no such announcement exists for the + * provided peer and txhash, nothing happens and false is returned. + * + * Returns whether an announcement was actually completed (a second call for the same peer and inv, without a + * re-announcement in between, returns false). This lets callers use it as an atomic did-we-ask-this-peer + * check when an object arrives. + * + * It should be called whenever a transaction or NOTFOUND was received from a peer. When the transaction is + * not needed entirely anymore, ForgetTxhash should be called instead of, or in addition to, this call. + */ + bool ReceivedResponse(NodeId peer, const CInv& txhash); + + // The operations below inspect the data structure. + + /** Count how many REQUESTED announcements a peer has. */ + size_t CountInFlight(NodeId peer) const; + + /** Count how many CANDIDATE announcements a peer has. */ + size_t CountCandidates(NodeId peer) const; + + /** Count how many announcements a peer has (REQUESTED, CANDIDATE, and COMPLETED combined). */ + size_t Count(NodeId peer) const; + + /** Count how many announcements are being tracked in total across all peers and transaction hashes. */ + size_t Size() const; + + /** Access to the internal priority computation (testing only) */ + uint64_t ComputePriority(const CInv& txhash, NodeId peer, bool preferred) const; + + /** Run internal consistency check (testing only). */ + void SanityCheck() const; + + /** Run a time-dependent internal consistency check (testing only). + * + * This can only be called immediately after GetRequestable, with the same 'now' parameter. + */ + void PostGetRequestableSanityCheck(std::chrono::microseconds now) const; +}; + +#endif // BITCOIN_TXREQUEST_H diff --git a/test/functional/p2p_tx_download.py b/test/functional/p2p_tx_download.py index 7bd616841556..aef451cccf27 100755 --- a/test/functional/p2p_tx_download.py +++ b/test/functional/p2p_tx_download.py @@ -37,16 +37,13 @@ def on_getdata(self, message): # Constants from net_processing GETDATA_TX_INTERVAL = 60 # seconds -MAX_GETDATA_RANDOM_DELAY = 2 # seconds -INBOUND_PEER_TX_DELAY = 2 # seconds -MAX_GETDATA_IN_FLIGHT = 100 -MAX_BLOCKS_IN_TRANSIT_PER_PEER = 16 -MAX_NOTFOUND_SIZE = MAX_GETDATA_IN_FLIGHT + MAX_BLOCKS_IN_TRANSIT_PER_PEER -TX_EXPIRY_INTERVAL = GETDATA_TX_INTERVAL * 10 +NONPREF_PEER_TX_DELAY = 2 # seconds +OVERLOADED_PEER_OBJECT_DELAY = 2 # seconds +MAX_PEER_OBJECT_REQUEST_IN_FLIGHT = 100 # Python test constants NUM_INBOUND = 10 -MAX_GETDATA_INBOUND_WAIT = GETDATA_TX_INTERVAL + MAX_GETDATA_RANDOM_DELAY + INBOUND_PEER_TX_DELAY +MAX_GETDATA_INBOUND_WAIT = GETDATA_TX_INTERVAL + NONPREF_PEER_TX_DELAY class TxDownloadTest(BitcoinTestFramework): @@ -59,7 +56,7 @@ def test_tx_requests(self): txid = 0xdeadbeef self.log.info("Announce the txid from each incoming peer to node 0") - msg = msg_inv([CInv(t=1, h=txid)]) + msg = msg_inv([CInv(t=MSG_TX, h=txid)]) for p in self.nodes[0].p2ps: p.send_and_ping(msg) @@ -86,11 +83,9 @@ def test_inv_block(self): self.log.info( "Announce the transaction to all nodes from all {} incoming peers, but never send it".format(NUM_INBOUND)) - msg = msg_inv([CInv(t=1, h=txid)]) + msg = msg_inv([CInv(t=MSG_TX, h=txid)]) for p in self.peers: p.send_and_ping(msg) - p.sync_with_ping() - self.bump_mocktime(1) self.log.info("Put the tx in node 0's mempool") self.nodes[0].sendrawtransaction(tx['hex']) @@ -103,62 +98,122 @@ def test_inv_block(self): # peer, plus # * the first time it is re-requested from the outbound peer, plus # * 2 seconds to avoid races - timeout = 2 + (MAX_GETDATA_RANDOM_DELAY + INBOUND_PEER_TX_DELAY) + ( - GETDATA_TX_INTERVAL + MAX_GETDATA_RANDOM_DELAY) - self.log.info("Tx should be received at node 1 after {} seconds".format(timeout)) - self.sync_mempools(timeout=timeout) + assert self.nodes[1].getpeerinfo()[0]['inbound'] == False + timeout = 2 + NONPREF_PEER_TX_DELAY + GETDATA_TX_INTERVAL + self.log.info("Tx should be received at node 1 after {} seconds of node time".format(timeout)) + # The relay needs several SendMessages cycles, so advance node time in 1 s steps rather than + # one jump -- but no further than the expected worst case. Once mocktime reaches the deadline + # we stop advancing it, so a relay-delay regression (tx needing more than `timeout` + # node-seconds) makes the test fail instead of silently advancing mocktime until it passes. + deadline = self.mocktime + timeout + def tx_relayed(): + if self.mocktime < deadline: + self.bump_mocktime(1) + return tx['txid'] in self.nodes[1].getrawmempool() + self.wait_until(tx_relayed) def test_in_flight_max(self): - self.log.info("Test that we don't request more than {} transactions from any peer, every {} minutes".format( - MAX_GETDATA_IN_FLIGHT, TX_EXPIRY_INTERVAL / 60)) - txids = [i for i in range(MAX_GETDATA_IN_FLIGHT + 2)] + self.log.info("Test that we don't load peers with more than {} transaction requests immediately".format(MAX_PEER_OBJECT_REQUEST_IN_FLIGHT)) + txids = [i for i in range(MAX_PEER_OBJECT_REQUEST_IN_FLIGHT + 2)] p = self.nodes[0].p2ps[0] with p2p_lock: p.tx_getdata_count = 0 - p.send_message(msg_inv([CInv(t=1, h=i) for i in txids])) - - def wait_for_tx_getdata(target): - self.bump_mocktime(1) - return p.tx_getdata_count >= target - - p.wait_until(lambda: wait_for_tx_getdata(MAX_GETDATA_IN_FLIGHT)) - + for i in range(MAX_PEER_OBJECT_REQUEST_IN_FLIGHT): + p.send_message(msg_inv([CInv(t=MSG_TX, h=txids[i])])) + p.sync_with_ping() + self.bump_mocktime(NONPREF_PEER_TX_DELAY) + p.wait_until(lambda: p.tx_getdata_count >= MAX_PEER_OBJECT_REQUEST_IN_FLIGHT) + for i in range(MAX_PEER_OBJECT_REQUEST_IN_FLIGHT, len(txids)): + p.send_message(msg_inv([CInv(t=MSG_TX, h=txids[i])])) + p.sync_with_ping() + self.log.info("No more than {} requests should be seen within {} seconds after announcement".format(MAX_PEER_OBJECT_REQUEST_IN_FLIGHT, NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_OBJECT_DELAY - 1)) + self.bump_mocktime(NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_OBJECT_DELAY - 1) + p.sync_with_ping() with p2p_lock: - assert_equal(p.tx_getdata_count, MAX_GETDATA_IN_FLIGHT) - - self.log.info("Now check that if we send a NOTFOUND for a transaction, we'll get one more request") - p.send_message(msg_notfound(vec=[CInv(t=1, h=txids[0])])) - p.wait_until(lambda: wait_for_tx_getdata(MAX_GETDATA_IN_FLIGHT + 1), timeout=10) + assert_equal(p.tx_getdata_count, MAX_PEER_OBJECT_REQUEST_IN_FLIGHT) + self.log.info("If we wait {} seconds after announcement, we should eventually get more requests".format(NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_OBJECT_DELAY)) + self.bump_mocktime(1) + p.wait_until(lambda: p.tx_getdata_count == len(txids)) + + def test_expiry_fallback(self): + self.log.info('Check that expiry will select another peer for download') + TXID = 0xffaa + peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) + peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) + for p in [peer1, peer2]: + p.send_and_ping(msg_inv([CInv(t=MSG_TX, h=TXID)])) + self.bump_mocktime(NONPREF_PEER_TX_DELAY) + # One of the peers is asked for the tx + peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1) with p2p_lock: - assert_equal(p.tx_getdata_count, MAX_GETDATA_IN_FLIGHT + 1) - - WAIT_TIME = TX_EXPIRY_INTERVAL // 2 + TX_EXPIRY_INTERVAL - self.log.info("if we wait about {} minutes, we should eventually get more requests".format(WAIT_TIME / 60)) - self.bump_mocktime(WAIT_TIME) - p.wait_until(lambda: wait_for_tx_getdata(MAX_GETDATA_IN_FLIGHT + 2)) + peer_expiry, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1) + assert_equal(peer_fallback.tx_getdata_count, 0) + self.bump_mocktime(GETDATA_TX_INTERVAL + 1) # Wait for request to peer_expiry to expire + peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) + + def test_disconnect_fallback(self): + self.log.info('Check that disconnect will select another peer for download') + TXID = 0xffbb + peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) + peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) + for p in [peer1, peer2]: + p.send_and_ping(msg_inv([CInv(t=MSG_TX, h=TXID)])) + self.bump_mocktime(NONPREF_PEER_TX_DELAY) + # One of the peers is asked for the tx + peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1) + with p2p_lock: + peer_disconnect, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1) + assert_equal(peer_fallback.tx_getdata_count, 0) + peer_disconnect.peer_disconnect() + peer_disconnect.wait_for_disconnect() + peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) + + def test_notfound_fallback(self): + self.log.info('Check that notfounds will select another peer for download immediately') + TXID = 0xffdd + peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) + peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) + for p in [peer1, peer2]: + p.send_and_ping(msg_inv([CInv(t=MSG_TX, h=TXID)])) + self.bump_mocktime(NONPREF_PEER_TX_DELAY) + # One of the peers is asked for the tx + peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1) + with p2p_lock: + peer_notfound, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1) + assert_equal(peer_fallback.tx_getdata_count, 0) + peer_notfound.send_and_ping(msg_notfound(vec=[CInv(MSG_TX, TXID)])) # Send notfound, so that fallback peer is selected + peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) + + def test_preferred_inv(self): + self.log.info('Check that invs from preferred peers are downloaded immediately') + self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1']) + peer = self.nodes[0].add_p2p_connection(TestP2PConn()) + peer.send_message(msg_inv([CInv(t=MSG_TX, h=0xff00ff00)])) + peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1) def test_spurious_notfound(self): self.log.info('Check that spurious notfound is ignored') - self.nodes[0].p2ps[0].send_message(msg_notfound(vec=[CInv(1, 1)])) - - def test_oversized_notfound(self): - self.log.info('Check that oversized notfound increases misbehavior score') - oversized_notfound_count = MAX_NOTFOUND_SIZE + 1 - invs = [CInv(t=1, h=i) for i in range(oversized_notfound_count)] - with self.nodes[0].assert_debug_log(["Misbehaving", f"notfound message size = {oversized_notfound_count}"]): - self.nodes[0].p2ps[0].send_message(msg_notfound(vec=invs)) - self.nodes[0].p2ps[0].sync_with_ping() + self.restart_node(0) + peer = self.nodes[0].add_p2p_connection(TestP2PConn()) + peer.send_and_ping(msg_notfound(vec=[CInv(MSG_TX, 1)])) def run_test(self): self.wallet = MiniWallet(self.nodes[0]) self.wallet.rescan_utxos() - # Run each test against new bitcoind instances, as setting mocktimes has long-term effects on when + # Run tests that only need one peer-connection first, to avoid restarting the nodes + self.test_expiry_fallback() + self.test_disconnect_fallback() + self.test_notfound_fallback() + self.test_preferred_inv() + self.test_spurious_notfound() + + # Run each test against new dashd instances, as setting mocktimes has long-term effects on when # the next trickle relay event happens. - for test in [self.test_spurious_notfound, self.test_oversized_notfound, self.test_in_flight_max, self.test_inv_block, self.test_tx_requests]: + for test in [self.test_in_flight_max, self.test_inv_block, self.test_tx_requests]: self.stop_nodes() self.start_nodes() self.connect_nodes(1, 0)