Merge branch 'master' into docs/parachains

.github/workflows/telegram_notify.yml

@@ -0,0 +1,20 @@
+name: Telegram Notification
+
+on:
+  pull_request:
+    types: [created, closed, opened, reopened, synchronize]
+  pull_request_target:
+    types: [ created, closed, opened, reopened, synchronize ]
+  issues:
+    types: [created, closed, opended, reopened]
+
+jobs:
+  notify:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Notify the commit on Telegram
+        uses: EverythingSuckz/github-telegram-notify@main
+        with:
+          bot_token: ${{ secrets.TELEGRAM_BOT_TOKEN }}
+          chat_id: ${{ secrets.TELEGRAM_CHAT_ID }}

cmake/Hunter/config.cmake

@@ -27,26 +27,26 @@ hunter_config(
 
 hunter_config(
     benchmark
-    URL https://github.com/google/benchmark/archive/refs/tags/v1.7.1.zip
-    SHA1 988246a257b0eeb1a8b112cff6ab3edfbe162912
+    URL https://github.com/google/benchmark/archive/refs/tags/v1.8.3.zip
+    SHA1 bf9870756ee3f8d2d3b346b24ee3600a41c74d3d
     CMAKE_ARGS BENCHMARK_ENABLE_TESTING=OFF
 )
 
 hunter_config(
     soralog
-    VERSION 0.2.2
+    VERSION 0.2.3
     KEEP_PACKAGE_SOURCES
 )
 
 hunter_config(
     libp2p
-    VERSION 0.1.18
+    VERSION 0.1.19
     KEEP_PACKAGE_SOURCES
 )
 
 hunter_config(
     rocksdb
-    VERSION 8.0.0
+    VERSION 9.0.0
     CMAKE_ARGS WITH_GFLAGS=OFF
 )
 

cmake/Hunter/hunter-gate-url.cmake

@@ -1,5 +1,5 @@
 HunterGate(
-  URL  https://github.com/qdrvm/hunter/archive/refs/tags/v0.25.3-qdrvm5.zip
-  SHA1 ae75fdf1850bd21dead239b8580012b6091be855
+  URL  https://github.com/qdrvm/hunter/archive/refs/tags/v0.25.3-qdrvm7.zip
+  SHA1 be5869134ef7448fe2420d60dbb9706596b1b8bd
   LOCAL
 )

core/log/configurator.cpp

@@ -101,14 +101,15 @@ namespace kagome::log {
                   - name: warp_sync_protocol
                   - name: parachain_protocols
                     children:
-                      - name: collation_protocol
-                      - name: validation_protocol
+                      - name: collation_protocol_vstaging
+                      - name: validation_protocol_vstaging
                       - name: req_collation_protocol
                       - name: req_chunk_protocol
                       - name: req_available_data_protocol
                       - name: req_statement_protocol
                       - name: req_pov_protocol
                       - name: dispute_protocol
+                      - name: req_attested_candidate_protocol
           - name: changes_trie
           - name: storage
             children:

core/network/helpers/stream_proxy_base.hpp

@@ -30,9 +30,6 @@ namespace libp2p::connection {
       stream->deferReadCallback(res, std::move(cb));
     }
 
-    void write(BytesIn in, size_t bytes, WriteCallbackFunc cb) override {
-      stream->write(in, bytes, std::move(cb));
-    }
     void writeSome(BytesIn in, size_t bytes, WriteCallbackFunc cb) override {
       stream->writeSome(in, bytes, std::move(cb));
     }

core/network/helpers/stream_read_buffer.hpp

@@ -10,6 +10,10 @@
 
 #include <libp2p/basic/read_return_size.hpp>
 #include <libp2p/common/ambigous_size.hpp>
+#include <libp2p/connection/stream_and_protocol.hpp>
+#include <thread>
+
+#include "log/logger.hpp"
 
 namespace libp2p::connection {
   /**
@@ -99,73 +103,78 @@ namespace kagome::network {
     StreamWrapper(std::shared_ptr<libp2p::connection::StreamReadBuffer> stream)
         : stream_{std::move(stream)} {}
 
-    bool isClosedForRead() const {
+    bool isClosedForRead() const override {
       return stream_->isClosedForRead();
     }
 
-    bool isClosedForWrite() const {
+    bool isClosedForWrite() const override {
       return stream_->isClosedForWrite();
     }
 
-    bool isClosed() const {
+    bool isClosed() const override {
       return stream_->isClosed();
     }
 
-    void close(VoidResultHandlerFunc cb) {
+    void close(VoidResultHandlerFunc cb) override {
       check();
       stream_->close(std::move(cb));
     }
 
-    void reset() {
+    void reset() override {
       check();
       stream_->reset();
     }
 
-    void adjustWindowSize(uint32_t new_size, VoidResultHandlerFunc cb) {
+    void adjustWindowSize(uint32_t new_size,
+                          VoidResultHandlerFunc cb) override {
       stream_->adjustWindowSize(new_size, std::move(cb));
     }
 
-    outcome::result<bool> isInitiator() const {
+    outcome::result<bool> isInitiator() const override {
       return stream_->isInitiator();
     }
 
-    outcome::result<libp2p::peer::PeerId> remotePeerId() const {
+    outcome::result<libp2p::peer::PeerId> remotePeerId() const override {
       return stream_->remotePeerId();
     }
 
-    outcome::result<libp2p::multi::Multiaddress> localMultiaddr() const {
+    outcome::result<libp2p::multi::Multiaddress> localMultiaddr()
+        const override {
       return stream_->localMultiaddr();
     }
 
-    outcome::result<libp2p::multi::Multiaddress> remoteMultiaddr() const {
+    outcome::result<libp2p::multi::Multiaddress> remoteMultiaddr()
+        const override {
       return stream_->remoteMultiaddr();
     }
 
-    void read(libp2p::BytesOut out, size_t bytes, ReadCallbackFunc cb) {
+    void read(libp2p::BytesOut out,
+              size_t bytes,
+              ReadCallbackFunc cb) override {
       check();
       stream_->read(out, bytes, std::move(cb));
     }
 
-    void readSome(libp2p::BytesOut out, size_t bytes, ReadCallbackFunc cb) {
+    void readSome(libp2p::BytesOut out,
+                  size_t bytes,
+                  ReadCallbackFunc cb) override {
       check();
       stream_->readSome(out, bytes, std::move(cb));
     }
 
-    void deferReadCallback(outcome::result<size_t> res, ReadCallbackFunc cb) {
+    void deferReadCallback(outcome::result<size_t> res,
+                           ReadCallbackFunc cb) override {
       stream_->deferReadCallback(std::move(res), std::move(cb));
     }
 
-    void write(libp2p::BytesIn in, size_t bytes, WriteCallbackFunc cb) {
-      check();
-      stream_->write(in, bytes, std::move(cb));
-    }
-
-    void writeSome(libp2p::BytesIn in, size_t bytes, WriteCallbackFunc cb) {
+    void writeSome(libp2p::BytesIn in,
+                   size_t bytes,
+                   WriteCallbackFunc cb) override {
       check();
       stream_->writeSome(in, bytes, std::move(cb));
     }
 
-    void deferWriteCallback(std::error_code ec, WriteCallbackFunc cb) {
+    void deferWriteCallback(std::error_code ec, WriteCallbackFunc cb) override {
       stream_->deferWriteCallback(ec, std::move(cb));
     }
   };

core/network/impl/peer_manager_impl.cpp

@@ -581,6 +581,15 @@ namespace kagome::network {
     return it->second;
   }
 
+  std::optional<std::reference_wrapper<const PeerState>>
+  PeerManagerImpl::getPeerState(const PeerId &peer_id) const {
+    auto it = peer_states_.find(peer_id);
+    if (it == peer_states_.end()) {
+      return std::nullopt;
+    }
+    return it->second;
+  }
+
   void PeerManagerImpl::processDiscoveredPeer(const PeerId &peer_id) {
     // Ignore himself
     if (isSelfPeer(peer_id)) {
@@ -726,41 +735,29 @@ namespace kagome::network {
 
       log_->trace("Try to open outgoing validation protocol.(peer={})",
                   peer_info.id);
-      openOutgoing(
-          stream_engine_,
-          validation_protocol,
-          peer_info,
-          [validation_protocol, peer_info, wptr{weak_from_this()}](
-              outcome::result<std::shared_ptr<Stream>> stream_result) {
-            auto self = wptr.lock();
-            if (not self) {
-              return;
-            }
-
-            auto &peer_id = peer_info.id;
-            if (!stream_result.has_value()) {
-              SL_TRACE(self->log_,
-                       "Unable to create stream {} with {}: {}",
-                       validation_protocol->protocolName(),
-                       peer_id,
-                       stream_result.error().message());
-              auto ps = self->getPeerState(peer_info.id);
-              if (ps) {
-                self->tryOpenValidationProtocol(
-                    peer_info, ps->get(), network::CollationVersion::V1);
-              } else {
-                SL_TRACE(
-                    self->log_,
-                    "No peer state to open V1 validation protocol {} with {}",
-                    validation_protocol->protocolName(),
-                    peer_id);
-              }
-              return;
-            }
-
-            self->stream_engine_->addOutgoing(stream_result.value(),
-                                              validation_protocol);
-          });
+      openOutgoing(stream_engine_,
+                   validation_protocol,
+                   peer_info,
+                   [validation_protocol, peer_info, wptr{weak_from_this()}](
+                       outcome::result<std::shared_ptr<Stream>> stream_result) {
+                     auto self = wptr.lock();
+                     if (not self) {
+                       return;
+                     }
+
+                     auto &peer_id = peer_info.id;
+                     if (!stream_result.has_value()) {
+                       SL_TRACE(self->log_,
+                                "Unable to create stream {} with {}: {}",
+                                validation_protocol->protocolName(),
+                                peer_id,
+                                stream_result.error().message());
+                       return;
+                     }
+
+                     self->stream_engine_->addOutgoing(stream_result.value(),
+                                                       validation_protocol);
+                   });
     }
   }
 
@@ -839,11 +836,14 @@ namespace kagome::network {
   }
 
   void PeerManagerImpl::reserveStatusStreams(const PeerId &peer_id) const {
-    auto proto_val_vstaging = router_->getValidationProtocolVStaging();
-    BOOST_ASSERT_MSG(proto_val_vstaging,
-                     "Router did not provide validation protocol vstaging");
+    if (auto ps = getPeerState(peer_id);
+        ps && ps->get().roles.flags.authority) {
+      auto proto_val_vstaging = router_->getValidationProtocolVStaging();
+      BOOST_ASSERT_MSG(proto_val_vstaging,
+                       "Router did not provide validation protocol vstaging");
 
-    stream_engine_->reserveStreams(peer_id, proto_val_vstaging);
+      stream_engine_->reserveStreams(peer_id, proto_val_vstaging);
+    }
   }
 
   void PeerManagerImpl::reserveStreams(const PeerId &peer_id) const {

core/network/impl/peer_manager_impl.hpp

@@ -139,6 +139,8 @@ namespace kagome::network {
     /** @see PeerManager::getPeerState */
     std::optional<std::reference_wrapper<PeerState>> getPeerState(
         const PeerId &peer_id) override;
+    std::optional<std::reference_wrapper<const PeerState>> getPeerState(
+        const PeerId &peer_id) const override;
 
    private:
     /// Right way to check self peer as it takes into account dev mode

core/network/impl/protocols/protocol_base_impl.hpp

@@ -55,7 +55,8 @@ namespace kagome::network {
     bool start(std::weak_ptr<T> wp) {
       host_.setProtocolHandler(
           protocols_,
-          [wp = std::move(wp), log = logger()](auto &&stream_and_proto) {
+          [wp = std::move(wp),
+           log = logger()](libp2p::StreamAndProtocol stream_and_proto) {
             if (auto self = wp.lock()) {
               BOOST_ASSERT(stream_and_proto.stream);
 
@@ -68,7 +69,7 @@ namespace kagome::network {
                 SL_TRACE(log,
                          "Handled {} protocol stream from {}",
                          protocol,
-                         peer_id);
+                         peer_id.error());
                 BOOST_ASSERT(stream);
                 self->onIncomingStream(std::move(stream));
                 return;

core/network/impl/router_libp2p.cpp

@@ -128,8 +128,8 @@ namespace kagome::network {
     // lazyStart(collation_protocol_);
     // lazyStart(validation_protocol_);
 
-    lazyStart(collation_protocol_);
-    lazyStart(validation_protocol_);
+    lazyStart(collation_protocol_vstaging_);
+    lazyStart(validation_protocol_vstaging_);
     lazyStart(req_collation_protocol_);
     lazyStart(req_pov_protocol_);
     lazyStart(fetch_chunk_protocol_);

core/network/notifications/wait_read_close.hpp

@@ -13,12 +13,12 @@ namespace kagome::network::notifications {
 
   inline void waitReadClose(std::shared_ptr<Stream> stream) {
     auto buf = std::make_shared<std::vector<uint8_t>>(1);
-    auto cb = [stream, buf](libp2p::outcome::result<size_t> r) {
+    auto cb = [stream, buf](outcome::result<size_t> r) {
       if (r) {
         stream->reset();
         return;
       }
-      stream->close([](libp2p::outcome::result<void>) {});
+      stream->close([](outcome::result<void>) {});
     };
     stream->read(*buf, buf->size(), std::move(cb));
   }

core/network/peer_manager.hpp

@@ -203,6 +203,8 @@ namespace kagome::network {
      */
     virtual std::optional<std::reference_wrapper<PeerState>> getPeerState(
         const PeerId &peer_id) = 0;
+    virtual std::optional<std::reference_wrapper<const PeerState>> getPeerState(
+        const PeerId &peer_id) const = 0;
 
     /**
      * @returns number of active peers

core/outcome/into.hpp

@@ -12,13 +12,13 @@ namespace outcome {
   /// Wraps value or returns original result
   template <typename T>
   struct Into {
-    static outcome::result<T> into(T &&r) {
-      return outcome::success(std::move(r));
+    static result<T> into(T &&r) {
+      return success(std::move(r));
     }
   };
   template <typename T>
-  struct Into<outcome::result<T>> {
-    static outcome::result<T> into(outcome::result<T> &&r) {
+  struct Into<result<T>> {
+    static result<T> into(result<T> &&r) {
       return std::move(r);
     }
   };

core/outcome/outcome.hpp

@@ -7,9 +7,3 @@
 #pragma once
 
 #include <libp2p/outcome/outcome.hpp>
-
-namespace outcome {
-  using libp2p::outcome::failure;
-  using libp2p::outcome::result;
-  using libp2p::outcome::success;
-}  // namespace outcome