fix: respect SENDDSQUEUE message, move DSQ relay into net processing / peerman

PastaPastaPasta · PastaPastaPasta · commit dafa7363a38f · 2024-11-23T00:01:11.000-06:00
in 6148, I broke the functionality where a peer must opt in / opt out of DSQUEUE messages. This was mostly ok, and not immediately detected, as with this bug, simply everyone would receive DSQ messages over inventory (or classically, old proto versions were not affected by this bug). But this still would result in quite a bit of wasted bandwidth for peers which may not care about DSQ at all.

This commit should restore the prior functionality, where a node should send the SENDDSQUEUE message if they wish to receive DSQs. Once they've sent that, depending on their protocol version, they will either have the messages pushed to them as available, or on modern protocols, they will thereafter receive DSQs over the inventory system.

NOTE: I also refactor the code in this commit, moving some network proccessing into.... wait for it... net_processing.cpp! This allowed us to remove some dependencies in coinjoin.h. DSQ messages are now relayed to peers by calling peer_manager.RelayDSQ
diff --git a/src/coinjoin/client.cpp b/src/coinjoin/client.cpp
@@ -132,7 +132,7 @@ PeerMsgRet CCoinJoinClientQueueManager::ProcessDSQueue(const CNode& peer, CDataS
             WITH_LOCK(cs_vecqueue, vecCoinJoinQueue.push_back(dsq));
         }
     } // cs_ProcessDSQueue
-    dsq.Relay(connman, *peerman);
+    peerman->RelayDSQ(dsq);
     return {};
 }
 
diff --git a/src/coinjoin/client.h b/src/coinjoin/client.h
@@ -28,6 +28,7 @@ class CMasternodeSync;
 class CNode;
 class CoinJoinWalletManager;
 class CTxMemPool;
+class PeerManager;
 
 class UniValue;
 
diff --git a/src/coinjoin/coinjoin.cpp b/src/coinjoin/coinjoin.cpp
@@ -14,7 +14,6 @@
 #include <masternode/node.h>
 #include <masternode/sync.h>
 #include <messagesigner.h>
-#include <net_processing.h>
 #include <netmessagemaker.h>
 #include <txmempool.h>
 #include <util/moneystr.h>
@@ -71,19 +70,6 @@ bool CCoinJoinQueue::CheckSignature(const CBLSPublicKey& blsPubKey) const
     return true;
 }
 
-bool CCoinJoinQueue::Relay(CConnman& connman, PeerManager& peerman)
-{
-    CInv inv(MSG_DSQ, GetHash());
-    peerman.RelayInv(inv, DSQ_INV_VERSION);
-    connman.ForEachNode([&connman, this](CNode* pnode) {
-        CNetMsgMaker msgMaker(pnode->GetCommonVersion());
-        if (pnode->fSendDSQueue && pnode->nVersion < DSQ_INV_VERSION) {
-            connman.PushMessage(pnode, msgMaker.Make(NetMsgType::DSQUEUE, (*this)));
-        }
-    });
-    return true;
-}
-
 bool CCoinJoinQueue::IsTimeOutOfBounds(int64_t current_time) const
 {
     return current_time - nTime > COINJOIN_QUEUE_TIMEOUT ||
diff --git a/src/coinjoin/coinjoin.h b/src/coinjoin/coinjoin.h
@@ -24,14 +24,12 @@
 
 class CActiveMasternodeManager;
 class CChainState;
-class CConnman;
 class CBLSPublicKey;
 class CBlockIndex;
 class ChainstateManager;
 class CMasternodeSync;
 class CTxMemPool;
 class TxValidationState;
-class PeerManager;
 
 namespace llmq {
 class CChainLocksHandler;
@@ -221,8 +219,6 @@ class CCoinJoinQueue
     /// Check if we have a valid Masternode address
     [[nodiscard]] bool CheckSignature(const CBLSPublicKey& blsPubKey) const;
 
-    bool Relay(CConnman& connman, PeerManager& peerman);
-
     /// Check if a queue is too old or too far into the future
     [[nodiscard]] bool IsTimeOutOfBounds(int64_t current_time = GetAdjustedTime()) const;
 
diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp
@@ -186,7 +186,7 @@ PeerMsgRet CCoinJoinServer::ProcessDSQUEUE(const CNode& peer, CDataStream& vRecv
         TRY_LOCK(cs_vecqueue, lockRecv);
         if (!lockRecv) return {};
         vecCoinJoinQueue.push_back(dsq);
-        dsq.Relay(connman, *m_peerman);
+        m_peerman->RelayDSQ(dsq);
     }
     return {};
 }
@@ -519,7 +519,7 @@ void CCoinJoinServer::CheckForCompleteQueue()
         LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckForCompleteQueue -- queue is ready, signing and relaying (%s) " /* Continued */
                                      "with %d participants\n", dsq.ToString(), vecSessionCollaterals.size());
         dsq.Sign(*m_mn_activeman);
-        dsq.Relay(connman, *m_peerman);
+        m_peerman->RelayDSQ(dsq);
     }
 }
 
@@ -732,7 +732,7 @@ bool CCoinJoinServer::CreateNewSession(const CCoinJoinAccept& dsa, PoolMessage&
                             GetAdjustedTime(), false);
         LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CreateNewSession -- signing and relaying new queue: %s\n", dsq.ToString());
         dsq.Sign(*m_mn_activeman);
-        dsq.Relay(connman, *m_peerman);
+        m_peerman->RelayDSQ(dsq);
         LOCK(cs_vecqueue);
         vecCoinJoinQueue.push_back(dsq);
     }
diff --git a/src/coinjoin/server.h b/src/coinjoin/server.h
@@ -11,6 +11,7 @@
 
 class CActiveMasternodeManager;
 class CCoinJoinServer;
+class CConnman;
 class CDataStream;
 class CDeterministicMNManager;
 class CDSTXManager;
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
@@ -368,6 +368,15 @@ struct Peer {
     /** Whether the peer has signaled support for receiving ADDRv2 (BIP155)
      *  messages, indicating a preference to receive ADDRv2 instead of ADDR ones. */
     std::atomic_bool m_wants_addrv2{false};
+
+    enum class WantsDSQ {
+        NONE, // Peer doesn't want DSQs
+        INV, // Peer will be notified of DSQs over Inventory System (see: DSQ_INV_VERSION)
+        ALL, // Peer will be notified of all DSQs, by simply sending them the DSQ
+    };
+
+    std::atomic<WantsDSQ> m_wants_dsq{WantsDSQ::NONE};
+
     /** Whether this peer has already sent us a getaddr message. */
     bool m_getaddr_recvd GUARDED_BY(NetEventsInterface::g_msgproc_mutex){false};
     /** Number of addresses that can be processed from this peer. Start at 1 to
@@ -610,6 +619,7 @@ class PeerManagerImpl final : public PeerManager
     void RelayInvFiltered(CInv &inv, const uint256 &relatedTxHash, const int minProtoVersion) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
     void RelayTransaction(const uint256& txid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
     void RelayRecoveredSig(const uint256& sigHash) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
+    void RelayDSQ(const CCoinJoinQueue& queue) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
     void SetBestHeight(int height) override { m_best_height = height; };
     void Misbehaving(const NodeId pnode, const int howmuch, const std::string& message = "") override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
     void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv,
@@ -2279,6 +2289,34 @@ void PeerManagerImpl::RelayInv(CInv &inv, const int minProtoVersion)
     });
 }
 
+void PeerManagerImpl::RelayDSQ(const CCoinJoinQueue& queue)
+{
+    CInv inv{MSG_DSQ, queue.GetHash()};
+    std::vector<NodeId> nodes_send_all;
+    {
+        LOCK(m_peer_mutex);
+        for (const auto& [nodeid, peer] : m_peer_map) {
+            switch (peer->m_wants_dsq) {
+            case Peer::WantsDSQ::NONE:
+                break;
+            case Peer::WantsDSQ::INV:
+                PushInv(*peer, inv);
+                break;
+            case Peer::WantsDSQ::ALL:
+                nodes_send_all.push_back(nodeid);
+                break;
+            }
+        }
+    }
+    for (auto nodeId : nodes_send_all) {
+        m_connman.ForNode(nodeId, [&](CNode* pnode) -> bool {
+            CNetMsgMaker msgMaker(pnode->GetCommonVersion());
+            m_connman.PushMessage(pnode, msgMaker.Make(NetMsgType::DSQUEUE, queue));
+            return true;
+        });
+    }
+}
+
 void PeerManagerImpl::RelayInvFiltered(CInv &inv, const CTransaction& relatedTx, const int minProtoVersion)
 {
     // TODO: Migrate to iteration through m_peer_map
@@ -3954,7 +3992,13 @@ void PeerManagerImpl::ProcessMessage(
     {
         bool b;
         vRecv >> b;
-        pfrom.fSendDSQueue = b;
+        if (!b) {
+            peer->m_wants_dsq = Peer::WantsDSQ::NONE;
+        } else if (pfrom.GetCommonVersion() < DSQ_INV_VERSION) {
+            peer->m_wants_dsq = Peer::WantsDSQ::ALL;
+        } else {
+            peer->m_wants_dsq = Peer::WantsDSQ::INV;
+        }
         return;
     }
 
diff --git a/src/net_processing.h b/src/net_processing.h
@@ -16,6 +16,7 @@
 class CActiveMasternodeManager;
 class AddrMan;
 class CTxMemPool;
+class CCoinJoinQueue;
 class CDeterministicMNManager;
 class CMasternodeMetaMan;
 class CMasternodeSync;
@@ -93,6 +94,9 @@ class PeerManager : public CValidationInterface, public NetEventsInterface
     /** Broadcast inventory message to a specific peer. */
     virtual void PushInventory(NodeId nodeid, const CInv& inv) = 0;
 
+    /** Relay DSQ based on peer preference */
+    virtual void RelayDSQ(const CCoinJoinQueue& queue) = 0;
+
     /** Relay inventories to all peers */
     virtual void RelayInv(CInv &inv, const int minProtoVersion = MIN_PEER_PROTO_VERSION) = 0;
     virtual void RelayInvFiltered(CInv &inv, const CTransaction &relatedTx,

Original file line number	Diff line number	Diff line change
`@@ -132,7 +132,7 @@ PeerMsgRet CCoinJoinClientQueueManager::ProcessDSQueue(const CNode& peer, CDataS`
`132`	`132`	`WITH_LOCK(cs_vecqueue, vecCoinJoinQueue.push_back(dsq));`
`133`	`133`	`}`
`134`	`134`	`} // cs_ProcessDSQueue`
`135`		`- dsq.Relay(connman, *peerman);`
	`135`	`+ peerman->RelayDSQ(dsq);`
`136`	`136`	`return {};`
`137`	`137`	`}`
`138`	`138`
Original file line number	Diff line number	Diff line change
`@@ -186,7 +186,7 @@ PeerMsgRet CCoinJoinServer::ProcessDSQUEUE(const CNode& peer, CDataStream& vRecv`
`186`	`186`	`TRY_LOCK(cs_vecqueue, lockRecv);`
`187`	`187`	`if (!lockRecv) return {};`
`188`	`188`	`vecCoinJoinQueue.push_back(dsq);`
`189`		`- dsq.Relay(connman, *m_peerman);`
	`189`	`+ m_peerman->RelayDSQ(dsq);`
`190`	`190`	`}`
`191`	`191`	`return {};`
`192`	`192`	`}`
`@@ -519,7 +519,7 @@ void CCoinJoinServer::CheckForCompleteQueue()`
`519`	`519`	`LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckForCompleteQueue -- queue is ready, signing and relaying (%s) " /* Continued */`
`520`	`520`	`"with %d participants\n", dsq.ToString(), vecSessionCollaterals.size());`
`521`	`521`	`dsq.Sign(*m_mn_activeman);`
`522`		`- dsq.Relay(connman, *m_peerman);`
	`522`	`+ m_peerman->RelayDSQ(dsq);`
`523`	`523`	`}`
`524`	`524`	`}`
`525`	`525`
`@@ -732,7 +732,7 @@ bool CCoinJoinServer::CreateNewSession(const CCoinJoinAccept& dsa, PoolMessage&`
`732`	`732`	`GetAdjustedTime(), false);`
`733`	`733`	`LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CreateNewSession -- signing and relaying new queue: %s\n", dsq.ToString());`
`734`	`734`	`dsq.Sign(*m_mn_activeman);`
`735`		`- dsq.Relay(connman, *m_peerman);`
	`735`	`+ m_peerman->RelayDSQ(dsq);`
`736`	`736`	`LOCK(cs_vecqueue);`
`737`	`737`	`vecCoinJoinQueue.push_back(dsq);`
`738`	`738`	`}`