tx relay v2: rework tx request management and relay logic

    Implement new relay logic with threshold-based peer dropping, single-peer
    tx hash requests, per-connection in-flight limits with queuing, and
    multiindex-based request tracking. Consolidate add/remove request paths,
    pass tx hashes directly into send_txs/relay_txs, and add unit tests for
    the request manager.

    Co-authored-by: j-berman <justinberman@protonmail.com>
    Co-authored-by: Boog900 <boog900@tutanota.com>

Author: navidR

contrib/epee/include/math_helper.h

@@ -27,6 +27,8 @@
 #pragma once
 
 #include <cstdint> // uint64_t
+#include <cstdlib> // NULL
+#include <utility>
 
 #ifdef _WIN32
 #include <sysinfoapi.h> // GetSystemTimeAsFileTime
@@ -41,6 +43,7 @@ namespace math_helper
 	template<typename get_interval, bool start_immediate = true>
 	class once_a_time
 	{
+    get_interval m_get_interval;
     uint64_t get_time() const
     {
 #ifdef _WIN32
@@ -61,11 +64,12 @@ namespace math_helper
 
     void set_next_interval()
     {
-      m_interval = get_interval()();
+      m_interval = m_get_interval();
     }
 
 	public:
-		once_a_time()
+    once_a_time(get_interval gi = get_interval())
+    : m_get_interval(std::move(gi))
 		{
 			m_last_worked_time = 0;
       if(!start_immediate)

contrib/epee/include/net/http_protocol_handler.h

@@ -36,6 +36,7 @@
 #include "net_utils_base.h"
 #include "http_auth.h"
 #include "http_base.h"
+#include "syncobj.h"
 
 #undef MONERO_DEFAULT_LOG_CATEGORY
 #define MONERO_DEFAULT_LOG_CATEGORY "net.http"

contrib/epee/include/net/levin_protocol_handler_async.h

@@ -40,6 +40,8 @@
 #include "time_helper.h"
 #include "int-util.h"
 
+#include "cryptonote_basic/connection_context.h"
+
 #include <random>
 #include <chrono>
 
@@ -480,7 +482,7 @@ class async_protocol_handler
 
             buff_to_invoke = {buff_to_invoke.data(), std::size_t(inner_size)};
 
-            const size_t max_bytes = m_connection_context.get_max_bytes(m_current_head.m_command);
+            const size_t max_bytes = cryptonote::get_command_max_bytes(m_current_head.m_command);
             if(buff_to_invoke.size() > std::min<size_t>(max_packet_size, max_bytes))
             {
               MERROR(m_connection_context << "Maximum packet size exceed!, m_max_packet_size = " << std::min<size_t>(max_packet_size, max_bytes)
@@ -580,7 +582,7 @@ class async_protocol_handler
           m_cache_in_buffer.erase(sizeof(bucket_head2));
           m_state = stream_state_body;
           m_oponent_protocol_ver = m_current_head.m_protocol_version;
-          const size_t max_bytes = m_connection_context.get_max_bytes(m_current_head.m_command);
+          const size_t max_bytes = cryptonote::get_command_max_bytes(m_current_head.m_command);
           if(m_current_head.m_cb > std::min<size_t>(max_packet_size, max_bytes))
           {
             LOG_ERROR_CC(m_connection_context, "Maximum packet size exceed!, m_max_packet_size = " << std::min<size_t>(max_packet_size, max_bytes)

src/blockchain_utilities/blockchain_import.cpp

@@ -175,7 +175,8 @@ int check_flush(cryptonote::core &core, std::vector<block_complete_entry> &block
       tx_verification_context tvc = AUTO_VAL_INIT(tvc);
       CHECK_AND_ASSERT_THROW_MES(tx_blob.prunable_hash == crypto::null_hash,
         "block entry must not contain pruned txs");
-      core.handle_incoming_tx(tx_blob.blob, tvc, relay_method::block, true);
+      crypto::hash txid;
+      core.handle_incoming_tx(tx_blob.blob, tvc, relay_method::block, true, txid);
       if(tvc.m_verifivation_failed)
       {
         cryptonote::transaction transaction;

src/cryptonote_basic/connection_context.cpp

@@ -30,47 +30,9 @@
 #include "connection_context.h"
 
 #include <boost/optional/optional.hpp>
-#include "cryptonote_protocol/cryptonote_protocol_defs.h"
-#include "p2p/p2p_protocol_defs.h"
 
 namespace cryptonote
 {
-  std::size_t cryptonote_connection_context::get_max_bytes(const int command) noexcept
-  {
-    switch (command)
-    {
-    case nodetool::COMMAND_HANDSHAKE_T<cryptonote::CORE_SYNC_DATA>::ID:
-      return 65536;
-    case nodetool::COMMAND_TIMED_SYNC_T<cryptonote::CORE_SYNC_DATA>::ID:
-      return 65536;
-    case nodetool::COMMAND_PING::ID:
-      return 4096;
-    case nodetool::COMMAND_REQUEST_SUPPORT_FLAGS::ID:
-      return 4096;
-    case cryptonote::NOTIFY_NEW_BLOCK::ID:
-      return 1024 * 1024 * 128; // 128 MB (max packet is a bit less than 100 MB though)
-    case cryptonote::NOTIFY_NEW_TRANSACTIONS::ID:
-      return 1024 * 1024 * 128; // 128 MB (max packet is a bit less than 100 MB though)
-    case cryptonote::NOTIFY_REQUEST_GET_OBJECTS::ID:
-      return 1024 * 1024 * 2; // 2 MB
-    case cryptonote::NOTIFY_RESPONSE_GET_OBJECTS::ID:
-      return 1024 * 1024 * 128; // 128 MB (max packet is a bit less than 100 MB though)
-    case cryptonote::NOTIFY_REQUEST_CHAIN::ID:
-      return 512 * 1024; // 512 kB
-    case cryptonote::NOTIFY_RESPONSE_CHAIN_ENTRY::ID:
-      return 1024 * 1024 * 4; // 4 MB
-    case cryptonote::NOTIFY_NEW_FLUFFY_BLOCK::ID:
-      return 1024 * 1024 * 4; // 4 MB, but it does not includes transaction data
-    case cryptonote::NOTIFY_REQUEST_FLUFFY_MISSING_TX::ID:
-      return 1024 * 1024; // 1 MB
-    case cryptonote::NOTIFY_GET_TXPOOL_COMPLEMENT::ID:
-      return 1024 * 1024 * 4; // 4 MB
-    default:
-      break;
-    };
-    return std::numeric_limits<size_t>::max();
-  }
-
   void cryptonote_connection_context::set_state_normal()
   {
     m_state = state_normal;

src/cryptonote_basic/connection_context.h

@@ -32,14 +32,59 @@
 #pragma once
 #include <unordered_set>
 #include <atomic>
-#include <algorithm>
+#include <memory>
+#include <shared_mutex>
 #include <boost/date_time/posix_time/posix_time.hpp>
 #include <boost/optional/optional_fwd.hpp>
-#include "net/net_utils_base.h"
 #include "crypto/hash.h"
+#include "cryptonote_protocol/cryptonote_protocol_defs.h"
+#include "net/net_utils_base.h"
+#include "p2p/p2p_protocol_defs.h"
+#include "syncobj.h"
 
 namespace cryptonote
 {
+  //! \return Maximum number of bytes permissible for `command`.
+  constexpr std::size_t get_command_max_bytes(const int command) noexcept
+  {
+    switch (command)
+    {
+    case nodetool::COMMAND_HANDSHAKE_T<cryptonote::CORE_SYNC_DATA>::ID:
+      return 65536;
+    case nodetool::COMMAND_TIMED_SYNC_T<cryptonote::CORE_SYNC_DATA>::ID:
+      return 65536;
+    case nodetool::COMMAND_PING::ID:
+      return 4096;
+    case nodetool::COMMAND_REQUEST_SUPPORT_FLAGS::ID:
+      return 4096;
+    case cryptonote::NOTIFY_NEW_BLOCK::ID:
+      return 1024 * 1024 * 128; // 128 MB (max packet is a bit less than 100 MB though)
+    case cryptonote::NOTIFY_NEW_TRANSACTIONS::ID:
+      return 1024 * 1024 * 128; // 128 MB (max packet is a bit less than 100 MB though)
+    case cryptonote::NOTIFY_REQUEST_GET_OBJECTS::ID:
+      return 1024 * 1024 * 2; // 2 MB
+    case cryptonote::NOTIFY_RESPONSE_GET_OBJECTS::ID:
+      return 1024 * 1024 * 128; // 128 MB (max packet is a bit less than 100 MB though)
+    case cryptonote::NOTIFY_REQUEST_CHAIN::ID:
+      return 512 * 1024; // 512 kB
+    case cryptonote::NOTIFY_RESPONSE_CHAIN_ENTRY::ID:
+      return 1024 * 1024 * 4; // 4 MB
+    case cryptonote::NOTIFY_NEW_FLUFFY_BLOCK::ID:
+      return 1024 * 1024 * 128; // 128 MB (max packet is a bit less than 100 MB though, fluffy blocks can be full)
+    case cryptonote::NOTIFY_REQUEST_FLUFFY_MISSING_TX::ID:
+      return 1024 * 1024; // 1 MB
+    case cryptonote::NOTIFY_GET_TXPOOL_COMPLEMENT::ID:
+      return 1024 * 1024 * 4; // 4 MB
+    case cryptonote::NOTIFY_TX_POOL_HASH::ID:
+      return 1024 * 1024 * 2; // 2 MB
+    case cryptonote::NOTIFY_REQUEST_TX_POOL_TXS::ID:
+      return 1024 * 1024 * 2; // 2 MB
+    default:
+      break;
+    };
+    return std::numeric_limits<size_t>::max();
+  }
+
   struct cryptonote_connection_context: public epee::net_utils::connection_context_base
   {
     cryptonote_connection_context(): m_state(state_before_handshake), m_remote_blockchain_height(0), m_last_response_height(0),
@@ -89,9 +134,6 @@ namespace cryptonote
     static constexpr int handshake_command() noexcept { return 1001; }
     bool handshake_complete() const noexcept { return m_state != state_before_handshake; }
 
-    //! \return Maximum number of bytes permissible for `command`.
-    static size_t get_max_bytes(int command) noexcept;
-
     //! Use this instead of `m_state = state_normal`.
     void set_state_normal();
 

src/cryptonote_config.h

@@ -145,6 +145,7 @@
 #define P2P_DEFAULT_PEERS_IN_HANDSHAKE                  250
 #define P2P_MAX_PEERS_IN_HANDSHAKE                      250
 #define P2P_DEFAULT_CONNECTION_TIMEOUT                  5000       //5 seconds
+#define P2P_DEFAULT_REQUEST_TIMEOUT                     (P2P_DEFAULT_CONNECTION_TIMEOUT*6) // 30 seconds
 #define P2P_DEFAULT_SOCKS_CONNECT_TIMEOUT               45         // seconds
 #define P2P_DEFAULT_PING_CONNECTION_TIMEOUT             2000       //2 seconds
 #define P2P_DEFAULT_INVOKE_TIMEOUT                      60*2*1000  //2 minutes
@@ -154,14 +155,17 @@
 #define P2P_DEFAULT_SYNC_SEARCH_CONNECTIONS_COUNT       2
 #define P2P_DEFAULT_LIMIT_RATE_UP                       8192       // kB/s
 #define P2P_DEFAULT_LIMIT_RATE_DOWN                     32768       // kB/s
+#define P2P_REQUEST_FAILURE_THRESHOLD_PERCENTAGE        70          // if more than 70% of requests fail, the peer is dropped
+#define P2P_MIN_SAMPLE_SIZE_FOR_DROPPING                5          // minimum number of requests to consider dropping a peer for failed requests
 
 #define P2P_FAILED_ADDR_FORGET_SECONDS                  (60*60)     //1 hour
 #define P2P_IP_BLOCKTIME                                (60*60*24)  //24 hour
 #define P2P_IP_FAILS_BEFORE_BLOCK                       10
 #define P2P_IDLE_CONNECTION_KILL_INTERVAL               (5*60) //5 minutes
 
 #define P2P_SUPPORT_FLAG_FLUFFY_BLOCKS                  0x01
-#define P2P_SUPPORT_FLAGS                               P2P_SUPPORT_FLAG_FLUFFY_BLOCKS
+#define P2P_SUPPORT_FLAG_TX_RELAY_V2                    0x02
+#define P2P_SUPPORT_FLAGS                               (P2P_SUPPORT_FLAG_FLUFFY_BLOCKS | P2P_SUPPORT_FLAG_TX_RELAY_V2)
 
 #define RPC_IP_FAILS_BEFORE_BLOCK                       3
 

src/cryptonote_core/cryptonote_core.cpp

@@ -135,7 +135,6 @@ namespace cryptonote
     "sync-pruned-blocks"
   , "Allow syncing from nodes with only pruned blocks"
   };
-
   static const command_line::arg_descriptor<bool> arg_test_drop_download = {
     "test-drop-download"
   , "For net tests: in download, discard ALL blocks instead checking/saving them (very fast)"
@@ -821,7 +820,7 @@ namespace cryptonote
     return false;
   }
   //-----------------------------------------------------------------------------------------------
-  bool core::handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, relay_method tx_relay, bool relayed)
+  bool core::handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, relay_method tx_relay, bool relayed, crypto::hash& txid)
   {
     tvc = {};
 
@@ -838,7 +837,6 @@ namespace cryptonote
     }
 
     transaction tx;
-    crypto::hash txid;
     if (!parse_and_validate_tx_from_blob(tx_blob, tx, txid))
     {
       LOG_PRINT_L1("Incoming transactions failed to parse, rejected");
@@ -1173,6 +1171,9 @@ namespace cryptonote
       NOTIFY_NEW_TRANSACTIONS::request public_req{};
       NOTIFY_NEW_TRANSACTIONS::request private_req{};
       NOTIFY_NEW_TRANSACTIONS::request stem_req{};
+      std::vector<crypto::hash> public_tx_hashes{};
+      std::vector<crypto::hash> private_tx_hashes{};
+      std::vector<crypto::hash> stem_tx_hashes{};
       for (auto& tx : txs)
       {
         switch (std::get<2>(tx))
@@ -1182,14 +1183,17 @@ namespace cryptonote
             break;
           case relay_method::local:
             private_req.txs.push_back(std::move(std::get<1>(tx)));
+            private_tx_hashes.push_back(std::move(std::get<0>(tx)));
             break;
           case relay_method::forward:
             stem_req.txs.push_back(std::move(std::get<1>(tx)));
+            stem_tx_hashes.push_back(std::move(std::get<0>(tx)));
             break;
           case relay_method::block:
           case relay_method::fluff:
           case relay_method::stem:
             public_req.txs.push_back(std::move(std::get<1>(tx)));
+            public_tx_hashes.push_back(std::move(std::get<0>(tx)));
             break;
         }
       }
@@ -1200,11 +1204,11 @@ namespace cryptonote
          re-relaying public and private _should_ be acceptable here. */
       const boost::uuids::uuid source = boost::uuids::nil_uuid();
       if (!public_req.txs.empty())
-        get_protocol()->relay_transactions(public_req, source, epee::net_utils::zone::public_, relay_method::fluff);
+        get_protocol()->relay_transactions(public_req, std::move(public_tx_hashes), source, epee::net_utils::zone::public_, relay_method::fluff);
       if (!private_req.txs.empty())
-        get_protocol()->relay_transactions(private_req, source, epee::net_utils::zone::invalid, relay_method::local);
+        get_protocol()->relay_transactions(private_req, std::move(private_tx_hashes), source, epee::net_utils::zone::invalid, relay_method::local);
       if (!stem_req.txs.empty())
-        get_protocol()->relay_transactions(stem_req, source, epee::net_utils::zone::public_, relay_method::stem);
+        get_protocol()->relay_transactions(stem_req, std::move(stem_tx_hashes), source, epee::net_utils::zone::public_, relay_method::stem);
     }
     return true;
   }
@@ -1918,9 +1922,9 @@ namespace cryptonote
     m_blockchain_storage.flush_invalid_blocks();
   }
   //-----------------------------------------------------------------------------------------------
-  bool core::get_txpool_complement(std::vector<crypto::hash> hashes, std::vector<cryptonote::blobdata> &txes)
+  bool core::get_txpool_complement(std::vector<crypto::hash> hashes, std::vector<crypto::hash> &inv_txes)
   {
-    return m_mempool.get_complement(std::move(hashes), txes);
+    return m_mempool.get_complement(std::move(hashes), inv_txes);
   }
   //-----------------------------------------------------------------------------------------------
   bool core::update_blockchain_pruning()

src/cryptonote_core/cryptonote_core.h

@@ -123,10 +123,11 @@ namespace cryptonote
       * @param tvc metadata about the transaction's validity
       * @param tx_relay how the transaction was received
       * @param relayed whether or not the transaction was relayed to us
+      * @param txid return by reference
       *
       * @return true if the transaction was accepted, false otherwise
       */
-     bool handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, relay_method tx_relay, bool relayed);
+     bool handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, relay_method tx_relay, bool relayed, crypto::hash& txid);
 
     /**
       * @brief handles a single incoming block
@@ -905,13 +906,13 @@ namespace cryptonote
      void flush_invalid_blocks();
 
      /**
-      * @brief returns the set of transactions in the txpool which are not in the argument
+      * @brief returns the set of transaction hashes in the txpool which are not in the argument
       *
       * @param hashes hashes of transactions to exclude from the result
       *
       * @return true iff success, false otherwise
       */
-     bool get_txpool_complement(std::vector<crypto::hash> hashes, std::vector<cryptonote::blobdata> &txes);
+     bool get_txpool_complement(std::vector<crypto::hash> hashes, std::vector<crypto::hash> &inv_txes);
 
      /**
       * @brief validates some simple properties of a transaction

src/cryptonote_core/tx_pool.cpp

@@ -684,43 +684,23 @@ namespace cryptonote
     return true;
   }
   //---------------------------------------------------------------------------------
-  bool tx_memory_pool::get_complement(std::vector<crypto::hash> hashes, std::vector<cryptonote::blobdata> &txes) const
+  bool tx_memory_pool::get_complement(std::vector<crypto::hash> hashes, std::vector<crypto::hash> &inv_txes) const
   {
     CRITICAL_REGION_LOCAL(m_transactions_lock);
     CRITICAL_REGION_LOCAL1(m_blockchain);
 
     // Sort so we can do binary search later
     std::sort(hashes.begin(), hashes.end());
 
-    m_blockchain.for_all_txpool_txes([this, &hashes, &txes](const crypto::hash &txid, const txpool_tx_meta_t &meta, const cryptonote::blobdata_ref*) {
-      const auto tx_relay_method = meta.get_relay_method();
-      if (tx_relay_method != relay_method::block && tx_relay_method != relay_method::fluff)
-        return true;
-
+    m_blockchain.for_all_txpool_txes([this, &hashes, &inv_txes](const crypto::hash &txid, const txpool_tx_meta_t &meta, const cryptonote::blobdata_ref*) {
       // Do binary search for our pool TXID in given list, skip to next if already present
       const auto hash_it = std::lower_bound(hashes.cbegin(), hashes.cend(), txid);
       if (hash_it != hashes.cend() && *hash_it == txid)
         return true;
 
-      {
-        cryptonote::blobdata bd;
-        try
-        {
-          if (!m_blockchain.get_txpool_tx_blob(txid, bd, cryptonote::relay_category::broadcasted))
-          {
-            MERROR("Failed to get blob for txpool transaction " << txid);
-            return true;
-          }
-          txes.emplace_back(std::move(bd));
-        }
-        catch (const std::exception &e)
-        {
-          MERROR("Failed to get blob for txpool transaction " << txid << ": " << e.what());
-          return true;
-        }
-      }
+      inv_txes.push_back(txid);
       return true;
-    }, false);
+    }, false, cryptonote::relay_category::broadcasted);
     return true;
   }
   //---------------------------------------------------------------------------------