fastsync working on rococo (#1164)

core/consensus/authority/impl/authority_manager_impl.cpp

@@ -699,10 +699,10 @@ namespace kagome::authority {
             return AuthorityUpdateObserverError::UNSUPPORTED_MESSAGE_TYPE;
           });
     } else {
-      SL_WARN(log_,
-              "Unknown consensus engine id in block {}: {}",
-              block,
-              message.consensus_engine_id.toString());
+      // SL_WARN(log_,
+      //         "Unknown consensus engine id in block {}: {}",
+      //         block,
+      //         message.consensus_engine_id.toString());
       return outcome::success();
     }
   }
@@ -878,7 +878,7 @@ namespace kagome::authority {
     count1 = 0;
     size_t count2 = 0;
     for (const auto &[hash, header] : boost::range::reverse(headers)) {
-      if (not(++count1 % 10000)) {
+      if (not(++count1 % 1000)) {
         SL_WARN(log_, "{} digests applied ({})", count1, count2);
         count2 = 0;
       }

core/consensus/babe/impl/babe_impl.cpp

@@ -316,6 +316,8 @@ namespace kagome::consensus::babe {
         [wp = weak_from_this(), announce = announce, peer_id](
             outcome::result<primitives::BlockInfo> block_res) mutable {
           if (auto self = wp.lock()) {
+            self->synchronizer_->endSync();
+
             if (block_res.has_error()) {
               return;
             }
@@ -326,8 +328,12 @@ namespace kagome::consensus::babe {
                   peer_id,
                   block,
                   common::Buffer(),
-                  [self](outcome::result<primitives::BlockInfo>
-                             block_res) mutable {
+                  [self, block = block, peer_id](
+                      outcome::result<primitives::BlockInfo>
+                          block_res) mutable {
+                    if (block_res.has_error()) {
+                      return;
+                    }
                     const auto &block = block_res.value();
                     SL_INFO(self->log_,
                             "Catching up is finished on block {}",
@@ -353,27 +359,34 @@ namespace kagome::consensus::babe {
     auto is_ran = synchronizer_->syncByBlockInfo(
         target_block,
         peer_id,
-        [wp = weak_from_this(),
-         bn = target_block.number](outcome::result<primitives::BlockInfo> res) {
+        [wp = weak_from_this(), bn = target_block.number, peer_id](
+            outcome::result<primitives::BlockInfo> res) {
           if (auto self = wp.lock()) {
+            self->synchronizer_->endSync();
             if (res.has_error()) {
               SL_DEBUG(self->log_,
-                       "Catching up to block #{} is failed: {}",
+                       "Catching up to block #{} on {} is failed: {}",
                        bn,
+                       peer_id.toBase58(),
                        res.error().message());
               return;
             }
 
-            SL_DEBUG(self->log_,
-                     "Catching up to block #{} is going (on block #{} now)",
-                     bn,
-                     res.value().number);
+            SL_DEBUG(
+                self->log_,
+                "Catching up to block #{} on {} is going (on block #{} now)",
+                bn,
+                peer_id.toBase58(),
+                res.value().number);
           }
         },
         false);
 
     if (is_ran) {
-      SL_VERBOSE(log_, "Catching up to block #{} is ran", target_block.number);
+      SL_VERBOSE(log_,
+                 "Catching up to block #{} on {} is ran",
+                 target_block.number,
+                 peer_id.toBase58());
       current_state_ = State::CATCHING_UP;
     }
   }

core/consensus/grandpa/impl/environment_impl.cpp

@@ -94,10 +94,16 @@ namespace kagome::consensus::grandpa {
       const libp2p::peer::PeerId &peer_id,
       MembershipCounter set_id,
       RoundNumber round_number) {
-    SL_DEBUG(
-        logger_, "Send Catch-Up-Request beginning with round {}", round_number);
     network::CatchUpRequest message{.round_number = round_number,
                                     .voter_set_id = set_id};
+    if (not recent_catchup_requests_.emplace(message.fingerprint()).second) {
+      SL_ERROR(logger_,
+               "Don't Send Duplicate Catch-Up-Request beginning with round {}",
+               round_number);
+      return outcome::success();
+    }
+    SL_DEBUG(
+        logger_, "Send Catch-Up-Request beginning with round {}", round_number);
     transmitter_->sendCatchUpRequest(peer_id, std::move(message));
     return outcome::success();
   }

core/consensus/grandpa/impl/environment_impl.hpp

@@ -105,6 +105,8 @@ namespace kagome::consensus::grandpa {
     std::shared_ptr<network::GrandpaTransmitter> transmitter_;
     std::weak_ptr<JustificationObserver> justification_observer_;
 
+    std::set<std::size_t> recent_catchup_requests_;
+
     log::Logger logger_;
   };
 

core/injector/application_injector.cpp

@@ -94,7 +94,7 @@
 #include "network/impl/router_libp2p.hpp"
 #include "network/impl/state_protocol_observer_impl.hpp"
 #include "network/impl/sync_protocol_observer_impl.hpp"
-#include "network/impl/synchronizer_impl.hpp"
+#include "network/impl/synchronizer_new_impl.hpp"
 #include "network/impl/transactions_transmitter_impl.hpp"
 #include "network/sync_protocol_observer.hpp"
 #include "offchain/impl/offchain_local_storage.hpp"
@@ -1239,7 +1239,7 @@ namespace {
           static auto last_finalized_hash = get_last_finalized_hash(injector);
           return get_babe_configuration(last_finalized_hash, babe_api);
         }),
-        di::bind<network::Synchronizer>.template to<network::SynchronizerImpl>(),
+        di::bind<network::Synchronizer>.template to<network::SynchronizerNewImpl>(),
         di::bind<consensus::grandpa::Environment>.template to<consensus::grandpa::EnvironmentImpl>(),
         di::bind<consensus::BlockValidator>.template to<consensus::BabeBlockValidator>(),
         di::bind<crypto::EcdsaProvider>.template to<crypto::EcdsaProviderImpl>(),

core/network/impl/CMakeLists.txt

@@ -6,7 +6,7 @@
 add_subdirectory(protocols)
 
 add_library(synchronizer
-    synchronizer_impl.cpp
+    synchronizer_new_impl.cpp
     )
 target_link_libraries(synchronizer
     logger