Merge branch 'refactor/notifications-skip' into 1948-ensure-parachain…

…s-data-is-temporary-stored-migration

.clang-tidy

@@ -160,8 +160,6 @@ CheckOptions:
     value:           '4294967295'
   - key:             hicpp-member-init.IgnoreArrays
     value:           '0'
-  - key:             hicpp-move-const-arg.CheckTriviallyCopyableMove
-    value:           '1'
   - key:             hicpp-multiway-paths-covered.WarnOnMissingElse
     value:           '0'
   - key:             hicpp-named-parameter.IgnoreFailedSplit
@@ -261,7 +259,7 @@ CheckOptions:
   - key:             performance-inefficient-vector-operation.VectorLikeClasses
     value:           '::std::vector'
   - key:             performance-move-const-arg.CheckTriviallyCopyableMove
-    value:           '1'
+    value:           '0'
   - key:             performance-move-constructor-init.IncludeStyle
     value:           google
   - key:             performance-type-promotion-in-math-fn.IncludeStyle

core/application/chain_spec.hpp

@@ -8,7 +8,7 @@
 
 #include <set>
 
-#include <libp2p/peer/peer_info.hpp>
+#include <libp2p/multi/multiaddress.hpp>
 
 #include "crypto/ed25519_types.hpp"
 #include "crypto/sr25519_types.hpp"
@@ -38,8 +38,8 @@ namespace kagome::application {
     virtual const std::vector<libp2p::multi::Multiaddress> &bootNodes()
         const = 0;
 
-    virtual const std::vector<std::pair<std::string, size_t>> &
-    telemetryEndpoints() const = 0;
+    virtual const std::vector<std::pair<std::string, size_t>>
+        &telemetryEndpoints() const = 0;
 
     virtual const std::string &protocolId() const = 0;
 

core/application/impl/chain_spec_impl.cpp

@@ -9,6 +9,7 @@
 #include <boost/property_tree/json_parser.hpp>
 #include <charconv>
 #include <libp2p/multi/multiaddress.hpp>
+#include <libp2p/peer/peer_id.hpp>
 #include <sstream>
 #include <system_error>
 

core/authority_discovery/query/query_impl.cpp

@@ -10,6 +10,7 @@
 #include "common/buffer_view.hpp"
 #include "common/bytestr.hpp"
 #include "crypto/sha/sha256.hpp"
+#include "network/impl/protocols/parachain.hpp"
 #include "utils/retain_if.hpp"
 
 OUTCOME_CPP_DEFINE_CATEGORY(kagome::authority_discovery, QueryImpl::Error, e) {
@@ -38,6 +39,7 @@ namespace kagome::authority_discovery {
       std::shared_ptr<application::AppStateManager> app_state_manager,
       std::shared_ptr<blockchain::BlockTree> block_tree,
       std::shared_ptr<runtime::AuthorityDiscoveryApi> authority_discovery_api,
+      LazySPtr<network::ValidationProtocol> validation_protocol,
       std::shared_ptr<crypto::KeyStore> key_store,
       std::shared_ptr<crypto::Sr25519Provider> sr_crypto_provider,
       std::shared_ptr<libp2p::crypto::CryptoProvider> libp2p_crypto_provider,
@@ -47,6 +49,7 @@ namespace kagome::authority_discovery {
       std::shared_ptr<libp2p::basic::Scheduler> scheduler)
       : block_tree_{std::move(block_tree)},
         authority_discovery_api_{std::move(authority_discovery_api)},
+        validation_protocol_{std::move(validation_protocol)},
         key_store_{std::move(key_store)},
         sr_crypto_provider_{std::move(sr_crypto_provider)},
         libp2p_crypto_provider_{std::move(libp2p_crypto_provider)},
@@ -158,6 +161,7 @@ namespace kagome::authority_discovery {
         ++it;
       } else {
         it = auth_to_peer_cache_.erase(it);
+        validation_protocol_.get()->reserve(it->second.peer.id, false);
       }
     }
     for (auto it = peer_to_auth_cache_.begin();
@@ -166,6 +170,7 @@ namespace kagome::authority_discovery {
         ++it;
       } else {
         it = peer_to_auth_cache_.erase(it);
+        validation_protocol_.get()->reserve(it->first, false);
       }
     }
     std::shuffle(authorities.begin(), authorities.end(), random_);
@@ -194,11 +199,24 @@ namespace kagome::authority_discovery {
       auto authority = queue_.back();
       queue_.pop_back();
 
+      static size_t skipped = 0;
+      if (auth_to_peer_cache_.contains(authority)) {
+        ++skipped;
+        if (skipped % 10 == 0) {
+          SL_DEBUG(log_, "skipped {} records", skipped);
+        } else {
+          --active_;
+          continue;
+        }
+      }
+      skipped = 0;
+
       scheduler_->schedule([wp{weak_from_this()},
                             hash = common::Buffer{crypto::sha256(authority)},
                             authority] {
         if (auto self = wp.lock()) {
-          SL_DEBUG(self->log_, "start lookup({})", common::hex_lower(authority));
+          SL_DEBUG(
+              self->log_, "start lookup({})", common::hex_lower(authority));
           std::ignore = self->kademlia_.get()->getValue(
               hash, [=](const outcome::result<std::vector<uint8_t>> &res) {
                 if (auto self = wp.lock()) {
@@ -216,9 +234,9 @@ namespace kagome::authority_discovery {
       const primitives::AuthorityDiscoveryId &authority,
       outcome::result<std::vector<uint8_t>> _res) {
     SL_TRACE(log_,
-            "lookup : add addresses for authority {}, _res {}",
-            common::hex_lower(authority),
-            _res.has_value() ? "ok" : "error: " + _res.error().message());
+             "lookup : add addresses for authority {}, _res {}",
+             common::hex_lower(authority),
+             _res.has_value() ? "ok" : "error: " + _res.error().message());
     OUTCOME_TRY(signed_record_pb, _res);
     auto it = auth_to_peer_cache_.find(authority);
     if (it != auth_to_peer_cache_.end()
@@ -271,9 +289,9 @@ namespace kagome::authority_discovery {
     libp2p::peer::PeerInfo peer{.id = std::move(peer_id)};
     auto peer_id_str = peer.id.toBase58();
     SL_TRACE(log_,
-            "lookup: adding {} addresses for authority {}",
-            record.addresses().size(),
-            authority);
+             "lookup: adding {} addresses for authority {}",
+             record.addresses().size(),
+             authority);
     for (auto &pb : record.addresses()) {
       OUTCOME_TRY(address, libp2p::multi::Multiaddress::create(str2byte(pb)));
       auto id = address.getPeerId();
@@ -316,9 +334,11 @@ namespace kagome::authority_discovery {
                                          Authority{
                                              .raw = std::move(signed_record_pb),
                                              .time = time,
-                                             .peer = std::move(peer),
+                                             .peer = peer,
                                          });
 
+    validation_protocol_.get()->reserve(peer.id, true);
+
     return outcome::success();
   }
 }  // namespace kagome::authority_discovery

core/authority_discovery/query/query_impl.hpp

@@ -26,6 +26,10 @@
 #include <mutex>
 #include <random>
 
+namespace kagome::network {
+  class ValidationProtocol;
+}  // namespace kagome::network
+
 namespace kagome::authority_discovery {
   class QueryImpl : public Query,
                     public libp2p::protocol::kademlia::Validator,
@@ -43,6 +47,7 @@ namespace kagome::authority_discovery {
         std::shared_ptr<application::AppStateManager> app_state_manager,
         std::shared_ptr<blockchain::BlockTree> block_tree,
         std::shared_ptr<runtime::AuthorityDiscoveryApi> authority_discovery_api,
+        LazySPtr<network::ValidationProtocol> validation_protocol,
         std::shared_ptr<crypto::KeyStore> key_store,
         std::shared_ptr<crypto::Sr25519Provider> sr_crypto_provider,
         std::shared_ptr<libp2p::crypto::CryptoProvider> libp2p_crypto_provider,
@@ -85,6 +90,7 @@ namespace kagome::authority_discovery {
 
     std::shared_ptr<blockchain::BlockTree> block_tree_;
     std::shared_ptr<runtime::AuthorityDiscoveryApi> authority_discovery_api_;
+    LazySPtr<network::ValidationProtocol> validation_protocol_;
     std::shared_ptr<crypto::KeyStore> key_store_;
     std::shared_ptr<crypto::Sr25519Provider> sr_crypto_provider_;
     std::shared_ptr<libp2p::crypto::CryptoProvider> libp2p_crypto_provider_;

core/blockchain/genesis_block_hash.hpp

@@ -6,7 +6,7 @@
 
 #pragma once
 
-#include "blockchain/block_tree.hpp"
+#include "primitives/common.hpp"
 
 namespace kagome::blockchain {
 

core/consensus/babe/impl/babe.cpp

@@ -35,7 +35,7 @@
 #include "offchain/offchain_worker_pool.hpp"
 #include "parachain/availability/bitfield/store.hpp"
 #include "parachain/parachain_inherent_data.hpp"
-#include "parachain/validator/parachain_processor.hpp"
+#include "parachain/validator/backed_candidates_source.hpp"
 #include "primitives/inherent_data.hpp"
 #include "runtime/runtime_api/babe_api.hpp"
 #include "runtime/runtime_api/offchain_worker_api.hpp"

core/consensus/babe/impl/babe.hpp

@@ -71,7 +71,7 @@ namespace kagome::offchain {
 
 namespace kagome::parachain {
   class BitfieldStore;
-  struct ParachainProcessorImpl;
+  class ParachainProcessorImpl;
   struct BackedCandidatesSource;
 }  // namespace kagome::parachain
 

core/consensus/babe/impl/babe_block_validator_impl.cpp

@@ -24,6 +24,7 @@
 #include "runtime/runtime_api/babe_api.hpp"
 #include "runtime/runtime_api/offchain_worker_api.hpp"
 #include "threshold_util.hpp"
+#include "utils/weak_macro.hpp"
 
 OUTCOME_CPP_DEFINE_CATEGORY(kagome::consensus::babe,
                             BabeBlockValidatorImpl::ValidationError,
@@ -77,22 +78,9 @@ namespace kagome::consensus::babe {
 
   void BabeBlockValidatorImpl::prepare() {
     sync_state_observer_ =
-        std::make_shared<primitives::events::SyncStateEventSubscriber>(
-            sync_state_observable_, false);
-    sync_state_observer_->subscribe(
-        sync_state_observer_->generateSubscriptionSetId(),
-        primitives::events::SyncStateEventType::kSyncState);
-    sync_state_observer_->setCallback(
-        [wp{weak_from_this()}](
-            auto /*set_id*/,
-            bool &synchronized,
-            auto /*event_type*/,
-            const primitives::events::SyncStateEventParams &event) mutable {
-          if (auto self = wp.lock()) {
-            if (event == consensus::SyncState::SYNCHRONIZED) {
-              self->was_synchronized_ = true;
-            }
-          }
+        primitives::events::onSync(sync_state_observable_, [WEAK_SELF] {
+          WEAK_LOCK(self);
+          self->was_synchronized_ = true;
         });
   }
 

core/consensus/babe/impl/babe_block_validator_impl.hpp

@@ -126,7 +126,7 @@ namespace kagome::consensus::babe {
     primitives::events::SyncStateSubscriptionEnginePtr sync_state_observable_;
 
     bool was_synchronized_ = false;
-    primitives::events::SyncStateEventSubscriberPtr sync_state_observer_;
+    std::shared_ptr<void> sync_state_observer_;
   };
 
 }  // namespace kagome::consensus::babe

core/consensus/beefy/impl/beefy_impl.cpp

@@ -38,18 +38,14 @@
 namespace kagome::network {
   constexpr std::chrono::minutes kRebroadcastAfter{1};
 
-  namespace {
-    // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
-    metrics::GaugeHelper metric_validator_set_id{
-        "kagome_beefy_validator_set_id",
-        "Current BEEFY active validator set id.",
-    };
-    // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
-    metrics::GaugeHelper metric_finalized{
-        "kagome_beefy_best_block",
-        "Best block finalized by BEEFY",
-    };
-  }  // namespace
+  static const metrics::GaugeHelper metric_validator_set_id{
+      "kagome_beefy_validator_set_id",
+      "Current BEEFY active validator set id.",
+  };
+  static const metrics::GaugeHelper metric_finalized{
+      "kagome_beefy_best_block",
+      "Best block finalized by BEEFY",
+  };
 
   BeefyImpl::BeefyImpl(
       std::shared_ptr<application::AppStateManager> app_state_manager,

core/dispute_coordinator/impl/dispute_coordinator_impl.cpp

@@ -15,6 +15,7 @@
 #include <libp2p/basic/scheduler/scheduler_impl.hpp>
 
 #include "application/app_state_manager.hpp"
+#include "application/chain_spec.hpp"
 #include "authority_discovery/query/query.hpp"
 #include "blockchain/block_header_repository.hpp"
 #include "common/main_thread_pool.hpp"
@@ -35,6 +36,7 @@
 #include "runtime/runtime_api/parachain_host.hpp"
 #include "utils/pool_handler_ready_make.hpp"
 #include "utils/tuple_hash.hpp"
+#include "utils/weak_macro.hpp"
 
 namespace kagome::dispute {
 
@@ -238,15 +240,12 @@ namespace kagome::dispute {
     active_heads_.insert(leaves.begin(), leaves.end());
 
     // subscribe to leaves update
-    my_view_sub_ = std::make_shared<network::PeerView::MyViewSubscriber>(
-        peer_view_->getMyViewObservable(), false);
-    primitives::events::subscribe(
-        *my_view_sub_,
+    my_view_sub_ = primitives::events::subscribe(
+        peer_view_->getMyViewObservable(),
         network::PeerView::EventType::kViewUpdated,
-        [wptr{weak_from_this()}](const network::ExView &event) {
-          if (auto self = wptr.lock()) {
-            self->on_active_leaves_update(event);
-          }
+        [WEAK_SELF](const network::ExView &event) {
+          WEAK_LOCK(self);
+          self->on_active_leaves_update(event);
         });
 
     // subscribe to finalization

core/dispute_coordinator/impl/dispute_coordinator_impl.hpp

@@ -72,7 +72,7 @@ namespace kagome::network {
 }  // namespace kagome::network
 
 namespace kagome::parachain {
-  struct ApprovalDistribution;
+  class ApprovalDistribution;
   class Recovery;
   class Pvf;
 }  // namespace kagome::parachain

core/host_api/host_api.hpp

@@ -413,7 +413,7 @@ namespace kagome::host_api {
     [[nodiscard]] virtual runtime::WasmSpan
     ext_crypto_ecdsa_sign_prehashed_version_1(runtime::WasmSize key_type,
                                               runtime::WasmPointer key,
-                                              runtime::WasmSpan msg_data) = 0;
+                                              runtime::WasmPointer msg_data) = 0;
 
     /**
      * @brief Generates an ecdsa key for the given key type using an optional

core/host_api/impl/crypto_extension.cpp

@@ -616,14 +616,14 @@ namespace kagome::host_api {
   runtime::WasmSpan CryptoExtension::ext_crypto_ecdsa_sign_prehashed_version_1(
       runtime::WasmPointer key_type_ptr,
       runtime::WasmPointer key,
-      runtime::WasmSpan msg) {
+      runtime::WasmPointer msg_data) {
     using ResultType = std::optional<crypto::EcdsaSignature>;
 
     crypto::KeyType key_type = loadKeyType(key_type_ptr);
     checkIfKeyIsSupported(key_type, logger_);
 
     auto public_buffer = getMemory().loadN(key, sizeof(crypto::EcdsaPublicKey));
-    auto [msg_data, msg_len] = runtime::PtrSize(msg);
+    runtime::WasmSize msg_len = 32;
     auto msg_buffer = getMemory().loadN(msg_data, msg_len);
 
     crypto::EcdsaPublicKey pk;

core/host_api/impl/crypto_extension.hpp

@@ -213,7 +213,7 @@ namespace kagome::host_api {
     runtime::WasmSpan ext_crypto_ecdsa_sign_prehashed_version_1(
         runtime::WasmPointer key_type,
         runtime::WasmPointer key,
-        runtime::WasmSpan msg);
+        runtime::WasmPointer msg);
 
     /**
      * @see HostApi::ext_crypto_ecdsa_generate

core/host_api/impl/host_api_impl.cpp

@@ -341,7 +341,7 @@ namespace kagome::host_api {
   runtime::WasmSpan HostApiImpl::ext_crypto_ecdsa_sign_prehashed_version_1(
       runtime::WasmSize key_type,
       runtime::WasmPointer key,
-      runtime::WasmSpan msg_data) {
+      runtime::WasmPointer msg_data) {
     return crypto_ext_.ext_crypto_ecdsa_sign_prehashed_version_1(
         key_type, key, msg_data);
   }

core/host_api/impl/host_api_impl.hpp

@@ -190,7 +190,7 @@ namespace kagome::host_api {
     runtime::WasmSpan ext_crypto_ecdsa_sign_prehashed_version_1(
         runtime::WasmSize key_type,
         runtime::WasmPointer key,
-        runtime::WasmSpan msg_data) override;
+        runtime::WasmPointer msg_data) override;
 
     runtime::WasmPointer ext_crypto_ecdsa_generate_version_1(
         runtime::WasmSize key_type_id, runtime::WasmSpan seed) override;