Merge branch 'master' into test/polkadot-functional-zombietests

cmake/Hunter/config.cmake

@@ -40,7 +40,7 @@ hunter_config(
 
 hunter_config(
     libp2p
-    VERSION 0.1.20
+    VERSION 0.1.21
     KEEP_PACKAGE_SOURCES
 )
 

cmake/Hunter/hunter-gate-url.cmake

@@ -1,5 +1,5 @@
 HunterGate(
-  URL  https://github.com/qdrvm/hunter/archive/refs/tags/v0.25.3-qdrvm9.zip
-  SHA1 7f3f8ee341aaac8c400e776c8a9f28e8fc458296
+  URL  https://github.com/qdrvm/hunter/archive/refs/tags/v0.25.3-qdrvm10.zip
+  SHA1 9571399d8d091420131eb81f884521326c9d3615
   LOCAL
 )

core/application/impl/kagome_application_impl.cpp

@@ -59,6 +59,7 @@ namespace kagome::application {
 
     kagome::telemetry::setTelemetryService(injector_.injectTelemetryService());
 
+    injector_.kademliaRandomWalk();
     injector_.injectAddressPublisher();
     injector_.injectTimeline();
 

core/dispute_coordinator/dispute_coordinator.hpp

@@ -11,6 +11,22 @@
 #include "dispute_coordinator/participation/types.hpp"
 #include "dispute_coordinator/types.hpp"
 
+/**
+* Disputes contains 2 main classes: dispute coordinator and participation.
+The main purpose of coordinator is to restore previous state once on
+startup(because of possible previous disputes on previous launch session), to
+initialize disputes based on approval result and to handle and collect
+statements from the whole validators set. Blocks with either active disputes or
+invalid are blocked for finalization, while valid disputes are allowed to be
+finalized. The main goal of participation class is to launch our validation
+process with predefined limits. It initializez recovery process followed by
+exhaustive validation and the result statement. Which will be imported the same
+way as statements from other nodes. So the disputes workflow is a loop which
+enters on disputes initialization either disputes request or approval result and
+works until enough votes collected to make a decision about block validity or
+another fork will be finalized.
+*/
+
 namespace kagome::dispute {
 
   class DisputeCoordinator {

core/dispute_coordinator/impl/dispute_coordinator_impl.cpp

@@ -2457,9 +2457,12 @@ namespace kagome::dispute {
     // auto &candidate_hash = signed_statement.candidate_hash;
 
     auto pending_confirmation =
-        [requesters(std::move(requesters))](outcome::result<void> res) {
-          for (auto &[peer, cb] : requesters) {
-            cb(res);
+        [wp{weak_from_this()},
+         requesters(std::move(requesters))](outcome::result<void> res) mutable {
+          if (auto self = wp.lock()) {
+            for (auto &[peer, cb] : requesters) {
+              self->sendDisputeResponse(res, std::move(cb));
+            }
           }
         };
 
@@ -2489,7 +2492,7 @@ namespace kagome::dispute {
     auto &[_, sending_dispute] = sending_disputes_.emplace_back(
         candidate_hash,
         std::make_unique<SendingDispute>(
-            authority_discovery_, protocol, request));
+            log_, main_pool_handler_, authority_discovery_, protocol, request));
 
     std::ignore =
         sending_dispute->refresh_sends(*runtime_info_, active_sessions_);

core/dispute_coordinator/impl/dispute_coordinator_impl.hpp

@@ -8,6 +8,7 @@
 
 #include "dispute_coordinator/dispute_coordinator.hpp"
 #include "network/dispute_request_observer.hpp"
+#include "network/types/dispute_messages.hpp"
 
 #include <list>
 
@@ -69,7 +70,6 @@ namespace kagome::dispute {
 }  // namespace kagome::dispute
 
 namespace kagome::network {
-  struct DisputeMessage;
   class Router;
   class PeerView;
 }  // namespace kagome::network

core/dispute_coordinator/impl/sending_dispute.cpp

@@ -8,16 +8,23 @@
 #include "authority_discovery/query/query.hpp"
 #include "dispute_coordinator/impl/runtime_info.hpp"
 #include "network/impl/protocols/send_dispute_protocol.hpp"
+#include "utils/pool_handler.hpp"
 
 namespace kagome::dispute {
 
   SendingDispute::SendingDispute(
+      log::Logger logger,
+      std::shared_ptr<PoolHandler> main_pool_handler,
       std::shared_ptr<authority_discovery::Query> authority_discovery,
       std::shared_ptr<network::SendDisputeProtocol> dispute_protocol,
       const network::DisputeMessage &request)
-      : authority_discovery_(std::move(authority_discovery)),
+      : logger_(std::move(logger)),
+        main_pool_handler_(std::move(main_pool_handler)),
+        authority_discovery_(std::move(authority_discovery)),
         dispute_protocol_(std::move(dispute_protocol)),
-        request_(std::move(request)) {
+        request_(request) {
+    BOOST_ASSERT(logger_ != nullptr);
+    BOOST_ASSERT(main_pool_handler_ != nullptr);
     BOOST_ASSERT(authority_discovery_ != nullptr);
     BOOST_ASSERT(dispute_protocol_.lock());
   }
@@ -51,15 +58,15 @@ namespace kagome::dispute {
 
     // Start any new tasks that are needed:
 
-    // SL_TRACE(log_,
-    //          "Starting new send requests for authorities. "
-    //          "(new_and_failed_authorities={},overall_authority_set_size={},"
-    //          "already_running_deliveries={})",
-    //          add_authorities.size(),
-    //          new_authorities.size(),
-    //          deliveries_.size());
+    SL_TRACE(logger_,
+             "Starting new send requests for authorities. "
+             "(new_and_failed_authorities={},overall_authority_set_size={},"
+             "already_running_deliveries={})",
+             add_authorities.size(),
+             new_authorities.size(),
+             deliveries_.size());
 
-    auto sent = send_requests(add_authorities);
+    auto sent = send_requests(std::move(add_authorities));
 
     return sent;
   }
@@ -104,7 +111,7 @@ namespace kagome::dispute {
   }
 
   bool SendingDispute::send_requests(
-      std::vector<primitives::AuthorityDiscoveryId> &authorities) {
+      std::vector<primitives::AuthorityDiscoveryId> &&authorities) {
     std::vector<
         std::tuple<primitives::AuthorityDiscoveryId, libp2p::peer::PeerId>>
         receivers;
@@ -115,7 +122,7 @@ namespace kagome::dispute {
     }
 
     if (receivers.empty()) {
-      // SL_WARN(log_, "No known peers to receive dispute request");
+      SL_WARN(logger_, "No known peers to receive dispute request");
       return false;
     }
 
@@ -127,28 +134,46 @@ namespace kagome::dispute {
 
     has_failed_sends_ = false;
 
+    auto size = receivers.size();
+
+    asyncSendRequests(std::move(protocol), std::move(receivers));
+
+    SL_TRACE(logger_, "Requests dispatched. Sent {} requests", size);
+
+    return true;
+  }
+
+  void SendingDispute::asyncSendRequests(
+      std::shared_ptr<network::SendDisputeProtocol> &&protocol,
+      std::vector<std::tuple<primitives::AuthorityDiscoveryId,
+                             libp2p::peer::PeerId>> &&receivers) {
+    REINVOKE(*main_pool_handler_,
+             asyncSendRequests,
+             std::move(protocol),
+             std::move(receivers));
+
     for (auto &[authority_id, peer_id] : receivers) {
       deliveries_.emplace(authority_id, DeliveryStatus::Pending);
       protocol->doRequest(
           peer_id,
           request_,
-          [wp{weak_from_this()}, authority_id(authority_id)](auto res) mutable {
+          [wp{weak_from_this()}, authority_id(authority_id), peer_id](
+              auto res) mutable {
             if (auto self = wp.lock()) {
               if (res.has_value()) {
                 self->deliveries_[authority_id] = DeliveryStatus::Succeeded;
               } else {
-                // LOG Can't sent dispute request to peer {}
+                SL_TRACE(self->logger_,
+                         "Can't send dispute request to peer {}: {}",
+                         peer_id,
+                         res.error());
+                // LOG
                 self->deliveries_.erase(authority_id);
                 self->has_failed_sends_ = true;
               }
             }
           });
     }
-
-    // SL_TRACE(log_, "Requests dispatched. Sent {} requests",
-    // receivers.size());
-
-    return true;
   }
 
 }  // namespace kagome::dispute

core/dispute_coordinator/impl/sending_dispute.hpp

@@ -9,6 +9,10 @@
 #include "dispute_coordinator/dispute_coordinator.hpp"
 #include "network/types/dispute_messages.hpp"
 
+namespace kagome {
+  class PoolHandler;
+}
+
 namespace kagome::authority_discovery {
   class Query;
 }
@@ -32,6 +36,8 @@ namespace kagome::dispute {
     /// employing a per-peer rate limit, we need to limit the construction of
     /// new `SendTask`s.
     SendingDispute(
+        log::Logger logger,
+        std::shared_ptr<PoolHandler> main_pool_handler,
         std::shared_ptr<authority_discovery::Query> authority_discovery,
         std::shared_ptr<network::SendDisputeProtocol> dispute_protocol,
         const network::DisputeMessage &request);
@@ -69,9 +75,18 @@ namespace kagome::dispute {
       Succeeded,
     };
 
+    /// Broadcasts dispute request to provided authorities
     bool send_requests(
-        std::vector<primitives::AuthorityDiscoveryId> &authorities);
+        std::vector<primitives::AuthorityDiscoveryId> &&authorities);
+
+    /// Asynchronous part of broadcasting dispute request
+    void asyncSendRequests(
+        std::shared_ptr<network::SendDisputeProtocol> &&protocol,
+        std::vector<std::tuple<primitives::AuthorityDiscoveryId,
+                               libp2p::peer::PeerId>> &&receivers);
 
+    log::Logger logger_;
+    std::shared_ptr<PoolHandler> main_pool_handler_;
     std::shared_ptr<authority_discovery::Query> authority_discovery_;
     std::weak_ptr<network::SendDisputeProtocol> dispute_protocol_;
 

core/injector/application_injector.cpp

@@ -133,6 +133,7 @@
 #include "network/impl/sync_protocol_observer_impl.hpp"
 #include "network/impl/synchronizer_impl.hpp"
 #include "network/impl/transactions_transmitter_impl.hpp"
+#include "network/kademlia_random_walk.hpp"
 #include "network/warp/cache.hpp"
 #include "network/warp/protocol.hpp"
 #include "network/warp/sync.hpp"
@@ -320,7 +321,7 @@ namespace {
     kademlia_config.protocols =
         network::make_protocols("/{}/kad", genesis, chain_spec);
     kademlia_config.maxBucketSize = 1000;
-    kademlia_config.randomWalk = {.interval = random_wak_interval};
+    kademlia_config.randomWalk.enabled = false;
 
     return std::make_shared<libp2p::protocol::kademlia::Config>(
         std::move(kademlia_config));
@@ -1106,4 +1107,7 @@ namespace kagome::injector {
     return pimpl_->injector_.template create<sptr<common::MainThreadPool>>();
   }
 
+  void KagomeNodeInjector::kademliaRandomWalk() {
+    pimpl_->injector_.create<sptr<KademliaRandomWalk>>();
+  }
 }  // namespace kagome::injector

core/injector/application_injector.hpp

@@ -145,6 +145,7 @@ namespace kagome::injector {
     std::shared_ptr<storage::SpacedStorage> injectStorage();
     std::shared_ptr<authority_discovery::AddressPublisher>
     injectAddressPublisher();
+    void kademliaRandomWalk();
 
     std::shared_ptr<application::mode::PrintChainInfoMode>
     injectPrintChainInfoMode();