Peer message flatbuffer parsing fixes. (#294)

src/msg/fbuf/p2pmsg_conversion.cpp

@@ -45,7 +45,7 @@ namespace msg::fbuf::p2pmsg
             const uint64_t time_now = util::get_epoch_milliseconds();
             if (p2p_msg->created_on() < (time_now - (conf::cfg.contract.roundtime * 4)))
             {
-                LOG_DEBUG << "Peer message is too old.";
+                LOG_DEBUG << "Peer message is too old. type:" << p2p_msg->content_type();
                 return p2p::peer_message_info{NULL, P2PMsgContent_NONE, 0};
             }
         }
@@ -224,7 +224,7 @@ namespace msg::fbuf::p2pmsg
         p2p::hpfs_log_response hpfs_log_response;
         hpfs_log_response.min_record_id = flatbuf_seqhash_to_seqhash(msg.min_record_id());
         hpfs_log_response.log_record_bytes.reserve(msg.log_record_bytes()->size());
-        for (const auto byte: *msg.log_record_bytes())
+        for (const auto byte : *msg.log_record_bytes())
             hpfs_log_response.log_record_bytes.push_back(byte);
         return hpfs_log_response;
     }

src/p2p/p2p.cpp

@@ -253,15 +253,15 @@ namespace p2p
             msg_type == p2pmsg::P2PMsgContent_NonUnlProposalMsg ||
             msg_type == p2pmsg::P2PMsgContent_NplMsg)
         {
-            return true;
-        }
+            // Checking the time to live of the message. The time to live for forwarding is three times the round time.
+            const uint64_t time_now = util::get_epoch_milliseconds();
+            if (originated_on < (time_now - (conf::cfg.contract.roundtime * 3)))
+            {
+                LOG_DEBUG << "Peer message is too old for forwarding. type:" << msg_type;
+                return false;
+            }
 
-        const uint64_t time_now = util::get_epoch_milliseconds();
-        // Checking the time to live of the message. The time to live for forwarding is three times the round time.
-        if (originated_on < (time_now - (conf::cfg.contract.roundtime * 3)))
-        {
-            LOG_DEBUG << "Peer message is too old for forwarding.";
-            return false;
+            return true;
         }
 
         return false;

src/p2p/peer_comm_server.cpp

@@ -211,6 +211,7 @@ namespace p2p
             {
                 is_weakly_connected = !is_weakly_connected;
                 send_peer_requirement_announcement(is_weakly_connected);
+                LOG_DEBUG << "Sent weakly connected announcement.";
             }
             connected_status_check_counter = 0;
         }

src/p2p/peer_session_handler.cpp

@@ -62,17 +62,13 @@ namespace p2p
         }
 
         const peer_message_info mi = p2pmsg::get_peer_message_info(message);
-
-        if (mi.type == p2pmsg::P2PMsgContent_NONE)
-        {
-            session.increment_metric(comm::SESSION_THRESHOLDS::MAX_BADMSGS_PER_MINUTE, 1);
-            LOG_DEBUG << "Received invalid peer message type. " << session.display_name();
+        if (!mi.p2p_msg) // Message buffer will be null if peer message was too old.
             return 0;
-        }
-        else if (!recent_peermsg_hashes.try_emplace(crypto::get_hash(message)))
+
+        if (!recent_peermsg_hashes.try_emplace(crypto::get_hash(message)))
         {
             session.increment_metric(comm::SESSION_THRESHOLDS::MAX_DUPMSGS_PER_MINUTE, 1);
-            LOG_DEBUG << "Duplicate peer message. " << session.display_name();
+            LOG_DEBUG << "Duplicate peer message. type:" << mi.type << " " << session.display_name();
             return 0;
         }
 
@@ -269,7 +265,7 @@ namespace p2p
         else
         {
             session.increment_metric(comm::SESSION_THRESHOLDS::MAX_BADMSGS_PER_MINUTE, 1);
-            LOG_DEBUG << "Received invalid peer message type. " << session.display_name();
+            LOG_DEBUG << "Received invalid peer message type [" << mi.type << "]. " << session.display_name();
         }
         return 0;
     }

src/sc/hpfs_log_sync.cpp

@@ -19,9 +19,6 @@ namespace sc::hpfs_log_sync
     // Max no. of repetitive reqeust resubmissions before abandoning the sync.
     constexpr uint16_t ABANDON_THRESHOLD = 10;
 
-    // No. of milliseconds to wait before resubmitting a request.
-    uint16_t REQUEST_RESUBMIT_TIMEOUT;
-
     sync_context sync_ctx;
     bool init_success = false;
 
@@ -33,8 +30,6 @@ namespace sc::hpfs_log_sync
      */
     int init()
     {
-        REQUEST_RESUBMIT_TIMEOUT = conf::cfg.contract.roundtime;
-
         sync_ctx.log_record_sync_thread = std::thread(hpfs_log_syncer_loop);
 
         genesis_seq_hash = {ledger::genesis.seq_no, hpfs::get_root_hash(ledger::genesis.config_hash, ledger::genesis.state_hash)};
@@ -143,10 +138,13 @@ namespace sc::hpfs_log_sync
      */
     void send_hpfs_log_sync_request()
     {
+        // No. of milliseconds to wait before resubmitting a request.
+        const uint32_t request_resubmit_timeout = hpfs::get_request_resubmit_timeout();
+
         // Check whether we need to send any requests or abandon the sync due to timeout.
         const uint64_t time_now = util::get_epoch_milliseconds();
         if ((sync_ctx.target_requested_on == 0) ||                                // Initial request.
-            (time_now - sync_ctx.target_requested_on) > REQUEST_RESUBMIT_TIMEOUT) // Request resubmission.
+            (time_now - sync_ctx.target_requested_on) > request_resubmit_timeout) // Request resubmission.
         {
             if (sync_ctx.request_submissions < ABANDON_THRESHOLD)
             {
@@ -315,7 +313,6 @@ namespace sc::hpfs_log_sync
         // In sync. No need to sync.
         if (last_from_index == last_from_ledger)
             return 1;
-        
 
         if (last_from_index.seq_no == last_from_ledger.seq_no)
         {