diff --git a/src/Makefile.test.include b/src/Makefile.test.include index 90a0ed75ed64..4ff1bb41c8dd 100644 --- a/src/Makefile.test.include +++ b/src/Makefile.test.include @@ -170,6 +170,7 @@ BITCOIN_TESTS =\ test/net_peer_eviction_tests.cpp \ test/net_tests.cpp \ test/netbase_tests.cpp \ + test/orphan_vote_cache_tests.cpp \ test/orphanage_tests.cpp \ test/pmt_tests.cpp \ test/policyestimator_tests.cpp \ diff --git a/src/cachemultimap.h b/src/cachemultimap.h index 88f3f12c2bea..53ce966a72d5 100644 --- a/src/cachemultimap.h +++ b/src/cachemultimap.h @@ -110,6 +110,15 @@ class CacheMultiMap return (mapIndex.find(key) != mapIndex.end()); } + bool HasEntry(const K& key, const V& value) const + { + map_cit mit = mapIndex.find(key); + if (mit == mapIndex.end()) { + return false; + } + return mit->second.count(value) > 0; + } + bool Get(const K& key, V& value) const { map_cit it = mapIndex.find(key); diff --git a/src/governance/governance.cpp b/src/governance/governance.cpp index 22dafc4214a1..d7bd06e9c7f5 100644 --- a/src/governance/governance.cpp +++ b/src/governance/governance.cpp @@ -65,7 +65,7 @@ GovernanceStore::GovernanceStore() : cs_store(), mapObjects(), mapErasedGovernanceObjects(), - cmmapOrphanVotes(MAX_CACHE_SIZE), + m_orphan_votes(MAX_ORPHAN_VOTES, MAX_ORPHAN_VOTES_PER_MN), mapLastMasternodeObject(), lastMNListForVotingKeys(std::make_shared()) { @@ -281,7 +281,7 @@ void CGovernanceManager::CheckOrphanVotes(CGovernanceObject& govobj) uint256 nHash = govobj.GetHash(); std::vector orphan_votes; - cmmapOrphanVotes.GetAll(nHash, orphan_votes); + m_orphan_votes.GetAll(nHash, orphan_votes); ScopedLockBool guard(cs_store, fRateChecksEnabled, false); @@ -298,7 +298,7 @@ void CGovernanceManager::CheckOrphanVotes(CGovernanceObject& govobj) fRemove = true; } if (fRemove) { - cmmapOrphanVotes.Erase(nHash, orphan_vote); + m_orphan_votes.Erase(nHash, orphan_vote); } } } @@ -415,6 +415,11 @@ void CGovernanceManager::CheckAndRemove() ScopedLockBool guard(cs_store, fRateChecksEnabled, false); + // Drop orphan votes whose parent never arrived. Votes for an object that did arrive are + // consumed by CheckOrphanVotes() at that point, so anything still here is either waiting or + // dead; this is the only thing that removes the latter. + ExpireOrphanVotes(); + // Clean up any expired or invalid triggers m_superblocks.Clean(nCachedBlockHeight); @@ -841,14 +846,29 @@ bool CGovernanceManager::ProcessVote(const CGovernanceVote& vote, CGovernanceExc exception = CGovernanceException(msg, GOVERNANCE_EXCEPTION_PERMANENT_ERROR, 20); return false; } + const auto insert_result = m_orphan_votes.Insert( + nHashGovobj, governance::OrphanVote{vote, Now() + GOVERNANCE_ORPHAN_EXPIRATION_TIME}); + if (insert_result == governance::OrphanVoteCache::InsertResult::MN_LIMIT) { + // This masternode already fills its share of the cache with votes for parents that + // never arrived; drop the vote and stop letting the key seed parent requests. Still no + // penalty: the sending peer may merely be relaying, and misbehaviour scores never decay. + std::string msg{strprintf("CGovernanceManager::%s -- Masternode %s is over its orphan-vote share, " + "dropping vote for unknown parent object %s", + __func__, vote.GetMasternodeOutpoint().ToStringShort(), nHashGovobj.ToString())}; + exception = CGovernanceException(msg, GOVERNANCE_EXCEPTION_WARNING); + LogPrint(BCLog::GOBJECT, "%s\n", msg); + return false; + } std::string msg{strprintf("CGovernanceManager::%s -- Unknown parent object %s, MN outpoint = %s", __func__, nHashGovobj.ToString(), vote.GetMasternodeOutpoint().ToStringShort())}; // No penalty: the vote is signed by a masternode, it just arrived before its parent object, // which routinely happens during governance sync. Misbehaviour scores never decay. exception = CGovernanceException(msg, GOVERNANCE_EXCEPTION_WARNING); - if (cmmapOrphanVotes.Insert(nHashGovobj, governance::OrphanVote{vote, Now() + GOVERNANCE_ORPHAN_EXPIRATION_TIME})) { - hashToRequest = nHashGovobj; // Caller should request this object - } + // Ask for the parent whether or not the vote itself was new to us. A vote we already hold, + // relayed by a second peer, is fresh evidence that this peer has the parent -- and it is the + // only evidence we will get, since a peer relays a given vote once. Suppressing the request + // on a duplicate would strand the parent whenever the first peer we asked fails to deliver. + hashToRequest = nHashGovobj; LogPrint(BCLog::GOBJECT, "%s\n", msg); return false; } @@ -1023,7 +1043,7 @@ void GovernanceStore::Clear() LOCK(cs_store); mapObjects.clear(); mapErasedGovernanceObjects.clear(); - cmmapOrphanVotes.Clear(); + m_orphan_votes.Clear(); mapLastMasternodeObject.clear(); lastMNListForVotingKeys = std::make_shared(); } @@ -1133,33 +1153,24 @@ void CGovernanceManager::UpdatedBlockTip(const CBlockIndex* pindex) m_superblocks.ExecuteBestSuperblock(m_dmnman.GetListAtChainTip(), pindex->nHeight); } -std::vector CGovernanceManager::GetOrphanVoteObjectHashes() +void CGovernanceManager::ExpireOrphanVotes() { - LOCK(cs_store); + AssertLockHeld(cs_store); const auto now{Now()}; - - // Clean up expired orphan votes - const vote_cmm_t::list_t& items = cmmapOrphanVotes.GetItemList(); + const auto& items = m_orphan_votes.GetItemList(); for (auto it = items.begin(); it != items.end();) { auto prevIt = it; ++it; if (prevIt->value.expiration < now) { - cmmapOrphanVotes.Erase(prevIt->key, prevIt->value); - } - } - - // Get hashes of objects we don't have yet - std::vector vecHashesFiltered; - std::vector vecHashes; - cmmapOrphanVotes.GetKeys(vecHashes); - for (const uint256& nHash : vecHashes) { - if (mapObjects.find(nHash) == mapObjects.end()) { - vecHashesFiltered.push_back(nHash); + m_orphan_votes.Erase(prevIt->key, prevIt->value); } } +} - return vecHashesFiltered; +size_t CGovernanceManager::GetOrphanVoteCount() const +{ + return WITH_LOCK(cs_store, return m_orphan_votes.GetSize()); } void CGovernanceManager::RemoveInvalidVotes() @@ -1201,7 +1212,6 @@ void CGovernanceManager::RemoveInvalidVotes() } for (const auto& voteHash : removed) { cmapVoteToObject.Erase(voteHash); - cmmapOrphanVotes.Erase(voteHash); } } } diff --git a/src/governance/governance.h b/src/governance/governance.h index 04536cb5ef04..b35dc741b68d 100644 --- a/src/governance/governance.h +++ b/src/governance/governance.h @@ -58,6 +58,89 @@ inline bool operator<(const OrphanVote& lhs, const OrphanVote& rhs) { return lhs.vote < rhs.vote; } + +/** Bounded holding area for votes whose parent object has not arrived yet. Entries are + * peer-supplied and evicted oldest-first, so two bounds are enforced together: a global one that + * caps memory, and a per-masternode one so that a single voting key cannot mint votes naming + * invented parents and flush everyone else's orphans on demand. */ +class OrphanVoteCache +{ +public: + using cache_t = CacheMultiMap; + + enum class InsertResult { + OK, + DUPLICATE, //!< this (parent, vote) pair is already cached + MN_LIMIT, //!< the vote's masternode already holds its full share of the cache + }; + + OrphanVoteCache(size_t max_total, size_t max_per_mn) : + m_max_total{max_total}, + m_max_per_mn{max_per_mn}, + // Eviction happens here, before the inner map is ever full, so its own self-pruning + // (which would bypass the per-masternode accounting) can never trigger. + m_cache{static_cast(max_total + 1)} + { + } + + InsertResult Insert(const uint256& parent_hash, const OrphanVote& orphan_vote) + { + // A pair we already hold is reported as a duplicate even when its masternode is over its + // share: it costs no capacity, and the caller treats a repeat relay as fresh evidence that + // the sending peer has the missing parent. + if (m_cache.HasEntry(parent_hash, orphan_vote)) { + return InsertResult::DUPLICATE; + } + const COutPoint outpoint{orphan_vote.vote.GetMasternodeOutpoint()}; + if (const auto it{m_counts.find(outpoint)}; it != m_counts.end() && it->second >= m_max_per_mn) { + return InsertResult::MN_LIMIT; + } + if (!m_cache.Insert(parent_hash, orphan_vote)) { + return InsertResult::DUPLICATE; + } + ++m_counts[outpoint]; + if (m_cache.GetSize() > m_max_total) { + // Insert prepends, so the back is the oldest entry cache-wide and never the one just + // added. Copied, not referenced: Erase destroys the node. + const auto oldest{m_cache.GetItemList().back()}; + Erase(oldest.key, oldest.value); + } + return InsertResult::OK; + } + + void Erase(const uint256& parent_hash, const OrphanVote& orphan_vote) + { + // Callers may pass a reference into the cache's own item list (e.g. the expiry sweep), + // which the erase below invalidates; take what we need first. + const COutPoint outpoint{orphan_vote.vote.GetMasternodeOutpoint()}; + const auto size_before{m_cache.GetSize()}; + m_cache.Erase(parent_hash, orphan_vote); + if (m_cache.GetSize() == size_before) return; + if (const auto it{m_counts.find(outpoint)}; it != m_counts.end() && --it->second == 0) { + m_counts.erase(it); + } + } + + bool GetAll(const uint256& parent_hash, std::vector& votes) + { + return m_cache.GetAll(parent_hash, votes); + } + + void Clear() + { + m_cache.Clear(); + m_counts.clear(); + } + + size_t GetSize() const { return m_cache.GetSize(); } + const cache_t::list_t& GetItemList() const { return m_cache.GetItemList(); } + +private: + const size_t m_max_total; + const size_t m_max_per_mn; + cache_t m_cache; + std::map m_counts; +}; } // namespace governance static constexpr int RATE_BUFFER_SIZE = 5; @@ -175,6 +258,17 @@ class GovernanceStore using txout_m_t = std::map; using vote_cmm_t = CacheMultiMap; +public: + /** Bounds for the orphan-vote cache, which is filled from the network by any peer with a + * parent object we do not have. The per-masternode bound is the security control: eviction + * is oldest-first, and every entry costs its sender a registered masternode voting key, so + * one key can occupy at most its share instead of flushing the whole cache; the value is + * far above the number of in-flight objects one masternode can plausibly have voted on. + * The global bound caps memory: CacheMultiMap stores each vote twice, so an entry costs + * roughly 750 bytes and this allows ~75 MB; filling it takes 200 distinct masternode keys. */ + static constexpr size_t MAX_ORPHAN_VOTES = 100'000; + static constexpr size_t MAX_ORPHAN_VOTES_PER_MN = 500; + protected: static constexpr int MAX_CACHE_SIZE = 1000000; static const std::string SERIALIZATION_VERSION_STRING; @@ -189,7 +283,7 @@ class GovernanceStore // key - governance object's hash // value - expiration time for deleted objects std::map mapErasedGovernanceObjects GUARDED_BY(cs_store); - vote_cmm_t cmmapOrphanVotes GUARDED_BY(cs_store); + governance::OrphanVoteCache m_orphan_votes GUARDED_BY(cs_store); txout_m_t mapLastMasternodeObject GUARDED_BY(cs_store); // used to check for changed voting keys std::shared_ptr lastMNListForVotingKeys GUARDED_BY(cs_store); @@ -202,9 +296,14 @@ class GovernanceStore void Serialize(Stream &s) const EXCLUSIVE_LOCKS_REQUIRED(!cs_store) { LOCK(cs_store); + // TODO: Remove the historical orphan-vote field on the next disk-format version bump. + // Unserialize discards the orphans, so write none. Builds that predate this still load the + // field, capacity included, as their live cache: keep the capacity they always wrote, since + // a CacheMultiMap with capacity 0 never prunes. + const vote_cmm_t empty_orphan_votes{MAX_CACHE_SIZE}; s << SERIALIZATION_VERSION_STRING << mapErasedGovernanceObjects - << cmmapOrphanVotes + << empty_orphan_votes << mapObjects << mapLastMasternodeObject << *lastMNListForVotingKeys; @@ -222,8 +321,14 @@ class GovernanceStore return; } + // The historical format stores CacheMultiMap's capacity with its entries. Consume that + // field to preserve the format, but keep both the stale orphan votes and their disk-supplied + // capacity out of the live cache. Orphans are a ten-minute recovery window invalidated by + // the restart; the live capacity is fixed at construction. + vote_cmm_t discarded_orphan_votes; + s >> mapErasedGovernanceObjects - >> cmmapOrphanVotes + >> discarded_orphan_votes >> mapObjects >> mapLastMasternodeObject >> *lastMNListForVotingKeys; @@ -358,8 +463,8 @@ class CGovernanceManager : public GovernanceStore // Used by NetGovernance std::vector FetchRelayInventory() EXCLUSIVE_LOCKS_REQUIRED(!cs_relay); void CheckAndRemove() EXCLUSIVE_LOCKS_REQUIRED(!cs_store); - /** Get hashes of governance objects for which we have orphan votes. Also cleans up expired orphans. */ - [[nodiscard]] std::vector GetOrphanVoteObjectHashes() EXCLUSIVE_LOCKS_REQUIRED(!cs_store); + /** Number of orphan votes currently held, so the MAX_ORPHAN_VOTES bound can be asserted. */ + [[nodiscard]] size_t GetOrphanVoteCount() const EXCLUSIVE_LOCKS_REQUIRED(!cs_store); std::pair, std::vector> FetchGovernanceObjectVotes( size_t peers_per_hash_max, int64_t now, std::map>& map_asked_recently) const EXCLUSIVE_LOCKS_REQUIRED(!cs_store); @@ -415,6 +520,10 @@ class CGovernanceManager : public GovernanceStore void CheckOrphanVotes(CGovernanceObject& govobj) EXCLUSIVE_LOCKS_REQUIRED(cs_store, !cs_relay); + /** Drop orphan votes whose parent object never arrived within GOVERNANCE_ORPHAN_EXPIRATION_TIME. */ + void ExpireOrphanVotes() + EXCLUSIVE_LOCKS_REQUIRED(cs_store); + void RebuildIndexes() EXCLUSIVE_LOCKS_REQUIRED(cs_store); diff --git a/src/governance/net_governance.cpp b/src/governance/net_governance.cpp index 2e8b5da79140..aa69b0c4cb29 100644 --- a/src/governance/net_governance.cpp +++ b/src/governance/net_governance.cpp @@ -45,23 +45,9 @@ void NetGovernance::Schedule(CScheduler& scheduler) [this]() -> void { if (!m_node_sync.IsSynced()) return; - // Request governance objects for orphan votes - auto vecOrphanHashes = m_gov_manager.GetOrphanVoteObjectHashes(); - if (!vecOrphanHashes.empty()) { - LogPrint(BCLog::GOBJECT, "NetGovernance::Schedule -- requesting %d orphan objects\n", - vecOrphanHashes.size()); - const CConnman::NodesSnapshot snap{m_connman, CConnman::FullyConnectedOnly}; - for (const uint256& nHash : vecOrphanHashes) { - for (CNode* pnode : snap.Nodes()) { - if (!pnode->CanRelay()) continue; - CNetMsgMaker msgMaker(pnode->GetCommonVersion()); - CBloomFilter filter; // Empty filter - we want the object, not votes - m_connman.PushMessage(pnode, msgMaker.Make(NetMsgType::MNGOVERNANCESYNC, nHash, filter)); - } - } - } - // CHECK AND REMOVE - REPROCESS GOVERNANCE OBJECTS + // Also expires orphan votes whose parent object never arrived. Fetching those parents + // is driven by the object request tracker from ProcessMessage(), not from here. m_gov_manager.CheckAndRemove(); }, std::chrono::minutes{5}); @@ -113,9 +99,9 @@ void NetGovernance::ProcessMessage(CNode& peer, const std::string& msg_type, CDa LogPrint(BCLog::GOBJECT, "MNGOVERNANCESYNC -- syncing governance objects to our peer %s\n", peer.GetLogString()); const bool full_sync{nProp == uint256()}; const bool object_fetch{!full_sync && IsEmptyBloomFilter(filter)}; - // Nonzero govsync with an empty filter is used to retry missing-object - // fetches for orphan votes. Only full sync and actual known-object vote - // sync are fulfilled-request limited. + // Nonzero govsync with an empty filter is how older peers retry + // missing-object fetches for orphan votes. Only full sync and actual + // known-object vote sync are fulfilled-request limited. const bool track_request{full_sync || (!object_fetch && m_gov_manager.HaveSyncableObjectForHash(nProp))}; const std::string fulfilled_request{full_sync ? NetMsgType::MNGOVERNANCESYNC : @@ -257,11 +243,13 @@ void NetGovernance::ProcessMessage(CNode& peer, const std::string& msg_type, CDa // m_peer_manager->PeerRelayInv(CInv{MSG_GOVERNANCE_OBJECT_VOTE, nHash}); } else { LogPrint(BCLog::GOBJECT, "MNGOVERNANCEOBJECTVOTE -- Rejected vote, error = %s\n", exception.what()); - if (hashToRequest != uint256()) { - // Orphan vote - request the missing governance object - CNetMsgMaker msgMaker(peer.GetCommonVersion()); - CBloomFilter filter; // Empty filter - we just want the object, not votes - m_connman.PushMessage(&peer, msgMaker.Make(NetMsgType::MNGOVERNANCESYNC, hashToRequest, filter)); + if (!hashToRequest.IsNull()) { + // Orphan vote: fetch the parent object through the request tracker, which owns + // GETDATA scheduling, per-peer in-flight limits, expiry and fallback to another + // peer. Register this peer explicitly -- holding a vote for the object is evidence + // it has the object, and it may never have announced the object to us. + m_peer_manager->PeerAskPeersForObject(CInv{MSG_GOVERNANCE_OBJECT, hashToRequest}, + peer.GetId()); } if ((exception.GetNodePenalty() != 0) && m_node_sync.IsSynced()) { m_peer_manager->PeerMisbehaving(peer.GetId(), exception.GetNodePenalty()); diff --git a/src/instantsend/net_instantsend.cpp b/src/instantsend/net_instantsend.cpp index b5871d326546..368b4a433368 100644 --- a/src/instantsend/net_instantsend.cpp +++ b/src/instantsend/net_instantsend.cpp @@ -419,7 +419,7 @@ void NetInstantSend::ProcessInstantSendLock(NodeId from, const uint256& hash, co m_peer_manager->PeerRelayInvFiltered(inv, *tx); } else { m_peer_manager->PeerRelayInvFiltered(inv, islock->txid); - m_peer_manager->PeerAskPeersForTransaction(islock->txid); + m_peer_manager->PeerAskPeersForObject(CInv{MSG_TX, islock->txid}); } } diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 2d84872ac1ce..579442b4015d 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -83,6 +83,10 @@ using node::ReadBlockFromDisk; /** 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; +/** How many peers to ask for an object we want but were never offered (see AskPeersForObject). + * Small on purpose: the request tracker retries and falls back to the next candidate on expiry, so + * this is the width of the initial attempt, not the number of chances to obtain the object. */ +static constexpr size_t MAX_PEERS_TO_ASK_FOR_OBJECT = 4; /** 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). */ @@ -649,8 +653,8 @@ class PeerManagerImpl final : public PeerManager void PeerRelayDSQ(const CCoinJoinQueue& queue) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void PeerRelayTransaction(const uint256& txid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void PeerRelayRecoveredSig(const llmq::CRecoveredSig& sig, bool proactive_relay) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); - void PeerAskPeersForTransaction(const uint256& txid) override - EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); + void PeerAskPeersForObject(const CInv& inv, NodeId explicit_peer) override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex, !cs_main); size_t PeerGetRequestedObjectCount(NodeId nodeid) const override EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); void PeerPostProcessMessage(MessageProcessingResult&& ret) override @@ -659,8 +663,9 @@ class PeerManagerImpl final : public PeerManager private: void _RelayTransaction(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(cs_main, !m_peer_mutex); - /** Ask peers that have a transaction in their inventory to relay it to us. */ - void AskPeersForTransaction(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); + /** Ask peers that have the object in their inventory to relay it to us, plus explicit_peer. */ + void AskPeersForObject(const CInv& inv, NodeId explicit_peer) + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex, !cs_main); /** Relay inventories to peers that find it relevant */ void RelayInvFiltered(const CInv& inv, const CTransaction& relatedTx) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); @@ -672,8 +677,11 @@ class PeerManagerImpl final : public PeerManager void RelayInvFiltered(const CInv& inv, const uint256& relatedTxHash) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); /** 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) + * request delay from the peer's preferredness and in-flight load. force_preferred is for + * announcements we synthesize because we want the object (see AskPeersForObject). Returns + * whether the announcement passed the per-peer accounting. */ + bool AddObjectAnnouncement(NodeId nodeid, const CInv& inv, std::chrono::microseconds current_time, + bool force_preferred = false) EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, ::cs_main); /** Delete all announcements of a transaction across all peers, under the inv types it may @@ -1631,21 +1639,24 @@ bool IsGetDataOnlyObject(int invType) } } -void PeerManagerImpl::AddObjectAnnouncement(const CNode& node, const CInv& inv, std::chrono::microseconds current_time) +bool PeerManagerImpl::AddObjectAnnouncement(NodeId nodeid, const CInv& inv, std::chrono::microseconds current_time, + bool force_preferred) { AssertLockHeld(cs_main); - const CNodeState* state = State(node.GetId()); - if (state == nullptr) return; + // nullptr if the peer disconnected; AskPeersForObject snapshots candidates outside cs_main. + const CNodeState* state = State(nodeid); + if (state == nullptr) return false; LOCK(m_object_request_mutex); - if (m_object_request.Count(node.GetId()) >= MAX_PEER_OBJECT_ANNOUNCEMENTS) { + if (m_object_request.Count(nodeid) >= MAX_PEER_OBJECT_ANNOUNCEMENTS) { // Too many queued announcements from this peer - return; + return false; } // Decide the TxRequestTracker parameters for this announcement: - // - "preferred": if fPreferredDownload is set (= outbound, or PF_NOBAN permission) + // - "preferred": if fPreferredDownload is set (= outbound, or PF_NOBAN permission), or if the + // caller forces it (an announcement we synthesized because we want the object) // - "reqtime": current time plus delays for: // - NONPREF_PEER_TX_DELAY for MSG_TX announcements from non-preferred connections. Other // object types -- including MSG_DSTX (used for the orphan-parent fetch, which wants the @@ -1653,13 +1664,14 @@ void PeerManagerImpl::AddObjectAnnouncement(const CNode& node, const CInv& inv, // neither is anything in masternode mode. This matches the pre-txrequest Dash behavior. // - 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; + const bool preferred = force_preferred || state->fPreferredDownload; auto delay{0us}; if (inv.IsMsgTx() && !preferred && m_nodeman == nullptr) delay += NONPREF_PEER_TX_DELAY; - const bool overloaded = m_object_request.CountInFlight(node.GetId()) >= MAX_PEER_OBJECT_REQUEST_IN_FLIGHT; + const bool overloaded = m_object_request.CountInFlight(nodeid) >= MAX_PEER_OBJECT_REQUEST_IN_FLIGHT; if (overloaded) delay += OVERLOADED_PEER_OBJECT_DELAY; - m_object_request.ReceivedInv(node.GetId(), inv, preferred, current_time + delay); + m_object_request.ReceivedInv(nodeid, inv, preferred, current_time + delay); + return true; } void PeerManagerImpl::ForgetTx(const uint256& txid) @@ -2467,44 +2479,49 @@ void PeerManagerImpl::SendPings() for(auto& it : m_peer_map) it.second->m_ping_queued = true; } -void PeerManagerImpl::AskPeersForTransaction(const uint256& txid) +void PeerManagerImpl::AskPeersForObject(const CInv& inv, NodeId explicit_peer) { - std::vector peersToAsk; - peersToAsk.reserve(4); - + // Every matching peer is collected, not just the first few: admission below needs cs_main, + // which cannot be taken under m_peer_mutex, so candidates that turn out to be disconnected or + // over their announcement cap are only discovered afterwards and must not consume the budget. + std::vector candidates; { READ_LOCK(m_peer_mutex); + candidates.reserve(m_peer_map.size()); + + // A peer that holds the object without having announced it is not in any inventory filter, + // so it can only be reached by being named explicitly. Candidate selection order remains + // the request tracker's responsibility. + if (explicit_peer != -1) { + if (auto it = m_peer_map.find(explicit_peer); it != m_peer_map.end()) { + candidates.emplace_back(it->second); + } + } + // TODO consider prioritizing MNs again, once that flag is moved into Peer for (const auto& [_, peer] : m_peer_map) { - if (peersToAsk.size() >= 4) { - break; - } - if (IsInvInFilter(*peer, txid)) { - peersToAsk.emplace_back(peer); + if (peer->m_id != explicit_peer && IsInvInFilter(*peer, inv.hash)) { + candidates.emplace_back(peer); } } } - { - LOCK(cs_main); - const auto current_time{GetTime()}; - // Register a fresh, preferred (undelayed) MSG_TX announcement from each peer we intend to - // ask, so the transaction is requested ASAP. We deliberately do not forget existing - // announcements for this txid: any live candidate/request from another peer must survive as - // a fallback, and there is nothing to "unstick" -- the tracker deletes a txid's COMPLETED - // announcements automatically once no live one remains, so a completed entry only lingers - // while some peer is still being tried. If a peer here already has an announcement, - // ReceivedInv is a no-op and the existing one (in flight or queued) keeps its place. - 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); - - WITH_LOCK(m_object_request_mutex, - m_object_request.ReceivedInv(peer->m_id, CInv(MSG_TX, txid), /*preferred=*/true, current_time)); + + LOCK(cs_main); + + const auto current_time{GetTime()}; + size_t asked_count{0}; + + // Synthesize an announcement from each peer we ask, forced preferred (we want this one ASAP) + // but through the same per-peer accounting as peer-sent announcements. Existing announcements + // for this hash are deliberately left alone: they must survive as fallbacks. + for (const auto& peer : candidates) { + if (asked_count >= MAX_PEERS_TO_ASK_FOR_OBJECT) { + break; + } + if (AddObjectAnnouncement(peer->m_id, inv, current_time, /*force_preferred=*/true)) { + LogPrint(BCLog::NET, "PeerManagerImpl::%s -- %s: asking peer %d\n", __func__, inv.ToString(), + peer->m_id); + ++asked_count; } } } @@ -4570,7 +4587,7 @@ void PeerManagerImpl::ProcessMessage( } bool allowWhileInIBD = allowWhileInIBDObjs.count(inv.type); if (allowWhileInIBD || !m_chainman.ActiveChainstate().IsInitialBlockDownload()) { - AddObjectAnnouncement(pfrom, inv, current_time); + AddObjectAnnouncement(pfrom.GetId(), inv, current_time); } } } @@ -4988,7 +5005,7 @@ void PeerManagerImpl::ProcessMessage( // 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)) AddObjectAnnouncement(pfrom, _inv, current_time); + if (!AlreadyHave(_inv)) AddObjectAnnouncement(pfrom.GetId(), _inv, current_time); } if (m_orphanage.AddTx(ptx, pfrom.GetId())) { @@ -6984,9 +7001,9 @@ void PeerManagerImpl::PeerRelayTransaction(const uint256& txid) RelayTransaction(txid); } -void PeerManagerImpl::PeerAskPeersForTransaction(const uint256& txid) +void PeerManagerImpl::PeerAskPeersForObject(const CInv& inv, NodeId explicit_peer) { - AskPeersForTransaction(txid); + AskPeersForObject(inv, explicit_peer); } size_t PeerManagerImpl::PeerGetRequestedObjectCount(NodeId nodeid) const diff --git a/src/net_processing.h b/src/net_processing.h index 713e1bad9d8e..052668c1e14f 100644 --- a/src/net_processing.h +++ b/src/net_processing.h @@ -116,7 +116,19 @@ class PeerManagerInternal virtual void PeerRelayTransaction(const uint256& txid) = 0; virtual void PeerRelayDSQ(const CCoinJoinQueue& queue) = 0; virtual void PeerRelayRecoveredSig(const llmq::CRecoveredSig& sig, bool proactive_relay) = 0; - virtual void PeerAskPeersForTransaction(const uint256& txid) = 0; + /** Ask a few peers for an object we want but have not been offered, by registering a synthetic + * announcement with the request tracker. The tracker then owns the fetch: GETDATA scheduling, + * per-peer in-flight limits, expiry, and fallback to the next candidate. + * + * Candidates are explicit_peer, if set, plus peers whose known-inventory filter already contains + * the hash. That filter is only consulted for peers that enabled transaction relay, so for an + * object type carried outside transaction relay -- and for any object nobody has announced to + * us -- explicit_peer may be the only candidate. Pass it whenever a specific peer demonstrably + * has the object without having announced it, such as one that sent a vote naming this parent. + * The request tracker chooses among the candidates; explicit_peer does not imply request order. + * + * Requires ::cs_main is NOT held. */ + virtual void PeerAskPeersForObject(const CInv& inv, NodeId explicit_peer = -1) = 0; virtual size_t PeerGetRequestedObjectCount(NodeId nodeid) const = 0; virtual void PeerPostProcessMessage(MessageProcessingResult&& ret) = 0; }; diff --git a/src/test/governance_inv_tests.cpp b/src/test/governance_inv_tests.cpp index 2d4caef5e95b..edec08a64bc9 100644 --- a/src/test/governance_inv_tests.cpp +++ b/src/test/governance_inv_tests.cpp @@ -498,7 +498,7 @@ BOOST_AUTO_TEST_CASE(orphan_votes_require_a_valid_masternode_signature) connman.FlushSendBuffer(*peer); ProcessGovernanceVote(net_gov, *peer, vote); - BOOST_CHECK(m_node.govman->GetOrphanVoteObjectHashes().empty()); + BOOST_CHECK_EQUAL(m_node.govman->GetOrphanVoteCount(), 0U); BOOST_CHECK_EQUAL(CountQueuedMessages(*peer, NetMsgType::MNGOVERNANCESYNC), 0U); AssertMisbehaviorScore(*m_node.peerman, *peer, 20); diff --git a/src/test/governance_vote_processing_tests.cpp b/src/test/governance_vote_processing_tests.cpp index 57c05c103b8f..4798b27a73fa 100644 --- a/src/test/governance_vote_processing_tests.cpp +++ b/src/test/governance_vote_processing_tests.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -15,6 +16,7 @@ #include #include #include +#include #include #include #include