refactor: reduce PeerManager usage in InstantSend worker functions

So far we cannot stop using `PeerManager` entirely the triggers are
internal (i.e. not coming from P2P interactions) but we can extend and
embrace MessageProcessingResult, to at least isolate its usage as
best we can.

Author: kwvg

src/instantsend/instantsend.cpp

@@ -172,7 +172,7 @@ MessageProcessingResult CInstantSendManager::ProcessMessageInstantSendLock(const
     return ret;
 }
 
-bool CInstantSendManager::ProcessPendingInstantSendLocks(PeerManager& peerman)
+bool CInstantSendManager::ProcessPendingInstantSendLocks(std::vector<std::pair<NodeId, MessageProcessingResult>>& peer_activity)
 {
     decltype(pendingInstantSendLocks) pend;
     bool fMoreWork{false};
@@ -217,7 +217,7 @@ bool CInstantSendManager::ProcessPendingInstantSendLocks(PeerManager& peerman)
     auto dkgInterval = llmq_params.dkgInterval;
 
     // First check against the current active set and don't ban
-    auto badISLocks = ProcessPendingInstantSendLocks(llmq_params, peerman, /*signOffset=*/0, pend, false);
+    auto badISLocks = ProcessPendingInstantSendLocks(llmq_params, /*signOffset=*/0, /*ban=*/false, pend, peer_activity);
     if (!badISLocks.empty()) {
         LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- doing verification on old active set\n", __func__);
 
@@ -230,16 +230,16 @@ bool CInstantSendManager::ProcessPendingInstantSendLocks(PeerManager& peerman)
             }
         }
         // Now check against the previous active set and perform banning if this fails
-        ProcessPendingInstantSendLocks(llmq_params, peerman, dkgInterval, pend, true);
+        ProcessPendingInstantSendLocks(llmq_params, dkgInterval, /*ban=*/true, pend, peer_activity);
     }
 
     return fMoreWork;
 }
 
 std::unordered_set<uint256, StaticSaltedHasher> CInstantSendManager::ProcessPendingInstantSendLocks(
-    const Consensus::LLMQParams& llmq_params, PeerManager& peerman, int signOffset,
+    const Consensus::LLMQParams& llmq_params, int signOffset, bool ban,
     const std::unordered_map<uint256, std::pair<NodeId, instantsend::InstantSendLockPtr>, StaticSaltedHasher>& pend,
-    bool ban)
+    std::vector<std::pair<NodeId, MessageProcessingResult>>& peer_activity)
 {
     CBLSBatchVerifier<NodeId, uint256> batchVerifier(false, true, 8);
     std::unordered_map<uint256, CRecoveredSig, StaticSaltedHasher> recSigs;
@@ -315,7 +315,7 @@ std::unordered_set<uint256, StaticSaltedHasher> CInstantSendManager::ProcessPend
         for (const auto& nodeId : batchVerifier.badSources) {
             // Let's not be too harsh, as the peer might simply be unlucky and might have sent us an old lock which
             // does not validate anymore due to changed quorums
-            peerman.Misbehaving(nodeId, 20);
+            peer_activity.push_back({nodeId, MisbehavingError{20}});
         }
     }
     for (const auto& p : pend) {
@@ -330,7 +330,7 @@ std::unordered_set<uint256, StaticSaltedHasher> CInstantSendManager::ProcessPend
             continue;
         }
 
-        ProcessInstantSendLock(nodeId, peerman, hash, islock);
+        peer_activity.push_back({nodeId, ProcessInstantSendLock(nodeId, hash, islock)});
 
         // See comment further on top. We pass a reconstructed recovered sig to the signing manager to avoid
         // double-verification of the sig.
@@ -348,8 +348,8 @@ std::unordered_set<uint256, StaticSaltedHasher> CInstantSendManager::ProcessPend
     return badISLocks;
 }
 
-void CInstantSendManager::ProcessInstantSendLock(NodeId from, PeerManager& peerman, const uint256& hash,
-                                                 const instantsend::InstantSendLockPtr& islock)
+MessageProcessingResult CInstantSendManager::ProcessInstantSendLock(NodeId from, const uint256& hash,
+                                                                    const instantsend::InstantSendLockPtr& islock)
 {
     LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- txid=%s, islock=%s: processing islock, peer=%d\n",
              __func__, islock->txid.ToString(), hash.ToString(), from);
@@ -358,12 +358,12 @@ void CInstantSendManager::ProcessInstantSendLock(NodeId from, PeerManager& peerm
         m_signer->ClearLockFromQueue(islock);
     }
     if (db.KnownInstantSendLock(hash)) {
-        return;
+        return {};
     }
 
     if (const auto sameTxIsLock = db.GetInstantSendLockByTxid(islock->txid)) {
         // can happen, nothing to do
-        return;
+        return {};
     }
     for (const auto& in : islock->inputs) {
         const auto sameOutpointIsLock = db.GetInstantSendLockByInput(in);
@@ -386,7 +386,7 @@ void CInstantSendManager::ProcessInstantSendLock(NodeId from, PeerManager& peerm
         if (pindexMined != nullptr && clhandler.HasChainLock(pindexMined->nHeight, pindexMined->GetBlockHash())) {
             LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- txlock=%s, islock=%s: dropping islock as it already got a ChainLock in block %s, peer=%d\n", __func__,
                      islock->txid.ToString(), hash.ToString(), hashBlock.ToString(), from);
-            return;
+            return {};
         }
     }
 
@@ -417,15 +417,17 @@ void CInstantSendManager::ProcessInstantSendLock(NodeId from, PeerManager& peerm
         mempool.AddTransactionsUpdated(1);
     }
 
+    MessageProcessingResult ret{/*is_masternode*/m_signer != nullptr};
     CInv inv(MSG_ISDLOCK, hash);
     if (found_transaction) {
-        peerman.RelayInvFiltered(inv, *tx, ISDLOCK_PROTO_VERSION);
+        ret.m_inv_filter = std::make_pair(inv, tx);
     } else {
         // we don't have the TX yet, so we only filter based on txid. Later when that TX arrives, we will re-announce
         // with the TX taken into account.
-        peerman.RelayInvFiltered(inv, islock->txid, ISDLOCK_PROTO_VERSION);
-        peerman.AskPeersForTransaction(islock->txid, /*is_masternode=*/m_signer != nullptr);
+        ret.m_inv_filter = std::make_pair(inv, islock->txid);
+        ret.m_request_tx = islock->txid;
     }
+    return ret;
 }
 
 void CInstantSendManager::TransactionAddedToMempool(const CTransactionRef& tx)
@@ -925,7 +927,11 @@ void CInstantSendManager::WorkThreadMain(PeerManager& peerman)
     while (!workInterrupt) {
         bool fMoreWork = [&]() -> bool {
             if (!IsInstantSendEnabled()) return false;
-            const bool more_work{ProcessPendingInstantSendLocks(peerman)};
+            std::vector<std::pair<NodeId, MessageProcessingResult>> peer_activity{};
+            const bool more_work{ProcessPendingInstantSendLocks(peer_activity)};
+            for (const auto& [node_id, mpr] : peer_activity) {
+                peerman.PostProcessMessage(std::move(mpr), node_id);
+            }
             if (!m_signer) return more_work;
             // Construct set of non-locked transactions that are pending to retry
             std::vector<CTransactionRef> txns{};

src/instantsend/instantsend.h

@@ -98,14 +98,15 @@ class CInstantSendManager final : public instantsend::InstantSendSignerParent
 
 private:
     MessageProcessingResult ProcessMessageInstantSendLock(const NodeId& pfrom, const instantsend::InstantSendLockPtr& islock);
-    bool ProcessPendingInstantSendLocks(PeerManager& peerman)
+    bool ProcessPendingInstantSendLocks(std::vector<std::pair<NodeId, MessageProcessingResult>>& peer_activity)
         EXCLUSIVE_LOCKS_REQUIRED(!cs_nonLocked, !cs_pendingLocks, !cs_pendingRetry);
 
     std::unordered_set<uint256, StaticSaltedHasher> ProcessPendingInstantSendLocks(
-        const Consensus::LLMQParams& llmq_params, PeerManager& peerman, int signOffset,
-        const std::unordered_map<uint256, std::pair<NodeId, instantsend::InstantSendLockPtr>, StaticSaltedHasher>& pend, bool ban)
+        const Consensus::LLMQParams& llmq_params, int signOffset, bool ban,
+        const std::unordered_map<uint256, std::pair<NodeId, instantsend::InstantSendLockPtr>, StaticSaltedHasher>& pend,
+        std::vector<std::pair<NodeId, MessageProcessingResult>>& peer_activity)
         EXCLUSIVE_LOCKS_REQUIRED(!cs_nonLocked, !cs_pendingLocks, !cs_pendingRetry);
-    void ProcessInstantSendLock(NodeId from, PeerManager& peerman, const uint256& hash, const instantsend::InstantSendLockPtr& islock)
+    MessageProcessingResult ProcessInstantSendLock(NodeId from, const uint256& hash, const instantsend::InstantSendLockPtr& islock)
         EXCLUSIVE_LOCKS_REQUIRED(!cs_nonLocked, !cs_pendingLocks, !cs_pendingRetry);
 
     void AddNonLockedTx(const CTransactionRef& tx, const CBlockIndex* pindexMined)

src/net_processing.cpp

@@ -651,7 +651,7 @@ class PeerManagerImpl final : public PeerManager
 
     /** Helpers to process result of external handlers of message */
     void ProcessPeerMsgRet(const PeerMsgRet& ret, CNode& pfrom) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
-    void PostProcessMessage(MessageProcessingResult&& ret, NodeId node) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
+    void PostProcessMessage(const MessageProcessingResult&& ret, NodeId node) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
 
     /** Consider evicting an outbound peer based on the amount of time they've been behind our tip */
     void ConsiderEviction(CNode& pto, Peer& peer, std::chrono::seconds time_in_seconds) EXCLUSIVE_LOCKS_REQUIRED(cs_main, g_msgproc_mutex);
@@ -3488,7 +3488,7 @@ void PeerManagerImpl::ProcessPeerMsgRet(const PeerMsgRet& ret, CNode& pfrom)
     if (!ret) Misbehaving(pfrom.GetId(), ret.error().score, ret.error().message);
 }
 
-void PeerManagerImpl::PostProcessMessage(MessageProcessingResult&& result, NodeId node)
+void PeerManagerImpl::PostProcessMessage(const MessageProcessingResult&& result, NodeId node)
 {
     if (result.m_error) {
         Misbehaving(node, result.m_error->score, result.m_error->message);
@@ -3502,6 +3502,17 @@ void PeerManagerImpl::PostProcessMessage(MessageProcessingResult&& result, NodeI
     if (result.m_inventory) {
         RelayInv(result.m_inventory.value());
     }
+    if (result.m_inv_filter) {
+        const auto& [inv, filter] = result.m_inv_filter.value();
+        if (std::holds_alternative<CTransactionRef>(filter)) {
+            RelayInvFiltered(inv, *std::get<CTransactionRef>(filter), ISDLOCK_PROTO_VERSION);
+        } else {
+            RelayInvFiltered(inv, std::get<uint256>(filter), ISDLOCK_PROTO_VERSION);
+        }
+    }
+    if (result.m_request_tx) {
+        AskPeersForTransaction(result.m_request_tx.value(), result.m_is_masternode);
+    }
 }
 
 void PeerManagerImpl::ProcessMessage(

src/net_processing.h

@@ -132,7 +132,7 @@ class PeerManager : public CValidationInterface, public NetEventsInterface
                                 const std::chrono::microseconds time_received, const std::atomic<bool>& interruptMsgProc) EXCLUSIVE_LOCKS_REQUIRED(g_msgproc_mutex) = 0;
 
     /** Finish message processing. Used for some specific messages */
-    virtual void PostProcessMessage(MessageProcessingResult&& ret, NodeId node = -1) = 0;
+    virtual void PostProcessMessage(const MessageProcessingResult&& ret, NodeId node = -1) = 0;
 
     /** This function is used for testing the stale tip eviction logic, see denialofservice_tests.cpp */
     virtual void UpdateLastBlockAnnounceTime(NodeId node, int64_t time_in_seconds) = 0;

src/protocol.h

@@ -17,6 +17,7 @@
 #include <cstdint>
 #include <limits>
 #include <string>
+#include <variant>
 
 /** Message header.
  * (4) message start.
@@ -597,19 +598,27 @@ using PeerMsgRet = tl::expected<void, MisbehavingError>;
  */
 struct MessageProcessingResult
 {
+    bool m_is_masternode{false};
+
     //! @m_error triggers Misbehaving error with score and optional message if not nullopt
     std::optional<MisbehavingError> m_error;
 
     //! @m_inventory will relay this inventory to connected peers if not nullopt
     std::optional<CInv> m_inventory;
 
+    //! @m_inv_filter will relay this inventory if filter matches to connected peers if not nullopt
+    std::optional<std::pair<CInv, std::variant<CTransactionRef, uint256>>> m_inv_filter;
+
+    //! @m_request_tx will ask connected peers to relay transaction if not nullopt
+    std::optional<uint256> m_request_tx;
+
     //! @m_transactions will relay transactions to peers which is ready to accept it (some peers does not accept transactions)
     std::vector<uint256> m_transactions;
 
     //! @m_to_erase triggers EraseObjectRequest from PeerManager for this inventory if not nullopt
     std::optional<CInv> m_to_erase;
 
-    MessageProcessingResult() = default;
+    MessageProcessingResult(bool is_masternode = false) : m_is_masternode{is_masternode} {}
     MessageProcessingResult(MisbehavingError error) :
         m_error(error)
     {}